Migrate Table → TableProvider; refactor registration and access, update

docs/tests, add DataFrame view support, and improve Send/concurrency
support.

migrates the codebase from using `Table` to a
`TableProvider`-based API, refactors registration and access paths to
simplify catalog/context interactions, and updates documentation and
examples. DataFrame view handling is improved (`into_view` is now
public), the test-suite is expanded to cover new registration and async
SQL scenarios, and `TableProvider` now supports the `Send` trait across
modules for safer concurrency. Minor import cleanup and utility
adjustments (including a refined `pyany_to_table_provider`) are
included.

Author: kosiew

docs/source/contributor-guide/ffi.rst

@@ -34,7 +34,7 @@ as performant as possible and to utilize the features of DataFusion, you may dec
 your source in Rust and then expose it through `PyO3 <https://pyo3.rs>`_ as a Python library.
 
 At first glance, it may appear the best way to do this is to add the ``datafusion-python``
-crate as a dependency, provide a ``PyTable``, and then to register it with the 
+crate as a dependency, produce a DataFusion table in Rust, and then register it with the
 ``SessionContext``. Unfortunately, this will not work.
 
 When you produce your code as a Python library and it needs to interact with the DataFusion

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

@@ -152,9 +152,11 @@ as Delta Lake. This will require a recent version of
 .. code-block:: python
 
     from deltalake import DeltaTable
+    from datafusion import TableProvider
 
     delta_table = DeltaTable("path_to_table")
-    ctx.register_table_provider("my_delta_table", delta_table)
+    provider = TableProvider.from_capsule(delta_table)
+    ctx.register_table("my_delta_table", provider)
     df = ctx.table("my_delta_table")
     df.show()
 

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

