pipedag compatibility with pydiverse.transform >=0.2

.github/workflows/nightly_tests.yml

@@ -58,7 +58,6 @@ jobs:
           - py312all
           - py311all
           - py310all
-          - py39all
           - py39pdsa1all
           - py311pdsa1all
     uses: ./.github/workflows/test.yml

.github/workflows/tests.yml

@@ -58,7 +58,6 @@ jobs:
         os:
           - ubuntu-latest
         environment:
-          - py39all
           - py310all
           - py311all
           - py312all
@@ -80,7 +79,7 @@ jobs:
         os:
           - ubuntu-latest
         environment:
-          - py39all
+          - py310all
           - py312all
     uses: ./.github/workflows/test.yml
     with:
@@ -98,7 +97,7 @@ jobs:
         os:
           - ubuntu-latest
         environment:
-          - py39ibm
+          - py310ibm
           - py312ibm
     uses: ./.github/workflows/test.yml
     with:
@@ -118,7 +117,7 @@ jobs:
         os:
           - ubuntu-latest
         environment:
-          - py39all
+          - py310all
           - py311all
     uses: ./.github/workflows/test.yml
     with:

README.md

@@ -308,6 +308,11 @@ Same happened for MacOS. The driver was installed in `/opt/homebrew/etc/odbcinst
 
 Furthermore, make sure you use 127.0.0.1 instead of localhost. It seems that /etc/hosts is ignored.
 
+### Incompatibility with pydiverse.transform 0.2.0
+
+Pydiverse.pipedag currently doesn't support pydiverse.transform 0.2.0, please consider using a newer Version,
+or downgrading to pydiverse.transform 0.1.6.
+
 ## Packaging and publishing to pypi and conda-forge using github actions
 
 - bump version number in [pyproject.toml](pyproject.toml)
@@ -345,4 +350,4 @@ Conda-forge packages are updated via:
 - https://github.com/conda-forge/pydiverse-pipedag-feedstock#updating-pydiverse-pipedag-feedstock
 - update `recipe/meta.yaml`
 - test meta.yaml in pipedag repo: `conda-build build ../pydiverse-pipedag-feedstock/recipe/meta.yaml`
-- commit `recipe/meta.yaml` to branch of fork and submit PR
+- commit `recipe/meta.yaml` to branch of fork and submit PR

docs/package/README.md

@@ -150,3 +150,9 @@ But `odbcinst -j` revealed that it installed the configuration in `/etc/unixODBC
 its own `odbcinst` executable and that shows odbc config files are expected in `/etc/*`. Symlinks were enough to fix the
 problem. Try `python -c 'import pyodbc;print(pyodbc.drivers())'` and see whether you get more than an empty list.
 Furthermore, make sure you use 127.0.0.1 instead of localhost. It seems that /etc/hosts is ignored.
+
+### Incompatibility with specific pydiverse.transform Versions
+
+pydiverse.pipedag currently doesn't support pydiverse.transfrom Versions (0.2.0, 0.2.1, 0.2.2), due to major
+differences to pdt 0.2.3 and pdt <0.2.
+However it does still work with pdt <0.2.

docs/source/changelog.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.9.9.1 (2025-02-11)
+- Fix incompatibility with pydiverse.transform 0.2.3 (<0.2.0 still supported) (0.2.0, 0.2.1, 0.2.2 still not compatible):
+  - Added new TableHook for new and kept the old.
+  - Added Version-Detecter (new/old) with try-except imports.
+  - Added same Version-Detector method to tests/test_table_hooks/test_pdtransform
+  - Added Version-Detecter to parquet.py as well as a new TableHook
+- Added NotImplementedError("pydiverse.transform 0.2.0 isn't supported") when pydiverse.transform 0.2.0 is being used.
+- Pinned prefect to Version ">=2.13.5, <3.0.0", because future versions were incompatible
+
 ## 0.9.9 (2025-02-05)
 - Fix incompatibility with DuckDB 1.1.
 

pixi.toml

@@ -37,7 +37,7 @@ kazoo = ">=2.8.0"
 dask = ">=2022.1.0"
 
 [feature.prefect.dependencies]
-prefect = ">=2.13.5"
+prefect = ">=2.13.5, <3.0.0"
 
 [feature.snowflake.dependencies]
 snowflake-sqlalchemy = ">=1.6.1"
@@ -58,7 +58,6 @@ tidypolars = ">=0.2.19"
 
 [feature.all-tests.dependencies]
 # Table Hooks
