Normalize & simplify TableProvider/DataFrame registration; add

utilities, docs, and robustness fixes

* Normalized table-provider handling and simplified registration flow
  across the codebase; multiple commits centralize provider coercion and
normalization.
* Introduced utility helpers (`coerce_table_provider`,
  `extract_table_provider`, `_normalize_table_provider`) to centralize
extraction, error handling, and improve clarity.
* Simplified `from_dataframe` / `into_view` behavior: clearer
  implementations, direct returns of DataFrame views where appropriate,
and added internal tests for DataFrame flows.
* Fixed DataFrame registration semantics: enforce `TypeError` for
  invalid registrations; added handling for `DataFrameWrapper` by
converting it to a view.
* Added tests, including a schema registration test using a PyArrow
  dataset and internal DataFrame tests to cover new flows.
* Documentation improvements: expanded `from_dataframe` docstrings with
  parameter details, added usage examples for `into_view`, and
documented deprecations (e.g., `register_table_provider` →
`register_table`).
* Warning and UX fixes: synchronized deprecation `stacklevel` so
  warnings point to caller code; improved `__dir__` to return sorted,
unique attributes.
* Cleanup: removed unused imports (including an unused error import from
  `utils.rs`) and other dead code to reduce noise.

Author: kosiew

dev/changelog/49.0.0.md

