Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 1 addition & 3 deletions datacontract/engines/checks/create_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}]*")
Expand Down
12 changes: 6 additions & 6 deletions datacontract/engines/checks/sql_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -44,7 +45,6 @@
"bigquery": "bigquery",
"databricks": "databricks",
"exasol": "exasol",
"mysql": "mysql",
"oracle": "oracle",
"postgres": "postgres",
"redshift": "redshift",
Expand Down
24 changes: 17 additions & 7 deletions datacontract/engines/ibis/connections/connect.py
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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)


Expand All @@ -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
Expand Down
7 changes: 3 additions & 4 deletions docs/docs/quality-rules/sql.md
Original file line number Diff line number Diff line change
Expand Up @@ -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` |
Expand All @@ -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

Expand Down
2 changes: 2 additions & 0 deletions docs/docs/testing/mysql.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
41 changes: 41 additions & 0 deletions tests/fixtures/quality-sql-dialects/iceberg.yaml
Original file line number Diff line number Diff line change
@@ -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
9 changes: 5 additions & 4 deletions tests/fixtures/quality-sql-dialects/mysql.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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
41 changes: 38 additions & 3 deletions tests/test_untrusted_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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."""
Expand Down
Loading