-pydiverse-transform = ">=0.1.3, <0.2"
 ibis-framework = ">=5.1.0, <9"
 ibis-mssql = ">=5.1.0, <9"
 ibis-postgres = ">=5.1.0, <9"
@@ -68,15 +67,21 @@ arrow-odbc = ">=7.0.4"
 pyodbc = ">=4.0.35"
 pytsql = ">=1.1.4"
 
+[feature.pdtransform-new.dependencies]
+pydiverse-transform = ">0.2.0"
+
+[feature.pdtransform-old.dependencies]
+pydiverse-transform = "<0.2"
+
 # IBM DB2
 [feature.ibm-db]
 # ibm-db is not available on Apple Silicon
 # https://ibm-data-and-ai.ideas.ibm.com/ideas/DB2CON-I-92
-platforms = ["linux-64", "osx-64", "win-64"]
-[feature.ibm-db.target.osx.pypi-dependencies]
-# ibm_db is not on conda-forge for macOS
-# https://github.com/ibmdb/db2drivers/issues/3
-ibm-db = ">=3.1.4"
+platforms = ["linux-64", "win-64"]  # TODO "osx-arm64"
+#[feature.ibm-db.target.osx.pypi-dependencies]
+## ibm_db is not on conda-forge for macOS
+## https://github.com/ibmdb/db2drivers/issues/3
+#ibm-db = ">=3.1.4"
 [feature.ibm-db.target.linux.dependencies]
 ibm_db = ">=3.1.4"
 [feature.ibm-db.target.win.dependencies]
@@ -132,8 +137,8 @@ connectorx = ">=0.3.3"
 connectorx = ">=0.3.3"
 
 [environments]
-default = ["py312", "all-tests", "dev", "filelock", "zookeeper", "dask", "prefect", "pd2", "sa2", "snowflake"]
-all-tests = ["py312", "all-tests", "dev", "filelock", "zookeeper", "dask", "prefect", "pd2", "sa2", "snowflake"]
+default = ["py312", "all-tests", "pdtransform-new", "dev", "filelock", "zookeeper", "dask", "prefect", "pd2", "sa2", "snowflake"]
+all-tests = ["py312", "all-tests", "pdtransform-new", "dev", "filelock", "zookeeper", "dask", "prefect", "pd2", "sa2", "snowflake"]
 docs = ["docs"]
 release = { features=["release"], no-default-feature=true }
 py39 = ["py39", "dev", "zookeeper", "pd2", "sa2", "dask", "duckdb1"]
@@ -143,11 +148,10 @@ py312 = ["py312", "dev", "zookeeper", "pd2", "sa2", "dask", "duckdb1"]
 py39pdsa1 = ["py39", "dev", "zookeeper", "pd1", "sa1", "dask", "connectorx-noarm", "duckdb0"]
 py311pdsa1 = ["py311", "dev", "zookeeper", "pd1", "sa1", "dask", "connectorx-noarm", "duckdb0"]
 py312pdsa1 = ["py39", "dev", "zookeeper", "pd1", "sa1", "dask", "connectorx-noarm", "duckdb0"]
