Revert "UNPICK changes to review"

This reverts commit 07d2718.

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.__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,20 +39,40 @@ 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()))
         }
     }
 
-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`.
+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
 
+    from datafusion import SessionContext, TableProvider
+
+    ctx = SessionContext()
     provider = MyTableProvider()
-    ctx.register_table_provider("my_table", provider)
 
-    ctx.table("my_table").show()
+    capsule = provider.__datafusion_table_provider__()
+    capsule_provider = TableProvider.from_capsule(capsule)
+
+    df = ctx.from_pydict({"a": [1]})
+    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_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/__init__.py

@@ -54,6 +54,7 @@
 from .io import read_avro, read_csv, read_json, read_parquet
 from .plan import ExecutionPlan, LogicalPlan
 from .record_batch import RecordBatch, RecordBatchStream
+from .table_provider import TableProvider
 from .user_defined import (
     Accumulator,
     AggregateUDF,
@@ -90,6 +91,7 @@
     "SessionContext",
     "Table",
     "TableFunction",
+    "TableProvider",
     "WindowFrame",
     "WindowUDF",
     "catalog",

python/datafusion/catalog.py

@@ -27,6 +27,9 @@
 if TYPE_CHECKING:
     import pyarrow as pa
 
+    from datafusion import TableProvider
+    from datafusion.context import TableProviderExportable
+
 try:
     from warnings import deprecated  # Python 3.13+
 except ImportError:
@@ -82,7 +85,11 @@ def database(self, name: str = "public") -> Schema:
         """Returns the database with the given ``name`` from this catalog."""
         return self.schema(name)
 
-    def register_schema(self, name, schema) -> Schema | None:
+    def register_schema(
+        self,
+        name: str,
+        schema: Schema | SchemaProvider | SchemaProviderExportable,
+    ) -> Schema | None:
         """Register a schema with this catalog."""
         if isinstance(schema, Schema):
             return self.catalog.register_schema(name, schema._raw_schema)
@@ -122,8 +129,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) -> None:
-        """Register a table provider in this schema."""
+    def register_table(
+        self, name: str, 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)
@@ -219,14 +232,19 @@ 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(  # 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
+    def deregister_table(self, name: str, cascade: bool) -> None:  # noqa: B027
         """Remove a table from this schema.
 
         This method is optional. If your schema provides a fixed list of tables, you do

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
 
 
@@ -734,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.
@@ -743,16 +744,31 @@ 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 | TableProviderExportable
+    ) -> 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.
+
+        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 table 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.
         """
-        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 +788,21 @@ def register_catalog_provider(
             self.ctx.register_catalog_provider(name, provider)
 
     def register_table_provider(
-        self, name: str, provider: TableProviderExportable
+        self, name: str, provider: Table | TableProvider | TableProviderExportable
     ) -> 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.
+
+        Objects implementing ``__datafusion_table_provider__`` are also supported
+        and treated as :class:`~datafusion.catalog.TableProvider` instances.
         """
-        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

@@ -53,6 +53,7 @@
     import pyarrow as pa
 
     from datafusion._internal import expr as expr_internal
+    from datafusion.table_provider import TableProvider
 
 from enum import Enum
 
@@ -307,9 +308,17 @@ 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."""
-        return self.df.into_view()
+    def into_view(self) -> TableProvider:
+        """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.
+        """
+        from datafusion.table_provider import TableProvider as _TableProvider
+
+        return _TableProvider(self.df.into_view())
 
     def __getitem__(self, key: str | list[str]) -> DataFrame:
         """Return a new :py:class`DataFrame` with the specified column or columns.

python/datafusion/table_provider.py

@@ -0,0 +1,104 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Wrapper helpers for :mod:`datafusion._internal.TableProvider`."""
+
+from __future__ import annotations
+
+from typing import Any
+import warnings
+
+import datafusion._internal as df_internal
+
+_InternalTableProvider = df_internal.TableProvider
+
+
+class TableProvider:
+    """High level wrapper around :mod:`datafusion._internal.TableProvider`."""
+
+    __slots__ = ("_table_provider",)
+
+    def __init__(self, table_provider: _InternalTableProvider) -> None:
+        """Wrap a low level :class:`~datafusion._internal.TableProvider`."""
+        if isinstance(table_provider, TableProvider):
+            table_provider = table_provider._table_provider
+
+        if not isinstance(table_provider, _InternalTableProvider):
+            msg = "Expected a datafusion._internal.TableProvider instance."
+            raise TypeError(msg)
+
+        self._table_provider = table_provider
+
+    # ------------------------------------------------------------------
+    # constructors
+    # ------------------------------------------------------------------
+    @classmethod
+    def _wrap(cls, provider: _InternalTableProvider | TableProvider) -> TableProvider:
+        if isinstance(provider, cls):
+            return provider
+        return cls(provider)
+
+    @classmethod
+    def from_capsule(cls, capsule: Any) -> TableProvider:
+        """Create a :class:`TableProvider` from a PyCapsule."""
+        provider = _InternalTableProvider.from_capsule(capsule)
+        return cls(provider)
+
+    @classmethod
+    def from_dataframe(cls, df: Any) -> TableProvider:
+        """Create a :class:`TableProvider` from a :class:`DataFrame`."""
+        from datafusion.dataframe import DataFrame as DataFrameWrapper
+
+        if isinstance(df, DataFrameWrapper):
+            df = df.df
+
+        provider = _InternalTableProvider.from_dataframe(df)
+        return cls(provider)
+
+    @classmethod
+    def from_view(cls, df: Any) -> TableProvider:
+        """Deprecated. Use :meth:`DataFrame.into_view` or :meth:`TableProvider.from_dataframe`."""
+        from datafusion.dataframe import DataFrame as DataFrameWrapper
+
+        if isinstance(df, DataFrameWrapper):
+            df = df.df
+
+        provider = _InternalTableProvider.from_view(df)
+        warnings.warn(
+            "TableProvider.from_view is deprecated; use DataFrame.into_view or "
+            "TableProvider.from_dataframe instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return cls(provider)
+
+    # ------------------------------------------------------------------
+    # passthrough helpers
+    # ------------------------------------------------------------------
+    def __getattr__(self, name: str) -> Any:
+        """Delegate attribute lookup to the wrapped provider."""
+        return getattr(self._table_provider, name)
+
+    def __repr__(self) -> str:  # pragma: no cover - simple delegation
+        """Return a representation of the wrapped provider."""
+        return repr(self._table_provider)
+
+    def __datafusion_table_provider__(self) -> Any:
+        """Expose the wrapped provider for FFI integrations."""
+        return self._table_provider.__datafusion_table_provider__()
+
+
+__all__ = ["TableProvider"]