Skip to content

Commit de25527

Browse files
fix(polars): fall back to plain collect when streaming engine is rejected (#432)
The native Polars backend requested collect(engine="streaming") and caught only TypeError, but polars 1.1-1.24 reject the value with ValueError (Invalid engine argument), which propagated and broke fd.clean on the declared polars floor. Broaden the catch to (TypeError, ValueError) so those versions fall back to a plain collect(); modern polars, which accepts the engine, is unchanged. Found by the production-readiness campaign (FDC-L1b-001).
1 parent 2d94d64 commit de25527

3 files changed

Lines changed: 34 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,11 @@ adheres to [Semantic Versioning](https://semver.org/).
77
## [Unreleased]
88

99
### Fixed
10+
- The native Polars backend no longer raises on polars versions that reject
11+
`collect(engine="streaming")` with a `ValueError` (polars 1.1–1.24): the
12+
streaming collect now falls back to a plain `collect()` on those versions,
13+
as it already did for the older `streaming=` keyword. Modern polars is
14+
unaffected.
1015
- Explicit imputation (`impute="mean"`, `"median"`, `"mode"`, `"auto"`,
1116
`"missforest"` or an `impute_strategy` entry) no longer fills the declared
1217
`id_columns` or `target_column`, as documented. Those columns keep their

‎src/freshdata/execution/backends/_polars.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -557,7 +557,10 @@ def _collect(self, lf: Any, engine_config: EngineConfig | None, pl: Any) -> Any:
557557
# keyword first, then the legacy one, then a plain collect.
558558
try:
559559
return lf.collect(engine="streaming")
560-
except TypeError:
560+
except (TypeError, ValueError):
561+
# Older polars either lacks the ``engine`` keyword (TypeError) or
562+
# rejects the ``"streaming"`` value (ValueError: Invalid engine
563+
# argument); fall through to the legacy switch, then plain collect.
561564
pass
562565
try:
563566
return lf.collect(streaming=True)

‎tests/test_execution/test_dispatch_and_config.py‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,3 +78,28 @@ def toPandas(self) -> pd.DataFrame: # noqa: N802 - matches pyspark's API
7878

7979
out = materialize_to_pandas(_FakeSparkFrame())
8080
pd.testing.assert_frame_equal(out, expected)
81+
82+
83+
def test_polars_collect_falls_back_when_streaming_engine_unsupported():
84+
"""Older polars rejects ``collect(engine="streaming")`` with ValueError (not
85+
TypeError); the backend must fall back to a plain collect, not propagate it."""
86+
from freshdata.execution._config import EngineConfig
87+
from freshdata.execution.backends._polars import PolarsEngine
88+
89+
calls: list = []
90+
91+
class _FakeLazy:
92+
def collect(self, **kwargs):
93+
calls.append(kwargs)
94+
if kwargs.get("engine") == "streaming":
95+
raise ValueError("Invalid engine argument engine='streaming'")
96+
if kwargs.get("streaming"):
97+
raise TypeError("unexpected keyword argument 'streaming'")
98+
return "collected"
99+
100+
class _Pl:
101+
pass
102+
103+
out = PolarsEngine()._collect(_FakeLazy(), EngineConfig(), _Pl())
104+
assert out == "collected"
105+
assert {} in calls # a plain collect() was reached

0 commit comments

Comments
 (0)