-py39all = ["py39", "dev", "zookeeper", "all-tests", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake", "tidypolars"]
-py310all = ["py310", "dev", "zookeeper", "all-tests", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake", "tidypolars"]
-py311all = ["py311", "dev", "zookeeper", "all-tests", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake"]
-py312all = ["py312", "dev", "zookeeper", "all-tests", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake"]
-py39pdsa1all = ["py39", "dev", "zookeeper", "pd1", "sa1", "connectorx-noarm", "all-tests", "filelock",  "dask", "prefect", "snowflake", "tidypolars"]
-py311pdsa1all = ["py311", "dev", "zookeeper", "pd1", "sa1", "connectorx-noarm", "all-tests", "filelock",  "dask", "prefect", "snowflake"]
-py39ibm = ["py39", "dev", "zookeeper", "pd2", "sa2", "dask", "ibm-db", "all-tests", "tidypolars"]
-py312ibm = ["py312", "dev", "zookeeper", "pd2", "sa2", "dask", "ibm-db", "all-tests"]
+py310all = ["py310", "dev", "zookeeper", "all-tests", "pdtransform-new", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake", "tidypolars"]
+py311all = ["py311", "dev", "zookeeper", "all-tests", "pdtransform-new", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake"]
+py312all = ["py312", "dev", "zookeeper", "all-tests", "pdtransform-new", "filelock", "dask", "prefect", "pd2", "sa2", "snowflake"]
+py39pdsa1all = ["py39", "dev", "zookeeper", "pd1", "sa1", "connectorx-noarm", "all-tests", "pdtransform-old", "filelock",  "dask", "prefect", "snowflake", "tidypolars"]
+py311pdsa1all = ["py311", "dev", "zookeeper", "pd1", "sa1", "connectorx-noarm", "all-tests", "pdtransform-old", "filelock",  "dask", "prefect", "snowflake"]
+py310ibm = ["py310", "dev", "zookeeper", "pd2", "sa2", "dask", "ibm-db", "all-tests", "pdtransform-old", "tidypolars"]
+py312ibm = ["py312", "dev", "zookeeper", "pd2", "sa2", "dask", "ibm-db", "all-tests", "pdtransform-old"]

src/pydiverse/pipedag/backend/table/cache/base.py

@@ -94,7 +94,7 @@ def _retrieve_table_obj(self, table: Table, as_type: type[T]) -> T:
             self.logger.info("Retrieved table from local table cache", table=table)
             return obj
         except Exception as e:
-            self.logger.info(
+            self.logger.warning(
                 "Failed to retrieve table from local table cache",
                 table=table,
                 cause=str(e),

src/pydiverse/pipedag/backend/table/cache/parquet.py

@@ -190,12 +190,37 @@ def retrieve(
 
 try:
     import pydiverse.transform as pdt
+
+    try:
+        from pydiverse.transform.eager import PandasTableImpl
+
+        _ = PandasTableImpl
+
+        pdt_old = pdt
+        pdt_new = None
+    except ImportError:
+        try:
+            # detect if 0.2 or >0.2 is active
+            # this import would only work in <=0.2
+            from pydiverse.transform.extended import Polars
+
+            # ensures a "used" state for the import, preventing black from deleting it
+            _ = Polars
+
+            pdt_old = None
+            pdt_new = pdt
+        except ImportError:
+            raise NotImplementedError(
+                "pydiverse.transform 0.2.0 isn't supported"
+            ) from None
 except ImportError:
     pdt = None
+    pdt_old = None
+    pdt_new = None
 
 
-@ParquetTableCache.register_table(pdt)
-class PydiverseTransformTableHook(TableHook[ParquetTableCache]):
+@ParquetTableCache.register_table(pdt_old)
+class PydiverseTransformTableHookOld(TableHook[ParquetTableCache]):
     @classmethod
     def can_materialize(cls, type_) -> bool:
         return issubclass(type_, pdt.Table)
@@ -240,3 +265,52 @@ def retrieve(
             return pdt.Table(PandasTableImpl(table.name, df))
 
         raise ValueError(f"Invalid type {as_type}")
+
+
+@ParquetTableCache.register_table(pdt_new)
+class PydiverseTransformTableHookNew(TableHook[ParquetTableCache]):
+    @classmethod
+    def can_materialize(cls, type_) -> bool:
+        return issubclass(type_, pdt.Table)
+
+    @classmethod
+    def can_retrieve(cls, type_) -> bool:
+        from pydiverse.transform.extended import Pandas
+
+        return issubclass(type_, Pandas)
+
+    @classmethod
+    def materialize(
+        cls, store: ParquetTableCache, table: Table[pdt.Table], stage_name: str
+    ):
+        from pydiverse.transform.extended import (
+            Polars,
+            export,
+        )
+
+        t = table.obj
+        table = table.copy_without_obj()
+
+        try:
+            table.obj = t >> export(Polars())
+            hook = store.get_hook_subclass(PolarsTableHook)
+            return hook.materialize(store, table, stage_name)
+        except Exception as e:
+            raise TypeError(f"Unsupported type {type(t._ast).__name__}") from e
+
+    @classmethod
+    def retrieve(
+        cls,
+        store: ParquetTableCache,
+        table: Table,
+        stage_name: str | None,
+        as_type: type,
+    ):
+        from pydiverse.transform.extended import Polars
+
+        if isinstance(as_type, Polars):
+            hook = store.get_hook_subclass(PandasTableHook)
+            df = hook.retrieve(store, table, stage_name, pd.DataFrame)
+            return pdt.Table(df)
+
+        raise ValueError(f"Invalid type {as_type}")