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/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/_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 d645d4b7..85577cef 100644 --- a/fluid_build/build_runners/_schema_evolution.py +++ b/fluid_build/build_runners/_schema_evolution.py @@ -66,6 +66,109 @@ 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. +#: 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, ...], ...] = ( + # 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] = { + 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. + + 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 + + +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 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 ( + 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 +199,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/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index 1bc7e203..4d7c63e2 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 @@ -1291,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. @@ -1323,16 +1324,172 @@ 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. + 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 +#: 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 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 + 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 + 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, +#: 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``. + + 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. + """ + try: + 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, 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 + 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). + + 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 + 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 temporal column takes this path. + """ + text = _canonical_cursor_text(value) + literal = quote_ansi_string_literal(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}" + + def _persist_cursor_after_run( ctx: RunContext, stream: str, @@ -1383,7 +1540,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(), ), ) @@ -1498,7 +1655,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/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/build_runners/test_duckdb_cursor_canonical.py b/tests/build_runners/test_duckdb_cursor_canonical.py new file mode 100644 index 00000000..3115ff25 --- /dev/null +++ b/tests/build_runners/test_duckdb_cursor_canonical.py @@ -0,0 +1,331 @@ +# 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_column_type, + _cursor_comparison, +) + + +@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. + + 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. + """ + 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.""" + 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 "'2026-05-01 14:00:00'" in sql, sql + + +@pytest.mark.parametrize( + "column_type, instant", + [ + # 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_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() + + +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 new file mode 100644 index 00000000..f8040a6d --- /dev/null +++ b/tests/build_runners/test_schema_evolution_type_family.py @@ -0,0 +1,270 @@ +# 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 + +import pytest + +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" + + +# ── 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" + + +# ── 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 new file mode 100644 index 00000000..6b23915e --- /dev/null +++ b/tests/cli/discover/test_discover_type_canonicalization.py @@ -0,0 +1,269 @@ +# 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, List + +import pytest + +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" + + +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 _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. + """ + contract = 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(version)) + + +@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(_emitted_version()) + 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(_emitted_version()) + 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(_emitted_version()) + 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}" + + +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.""" + 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 = 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"