fix: Fix materialization details error handling (#282)

* fix: Fix materialization details error handling

* pixi update

* Somehow sa.select("*") can create a race condition with prefect backend. Thus changed to sa.select(sa.text("*")).

This error vanished always on second run:

Split tests into 2 parallelization groups: dask_engine, prefect_engine
..............F.................................................ssssssss [ 52%]
s..................................................sssssssss....         [100%]
=================================== FAILURES ===================================
______________ test_materialize_view_union[prefect_engine-False] _______________

instance = '*'

    @inspection._inspects(object)
    def _inspect_mapped_object(instance: _T) -> Optional[InstanceState[_T]]:
        try:
>           return instance_state(instance)
                   ^^^^^^^^^^^^^^^^^^^^^^^^
E           AttributeError: 'str' object has no attribute '_sa_instance_state'

.pixi/envs/py311all/lib/python3.11/site-packages/sqlalchemy/orm/base.py:430: AttributeError

* Prevent version conflict for ibm_db=3.2.7

* Convert auto_table error to warning in pipedag tests.

It is a bit lame to catch one specific type but this type is helpful in the general auto_table arguments and this is hard to make dependent on pydiverse.transform being installed.

* Convert auto_table error to warning in pipedag tests.

Prepare release 0.11.0

---------

Co-authored-by: windiana42 <windiana@users.sf.net>

Author: nicolasmueller

pixi.lock

@@ -101,7 +101,7 @@ dataframely = ">=1.2.1,<2"
 platforms = ["linux-64", "osx-arm64", "win-64"]
 
 [feature.ibm-db.dependencies]
-ibm_db = ">=3.2.5"
+ibm_db = ">=3.2.5, <3.2.7"  # somehoew there is a version conflict for 3.2.7 that fails empty queries
 ibm_db_sa = ">=0.3.8"
 
 # S3
@@ -127,6 +127,7 @@ sphinx-copybutton = ">=0.5.2"
 [tasks]
 postinstall = "pip install --no-build-isolation --no-deps --disable-pip-version-check -e ."
 zip-examples = "sh -c 'set -ex; for d in example*; do if test -f $d/run_pipeline.py; then echo $d; cd $d; pixi install; cd -; rm -r $d/.pixi $d/__pycache__ || echo ignore; rm docs/source/examples/zip/$d.zip || echo $d.zip is new; zip -r docs/source/examples/zip/$d.zip $d/; fi; done'"
+install-all = "sh -c 'cd .pixi/envs/ && for i in */; do pixi install -e $(basename $i /); done && cd -'"
 
 [feature.docs.tasks]
 docs = "cd docs && make html "

pixi.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pydiverse-pipedag"
-version = "0.10.12"
+version = "0.11.0"
 description = "A pipeline orchestration library executing tasks within one python session. It takes care of SQL table (de)materialization, caching and cache invalidation. Blob storage is supported as well for example for storing model files."
 authors = [
   { name = "QuantCo, Inc." },

pyproject.toml

@@ -876,7 +876,7 @@ def insert_into_in_query(select_sql, schema, table):
 def visit_create_alias(create_alias: CreateAlias, compiler, **kw):
     from_name = compiler.preparer.quote(create_alias.from_name)
     from_schema = compiler.preparer.format_schema(create_alias.from_schema.get())
-    query = sa.select("*").select_from(sa.text(f"{from_schema}.{from_name}"))
+    query = sa.select(sa.text("*")).select_from(sa.text(f"{from_schema}.{from_name}"))
     return compiler.process(CreateViewAsSelect(create_alias.to_name, create_alias.to_schema, query), **kw)
 
 
@@ -912,7 +912,7 @@ def visit_create_alias(create_alias: CreateAlias, compiler, **kw):
 def visit_copy_table(copy_table: CopyTable, compiler, **kw):
     from_name = compiler.preparer.quote(copy_table.from_name)
     from_schema = compiler.preparer.format_schema(copy_table.from_schema.get())
-    query = sa.select("*").select_from(sa.text(f"{from_schema}.{from_name}"))
+    query = sa.select(sa.text("*")).select_from(sa.text(f"{from_schema}.{from_name}"))
     create = CreateTableAsSelect(
         copy_table.to_name,
         copy_table.to_schema,

src/pydiverse/pipedag/backend/table/sql/ddl.py

@@ -186,12 +186,10 @@ def resolve_alias(self, table: Table, stage_name: str) -> tuple[str, str]:
         table_name, schema = super().resolve_alias(table, stage_name)
         return PipedagDB2Reflection.resolve_alias(self.engine, table_name, schema)
 
-    def check_materialization_details_supported(self, label: str | None) -> None:
-        _ = label
-        return
-
-    def _set_materialization_details(self, materialization_details: dict[str, dict[str | list[str]]] | None) -> None:
-        self.materialization_details = IBMDB2MaterializationDetails.create_materialization_details_dict(
+    def _create_materialization_details(
+        self, materialization_details: dict[str, dict[str | list[str]]] | None
+    ) -> BaseMaterializationDetails:
+        return IBMDB2MaterializationDetails.create_materialization_details_dict(
             materialization_details,
             self.strict_materialization_details,
             self.default_materialization_details,

src/pydiverse/pipedag/backend/table/sql/dialects/ibm_db2.py

@@ -372,8 +372,10 @@ def resolve_alias(self, table: Table, stage_name: str):
         table_name, schema = super().resolve_alias(table, stage_name)
         return PipedagMSSqlReflection.resolve_alias(self.engine, table_name, schema)
 
-    def _set_materialization_details(self, materialization_details: dict[str, dict[str | list[str]]] | None) -> None:
-        self.materialization_details = MSSqlMaterializationDetails.create_materialization_details_dict(
+    def _create_materialization_details(
+        self, materialization_details: dict[str, dict[str | list[str]]] | None
+    ) -> BaseMaterializationDetails:
+        return MSSqlMaterializationDetails.create_materialization_details_dict(
             materialization_details,
             self.strict_materialization_details,
             self.default_materialization_details,

src/pydiverse/pipedag/backend/table/sql/dialects/mssql.py

@@ -109,12 +109,10 @@ def lock_source_table(self, table: Table | str, schema: Schema | str, conn: Any
             self.execute(stmt, conn=conn)
         return [stmt]
 
-    def check_materialization_details_supported(self, label: str | None) -> None:
-        _ = label
-        return
-
-    def _set_materialization_details(self, materialization_details: dict[str, dict[str | list[str]]] | None) -> None:
-        self.materialization_details = PostgresMaterializationDetails.create_materialization_details_dict(
+    def _create_materialization_details(
+        self, materialization_details: dict[str, dict[str | list[str]]] | None
+    ) -> BaseMaterializationDetails:
+        return PostgresMaterializationDetails.create_materialization_details_dict(
             materialization_details,
             self.strict_materialization_details,
             self.default_materialization_details,

src/pydiverse/pipedag/backend/table/sql/dialects/postgres.py

@@ -405,7 +405,7 @@ def materialize(
     ):
         query = table.obj
         if isinstance(table.obj, (sa.Table, sa.sql.expression.Alias)):
-            query = sa.select("*").select_from(table.obj)
+            query = sa.select(sa.text("*")).select_from(table.obj)
             tbl = table.obj
             while hasattr(tbl, "original"):
                 tbl = tbl.original
@@ -617,7 +617,7 @@ def retrieve(
     @classmethod
     def lazy_query_str(cls, store, obj) -> str:
         if isinstance(obj, sa.sql.expression.FromClause):
-            query = sa.select("*").select_from(obj)
+            query = sa.select(sa.text("*")).select_from(obj)
         else:
             query = obj
         query_str = str(query.compile(store.engine, compile_kwargs={"literal_binds": True}))
@@ -1062,7 +1062,7 @@ def _build_retrieve_query(
         cols, dtypes = cls._adjust_cols_retrieve(store, cols, dtypes)
 
         if cols is None:
-            query = sa.select("*").select_from(query)
+            query = sa.select(sa.text("*")).select_from(query)
         else:
             query = sa.select(*cols.values()).select_from(query)
         if limit is not None:

src/pydiverse/pipedag/backend/table/sql/hooks.py

@@ -8,15 +8,15 @@ class PipedagDB2Reflection:
     @staticmethod
     def get_alias_names(engine: Engine, schema: str) -> list[str]:
         """Returns all aliases in a schema"""
-        return PipedagDB2Reflection._get_tabname(engine, schema, "A")
+        return PipedagDB2Reflection._get_tabnames(engine, schema, "A")
 
     @staticmethod
     def get_nickname_names(engine: Engine, schema: str) -> list[str]:
         """Returns all nicknames in a schema"""
-        return PipedagDB2Reflection._get_tabname(engine, schema, "N")
+        return PipedagDB2Reflection._get_tabnames(engine, schema, "N")
 
     @staticmethod
-    def _get_tabname(engine: Engine, schema: str, _type: str):
+    def _get_tabnames(engine: Engine, schema: str, _type: str):
         schema = engine.dialect.denormalize_name(schema)
         query = f"""
         SELECT TABNAME

src/pydiverse/pipedag/backend/table/sql/reflection.py

@@ -40,7 +40,7 @@
 )
 from pydiverse.pipedag.context.run_context import DeferredTableStoreOp
 from pydiverse.pipedag.errors import CacheError
-from pydiverse.pipedag.materialize.details import resolve_materialization_details_label
+from pydiverse.pipedag.materialize.details import BaseMaterializationDetails, resolve_materialization_details_label
 from pydiverse.pipedag.materialize.materializing_task import MaterializingTask
 from pydiverse.pipedag.materialize.metadata import (
     LazyTableMetadata,
@@ -344,7 +344,7 @@ def __init__(
 
         self.default_materialization_details = default_materialization_details
 
-        self._set_materialization_details(materialization_details)
+        self.materialization_details = self._create_materialization_details(materialization_details)
 
         self.logger.info(
             "Initialized SQL Table Store",
@@ -611,18 +611,27 @@ def drop_subquery_table(
     def check_materialization_details_supported(self, label: str | None) -> None:
         if label is None:
             return
-        error_msg = f"Materialization details are not supported for store {type(self).__name__}."
+        if self.materialization_details and label in self.materialization_details:
+            return
+        msg = f"{label} is an unknown materialization details label."
         if self.strict_materialization_details:
-            raise ValueError(f"{error_msg} To silence this exception set strict_materialization_details=False")
+            raise ValueError(f"{msg} To silence this exception set strict_materialization_details=False")
         else:
-            self.logger.warning(f"{error_msg}")
+            self.logger.warning(msg)
+
+    def _create_materialization_details(
+        self, materialization_details: dict[str, dict[str | list[str]]] | None
+    ) -> BaseMaterializationDetails | None:
+        """This function causes an error or exception if materialization details are specified for a table
+        store that does not support them.
 
-    def _set_materialization_details(self, materialization_details: dict[str, dict[str | list[str]]] | None) -> None:
+        Any table store that supports materialization details should override this method."""
         if materialization_details is not None or self.default_materialization_details is not None:
-            error_msg = f"{type(self).__name__} does not support materialization details."
+            msg = f"{type(self).__name__} does not support materialization details."
             if self.strict_materialization_details:
-                raise TypeError(f"{error_msg} To suppress this exception, use strict_materialization_details=False")
-            self.logger.error(error_msg)
+                raise ValueError(f"{msg} To silence this exception, use strict_materialization_details=False")
+            self.logger.warning(msg)
+        return None  # no materialization details object
 
     def get_unlogged(self, materialization_details_label: str | None) -> bool:
         _ = materialization_details_label