From 45bf571d7565235b3e9ac6d6f6d6653245f1a5e6 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 7 Oct 2026 11:20:28 +0100 Subject: [PATCH 1/2] fix(compiler): keep shared-analysis lineage and stage captures schedule-independent A model that reused another model's shared analysis lost its own resource names when the template facts were already projected, so fast column lineage named the representative's source. Pass the member's resource names on the cached-template path too. Compiled-project stage captures also depended on how the binding dataflow batched models: compact lineage facts index a batch-wide string pool, and the shared-analysis memo keeps whichever representative and lazy caches a schedule produced. Capture compact lineage as its decoded facts and leave the memo out, so python-vs-python captures are identical under any schedule. Bump the analysis cache version so entries written with the wrong names are recomputed. --- .../compile/_helpers/analysis/cache.py | 2 +- .../compile/_helpers/analysis/compact.py | 1 + .../frontier/classes/stage_capture_encoder.py | 10 ++- src/sqlbuild/compiler/frontier/constants.py | 13 +++- .../test_compiler_differential.py | 41 ++++++++++ .../sqlbuild/compiler/pipeline/_test_types.py | 7 ++ .../src/sqlbuild/compiler/pipeline/helpers.py | 28 +++++++ .../test_dataflow_analysis_schedule.py | 59 +++++++++++++++ .../pipeline/test_shared_binding_queries.py | 52 +++++++++++++ .../compile/_helpers/test_analysis_cache.py | 12 +-- .../src/sqlbuild/compiler/frontier/helpers.py | 35 +++++++++ .../compiler/frontier/test_stage_capture.py | 74 +++++++++++++++++++ 12 files changed, 324 insertions(+), 10 deletions(-) diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py index 54418cdf95..10a92d3e45 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py @@ -46,7 +46,7 @@ from sqlbuild.compiler.sql_analysis.constants import BINDING_SEVERITIES, TYPE_CHECKED_DIALECTS from sqlbuild.compiler.sql_analysis.models import SqlBindingDiagnostic -_ANALYSIS_CACHE_VERSION: int = 12 +_ANALYSIS_CACHE_VERSION: int = 13 _ANALYSIS_ALGORITHM_FINGERPRINT: str = "model-sql-analysis-v21-completed-star-shapes" _LINEAGE_COLUMN_VALUE_COUNT: int = 4 _LINEAGE_SOURCE_VALUE_COUNT: int = 3 diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py index 3833027435..c2b40d58ef 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py @@ -1274,6 +1274,7 @@ def _compact_projected_analysis_result( string_pool=string_pool, rows=cached_facts.lineage_rows, resource_name_indexes=request.resource_name_indexes, + resource_names=request.resource_names, ), has_star=bool(request.analysis and request.analysis.get("hasStar")), star_resolved=bool(request.analysis and request.analysis.get("starResolved")), diff --git a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py index b7d31c891a..0a24086bec 100644 --- a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py +++ b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py @@ -12,12 +12,14 @@ import math import re import types -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from collections.abc import Set as AbstractSet from pathlib import PurePath from typing import Any from sqlbuild.compiler.frontier.constants import ( + STAGE_CAPTURE_DECODED_SEQUENCES, + STAGE_CAPTURE_OMITTED_ATTRIBUTES, STAGE_CAPTURE_SKIPPED_SLOTS, STAGE_CAPTURE_UNORDERED_ATTRIBUTES, ) @@ -59,6 +61,9 @@ def encode(self, value: object) -> object: self._active.discard(identity) def _encode_container(self, value: object) -> object: + type_name: str = qualified_name(type(value)) + if type_name in STAGE_CAPTURE_DECODED_SEQUENCES and isinstance(value, Sequence): + return [self.encode(item) for item in value] if dataclasses.is_dataclass(value) and not isinstance(value, type): return { "__type__": qualified_name(type(value)), @@ -80,15 +85,16 @@ def _encode_container(self, value: object) -> object: "keywords": self.encode(value.keywords), } attributes: dict[str, object] | None = _instance_attributes(value) - type_name: str = qualified_name(type(value)) if attributes is None: return {"__opaque__": type_name} unordered: frozenset[str] = STAGE_CAPTURE_UNORDERED_ATTRIBUTES.get(type_name, frozenset()) + omitted: frozenset[str] = STAGE_CAPTURE_OMITTED_ATTRIBUTES.get(type_name, frozenset()) return { "__type__": type_name, **{ name: self._encode_attribute(value=item, unordered=name in unordered) for name, item in attributes.items() + if name not in omitted }, } diff --git a/src/sqlbuild/compiler/frontier/constants.py b/src/sqlbuild/compiler/frontier/constants.py index 68818cdc2a..8f6ed2fd05 100644 --- a/src/sqlbuild/compiler/frontier/constants.py +++ b/src/sqlbuild/compiler/frontier/constants.py @@ -15,6 +15,17 @@ STAGE_CAPTURE_SKIPPED_SLOTS: frozenset[str] = frozenset({"__dict__", "__weakref__"}) STAGE_CAPTURE_UNORDERED_ATTRIBUTES: dict[str, frozenset[str]] = { "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset( - {"schemas", "analysis_shapes", "expression_shapes", "shared_analyses"} + {"schemas", "analysis_shapes", "expression_shapes"} ) } +# The shared-analysis memo is a reuse cache whose representatives and lazily filled caches +# depend on analysis batching; compact lineage facts index a batch-wide string pool, so they +# are captured by their decoded facts. +STAGE_CAPTURE_OMITTED_ATTRIBUTES: dict[str, frozenset[str]] = { + "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset( + {"shared_analyses"} + ) +} +STAGE_CAPTURE_DECODED_SEQUENCES: frozenset[str] = frozenset( + {"sqlbuild.compiler.compile.models:CompactLineageFacts"} +) diff --git a/tests/e2e/scripts/compiler_differential/test_compiler_differential.py b/tests/e2e/scripts/compiler_differential/test_compiler_differential.py index ebf2ba845c..f66275611d 100644 --- a/tests/e2e/scripts/compiler_differential/test_compiler_differential.py +++ b/tests/e2e/scripts/compiler_differential/test_compiler_differential.py @@ -268,6 +268,47 @@ def test_given_required_coverage_missing_when_comparing_then_harness_fails_and_s assert not any(text in output for text in test_case.expected_absent), output +@pytest.mark.parametrize( + "test_case", + [ + HarnessRunTestCase( + description="shared_analysis_seed_against_itself", + extra_arguments=( + "--corpus", + "seeds", + "--seed-start", + "29", + "--seeds", + "1", + "--engines", + "python", + "python", + "--stage-captures", + ), + expected_exit_code=0, + expected_lines=( + "OK seed/29", + "Compiler differential passed: 1 projects identical (python vs python)", + ), + expected_patterns=(), + expected_absent=("DIFF",), + ) + ], + ids=lambda case: case.description, +) +def test_given_python_engine_twice_when_capturing_stages_then_captures_are_identical( + test_case: HarnessRunTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + exit_code: int = run_compiler_differential( + ["--jobs", "1", "--work-dir", str(tmp_path / "work"), *test_case.extra_arguments] + ) + + output: str = capsys.readouterr().out + assert exit_code == test_case.expected_exit_code, output + assert all(line in output for line in test_case.expected_lines), output + assert not any(text in output for text in test_case.expected_absent), output + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py index 81a1f14d2d..0d3737b292 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py @@ -440,6 +440,13 @@ class DataflowScheduleCase: expected_edit_exit_code: int = 1 +@dataclass(frozen=True) +class DataflowCaptureCase: + description: str + write_project: Callable[[Path], object] + schedules: tuple[DataflowScheduleCase, ...] + + @dataclass(frozen=True) class DataflowFailureCase: description: str diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py index 66f7ada616..ad98206841 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py @@ -20,6 +20,7 @@ from scripts.cold_compile_performance._helpers.dense_project import write_dense_compile_project from scripts.cold_compile_performance._helpers.random_dag_project import write_random_dag_project from scripts.cold_compile_performance.models import RandomDagProject +from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.cli.commands.main.entrypoint.entry import main @@ -41,6 +42,8 @@ ) from sqlbuild.compiler.discovery.main.discover import discover_project_inputs from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs +from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_DIR_ENV_VAR +from sqlbuild.compiler.frontier.types import CompilerStage from sqlbuild.compiler.manifest.main.build import build_manifest from sqlbuild.compiler.pipeline.main.compile import run_compile_pipeline from sqlbuild.compiler.pipeline.main.project import compile_project @@ -306,6 +309,23 @@ def timed_compile_outcome( ), timings +def compiled_project_capture( + *, + project_dir: Path, + capture_dir: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> str: + """Cold-compile once with stage captures; return the normalized compiled-project capture.""" + + with monkeypatch.context() as capture_patch: + capture_patch.setenv(STAGE_CAPTURE_DIR_ENV_VAR, str(capture_dir)) + _ = main(["--project-dir", str(project_dir), "compile", "--json", "--no-cache"]) + _ = capsys.readouterr() + capture: Path = next(capture_dir.glob(f"*-{CompilerStage.COMPILED_PROJECT.value}.json")) + return normalize_artifact_text(capture.read_text(encoding="utf-8")) + + def reshape_models(*, project_dir: Path, names: tuple[str, ...], indexes: tuple[int, ...]) -> None: """Reshape the models at `indexes` of the generated topological name order.""" @@ -319,6 +339,14 @@ def use_wave_analysis(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(project_assembly, "analyze_binding_dataflow", analyze_binding_waves) +def analyze_single_model_batches(monkeypatch: pytest.MonkeyPatch) -> None: + """Analyze every model in its own native batch on one worker.""" + + monkeypatch.setattr(binding_waves, "_DATAFLOW_WORKERS", 1) + monkeypatch.setattr(binding_waves, "_DATAFLOW_BATCH_MIN", 1) + monkeypatch.setattr(binding_waves, "_DATAFLOW_BATCH_LIMIT", 1) + + def perturb_dataflow_schedule( *, monkeypatch: pytest.MonkeyPatch, case: DataflowScheduleCase, seed: int ) -> None: diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py index 4fb2220e19..a9906f7bbf 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py @@ -9,9 +9,12 @@ from scripts.cold_compile_performance._helpers.random_dag_project import write_random_dag_project from scripts.cold_compile_performance.models import RandomDagProject +from scripts.compiler_differential._helpers.comparing.compare import first_document_difference +from scripts.compiler_differential.main.generate_project import generate_project from sqlbuild.cli.commands.main.entrypoint.entry import main from tests.integration.src.sqlbuild.compiler.pipeline._test_types import ( CompileOutcome, + DataflowCaptureCase, DataflowFailureCase, DataflowInterruptAfterFaultCase, DataflowInterruptCase, @@ -22,6 +25,7 @@ from tests.integration.src.sqlbuild.compiler.pipeline.helpers import ( analysis_pool_thread_ids, compile_outcome, + compiled_project_capture, fail_model_analysis, fail_worker_start, interrupt_after_analysis_fault, @@ -33,12 +37,22 @@ ) _PROJECT: RandomDagProject = RandomDagProject(seed=97, model_count=40, errors=True) +_SHARED_ANALYSIS_SEED: int = 29 +_CAPTURE_SCHEDULES: tuple[DataflowScheduleCase, ...] = ( + DataflowScheduleCase("one worker with single models", 1, 1, 0.0), + DataflowScheduleCase("four workers with small batches", 4, 2, 0.004), + DataflowScheduleCase("four workers with wide batches", 4, 64, 0.002), +) _EDITED: tuple[int, ...] = (10, 25) _REPEATS: tuple[int, ...] = (0, 1) _SETTLE_SECONDS: float = 0.3 _INTERRUPT_NOTICE: str = "Interrupted; finishing in-flight model analysis..." +def _write_random_dag(project_dir: Path) -> None: + _ = write_random_dag_project(project_dir=project_dir, project=_PROJECT) + + @pytest.mark.parametrize( "test_case", [ @@ -81,6 +95,51 @@ def test_given_perturbed_schedule_when_compiling_repeatedly_then_output_is_ident assert actual_edit[0] == test_case.expected_edit_exit_code +@pytest.mark.parametrize( + "test_case", + [ + DataflowCaptureCase( + "random dag", + _write_random_dag, + _CAPTURE_SCHEDULES, + ), + DataflowCaptureCase( + "differential seed with shared analyses", + generate_project(seed=_SHARED_ANALYSIS_SEED).write, + _CAPTURE_SCHEDULES, + ), + ], + ids=lambda case: case.description, +) +def test_given_perturbed_schedule_when_capturing_compiled_project_then_capture_is_identical( + test_case: DataflowCaptureCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + project_dir: Path = tmp_path / "project" + _ = test_case.write_project(project_dir) + with monkeypatch.context() as waves_patch: + use_wave_analysis(waves_patch) + reference: str = compiled_project_capture( + project_dir=project_dir, + capture_dir=tmp_path / "reference", + capsys=capsys, + monkeypatch=monkeypatch, + ) + for index, schedule in enumerate(test_case.schedules): + with monkeypatch.context() as schedule_patch: + perturb_dataflow_schedule(monkeypatch=schedule_patch, case=schedule, seed=index) + actual: str = compiled_project_capture( + project_dir=project_dir, + capture_dir=tmp_path / f"schedule-{index}", + capsys=capsys, + monkeypatch=monkeypatch, + ) + + assert first_document_difference(left=reference, right=actual) is None, schedule + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py index cd1d1b2cd1..6a07dbe703 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_shared_binding_queries.py @@ -10,6 +10,7 @@ from sqlbuild.compiler.compile.models import CompactBatchPreparation from tests.integration.src.sqlbuild.compiler.pipeline._test_types import SharedBindingQueryCase from tests.integration.src.sqlbuild.compiler.pipeline.helpers import ( + analyze_single_model_batches, trace_native_compact_batches, write_shared_binding_project, ) @@ -229,6 +230,57 @@ def test_given_later_wave_members_with_equal_keys_when_compiling_then_reuses_sha assert [step["source"]["resource_name"] for step in lineage["trace"]] == [upstream] +@pytest.mark.parametrize( + "test_case", + [ + SharedBindingQueryCase( + description="members analyzed in separate batches keep their own sources", + orders_summary_sql='MODEL (description "Test model.", materialized view); SELECT id, amount FROM __ref("orders")', + customers_summary_sql=( + 'MODEL (description "Test model.", materialized view); SELECT id, amount FROM __ref("customers")' + ), + expected_shared_queries=0, + expected_lineage=( + ("orders_summary", "orders"), + ("customers_summary", "customers"), + ), + ) + ], + ids=lambda case: case.description, +) +def test_given_equal_keys_in_separate_batches_when_tracing_lineage_then_each_model_keeps_its_source( + test_case: SharedBindingQueryCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + write_shared_binding_project(project_dir=tmp_path, test_case=test_case) + analyze_single_model_batches(monkeypatch) + + for model, upstream in test_case.expected_lineage: + assert ( + main( + [ + "--project-dir", + str(tmp_path), + "lineage", + f"{model}.amount", + "--format", + "json", + "--direction", + "upstream", + "--depth", + "1", + "--mode", + "fast", + ] + ) + == 0 + ) + lineage: dict[str, Any] = json.loads(capsys.readouterr().out) + assert [step["source"]["resource_name"] for step in lineage["trace"]] == [upstream] + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py index f96a30d800..ea1e0d943c 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_analysis_cache.py @@ -194,7 +194,7 @@ def test_given_exact_compact_batch_when_entry_rows_are_absent_then_warm_compile_ write_repo_files(tmp_path, _CACHE_REPO_FILES) monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1) cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) - cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3" + cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: batch_count: int = connection.execute( "SELECT COUNT(*) FROM model_analysis_compact_batch" @@ -227,7 +227,7 @@ def test_given_one_changed_model_when_entry_rows_are_absent_then_compact_batch_r write_repo_files(tmp_path, _SELECTION_REPO_FILES) monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1) cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) - cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3" + cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: _ = connection.execute("DELETE FROM model_analysis") (tmp_path / "models" / "unrelated.sql").write_text( @@ -265,7 +265,7 @@ def test_given_corrupt_compact_batch_when_entry_rows_are_valid_then_warm_compile write_repo_files(tmp_path, _CACHE_REPO_FILES) monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1) cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) - cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3" + cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: _ = connection.execute( "UPDATE model_analysis_compact_batch SET payload = ?", @@ -316,7 +316,7 @@ def test_given_corrupt_analysis_when_compiling_then_reanalyzes_and_repairs_the_e ) -> None: write_repo_files(tmp_path, _CACHE_REPO_FILES) cold_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) - cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3" + cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: persisted_contents: str = connection.execute( "SELECT payload FROM model_analysis" @@ -339,7 +339,7 @@ def test_given_corrupt_analysis_when_compiling_then_reanalyzes_and_repairs_the_e _digest, _separator, serialized_payload = repaired_contents.partition("\n") repaired_payload: dict[str, object] = json.loads(serialized_payload) assert repaired_project.models == cold_project.models - assert repaired_payload["v"] == 12 + assert repaired_payload["v"] == 13 assert isinstance(repaired_payload["s"], str) assert analyzer.call_count == test_case.expected_count @@ -357,7 +357,7 @@ def test_given_non_text_analysis_cache_when_compiling_then_reanalyzes_safely( ) -> None: write_repo_files(tmp_path, _CACHE_REPO_FILES) _ = compile_project_with_cache(project_dir=tmp_path) - cache_path: Path = compiler_cache_directory(tmp_path) / "v12" / "model-analysis.sqlite3" + cache_path: Path = compiler_cache_directory(tmp_path) / "v13" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: _ = connection.execute( "UPDATE model_analysis SET payload = ?", diff --git a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py index 5259371578..a0a19a1bc9 100644 --- a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py @@ -6,9 +6,11 @@ from enum import StrEnum from pathlib import Path, PurePosixPath +from sqlbuild.compiler.compile.models import CompactLineageFacts from sqlbuild.compiler.fact_cache.main._compile_cache_root import compile_cache_root from sqlbuild.compiler.frontier.main.compiler_cache_directory import compiler_cache_directory from sqlbuild.compiler.scopes.constants import SCOPE_CACHE_DIRECTORY_NAME +from sqlbuild.compiler.sql_analysis.classes.binding_catalog import BindingCatalog from sqlbuild.rule_engine._helpers.run.cache_paths import rules_bulk_cache_path HELPERS_MODULE: str = "tests.unit.src.sqlbuild.compiler.frontier.helpers" @@ -68,6 +70,39 @@ def memo_catalog(*, memo_keys: tuple[str, ...], ordered_keys: tuple[str, ...]) - ) +def orders_lineage(*, string_pool: tuple[str, ...], source_index: int) -> CompactLineageFacts: + """Return `order_id` read directly from a source, indexed into a batch-wide string pool.""" + + position: dict[str, int] = {value: index for index, value in enumerate(string_pool)} + return CompactLineageFacts( + string_pool=string_pool, + rows=( + ( + position["order_id"], + 0, + 1, + ((position["source"], source_index, position["order_id"]),), + ), + ), + ) + + +def orders_binding_catalog( + *, shared_analyses: tuple[tuple[str, str], ...], relations: tuple[str, ...] +) -> BindingCatalog: + """Return a catalog whose memo and relation map were filled in the given orders.""" + + catalog: BindingCatalog = BindingCatalog( + dialect="duckdb", + quoted_ignore_case=False, + known_functions=(), + known_types=(), + relations={name: {"order_id": "INTEGER"} for name in relations}, + ) + catalog.shared_analyses.update(shared_analyses) + return catalog + + def order_line() -> OrderLine: """Return one neutral order line.""" diff --git a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py index 6b2c2384f9..8592fc9e08 100644 --- a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py +++ b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py @@ -27,8 +27,14 @@ memo_catalog, order_line, order_total, + orders_binding_catalog, + orders_lineage, ) +_COMPILE_MODELS: str = "sqlbuild.compiler.compile.models" +_COMPILE_TYPES: str = "sqlbuild.compiler.compile.types" +_LINEAGE_TYPES: str = "sqlbuild.compiler.lineage.types" + @pytest.mark.parametrize( "test_case", @@ -67,6 +73,30 @@ "peer": {"__cycle__": f"{HELPERS_MODULE}:LinkedCustomer"}, }, ), + StageCaptureTestCase( + description="compact_lineage_is_captured_as_decoded_facts", + value=lambda: orders_lineage( + string_pool=("customers", "order_id", "source", "orders"), source_index=3 + ), + expected_capture=[ + { + "__type__": f"{_COMPILE_MODELS}:CompiledLineageColumnFact", + "output_column": "order_id", + "upstream_columns": [ + { + "__type__": f"{_COMPILE_MODELS}:CompiledLineageSourceFact", + "resource_type": { + "__enum__": f"{_COMPILE_TYPES}:CompiledResourceType.SOURCE" + }, + "resource_name": "orders", + "column_name": "order_id", + } + ], + "transform_kind": {"__enum__": f"{_LINEAGE_TYPES}:ColumnTransformKind.DIRECT"}, + "confidence": {"__enum__": f"{_LINEAGE_TYPES}:ColumnLineageConfidence.HIGH"}, + } + ], + ), StageCaptureTestCase( description="non_finite_float_and_bytes", value=lambda: (float("inf"), b"orders"), @@ -114,6 +144,50 @@ def test_given_frontier_value_when_rendering_capture_then_json_is_canonical( second=lambda: frozenset(("products", "customers", "orders")), expected_identical=True, ), + StageCaptureOrderTestCase( + description="batch_string_pool_layout_is_not_visible", + first=lambda: orders_lineage( + string_pool=("order_id", "source", "orders"), source_index=2 + ), + second=lambda: orders_lineage( + string_pool=("customers", "source", "orders", "amount", "order_id"), + source_index=2, + ), + expected_identical=True, + ), + StageCaptureOrderTestCase( + description="decoded_lineage_source_is_visible", + first=lambda: orders_lineage( + string_pool=("order_id", "source", "orders", "customers"), source_index=2 + ), + second=lambda: orders_lineage( + string_pool=("order_id", "source", "orders", "customers"), source_index=3 + ), + expected_identical=False, + ), + StageCaptureOrderTestCase( + description="shared_analysis_memo_fill_is_not_visible", + first=lambda: orders_binding_catalog( + shared_analyses=(("orders_key", "orders"), ("customers_key", "customers")), + relations=("orders", "customers"), + ), + second=lambda: orders_binding_catalog( + shared_analyses=( + ("customers_key", "customers_copy"), + ("orders_key", "orders_copy"), + ), + relations=("customers", "orders"), + ), + expected_identical=True, + ), + StageCaptureOrderTestCase( + description="binding_catalog_relations_stay_visible", + first=lambda: orders_binding_catalog( + shared_analyses=(), relations=("orders", "customers") + ), + second=lambda: orders_binding_catalog(shared_analyses=(), relations=("orders",)), + expected_identical=False, + ), ], ids=lambda case: case.description, ) From 4e99eb1259c12e65eb6bb77f1207e3596e2e49fe Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Wed, 7 Oct 2026 12:52:58 +0100 Subject: [PATCH 2/2] test(compiler): compare large stage captures without loading them whole Captures embedded every repeated subtree in full, so their size grew quadratically with sources per file and the harness held and parsed both engines' captures in memory; dense 3k ran out of memory. The capture writer now stores every container or string of at least 1 KiB once under the SHA-256 of its compact JSON, with invocation ids and native store suffixes masked, and refers to it as {"__shared__": digest}. Sets and unordered maps sort by the same masked form, so ordering never depends on noise the comparison ignores, and authored mappings using a reserved marker key are encoded as pairs. Each shared node and the root sit on their own line, so the harness indexes a capture by digest, treats equal digests as equal without reading them, and parses only the nodes on a differing path while still reporting the logical JSON pointer and resolved previews. Captures stay on disk, one directory per compared side, instead of in memory. --- .../_helpers/comparing/capture.py | 53 ++++++ .../_helpers/comparing/compare.py | 96 ++++++++--- .../_helpers/comparing/comparison.py | 14 +- .../_helpers/comparing/normalize.py | 6 +- .../_helpers/comparing/project_comparison.py | 9 +- .../_helpers/corpus/emitted_codes.py | 2 + .../_helpers/coverage/capture_values.py | 4 +- .../_helpers/coverage/discovery.py | 13 +- .../_helpers/coverage/render.py | 11 +- .../_helpers/running/execution.py | 21 +-- .../classes/capture_file.py | 92 +++++++++++ scripts/compiler_differential/constants.py | 5 +- scripts/compiler_differential/models.py | 2 +- scripts/compiler_differential/types.py | 5 + .../frontier/_helpers/stage_capture.py | 28 ++-- .../frontier/classes/stage_capture_encoder.py | 65 +++++++- .../frontier/classes/stage_capture_writer.py | 39 +++++ src/sqlbuild/compiler/frontier/constants.py | 15 +- .../sqlbuild/compiler/pipeline/_test_types.py | 1 + .../src/sqlbuild/compiler/pipeline/helpers.py | 8 +- .../test_dataflow_analysis_schedule.py | 17 +- .../_helpers/comparing/_test_types.py | 32 ++++ .../_helpers/comparing/helpers.py | 80 +++++++++ .../_helpers/comparing/test_capture.py | 154 ++++++++++++++++++ .../_helpers/coverage/test_discovery.py | 15 +- .../_helpers/coverage/test_render.py | 7 +- .../sqlbuild/compiler/frontier/_test_types.py | 9 + .../src/sqlbuild/compiler/frontier/helpers.py | 16 ++ .../compiler/frontier/test_stage_capture.py | 39 +++++ 29 files changed, 747 insertions(+), 111 deletions(-) create mode 100644 scripts/compiler_differential/_helpers/comparing/capture.py create mode 100644 scripts/compiler_differential/classes/capture_file.py create mode 100644 scripts/compiler_differential/types.py create mode 100644 src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py create mode 100644 tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py create mode 100644 tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py diff --git a/scripts/compiler_differential/_helpers/comparing/capture.py b/scripts/compiler_differential/_helpers/comparing/capture.py new file mode 100644 index 0000000000..4f35198c9c --- /dev/null +++ b/scripts/compiler_differential/_helpers/comparing/capture.py @@ -0,0 +1,53 @@ +"""Compare two stage captures on disk without loading either one whole.""" + +from __future__ import annotations + +import itertools +import json +from collections.abc import Iterator +from pathlib import Path + +from scripts.compiler_differential._helpers.comparing.compare import ( + first_json_difference, + preview, +) +from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text +from scripts.compiler_differential.classes.capture_file import CaptureFile +from scripts.compiler_differential.constants import MISSING_VALUE +from scripts.compiler_differential.models import Divergence + + +def first_capture_difference(*, left: Path, right: Path) -> Divergence | None: + """Return the first logical JSON pointer, or line for invalid JSON, where captures differ.""" + + try: + left_capture: CaptureFile = CaptureFile(left) + right_capture: CaptureFile = CaptureFile(right) + return first_json_difference( + left=left_capture.root, + right=right_capture.root, + resolve=(left_capture.resolve, right_capture.resolve), + ) + except (json.JSONDecodeError, UnicodeDecodeError): + return _first_line_difference(left=left, right=right) + + +def _first_line_difference(*, left: Path, right: Path) -> Divergence | None: + with left.open("rb") as left_stream, right.open("rb") as right_stream: + lines: Iterator[tuple[bytes | None, bytes | None]] = itertools.zip_longest( + left_stream, right_stream + ) + for number, (left_line, right_line) in enumerate(lines, start=1): + left_text: str = _line_text(left_line) + right_text: str = _line_text(right_line) + if left_text != right_text: + return Divergence( + location=f"line {number}", left=preview(left_text), right=preview(right_text) + ) + return None + + +def _line_text(line: bytes | None) -> str: + if line is None: + return MISSING_VALUE + return normalize_artifact_text(line.decode("utf-8", "surrogateescape").rstrip("\n")) diff --git a/scripts/compiler_differential/_helpers/comparing/compare.py b/scripts/compiler_differential/_helpers/comparing/compare.py index 3ee796c4c3..4d6af02808 100644 --- a/scripts/compiler_differential/_helpers/comparing/compare.py +++ b/scripts/compiler_differential/_helpers/comparing/compare.py @@ -3,25 +3,37 @@ from __future__ import annotations import json +from collections.abc import Callable from typing import cast from scripts.compiler_differential.constants import MISSING_VALUE, VALUE_PREVIEW_CHARACTERS from scripts.compiler_differential.models import Divergence +from scripts.compiler_differential.types import NodeResolvers +from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_SHARED_MARKER -def first_json_difference(*, left: object, right: object, pointer: str = "") -> Divergence | None: - """Return the first JSON pointer whose value, type, or object key order differs.""" +def first_json_difference( + *, left: object, right: object, pointer: str = "", resolve: NodeResolvers | None = None +) -> Divergence | None: + """Return the first JSON pointer whose value, type, or key order differs, via `resolve`.""" + if resolve is not None: + left_digest: str | None = shared_digest(left) + if left_digest is not None and left_digest == shared_digest(right): + return None + left, right = resolve[0](left), resolve[1](right) if type(left) is not type(right): return _divergence(location=pointer or "/", left=left, right=right) left_object: dict[str, object] | None = as_json_object(left) right_object: dict[str, object] | None = as_json_object(right) if left_object is not None and right_object is not None: - return _first_object_difference(left=left_object, right=right_object, pointer=pointer) + return _first_object_difference( + left=left_object, right=right_object, pointer=pointer, resolve=resolve + ) if isinstance(left, list) and isinstance(right, list): for index, (left_item, right_item) in enumerate(zip(left, right, strict=False)): found: Divergence | None = first_json_difference( - left=left_item, right=right_item, pointer=f"{pointer}/{index}" + left=left_item, right=right_item, pointer=f"{pointer}/{index}", resolve=resolve ) if found is not None: return found @@ -29,8 +41,12 @@ def first_json_difference(*, left: object, right: object, pointer: str = "") -> index: int = min(len(left), len(right)) return _divergence( location=f"{pointer}/{index}", - left=left[index] if index < len(left) else MISSING_VALUE, - right=right[index] if index < len(right) else MISSING_VALUE, + left=_shown(value=left[index], resolve=resolve, side=0) + if index < len(left) + else MISSING_VALUE, + right=_shown(value=right[index], resolve=resolve, side=1) + if index < len(right) + else MISSING_VALUE, ) return None if left != right: @@ -50,20 +66,19 @@ def first_text_difference(*, left: str, right: str) -> Divergence | None: right_line: str = right_lines[index] if index < len(right_lines) else MISSING_VALUE if left_line != right_line: return Divergence( - location=f"line {index + 1}", left=_preview(left_line), right=_preview(right_line) + location=f"line {index + 1}", left=preview(left_line), right=preview(right_line) ) return Divergence(location="end of file", left="", right="") -def first_document_difference(*, left: str, right: str) -> Divergence | None: - """Compare two documents as JSON when both parse, otherwise line by line.""" +def shared_digest(value: object) -> str | None: + """Return the digest a stage-capture `{"__shared__": digest}` reference names.""" - try: - left_payload: object = json.loads(left) - right_payload: object = json.loads(right) - except json.JSONDecodeError: - return first_text_difference(left=left, right=right) - return first_json_difference(left=left_payload, right=right_payload) + mapping: dict[str, object] | None = as_json_object(value) + if mapping is None or len(mapping) != 1: + return None + digest: object = mapping.get(STAGE_CAPTURE_SHARED_MARKER) + return digest if isinstance(digest, str) else None def as_json_object(value: object) -> dict[str, object] | None: @@ -79,35 +94,70 @@ def json_pointer_token(key: str) -> str: def _first_object_difference( - *, left: dict[str, object], right: dict[str, object], pointer: str + *, + left: dict[str, object], + right: dict[str, object], + pointer: str, + resolve: NodeResolvers | None, ) -> Divergence | None: for key in (*left, *(key for key in right if key not in left)): location: str = f"{pointer}/{json_pointer_token(key)}" if key not in left or key not in right: return _divergence( location=location, - left=left.get(key, MISSING_VALUE), - right=right.get(key, MISSING_VALUE), + left=_shown(value=left.get(key, MISSING_VALUE), resolve=resolve, side=0), + right=_shown(value=right.get(key, MISSING_VALUE), resolve=resolve, side=1), ) found: Divergence | None = first_json_difference( - left=left[key], right=right[key], pointer=location + left=left[key], right=right[key], pointer=location, resolve=resolve ) if found is not None: return found if list(left) != list(right): return Divergence( location=f"{pointer or '/'} (key order)", - left=_preview(list(left)), - right=_preview(list(right)), + left=preview(list(left)), + right=preview(list(right)), ) return None +def _shown(*, value: object, resolve: NodeResolvers | None, side: int) -> object: + if resolve is None: + return value + return _previewed(value=value, resolve=resolve[side], remaining=VALUE_PREVIEW_CHARACTERS)[0] + + +def _previewed( + *, value: object, resolve: Callable[[object], object], remaining: int +) -> tuple[object, int]: + if remaining <= 0: + return value, remaining + resolved: object = resolve(value) + mapping: dict[str, object] | None = as_json_object(resolved) + if mapping is not None: + shown: dict[str, object] = {} + for key, item in mapping.items(): + shown[key], remaining = _previewed( + value=item, resolve=resolve, remaining=remaining - len(key) + ) + return shown, remaining + if isinstance(resolved, list): + items: list[object] = [] + for item in resolved: + shown_item, remaining = _previewed(value=item, resolve=resolve, remaining=remaining) + items.append(shown_item) + return items, remaining + return resolved, remaining - len(json.dumps(resolved, ensure_ascii=False)) + + def _divergence(*, location: str, left: object, right: object) -> Divergence: - return Divergence(location=location, left=_preview(left), right=_preview(right)) + return Divergence(location=location, left=preview(left), right=preview(right)) + +def preview(value: object) -> str: + """Render a value for a report, shortened to the preview width.""" -def _preview(value: object) -> str: rendered: str = ( value if isinstance(value, str) and value == MISSING_VALUE diff --git a/scripts/compiler_differential/_helpers/comparing/comparison.py b/scripts/compiler_differential/_helpers/comparing/comparison.py index 97dc2c2eea..f10225a405 100644 --- a/scripts/compiler_differential/_helpers/comparing/comparison.py +++ b/scripts/compiler_differential/_helpers/comparing/comparison.py @@ -4,10 +4,11 @@ import json from collections.abc import Callable +from pathlib import Path +from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference from scripts.compiler_differential._helpers.comparing.compare import ( as_json_object, - first_document_difference, first_json_difference, first_text_difference, ) @@ -237,12 +238,12 @@ def _loads(text: str | None) -> object: def _capture_differences( - *, left: dict[str, dict[str, str]], right: dict[str, dict[str, str]] + *, left: dict[str, dict[str, Path]], right: dict[str, dict[str, Path]] ) -> list[tuple[str, Divergence | None]]: differences: list[tuple[str, Divergence | None]] = [] for command in sorted(set(left) | set(right)): - left_files: dict[str, str] = left.get(command, {}) - right_files: dict[str, str] = right.get(command, {}) + left_files: dict[str, Path] = left.get(command, {}) + right_files: dict[str, Path] = right.get(command, {}) for name in sorted(set(left_files) | set(right_files)): artifact: str = f"stage capture {command}/{name}" if name not in left_files or name not in right_files: @@ -258,10 +259,7 @@ def _capture_differences( differences.append( ( artifact, - first_document_difference( - left=normalize_artifact_text(left_files[name]), - right=normalize_artifact_text(right_files[name]), - ), + first_capture_difference(left=left_files[name], right=right_files[name]), ) ) return differences diff --git a/scripts/compiler_differential/_helpers/comparing/normalize.py b/scripts/compiler_differential/_helpers/comparing/normalize.py index 728ed1c84f..d976cecb28 100644 --- a/scripts/compiler_differential/_helpers/comparing/normalize.py +++ b/scripts/compiler_differential/_helpers/comparing/normalize.py @@ -6,18 +6,16 @@ from scripts.compiler_differential.constants import ( ELAPSED_TIME_MASK, ELAPSED_TIME_PATTERN, - ENGINE_NAMESPACE_PATTERN, - INVOCATION_ID_MASK, - INVOCATION_ID_PATTERN, MANIFEST_VOLATILE_METADATA, STRIPPED_REPORT_FIELDS, ) +from sqlbuild.compiler.frontier.classes.stage_capture_encoder import mask_capture_noise def normalize_artifact_text(text: str) -> str: """Mask invocation ids and engine store suffixes, which differ by design.""" - return ENGINE_NAMESPACE_PATTERN.sub("", INVOCATION_ID_PATTERN.sub(INVOCATION_ID_MASK, text)) + return mask_capture_noise(text) def normalize_stderr(text: str) -> str: diff --git a/scripts/compiler_differential/_helpers/comparing/project_comparison.py b/scripts/compiler_differential/_helpers/comparing/project_comparison.py index c1e2f3af67..7e5c1397c4 100644 --- a/scripts/compiler_differential/_helpers/comparing/project_comparison.py +++ b/scripts/compiler_differential/_helpers/comparing/project_comparison.py @@ -13,7 +13,12 @@ from scripts.compiler_differential._helpers.coverage.discovery import project_discovery_kinds from scripts.compiler_differential._helpers.coverage.render import project_render_kinds from scripts.compiler_differential._helpers.running.execution import run_engine -from scripts.compiler_differential.constants import ERROR_SEVERITY, WARNING_SEVERITY +from scripts.compiler_differential.constants import ( + ERROR_SEVERITY, + LEFT_SIDE, + RIGHT_SIDE, + WARNING_SEVERITY, +) from scripts.compiler_differential.models import ( CommandOutcome, CorpusProject, @@ -42,6 +47,7 @@ def compare_project(*, project: CorpusProject, options: DifferentialOptions) -> source_dir=source_dir, case_dir=case_dir, engine=left_engine, + side=LEFT_SIDE, options=options, ) right: EngineRun = run_engine( @@ -49,6 +55,7 @@ def compare_project(*, project: CorpusProject, options: DifferentialOptions) -> source_dir=source_dir, case_dir=case_dir, engine=right_engine, + side=RIGHT_SIDE, options=options, ) counted: bool = project.seed_coverage and options.stage_captures diff --git a/scripts/compiler_differential/_helpers/corpus/emitted_codes.py b/scripts/compiler_differential/_helpers/corpus/emitted_codes.py index 810e3f1480..90b28c7d88 100644 --- a/scripts/compiler_differential/_helpers/corpus/emitted_codes.py +++ b/scripts/compiler_differential/_helpers/corpus/emitted_codes.py @@ -17,6 +17,7 @@ DIAGNOSTIC_CODE_PATTERN, ERROR_SEVERITY, EXPECT_SUCCESS, + LEFT_SIDE, RENDER_CODE_SCAN_EXCLUDED, RENDER_RAISED_CODES, WARNING_SEVERITY, @@ -139,6 +140,7 @@ def _emitted_codes(*, project: CorpusProject, options: DifferentialOptions) -> E source_dir=source_dir, case_dir=case_dir, engine=options.engines[0], + side=LEFT_SIDE, options=options, ) return EmittedCodes( diff --git a/scripts/compiler_differential/_helpers/coverage/capture_values.py b/scripts/compiler_differential/_helpers/coverage/capture_values.py index 40e2d90287..16d29d8956 100644 --- a/scripts/compiler_differential/_helpers/coverage/capture_values.py +++ b/scripts/compiler_differential/_helpers/coverage/capture_values.py @@ -2,11 +2,11 @@ from __future__ import annotations -import json import re from collections.abc import Iterator from scripts.compiler_differential._helpers.comparing.compare import as_json_object +from scripts.compiler_differential.classes.capture_file import expand_capture_text from scripts.compiler_differential.constants import ( CALLABLE_MARKER, COLLECTION_CAPTURE_MARKERS, @@ -87,7 +87,7 @@ def capture_problems( ) -> tuple[str, ...]: """Return why a capture is incomplete or not canonical; empty when it is sound.""" - capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {} + capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {} problems: list[str] = [] captured: tuple[str, ...] = tuple(key for key in capture if key != TYPE_MARKER) if captured != fields: diff --git a/scripts/compiler_differential/_helpers/coverage/discovery.py b/scripts/compiler_differential/_helpers/coverage/discovery.py index 37e800e26f..bb4d871983 100644 --- a/scripts/compiler_differential/_helpers/coverage/discovery.py +++ b/scripts/compiler_differential/_helpers/coverage/discovery.py @@ -17,6 +17,7 @@ named_values, records, ) +from scripts.compiler_differential.classes.capture_file import expand_capture_text from scripts.compiler_differential.constants import ( CALLABLE_CAPTURE_FIELDS, CRLF_BYTES, @@ -59,7 +60,7 @@ def discovery_collection_kinds() -> tuple[str, ...]: def project_discovery_kinds( - *, captures: dict[str, dict[str, str]], source_dir: Path + *, captures: dict[str, dict[str, Path]], source_dir: Path ) -> frozenset[str]: """Return the kinds of every discovery capture plus CRLF, which only the bytes show.""" @@ -75,7 +76,7 @@ def project_discovery_kinds( def discovered_relative_paths(capture_text: str) -> frozenset[str]: """Return the project-relative path of every file a discovery capture says it read.""" - capture: object = json.loads(capture_text) + capture: object = expand_capture_text(capture_text) return frozenset( str(path) for _, value in named_values(value=capture, names=RELATIVE_PATH_FIELDS) @@ -94,17 +95,17 @@ def authored_byte_kinds(*, source_dir: Path, relative_paths: frozenset[str]) -> ) -def _discovery_capture_texts(captures: dict[str, dict[str, str]]) -> Iterator[str]: +def _discovery_capture_texts(captures: dict[str, dict[str, Path]]) -> Iterator[str]: for files in captures.values(): - for name, text in files.items(): + for name, path in files.items(): if name.endswith(DISCOVERY_STAGE_CAPTURE_SUFFIX): - yield text + yield path.read_text(encoding="utf-8") def discovered_input_kinds(capture_text: str) -> frozenset[str]: """Return the input kinds one canonical discovery capture proves were discovered.""" - capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {} + capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {} kinds: set[str] = { kind for kind in discovery_collection_kinds() if has_content(capture.get(kind)) } diff --git a/scripts/compiler_differential/_helpers/coverage/render.py b/scripts/compiler_differential/_helpers/coverage/render.py index ffd99a14da..828659d4a4 100644 --- a/scripts/compiler_differential/_helpers/coverage/render.py +++ b/scripts/compiler_differential/_helpers/coverage/render.py @@ -6,7 +6,7 @@ import json import re from collections.abc import Iterator -from pathlib import PurePosixPath +from pathlib import Path, PurePosixPath from scripts.compiler_differential._helpers.comparing.compare import as_json_object from scripts.compiler_differential._helpers.coverage.capture_values import ( @@ -20,6 +20,7 @@ macro_function_source, takes_typed_reference, ) +from scripts.compiler_differential.classes.capture_file import expand_capture_text from scripts.compiler_differential.constants import ( RENDER_CALLABLE_CAPTURE_FIELDS, RENDER_DETAIL_KINDS, @@ -77,14 +78,14 @@ def render_collection_kinds() -> tuple[str, ...]: ) -def project_render_kinds(captures: dict[str, dict[str, str]]) -> frozenset[str]: +def project_render_kinds(captures: dict[str, dict[str, Path]]) -> frozenset[str]: """Return the render kinds every `CompileProjectInputs` capture of one project proves.""" kinds: set[str] = set() for files in captures.values(): - for name, text in files.items(): + for name, path in files.items(): if name.endswith(RENDER_STAGE_CAPTURE_SUFFIX): - kinds.update(rendered_input_kinds(text)) + kinds.update(rendered_input_kinds(path.read_text(encoding="utf-8"))) return frozenset(kinds) @@ -101,7 +102,7 @@ def render_capture_problems(capture_text: str) -> tuple[str, ...]: def rendered_input_kinds(capture_text: str) -> frozenset[str]: """Return the render input kinds one canonical `CompileProjectInputs` capture proves.""" - capture: dict[str, object] = as_json_object(json.loads(capture_text)) or {} + capture: dict[str, object] = as_json_object(expand_capture_text(capture_text)) or {} kinds: set[str] = {kind for kind in render_collection_kinds() if has_content(capture.get(kind))} usages: list[tuple[str, dict[str, object]]] = list(_usages(capture)) kinds.update(_macro_kinds(capture=capture, usages=usages)) diff --git a/scripts/compiler_differential/_helpers/running/execution.py b/scripts/compiler_differential/_helpers/running/execution.py index 4835622891..5e1b13c5c4 100644 --- a/scripts/compiler_differential/_helpers/running/execution.py +++ b/scripts/compiler_differential/_helpers/running/execution.py @@ -45,9 +45,10 @@ def run_engine( source_dir: Path, case_dir: Path, engine: str, + side: str, options: DifferentialOptions, ) -> EngineRun: - """Run the project's commands under one engine in a fresh copy at a fixed path.""" + """Run the project's commands under one engine in a fresh copy, keeping captures on disk.""" workspace: Path = case_dir / PROJECT_DIRECTORY shutil.rmtree(workspace, ignore_errors=True) @@ -57,7 +58,7 @@ def run_engine( if project.project_subdirectory is None else workspace / project.project_subdirectory ) - capture_root: Path = case_dir / f"{engine}-{CAPTURES_DIRECTORY}" + capture_root: Path = case_dir / f"{side}-{engine}-{CAPTURES_DIRECTORY}" shutil.rmtree(capture_root, ignore_errors=True) outcomes: list[CommandOutcome] = [] for index, command in enumerate(project.commands): @@ -166,16 +167,10 @@ def _optional_text(path: Path) -> str | None: return path.read_text(encoding="utf-8", errors="surrogateescape") if path.is_file() else None -def _captures(root: Path) -> dict[str, dict[str, str]]: - captures: dict[str, dict[str, str]] = {} +def _captures(root: Path) -> dict[str, dict[str, Path]]: + captures: dict[str, dict[str, Path]] = {} for command_dir in sorted(root.iterdir()) if root.is_dir() else (): - captures[command_dir.name] = _capture_files(command_dir) + captures[command_dir.name] = { + capture.name: capture for capture in sorted(command_dir.iterdir()) if capture.is_file() + } return captures - - -def _capture_files(command_dir: Path) -> dict[str, str]: - return { - capture.name: capture.read_text(encoding="utf-8") - for capture in sorted(command_dir.iterdir()) - if capture.is_file() - } diff --git a/scripts/compiler_differential/classes/capture_file.py b/scripts/compiler_differential/classes/capture_file.py new file mode 100644 index 0000000000..bc73aaa489 --- /dev/null +++ b/scripts/compiler_differential/classes/capture_file.py @@ -0,0 +1,92 @@ +"""Read one stage capture from disk node by node instead of parsing it whole.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from scripts.compiler_differential._helpers.comparing.compare import as_json_object, shared_digest +from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text +from sqlbuild.compiler.frontier.constants import ( + STAGE_CAPTURE_ROOT_KEY, + STAGE_CAPTURE_SHARED_NODES_KEY, +) + +_SHARED_HEADER: bytes = f"{{{json.dumps(STAGE_CAPTURE_SHARED_NODES_KEY)}: {{\n".encode() +_ROOT_PREFIX: bytes = f"{json.dumps(STAGE_CAPTURE_ROOT_KEY)}: ".encode() +_SHARED_DOCUMENT_KEYS: frozenset[str] = frozenset( + {STAGE_CAPTURE_SHARED_NODES_KEY, STAGE_CAPTURE_ROOT_KEY} +) +_DIGEST_CHARACTERS: int = 64 +_NODE_PREFIX_BYTES: int = _DIGEST_CHARACTERS + len(b'"": ') + + +class CaptureFile: + """A normalized stage capture whose shared nodes are parsed only when resolved.""" + + def __init__(self, path: Path) -> None: + self._path: Path = path + self._offsets: dict[str, int] = {} + self.root: object = self._index() + + def resolve(self, value: object) -> object: + """Return the shared node a reference names, or the value itself.""" + + digest: str | None = shared_digest(value) + return value if digest is None else self._node(digest) + + def _index(self) -> object: + with self._path.open("rb") as stream: + if stream.readline() != _SHARED_HEADER: + return _parse(self._path.read_bytes()) + root: object = None + offset: int = stream.tell() + for line in iter(stream.readline, b""): + if line.startswith(_ROOT_PREFIX): + root = _parse(line[len(_ROOT_PREFIX) :]) + elif line.startswith(b'"') and len(line) > _NODE_PREFIX_BYTES: + self._offsets[line[1 : _DIGEST_CHARACTERS + 1].decode("ascii")] = offset + offset += len(line) + return root + + def _node(self, digest: str) -> object: + offset: int | None = self._offsets.get(digest) + if offset is None: + raise json.JSONDecodeError(f"missing shared node {digest}", str(self._path), 0) + with self._path.open("rb") as stream: + _ = stream.seek(offset) + line: bytes = stream.readline() + return _parse(line[_NODE_PREFIX_BYTES:]) + + +def expand_capture_text(text: str) -> object: + """Parse a whole capture document, replacing every shared reference by its node.""" + + document: object = json.loads(text) + mapping: dict[str, object] | None = as_json_object(document) + nodes: dict[str, object] | None = as_json_object( + None if mapping is None else mapping.get(STAGE_CAPTURE_SHARED_NODES_KEY) + ) + if mapping is None or nodes is None or set(mapping) != _SHARED_DOCUMENT_KEYS: + return document + expanded: dict[str, object] = {} + for digest, node in nodes.items(): + expanded[digest] = _expanded(value=node, nodes=expanded) + return _expanded(value=mapping[STAGE_CAPTURE_ROOT_KEY], nodes=expanded) + + +def _expanded(*, value: object, nodes: dict[str, object]) -> object: + digest: str | None = shared_digest(value) + if digest is not None: + return nodes[digest] + items: dict[str, object] | None = as_json_object(value) + if items is not None: + return {key: _expanded(value=item, nodes=nodes) for key, item in items.items()} + if isinstance(value, list): + return [_expanded(value=item, nodes=nodes) for item in value] + return value + + +def _parse(data: bytes) -> object: + text: str = data.decode("utf-8").rstrip("\n") + return json.loads(normalize_artifact_text(text.removesuffix(","))) diff --git a/scripts/compiler_differential/constants.py b/scripts/compiler_differential/constants.py index e9429d3521..8b72d0b5c7 100644 --- a/scripts/compiler_differential/constants.py +++ b/scripts/compiler_differential/constants.py @@ -17,15 +17,14 @@ DUCKDB_ADAPTER: str = "duckdb" PROJECT_DIRECTORY: str = "project" CAPTURES_DIRECTORY: str = "captures" +LEFT_SIDE: str = "left" +RIGHT_SIDE: str = "right" TARGET_DIRECTORY: str = "target" COMPILED_DIRECTORY: str = "target/compiled" MANIFEST_FILE: str = "target/manifest.json" DAG_FILE: str = "target/sqlbuild_dag.json" STRIPPED_REPORT_FIELDS: frozenset[str] = frozenset({"compile_timings", "compiler_engine"}) MANIFEST_VOLATILE_METADATA: frozenset[str] = frozenset({"generated_at", "invocation_id"}) -INVOCATION_ID_PATTERN: re.Pattern[str] = re.compile(r"\b\d{8}T\d{6}Z_[0-9a-f]{12}\b") -INVOCATION_ID_MASK: str = "" -ENGINE_NAMESPACE_PATTERN: re.Pattern[str] = re.compile(r"-native-v\d+\b") ELAPSED_TIME_PATTERN: re.Pattern[str] = re.compile(r"\b\d+(?:\.\d+)?\s?m?s\b") ELAPSED_TIME_MASK: str = "" COMMAND_TIMEOUT_SECONDS: float = 900.0 diff --git a/scripts/compiler_differential/models.py b/scripts/compiler_differential/models.py index 9237e10793..5a399c2e9c 100644 --- a/scripts/compiler_differential/models.py +++ b/scripts/compiler_differential/models.py @@ -57,7 +57,7 @@ class EngineRun: compiled: dict[str, bytes] manifest: str | None dag: str | None - captures: dict[str, dict[str, str]] = field(default_factory=dict) + captures: dict[str, dict[str, Path]] = field(default_factory=dict) @dataclass(frozen=True) diff --git a/scripts/compiler_differential/types.py b/scripts/compiler_differential/types.py new file mode 100644 index 0000000000..7ee809df54 --- /dev/null +++ b/scripts/compiler_differential/types.py @@ -0,0 +1,5 @@ +"""Type aliases for the compiler differential harness.""" + +from collections.abc import Callable + +type NodeResolvers = tuple[Callable[[object], object], Callable[[object], object]] diff --git a/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py b/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py index da8a53de2a..fc356661bd 100644 --- a/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py +++ b/src/sqlbuild/compiler/frontier/_helpers/stage_capture.py @@ -1,11 +1,13 @@ """Dump frontier objects as canonical JSON so two engines can be compared stage by stage.""" -import json +import io import os from pathlib import Path +from typing import TextIO from sqlbuild.compiler.frontier.classes.stage_capture_encoder import StageCaptureEncoder from sqlbuild.compiler.frontier.classes.stage_capture_sequence import StageCaptureSequence +from sqlbuild.compiler.frontier.classes.stage_capture_writer import StageCaptureWriter from sqlbuild.compiler.frontier.constants import STAGE_CAPTURE_DIR_ENV_VAR, STAGE_CAPTURE_SUFFIX from sqlbuild.compiler.frontier.types import CompilerStage @@ -20,18 +22,18 @@ def write_stage_capture(*, stage: CompilerStage, value: object) -> None: f"{StageCaptureSequence.next_value():03d}-{stage.value}{STAGE_CAPTURE_SUFFIX}" ) path.parent.mkdir(parents=True, exist_ok=True) - _ = path.write_text(render_stage_capture(value), encoding="utf-8") + with path.open("w", encoding="utf-8", newline="\n") as stream: + _write_capture(value=value, stream=stream) def render_stage_capture(value: object) -> str: - """Render a value as indented JSON that keeps insertion order and sorts only sets.""" - - return ( - json.dumps( - StageCaptureEncoder().encode(value), - indent=1, - ensure_ascii=False, - allow_nan=False, - ) - + "\n" - ) + """Render a value as the JSON document `write_stage_capture` writes.""" + + stream: io.StringIO = io.StringIO() + _write_capture(value=value, stream=stream) + return stream.getvalue() + + +def _write_capture(*, value: object, stream: TextIO) -> None: + writer: StageCaptureWriter = StageCaptureWriter(stream) + writer.finish(StageCaptureEncoder(on_shared=writer.write_shared).encode(value)) diff --git a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py index 0a24086bec..4807ca8f74 100644 --- a/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py +++ b/src/sqlbuild/compiler/frontier/classes/stage_capture_encoder.py @@ -12,14 +12,20 @@ import math import re import types -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from collections.abc import Set as AbstractSet from pathlib import PurePath from typing import Any from sqlbuild.compiler.frontier.constants import ( STAGE_CAPTURE_DECODED_SEQUENCES, + STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN, + STAGE_CAPTURE_INVOCATION_ID_MASK, + STAGE_CAPTURE_INVOCATION_ID_PATTERN, STAGE_CAPTURE_OMITTED_ATTRIBUTES, + STAGE_CAPTURE_RESERVED_KEYS, + STAGE_CAPTURE_SHARED_MARKER, + STAGE_CAPTURE_SHARED_MIN_BYTES, STAGE_CAPTURE_SKIPPED_SLOTS, STAGE_CAPTURE_UNORDERED_ATTRIBUTES, ) @@ -34,17 +40,29 @@ class StageCaptureEncoder: - """Encode one object graph deterministically, replacing callables with qualified names.""" - - def __init__(self) -> None: + """Encode an object graph deterministically, sharing large subtrees by content digest.""" + + def __init__( + self, + *, + on_shared: Callable[..., None] | None = None, + share_min_bytes: int = STAGE_CAPTURE_SHARED_MIN_BYTES, + ) -> None: self._active: set[int] = set() + self._cycles: int = 0 + self._encoded: dict[int, tuple[object, object]] = {} + self._on_shared: Callable[..., None] | None = on_shared + self._share_min_bytes: int = share_min_bytes + self._digests: set[str] = set() def encode(self, value: object) -> object: """Return a JSON-compatible value that keeps insertion order and sorts only sets.""" if isinstance(value, enum.Enum): return {"__enum__": f"{qualified_name(type(value))}.{value.name}"} - if value is None or isinstance(value, (bool, int, str)): + if isinstance(value, str): + return value if len(value) < self._share_min_bytes else self._share(value) + if value is None or isinstance(value, (bool, int)): return value if isinstance(value, float): return value if math.isfinite(value) else {"__float__": repr(value)} @@ -53,12 +71,32 @@ def encode(self, value: object) -> object: return scalar identity: int = id(value) if identity in self._active: + self._cycles += 1 return {"__cycle__": qualified_name(type(value))} + known: tuple[object, object] | None = self._encoded.get(identity) + if known is not None: + return known[1] + cycles: int = self._cycles self._active.add(identity) try: - return self._encode_container(value) + encoded: object = self._share(self._encode_container(value)) finally: self._active.discard(identity) + if self._cycles == cycles: + self._encoded[identity] = (value, encoded) + return encoded + + def _share(self, encoded: object) -> object: + if self._on_shared is None or not isinstance(encoded, (dict, list, str)): + return encoded + text: str = json.dumps(encoded, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + if len(text) < self._share_min_bytes: + return encoded + digest: str = hashlib.sha256(mask_capture_noise(text).encode("utf-8")).hexdigest() + if digest not in self._digests: + self._digests.add(digest) + self._on_shared(digest=digest, text=text) + return {STAGE_CAPTURE_SHARED_MARKER: digest} def _encode_container(self, value: object) -> object: type_name: str = qualified_name(type(value)) @@ -104,7 +142,9 @@ def _encode_attribute(self, *, value: object, unordered: bool) -> object: return {"__unordered_mapping__": sorted(self._mapping_pairs(value), key=_sort_key)} def _encode_mapping(self, value: Mapping[Any, object]) -> object: - if all(isinstance(key, str) for key in value): + if all(isinstance(key, str) for key in value) and STAGE_CAPTURE_RESERVED_KEYS.isdisjoint( + value + ): return {str(key): self.encode(item) for key, item in value.items()} return {"__mapping__": self._mapping_pairs(value)} @@ -141,8 +181,17 @@ def _encode_scalar(value: object) -> object | None: return None +def mask_capture_noise(text: str) -> str: + """Mask invocation ids and native store suffixes, which differ between identical compiles.""" + + return STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN.sub( + "", + STAGE_CAPTURE_INVOCATION_ID_PATTERN.sub(STAGE_CAPTURE_INVOCATION_ID_MASK, text), + ) + + def _sort_key(item: object) -> str: - return json.dumps(item, sort_keys=True, ensure_ascii=False) + return mask_capture_noise(json.dumps(item, sort_keys=True, ensure_ascii=False)) def _instance_attributes(value: object) -> dict[str, object] | None: diff --git a/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py b/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py new file mode 100644 index 0000000000..1dd17b9ad1 --- /dev/null +++ b/src/sqlbuild/compiler/frontier/classes/stage_capture_writer.py @@ -0,0 +1,39 @@ +"""Stream one stage capture so shared subtrees reach the file as they are encoded.""" + +from __future__ import annotations + +import json +from typing import TextIO + +from sqlbuild.compiler.frontier.constants import ( + STAGE_CAPTURE_ROOT_KEY, + STAGE_CAPTURE_SHARED_NODES_KEY, +) + + +class StageCaptureWriter: + """Write a capture as one JSON document with each shared node and the root on its own line.""" + + def __init__(self, stream: TextIO) -> None: + self._stream: TextIO = stream + self._shared: bool = False + + def write_shared(self, *, digest: str, text: str) -> None: + """Append one shared node, given as compact JSON.""" + + prefix: str = ( + ",\n" if self._shared else f"{{{json.dumps(STAGE_CAPTURE_SHARED_NODES_KEY)}: {{\n" + ) + self._shared = True + _ = self._stream.write(f"{prefix}{json.dumps(digest)}: {text}") + + def finish(self, root: object) -> None: + """Write the root and close the document.""" + + if not self._shared: + _ = self._stream.write( + json.dumps(root, indent=1, ensure_ascii=False, allow_nan=False) + "\n" + ) + return + compact: str = json.dumps(root, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + _ = self._stream.write(f"\n}},\n{json.dumps(STAGE_CAPTURE_ROOT_KEY)}: {compact}\n}}\n") diff --git a/src/sqlbuild/compiler/frontier/constants.py b/src/sqlbuild/compiler/frontier/constants.py index 8f6ed2fd05..09445fd0de 100644 --- a/src/sqlbuild/compiler/frontier/constants.py +++ b/src/sqlbuild/compiler/frontier/constants.py @@ -1,5 +1,7 @@ """Compiler engine switch, cache namespace, and stage capture constants.""" +import re + from sqlbuild.compiler.frontier.types import CompilerEngine COMPILER_ENGINE_ENV_VAR: str = "SQLBUILD_COMPILER_ENGINE" @@ -12,15 +14,22 @@ COMPILER_CACHE_DIRECTORY_NAME: str = "compiler" STAGE_CAPTURE_DIR_ENV_VAR: str = "SQLBUILD_COMPILER_STAGE_CAPTURE_DIR" STAGE_CAPTURE_SUFFIX: str = ".json" +STAGE_CAPTURE_SHARED_MARKER: str = "__shared__" +STAGE_CAPTURE_ROOT_KEY: str = "__capture__" +STAGE_CAPTURE_SHARED_NODES_KEY: str = "__shared_nodes__" +STAGE_CAPTURE_SHARED_MIN_BYTES: int = 1024 +STAGE_CAPTURE_RESERVED_KEYS: frozenset[str] = frozenset( + {STAGE_CAPTURE_SHARED_MARKER, STAGE_CAPTURE_ROOT_KEY, STAGE_CAPTURE_SHARED_NODES_KEY} +) +STAGE_CAPTURE_INVOCATION_ID_PATTERN: re.Pattern[str] = re.compile(r"\b\d{8}T\d{6}Z_[0-9a-f]{12}\b") +STAGE_CAPTURE_INVOCATION_ID_MASK: str = "" +STAGE_CAPTURE_ENGINE_NAMESPACE_PATTERN: re.Pattern[str] = re.compile(r"-native-v\d+\b") STAGE_CAPTURE_SKIPPED_SLOTS: frozenset[str] = frozenset({"__dict__", "__weakref__"}) STAGE_CAPTURE_UNORDERED_ATTRIBUTES: dict[str, frozenset[str]] = { "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset( {"schemas", "analysis_shapes", "expression_shapes"} ) } -# The shared-analysis memo is a reuse cache whose representatives and lazily filled caches -# depend on analysis batching; compact lineage facts index a batch-wide string pool, so they -# are captured by their decoded facts. STAGE_CAPTURE_OMITTED_ATTRIBUTES: dict[str, frozenset[str]] = { "sqlbuild.compiler.sql_analysis.classes.binding_catalog:BindingCatalog": frozenset( {"shared_analyses"} diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py index 0d3737b292..0e6963f61c 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py @@ -445,6 +445,7 @@ class DataflowCaptureCase: description: str write_project: Callable[[Path], object] schedules: tuple[DataflowScheduleCase, ...] + expected_difference: str | None = None @dataclass(frozen=True) diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py index ad98206841..52cea2d482 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/helpers.py @@ -20,7 +20,6 @@ from scripts.cold_compile_performance._helpers.dense_project import write_dense_compile_project from scripts.cold_compile_performance._helpers.random_dag_project import write_random_dag_project from scripts.cold_compile_performance.models import RandomDagProject -from scripts.compiler_differential._helpers.comparing.normalize import normalize_artifact_text from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.cli.commands.main.entrypoint.entry import main @@ -315,15 +314,14 @@ def compiled_project_capture( capture_dir: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, -) -> str: - """Cold-compile once with stage captures; return the normalized compiled-project capture.""" +) -> Path: + """Cold-compile once with stage captures; return the compiled-project capture file.""" with monkeypatch.context() as capture_patch: capture_patch.setenv(STAGE_CAPTURE_DIR_ENV_VAR, str(capture_dir)) _ = main(["--project-dir", str(project_dir), "compile", "--json", "--no-cache"]) _ = capsys.readouterr() - capture: Path = next(capture_dir.glob(f"*-{CompilerStage.COMPILED_PROJECT.value}.json")) - return normalize_artifact_text(capture.read_text(encoding="utf-8")) + return next(capture_dir.glob(f"*-{CompilerStage.COMPILED_PROJECT.value}.json")) def reshape_models(*, project_dir: Path, names: tuple[str, ...], indexes: tuple[int, ...]) -> None: diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py index a9906f7bbf..6b951d7fa0 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_dataflow_analysis_schedule.py @@ -9,8 +9,9 @@ from scripts.cold_compile_performance._helpers.random_dag_project import write_random_dag_project from scripts.cold_compile_performance.models import RandomDagProject -from scripts.compiler_differential._helpers.comparing.compare import first_document_difference +from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference from scripts.compiler_differential.main.generate_project import generate_project +from scripts.compiler_differential.models import Divergence from sqlbuild.cli.commands.main.entrypoint.entry import main from tests.integration.src.sqlbuild.compiler.pipeline._test_types import ( CompileOutcome, @@ -32,6 +33,7 @@ interrupt_reacquiring_analysis, live_model_analysis_threads, perturb_dataflow_schedule, + random_dag_writer, reshape_models, use_wave_analysis, ) @@ -49,10 +51,6 @@ _INTERRUPT_NOTICE: str = "Interrupted; finishing in-flight model analysis..." -def _write_random_dag(project_dir: Path) -> None: - _ = write_random_dag_project(project_dir=project_dir, project=_PROJECT) - - @pytest.mark.parametrize( "test_case", [ @@ -100,7 +98,7 @@ def test_given_perturbed_schedule_when_compiling_repeatedly_then_output_is_ident [ DataflowCaptureCase( "random dag", - _write_random_dag, + random_dag_writer(_PROJECT), _CAPTURE_SCHEDULES, ), DataflowCaptureCase( @@ -121,7 +119,7 @@ def test_given_perturbed_schedule_when_capturing_compiled_project_then_capture_i _ = test_case.write_project(project_dir) with monkeypatch.context() as waves_patch: use_wave_analysis(waves_patch) - reference: str = compiled_project_capture( + reference: Path = compiled_project_capture( project_dir=project_dir, capture_dir=tmp_path / "reference", capsys=capsys, @@ -130,14 +128,15 @@ def test_given_perturbed_schedule_when_capturing_compiled_project_then_capture_i for index, schedule in enumerate(test_case.schedules): with monkeypatch.context() as schedule_patch: perturb_dataflow_schedule(monkeypatch=schedule_patch, case=schedule, seed=index) - actual: str = compiled_project_capture( + actual: Path = compiled_project_capture( project_dir=project_dir, capture_dir=tmp_path / f"schedule-{index}", capsys=capsys, monkeypatch=monkeypatch, ) - assert first_document_difference(left=reference, right=actual) is None, schedule + difference: Divergence | None = first_capture_difference(left=reference, right=actual) + assert getattr(difference, "location", None) == test_case.expected_difference, schedule @pytest.mark.parametrize( diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py index f799bdcdf4..600124eabf 100644 --- a/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py +++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/_test_types.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass @@ -52,3 +53,34 @@ class PayloadStripTestCase: description: str payload: object expected_payload: object + + +@dataclass(frozen=True) +class CaptureDifferenceTestCase: + """Two frontier values captured to disk and where their first difference must be reported.""" + + description: str + left: Callable[[], object] + right: Callable[[], object] + expected_location: str | None + + +@dataclass(frozen=True) +class CaptureTextDifferenceTestCase: + """Two capture files that are not both JSON and the first differing line to report.""" + + description: str + left: str + right: str + expected_location: str | None + + +@dataclass(frozen=True) +class CapturePreviewTestCase: + """Two captures whose first difference is a missing value and the preview it must show.""" + + description: str + left: Callable[[], object] + right: Callable[[], object] + expected_location: str + expected_left_prefix: str diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py new file mode 100644 index 0000000000..09c826ff02 --- /dev/null +++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/helpers.py @@ -0,0 +1,80 @@ +"""Neutral frontier-like values whose captures share large repeated subtrees.""" + +from __future__ import annotations + +from typing import cast + +_LINE_COUNT: int = 40 +_ORDER_COUNT: int = 6 +_RUN_ID: str = "20261006T172510Z_dee386e8ffa7" +_OTHER_RUN_ID: str = "20261007T081500Z_0a1b2c3d4e5f" +_CHANGED_QUANTITY: int = 99 + + +def orders_batch() -> dict[str, object]: + """Return orders that repeat two identical large orders.""" + + return { + "run_id": _RUN_ID, + "orders": [_order(order_id=index % 2) for index in range(_ORDER_COUNT)], + } + + +def orders_batch_with_changed_line(*, order_index: int, line_index: int) -> dict[str, object]: + """Return the repeated orders with one quantity of one order changed.""" + + batch: dict[str, object] = orders_batch() + orders: list[dict[str, object]] = cast(list[dict[str, object]], batch["orders"]) + lines: list[dict[str, object]] = cast(list[dict[str, object]], orders[order_index]["lines"]) + lines[line_index]["quantity"] = _CHANGED_QUANTITY + return batch + + +def orders_batch_for_run() -> dict[str, object]: + """Return the repeated orders, each recording the default run id.""" + + return _orders_recording(run_id=_RUN_ID) + + +def orders_batch_from_other_run() -> dict[str, object]: + """Return the same orders, each recording a different run id.""" + + return _orders_recording(run_id=_OTHER_RUN_ID) + + +def _orders_recording(*, run_id: str) -> dict[str, object]: + return { + "run_id": run_id, + "orders": [ + {**_order(order_id=index % 2), "run_id": run_id} for index in range(_ORDER_COUNT) + ], + } + + +def _order(*, order_id: int) -> dict[str, object]: + return {"order_id": order_id, "lines": _order_lines()} + + +def _order_lines() -> list[dict[str, object]]: + return [{"sku": f"product-{line:04d}", "quantity": line} for line in range(_LINE_COUNT)] + + +def store_paths(*, native_suffix: str) -> dict[str, object]: + """Return a set of two large store paths, one carrying an engine namespace suffix.""" + + return {"paths": frozenset({"a" * 1100 + f"/store{native_suffix}", "b" * 1100})} + + +def marker_shaped_orders(*, name: str) -> dict[str, object]: + """Return authored data whose only key looks like a shared-node reference marker.""" + + return {"__shared__": name} + + +def orders_batch_without_last() -> dict[str, object]: + """Return the repeated orders with the last one removed.""" + + batch: dict[str, object] = orders_batch() + orders: list[dict[str, object]] = cast(list[dict[str, object]], batch["orders"]) + batch["orders"] = orders[:-1] + return batch diff --git a/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py b/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py new file mode 100644 index 0000000000..2a23951b6b --- /dev/null +++ b/tests/unit/scripts/compiler_differential/_helpers/comparing/test_capture.py @@ -0,0 +1,154 @@ +"""Stage captures on disk are compared node by node with exact first-difference pointers.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from scripts.compiler_differential._helpers.comparing.capture import first_capture_difference +from scripts.compiler_differential.models import Divergence +from sqlbuild.compiler.frontier._helpers.stage_capture import render_stage_capture +from tests.unit.scripts.compiler_differential._helpers.comparing._test_types import ( + CaptureDifferenceTestCase, + CapturePreviewTestCase, + CaptureTextDifferenceTestCase, +) +from tests.unit.scripts.compiler_differential._helpers.comparing.helpers import ( + marker_shaped_orders, + orders_batch, + orders_batch_for_run, + orders_batch_from_other_run, + orders_batch_with_changed_line, + orders_batch_without_last, + store_paths, +) + + +@pytest.mark.parametrize( + "test_case", + [ + CaptureDifferenceTestCase( + description="identical_shared_captures", + left=orders_batch, + right=orders_batch, + expected_location=None, + ), + CaptureDifferenceTestCase( + description="change_inside_a_shared_node", + left=orders_batch, + right=lambda: orders_batch_with_changed_line(order_index=3, line_index=7), + expected_location="/orders/3/lines/7/quantity", + ), + CaptureDifferenceTestCase( + description="change_to_one_copy_of_a_repeated_node", + left=orders_batch, + right=lambda: orders_batch_with_changed_line(order_index=4, line_index=0), + expected_location="/orders/4/lines/0/quantity", + ), + CaptureDifferenceTestCase( + description="run_identity_inside_shared_nodes_is_masked", + left=orders_batch_for_run, + right=orders_batch_from_other_run, + expected_location=None, + ), + CaptureDifferenceTestCase( + description="large_set_elements_differing_only_in_masked_noise", + left=lambda: store_paths(native_suffix="-native-v3"), + right=lambda: store_paths(native_suffix=""), + expected_location=None, + ), + CaptureDifferenceTestCase( + description="authored_marker_shaped_data_is_not_a_reference", + left=lambda: marker_shaped_orders(name="orders"), + right=lambda: marker_shaped_orders(name="customers"), + expected_location="/__mapping__/0/1", + ), + CaptureDifferenceTestCase( + description="identical_marker_shaped_data", + left=lambda: marker_shaped_orders(name="orders"), + right=lambda: marker_shaped_orders(name="orders"), + expected_location=None, + ), + CaptureDifferenceTestCase( + description="shared_node_against_inline_value", + left=orders_batch, + right=lambda: {**orders_batch(), "orders": [{"order_id": 0}]}, + expected_location="/orders/0/lines", + ), + ], + ids=lambda case: case.description, +) +def test_given_two_captures_when_comparing_then_first_logical_difference_is_reported( + test_case: CaptureDifferenceTestCase, tmp_path: Path +) -> None: + left: Path = tmp_path / "left.json" + right: Path = tmp_path / "right.json" + _ = left.write_text(render_stage_capture(test_case.left()), encoding="utf-8") + _ = right.write_text(render_stage_capture(test_case.right()), encoding="utf-8") + + found: Divergence | None = first_capture_difference(left=left, right=right) + + assert getattr(found, "location", None) == test_case.expected_location + + +@pytest.mark.parametrize( + "test_case", + [ + CapturePreviewTestCase( + description="missing_list_item_previews_the_resolved_node", + left=orders_batch, + right=orders_batch_without_last, + expected_location="/orders/5", + expected_left_prefix='{"order_id": 1, "lines": [{"sku": "product-0000"', + ) + ], + ids=lambda case: case.description, +) +def test_given_missing_shared_value_when_comparing_then_preview_shows_its_content( + test_case: CapturePreviewTestCase, tmp_path: Path +) -> None: + left: Path = tmp_path / "left.json" + right: Path = tmp_path / "right.json" + _ = left.write_text(render_stage_capture(test_case.left()), encoding="utf-8") + _ = right.write_text(render_stage_capture(test_case.right()), encoding="utf-8") + + found: Divergence | None = first_capture_difference(left=left, right=right) + + assert getattr(found, "location", None) == test_case.expected_location + assert getattr(found, "left", "").startswith(test_case.expected_left_prefix) + + +@pytest.mark.parametrize( + "test_case", + [ + CaptureTextDifferenceTestCase( + description="truncated_capture", + left='{"orders": [\n1,\n2\n]}\n', + right='{"orders": [\n1,\n3', + expected_location="line 3", + ), + CaptureTextDifferenceTestCase( + description="identical_invalid_captures", + left="not json\n", + right="not json\n", + expected_location=None, + ), + ], + ids=lambda case: case.description, +) +def test_given_capture_that_is_not_json_when_comparing_then_first_differing_line_is_reported( + test_case: CaptureTextDifferenceTestCase, tmp_path: Path +) -> None: + left: Path = tmp_path / "left.json" + right: Path = tmp_path / "right.json" + _ = left.write_text(test_case.left, encoding="utf-8") + _ = right.write_text(test_case.right, encoding="utf-8") + + found: Divergence | None = first_capture_difference(left=left, right=right) + + assert getattr(found, "location", None) == test_case.expected_location + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py index 8acee8e92d..81d0289c37 100644 --- a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py +++ b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_discovery.py @@ -236,14 +236,19 @@ def test_given_authored_bytes_when_classifying_then_crlf_counts_only_for_discove path: Path = tmp_path / relative_path path.parent.mkdir(parents=True, exist_ok=True) _ = path.write_bytes(contents) - capture: str = json.dumps( - empty_inputs_capture( - model_files=[{"relative_path": {"__path__": path}} for path in test_case.read_paths] - ) + capture: Path = tmp_path / "captures" / "001-discovered_project_inputs.json" + capture.parent.mkdir() + _ = capture.write_text( + json.dumps( + empty_inputs_capture( + model_files=[{"relative_path": {"__path__": path}} for path in test_case.read_paths] + ) + ), + encoding="utf-8", ) kinds: frozenset[str] = project_discovery_kinds( - captures={"0-compile": {"001-discovered_project_inputs.json": capture}}, source_dir=tmp_path + captures={"0-compile": {capture.name: capture}}, source_dir=tmp_path ) assert ("crlf" in kinds) is test_case.expected_crlf diff --git a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py index 5f84d87d99..4aebac3039 100644 --- a/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py +++ b/tests/unit/scripts/compiler_differential/_helpers/coverage/test_render.py @@ -4,6 +4,7 @@ import dataclasses import json +from pathlib import Path import pytest @@ -418,12 +419,14 @@ def test_given_render_capture_when_classifying_then_only_rendered_inputs_are_cre ids=lambda case: case.description, ) def test_given_encoded_compile_inputs_when_classifying_then_encoded_empty_values_prove_nothing( - test_case: RenderEncodedInputsTestCase, + test_case: RenderEncodedInputsTestCase, tmp_path: Path ) -> None: capture: str = render_stage_capture(test_case.inputs) + capture_path: Path = tmp_path / "002-compile_project_inputs.json" + _ = capture_path.write_text(capture, encoding="utf-8") assert rendered_input_kinds(capture) == test_case.expected_kinds - assert project_render_kinds({"0-compile": {"002-compile_project_inputs.json": capture}}) == ( + assert project_render_kinds({"0-compile": {capture_path.name: capture_path}}) == ( test_case.expected_kinds ) diff --git a/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py b/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py index 7049921840..b35c5513e4 100644 --- a/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/frontier/_test_types.py @@ -96,3 +96,12 @@ class FrontierCaptureTestCase: stages: tuple[CompilerStage, ...] expected_results: tuple[object, ...] expected_files: tuple[str, ...] + + +@dataclass(frozen=True) +class SharedCaptureTestCase: + """A value with repeated large subtrees and how many distinct shared nodes its capture holds.""" + + description: str + value: Callable[[], object] + expected_shared_nodes: int diff --git a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py index a0a19a1bc9..0507124821 100644 --- a/tests/unit/src/sqlbuild/compiler/frontier/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/frontier/helpers.py @@ -103,6 +103,22 @@ def orders_binding_catalog( return catalog +def repeated_orders() -> dict[str, object]: + """Return equal large order lines and one long SQL text, each repeated several times.""" + + query_sql: str = "SELECT " + ", ".join(f"amount_{index:03d}" for index in range(120)) + return { + "orders": [ + {"order_id": index, "lines": _order_lines(), "query_sql": query_sql} + for index in range(3) + ], + } + + +def _order_lines() -> list[dict[str, object]]: + return [{"sku": f"product-{line:03d}", "quantity": line} for line in range(40)] + + def order_line() -> OrderLine: """Return one neutral order line.""" diff --git a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py index 8592fc9e08..e589d0e884 100644 --- a/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py +++ b/tests/unit/src/sqlbuild/compiler/frontier/test_stage_capture.py @@ -2,21 +2,26 @@ from __future__ import annotations +import hashlib import json from pathlib import Path import pytest +from scripts.compiler_differential.classes.capture_file import expand_capture_text from sqlbuild.compiler.frontier._helpers.stage_capture import render_stage_capture +from sqlbuild.compiler.frontier.classes.stage_capture_encoder import StageCaptureEncoder from sqlbuild.compiler.frontier.constants import ( COMPILER_ENGINE_ENV_VAR, STAGE_CAPTURE_DIR_ENV_VAR, + STAGE_CAPTURE_SHARED_NODES_KEY, STAGE_CAPTURE_UNORDERED_ATTRIBUTES, ) from sqlbuild.compiler.frontier.main._compile_frontier import compile_frontier from sqlbuild.compiler.frontier.types import CompilerEngine, CompilerStage from tests.unit.src.sqlbuild.compiler.frontier._test_types import ( FrontierCaptureTestCase, + SharedCaptureTestCase, StageCaptureOrderTestCase, StageCaptureTestCase, UnorderedAttributeTestCase, @@ -29,6 +34,7 @@ order_total, orders_binding_catalog, orders_lineage, + repeated_orders, ) _COMPILE_MODELS: str = "sqlbuild.compiler.compile.models" @@ -97,6 +103,11 @@ } ], ), + StageCaptureTestCase( + description="mapping_with_a_reserved_key_is_encoded_as_pairs", + value=lambda: {"__shared__": "orders", "status": "placed"}, + expected_capture={"__mapping__": [["__shared__", "orders"], ["status", "placed"]]}, + ), StageCaptureTestCase( description="non_finite_float_and_bytes", value=lambda: (float("inf"), b"orders"), @@ -239,6 +250,34 @@ def test_given_declared_memo_attribute_when_rendering_capture_then_only_it_is_so assert (first == second) is test_case.expected_identical +@pytest.mark.parametrize( + "test_case", + [ + SharedCaptureTestCase( + description="repeated_lines_and_sql_are_stored_once", + value=repeated_orders, + expected_shared_nodes=2, + ), + ], + ids=lambda case: case.description, +) +def test_given_repeated_large_subtrees_when_rendering_capture_then_each_is_stored_once( + test_case: SharedCaptureTestCase, +) -> None: + text: str = render_stage_capture(test_case.value()) + + shared: dict[str, object] = json.loads(text)[STAGE_CAPTURE_SHARED_NODES_KEY] + assert len(shared) == test_case.expected_shared_nodes + assert expand_capture_text(text) == StageCaptureEncoder().encode(test_case.value()) + assert { + hashlib.sha256( + json.dumps(node, ensure_ascii=False, separators=(",", ":")).encode() + ).hexdigest() + for node in shared.values() + } == set(shared) + assert len(text.splitlines()) == test_case.expected_shared_nodes + 4 + + @pytest.mark.parametrize( "test_case", [