From cd0dbb8390d562a45d6c22cf0b0b665167b5ff73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jakob=20Sch=C3=B6dl?= Date: Fri, 25 Sep 2026 14:03:34 +0200 Subject: [PATCH] fix: read SQL quality rules on mysql/iceberg as DuckDB SQL and keep contract SQL away from MySQL --- CHANGELOG.md | 1 + datacontract/engines/checks/create_checks.py | 4 +- datacontract/engines/checks/sql_guard.py | 12 +++--- .../engines/ibis/connections/connect.py | 24 +++++++---- docs/docs/quality-rules/sql.md | 7 ++-- docs/docs/testing/mysql.md | 2 + .../quality-sql-dialects/iceberg.yaml | 41 +++++++++++++++++++ .../fixtures/quality-sql-dialects/mysql.yaml | 9 ++-- tests/test_untrusted_contract.py | 41 +++++++++++++++++-- 9 files changed, 114 insertions(+), 27 deletions(-) create mode 100644 tests/fixtures/quality-sql-dialects/iceberg.yaml diff --git a/CHANGELOG.md b/CHANGELOG.md index 72c486d6f..5de32ad6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `datacontract api`: - a posted data contract resolves `${VAR}` only from the variables allow-listed with `--contract-variables` - a `${X:-local}` server type no longer slips past the checks for local files and environment-held credentials +- `datacontract test`: a SQL quality rule on a `mysql` server can no longer reach the MySQL server or its credentials; rules on `mysql` and `iceberg` servers are read as DuckDB SQL ### Fixed - Loading a DCS contract dropped quality rule `arguments`, so `invalidValues` and `missingValues` rules lost their configuration diff --git a/datacontract/engines/checks/create_checks.py b/datacontract/engines/checks/create_checks.py index eaf740dda..04eb2b5cd 100644 --- a/datacontract/engines/checks/create_checks.py +++ b/datacontract/engines/checks/create_checks.py @@ -196,9 +196,7 @@ def prepare_query( if not quality.query: return None - server_type = get_server_type(server) - # mysql is attached through duckdb, which runs the query - dialect = sqlglot_dialect_by_name("duckdb" if server_type == "mysql" else dialect_for_server_type(server_type)) + dialect = sqlglot_dialect_by_name(dialect_for_server_type(get_server_type(server))) # the dialect's extra name characters, e.g. `$` in Snowflake's `amount$usd` name_chars = re.escape("".join(Dialect.get_or_raise(dialect).tokenizer_class.VAR_SINGLE_TOKENS)) bare = re.compile(rf"[_a-zA-Z][\w{name_chars}]*") diff --git a/datacontract/engines/checks/sql_guard.py b/datacontract/engines/checks/sql_guard.py index e4bcab233..6599c66f7 100644 --- a/datacontract/engines/checks/sql_guard.py +++ b/datacontract/engines/checks/sql_guard.py @@ -20,19 +20,20 @@ # dialect-specific query is parsed as generic SQL and refused for syntax its own # data source accepts -- BigQuery's backticks, Snowflake's SAMPLE, SQL Server's TOP. # -# The value is the dialect the *rule author* writes in, which follows from the -# server type they declared. It is not always the engine that ends up running the -# query: the file, kafka and api server types are read through duckdb, and mysql is -# attached through duckdb, but a rule on a mysql server is still written as MySQL. +# It is the dialect the query is run in, which is not always the server's own: +# the guard must read the query exactly as the engine running it will. _DIALECT_BY_SERVER_TYPE = { - # read through duckdb, and written as duckdb + # read through duckdb "local": "duckdb", "s3": "duckdb", "gcs": "duckdb", "azure": "duckdb", "kafka": "duckdb", "api": "duckdb", + "iceberg": "duckdb", "duckdb": "duckdb", + # copied into duckdb + "mysql": "duckdb", # spark session backends "dataframe": "spark", # named by a different spelling in sqlglot @@ -44,7 +45,6 @@ "bigquery": "bigquery", "databricks": "databricks", "exasol": "exasol", - "mysql": "mysql", "oracle": "oracle", "postgres": "postgres", "redshift": "redshift", diff --git a/datacontract/engines/ibis/connections/connect.py b/datacontract/engines/ibis/connections/connect.py index 042c03258..ad9745f6e 100644 --- a/datacontract/engines/ibis/connections/connect.py +++ b/datacontract/engines/ibis/connections/connect.py @@ -156,7 +156,7 @@ def connect_ibis( return con if server_type == "mysql": - return _connect_mysql_via_duckdb(ibis, data_contract, server, run, schema_name, config) + return _connect_mysql_via_duckdb(ibis, data_contract, server, run, schema_name, config, untrusted_contract) if server_type == "snowflake": return ibis.snowflake.connect(**_snowflake_connection_kwargs(server, run, config)) @@ -583,7 +583,9 @@ def _connect_duckdb_database(ibis, server: Server, run: Run, config: Config): return con -def _connect_mysql_via_duckdb(ibis, data_contract, server: Server, run: Run, schema_name: str, config: Config): +def _connect_mysql_via_duckdb( + ibis, data_contract, server: Server, run: Run, schema_name: str, config: Config, untrusted_contract: bool +): """Connect to MySQL through DuckDB's ``mysql`` extension. ibis's native MySQL backend requires ``mysqlclient`` (a C extension with no @@ -594,7 +596,7 @@ def _connect_mysql_via_duckdb(ibis, data_contract, server: Server, run: Run, sch """ import duckdb - from datacontract.engines.ibis.connections.duckdb_connection import _load_extension + from datacontract.engines.ibis.connections.duckdb_connection import _load_extension, restrict_to_paths user = config.get_mysql_username(required=True) password = config.get_mysql_password(required=True) @@ -619,6 +621,11 @@ def _connect_mysql_via_duckdb(ibis, data_contract, server: Server, run: Run, sch model = schema_obj.physicalName or schema_obj.name _materialize_attached_table(con, "mysqldb", database, model) + # the checks read the local copies; contract SQL must not reach MySQL with these credentials + con.execute("DETACH mysqldb") + if untrusted_contract: + restrict_to_paths(con, []) + return ibis.duckdb.from_connection(con) @@ -630,14 +637,17 @@ def _materialize_attached_table(con, catalog: str, database: str | None, model: the DuckDB MySQL scanner can trigger DuckDB binder errors (e.g. on the grouped duplicate-count query), so we read the rows once and check locally. """ + # both come from the contract + safe_model = model.replace('"', '""') + safe_database = database.replace('"', '""') if database else None candidates = [] - if database: - candidates.append(f'{catalog}."{database}"."{model}"') - candidates.append(f'{catalog}."{model}"') + if safe_database: + candidates.append(f'{catalog}."{safe_database}"."{safe_model}"') + candidates.append(f'{catalog}."{safe_model}"') last_error = None for src in candidates: try: - con.execute(f'CREATE OR REPLACE TABLE "{model}" AS SELECT * FROM {src}') + con.execute(f'CREATE OR REPLACE TABLE "{safe_model}" AS SELECT * FROM {src}') return except Exception as e: # noqa: BLE001 - try the next naming candidate last_error = e diff --git a/docs/docs/quality-rules/sql.md b/docs/docs/quality-rules/sql.md index 75327493a..38d731ca9 100644 --- a/docs/docs/quality-rules/sql.md +++ b/docs/docs/quality-rules/sql.md @@ -94,14 +94,13 @@ This applies to every data source, and to every way the CLI is run. A data contr ## SQL dialect -There is no `dialect` field on a quality rule. The dialect is **derived from the type of the server the rule runs against**, so a query is read the same way the data source will read it — BigQuery's backtick-quoted table names, Snowflake's `SAMPLE`, SQL Server's `TOP` and Postgres' `->>` are all understood without any declaration: +There is no `dialect` field on a quality rule. The dialect is **derived from the type of the server the rule runs against**, so a query is read the same way the engine that runs it will read it — BigQuery's backtick-quoted table names, Snowflake's `SAMPLE`, SQL Server's `TOP` and Postgres' `->>` are all understood without any declaration: | Server type | SQL dialect | |---|---| -| `local`, `s3`, `gcs`, `azure`, `kafka`, `api`, `duckdb` | `duckdb` | +| `local`, `s3`, `gcs`, `azure`, `kafka`, `api`, `iceberg`, `duckdb`, `mysql` | `duckdb` | | `postgres` | `postgres` | | `redshift` | `redshift` | -| `mysql` | `mysql` | | `oracle` | `oracle` | | `sqlserver` | `tsql` | | `snowflake` | `snowflake` | @@ -115,7 +114,7 @@ There is no `dialect` field on a quality rule. The dialect is **derived from the The ODCS synonyms resolve to the spelling above before the dialect is looked up, so `postgresql` is read as `postgres`. A server declared as `type: custom` with `customType: mssql` is read as `tsql`, like `sqlserver`. -Files, Kafka topics and API responses are read through DuckDB, so a rule on those server types is written in DuckDB SQL — including its functions, such as `read_parquet` or `list_contains`. +Files, Kafka topics, API responses and Iceberg tables are read through DuckDB, and MySQL tables are copied into DuckDB before the checks run, so a rule on those server types is written in DuckDB SQL — including its functions, such as `read_parquet` or `list_contains`. MySQL-only syntax, such as backtick-quoted names or `DATE_SUB`, does not work there. ## Notes diff --git a/docs/docs/testing/mysql.md b/docs/docs/testing/mysql.md index 164289d89..10ac23ef0 100644 --- a/docs/docs/testing/mysql.md +++ b/docs/docs/testing/mysql.md @@ -76,6 +76,8 @@ schema: mustBe: 0 ``` +The tables are copied into DuckDB before the checks run, so write the query in [DuckDB SQL](../quality-rules/sql.md#sql-dialect), not MySQL. + Run `datacontract test datacontract.yaml` again: every violation is listed as an error, and the command exits with code `1` — ready for [CI/CD and scheduled runs](../scheduling/index.md) so you catch drift before your consumers do. ## Reference diff --git a/tests/fixtures/quality-sql-dialects/iceberg.yaml b/tests/fixtures/quality-sql-dialects/iceberg.yaml new file mode 100644 index 000000000..3f262d21b --- /dev/null +++ b/tests/fixtures/quality-sql-dialects/iceberg.yaml @@ -0,0 +1,41 @@ +apiVersion: v3.2.0 +kind: DataContract +id: orders-quality-iceberg +name: Orders quality rules (iceberg) +version: 1.0.0 +status: active +description: + purpose: >- + Custom SQL quality rules on an iceberg server, written in duckdb SQL because + duckdb reads the tables to run them. Used by + tests/test_quality_sql_dialects.py to keep the read-only SQL guard from + refusing SQL that a real data source accepts. +servers: + - server: production + type: iceberg + catalog: main + catalogUrl: https://polaris.acme.internal/api/catalog + namespace: sales + warehouse: s3://warehouse +schema: + - name: orders + physicalName: orders + properties: + - name: order_id + logicalType: string + physicalType: string + - name: order_total + logicalType: integer + physicalType: integer + - name: order_status + logicalType: string + physicalType: string + - name: order_timestamp + logicalType: date + physicalType: timestamp + quality: + - type: sql + description: "Rows were read from the table" + query: | + SELECT count(*) FROM ${model} + mustBeGreaterThan: 0 diff --git a/tests/fixtures/quality-sql-dialects/mysql.yaml b/tests/fixtures/quality-sql-dialects/mysql.yaml index 48cb1ed51..b5490dbaf 100644 --- a/tests/fixtures/quality-sql-dialects/mysql.yaml +++ b/tests/fixtures/quality-sql-dialects/mysql.yaml @@ -6,7 +6,8 @@ version: 1.0.0 status: active description: purpose: >- - Custom SQL quality rules written in the dialect of a mysql server. Used by + Custom SQL quality rules on a mysql server, written in duckdb SQL because + the tables are copied into duckdb to run them. Used by tests/test_quality_sql_dialects.py to keep the read-only SQL guard from refusing SQL that a real data source accepts. servers: @@ -33,12 +34,12 @@ schema: physicalType: timestamp quality: - type: sql - description: "Backtick-quoted table name (DIALECT-ONLY: backtick quoting)" + description: "Orders were copied from MySQL" query: | - SELECT count(*) FROM `orders` + SELECT count(*) FROM ${model} mustBeGreaterThan: 0 - type: sql description: "Orders arrived in the last day" query: | - SELECT count(*) FROM ${model} WHERE order_timestamp > DATE_SUB(NOW(), INTERVAL 1 DAY) + SELECT count(*) FROM ${model} WHERE order_timestamp > now() - INTERVAL 1 DAY mustBeGreaterThan: 0 diff --git a/tests/test_untrusted_contract.py b/tests/test_untrusted_contract.py index 4b6ba6d32..1f12a7297 100644 --- a/tests/test_untrusted_contract.py +++ b/tests/test_untrusted_contract.py @@ -6,12 +6,16 @@ 2. duckdb is confined to the contract's own data locations """ +import duckdb import pytest from fastapi.testclient import TestClient +from open_data_contract_standard.model import DataQuality, Server from datacontract.api import ALLOW_LOCAL_FILES_ENV, app from datacontract.data_contract import DataContract +from datacontract.engines.checks.create_checks import prepare_query from datacontract.engines.checks.sql_guard import dialect_for_server_type, refusal_reason +from datacontract.engines.ibis.connections.connect import _materialize_attached_table from datacontract.engines.ibis.connections.duckdb_connection import restrict_to_paths from datacontract.model.run import ResultEnum @@ -131,12 +135,43 @@ def test_the_server_dialect_still_refuses_what_is_not_a_query(): assert refusal_reason("DROP TABLE orders", dialect_for_server_type(server_type)) is not None -def test_file_server_types_are_read_as_duckdb(): - """local/s3/gcs/azure are read through duckdb, so a rule on them is duckdb SQL.""" - for server_type in ("local", "s3", "gcs", "azure"): +def test_server_types_run_in_duckdb_are_read_as_duckdb(): + """local/s3/gcs/azure/iceberg are read through duckdb and mysql tables are copied + into it, so a rule on them is duckdb SQL.""" + for server_type in ("local", "s3", "gcs", "azure", "iceberg", "mysql"): assert dialect_for_server_type(server_type) == "duckdb" +def test_a_backslash_escape_cannot_hide_a_second_statement_on_mysql(): + """Read as MySQL, `\\'` is an escaped quote and this is one SELECT; duckdb, which + runs it, ends the string there and runs the DROP.""" + query = "SELECT 1 AS x, 'a\\'; DROP TABLE orders; SELECT 1 AS y -- ' AS z" + + assert refusal_reason(query, dialect_for_server_type("mysql")) is not None + + +def test_a_backslash_in_a_name_cannot_hide_a_second_statement_on_mysql(): + quality = DataQuality( + type="sql", query='SELECT COUNT(*) AS {field} FROM {model}; CREATE TABLE pwn AS SELECT 1; --"' + ) + query = prepare_query(quality, "orders", "a\\", Server(server="production", type="mysql")) + + assert refusal_reason(query, dialect_for_server_type("mysql")) is not None + + +def test_a_quote_in_a_table_name_cannot_start_a_second_statement_on_mysql(): + """The table name comes from the contract and runs while MySQL is still attached.""" + model = 'orders" AS SELECT 1; CREATE TABLE pwned AS SELECT 1; --' + con = duckdb.connect() + con.execute("ATTACH ':memory:' AS mysqldb") + con.execute('CREATE TABLE mysqldb."orders"" AS SELECT 1; CREATE TABLE pwned AS SELECT 1; --" AS SELECT 1 AS id') + + _materialize_attached_table(con, "mysqldb", None, model) + + tables = con.execute("SELECT table_name FROM duckdb_tables() WHERE database_name = 'memory'").fetchall() + assert tables == [(model,)] + + def test_the_documented_dialect_mapping_matches_the_code(): """The mapping is a table in docs/docs/quality-rules/sql.md, so it drifts silently unless something reads both."""