diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 49b2bc8826..86739641ba 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -84,6 +84,11 @@ jobs: broad=true performance=true ;; + src/sqlbuild/cli/commands/_helpers/lineage/*|src/sqlbuild/cli/commands/_helpers/plan/*|src/sqlbuild/cli/commands/_helpers/scope/*|src/sqlbuild/cli/commands/main/inspection/*|src/sqlbuild/cli/commands/main/project/*|src/sqlbuild/cli/output/*|src/sqlbuild/presentation/*) + # Inspection command guards run inside the fresh-process compile jobs. + broad=true + performance=true + ;; src/sqlbuild/rule_engine/*|src/sqlbuild/lint/*|scripts/rules_benchmark/*|scripts/benchmark_rules_ci.py|tests/performance/*) broad=true performance=true diff --git a/Makefile b/Makefile index 15ac68a182..c2513fe738 100644 --- a/Makefile +++ b/Makefile @@ -236,6 +236,7 @@ test-e2e-cold-compile-performance: test -n "$(SQLBUILD_BENCHMARK_MODELS)" env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=4 uv run pytest \ tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_performance.py \ + tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py \ -m cold_compile_performance -k "models_$(SQLBUILD_BENCHMARK_MODELS)" \ -vv --log-cli-level=INFO --color=yes diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py index 4d12fbf4dd..d7b00ebeba 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py @@ -796,6 +796,21 @@ class DagJsonBuildE2ETestCase: expected_absent_fragments: tuple[str, ...] +@dataclass(frozen=True) +class DiamondDagE2ETestCase: + description: str + command: tuple[str, ...] + expected_fragments: tuple[str, ...] + expected_max_lines_per_node: int + + +@dataclass(frozen=True) +class DiamondDagJsonE2ETestCase: + description: str + expected_node_ids: tuple[str, ...] + expected_edge_count: int + + @dataclass(frozen=True) class PlanCommandBuildE2ETestCase: """Test case for plan command surface behavior.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py index 9c333dec08..dd883884a8 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py @@ -10,6 +10,13 @@ from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( DagJsonBuildE2ETestCase, + DiamondDagE2ETestCase, + DiamondDagJsonE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.lineage.helpers import ( + DIAMOND_EDGE_COUNT, + diamond_model_names, + prepare_diamond_lineage_project, ) from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_waffle_shop, run_sqb @@ -84,3 +91,73 @@ def test_given_waffle_shop_when_running_dag_json_then_it_reports_static_graph( assert fragment not in result.stdout assert "query_sql" not in result.stdout assert "action" not in result.stdout + + +@pytest.mark.parametrize( + "test_case", + [ + DiamondDagE2ETestCase( + description="dag text summarizes a layered diamond graph", + command=("--no-color", "dag"), + expected_fragments=( + f"DAG ready ({len(diamond_model_names())} nodes, " + f"{DIAMOND_EDGE_COUNT} edges, 0 checks)", + ), + expected_max_lines_per_node=1, + ), + DiamondDagE2ETestCase( + description="dag json lists a layered diamond graph once per node and edge", + command=("--no-color", "dag", "--json"), + expected_fragments=('"project_name": "diamond_lineage"',), + expected_max_lines_per_node=40, + ), + ], + ids=lambda case: case.description, +) +def test_given_layered_diamonds_when_running_dag_then_output_stays_linear( + test_case: DiamondDagE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diamond_lineage_project(tmp_path=tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=test_case.command, project_dir=project_dir + ) + + assert result.returncode == 0, result.stdout + result.stderr + for fragment in test_case.expected_fragments: + assert fragment in result.stdout, result.stdout + assert len(result.stdout.splitlines()) <= test_case.expected_max_lines_per_node * len( + diamond_model_names() + ) + + +@pytest.mark.parametrize( + "test_case", + [ + DiamondDagJsonE2ETestCase( + description="dag json lists every diamond model and dependency exactly once", + expected_node_ids=tuple(f"model:{name}" for name in diamond_model_names()), + expected_edge_count=DIAMOND_EDGE_COUNT, + ) + ], + ids=lambda case: case.description, +) +def test_given_layered_diamonds_when_running_dag_json_then_lists_each_node_and_edge_once( + test_case: DiamondDagJsonE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diamond_lineage_project(tmp_path=tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "dag", "--json"), project_dir=project_dir + ) + + assert result.returncode == 0, result.stdout + result.stderr + payload: dict[str, list[dict[str, object]]] = json.loads(result.stdout) + node_ids: list[str] = [str(node["id"]) for node in payload["nodes"]] + edge_pairs: list[tuple[str, str]] = [ + (str(edge["from_id"]), str(edge["to_id"])) for edge in payload["edges"] + ] + assert sorted(node_ids) == sorted(test_case.expected_node_ids) + assert len(edge_pairs) == len(set(edge_pairs)) == test_case.expected_edge_count diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md b/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md index 2d271c0008..92bf38481a 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/COMPILE_BENCHMARKS.md @@ -53,3 +53,19 @@ Both guards initialize the process-local SQL runtime with an unrelated 32-model measurement. “Cold” means the measured project has no compiler cache or target artifacts; it does not include one-time lazy initialization noise from the Python test worker. Phase timings are attribution spans rather than additive buckets. + +## Inspection command guard + +The semantic benchmark's model graph is a spine plus independent chains, so it has no shared +dependencies. `test_inspection_command_performance.py` adds an opt-in lattice to the 3,000-model +semantic project: `shared_orders_hub` reads the end of the spine and fans out to eight models, +sixteen layers each join two neighbouring slots of the previous layer, and `shared_orders_rollup` +fans the last layer back in. Every lattice model is reachable along exponentially many paths. The +compile guards do not enable the lattice, so their fingerprints and budgets are unchanged. + +After one compile and warm lineage and scope caches, the guard runs `sqb lineage` downstream from +the hub and upstream from the rollup as text and JSON, `sqb dag --json` and `sqb scope --json`, +each in a fresh process. Every command +has a wall-time ceiling, a peak-RSS ceiling and an output-line bound linear in the model count, so +a renderer or graph walk that re-expands shared nodes per path fails on size before it can hide +behind a loose timing budget. It runs in the 3,000-model fresh-process compile job. 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 b5905600f1..f309d36440 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 @@ -279,3 +279,13 @@ class DiagnosticPerformanceCase: expected_max_wall_seconds: float = 20.0 expected_timeout_seconds: float = 65.0 diagnostic_count: int = 100 + + +@dataclass(frozen=True) +class InspectionCommandPerformanceGuardTestCase: + description: str + sqb_args: tuple[str, ...] + expected_fragments: tuple[str, ...] + expected_max_output_lines: int + expected_max_wall_seconds: float + expected_max_rss_bytes: int diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py index a15065c86c..01872affcb 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/conftest.py @@ -5,6 +5,12 @@ import pytest +from scripts.cold_compile_performance.main.assert_required_cgroup_memory_limit import ( + assert_required_cgroup_memory_limit, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + prepare_inspection_benchmark_project, +) from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb @@ -21,3 +27,11 @@ def semantic_playground(tmp_path_factory: pytest.TempPathFactory) -> Path: typed.read_text(encoding="utf-8"), encoding="utf-8" ) return project + + +@pytest.fixture(scope="module") +def inspection_benchmark_project(tmp_path_factory: pytest.TempPathFactory) -> Path: + assert_required_cgroup_memory_limit() + project_dir: Path = tmp_path_factory.mktemp("inspection") / "semantic_inspection" + prepare_inspection_benchmark_project(project_dir=project_dir) + return project_dir 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 c076c536e6..23e70d68fc 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -602,8 +602,48 @@ def _run_fresh_process_compile_benchmark( expected_max_wall_seconds: float, compile_args: tuple[str, ...], ) -> FreshProcessCompileBenchmarkResult: + run: _TimedSqbRun = _run_timed_sqb( + project_dir=project_dir, + label=label, + sqb_args=("compile", "--json", *compile_args), + expected_max_wall_seconds=expected_max_wall_seconds, + ) + assert isinstance(run.payload, dict) + payload: dict[str, object] = cast(dict[str, object], run.payload) + compiled_dir: Path = project_dir / "target" / "compiled" + semantic_fingerprint: str = semantic_compile_fingerprint( + payload=payload, compiled_dir=compiled_dir + ) + return FreshProcessCompileBenchmarkResult( + elapsed_seconds=run.elapsed_seconds, + peak_rss_bytes=run.peak_rss_bytes, + semantic_fingerprint=semantic_fingerprint, + payload=payload, + cpu_seconds=run.cpu_seconds, + major_page_faults=run.major_page_faults, + minor_page_faults=run.minor_page_faults, + ) + + +class _TimedSqbRun(NamedTuple): + elapsed_seconds: float + peak_rss_bytes: int + cpu_seconds: float + major_page_faults: int + minor_page_faults: int + payload: object + output_path: Path + + +def _run_timed_sqb( + *, + project_dir: Path, + label: str, + sqb_args: tuple[str, ...], + expected_max_wall_seconds: float, +) -> _TimedSqbRun: measurement_path: Path = project_dir.parent / f"{label}-measurement.txt" - output_path: Path = project_dir.parent / f"{label}.json" + output_path: Path = project_dir.parent / f"{label}.out" stderr_path: Path = project_dir.parent / f"{label}.stderr" command: list[str] = [ "/usr/bin/time", @@ -616,9 +656,7 @@ def _run_fresh_process_compile_benchmark( "--project-dir", str(project_dir), "--no-color", - "compile", - "--json", - *compile_args, + *sqb_args, ] started: float = time.monotonic() try: @@ -639,7 +677,7 @@ def _run_fresh_process_compile_benchmark( process.wait() raise finally: - payload_object, measurement = read_compile_measurement( + payload, measurement = read_compile_measurement( label=label, measurement_path=measurement_path, output_path=output_path, @@ -647,25 +685,94 @@ def _run_fresh_process_compile_benchmark( ) assert returncode == 0, stderr_path.read_text(encoding="utf-8") elapsed_text, peak_rss_kib_text, user_text, system_text, major_text, minor_text = measurement - elapsed_seconds: float = float(elapsed_text) - assert isinstance(payload_object, dict) - payload: dict[str, object] = cast(dict[str, object], payload_object) - peak_rss_bytes: int = int(peak_rss_kib_text) * 1024 - compiled_dir: Path = project_dir / "target" / "compiled" - semantic_fingerprint: str = semantic_compile_fingerprint( - payload=payload, compiled_dir=compiled_dir - ) - return FreshProcessCompileBenchmarkResult( - elapsed_seconds=elapsed_seconds, - peak_rss_bytes=peak_rss_bytes, - semantic_fingerprint=semantic_fingerprint, - payload=payload, + return _TimedSqbRun( + elapsed_seconds=float(elapsed_text), + peak_rss_bytes=int(peak_rss_kib_text) * 1024, cpu_seconds=float(user_text) + float(system_text), major_page_faults=int(major_text), minor_page_faults=int(minor_text), + payload=payload, + output_path=output_path, ) +class InspectionCommandMeasurement(NamedTuple): + elapsed_seconds: float + peak_rss_bytes: int + cpu_seconds: float + payload: object + output: str + + +def run_fresh_process_inspection_command( + *, + project_dir: Path, + label: str, + sqb_args: tuple[str, ...], + expected_max_wall_seconds: float, +) -> InspectionCommandMeasurement: + """Run one read-only inspection command in a fresh measured process.""" + + run: _TimedSqbRun = _run_timed_sqb( + project_dir=project_dir, + label=label, + sqb_args=sqb_args, + expected_max_wall_seconds=expected_max_wall_seconds, + ) + return InspectionCommandMeasurement( + elapsed_seconds=run.elapsed_seconds, + peak_rss_bytes=run.peak_rss_bytes, + cpu_seconds=run.cpu_seconds, + payload=run.payload, + output=run.output_path.read_text(encoding="utf-8"), + ) + + +INSPECTION_DIAMOND_LAYERS: int = 16 +INSPECTION_DIAMOND_WIDTH: int = 8 +INSPECTION_BENCHMARK_MODEL_COUNT: int = ( + 3_000 + INSPECTION_DIAMOND_LAYERS * INSPECTION_DIAMOND_WIDTH + 2 +) + + +def prepare_inspection_benchmark_project(*, project_dir: Path) -> None: + """Write and compile the shared-dependency benchmark, then warm lineage and scope caches.""" + + write_semantic_compile_project( + project_dir=project_dir, + model_count=3_000, + source_count=713, + seed_count=141, + function_count=71, + macro_count=37, + test_count=2_945, + audit_count=5_056, + shared_diamond_layers=INSPECTION_DIAMOND_LAYERS, + shared_diamond_width=INSPECTION_DIAMOND_WIDTH, + ) + compiled: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=project_dir, + label="inspection-compile", + sqb_args=("compile", "--json"), + expected_max_wall_seconds=60.0, + ) + payload: dict[str, Any] = cast(dict[str, Any], compiled.payload) + assert payload["diagnostics"] == [] + assert payload["summary"]["models"] == INSPECTION_BENCHMARK_MODEL_COUNT + assert payload["summary"]["execution_layers"] == _SPINE_DEPTH + INSPECTION_DIAMOND_LAYERS + 2 + warmups: tuple[tuple[str, ...], ...] = ( + ("lineage", SHARED_DIAMOND_HUB, "--depth", "1"), + ("scope", f"model:{SHARED_DIAMOND_ROLLUP}", "--json"), + ) + for index, warmup_args in enumerate(warmups): + run_fresh_process_inspection_command( + project_dir=project_dir, + label=f"inspection-warmup-{index}", + sqb_args=warmup_args, + expected_max_wall_seconds=60.0, + ) + + def _append_benchmark_edit(path: Path, label: str) -> None: path.write_text( path.read_text(encoding="utf-8") + f"\n-- one {label} edit\n", @@ -1158,8 +1265,14 @@ def write_semantic_compile_project( audit_count: int, macro_call_interval: int = _MACRO_INTERVAL, scoped_macros: bool = False, + shared_diamond_layers: int = 0, + shared_diamond_width: int = 0, ) -> None: - """Write a neutral project with broad resources and dense SQL semantics.""" + """Write a neutral project with broad resources and dense SQL semantics. + + A positive `shared_diamond_layers` adds the shared-dependency lattice from + `write_shared_diamond_models`; the default leaves the compile guard fixtures unchanged. + """ _layered_write_project_config(project_dir=project_dir) _layered_write_sources(project_dir=project_dir, source_count=source_count) @@ -1176,6 +1289,15 @@ def write_semantic_compile_project( audit_count=audit_count, macro_call_interval=macro_call_interval, ) + diamond_writers: dict[bool, Callable[[], None]] = { + True: lambda: write_shared_diamond_models( + project_dir=project_dir, + layer_count=shared_diamond_layers, + width=shared_diamond_width, + ), + False: lambda: None, + } + diamond_writers[shared_diamond_layers > 0]() {True: _scope_single_folder_macros, False: _keep_project_macros}[scoped_macros]( project_dir=project_dir ) @@ -1713,6 +1835,79 @@ def _semantic_set_operation_model_sql( """ +SHARED_DIAMOND_HUB: str = "shared_orders_hub" +SHARED_DIAMOND_ROLLUP: str = "shared_orders_rollup" +_SHARED_DIAMOND_FOLDER: str = "intermediate/shared_orders" +_SHARED_DIAMOND_UPSTREAM_MODEL: str = f"model_{_SPINE_DEPTH - 1:05d}" +_SHARED_DIAMOND_HEADER: str = """MODEL ( + columns ( + id (type INTEGER, nullable false), + amount (type DOUBLE), + status (type VARCHAR) + ), +);""" + + +def shared_diamond_model_name(*, layer: int, slot: int) -> str: + """Return the lattice model name for one layer and slot.""" + + return f"shared_orders_l{layer:02d}_s{slot:02d}" + + +def write_shared_diamond_models(*, project_dir: Path, layer_count: int, width: int) -> None: + """Write a shared-dependency lattice with repeated fan-out and fan-in. + + The hub reads the end of the spine and fans out to `width` first-layer models. Every later + slot joins the two neighbouring slots of the previous layer, so each model is reachable along + exponentially many paths, and the rollup fans the final layer back in. + """ + + model_dir: Path = project_dir / "models" / _SHARED_DIAMOND_FOLDER + model_dir.mkdir(parents=True) + files: dict[str, str] = { + SHARED_DIAMOND_HUB: _shared_diamond_sql(inputs=(_SHARED_DIAMOND_UPSTREAM_MODEL,)), + SHARED_DIAMOND_ROLLUP: _shared_diamond_sql( + inputs=tuple( + shared_diamond_model_name(layer=layer_count - 1, slot=slot) for slot in range(width) + ) + ), + } + for layer in range(layer_count): + for slot in range(width): + files[shared_diamond_model_name(layer=layer, slot=slot)] = _shared_diamond_sql( + inputs=_shared_diamond_inputs(layer=layer, slot=slot, width=width) + ) + for name, sql in files.items(): + (model_dir / f"{name}.sql").write_text(sql, encoding="utf-8") + + +def _shared_diamond_inputs(*, layer: int, slot: int, width: int) -> tuple[str, ...]: + builders: dict[bool, Callable[[], tuple[str, ...]]] = { + True: lambda: (SHARED_DIAMOND_HUB,), + False: lambda: ( + shared_diamond_model_name(layer=layer - 1, slot=slot), + shared_diamond_model_name(layer=layer - 1, slot=(slot + 1) % width), + ), + } + return builders[layer == 0]() + + +def _shared_diamond_sql(*, inputs: tuple[str, ...]) -> str: + amount: str = " + ".join(f"input_{index:02d}.amount" for index in range(len(inputs))) + joins: str = "".join( + f'\nJOIN __ref("{name}") AS input_{index:02d} ON input_{index:02d}.id = input_00.id' + for index, name in enumerate(inputs[1:], start=1) + ) + return f"""{_SHARED_DIAMOND_HEADER} + +SELECT + CAST(input_00.id AS INTEGER) AS id, + CAST({amount} AS DOUBLE) AS amount, + CAST(input_00.status AS VARCHAR) AS status +FROM __ref("{inputs[0]}") AS input_00{joins} +""" + + def _layered_write_tests( *, project_dir: Path, diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py new file mode 100644 index 0000000000..40a0c8719e --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_inspection_command_performance.py @@ -0,0 +1,156 @@ +"""Fresh-process speed and memory guards for read-only inspection commands. + +The guarded project is the 3,000-model semantic benchmark plus a shared-dependency lattice: a hub +that fans out to eight models, sixteen layers where every slot joins two neighbouring slots of the +previous layer, and a rollup that fans the last layer back in. Graph walkers and tree renderers +that re-expand shared nodes per path blow up here exponentially, so their wall time, peak RSS and +output size are bounded against the project size. +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( + InspectionCommandPerformanceGuardTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + INSPECTION_BENCHMARK_MODEL_COUNT, + SHARED_DIAMOND_HUB, + SHARED_DIAMOND_ROLLUP, + InspectionCommandMeasurement, + run_fresh_process_inspection_command, +) + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_MIB: int = 1024 * 1024 +_MODEL_COUNT: int = INSPECTION_BENCHMARK_MODEL_COUNT + + +@pytest.mark.performance +@pytest.mark.cold_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + InspectionCommandPerformanceGuardTestCase( + description="models_3000_lineage_text_downstream_from_fan_out_hub", + sqb_args=("lineage", SHARED_DIAMOND_HUB, "--direction", "downstream"), + expected_fragments=( + f"Lineage model {SHARED_DIAMOND_HUB}", + f"model {SHARED_DIAMOND_ROLLUP}", + "(already shown)", + ), + expected_max_output_lines=_MODEL_COUNT, + expected_max_wall_seconds=2.5, + expected_max_rss_bytes=128 * _MIB, + ), + InspectionCommandPerformanceGuardTestCase( + description="models_3000_lineage_text_upstream_from_fan_in_rollup", + sqb_args=("lineage", SHARED_DIAMOND_ROLLUP, "--direction", "upstream"), + expected_fragments=( + f"Lineage model {SHARED_DIAMOND_ROLLUP}", + f"model {SHARED_DIAMOND_HUB}", + "model model_00000", + "(already shown)", + ), + expected_max_output_lines=_MODEL_COUNT, + expected_max_wall_seconds=2.5, + expected_max_rss_bytes=128 * _MIB, + ), + InspectionCommandPerformanceGuardTestCase( + description="models_3000_lineage_json_downstream_from_fan_out_hub", + sqb_args=( + "lineage", + SHARED_DIAMOND_HUB, + "--direction", + "downstream", + "--format", + "json", + ), + expected_fragments=( + '"direction": "downstream"', + f'"id": "model:{SHARED_DIAMOND_ROLLUP}"', + ), + expected_max_output_lines=20 * _MODEL_COUNT, + expected_max_wall_seconds=2.5, + expected_max_rss_bytes=128 * _MIB, + ), + InspectionCommandPerformanceGuardTestCase( + description="models_3000_lineage_json_upstream_from_fan_in_rollup", + sqb_args=( + "lineage", + SHARED_DIAMOND_ROLLUP, + "--direction", + "upstream", + "--format", + "json", + ), + expected_fragments=( + '"direction": "upstream"', + f'"id": "model:{SHARED_DIAMOND_HUB}"', + '"id": "model:model_00000"', + ), + expected_max_output_lines=20 * _MODEL_COUNT, + expected_max_wall_seconds=2.5, + expected_max_rss_bytes=128 * _MIB, + ), + InspectionCommandPerformanceGuardTestCase( + description="models_3000_dag_json", + sqb_args=("dag", "--json"), + expected_fragments=( + '"project_name": "layered_production_performance_guard"', + f'"id": "model:{SHARED_DIAMOND_ROLLUP}"', + ), + expected_max_output_lines=250 * _MODEL_COUNT, + expected_max_wall_seconds=14.0, + expected_max_rss_bytes=768 * _MIB, + ), + InspectionCommandPerformanceGuardTestCase( + description="models_3000_scope_json_for_fan_in_rollup", + sqb_args=( + "scope", + f"model:{SHARED_DIAMOND_ROLLUP}", + "--json", + "--globals", + "all", + "--include-nearby", + ), + expected_fragments=( + f'"identity":"model:{SHARED_DIAMOND_ROLLUP}"', + '"complete":true', + ), + expected_max_output_lines=_MODEL_COUNT, + expected_max_wall_seconds=2.5, + expected_max_rss_bytes=144 * _MIB, + ), + ], + ids=lambda case: case.description, +) +def test_given_shared_dependency_benchmark_when_inspecting_then_preserves_resource_budgets( + inspection_benchmark_project: Path, + test_case: InspectionCommandPerformanceGuardTestCase, +) -> None: + result: InspectionCommandMeasurement = run_fresh_process_inspection_command( + project_dir=inspection_benchmark_project, + label=test_case.description, + sqb_args=test_case.sqb_args, + expected_max_wall_seconds=test_case.expected_max_wall_seconds, + ) + output_lines: int = len(result.output.splitlines()) + _LOGGER.info( + "inspection command %s wall=%.3fs cpu=%.3fs peak_rss_bytes=%d output_lines=%d", + test_case.description, + result.elapsed_seconds, + result.cpu_seconds, + result.peak_rss_bytes, + output_lines, + ) + + for fragment in test_case.expected_fragments: + assert fragment in result.output + assert output_lines <= test_case.expected_max_output_lines + assert result.elapsed_seconds < test_case.expected_max_wall_seconds + assert result.peak_rss_bytes < test_case.expected_max_rss_bytes diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/lineage/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/lineage/helpers.py index afb6b5fce3..3ec718855d 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/lineage/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/lineage/helpers.py @@ -31,6 +31,7 @@ def lineage_node_ids(*, payload: dict[str, object]) -> tuple[str, ...]: DIAMOND_LAYERS: int = 14 +DIAMOND_EDGE_COUNT: int = 4 * DIAMOND_LAYERS def diamond_model_names() -> tuple[str, ...]: diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/lineage/test_lineage.py b/tests/e2e/src/sqlbuild/cli/commands/main/lineage/test_lineage.py index 0602dba144..e42776e88d 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/lineage/test_lineage.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/lineage/test_lineage.py @@ -15,6 +15,7 @@ LineageErrorCliTestCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.lineage.helpers import ( + DIAMOND_EDGE_COUNT, DIAMOND_LAYERS, diamond_model_names, lineage_node_ids, @@ -458,7 +459,7 @@ def test_given_invalid_lineage_target_when_running_then_explains_the_error( description="downstream tree expands each shared model once", command=("--no-color", "lineage", "orders_0", "--direction", "downstream"), expected_expanded_names=diamond_model_names()[1:], - expected_max_lines=4 * len(diamond_model_names()), + expected_max_lines=DIAMOND_EDGE_COUNT, ), DiamondLineageTreeCliTestCase( description="upstream tree expands each shared model once", @@ -470,10 +471,25 @@ def test_given_invalid_lineage_target_when_running_then_explains_the_error( "upstream", ), expected_expanded_names=diamond_model_names()[:-1], - expected_max_lines=4 * len(diamond_model_names()), + expected_max_lines=DIAMOND_EDGE_COUNT, ), DiamondLineageTreeCliTestCase( - description="column trace expands each shared column once", + description="both-direction tree expands each shared model once", + command=( + "--no-color", + "lineage", + f"orders_{DIAMOND_LAYERS // 2}", + "--direction", + "both", + ), + expected_expanded_names=( + diamond_model_names()[: 3 * (DIAMOND_LAYERS // 2)] + + diamond_model_names()[3 * (DIAMOND_LAYERS // 2) + 1 :] + ), + expected_max_lines=DIAMOND_EDGE_COUNT, + ), + DiamondLineageTreeCliTestCase( + description="upstream column trace expands each shared column once", command=( "--no-color", "lineage", @@ -486,7 +502,23 @@ def test_given_invalid_lineage_target_when_running_then_explains_the_error( f"orders_right_{DIAMOND_LAYERS}.amount", f"orders_{DIAMOND_LAYERS - 1}.amount", ), - expected_max_lines=4 * len(diamond_model_names()), + expected_max_lines=DIAMOND_EDGE_COUNT, + ), + DiamondLineageTreeCliTestCase( + description="downstream column trace expands each shared column once", + command=( + "--no-color", + "lineage", + "orders_0.amount", + "--direction", + "downstream", + ), + expected_expanded_names=( + "orders_left_1.amount", + "orders_right_1.amount", + "orders_1.amount", + ), + expected_max_lines=DIAMOND_EDGE_COUNT, ), ), ids=lambda case: case.description, diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/plan/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/plan/_test_types.py index 7ea97ccdd3..76027ea038 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/plan/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/plan/_test_types.py @@ -33,3 +33,16 @@ class TerminalPlanProgressE2ETestCase: columns: int expected_screen_fragments: tuple[str, ...] expected_raw_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class DiamondPlanE2ETestCase: + description: str + expected_fragments: tuple[str, ...] + expected_max_lines_per_model: int + + +@dataclass(frozen=True) +class DiamondPlanJsonE2ETestCase: + description: str + expected_reasons: dict[str, str] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/plan/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/plan/helpers.py new file mode 100644 index 0000000000..927dc90025 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/plan/helpers.py @@ -0,0 +1,105 @@ +"""Project builders for plan command e2e tests.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +from tests.e2e.src.sqlbuild.cli.commands.main.lineage.helpers import DIAMOND_LAYERS +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + execute_duckdb, + prepare_inline_project, + run_sqb, +) + +DIAMOND_DATABASE_FILE: str = "orders.duckdb" +_DIAMOND_PROJECT_TOML: str = ( + 'name = "incremental_diamond"\nadapter = "duckdb"\n\n' + f'[connection]\ndatabase = "{DIAMOND_DATABASE_FILE}"\n' +) +_DIAMOND_SOURCES_YML: str = ( + "sources:\n - name: raw_orders\n schema: main\n table: raw_orders\n" +) +_DIAMOND_ROOT_PATH: str = "models/orders_0.sql" + + +def _incremental_header(*, extra_config: str) -> str: + return ( + "MODEL (\n" + " materialized incremental,\n" + " incremental_strategy delete_insert,\n" + " unique_key order_id,\n" + " cursor order_date,\n" + " cursor_type timestamp,\n" + " cursor_grain day,\n" + ' cursor_start "2026-01-01",\n' + f"{extra_config}" + ");\n\n" + ) + + +def _diamond_root_sql(*, amount_offset: int) -> str: + return ( + _incremental_header(extra_config=" replay_on_change full,\n") + + "SELECT order_id, order_date, amount_cents + " + f'{amount_offset} AS amount_cents FROM __source("raw_orders")\n' + ) + + +def prepare_incremental_diamond_project(*, tmp_path: Path) -> Path: + """Write incremental layered diamonds whose root replays fully on change.""" + + files: dict[str, str] = { + "sqlbuild_project.toml": _DIAMOND_PROJECT_TOML, + "sources/raw.yml": _DIAMOND_SOURCES_YML, + _DIAMOND_ROOT_PATH: _diamond_root_sql(amount_offset=0), + } + for layer in range(1, DIAMOND_LAYERS + 1): + for side in ("left", "right"): + files[f"models/orders_{side}_{layer}.sql"] = ( + _incremental_header(extra_config="") + "SELECT order_id, order_date, amount_cents " + f'FROM __ref("orders_{layer - 1}")\n' + ) + files[f"models/orders_{layer}.sql"] = ( + _incremental_header( + extra_config=( + f" cursor_inputs (orders_left_{layer} order_date, " + f"orders_right_{layer} order_date,),\n" + ) + ) + + "SELECT l.order_id, l.order_date, l.amount_cents + r.amount_cents AS amount_cents\n" + f'FROM __ref("orders_left_{layer}") AS l\n' + f'JOIN __ref("orders_right_{layer}") AS r ON l.order_id = r.order_id\n' + ) + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, project_name="incremental_diamond", repo_files=files + ) + execute_duckdb( + db_path=project_dir / DIAMOND_DATABASE_FILE, + sql=( + "CREATE TABLE main.raw_orders AS SELECT i AS order_id, " + "TIMESTAMP '2026-01-01' + to_days(CAST(i - 1 AS INTEGER)) AS order_date, " + "100 + i AS amount_cents FROM range(1, 4) AS t(i)" + ), + ) + return project_dir + + +def change_incremental_diamond_root(*, project_dir: Path) -> None: + """Change the root query so every downstream model inherits a full replay.""" + + (project_dir / _DIAMOND_ROOT_PATH).write_text( + _diamond_root_sql(amount_offset=1), encoding="utf-8" + ) + + +def prepare_changed_incremental_diamond_project(*, tmp_path: Path) -> Path: + """Build the incremental diamonds, then change the root so the plan cascades.""" + + project_dir: Path = prepare_incremental_diamond_project(tmp_path=tmp_path) + build: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build"), project_dir=project_dir + ) + assert build.returncode == 0, build.stdout + build.stderr + change_incremental_diamond_root(project_dir=project_dir) + return project_dir diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_diamond_plan.py b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_diamond_plan.py new file mode 100644 index 0000000000..b479d6fa54 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/plan/test_diamond_plan.py @@ -0,0 +1,91 @@ +"""E2E coverage for plan output over layered shared dependencies.""" + +from __future__ import annotations + +import json +import re +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.lineage.helpers import ( + DIAMOND_LAYERS, + diamond_model_names, +) +from tests.e2e.src.sqlbuild.cli.commands.main.plan._test_types import ( + DiamondPlanE2ETestCase, + DiamondPlanJsonE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.plan.helpers import ( + prepare_changed_incremental_diamond_project, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb + +_PLAN_ENTRY: re.Pattern[str] = re.compile(r"^[├└]── (orders_\w+)\s", re.MULTILINE) + + +@pytest.mark.parametrize( + "test_case", + [ + DiamondPlanE2ETestCase( + description="plan tree lists each diamond model once with its cascade cause", + expected_fragments=( + "Query changed (1)", + f"Upstream changed ({len(diamond_model_names()) - 1})", + f"└── orders_{DIAMOND_LAYERS} ", + "cause orders_0 (query changed)", + ), + expected_max_lines_per_model=4, + ) + ], + ids=lambda case: case.description, +) +def test_given_changed_diamond_root_when_planning_then_tree_lists_each_model_once( + test_case: DiamondPlanE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_changed_incremental_diamond_project(tmp_path=tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "plan"), project_dir=project_dir + ) + + assert result.returncode == 0, result.stdout + result.stderr + for fragment in test_case.expected_fragments: + assert fragment in result.stdout, result.stdout + plan_output: str = result.stdout[result.stdout.index("Plan ready") :] + assert sorted(_PLAN_ENTRY.findall(plan_output)) == sorted(diamond_model_names()) + assert len(plan_output.splitlines()) <= test_case.expected_max_lines_per_model * len( + diamond_model_names() + ) + + +@pytest.mark.parametrize( + "test_case", + [ + DiamondPlanJsonE2ETestCase( + description="plan json lists each diamond model once with its cascade reason", + expected_reasons={ + "orders_0": "query_changed", + **{name: "upstream_changed" for name in diamond_model_names()[1:]}, + }, + ) + ], + ids=lambda case: case.description, +) +def test_given_changed_diamond_root_when_planning_json_then_lists_each_model_once( + test_case: DiamondPlanJsonE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_changed_incremental_diamond_project(tmp_path=tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "plan", "--json"), project_dir=project_dir + ) + + assert result.returncode == 0, result.stdout + result.stderr + models: list[dict[str, object]] = json.loads(result.stdout)["models"] + reasons: dict[str, object] = {str(model["name"]): model["reason"] for model in models} + assert len(models) == len(reasons) + assert reasons == test_case.expected_reasons