Refactors and bug fixes around TableProvider registration and

DataFrame→TableProvider conversion, plus tests and FFI/pycapsule
improvements.

-- Registration logic & API

* Refactor of table provider registration logic for improved clarity and
  simpler call sites.
* Remove PyTableProvider registration from an internal module (reduces
  surprising side effects).
* Update table registration method to call `register_table` instead of
  `register_table_provider`.
* Extend `register_table` to support `TableProviderExportable` so more
  provider types can be registered uniformly.
* Improve error messages related to registration failures (missing
  PyCapsule name and DataFrame registration errors).

-- DataFrame ↔ TableProvider conversions

* Introduce utility functions to simplify table provider conversions and
  centralize conversion logic.
* Rename `into_view_provider` → `to_view_provider` for clearer intent.
* Fix `from_dataframe` to return the correct type and update
  `DataFrame.into_view` to import the correct `TableProvider`.
* Remove an obsolete `dataframe_into_view` test case after the refactor.

-- FFI / PyCapsule handling

* Update `FFI_TableProvider` initialization to accept an optional
  parameter (improves FFI ergonomics).
* Introduce `table_provider_from_pycapsule` utility to standardize
  pycapsule-based construction.
* Improve the error message when a PyCapsule name is missing to help
  debugging.

-- DeltaTable & specific integrations

* Update TableProvider registration for `DeltaTable` to use the correct
  registration method (matches the new API surface).

-- Tests, docs & minor fixes

* Add tests for registering a `TableProvider` from a `DataFrame` and
  from a capsule to ensure conversion paths are covered.
* Fix a typo in the `register_view` docstring and another typo in the
  error message for unsupported volatility type.
* Simplify version retrieval by removing exception handling around
  `PackageNotFoundError` (streamlines code path).

Author: kosiew

docs/source/user-guide/data-sources.rst

@@ -155,7 +155,7 @@ as Delta Lake. This will require a recent version of
     from datafusion import TableProvider
 
     delta_table = DeltaTable("path_to_table")
-    provider = TableProvider.from_capsule(delta_table)
+    provider = TableProvider.from_capsule(delta_table.__datafusion_table_provider__())
     ctx.register_table("my_delta_table", provider)
     df = ctx.table("my_delta_table")
     df.show()

docs/source/user-guide/io/table_provider.rst

@@ -39,9 +39,8 @@ A complete example can be found in the `examples folder <https://github.com/apac
         ) -> PyResult<Bound<'py, PyCapsule>> {
             let name = CString::new("datafusion_table_provider").unwrap();
 
-            let provider = Arc::new(self.clone())
-                .map_err(|e| PyRuntimeError::new_err(e.to_string()))?;
-            let provider = FFI_TableProvider::new(Arc::new(provider), false);
+            let provider = Arc::new(self.clone());
+            let provider = FFI_TableProvider::new(provider, false, None);
 
             PyCapsule::new_bound(py, provider, Some(name.clone()))
         }
@@ -50,8 +49,10 @@ A complete example can be found in the `examples folder <https://github.com/apac
 Once you have this library available, you can construct a
 :py:class:`~datafusion.TableProvider` in Python and register it with the
 ``SessionContext``.  Table providers can be created either from the PyCapsule exposed by
-your Rust provider or from an existing :py:class:`~datafusion.dataframe.DataFrame`
-using ``TableProvider.from_view()``.
+your Rust provider or from an existing :py:class:`~datafusion.dataframe.DataFrame`.
+Call the provider's ``__datafusion_table_provider__()`` method to obtain the capsule
+before constructing a ``TableProvider``. The ``TableProvider.from_view()`` helper is
+deprecated; instead use ``TableProvider.from_dataframe()`` or ``DataFrame.into_view()``.
 
 .. code-block:: python
 
@@ -60,16 +61,18 @@ using ``TableProvider.from_view()``.
     ctx = SessionContext()
     provider = MyTableProvider()
 
-    capsule_provider = TableProvider.from_capsule(provider)
+    capsule = provider.__datafusion_table_provider__()
+    capsule_provider = TableProvider.from_capsule(capsule)
 
     df = ctx.from_pydict({"a": [1]})
-    view_provider = TableProvider.from_view(df)
+    view_provider = TableProvider.from_dataframe(df)
+    # or: view_provider = df.into_view()
 
     ctx.register_table("capsule_table", capsule_provider)
     ctx.register_table("view_table", view_provider)
 
     ctx.table("capsule_table").show()
     ctx.table("view_table").show()
 
-Both ``TableProvider.from_capsule()`` and ``TableProvider.from_view()`` create
+Both ``TableProvider.from_capsule()`` and ``TableProvider.from_dataframe()`` create
 table providers that can be registered with the SessionContext using ``register_table()``.

