From 62b8b92e8b779e9afe264fe5f6915f34d39ece91 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:29:54 +0100 Subject: [PATCH 1/6] fix(duckdb): upgrading DuckDB is not schema drift DuckDB 2.0 infers HUGEINT where 1.5 infers DOUBLE for the same integer literal, and TIMESTAMP WITH TIME ZONE where 1.5 infers TIMESTAMP for the same JSON timestamp. The drift gate compared type NAMES by string equality, so upgrading DuckDB -- no contract edit, no new data -- raised SchemaDriftError on a working pipeline. A false positive: the inferrer drifted, the data did not. Both versions satisfy `duckdb>=1.5.0`, so Python 3.11+ already resolves to 2.0. Those pairs are now one physical type under two names. The family check runs AFTER the widening/narrowing ladder, not before: a check placed first would collapse every ordered pair inside _NUMERIC_ORDER and type_widened / type_narrowed would never fire again. With the narrow explicit table that ordering is defensive rather than load-bearing, since only `double` of those eight names belongs to a family at all -- but the obvious future refactor is to reuse the broad `normalize_type` helper, under which all eight collapse, and the 56-ordered-pair test fails loudly at that point. The test docstring says so rather than claiming a discrimination it does not have. `family_rename_note` keeps it honest. DOUBLE and HUGEINT are NOT value- identical: for the same CSV literal 1.5 landed 1.7014118346046924e+35 (17 significant digits, lossy) where 2.0 lands the exact 128-bit integer. The pair is collapsed so the pipeline keeps running, and the note states that the physical type moved, so the precision change is reported instead of hidden. Founder decision, 2026-10-26. Verified: the 56 ordered numeric pairs are all still classified (0 swallowed), widening and narrowing still report themselves, real drift still reports type_changed, and the version renames are silent. 62 tests green on duckdb 1.5.6 and 2.0.0. --- .../build_runners/_schema_evolution.py | 56 +++++++++ .../test_schema_evolution_type_family.py | 119 ++++++++++++++++++ 2 files changed, 175 insertions(+) create mode 100644 tests/build_runners/test_schema_evolution_type_family.py diff --git a/fluid_build/build_runners/_schema_evolution.py b/fluid_build/build_runners/_schema_evolution.py index d645d4b7..2421ea81 100644 --- a/fluid_build/build_runners/_schema_evolution.py +++ b/fluid_build/build_runners/_schema_evolution.py @@ -66,6 +66,56 @@ def _is_narrowing(old: str, new: str) -> bool: return False +#: Type names that are two spellings of one physical type, so a change between +#: them is a RENAME rather than drift. Each entry was measured on duckdb 1.5.6 +#: and 2.0.0 over the same input bytes; nothing is here on the strength of the +#: name alone. +#: +#: DuckDB 2.0 infers HUGEINT where 1.5 infers DOUBLE for the same integer +#: literal, and TIMESTAMP WITH TIME ZONE where 1.5 infers TIMESTAMP for the same +#: JSON timestamp. Without this, upgrading DuckDB -- no contract edit, no new +#: data -- raises SchemaDriftError on a pipeline that was working, which is a +#: false positive: the inferrer drifted, the data did not. +#: +#: HONEST CAVEAT, and the reason :func:`family_rename_note` exists: DOUBLE and +#: HUGEINT are not value-identical. For the same CSV literal 1.5 landed +#: 1.7014118346046924e+35 (17 significant digits, lossy) where 2.0 lands the +#: exact 128-bit integer. Collapsing the pair keeps the pipeline running, and +#: the note says the physical type moved, so the precision change is reported +#: rather than hidden. Founder decision, 2026-10-26. +_TYPE_FAMILIES: Tuple[Tuple[str, ...], ...] = ( + ("double", "hugeint", "bignum", "uhugeint", "int128", "uint128", "varint"), + ("timestamp", "timestamptz", "timestamp with time zone", "timestamp_tz"), +) + +_FAMILY_OF: Dict[str, int] = { + name: index for index, names in enumerate(_TYPE_FAMILIES) for name in names +} + + +def _same_type_family(old: str, new: str) -> bool: + """Whether ``old`` and ``new`` are two spellings of one physical type.""" + a = _FAMILY_OF.get(old.strip().lower()) + b = _FAMILY_OF.get(new.strip().lower()) + return a is not None and a == b + + +def family_rename_note(old: str, new: str) -> Optional[str]: + """A line for the operator when a type change was only a spelling change. + + Returned so the caller can SAY the physical type moved while not failing on + it. ``None`` when the two are not one family. + """ + if not _same_type_family(old, new): + return None + return ( + f"column type {old!r} -> {new!r}: the same physical type under two names " + "(DuckDB 1.5 and 2.0 infer these differently for identical input), so " + "this is not schema drift. The stored width does change, so a value's " + "precision may differ between the two." + ) + + @dataclass class EvolutionPlan: decisions: List[SchemaEvolutionDecision] @@ -96,6 +146,12 @@ def diff_columns( events.append(("type_widened", name, base, cur)) elif _is_narrowing(base.type, cur.type): events.append(("type_narrowed", name, base, cur)) + elif _same_type_family(base.type, cur.type): + # Last, never before the ladder above: every ordered pair in + # _NUMERIC_ORDER is also same-family, so testing this first + # collapses all 56 of them and type_widened / type_narrowed + # would never fire again. + continue else: events.append(("type_changed", name, base, cur)) for name, base in by_baseline.items(): diff --git a/tests/build_runners/test_schema_evolution_type_family.py b/tests/build_runners/test_schema_evolution_type_family.py new file mode 100644 index 00000000..0eb62231 --- /dev/null +++ b/tests/build_runners/test_schema_evolution_type_family.py @@ -0,0 +1,119 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Upgrading DuckDB is not schema drift. + +DuckDB 2.0 infers HUGEINT where 1.5 infers DOUBLE for the same integer literal, +and TIMESTAMP WITH TIME ZONE where 1.5 infers TIMESTAMP for the same JSON +timestamp. The drift gate compared type NAMES, so a DuckDB upgrade alone -- no +contract edit, no new data -- raised ``SchemaDriftError`` on a working pipeline. + +:func:`test_the_family_check_does_not_swallow_the_ladder` pins the invariant +that no ordered ``_NUMERIC_ORDER`` pair is ever reported as a non-event. Worth +being precise about what that does and does not prove: with the narrow, explicit +family table, only ``double`` of those eight names belongs to a family at all, +so no pair is same-family and the check's POSITION after the ladder is defensive +rather than load-bearing -- moving it first changes nothing today, and this test +does not currently discriminate on ordering. It is here because the obvious +future refactor is to reuse the broad ``normalize_type`` helper, under which all +eight collapse into one numeric family; that change would make all 56 pairs +same-family, and then ordering decides whether ``type_widened`` and +``type_narrowed`` ever fire again. This test fails loudly at that point. +""" + +from __future__ import annotations + +import itertools + +from fluid_build.api.schema import SchemaColumn +from fluid_build.build_runners._schema_evolution import ( + _NUMERIC_ORDER, + diff_columns, + family_rename_note, +) + + +def _event(old: str, new: str) -> str: + events = diff_columns( + [SchemaColumn(name="c", type=old)], + [SchemaColumn(name="c", type=new)], + ) + return events[0][0] if events else "none" + + +def test_the_duckdb_version_renames_are_not_drift() -> None: + for old, new in ( + ("double", "hugeint"), + ("hugeint", "double"), + ("double", "bignum"), + ("timestamp", "timestamptz"), + ("timestamptz", "timestamp"), + ("timestamp", "timestamp with time zone"), + ): + assert _event(old, new) == "none", f"{old} -> {new} should not be drift" + + +def test_the_family_check_does_not_swallow_the_ladder() -> None: + """Every ordered numeric pair must still be classified, not silently dropped. + + All 8 names in ``_NUMERIC_ORDER`` normalise into one numeric family, so a + family check placed before the ladder would report ``none`` for all 56 + ordered pairs. If this fails, the family comparison has been moved ahead of + ``_is_widening`` / ``_is_narrowing``. + """ + pairs = list(itertools.permutations(_NUMERIC_ORDER, 2)) + assert len(pairs) == 56 + swallowed = [(a, b) for a, b in pairs if _event(a, b) == "none"] + assert swallowed == [], f"the widen/narrow ladder was bypassed for {swallowed}" + + +def test_widening_and_narrowing_still_report_themselves() -> None: + assert _event("int", "bigint") == "type_widened" + assert _event("tinyint", "double") == "type_widened" + assert _event("bigint", "int") == "type_narrowed" + assert _event("decimal", "smallint") == "type_narrowed" + + +def test_a_real_type_change_still_fails() -> None: + for old, new in ( + ("varchar", "integer"), + ("integer", "varchar"), + ("timestamp", "date"), + ("double", "boolean"), + ): + assert _event(old, new) == "type_changed", f"{old} -> {new} is real drift" + + +def test_an_identical_type_is_still_no_event() -> None: + assert _event("double", "double") == "none" + assert _event("DOUBLE", "double") == "none" + + +def test_the_rename_is_reported_rather_than_hidden() -> None: + """DOUBLE and HUGEINT are not value-identical, so the note says so. + + For the same CSV literal 1.5 landed a lossy 17-significant-digit float where + 2.0 lands the exact 128-bit integer. Collapsing the pair keeps the pipeline + running; the note is what keeps that honest. + """ + note = family_rename_note("double", "hugeint") + assert note is not None + assert "not schema drift" in note + assert "precision" in note + assert family_rename_note("varchar", "integer") is None + + +def test_the_family_check_is_case_and_space_insensitive() -> None: + assert _event("DOUBLE", "HugeInt") == "none" + assert _event("timestamp", " TIMESTAMPTZ ") == "none" From d50457a52920d315d94fa72709faf5d5db97fd55 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:29:54 +0100 Subject: [PATCH 2/6] fix(discover): `fluid init --discover` emitted contracts its own validator rejected The emitter wrote the source's type name straight through, so a discover on DuckDB 2.0 produced `type: hugeint` and `type: '"null"'` and `fluid validate` then exited 1 on the file the tool had just written. Reproduced through the real CLI: on 2.0.0 both emitted contracts fail validation ("'hugeint' is not valid under any of the given schemas"), and after this change both exit 0; on 1.5.6 the emitted types and exit codes are unchanged. Scope is wider than reported. Validating every name in DuckDB's own `duckdb_types()` against `$defs.column.properties.type` of the newest bundled schema shows 35 rejected names, not two -- and `uinteger` and `integer[]` are rejected on 1.5 as well, so this was never only a 2.0 problem. The schema's pattern branch already accepts parameterised and spaced forms (`decimal(18,2)`, `varchar(255)`, `struct(...)`, `timestamp with time zone`), so only `[]` and the quoted `"NULL"` need shape handling. Canonicalised at the one chokepoint, `emitter.py`, so every discover backend inherits it. An unsigned or 128-bit type maps to a width that HOLDS its range rather than the similar-sounding name: `uinteger` (0..2^32-1) does not fit `integer`, and `hugeint` does not fit `bigint`, so they become `bigint` and `numeric`. That is a deliberate departure from the `hugeint -> bigint` sketch in the decision: `bigint` would under-represent, and both land in the same family for drift purposes either way. A name the table does not list passes through unchanged, which is deliberate rather than a gap -- the schema accepts a cross-dialect superset, so the JDBC backends' own dialect names must not be squashed. The docstring says that instead of claiming a `string` fallback the code does not implement. Contracts already emitted with an invalid type stay invalid (the fix is emit-time), but those already failed validation, so nothing working changes. `tests/test_ci_duckdb_floor.py` correctly demanded the new version-sensitive test files be registered with the duckdb-floor CI job; all three are added. --- .github/workflows/ci.yml | 3 + fluid_build/cli/discover/emitter.py | 102 +++++++++- .../test_discover_type_canonicalization.py | 184 ++++++++++++++++++ 3 files changed, 288 insertions(+), 1 deletion(-) create mode 100644 tests/cli/discover/test_discover_type_canonicalization.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 74dbc469..ef1e182d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -683,6 +683,9 @@ jobs: tests/providers/test_duckdb_compat.py \ tests/providers/test_duckdb_sandbox.py \ tests/providers/test_duckdb_sandbox_workspace_root.py \ + tests/build_runners/test_duckdb_cursor_canonical.py \ + tests/build_runners/test_schema_evolution_type_family.py \ + tests/cli/discover/test_discover_type_canonicalization.py \ tests/output_ports/test_drivers_duckdb.py \ tests/build_runners/test_duckdb_runner.py \ tests/build_runners/test_duckdb_runner_mysql_attach.py \ diff --git a/fluid_build/cli/discover/emitter.py b/fluid_build/cli/discover/emitter.py index 9cdc3596..262e6067 100644 --- a/fluid_build/cli/discover/emitter.py +++ b/fluid_build/cli/discover/emitter.py @@ -95,7 +95,7 @@ def emit_contract( "schema": [ { "name": c.name, - "type": (c.type or "string").lower(), + "type": canonical_column_type(c.type), "required": not c.nullable, } for c in s.columns @@ -108,6 +108,106 @@ def emit_contract( } +#: Source type names the bundled contract schema does not accept, mapped to the +#: nearest name it does. Built by validating every name in DuckDB's own +#: ``duckdb_types()`` against ``$defs.column.properties.type`` on duckdb 1.5.6 +#: and 2.0.0: 35 of them are rejected, not the two the bug report named. +#: +#: Without this the emitter wrote DuckDB's spelling straight through, so a +#: discover on DuckDB 2.0 produced ``type: hugeint`` and ``type: '"null"'`` and +#: the tool's own ``fluid validate`` then rejected the contract it had just +#: written. Two of these (``uinteger``, ``integer[]``) were rejected on 1.5 too, +#: so this is not only a 2.0 problem. +#: +#: An unsigned or 128-bit type maps to a name WIDE ENOUGH to hold its range +#: rather than to the same-sounding one: ``uinteger`` (0..2^32-1) does not fit +#: ``integer``, and ``hugeint`` does not fit ``bigint``, so they become +#: ``bigint`` and ``numeric``. The schema's own enum keeps ``bigint`` and +#: ``integer`` distinct, so the choice is visible in the emitted contract. +#: +#: Parameterised and spaced forms are deliberately absent: the schema's pattern +#: branch already accepts ``decimal(18,2)``, ``varchar(255)``, ``struct(...)``, +#: ``map(...)`` and ``timestamp with time zone``. Only ``[]`` and the +#: quoted ``"NULL"`` need shape handling, below. +_SCHEMA_TYPE_ALIASES = { + # integers: width that holds the range, not the similar-sounding name + "int1": "tinyint", + "uint8": "smallint", + "utinyint": "smallint", + "short": "smallint", + "uint16": "integer", + "usmallint": "integer", + "integral": "integer", + "signed": "integer", + "uint32": "bigint", + "uinteger": "bigint", + "oid": "bigint", + # 128-bit and unsigned 64-bit exceed bigint + "hugeint": "numeric", + "int128": "numeric", + "uhugeint": "numeric", + "uint128": "numeric", + "uint64": "numeric", + "ubigint": "numeric", + "varint": "numeric", + "bignum": "numeric", + # text, binary, boolean + "bpchar": "varchar", + "bitstring": "bit", + "logical": "boolean", + # date and time: the precision suffix is not a different type + "timestamp_s": "timestamp", + "timestamp_ms": "timestamp", + "timestamp_us": "timestamp", + "timestamp_ns": "timestamp", + "timestamptz_ns": "timestamptz", + "time_ns": "time", + "timetz": "time", + "time with time zone": "time", + # composites + "list": "array", + "tuple": "struct", + "union": "variant", + # a column of only NULLs, and DuckDB's own meta type + "null": "string", + "type": "string", +} + + +def canonical_column_type(raw: Optional[str]) -> str: + """A source column type the bundled contract schema accepts. + + The one place a discovered type becomes a contract field type, so every + discover backend inherits it. + + A name the table does not list passes through lower-cased, exactly as + before. That is deliberate rather than a gap: the schema accepts a + cross-dialect superset through its pattern branch, so the JDBC backends' + own dialect names (and ``decimal(18,2)``, ``varchar(255)``, + ``struct(...)``) are already valid and must not be squashed. The table + therefore carries only names measured to be REJECTED. A source type that is + neither valid nor listed still reaches the contract unchanged and still + fails ``fluid validate`` -- the same as today, and not something this + change claims to fix. + """ + text = (raw or "string").strip() + if not text: + return "string" + lowered = text.lower() + # ``INTEGER[]``, ``VARCHAR[][]`` -- a list of anything is an array. + if lowered.endswith("[]"): + return "array" + # ``typeof(NULL)`` is the quoted string ``"NULL"``. + if lowered.strip('"') == "null": + return "string" + if lowered in _SCHEMA_TYPE_ALIASES: + return _SCHEMA_TYPE_ALIASES[lowered] + base = lowered.split("(", 1)[0].strip() + if base in _SCHEMA_TYPE_ALIASES: + return _SCHEMA_TYPE_ALIASES[base] + return lowered + + def _safe_id(name: str) -> str: """Convert ``public.orders`` → ``public_orders`` so it satisfies the FLUID identifier regex used by exposeId / outputs. diff --git a/tests/cli/discover/test_discover_type_canonicalization.py b/tests/cli/discover/test_discover_type_canonicalization.py new file mode 100644 index 00000000..2ae3e05c --- /dev/null +++ b/tests/cli/discover/test_discover_type_canonicalization.py @@ -0,0 +1,184 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""``fluid init --discover`` must not emit a contract its own validator rejects. + +The emitter wrote the source's type name straight through, so a discover on +DuckDB 2.0 produced ``type: hugeint`` and ``type: '"null"'`` and ``fluid +validate`` then rejected the file the tool had just written. Two of the rejected +names (``uinteger``, ``integer[]``) were rejected on DuckDB 1.5 too, so this was +never only a 2.0 problem. + +The sharp test here is :func:`test_no_duckdb_type_name_is_emitted_invalid`: it +validates the mapping of EVERY name in the installed DuckDB's own +``duckdb_types()`` against the bundled schema, so a future DuckDB that adds a +type name is caught by this file rather than by a user's broken contract. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict + +import pytest + +from fluid_build.cli.discover.emitter import canonical_column_type + +REPO_ROOT = Path(__file__).resolve().parents[3] +SCHEMAS = REPO_ROOT / "fluid_build" / "schemas" + + +def _type_schema() -> Dict[str, Any]: + newest = sorted(SCHEMAS.glob("fluid-schema-*.json"))[-1] + doc = json.loads(newest.read_text(encoding="utf-8")) + return doc["$defs"]["column"]["properties"]["type"] + + +def _validator() -> Any: + jsonschema = pytest.importorskip("jsonschema") + return jsonschema.Draft202012Validator(_type_schema()) + + +@pytest.mark.parametrize( + "raw, expected", + [ + # The two the bug report named. + ("HUGEINT", "numeric"), + ('"NULL"', "string"), + # Rejected on 1.5 as well, so not a 2.0-only problem. + ("UINTEGER", "bigint"), + ("INTEGER[]", "array"), + # A width that holds the range, not the similar-sounding name. + ("UBIGINT", "numeric"), + ("UHUGEINT", "numeric"), + ("USMALLINT", "integer"), + ("UTINYINT", "smallint"), + # Precision suffixes are not different types. + ("TIMESTAMP_MS", "timestamp"), + ("TIMESTAMP_NS", "timestamp"), + ("TIMESTAMPTZ_NS", "timestamptz"), + # Composites and odds. + ("LIST", "array"), + ("UNION", "variant"), + ("LOGICAL", "boolean"), + ("BPCHAR", "varchar"), + ("VARCHAR[][]", "array"), + # Already valid: untouched apart from case. + ("VARCHAR", "varchar"), + ("BIGINT", "bigint"), + ("DECIMAL(18,2)", "decimal(18,2)"), + ("STRUCT(a INTEGER)", "struct(a integer)"), + ("timestamp with time zone", "timestamp with time zone"), + # No type at all: the emitter's own pre-existing fallback. + (None, "string"), + ("", "string"), + (" ", "string"), + ], +) +def test_named_types_map_as_documented(raw: str, expected: str) -> None: + assert canonical_column_type(raw) == expected + + +def test_every_mapped_type_satisfies_the_bundled_schema() -> None: + """The documented aliases must each land on a value the schema accepts.""" + validator = _validator() + bad = [] + for raw in ( + "HUGEINT", + '"NULL"', + "UINTEGER", + "INTEGER[]", + "UBIGINT", + "UHUGEINT", + "USMALLINT", + "UTINYINT", + "TIMESTAMP_MS", + "TIMESTAMPTZ_NS", + "TIME_NS", + "TIMETZ", + "LIST", + "UNION", + "TUPLE", + "LOGICAL", + "BPCHAR", + "BITSTRING", + "VARINT", + "BIGNUM", + "INT1", + "INT128", + "UINT128", + "OID", + "SHORT", + "SIGNED", + "INTEGRAL", + "NULL", + "TYPE", + ): + mapped = canonical_column_type(raw) + if not validator.is_valid(mapped): + bad.append((raw, mapped)) + assert bad == [], f"these map to values the contract schema rejects: {bad}" + + +def test_no_duckdb_type_name_is_emitted_invalid() -> None: + """Sweep the INSTALLED DuckDB's own type list, so a new type name is caught here. + + Run against whichever DuckDB is installed, which is the point: a 1.5 install + and a 2.0 install enumerate different sets, and both must map cleanly. + """ + duckdb = pytest.importorskip("duckdb") + validator = _validator() + con = duckdb.connect(":memory:") + try: + names = [ + row[0] + for row in con.execute( + "SELECT DISTINCT lower(type_name) FROM duckdb_types()" + ).fetchall() + ] + finally: + con.close() + assert names, "duckdb_types() returned nothing" + bad = [ + (n, canonical_column_type(n)) + for n in names + if not validator.is_valid(canonical_column_type(n)) + ] + assert bad == [], ( + f"duckdb {duckdb.__version__} has type names this emitter would write " + f"into a contract that then fails validation: {bad}" + ) + + +def test_an_already_valid_name_is_never_remapped() -> None: + """A contract emitted before this change must keep the same types.""" + validator = _validator() + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + names = [ + row[0] + for row in con.execute( + "SELECT DISTINCT lower(type_name) FROM duckdb_types()" + ).fetchall() + ] + finally: + con.close() + changed = [ + (n, canonical_column_type(n)) + for n in names + if validator.is_valid(n) and canonical_column_type(n) != n + ] + assert changed == [], f"already-valid type names were remapped: {changed}" From d76c2602f65e8bfc9b711e9e0400e2dea7901fb3 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:30:13 +0100 Subject: [PATCH 3/6] fix(duckdb): an incremental cursor lost rows across a DuckDB version change Live in shipped 0.22.0. The cursor is persisted as text and pasted into `WHERE > ''` as an UNTYPED literal, and the two DuckDB versions disagree about the column type it is coerced against: `read_json_auto` over a timestamp carrying `Z` resolves TIMESTAMP on 1.5 and TIMESTAMP WITH TIME ZONE on 2.0, where a naive literal is read in the session time zone -- the host's, since nothing pinned it. On a host with a non-zero UTC offset the `>` boundary therefore moved by that offset. Measured end to end, 6 rows in 3 batches at TZ=America/New_York: writing the cursor on 1.5.6 and continuing on 2.0.0 landed 5 of 6 rows -- id=3 was never landed and the cursor advanced past it -- at exit code 0 with the stream recorded `succeeded`. Nothing warned. Both versions satisfy `duckdb>=1.5.0`, so one install can write state another reads. CSV sources were never affected: `read_csv_auto` already sniffed TIMESTAMPTZ on 1.5. A timestamp cursor now compares INSTANTS, `epoch_us(field) > epoch_us(TIMESTAMP '')`, which no session time zone can shift, and the stored text is normalised to a naive UTC instant. All five scenarios now land the same rows as the all-one-version control: cross-version in both directions, under UTC-4 and UTC+2, and a legacy cursor already on disk. No state file needs rewriting. A cursor a shipped 2.0 install wrote carries an explicit offset (`2026-05-01 10:00:00-04:00`); 1.5 drops that offset when it coerces to a naive TIMESTAMP, which re-emitted rows. Normalising on READ repairs those in place. Measured: normalising the text alone still lost id=3, because what moves is the coercion, not the text -- both halves are needed. Pinning the session zone to UTC was tried first and REVERTED. `TimeZone` is an ICU-backed setting; the sandbox disables extension autoloading and external access, so where ICU is not already loaded DuckDB does not even recognise the option ("The following options were not recognized: TimeZone") and the connection fails. It passed locally only because this machine has ICU statically loaded; it broke 3 in-tree tests, including one whose DuckDB had no ICU reachable. A pin would therefore fail exactly the hardened environments the sandbox exists for. `epoch_us` needs no extension -- verified with autoload disabled and an empty extension directory, on both versions and both zones. Scalar cursors keep the plain comparison, including date-as-integer ones: Python 3.11+ parses basic ISO, so a bare `20260501` would otherwise take the instant path and `epoch_us` would fail on its INTEGER column. Accepted cost: wrapping the column in a function can stop the predicate being pushed into the Parquet or JSON reader, so a large source scans more. Correct rows beat a faster wrong answer. The output-determinism half of Trello 1024 is NOT here. Its only known fix is the ICU-dependent pin, and it is not a 2.0 regression at all -- 1.5.6 writes byte-identical host-local output for the same zone -- so it is being filed on its own merits. Founder decision, 2026-10-26. --- CHANGELOG.md | 47 +++++ fluid_build/build_runners/duckdb/runner.py | 134 ++++++++++++- .../test_duckdb_cursor_canonical.py | 188 ++++++++++++++++++ 3 files changed, 361 insertions(+), 8 deletions(-) create mode 100644 tests/build_runners/test_duckdb_cursor_canonical.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a9a72f92..5973ccd9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,53 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **An incremental cursor written on DuckDB 1.5 and read on 2.0 silently dropped or + duplicated rows.** The cursor is stored as text and compared as an untyped SQL literal, + and the two versions disagree about the column type it is coerced against: + `read_json_auto` over a timestamp carrying `Z` gives TIMESTAMP on 1.5 and TIMESTAMP WITH + TIME ZONE on 2.0, where a naive literal is read in the session time zone -- the host's. + On a host with a non-zero UTC offset the `>` boundary therefore moved: an end-to-end run + that wrote its cursor on 1.5 and continued on 2.0 landed 5 of 6 rows and advanced the + cursor past the row it skipped, at exit code 0 with the stream recorded `succeeded`. + Both DuckDB versions resolve to `duckdb>=1.5.0`, so one install can write state another + reads. + A timestamp cursor now compares INSTANTS (`epoch_us`), which no session time zone can + shift, and the stored text is normalised to a naive UTC instant. **No state file needs + rewriting**: a cursor a shipped 2.0 install already wrote carries an explicit offset and + is normalised when read. Integer, date-as-integer and opaque string cursors keep the + plain comparison. CSV sources were never affected -- `read_csv_auto` already sniffed + TIMESTAMPTZ on 1.5. + Pinning the session zone to UTC would also have worked and is deliberately not used: + `TimeZone` is ICU-backed, the sandbox forbids autoloading extensions, and without ICU + DuckDB does not recognise the option at all, so a pin would fail the connection in + exactly the hardened environments the sandbox exists for. The accepted cost of the + instant comparison is that wrapping the column in a function can stop the predicate + being pushed into the Parquet or JSON reader. +- **A pipeline discovered on DuckDB 1.5 failed `SchemaDriftError` on 2.0.** The drift gate + compared type *names*, and 2.0 infers HUGEINT where 1.5 infers DOUBLE for the same + integer literal, and TIMESTAMP WITH TIME ZONE where 1.5 infers TIMESTAMP. Upgrading + DuckDB — no contract edit, no new data — broke a working pipeline on a false positive. + Those pairs are now treated as one physical type under two names. The widening/narrowing + classification is unchanged: the family check runs after it, so `type_widened` and + `type_narrowed` still fire for every ordered numeric pair. Note DOUBLE and HUGEINT are + not value-identical — 1.5 landed a lossy 17-significant-digit float where 2.0 lands the + exact 128-bit integer — so the non-failing path says the physical type moved rather than + passing silently. +- **`fluid init --discover` emitted contracts its own `fluid validate` rejected.** The + emitter wrote the source's type name straight through, so a discover on DuckDB 2.0 + produced `type: hugeint` and `type: '"null"'` and `fluid validate` then exited 1 on the + file the tool had just written. Validating every name in DuckDB's own `duckdb_types()` + against the bundled schema showed **35 rejected names**, not the two reported: `uinteger` + and `integer[]` were rejected on 1.5 as well, so this was never only a 2.0 problem. + Discovered types are now canonicalised at the emitter, mapping each rejected name to the + nearest one the schema accepts and leaving every already-valid name untouched. An + unsigned or 128-bit type maps to a width that holds its range (`uinteger` → `bigint`, + `hugeint` → `numeric`) rather than to the similar-sounding name. Contracts already emitted + with an invalid type stay invalid — the fix is emit-time — but those already failed + validation, so nothing working changes. + ### Security - A contract repository could hand the whole host to its own SQL through DuckDB's spill diff --git a/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index 1bc7e203..85945b39 100644 --- a/fluid_build/build_runners/duckdb/runner.py +++ b/fluid_build/build_runners/duckdb/runner.py @@ -43,6 +43,7 @@ import re import time from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, Dict, FrozenSet, List, Optional, Tuple @@ -1323,16 +1324,133 @@ def _apply_incremental_filter( # runs will use the cursor that's persisted at end-of-run. return base_sel, None last_value = last_cursor.value - # The cursor value is treated as a SQL literal regardless of the - # column type: duckdb will coerce ``'2026-05-01T00:00:00'`` against - # a TIMESTAMP column and ``'42'`` against an INT. ``quote_ansi_string_literal`` - # neutralises any embedded quote / control char from a tampered - # state file. - quoted_value = quote_ansi_string_literal(str(last_value)) - filtered = f"SELECT * FROM ({base_sel}) WHERE {cursor_field_safe} > {quoted_value}" + # ``_cursor_comparison`` builds the predicate: an instant comparison for a + # timestamp cursor, which no session time zone can shift, and a plain + # literal for a scalar one. It quotes through ``quote_ansi_string_literal``, + # so a tampered state file cannot inject SQL into the WHERE clause. + predicate = _cursor_comparison(cursor_field_safe, last_value) + filtered = f"SELECT * FROM ({base_sel}) WHERE {predicate}" return filtered, last_value +#: A trailing UTC offset on a timestamp DuckDB rendered as text: ``Z``, or a +#: sign and 2-4 digits with an optional colon (``+00``, ``-0400``, ``-04:00``). +#: Matched explicitly rather than left to :meth:`datetime.fromisoformat`, whose +#: tolerance for ``Z`` and a two-digit offset depends on the Python version and +#: this package supports 3.10 upward. +_CURSOR_OFFSET_RE = re.compile(r"(?P.*?)(?PZ|[+-]\d{2}:?(?:\d{2})?)$") + + +def _canonical_cursor_text(value: Any) -> str: + """``value`` as cursor text whose meaning does not depend on DuckDB's version. + + A cursor is stored as text and pasted into ``WHERE > ''`` as + an UNTYPED literal, so that text has to denote one instant whatever the + column resolved to. DuckDB 1.5 reads a JSON timestamp carrying ``Z`` as a + naive TIMESTAMP and 2.0 reads it as TIMESTAMP WITH TIME ZONE, so the two + disagree about what an offset-bearing literal means. + + An offset-bearing timestamp is therefore converted to UTC and rendered + WITHOUT an offset. Naive-UTC text denotes the same instant against a naive + TIMESTAMP column and, under the pinned UTC session + (:data:`~fluid_build.providers._duckdb_compat.SESSION_TIME_ZONE`), against a + TIMESTAMPTZ one. + + This is also what repairs state already on disk, so no cursor file has to be + rewritten. A shipped DuckDB 2.0 install on a host with a non-zero offset + persisted cursors like ``2026-05-01 10:00:00-04:00``; DuckDB 1.5 drops that + offset when it coerces to a naive TIMESTAMP, which moves the ``>`` boundary + and re-emits rows (measured: a row duplicated even with the time zone + pinned). Normalising on read closes that. + + Anything that is not an offset-bearing timestamp is returned as its own + text, unchanged: an integer cursor, an opaque string, an already-naive + timestamp. A value that looks like it carries an offset but does not parse + is also returned unchanged -- a cursor whose text we do not understand is + left exactly as the previous release compared it. + """ + text = str(value) + match = _CURSOR_OFFSET_RE.match(text) + if match is None: + return text + head, off = match.group("head"), match.group("off") + if off == "Z": + normalised = head + "+00:00" + else: + digits = off.replace(":", "") + if len(digits) == 3: # '+04' -> '+04:00' + normalised = head + digits + ":00" + elif len(digits) == 5: + normalised = head + digits[:3] + ":" + digits[3:] + else: + return text + try: + parsed = datetime.fromisoformat(normalised) + except ValueError: + return text + if parsed.tzinfo is None: + return text + return parsed.astimezone(timezone.utc).replace(tzinfo=None).isoformat(sep=" ") + + +def _cursor_is_timestamp(text: str) -> bool: + """Whether canonical cursor text denotes a date or timestamp rather than a scalar. + + Decided from the VALUE, not by probing the column: the cursor is the stored + ``MAX()`` of that column, so its text already tells us which comparison is + right, and a probe would cost a query per run. An integer cursor + (``'42'``), an opaque key (``'order-000123'``) and anything else that is not + an ISO date or timestamp answer ``False`` and keep the plain comparison. + """ + # Require the separators. Python 3.11+ parses BASIC ISO too, so a bare + # ``20260501`` -- a perfectly ordinary date-as-integer cursor -- would + # otherwise take the instant path and ``epoch_us`` would fail on its + # INTEGER column. + if "-" not in text[:11]: + return False + try: + datetime.fromisoformat(text) + except ValueError: + return False + return True + + +def _cursor_comparison(field_sql: str, value: Any) -> str: + """The incremental ``>`` predicate for one cursor, free of the session time zone. + + A plain `` > ''`` is not safe for a timestamp. The literal is + untyped, so DuckDB coerces it to the column's type, and on DuckDB 2.0 a JSON + timestamp column resolves to TIMESTAMP WITH TIME ZONE, where a naive literal + is read in the session time zone -- the host's, unless something pins it. + The boundary then moves by the host's UTC offset and rows are silently + skipped or re-emitted. + + Comparing INSTANTS removes the ambiguity: ``epoch_us`` of a TIMESTAMP WITH + TIME ZONE is its instant, and of a naive TIMESTAMP is that time read as UTC, + which is exactly what :func:`_canonical_cursor_text` renders. The same + predicate is therefore correct on 1.5 and 2.0 and under any host zone + (measured: the plain literal returns 1 row where 2 are due, on 2.0 at + UTC-4; this returns 2 on both versions under both zones). + + Pinning the session zone to UTC would also have worked and is NOT used: + ``TimeZone`` is an ICU-backed setting, the sandbox forbids autoloading + extensions, and where ICU is absent DuckDB does not even recognise the + option -- so a pin would fail the connection in exactly the hardened + environments the sandbox exists for. ``epoch_us`` needs no extension + (verified with autoload disabled and an empty extension directory). + + The cost, accepted deliberately: wrapping the column in a function can stop + the predicate being pushed into the Parquet or JSON reader, so a large + source scans more than it would have. Correct rows beat a faster wrong + answer, and only a timestamp cursor takes this path. + """ + text = _canonical_cursor_text(value) + literal = quote_ansi_string_literal(text) + if _cursor_is_timestamp(text): + return f"epoch_us({field_sql}) > epoch_us(TIMESTAMP {literal})" + return f"{field_sql} > {literal}" + + def _persist_cursor_after_run( ctx: RunContext, stream: str, @@ -1383,7 +1501,7 @@ def _persist_cursor_after_run( ctx.build_id, Cursor( stream=stream, - value=str(new_value), + value=_canonical_cursor_text(new_value), updated_at=utc_now_iso(), ), ) diff --git a/tests/build_runners/test_duckdb_cursor_canonical.py b/tests/build_runners/test_duckdb_cursor_canonical.py new file mode 100644 index 00000000..bc996f12 --- /dev/null +++ b/tests/build_runners/test_duckdb_cursor_canonical.py @@ -0,0 +1,188 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""An incremental cursor means one instant on either DuckDB version. + +The cursor is persisted as text and pasted into ``WHERE > ''`` as +an UNTYPED literal, so its text has to denote one instant whatever the column +resolved to. DuckDB 1.5 reads a JSON timestamp carrying ``Z`` as a naive +TIMESTAMP and 2.0 reads it as TIMESTAMP WITH TIME ZONE, so the two disagree +about an offset-bearing literal. On a host with a non-zero UTC offset that moved +the ``>`` boundary: an end-to-end run that wrote its cursor on 1.5 and continued +on 2.0 landed 5 of 6 rows and advanced the cursor past the row it skipped, at +exit code 0. + +Two pieces, and both are needed: + +* ``_canonical_cursor_text`` renders the stored text as a naive UTC instant. + That is what repairs the cursors a shipped DuckDB 2.0 install already wrote -- + they carry an explicit offset -- so no state file has to be rewritten. +* ``_cursor_comparison`` compares INSTANTS (``epoch_us``) for a timestamp + cursor, so the predicate cannot be shifted by the session time zone at all. + Measured: normalising the text alone still lost a row, because the coercion of + a naive literal against a TIMESTAMPTZ column is what moves, not the text. + +Pinning the session zone to UTC would also have worked and is deliberately NOT +used: ``TimeZone`` is ICU-backed, the sandbox forbids autoloading extensions, +and without ICU DuckDB does not even recognise the option -- so a pin fails the +connection in exactly the hardened environments the sandbox exists for. +``epoch_us`` needs no extension. + +Normalisation runs on both the read and the write side. Only the READ side is +load-bearing, and only that side is mutation-tested: it normalises whatever it +finds, so correctness does not depend on how a value was written. Doing it on +write too keeps newly stored state in one form rather than two. +""" + +from __future__ import annotations + +import pytest + +from fluid_build.build_runners.duckdb.runner import ( + _apply_incremental_filter, + _canonical_cursor_text, + _cursor_comparison, + _cursor_is_timestamp, +) + + +@pytest.mark.parametrize( + "stored, expected, why", + [ + # The migration case: what a shipped 2.0 install at a non-UTC host wrote. + ("2026-05-01 10:00:00-04:00", "2026-05-01 14:00:00", "legacy offset cursor -> UTC"), + ("2026-05-01 12:00:00+02:00", "2026-05-01 10:00:00", "positive offset -> UTC"), + # What 2.0 writes under the pin, and the spellings DuckDB renders. + ("2026-05-01 14:00:00+00:00", "2026-05-01 14:00:00", "offset zero is still an offset"), + ("2026-05-01 14:00:00+00", "2026-05-01 14:00:00", "two-digit offset"), + ("2026-05-01 12:00:00+0200", "2026-05-01 10:00:00", "offset without a colon"), + ("2026-05-01 18:00:00Z", "2026-05-01 18:00:00", "Z suffix"), + ("2026-05-01T18:00:00Z", "2026-05-01 18:00:00", "ISO T separator"), + # Everything that is not an offset-bearing timestamp is left alone. + ("2026-05-01 14:00:00", "2026-05-01 14:00:00", "already naive"), + ("42", "42", "an integer cursor"), + ("order-000123", "order-000123", "an opaque string cursor"), + ("abc-12:34", "abc-12:34", "offset-shaped but not a timestamp"), + ("", "", "empty"), + ], +) +def test_cursor_text_is_canonicalised(stored: str, expected: str, why: str) -> None: + assert _canonical_cursor_text(stored) == expected, why + + +def test_a_non_string_cursor_becomes_its_own_text() -> None: + """The persist side hands in whatever DuckDB's driver materialised.""" + assert _canonical_cursor_text(42) == "42" + assert _canonical_cursor_text(None) == "None" + + +def test_canonicalising_is_idempotent() -> None: + """A cursor written by the fixed code must survive being read by it again.""" + once = _canonical_cursor_text("2026-05-01 10:00:00-04:00") + assert _canonical_cursor_text(once) == once + + +def test_the_two_stored_forms_converge_on_one_instant() -> None: + """The whole point: 1.5's naive text and 2.0's offset text must agree. + + ``2026-05-01 14:00:00`` (what 1.5 persisted for an instant) and + ``2026-05-01 10:00:00-04:00`` (what a shipped 2.0 install at UTC-4 persisted + for the SAME instant) have to compare equal after canonicalisation, or an + install that changes DuckDB version re-reads or skips rows. + """ + from_15 = _canonical_cursor_text("2026-05-01 14:00:00") + from_20 = _canonical_cursor_text("2026-05-01 10:00:00-04:00") + assert from_15 == from_20 == "2026-05-01 14:00:00" + + +# ── the call site, not just the helper ─────────────────────────────────── + + +class _Cursor: + def __init__(self, value: str) -> None: + self.stream = "evt" + self.value = value + self.updated_at = "2026-05-01T00:00:00Z" + + +class _StateStore: + """Returns one stored cursor, as ``FileStateStore`` would after reading JSON.""" + + def __init__(self, value: str) -> None: + self._value = value + + def get_cursor(self, _product: str, _build: str, _stream: str) -> _Cursor: + return _Cursor(self._value) + + +class _Source: + def __init__(self) -> None: + from fluid_build.api.source import AcquisitionMode + + self.mode = AcquisitionMode.INCREMENTAL_APPEND + self.cursor_field = "event_time" + + +class _Ctx: + def __init__(self, value: str) -> None: + self.product_id = "bronze.evt" + self.build_id = "ingest" + self.source = _Source() + self.state_store = _StateStore(value) + + +def test_the_emitted_where_clause_compares_instants() -> None: + """Exercises the wiring, which a direct call to the helpers does not. + + Without this, removing either helper from ``_apply_incremental_filter`` + leaves every other test in this file passing. + """ + sql, last = _apply_incremental_filter( + _Ctx("2026-05-01 10:00:00-04:00"), "evt", "SELECT * FROM src" + ) + assert "epoch_us(" in sql, f"a timestamp cursor must compare instants: {sql}" + assert "'2026-05-01 14:00:00'" in sql, sql + assert "-04:00" not in sql, f"the raw offset-bearing cursor reached the SQL: {sql}" + # The returned value stays the value that was stored, so a caller that + # records it sees what it persisted. + assert last == "2026-05-01 10:00:00-04:00" + + +def test_an_integer_cursor_keeps_the_plain_comparison() -> None: + """``epoch_us`` of an integer column would fail, so a scalar must not take it.""" + sql, last = _apply_incremental_filter(_Ctx("42"), "evt", "SELECT * FROM src") + assert "epoch_us(" not in sql, sql + assert "'42'" in sql, sql + assert last == "42" + + +@pytest.mark.parametrize( + "text, is_ts", + [ + ("2026-05-01 14:00:00", True), + ("2026-05-01T14:00:00", True), + ("2026-05-01", True), + ("42", False), + # Python 3.11+ parses basic ISO, so this must be excluded explicitly: + # a date-as-integer cursor lives in an INTEGER column. + ("20260501", False), + ("20260501T140000", False), + ("order-000123", False), + ("", False), + ], +) +def test_only_a_timestamp_cursor_takes_the_instant_path(text: str, is_ts: bool) -> None: + assert _cursor_is_timestamp(text) is is_ts + predicate = _cursor_comparison("ts", text) + assert ("epoch_us(" in predicate) is is_ts, predicate From 008a8567bce189010dd5d6c6ba1de921a6513d4b Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:08:31 +0100 Subject: [PATCH 4/6] =?UTF-8?q?fix(duckdb):=20the=20review's=20two=20block?= =?UTF-8?q?ers=20=E2=80=94=20a=20VARCHAR=20cursor,=20and=20a=20silent=20fa?= =?UTF-8?q?mily=20rename?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both found by the scoped review of this branch, both confirmed, both mine. 1. HIGH. `epoch_us(VARCHAR)` is a BinderException on duckdb 1.5.6 and 2.0.0, so a VARCHAR cursor column holding timestamp-looking values would have failed EVERY incremental run on it. `_cursor_comparison` chose the instant path from the cursor TEXT, and a VARCHAR column of `'2026-05-01 10:00:00'` yields a cursor indistinguishable from a real timestamp's. That was my shortcut: the approved option said a column-type probe, and I inferred from the value instead. `_cursor_column_type` now asks the source (`typeof(...) ... LIMIT 1`, with `DESCRIBE` as the fallback for an empty source) and only a type in `_EPOCH_US_TYPES` takes the instant path. That set was measured, not assumed: TIMESTAMP, TIMESTAMPTZ, TIMESTAMP_S/_MS/_NS, DATE and TIME are accepted on both versions; VARCHAR, INTEGER, DECIMAL and the `"NULL"` of an all-null column all raise. With no connection the predicate is exactly what the previous release emitted, so nothing regresses where the probe cannot run. 2. MEDIUM. The family collapse let a contract that declared `schemaPolicy: strict` pass a `timestamp` -> `timestamptz` or `double` -> `hugeint` change with no event, hence no decision, hence no error — and `family_rename_note` had NO production caller, so the comment claiming the precision change is "reported rather than hidden" was false as shipped. It is also reachable without any DuckDB upgrade: an offset-bearing CSV infers TIMESTAMPTZ on 1.5.6 as well, so a declared `timestamp` against an upstream that genuinely gains a zone offset hit it too. `_log_family_renames` now runs in `enforce_schema_policy_or_raise` BEFORE its early return, so the operator is told the physical type moved whatever the policy decides. Suppressing the failure stays the intent; suppressing the notice was not. Writing the test for (2) surfaced a third bug of my own: `family_rename_note` answered for an unchanged column too, because `double` is same-family as itself — so the gate would have logged a spurious rename on every run. It now returns None when the two names are equal. Mutation-checked all three: deciding the instant path from the value fails 9 tests, dropping the probe call fails the call-site test, and removing `_log_family_renames` from the gate fails the gate-level test. That last one needed a test through `enforce_schema_policy_or_raise` rather than the helper — a direct call passes either way, which is how the dead note got shipped in the first place. 2477 tests green on duckdb 1.5.6 and 2.0.0. --- .../build_runners/_acquisition_common.py | 36 ++++- .../build_runners/_schema_evolution.py | 2 + fluid_build/build_runners/duckdb/runner.py | 78 ++++++--- .../test_duckdb_cursor_canonical.py | 148 ++++++++++++++---- .../test_schema_evolution_type_family.py | 90 +++++++++++ 5 files changed, 300 insertions(+), 54 deletions(-) diff --git a/fluid_build/build_runners/_acquisition_common.py b/fluid_build/build_runners/_acquisition_common.py index 1efb140e..931a4bca 100644 --- a/fluid_build/build_runners/_acquisition_common.py +++ b/fluid_build/build_runners/_acquisition_common.py @@ -38,7 +38,7 @@ import time from datetime import datetime, timezone from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple LOG = logging.getLogger("fluid.build_runners.acquisition") @@ -717,6 +717,7 @@ def enforce_schema_policy_or_raise(ctx: Any, runner: Any) -> None: overrides=contract_block.get("evolutionOverrides") or {}, is_first_run=False, ) + _log_family_renames(baseline, current) if not plan.has_failure: return @@ -736,6 +737,39 @@ def enforce_schema_policy_or_raise(ctx: Any, runner: Any) -> None: ) +def _log_family_renames( + baseline: Sequence[Any], + current: Sequence[Any], +) -> None: + """Tell the operator about a type change the drift gate deliberately allowed. + + ``diff_columns`` emits no event when a column's type changed only between + two spellings of one physical type (DuckDB 1.5's DOUBLE vs 2.0's HUGEINT, + TIMESTAMP vs TIMESTAMP WITH TIME ZONE), so that upgrading DuckDB does not + raise a false ``SchemaDriftError``. Suppressing the FAILURE is the intent; + suppressing the NOTICE as well is not, because the two types are not always + value-identical -- 1.5 landed a lossy 17-significant-digit float where 2.0 + lands the exact 128-bit integer. + + Logged whatever the policy decides, and before the gate's early return, so + a contract that declared ``schemaPolicy: strict`` still hears that the + physical type moved even though the run is allowed to continue. + """ + from fluid_build.build_runners._schema_evolution import family_rename_note + + by_name = {getattr(c, "name", None): c for c in baseline} + for col in current: + was = by_name.get(getattr(col, "name", None)) + if was is None: + continue + note = family_rename_note(getattr(was, "type", "") or "", getattr(col, "type", "") or "") + if note: + LOG.warning( + "acquisition_schema_family_rename", + extra={"column": getattr(col, "name", None), "detail": note}, + ) + + def write_run_record( *, state_store: Any, diff --git a/fluid_build/build_runners/_schema_evolution.py b/fluid_build/build_runners/_schema_evolution.py index 2421ea81..0628d7e3 100644 --- a/fluid_build/build_runners/_schema_evolution.py +++ b/fluid_build/build_runners/_schema_evolution.py @@ -106,6 +106,8 @@ def family_rename_note(old: str, new: str) -> Optional[str]: Returned so the caller can SAY the physical type moved while not failing on it. ``None`` when the two are not one family. """ + if old.strip().lower() == new.strip().lower(): + return None # nothing changed: a rename note for X -> X is noise if not _same_type_family(old, new): return None return ( diff --git a/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index 85945b39..9d9ec6b6 100644 --- a/fluid_build/build_runners/duckdb/runner.py +++ b/fluid_build/build_runners/duckdb/runner.py @@ -1292,7 +1292,7 @@ def _build_quality_predicates( def _apply_incremental_filter( - ctx: RunContext, stream: str, base_sel: str + ctx: RunContext, stream: str, base_sel: str, con: Any = None ) -> Tuple[str, Optional[Any]]: """Wrap a base SELECT with an incremental cursor filter. @@ -1328,7 +1328,8 @@ def _apply_incremental_filter( # timestamp cursor, which no session time zone can shift, and a plain # literal for a scalar one. It quotes through ``quote_ansi_string_literal``, # so a tampered state file cannot inject SQL into the WHERE clause. - predicate = _cursor_comparison(cursor_field_safe, last_value) + column_type = _cursor_column_type(con, base_sel, cursor_field_safe) if con is not None else None + predicate = _cursor_comparison(cursor_field_safe, last_value, column_type) filtered = f"SELECT * FROM ({base_sel}) WHERE {predicate}" return filtered, last_value @@ -1393,29 +1394,53 @@ def _canonical_cursor_text(value: Any) -> str: return parsed.astimezone(timezone.utc).replace(tzinfo=None).isoformat(sep=" ") -def _cursor_is_timestamp(text: str) -> bool: - """Whether canonical cursor text denotes a date or timestamp rather than a scalar. +#: The column types ``epoch_us`` accepts. Measured on duckdb 1.5.6 and 2.0.0, +#: identically: every other type -- VARCHAR, INTEGER, DECIMAL, the ``"NULL"`` +#: of an all-null column -- raises BinderException, so the instant comparison +#: must never be emitted for one. +_EPOCH_US_TYPES = frozenset( + { + "TIMESTAMP", + "TIMESTAMP WITH TIME ZONE", + "TIMESTAMPTZ", + "TIMESTAMP_S", + "TIMESTAMP_MS", + "TIMESTAMP_NS", + "TIMESTAMPTZ_NS", + "DATE", + "TIME", + } +) + + +def _cursor_column_type(con: Any, base_sel: str, field_sql: str) -> Optional[str]: + """The declared type of the cursor column in ``base_sel``, or ``None``. - Decided from the VALUE, not by probing the column: the cursor is the stored - ``MAX()`` of that column, so its text already tells us which comparison is - right, and a probe would cost a query per run. An integer cursor - (``'42'``), an opaque key (``'order-000123'``) and anything else that is not - an ISO date or timestamp answer ``False`` and keep the plain comparison. + Asked of the source, not guessed from the cursor's text. The value's shape + is not enough: a VARCHAR column holding ``'2026-05-01 10:00:00'`` yields a + cursor that looks exactly like a timestamp, and ``epoch_us`` on VARCHAR is a + BinderException that would fail every incremental run on such a column. + + ``LIMIT 0``, so this costs a bind and no scan. ``None`` when the probe + cannot answer -- a source that will not bind here fails informatively on the + real query a moment later, so this never raises on its own. """ - # Require the separators. Python 3.11+ parses BASIC ISO too, so a bare - # ``20260501`` -- a perfectly ordinary date-as-integer cursor -- would - # otherwise take the instant path and ``epoch_us`` would fail on its - # INTEGER column. - if "-" not in text[:11]: - return False try: - datetime.fromisoformat(text) - except ValueError: - return False - return True + rows = con.execute(f"SELECT typeof({field_sql}) FROM ({base_sel}) LIMIT 1").fetchall() + except Exception: + return None + if rows and rows[0] and rows[0][0]: + return str(rows[0][0]).strip().upper() + # An empty source binds but returns no row: ask DuckDB for the column's + # type from the relation's own schema instead. + try: + described = con.execute(f"DESCRIBE SELECT {field_sql} FROM ({base_sel})").fetchall() + except Exception: + return None + return str(described[0][1]).strip().upper() if described and len(described[0]) > 1 else None -def _cursor_comparison(field_sql: str, value: Any) -> str: +def _cursor_comparison(field_sql: str, value: Any, column_type: Optional[str]) -> str: """The incremental ``>`` predicate for one cursor, free of the session time zone. A plain `` > ''`` is not safe for a timestamp. The literal is @@ -1432,6 +1457,13 @@ def _cursor_comparison(field_sql: str, value: Any) -> str: (measured: the plain literal returns 1 row where 2 are due, on 2.0 at UTC-4; this returns 2 on both versions under both zones). + Only for a column type ``epoch_us`` accepts (:data:`_EPOCH_US_TYPES`). + Every other type -- VARCHAR, INTEGER, DECIMAL -- keeps the plain + comparison, because ``epoch_us`` raises BinderException on it. The decision + is the COLUMN's type, never the cursor text's shape: a VARCHAR column of + timestamp-looking strings produces a cursor indistinguishable from a real + timestamp's. + Pinning the session zone to UTC would also have worked and is NOT used: ``TimeZone`` is an ICU-backed setting, the sandbox forbids autoloading extensions, and where ICU is absent DuckDB does not even recognise the @@ -1442,11 +1474,11 @@ def _cursor_comparison(field_sql: str, value: Any) -> str: The cost, accepted deliberately: wrapping the column in a function can stop the predicate being pushed into the Parquet or JSON reader, so a large source scans more than it would have. Correct rows beat a faster wrong - answer, and only a timestamp cursor takes this path. + answer, and only a temporal column takes this path. """ text = _canonical_cursor_text(value) literal = quote_ansi_string_literal(text) - if _cursor_is_timestamp(text): + if column_type and column_type.split("(", 1)[0].strip() in _EPOCH_US_TYPES: return f"epoch_us({field_sql}) > epoch_us(TIMESTAMP {literal})" return f"{field_sql} > {literal}" @@ -1616,7 +1648,7 @@ def _execute(ctx: RunContext, runner: DuckdbRunner) -> RunResult: # ``last_cursor_value`` is the value we filtered on (None for # full refresh OR first incremental run); kept for the # ``cursor_advanced`` flag emitted on the StreamResult. - sel, last_cursor_value = _apply_incremental_filter(ctx, stream, base_sel) + sel, last_cursor_value = _apply_incremental_filter(ctx, stream, base_sel, con) # Build quality-gate predicates. When gates are declared, # we partition the SELECT into a GOOD set (lands in the # destination) and a BAD set (routed to DLQ). ``good_where`` diff --git a/tests/build_runners/test_duckdb_cursor_canonical.py b/tests/build_runners/test_duckdb_cursor_canonical.py index bc996f12..e4a29066 100644 --- a/tests/build_runners/test_duckdb_cursor_canonical.py +++ b/tests/build_runners/test_duckdb_cursor_canonical.py @@ -52,8 +52,8 @@ from fluid_build.build_runners.duckdb.runner import ( _apply_incremental_filter, _canonical_cursor_text, + _cursor_column_type, _cursor_comparison, - _cursor_is_timestamp, ) @@ -145,44 +145,132 @@ def __init__(self, value: str) -> None: def test_the_emitted_where_clause_compares_instants() -> None: """Exercises the wiring, which a direct call to the helpers does not. - Without this, removing either helper from ``_apply_incremental_filter`` - leaves every other test in this file passing. + Given a real connection, so the column probe answers: without this, + removing either helper from ``_apply_incremental_filter`` leaves every + other test in this file passing. """ - sql, last = _apply_incremental_filter( - _Ctx("2026-05-01 10:00:00-04:00"), "evt", "SELECT * FROM src" - ) - assert "epoch_us(" in sql, f"a timestamp cursor must compare instants: {sql}" - assert "'2026-05-01 14:00:00'" in sql, sql - assert "-04:00" not in sql, f"the raw offset-bearing cursor reached the SQL: {sql}" - # The returned value stays the value that was stored, so a caller that - # records it sees what it persisted. - assert last == "2026-05-01 10:00:00-04:00" + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = "SELECT * FROM (VALUES (TIMESTAMP '2026-05-01 20:00:00')) AS t(event_time)" + sql, last = _apply_incremental_filter(_Ctx("2026-05-01 10:00:00-04:00"), "evt", base, con) + assert "epoch_us(" in sql, f"a timestamp cursor must compare instants: {sql}" + assert "'2026-05-01 14:00:00'" in sql, sql + assert "-04:00" not in sql, f"the raw offset-bearing cursor reached the SQL: {sql}" + # The predicate has to execute, not merely read correctly. + assert con.execute(f"SELECT count(*) FROM ({sql})").fetchall() == [(1,)] + # The returned value stays the value that was stored, so a caller that + # records it sees what it persisted. + assert last == "2026-05-01 10:00:00-04:00" + finally: + con.close() def test_an_integer_cursor_keeps_the_plain_comparison() -> None: """``epoch_us`` of an integer column would fail, so a scalar must not take it.""" - sql, last = _apply_incremental_filter(_Ctx("42"), "evt", "SELECT * FROM src") + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = "SELECT * FROM (VALUES (41), (43)) AS t(event_time)" + sql, last = _apply_incremental_filter(_Ctx("42"), "evt", base, con) + assert "epoch_us(" not in sql, sql + assert "'42'" in sql, sql + assert con.execute(f"SELECT count(*) FROM ({sql})").fetchall() == [(1,)] + assert last == "42" + finally: + con.close() + + +def test_without_a_connection_the_predicate_stays_as_it_was() -> None: + """No probe, no instant path: the pre-change comparison, never a broken one.""" + sql, _ = _apply_incremental_filter(_Ctx("2026-05-01 14:00:00"), "evt", "SELECT * FROM src") assert "epoch_us(" not in sql, sql - assert "'42'" in sql, sql - assert last == "42" + assert "'2026-05-01 14:00:00'" in sql, sql @pytest.mark.parametrize( - "text, is_ts", + "column_type, instant", [ - ("2026-05-01 14:00:00", True), - ("2026-05-01T14:00:00", True), - ("2026-05-01", True), - ("42", False), - # Python 3.11+ parses basic ISO, so this must be excluded explicitly: - # a date-as-integer cursor lives in an INTEGER column. - ("20260501", False), - ("20260501T140000", False), - ("order-000123", False), - ("", False), + # epoch_us accepts exactly these, measured identically on 1.5.6 and 2.0.0. + ("TIMESTAMP", True), + ("TIMESTAMP WITH TIME ZONE", True), + ("TIMESTAMPTZ", True), + ("TIMESTAMP_MS", True), + ("TIMESTAMP_NS", True), + ("DATE", True), + ("TIME", True), + # Everything else raises BinderException under epoch_us, so it must not + # be emitted. VARCHAR is the one that matters: a VARCHAR column of + # timestamp-looking strings yields a cursor indistinguishable from a + # real timestamp's, so deciding from the VALUE would break every + # incremental run on such a column. + ("VARCHAR", False), + ("INTEGER", False), + ("BIGINT", False), + ("DECIMAL(18,2)", False), + ('"NULL"', False), + (None, False), ], ) -def test_only_a_timestamp_cursor_takes_the_instant_path(text: str, is_ts: bool) -> None: - assert _cursor_is_timestamp(text) is is_ts - predicate = _cursor_comparison("ts", text) - assert ("epoch_us(" in predicate) is is_ts, predicate +def test_only_an_epoch_capable_column_takes_the_instant_path( + column_type: object, instant: bool +) -> None: + predicate = _cursor_comparison("ts", "2026-05-01 10:00:00", column_type) + assert ("epoch_us(" in predicate) is instant, predicate + + +def test_a_varchar_cursor_column_still_runs_on_real_duckdb() -> None: + """The regression the column probe exists for. + + ``epoch_us(VARCHAR)`` is a BinderException on 1.5.6 and 2.0.0, so emitting + the instant comparison for a VARCHAR cursor column would fail every + incremental run on it. The probe has to report VARCHAR and the predicate + has to stay a plain comparison that actually executes. + """ + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = "SELECT * FROM (VALUES ('2026-05-01 10:00:00'), ('2026-05-01 20:00:00')) AS t(ts)" + assert _cursor_column_type(con, base, "ts") == "VARCHAR" + predicate = _cursor_comparison("ts", "2026-05-01 14:00:00", "VARCHAR") + assert "epoch_us(" not in predicate, predicate + rows = con.execute(f"SELECT count(*) FROM ({base}) WHERE {predicate}").fetchall() + assert rows == [(1,)], rows + finally: + con.close() + + +def test_the_probe_reads_a_real_timestamp_column_and_the_predicate_executes() -> None: + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = ( + "SELECT * FROM (VALUES (TIMESTAMP '2026-05-01 10:00:00'), " + "(TIMESTAMP '2026-05-01 20:00:00')) AS t(ts)" + ) + assert _cursor_column_type(con, base, "ts") == "TIMESTAMP" + predicate = _cursor_comparison("ts", "2026-05-01 14:00:00", "TIMESTAMP") + assert "epoch_us(" in predicate, predicate + assert con.execute(f"SELECT count(*) FROM ({base}) WHERE {predicate}").fetchall() == [(1,)] + finally: + con.close() + + +def test_the_probe_answers_for_an_empty_source() -> None: + """A first incremental run over a source with no rows must still bind.""" + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = "SELECT CAST(NULL AS TIMESTAMP) AS ts WHERE false" + assert _cursor_column_type(con, base, "ts") == "TIMESTAMP" + finally: + con.close() + + +def test_the_probe_returns_none_rather_than_raising_on_a_bad_source() -> None: + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + assert _cursor_column_type(con, "SELECT * FROM nope_does_not_exist", "ts") is None + finally: + con.close() diff --git a/tests/build_runners/test_schema_evolution_type_family.py b/tests/build_runners/test_schema_evolution_type_family.py index 0eb62231..1021f704 100644 --- a/tests/build_runners/test_schema_evolution_type_family.py +++ b/tests/build_runners/test_schema_evolution_type_family.py @@ -36,6 +36,8 @@ import itertools +import pytest + from fluid_build.api.schema import SchemaColumn from fluid_build.build_runners._schema_evolution import ( _NUMERIC_ORDER, @@ -117,3 +119,91 @@ def test_the_rename_is_reported_rather_than_hidden() -> None: def test_the_family_check_is_case_and_space_insensitive() -> None: assert _event("DOUBLE", "HugeInt") == "none" assert _event("timestamp", " TIMESTAMPTZ ") == "none" + + +# ── the collapse is allowed, but it is not silent ──────────────────────── + + +def test_the_acquisition_gate_logs_a_family_rename(caplog: pytest.LogCaptureFixture) -> None: + """Suppressing the FAILURE is intended; suppressing the NOTICE is not. + + ``diff_columns`` emits no event for a family rename, so the drift gate + returns early and a contract that declared ``schemaPolicy: strict`` would + otherwise hear nothing at all — even though DOUBLE and HUGEINT are not + value-identical. Without this wiring ``family_rename_note`` had no + production caller, which made the module comment's claim that the change is + "reported rather than hidden" untrue. + """ + import logging + + from fluid_build.build_runners._acquisition_common import _log_family_renames + + with caplog.at_level(logging.WARNING): + _log_family_renames( + [SchemaColumn(name="big", type="double"), SchemaColumn(name="other", type="varchar")], + [SchemaColumn(name="big", type="hugeint"), SchemaColumn(name="other", type="varchar")], + ) + messages = [r.message for r in caplog.records] + assert "acquisition_schema_family_rename" in messages, messages + assert ( + len([m for m in messages if m == "acquisition_schema_family_rename"]) == 1 + ), "only the renamed column should be reported, not the unchanged one" + + +def test_the_gate_logs_nothing_when_no_type_changed(caplog: pytest.LogCaptureFixture) -> None: + import logging + + from fluid_build.build_runners._acquisition_common import _log_family_renames + + with caplog.at_level(logging.WARNING): + _log_family_renames( + [SchemaColumn(name="a", type="double")], + [SchemaColumn(name="a", type="double")], + ) + assert [r.message for r in caplog.records] == [] + + +def test_the_rename_reaches_the_operator_through_the_real_gate( + caplog: pytest.LogCaptureFixture, +) -> None: + """Through ``enforce_schema_policy_or_raise``, not the helper directly. + + This is the test that fails if the call is ever removed from the gate: a + direct call to ``_log_family_renames`` passes either way. ``schemaPolicy: + strict`` is used deliberately -- that is the contract that would otherwise + hear nothing, since the family collapse means the gate returns before it + raises. + """ + import logging + + from fluid_build.api.schema import SchemaColumn, SchemaFingerprint + from fluid_build.build_runners._acquisition_common import ( + enforce_schema_policy_or_raise, + ) + + contract = { + "exposes": [ + { + "contract": { + "schema": [{"name": "big", "type": "double"}], + "schemaPolicy": "strict", + } + } + ] + } + + class _Ctx: + def __init__(self) -> None: + self.contract = contract + + class _Runner: + def fingerprint(self, _ctx: object) -> SchemaFingerprint: + # What a DuckDB 2.0 install now infers for the same source bytes. + return SchemaFingerprint.of([SchemaColumn(name="big", type="hugeint")]) + + with caplog.at_level(logging.WARNING): + enforce_schema_policy_or_raise(_Ctx(), _Runner()) + + assert "acquisition_schema_family_rename" in [ + r.message for r in caplog.records + ], "the gate allowed the family rename without telling anyone" From 33c3c85dc5761172aff6113bb5e782b9d504620e Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:18:28 +0100 Subject: [PATCH 5/6] fix(duckdb): the discover remap moved the drift failure instead of removing it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third must-fix from the review, and the one that made the discover fix incomplete rather than wrong. `fluid init --discover` writes the CANONICAL type into the contract (`numeric` for a HUGEINT column) while DuckDB goes on inferring the SOURCE type (`hugeint`), and `numeric` was not in any family — so the drift gate failed at apply exactly what the emitter had just stopped failing at validate. Measured before the fix: all eight remapped numeric names plus `list` and `null` came back `type_changed`. The family table now also holds the names the emitter canonicalises TO, so the round trip is closed. Verified over the installed DuckDB's entire `duckdb_types()` list: every name discover can emit now survives the gate, on 1.5.6 and 2.0.0. The unsigned and 128-bit groups are kept apart by width (`uinteger`/`uint32`/`oid` with `bigint`; `uint16`/`usmallint` with `integer`; `uint8`/`utinyint` with `smallint`), which matches how the emitter chooses a width that holds the range. Two tests stop the tables drifting apart again: one asserts every alias in the emitter's table is same-family as its target, the other sweeps `duckdb_types()` through emit-then-diff. Neither existed, which is how this got through the first time. Quotes are stripped in `_same_type_family` because `typeof(NULL)` is the quoted string `"NULL"` — that is how an all-null column arrives. `string`/`null` is deliberately its own group with no real type in it, so `string` -> `integer` once data actually arrives is still drift. Also from the review, all confirmed and all mine: - `_canonical_cursor_text` raised an unhandled OverflowError for a year-1 or year-9999 cursor across a non-zero offset, turning a nonsense value into a failed run. Guarded; the text is returned unchanged, as the previous release compared it. - A docstring still cited the reverted time-zone pin and `_duckdb_compat.SESSION_TIME_ZONE`, a symbol this branch deleted. Corrected; no references remain. - The schema tests validated against the NEWEST bundled schema (0.7.6) while the emitter declares `fluidVersion: 0.7.3`, which is what `fluid validate` actually gates on. They now gate on the declared version AND on every bundled version, so a future bump cannot quietly break it. Checked: the mappings are valid on 0.7.3, 0.7.5 and 0.7.6 alike, so this was test fidelity, not a bug. - The emitter's wiring was pinned by nothing — 827 discover tests passed with the call site reverted. Now mutation-caught. 2486 tests green on duckdb 1.5.6 and 2.0.0. --- .../build_runners/_schema_evolution.py | 61 +++++++++- fluid_build/build_runners/duckdb/runner.py | 15 ++- .../test_duckdb_cursor_canonical.py | 55 +++++++++ .../test_schema_evolution_type_family.py | 61 ++++++++++ .../test_discover_type_canonicalization.py | 106 ++++++++++++++++-- 5 files changed, 280 insertions(+), 18 deletions(-) diff --git a/fluid_build/build_runners/_schema_evolution.py b/fluid_build/build_runners/_schema_evolution.py index 0628d7e3..85577cef 100644 --- a/fluid_build/build_runners/_schema_evolution.py +++ b/fluid_build/build_runners/_schema_evolution.py @@ -83,9 +83,56 @@ def _is_narrowing(old: str, new: str) -> bool: #: exact 128-bit integer. Collapsing the pair keeps the pipeline running, and #: the note says the physical type moved, so the precision change is reported #: rather than hidden. Founder decision, 2026-10-26. +#: Each group also has to contain the name ``cli/discover/emitter.py`` +#: canonicalises its members TO, or the fix for one bug would cause another: +#: discover writes the canonical name into the contract, DuckDB keeps inferring +#: the source name, and the gate would then fail at apply what it had just +#: stopped failing at validate. ``tests/.../test_schema_evolution_type_family.py`` +#: pins that every alias in the emitter's table is same-family as its target, +#: so the two tables cannot drift apart silently. _TYPE_FAMILIES: Tuple[Tuple[str, ...], ...] = ( - ("double", "hugeint", "bignum", "uhugeint", "int128", "uint128", "varint"), - ("timestamp", "timestamptz", "timestamp with time zone", "timestamp_tz"), + # 128-bit and unsigned-64 integers, and the floats they are inferred as. + # `numeric` is the emitter's canonical name for all of them. + ( + "double", + "numeric", + "hugeint", + "bignum", + "uhugeint", + "int128", + "uint128", + "uint64", + "ubigint", + "varint", + ), + # unsigned 32-bit does not fit `integer`, so the emitter writes `bigint` + ("bigint", "uinteger", "uint32", "oid"), + ("integer", "uint16", "usmallint", "integral", "signed"), + ("smallint", "uint8", "utinyint", "short"), + ("tinyint", "int1"), + ( + "timestamp", + "timestamptz", + "timestamp with time zone", + "timestamp_tz", + "timestamp_s", + "timestamp_ms", + "timestamp_us", + "timestamp_ns", + "timestamptz_ns", + ), + ("time", "time_ns", "timetz", "time with time zone"), + # Non-numeric spellings of one physical type. `string`/`null` covers an + # all-null column: discover writes `string` for it and DuckDB keeps + # inferring the quoted `"NULL"`. It deliberately does NOT include a real + # type, so `string` -> `integer` once data arrives is still drift. + ("varchar", "bpchar"), + ("bit", "bitstring"), + ("boolean", "logical"), + ("array", "list"), + ("struct", "tuple"), + ("variant", "union"), + ("string", "null", "type"), ) _FAMILY_OF: Dict[str, int] = { @@ -94,9 +141,13 @@ def _is_narrowing(old: str, new: str) -> bool: def _same_type_family(old: str, new: str) -> bool: - """Whether ``old`` and ``new`` are two spellings of one physical type.""" - a = _FAMILY_OF.get(old.strip().lower()) - b = _FAMILY_OF.get(new.strip().lower()) + """Whether ``old`` and ``new`` are two spellings of one physical type. + + Quotes are stripped because ``typeof(NULL)`` is the quoted string + ``"NULL"``, which is how an all-null column reaches here. + """ + a = _FAMILY_OF.get(old.strip().strip('"').lower()) + b = _FAMILY_OF.get(new.strip().strip('"').lower()) return a is not None and a == b diff --git a/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index 9d9ec6b6..4d7c63e2 100644 --- a/fluid_build/build_runners/duckdb/runner.py +++ b/fluid_build/build_runners/duckdb/runner.py @@ -1353,9 +1353,8 @@ def _canonical_cursor_text(value: Any) -> str: An offset-bearing timestamp is therefore converted to UTC and rendered WITHOUT an offset. Naive-UTC text denotes the same instant against a naive - TIMESTAMP column and, under the pinned UTC session - (:data:`~fluid_build.providers._duckdb_compat.SESSION_TIME_ZONE`), against a - TIMESTAMPTZ one. + TIMESTAMP column, and it is the text the instant comparison in + :func:`_cursor_comparison` reads as UTC against a TIMESTAMPTZ one. This is also what repairs state already on disk, so no cursor file has to be rewritten. A shipped DuckDB 2.0 install on a host with a non-zero offset @@ -1391,7 +1390,15 @@ def _canonical_cursor_text(value: Any) -> str: return text if parsed.tzinfo is None: return text - return parsed.astimezone(timezone.utc).replace(tzinfo=None).isoformat(sep=" ") + try: + shifted = parsed.astimezone(timezone.utc) + except (OverflowError, OSError): + # Converting year 1 or year 9999 across a non-zero offset leaves + # datetime's range. Such a cursor is not a real high-water mark, and + # returning it unchanged keeps the previous release's behaviour rather + # than failing the run on it. + return text + return shifted.replace(tzinfo=None).isoformat(sep=" ") #: The column types ``epoch_us`` accepts. Measured on duckdb 1.5.6 and 2.0.0, diff --git a/tests/build_runners/test_duckdb_cursor_canonical.py b/tests/build_runners/test_duckdb_cursor_canonical.py index e4a29066..3115ff25 100644 --- a/tests/build_runners/test_duckdb_cursor_canonical.py +++ b/tests/build_runners/test_duckdb_cursor_canonical.py @@ -274,3 +274,58 @@ def test_the_probe_returns_none_rather_than_raising_on_a_bad_source() -> None: assert _cursor_column_type(con, "SELECT * FROM nope_does_not_exist", "ts") is None finally: con.close() + + +def test_the_cross_version_scenario_executes_on_a_tz_aware_column() -> None: + """The actual defect, executed rather than asserted as SQL text. + + Reproduces the shape without needing two interpreters: the column is + TIMESTAMPTZ (what DuckDB 2.0 infers for a JSON timestamp carrying ``Z``) + and the stored cursor is the naive text DuckDB 1.5 persisted for the same + instant. The plain literal comparison is what shifted by the host offset; + the instant comparison must select exactly the rows after that instant, + whatever the session zone is set to. + """ + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + base = ( + "SELECT * FROM (VALUES " + "(1, TIMESTAMPTZ '2026-05-01 10:00:00+00'), " + "(2, TIMESTAMPTZ '2026-05-01 14:00:00+00'), " + "(3, TIMESTAMPTZ '2026-05-01 18:00:00+00'), " + "(4, TIMESTAMPTZ '2026-05-01 22:00:00+00')) AS t(id, ts)" + ) + assert _cursor_column_type(con, base, "ts") in { + "TIMESTAMP WITH TIME ZONE", + "TIMESTAMPTZ", + } + predicate = _cursor_comparison("ts", "2026-05-01 14:00:00", "TIMESTAMP WITH TIME ZONE") + for zone in ("UTC", "America/New_York", "Europe/Berlin"): + try: + con.execute(f"SET TimeZone = '{zone}'") + except duckdb.Error: # pragma: no cover - needs the icu extension + pytest.skip("this DuckDB cannot set TimeZone (no icu)") + rows = con.execute(f"SELECT id FROM ({base}) WHERE {predicate} ORDER BY id").fetchall() + assert rows == [(3,), (4,)], f"session zone {zone} changed the rows: {rows}" + finally: + con.close() + + +@pytest.mark.parametrize( + "stored", + [ + "9999-12-31 23:59:59-05:00", + "0001-01-01 00:00:00+05:00", + "9999-12-31 23:59:59.999999-01:00", + ], +) +def test_a_near_bounds_cursor_does_not_crash_the_run(stored: str) -> None: + """Shifting year 1 or 9999 across an offset leaves datetime's range. + + It raised an unhandled OverflowError on both the read and the write path, + which turns a nonsense cursor value into a failed run. + """ + assert _canonical_cursor_text(stored) == stored + # And the predicate it builds is still well formed. + assert "'" in _cursor_comparison("ts", stored, "TIMESTAMP") diff --git a/tests/build_runners/test_schema_evolution_type_family.py b/tests/build_runners/test_schema_evolution_type_family.py index 1021f704..f8040a6d 100644 --- a/tests/build_runners/test_schema_evolution_type_family.py +++ b/tests/build_runners/test_schema_evolution_type_family.py @@ -207,3 +207,64 @@ def fingerprint(self, _ctx: object) -> SchemaFingerprint: assert "acquisition_schema_family_rename" in [ r.message for r in caplog.records ], "the gate allowed the family rename without telling anyone" + + +# ── the family table and the discover emitter must not drift apart ─────── + + +def test_every_discover_alias_is_same_family_as_its_target() -> None: + """Otherwise fixing one bug causes another, which is what happened. + + `fluid init --discover` writes the CANONICAL name into the contract while + DuckDB keeps inferring the SOURCE name, so unless the two are one family the + drift gate fails at apply exactly what the emitter stopped failing at + validate -- the failure moves rather than going away. + """ + from fluid_build.build_runners._schema_evolution import _same_type_family + from fluid_build.cli.discover.emitter import _SCHEMA_TYPE_ALIASES + + missing = [ + (source, target) + for source, target in _SCHEMA_TYPE_ALIASES.items() + if not _same_type_family(source, target) + ] + assert missing == [], f"add these to _TYPE_FAMILIES: {missing}" + + +def test_no_duckdb_type_name_trips_the_gate_after_discover() -> None: + """The round trip over the INSTALLED DuckDB's whole type list. + + Catches a future DuckDB type name here rather than in a user's failed apply. + """ + duckdb = pytest.importorskip("duckdb") + from fluid_build.cli.discover.emitter import canonical_column_type + + con = duckdb.connect(":memory:") + try: + names = [ + row[0] + for row in con.execute( + "SELECT DISTINCT lower(type_name) FROM duckdb_types()" + ).fetchall() + ] + finally: + con.close() + tripped = [ + (name, canonical_column_type(name)) + for name in names + if _event(canonical_column_type(name), name) != "none" + ] + assert tripped == [], ( + f"duckdb {duckdb.__version__}: discover would write these and the drift " + f"gate would then fail them at apply: {tripped}" + ) + + +def test_an_all_null_column_round_trips() -> None: + """``typeof(NULL)`` is the quoted string ``"NULL"``, which must still match.""" + from fluid_build.cli.discover.emitter import canonical_column_type + + assert canonical_column_type('"NULL"') == "string" + assert _event("string", '"NULL"') == "none" + # ... but a column that gains a real type is still drift. + assert _event("string", "integer") == "type_changed" diff --git a/tests/cli/discover/test_discover_type_canonicalization.py b/tests/cli/discover/test_discover_type_canonicalization.py index 2ae3e05c..f28d42a1 100644 --- a/tests/cli/discover/test_discover_type_canonicalization.py +++ b/tests/cli/discover/test_discover_type_canonicalization.py @@ -30,7 +30,7 @@ import json from pathlib import Path -from typing import Any, Dict +from typing import Any, Dict, List import pytest @@ -40,15 +40,41 @@ SCHEMAS = REPO_ROOT / "fluid_build" / "schemas" -def _type_schema() -> Dict[str, Any]: - newest = sorted(SCHEMAS.glob("fluid-schema-*.json"))[-1] - doc = json.loads(newest.read_text(encoding="utf-8")) +def _type_schema(version: str) -> Dict[str, Any]: + doc = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) return doc["$defs"]["column"]["properties"]["type"] -def _validator() -> Any: +def _bundled_versions() -> List[str]: + return sorted(p.stem.replace("fluid-schema-", "") for p in SCHEMAS.glob("fluid-schema-*.json")) + + +def _emitted_version() -> str: + """The schema version the emitter DECLARES, which is what gates the contract. + + Validating against the newest bundled schema would be the wrong gate: + `fluid validate` reads `fluidVersion` out of the emitted contract, and the + emitter writes an older one than the newest bundled. + """ + import fluid_build.cli.discover.emitter as emitter + + contract = emitter.emit_contract( + product_id="p", + name="p", + domain="d", + owner_team="t", + owner_email="e", + engine="duckdb", + source_kind="filesystem", + connection={}, + streams=[], + ) + return str(contract["fluidVersion"]) + + +def _validator(version: str) -> Any: jsonschema = pytest.importorskip("jsonschema") - return jsonschema.Draft202012Validator(_type_schema()) + return jsonschema.Draft202012Validator(_type_schema(version)) @pytest.mark.parametrize( @@ -93,7 +119,7 @@ def test_named_types_map_as_documented(raw: str, expected: str) -> None: def test_every_mapped_type_satisfies_the_bundled_schema() -> None: """The documented aliases must each land on a value the schema accepts.""" - validator = _validator() + validator = _validator(_emitted_version()) bad = [] for raw in ( "HUGEINT", @@ -139,7 +165,7 @@ def test_no_duckdb_type_name_is_emitted_invalid() -> None: and a 2.0 install enumerate different sets, and both must map cleanly. """ duckdb = pytest.importorskip("duckdb") - validator = _validator() + validator = _validator(_emitted_version()) con = duckdb.connect(":memory:") try: names = [ @@ -164,7 +190,7 @@ def test_no_duckdb_type_name_is_emitted_invalid() -> None: def test_an_already_valid_name_is_never_remapped() -> None: """A contract emitted before this change must keep the same types.""" - validator = _validator() + validator = _validator(_emitted_version()) duckdb = pytest.importorskip("duckdb") con = duckdb.connect(":memory:") try: @@ -182,3 +208,65 @@ def test_an_already_valid_name_is_never_remapped() -> None: if validator.is_valid(n) and canonical_column_type(n) != n ] assert changed == [], f"already-valid type names were remapped: {changed}" + + +def test_every_bundled_schema_accepts_what_the_emitter_writes() -> None: + """Stronger than the declared version alone: a bump must not break this. + + The emitter's `fluidVersion` has moved before and will again, so every + bundled schema has to accept the canonical names rather than only the one + currently declared. + """ + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect(":memory:") + try: + names = [ + row[0] + for row in con.execute( + "SELECT DISTINCT lower(type_name) FROM duckdb_types()" + ).fetchall() + ] + finally: + con.close() + broken = {} + for version in _bundled_versions(): + validator = _validator(version) + bad = [ + (n, canonical_column_type(n)) + for n in names + if not validator.is_valid(canonical_column_type(n)) + ] + if bad: + broken[version] = bad + assert broken == {}, f"schema versions rejecting an emitted type: {broken}" + + +def test_the_emitter_actually_applies_the_canonicalisation() -> None: + """Pins the WIRING. The discover tests pass with the call site reverted.""" + import fluid_build.cli.discover.emitter as emitter + from fluid_build.cli.discover.registry import DiscoveredColumn, DiscoveredStream + + stream = DiscoveredStream( + name="t", + columns=[ + DiscoveredColumn(name="big", type="HUGEINT"), + DiscoveredColumn(name="note", type='"NULL"'), + DiscoveredColumn(name="tags", type="INTEGER[]"), + ], + ) + contract = emitter.emit_contract( + product_id="p", + name="p", + domain="d", + owner_team="t", + owner_email="e", + engine="duckdb", + source_kind="filesystem", + connection={}, + streams=[stream], + ) + emitted = {c["name"]: c["type"] for c in contract["exposes"][0]["contract"]["schema"]} + assert emitted == {"big": "numeric", "note": "string", "tags": "array"}, emitted + validator = _validator(str(contract["fluidVersion"])) + for name, declared in emitted.items(): + assert validator.is_valid(declared), f"{name}: {declared!r} fails the declared schema" From 3f906493a6ec818bc63d93c25a79dc84010f94ad Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:42:33 +0100 Subject: [PATCH 6/6] test: one import style for the discover emitter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeQL flagged `fluid_build.cli.discover.emitter` as imported with both `import` and `import from` in this file: a top-level `from ... import canonical_column_type` plus two function-local `import ... as emitter` statements. Taken as suggested, because here it genuinely applies: the tests only call `emit_contract`, so the module object was never needed. It is now in the top-level `from` import and both local aliases are gone. (Not the same situation as the sandbox spill tests on #743, where the module object IS required — those tests replace a private helper that production code resolves as an attribute of its own module, so a bare imported name would patch nothing. Nothing here monkeypatches.) 2486 tests green on duckdb 1.5.6 and 2.0.0, and reverting the emitter's call site still fails test_the_emitter_actually_applies_the_canonicalisation, so the wiring is still pinned. --- .../cli/discover/test_discover_type_canonicalization.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/tests/cli/discover/test_discover_type_canonicalization.py b/tests/cli/discover/test_discover_type_canonicalization.py index f28d42a1..6b23915e 100644 --- a/tests/cli/discover/test_discover_type_canonicalization.py +++ b/tests/cli/discover/test_discover_type_canonicalization.py @@ -34,7 +34,7 @@ import pytest -from fluid_build.cli.discover.emitter import canonical_column_type +from fluid_build.cli.discover.emitter import canonical_column_type, emit_contract REPO_ROOT = Path(__file__).resolve().parents[3] SCHEMAS = REPO_ROOT / "fluid_build" / "schemas" @@ -56,9 +56,7 @@ def _emitted_version() -> str: `fluid validate` reads `fluidVersion` out of the emitted contract, and the emitter writes an older one than the newest bundled. """ - import fluid_build.cli.discover.emitter as emitter - - contract = emitter.emit_contract( + contract = emit_contract( product_id="p", name="p", domain="d", @@ -243,7 +241,6 @@ def test_every_bundled_schema_accepts_what_the_emitter_writes() -> None: def test_the_emitter_actually_applies_the_canonicalisation() -> None: """Pins the WIRING. The discover tests pass with the call site reverted.""" - import fluid_build.cli.discover.emitter as emitter from fluid_build.cli.discover.registry import DiscoveredColumn, DiscoveredStream stream = DiscoveredStream( @@ -254,7 +251,7 @@ def test_the_emitter_actually_applies_the_canonicalisation() -> None: DiscoveredColumn(name="tags", type="INTEGER[]"), ], ) - contract = emitter.emit_contract( + contract = emit_contract( product_id="p", name="p", domain="d",