diff --git a/CHANGELOG.md b/CHANGELOG.md index 3da76e48df..a559c418a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -147,6 +147,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 An enabled primary key that is not a prefix of the tuned sort key now fails before any table is created, with a message that says to put the key columns first or disable `primary_keys`. +- **DuckDB `auto` and `tuned-fallback` runs no longer enable CHECK + constraints.** CHECK constraints cannot be confirmed from the catalog, so + these DuckDB configurations no longer request them, matching the shipped + DuckDB templates. No shipped benchmark schema emits a CHECK constraint, so + the tables are unchanged, and other platforms are unchanged. These DuckDB + runs record a different `requested_config_hash`, and their cached database + is rebuilt on the next run (for example `tpch_sf001_custom_pk_fk_uniq_check` + becomes `tpch_sf001_custom_pk_fk_uniq`). - **Spark, LakeSail and Velox per-query times no longer include `clearCache()`.** With `disable_cache` on (the default), these adapters cleared the session cache inside each query's timer. They now clear it before the timer starts, @@ -283,6 +291,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the ledger records it as satisfied by that statement instead of dropped. Sorts that no executed statement realized are still dropped. This lets ClickHouse tuned runs reach `applied_verified`. +- **DuckDB constraints are checked against the catalog.** Primary keys, + unique constraints and foreign keys in a tuned DuckDB `CREATE TABLE` are + compared with `duckdb_constraints()`. Keys must match on the same columns in + the same order, and a foreign key must also reference the same table and + columns. A tuned DuckDB run with the shipped TPC-H template now reaches + `applied_verified`. A `CHECK` clause in a `CREATE TABLE`, and constraints on + every other platform, still keep a run at `applied_unverified`. So does a + `CREATE TABLE` holding a backslash directly before a single or double + quote, which some engines read as an escaped quote the scanner cannot + mask, or holding a dollar-quoted string whose tag carries non-ASCII + characters; so do a + DuckDB `CREATE TABLE` whose constraints sit next to list syntax (`[...]`), + and a constrained table whose name also exists in another schema or + attached database. On those + platforms this now also holds when the same `CREATE TABLE` carries a + verified sort, partition or cluster key; before, the constraint was ignored + there. Receipt entries for constraints carry `kind: "constraint"` and a + `constraint_type`. Quoted non-ASCII table names, such as "É", are now found + consistently with DuckDB's `lower(table_name)` catalog filters. ### Changed diff --git a/benchbox/cli/commands/run.py b/benchbox/cli/commands/run.py index c4fccc0e19..5e6215c8f1 100644 --- a/benchbox/cli/commands/run.py +++ b/benchbox/cli/commands/run.py @@ -82,6 +82,7 @@ ) from benchbox.cli.tuning_runtime import ( build_baseline_unified_config, + build_default_unified_config, infer_runtime_tuning_mode, resolve_dataframe_tuning_config, ) @@ -764,8 +765,6 @@ def _resolve_tuning(s: types.SimpleNamespace, *, non_interactive_override: bool def _load_unified_tuning_config(s: types.SimpleNamespace, *, non_interactive_override: bool | None = None) -> None: s.loaded_unified_config = None - from benchbox.core.tuning.interface import UnifiedTuningConfiguration - tuning_resolution = s.tuning_resolution if tuning_resolution.config_file: if getattr(s, "resolved_mode", None) == "sql" and is_dataframe_tuning_file(tuning_resolution.config_file): @@ -814,13 +813,13 @@ def _load_unified_tuning_config(s: types.SimpleNamespace, *, non_interactive_ove tuning_resolution.source = TuningSource.BASELINE console.print("[green]✅ Tuning configuration completed[/green]") else: - s.loaded_unified_config = UnifiedTuningConfiguration() + s.loaded_unified_config = build_default_unified_config(s.platform) else: - s.loaded_unified_config = UnifiedTuningConfiguration() + s.loaded_unified_config = build_default_unified_config(s.platform) if s.logger: s.logger.debug("Using basic constraints-only configuration (fallback)") else: - s.loaded_unified_config = UnifiedTuningConfiguration() + s.loaded_unified_config = build_default_unified_config(s.platform) warn_sql_auto_mode(tuning_resolution, s.resolved_mode, console, s.logger, quiet=bool(s.quiet)) s.tuning_source_value = tuning_resolution.source.value diff --git a/benchbox/cli/tuning_resolver.py b/benchbox/cli/tuning_resolver.py index 79fa1f20fe..8aa058f557 100644 --- a/benchbox/cli/tuning_resolver.py +++ b/benchbox/cli/tuning_resolver.py @@ -428,7 +428,8 @@ def warn_sql_auto_mode( console.print( "[yellow]Warning: --tuning auto smart defaults are DataFrame-only today; " "this SQL run proceeds with a basic constraints-only configuration " - "(primary/foreign/unique/check constraints enabled; no other tunings applied), " + "(primary/foreign/unique constraints enabled, plus check constraints except on DuckDB; " + "no other tunings applied), " "not an untuned baseline.[/yellow]" ) if logger: diff --git a/benchbox/cli/tuning_runtime.py b/benchbox/cli/tuning_runtime.py index 8189c150ef..0f9049ae95 100644 --- a/benchbox/cli/tuning_runtime.py +++ b/benchbox/cli/tuning_runtime.py @@ -4,9 +4,17 @@ from typing import Any, Literal from benchbox.core.dataframe.tuning.interface import DataFrameTuningConfiguration +from benchbox.core.tuning.capability_registry import resolve_platform_key from benchbox.core.tuning.interface import UnifiedTuningConfiguration +def build_default_unified_config(platform: str | None) -> UnifiedTuningConfiguration: + config = UnifiedTuningConfiguration() + if platform and resolve_platform_key(str(platform).split(":", 1)[0]) == "duckdb": + config.check_constraints.enabled = False + return config + + def build_baseline_unified_config() -> UnifiedTuningConfiguration: baseline_config = UnifiedTuningConfiguration() baseline_config.disable_all_constraints() diff --git a/benchbox/core/tuning/introspection.py b/benchbox/core/tuning/introspection.py index 208828e12d..c7e3578e24 100644 --- a/benchbox/core/tuning/introspection.py +++ b/benchbox/core/tuning/introspection.py @@ -6,6 +6,7 @@ import logging import re +import string from dataclasses import dataclass, field from typing import Any, Protocol, runtime_checkable @@ -33,6 +34,11 @@ KIND_SORT_KEY = "sort_key" KIND_PARTITION_KEY = "partition_key" KIND_CLUSTER_KEY = "cluster_key" +KIND_CONSTRAINT = "constraint" + +CONSTRAINT_PRIMARY_KEY = "PRIMARY KEY" +CONSTRAINT_UNIQUE = "UNIQUE" +CONSTRAINT_FOREIGN_KEY = "FOREIGN KEY" _DDL_PHASES = frozenset({PHASE_DDL, PHASE_POST_LOAD}) @@ -44,6 +50,9 @@ class IntrospectedObject: columns: tuple[str, ...] = () name: str | None = None evidence: dict[str, Any] = field(default_factory=dict) + constraint_type: str | None = None + referenced_table: str | None = None + referenced_columns: tuple[str, ...] = () @dataclass @@ -52,6 +61,7 @@ class IntrospectedState: objects: list[IntrospectedObject] = field(default_factory=list) error: str | None = None truncated: bool = False + constraint_types: frozenset[str] = frozenset() @runtime_checkable @@ -75,6 +85,7 @@ class ReceiptEntry: reason: str | None = None evidence: dict[str, Any] | None = None detail: str | None = None + constraint_type: str | None = None def to_dict(self) -> dict[str, Any]: payload: dict[str, Any] = { @@ -84,6 +95,8 @@ def to_dict(self) -> dict[str, Any]: } if self.kind: payload["kind"] = self.kind + if self.constraint_type: + payload["constraint_type"] = self.constraint_type if self.table: payload["table"] = self.table if self.name: @@ -131,6 +144,7 @@ def to_payload(self) -> dict[str, Any]: payload["observed"] = [ { "kind": obj.kind, + **({"constraint_type": obj.constraint_type} if obj.constraint_type else {}), "table": obj.table, "columns": list(obj.columns), **({"name": obj.name} if obj.name else {}), @@ -151,6 +165,9 @@ class _Intent: table: str | None columns: tuple[str, ...] name: str | None = None + constraint_type: str | None = None + referenced_table: str | None = None + referenced_columns: tuple[str, ...] = () _CREATE_INDEX_RE = re.compile( @@ -173,6 +190,36 @@ class _Intent: _CLUSTER_BY_KEYWORD_RE = re.compile(r"\bcluster\s+by\b", re.IGNORECASE) _ALTER_TABLE_RE = re.compile(r"^\s*alter\s+table\s+(?P[^\s(]+)", re.IGNORECASE) _RECLUSTER_RE = re.compile(r"^\s*alter\s+table\s+\S+\s+(?:(?:resume|suspend)\s+)?recluster\b", re.IGNORECASE) +_CONSTRAINT_KEYWORD_RE = re.compile( + r"\b(?:primary\s+key|foreign\s+key|unique|references|check|constraint)\b", re.IGNORECASE +) +_FOLLOWING_STATEMENT_RE = re.compile(r";\s*\S") +_REFERENTIAL_ACTION_RE = re.compile(r"\bon\s+(?:delete|update)\b|\bmatch\b", re.IGNORECASE) +_TABLE_PRIMARY_KEY_RE = re.compile(r"\s*primary\s+key\s*\((?P[^()]*)\)\s*", re.IGNORECASE) +_TABLE_UNIQUE_RE = re.compile(r"\s*unique\s*\((?P[^()]*)\)\s*", re.IGNORECASE) +_TABLE_FOREIGN_KEY_RE = re.compile( + r"\s*foreign\s+key\s*\((?P[^()]*)\)\s*references\s+(?P
[^\s(]+)\s*\((?P[^()]*)\)\s*", + re.IGNORECASE, +) +_INLINE_REFERENCES_RE = re.compile(r"references\s+(?P
[^\s(]+)\s*\((?P[^()]*)\)", re.IGNORECASE) +_COLUMN_NAME_RE = re.compile(r'\s*(?P"(?:[^"]|"")+"|[^\s"(),]+)') +_CREATE_TABLE_PREFIX_RE = re.compile( + r"\s*create\s+(?:or\s+replace\s+)?table\s+(?:if\s+not\s+exists\s+)?", re.IGNORECASE +) +_QUOTED_IDENTIFIER_RE = re.compile(r'"(?P(?:[^"]|"")+)"') +_BARE_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_$]*") +_IDENTIFIER_LIST_ITEM_RE = re.compile(r'\s*(?P"(?:[^"]|"")*"|[^",\s]+)\s*(?P,|$)') +_NON_DOUBLE_QUOTE_OPENERS = ("`", "[", "'") +_UNMASKED_LITERAL_RES = ( + re.compile(r"\$(?:[^\W$]\w*)?\$"), + re.compile(r"(? str | None: return None +def constraint_tables(ledger: AppliedTuningLedger) -> set[str]: + tables: set[str] = set() + for statement in ledger.executed_statements: + _klass, intents = _classify(statement) + tables.update(intent.table for intent in intents if intent.kind == KIND_CONSTRAINT and intent.table) + return tables + + def statement_table(statement: AppliedStatement) -> str | None: return _statement_table(statement) @@ -325,6 +380,191 @@ def ledger_tables(ledger: AppliedTuningLedger) -> set[str]: return tables +def fold_physical_identifier(value: Any) -> str: + return str(value).translate(_ASCII_LOWER) + + +def _physical_identifier(text: str) -> str | None: + candidate = text.strip() + quoted = _QUOTED_IDENTIFIER_RE.fullmatch(candidate) + if quoted is not None: + return fold_physical_identifier(quoted.group("body").replace('""', '"')) + if _BARE_IDENTIFIER_RE.fullmatch(candidate): + return fold_physical_identifier(candidate) + return None + + +def _physical_identifier_list(text: str) -> tuple[str, ...] | None: + if not text.strip(): + return None + identifiers: list[str] = [] + position = 0 + while position < len(text): + item = _IDENTIFIER_LIST_ITEM_RE.match(text, position) + if item is None: + return None + identifier = _physical_identifier(item.group("item")) + if identifier is None: + return None + identifiers.append(identifier) + position = item.end() + if not item.group("sep"): + return tuple(identifiers) + return None + + +def _create_table_body_span(stripped: str, start: int) -> tuple[int, int] | None: + index = start + while index < len(stripped) and stripped[index].isspace(): + index += 1 + if index >= len(stripped) or stripped[index] != "(": + return None + depth = 0 + for position in range(index, len(stripped)): + if stripped[position] == "(": + depth += 1 + elif stripped[position] == ")": + depth -= 1 + if depth == 0: + return index + 1, position + return index + 1, -1 + + +def _top_level_element_spans(stripped: str, begin: int, end: int) -> list[tuple[int, int]]: + spans: list[tuple[int, int]] = [] + depth = 0 + element_start = begin + for position in range(begin, end): + char = stripped[position] + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + elif char == "," and depth == 0: + spans.append((element_start, position)) + element_start = position + 1 + spans.append((element_start, end)) + return spans + + +def _constraint_intent( + constraint_type: str, + table: str, + columns: tuple[str, ...], + referenced_table: str | None = None, + referenced_columns: tuple[str, ...] = (), +) -> _Intent: + return _Intent( + kind=KIND_CONSTRAINT, + table=table, + columns=columns, + constraint_type=constraint_type, + referenced_table=referenced_table, + referenced_columns=referenced_columns, + ) + + +def _table_level_constraint(element: str, keywords: list[str], table: str) -> _Intent | None: + if keywords == ["primary key"]: + match = _TABLE_PRIMARY_KEY_RE.fullmatch(element) + constraint_type = CONSTRAINT_PRIMARY_KEY + elif keywords == ["unique"]: + match = _TABLE_UNIQUE_RE.fullmatch(element) + constraint_type = CONSTRAINT_UNIQUE + elif keywords == ["foreign key", "references"]: + match = _TABLE_FOREIGN_KEY_RE.fullmatch(element) + constraint_type = CONSTRAINT_FOREIGN_KEY + else: + return None + if match is None: + return None + columns = _physical_identifier_list(match.group("cols")) + if columns is None: + return None + if constraint_type != CONSTRAINT_FOREIGN_KEY: + return _constraint_intent(constraint_type, table, columns) + referenced_table = _physical_identifier(match.group("table")) + referenced_columns = _physical_identifier_list(match.group("ref")) + if referenced_table is None or referenced_columns is None: + return None + return _constraint_intent(constraint_type, table, columns, referenced_table, referenced_columns) + + +def _column_level_constraints(element: str, keyword_matches: list[re.Match[str]], table: str) -> list[_Intent] | None: + name = _COLUMN_NAME_RE.match(element) + if name is None: + return None + column = _physical_identifier(name.group("name")) + if column is None: + return None + intents: list[_Intent] = [] + for keyword_match in keyword_matches: + keyword = re.sub(r"\s+", " ", keyword_match.group(0).lower()) + if keyword == "primary key": + intents.append(_constraint_intent(CONSTRAINT_PRIMARY_KEY, table, (column,))) + elif keyword == "unique": + intents.append(_constraint_intent(CONSTRAINT_UNIQUE, table, (column,))) + elif keyword == "references": + reference = _INLINE_REFERENCES_RE.match(element, keyword_match.start()) + if reference is None: + return None + referenced_table = _physical_identifier(reference.group("table")) + referenced_columns = _physical_identifier_list(reference.group("ref")) + if referenced_table is None or referenced_columns is None: + return None + intents.append( + _constraint_intent(CONSTRAINT_FOREIGN_KEY, table, (column,), referenced_table, referenced_columns) + ) + else: + return None + return intents + + +def _constraint_intents(statement_text: str) -> list[_Intent] | None: + stripped = _strip_sql_literals_and_comments(statement_text) + quoted = _strip_sql_literals_and_comments(statement_text, keep_double_quoted=True) + header = _CREATE_TABLE_RE.match(quoted) + prefix = _CREATE_TABLE_PREFIX_RE.match(quoted) + if header is None or prefix is None: + return [] + body = _create_table_body_span(stripped, header.end()) + if body is None: + return [] + if any(pattern.search(statement_text) for pattern in _UNMASKED_LITERAL_RES): + return None + if _DIALECT_DEPENDENT_QUOTE_RE.search(statement_text) and _CONSTRAINT_KEYWORD_RE.search(statement_text): + return None + begin, end = body + if end < 0: + return None if _CONSTRAINT_KEYWORD_RE.search(stripped, begin) else [] + if _FOLLOWING_STATEMENT_RE.search(stripped, end + 1): + return None + table = _physical_identifier(quoted[prefix.end() : begin - 1]) + intents: list[_Intent] = [] + for element_begin, element_end in _top_level_element_spans(stripped, begin, end): + stripped_element = stripped[element_begin:element_end] + keyword_matches = list(_CONSTRAINT_KEYWORD_RE.finditer(stripped_element)) + if not keyword_matches: + continue + if table is None or _REFERENTIAL_ACTION_RE.search(stripped_element): + return None + if statement_text[element_begin:element_end].lstrip()[:1] in _NON_DOUBLE_QUOTE_OPENERS: + return None + element = quoted[element_begin:element_end] + if stripped_element[: keyword_matches[0].start()].strip(): + column_intents = _column_level_constraints(element, keyword_matches, table) + if column_intents is None: + return None + intents.extend(column_intents) + continue + keywords = [re.sub(r"\s+", " ", match.group(0).lower()) for match in keyword_matches] + intent = _table_level_constraint(element, keywords, table) + if intent is None: + return None + intents.append(intent) + return intents + + def _classify(statement: AppliedStatement) -> tuple[str, list[_Intent]]: text = _strip_sql_literals_and_comments(str(statement.statement or "")).strip() lowered = text.lower() @@ -348,26 +588,38 @@ def _classify(statement: AppliedStatement) -> tuple[str, list[_Intent]]: ] ct = _CREATE_TABLE_RE.match(text) - if ct: - table = normalize_identifier(ct.group("table")) + quoted_ct = _CREATE_TABLE_RE.match( + _strip_sql_literals_and_comments(str(statement.statement or ""), keep_double_quoted=True).strip() + ) + if ct or quoted_ct: intents: list[_Intent] = [] - order = _ORDER_BY_RE.search(text) - if order: - intents.append(_Intent(kind=KIND_SORT_KEY, table=table, columns=normalize_columns(order.group("cols")))) - part = _PARTITION_BY_RE.search(text) - if part: - intents.append(_Intent(kind=KIND_PARTITION_KEY, table=table, columns=normalize_columns(part.group("cols")))) - cluster = _CLUSTER_BY_RE.search(text) - if cluster: - intents.append( - _Intent(kind=KIND_CLUSTER_KEY, table=table, columns=normalize_columns(cluster.group("cols"))) - ) - if order is None and _ORDER_BY_KEYWORD_RE.search(text) and not _ORDER_BY_NOOP_RE.search(text): - return UNVERIFIABLE, [] - if cluster is None and _CLUSTER_BY_KEYWORD_RE.search(text): + if ct: + table = normalize_identifier(ct.group("table")) + order = _ORDER_BY_RE.search(text) + if order: + intents.append(_Intent(kind=KIND_SORT_KEY, table=table, columns=normalize_columns(order.group("cols")))) + part = _PARTITION_BY_RE.search(text) + if part: + intents.append( + _Intent(kind=KIND_PARTITION_KEY, table=table, columns=normalize_columns(part.group("cols"))) + ) + cluster = _CLUSTER_BY_RE.search(text) + if cluster: + intents.append( + _Intent(kind=KIND_CLUSTER_KEY, table=table, columns=normalize_columns(cluster.group("cols"))) + ) + if order is None and _ORDER_BY_KEYWORD_RE.search(text) and not _ORDER_BY_NOOP_RE.search(text): + return UNVERIFIABLE, [] + if cluster is None and _CLUSTER_BY_KEYWORD_RE.search(text): + return UNVERIFIABLE, [] + if part is None and _PARTITION_BY_KEYWORD_RE.search(text): + return UNVERIFIABLE, [] + elif any(pattern.search(text) for pattern in _LAYOUT_KEYWORD_RES): return UNVERIFIABLE, [] - if part is None and _PARTITION_BY_KEYWORD_RE.search(text): + constraints = _constraint_intents(str(statement.statement or "")) + if constraints is None: return UNVERIFIABLE, [] + intents.extend(constraints) if intents: return "verifiable", intents return UNVERIFIABLE, [] @@ -400,7 +652,63 @@ def _short_diff(expected: tuple[str, ...], observed: tuple[str, ...], table: str return diff +def _references(columns: tuple[str, ...], table: str | None, referenced: tuple[str, ...]) -> str: + return f"{list(columns)} references {table}{list(referenced)}" + + +def _foreign_key_diff(intent: _Intent, fact: IntrospectedObject | None) -> str: + expected = _references(intent.columns, intent.referenced_table, intent.referenced_columns) + if fact is None: + return f"expected {expected} != observed nothing" + observed_table = fold_physical_identifier(fact.referenced_table) if fact.referenced_table else None + return f"expected {expected} != observed {_references(fact.columns, observed_table, fact.referenced_columns)}" + + +def _folded(identifiers: tuple[str, ...]) -> tuple[str, ...]: + return tuple(fold_physical_identifier(identifier) for identifier in identifiers) + + +def _has_reference(fact: IntrospectedObject) -> bool: + return bool(fact.referenced_table) and bool(fact.referenced_columns) + + +def _match_constraint(intent: _Intent, state: IntrospectedState) -> tuple[str, IntrospectedObject | None, str | None]: + if intent.constraint_type not in state.constraint_types: + return UNVERIFIABLE, None, f"{state.platform} introspection reads no {intent.constraint_type} constraints" + same_type = [ + obj + for obj in state.objects + if obj.kind == KIND_CONSTRAINT + and obj.constraint_type == intent.constraint_type + and obj.table is not None + and fold_physical_identifier(obj.table) == intent.table + ] + if not same_type: + return ABSENT, None, f"no {intent.constraint_type} constraint found in catalog" + if not intent.columns: + return MISMATCH, same_type[0], None + same_columns = [obj for obj in same_type if _folded(obj.columns) == intent.columns] + if intent.constraint_type != CONSTRAINT_FOREIGN_KEY: + return (CORROBORATED, same_columns[0], None) if same_columns else (MISMATCH, same_type[0], None) + if not intent.referenced_table or not intent.referenced_columns: + return UNVERIFIABLE, None, "foreign key statement names no referenced table or columns" + for obj in same_columns: + if ( + _has_reference(obj) + and fold_physical_identifier(obj.referenced_table) == intent.referenced_table + and _folded(obj.referenced_columns) == intent.referenced_columns + ): + return CORROBORATED, obj, None + incomplete = next((obj for obj in same_columns if not _has_reference(obj)), None) + if incomplete is not None: + return UNVERIFIABLE, incomplete, "catalog foreign key fact has no referenced table or columns" + return MISMATCH, (same_columns or same_type)[0], None + + def _match_object(intent: _Intent, state: IntrospectedState) -> tuple[str, IntrospectedObject | None]: + if intent.kind == KIND_CONSTRAINT: + verdict, fact, _reason = _match_constraint(intent, state) + return verdict, fact same_table = [ obj for obj in state.objects @@ -498,11 +806,16 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta expected_columns=intent.columns, reason="introspection degraded", detail=state_error, + constraint_type=intent.constraint_type, ) ) continue - verdict, fact = _match_object(intent, introspected_state) + if intent.kind == KIND_CONSTRAINT: + verdict, fact, reason = _match_constraint(intent, introspected_state) + else: + verdict, fact = _match_object(intent, introspected_state) + reason = f"no {intent.kind} found in catalog" if verdict == ABSENT else None entry = ReceiptEntry( statement=stmt.statement, phase=stmt.phase, @@ -511,15 +824,18 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta table=intent.table, name=intent.name, expected_columns=intent.columns, + reason=reason, + constraint_type=intent.constraint_type, ) if fact is not None: entry.observed_columns = fact.columns entry.evidence = dict(fact.evidence) if fact.evidence else None if verdict == MISMATCH: - observed = fact.columns if fact is not None else () - entry.diff = _short_diff(intent.columns, observed, intent.table) - elif verdict == ABSENT: - entry.reason = f"no {intent.kind} found in catalog" + if intent.constraint_type == CONSTRAINT_FOREIGN_KEY: + entry.diff = _foreign_key_diff(intent, fact) + else: + observed = fact.columns if fact is not None else () + entry.diff = _short_diff(intent.columns, observed, intent.table) entries.append(entry) verifiable = [e for e in entries if e.verdict in _VERIFIABLE_VERDICTS] @@ -541,6 +857,9 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta __all__ = [ "ABSENT", + "CONSTRAINT_FOREIGN_KEY", + "CONSTRAINT_PRIMARY_KEY", + "CONSTRAINT_UNIQUE", "CORROBORATED", "Introspector", "IntrospectedObject", @@ -548,6 +867,7 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta "IntrospectionReceipt", "KIND_INDEX", "KIND_CLUSTER_KEY", + "KIND_CONSTRAINT", "KIND_PARTITION_KEY", "KIND_SORT_KEY", "MAINTENANCE", @@ -555,7 +875,9 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta "ReceiptEntry", "TRANSIENT", "UNVERIFIABLE", + "constraint_tables", "corroborate", + "fold_physical_identifier", "ledger_tables", "normalize_columns", "normalize_identifier", diff --git a/benchbox/platforms/duckdb_introspection.py b/benchbox/platforms/duckdb_introspection.py index 96c03d1484..668a3bfa2c 100644 --- a/benchbox/platforms/duckdb_introspection.py +++ b/benchbox/platforms/duckdb_introspection.py @@ -9,9 +9,15 @@ from benchbox.core.tuning.applied_ledger import AppliedTuningLedger from benchbox.core.tuning.introspection import ( + CONSTRAINT_FOREIGN_KEY, + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_UNIQUE, + KIND_CONSTRAINT, KIND_INDEX, IntrospectedObject, IntrospectedState, + constraint_tables, + fold_physical_identifier, ledger_tables, normalize_columns, normalize_identifier, @@ -20,6 +26,9 @@ logger = logging.getLogger(__name__) _MAX_INDEX_ROWS = 1000 +_MAX_CONSTRAINT_ROWS = 1000 +_CATALOG_CONSTRAINT_TYPES = frozenset({CONSTRAINT_PRIMARY_KEY, CONSTRAINT_UNIQUE, CONSTRAINT_FOREIGN_KEY}) +_REQUIRED_CONSTRAINT_COLUMNS = ("table_name", "constraint_type", "constraint_column_names") class DuckDBTuningIntrospector: @@ -66,7 +75,97 @@ def introspect(self, connection: Any, ledger: AppliedTuningLedger) -> Introspect }, ) ) - return IntrospectedState(platform=self.platform, objects=objects, truncated=truncated) + + tables_with_constraints = constraint_tables(ledger) + try: + ambiguous = _tables_in_several_schemas(connection, tables_with_constraints) + constraints, constraints_truncated = _read_constraints(connection, tables_with_constraints) + except Exception as exc: + logger.debug("duckdb constraint introspection degraded: %s", exc) + return IntrospectedState(platform=self.platform, error=f"duckdb_constraints read failed: {exc}") + if ambiguous: + return IntrospectedState( + platform=self.platform, + error=f"constraint tables {ambiguous} exist in more than one database schema", + ) + objects.extend(constraints) + return IntrospectedState( + platform=self.platform, + objects=objects, + truncated=truncated or constraints_truncated, + constraint_types=_CATALOG_CONSTRAINT_TYPES, + ) + + +def _tables_in_several_schemas(connection: Any, tables: set[str]) -> list[str]: + if not tables: + return [] + ordered_tables = sorted(tables) + placeholders = ", ".join("lower(?)" for _ in ordered_tables) + rows = connection.execute( + "SELECT lower(table_name) FROM duckdb_tables() " + f"WHERE lower(table_name) IN ({placeholders}) " + "GROUP BY lower(table_name) HAVING count(*) > 1", + ordered_tables, + ).fetchall() + return sorted(str(row[0]) for row in rows) + + +def _read_constraints(connection: Any, tables: set[str]) -> tuple[list[IntrospectedObject], bool]: + if not tables: + return [], False + ordered_tables = sorted(tables) + placeholders = ", ".join("lower(?)" for _ in ordered_tables) + type_list = ", ".join(f"'{constraint_type}'" for constraint_type in sorted(_CATALOG_CONSTRAINT_TYPES)) + cursor = connection.execute( + "SELECT * FROM duckdb_constraints() " + f"WHERE constraint_type IN ({type_list}) " + "AND database_name = current_database() AND schema_name = current_schema() " + f"AND lower(table_name) IN ({placeholders}) " + f"LIMIT {_MAX_CONSTRAINT_ROWS}", + ordered_tables, + ) + names = [str(description[0]).lower() for description in cursor.description or ()] + missing = [column for column in _REQUIRED_CONSTRAINT_COLUMNS if column not in names] + if missing: + raise ValueError(f"duckdb_constraints() lacks columns {missing}") + rows = cursor.fetchall() + objects: list[IntrospectedObject] = [] + for row in rows: + fact = dict(zip(names, row, strict=False)) + table_name = fact["table_name"] + if not isinstance(table_name, str) or fold_physical_identifier(table_name) not in tables: + continue + constraint_type = str(fact["constraint_type"] or "").strip().upper() + if constraint_type not in _CATALOG_CONSTRAINT_TYPES: + continue + referenced_table = fact.get("referenced_table") or None + referenced_columns = _physical_columns(fact.get("referenced_column_names")) + objects.append( + IntrospectedObject( + kind=KIND_CONSTRAINT, + table=table_name, + columns=_physical_columns(fact["constraint_column_names"]), + name=fact.get("constraint_name"), + constraint_type=constraint_type, + referenced_table=referenced_table, + referenced_columns=referenced_columns, + evidence={ + "constraint_name": fact.get("constraint_name"), + "constraint_type": constraint_type, + "constraint_column_names": list(fact["constraint_column_names"] or []), + "referenced_table": referenced_table, + "referenced_column_names": list(fact.get("referenced_column_names") or []), + }, + ) + ) + return objects, len(rows) >= _MAX_CONSTRAINT_ROWS + + +def _physical_columns(names: Any) -> tuple[str, ...]: + if not isinstance(names, (list, tuple)) or any(not isinstance(name, str) or not name for name in names): + return () + return tuple(fold_physical_identifier(name) for name in names) __all__ = ["DuckDBTuningIntrospector"] diff --git a/docs/development/tuning-introspection-receipts.md b/docs/development/tuning-introspection-receipts.md index c64b585302..ceab2d5b3a 100644 --- a/docs/development/tuning-introspection-receipts.md +++ b/docs/development/tuning-introspection-receipts.md @@ -156,6 +156,8 @@ platform-agnostic; the per-platform catalog *reads* live in the | `CREATE TABLE ... ORDER BY (cols)` | ddl/post_load | `sort_key` | a catalog `sort_key` fact on `T` whose columns equal `cols` | yes | | `CREATE TABLE ... PARTITION BY (cols)` | ddl/post_load | `partition_key` | a catalog `partition_key` fact on `T` whose columns equal `cols` | yes | | `ALTER TABLE T CLUSTER BY (cols)`, `CREATE TABLE ... CLUSTER BY (cols)` | ddl/post_load | `cluster_key` | a catalog `cluster_key` fact on `T` whose columns equal `cols`, in order | yes | +| `CREATE TABLE T (..., PRIMARY KEY (cols))`, `UNIQUE (cols)`, or column-level `PRIMARY KEY` / `UNIQUE` | ddl/post_load | `constraint` | a catalog `constraint` fact on `T` of the same `constraint_type` whose columns equal `cols`, in order; only when the introspector reads that constraint type | yes | +| `CREATE TABLE T (..., FOREIGN KEY (cols) REFERENCES R (rcols))`, or column-level `REFERENCES R (rcols)` | ddl/post_load | `constraint` | a catalog `FOREIGN KEY` fact on `T` whose child columns equal `cols`, referenced table equals `R` and referenced columns equal `rcols`, all in order | yes | | `SET ...`, `PRAGMA ...` | any | `transient` | session/config, no persistent catalog footprint -- noted, **non-blocking** | no | | `OPTIMIZE TABLE ...` / maintenance | ddl/post_load | `maintenance` | merge/compaction op, no distinct catalog footprint -- noted, **non-blocking** | no | | `ALTER TABLE T [RESUME/SUSPEND] RECLUSTER` | ddl/post_load | `maintenance` | reorganizes existing data; the clustering KEY is the catalog footprint -- **non-blocking** | no | @@ -176,6 +178,43 @@ and emits one receipt entry per clause, so each key must corroborate on its own. Corroborating only the first clause would let a run reach `applied_verified` while the other configured key never applied. +Constraint clauses are read only from the column list in parentheses that +follows the table name; a clause outside it, such as a ClickHouse +`PRIMARY KEY` engine clause, is not a constraint. Each constraint in a +`CREATE TABLE` gets its own receipt entry with `kind: "constraint"` and a +`constraint_type` of `PRIMARY KEY`, `UNIQUE` or `FOREIGN KEY`. The statement +stays blocking `unverifiable` as a whole when the column list holds +constraint text that is not fully parsed: a `CHECK` constraint, a named +`CONSTRAINT`, a referential action (`ON DELETE`, `ON UPDATE`, `MATCH`), a +`REFERENCES` without a column list, trailing clause text, an unbalanced +list, a second SQL statement in the same ledger entry, or a literal the +scanner cannot mask exactly: a dollar-quoted string (`$$...$$` or +`$tag$...$tag$`, where a tag is any run of word characters, so non-ASCII +tags count), an escape string (`E'...'`), a nested block comment, a +backslash directly before a single or double quote (some engines read `\'` +or `\"` as an escaped quote, so the scanner cannot tell where the string +or quoted identifier ends), or a +`[` or backtick anywhere in a statement that carries constraint text. DuckDB +reads `[` as list syntax and a backtick as an operator, not as identifier +quotes, so the scanner cannot tell which text they enclose. +Such a statement is `unverifiable` as a whole, including any sort, partition +or cluster clause in it. A constraint entry is `unverifiable` when the introspector does not read +that constraint type, which is every platform except DuckDB, and when a +catalog foreign-key fact with the same child columns lacks its referenced +table or columns. This applies even when the same statement also carries a +corroborated sort, partition or cluster key, so a constraint never rides on +another clause's verdict. Constraint identifiers compare by their physical +contents, the way DuckDB resolves them: ASCII letters fold to lower case, and +everything else, including whitespace inside a quoted identifier, is kept, so +a key on `" a "` never corroborates a key on `a`. The catalog's names are +compared without the quote and whitespace stripping that index and sort-key +columns use. A table, column or referenced name that is not one bare +identifier or one double-quoted identifier, such as a schema-qualified name or +a backtick- or bracket-quoted one, keeps the statement `unverifiable`. +A quoted table name binds the statement's +constraints to that table; sort, partition and cluster clauses of a +`CREATE TABLE` with a quoted table name stay `unverifiable` as before. + **Verdicts** (per statement): `corroborated`, `absent` (expected object not in catalog), `mismatch` (object present, columns differ -- carries a short diff), plus the non-blocking notes `transient` / `maintenance` and the @@ -221,7 +260,8 @@ which every attempted ddl/post_load statement failed; that ledger derives Each platform's `Introspector.introspect(connection, ledger) -> IntrospectedState` returns structured catalog facts (`kind`, `table`, `columns`, `name`, -`evidence`), bounded to the tables the ledger touched and wrapped +`evidence`, and for constraints `constraint_type`, `referenced_table` and +`referenced_columns`), bounded to the tables the ledger touched and wrapped non-fatal. **No screen-scraping of engine DDL text** where a structured catalog exists. ClickHouse and Snowflake keep a hard catalog-row limit, then filter the bounded result to table names extracted from every supported @@ -237,6 +277,23 @@ that itself reaches the cap remains explicitly truncated. bounded query, filtered to the ledger's tables. Corroborates each `CREATE INDEX` ledger entry against its `index` row. + It also reads `duckdb_constraints()` for `PRIMARY KEY`, `UNIQUE` and + `FOREIGN KEY` rows in a second bounded query that filters inside its `WHERE` + clause to those types, the current database and schema, and the tables + that carry constraint intents. Those table names come from the same parser + as the intents, so a quoted name such as `"my t"` or `"q""t"` is looked up + by its physical name. Because those names are unqualified, a table name + that also exists in another schema or attached database, including a + temporary table, degrades the whole state: the catalog cannot show which of + them the statement changed, for example after `SET schema` or `USE`. + `NOT NULL` and `CHECK` rows are never read, so `CHECK` constraints + stay `unverifiable`. Each row becomes a `constraint` fact carrying + `constraint_column_names` as its columns and, for foreign keys, + `referenced_table` and `referenced_column_names`. A DuckDB build whose + `duckdb_constraints()` lacks the referenced columns leaves foreign keys + `unverifiable`; a build that lacks the required columns, or a failed read, + degrades the whole state, and hitting the row bound marks it truncated. + DuckDB index statements are corroborated against the physical identifiers `create_schema` produced. The tuning builder resolves each logical table and column through `duckdb_tables()` / `duckdb_columns()` (case-insensitive, @@ -261,8 +318,11 @@ that itself reaches the cap remains explicitly truncated. A corroborated index is evidence of the index, not of row order: an index can exist when the data was never sorted. Reaching `applied_verified` also needs every other statement in the ledger to corroborate, so a run that also - requests constraints, partitioning or CHECK constraints stays - `applied_unverified` until those have catalog rules. A sort-only + requests partitioning or CHECK constraints stays `applied_unverified`. + Primary, unique and foreign keys corroborate against `duckdb_constraints()`; + the shipped DuckDB TPC-H template reaching `applied_verified` end to end is + pinned by + `tests/integration/tuning/test_duckdb_constraint_corroboration_e2e.py`. A sort-only configuration built through the real `create_schema` is pinned by `tests/unit/platforms/test_physical_identifier_resolution.py::TestDuckDBPhysicalIdentifiers::test_sort_only_config_reaches_applied_verified_without_mismatch`. diff --git a/tests/integration/tuning/test_duckdb_constraint_corroboration_e2e.py b/tests/integration/tuning/test_duckdb_constraint_corroboration_e2e.py new file mode 100644 index 0000000000..63b10e331d --- /dev/null +++ b/tests/integration/tuning/test_duckdb_constraint_corroboration_e2e.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from benchbox.cli.config import ConfigManager +from benchbox.core.tpch.benchmark import TPCHBenchmark +from benchbox.core.tuning.applied_ledger import APPLIED_VERIFIED +from benchbox.core.tuning.interface import TuningType +from benchbox.core.tuning.introspection import ( + CONSTRAINT_FOREIGN_KEY, + CONSTRAINT_PRIMARY_KEY, + CORROBORATED, + KIND_CONSTRAINT, + UNVERIFIABLE, +) +from benchbox.platforms.duckdb import DuckDBAdapter + +pytestmark = [ + pytest.mark.integration, + pytest.mark.slow, +] + +SHIPPED_TPCH_TEMPLATE = ( + Path(__file__).resolve().parents[3] / "benchbox" / "core" / "tuning" / "templates" / "duckdb" / "tpch_tuned.yaml" +) + + +@pytest.fixture(scope="module") +def tuned_tpch_result(tmp_path_factory: pytest.TempPathFactory): + base = tmp_path_factory.mktemp("constraint_corroboration_tpch") + benchmark = TPCHBenchmark(scale_factor=0.01, output_dir=str(base / "data")) + benchmark.generate_data() + config = ConfigManager().load_unified_tuning_config(SHIPPED_TPCH_TEMPLATE, platform="duckdb") + adapter = DuckDBAdapter( + type="duckdb", database_path=str(base / "tpch.duckdb"), tuning_enabled=True, tuning_config=config + ) + return config, adapter.run_benchmark(benchmark, benchmark_type="olap", query_subset=["1"]) + + +def test_shipped_template_requests_constraints_without_partitioning_or_check(tuned_tpch_result): + config, _result = tuned_tpch_result + assert config.primary_keys.enabled and config.foreign_keys.enabled + assert config.check_constraints.enabled is False + assert not any(tuning.get_columns_by_type(TuningType.PARTITIONING) for tuning in config.table_tunings.values()) + + +def test_shipped_tpch_template_has_no_unverifiable_constraint_entries(tuned_tpch_result): + _config, result = tuned_tpch_result + entries = result.applied_tuning_ledger["receipt"]["entries"] + constraint_entries = [entry for entry in entries if entry.get("kind") == KIND_CONSTRAINT] + + assert {entry["constraint_type"] for entry in constraint_entries} == { + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_FOREIGN_KEY, + } + assert len(constraint_entries) == 17 + assert all(entry["verdict"] == CORROBORATED for entry in constraint_entries) + assert not [entry for entry in entries if entry["verdict"] == UNVERIFIABLE] + + +def test_shipped_tpch_template_reaches_applied_verified(tuned_tpch_result): + _config, result = tuned_tpch_result + assert result.applied_tuning_ledger["receipt"]["corroborated"] is True + assert result.tuning_validation_status == APPLIED_VERIFIED diff --git a/tests/unit/cli/commands/test_run_tuning_mode_guard.py b/tests/unit/cli/commands/test_run_tuning_mode_guard.py index acaebf9b82..5dda4d6845 100644 --- a/tests/unit/cli/commands/test_run_tuning_mode_guard.py +++ b/tests/unit/cli/commands/test_run_tuning_mode_guard.py @@ -6,7 +6,7 @@ import pytest from benchbox.cli.commands.run import _load_unified_tuning_config -from benchbox.cli.tuning_resolver import TuningSource +from benchbox.cli.tuning_resolver import TuningMode, TuningSource pytestmark = [ pytest.mark.unit, @@ -46,3 +46,26 @@ def test_dataframe_run_loads_without_the_sql_guard(tmp_path): _load_unified_tuning_config(s) assert s.loaded_unified_config == "loaded" + + +@pytest.mark.parametrize( + ("mode", "source"), + [(TuningMode.TUNED, TuningSource.FALLBACK), (TuningMode.AUTO, TuningSource.SMART_DEFAULTS)], +) +@pytest.mark.parametrize(("platform", "check_enabled"), [("duckdb", False), ("postgresql", True)]) +def test_fallback_and_auto_request_check_constraints_except_on_duckdb(mode, source, platform, check_enabled): + s = SimpleNamespace( + tuning_resolution=SimpleNamespace(config_file=None, source=source, mode=mode), + resolved_mode="sql", + config=MagicMock(), + logger=None, + ctx=MagicMock(), + platform=platform, + non_interactive=True, + quiet=True, + ) + + _load_unified_tuning_config(s) + + assert s.loaded_unified_config.check_constraints.enabled is check_enabled + assert s.loaded_unified_config.primary_keys.enabled is True diff --git a/tests/unit/cli/test_tuning_runtime.py b/tests/unit/cli/test_tuning_runtime.py index d09a683681..6943258492 100644 --- a/tests/unit/cli/test_tuning_runtime.py +++ b/tests/unit/cli/test_tuning_runtime.py @@ -2,7 +2,11 @@ import pytest -from benchbox.cli.tuning_runtime import build_baseline_unified_config, infer_runtime_tuning_mode +from benchbox.cli.tuning_runtime import ( + build_baseline_unified_config, + build_default_unified_config, + infer_runtime_tuning_mode, +) from benchbox.core.tuning.interface import TuningType, UnifiedTuningConfiguration pytestmark = [ @@ -53,3 +57,18 @@ def test_build_baseline_unified_config_reports_no_clustering_strategy() -> None: assert config.platform_optimizations.databricks_clustering_strategy == "none" assert config.platform_optimizations.z_ordering_enabled is False assert config.platform_optimizations.liquid_clustering_enabled is False + + +@pytest.mark.parametrize("platform", ["duckdb", "DuckDB", "duckdb:memory"]) +def test_default_unified_config_disables_check_constraints_on_duckdb(platform: str) -> None: + config = build_default_unified_config(platform) + + assert config.check_constraints.enabled is False + assert config.primary_keys.enabled is True + assert config.foreign_keys.enabled is True + assert config.unique_constraints.enabled is True + + +@pytest.mark.parametrize("platform", [None, "motherduck", "postgresql", "snowflake"]) +def test_default_unified_config_keeps_check_constraints_elsewhere(platform: str | None) -> None: + assert build_default_unified_config(platform).to_dict() == UnifiedTuningConfiguration().to_dict() diff --git a/tests/unit/core/tuning/test_constraint_corroboration.py b/tests/unit/core/tuning/test_constraint_corroboration.py new file mode 100644 index 0000000000..9f97c17380 --- /dev/null +++ b/tests/unit/core/tuning/test_constraint_corroboration.py @@ -0,0 +1,510 @@ +from __future__ import annotations + +import pytest + +from benchbox.core.tuning.applied_ledger import PHASE_DDL, AppliedStatement, AppliedTuningLedger +from benchbox.core.tuning.introspection import ( + ABSENT, + CONSTRAINT_FOREIGN_KEY, + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_UNIQUE, + CORROBORATED, + KIND_CONSTRAINT, + KIND_SORT_KEY, + MISMATCH, + UNVERIFIABLE, + IntrospectedObject, + IntrospectedState, + _classify, + _match_object, + constraint_tables, + corroborate, +) + +pytestmark = [pytest.mark.unit, pytest.mark.fast] + +ALL_TYPES = frozenset({CONSTRAINT_PRIMARY_KEY, CONSTRAINT_UNIQUE, CONSTRAINT_FOREIGN_KEY}) + +ORDERS_DDL = """CREATE TABLE orders ( + o_orderkey INTEGER NOT NULL, + o_custkey INTEGER NOT NULL, + o_comment VARCHAR(79) NOT NULL, + PRIMARY KEY (o_orderkey), + UNIQUE (o_custkey, o_comment), + FOREIGN KEY (o_custkey) REFERENCES customer(c_custkey) +);""" + + +def _fact(constraint_type, columns, table="orders", referenced_table=None, referenced_columns=()): + return IntrospectedObject( + kind=KIND_CONSTRAINT, + table=table, + columns=tuple(columns), + constraint_type=constraint_type, + referenced_table=referenced_table, + referenced_columns=tuple(referenced_columns), + ) + + +def _orders_facts(**overrides): + facts = { + CONSTRAINT_PRIMARY_KEY: _fact(CONSTRAINT_PRIMARY_KEY, ["o_orderkey"]), + CONSTRAINT_UNIQUE: _fact(CONSTRAINT_UNIQUE, ["o_custkey", "o_comment"]), + CONSTRAINT_FOREIGN_KEY: _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders", "customer", ["c_custkey"]), + } + facts.update(overrides) + return [fact for fact in facts.values() if fact is not None] + + +def _stmt(ddl): + return AppliedStatement(statement=ddl, phase=PHASE_DDL) + + +def _state(objects, constraint_types=ALL_TYPES, platform="duckdb"): + return IntrospectedState(platform=platform, objects=list(objects), constraint_types=constraint_types) + + +def _receipt(ddl, state): + ledger = AppliedTuningLedger() + ledger.record(ddl, PHASE_DDL) + return corroborate(ledger, state) + + +def _verdicts(receipt): + return {entry.constraint_type: entry.verdict for entry in receipt.entries if entry.kind == KIND_CONSTRAINT} + + +class TestClassification: + def test_table_level_constraints_become_typed_intents(self): + klass, intents = _classify(_stmt(ORDERS_DDL)) + assert klass == "verifiable" + assert [(i.kind, i.constraint_type, i.table, i.columns) for i in intents] == [ + (KIND_CONSTRAINT, CONSTRAINT_PRIMARY_KEY, "orders", ("o_orderkey",)), + (KIND_CONSTRAINT, CONSTRAINT_UNIQUE, "orders", ("o_custkey", "o_comment")), + (KIND_CONSTRAINT, CONSTRAINT_FOREIGN_KEY, "orders", ("o_custkey",)), + ] + assert (intents[2].referenced_table, intents[2].referenced_columns) == ("customer", ("c_custkey",)) + assert intents[0].referenced_table is None and intents[0].referenced_columns == () + + def test_column_level_constraints_become_typed_intents(self): + ddl = ( + "CREATE TABLE part (p_partkey INTEGER PRIMARY KEY, p_name VARCHAR UNIQUE NOT NULL, " + "p_suppkey INTEGER REFERENCES supplier (s_suppkey), p_price DECIMAL(15,2) NOT NULL)" + ) + _klass, intents = _classify(_stmt(ddl)) + assert [(i.constraint_type, i.columns, i.referenced_table, i.referenced_columns) for i in intents] == [ + (CONSTRAINT_PRIMARY_KEY, ("p_partkey",), None, ()), + (CONSTRAINT_UNIQUE, ("p_name",), None, ()), + (CONSTRAINT_FOREIGN_KEY, ("p_suppkey",), "supplier", ("s_suppkey",)), + ] + + def test_quoted_identifiers_keep_their_names(self): + ddl = 'CREATE TABLE t ("Key A" INTEGER, "b" INTEGER, PRIMARY KEY ("Key A", "b"))' + _klass, intents = _classify(_stmt(ddl)) + assert [i.columns for i in intents] == [("key a", "b")] + + @pytest.mark.parametrize( + "ddl", + [ + 'CREATE TABLE "Region" ("r_key" INT NOT NULL, PRIMARY KEY ("r_key"))', + 'CREATE TABLE IF NOT EXISTS "Region" ("r_key" INT NOT NULL, PRIMARY KEY ("r_key"))', + ], + ) + def test_quoted_table_name_binds_constraints_to_that_table(self, ddl): + klass, intents = _classify(_stmt(ddl)) + assert klass == "verifiable" + assert [(i.table, i.columns) for i in intents] == [("region", ("r_key",))] + + def test_quoted_table_name_does_not_widen_layout_clauses(self): + ddl = 'CREATE TABLE "t" ("a" Int64, PRIMARY KEY ("a")) ENGINE = MergeTree() ORDER BY ("a")' + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + + def test_trailing_semicolon_alone_is_not_a_second_statement(self): + klass, intents = _classify(_stmt("CREATE TABLE t (a INTEGER PRIMARY KEY); \n")) + assert klass == "verifiable" + assert [i.columns for i in intents] == [("a",)] + + def test_quoted_table_name_with_a_space_binds_to_that_exact_table(self): + ddl = 'CREATE TABLE "my t" (a INTEGER, PRIMARY KEY (a))' + assert _receipt(ddl, _state([_fact(CONSTRAINT_PRIMARY_KEY, ["a"], table="my t")])).corroborated is True + for other in ("my", "myt", " my t"): + assert _receipt(ddl, _state([_fact(CONSTRAINT_PRIMARY_KEY, ["a"], table=other)])).corroborated is False + + def test_quoted_column_with_inline_key_uses_the_quoted_name(self): + ddl = 'CREATE TABLE t ("Unique" INTEGER PRIMARY KEY)' + _klass, intents = _classify(_stmt(ddl)) + assert [(i.constraint_type, i.columns) for i in intents] == [(CONSTRAINT_PRIMARY_KEY, ("unique",))] + + def test_keywords_in_literals_and_comments_create_no_intent(self): + ddl = ( + "CREATE TABLE t (a INTEGER DEFAULT 'PRIMARY KEY (b)', -- UNIQUE (a)\n" + "b VARCHAR /* FOREIGN KEY (b) REFERENCES x(y) */, PRIMARY KEY (a))" + ) + _klass, intents = _classify(_stmt(ddl)) + assert [(i.constraint_type, i.columns) for i in intents] == [(CONSTRAINT_PRIMARY_KEY, ("a",))] + + def test_engine_primary_key_outside_the_column_list_is_not_a_constraint(self): + ddl = "CREATE TABLE t (a Int64, b Int64) ENGINE = MergeTree() ORDER BY (a, b) PRIMARY KEY (a)" + _klass, intents = _classify(_stmt(ddl)) + assert [(i.kind, i.columns) for i in intents] == [(KIND_SORT_KEY, ("a", "b"))] + + def test_create_table_with_only_not_null_columns_has_no_rule(self): + klass, intents = _classify(_stmt("CREATE TABLE t (a INTEGER NOT NULL)")) + assert (klass, intents) == (UNVERIFIABLE, []) + + @pytest.mark.parametrize( + "ddl", + [ + "CREATE TABLE t (a INTEGER, CHECK (a > 0), PRIMARY KEY (a))", + "CREATE TABLE t (a INTEGER CHECK (a > 0) PRIMARY KEY)", + "CREATE TABLE t (a INTEGER, CONSTRAINT pk_t PRIMARY KEY (a))", + "CREATE TABLE t (a INTEGER, FOREIGN KEY (a) REFERENCES u(b) ON DELETE CASCADE)", + "CREATE TABLE t (a INTEGER REFERENCES u(b) ON UPDATE RESTRICT)", + "CREATE TABLE t (a INTEGER REFERENCES u)", + "CREATE TABLE t (a INTEGER, FOREIGN KEY (a) REFERENCES u)", + "CREATE TABLE t (a INTEGER, PRIMARY KEY ())", + "CREATE TABLE t (a INTEGER, PRIMARY KEY a)", + "CREATE TABLE t (a INTEGER, UNIQUE (a) NULLS NOT DISTINCT)", + "CREATE TABLE t (a INTEGER, PRIMARY KEY (a)", + "CREATE TABLE t (a INTEGER PRIMARY KEY); CREATE TABLE u (b INTEGER CHECK (b > 0));", + "CREATE TABLE t (a INTEGER PRIMARY KEY);\nALTER TABLE t ADD COLUMN c INTEGER", + ], + ids=[ + "check", + "inline-check", + "named-constraint", + "on-delete", + "inline-on-update", + "inline-references-without-columns", + "references-without-columns", + "empty-key", + "unparenthesized-key", + "trailing-clause", + "unbalanced-body", + "second-statement", + "trailing-alter", + ], + ) + def test_constraint_text_that_is_not_fully_parsed_blocks_the_statement(self, ddl): + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + + def test_unparsed_constraint_also_blocks_a_parsed_sort_key_in_the_same_statement(self): + ddl = "CREATE TABLE t (a Int64, CHECK (a > 0)) ENGINE = MergeTree() ORDER BY (a)" + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + + +class TestPositiveMatches: + @pytest.mark.parametrize("constraint_type", [CONSTRAINT_PRIMARY_KEY, CONSTRAINT_UNIQUE, CONSTRAINT_FOREIGN_KEY]) + def test_each_constraint_kind_corroborates(self, constraint_type): + receipt = _receipt(ORDERS_DDL, _state(_orders_facts())) + assert _verdicts(receipt)[constraint_type] == CORROBORATED + + def test_all_constraints_corroborated_verifies_the_statement(self): + receipt = _receipt(ORDERS_DDL, _state(_orders_facts())) + assert receipt.corroborated is True + assert receipt.summary == {"corroborated": 3, "gate_relevant_total": 3} + + def test_identifier_case_and_quotes_are_normalized(self): + ddl = 'CREATE TABLE Orders (A INTEGER, B INTEGER, FOREIGN KEY ("A", B) REFERENCES "Customer"(X, "Y"))' + fact = _fact(CONSTRAINT_FOREIGN_KEY, ["a", "b"], "ORDERS", "customer", ["x", "y"]) + assert _verdicts(_receipt(ddl, _state([fact]))) == {CONSTRAINT_FOREIGN_KEY: CORROBORATED} + + +class TestNegativeMatches: + def test_primary_key_on_different_columns_is_mismatch(self): + state = _state(_orders_facts(**{CONSTRAINT_PRIMARY_KEY: _fact(CONSTRAINT_PRIMARY_KEY, ["o_custkey"])})) + receipt = _receipt(ORDERS_DDL, state) + assert _verdicts(receipt)[CONSTRAINT_PRIMARY_KEY] == MISMATCH + assert receipt.corroborated is False + entry = next(e for e in receipt.entries if e.constraint_type == CONSTRAINT_PRIMARY_KEY) + assert entry.diff == "expected ['o_orderkey'] != observed ['o_custkey']" + + def test_foreign_key_with_same_child_columns_but_different_referenced_table_is_mismatch(self): + wrong = _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders", "supplier", ["c_custkey"]) + receipt = _receipt(ORDERS_DDL, _state(_orders_facts(**{CONSTRAINT_FOREIGN_KEY: wrong}))) + assert _verdicts(receipt)[CONSTRAINT_FOREIGN_KEY] == MISMATCH + assert receipt.corroborated is False + entry = next(e for e in receipt.entries if e.constraint_type == CONSTRAINT_FOREIGN_KEY) + assert "references customer['c_custkey']" in entry.diff + assert "references supplier['c_custkey']" in entry.diff + + def test_foreign_key_with_different_referenced_columns_is_mismatch(self): + wrong = _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders", "customer", ["c_nationkey"]) + receipt = _receipt(ORDERS_DDL, _state(_orders_facts(**{CONSTRAINT_FOREIGN_KEY: wrong}))) + assert _verdicts(receipt)[CONSTRAINT_FOREIGN_KEY] == MISMATCH + + def test_foreign_key_with_different_child_columns_is_mismatch(self): + wrong = _fact(CONSTRAINT_FOREIGN_KEY, ["o_orderkey"], "orders", "customer", ["c_custkey"]) + receipt = _receipt(ORDERS_DDL, _state(_orders_facts(**{CONSTRAINT_FOREIGN_KEY: wrong}))) + assert _verdicts(receipt)[CONSTRAINT_FOREIGN_KEY] == MISMATCH + + def test_key_column_order_is_significant(self): + reordered = _fact(CONSTRAINT_UNIQUE, ["o_comment", "o_custkey"]) + receipt = _receipt(ORDERS_DDL, _state(_orders_facts(**{CONSTRAINT_UNIQUE: reordered}))) + assert _verdicts(receipt)[CONSTRAINT_UNIQUE] == MISMATCH + + def test_unique_fact_does_not_stand_in_for_a_primary_key(self): + unique_instead = _fact(CONSTRAINT_UNIQUE, ["o_orderkey"]) + state = _state(_orders_facts(**{CONSTRAINT_PRIMARY_KEY: None}) + [unique_instead]) + receipt = _receipt(ORDERS_DDL, state) + assert _verdicts(receipt)[CONSTRAINT_PRIMARY_KEY] == ABSENT + + def test_fact_on_another_table_does_not_corroborate(self): + elsewhere = _fact(CONSTRAINT_PRIMARY_KEY, ["o_orderkey"], table="orders_archive") + state = _state(_orders_facts(**{CONSTRAINT_PRIMARY_KEY: None}) + [elsewhere]) + receipt = _receipt(ORDERS_DDL, state) + entry = next(e for e in receipt.entries if e.constraint_type == CONSTRAINT_PRIMARY_KEY) + assert (entry.verdict, entry.reason) == (ABSENT, "no PRIMARY KEY constraint found in catalog") + + def test_index_fact_does_not_stand_in_for_a_constraint(self): + index = IntrospectedObject(kind="index", table="orders", columns=("o_orderkey",)) + receipt = _receipt("CREATE TABLE orders (o_orderkey INTEGER, PRIMARY KEY (o_orderkey))", _state([index])) + assert _verdicts(receipt) == {CONSTRAINT_PRIMARY_KEY: ABSENT} + + +class TestUnverifiable: + @pytest.mark.parametrize( + ("referenced_table", "referenced_columns"), + [(None, ["c_custkey"]), ("customer", []), (None, [])], + ) + def test_foreign_key_fact_without_referenced_fields_is_unverifiable(self, referenced_table, referenced_columns): + bare = _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders", referenced_table, referenced_columns) + receipt = _receipt(ORDERS_DDL, _state(_orders_facts(**{CONSTRAINT_FOREIGN_KEY: bare}))) + entry = next(e for e in receipt.entries if e.constraint_type == CONSTRAINT_FOREIGN_KEY) + assert entry.verdict == UNVERIFIABLE + assert entry.reason == "catalog foreign key fact has no referenced table or columns" + assert receipt.corroborated is False + + def test_complete_fact_still_wins_over_an_incomplete_duplicate(self): + bare = _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders") + receipt = _receipt(ORDERS_DDL, _state([bare, *_orders_facts()])) + assert _verdicts(receipt)[CONSTRAINT_FOREIGN_KEY] == CORROBORATED + + @pytest.mark.parametrize("platform", ["snowflake", "clickhouse"]) + def test_platform_without_a_constraint_source_stays_unverifiable(self, platform): + state = _state(_orders_facts(), constraint_types=frozenset(), platform=platform) + receipt = _receipt(ORDERS_DDL, state) + assert set(_verdicts(receipt).values()) == {UNVERIFIABLE} + assert receipt.corroborated is False + assert receipt.entries[0].reason == f"{platform} introspection reads no PRIMARY KEY constraints" + + def test_constraint_type_the_introspector_does_not_read_stays_unverifiable(self): + state = _state(_orders_facts(), constraint_types=frozenset({CONSTRAINT_PRIMARY_KEY, CONSTRAINT_UNIQUE})) + receipt = _receipt(ORDERS_DDL, state) + assert _verdicts(receipt) == { + CONSTRAINT_PRIMARY_KEY: CORROBORATED, + CONSTRAINT_UNIQUE: CORROBORATED, + CONSTRAINT_FOREIGN_KEY: UNVERIFIABLE, + } + assert receipt.corroborated is False + + def test_constraint_next_to_a_corroborated_sort_key_still_blocks_without_a_source(self): + ddl = "CREATE TABLE t (a Int64, PRIMARY KEY (a)) ENGINE = MergeTree() ORDER BY (a)" + sort_key = IntrospectedObject(kind=KIND_SORT_KEY, table="t", columns=("a",)) + receipt = _receipt(ddl, IntrospectedState(platform="clickhouse", objects=[sort_key])) + assert [(e.kind, e.verdict) for e in receipt.entries] == [ + (KIND_SORT_KEY, CORROBORATED), + (KIND_CONSTRAINT, UNVERIFIABLE), + ] + assert receipt.corroborated is False + + def test_degraded_state_marks_constraints_unverifiable(self): + state = IntrospectedState(platform="duckdb", error="catalog read failed", constraint_types=ALL_TYPES) + receipt = _receipt(ORDERS_DDL, state) + assert set(_verdicts(receipt).values()) == {UNVERIFIABLE} + assert {e.reason for e in receipt.entries} == {"introspection degraded"} + + def test_match_object_applies_the_constraint_rules(self): + _klass, intents = _classify(_stmt(ORDERS_DDL)) + fk = intents[2] + wrong = _fact(CONSTRAINT_FOREIGN_KEY, ["o_custkey"], "orders", "supplier", ["c_custkey"]) + assert _match_object(fk, _state([wrong]))[0] == MISMATCH + assert _match_object(fk, _state([wrong], constraint_types=frozenset()))[0] == UNVERIFIABLE + + +class TestPayload: + def test_entries_and_observed_facts_carry_the_constraint_type(self): + payload = _receipt(ORDERS_DDL, _state(_orders_facts())).to_payload() + assert [entry["constraint_type"] for entry in payload["entries"]] == [ + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_UNIQUE, + CONSTRAINT_FOREIGN_KEY, + ] + assert {item["constraint_type"] for item in payload["observed"]} == ALL_TYPES + + def test_non_constraint_entries_have_no_constraint_type_key(self): + ddl = "CREATE TABLE t (a Int64) ENGINE = MergeTree() ORDER BY (a)" + sort_key = IntrospectedObject(kind=KIND_SORT_KEY, table="t", columns=("a",)) + payload = _receipt(ddl, IntrospectedState(platform="clickhouse", objects=[sort_key])).to_payload() + assert "constraint_type" not in payload["entries"][0] + assert "constraint_type" not in payload["observed"][0] + + +class TestPhysicalIdentifiers: + _DDL = ( + 'CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, " a " INTEGER, b INTEGER, ' + "UNIQUE (a, b), FOREIGN KEY (a) REFERENCES u (k))" + ) + + def _verdict(self, constraint_type, fact): + return _verdicts(_receipt(self._DDL, _state([fact])))[constraint_type] + + def test_padded_quoted_identifier_keeps_its_whitespace(self): + _klass, intents = _classify(_stmt('CREATE TABLE t (" a " INTEGER, PRIMARY KEY (" a "))')) + assert [i.columns for i in intents] == [(" a ",)] + + def test_primary_key_on_a_padded_column_does_not_corroborate_a_key_on_a(self): + assert self._verdict(CONSTRAINT_PRIMARY_KEY, _fact(CONSTRAINT_PRIMARY_KEY, [" a "], table="t")) == MISMATCH + assert self._verdict(CONSTRAINT_PRIMARY_KEY, _fact(CONSTRAINT_PRIMARY_KEY, ["A"], table="T")) == CORROBORATED + + def test_unique_on_a_padded_column_does_not_corroborate_unique_on_a(self): + padded = _fact(CONSTRAINT_UNIQUE, [" a ", "b"], table="t") + assert self._verdict(CONSTRAINT_UNIQUE, padded) == MISMATCH + assert self._verdict(CONSTRAINT_UNIQUE, _fact(CONSTRAINT_UNIQUE, ["a", "b"], table="t")) == CORROBORATED + + @pytest.mark.parametrize( + ("columns", "referenced_table", "referenced_columns"), + [([" a "], "u", ["k"]), (["a"], "u", [" k "]), (["a"], " u ", ["k"]), (["a"], '"u"', ["k"])], + ids=["padded-child", "padded-referenced-column", "padded-referenced-table", "quote-characters-in-name"], + ) + def test_foreign_key_identifiers_compare_by_physical_contents(self, columns, referenced_table, referenced_columns): + fact = _fact(CONSTRAINT_FOREIGN_KEY, columns, "t", referenced_table, referenced_columns) + assert self._verdict(CONSTRAINT_FOREIGN_KEY, fact) == MISMATCH + + def test_foreign_key_with_exact_identifiers_corroborates(self): + fact = _fact(CONSTRAINT_FOREIGN_KEY, ["A"], "t", "U", ["K"]) + assert self._verdict(CONSTRAINT_FOREIGN_KEY, fact) == CORROBORATED + + def test_padded_table_name_does_not_corroborate_a_key_on_t(self): + assert self._verdict(CONSTRAINT_PRIMARY_KEY, _fact(CONSTRAINT_PRIMARY_KEY, ["a"], table=" t ")) == ABSENT + + def test_case_folding_is_ascii_only(self): + ddl = 'CREATE TABLE t ("É" INTEGER, PRIMARY KEY ("É"))' + assert _verdicts(_receipt(ddl, _state([_fact(CONSTRAINT_PRIMARY_KEY, ["é"], table="t")]))) == { + CONSTRAINT_PRIMARY_KEY: MISMATCH + } + + def test_escaped_quote_is_unescaped(self): + _klass, intents = _classify(_stmt('CREATE TABLE t ("a""b" INTEGER PRIMARY KEY)')) + assert [i.columns for i in intents] == [('a"b',)] + + @pytest.mark.parametrize( + "ddl", + [ + "CREATE TABLE main.t (a INTEGER, PRIMARY KEY (a))", + 'CREATE TABLE "main"."t" (a INTEGER, PRIMARY KEY (a))', + "CREATE TABLE t (a INTEGER, PRIMARY KEY (`a`))", + "CREATE TABLE t (a INTEGER, PRIMARY KEY ([a]))", + "CREATE TABLE t (a INTEGER, PRIMARY KEY (a b))", + "CREATE TABLE t (a INTEGER, PRIMARY KEY (a,))", + 'CREATE TABLE t (a INTEGER, PRIMARY KEY (""))', + "CREATE TABLE t (a INTEGER, FOREIGN KEY (a) REFERENCES main.u (k))", + "CREATE TABLE t (a INTEGER REFERENCES u (k + 1))", + "CREATE TABLE t (`a` INTEGER PRIMARY KEY)", + ], + ids=[ + "qualified-table", + "quoted-qualified-table", + "backtick-column", + "bracket-column", + "two-tokens", + "dangling-comma", + "empty-quoted", + "qualified-referenced-table", + "expression-reference", + "backtick-inline-column", + ], + ) + def test_identifiers_that_do_not_parse_unambiguously_block_the_statement(self, ddl): + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + + def test_index_and_sort_key_columns_keep_their_existing_normalization(self): + _klass, [index] = _classify(_stmt('CREATE INDEX i ON t (" A ", B)')) + assert index.columns == ("a", "b") + _klass, [sort_key] = _classify(_stmt("CREATE TABLE t (a Int64) ENGINE = MergeTree() ORDER BY (A, b)")) + assert sort_key.columns == ("a", "b") + + +class TestLiteralsTheScannerCannotMask: + @pytest.mark.parametrize( + "ddl", + [ + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT $$)$$ CHECK (length(b) > 0))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT $tag$)$tag$ CHECK (length(b) > 0))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT E'\\')' CHECK (length(b) > 0))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT e'x' )", + "CREATE TABLE t (a INTEGER PRIMARY KEY /* x /* y */ ) */, b INTEGER CHECK (b > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR[] DEFAULT [']', ')', 'x'] " + "CHECK (length(b) > 0))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b INTEGER[] DEFAULT [1] CHECK (len(b) > 0))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT ']' CHECK (b <> ''))", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT 1 ` 2 CHECK (b <> ''))", + "CREATE TABLE IF NOT EXISTS t (a Int64, b String DEFAULT '\\')\\'', CONSTRAINT positive " + "CHECK a > 0) ENGINE = MergeTree() ORDER BY (a)", + "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT 'it\\'s ) CHECK (b <> ''))", + 'CREATE TABLE t (a Int64, "b\\")\\"x" Int64, CONSTRAINT positive CHECK a > 0) ' + "ENGINE = MergeTree() ORDER BY (a)", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT $\u00e9$)$\u00e9$ " + "CHECK (length(b) > 0))", + ], + ids=[ + "dollar", + "tagged-dollar", + "escape-string", + "lowercase-escape-string", + "nested-block-comment", + "list-literal-with-quoted-brackets", + "list-literal", + "bracket-in-string", + "backtick-operator", + "backslash-escaped-clickhouse-default", + "backslash-escaped-quote", + "backslash-escaped-double-quote", + "non-ascii-dollar-tag", + ], + ) + def test_statement_with_an_unmaskable_literal_is_unverifiable(self, ddl): + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + receipt = _receipt(ddl, _state([_fact(CONSTRAINT_PRIMARY_KEY, ["a"], table="t")])) + assert receipt.corroborated is False + + def test_unclosed_block_comment_is_unverifiable(self): + ddl = "CREATE TABLE t (a INTEGER PRIMARY KEY) /* unclosed" + assert _classify(_stmt(ddl)) == (UNVERIFIABLE, []) + receipt = _receipt(ddl, _state([_fact(CONSTRAINT_PRIMARY_KEY, ["a"], table="t")])) + assert receipt.corroborated is False + + def test_brackets_without_constraint_keywords_keep_layout_intents(self): + klass, intents = _classify(_stmt("CREATE TABLE t (`a` Array(Int64) DEFAULT [1]) ORDER BY (a)")) + assert klass == "verifiable" + assert [(i.kind, i.columns) for i in intents] == [(KIND_SORT_KEY, ("a",))] + + def test_masked_comments_and_strings_still_parse(self): + ddl = ( + "CREATE TABLE t (a INTEGER PRIMARY KEY, -- ) CHECK (\nb VARCHAR DEFAULT 'it''s )' /* ) */, \"c)\" INTEGER)" + ) + klass, intents = _classify(_stmt(ddl)) + assert klass == "verifiable" + assert [(i.constraint_type, i.columns) for i in intents] == [(CONSTRAINT_PRIMARY_KEY, ("a",))] + + def test_backslash_away_from_a_quote_still_parses(self): + ddl = "CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT 'C:\\data')" + klass, intents = _classify(_stmt(ddl)) + assert klass == "verifiable" + assert [(i.constraint_type, i.columns) for i in intents] == [(CONSTRAINT_PRIMARY_KEY, ("a",))] + + def test_backslash_away_from_a_double_quote_still_parses(self): + ddl = 'CREATE TABLE t ("a\\b" INTEGER PRIMARY KEY)' + klass, intents = _classify(_stmt(ddl)) + assert klass == "verifiable" + assert [(i.constraint_type, i.columns) for i in intents] == [(CONSTRAINT_PRIMARY_KEY, ("a\\b",))] + + +class TestConstraintTables: + def test_constraint_tables_use_the_physical_table_name(self): + ledger = AppliedTuningLedger() + ledger.record('CREATE TABLE "my t" (a INTEGER PRIMARY KEY)', PHASE_DDL) + ledger.record('CREATE TABLE "q""T" (a INTEGER, UNIQUE (a))', PHASE_DDL) + ledger.record("CREATE INDEX i ON other (a)", PHASE_DDL) + ledger.record("CREATE TABLE plain (a INTEGER NOT NULL)", PHASE_DDL) + assert constraint_tables(ledger) == {"my t", 'q"t'} diff --git a/tests/unit/platforms/test_duckdb_introspection.py b/tests/unit/platforms/test_duckdb_introspection.py index e42833cbc9..0386c1a4c0 100644 --- a/tests/unit/platforms/test_duckdb_introspection.py +++ b/tests/unit/platforms/test_duckdb_introspection.py @@ -4,24 +4,44 @@ from __future__ import annotations +from pathlib import Path + import duckdb import pytest +from benchbox import SSB +from benchbox.cli.config import ConfigManager +from benchbox.core.tpch.benchmark import TPCHBenchmark from benchbox.core.tuning.applied_ledger import ( APPLIED_UNVERIFIED, APPLIED_VERIFIED, PHASE_DDL, PHASE_POST_LOAD, + PHASE_SESSION, AppliedTuningLedger, recording_connection, ) from benchbox.core.tuning.interface import TableTuning, TuningColumn, UnifiedTuningConfiguration -from benchbox.core.tuning.introspection import ABSENT, CORROBORATED, KIND_INDEX, corroborate +from benchbox.core.tuning.introspection import ( + ABSENT, + CONSTRAINT_FOREIGN_KEY, + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_UNIQUE, + CORROBORATED, + KIND_CONSTRAINT, + KIND_INDEX, + MISMATCH, + UNVERIFIABLE, + corroborate, +) +from benchbox.platforms import duckdb_introspection from benchbox.platforms.duckdb import DuckDBAdapter from benchbox.platforms.duckdb_introspection import DuckDBTuningIntrospector pytestmark = [pytest.mark.unit, pytest.mark.fast] +TEMPLATES = Path(__file__).resolve().parents[3] / "benchbox" / "core" / "tuning" / "templates" / "duckdb" + def _con_with_indexes() -> duckdb.DuckDBPyConnection: con = duckdb.connect(":memory:") @@ -221,3 +241,350 @@ def test_dry_run_does_not_recreate_index(self): adapter.apply_ctas_sort("LINEITEM", config, con) assert not [op for op in adapter._applied_layout_operations if op["mechanism"] == "sort_index"] + + +_CONSTRAINED_DDL = ( + "CREATE TABLE customer (c_custkey INTEGER PRIMARY KEY, c_name VARCHAR NOT NULL)", + "CREATE TABLE orders (o_orderkey INTEGER NOT NULL, o_custkey INTEGER, o_note VARCHAR, " + "o_total INTEGER CHECK (o_total >= 0), PRIMARY KEY (o_orderkey), UNIQUE (o_custkey, o_note), " + "FOREIGN KEY (o_custkey) REFERENCES customer(c_custkey))", +) +_PARSED_ORDERS_DDL = ( + "CREATE TABLE orders (o_orderkey INTEGER NOT NULL, o_custkey INTEGER, o_note VARCHAR, " + "PRIMARY KEY (o_orderkey), UNIQUE (o_custkey, o_note), FOREIGN KEY (o_custkey) REFERENCES customer(c_custkey))" +) + + +def _constrained_connection() -> duckdb.DuckDBPyConnection: + con = duckdb.connect(":memory:") + for statement in _CONSTRAINED_DDL: + con.execute(statement) + return con + + +def _constraint_facts(state) -> dict[tuple[str, str], object]: + return {(obj.table.lower(), obj.constraint_type): obj for obj in state.objects if obj.kind == KIND_CONSTRAINT} + + +class TestDuckDBConstraintFacts: + def test_reads_primary_unique_and_foreign_keys_as_structured_facts(self): + con = _constrained_connection() + state = DuckDBTuningIntrospector().introspect(con, _ledger_for(_CONSTRAINED_DDL[0], _PARSED_ORDERS_DDL)) + + assert state.error is None + assert state.constraint_types == {CONSTRAINT_PRIMARY_KEY, CONSTRAINT_UNIQUE, CONSTRAINT_FOREIGN_KEY} + facts = _constraint_facts(state) + assert set(facts) == { + ("customer", CONSTRAINT_PRIMARY_KEY), + ("orders", CONSTRAINT_PRIMARY_KEY), + ("orders", CONSTRAINT_UNIQUE), + ("orders", CONSTRAINT_FOREIGN_KEY), + } + assert facts[("orders", CONSTRAINT_UNIQUE)].columns == ("o_custkey", "o_note") + fk = facts[("orders", CONSTRAINT_FOREIGN_KEY)] + assert (fk.columns, fk.referenced_table, fk.referenced_columns) == (("o_custkey",), "customer", ("c_custkey",)) + assert fk.evidence["referenced_column_names"] == ["c_custkey"] + + def test_quoted_unicode_table_name_constraints_are_read(self): + con = duckdb.connect(":memory:") + ddl = 'CREATE TABLE "É" (id INTEGER PRIMARY KEY)' + con.execute(ddl) + ledger = _ledger_for(ddl) + + state = DuckDBTuningIntrospector().introspect(con, ledger) + + assert state.error is None + [fact] = [obj for obj in state.objects if obj.kind == KIND_CONSTRAINT] + assert (fact.table, fact.columns) == ("É", ("id",)) + assert corroborate(ledger, state).corroborated + + def test_tables_whose_statement_yields_no_constraint_intent_are_not_read(self): + con = _constrained_connection() + state = DuckDBTuningIntrospector().introspect(con, _ledger_for(*_CONSTRAINED_DDL)) + assert {obj.table for obj in state.objects} == {"customer"} + + def test_not_null_and_check_rows_are_not_read(self): + con = _constrained_connection() + state = DuckDBTuningIntrospector().introspect(con, _ledger_for(_CONSTRAINED_DDL[0], _PARSED_ORDERS_DDL)) + assert {obj.constraint_type for obj in state.objects} <= { + CONSTRAINT_PRIMARY_KEY, + CONSTRAINT_UNIQUE, + CONSTRAINT_FOREIGN_KEY, + } + + def test_bounded_to_ledger_tables_in_the_current_schema(self): + con = _constrained_connection() + con.execute("CREATE TABLE part (p_partkey INTEGER PRIMARY KEY)") + con.execute("CREATE SCHEMA other") + con.execute("CREATE TABLE other.nation (n_nationkey INTEGER PRIMARY KEY)") + + state = DuckDBTuningIntrospector().introspect( + con, _ledger_for(_CONSTRAINED_DDL[0], "CREATE TABLE nation (n_nationkey INTEGER PRIMARY KEY)") + ) + + assert state.error is None + assert [(obj.table, obj.columns) for obj in state.objects] == [("customer", ("c_custkey",))] + + def test_ledger_statements_corroborate_against_the_catalog(self): + con = _constrained_connection() + ledger = _ledger_for(_CONSTRAINED_DDL[0], _PARSED_ORDERS_DDL) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert receipt.corroborated is True + assert receipt.summary == {"corroborated": 4, "gate_relevant_total": 4} + + def test_check_constraint_statement_stays_unverifiable(self): + con = _constrained_connection() + ledger = _ledger_for(*_CONSTRAINED_DDL) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries] == [CORROBORATED, UNVERIFIABLE] + + def test_foreign_key_to_another_table_in_the_catalog_is_mismatch(self): + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE customer (c_custkey INTEGER PRIMARY KEY)") + con.execute("CREATE TABLE supplier (s_suppkey INTEGER PRIMARY KEY)") + con.execute("CREATE TABLE orders (o_custkey INTEGER, FOREIGN KEY (o_custkey) REFERENCES supplier(s_suppkey))") + ledger = _ledger_for( + "CREATE TABLE orders (o_custkey INTEGER, FOREIGN KEY (o_custkey) REFERENCES customer(c_custkey))" + ) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + [entry] = receipt.entries + assert (entry.constraint_type, entry.verdict) == (CONSTRAINT_FOREIGN_KEY, MISMATCH) + + def test_missing_referenced_catalog_columns_leave_foreign_keys_unverifiable(self): + con = _constrained_connection() + + class _WithoutReferences: + def execute(self, sql, params=None): + if "duckdb_constraints()" in sql: + sql = sql.replace("SELECT *", "SELECT * EXCLUDE (referenced_table, referenced_column_names)") + return con.execute(sql, params) if params is not None else con.execute(sql) + + ledger = _ledger_for(_CONSTRAINED_DDL[0], _PARSED_ORDERS_DDL) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(_WithoutReferences(), ledger)) + verdicts = {entry.constraint_type: entry.verdict for entry in receipt.entries if entry.table == "orders"} + assert verdicts == { + CONSTRAINT_PRIMARY_KEY: CORROBORATED, + CONSTRAINT_UNIQUE: CORROBORATED, + CONSTRAINT_FOREIGN_KEY: UNVERIFIABLE, + } + assert receipt.corroborated is False + + def test_constraint_read_failure_degrades_the_whole_state(self): + con = _constrained_connection() + + class _ConstraintsFail: + def execute(self, sql, params=None): + if "duckdb_constraints()" in sql: + raise RuntimeError("constraint catalog exploded") + return con.execute(sql, params) if params is not None else con.execute(sql) + + state = DuckDBTuningIntrospector().introspect(_ConstraintsFail(), _ledger_for(*_CONSTRAINED_DDL)) + assert state.error is not None and "duckdb_constraints" in state.error + assert state.objects == [] + + def test_constraint_row_bound_marks_the_state_truncated(self, monkeypatch): + monkeypatch.setattr(duckdb_introspection, "_MAX_CONSTRAINT_ROWS", 2) + con = _constrained_connection() + ledger = _ledger_for(_CONSTRAINED_DDL[0], _PARSED_ORDERS_DDL) + state = DuckDBTuningIntrospector().introspect(con, ledger) + assert state.truncated is True + assert corroborate(ledger, state).corroborated is False + + +def _template_config(benchmark_name: str) -> UnifiedTuningConfiguration: + return ConfigManager().load_unified_tuning_config(TEMPLATES / f"{benchmark_name}_tuned.yaml", platform="duckdb") + + +_SCHEMA_BENCHMARKS = {"tpch": TPCHBenchmark, "ssb": SSB} + + +@pytest.mark.parametrize("benchmark_name", sorted(_SCHEMA_BENCHMARKS)) +def test_shipped_template_schema_constraints_all_corroborate(benchmark_name, tmp_path): + con = duckdb.connect(":memory:") + adapter = DuckDBAdapter() + adapter.unified_tuning_configuration = _template_config(benchmark_name) + ledger = AppliedTuningLedger() + schema_benchmark = _SCHEMA_BENCHMARKS[benchmark_name](scale_factor=0.01, output_dir=tmp_path) + + adapter.create_schema(schema_benchmark, recording_connection(con, ledger, PHASE_DDL)) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + + constraint_entries = [entry for entry in receipt.entries if entry.kind == KIND_CONSTRAINT] + assert constraint_entries + assert {entry.verdict for entry in constraint_entries} == {CORROBORATED} + assert {entry.verdict for entry in receipt.entries} == {CORROBORATED} + + +class TestDuckDBPhysicalConstraintIdentifiers: + def _receipt(self, existing: list[str], executed: str): + con = duckdb.connect(":memory:") + con.execute('CREATE TABLE u (k INTEGER PRIMARY KEY, " k " INTEGER UNIQUE)') + for statement in existing: + con.execute(statement) + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute(executed) + return corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + + def test_catalog_keeps_whitespace_in_constraint_columns(self): + con = duckdb.connect(":memory:") + statement = 'CREATE TABLE t (a INTEGER, " a " INTEGER, PRIMARY KEY (" a "))' + con.execute(statement) + state = DuckDBTuningIntrospector().introspect(con, _ledger_for(statement)) + assert [obj.columns for obj in state.objects if obj.kind == KIND_CONSTRAINT] == [(" a ",)] + + def test_existing_key_on_a_padded_column_does_not_corroborate_a_requested_primary_key(self): + receipt = self._receipt( + ['CREATE TABLE t (a INTEGER, " a " INTEGER, PRIMARY KEY (" a "))'], + 'CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, " a " INTEGER)', + ) + assert [(e.constraint_type, e.verdict) for e in receipt.entries] == [(CONSTRAINT_PRIMARY_KEY, MISMATCH)] + assert receipt.corroborated is False + + def test_existing_unique_on_a_padded_column_does_not_corroborate_a_requested_unique(self): + receipt = self._receipt( + ['CREATE TABLE t (a INTEGER, " a " INTEGER UNIQUE)'], + 'CREATE TABLE IF NOT EXISTS t (a INTEGER UNIQUE, " a " INTEGER)', + ) + assert [(e.constraint_type, e.verdict) for e in receipt.entries] == [(CONSTRAINT_UNIQUE, MISMATCH)] + + def test_existing_foreign_key_from_a_padded_column_does_not_corroborate(self): + receipt = self._receipt( + ['CREATE TABLE t (a INTEGER, " a " INTEGER REFERENCES u (k))'], + 'CREATE TABLE IF NOT EXISTS t (a INTEGER REFERENCES u (k), " a " INTEGER)', + ) + assert [(e.constraint_type, e.verdict) for e in receipt.entries] == [(CONSTRAINT_FOREIGN_KEY, MISMATCH)] + + def test_existing_foreign_key_to_a_padded_referenced_column_does_not_corroborate(self): + receipt = self._receipt( + ['CREATE TABLE t (a INTEGER REFERENCES u (" k "))'], + "CREATE TABLE IF NOT EXISTS t (a INTEGER REFERENCES u (k))", + ) + assert [(e.constraint_type, e.verdict) for e in receipt.entries] == [(CONSTRAINT_FOREIGN_KEY, MISMATCH)] + + def test_matching_physical_identifiers_corroborate(self): + receipt = self._receipt( + [], + 'CREATE TABLE t (A INTEGER PRIMARY KEY, " a " INTEGER UNIQUE REFERENCES u (" k "), ' + "FOREIGN KEY (A) REFERENCES U (K))", + ) + assert receipt.corroborated is True + assert len(receipt.entries) == 4 + + +class TestDuckDBConstraintTableLookup: + @pytest.mark.parametrize( + ("ddl", "catalog_name"), + [ + ('CREATE TABLE "my t" (a INTEGER PRIMARY KEY)', "my t"), + ('CREATE TABLE "q""t" (a INTEGER PRIMARY KEY)', 'q"t'), + ('CREATE TABLE IF NOT EXISTS "Mixed Case" (a INTEGER PRIMARY KEY)', "Mixed Case"), + ], + ids=["space", "escaped-quote", "mixed-case"], + ) + def test_quoted_table_names_are_read_and_corroborate(self, ddl, catalog_name): + con = duckdb.connect(":memory:") + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute(ddl) + state = DuckDBTuningIntrospector().introspect(con, ledger) + assert [obj.table for obj in state.objects if obj.kind == KIND_CONSTRAINT] == [catalog_name] + receipt = corroborate(ledger, state) + assert [(entry.constraint_type, entry.verdict) for entry in receipt.entries] == [ + (CONSTRAINT_PRIMARY_KEY, CORROBORATED) + ] + + def test_a_similarly_named_table_does_not_stand_in(self): + con = duckdb.connect(":memory:") + con.execute('CREATE TABLE "my" (a INTEGER PRIMARY KEY)') + con.execute('CREATE TABLE "my t" (a INTEGER)') + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute( + 'CREATE TABLE IF NOT EXISTS "my t" (a INTEGER PRIMARY KEY)' + ) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert [entry.verdict for entry in receipt.entries] == [ABSENT] + + +class TestDuckDBUnmaskableLiterals: + @pytest.mark.parametrize( + "executed", + [ + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT $$)$$ CHECK (length(b) > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT $x$)$x$ CHECK (length(b) > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT E'\\')' CHECK (length(b) > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY /* x /* y */ ) */, b VARCHAR CHECK (length(b) > 0))", + ], + ids=["dollar", "tagged-dollar", "escape-string", "nested-block-comment"], + ) + def test_hidden_check_does_not_let_an_existing_key_corroborate(self, executed): + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR)") + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute(executed) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries] == [UNVERIFIABLE] + + +class TestDuckDBListLiterals: + @pytest.mark.parametrize( + "executed", + [ + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR[] DEFAULT [']', ')', 'x'] " + "CHECK (length(b) > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b INTEGER[] DEFAULT [1, 2] CHECK (len(b) > 0))", + "CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY, b VARCHAR DEFAULT ([')'])[1] CHECK (b <> ''))", + ], + ids=["quoted-brackets-in-list", "plain-list", "subscripted-list"], + ) + def test_list_syntax_does_not_let_an_existing_key_corroborate(self, executed): + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE t (a INTEGER PRIMARY KEY, b VARCHAR[])") + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute(executed) + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries] == [UNVERIFIABLE] + + +class TestDuckDBNameResolution: + @pytest.mark.parametrize( + ("setup", "switch"), + [ + ("CREATE SCHEMA other; CREATE TABLE other.t (a INTEGER PRIMARY KEY)", "SET schema = 'other'"), + ("CREATE SCHEMA other; CREATE TABLE other.t (a INTEGER PRIMARY KEY)", "SET search_path = 'other'"), + ("CREATE SCHEMA other; CREATE TABLE other.t (a INTEGER PRIMARY KEY)", "USE memory.other"), + ("ATTACH ':memory:' AS other_db; CREATE TABLE other_db.main.t (a INTEGER PRIMARY KEY)", "USE other_db"), + ], + ids=["set-schema", "set-search-path", "use-schema", "use-database"], + ) + def test_table_in_the_switched_to_schema_does_not_corroborate_the_target(self, setup, switch): + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE main.t (a INTEGER)") + con.execute(setup) + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute("CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY)") + recording_connection(con, ledger, PHASE_SESSION).execute(switch) + + state = DuckDBTuningIntrospector().introspect(con, ledger) + receipt = corroborate(ledger, state) + + assert state.error is not None and "['t']" in state.error + assert receipt.corroborated is False + assert [entry.verdict for entry in receipt.entries if entry.kind == KIND_CONSTRAINT] == [UNVERIFIABLE] + + def test_temporary_table_with_the_same_name_blocks_corroboration(self): + con = duckdb.connect(":memory:") + con.execute("CREATE TABLE t (a INTEGER PRIMARY KEY)") + con.execute("CREATE TEMP TABLE t (a INTEGER)") + ledger = _ledger_for("CREATE TABLE IF NOT EXISTS t (a INTEGER PRIMARY KEY)") + assert corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)).corroborated is False + + def test_target_in_the_current_schema_alone_still_corroborates(self): + con = duckdb.connect(":memory:") + con.execute("CREATE SCHEMA other") + con.execute("CREATE TABLE other.u (a INTEGER PRIMARY KEY)") + ledger = AppliedTuningLedger() + recording_connection(con, ledger, PHASE_DDL).execute("CREATE TABLE t (a INTEGER PRIMARY KEY)") + receipt = corroborate(ledger, DuckDBTuningIntrospector().introspect(con, ledger)) + assert receipt.corroborated is True