examples/datafusion-ffi-example/python/tests/_test_table_function.py

@@ -53,7 +53,7 @@ def test_ffi_table_function_call_directly():
     table_udtf = udtf(table_func, "my_table_func")
 
     my_table = table_udtf()
-    ctx.register_table_provider("t", my_table)
+    ctx.register_table("t", my_table)
     result = ctx.table("t").collect()
 
     assert len(result) == 2

examples/datafusion-ffi-example/python/tests/_test_table_provider.py

@@ -18,14 +18,16 @@
 from __future__ import annotations
 
 import pyarrow as pa
-from datafusion import SessionContext
+from datafusion import SessionContext, TableProvider
 from datafusion_ffi_example import MyTableProvider
 
 
 def test_table_loading():
     ctx = SessionContext()
     table = MyTableProvider(3, 2, 4)
-    ctx.register_table_provider("t", table)
+    ctx.register_table(
+        "t", TableProvider.from_capsule(table.__datafusion_table_provider__())
+    )
     result = ctx.table("t").collect()
 
     assert len(result) == 4

python/datafusion/catalog.py

@@ -28,6 +28,7 @@
     import pyarrow as pa
 
     from datafusion import TableProvider
+    from datafusion.context import TableProviderExportable
 
 try:
     from warnings import deprecated  # Python 3.13+
@@ -124,8 +125,14 @@ def table(self, name: str) -> Table:
         """Return the table with the given ``name`` from this schema."""
         return Table(self._raw_schema.table(name))
 
-    def register_table(self, name, table: Table | TableProvider) -> None:
-        """Register a table or table provider in this schema."""
+    def register_table(
+        self, name, table: Table | TableProvider | TableProviderExportable
+    ) -> None:
+        """Register a table or table provider in this schema.
+
+        Objects implementing ``__datafusion_table_provider__`` are also supported
+        and treated as :class:`TableProvider` instances.
+        """
         if isinstance(table, Table):
             return self._raw_schema.register_table(name, table.table)
         return self._raw_schema.register_table(name, table)
@@ -221,11 +228,16 @@ def table(self, name: str) -> Table | None:
         """Retrieve a specific table from this schema."""
         ...
 
-    def register_table(self, name: str, table: Table | TableProvider) -> None:  # noqa: B027
+    def register_table(  # noqa: B027
+        self, name: str, table: Table | TableProvider | TableProviderExportable
+    ) -> None:
         """Add a table to this schema.
 
         This method is optional. If your schema provides a fixed list of tables, you do
         not need to implement this method.
+
+        Objects implementing ``__datafusion_table_provider__`` are also supported
+        and treated as :class:`TableProvider` instances.
         """
 
     def deregister_table(self, name, cascade: bool) -> None:  # noqa: B027

python/datafusion/context.py

@@ -735,7 +735,7 @@ def from_polars(self, data: pl.DataFrame, name: str | None = None) -> DataFrame:
     # https://github.com/apache/datafusion-python/pull/1016#discussion_r1983239116
     # is the discussion on how we arrived at adding register_view
     def register_view(self, name: str, df: DataFrame) -> None:
-        """Register a :py:class: `~datafusion.detaframe.DataFrame` as a view.
+        """Register a :py:class:`~datafusion.dataframe.DataFrame` as a view.
 
         Args:
             name (str): The name to register the view under.
@@ -744,16 +744,26 @@ def register_view(self, name: str, df: DataFrame) -> None:
         view = df.into_view()
         self.ctx.register_table(name, view)
 
-    def register_table(self, name: str, table: Table | TableProvider) -> None:
+    def register_table(
+        self, name: str, table: Table | TableProvider | TableProviderExportable
+    ) -> None:
         """Register a :py:class:`~datafusion.catalog.Table` or ``TableProvider``.
 
         The registered table can be referenced from SQL statements executed against
         this context.
 
+        Plain :py:class:`~datafusion.dataframe.DataFrame` objects are not supported;
+        convert them first with :meth:`datafusion.dataframe.DataFrame.into_view` or
+        :meth:`datafusion.catalog.TableProvider.from_dataframe`.
+
+        Objects implementing ``__datafusion_table_provider__`` are also supported
+        and treated as :class:`~datafusion.catalog.TableProvider` instances.
+
         Args:
             name: Name of the resultant table.
-            table: DataFusion :class:`Table` or :class:`TableProvider` to add to the
-                session context.
+            table: DataFusion :class:`Table`, :class:`TableProvider`, or any object
+                implementing ``__datafusion_table_provider__`` to add to the session
+                context.
         """
         if isinstance(table, Table):
             self.ctx.register_table(name, table.table)
