add first simple version of using unbound ibis tables in transformations

Author: sh-rp

.github/workflows/test_common.yml

@@ -110,7 +110,7 @@ jobs:
           pytest tests/pipeline/test_pipeline_extra.py -k arrow ${{ matrix.pytest_args }}
 
       - name: Install pipeline and sources dependencies
-        run: uv sync ${{ matrix.uv_sync_args }} --extra duckdb --extra cli --extra parquet --extra deltalake --extra sql_database --group sentry-sdk --group pipeline --group sources
+        run: uv sync ${{ matrix.uv_sync_args }} --extra duckdb --extra cli --extra parquet --extra deltalake --extra sql_database --group sentry-sdk --group pipeline --group sources --group ibis
 
       - name: Run extract and pipeline tests
         run: |

dlt/common/destination/dataset.py

@@ -22,11 +22,12 @@
 if TYPE_CHECKING:
     from dlt.common.libs.pandas import DataFrame
     from dlt.common.libs.pyarrow import Table as ArrowTable
-    from dlt.helpers.ibis import BaseBackend as IbisBackend
+    from dlt.helpers.ibis import BaseBackend as IbisBackend, Table as IbisTable
 else:
     DataFrame = Any
     ArrowTable = Any
     IbisBackend = Any
+    IbisTable = Any
 
 
 class SupportsReadableRelation:
@@ -193,6 +194,14 @@ def select(self, *columns: str) -> Self:
         """
         raise NotImplementedError("`select()` method is not supported for this relation")
 
+    def ibis(self) -> IbisTable:
+        """Returns an undbound ibis table representing the relation.
+
+        Returns:
+            IbisTable: The ibis table for the relation
+        """
+        raise NotImplementedError("`ibis()` method is not supported for this relation")
+
     @overload
     def __getitem__(self, column: str) -> Self: ...
 

dlt/destinations/dataset/ibis_relation.py

@@ -138,3 +138,7 @@ def __repr__(self) -> str:
 
     def __str__(self) -> str:
         return self._ibis_object.__str__()  # type: ignore[no-any-return]
+
+    def ibis(self) -> IbisTable:
+        """Returns an undbound ibis table representing the relation."""
+        return self._ibis_object

dlt/destinations/dataset/relation.py

@@ -13,7 +13,6 @@
 
 from dlt.common.schema.typing import TTableSchemaColumns
 from dlt.common.typing import Self
-from dlt.common.exceptions import TypeErrorWithKnownTypes
 from dlt.transformations import lineage
 from dlt.destinations.sql_client import SqlClientBase, WithSqlClient
 from dlt.destinations.dataset.exceptions import (
@@ -23,8 +22,10 @@
 
 if TYPE_CHECKING:
     from dlt.destinations.dataset.dataset import ReadableDBAPIDataset
+    from dlt.helpers.ibis import Table as IbisTable
 else:
     ReadableDBAPIDataset = Any
+    IbisTable = Any
 
 
 class BaseReadableDBAPIRelation(SupportsReadableRelation, WithSqlClient):
@@ -326,3 +327,13 @@ def __getitem__(self, columns: Sequence[str]) -> Self:
 
     def head(self, limit: int = 5) -> Self:
         return self.limit(limit)
+
+    def ibis(self) -> IbisTable:
+        """Returns an undbound ibis table representing the relation."""
+        from dlt.helpers.ibis import create_unbound_ibis_table
+
+        return create_unbound_ibis_table(
+            self._dataset.schema,
+            self._dataset.dataset_name,
+            self._table_name,
+        )

dlt/transformations/transformation.py

@@ -18,6 +18,8 @@
     TransformationException,
     IncompatibleDatasetsException,
 )
+
+from dlt.common.exceptions import MissingDependencyException
 from dlt.pipeline.exceptions import PipelineConfigMissing
 from dlt.destinations.dataset import ReadableDBAPIDataset
 from dlt.common.schema.typing import (
@@ -33,6 +35,12 @@
 from dlt.extract.exceptions import CurrentSourceNotAvailable
 from dlt.extract.pipe_iterator import DataItemWithMeta
 
+try:
+    from dlt.helpers.ibis import Expr as IbisExpr
+    from dlt.helpers.ibis import compile_ibis_to_sqlglot
+except (ImportError, MissingDependencyException):
+    IbisExpr = None
+
 
 class MaterializableSqlModel(SqlModel, WithComputableHints):
     # NOTE: we could forward all data access methods to this class
@@ -125,6 +133,10 @@ def transformation_function(*args: Any, **kwargs: Any) -> Iterator[TDataItems]:
                     relation = datasets[0](unwrapped_item)
             except sqlglot.errors.ParseError:
                 pass
+        # TODO: after merge of sqlglot based readble relation, we want to nativly support ibis expressions in the constructor of the relation
+        elif IbisExpr and isinstance(unwrapped_item, IbisExpr):
+            sql_query = compile_ibis_to_sqlglot(unwrapped_item, datasets[0].sqlglot_dialect)
+            relation = datasets[0](sql_query.sql(datasets[0].sqlglot_dialect))
 
         # we have something else, so fall back to regular resource behavior
         if not relation:

tests/transformations/test_transformations.py

@@ -175,3 +175,33 @@ def materializable_sql_model(dataset: SupportsReadableDataset[Any]) -> Any:
         "id": {"name": "id", "data_type": "bigint", "nullable": False},
         "name": {"name": "name", "x-annotation-pii": True, "data_type": "text", "nullable": True},
     }
+
+
+@pytest.mark.parametrize(
+    "destination_config",
+    transformation_configs(only_duckdb=True),
+    ids=lambda x: x.name,
+)
+def test_ibis_unbound_table_transformation(
+    destination_config: DestinationTestConfiguration,
+) -> None:
+    fruit_p, dest_p = setup_transformation_pipelines(destination_config)
+    fruit_p.run(fruitshop_source())
+
+    @dlt.transformation()
+    def materializable_sql_model(dataset: SupportsReadableDataset[Any]) -> Any:
+        purchases = dataset.purchases.ibis()
+        customers = dataset.customers.ibis()
+        yield purchases.join(customers, purchases.customer_id == customers.id)[
+            ["id", "customer_id", "inventory_id", "quantity", "name"]
+        ]
+
+    model = list(materializable_sql_model(fruit_p.dataset()))[0]
+    assert model.relation.arrow().column_names == [
+        "id",
+        "customer_id",
+        "inventory_id",
+        "quantity",
+        "name",
+    ]
+    assert model.relation.arrow().shape == (3, 5)