diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6ab6d257..1d84f546 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -623,6 +623,79 @@ jobs: if: always() run: python scripts/cleanup_local_test_artifacts.py + # ────────────────────────────────────────────────────────────── + # DuckDB floor. pyproject.toml takes duckdb>=1.5.0 with no ceiling, and + # DuckDB 2.0 needs Python 3.11, so the only other job on 1.5.x is the + # Python 3.10 test leg, which gets the newest 1.5 patch, never the floor. + # This installs exactly the floor the sandbox enforces (MIN_DUCKDB_VERSION, + # read from the code, so the two cannot drift) on a supported Python and + # runs what differs between DuckDB versions: the compat seam + # (providers/_duckdb_compat.py), the sandbox, the MCP DuckDB driver, the + # DuckDB acquisition runner (its database-source attaches) and every suite + # the DuckDB end-to-end job runs. tests/test_ci_duckdb_floor.py fails when a + # file the end-to-end job runs, or a test that branches on the DuckDB + # version, is missing here. + # ────────────────────────────────────────────────────────────── + duckdb-floor: + name: DuckDB floor (Python 3.12) + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + cache: pip + - name: Install with local extras, then the DuckDB floor + env: + # Same pip timeout and retried install as the test matrix (see the + # comments there). + PIP_TIMEOUT: "30" + run: | + python -m pip install --upgrade pip + for attempt in 1 2 3; do + python -m pip install -e ".[dev,local]" && break + if [ "$attempt" -eq 3 ]; then + echo "::error::pip install -e .[dev,local] failed 3 times; this is an install failure, not a test failure." + exit 1 + fi + echo "::warning::pip install -e .[dev,local] failed (attempt ${attempt}/3); retrying in 30s." + sleep 30 + done + floor=$(python -c "from fluid_build.providers._duckdb_sandbox import MIN_DUCKDB_VERSION as v; print('.'.join(map(str, v)))") + for attempt in 1 2 3; do + python -m pip install "duckdb==${floor}" && break + if [ "$attempt" -eq 3 ]; then + echo "::error::pip install duckdb==${floor} failed 3 times; this is an install failure, not a test failure." + exit 1 + fi + echo "::warning::pip install duckdb==${floor} failed (attempt ${attempt}/3); retrying in 30s." + sleep 30 + done + python -c "import duckdb, sys; print('duckdb', duckdb.__version__); sys.exit(duckdb.__version__ != '${floor}')" + - name: Run the DuckDB-version-sensitive tests on the floor + env: + # The quickstart and example suites fail, rather than skip, without + # duckdb (as in the DuckDB end-to-end job). + FLUID_REQUIRE_DUCKDB: "1" + run: | + python -m pytest -q \ + tests/providers/test_duckdb_compat.py \ + tests/providers/test_duckdb_sandbox.py \ + tests/providers/test_duckdb_sandbox_workspace_root.py \ + tests/output_ports/test_drivers_duckdb.py \ + tests/build_runners/test_duckdb_runner.py \ + tests/build_runners/test_duckdb_runner_mysql_attach.py \ + tests/test_e2e_local.py \ + tests/test_e2e_local_apply.py \ + tests/test_e2e_local_negative.py \ + tests/providers/test_local_live_happy_path.py \ + tests/test_quickstart_templates_execute.py \ + tests/test_examples_execute.py + - name: Cleanup local artifacts (defensive) + if: always() + run: python scripts/cleanup_local_test_artifacts.py + # ────────────────────────────────────────────────────────────── # dbt v2 canary. The `test` and `coverage-and-audit` jobs install # `dbt-core<2` + dbt-duckdb, so they exercise the 1.x line only. Every diff --git a/.secrets.baseline b/.secrets.baseline index e8381aa8..5e153ce6 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1350,7 +1350,7 @@ "filename": "tests/security/test_security_regressions.py", "hashed_secret": "7c9ce88af5f76ecad340e5e6a6e40bca11c3d234", "is_verified": false, - "line_number": 213 + "line_number": 217 } ], "tests/security/test_security_regressions_round2.py": [ @@ -2154,5 +2154,5 @@ } ] }, - "generated_at": "2026-10-13T20:49:43Z" + "generated_at": "2026-10-22T06:15:34Z" } diff --git a/CHANGELOG.md b/CHANGELOG.md index b5643c62..543d2ea5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Upgrade notes +- forge-cli runs on DuckDB 2.x as well as 1.5.x (`duckdb>=1.5.0`, no ceiling; DuckDB 2.0 + needs Python 3.11, so a Python 3.10 install stays on 1.5.x). On 2.x, a DuckDB database + file forge-cli creates (`fluid demo`'s `.fluid/db.duckdb`, the local provider's session + and `persist` databases, a Meltano acquisition's DuckDB destination) is written in the + format DuckDB 1.5 writes, which DuckDB 1.0 and later open, so a type that needs a newer + format, such as a `VARIANT` column, is refused in it on 2.x as it is on 1.5; a file + that already exists keeps its format. + DuckDB 2.x's own default format cannot be read by 1.5: a file another 2.x tool created + (dbt-duckdb and dlt included), or forge-cli 0.21.0 or older on DuckDB 2.0, opens only + on 2.x. To convert one, run on 2.x: `ATTACH 'old.duckdb' AS src (READ_ONLY); ATTACH + 'new.duckdb' AS dst (STORAGE_VERSION 'v1.0.0'); COPY FROM DATABASE src TO dst;`. +- Contract SQL runs on the DuckDB that is installed, and DuckDB 2.0 changed some SQL: the + `x -> x + 1` lambda is an error (write `lambda x: x + 1`), and integer `//` and `%` by + zero raise an error instead of returning NULL (wrap them in `TRY(...)`). See + https://duckdb.org/docs/2.0/operations_manual/migration.html. +- A DuckDB acquisition build now resolves a Postgres source's `connection.secretRef`. It + ignored it, and libpq's `PGPASSWORD` or `~/.pgpass` supplied the password, so a + `secretRef` that does not resolve where `fluid apply` runs now fails the build. To keep + using `PGPASSWORD`, remove the `secretRef`; an inline `password` still wins over it. - `fluid validate --env --strict` (on one contract or a whole workspace), `fluid contract-validation --env --strict` and `fluid test --env --strict` (the native engine; `--engine soda` does not read `--strict`) now fail when an overlay patch @@ -52,6 +71,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 here`, or the rule's own `description`, instead of quoting the whole object and the subschema it must not match. +### Fixed + +- **DuckDB 2.0.** `fluid verify` reported every local file as `error` ("Relations no + longer have fetchone()"); `fluid apply`'s local SQL and materialize steps, and the + actions of the legacy `fluid_build.contract_tests` module, lost their row counts (a + materialize step logged `local_materialize_fallback`, a `contract_tests` action + reported `-1` rows). DuckDB 2.0 relations have no `fetchone()`, and a query result + keeps its connection busy until it is read to the end, which fails the next step on + that connection. Every DuckDB result is now read to the end, the same way on 1.5.x and + 2.x. +- **DuckDB 2.0: Postgres and MySQL sources.** A DuckDB acquisition build reading + Postgres failed on DuckDB 2.0 (`Table Function with name postgres_scan does not + exist!`): the postgres extension dropped `postgres_scan`. A MySQL or MariaDB source + failed on every DuckDB version, before connecting (`Unknown secret storage found: + 'local_file'`): its `ATTACH` looked for a default secret in the persistent storage the + sandbox turns off. Both are now attached read-only before the sandbox locks, each with + an empty temporary secret named in the `ATTACH` (the DSN carries the credentials), and + read as `..` on 1.5.x and 2.x; a Postgres source's + `connection.secretRef` is resolved into the password, as a MySQL source's already was. + A Postgres enum column still reads, and lands, as text. A stream name is looked up as + DuckDB looks up names, ignoring case: where two tables' names differ only in case, the + build can read the other one. + A source the runner cannot attach (unreachable, refused, a `secretRef` that does not + resolve) fails the build with the reason, its password masked, and writes a run + record; a MySQL or MariaDB source used to end `fluid apply` with `Unexpected error + during execution` instead. +- A local SQL action, or a legacy `fluid_build.contract_tests` action, whose SQL ends in + a statement that returns no rows (`CREATE TABLE ... AS`, an `INSERT` without + `RETURNING`) now fails with a message saying to write just the `SELECT`. DuckDB 1.5 + failed there on `'NoneType' object has no attribute 'write_csv'`, and DuckDB 2.0 + reported success with a one-row `Count` file, an empty `Success` file (a `CREATE TABLE + ... AS PIVOT`) or the rows of an earlier `SELECT` in the same SQL as the output. +- **DuckDB 2.0: contract SQL.** A local SQL action whose SQL ends in a comment after its + last `;` failed to write its output (`syntax error at or near ";"`); SQL with a + statement that does not parse ran the statements ahead of it before failing, where + DuckDB 1.5 runs none; and the result of a `PIVOT` without an `IN` list, or of a query + calling `nextval()`, could not be written to an `s3://` output or, in + `fluid_build.contract_tests`, to Parquet (`Cannot create the view ...: it reads rows + held in this process`). Each now behaves as on 1.5. + ## [0.21.0] — 2026-10-19 ### Upgrade notes diff --git a/examples/source-aligned-postgres-duckdb/README.md b/examples/source-aligned-postgres-duckdb/README.md index 189d7448..5121ea93 100644 --- a/examples/source-aligned-postgres-duckdb/README.md +++ b/examples/source-aligned-postgres-duckdb/README.md @@ -1,8 +1,8 @@ # Source-aligned Postgres → DuckDB → Parquet A minimal source-aligned data product (`Bronze` / `SDP`) that acquires -`public.orders` from a Postgres source via DuckDB's `postgres_scan` and -lands it as Parquet. +`public.orders` from a Postgres source through DuckDB's postgres extension +(the source is attached read-only) and lands it as Parquet. **Vocabulary note.** v0.7.3 carries two interchangeable classifications: the medallion `metadata.layer` (`Bronze`/`Silver`/`Gold`) and the Data @@ -34,7 +34,7 @@ The pipeline: ``` fluid validate contract.fluid.yaml -fluid apply --build ingest_orders contract.fluid.yaml +fluid apply contract.fluid.yaml --mode amend-and-build --build-id ingest_orders ``` `apply` detects `pattern: acquisition` + `engine: duckdb`, dispatches to @@ -45,7 +45,10 @@ table to `out/orders.parquet`, and persists a run record under ## What this exercises - v0.7.3 `acquisition` build pattern + `duckdb` engine -- Postgres source via `postgres_scan` (no Airbyte server needed) +- Postgres source attached by DuckDB's postgres extension (no Airbyte server needed). + The attach is `READ_ONLY`, which refuses writes through DuckDB's catalog but + is no write barrier on the server; outside this demo, connect as a role that + can only read (for example one granted `pg_read_all_data`) - Parquet sink via `COPY (...) TO ... (FORMAT 'parquet')` - Run-record persistence + lock acquisition - Pre-land hook chain (`dlp_scan`, `quality_gate`) — non-blocking on this diff --git a/examples/source-aligned-postgres-duckdb/contract.fluid.yaml b/examples/source-aligned-postgres-duckdb/contract.fluid.yaml index a848c18b..0ac03e33 100644 --- a/examples/source-aligned-postgres-duckdb/contract.fluid.yaml +++ b/examples/source-aligned-postgres-duckdb/contract.fluid.yaml @@ -4,7 +4,7 @@ id: bronze.crm_orders name: CRM Orders Bronze description: | Source-aligned Bronze data product. Acquires raw orders from a Postgres - source via DuckDB's postgres_scan and lands them as Parquet for downstream + source through DuckDB's postgres extension and lands them as Parquet for downstream Silver/Gold consumption. domain: sales diff --git a/fluid_build/build_runners/_embedded_sql_io.py b/fluid_build/build_runners/_embedded_sql_io.py index 6db965aa..95d78ec6 100644 --- a/fluid_build/build_runners/_embedded_sql_io.py +++ b/fluid_build/build_runners/_embedded_sql_io.py @@ -471,12 +471,13 @@ def relations_read(sql: str) -> Optional[FrozenSet[str]]: if not str(sql or "").strip(): return None try: + from fluid_build.providers._duckdb_compat import first_row from fluid_build.providers._duckdb_sandbox import DuckDBAllowlist, secure_duckdb_connect # Parsed only, never bound or run: no file, URL or extension at all. con = secure_duckdb_connect(":memory:", allow=DuckDBAllowlist.none()) try: - row = con.execute("SELECT json_serialize_sql(?)", [str(sql)]).fetchone() + row = first_row(con.execute("SELECT json_serialize_sql(?)", [str(sql)])) finally: con.close() doc = json.loads(row[0]) if row and row[0] else {} diff --git a/fluid_build/build_runners/_late_arrival.py b/fluid_build/build_runners/_late_arrival.py index 95209bd6..151cd347 100644 --- a/fluid_build/build_runners/_late_arrival.py +++ b/fluid_build/build_runners/_late_arrival.py @@ -370,6 +370,7 @@ def split_late_events_in_duckdb( can call this too. """ # ── input validation ──────────────────────────────────────────── + from fluid_build.providers._duckdb_compat import first_row, one_row from fluid_build.providers._sql_safety import validate_ident validate_ident(source_relation) @@ -380,9 +381,9 @@ def split_late_events_in_duckdb( return {"on_time": 0, "late": 0} # ── compute watermark ──────────────────────────────────────────── - watermark_row = con.execute( - f"SELECT max({event_time_column}) FROM {source_relation}" - ).fetchone() + watermark_row = first_row( + con.execute(f"SELECT max({event_time_column}) FROM {source_relation}") + ) watermark = watermark_row[0] if watermark_row else None if watermark is None: # Empty source — no enforcement possible. @@ -401,9 +402,9 @@ def split_late_events_in_duckdb( ) # ── append late rows to side-output ────────────────────────────── - late_count = con.execute( - f"SELECT count(*) FROM {source_relation} WHERE {threshold_expr}" - ).fetchone()[0] + late_count = one_row( + con.execute(f"SELECT count(*) FROM {source_relation} WHERE {threshold_expr}") + )[0] if late_count > 0: con.execute( f"INSERT INTO {side_output_relation} " @@ -411,7 +412,7 @@ def split_late_events_in_duckdb( ) con.execute(f"DELETE FROM {source_relation} WHERE {threshold_expr}") - on_time_count = con.execute(f"SELECT count(*) FROM {source_relation}").fetchone()[0] + on_time_count = one_row(con.execute(f"SELECT count(*) FROM {source_relation}"))[0] return {"on_time": int(on_time_count), "late": int(late_count)} diff --git a/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index 7dd7d0df..1bc7e203 100644 --- a/fluid_build/build_runners/duckdb/runner.py +++ b/fluid_build/build_runners/duckdb/runner.py @@ -21,9 +21,9 @@ Reader dispatch by ``source.kind``: - ``filesystem`` -> ``read_csv`` / ``read_parquet`` / ``read_json`` - - ``postgres`` -> ``postgres_scan(...)`` - - ``mysql`` -> ``mysql_scan(...)`` - - ``sqlite`` -> ``sqlite_scan(...)`` + - ``postgres`` -> ``ATTACH ... (TYPE postgres, READ_ONLY)``, then ``..
`` + - ``mysql`` -> ``ATTACH ... (TYPE mysql, READ_ONLY)``, then ``..
`` + - ``sqlite`` -> ``ATTACH ... (TYPE sqlite)``, then ``.
`` - ``http`` -> ``read_csv_auto('https://...')`` Loads required extensions on demand (``httpfs``, ``postgres``, ``mysql``, @@ -57,6 +57,7 @@ ) from fluid_build.api.schema import SchemaFingerprint from fluid_build.api.source import AcquisitionMode +from fluid_build.providers._duckdb_compat import first_row from fluid_build.providers._duckdb_sandbox import ( DuckDBAllowlist, confine_declared, @@ -396,37 +397,109 @@ def _build_select_for_filesystem(uri: str, fmt: str, opts: Dict[str, Any]) -> st raise ValueError(f"unsupported reader.format for filesystem: {fmt}") -def _build_select_for_postgres(connection: Dict[str, Any], stream: str) -> str: - """``postgres_scan('host=… port=… dbname=… user=… password=…', schema, table)``. +def _postgres_attach_sql(connection: Dict[str, Any], *, alias: str) -> str: + """``ATTACH '' AS (TYPE postgres, SECRET , READ_ONLY)``. - Schema/table are validated as identifiers (so ``stream='x; DROP TABLE y'`` - is rejected at build time, not executed). DSN values are passed through - ``quote_ansi_string_literal`` so embedded single quotes can't escape the - outer SQL string. The DSN itself uses libpq's ``key=value`` whitespace - format; the entire DSN is a single SQL literal at the boundary. + DuckDB 1.5's ``postgres_scan(dsn, schema, table)`` is gone from DuckDB 2.0's + postgres extension; ``ATTACH`` works on both. ``READ_ONLY`` refuses a write + through the attached catalog (``INSERT``, ``CREATE``) and runs the + connection in a read-only transaction, but it is no write barrier: + ``postgres_execute`` sends its SQL to the server, which can end that + transaction (a ``COMMIT``) and then write. Only the engine's own SQL runs on + this connection. The DSN is the shared :func:`build_libpq_dsn` one (control + characters and a non-numeric port refused), one SQL literal at the boundary. + ``SECRET `` names the empty secret :func:`_empty_secret_sql` makes. + """ + alias = validate_ident(alias) + dsn = build_libpq_dsn(connection, database_key="dbname") + return ( + f"ATTACH {quote_ansi_string_literal(dsn)} AS {alias} " + f"(TYPE postgres, SECRET {alias}, READ_ONLY)" + ) + + +def _empty_secret_sql(secret_type: str, *, alias: str) -> str: + """``CREATE TEMPORARY SECRET (TYPE postgres|mysql)``: empty, the DSN carries the credentials. - Delegates DSN construction to the shared :func:`build_libpq_dsn` so - postgres and mysql don't drift over time. + Without a secret named in the ``ATTACH``, the postgres and mysql extensions + look for a default one in every secret storage, and the persistent storage + the sandbox turns off answers ``Unknown secret storage found: 'local_file'`` + (DuckDB 1.5 and 2.0 alike), so no postgres or mysql source could attach. A + temporary secret lives in memory, on this connection only. + """ + if secret_type not in ("postgres", "mysql"): + raise ValueError(f"no attach secret for a {secret_type!r} source") + return f"CREATE TEMPORARY SECRET {validate_ident(alias)} (TYPE {secret_type})" + + +def _mysql_attach_sql(connection: Dict[str, Any], *, alias: str) -> str: + """``ATTACH '' AS (TYPE mysql, SECRET , READ_ONLY)``. + + mysql_scanner's own DSN syntax, not libpq's: libpq quoting let a value with + a space add ``socket=`` or ``ssl_key=``, and a ``?`` turn the rest into URI + options. :func:`build_mysql_dsn` double-quotes each value and refuses a + ``?``, a ``/`` in the host and a control character. The extension type is + ``mysql`` for a MariaDB upstream too. ``SECRET `` names the empty + secret :func:`_empty_secret_sql` makes; ``READ_ONLY``, as a postgres source + is attached, since acquisition only reads. + """ + alias = validate_ident(alias) + dsn = build_mysql_dsn(connection, database_key="database") + return ( + f"ATTACH {quote_ansi_string_literal(dsn)} AS {alias} " + f"(TYPE mysql, SECRET {alias}, READ_ONLY)" + ) + + +def _build_select_for_postgres(stream: str, *, alias: str) -> str: + """``SELECT * FROM ..
`` over the ``ATTACH``-ed postgres source. + + ``DuckdbRunner._attach_external_databases`` attaches it first + (:func:`_postgres_attach_sql`). Schema and table are validated as + identifiers, so ``stream='x; DROP TABLE y'`` is rejected at build time; a + bare ``"table"`` is read from ``public``. """ if "." in stream: schema, table = stream.split(".", 1) else: schema, table = "public", stream + alias = validate_ident(alias) schema = validate_ident(schema) table = validate_ident(table) - dsn = build_libpq_dsn(connection, database_key="dbname") - return ( - f"SELECT * FROM postgres_scan({quote_ansi_string_literal(dsn)}, " - f"{quote_ansi_string_literal(schema)}, {quote_ansi_string_literal(table)})" - ) + return f"SELECT * FROM {alias}.{schema}.{table}" + + +#: How an ``ATTACH``-ed Postgres catalog types an enum column, ``ENUM('a', 'b')``, +#: or an array of one, ``ENUM('a', 'b')[]``. Group 1 is the array suffix. +_ENUM_TYPE = re.compile(r"ENUM\(.*\)((?:\[\])*)", re.DOTALL) + + +def _postgres_enums_as_varchar(con: Any, select_sql: str) -> str: + """``select_sql`` with each enum column read as ``VARCHAR`` (an enum array as ``VARCHAR[]``). + + ``postgres_scan``, which read a Postgres source until DuckDB 2.0 dropped + it, typed an enum column ``VARCHAR``; the ``ATTACH``-ed catalog types it + ``ENUM(...)``. Read as before, an enum column is no type change to the + declared schema the gate compares, a ``range`` gate compares it as text + rather than in declaration order, and it lands as the text it always did. + The same ``SELECT * REPLACE`` the masking projection uses. + """ + replaced = [] + for row in con.execute(f"DESCRIBE {select_sql}").fetchall(): + match = _ENUM_TYPE.fullmatch(str(row[1])) + if match: + ident = '"' + str(row[0]).replace('"', '""') + '"' + replaced.append(f"CAST({ident} AS VARCHAR{match.group(1)}) AS {ident}") + if not replaced: + return select_sql + return f"SELECT * REPLACE ({', '.join(replaced)}) FROM ({select_sql})" def _build_select_for_mysql(connection: Dict[str, Any], stream: str, *, alias: str) -> str: """Read from MySQL via the duckdb ``mysql`` extension. - Requires that ``ATTACH '' AS (TYPE mysql)`` has already - been issued in the same connection (handled by - ``DuckdbRunner._attach_external_databases``). + Requires the ``ATTACH`` of :func:`_mysql_attach_sql` on the same + connection (``DuckdbRunner._attach_external_databases`` makes it). The ``alias`` is per-build (derived from the build_id) so a contract with two mysql sources doesn't collide on a hardcoded ``mysql_db`` @@ -486,6 +559,13 @@ def _mysql_alias_for_build(build_id: str) -> str: return validate_ident(candidate) +def _postgres_alias_for_build(build_id: str) -> str: + """Per-build ATTACH alias for postgres — same pattern as mysql.""" + safe = re.sub(r"[^A-Za-z0-9_]", "_", build_id) or "default" + candidate = f"postgres_{safe}" + return validate_ident(candidate) + + def _sqlite_alias_for_build(build_id: str) -> str: """Per-build ATTACH alias for sqlite — same pattern as mysql.""" safe = re.sub(r"[^A-Za-z0-9_]", "_", build_id) or "default" @@ -606,9 +686,9 @@ def _warn_workspace_not_granted(message: str) -> None: def _grant_source(allow: DuckDBAllowlist, ctx: RunContext, streams: List[str]) -> DuckDBAllowlist: """``allow`` plus where a filesystem / http source reads from. - A database source needs no grant: postgres is scanned through its DSN, and - mysql / sqlite are attached before the lock (``_attach_external_databases``), - a sqlite file only inside :func:`_declared_roots` (``confine_declared``). + A database source needs no grant: postgres / mysql / sqlite are attached + before the lock (``_attach_external_databases``), a sqlite file only inside + :func:`_declared_roots` (``confine_declared``). A local source is granted only inside :func:`_declared_roots` (a glob as the directory above its first wildcard): the contract names it, so naming ``~/.ssh/*`` must not make it readable. @@ -707,7 +787,7 @@ def fingerprint(self, ctx: RunContext) -> SchemaFingerprint: before_lock=lambda c: self._load_extensions(c, ctx), ) try: - select_sql = _select_for_first_stream(ctx) + select_sql = _select_for_first_stream(con, ctx) con.execute(f"CREATE TEMP VIEW _fp AS {select_sql}") rows = con.execute("DESCRIBE _fp").fetchall() # What lands, not what the source holds: a masked column lands as @@ -750,56 +830,59 @@ def _load_extensions(self, con: Any, ctx: RunContext) -> None: self._attach_external_databases(con, ctx) def _attach_external_databases(self, con: Any, ctx: RunContext) -> None: - """ATTACH any source kind that requires a named database alias. + """ATTACH the source database a postgres / mysql / sqlite build reads. - Postgres uses ``postgres_scan(dsn, schema, table)`` directly with - the DSN inline, so no ATTACH is needed. The mysql / mariadb / - sqlite extensions only expose ATTACH-then-reference semantics — - there's no ``mysql_scan(dsn, …)`` function — so we ATTACH here - at extension-load time and ``_build_select_for_`` helpers - reference ``..
`` (or ``.
`` - for sqlite). + These extensions read through ATTACH-then-reference: the attach runs + here, at extension-load time (before the sandbox's lock, after which an + ATTACH is refused), and ``_build_select_for_`` references + ``..
`` (or ``.
`` for sqlite). + Postgres used ``postgres_scan(dsn, schema, table)`` until DuckDB 2.0 + dropped it; it is attached ``READ_ONLY``. - The alias is per-build (``mysql_`` / ``sqlite_``) - so two sources in the same contract / interpreter don't collide - on a hardcoded name. + The alias is per-build (``postgres_`` / ``mysql_`` + / ``sqlite_``) so two sources in the same contract / + interpreter don't collide on a hardcoded name. - Best-effort: if the connection DSN is malformed or the upstream - is unreachable, the ATTACH raises and the runner surfaces the - error in the per-stream try/except. + A connection block the DSN builder refuses, a ``secretRef`` that does + not resolve, or an upstream that cannot be reached raises a + :class:`_SourceAttachError` (the password masked), which fails the run + with that reason and a run record (``_execute``). A sqlite file is confined to :func:`_declared_roots` plus the operator's ``FLUID_DUCKDB_ALLOWED_DIRS`` first, and refused with - ``DuckDBSandboxError`` outside them: the sqlite scanner is not bounded - by the sandbox's allowlist, so nothing else would stop it. + ``DuckDBSandboxError`` outside them: nothing else bounds an ATTACH + made before the lock. """ kind = ctx.source.kind - if kind in ("mysql", "mariadb"): - # Resolve secretRef → password before build_mysql_dsn reads - # connection.get("password"). Inline literal values still win. - conn = resolve_connection_secrets(dict(ctx.source.connection.raw)) - # mysql_scanner's own DSN syntax, not libpq's: libpq quoting let a - # value with a space add socket= or ssl_key=, and a '?' turn the - # rest into URI options. build_mysql_dsn double-quotes each value - # and refuses a '?', a '/' in the host and a control character. - dsn = build_mysql_dsn(conn, database_key="database") - alias = _mysql_alias_for_build(ctx.build_id) - password = conn.get("password") - masked: Optional[str] = None + if kind in ("postgres", "mysql", "mariadb"): + import duckdb + + postgres = kind == "postgres" try: - # mysql extension supports both mysql:// and mariadb:// upstreams; - # the duckdb extension type literal is always ``mysql``. - con.execute(f"ATTACH {quote_ansi_string_literal(dsn)} AS {alias} (TYPE mysql)") + # Resolve secretRef → password before the DSN is built; an + # inline literal still wins. + conn = resolve_connection_secrets(dict(ctx.source.connection.raw)) except Exception as exc: - if not password: - raise - # mysql_scanner quotes the whole DSN in a connect error, and - # ``fluid apply`` prints the error: keep it, without the password. - masked = mask_dsn_password(str(exc), str(password)) - if masked is not None: - # Raised after the handler has ended, so the unmasked error is - # neither the cause nor the context of this one (PEP 415). - raise RuntimeError(masked) + # An unset variable or an unknown scheme (ValueError), or the + # secret backend's own error (ConfigurationError from vault, + # aws, gcp, azure or file): each names the reference, never + # the secret. + raise _SourceSetupError(str(exc)) from exc + try: + if postgres: + alias = _postgres_alias_for_build(ctx.build_id) + attach_sql = _postgres_attach_sql(conn, alias=alias) # validates first + else: + alias = _mysql_alias_for_build(ctx.build_id) + attach_sql = _mysql_attach_sql(conn, alias=alias) # validates first + con.execute(_empty_secret_sql("postgres" if postgres else "mysql", alias=alias)) + except ValueError as exc: + # A DSN value the builder refuses: the message names the field + # (and quotes a refused port), never a credential. + raise _SourceSetupError(str(exc)) from exc + except duckdb.Error as exc: # the extension did not load + raise _SourceAttachError(str(exc)) from exc + _attach_with_password_masked(con, attach_sql, conn.get("password")) elif kind == "sqlite": conn = dict(ctx.source.connection.raw) # SQLite uses a path, not a libpq-style DSN. The path comes @@ -808,21 +891,52 @@ def _attach_external_databases(self, con: Any, ctx: RunContext) -> None: path = conn.get("uri") or conn.get("path") or conn.get("database") or "" if not path: raise ValueError("sqlite source requires connection.uri, .path, or .database") - # The sqlite scanner opens the file through its own client library, - # which allowed_directories never bounds: this confinement is the - # only check on it. Without it a contract could land any SQLite - # file on the host (a browser's cookie store, another app's data). - # The checked realpath is what is attached. + # The attach runs before the sandbox's lock, when nothing bounds the + # open (and through DuckDB 1.5 the sqlite scanner's own client + # library is never bounded): this confinement is the only check on + # it. Without it a contract could land any SQLite file on the host + # (a browser's cookie store, another app's data). The checked + # realpath is what is attached. real = confine_declared(str(path), within=_declared_roots(ctx)) alias = _sqlite_alias_for_build(ctx.build_id) con.execute(f"ATTACH {quote_ansi_string_literal(real)} AS {alias} (TYPE sqlite)") -def _select_for_first_stream(ctx: RunContext) -> str: - return _select_for_stream(ctx, ctx.source.streams[0] if ctx.source.streams else "data") +class _SourceAttachError(RuntimeError): + """The source database could not be attached; the message holds no password.""" + + +class _SourceSetupError(_SourceAttachError, ValueError): + """The source's connection block was refused before any attach was tried.""" + + +def _attach_with_password_masked(con: Any, statement: str, password: Any) -> None: + """Run an ``ATTACH`` that carries a DSN; its connect error loses the password. + The postgres and mysql scanners quote the whole DSN they were given in a + connect error, and ``fluid apply`` prints the error: keep it, without the + password (:func:`mask_dsn_password`), as a :class:`_SourceAttachError` the + run reports as the build's failure. + """ + masked: Optional[str] = None + try: + con.execute(statement) + except Exception as exc: + if not password: + raise _SourceAttachError(str(exc)) from exc + masked = mask_dsn_password(str(exc), str(password)) + if masked is not None: + # Raised after the handler has ended, so the unmasked error is + # neither the cause nor the context of this one (PEP 415). + raise _SourceAttachError(masked) + + +def _select_for_first_stream(con: Any, ctx: RunContext) -> str: + return _select_for_stream(con, ctx, ctx.source.streams[0] if ctx.source.streams else "data") -def _select_for_stream(ctx: RunContext, stream: str) -> str: + +def _select_for_stream(con: Any, ctx: RunContext, stream: str) -> str: + """The ``SELECT`` of ``stream`` on ``con``, the run's connection with the source attached.""" kind = ctx.source.kind conn = dict(ctx.source.connection.raw) if kind == "filesystem": @@ -836,7 +950,10 @@ def _select_for_stream(ctx: RunContext, stream: str) -> str: opts = dict(ctx.source.reader.options) if ctx.source.reader else {} return _build_select_for_http(uri, fmt or "csv", opts) if kind == "postgres": - return _build_select_for_postgres(conn, stream) + select_sql = _build_select_for_postgres( + stream, alias=_postgres_alias_for_build(ctx.build_id) + ) + return _postgres_enums_as_varchar(con, select_sql) if kind in ("mysql", "mariadb"): # MariaDB uses the same duckdb mysql extension; the discoverer # already accepts both schemes (cli/discover/mysql.py), keep @@ -1253,9 +1370,9 @@ def _persist_cursor_after_run( quoted_path = quote_ansi_string_literal(out_path) con = _connect_for_destination(ctx) try: - row = con.execute( - f"SELECT MAX({cursor_field_safe}) FROM {reader}({quoted_path})" - ).fetchone() + row = first_row( + con.execute(f"SELECT MAX({cursor_field_safe}) FROM {reader}({quoted_path})") + ) finally: con.close() new_value = row[0] if row is not None else None @@ -1359,18 +1476,24 @@ def _execute(ctx: RunContext, runner: DuckdbRunner) -> RunResult: dlq_writer: Optional[DLQWriter] = None dlq_total_records = 0 - con = secure_duckdb_connect( - ":memory:", - allow=_run_allowlist(ctx, streams_to_run, sink_format), - before_lock=lambda c: runner._load_extensions(c, ctx), - ) + try: + con = secure_duckdb_connect( + ":memory:", + allow=_run_allowlist(ctx, streams_to_run, sink_format), + before_lock=lambda c: runner._load_extensions(c, ctx), + ) + except _SourceAttachError as exc: + # The source database is attached as the connection opens. Its connect + # error fails the run as a stream's error does, so the run record and + # the build's summary carry it (without the password). + return _refused_run(ctx, streams_to_run, started_at, t_start, f"source: {exc}") try: if masker is not None: masker.install(con) for stream in streams_to_run: t_stream = time.time() try: - base_sel = _select_for_stream(ctx, stream) + base_sel = _select_for_stream(con, ctx, stream) # Apply incremental cursor filter if the contract asked for it. # ``last_cursor_value`` is the value we filtered on (None for # full refresh OR first incremental run); kept for the @@ -1409,14 +1532,14 @@ def _execute(ctx: RunContext, runner: DuckdbRunner) -> RunResult: # select re-reads the source, which a live database may have # changed since, and a table holding exactly what the COPY # wrote then failed verify against the larger number. - written = con.execute(copy_sql).fetchone() + written = first_row(con.execute(copy_sql)) if written is not None and isinstance(written[0], int): n = written[0] else: counted_at_write = False # Best-effort via COUNT on the same select. try: - row = con.execute(f"SELECT COUNT(*) FROM ({landed_sel})").fetchone() + row = first_row(con.execute(f"SELECT COUNT(*) FROM ({landed_sel})")) n = row[0] if row is not None else 0 except Exception: n = 0 @@ -1489,8 +1612,8 @@ def _execute(ctx: RunContext, runner: DuckdbRunner) -> RunResult: on_error, ) except Exception as exc: # noqa: BLE001 - # The COPY SQL embeds DSN strings (postgres_scan dsn=…), - # which DuckDB echoes in its error messages. Don't pass + # DuckDB echoes the statement, and a source database's own + # error, in its error messages. Don't pass # `exc` directly and don't dump the full traceback via # exc_info — both can carry password/key data. LOG.error( @@ -1857,7 +1980,7 @@ def _artifact_exists(con: Any, path: str, reader: str) -> bool: if not _is_remote_uri(path): return Path(path).exists() try: - con.execute(f"SELECT 1 FROM {reader}({quote_ansi_string_literal(path)}) LIMIT 1").fetchone() + first_row(con.execute(f"SELECT 1 FROM {reader}({quote_ansi_string_literal(path)}) LIMIT 1")) return True except Exception: # noqa: BLE001 return False @@ -1944,9 +2067,9 @@ def _count_file_rows(path: str, sink_format: str) -> int: ] con = secure_duckdb_connect(":memory:", allow=DuckDBAllowlist.none().with_paths(path)) try: - row = con.execute( - f"SELECT COUNT(*) FROM {reader}({quote_ansi_string_literal(path)})" - ).fetchone() + row = first_row( + con.execute(f"SELECT COUNT(*) FROM {reader}({quote_ansi_string_literal(path)})") + ) finally: con.close() return int(row[0]) if row else 0 diff --git a/fluid_build/cli/_diff_live.py b/fluid_build/cli/_diff_live.py index eca88b6d..3b117b62 100644 --- a/fluid_build/cli/_diff_live.py +++ b/fluid_build/cli/_diff_live.py @@ -644,6 +644,7 @@ def _duckdb_table_exists(path: Path, schema_name: str, table: str) -> bool: checked by the provider's own allowlist first and are bound as parameters, never spliced into the SQL. """ + from fluid_build.providers._duckdb_compat import first_row from fluid_build.providers._duckdb_sandbox import DuckDBAllowlist, secure_duckdb_connect from fluid_build.providers.local_validation import _build_duckdb_table_ref @@ -652,11 +653,13 @@ def _duckdb_table_exists(path: Path, schema_name: str, table: str) -> bool: # access beyond the database file itself. con = secure_duckdb_connect(path, allow=DuckDBAllowlist.none(), read_only=True) try: - row = con.execute( - "SELECT count(*) FROM information_schema.tables " - "WHERE lower(table_schema) = lower(?) AND lower(table_name) = lower(?)", - [schema_name, table], - ).fetchone() + row = first_row( + con.execute( + "SELECT count(*) FROM information_schema.tables " + "WHERE lower(table_schema) = lower(?) AND lower(table_name) = lower(?)", + [schema_name, table], + ) + ) finally: con.close() return bool(row and row[0]) diff --git a/fluid_build/cli/_verify_masking.py b/fluid_build/cli/_verify_masking.py index 61985753..702ace4c 100644 --- a/fluid_build/cli/_verify_masking.py +++ b/fluid_build/cli/_verify_masking.py @@ -215,7 +215,9 @@ def local_dimension( ) values: List[int] = [] if selects: - row = con.execute(f"SELECT {', '.join(selects)} FROM ({relation_sql})").fetchone() + from fluid_build.providers._duckdb_compat import one_row + + row = one_row(con.execute(f"SELECT {', '.join(selects)} FROM ({relation_sql})")) values = [int(v or 0) for v in row] return finish(values) diff --git a/fluid_build/cli/discover/__init__.py b/fluid_build/cli/discover/__init__.py index ccd7e426..1f557c93 100644 --- a/fluid_build/cli/discover/__init__.py +++ b/fluid_build/cli/discover/__init__.py @@ -17,9 +17,9 @@ Introspects a live source and emits a deterministic Bronze contract for every discoverable stream. Supported source kinds: - - ``postgres://...`` — full schema/table introspection via DuckDB postgres_scan - - ``mysql://...`` — same via mysql_scan - - ``sqlite:///...`` — via sqlite_scan + - ``postgres://...`` — full schema/table introspection via DuckDB's postgres extension (ATTACH) + - ``mysql://...`` — same via the mysql extension + - ``sqlite:///...`` — via the sqlite extension - ``s3://`` / ``file://`` / ``https://`` — filesystem (CSV / Parquet / JSONL) The output is a list of ``contract..fluid.yaml`` files plus an optional diff --git a/fluid_build/cli/verify.py b/fluid_build/cli/verify.py index 8cec840e..6c9c2aeb 100644 --- a/fluid_build/cli/verify.py +++ b/fluid_build/cli/verify.py @@ -960,6 +960,7 @@ def _verify_local_file( # Introspect with DuckDB. try: + from fluid_build.providers._duckdb_compat import one_row from fluid_build.providers._duckdb_sandbox import ( DuckDBAllowlist, secure_duckdb_connect, @@ -968,25 +969,28 @@ def _verify_local_file( # The masking check below runs SQL built from the contract: it reads # the one file being verified and nothing else. con = secure_duckdb_connect(allow=DuckDBAllowlist.none().with_paths(file_path)) - # A SQL string literal, not Python ``repr``: the path now carries the - # contract's directory, and ``repr`` switches to double quotes (a - # DuckDB identifier) or backslash escapes (literal in DuckDB) the - # moment a directory name holds a quote. - literal = quote_ansi_string_literal(file_path.as_posix()) - if actual_fmt == "parquet": - relation_sql = f"SELECT * FROM read_parquet({literal})" - else: - relation_sql = f"SELECT * FROM read_csv_auto({literal})" - rel = con.sql(relation_sql) - - row_count = rel.aggregate("count(*)").fetchone()[0] - actual_columns = [col for col in rel.columns] - # Did the columns policy.privacy.masking names land treated? Counts only: - # a value that fails the check is cleartext and stays out of the report. - from fluid_build.cli._verify_masking import local_dimension - - masking = local_dimension(con, relation_sql, actual_columns, expose_config) - con.close() + try: + # A SQL string literal, not Python ``repr``: the path now carries the + # contract's directory, and ``repr`` switches to double quotes (a + # DuckDB identifier) or backslash escapes (literal in DuckDB) the + # moment a directory name holds a quote. + literal = quote_ansi_string_literal(file_path.as_posix()) + if actual_fmt == "parquet": + relation_sql = f"SELECT * FROM read_parquet({literal})" + else: + relation_sql = f"SELECT * FROM read_csv_auto({literal})" + rel = con.sql(relation_sql) + + row_count = one_row(rel.aggregate("count(*)"))[0] + actual_columns = [col for col in rel.columns] + # Did the columns policy.privacy.masking names land treated? Counts + # only: a value that fails the check is cleartext and stays out of + # the report. + from fluid_build.cli._verify_masking import local_dimension + + masking = local_dimension(con, relation_sql, actual_columns, expose_config) + finally: + con.close() except Exception as exc: return { "status": "error", diff --git a/fluid_build/contract_tests.py b/fluid_build/contract_tests.py index f36759c6..88f3cfcb 100644 --- a/fluid_build/contract_tests.py +++ b/fluid_build/contract_tests.py @@ -27,7 +27,8 @@ - The key is an alias; referenced as table/view name in SQL. - `path` can be a string (supports globs) or list of strings. - If `name` is set, it overrides the alias for the table/view name. -- sql: the SQL string to execute (must be a SELECT or CREATE TABLE AS ... style) +- sql: the SQL string to execute; it must end in a SELECT of the result + (for CREATE TABLE t AS SELECT ..., write just the SELECT ...) - outputs: dict with one or more outputs. Supported: { "path": "runtime/out/high_value_churn_risk.csv", @@ -53,6 +54,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, Union +from fluid_build.providers._duckdb_compat import NO_ROWS_TO_WRITE, one_row, query_relation from fluid_build.providers._duckdb_sandbox import ( DuckDBAllowlist, DuckDBSandboxError, @@ -208,10 +210,9 @@ def _execute_sql(con, sql: str): if not sql_stripped: raise LocalProviderError("No SQL provided for action") try: - # DuckDB returns a Relation for SELECT; for CREATE TABLE AS, we can - # run an additional SELECT if needed. We assume a SELECT for portability. - rel = con.sql(sql_stripped) - return rel + # A relation for a query; None when the last statement returns no rows + # (CREATE TABLE AS, INSERT), on DuckDB 1.5 and 2.x alike. + rel = query_relation(con, sql_stripped) except Exception as e: # Try to be helpful if the error hints at missing tables msg = str(e) @@ -221,6 +222,9 @@ def _execute_sql(con, sql: str): "the SQL references the correct table/view names. Error: " + msg ) from e raise LocalProviderError("SQL execution failed: " + msg) from e + if rel is None: + raise LocalProviderError(NO_ROWS_TO_WRITE) + return rel def _write_output(rel, outputs: Dict[str, Any], dry_run: bool) -> Tuple[int, Optional[str]]: @@ -243,7 +247,7 @@ def _write_output(rel, outputs: Dict[str, Any], dry_run: bool) -> Tuple[int, Opt if dry_run: # Estimate count without writing try: - cnt = rel.count("*").fetchone()[0] + cnt = one_row(rel.count("*"))[0] except Exception as e: LOGGER.warning(f"Failed to count rows in dry run: {e}") cnt = -1 @@ -266,7 +270,13 @@ def _write_output(rel, outputs: Dict[str, Any], dry_run: bool) -> Tuple[int, Opt # Create a temp view and then COPY from it. tmp_view = "__fluid_tmp_out" - rel.create_view(tmp_view, replace=True) + try: + rel.create_view(tmp_view, replace=True) + except Exception as e: + # DuckDB 2.0 refuses a view over rows it already holds (a PIVOT, + # nextval(), RETURNING); the COPY below then fails too, and its + # pandas fallback writes the file. + LOGGER.debug(f"Failed to create temp view {tmp_view}: {e}") if mode == "append": # Append parquet: create if not exists then append is not trivial. # DuckDB supports COPY with APPEND (v0.10+). Try; else fallback. @@ -309,7 +319,7 @@ def _write_output(rel, outputs: Dict[str, Any], dry_run: bool) -> Tuple[int, Opt # Count rows try: - cnt = rel.count("*").fetchone()[0] + cnt = one_row(rel.count("*"))[0] except Exception as e: LOGGER.warning(f"Failed to count output rows: {e}") cnt = -1 diff --git a/fluid_build/output_ports/mcp/drivers/duckdb.py b/fluid_build/output_ports/mcp/drivers/duckdb.py index 6b97c82d..8b767f4c 100644 --- a/fluid_build/output_ports/mcp/drivers/duckdb.py +++ b/fluid_build/output_ports/mcp/drivers/duckdb.py @@ -193,8 +193,13 @@ def execute( def health_check(self) -> Dict[str, Any]: started = time.monotonic() try: - connection = self._get_connection() - connection.execute("SELECT 1").fetchall() + # A cursor of its own, as execute() uses: on DuckDB 2.0 a call on + # the shared connection refuses while another thread is on it. + cursor = self._get_connection().cursor() + try: + cursor.execute("SELECT 1").fetchall() + finally: + cursor.close() except Exception as exc: # noqa: BLE001 return { "status": "unavailable", diff --git a/fluid_build/providers/_duckdb_compat.py b/fluid_build/providers/_duckdb_compat.py new file mode 100644 index 00000000..baadf647 --- /dev/null +++ b/fluid_build/providers/_duckdb_compat.py @@ -0,0 +1,219 @@ +# 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. + +"""Version-compat seam for the DuckDB Python client (1.5.x and 2.x). + +forge-cli takes ``duckdb>=1.5.0`` with no ceiling (the floor is the sandbox's, +:data:`fluid_build.providers._duckdb_sandbox.MIN_DUCKDB_VERSION`). DuckDB 2.0 +needs Python 3.11, so a Python 3.10 install stays on 1.5.x while a newer one +gets 2.x: the engine runs on both, and what differs lives here. + +These 2.0 changes reach the engine (duckdb-python v2.0.0 release notes, +"Upgrading from 1.x"; https://duckdb.org/docs/2.0/clients/python/migration.html): + +* A ``con.execute()`` result streams, and owns its connection until it is read + to the end. ``fetchone()`` on a one-row ``COUNT(*)`` leaves it open, and the + next relation read, ``register`` or ``create_function`` on that connection + raises ``ResourceInUseException``. Relations lost ``fetchone()`` and + ``fetchmany()`` outright. :func:`first_row` reads to the end on both. +* ``con.sql()`` of SQL whose last statement returns no rows (``CREATE TABLE + ... AS``, an ``INSERT`` without ``RETURNING``) returns a relation where 1.5 + returned ``None``: over the statement's row count, or, after an earlier + query in the same SQL, over that query's rows. Written out as a contract's + result, that is a one-row ``Count`` file or stale rows. It also runs the + statements ahead of one that does not parse, where 1.5 runs none, and a + relation of SQL that ends in a comment after its ``;`` fails to aggregate, + limit or write ("syntax error at or near ';'"). :func:`query_relation` + answers as 1.5 does on both. +* A relation whose rows 2.0 already holds (its SQL ran when the relation was + made: a ``PIVOT`` without an ``IN`` list, a ``SELECT`` calling + ``nextval()``, ``... RETURNING``) cannot back a view ("it reads rows held in + this process"). :func:`relation_view` puts those rows in a table instead. +* A database file 2.0 creates uses storage version 2.0, which no 1.5 process can + open ("can only read versions between 64 and 68"). :func:`storage_compat_config` + has 2.x write the format 1.5 itself writes, so a file forge-cli creates is the + same whichever DuckDB created it. + +Nothing here imports duckdb at module level: callers pass what they hold, and +:func:`query_relation` reads the installed version when it runs. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any, Dict, Iterator, Optional, Tuple + +#: Storage version 64, the one DuckDB 1.5 writes by default (its own setting +#: reads ``v0.10.2``, which writes the same version), and which DuckDB 1.0 and +#: later open. Database files forge-cli creates on DuckDB 2.x are written in +#: it, so a 1.5 install (Python 3.10, or a tool pinned below 2) opens them, and +#: a 2.x connection accepts the same SQL on them as 1.5 does (a VARIANT column, +#: which needs storage v1.5.0, is refused on both). +STORAGE_COMPATIBILITY_VERSION = "v1.0.0" + + +def first_row(result: Any) -> Optional[Tuple[Any, ...]]: + """The first row of ``result`` (a ``con.execute()`` result or a relation), or ``None``. + + Reads the whole result, so give it a query of one row (``COUNT(*)``, + ``LIMIT 1``). ``fetchone()`` does not do: on DuckDB 2.0 a result read only + partway keeps its connection busy, and a relation has no ``fetchone()``. + """ + rows = result.fetchall() + return rows[0] if rows else None + + +def one_row(result: Any) -> Tuple[Any, ...]: + """The row of a query that always returns exactly one (an aggregate, ``COUNT(*)``). + + :func:`first_row`, for the callers that index the row straight away: an + aggregate without ``GROUP BY`` returns one row even over no input, so an + empty result is a bug and raises ``ValueError`` rather than passing ``None``. + """ + row = first_row(result) + if row is None: + raise ValueError("an aggregate query returned no row") + return row + + +#: Statement types whose result is a query result, whatever it holds: DuckDB 1.5 +#: returns a relation for these even when it is a ``Success`` flag +#: (``CHECKPOINT`` is a ``CALL``). Not ``MULTI``, 2.0's type for a ``PIVOT`` +#: with columns it must look up first: a bare ``PIVOT`` returns rows, a +#: ``CREATE TABLE ... AS PIVOT`` does not, so its shape decides. +_QUERY_STATEMENTS = frozenset({"SELECT", "EXPLAIN", "CALL", "PRAGMA"}) +#: What 2.0 holds for a statement with no query result: the changed-row count +#: (``INSERT``, ``UPDATE``, ``DELETE``, ``MERGE``, ``COPY``), or a success flag. +_NO_RESULT_SHAPES = frozenset({(("Count",), ("BIGINT",)), (("Success",), ("BOOLEAN",))}) + + +#: What the engine says when SQL meant to produce a result ends in a statement +#: that returns no rows (``query_relation`` answered ``None``). +NO_ROWS_TO_WRITE = ( + "The SQL ends in a statement that returns no rows (CREATE, INSERT, UPDATE, ...), so " + "there is no result to write. End it with the SELECT of the result: for " + "CREATE TABLE t AS SELECT ..., write just the SELECT ..." +) + + +def query_relation(con: Any, sql: str) -> Any: + """``con.sql(sql)``, or ``None`` when the SQL's last statement returns no rows. + + What DuckDB 1.5 answers itself: a relation for a query (``SELECT``, + ``CALL``, ``EXPLAIN``, ``PRAGMA``, a ``RETURNING`` clause), ``None`` for a + statement that only changes things, so on 1.5 this is ``con.sql``. DuckDB + 2.0 returns a relation for both, and for SQL of several statements the + relation of the last one that returned rows, which need not be the last + statement. On 2.x every statement but the last therefore runs first, and + the last one alone decides: a query type keeps its relation, and any other + statement is judged by the shape 2.0 gives one without a query result (a + ``Count`` of changed rows, or a ``Success`` flag; every DDL, ``SET``, + ``ATTACH``, ``LOAD`` and transaction statement returns one of the two). + Each statement runs once, and SQL that does not parse raises before any of + it runs, as on 1.5. One case stays apart: a ``RETURNING`` clause whose only + column is a ``BIGINT`` named ``Count``, or a ``BOOLEAN`` named ``Success``, + reads as no rows on 2.x. + """ + import duckdb + + from ._duckdb_sandbox import duckdb_version # the sandbox imports this module + + if duckdb_version(duckdb) < (2, 0, 0): + return con.sql(sql) + # Parses the whole SQL first: a statement that does not parse raises here, + # before any runs (2.0's con.sql() would run the statements ahead of it). + statements = con.extract_statements(sql) + if not statements: # empty, or only comments + return con.sql(sql) + for statement in statements[:-1]: + con.execute(statement.query) + # To its end without reading its rows into Python (2.0's complete()): + # a SELECT ahead of the last statement is run for its effects only. + con.complete() + last = statements[-1] + rel = con.sql(_without_trailing_comments(last.query)) + if last.type.name in _QUERY_STATEMENTS: + return rel + shape = (tuple(rel.columns), tuple(str(t) for t in rel.types)) + return None if shape in _NO_RESULT_SHAPES else rel + + +def _without_trailing_comments(statement: str) -> str: + """``statement`` without the comments that end it. + + DuckDB 2.0 keeps a statement's text with the relation ``con.sql()`` makes + and reads it again inside the SQL of ``aggregate``, ``limit`` or + ``write_csv``, where a comment after the closing ``;`` no longer parses. + ``duckdb.tokenize`` tells a comment from a ``--`` inside a string, and + gives each token's offset in bytes. + """ + import duckdb + + cut = None + for offset, kind in reversed(duckdb.tokenize(statement)): + if kind != duckdb.token_type.comment: + break + cut = offset + if cut is None: + return statement + return statement.encode("utf-8")[:cut].decode("utf-8").rstrip() + + +@contextmanager +def relation_view(con: Any, rel: Any, name: str) -> Iterator[str]: + """``name`` reads ``rel`` while the block runs (a view), for SQL such as ``COPY``. + + DuckDB 2.0 refuses a view over a relation whose rows it already holds (see + the module docstring); those rows go to a table of that name instead. The + view or table is dropped when the block ends. ``name`` is an identifier the + caller owns. + """ + import duckdb + + try: + rel.create_view(name, replace=True) + kind = "VIEW" + except duckdb.InvalidInputException: + con.execute(f"DROP TABLE IF EXISTS {name}") + rel.to_table(name) + kind = "TABLE" + try: + yield name + finally: + con.execute(f"DROP {kind} IF EXISTS {name}") + + +def storage_compat_config(version: Tuple[int, int, int]) -> Dict[str, str]: + """``duckdb.connect`` config that keeps files DuckDB ``version`` creates readable by 1.5. + + Empty before 2.0, where the client's own default is the format 1.5 writes. + From 2.0, ``storage_compatibility_version`` set at connect (a ``SET`` once + the file is open is too late) makes each database file the connection + creates, ``ATTACH`` included, use that format. A file that exists keeps the + format it has. + """ + if version >= (2, 0, 0): + return {"storage_compatibility_version": STORAGE_COMPATIBILITY_VERSION} + return {} + + +__all__ = [ + "NO_ROWS_TO_WRITE", + "STORAGE_COMPATIBILITY_VERSION", + "first_row", + "one_row", + "query_relation", + "relation_view", + "storage_compat_config", +] diff --git a/fluid_build/providers/_duckdb_sandbox.py b/fluid_build/providers/_duckdb_sandbox.py index c23ac606..8bbe73ab 100644 --- a/fluid_build/providers/_duckdb_sandbox.py +++ b/fluid_build/providers/_duckdb_sandbox.py @@ -100,13 +100,17 @@ load (``sqlite_scan``, ``read_xlsx``, ``ST_Read``, ``delta_scan``, ``iceberg_scan``) is not in the catalog. :func:`is_sandbox_refusal` treats that error as a refusal. -* The ``sqlite``, ``postgres`` and ``mysql`` scanners open files and sockets - through their own client libraries, not through DuckDB's file system, so - neither ``allowed_directories`` nor ``enable_external_access`` bounds them: - once ``sqlite`` is loaded, ``sqlite_scan`` / ``ATTACH ... (TYPE sqlite)`` - open any SQLite file the process can read, and once ``postgres`` is loaded, - ``postgres_scan`` reaches any host the process can. Only the engine's own - SQL runs on a connection that loads them; contract SQL never does. +* A database scanner the call site loads widens what SQL on that connection + reaches, and how far depends on the DuckDB version (each case replayed on + 1.5.0, 1.5.6 and 2.0.0). Through 1.5, once ``postgres`` is loaded, + ``postgres_scan(dsn, ...)`` connects to any host the process can reach, and + once ``sqlite`` is loaded, ``sqlite_scan`` / ``ATTACH ... (TYPE sqlite)`` open + any SQLite file it can read, whatever the allowlist says. DuckDB 2.0 has no + ``postgres_scan``, and refuses a SQLite file outside the allowlist as it + refuses any other read. On both, an ``ATTACH`` of a postgres or mysql + database after the lock is refused, so ``postgres_query`` / ``mysql_query`` + reach only what the engine attached before it. Only the engine's own SQL runs + on a connection that loads a scanner; contract SQL never does. * DuckDB shares one database instance per file per process, and the lock is instance-wide: a second connection to a file database that is already open cannot be sandboxed, and is refused with :class:`DuckDBSandboxError`. Each @@ -145,6 +149,7 @@ from fluid_build._errors import FluidUserError, doc_url +from ._duckdb_compat import storage_compat_config from ._sql_safety import quote_ansi_string_literal, validate_ident if TYPE_CHECKING: # pragma: no cover - typing only @@ -451,9 +456,9 @@ def confine_declared( """The ``realpath`` of a declared local ``location``, refused outside ``within``. For a location the engine opens itself rather than grants to the SQL, such - as a SQLite source it ``ATTACH``es before the lock: the sqlite scanner opens - files through its own client library, so ``allowed_directories`` never - bounds it, and this check is the only one. Confined as + as a SQLite source it ``ATTACH``es before the lock: nothing bounds that open + (and through DuckDB 1.5 the sqlite scanner's own client library is never + bounded), so this check is the only one. Confined as :meth:`DuckDBAllowlist.with_declared` confines (``within`` plus ``FLUID_DUCKDB_ALLOWED_DIRS``); a relative ``location`` is resolved against ``base``, else the working directory. Open the returned path, the one that @@ -796,7 +801,8 @@ def duckdb_version(module: Any) -> Tuple[int, int, int]: return int(match.group(1)), int(match.group(2)), int(match.group(3)) -def _require_supported(module: Any) -> None: +def _require_supported(module: Any) -> Tuple[int, int, int]: + """``module``'s version, once it is one the sandbox can confine.""" version = duckdb_version(module) if version < MIN_DUCKDB_VERSION: want = ".".join(str(n) for n in MIN_DUCKDB_VERSION) @@ -810,6 +816,7 @@ def _require_supported(module: Any) -> None: ), fix=f"pip install 'duckdb>={want}'", ) + return version def _sql_list(values: Iterable[str]) -> str: @@ -843,7 +850,10 @@ def secure_duckdb_connect( in, for set-up that needs access the SQL must not have, such as attaching a declared source database or creating an object-store secret. ``config`` takes scalar start-up options (``threads``, ``TimeZone``), which cannot be - changed once the configuration is locked. + changed once the configuration is locked. On DuckDB 2.x a database file the + connection creates is written in the storage format DuckDB 1.5 writes + (:func:`~fluid_build.providers._duckdb_compat.storage_compat_config`), unless + ``config`` names ``storage_compatibility_version`` itself. Raises :class:`DuckDBSandboxError` for an older DuckDB, a bad allowlist entry, or a file ``database`` this process already has open (its one @@ -853,10 +863,13 @@ def secure_duckdb_connect( """ import duckdb - _require_supported(duckdb) + version = _require_supported(duckdb) target = os.fspath(database) + # A database file this creates stays readable by the oldest supported + # DuckDB (see _duckdb_compat); a caller's own setting wins. + options = {**storage_compat_config(version), **dict(config or {})} try: - con = duckdb.connect(target, read_only=read_only, config=dict(config or {})) + con = duckdb.connect(target, read_only=read_only, config=options) except duckdb.Error as exc: if "same database file with a different configuration" in str(exc): raise _already_open(target) from exc diff --git a/fluid_build/providers/_sql_safety.py b/fluid_build/providers/_sql_safety.py index 7c14fe8d..6722aac8 100644 --- a/fluid_build/providers/_sql_safety.py +++ b/fluid_build/providers/_sql_safety.py @@ -225,8 +225,8 @@ def libpq_escape(value: Any) -> str: Used by: - * The duckdb acquisition runner when building ``postgres_scan(...)`` - DSNs. + * The duckdb acquisition runner when building its postgres ``ATTACH`` + DSN. * The JDBC attach builder (``cli/discover/_jdbc_introspect.py``) for postgres. """ diff --git a/fluid_build/providers/local/local.py b/fluid_build/providers/local/local.py index 52e1bee5..0bc0dc98 100644 --- a/fluid_build/providers/local/local.py +++ b/fluid_build/providers/local/local.py @@ -39,6 +39,12 @@ from typing import Any, Dict, List, Optional, Tuple, Union from fluid_build.observability.secret_redactor import redact_secret_text, redact_value +from fluid_build.providers._duckdb_compat import ( + NO_ROWS_TO_WRITE, + one_row, + query_relation, + relation_view, +) from fluid_build.providers._duckdb_read import build_register_view_sql from fluid_build.providers._duckdb_sandbox import ( DuckDBAllowlist, @@ -845,9 +851,7 @@ def _register_with_retry(): with_retry(_register_with_retry, logger=self.logger, max_attempts=3) # Get row count for reporting - rowcount = con.execute(f"SELECT COUNT(*) FROM {validate_ident(table_name)}").fetchone()[ - 0 - ] + rowcount = one_row(con.execute(f"SELECT COUNT(*) FROM {validate_ident(table_name)}"))[0] self._log_info( "local_load_data_complete", @@ -936,7 +940,7 @@ def _run_sql_on( try: # Execute SQL with retry logic (handles transient DuckDB errors) def _execute_sql(): - return con.sql(sql) + return query_relation(con, sql) rel = with_retry(_execute_sql, logger=self.logger, max_attempts=3) @@ -957,6 +961,10 @@ def _execute_sql(): # error (an error -> traceback -> frame -> error cycle would keep # the connection alive until the cyclic GC runs). raise self._sandbox_refusal(e) from e + if rel is None: + # The same on DuckDB 1.5 and 2.x (see query_relation); 1.5 used to + # fail below on ``None.write_csv``. + raise ValueError(NO_ROWS_TO_WRITE) # If an output_table is specified, persist the result as a DuckDB table # so downstream materialize/copy steps can reference it. @@ -987,7 +995,7 @@ def _write_preview(): rowcount = None try: - rowcount = rel.aggregate("count(*)").fetchone()[0] + rowcount = one_row(rel.aggregate("count(*)"))[0] except Exception: pass @@ -1136,12 +1144,11 @@ def _copy_relation_to_uri(self, con: Any, rel: Any, uri: str, fmt: str) -> None: """ from fluid_build.build_runners.duckdb.runner import _build_copy_destination - view = "__fluid_embedded_result" - rel.create_view(view, replace=True) # ``replace`` on the first ``{select}``, not ``str.format``: the quoted # URI follows it in the template and may itself hold braces. template = _build_copy_destination(uri, fmt) - con.execute(template.replace("{select}", f"SELECT * FROM {view}", 1)) + with relation_view(con, rel, "__fluid_embedded_result") as view: + con.execute(template.replace("{select}", f"SELECT * FROM {view}", 1)) # ------------------ COPY / materialize (helper) --------------------- # @@ -1163,7 +1170,7 @@ def _run_copy_action(self, idx: int, action: Dict[str, Any]) -> Dict[str, Any]: try: rel = con.sql(f"SELECT * FROM {validate_ident(source_table)}") self._write_relation(rel, dst, fmt) - rowcount = rel.count("*").fetchone()[0] if hasattr(rel, "count") else -1 + rowcount = one_row(rel.count("*"))[0] if hasattr(rel, "count") else -1 del rel finally: # Closed on every path (see _run_load_data_action). diff --git a/fluid_build/providers/local/mocks.py b/fluid_build/providers/local/mocks.py index 614747d1..c2957d81 100644 --- a/fluid_build/providers/local/mocks.py +++ b/fluid_build/providers/local/mocks.py @@ -212,8 +212,8 @@ def connect(self) -> Any: self.db_path, allow=DuckDBAllowlist.none(), config={"TimeZone": "UTC"} ) - # Create standard Snowflake schemas if they don't exist - self._con.execute("CREATE SCHEMA IF NOT EXISTS INFORMATION_SCHEMA") + # Snowflake's PUBLIC schema. DuckDB has its own information_schema, + # and DuckDB 2.0 refuses to create one ("system catalog"). self._con.execute("CREATE SCHEMA IF NOT EXISTS PUBLIC") self.logger.info(f"Connected to Snowflake mock at {self.db_path}") diff --git a/fluid_build/providers/local_validation.py b/fluid_build/providers/local_validation.py index ce176596..39bc48b0 100644 --- a/fluid_build/providers/local_validation.py +++ b/fluid_build/providers/local_validation.py @@ -26,6 +26,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional +from fluid_build.providers._duckdb_compat import first_row from fluid_build.providers._duckdb_sandbox import DuckDBAllowlist, secure_duckdb_connect from fluid_build.providers._sql_safety import quote_ansi_string_literal, validate_ident from fluid_build.providers.quality_engine import ( @@ -132,7 +133,7 @@ def validate_connection(self) -> bool: """Local provider is always reachable.""" try: conn = self._connect(allow=DuckDBAllowlist.none()) - conn.execute("SELECT 1") + conn.execute("SELECT 1").fetchall() conn.close() return True except Exception: @@ -175,7 +176,7 @@ def get_resource_schema(self, resource_spec: Dict[str, Any]) -> Optional[Resourc fields: List[FieldSchema] = [] for row in rows: fields.append(FieldSchema(name=row[0], type=row[1], mode="NULLABLE")) - count_row = conn.execute(f"SELECT COUNT(*) FROM {table_ref}").fetchone() + count_row = first_row(conn.execute(f"SELECT COUNT(*) FROM {table_ref}")) row_count = count_row[0] if count_row else None size_bytes = path.stat().st_size conn.close() @@ -219,7 +220,7 @@ def get_resource_schema(self, resource_spec: Dict[str, Any]) -> Optional[Resourc count_sql = "SELECT COUNT(*) FROM {fn}({p})".format( fn=read_fn, p=quote_ansi_string_literal(abs_path) ) - count_row = conn.execute(count_sql).fetchone() + count_row = first_row(conn.execute(count_sql)) row_count = count_row[0] if count_row else None # File size diff --git a/scripts/verify_engines.sh b/scripts/verify_engines.sh index 0c0d9c65..f485293c 100755 --- a/scripts/verify_engines.sh +++ b/scripts/verify_engines.sh @@ -65,7 +65,7 @@ YAML ( cd "$WORK/duckdb" && python -m fluid_build.cli apply --build ingest contract.fluid.yaml >/repo/.verify-logs/duckdb.out 2>&1 ) state=$(run_state "$WORK/duckdb") [ -f "$WORK/duckdb/out/orders.parquet" ] \ - && rows=$(python -c "import duckdb;print(duckdb.sql(\"select count(*) from '$WORK/duckdb/out/orders.parquet'\").fetchone()[0])" 2>/dev/null) \ + && rows=$(python -c "import duckdb;print(duckdb.sql(\"select count(*) from '$WORK/duckdb/out/orders.parquet'\").fetchall()[0][0])" 2>/dev/null) \ && [ "$rows" = "5" ] \ && ok "5 rows landed in parquet (state=$state)" duckdb \ || miss "duckdb state=$state rows=${rows:-?} err=$(run_error "$WORK/duckdb")" duckdb diff --git a/tests/build_runners/test_consumes_resolution.py b/tests/build_runners/test_consumes_resolution.py index 91e9ffc9..7b6157a4 100644 --- a/tests/build_runners/test_consumes_resolution.py +++ b/tests/build_runners/test_consumes_resolution.py @@ -1131,7 +1131,7 @@ def test_the_high_value_churn_example_builds_as_it_did_before_consumes_resolved( ) assert run_builds_from_args(args, logging.getLogger("t")) == 0, printed out = copy / "output" / "high_value_churn.parquet" - assert duckdb.sql(f"SELECT COUNT(*) FROM '{out}'").fetchone()[0] > 0 + assert duckdb.sql(f"SELECT COUNT(*) FROM '{out}'").fetchall()[0][0] > 0 assert any( "consumes bronze.local.high_value_customers_v1/hv_customers" in line and "lineage only" in line diff --git a/tests/build_runners/test_duckdb_runner.py b/tests/build_runners/test_duckdb_runner.py index d54de613..17199b61 100644 --- a/tests/build_runners/test_duckdb_runner.py +++ b/tests/build_runners/test_duckdb_runner.py @@ -14,8 +14,9 @@ """DuckDB acquisition runner — unit + integration tests. -Filesystem-based tests run unconditionally; postgres tests require a live -Postgres (docker-compose / Testcontainers) and are gated by an env var. +Filesystem-based tests run unconditionally; the live postgres and mysql round +trips need a server and run only when their FLUID_TEST_* variables are set (no +CI job sets them). """ from __future__ import annotations @@ -209,16 +210,249 @@ def test_missing_uri_fails(self, tmp_path): assert rc != 0 +# ── postgres, no server ─────────────────────────────────────────────────── + + +_UNSET_SECRET_REF = "env://FLUID_TEST_UNSET_SOURCE_SECRET" # pragma: allowlist secret +_REFUSED_SECRET_REF = "file://a/b" # pragma: allowlist secret + +_CONNECT_FAILED = { + "postgres": "Unable to connect to Postgres", + "mysql": "Failed to connect to MySQL", +} + + +def _require_extension(kind: str) -> None: + duckdb = pytest.importorskip("duckdb") + try: + probe = duckdb.connect() + probe.execute(f"INSTALL {kind}; LOAD {kind}") + probe.close() + except duckdb.Error: + pytest.skip(f"the {kind} extension is not installable here (offline)") + + +def _database_source_contract(kind: str, connection: Dict[str, Any], out_path: Path) -> dict: + """One acquisition build reading ``app.orders`` from a ``kind`` database.""" + return { + "fluidVersion": "0.7.3", + "id": "bronze.db_orders", + "builds": [ + { + "id": "ingest_db", + "pattern": "acquisition", + "engine": "duckdb", + "properties": { + "source": { + "kind": kind, + "connection": connection, + "mode": "full_refresh", + "streams": ["app.orders"], + }, + "sink": {"format": "parquet"}, + }, + "outputs": ["orders"], + } + ], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "local", + "format": "parquet", + "location": {"path": str(out_path)}, + }, + "contract": {"schema": []}, + } + ], + } + + +@pytest.mark.parametrize("with_password", [True, False], ids=["password", "no-password"]) +@pytest.mark.parametrize("kind", ["postgres", "mysql"]) +def test_an_unreachable_database_source_fails_the_build_not_the_process( + tmp_path, caplog, kind, with_password +): + """The source is attached as the run's connection opens (DuckDB 2.0 has no + ``postgres_scan``), so its connect error is reported as the build's failure, + with the password masked. The error is a connection failure: not a function + DuckDB 2.0 lacks, nor the "Unknown secret storage found: 'local_file'" that + an ATTACH without a named secret hits under the sandbox.""" + _require_extension(kind) + password = "runner-unreachable-db-pw-6T" # pragma: allowlist secret + out_path = tmp_path / "orders.parquet" + connection = { + "host": "127.0.0.1", + "port": 9, # discard: nothing listens + "user": "fluid", + **({"password": password} if with_password else {}), + "database": "app", + } + contract = _database_source_contract(kind, connection, out_path) + caplog.set_level("DEBUG") + rc = execute_duckdb_build(contract["builds"][0], contract, tmp_path, dry_run=False) + assert rc == 1 + assert not out_path.exists() + assert _CONNECT_FAILED[kind] in caplog.text, caplog.text[-2000:] + assert password not in caplog.text + for written in tmp_path.rglob("*"): + if written.is_file(): + assert password not in written.read_text(encoding="utf-8", errors="replace"), written + + +@pytest.mark.parametrize( + "refused, reason", + [ + ({"port": "abc"}, "connection.port must be numeric"), + # The run record's redactor reads "_SECRET: environment" as an + # assignment and masks its value, so the record says "variable not set". + ({"secretRef": _UNSET_SECRET_REF}, "variable not set"), + # The file backend refuses a name with a separator: a ConfigurationError, + # as every secret backend but env:// raises, not a ValueError. + ({"secretRef": _REFUSED_SECRET_REF}, "Invalid secret name"), + ], + ids=["port", "unset-secret-ref", "secret-backend-error"], +) +@pytest.mark.parametrize("kind", ["postgres", "mysql"]) +def test_a_source_the_runner_cannot_set_up_fails_the_build_with_a_run_record( + tmp_path, monkeypatch, kind, refused, reason +): + """A connection block refused before any attach (a port that is not a + number, a secretRef that does not resolve) fails the build as a connect + error does: exit 1 and a failed run record, not an exception out of the + runner that ends `fluid apply` before its other builds run.""" + import json + + _require_extension(kind) + monkeypatch.delenv("FLUID_TEST_UNSET_SOURCE_SECRET", raising=False) + out_path = tmp_path / "orders.parquet" + connection = {"host": "127.0.0.1", "port": 9, "user": "fluid", "database": "app", **refused} + contract = _database_source_contract(kind, connection, out_path) + rc = execute_duckdb_build(contract["builds"][0], contract, tmp_path, dry_run=False) + assert rc == 1 + assert not out_path.exists() + runs = tmp_path / ".fluid" / "runs" / "bronze.db_orders" / "ingest_db" / "runs" + (record,) = [json.loads(p.read_text(encoding="utf-8")) for p in runs.glob("*.json")] + assert record["state"] == "failed" + (stream,) = record["streams"] + assert stream["state"] == "failed" + assert stream["error"].startswith("source: ") and reason in stream["error"], stream + + +def test_a_postgres_enum_column_reads_as_text_as_postgres_scan_read_it(): + """``postgres_scan`` typed a Postgres enum ``VARCHAR``; the ``ATTACH``-ed + catalog types it ``ENUM(...)``, which the schema gate took for a type + change against a declared ``VARCHAR``. DuckDB's own enum stands in here + for the attached one: both describe as ``ENUM('a', 'b')``.""" + import types + + duckdb = pytest.importorskip("duckdb") + from fluid_build.build_runners.duckdb.runner import ( + _postgres_alias_for_build, + _postgres_enums_as_varchar, + _select_for_stream, + ) + + con = duckdb.connect() + # Under the alias the runner attaches the source as. + alias = _postgres_alias_for_build("b1") + con.execute(f"ATTACH ':memory:' AS {alias}") + con.execute(f"CREATE SCHEMA {alias}.app") + con.execute( + f"CREATE TABLE {alias}.app.src (id INTEGER, mood ENUM('sad', 'ok', 'happy'), " + "moods ENUM('sad', 'ok', 'happy')[], \"Odd \"\"Name\" ENUM('a)b', 'c'))" + ) + con.execute(f"INSERT INTO {alias}.app.src VALUES (1, 'ok', ['sad', 'happy'], 'a)b')") + ctx = types.SimpleNamespace( + build_id="b1", + source=types.SimpleNamespace(kind="postgres", connection=types.SimpleNamespace(raw={})), + ) + select_sql = _select_for_stream(con, ctx, "app.src") + assert [(row[0], row[1]) for row in con.execute(f"DESCRIBE {select_sql}").fetchall()] == [ + ("id", "INTEGER"), + ("mood", "VARCHAR"), + ("moods", "VARCHAR[]"), + ('Odd "Name', "VARCHAR"), + ] + assert con.execute(select_sql).fetchall() == [(1, "ok", ["sad", "happy"], "a)b")] + # Text order, not declaration order: 'ok' < 'sad' as text, not as this enum. + assert con.execute(f"SELECT mood < 'sad' FROM ({select_sql})").fetchall() == [(True,)] + no_enum = f"SELECT id FROM {alias}.app.src" + assert _postgres_enums_as_varchar(con, no_enum) == no_enum + + +def test_a_database_source_that_drifts_from_its_declared_schema_is_refused(tmp_path): + """The schema gate fingerprints the source on a connection of its own, + which must attach the source as the run's does: a database source exists + only through that ``ATTACH``. The gate swallows a fingerprint that fails, + so a fingerprint that cannot see the source would land drifted data.""" + import sqlite3 + + from fluid_build._errors import SchemaDriftError + + _require_extension("sqlite") + db = tmp_path / "src.sqlite" + source = sqlite3.connect(db) + source.execute("CREATE TABLE orders (id INTEGER, name TEXT)") + source.execute("INSERT INTO orders VALUES (1, 'a'), (2, 'b')") + source.commit() + source.close() + out = tmp_path / "orders.parquet" + contract = { + "fluidVersion": "0.7.3", + "id": "bronze.sqlite_orders", + "builds": [ + { + "id": "ingest_sqlite", + "pattern": "acquisition", + "engine": "duckdb", + "properties": { + "source": { + "kind": "sqlite", + "connection": {"path": str(db)}, + "mode": "full_refresh", + "streams": ["orders"], + }, + "sink": {"format": "parquet"}, + }, + "outputs": ["orders"], + } + ], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "local", + "format": "parquet", + "location": {"path": str(out)}, + }, + "contract": { + "schemaPolicy": "strict", + "schema": [ + {"name": "id", "type": "integer"}, + {"name": "not_in_the_source", "type": "string"}, + ], + }, + } + ], + } + with pytest.raises(SchemaDriftError): + execute_duckdb_build(contract["builds"][0], contract, tmp_path, dry_run=False) + assert not out.exists() + + # ── postgres (docker-required) ──────────────────────────────────────────── @pytest.mark.integration @pytest.mark.skipif( os.environ.get("FLUID_TEST_POSTGRES_HOST") is None, - reason="Set FLUID_TEST_POSTGRES_HOST/PORT/USER/PASSWORD/DB to run; CI uses docker-compose.", + reason="Set FLUID_TEST_POSTGRES_HOST/PORT/USER/PASSWORD/DB to run (no CI job sets them).", ) class TestDuckdbPostgresLive: - def test_postgres_scan_round_trip(self, tmp_path): + def test_postgres_round_trip(self, tmp_path): host = os.environ["FLUID_TEST_POSTGRES_HOST"] port = int(os.environ.get("FLUID_TEST_POSTGRES_PORT", "5432")) user = os.environ.get("FLUID_TEST_POSTGRES_USER", "postgres") @@ -309,7 +543,7 @@ def test_postgres_scan_round_trip(self, tmp_path): @pytest.mark.integration @pytest.mark.skipif( os.environ.get("FLUID_TEST_MYSQL_HOST") is None, - reason="Set FLUID_TEST_MYSQL_HOST/PORT/USER/PASSWORD/DB to run; CI uses docker-compose.", + reason="Set FLUID_TEST_MYSQL_HOST/PORT/USER/PASSWORD/DB to run (no CI job sets them).", ) class TestDuckdbMysqlLive: def test_mysql_scan_round_trip(self, tmp_path): diff --git a/tests/build_runners/test_duckdb_runner_mysql_attach.py b/tests/build_runners/test_duckdb_runner_mysql_attach.py index cf679f3c..fe9b31c4 100644 --- a/tests/build_runners/test_duckdb_runner_mysql_attach.py +++ b/tests/build_runners/test_duckdb_runner_mysql_attach.py @@ -50,8 +50,8 @@ def _attach_on(con: Any, raw: Dict[str, Any], kind: str = "mysql") -> None: runner._attach_external_databases(con, ctx) # type: ignore[arg-type] -def _attach(raw: Dict[str, Any], kind: str = "mysql") -> List[str]: - """Run the runner's mysql attach on a recording connection; return the SQL run.""" +def _executed(raw: Dict[str, Any], kind: str = "mysql") -> List[str]: + """Run the runner's attach on a recording connection; return every statement run.""" seen: List[str] = [] class _Con: @@ -62,6 +62,11 @@ def execute(self, sql: str, *_a: Any, **_k: Any) -> None: return seen +def _attach(raw: Dict[str, Any], kind: str = "mysql") -> List[str]: + """The ``ATTACH`` statements :func:`_executed` saw (the empty secret left out).""" + return [sql for sql in _executed(raw, kind) if not sql.startswith("CREATE TEMPORARY SECRET")] + + def _dsn_of(attach_sql: str) -> str: """The DSN inside ``ATTACH '' AS ...`` with the SQL literal unquoted.""" assert attach_sql.startswith("ATTACH '"), attach_sql @@ -124,7 +129,15 @@ def test_one_keyword_per_field(self, field, value): def test_a_plain_connection(self): (sql,) = _attach(_BASE) assert dict(_parse_mysql_dsn(_dsn_of(sql))) == _BASE - assert sql.endswith(" AS mysql_b1 (TYPE mysql)") + assert sql.endswith(" AS mysql_b1 (TYPE mysql, SECRET mysql_b1, READ_ONLY)") + + def test_an_empty_temporary_secret_is_named_in_the_attach(self): + """Without it the extension looks for a default secret in the persistent + storage the sandbox turned off: ``Unknown secret storage found: + 'local_file'``, and no mysql source attached at all.""" + store_sql, sql = _executed(_BASE) + assert store_sql == "CREATE TEMPORARY SECRET mysql_b1 (TYPE mysql)" + assert sql.startswith("ATTACH '") class TestRefusedBeforeTheAttach: @@ -187,22 +200,92 @@ def test_other_non_ascii_text_is_accepted(self, value): assert dict(_parse_mysql_dsn(_dsn_of(sql)))["password"] == value -class TestPostgresScanDsnRefusesControlCharacters: - """The runner's postgres source is read through ``postgres_scan(dsn, ...)``, - whose DSN the same shared builder makes. libpq ends a bare value at any C - ``isspace()`` character, and libpq_escape quotes only a value with a space - or a quote, so a tab or newline there started a second keyword.""" +class TestPostgresAttachDsnRefusesControlCharacters: + """The runner's postgres source is attached with ``ATTACH '' ... (TYPE + postgres)``, whose DSN the same shared builder makes. libpq ends a bare + value at any C ``isspace()`` character, and libpq_escape quotes only a value + with a space or a quote, so a tab or newline there started a second keyword.""" @pytest.mark.parametrize("field", ["host", "user", "password", "database"]) @pytest.mark.parametrize("ch", _CONTROLS, ids=lambda c: f"U+{ord(c):04X}") def test_refused(self, field, ch): - from fluid_build.build_runners.duckdb.runner import _build_select_for_postgres - conn = {"host": "db.test", "port": "5432", "user": "u", "password": FAKE_PW} conn["database"] = "app" conn[field] = f"ab{ch}sslmode=disable" with pytest.raises(ValueError, match=f"{field}.*control characters"): - _build_select_for_postgres(conn, "public.orders") + _attach(conn, kind="postgres") + + +class TestPostgresAttach: + """DuckDB 2.0's postgres extension has no ``postgres_scan``, so the runner + attaches a postgres source before the sandbox's lock, as it attaches mysql + and sqlite sources, and attaches it read-only.""" + + def test_the_source_is_attached_read_only_under_a_per_build_alias(self): + store_sql, sql = _executed(dict(_BASE, port="5432"), kind="postgres") + # An empty temporary secret: the sandbox's disabled persistent storage + # otherwise fails the ATTACH's default-secret lookup. + assert store_sql == "CREATE TEMPORARY SECRET postgres_b1 (TYPE postgres)" + assert sql.startswith("ATTACH '") + assert sql.endswith(" AS postgres_b1 (TYPE postgres, SECRET postgres_b1, READ_ONLY)") + assert "dbname=app" in _dsn_of(sql) + + def test_the_select_reads_the_attached_alias(self): + from fluid_build.build_runners.duckdb.runner import _build_select_for_postgres + + assert ( + _build_select_for_postgres("sales.orders", alias="postgres_b1") + == "SELECT * FROM postgres_b1.sales.orders" + ) + assert ( + _build_select_for_postgres("orders", alias="postgres_b1") + == "SELECT * FROM postgres_b1.public.orders" + ) + + def test_a_secret_ref_password_is_resolved(self, monkeypatch): + monkeypatch.setenv("RUNNER_FAKE_PG_SECRET", "runner-pg-ref-pw-3J") + raw = {k: v for k, v in _BASE.items() if k != "password"} + raw["secretRef"] = "env://RUNNER_FAKE_PG_SECRET" # pragma: allowlist secret + (sql,) = _attach(raw, kind="postgres") + assert "password=runner-pg-ref-pw-3J" in _dsn_of(sql) + + @pytest.mark.parametrize("kind", ["postgres", "mysql"]) + def test_a_secret_ref_that_does_not_resolve_fails_the_source_setup(self, kind, monkeypatch): + from fluid_build.build_runners.duckdb.runner import _SourceAttachError + + monkeypatch.delenv("RUNNER_FAKE_UNSET_SECRET", raising=False) + raw = {k: v for k, v in _BASE.items() if k != "password"} + raw["secretRef"] = "env://RUNNER_FAKE_UNSET_SECRET" # pragma: allowlist secret + with pytest.raises(_SourceAttachError, match="RUNNER_FAKE_UNSET_SECRET") as caught: + _executed(raw, kind=kind) + # Still a ValueError for callers that catch the builder's refusal. + assert isinstance(caught.value, ValueError) + + @pytest.mark.parametrize("kind", ["postgres", "mysql"]) + def test_a_secret_backend_error_fails_the_source_setup(self, kind): + """Every backend but env:// raises ConfigurationError, not ValueError; + the file backend refuses a name with a separator before reading.""" + from fluid_build.build_runners.duckdb.runner import _SourceSetupError + + raw = {k: v for k, v in _BASE.items() if k != "password"} + raw["secretRef"] = "file://a/b" # pragma: allowlist secret + with pytest.raises(_SourceSetupError, match="Invalid secret name"): + _executed(raw, kind=kind) + + def test_a_connect_error_has_the_password_masked(self): + class _Con: + def execute(self, sql: str, *_a: Any, **_k: Any) -> None: + if sql.startswith("ATTACH"): + # What postgres_scanner says: it quotes the DSN it was given. + raise RuntimeError( + f'IO Error: Unable to connect to Postgres at "{_dsn_of(sql)}"' + ) + + with pytest.raises(RuntimeError) as caught: + _attach_on(_Con(), dict(_BASE, port="5432"), kind="postgres") + assert FAKE_PW not in str(caught.value) + assert "password=***" in str(caught.value) + assert caught.value.__context__ is None and caught.value.__cause__ is None class TestConnectErrorHasThePasswordMasked: @@ -215,6 +298,8 @@ class TestConnectErrorHasThePasswordMasked: def _failing_con() -> Any: class _Con: def execute(self, sql: str, *_a: Any, **_k: Any) -> None: + if not sql.startswith("ATTACH"): + return # What mysql_scanner 1.5.6 says, quoting the DSN and the SQL. raise RuntimeError( "IO Error: Failed to connect to MySQL database with parameters " diff --git a/tests/cli/discover/test_jdbc_attach_safety.py b/tests/cli/discover/test_jdbc_attach_safety.py index f799b418..8d150c7a 100644 --- a/tests/cli/discover/test_jdbc_attach_safety.py +++ b/tests/cli/discover/test_jdbc_attach_safety.py @@ -137,7 +137,7 @@ def test_other_non_ascii_text_is_accepted(self, value, postgres): class TestSharedBuildersRejectControlChars: """The check lives in the shared DSN builders, so every caller gets it: the - JDBC attach, and the duckdb build runner's mysql ATTACH and postgres_scan DSN, + JDBC attach, and the duckdb build runner's mysql and postgres ATTACH DSNs, which call the builders directly.""" @pytest.mark.parametrize("ch", ["\x00", "\t", "\n", "\x1b", "\x7f", "\x85"], ids=repr) @@ -182,7 +182,7 @@ def test_build_attach_dsn_refuses_a_non_ascii_digit_port(self, name): @pytest.mark.parametrize("name", sorted(_PORTS)) def test_build_libpq_dsn_refuses_a_non_ascii_digit_port(self, name): - # The shared builder: the duckdb runner's postgres_scan DSN uses it too. + # The shared builder: the duckdb runner's postgres ATTACH DSN uses it too. from fluid_build.providers._sql_safety import build_libpq_dsn with pytest.raises(ValueError, match="port"): diff --git a/tests/forge/core/pipeline_systems/test_plan_apply_mode_agreement.py b/tests/forge/core/pipeline_systems/test_plan_apply_mode_agreement.py index 52d808d9..df306213 100644 --- a/tests/forge/core/pipeline_systems/test_plan_apply_mode_agreement.py +++ b/tests/forge/core/pipeline_systems/test_plan_apply_mode_agreement.py @@ -228,7 +228,7 @@ def test_generated_plan_and_apply_agree_on_the_mode( assert _flag(apply_argv, "--mode") == apply_mode landed = ws / "contracts" / "p" / "out" / "rows.parquet" if apply_mode in _BUILD_MODES: - assert duckdb.sql(f"SELECT count(*) FROM '{landed}'").fetchone() == (1,) + assert duckdb.sql(f"SELECT count(*) FROM '{landed}'").fetchall() == [(1,)] # A build id reaches ``fluid apply`` only with a build mode. expected_build_id = build_id if (build_id and apply_mode in _BUILD_MODES) else None assert _flag(apply_argv, "--build-id") == expected_build_id @@ -257,7 +257,7 @@ def test_a_parameterless_run_uses_the_declared_defaults( assert _flag(apply_argv, "--build-id") == expected_build_id landed = ws / "contracts" / "p" / "out" / "rows.parquet" if expected_mode in _BUILD_MODES: - assert duckdb.sql(f"SELECT count(*) FROM '{landed}'").fetchone() == (1,) + assert duckdb.sql(f"SELECT count(*) FROM '{landed}'").fetchall() == [(1,)] if expected_mode == "dry-run": assert not landed.exists() # a dry run writes nothing @@ -301,7 +301,7 @@ def test_a_run_on_defaults_applies_the_fluid_env_default( apply_argv = _run_chain(ws, stages, exported, default_env="staging") assert _flag(apply_argv, "--env") == "staging" out = ws / "contracts" / "p" / "out" - assert duckdb.sql(f"SELECT count(*) FROM '{out / 'staging' / 'rows.parquet'}'").fetchone() == ( - 1, - ) + assert duckdb.sql(f"SELECT count(*) FROM '{out / 'staging' / 'rows.parquet'}'").fetchall() == [ + (1,) + ] assert not (out / "rows.parquet").exists() # the base binding was not applied diff --git a/tests/output_ports/test_drivers_duckdb.py b/tests/output_ports/test_drivers_duckdb.py index 5f329c49..9aa6a0aa 100644 --- a/tests/output_ports/test_drivers_duckdb.py +++ b/tests/output_ports/test_drivers_duckdb.py @@ -82,6 +82,34 @@ def test_health_check_succeeds(tmp_path): assert "latency_ms" in health +def test_concurrent_calls_on_one_driver_all_succeed(tmp_path): + """DuckDB 2.0 refuses a second thread on one connection where 1.5 waited. + health_check and execute each take a cursor of their own, so concurrent + calls on one driver succeed on both versions.""" + import threading + + driver, _ = _build_duckdb_driver(tmp_path) + assert driver.health_check()["status"] == "ok" # the connection now exists + errors = [] + + def work(): + try: + for _ in range(20): + health = driver.health_check() + if health["status"] != "ok": + errors.append(health["detail"]) + driver.sample(limit=1) + except Exception as exc: # noqa: BLE001 - collected for the assert + errors.append(repr(exc)) + + threads = [threading.Thread(target=work) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert errors == [] + + def test_sample_returns_three_rows(tmp_path): driver, _ = _build_duckdb_driver(tmp_path) result = driver.sample(limit=10) diff --git a/tests/providers/test_duckdb_compat.py b/tests/providers/test_duckdb_compat.py new file mode 100644 index 00000000..a2cd6417 --- /dev/null +++ b/tests/providers/test_duckdb_compat.py @@ -0,0 +1,725 @@ +# 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. + +"""One engine on DuckDB 1.5.x and 2.x: ``providers/_duckdb_compat.py``. + +Each test runs on whichever DuckDB is installed and states what must hold on +both. CI runs the suite on 1.5.x (Python 3.10, which DuckDB 2.0 does not +support) and on 2.x (every newer Python). What 2.0 changed is in the module's +docstring; the cases here are the ones that broke forge-cli's own code. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path +from typing import Any, Dict, List + +import pytest + +duckdb = pytest.importorskip("duckdb") + +from fluid_build.providers._duckdb_compat import ( # noqa: E402 + STORAGE_COMPATIBILITY_VERSION, + first_row, + one_row, + query_relation, + relation_view, + storage_compat_config, +) +from fluid_build.providers._duckdb_sandbox import ( # noqa: E402 + DuckDBAllowlist, + duckdb_version, + secure_duckdb_connect, +) + +REPO_ROOT = Path(__file__).resolve().parents[2] +PACKAGE = REPO_ROOT / "fluid_build" +DUCKDB_2 = duckdb_version(duckdb) >= (2, 0, 0) + + +# ── first_row ──────────────────────────────────────────────────────────── + + +def test_first_row_of_an_execute_result(): + assert first_row(duckdb.connect().execute("SELECT 41 + 1, 'x'")) == (42, "x") + + +def test_first_row_of_a_relation(): + """DuckDB 2.0 removed ``fetchone()`` from relations.""" + con = duckdb.connect() + rel = con.sql("SELECT range AS a FROM range(3)") + assert first_row(rel.aggregate("count(*)")) == (3,) + assert first_row(rel.count("*")) == (3,) + + +def test_first_row_is_the_first_of_several_rows(): + con = duckdb.connect() + assert first_row(con.execute("SELECT range FROM range(3) ORDER BY 1")) == (0,) + assert first_row(con.sql("SELECT range FROM range(3) ORDER BY 1")) == (0,) + + +def test_first_row_of_no_rows_is_none(): + con = duckdb.connect() + assert first_row(con.execute("SELECT 1 WHERE false")) is None + assert first_row(con.sql("SELECT 1 WHERE false")) is None + + +def test_one_row_is_the_row_of_an_aggregate(): + con = duckdb.connect() + assert one_row(con.execute("SELECT count(*) FROM range(0)")) == (0,) + assert one_row(con.sql("SELECT max(range) FROM range(4)")) == (3,) + + +def test_one_row_refuses_an_empty_result(): + """An aggregate cannot return no row; a caller indexing [0] must not get None.""" + with pytest.raises(ValueError, match="no row"): + one_row(duckdb.connect().execute("SELECT 1 WHERE false")) + + +def test_the_connection_is_free_after_first_row(): + """On DuckDB 2.0 a ``con.execute()`` result read only partway (``fetchone()`` + on a one-row ``COUNT(*)``) keeps the connection busy, and the next relation + read, ``register`` or ``create_function`` on it raises + ``ResourceInUseException``. ``first_row`` reads to the end.""" + pd = pytest.importorskip("pandas") + pytest.importorskip("numpy") # DuckDB's create_function needs it + con = duckdb.connect() + con.execute("CREATE TABLE t AS SELECT range AS a FROM range(5)") + + assert first_row(con.execute("SELECT count(*) FROM t")) == (5,) + assert con.sql("SELECT sum(a) FROM t").fetchall() == [(10,)] + assert first_row(con.execute("SELECT a FROM t ORDER BY a LIMIT 1")) == (0,) + con.register("frame", pd.DataFrame({"b": [1, 2]})) + assert first_row(con.execute("SELECT max(a) FROM t")) == (4,) + con.create_function("twice", lambda x: x * 2, ["BIGINT"], "BIGINT") + assert con.sql("SELECT twice(b) FROM frame ORDER BY b").fetchall() == [(2,), (4,)] + + +@pytest.mark.skipif(not DUCKDB_2, reason="the hazard first_row avoids is DuckDB 2.0's") +def test_on_duckdb_2_a_partial_read_keeps_the_connection_busy(): + """What first_row exists for, pinned: if a later DuckDB relaxes this, the + module docstring and this test go together.""" + con = duckdb.connect() + con.execute("CREATE TABLE t AS SELECT range AS a FROM range(5)") + assert con.execute("SELECT count(*) FROM t").fetchone() == (5,) + with pytest.raises(duckdb.Error, match="(?i)resource in use|busy|open result"): + con.sql("SELECT 1").fetchall() + + +# ── query_relation ─────────────────────────────────────────────────────── + + +def _table_t() -> Any: + con = duckdb.connect() + con.execute("CREATE TABLE t AS SELECT range AS a FROM range(3)") + return con + + +@pytest.mark.parametrize( + "sql, columns", + [ + ("SELECT 1 AS a", ["a"]), + ("WITH x AS (SELECT 1 AS a) SELECT * FROM x", ["a"]), + ("FROM t", ["a"]), + ("CREATE TABLE made AS SELECT 1 AS q; SELECT * FROM made", ["q"]), + ("INSERT INTO t VALUES (7) RETURNING a", ["a"]), + ("UPDATE t SET a = a RETURNING a", ["a"]), + ('SELECT count(*) AS "Count" FROM t', ["Count"]), + ("PIVOT t ON a USING count(*)", ["0", "1", "2"]), + ("DESCRIBE t", None), + ("CALL pragma_version()", None), + ("EXPLAIN SELECT 1", None), + # A CALL whose result is a Success flag: 1.5 returns it as a relation. + ("CHECKPOINT", ["Success"]), + ], +) +def test_query_relation_keeps_a_query(sql, columns): + rel = query_relation(_table_t(), sql) + assert rel is not None + if columns is not None: + assert rel.columns == columns + + +@pytest.mark.parametrize( + "sql", + [ + "CREATE TABLE made AS SELECT 1 AS q", + "CREATE OR REPLACE VIEW v AS SELECT 1 AS q", + "INSERT INTO t VALUES (7)", + "UPDATE t SET a = a", + "DELETE FROM t WHERE a = 100", + "MERGE INTO t USING (SELECT 1 AS a) s ON t.a = s.a " + + "WHEN NOT MATCHED THEN INSERT VALUES (s.a)", + "DROP TABLE t", + "SET threads = 2", + "SELECT 1 AS x; CREATE TABLE made AS SELECT 1 AS q", + # DuckDB 2.0 types a PIVOT whose columns it looks up as MULTI, a query + # or not: as DDL it returns a Success flag. + "CREATE TABLE p AS PIVOT t ON a USING count(*)", + # SQL ending in DML after a query: 2.0's con.sql() returns the query's rows. + "CREATE TABLE o2 AS SELECT 1 AS id; SELECT * FROM o2; INSERT INTO o2 VALUES (3)", + "SELECT a FROM t; DELETE FROM t WHERE a = 1", + "SELECT a FROM t; UPDATE t SET a = a", + "SELECT a FROM t; ALTER TABLE t ADD COLUMN b INTEGER", + "SELECT 1 AS x; SET threads = 2", + "SELECT 1 AS x; DROP TABLE t", + "BEGIN; SELECT * FROM t; COMMIT", + ], +) +def test_query_relation_is_none_for_a_statement_that_returns_no_rows(sql): + """DuckDB 1.5 returned None here; 2.0 returns a relation over a row count, + which an engine writing "the result" would write as a one-row Count file.""" + assert query_relation(_table_t(), sql) is None + + +def test_query_relation_of_a_copy_is_none(tmp_path): + con = _table_t() + assert query_relation(con, f"COPY t TO '{tmp_path / 'x.csv'}'") is None + assert (tmp_path / "x.csv").read_text(encoding="utf-8").splitlines() == ["a", "0", "1", "2"] + + +def test_query_relation_runs_each_statement_once(): + con = _table_t() + assert query_relation(con, "INSERT INTO t VALUES (7)") is None + assert first_row(con.execute("SELECT count(*) FROM t WHERE a = 7")) == (1,) + assert query_relation(con, "CREATE TABLE made AS SELECT 1 AS q") is None + assert con.execute("SELECT q FROM made").fetchall() == [(1,)] + rel = query_relation(con, "INSERT INTO t VALUES (8); SELECT count(*) AS n FROM t WHERE a = 8") + assert rel.fetchall() == [(1,)] + assert query_relation(con, "SELECT a FROM t; DELETE FROM t WHERE a = 8") is None + assert first_row(con.execute("SELECT count(*) FROM t WHERE a = 8")) == (0,) + + +def test_query_relation_of_sql_that_does_not_parse_runs_none_of_it(): + """DuckDB 2.0's con.sql() runs the statements ahead of one that does not + parse before it raises; 1.5 parses the whole SQL first and runs none.""" + con = _table_t() + with pytest.raises(duckdb.ParserException): + query_relation(con, "INSERT INTO t VALUES (7); SELEC 1") + assert first_row(con.execute("SELECT count(*) FROM t WHERE a = 7")) == (0,) + + +@pytest.mark.parametrize("sql", ["", "-- nothing", "/* nothing */"]) +def test_query_relation_of_no_statement_is_none(sql): + assert query_relation(_table_t(), sql) is None + + +@pytest.mark.parametrize( + "sql, value", + [ + ("SELECT 'x' AS v FROM t; -- the result", "x"), + ("SELECT 'x' AS v FROM t; /* the result */", "x"), + ("SELECT 'x' AS v FROM t -- the result", "x"), + ("SELECT 'x' AS v FROM t;\n-- one\n-- two\n", "x"), + # A ';' or '--' inside a string is not the end, and the offsets + # DuckDB's tokenizer gives are bytes, which these characters are not. + ("SELECT '; --é€' AS v FROM t; -- ü", "; --é€"), + ], +) +def test_a_comment_after_the_last_statement_leaves_the_relation_usable(sql, value, tmp_path): + """DuckDB 2.0 reads the statement's text again inside the SQL of a + relation's aggregate, limit or write, where a comment after the closing + ';' fails to parse; 1.5 reads it fine.""" + rel = query_relation(_table_t(), sql) + assert one_row(rel.aggregate("count(*)")) == (3,) + assert rel.limit(1).fetchall() == [(value,)] + out = tmp_path / "out.csv" + rel.write_csv(str(out)) + assert out.read_text(encoding="utf-8").splitlines()[1] == value + + +@pytest.mark.parametrize( + "sql, rows", + [ + # DuckDB 2.0 runs these as the relation is made and holds their rows. + ("PIVOT t ON a USING count(*)", [(1, 1, 1)]), + ("SELECT nextval('s') AS n FROM t ORDER BY n", [(1,), (2,), (3,)]), + ("SELECT a FROM t ORDER BY a", [(0,), (1,), (2,)]), + ], + ids=["pivot", "nextval", "lazy"], +) +def test_relation_view_reads_rows_duckdb_already_holds(sql, rows, tmp_path): + """DuckDB 2.0 refuses a view over such a relation ("it reads rows held in + this process"); relation_view keeps the COPY the engine runs from it.""" + con = _table_t() + con.execute("CREATE SEQUENCE s") + rel = con.sql(sql) + out = tmp_path / "out.parquet" + with relation_view(con, rel, "__fluid_test_result") as name: + con.execute(f"COPY (SELECT * FROM {name}) TO '{out}' (FORMAT parquet)") + assert duckdb.connect().execute(f"SELECT * FROM '{out}'").fetchall() == rows + # The query ran once (the sequence moved by three, not six) and nothing is left. + if "nextval" in sql: + assert con.execute("SELECT currval('s')").fetchall() == [(3,)] + left = "SELECT count(*) FROM duckdb_{0}s() WHERE {0}_name = '__fluid_test_result'" + assert con.execute(left.format("view")).fetchall() == [(0,)] + assert con.execute(left.format("table")).fetchall() == [(0,)] + + +# ── the engine, end to end ─────────────────────────────────────────────── + + +@pytest.fixture +def product(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + contract_dir = tmp_path / "product" + (contract_dir / "data").mkdir(parents=True) + (contract_dir / "data" / "orders.csv").write_text("id,amount\n1,10\n2,20\n", encoding="utf-8") + (tmp_path / "home").mkdir() + monkeypatch.setenv("HOME", str(tmp_path / "home")) + monkeypatch.chdir(contract_dir) + monkeypatch.delenv("FLUID_UPSTREAM_CONTRACTS", raising=False) + return contract_dir + + +def _local_apply(product: Path, *actions: Dict[str, Any]) -> List[Dict[str, Any]]: + from fluid_build.providers.local.local import LocalProvider + + provider = LocalProvider(project="local", region="local", anchor_dir=product) + load = {"op": "load_data", "path": str(product / "data" / "orders.csv"), "table_name": "orders"} + return provider.apply(actions=[load, *actions])["results"] + + +def test_local_apply_counts_rows_and_materializes(product): + """On DuckDB 2.0 the SQL action's row count was lost (``rows: None``) and a + materialize step fell back to a skipped ``noop`` after writing its file.""" + out = product / "out" + results = _local_apply( + product, + { + "op": "sql", + "sql": "SELECT * FROM orders", + "output_table": "t1", + "outputs": [str(out / "t1.parquet")], + }, + {"op": "materialize", "source_table": "t1", "dst": str(out / "copy.csv")}, + ) + assert [r["status"] for r in results] == ["ok", "ok", "ok"], results + assert results[0]["rows"] == 2 + assert results[1]["rows"] == 2 + assert results[2]["op"] == "materialize" + assert (out / "copy.csv").read_text(encoding="utf-8").splitlines() == [ + "id,amount", + "1,10", + "2,20", + ] + + +@pytest.mark.parametrize( + "sql", + [ + "CREATE TABLE made AS SELECT * FROM orders", + "CREATE TABLE made AS PIVOT orders ON id USING sum(amount)", + "CREATE TABLE made AS SELECT * FROM orders; SELECT * FROM made; " + + "INSERT INTO made VALUES (3, 30)", + ], + ids=["ctas", "pivot-ctas", "dml-after-a-query"], +) +def test_local_sql_that_returns_no_rows_fails_the_same_on_both_versions(product, sql): + """1.5 failed on ``None.write_csv``; 2.0 wrote a one-row ``Count`` file (or + a ``Success`` header, or the rows before the INSERT) and reported ok. Both + now refuse it and say what to write instead.""" + out = product / "out" / "made.csv" + results = _local_apply(product, {"op": "sql", "sql": sql, "outputs": [str(out)]}) + assert results[1]["status"] == "error", results + assert "returns no rows" in results[1]["error"] + assert "write just the SELECT" in results[1]["error"] + assert not out.exists() + + +def test_local_sql_ending_in_a_comment_writes_its_result(product): + out = product / "out" / "orders.csv" + sql = "SELECT * FROM orders ORDER BY id; -- every order" + results = _local_apply(product, {"op": "sql", "sql": sql, "outputs": [str(out)]}) + assert results[1]["status"] == "ok", results + assert out.read_text(encoding="utf-8").splitlines() == ["id,amount", "1,10", "2,20"] + + +def test_local_sql_with_a_statement_that_does_not_parse_runs_none_of_it(product): + """On DuckDB 2.0 the statement ahead of the bad one ran, and the next + action in the same apply saw the table it made.""" + out = product / "out" + results = _local_apply( + product, + { + "op": "sql", + "sql": "CREATE TABLE made AS SELECT * FROM orders; SELEC 1", + "outputs": [str(out / "bad.csv")], + }, + { + "op": "sql", + "sql": "SELECT count(*) AS n FROM duckdb_tables() WHERE table_name = 'made'", + "outputs": [str(out / "made.csv")], + }, + ) + assert [r["status"] for r in results] == ["ok", "error", "ok"], results + assert (out / "made.csv").read_text(encoding="utf-8").splitlines() == ["n", "0"] + + +def _sales(con: Any) -> Any: + con.execute( + "CREATE TABLE sales AS SELECT * FROM (VALUES ('jan', 1), ('feb', 2), ('jan', 3)) " + "s(month, amount)" + ) + return con + + +def test_a_local_result_duckdb_already_holds_is_copied_to_a_remote_output(tmp_path): + """``_copy_relation_to_uri`` (an ``s3://`` output) made a view of the + result, which DuckDB 2.0 refuses for a PIVOT without an IN list. A local + path stands in for the bucket: the COPY is the same statement.""" + from fluid_build.providers.local.local import LocalProvider + + con = _sales(duckdb.connect()) + rel = con.sql("PIVOT sales ON month USING sum(amount)") + out = tmp_path / "pivot.parquet" + provider = LocalProvider(project="local", region="local", anchor_dir=tmp_path) + provider._copy_relation_to_uri(con, rel, str(out), "parquet") + assert duckdb.connect().execute(f"SELECT feb, jan FROM '{out}'").fetchall() == [(2, 4)] + + +def test_contract_tests_writes_a_result_duckdb_already_holds_to_parquet(tmp_path): + pytest.importorskip("pandas") + pytest.importorskip("pyarrow") + from fluid_build.contract_tests import _execute_sql, _write_output + + rel = _execute_sql(_sales(duckdb.connect()), "PIVOT sales ON month USING sum(amount)") + out = tmp_path / "pivot.parquet" + assert _write_output(rel, {"path": str(out), "format": "parquet"}, dry_run=False) == ( + 1, + str(out.resolve()), + ) + assert duckdb.connect().execute(f"SELECT feb, jan FROM '{out}'").fetchall() == [(2, 4)] + + +def test_contract_tests_sql_that_returns_no_rows_fails_clearly(): + from fluid_build.contract_tests import LocalProviderError, _execute_sql + + with pytest.raises(LocalProviderError, match="returns no rows"): + _execute_sql(duckdb.connect(), "CREATE TABLE made AS SELECT 1 AS q") + assert _execute_sql(duckdb.connect(), "SELECT 1 AS q").fetchall() == [(1,)] + + +def test_contract_tests_dry_run_counts_a_real_relation(tmp_path): + from fluid_build.contract_tests import _write_output + + rel = duckdb.connect().sql("SELECT range AS a FROM range(42)") + out = tmp_path / "out.csv" + assert _write_output(rel, {"path": str(out), "format": "csv"}, dry_run=True) == (42, None) + assert not out.exists() + + +def test_a_remote_landing_is_probed_without_keeping_the_connection_busy(tmp_path, monkeypatch): + """``_artifact_exists`` probes a remote landing with ``SELECT 1 ... LIMIT 1`` + on the run's own connection; on DuckDB 2.0 a result read only partway keeps + the connection busy for the next relation read on it.""" + from fluid_build.build_runners.duckdb import runner + + landed = tmp_path / "landed.parquet" + duckdb.connect().execute(f"COPY (SELECT 1 AS a) TO '{landed}' (FORMAT parquet)") + monkeypatch.setattr(runner, "_is_remote_uri", lambda _path: True) + con = duckdb.connect() + assert runner._artifact_exists(con, str(landed), "read_parquet") is True + assert con.sql(f"SELECT a FROM read_parquet('{landed}')").fetchall() == [(1,)] + assert runner._artifact_exists(con, str(tmp_path / "missing.parquet"), "read_parquet") is False + + +@pytest.mark.parametrize("function", ["postgres_query", "mysql_query", "sqlite_scan", "read_xlsx"]) +def test_a_function_of_an_extension_the_sandbox_did_not_load_reads_as_a_refusal(function): + from fluid_build.providers._duckdb_sandbox import is_sandbox_refusal, is_unloaded_function + + con = secure_duckdb_connect(allow=DuckDBAllowlist.none()) + with pytest.raises(duckdb.Error) as refused: + con.execute(f"SELECT * FROM {function}('x', 'y')") + assert is_unloaded_function(refused.value), refused.value + assert is_sandbox_refusal(refused.value) + + +@pytest.mark.skipif(not DUCKDB_2, reason="postgres_scan exists through DuckDB 1.5") +def test_duckdb_2_has_no_postgres_scan_even_with_the_extension_loaded(): + """Why the duckdb runner reads a Postgres source through ATTACH: DuckDB 2.0's + postgres extension dropped ``postgres_scan``. Not a sandbox refusal either: + loading the extension would not bring it back.""" + from fluid_build.providers._duckdb_sandbox import is_sandbox_refusal + + try: + con = secure_duckdb_connect(allow=DuckDBAllowlist.none(), extensions=["postgres"]) + except duckdb.Error as exc: + raise pytest.skip.Exception("the postgres extension is not installable here") from exc + with pytest.raises(duckdb.CatalogException) as missing: + con.execute("SELECT * FROM postgres_scan('host=127.0.0.1 port=9', 'public', 't')") + assert "does not exist" in str(missing.value) + assert not is_sandbox_refusal(missing.value) + + +def test_an_unknown_function_is_not_a_sandbox_refusal(): + from fluid_build.providers._duckdb_sandbox import is_sandbox_refusal + + con = secure_duckdb_connect(allow=DuckDBAllowlist.none()) + with pytest.raises(duckdb.Error) as missing: + con.execute("SELECT * FROM no_such_function_anywhere()") + assert not is_sandbox_refusal(missing.value) + + +def test_the_snowflake_mock_connects(tmp_path): + """DuckDB 2.0 refuses ``CREATE SCHEMA information_schema`` (a system catalog).""" + from fluid_build.providers.local.mocks import SnowflakeDuckDBMock + + mock = SnowflakeDuckDBMock(db_path=str(tmp_path / "sf.duckdb")) + con = mock.connect() + assert con.execute("SELECT count(*) FROM information_schema.schemata").fetchall()[0][0] > 0 + assert con.execute( + "SELECT count(*) FROM information_schema.schemata WHERE schema_name = 'PUBLIC'" + ).fetchall() == [(1,)] + con.close() + + +# ── storage format ─────────────────────────────────────────────────────── + + +def _storage(con, name: str) -> str: + tags = con.execute( + "SELECT tags FROM duckdb_databases() WHERE database_name = ?", [name] + ).fetchall()[0][0] + return dict(tags)["storage_version"] + + +def test_storage_compat_config_pins_only_from_duckdb_2(): + assert storage_compat_config((1, 5, 0)) == {} + assert storage_compat_config((1, 99, 99)) == {} + for version in ((2, 0, 0), (2, 3, 1), (3, 0, 0)): + assert storage_compat_config(version) == { + "storage_compatibility_version": STORAGE_COMPATIBILITY_VERSION + } + # What DuckDB 1.5 writes by default: storage version 64, "v1.0.0+". + assert STORAGE_COMPATIBILITY_VERSION == "v1.0.0" + + +def test_a_database_file_the_engine_creates_is_readable_by_duckdb_1_5(tmp_path): + """A 2.x install must not write files a 1.5 one (Python 3.10, or a tool still + pinned below 2) cannot open: 1.5.6 reads storage versions 64 to 68 only.""" + db = tmp_path / "new.duckdb" + con = secure_duckdb_connect(db, allow=DuckDBAllowlist.none()) + con.execute("CREATE TABLE t AS SELECT 1 AS a") + # The same format on both: 1.5's own default, which 2.x is pinned to. + assert _storage(con, "new") == "v1.0.0+" + assert con.execute("SELECT current_setting('storage_compatibility_version')").fetchall() == [ + (STORAGE_COMPATIBILITY_VERSION if DUCKDB_2 else "v0.10.2",) + ] + + +def test_a_new_database_file_has_the_header_duckdb_1_5_writes(tmp_path): + """Byte-level: storage version 64 in the main header, which DuckDB 1.0 and + later read; 2.0's own default writes 999 there, which no 1.5 opens.""" + db = tmp_path / "header.duckdb" + con = secure_duckdb_connect(db, allow=DuckDBAllowlist.none()) + con.execute("CREATE TABLE t AS SELECT 1 AS a") + con.close() + header = db.read_bytes()[:20] + assert header[8:12] == b"DUCK" + assert int.from_bytes(header[12:20], "little") == 64 + + +def test_an_attached_new_database_follows_the_same_format(tmp_path): + con = secure_duckdb_connect( + allow=DuckDBAllowlist.none(), + before_lock=lambda c: c.execute(f"ATTACH '{tmp_path / 'att.duckdb'}' AS att"), + ) + con.execute("CREATE TABLE att.t AS SELECT 1 AS a") + assert _storage(con, "att") == "v1.0.0+" + + +def test_a_database_file_that_exists_keeps_its_format(tmp_path): + db = tmp_path / "theirs.duckdb" + raw = duckdb.connect(str(db)) + raw.execute("CREATE TABLE t AS SELECT 1 AS a") + before = _storage(raw, "theirs") + raw.close() + + con = secure_duckdb_connect(db, allow=DuckDBAllowlist.none()) + con.execute("INSERT INTO t VALUES (2)") + assert _storage(con, "theirs") == before + assert con.execute("SELECT count(*) FROM t").fetchall() == [(2,)] + con.close() + con = secure_duckdb_connect(db, allow=DuckDBAllowlist.none(), read_only=True) + assert con.execute("SELECT count(*) FROM t").fetchall() == [(2,)] + + +@pytest.mark.skipif(not DUCKDB_2, reason="only 2.x has a newer format to choose") +def test_a_callers_own_storage_setting_wins(tmp_path): + db = tmp_path / "latest.duckdb" + con = secure_duckdb_connect( + db, allow=DuckDBAllowlist.none(), config={"storage_compatibility_version": "latest"} + ) + con.execute("CREATE TABLE t AS SELECT 1 AS a") + assert _storage(con, "latest").startswith("v2.") + + +def test_the_storage_setting_is_locked_with_the_rest(tmp_path): + con = secure_duckdb_connect(allow=DuckDBAllowlist.none()) + with pytest.raises(duckdb.Error): + con.execute("SET storage_compatibility_version = 'latest'") + + +# ── no partial reads in DuckDB code ────────────────────────────────────── + +#: Reads that DuckDB 2.0 removed from relations or ``con.execute()`` results, or +#: that leave a result open (the connection stays busy until it is drained): +#: ``fetchone``/``fetchmany`` (``first_row`` replaces them), the Arrow batch +#: readers, ``rel.arrow()``. Caught separately: a relation's argument-less +#: ``execute()`` (gone), and an ``execute()`` of a query whose result is thrown +#: away unread (open until the next ``execute()``). ``rel.pl()`` is not listed: +#: relations keep it in 2.0. +_PARTIAL_OR_GONE = frozenset( + {"fetchone", "fetchmany", "fetch_df_chunk", "fetch_arrow_table", "fetch_record_batch"} + | {"fetch_arrow_reader", "record_batch", "arrow"} +) + +#: Modules whose ``fetchone`` (or ``execute``) belongs to another client, never +#: DuckDB's; each must keep importing no DuckDB. +_OTHER_CLIENTS = { + "fluid_build/cli/doctor.py": "sqlite3 / Snowflake connector cursors", + "fluid_build/cli/workspace.py": "sqlite3 workspace registry", + "fluid_build/copilot/catalog/snowflake.py": "Snowflake connector cursor", + "fluid_build/copilot/store/backends/postgres.py": "psycopg cursor", + "fluid_build/copilot/store/backends/sqlite.py": "sqlite3 connection", + "fluid_build/copilot/store/backends/vector.py": "sqlite3 + sqlite-vec probe", + "fluid_build/forge_datamodel/from_ddl/snowflake_dumper.py": "Snowflake connector cursor", + "fluid_build/providers/gcp/util/auth.py": "googleapiclient request .execute()", + "fluid_build/output_ports/mcp/drivers/postgres.py": "psycopg cursor health probe", + "fluid_build/output_ports/mcp/drivers/snowflake.py": "Snowflake cursor health probe", +} + +#: The start of a statement that returns rows, for an ``execute()`` whose result +#: is thrown away. +_QUERY_START = re.compile( + r"\s*(SELECT|WITH|FROM|TABLE|VALUES|DESCRIBE|SHOW|SUMMARIZE|PRAGMA|CALL|EXPLAIN|PIVOT)\b", + re.IGNORECASE, +) + + +def _literal_start(node: ast.expr) -> str: + """The leading literal text of a ``str`` or f-string argument, else ``""``.""" + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.JoinedStr) and node.values: + first = node.values[0] + if isinstance(first, ast.Constant) and isinstance(first.value, str): + return first.value + return "" + + +def _unread_query(node: ast.AST) -> bool: + """``.execute("SELECT ...")`` as a statement of its own: its rows are never read.""" + if not (isinstance(node, ast.Expr) and isinstance(node.value, ast.Call)): + return False + call = node.value + return ( + isinstance(call.func, ast.Attribute) + and call.func.attr == "execute" + and bool(call.args) + and bool(_QUERY_START.match(_literal_start(call.args[0]))) + ) + + +def _partial_reads(path: Path) -> List[str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + where = path.relative_to(REPO_ROOT).as_posix() if REPO_ROOT in path.parents else path.name + found = [ + f"{where}:{node.lineno} .{node.func.attr}()" + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and ( + node.func.attr in _PARTIAL_OR_GONE + or (node.func.attr == "execute" and not node.args and not node.keywords) + ) + ] + found += [ + f"{where}:{node.lineno} unread .execute()" for node in ast.walk(tree) if _unread_query(node) + ] + return found + + +@pytest.mark.parametrize( + "shape", + [ + "con.execute('select 1').fetchone()\n", + "rel.aggregate('count(*)').fetchone()[0]\n", + "con.execute('select 1').fetchmany(2)\n", + "con.execute('select 1').fetch_arrow_table()\n", + "rel.fetch_df_chunk()\n", + "con.execute('select 1').fetch_record_batch(10)\n", + "rel.fetch_arrow_reader(1000)\n", + "rel.record_batch()\n", + "rel.arrow()\n", + "rel.execute()\n", + "con.execute('SELECT 1')\n", + "con.execute(f'select count(*) FROM {table}')\n", + "con.execute('DESCRIBE t')\n", + ], +) +def test_the_partial_read_detector_catches_each_shape(shape, tmp_path): + sample = tmp_path / "probe.py" + sample.write_text(shape, encoding="utf-8") + assert _partial_reads(sample), shape + + +@pytest.mark.parametrize( + "shape", + [ + "con.execute('SELECT 1').fetchall()\n", + "rows = con.execute('SELECT 1')\n", + "con.execute('CREATE TABLE t AS SELECT 1')\n", + "con.execute(f'COPY ({select}) TO {path}')\n", + "con.execute(sql)\n", + ], +) +def test_the_partial_read_detector_leaves_a_read_or_a_write_alone(shape, tmp_path): + sample = tmp_path / "probe.py" + sample.write_text(shape, encoding="utf-8") + assert not _partial_reads(sample), shape + + +def test_no_partial_reads_in_duckdb_code(): + offenders, scanned = [], 0 + for path in sorted(PACKAGE.rglob("*.py")): + if path.relative_to(REPO_ROOT).as_posix() in _OTHER_CLIENTS: + continue + scanned += 1 + offenders.extend(_partial_reads(path)) + # A fence that scans nothing passes; it must see the DuckDB modules. + assert scanned > 500, scanned + assert not offenders, ( + "On DuckDB 2.0 a relation has no fetchone/fetchmany, and a con.execute() " + "result read partway keeps its connection busy (the next relation read, " + "register or create_function raises ResourceInUseException). Use " + "fluid_build.providers._duckdb_compat.first_row with a one-row query; if the " + "call is another database's client, list the module in _OTHER_CLIENTS:\n " + + "\n ".join(offenders) + ) + + +def test_every_other_client_module_still_needs_its_exemption(): + """An exemption outlives its reason silently; a stale one would hide DuckDB code.""" + for module in _OTHER_CLIENTS: + path = REPO_ROOT / module + assert path.is_file(), module + assert _partial_reads(path), f"{module} no longer needs its exemption; drop it" + # The exemption is for another client's calls: DuckDB code there would + # go unchecked. + assert not re.search( + r"import duckdb|from duckdb|secure_duckdb_connect", path.read_text(encoding="utf-8") + ), f"{module} now uses DuckDB; it cannot stay exempt" diff --git a/tests/providers/test_duckdb_sandbox.py b/tests/providers/test_duckdb_sandbox.py index f5796ecb..6b75c904 100644 --- a/tests/providers/test_duckdb_sandbox.py +++ b/tests/providers/test_duckdb_sandbox.py @@ -522,9 +522,10 @@ def test_the_copy_sibling_is_one_more_file_confined_like_the_landing(layout, mon def test_a_sqlite_source_outside_the_allowed_directories_is_refused(layout, monkeypatch): - """The sqlite scanner opens files through its own library, which DuckDB's - allowlist never bounds, so the declared source path was attached wherever - it pointed: any SQLite file on the host could be landed.""" + """The engine attaches a SQLite source before the sandbox's lock, when no + allowlist applies yet (and through DuckDB 1.5 the sqlite scanner did not + honour one), so the declared source path was attached wherever it pointed: + any SQLite file on the host could be landed.""" import sqlite3 from fluid_build.build_runners.duckdb.runner import execute_duckdb_build @@ -1133,8 +1134,8 @@ def test_a_function_from_an_unloaded_extension_reads_as_a_sandbox_refusal(layout def test_contract_sql_runs_on_a_connection_with_no_database_scanner(layout, printed): - """sqlite/postgres/mysql scanners are not bounded by the allowlist (next - test), so the connection contract SQL runs on must never have one loaded.""" + """A loaded database scanner widens what SQL on the connection reaches (the + tests below), so the connection contract SQL runs on must never have one.""" rc = _build( "SELECT count(*) AS n FROM duckdb_functions() WHERE function_name IN " "('sqlite_scan', 'sqlite_attach', 'postgres_scan', 'postgres_query', 'mysql_query')", @@ -1154,11 +1155,10 @@ def _with_extension(name: str) -> Any: ) from exc -def test_a_loaded_sqlite_scanner_is_not_bounded_by_the_allowlist(tmp_path): - """Pins a documented limit: once ``sqlite`` is loaded, any SQLite file the - process can read is reachable, whatever the allowlist says. A call site that - loads it must keep contract SQL off the connection.""" - db = tmp_path / "elsewhere.sqlite" +_DUCKDB_2 = sandbox.duckdb_version(duckdb) >= (2, 0, 0) + + +def _seed_sqlite(db: Path) -> None: seed = duckdb.connect() try: seed.execute("INSTALL sqlite; LOAD sqlite") @@ -1168,13 +1168,58 @@ def test_a_loaded_sqlite_scanner_is_not_bounded_by_the_allowlist(tmp_path): seed.execute(f"CREATE TABLE s.t AS SELECT '{SECRET}' AS v") seed.close() + +@pytest.mark.skipif(_DUCKDB_2, reason="DuckDB 2.0 bounds the sqlite scanner (next test)") +def test_a_loaded_sqlite_scanner_is_not_bounded_by_the_allowlist(tmp_path): + """Pins a documented limit through DuckDB 1.5: once ``sqlite`` is loaded, + any SQLite file the process can read is reachable, whatever the allowlist + says. A call site that loads it must keep contract SQL off the connection.""" + db = tmp_path / "elsewhere.sqlite" + _seed_sqlite(db) + con = _with_extension("sqlite") assert con.execute(f"SELECT v FROM sqlite_scan('{db}', 't')").fetchall() == [(SECRET,)] +@pytest.mark.skipif(not _DUCKDB_2, reason="through DuckDB 1.5 the scanner is unbounded") +def test_from_duckdb_2_a_loaded_sqlite_scanner_is_bounded_by_the_allowlist(tmp_path): + """DuckDB 2.0's sqlite scanner opens files through DuckDB's file system, so + the allowlist bounds it as it bounds ``read_csv``: a file outside, ``./..`` + out of an allowed directory and a symlink out of one are each refused, + while a file inside is read. The engine still keeps contract SQL off a + connection that loads a scanner (``postgres_query`` reaches what the engine + attached before the lock).""" + inside, outside = tmp_path / "allowed", tmp_path / "outside" + inside.mkdir() + outside.mkdir() + _seed_sqlite(outside / "x.sqlite") + _seed_sqlite(inside / "x.sqlite") + (inside / "link.sqlite").symlink_to(outside / "x.sqlite") + + try: + con = secure_duckdb_connect( + allow=DuckDBAllowlist.none().with_dirs(inside), extensions=["sqlite"] + ) + except duckdb.Error as exc: + raise pytest.skip.Exception("the sqlite extension is not installable here") from exc + assert con.execute(f"SELECT v FROM sqlite_scan('{inside}/x.sqlite', 't')").fetchall() == [ + (SECRET,) + ] + for refused in ( + f"SELECT v FROM sqlite_scan('{outside}/x.sqlite', 't')", + f"SELECT v FROM sqlite_scan('{inside}/./../outside/x.sqlite', 't')", + f"SELECT v FROM sqlite_scan('{inside}/link.sqlite', 't')", + f"ATTACH '{outside}/x.sqlite' AS o (TYPE sqlite)", + ): + with pytest.raises(duckdb.PermissionException): + con.execute(refused).fetchall() + + +@pytest.mark.skipif(_DUCKDB_2, reason="DuckDB 2.0's postgres extension has no postgres_scan") def test_a_loaded_postgres_scanner_is_not_bounded_by_the_allowlist(): - """Pins a documented limit: ``postgres_scan`` opens its own socket, so the - failure is the server's (nothing listens), not the sandbox's refusal.""" + """Pins a documented limit through DuckDB 1.5: ``postgres_scan`` opens its + own socket, so the failure is the server's (nothing listens), not the + sandbox's refusal.""" con = _with_extension("postgres") with pytest.raises(duckdb.Error) as failed: con.execute( @@ -1184,6 +1229,16 @@ def test_a_loaded_postgres_scanner_is_not_bounded_by_the_allowlist(): assert "Permission Error" not in str(failed.value) +@pytest.mark.parametrize("extension", ["postgres", "mysql"]) +def test_after_the_lock_no_database_can_be_attached_through_a_loaded_scanner(extension): + """On every version: ``postgres_query`` / ``mysql_query`` reach only the + databases the engine attached before the lock, which is how the duckdb + runner reads a postgres or mysql source.""" + con = _with_extension(extension) + with pytest.raises(duckdb.PermissionException, match="disabled through configuration"): + con.execute(f"ATTACH 'host=127.0.0.1 port=9 connect_timeout=1' AS x (TYPE {extension})") + + # ── the helper's own guarantees ────────────────────────────────────────── @@ -1402,11 +1457,20 @@ def _duckdb_aliases(tree: ast.AST) -> tuple: return modules, members +#: Functions of the duckdb module that take no connection and read nothing but +#: their argument. ``tokenize`` splits a SQL string into token offsets and +#: types (``_duckdb_compat`` finds a statement's trailing comments with it); +#: unlike ``duckdb.sql`` and the other module-level functions, it has no +#: ``connection`` parameter and runs no SQL. +_CONNECTIONLESS = frozenset({"tokenize"}) + + def _bypasses(path: Path) -> List[str]: """Each place ``path`` opens or queries DuckDB without the helper. * any call on the duckdb module (``duckdb.connect``, ``duckdb.sql`` and the - other module-level functions run on an unsandboxed default connection); + other module-level functions run on an unsandboxed default connection), + except the :data:`_CONNECTIONLESS` ones; * a function imported from duckdb (``from duckdb import connect``); * ``.connect(...)``, which catches the module held in a variable (``self._duckdb.connect``, ``_Duck.get().connect``). @@ -1425,7 +1489,8 @@ def _bypasses(path: Path) -> List[str]: if isinstance(func, ast.Attribute): receiver = func.value if isinstance(receiver, ast.Name) and receiver.id in modules: - found.append(f"{where} duckdb.{func.attr}(...)") + if func.attr not in _CONNECTIONLESS: + found.append(f"{where} duckdb.{func.attr}(...)") elif func.attr == "connect" and "duck" in ast.unparse(receiver).lower(): found.append(f"{where} {ast.unparse(func)}(...)") elif isinstance(func, ast.Name) and func.id in members: @@ -1433,6 +1498,19 @@ def _bypasses(path: Path) -> List[str]: return found +def test_a_connectionless_duckdb_function_is_not_a_bypass(tmp_path): + """``duckdb.tokenize`` runs no SQL; the exemption is that name alone.""" + # A default-connection function documents its ``connection`` parameter; + # tokenize has none, on 1.5 and 2.x. + assert "connection" in (duckdb.execute.__doc__ or "") + assert "connection" not in (duckdb.tokenize.__doc__ or "") + sample = tmp_path / "probe.py" + sample.write_text("import duckdb\nduckdb.tokenize('SELECT 1 -- c')\n", encoding="utf-8") + assert _bypasses(sample) == [] + sample.write_text("import duckdb\nduckdb.tokenizer('SELECT 1')\n", encoding="utf-8") + assert _bypasses(sample) + + def test_the_bypass_detector_sees_the_helpers_own_connect(): """A guard that has never fired proves nothing: it must see the one real call.""" assert [b for b in _bypasses(_HELPER) if "duckdb.connect(" in b] diff --git a/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py b/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py index c1b4a913..4ab34afd 100644 --- a/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py +++ b/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py @@ -383,6 +383,10 @@ def __init__(self, message: str) -> None: def sql(self, _sql): raise RuntimeError(self.message) + def extract_statements(self, _sql): + # On DuckDB 2.x query_relation parses the SQL first (_duckdb_compat). + raise RuntimeError(self.message) + def test_no_local_provider_log_call_carries_a_raw_exception_text(): """Every ``"error": str(e)`` the local provider logs goes through the shared diff --git a/tests/security/test_security_regressions.py b/tests/security/test_security_regressions.py index 0165d704..d06e7df7 100644 --- a/tests/security/test_security_regressions.py +++ b/tests/security/test_security_regressions.py @@ -39,6 +39,7 @@ _build_select_for_filesystem, _build_select_for_postgres, _csv_options_clause, + _postgres_attach_sql, ) from fluid_build.build_runners.hooks.tokenize_pii import ( _TOKEN_HEX_LEN, @@ -78,15 +79,18 @@ def test_filesystem_options_key_must_be_identifier(self): def test_postgres_stream_must_be_identifier(self): with pytest.raises(ValueError): - _build_select_for_postgres( - {"host": "h", "port": 5432, "user": "u", "password": "p", "database": "d"}, - "public.orders; DROP TABLE secrets; --", - ) + _build_select_for_postgres("public.orders; DROP TABLE secrets; --", alias="pg") + + def test_postgres_alias_must_be_identifier(self): + with pytest.raises(ValueError): + _build_select_for_postgres("orders", alias="pg; DROP TABLE secrets") + with pytest.raises(ValueError): + _postgres_attach_sql({"host": "h"}, alias="pg; DROP TABLE secrets") def test_postgres_password_with_quote_neutralized(self): # Password contains a single quote; the runner must escape it twice - # (libpq + SQL boundary) so the resulting SQL is still well-formed. - sql = _build_select_for_postgres( + # (libpq + SQL boundary) so the ATTACH is still one well-formed statement. + sql = _postgres_attach_sql( { "host": "h", "port": 5432, @@ -94,18 +98,18 @@ def test_postgres_password_with_quote_neutralized(self): "password": "p'; DROP TABLE secrets; --", "database": "d", }, - "orders", + alias="pg", ) # Single quote was doubled at the SQL boundary. assert "''" in sql - # Exactly one postgres_scan call is emitted. - assert sql.count("postgres_scan(") == 1 + assert sql.count("ATTACH ") == 1 + assert sql.endswith(" AS pg (TYPE postgres, SECRET pg, READ_ONLY)") def test_postgres_port_must_be_numeric(self): with pytest.raises(ValueError): - _build_select_for_postgres( + _postgres_attach_sql( {"host": "h", "port": "5432; DROP", "user": "u", "password": "p", "database": "d"}, - "orders", + alias="pg", ) def test_copy_destination_path_with_quote_is_quoted(self): diff --git a/tests/test_ci_duckdb_floor.py b/tests/test_ci_duckdb_floor.py new file mode 100644 index 00000000..aa049b72 --- /dev/null +++ b/tests/test_ci_duckdb_floor.py @@ -0,0 +1,72 @@ +# 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. + +"""The ``duckdb-floor`` CI job runs what differs between DuckDB versions. + +pyproject.toml takes ``duckdb>=1.5.0`` with no ceiling, and that job is the +only one that installs the floor itself. Its test list is written out, and a +written-out list drifts: the MCP SDK job's had come to name 3 of the 12 MCP +test files when it was replaced by ``-k mcp``. So the list must hold every +file the DuckDB end-to-end job runs, and every test file that branches on the +installed DuckDB version. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +import pytest +import yaml + +pytestmark = pytest.mark.unit + +REPO_ROOT = Path(__file__).resolve().parents[1] +CI = REPO_ROOT / ".github" / "workflows" / "ci.yml" + +_TEST_FILE = re.compile(r"\btests/[\w/]+\.py\b") +#: How a test reads the installed DuckDB version (the compat and sandbox tests +#: derive their ``_DUCKDB_2`` flag from one of these). +_READS_THE_VERSION = re.compile(r"duckdb\.__version__|\bduckdb_version\(|\bMIN_DUCKDB_VERSION\b") + + +def _files_run_by(job: str) -> set[str]: + steps = yaml.safe_load(CI.read_text(encoding="utf-8"))["jobs"][job]["steps"] + scripts = [step.get("run", "") for step in steps] + return {path for script in scripts if "pytest" in script for path in _TEST_FILE.findall(script)} + + +def test_the_floor_runs_every_suite_the_duckdb_end_to_end_job_runs(): + end_to_end = _files_run_by("duckdb-integration") + assert end_to_end, "found no test file in the DuckDB end-to-end job's pytest step" + missing = sorted(end_to_end - _files_run_by("duckdb-floor")) + assert not missing, f"add to the duckdb-floor job in ci.yml: {missing}" + + +def test_the_floor_runs_every_test_that_branches_on_the_duckdb_version(): + this_file = Path(__file__).resolve() + branching = { + path.relative_to(REPO_ROOT).as_posix() + for path in (REPO_ROOT / "tests").rglob("test_*.py") + if path.resolve() != this_file + and _READS_THE_VERSION.search(path.read_text(encoding="utf-8")) + } + assert branching, "found no test that reads the DuckDB version; the scan is broken" + missing = sorted(branching - _files_run_by("duckdb-floor")) + assert not missing, f"add to the duckdb-floor job in ci.yml: {missing}" + + +def test_every_file_the_floor_names_exists(): + absent = sorted(p for p in _files_run_by("duckdb-floor") if not (REPO_ROOT / p).is_file()) + assert not absent, f"the duckdb-floor job names files that do not exist: {absent}" diff --git a/tests/test_contract_tests_branches.py b/tests/test_contract_tests_branches.py index ac3efb86..944f8587 100644 --- a/tests/test_contract_tests_branches.py +++ b/tests/test_contract_tests_branches.py @@ -230,15 +230,6 @@ def test_invalid_format(self): with pytest.raises(LocalProviderError, match="format must be"): _write_output(rel, {"path": "/tmp/out.txt", "format": "json"}, dry_run=False) - def test_csv_dry_run(self): - rel = MagicMock() - count_rel = MagicMock() - count_rel.fetchone.return_value = (42,) - rel.count.return_value = count_rel - cnt, path = _write_output(rel, {"path": "/tmp/out.csv", "format": "csv"}, dry_run=True) - assert cnt == 42 - assert path is None - def test_csv_dry_run_count_error(self): rel = MagicMock() rel.count.side_effect = RuntimeError("count fail") diff --git a/tests/test_e2e_local_apply.py b/tests/test_e2e_local_apply.py index e2cb09a8..e2d30016 100644 --- a/tests/test_e2e_local_apply.py +++ b/tests/test_e2e_local_apply.py @@ -209,7 +209,7 @@ def test_two_apply_runs_produce_consistent_row_count(self, hello_world_workspace output_csv = hello_world_workspace / "runtime" / "out" / "hello-world-v1.csv" # ``all_varchar=True`` for the same reason as the schema test — # avoids materialising TIMESTAMPTZ which needs pytz at runtime. - row_count = duckdb.read_csv(str(output_csv), all_varchar=True).count("*").fetchone()[0] + row_count = duckdb.read_csv(str(output_csv), all_varchar=True).count("*").fetchall()[0][0] assert row_count == 1, f"expected 1 row after two applies; got {row_count}" diff --git a/tests/test_fix_a4_bugs.py b/tests/test_fix_a4_bugs.py index 3672c635..54347818 100644 --- a/tests/test_fix_a4_bugs.py +++ b/tests/test_fix_a4_bugs.py @@ -356,7 +356,7 @@ def test_parquet_apply_creates_parquet_file(self, tmp_path: Path, monkeypatch): assert Path(out_path).exists(), "parquet file must exist after apply" # Confirm it's readable parquet con = duckdb.connect(":memory:") - rows = con.sql(f"SELECT COUNT(*) FROM read_parquet({out_path!r})").fetchone()[0] + rows = con.sql(f"SELECT COUNT(*) FROM read_parquet({out_path!r})").fetchall()[0][0] assert rows >= 1 diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index 5c11e0dc..c8d823b9 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -427,6 +427,20 @@ def _write_product(tmp_path: Path, version: str, overlay: Dict[str, Any]) -> Non #: Written against ``[ingest, transform]``: its second entry patches transform. PROD_BY_POSITION = {"builds": [{}, {"id": "transform", "description": "8 executors"}]} +#: Environment variables a CLI subprocess here must not inherit: API keys, and +#: the debug level ``tests/cli/conftest.py`` sets, which some xdist workers keep. +#: At DEBUG the loader also logs a warning the command reports, and the tests +#: below count that the terminal shows it once. +_NOT_INHERITED = ("FLUID_LOG_LEVEL", "FLUID_DEBUG", "FLUID_SHOW_REGISTRATION") + + +def _cli_env() -> Dict[str, str]: + return { + k: v + for k, v in os.environ.items() + if not k.endswith("_API_KEY") and k not in _NOT_INHERITED + } + def _validate(tmp_path: Path, *argv: str) -> "subprocess.CompletedProcess[str]": return subprocess.run( @@ -435,7 +449,7 @@ def _validate(tmp_path: Path, *argv: str) -> "subprocess.CompletedProcess[str]": cwd=str(tmp_path), capture_output=True, text=True, - env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + env=_cli_env(), ) @@ -515,7 +529,7 @@ def test_workspace_validate_strict_fails_on_it(tmp_path: Path) -> None: product.mkdir(parents=True) _write_product(product, "0.7.5", PROD_BY_POSITION) run = [sys.executable, "-m", "fluid_build.cli", "validate", "--env", "prod", "--offline"] - env = {k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")} + env = _cli_env() lenient = subprocess.run(run, cwd=tmp_path, capture_output=True, text=True, env=env) strict = subprocess.run( run + ["--strict"], cwd=tmp_path, capture_output=True, text=True, env=env