diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py index b9755c446a..9dc9c4d0c6 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py @@ -31,6 +31,7 @@ binding_query_fields, lineage_reference_map, prepare_binding_queries, + relation_stubs_collide, remembered_shared_results, reused_shared_results, shared_result_is_exact, @@ -723,22 +724,24 @@ def _prepare_compact_analysis_batch( shared: SharedBindingQuery | None = ( binding_query.shared if binding_query is not None else None ) - qualified_reference_names: frozenset[str] = ( - _qualified_reference_names( + stubbed_reference_names: frozenset[str] = ( + frozenset() + if binding_schema is not None + or relation_stubs_collide(cleaned_sql=cleaned_sql, names=lineage_references) + else frozenset(lineage_references) + - _qualified_reference_names( query_sql=cleaned_sql, reference_names=lineage_references.keys(), ) - if binding_schema is None - else frozenset() ) canonical_stubs: dict[str, str] = ( shared.stubs if shared is not None else { name: ( - name - if binding_schema is not None or name in qualified_reference_names - else f"{COMPACT_RELATION_STUB_PREFIX}{index}" + f"{COMPACT_RELATION_STUB_PREFIX}{index}" + if name in stubbed_reference_names + else name ) for index, name in enumerate(lineage_references) } diff --git a/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py b/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py index dad614ff6b..24b990bd30 100644 --- a/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py +++ b/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py @@ -4,7 +4,7 @@ import re from collections import Counter -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from dataclasses import replace from typing import Any, cast @@ -190,6 +190,15 @@ def shared_result_is_exact(*, validation: object, template: object) -> bool: return isinstance(validation, dict) and cast(dict[str, object], validation).get("errors") == [] +def relation_stubs_collide(*, cleaned_sql: str, names: Iterable[str]) -> bool: + """Return whether authored SQL or relation names already use the relation stub prefix.""" + + return ( + any(name.startswith(COMPACT_RELATION_STUB_PREFIX) for name in names) + or COMPACT_RELATION_STUB_PREFIX.casefold() in cleaned_sql.casefold() + ) + + def _shared_reference_names( *, cleaned_sql: str, @@ -199,9 +208,7 @@ def _shared_reference_names( names: list[str] = list(lineage_reference_map(references)) if not names or len({name.casefold() for name in names}) != len(names): return None - if any(name.startswith(COMPACT_RELATION_STUB_PREFIX) for name in names): - return None - if COMPACT_RELATION_STUB_PREFIX.casefold() in cleaned_sql.casefold(): + if relation_stubs_collide(cleaned_sql=cleaned_sql, names=names): return None if SQL_QUOTED_IDENTIFIER_DELIMITER in cleaned_sql: return None diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py index 863289062a..e2ae536ae9 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py @@ -12,6 +12,14 @@ class SourceSemanticBindingCase: expected_code: str = "B002" +@dataclass(frozen=True) +class RelationStubNameCase: + description: str + stub_cte_name: str + expected_downstream_type: str + expected_ghost_column_diagnostics: frozenset[tuple[str, str]] + + @dataclass(frozen=True) class SemanticCorpusCase: description: str @@ -106,6 +114,19 @@ class LayeredProductionCompilePerformanceGuardTestCase: expected_config_edit_max_seconds: float +@dataclass(frozen=True) +class LayeredProductionCompileTestCase: + description: str + model_count: int + source_count: int + seed_count: int + function_count: int + macro_count: int + test_count: int + audit_count: int + expected_diagnostic_codes: tuple[str, ...] + + @dataclass(frozen=True) class SemanticCompilePerformanceGuardTestCase: description: str diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py index 510052ba73..237b3bc064 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -30,6 +30,7 @@ FreshProcessCompileCachePerformanceGuardTestCase, SemanticCorpusCase, ) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_inline_project _DBT_SHAPED_SQL_SIZE_PROFILE: tuple[tuple[float, int], ...] = ( (0.50, 1_800), @@ -1474,8 +1475,12 @@ def _scope_single_folder_macros(*, project_dir: Path) -> None: def _layered_write_schemas(*, project_dir: Path) -> None: - schemas_dir: Path = project_dir / "schemas" - schemas_dir.mkdir() + """Write the contract schema beside its only consumer, the first staging model.""" + + schemas_dir: Path = ( + project_dir / "models" / _layered_model_folder(index=0) / "_sqlbuild" / "_schemas" + ) + schemas_dir.mkdir(parents=True) (schemas_dir / "benchmark_row.sql").write_text( """SCHEMA ( name benchmark_row, @@ -1619,16 +1624,17 @@ def _layered_dependent_model_sql( previous_name: str = f"model_{index - 1:05d}" macro_index: int = (index // _MACRO_INTERVAL) % macro_count id_expression: str = { - True: f'@macro_{macro_index:05d}("id")', - False: f"id + {index % 7}", + True: f'@macro_{macro_index:05d}("previous.id")', + False: f"previous.id + {index % 7}", }[index % _MACRO_INTERVAL == 0] generated_amount_expression: str = ( - f"amount + {index % 11} + {_layered_generated_mapping_expression()} " + f"previous.amount + {index % 11} + " + f"{_layered_generated_mapping_expression(column='previous.id')} " "+ CAST(@@benchmark_revision AS INTEGER)" ) function_index: int = index % function_count amount_expression: str = { - True: f'__udf("fn_{function_index:05d}")(amount)', + True: f'__udf("fn_{function_index:05d}")(previous.amount)', False: generated_amount_expression, }[index % _FUNCTION_INTERVAL == 0] seed_index: int = index % seed_count @@ -1639,7 +1645,7 @@ def _layered_dependent_model_sql( query: str = f'''SELECT {id_expression} AS id, {amount_expression} AS amount, - CASE WHEN id % 2 = 0 THEN 'even' ELSE 'odd' END AS status + CASE WHEN previous.id % 2 = 0 THEN 'even' ELSE 'odd' END AS status FROM __ref("{previous_name}") AS previous{join_sql} ''' direct_sql: str = f"{_layered_model_header(index=index, audit_count=audit_count)}\n\n{query}" @@ -2061,8 +2067,10 @@ def _layered_model_sql_size_target(*, index: int, model_count: int) -> int: return round(lower_size + position * (upper_size - lower_size)) -def _layered_generated_mapping_expression() -> str: - clauses: str = "".join(f"WHEN id = {value:05d} THEN {value % 13:02d} " for value in range(45)) +def _layered_generated_mapping_expression(*, column: str = "id") -> str: + clauses: str = "".join( + f"WHEN {column} = {value:05d} THEN {value % 13:02d} " for value in range(45) + ) return f"CASE {clauses} ELSE 0 END" @@ -2276,3 +2284,44 @@ def plan_reasons_and_migrations( def _query_changed(model: dict[str, Any]) -> bool: return model["reason"] == "query_changed" + + +def write_relation_stub_project( + *, tmp_path: Path, stub_cte_name: str, selected_column: str +) -> Path: + """Write a project whose middle model authors a CTE named like a relation stub.""" + + return prepare_inline_project( + tmp_path=tmp_path, + project_name="relation_stub_names", + repo_files={ + "sqlbuild_project.toml": ( + 'name = "relation_stub_names"\n' + 'adapter = "duckdb"\n' + 'default_target = "dev"\n\n' + "[connection]\n" + 'database = "orders.duckdb"\n\n' + "[targets.dev]\n" + 'schema = "main"\n' + ), + "models/orders.sql": ( + "MODEL (materialized table);\n\nSELECT CAST(7 AS INTEGER) AS order_id\n" + ), + "models/order_status.sql": ( + "MODEL (materialized table);\n\n" + f"WITH {stub_cte_name} AS (\n" + " SELECT 'pending' AS order_id, 'pending' AS ghost_status\n" + ")\n" + f'SELECT {selected_column} FROM __ref("orders")\n' + ), + "models/next_order.sql": ( + "MODEL (\n" + " materialized table,\n" + " contract enforced,\n" + " columns (order_id (type INTEGER), next_order_id (type INTEGER)),\n" + ");\n\n" + "SELECT order_id, order_id + 1 AS next_order_id\n" + 'FROM __ref("order_status")\n' + ), + }, + ) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_performance.py index 6e56892d99..d0f011d914 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_compile_performance.py @@ -237,7 +237,7 @@ def test_given_test_heavy_project_when_compiling_then_finishes_within_budget( macro_count=12, test_count=130, expected_audit_count=700, - expected_hook_count=2, + expected_hook_count=0, expected_min_model_sql_bytes=5_500_000, expected_max_model_sql_bytes=7_000_000, expected_min_compiled_test_bytes=3_000_000, diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_layered_production_project.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_layered_production_project.py new file mode 100644 index 0000000000..c4b684b28a --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_layered_production_project.py @@ -0,0 +1,77 @@ +"""E2E coverage that the layered production benchmark project compiles cleanly.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( + LayeredProductionCompileTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + write_layered_production_compile_project, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb + + +@pytest.mark.parametrize( + "test_case", + [ + LayeredProductionCompileTestCase( + description="small layered project with seed joins, macros, functions and contract", + model_count=350, + source_count=83, + seed_count=16, + function_count=8, + macro_count=4, + test_count=47, + audit_count=251, + expected_diagnostic_codes=(), + ), + ], + ids=lambda case: case.description, +) +def test_given_layered_production_project_when_compiling_then_reports_no_diagnostics( + test_case: LayeredProductionCompileTestCase, + tmp_path: Path, +) -> None: + project_dir: Path = tmp_path / "layered_production" + write_layered_production_compile_project( + project_dir=project_dir, + model_count=test_case.model_count, + source_count=test_case.source_count, + seed_count=test_case.seed_count, + function_count=test_case.function_count, + macro_count=test_case.macro_count, + test_count=test_case.test_count, + audit_count=test_case.audit_count, + ) + model_sql: str = "".join( + path.read_text(encoding="utf-8") for path in (project_dir / "models").rglob("*.sql") + ) + for fragment in ("__seed(", "__udf(", "@macro_", "model_schema benchmark_row"): + assert fragment in model_sql + + result: subprocess.CompletedProcess[str] = run_sqb( + project_dir=project_dir, + command=("compile", "--no-cache", "--json"), + ) + + assert result.returncode == 0, result.stdout + result.stderr + payload: dict[str, Any] = json.loads(result.stdout) + diagnostic_codes: tuple[str, ...] = tuple( + str(diagnostic["code"]) for diagnostic in payload["diagnostics"] + ) + assert diagnostic_codes == test_case.expected_diagnostic_codes, payload["diagnostics"] + assert payload["summary"]["errors"] == 0 + assert payload["summary"]["warnings"] == 0 + assert payload["summary"]["models"] == test_case.model_count + assert payload["summary"]["tests"] == test_case.test_count + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_relation_stub_names.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_relation_stub_names.py new file mode 100644 index 0000000000..cae9128ea7 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_relation_stub_names.py @@ -0,0 +1,113 @@ +"""Authored CTEs named like internal relation stubs must not capture model references.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import RelationStubNameCase +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import write_relation_stub_project +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import query_duckdb, run_sqb + + +@pytest.mark.parametrize( + "test_case", + [ + RelationStubNameCase( + description="lowercase stub-named CTE", + stub_cte_name="__sqlbuild_project_input_0", + expected_downstream_type="INTEGER", + expected_ghost_column_diagnostics=frozenset( + {("B002", "order_status"), ("B002", "next_order")} + ), + ), + RelationStubNameCase( + description="uppercase stub-named CTE", + stub_cte_name="__SQLBUILD_PROJECT_INPUT_0", + expected_downstream_type="INTEGER", + expected_ghost_column_diagnostics=frozenset( + {("B002", "order_status"), ("B002", "next_order")} + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_stub_named_cte_when_compiling_and_building_then_reference_keeps_input_types( + test_case: RelationStubNameCase, + tmp_path: Path, +) -> None: + project: Path = write_relation_stub_project( + tmp_path=tmp_path, stub_cte_name=test_case.stub_cte_name, selected_column="order_id" + ) + + compiled: subprocess.CompletedProcess[str] = run_sqb( + project_dir=project, command=("compile", "--no-cache", "--json") + ) + + assert compiled.returncode == 0, compiled.stdout + compiled.stderr + payload: dict[str, Any] = json.loads(compiled.stdout) + assert payload["diagnostics"] == [] + built: subprocess.CompletedProcess[str] = run_sqb(project_dir=project, command=("build",)) + assert built.returncode == 0, built.stdout + built.stderr + assert query_duckdb( + db_path=project / "orders.duckdb", + sql=( + "SELECT column_name, data_type FROM information_schema.columns " + "WHERE table_name = 'next_order' ORDER BY ordinal_position" + ), + ) == [ + ("order_id", test_case.expected_downstream_type), + ("next_order_id", test_case.expected_downstream_type), + ] + assert query_duckdb( + db_path=project / "orders.duckdb", sql="SELECT order_id, next_order_id FROM next_order" + ) == [(7, 8)] + + +@pytest.mark.parametrize( + "test_case", + [ + RelationStubNameCase( + description="lowercase stub-named CTE", + stub_cte_name="__sqlbuild_project_input_0", + expected_downstream_type="INTEGER", + expected_ghost_column_diagnostics=frozenset( + {("B002", "order_status"), ("B002", "next_order")} + ), + ), + RelationStubNameCase( + description="uppercase stub-named CTE", + stub_cte_name="__SQLBUILD_PROJECT_INPUT_0", + expected_downstream_type="INTEGER", + expected_ghost_column_diagnostics=frozenset( + {("B002", "order_status"), ("B002", "next_order")} + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_column_only_in_stub_named_cte_when_compiling_then_reports_unknown_input_column( + test_case: RelationStubNameCase, + tmp_path: Path, +) -> None: + project: Path = write_relation_stub_project( + tmp_path=tmp_path, stub_cte_name=test_case.stub_cte_name, selected_column="ghost_status" + ) + + compiled: subprocess.CompletedProcess[str] = run_sqb( + project_dir=project, command=("compile", "--no-cache", "--json") + ) + + assert compiled.returncode == 1, compiled.stdout + compiled.stderr + payload: dict[str, Any] = json.loads(compiled.stdout) + assert { + (diagnostic["code"], diagnostic["resource_name"]) for diagnostic in payload["diagnostics"] + } == test_case.expected_ghost_column_diagnostics + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py index 105003b8b3..e489937fb9 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py @@ -230,6 +230,73 @@ def test_given_cte_cast_when_batch_analyzing_then_cast_type_overrides_input_type assert result.columns == test_case.expected_columns +@pytest.mark.parametrize( + "test_case", + ( + InferColumnsTestCase( + description="lowercase authored stub-named CTE", + query_sql=( + "WITH __sqlbuild_project_input_0 AS (SELECT 'pending' AS order_id) " + 'SELECT order_id FROM __ref("orders")' + ), + expected_columns=( + InferredColumn( + name="order_id", type="BIGINT", nullability=InferredNullability.NON_NULL + ), + ), + ), + InferColumnsTestCase( + description="uppercase authored stub-named CTE", + query_sql=( + "WITH __SQLBUILD_PROJECT_INPUT_0 AS (SELECT 'pending' AS order_id) " + 'SELECT order_id FROM __ref("orders")' + ), + expected_columns=( + InferredColumn( + name="order_id", type="BIGINT", nullability=InferredNullability.NON_NULL + ), + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_authored_stub_named_cte_when_batch_analyzing_without_binding_then_uses_input( + test_case: InferColumnsTestCase, +) -> None: + references: tuple[CompileSqlReference, ...] = ( + CompileSqlReference(ref_kind=SqlReferenceKind.REF, ref_name="orders"), + ) + nullability: dict[str, dict[str, InferredNullability]] = { + "orders": {"order_id": InferredNullability.NON_NULL} + } + types: dict[str, dict[str, str]] = {"orders": {"order_id": "BIGINT"}} + profile: ExpressionInferenceProfile = ExpressionInferenceProfile(sql_analysis_dialect="duckdb") + + prepared: NativeCompactAnalysis = analyze_queries_with_compact_polyglot_batch( + query_sqls=(test_case.query_sql,), + references=(references,), + placeholders=(None,), + column_nullability_by_table=nullability, + column_types_by_table=types, + inference_profile=profile, + recover_cte_facts=(False,), + )[0] + result: PolyglotAnalysisResult = analyze_columns_and_lineage_with_polyglot( + query_sql=test_case.query_sql, + references=references, + column_nullability_by_table=nullability, + column_types_by_table=types, + inference_profile=profile, + allow_compact_analysis=True, + precomputed=prepared, + ) + + assert result.analysis_succeeded + assert result.columns == test_case.expected_columns + lineage_column: CompiledLineageColumnFact = tuple(result.lineage_columns)[0] + assert [source.resource_name for source in lineage_column.upstream_columns] == ["orders"] + + @pytest.mark.parametrize( "test_case", [