@@ -778,11 +788,14 @@ def register_catalog_provider(
             self.ctx.register_catalog_provider(name, provider)
 
     def register_table_provider(
-        self, name: str, provider: TableProviderExportable | TableProvider
+        self, name: str, provider: Table | TableProvider | TableProviderExportable
     ) -> None:
         """Register a table provider.
 
         Deprecated: use :meth:`register_table` instead.
+
+        Objects implementing ``__datafusion_table_provider__`` are also supported
+        and treated as :class:`~datafusion.catalog.TableProvider` instances.
         """
         warnings.warn(
             "register_table_provider is deprecated; use register_table",

python/datafusion/dataframe.py

@@ -40,7 +40,6 @@
 from datafusion._internal import DataFrame as DataFrameInternal
 from datafusion._internal import ParquetColumnOptions as ParquetColumnOptionsInternal
 from datafusion._internal import ParquetWriterOptions as ParquetWriterOptionsInternal
-from datafusion._internal import TableProvider as TableProviderInternal
 from datafusion.expr import Expr, SortExpr, sort_or_default
 from datafusion.plan import ExecutionPlan, LogicalPlan
 from datafusion.record_batch import RecordBatchStream
@@ -53,6 +52,7 @@
     import polars as pl
     import pyarrow as pa
 
+    from datafusion._internal import TableProvider
     from datafusion._internal import expr as expr_internal
 
 from enum import Enum
@@ -308,7 +308,7 @@ def __init__(self, df: DataFrameInternal) -> None:
         """
         self.df = df
 
-    def into_view(self) -> TableProviderInternal:
+    def into_view(self) -> TableProvider:
         """Convert ``DataFrame`` into a ``TableProvider`` view for registration.
 
         This is the preferred way to obtain a view for

python/tests/test_context.py

@@ -349,16 +349,45 @@ def test_table_provider_from_capsule(ctx):
     assert [b.to_pydict() for b in result] == [{"a": [1, 2]}]
 
 
+def test_table_provider_from_dataframe(ctx):
+    df = ctx.from_pydict({"a": [1, 2]}).df
+    provider = TableProvider.from_dataframe(df)
+    ctx.register_table("from_dataframe_tbl", provider)
+    result = ctx.sql("SELECT * FROM from_dataframe_tbl").collect()
+    assert [b.to_pydict() for b in result] == [{"a": [1, 2]}]
+
+
+def test_register_table_capsule_direct(ctx):
+    df = ctx.from_pydict({"a": [1, 2]})
+    provider = df.into_view()
+
+    class CapsuleProvider:
+        def __init__(self, inner):
+            self._inner = inner
+
+        def __datafusion_table_provider__(self):
+            return self._inner.__datafusion_table_provider__()
+
+    ctx.register_table("capsule_direct_tbl", CapsuleProvider(provider))
+    result = ctx.sql("SELECT * FROM capsule_direct_tbl").collect()
+    assert [b.to_pydict() for b in result] == [{"a": [1, 2]}]
+
+
 def test_table_provider_from_capsule_invalid():
     with pytest.raises(Exception):  # noqa: B017
         TableProvider.from_capsule(object())
 
 
 def test_register_table_with_dataframe_errors(ctx):
     df = ctx.from_pydict({"a": [1]})
-    with pytest.raises(Exception):  # noqa: B017
+    with pytest.raises(Exception) as exc_info:  # noqa: B017
         ctx.register_table("bad", df)
 
+    assert (
+        str(exc_info.value)
+        == 'Expected a Table or TableProvider. Convert DataFrames with "DataFrame.into_view()" or "TableProvider.from_dataframe()".'
+    )
+
 
 def test_register_dataset(ctx):
     # create a RecordBatch and register it as a pyarrow.dataset.Dataset

src/catalog.rs

@@ -18,7 +18,10 @@
 use crate::dataset::Dataset;
 use crate::errors::{py_datafusion_err, to_datafusion_err, PyDataFusionError, PyDataFusionResult};
 use crate::table::PyTableProvider;
-use crate::utils::{validate_pycapsule, wait_for_future};
+use crate::utils::{
+    table_provider_from_pycapsule, table_provider_send_to_table_provider, table_provider_to_send,
+    validate_pycapsule, wait_for_future,
+};
 use async_trait::async_trait;
 use datafusion::catalog::{MemoryCatalogProvider, MemorySchemaProvider};
 use datafusion::common::DataFusionError;
@@ -28,7 +31,6 @@ use datafusion::{
     datasource::{TableProvider, TableType},
 };
 use datafusion_ffi::schema_provider::{FFI_SchemaProvider, ForeignSchemaProvider};
-use datafusion_ffi::table_provider::{FFI_TableProvider, ForeignTableProvider};
 use pyo3::exceptions::PyKeyError;
 use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
@@ -197,28 +199,16 @@ impl PySchema {
     }
 
     fn register_table(&self, name: &str, table_provider: Bound<'_, PyAny>) -> PyResult<()> {
-        let provider = if table_provider.hasattr("__datafusion_table_provider__")? {
-            let capsule = table_provider
-                .getattr("__datafusion_table_provider__")?
-                .call0()?;
-            let capsule = capsule.downcast::<PyCapsule>().map_err(py_datafusion_err)?;
-            validate_pycapsule(capsule, "datafusion_table_provider")?;
-
-            let provider = unsafe { capsule.reference::<FFI_TableProvider>() };
-            let provider: ForeignTableProvider = provider.into();
-            Arc::new(provider) as Arc<dyn TableProvider + Send>
+        let provider = if let Ok(py_table) = table_provider.extract::<PyTable>() {
+            py_table.table
+        } else if let Ok(py_provider) = table_provider.extract::<PyTableProvider>() {
+            py_provider.into_inner()
+        } else if let Some(provider) = table_provider_from_pycapsule(&table_provider)? {
+            provider
         } else {
-            match table_provider.extract::<PyTable>() {
-                Ok(py_table) => py_table.table,
-                Err(_) => match table_provider.extract::<PyTableProvider>() {
-                    Ok(py_provider) => py_provider.into_inner(),
-                    Err(_) => {
-                        let py = table_provider.py();
-                        let provider = Dataset::new(&table_provider, py)?;
-                        Arc::new(provider) as Arc<dyn TableProvider + Send>
-                    }
-                },
-            }
+            let py = table_provider.py();
+            let provider = Dataset::new(&table_provider, py)?;
+            Arc::new(provider) as Arc<dyn TableProvider + Send>
         };
 
         let _ = self
@@ -308,15 +298,8 @@ impl RustWrappedPySchemaProvider {
                 return Ok(None);
             }
 
-            if py_table.hasattr("__datafusion_table_provider__")? {
-                let capsule = py_table.getattr("__datafusion_table_provider__")?.call0()?;
-                let capsule = capsule.downcast::<PyCapsule>().map_err(py_datafusion_err)?;
-                validate_pycapsule(capsule, "datafusion_table_provider")?;
-
-                let provider = unsafe { capsule.reference::<FFI_TableProvider>() };
-                let provider: ForeignTableProvider = provider.into();
-
-                Ok(Some(Arc::new(provider) as Arc<dyn TableProvider + Send>))
+            if let Some(provider) = table_provider_from_pycapsule(&py_table)? {
+                Ok(Some(provider))
             } else {
                 if let Ok(inner_table) = py_table.getattr("table") {
                     if let Ok(inner_table) = inner_table.extract::<PyTable>() {
@@ -370,13 +353,7 @@ impl SchemaProvider for RustWrappedPySchemaProvider {
     ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>, DataFusionError> {
         // Convert from our internal Send type to the trait expected type
         match self.table_inner(name).map_err(to_datafusion_err)? {
-            Some(table) => {
-                // Safe conversion: we're widening the bounds (removing Send)
-                let raw = Arc::into_raw(table);
-                let wide: *const dyn TableProvider = raw as *const _;
-                let arc = unsafe { Arc::from_raw(wide) };
-                Ok(Some(arc))
-            }
+            Some(table) => Ok(Some(table_provider_send_to_table_provider(table))),
             None => Ok(None),
         }
     }
@@ -387,11 +364,7 @@ impl SchemaProvider for RustWrappedPySchemaProvider {
         table: Arc<dyn TableProvider>,
     ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
         // Convert from trait type to our internal Send type
-        let send_table = {
-            let raw = Arc::into_raw(table);
-            let send: *const (dyn TableProvider + Send) = raw as *const _;
-            unsafe { Arc::from_raw(send) }
-        };
+        let send_table = table_provider_to_send(table);
 
         let py_table = PyTable::new(send_table);
         Python::with_gil(|py| {
@@ -423,12 +396,8 @@ impl SchemaProvider for RustWrappedPySchemaProvider {
             // Otherwise, return None.
             let dataset = match Dataset::new(&table, py) {
                 Ok(dataset) => {
-                    // Convert from our internal Send type to trait expected type
                     let send_table = Arc::new(dataset) as Arc<dyn TableProvider + Send>;
-                    let raw = Arc::into_raw(send_table);
-                    let wide: *const dyn TableProvider = raw as *const _;
-                    let arc = unsafe { Arc::from_raw(wide) };
-                    Some(arc)
+                    Some(table_provider_send_to_table_provider(send_table))
                 }
                 Err(_) => None,
             };