@@ -47,12 +47,29 @@ A complete example can be found in the `examples folder <https://github.com/apac
         }
     }
 
-Once you have this library available, in python you can register your table provider
-to the ``SessionContext``.
+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()``.
 
 .. code-block:: python
 
+    from datafusion import SessionContext, TableProvider
+
+    ctx = SessionContext()
     provider = MyTableProvider()
-    ctx.register_table_provider("my_table", provider)
 
-    ctx.table("my_table").show()
+    capsule_provider = TableProvider.from_capsule(provider)
+
+    df = ctx.from_pydict({"a": [1]})
+    view_provider = TableProvider.from_view(df)
+
+    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
+table providers that can be registered with the SessionContext using ``register_table()``.

python/datafusion/__init__.py

@@ -33,7 +33,7 @@
 from . import functions, object_store, substrait, unparser
 
 # The following imports are okay to remain as opaque to the user.
-from ._internal import Config
+from ._internal import Config, TableProvider
 from .catalog import Catalog, Database, Table
 from .col import col, column
 from .common import (
@@ -90,6 +90,7 @@
     "SessionContext",
     "Table",
     "TableFunction",
+    "TableProvider",
     "WindowFrame",
     "WindowUDF",
     "catalog",

python/datafusion/catalog.py

@@ -27,6 +27,8 @@
 if TYPE_CHECKING:
     import pyarrow as pa
 
+    from datafusion import TableProvider
+
 try:
     from warnings import deprecated  # Python 3.13+
 except ImportError:
@@ -122,8 +124,8 @@ 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) -> None:
-        """Register a table provider in this schema."""
+    def register_table(self, name, table: Table | TableProvider) -> None:
+        """Register a table or table provider in this schema."""
         if isinstance(table, Table):
             return self._raw_schema.register_table(name, table.table)
         return self._raw_schema.register_table(name, table)
@@ -219,8 +221,8 @@ def table(self, name: str) -> Table | None:
         """Retrieve a specific table from this schema."""
         ...
 
-    def register_table(self, name: str, table: Table) -> None:  # noqa: B027
-        """Add a table from this schema.
+    def register_table(self, name: str, table: Table | TableProvider) -> None:  # noqa: B027
+        """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.

python/datafusion/context.py

@@ -46,6 +46,7 @@
     import pandas as pd
     import polars as pl
 
+    from datafusion import TableProvider
     from datafusion.plan import ExecutionPlan, LogicalPlan
 
 
@@ -743,16 +744,21 @@ 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) -> None:
-        """Register a :py:class: `~datafusion.catalog.Table` as a table.
+    def register_table(self, name: str, table: Table | TableProvider) -> None:
+        """Register a :py:class:`~datafusion.catalog.Table` or ``TableProvider``.
 
-        The registered table can be referenced from SQL statement executed against.
+        The registered table can be referenced from SQL statements executed against
+        this context.
 
         Args:
             name: Name of the resultant table.
-            table: DataFusion table to add to the session context.
+            table: DataFusion :class:`Table` or :class:`TableProvider` to add to the
+                session context.
         """
-        self.ctx.register_table(name, table.table)
+        if isinstance(table, Table):
+            self.ctx.register_table(name, table.table)
+        else:
+            self.ctx.register_table(name, table)
 
     def deregister_table(self, name: str) -> None:
         """Remove a table from the session."""
@@ -772,14 +778,18 @@ def register_catalog_provider(
             self.ctx.register_catalog_provider(name, provider)
 
     def register_table_provider(
-        self, name: str, provider: TableProviderExportable
+        self, name: str, provider: TableProviderExportable | TableProvider
     ) -> None:
         """Register a table provider.
 
-        This table provider must have a method called ``__datafusion_table_provider__``
-        which returns a PyCapsule that exposes a ``FFI_TableProvider``.
+        Deprecated: use :meth:`register_table` instead.
         """
-        self.ctx.register_table_provider(name, provider)
+        warnings.warn(
+            "register_table_provider is deprecated; use register_table",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        self.register_table(name, provider)
 
     def register_udtf(self, func: TableFunction) -> None:
         """Register a user defined table function."""

python/datafusion/dataframe.py

@@ -40,6 +40,7 @@
 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
@@ -307,8 +308,14 @@ def __init__(self, df: DataFrameInternal) -> None:
         """
         self.df = df
 
-    def into_view(self) -> pa.Table:
-        """Convert DataFrame as a ViewTable which can be used in register_table."""
+    def into_view(self) -> TableProviderInternal:
+        """Convert ``DataFrame`` into a ``TableProvider`` view for registration.
+
+        This is the preferred way to obtain a view for
+        :py:meth:`~datafusion.context.SessionContext.register_table`.
+        ``TableProvider.from_dataframe`` calls this method under the hood,
+        and the older ``TableProvider.from_view`` helper is deprecated.
+        """
         return self.df.into_view()
 
     def __getitem__(self, key: str | list[str]) -> DataFrame:

python/tests/test_context.py

@@ -27,6 +27,7 @@
     SessionConfig,
     SessionContext,
     SQLOptions,
+    TableProvider,
     column,
     literal,
 )
@@ -330,6 +331,35 @@ def test_deregister_table(ctx, database):
     assert public.names() == {"csv1", "csv2"}
 
 
+def test_register_table_from_dataframe_into_view(ctx):
+    df = ctx.from_pydict({"a": [1, 2]})
+    provider = df.into_view()
+    ctx.register_table("view_tbl", provider)
+    result = ctx.sql("SELECT * FROM view_tbl").collect()
+    assert [b.to_pydict() for b in result] == [{"a": [1, 2]}]
+
+
+def test_table_provider_from_capsule(ctx):
+    df = ctx.from_pydict({"a": [1, 2]})
+    provider = df.into_view()
+    capsule = provider.__datafusion_table_provider__()
+    provider2 = TableProvider.from_capsule(capsule)
+    ctx.register_table("capsule_tbl", provider2)
+    result = ctx.sql("SELECT * FROM capsule_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
+        ctx.register_table("bad", df)
+
+
 def test_register_dataset(ctx):
     # create a RecordBatch and register it as a pyarrow.dataset.Dataset
     batch = pa.RecordBatch.from_arrays(

src/catalog.rs

@@ -17,6 +17,7 @@
 
 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 async_trait::async_trait;
 use datafusion::catalog::{MemoryCatalogProvider, MemorySchemaProvider};
@@ -51,7 +52,7 @@ pub struct PySchema {
 #[pyclass(name = "RawTable", module = "datafusion.catalog", subclass)]
 #[derive(Clone)]
 pub struct PyTable {
-    pub table: Arc<dyn TableProvider>,
+    pub table: Arc<dyn TableProvider + Send>,
 }
 
 impl From<Arc<dyn CatalogProvider>> for PyCatalog {
@@ -67,11 +68,11 @@ impl From<Arc<dyn SchemaProvider>> for PySchema {
 }
 
 impl PyTable {
-    pub fn new(table: Arc<dyn TableProvider>) -> Self {
+    pub fn new(table: Arc<dyn TableProvider + Send>) -> Self {
         Self { table }
     }
 
-    pub fn table(&self) -> Arc<dyn TableProvider> {
+    pub fn table(&self) -> Arc<dyn TableProvider + Send> {
         self.table.clone()
     }
 }
@@ -205,15 +206,18 @@ impl PySchema {
 
             let provider = unsafe { capsule.reference::<FFI_TableProvider>() };
             let provider: ForeignTableProvider = provider.into();
-            Arc::new(provider) as Arc<dyn TableProvider>
+            Arc::new(provider) as Arc<dyn TableProvider + Send>
         } else {
             match table_provider.extract::<PyTable>() {
                 Ok(py_table) => py_table.table,
-                Err(_) => {
-                    let py = table_provider.py();
-                    let provider = Dataset::new(&table_provider, py)?;
-                    Arc::new(provider) as Arc<dyn TableProvider>
-                }
+                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>
+                    }
+                },
             }
         };
 
@@ -294,7 +298,7 @@ impl RustWrappedPySchemaProvider {
         }
     }
 
-    fn table_inner(&self, name: &str) -> PyResult<Option<Arc<dyn TableProvider>>> {
+    fn table_inner(&self, name: &str) -> PyResult<Option<Arc<dyn TableProvider + Send>>> {
         Python::with_gil(|py| {
             let provider = self.schema_provider.bind(py);
             let py_table_method = provider.getattr("table")?;
@@ -305,26 +309,30 @@ impl RustWrappedPySchemaProvider {
             }
 
             if py_table.hasattr("__datafusion_table_provider__")? {
-                let capsule = provider.getattr("__datafusion_table_provider__")?.call0()?;
+                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>))
+                Ok(Some(Arc::new(provider) as Arc<dyn TableProvider + Send>))
             } else {
                 if let Ok(inner_table) = py_table.getattr("table") {
                     if let Ok(inner_table) = inner_table.extract::<PyTable>() {
                         return Ok(Some(inner_table.table));
                     }
                 }
 
+                if let Ok(py_provider) = py_table.extract::<PyTableProvider>() {
+                    return Ok(Some(py_provider.into_inner()));
+                }
+
                 match py_table.extract::<PyTable>() {
                     Ok(py_table) => Ok(Some(py_table.table)),
                     Err(_) => {
                         let ds = Dataset::new(&py_table, py).map_err(py_datafusion_err)?;
-                        Ok(Some(Arc::new(ds) as Arc<dyn TableProvider>))
+                        Ok(Some(Arc::new(ds) as Arc<dyn TableProvider + Send>))
                     }
                 }
             }
@@ -360,15 +368,32 @@ impl SchemaProvider for RustWrappedPySchemaProvider {
         &self,
         name: &str,
     ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>, DataFusionError> {
-        self.table_inner(name).map_err(to_datafusion_err)
+        // 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))
+            }
+            None => Ok(None),
+        }
     }
 
     fn register_table(
         &self,
         name: String,
         table: Arc<dyn TableProvider>,
     ) -> datafusion::common::Result<Option<Arc<dyn TableProvider>>> {
-        let py_table = PyTable::new(table);
+        // 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 py_table = PyTable::new(send_table);
         Python::with_gil(|py| {
             let provider = self.schema_provider.bind(py);
             let _ = provider
@@ -397,7 +422,14 @@ impl SchemaProvider for RustWrappedPySchemaProvider {
             // If we can turn this table provider into a `Dataset`, return it.
             // Otherwise, return None.
             let dataset = match Dataset::new(&table, py) {
-                Ok(dataset) => Some(Arc::new(dataset) as Arc<dyn TableProvider>),
+                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)
+                }
                 Err(_) => None,
             };