feat(catalog): Integrate session and catalog actions alongside existing APIs [2/3] (#3825)

This is PR [2/3] for integrating #3805 to Daft/main. This PR adds
several implementations with basic end-to-end sanity tests. Most
notably, this PR includes the bi-directional implementations for
Catalogs and Tables. These enable us to use python-backed
implementations in rust (for daft-connect and daft-sql) while also being
able to return Catalog/Table python objects for use with the python
APIs.

**Changes**
1. Adds several internal factories for attach/detaching supported
objects to the session.
2. Adds `Catalog.from_pydict` to cover basic needs and some backwards
compatibility.
3. Adds `Identifier.from_str` for pyiceberg style identifiers (these are
entirely optional).
4. Flesh out the [minimal session
actions](https://github.com/Eventual-Inc/Daft/compare/rchowell/catalog-1-of-3...rchowell/catalog-2-of-3?expand=1#diff-fc2305e1560f8f7d5a974f58325f4b231bc638d6bacebbc795b6f8fb924114ccR1884-R1895)
for a release.
5. Adds Bindings for a single name resolution abstraction, will come in
handy for case-normalization in the near future.
6. Updates daft-connect and daft-sql to use `attach_table` 

Part [3/3] will cutover all deprecated APIs to use these new APIs and
remove the unused artifacts. These APIs are enumerated in #3820

Author: rchowell

Cargo.lock

@@ -81,7 +81,6 @@ python = [
   "daft-stats/python",
   "daft-recordbatch/python",
   "daft-writers/python",
-  "dep:daft-catalog-python-catalog",
   "dep:daft-connect",
   "daft-connect/python",
   "dep:pyo3",

Cargo.toml

@@ -8,7 +8,7 @@
 from pyiceberg.catalog import Catalog as InnerCatalog
 from pyiceberg.table import Table as InnerTable
 
-from daft.catalog import Catalog, Table
+from daft.catalog import Catalog, Identifier, Table
 
 if TYPE_CHECKING:
     from daft.dataframe import DataFrame
@@ -34,16 +34,13 @@ def _from_obj(obj: object) -> IcebergCatalog:
             return c
         raise ValueError(f"Unsupported iceberg catalog type: {type(obj)}")
 
-    @property
-    def inner(self) -> InnerCatalog:
-        """Returns the inner iceberg catalog."""
-        return self._inner
-
     ###
     # get_*
     ###
 
-    def get_table(self, name: str) -> IcebergTable:
+    def get_table(self, name: str | Identifier) -> IcebergTable:
+        if isinstance(name, Identifier):
+            name = tuple(name)  # type: ignore
         return IcebergTable(self._inner.load_table(name))
 
     ###
@@ -59,20 +56,29 @@ class IcebergTable(Table):
     _inner: InnerTable
 
     def __init__(self, inner: InnerTable):
+        """DEPRECATED: Please use `Table.from_iceberg`; version 0.5.0!"""
+        warnings.warn(
+            "This is deprecated and will be removed in daft >= 0.5.0, please prefer using `Table.from_iceberg` instead; version 0.5.0!",
+            category=DeprecationWarning,
+        )
         self._inner = inner
 
+    @staticmethod
+    def _from_obj(obj: object) -> IcebergTable | None:
+        """Returns an IcebergTable if the given object can be adapted so."""
+        if isinstance(obj, InnerTable):
+            t = IcebergTable.__new__(IcebergTable)
+            t._inner = obj
+            return t
+        raise ValueError(f"Unsupported iceberg table type: {type(obj)}")
+
     @staticmethod
     def _try_from(obj: object) -> IcebergTable | None:
         """Returns an IcebergTable if the given object can be adapted so."""
         if isinstance(obj, InnerTable):
             return IcebergTable(obj)
         return None
 
-    @property
-    def inner(self) -> InnerTable:
-        """Returns the inner iceberg table."""
-        return self._inner
-
     def read(self) -> DataFrame:
         import daft
 

daft/catalog/__iceberg.py

@@ -42,14 +42,16 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
-from daft.daft import catalog as native_catalog
+from daft.daft import PyTableSource, catalog as native_catalog
 from daft.daft import PyIdentifier
 from daft.logical.builder import LogicalPlanBuilder
 
 from daft.dataframe import DataFrame
 
 from typing import TYPE_CHECKING
 
+from daft.logical.schema import Schema
+
 if TYPE_CHECKING:
     from daft.dataframe.dataframe import ColumnInputType
 
@@ -142,9 +144,12 @@ def register_python_catalog(catalog: object, name: str | None = None) -> str:
 class Catalog(ABC):
     """Interface for python catalog implementations."""
 
-    @property
-    def inner(self) -> object | None:
-        """Returns the inner catalog object if this is an adapter."""
+    @staticmethod
+    def from_pydict(tables: dict[str, Table]) -> Catalog:
+        """Returns an in-memory catalog from the dictionary."""
+        from daft.catalog.__memory import MemoryCatalog
+
+        return MemoryCatalog(tables)
 
     @staticmethod
     def from_iceberg(obj: object) -> Catalog:
@@ -192,7 +197,7 @@ def list_tables(self, pattern: str | None = None) -> list[str]: ...
     ###
 
     @abstractmethod
-    def get_table(self, name: str) -> Table: ...
+    def get_table(self, name: str | Identifier) -> Table: ...
 
     # TODO deprecated catalog APIs #3819
     def load_table(self, name: str) -> Table:
@@ -227,6 +232,12 @@ def __init__(self, *parts: str):
             raise ValueError("Identifier requires at least one part.")
         self._identifier = PyIdentifier(parts[:-1], parts[-1])
 
+    @staticmethod
+    def _from_pyidentifier(identifier: PyIdentifier) -> Identifier:
+        i = Identifier.__new__(Identifier)
+        i._identifier = identifier
+        return i
+
     @staticmethod
     def from_sql(input: str, normalize: bool = False) -> Identifier:
         """Parses an Identifier from an SQL string, normalizing to lowercase if specified.
@@ -243,6 +254,11 @@ def from_sql(input: str, normalize: bool = False) -> Identifier:
         i._identifier = PyIdentifier.from_sql(input, normalize)
         return i
 
+    @staticmethod
+    def from_str(input: str) -> Identifier:
+        """Parses an Identifier from a dot-delimited Python string without normalization."""
+        return Identifier(*input.split("."))
+
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Identifier):
             return False
@@ -260,13 +276,60 @@ def __len__(self) -> int:
     def __repr__(self) -> str:
         return f"Identifier('{self._identifier.__repr__()}')"
 
+    def __str__(self) -> str:
+        return ".".join(self)
+
 
 class Table(ABC):
     """Interface for python table implementations."""
 
-    @property
-    def inner(self) -> object | None:
-        """Returns the inner table object if this is an adapter."""
+    @staticmethod
+    def from_df(name: str, dataframe: DataFrame) -> Table:
+        """Returns a read-only table backed by the DataFrame."""
+        from daft.catalog.__memory import MemoryTable
+
+        return MemoryTable(name, dataframe)
+
+    @staticmethod
+    def from_iceberg(obj: object) -> Table:
+        """Returns a Daft Table instance from an Iceberg table."""
+        try:
+            from daft.catalog.__iceberg import IcebergTable
+
+            return IcebergTable._from_obj(obj)
+        except ImportError:
+            raise ImportError("Iceberg support not installed: pip install -U 'getdaft[iceberg]'")
+
+    @staticmethod
+    def from_unity(obj: object) -> Table:
+        """Returns a Daft Table instance from a Unity table."""
+        try:
+            from daft.catalog.__unity import UnityTable
+
+            return UnityTable._from_obj(obj)
+        except ImportError:
+            raise ImportError("Unity support not installed: pip install -U 'getdaft[unity]'")
+
+    @staticmethod
+    def _from_obj(obj: object) -> Table:
+        """Returns a Daft Table from a supported object type or raises an error."""
+        raise ValueError(f"Unsupported table type: {type(obj)}")
+
+    # TODO catalog APIs part 3
+    # @property
+    # @abstractmethod
+    # def name(self) -> str:
+    #     """Returns the table name."""
+
+    # TODO catalog APIs part 3
+    # @property
+    # @abstractmethod
+    # def inner(self) -> object | None:
+    #     """Returns the inner table object if this is an adapter."""
+
+    @abstractmethod
+    def read(self) -> DataFrame:
+        """Returns a DataFrame from this table."""
 
     # TODO deprecated catalog APIs #3819
     def to_dataframe(self) -> DataFrame:
@@ -277,14 +340,62 @@ def to_dataframe(self) -> DataFrame:
         )
         return self.read()
 
-    @abstractmethod
-    def read(self) -> DataFrame:
-        """Returns a DataFrame from this table."""
-
     def select(self, *columns: ColumnInputType) -> DataFrame:
         """Returns a DataFrame from this table with the selected columns."""
         return self.read().select(*columns)
 
     def show(self, n: int = 8) -> None:
         """Shows the first n rows from this table."""
         return self.read().show(n)
+
+
+class TableSource:
+    """A TableSource is used to create a new table; this could be a Schema or DataFrame."""
+
+    _source: PyTableSource
+
+    def __init__(self) -> None:
+        raise ValueError("We do not support creating a TableSource via __init__")
+
+    @staticmethod
+    def from_df(df: DataFrame) -> TableSource:
+        s = TableSource.__new__(TableSource)
+        s._source = PyTableSource.from_builder(df._builder._builder)
+        return s
+
+    @staticmethod
+    def _from_obj(obj: object = None) -> TableSource:
+        # TODO for future sources, consider https://github.com/Eventual-Inc/Daft/pull/2864
+        if obj is None:
+            return TableSource._from_none()
+        elif isinstance(obj, DataFrame):
+            return TableSource.from_df(obj)
+        elif isinstance(obj, str):
+            return TableSource._from_path(obj)
+        elif isinstance(obj, Schema):
+            return TableSource._from_schema(obj)
+        else:
+            raise Exception(f"Unknown table source: {obj}")
+
+    @staticmethod
+    def _from_none() -> TableSource:
+        # for creating temp mutable tables, but we don't have those yet
+        # s = TableSource.__new__(TableSource)
+        # s._source = PyTableSource.empty()
+        # return s
+        # todo temp workaround just use an empty schema
+        return TableSource._from_schema(Schema._from_fields([]))
+
+    @staticmethod
+    def _from_schema(schema: Schema) -> TableSource:
+        # we don't have mutable temp tables, so just make an empty view
+        # s = TableSource.__new__(TableSource)
+        # s._source = PyTableSource.from_schema(schema._schema)
+        # return s
+        # todo temp workaround until create_table is wired
+        return TableSource.from_df(DataFrame._from_pylist([]))
+
+    @staticmethod
+    def _from_path(path: str) -> TableSource:
+        # for supporting daft.create_table("t", "/path/to/data") <-> CREATE TABLE t AS '/path/to/my.data'
+        raise NotImplementedError("creating a table source from a path is not yet supported.")

daft/catalog/__init__.py

@@ -0,0 +1,56 @@
+"""An in-memory implementation for the daft catalog abstractions."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from daft.catalog import Catalog, Identifier, Table
+
+if TYPE_CHECKING:
+    from daft.dataframe.dataframe import DataFrame
+
+
+class MemoryCatalog(Catalog):
+    """An in-memory catalog is backed by a dictionary."""
+
+    _tables: dict[str, Table]
+
+    def __init__(self, tables: dict[str, Table]):
+        self._tables = tables
+
+    ###
+    # list_*
+    ###
+
+    def list_tables(self, pattern: str | None = None) -> list[str]:
+        if pattern is None:
+            return list(self._tables.keys())
+        return [path for path in self._tables.keys() if path.startswith(pattern)]
+
+    ###
+    # get_*
+    ###
+
+    def get_table(self, name: str | Identifier) -> Table:
+        path = str(name)
+        if path not in self._tables:
+            raise ValueError(f"Table {path} does not exist.")
+        return self._tables[path]
+
+
+class MemoryTable(Table):
+    """An in-memory table holds a reference to an existing dataframe."""
+
+    _name: str
+    _inner: DataFrame
+
+    def __init__(self, name: str, inner: DataFrame):
+        self._name = name
+        self._inner = inner
+
+    ###
+    # DataFrame Methods
+    ###
+
+    def read(self) -> DataFrame:
+        return self._inner

daft/catalog/__memory.py

@@ -5,7 +5,7 @@
 import warnings
 from typing import TYPE_CHECKING
 
-from daft.catalog import Catalog, Table
+from daft.catalog import Catalog, Identifier, Table
 from daft.unity_catalog import UnityCatalog as InnerCatalog  # noqa: TID253
 from daft.unity_catalog import UnityCatalogTable as InnerTable  # noqa: TID253
 
@@ -33,16 +33,13 @@ def _from_obj(obj: object) -> UnityCatalog:
             return c
         raise ValueError(f"Unsupported unity catalog type: {type(obj)}")
 
-    @property
-    def inner(self) -> InnerCatalog:
-        """Returns the inner unity catalog."""
-        return self._inner
-
     ###
     # get_*
     ###
 
-    def get_table(self, name: str) -> UnityTable:
+    def get_table(self, name: str | Identifier) -> UnityTable:
+        if isinstance(name, Identifier):
+            name = ".".join(name)  # TODO unity qualified identifiers
         return UnityTable(self._inner.load_table(name))
 
     ###
@@ -74,8 +71,22 @@ class UnityTable(Table):
     _inner: InnerTable
 
     def __init__(self, unity_table: InnerTable):
+        """DEPRECATED: Please use `Table.from_unity`; version 0.5.0!"""
+        warnings.warn(
+            "This is deprecated and will be removed in daft >= 0.5.0, please prefer using `Table.from_unity` instead; version 0.5.0!",
+            category=DeprecationWarning,
+        )
         self._inner = unity_table
 
+    @staticmethod
+    def _from_obj(obj: object) -> UnityTable | None:
+        """Returns a UnityTable if the given object can be adapted so."""
+        if isinstance(obj, InnerTable):
+            t = UnityTable.__new__(UnityTable)
+            t._inner = obj
+            return t
+        raise ValueError(f"Unsupported unity table type: {type(obj)}")
+
     @staticmethod
     def _try_from(obj: object) -> UnityTable | None:
         """Returns an UnityTable if the given object can be adapted so."""