Merge remote-tracking branch 'origin/main' into reduce-errors

windiana42 · windiana42 · commit 33e833ae7f84 · 2025-10-10T12:03:38.000+02:00
diff --git a/docs/package/README.md b/docs/package/README.md
@@ -13,8 +13,26 @@ comment on existing issues to extend them to your needs or to add solution ideas
 ## Usage
 
 pydiverse.pipedag can either be installed via pypi with `pip install pydiverse-pipedag pydot` or via
-conda-forge with `conda install pydiverse-pipedag pydot -c conda-forge`. If you don't use duckdb for
-testing, you can obmit it here. However, it is needed to run the following example.
+conda-forge with `conda install pydiverse-pipedag pydot -c conda-forge`. Our recommendation would be
+to use [pixi](https://pixi.sh/latest/) which is also based on conda-forge:
+
+```bash
+mkdir my_project
+pixi init
+pixi add pydiverse-pipedag pydot
+```
+
+With pixi, you run python like this:
+
+```bash
+pixi run python -c 'import pydiverse.pipedag'
+```
+
+or this:
+
+```bash
+pixi run python my_script.py
+```
 
 ## Example
 
diff --git a/docs/source/changelog.md b/docs/source/changelog.md
@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.12.0 (2025-10-07)
+- Support pydiverse.common 0.4.1, pydiverse.transform 0.6.0, pydiverse.colspec 0.3.0.
+- Structlog logger initialization changed to stdlib logger factory to support dynamic loglevel filter in tests.
+- Switch cache misses from warning to info log level.
+
 ## 0.11.0 (2025-10-01)
 - Support View as task output to allow multi-parquet fusion in ParquetTableStore or basic column selection/renaming
   outside consumer task.
diff --git a/docs/source/examples/multi_instance_pipeline.md b/docs/source/examples/multi_instance_pipeline.md
@@ -95,7 +95,7 @@ technical_setups:
 
       args:
         create_database_if_not_exists: True
-        # print select statements before being encapsualted in materialize expressions and tables before writing to
+        # print select statements before being encapsulated in materialize expressions and tables before writing to
         # database
         print_materialize: true
         # print final sql statements
diff --git a/docs/source/reference/config.md b/docs/source/reference/config.md
@@ -689,7 +689,7 @@ technical_setups:
 
       args:
         create_database_if_not_exists: True
-        # print select statements before being encapsualted in materialize expressions and tables before writing to
+        # print select statements before being encapsulated in materialize expressions and tables before writing to
         # database
         print_materialize: true
         # print final sql statements
diff --git a/pixi.lock b/pixi.lock
diff --git a/pixi.toml b/pixi.toml
@@ -5,7 +5,7 @@ platforms = ["linux-64", "osx-64", "osx-arm64", "win-64"] # "linux-aarch64"
 
 [dependencies]
 python = ">=3.10.18,<3.14"
-pydiverse-common = ">=0.3.15,<0.4"
+pydiverse-common = ">=0.4.1,<0.5"
 typing-extensions = ">=4.14.1,<5"
 networkx = ">=3.4,<4"
 attrs = ">=25.3.0,<26"
@@ -15,17 +15,17 @@ msgpack-python = ">=1.1.1,<2"
 packaging = ">=24.2,<26"
 python-box = ">=7.3.2,<8"
 pyyaml = ">=6.0.2,<7"
-cryptography = ">=45.0.5,<46"
+cryptography = ">=45.0.5,<47"
 click = ">=8.1.8,<9"
 pyparsing = ">=3.2.3,<4"
 sqlalchemy = ">=1.4.54,<3"
 pandas = ">=1.5.3,<3"
-pyarrow = ">=20.0.0,<21"
-duckdb = ">=0.10.3,<2"
+pyarrow = ">=20.0.0,<22"
+duckdb = ">=0.10.3,<1.4"
 duckdb-engine = ">=0.15.0,<0.18"
 polars = ">=1.2.1,<2"
 fsspec = ">=2025.5.1,<2026"
-universal_pathlib = ">=0.2.6,<0.3"
+universal_pathlib = ">=0.2.6,<0.4"
 
 
 [target.osx-arm64.dependencies]
@@ -54,12 +54,11 @@ kazoo = ">=2.8.0"
 dask = ">=2022.1.0"
 
 [feature.prefect2.dependencies]
-prefect = ">=2.13.5,<3a0"
+prefect = ">=2.14.2,<3a0"
 pydantic = ">=1.10.9"
 
 [feature.prefect3.dependencies]
-prefect = ">=3.4.6,<4a0"
-whenever = ">=0.8.5,<1.0"
+prefect = ">=3.4.22,<4a0"
 
 [feature.snowflake.dependencies]
 snowflake-sqlalchemy = ">=1.6.1"
@@ -85,13 +84,13 @@ pytsql = ">=1.4.0,<2"
 bcpandas = ">=2.6.3,<3"
 
 [feature.pdtransform-new.dependencies]
-pydiverse-transform = ">=0.5.6,<0.6"
+pydiverse-transform = ">=0.6.0,<0.7"
 
 [feature.pdtransform-old.dependencies]
 pydiverse-transform = "<0.2"
 
 [feature.colspec.dependencies]
-pydiverse-colspec = ">=0.2.8,<0.3"
+pydiverse-colspec = ">=0.3.0,<0.4"
 
 [feature.dataframely.dependencies]
 dataframely = ">=1.2.1,<2"
@@ -101,7 +100,8 @@ dataframely = ">=1.2.1,<2"
 platforms = ["linux-64", "osx-arm64", "win-64"]
 
 [feature.ibm-db.dependencies]
-ibm_db = ">=3.2.5, <3.2.7"  # somehoew there is a version conflict for 3.2.7 that fails empty queries
+#Current pixi install keeps ibm_db 3.2.6 which works.
+ibm_db = ">=3.2.5, <3.2.7"  # somehow there is a version conflict for 3.2.7 that fails empty queries
 ibm_db_sa = ">=0.3.8"
 
 # S3
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "pydiverse-pipedag"
-version = "0.11.0"
+version = "0.12.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." },
@@ -30,7 +30,7 @@ classifiers = [
 ]
 
 dependencies = [
-  "pydiverse-common >=0.3.15,<0.4",
+  "pydiverse-common >=0.4.1,<0.5",
   "typing-extensions >=4.14.1,<5",
   "networkx >=3.4,<4",
   "attrs >=25.3.0,<26",
@@ -40,17 +40,17 @@ dependencies = [
   "packaging >=24.2,<26",
   "python-box >=7.3.2,<8",
   "pyyaml >=6.0.2,<7",
-  "cryptography >=45.0.5,<46",
+  "cryptography >=45.0.5,<47",
   "click >=8.1.8,<9",
   "pyparsing >=3.2.3,<4",
   "sqlalchemy >=1.4.54,<3",
   "pandas >=1.5.3,<3",
-  "pyarrow >=20.0.0,<21",
-  "duckdb >=0.10.3,<2",             # py311pdsa1 requires < 1.3.0
+  "pyarrow >=20.0.0,<22",
+  "duckdb >=0.10.3,<1.4",           # 1.4.0 has bug with sqlalchemy; py311pdsa1 requires < 1.3.0
   "duckdb-engine >=0.15.0,<0.18",
   "polars >=1.2.1,<2",              # py311pdsa1all requires < 1.30.0
   "fsspec >=2025.5.1,<2026",
-  "universal_pathlib >=0.2.6,<0.3",
+  "universal_pathlib >=0.2.6,<0.4",
 ]
 
 [tool.hatch.build.targets.wheel]
diff --git a/pytest.ini b/pytest.ini
@@ -16,7 +16,7 @@ markers =
     dask: a test that requires dask [DaskEngine]
     prefect: a test that requires prefect [PrefectEngine]
 
-    lock_tests: tests exercising lock manager implementations (enabled by default, can be disabled with --no-lock-tests)
+    lock_tests: tests exercising lock manager implementations (enabled by default, can be disabled with --no-lock_tests)
 
     instances: marker used to run an test with different instances
     skip_instances: fixture used to skip running test for a list of instances
diff --git a/src/pydiverse/pipedag/backend/table/cache/parquet.py b/src/pydiverse/pipedag/backend/table/cache/parquet.py
@@ -206,7 +206,7 @@ def materialize(
 
         if isinstance(df, pl.LazyFrame):
             df = df.collect()
-        df.write_parquet(path)
+        df.write_parquet(str(path))
         # intentionally don't apply annotation checks because they might also be done
         # within polars table hook of actual table store
 
@@ -222,7 +222,7 @@ def _execute_query(
     ) -> pl.DataFrame:
         _ = as_type
         path = store.get_table_path(table, ".parquet")
-        df = pl.read_parquet(path, n_rows=limit)
+        df = pl.read_parquet(str(path), n_rows=limit)
         if issubclass(as_type, pl.LazyFrame):
             return df.lazy()
         return df
diff --git a/src/pydiverse/pipedag/backend/table/sql/dialects/duckdb_parquet.py b/src/pydiverse/pipedag/backend/table/sql/dialects/duckdb_parquet.py
@@ -911,7 +911,9 @@ def materialize(
             )
 
         df = table.obj
-        df.to_parquet(file_path, index=False, storage_options=store.get_storage_options("fsspec", file_path.protocol))
+        df.to_parquet(
+            str(file_path), index=False, storage_options=store.get_storage_options("fsspec", file_path.protocol)
+        )
         store.execute(CreateViewAsSelect(table.name, schema, store._read_parquet_query(file_path)))
         store.metadata_track_view(table.name, schema.get(), file_path.as_uri(), "parquet")
 
@@ -968,7 +970,7 @@ def retrieve(
             if limit is not None:
                 df = ds.dataset(pyarrow_path, filesystem=pyarrow_fs).scanner().head(limit).to_pandas()
             else:
-                df = pd.read_parquet(path, storage_options=store.get_storage_options("fsspec", path.protocol))
+                df = pd.read_parquet(str(path), storage_options=store.get_storage_options("fsspec", path.protocol))
         import pyarrow.parquet
 
         # attention: with categorical columns, it might be necessary to fuse dictionaries of all parquet files
@@ -1050,9 +1052,10 @@ def _execute_materialize_polars(table, store, stage_name):
                     "Storing polars tables with custom storage options is not supported for polars < 1.3.0. "
                     f"Current version is {pl.__version__}: {options}"
                 )
-            df.write_parquet(file_path, storage_options=options)
+            # at some point polars supported UPath, but 1.33.1 does not
+            df.write_parquet(str(file_path), storage_options=options)
         else:
-            df.write_parquet(file_path)
+            df.write_parquet(str(file_path))
         store.execute(CreateViewAsSelect(table.name, schema, store._read_parquet_query(file_path)))
         store.metadata_track_view(table.name, schema.get(), file_path.as_uri(), "parquet")
 
@@ -1071,9 +1074,11 @@ def _execute_query(
             if isinstance(view.src, Iterable):
                 file_paths = [store.get_table_path(tbl) for tbl in view.src]
                 protocol = file_paths[0].protocol
+                file_paths = [str(path) for path in file_paths]
             else:
                 file_paths = store.get_table_path(view.src)
                 protocol = file_paths.protocol
+                file_paths = str(file_paths)
             lf = pl.scan_parquet(
                 file_paths, n_rows=limit, storage_options=store.get_storage_options("polars", protocol)
             )
@@ -1090,7 +1095,7 @@ def _execute_query(
         else:
             file_path = store.get_table_path(table)
             df = pl.read_parquet(
-                file_path, n_rows=limit, storage_options=store.get_storage_options("polars", file_path.protocol)
+                str(file_path), n_rows=limit, storage_options=store.get_storage_options("polars", file_path.protocol)
             )
             return df
 
diff --git a/src/pydiverse/pipedag/backend/table/sql/sql.py b/src/pydiverse/pipedag/backend/table/sql/sql.py
@@ -855,14 +855,14 @@ def copy_indexes(self, src_table: str, src_schema: Schema, dest_table: str, dest
         for index in indexes:
             if len(index["include_columns"]) > 0:
                 self.logger.warning(
-                    "Perfect index recreation in caching is not net implemented",
+                    "Perfect index recreation in caching is not implemented, yet",
                     table=dest_table,
                     schema=dest_schema.get(),
                     include_columns=index["include_columns"],
                 )
             if any(not isinstance(val, list) or len(val) > 0 for val in index["dialect_options"].values()):
                 self.logger.warning(
-                    "Perfect index recreation in caching is not net implemented",
+                    "Perfect index recreation in caching is not implemented, yet",
                     table=dest_table,
                     schema=dest_schema.get(),
                     dialect_options=index["dialect_options"],
diff --git a/src/pydiverse/pipedag/engine/dask.py b/src/pydiverse/pipedag/engine/dask.py
@@ -1,7 +1,6 @@
 # Copyright (c) QuantCo and pydiverse contributors 2025-2025
 # SPDX-License-Identifier: BSD-3-Clause
 
-import sys
 
 import structlog
 
@@ -62,7 +61,7 @@ def run(parent_futures, **kwargs):
                 _ = parent_futures
 
                 # TODO: Don't just assume a logger factory...
-                structlog_config["logger_factory"] = structlog.PrintLoggerFactory(sys.stderr)
+                structlog_config["logger_factory"] = structlog.stdlib.LoggerFactory()
                 structlog.configure(**structlog_config)
 
                 with structlog.contextvars.bound_contextvars(**structlog_context):
diff --git a/src/pydiverse/pipedag/materialize/store.py b/src/pydiverse/pipedag/materialize/store.py
@@ -188,7 +188,7 @@ def store_table_lazy(
         except CacheError as e:
             # Either not found in cache, or copying failed
             # -> Store using default method
-            self.logger.warning("Cache miss", table=table.name, stage=table.stage.name, cause=str(e))
+            self.logger.info("Cache miss", table=table.name, stage=table.stage.name, cause=str(e))
             TaskContext.get().is_cache_valid = False
             RunContext.get().trace_hook.query_cache_status(
                 task, table, task_cache_info, query_hash, query_str, cache_valid=False
@@ -257,7 +257,7 @@ def store_raw_sql(self, raw_sql: RawSql, task: MaterializingTask, task_cache_inf
         except CacheError as e:
             # Either not found in cache, or copying failed
             # -> Store using default method
-            self.logger.warning("Cache miss for raw-SQL", cause=str(e))
+            self.logger.info("Cache miss for raw-SQL", cause=str(e))
 
             TaskContext.get().is_cache_valid = False
             RunContext.get().set_stage_has_changed(task._stage)
@@ -562,13 +562,19 @@ def _retrieve_table_obj(self, table: Table, as_type: type[T]) -> T:
         except StoreIncompatibleException:
             # This is expected for example when ParquetTableCache is asked to retrieve a SQL reference
             return None
+        except CacheError as e:
+            self.logger.info(
+                "Cache miss in local table cache",
+                table=table,
+                cause=str(e),
+            )
         except Exception as e:
             self.logger.warning(
                 "Failed to retrieve table from local table cache",
                 table=table,
                 cause=str(e),
             )
-            return None
+        return None
 
     @abstractmethod
     def _has_table(self, table: Table, as_type: type) -> bool:
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -50,6 +50,7 @@ def setup_environ():
     if os.environ.get("DEBUG", "0") == "0"
     else logging.DEBUG
 )
+log_level = logging.ERROR
 setup_logging(log_level=log_level)
 
 
diff --git a/tests/parallelize/plugin.py b/tests/parallelize/plugin.py
@@ -3,7 +3,7 @@
 
 import pytest
 
-from .sesson import Session
+from .session import Session
 from .util import parse_config
 
 
diff --git a/tests/parallelize/session.py b/tests/parallelize/session.py
diff --git a/tests/parallelize/worker.py b/tests/parallelize/worker.py
@@ -1,7 +1,6 @@
 # Copyright (c) QuantCo and pydiverse contributors 2025-2025
 # SPDX-License-Identifier: BSD-3-Clause
 
-from abc import ABC
 from multiprocessing import Queue
 
 import pytest
@@ -14,19 +13,21 @@ def start_worker(worker_id: int, work_queue: Queue, msg_queue: Queue, args: list
     config = Config.fromdictargs(option_dict, args)
     config.args = args
 
-    from typing import TextIO
-
-    class DontPrint(TextIO, ABC):
-        def write(*_):
-            pass
-
-    # fix assert inspection code of pytest raised in threads
-    # register dummy terminal reporter since it is needed by pytest even with
-    # plugins:"no:terminal" option
-    from _pytest.terminal import TerminalReporter
-
-    terminal_reporter = TerminalReporter(config, DontPrint())
-    config.pluginmanager.register(terminal_reporter, "terminalreporter")
+    # # This code was needed in the past to enable debugging in workers.
+    # # Currently, it doesn't seem to be needed any more.
+    # from typing import TextIO
+    #
+    # class DontPrint(TextIO, ABC):
+    #     def write(*_):
+    #         pass
+    #
+    # # fix assert inspection code of pytest raised in threads
+    # # register dummy terminal reporter since it is needed by pytest even with
+    # # plugins:"no:terminal" option
+    # from _pytest.terminal import TerminalReporter
+    #
+    # terminal_reporter = TerminalReporter(config, DontPrint())
+    # config.pluginmanager.register(terminal_reporter, "terminalreporter")
 
     # Remove workers option to prevent triggering main plugin
     config.option.workers = None
diff --git a/tests/test_flows/complex_config_flows/pipedag_anchor.yaml b/tests/test_flows/complex_config_flows/pipedag_anchor.yaml
@@ -57,7 +57,7 @@ _technical_setups:
         #<<: *db_mssql
 
         create_database_if_not_exists: True
-        # print select statements before being encapsualted in materialize expressions and tables before writing to
+        # print select statements before being encapsulated in materialize expressions and tables before writing to
         # database
         print_materialize: true
         # print final sql statements
diff --git a/tests/test_flows/complex_config_flows/pipedag_complex.yaml b/tests/test_flows/complex_config_flows/pipedag_complex.yaml
diff --git a/tests/test_lock_manager.py b/tests/test_lock_manager.py
diff --git a/tests/test_materializing_task.py b/tests/test_materializing_task.py