@@ -25,6 +25,10 @@ This release consists of 16 commits from 7 contributors. See credits at the end
 
 - fix(build): Include build.rs in published crates [#1199](https://github.com/apache/datafusion-python/pull/1199) (colinmarc)
 
+**Deprecations:**
+
+- Document that `SessionContext.register_table_provider` is deprecated in favor of `SessionContext.register_table`.
+
 **Other:**
 
 - 48.0.0 Release [#1175](https://github.com/apache/datafusion-python/pull/1175) (timsaucer)

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

@@ -160,7 +160,14 @@ as Delta Lake. This will require a recent version of
     df = ctx.table("my_delta_table")
     df.show()
 
-On older versions of ``deltalake`` (prior to 0.22) you can use the 
+.. note::
+
+   :py:meth:`~datafusion.context.SessionContext.register_table_provider` is
+   deprecated. Use
+   :py:meth:`~datafusion.context.SessionContext.register_table` with a
+   :py:class:`~datafusion.TableProvider` instead.
+
+On older versions of ``deltalake`` (prior to 0.22) you can use the
 `Arrow DataSet <https://arrow.apache.org/docs/python/generated/pyarrow.dataset.Dataset.html>`_
 interface to import to DataFusion, but this does not support features such as filter push down
 which can lead to a significant performance difference.

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

@@ -54,6 +54,13 @@ Call the provider's ``__datafusion_table_provider__()`` method to obtain the cap
 before constructing a ``TableProvider``. The ``TableProvider.from_view()`` helper is
 deprecated; instead use ``TableProvider.from_dataframe()`` or ``DataFrame.into_view()``.
 
+.. note::
+
+   :py:meth:`~datafusion.context.SessionContext.register_table_provider` is
+   deprecated. Use
+   :py:meth:`~datafusion.context.SessionContext.register_table` with the
+   resulting :py:class:`~datafusion.TableProvider` instead.
+
 .. code-block:: python
 
     from datafusion import SessionContext, TableProvider

python/datafusion/catalog.py

@@ -23,6 +23,7 @@
 from typing import TYPE_CHECKING, Protocol
 
 import datafusion._internal as df_internal
+from datafusion.utils import _normalize_table_provider
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -137,9 +138,8 @@ def register_table(
         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)
+        provider = _normalize_table_provider(table)
+        return self._raw_schema.register_table(name, provider)
 
     def deregister_table(self, name: str) -> None:
         """Deregister a table provider from this schema."""

python/datafusion/context.py

@@ -34,6 +34,7 @@
 from datafusion.expr import Expr, SortExpr, sort_list_to_raw_sort_list
 from datafusion.record_batch import RecordBatchStream
 from datafusion.user_defined import AggregateUDF, ScalarUDF, TableFunction, WindowUDF
+from datafusion.utils import _normalize_table_provider
 
 from ._internal import RuntimeEnvBuilder as RuntimeEnvBuilderInternal
 from ._internal import SessionConfig as SessionConfigInternal
@@ -735,7 +736,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.dataframe.DataFrame` as a view.
+        """Register a :py:class: `~datafusion.detaframe.DataFrame` as a view.
 
         Args:
             name (str): The name to register the view under.
@@ -747,28 +748,27 @@ def register_view(self, name: str, df: DataFrame) -> None:
     def register_table(
         self, name: str, table: Table | TableProvider | TableProviderExportable
     ) -> None:
-        """Register a :py:class:`~datafusion.catalog.Table` or ``TableProvider``.
+        """Register a :py:class:`~datafusion.catalog.Table` or
+        :py:class:`~datafusion.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`.
+        :meth:`datafusion.TableProvider.from_dataframe`.
 
         Objects implementing ``__datafusion_table_provider__`` are also supported
-        and treated as :class:`~datafusion.catalog.TableProvider` instances.
+        and treated as :py:class:`~datafusion.TableProvider` instances.
 
         Args:
             name: Name of the resultant table.
             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)
-        else:
-            self.ctx.register_table(name, table)
+        provider = _normalize_table_provider(table)
+        self.ctx.register_table(name, provider)
 
     def deregister_table(self, name: str) -> None:
         """Remove a table from the session."""
@@ -795,7 +795,7 @@ def register_table_provider(
         Deprecated: use :meth:`register_table` instead.
 
         Objects implementing ``__datafusion_table_provider__`` are also supported
-        and treated as :class:`~datafusion.catalog.TableProvider` instances.
+        and treated as :py:class:`~datafusion.TableProvider` instances.
         """
         warnings.warn(
             "register_table_provider is deprecated; use register_table",

python/datafusion/dataframe.py

@@ -313,8 +313,20 @@ def into_view(self) -> TableProvider:
 
         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,
+        ``datafusion.TableProvider.from_dataframe`` calls this method under the hood,
         and the older ``TableProvider.from_view`` helper is deprecated.
+
+        The ``DataFrame`` remains valid after conversion, so it can still be used for
+        additional queries alongside the returned view.
+
+        Examples:
+            >>> from datafusion import SessionContext
+            >>> ctx = SessionContext()
+            >>> df = ctx.sql("SELECT 1 AS value")
+            >>> provider = df.into_view()
+            >>> ctx.register_table("values_view", provider)
+            >>> df.collect()  # The DataFrame is still usable
+            >>> ctx.sql("SELECT value FROM values_view").collect()
         """
         from datafusion.table_provider import TableProvider as _TableProvider
 

python/datafusion/table_provider.py

@@ -26,6 +26,9 @@
 
 _InternalTableProvider = df_internal.TableProvider
 
+# Keep in sync with ``datafusion._internal.TableProvider.from_view``.
+_FROM_VIEW_WARN_STACKLEVEL = 2
+
 
 class TableProvider:
     """High level wrapper around :mod:`datafusion._internal.TableProvider`."""
@@ -50,14 +53,26 @@ def from_capsule(cls, capsule: Any) -> TableProvider:
 
     @classmethod
     def from_dataframe(cls, df: Any) -> TableProvider:
-        """Create a :class:`TableProvider` from a :class:`DataFrame`."""
+        """Create a :class:`TableProvider` from tabular data.
+
+        Parameters
+        ----------
+        df:
+            Either a :class:`~datafusion.dataframe.DataFrame` wrapper or the
+            corresponding :class:`~datafusion._internal.DataFrame`. When
+            working with third-party DataFrame libraries, convert them via
+            :meth:`~datafusion.SessionContext.from_arrow` before calling
+            :meth:`~datafusion.dataframe.DataFrame.into_view` or this
+            constructor.
+        """
         from datafusion.dataframe import DataFrame as DataFrameWrapper
 
         if isinstance(df, DataFrameWrapper):
-            df = df.df
+            dataframe = df
+        else:
+            dataframe = DataFrameWrapper(df)
 
-        provider = _InternalTableProvider.from_dataframe(df)
-        return cls(provider)
+        return dataframe.into_view()
 
     @classmethod
     def from_view(cls, df: Any) -> TableProvider:
@@ -74,8 +89,8 @@ def from_view(cls, df: Any) -> TableProvider:
         warnings.warn(
             "TableProvider.from_view is deprecated; use DataFrame.into_view or "
             "TableProvider.from_dataframe instead.",
-            DeprecationWarning,
-            stacklevel=2,
+            category=DeprecationWarning,
+            stacklevel=_FROM_VIEW_WARN_STACKLEVEL,
         )
         return cls(provider)
 
@@ -88,7 +103,7 @@ def __getattr__(self, name: str) -> Any:
 
     def __dir__(self) -> list[str]:
         """Expose delegated attributes via :func:`dir`."""
-        return dir(self._table_provider) + super().__dir__()
+        return sorted(set(super().__dir__()) | set(dir(self._table_provider)))
 
     def __repr__(self) -> str:  # pragma: no cover - simple delegation
         """Return a representation of the wrapped provider."""

python/datafusion/utils.py

@@ -0,0 +1,80 @@
+# 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.
+"""Miscellaneous helper utilities for DataFusion's Python bindings."""
+
+from __future__ import annotations
+
+from importlib import import_module, util
+from typing import TYPE_CHECKING, Any
+
+from datafusion._internal import EXPECTED_PROVIDER_MSG
+
+_PYARROW_DATASET_TYPES: tuple[type[Any], ...]
+_dataset_spec = util.find_spec("pyarrow.dataset")
+if _dataset_spec is None:  # pragma: no cover - optional dependency at runtime
+    _PYARROW_DATASET_TYPES = ()
+else:  # pragma: no cover - exercised in environments with pyarrow installed
+    _dataset_module = import_module("pyarrow.dataset")
+    dataset_base = getattr(_dataset_module, "Dataset", None)
+    dataset_types: set[type[Any]] = set()
+    if isinstance(dataset_base, type):
+        dataset_types.add(dataset_base)
+        for value in vars(_dataset_module).values():
+            if isinstance(value, type) and issubclass(value, dataset_base):
+                dataset_types.add(value)
+    _PYARROW_DATASET_TYPES = tuple(dataset_types)
+
+if TYPE_CHECKING:  # pragma: no cover - imported for typing only
+    from datafusion import TableProvider
+    from datafusion.catalog import Table
+    from datafusion.context import TableProviderExportable
+
+
+def _normalize_table_provider(
+    table: Table | TableProvider | TableProviderExportable,
+) -> Any:
+    """Return the underlying provider for supported table inputs.
+
+    Args:
+        table: A :class:`~datafusion.catalog.Table`,
+            :class:`~datafusion.table_provider.TableProvider`, or object exporting a
+            DataFusion table provider via ``__datafusion_table_provider__``.
+
+    Returns:
+        The object expected by the Rust bindings for table registration.
+
+    Raises:
+        TypeError: If ``table`` is not a supported table provider input.
+    """
+
+    from datafusion.catalog import Table as _Table
+    from datafusion.table_provider import TableProvider as _TableProvider
+
+    if isinstance(table, _Table):
+        return table.table
+
+    if isinstance(table, _TableProvider):
+        return table._table_provider
+
+    if _PYARROW_DATASET_TYPES and isinstance(table, _PYARROW_DATASET_TYPES):
+        return table
+
+    provider_factory = getattr(table, "__datafusion_table_provider__", None)
+    if callable(provider_factory):
+        return table
+
+    raise TypeError(EXPECTED_PROVIDER_MSG)

python/tests/test_catalog.py

@@ -20,7 +20,7 @@
 import pyarrow as pa
 import pyarrow.dataset as ds
 import pytest
-from datafusion import SessionContext, Table
+from datafusion import EXPECTED_PROVIDER_MSG, SessionContext, Table
 
 
 # Note we take in `database` as a variable even though we don't use
@@ -164,6 +164,38 @@ def test_python_table_provider(ctx: SessionContext):
     assert schema.table_names() == {"table4"}
 
 
+def test_schema_register_table_with_pyarrow_dataset(ctx: SessionContext):
+    schema = ctx.catalog().schema()
+    batch = pa.RecordBatch.from_arrays(
+        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
+        names=["a", "b"],
+    )
+    dataset = ds.dataset([batch])
+    table_name = "pa_dataset"
+
+    try:
+        schema.register_table(table_name, dataset)
+        assert table_name in schema.table_names()
+
+        result = ctx.sql(f"SELECT a, b FROM {table_name}").collect()
+
+        assert len(result) == 1
+        assert result[0].column(0) == pa.array([1, 2, 3])
+        assert result[0].column(1) == pa.array([4, 5, 6])
+    finally:
+        schema.deregister_table(table_name)
+
+
+def test_schema_register_table_with_dataframe_errors(ctx: SessionContext):
+    schema = ctx.catalog().schema()
+    df = ctx.from_pydict({"a": [1]})
+
+    with pytest.raises(Exception) as exc_info:
+        schema.register_table("bad", df)
+
+    assert str(exc_info.value) == EXPECTED_PROVIDER_MSG
+
+
 def test_in_end_to_end_python_providers(ctx: SessionContext):
     """Test registering all python providers and running a query against them."""