From a5709476ebca865ffa298ecb81d90b8fd996144f Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Mon, 21 Sep 2026 08:54:27 +0100 Subject: [PATCH] perf: accelerate warm incremental compiles --- .github/workflows/verify.yml | 50 +++ Makefile | 11 +- docs/compiler-cache.md | 141 ++++++++ pyproject.toml | 1 + .../_helpers/compile/target_writer.py | 15 +- .../compile/_helpers/analysis/cache.py | 334 ++++++++++++++++- .../compile/_helpers/analysis/compact.py | 41 ++- .../compile/_helpers/assembly/project.py | 340 ++++++++---------- .../compile/_helpers/assembly/targets.py | 210 +++++++++++ src/sqlbuild/compiler/compile/models.py | 28 ++ src/sqlbuild/compiler/compile/types.py | 11 +- .../main/execution/sql_test_model_chain.py | 22 +- src/sqlbuild/compiler/profiling/constants.py | 9 +- .../compiler/profiling/main/_metric.py | 13 + src/sqlbuild/compiler/profiling/models.py | 12 +- src/sqlbuild/compiler/profiling/types.py | 7 + .../main/build/test_compile_json_behavior.py | 4 + .../cli/commands/main/compile/_test_types.py | 21 ++ .../cli/commands/main/compile/helpers.py | 210 ++++++++++- ...fresh_process_compile_cache_performance.py | 224 ++++++++++++ .../compile/_helpers/test_analysis_cache.py | 179 ++++++++- 21 files changed, 1635 insertions(+), 248 deletions(-) create mode 100644 docs/compiler-cache.md create mode 100644 src/sqlbuild/compiler/compile/_helpers/assembly/targets.py create mode 100644 src/sqlbuild/compiler/profiling/main/_metric.py create mode 100644 tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 25b42435f0..3ee8274f1b 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -281,6 +281,53 @@ jobs: make test-e2e-cold-compile-performance \ SQLBUILD_BENCHMARK_MODELS="${BENCHMARK_MODELS}" + cache_compile_performance: + name: Fresh-process compile cache (${{ matrix.models }} models) + needs: [changes, lockfile] + if: needs.changes.outputs.performance == 'true' + runs-on: ubuntu-24.04 + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + models: [3000, 5000, 10000] + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + with: + version: "0.12.3" + enable-cache: true + cache-dependency-glob: | + uv.lock + Cargo.lock + - uses: dtolnay/rust-toolchain@master + with: + toolchain: stable + - name: Install Python + run: uv python install 3.12 + - name: Install dependencies + run: uv sync --locked --all-groups --all-extras + - name: Run fresh-process compile cache guard under 4 GiB + env: + BENCHMARK_MODELS: ${{ matrix.models }} + MEMORY_LIMIT_BYTES: "4294967296" + run: | + sudo systemd-run \ + --unit="sqlbuild-cache-${BENCHMARK_MODELS}-${GITHUB_RUN_ID}" \ + --uid="$(id -u)" \ + --gid="$(id -g)" \ + --same-dir \ + --setenv="HOME=${HOME}" \ + --setenv="PATH=${PATH}" \ + --setenv="SQLBUILD_REQUIRE_MEMORY_LIMIT_BYTES=${MEMORY_LIMIT_BYTES}" \ + --property="MemoryMax=${MEMORY_LIMIT_BYTES}" \ + --property="MemorySwapMax=0" \ + --wait \ + --collect \ + --pipe \ + make test-e2e-cache-compile-performance \ + SQLBUILD_BENCHMARK_MODELS="${BENCHMARK_MODELS}" + rules_performance: name: Rules performance guards needs: [changes, lockfile] @@ -373,6 +420,7 @@ jobs: e2e_shards, performance, cold_compile_performance, + cache_compile_performance, rules_performance, postgres, ] @@ -386,6 +434,7 @@ jobs: E2E_SHARDS_RESULT: ${{ needs.e2e_shards.result }} PERFORMANCE_RESULT: ${{ needs.performance.result }} COLD_COMPILE_PERFORMANCE_RESULT: ${{ needs.cold_compile_performance.result }} + CACHE_COMPILE_PERFORMANCE_RESULT: ${{ needs.cache_compile_performance.result }} RULES_PERFORMANCE_RESULT: ${{ needs.rules_performance.result }} POSTGRES_RESULT: ${{ needs.postgres.result }} run: | @@ -399,6 +448,7 @@ jobs: "${E2E_SHARDS_RESULT}" \ "${PERFORMANCE_RESULT}" \ "${COLD_COMPILE_PERFORMANCE_RESULT}" \ + "${CACHE_COMPILE_PERFORMANCE_RESULT}" \ "${RULES_PERFORMANCE_RESULT}" \ "${POSTGRES_RESULT}"; do if [[ "${result}" != "success" && "${result}" != "skipped" ]]; then diff --git a/Makefile b/Makefile index 6fa23fa4b1..11c675682d 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ SHELL := /bin/bash test-e2e-duckdb-build-incremental test-e2e-duckdb-build-virtual \ test-e2e-duckdb-cli-data test-e2e-duckdb-cli test-e2e-duckdb-virtual \ test-e2e-duckdb-integrations test-e2e-performance \ - test-e2e-cold-compile-performance + test-e2e-cold-compile-performance test-e2e-cache-compile-performance format: uv run ruff format . @@ -224,7 +224,7 @@ test-e2e-performance: @log="/tmp/opencode/test-e2e-performance-$$(date +%Y%m%d-%H%M%S).log"; \ echo "Logging to $$log"; \ env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=$(SQLBUILD_CONCURRENCY) uv run pytest tests/e2e \ - -m "performance and not cold_compile_performance and not real_warehouse and not dbt" \ + -m "performance and not cold_compile_performance and not cache_compile_performance and not real_warehouse and not dbt" \ -vv --log-cli-level=INFO --color=yes 2>&1 | tee "$$log"; \ status=$${PIPESTATUS[0]}; \ echo "TEST_E2E_PERFORMANCE_EXIT=$$status (log: $$log)" | tee -a "$$log"; \ @@ -238,6 +238,13 @@ test-e2e-cold-compile-performance: -m cold_compile_performance -k "models_$(SQLBUILD_BENCHMARK_MODELS)" \ -vv --log-cli-level=INFO --color=yes +test-e2e-cache-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_cache_performance.py \ + -m cache_compile_performance -k "models_$(SQLBUILD_BENCHMARK_MODELS)" \ + -vv --log-cli-level=INFO --color=yes + test-virtual: @mkdir -p /tmp/opencode diff --git a/docs/compiler-cache.md b/docs/compiler-cache.md new file mode 100644 index 0000000000..8093e9e59c --- /dev/null +++ b/docs/compiler-cache.md @@ -0,0 +1,141 @@ +# Compiler cache architecture + +The compiler cache exists to avoid proving unchanged facts again. It must never turn an uncertain +input into a cache hit, and a missing, stale, corrupt, oversized, or contended cache must behave like +an ordinary uncached compile. + +This document defines the target architecture. The current vertical slice implements versioned +model-analysis entries, dependency-aware compact analysis generations, shared SQL-test closure +indexes, bounded compact-generation retention, corruption fallback, and analysis hit/miss telemetry. +Project inventory, discovery/attachment reuse, and command snapshots remain later layers; their +sections below are design constraints rather than claims of current implementation. + +## Goals + +- Make unchanged fresh-process compiles proportional to project inventory and requested output, not + total SQL complexity. +- Make a small edit proportional to the changed resource and its semantic dependents. +- Reuse downstream analysis when an upstream implementation changes without changing its exported + column interface. +- Preserve diagnostics, bindings, types and nullability, lineage, tests, audits, hooks, macros, + ordering, artifacts, and exit codes exactly. +- Keep cache data project-local, bounded, inspectable, non-executable, and safe under concurrent + compiler processes. + +## Non-goals + +- The cache is not a second source of project truth. +- The cache does not make arbitrary Python execution deterministic. A stage with dynamic inputs that + are not represented in its identity must bypass persistent reuse. +- The cache does not store pickles, code objects, callables, imported modules, adapter instances, or + open connections. +- The cache does not replace warehouse execution state or change build selection semantics. + +## Layers + +### 1. Project inventory + +The inventory records every compiler-owned authored input as a normalized project-relative path, +resource role, content digest, size, and filesystem identity metadata. Configuration, ignore files, +adapter identity, compiler and SQL-analysis engine versions, selected target, CLI variables, and +compile modes are explicit identity inputs. + +Filesystem metadata may identify an unchanged candidate, but reusable semantic facts remain +content-addressed. Ambiguous or concurrently changing observations are misses. Added, removed, +renamed, and role-changing paths are first-class changes. + +### 2. Static discovery facts + +Parsing results for SQL, YAML, CSV metadata, and statically inspected Python declarations are keyed +by content digest, parser version, resource role, and parse options. These records contain only +validated JSON-compatible values. Runtime Python discovery still imports current project code and +is never reconstructed from executable cache data. + +### 3. Attached resource facts + +Attachment resolves configuration, declarations, macros, hooks, schemas, tests, and references. +Each record carries the exact identities it observed: + +- authored resource content; +- effective target, variables, and relevant configuration; +- visible enum and constant declarations; +- macro implementation and dependency closure; +- referenced schema or hook declarations; +- adapter rendering contract. + +If any observed identity is absent or unstable, that resource is attached normally. A global +configuration change is broad only when the affected value was observed broadly. + +### 4. Semantic analysis facts + +Model SQL analysis remains keyed by expanded SQL, analysis profile, referenced relation interfaces, +binding schema, and engine versions. The persistent representation should retain compact interned +facts so warm reads do not reconstruct duplicate lineage objects entry by entry. + +Each model publishes a deterministic interface signature containing the output columns, types, +nullability, wildcard status, and any other fact consumed by downstream analysis. A changed model +invalidates downstream semantic analysis only when this exported interface changes. Its own compiled +artifact still changes whenever its implementation changes. + +### 5. Shared derived indexes + +Project-wide indexes are built once per compile and shared by all consumers. They include model +dependencies, test model closures, declaration visibility, selector indexes, and artifact identities. +No per-test, per-rule, or per-artifact path may rebuild an unchanged whole-project index. + +### 6. Command snapshots + +A command snapshot records validated, non-executable output facts after all lower layers complete. +It may provide a no-op fast path only when every input observed by that command is represented in +the snapshot identity. Commands involving untracked runtime values bypass this layer while still +using safe lower-layer facts. + +## Invalidation + +Invalidation follows recorded dependency edges rather than directory-wide guesses: + +1. Compare the current inventory with the last committed inventory. +2. Invalidate directly changed static facts. +3. Follow declaration, macro, configuration, schema, hook, and test dependency edges. +4. Reattach directly affected resources. +5. Reanalyze changed models and compare exported interface signatures. +6. Reanalyze downstream models only when a consumed interface changed. +7. Rebuild tests, audits, rules, and artifacts only when their recorded inputs changed. + +Removing an edge is itself an invalidating change. Cache readers must not infer validity from the +absence of a newly introduced dependency record. + +## Storage and publication + +- Records are schema-versioned and include algorithm and package fingerprints. +- Payloads have explicit size and collection-count limits before semantic acceptance. +- Digests cover keys and canonical serialized payloads. +- Writers publish in one transaction or by atomic replacement; readers observe one committed + generation. +- Concurrent lock or I/O failure is a cache miss, never a compile failure. +- Cache growth is bounded by generation and byte limits. Pruning removes old immutable records and + does not mutate current semantic state. +- `--no-cache`, target configuration, and the compile-cache environment override bypass every + persistent layer and do not publish new records. + +## Observability and guards + +The target machine-readable contract reports per-layer hits, misses, invalidations, bytes read and +written, and phase timings. The current vertical slice reports model-analysis batch hits, entry +hits, misses, bypasses, and phase timings. Counts describe semantic records, not low-level database +operations. + +Fresh-process performance guards use representative semantic projects and cover: + +- cold compile with a clean target; +- unchanged warm compile; +- leaf-model implementation edit; +- upstream implementation edit with an unchanged interface; +- interface-changing upstream edit; +- test-only and macro edits; +- broad project-configuration edit; +- cache corruption and `--no-cache` fallback. + +Every path verifies exact resource counts, zero unexpected diagnostics, deterministic semantic +fingerprints, bounded peak RSS, and expected hit and invalidation counts. Wall-time ceilings are +calibrated only after the behavior and identity assertions are stable. diff --git a/pyproject.toml b/pyproject.toml index 39ceb9e92e..ceb2733f7b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -116,6 +116,7 @@ markers = [ "motherduck: tests that require live MotherDuck credentials", "performance: opt-in performance guard tests", "cold_compile_performance: fresh-process semantic compile scaling guards", + "cache_compile_performance: fresh-process warm and incremental compile cache guards", ] [tool.ruff] diff --git a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py index 3367350130..55e46f548b 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py @@ -27,13 +27,13 @@ compiled_sql_test_output_path, ) from sqlbuild.cli.paths.main._sql_test_output_path import sql_test_output_path -from sqlbuild.compiler.compile.models import CompiledModel, CompiledProject, CompiledSqlTest +from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledProject, CompiledSqlTest from sqlbuild.compiler.compile.types import FunctionLanguage from sqlbuild.compiler.planner.main.execution.sql_test_artifacts import ( plan_and_render_sql_test_artifacts, ) from sqlbuild.compiler.planner.main.execution.sql_test_model_chain import ( - sql_test_model_chain_names, + sql_test_model_chain_names_by_key, ) from sqlbuild.compiler.planner.models import AuditPlanEntry, NativeSqlTestArtifact, PlanOutput from sqlbuild.compiler.profiling.main.record import record_compile_timing @@ -332,7 +332,12 @@ def _write_static_tests( if project.compile_cache_dir is not None else None ) - model_map: dict[str, CompiledModel] = {model.name: model for model in project.models} + model_chain_names_by_key: dict[CompiledObjectKey, tuple[str, ...]] = ( + sql_test_model_chain_names_by_key( + project=project, + tests=project.sql_tests, + ) + ) pending: list[tuple[CompiledSqlTest, str | None, str | None]] = [] for test in project.sql_tests: record_key: str | None = None @@ -341,9 +346,7 @@ def _write_static_tests( record_key = sql_test_artifact_record_key(test=test) artifact_identity = sql_test_artifact_identity( test=test, - model_chain_names=sql_test_model_chain_names( - test=test, project=project, model_map=model_map - ), + model_chain_names=model_chain_names_by_key.get(test.key, ()), context=identity_context, ) cached_record: SqlTestArtifactCacheRecord | None = cached_records.get(record_key) diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py index 1414a52af3..bff28c575e 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py @@ -8,32 +8,43 @@ import json import platform import sqlite3 +import time +from collections import defaultdict, deque from importlib.metadata import PackageNotFoundError, version from pathlib import Path from string import hexdigits from typing import Any, cast +import orjson + from sqlbuild.adapter.contract.models import ExpressionInferenceProfile from sqlbuild.compiler.compile._helpers.analysis.columns import table_function_analysis_name from sqlbuild.compiler.compile.exceptions import AnalysisCacheEntryError from sqlbuild.compiler.compile.models import ( AnalysisCacheContext, + CompactAnalysisCacheCandidate, + CompactAnalysisCacheModel, + CompactAnalysisCachePlan, + CompactBatchPreparation, CompiledLineageColumnFact, CompiledLineageSourceFact, CompileSqlReference, InferredColumn, PolyglotAnalysisResult, ) +from sqlbuild.compiler.compile.types import CompactBatchResponseCallback from sqlbuild.compiler.lineage.types import ( ColumnLineageConfidence, ColumnTransformKind, InferredNullability, ) +from sqlbuild.compiler.profiling.main._metric import record_compile_metric +from sqlbuild.compiler.profiling.main.record import record_compile_timing from sqlbuild.compiler.references.types import SqlReferenceKind from sqlbuild.compiler.sql_analysis.models import SqlBindingDiagnostic -_ANALYSIS_CACHE_VERSION: int = 9 -_ANALYSIS_ALGORITHM_FINGERPRINT: str = "model-sql-analysis-v9-polyglot-scope-fix" +_ANALYSIS_CACHE_VERSION: int = 10 +_ANALYSIS_ALGORITHM_FINGERPRINT: str = "model-sql-analysis-v10-incremental-compact-batch" _MAX_CACHE_ENTRY_BYTES: int = 10_000_000 _SHA256_HEX_LENGTH: int = 64 _CACHE_ENTRY_SEPARATOR: str = "\n" @@ -41,6 +52,9 @@ _SQLITE_QUERY_CHUNK_SIZE: int = 500 _SQLITE_TIMEOUT_SECONDS: float = 0.1 _CACHE_DATABASE_NAME: str = "model-analysis.sqlite3" +_COMPACT_BATCH_CACHE_VERSION: int = 1 +_MAX_COMPACT_BATCH_BYTES: int = 128 * 1024 * 1024 +_MAX_COMPACT_BATCH_RECORDS: int = 2 _CREATE_CACHE_TABLE_SQL: str = """ CREATE TABLE IF NOT EXISTS model_analysis ( cache_key TEXT PRIMARY KEY, @@ -65,6 +79,14 @@ PRIMARY KEY (signature_namespace, cache_key, upstream_model_name) ) """ +_CREATE_COMPACT_BATCH_TABLE_SQL: str = """ +CREATE TABLE IF NOT EXISTS model_analysis_compact_batch ( + batch_key TEXT PRIMARY KEY, + shared_fingerprint TEXT NOT NULL, + payload BLOB NOT NULL, + created_ns INTEGER NOT NULL +) +""" def build_analysis_cache_context( @@ -143,6 +165,314 @@ def model_analysis_cache_key( return _payload_digest(payload) +def compact_analysis_batch_cache_key( + *, context: AnalysisCacheContext, reuse_keys: tuple[str, ...] +) -> str: + """Return the exact identity of one ordered whole-project compact analysis batch.""" + + return _payload_digest( + { + "cache_version": _ANALYSIS_CACHE_VERSION, + "compact_batch_version": _COMPACT_BATCH_CACHE_VERSION, + "shared_fingerprint": context.shared_fingerprint, + "reuse_keys": reuse_keys, + } + ) + + +def compact_analysis_model_reuse_key( + *, cache_key: str, upstream_reuse_keys: tuple[tuple[str, str], ...] +) -> str: + """Return one model key that changes with every transitive model dependency.""" + + return _payload_digest( + { + "cache_key": cache_key, + "upstream": upstream_reuse_keys, + } + ) + + +def build_compact_analysis_cache_plan( + *, + context: AnalysisCacheContext | None, + models: tuple[CompactAnalysisCacheModel, ...], + min_model_count: int, +) -> CompactAnalysisCachePlan | None: + """Build a dependency-aware ordered batch identity when every model is cacheable.""" + + if context is None or len(models) < min_model_count: + return None + models_by_name: dict[str, CompactAnalysisCacheModel] = {model.name: model for model in models} + dependencies_by_name: dict[str, tuple[str, ...]] = {} + for model in models: + dependencies_by_name[model.name] = tuple( + name for name in model.upstream_names if name in models_by_name + ) + dependents_by_name: dict[str, list[str]] = defaultdict(list) + remaining_dependency_count: dict[str, int] = {} + for model_name, dependencies in dependencies_by_name.items(): + remaining_dependency_count[model_name] = len(dependencies) + for dependency in dependencies: + dependents_by_name[dependency].append(model_name) + resolved: dict[str, str] = {} + ready: deque[str] = deque( + sorted(name for name, count in remaining_dependency_count.items() if count == 0) + ) + while ready: + model_name: str = ready.popleft() + model: CompactAnalysisCacheModel = models_by_name[model_name] + if model.cache_key is None: + return None + resolved[model_name] = compact_analysis_model_reuse_key( + cache_key=model.cache_key, + upstream_reuse_keys=tuple( + (upstream_name, resolved[upstream_name]) + for upstream_name in dependencies_by_name[model_name] + ), + ) + for dependent_name in sorted(dependents_by_name.get(model_name, ())): + remaining_dependency_count[dependent_name] -= 1 + if remaining_dependency_count[dependent_name] == 0: + ready.append(dependent_name) + if len(resolved) != len(models): + return None + cache_keys: tuple[str, ...] = tuple( + model.cache_key for model in models if model.cache_key is not None + ) + if len(cache_keys) != len(models): + return None + reuse_keys: tuple[str, ...] = tuple(resolved[model.name] for model in models) + return CompactAnalysisCachePlan( + context=context, + batch_key=compact_analysis_batch_cache_key(context=context, reuse_keys=reuse_keys), + cache_keys=cache_keys, + reuse_keys=reuse_keys, + ) + + +def read_compact_analysis_cache_candidate( + *, plan: CompactAnalysisCachePlan, expected_count: int, min_model_count: int +) -> CompactAnalysisCacheCandidate | None: + """Read one compatible generation and locate dependency-safe matching positions.""" + + cached: tuple[tuple[str, ...], tuple[str, ...], tuple[str, ...], object] | None = ( + read_compact_analysis_batch(context=plan.context, batch_key=plan.batch_key) + ) + if cached is None: + return None + stored_cache_keys, stored_reuse_keys, cleaned_sql, response = cached + matching_indexes: tuple[int, ...] = tuple( + index + for index, (stored_key, current_key) in enumerate( + zip(stored_reuse_keys, plan.reuse_keys, strict=False) + ) + if stored_key == current_key + ) + if len(stored_cache_keys) != expected_count or len(matching_indexes) < min_model_count: + return None + return CompactAnalysisCacheCandidate( + matching_indexes=matching_indexes, + preparation=CompactBatchPreparation( + cleaned_sql=cleaned_sql, + queries=(), + templates=(), + projections=(), + ), + response=response, + ) + + +def compact_analysis_batch_response_writer( + *, plan: CompactAnalysisCachePlan, count: int +) -> CompactBatchResponseCallback: + """Build a callback that publishes one complete compact generation.""" + + def write(*, preparation: CompactBatchPreparation, response: object) -> None: + if len(preparation.cleaned_sql) != count: + return + with record_compile_timing("cache_publication_ms"): + write_compact_analysis_batch( + context=plan.context, + batch_key=plan.batch_key, + cache_keys=plan.cache_keys, + reuse_keys=plan.reuse_keys, + cleaned_sql=preparation.cleaned_sql, + response=response, + ) + + return write + + +def record_analysis_cache_metrics( + *, batch_hits: int, entry_hits: int, misses: int, bypasses: int +) -> None: + """Record semantic analysis cache outcomes for one invocation.""" + + record_compile_metric(metric="analysis_batch_cache_hits", value=batch_hits) + record_compile_metric(metric="analysis_entry_cache_hits", value=entry_hits) + record_compile_metric(metric="analysis_cache_misses", value=misses) + record_compile_metric(metric="analysis_cache_bypasses", value=bypasses) + + +def read_compact_analysis_batch( + *, context: AnalysisCacheContext, batch_key: str +) -> tuple[tuple[str, ...], tuple[str, ...], tuple[str, ...], object] | None: + """Read the newest compatible compact generation, treating every fault as a miss.""" + + database_path: Path = _cache_database_path(context=context) + if not database_path.is_file(): + return None + try: + connection_uri: str = f"file:{database_path}?mode=ro" + with sqlite3.connect( + connection_uri, + uri=True, + timeout=_SQLITE_TIMEOUT_SECONDS, + ) as connection: + metadata_row: tuple[str, int] | None = connection.execute( + "SELECT batch_key, length(payload) FROM model_analysis_compact_batch " + "WHERE batch_key = ?", + (batch_key,), + ).fetchone() + if metadata_row is None: + metadata_row = connection.execute( + "SELECT batch_key, length(payload) FROM model_analysis_compact_batch " + "WHERE shared_fingerprint = ? ORDER BY created_ns DESC LIMIT 1", + (context.shared_fingerprint,), + ).fetchone() + if metadata_row is None: + return None + stored_batch_key, payload_bytes = metadata_row + if payload_bytes < 0 or payload_bytes > _MAX_COMPACT_BATCH_BYTES: + return None + payload_row: tuple[bytes] | None = connection.execute( + "SELECT payload FROM model_analysis_compact_batch WHERE batch_key = ?", + (stored_batch_key,), + ).fetchone() + if payload_row is None: + return None + contents: bytes = payload_row[0] + if not isinstance(contents, bytes) or len(contents) != payload_bytes: + return None + envelope: object = orjson.loads(contents) + if not isinstance(envelope, dict): + return None + values: dict[str, Any] = cast(dict[str, Any], envelope) + cache_keys_value: object = values.get("cache_keys") + reuse_keys_value: object = values.get("reuse_keys") + cleaned_sql_value: object = values.get("cleaned_sql") + response: object = values.get("response") + if ( + not isinstance(cache_keys_value, list) + or not all(isinstance(value, str) for value in cache_keys_value) + or not isinstance(reuse_keys_value, list) + or not all(isinstance(value, str) for value in reuse_keys_value) + or not isinstance(cleaned_sql_value, list) + or not all(isinstance(value, str) for value in cleaned_sql_value) + ): + return None + cache_keys: tuple[str, ...] = tuple(cache_keys_value) + reuse_keys: tuple[str, ...] = tuple(reuse_keys_value) + cleaned_sql: tuple[str, ...] = tuple(cleaned_sql_value) + if ( + values.get("version") != _COMPACT_BATCH_CACHE_VERSION + or values.get("batch_key") != stored_batch_key + or values.get("shared_fingerprint") != context.shared_fingerprint + or values.get("count") != len(cache_keys) + or len(reuse_keys) != len(cache_keys) + or len(cleaned_sql) != len(cache_keys) + or not isinstance(values.get("facts_sha256"), str) + or not hmac.compare_digest( + values["facts_sha256"], + _compact_batch_facts_digest( + cache_keys=cache_keys, + reuse_keys=reuse_keys, + cleaned_sql=cleaned_sql, + response=response, + ), + ) + ): + return None + return cache_keys, reuse_keys, cleaned_sql, response + except (OSError, sqlite3.DatabaseError, TypeError, ValueError, orjson.JSONDecodeError): + return None + + +def write_compact_analysis_batch( + *, + context: AnalysisCacheContext, + batch_key: str, + cache_keys: tuple[str, ...], + reuse_keys: tuple[str, ...], + cleaned_sql: tuple[str, ...], + response: object, +) -> None: + """Persist one compact native response transactionally, with bounded logical retention.""" + + try: + if not (len(cache_keys) == len(reuse_keys) == len(cleaned_sql)): + return + contents: bytes = orjson.dumps( + { + "version": _COMPACT_BATCH_CACHE_VERSION, + "batch_key": batch_key, + "shared_fingerprint": context.shared_fingerprint, + "count": len(cleaned_sql), + "facts_sha256": _compact_batch_facts_digest( + cache_keys=cache_keys, + reuse_keys=reuse_keys, + cleaned_sql=cleaned_sql, + response=response, + ), + "cache_keys": cache_keys, + "reuse_keys": reuse_keys, + "cleaned_sql": cleaned_sql, + "response": response, + }, + option=orjson.OPT_SORT_KEYS, + ) + if len(contents) > _MAX_COMPACT_BATCH_BYTES: + return + database_path: Path = _cache_database_path(context=context) + database_path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(database_path, timeout=_SQLITE_TIMEOUT_SECONDS) as connection: + _ = connection.execute(_CREATE_COMPACT_BATCH_TABLE_SQL) + _ = connection.execute( + "INSERT OR REPLACE INTO model_analysis_compact_batch " + "(batch_key, shared_fingerprint, payload, created_ns) VALUES (?, ?, ?, ?)", + (batch_key, context.shared_fingerprint, contents, time.time_ns()), + ) + _ = connection.execute( + "DELETE FROM model_analysis_compact_batch WHERE batch_key NOT IN " + "(SELECT batch_key FROM model_analysis_compact_batch " + "ORDER BY created_ns DESC LIMIT ?)", + (_MAX_COMPACT_BATCH_RECORDS,), + ) + except (OSError, sqlite3.DatabaseError, TypeError, ValueError): + return + + +def _compact_batch_facts_digest( + *, + cache_keys: tuple[str, ...], + reuse_keys: tuple[str, ...], + cleaned_sql: tuple[str, ...], + response: object, +) -> str: + digest: Any = hashlib.sha256() + for label, value in ( + (b"cache_keys\0", cache_keys), + (b"reuse_keys\0", reuse_keys), + (b"cleaned_sql\0", cleaned_sql), + (b"response\0", response), + ): + digest.update(label) + digest.update(orjson.dumps(value, option=orjson.OPT_SORT_KEYS)) + digest.update(b"\0") + return digest.hexdigest() + + def read_model_analyses( *, context: AnalysisCacheContext, diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py index 26f115b425..0baafc2e41 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py @@ -50,7 +50,7 @@ PolyglotAnalysisResult, ProjectedAnalysisRequest, ) -from sqlbuild.compiler.compile.types import CompiledResourceType +from sqlbuild.compiler.compile.types import CompactBatchResponseCallback, CompiledResourceType from sqlbuild.compiler.lineage.types import ( ColumnLineageConfidence, ColumnTransformKind, @@ -373,6 +373,8 @@ def analyze_queries_with_compact_polyglot_batch( inference_profile: ExpressionInferenceProfile, recover_cte_facts: tuple[bool, ...], rich_type_inference: bool = True, + cached_batch: tuple[CompactBatchPreparation, object] | None = None, + on_response: CompactBatchResponseCallback | None = None, ) -> tuple[NativeCompactAnalysis, ...]: """Analyze rendered SQL in one bounded native call, preserving input order.""" @@ -380,21 +382,36 @@ def analyze_queries_with_compact_polyglot_batch( raise CompactAnalysisInputError( "compact query-analysis batch inputs must have equal lengths" ) - preparation: CompactBatchPreparation = _prepare_compact_analysis_batch( - query_sqls=query_sqls, - references=references, - placeholders=placeholders, - column_nullability_by_table=column_nullability_by_table, - column_types_by_table=column_types_by_table, - inference_profile=inference_profile, - recover_cte_facts=recover_cte_facts, - rich_type_inference=rich_type_inference, + preparation: CompactBatchPreparation = ( + cached_batch[0] + if cached_batch is not None + else _prepare_compact_analysis_batch( + query_sqls=query_sqls, + references=references, + placeholders=placeholders, + column_nullability_by_table=column_nullability_by_table, + column_types_by_table=column_types_by_table, + inference_profile=inference_profile, + recover_cte_facts=recover_cte_facts, + rich_type_inference=rich_type_inference, + ) + ) + if len(preparation.cleaned_sql) != len(query_sqls): + raise CompactAnalysisInputError( + "cached compact query-analysis preparation has an invalid length" + ) + response_payload: object = ( + cached_batch[1] + if cached_batch is not None + else _run_compact_analysis_batch(preparation=preparation) ) - response_payload: object = _run_compact_analysis_batch(preparation=preparation) - return _project_compact_analysis_batch( + projected: tuple[NativeCompactAnalysis, ...] = _project_compact_analysis_batch( preparation=preparation, response_payload=response_payload, ) + if cached_batch is None and on_response is not None: + on_response(preparation=preparation, response=response_payload) + return projected def _prepare_compact_analysis_batch( diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py index f217d40d57..416e734feb 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py @@ -12,9 +12,13 @@ from sqlbuild.adapter.contract.types import BuiltinAdapter from sqlbuild.compiler.compile._helpers.analysis.cache import ( build_analysis_cache_context, + build_compact_analysis_cache_plan, + compact_analysis_batch_response_writer, model_analysis_cache_key, model_analysis_output_signature, + read_compact_analysis_cache_candidate, read_model_analyses, + record_analysis_cache_metrics, write_model_analyses, ) from sqlbuild.compiler.compile._helpers.analysis.columns import ( @@ -35,6 +39,10 @@ validate_hook_sql_syntax, validate_sql_syntax, ) +from sqlbuild.compiler.compile._helpers.assembly.targets import ( + build_model_relation_target, + build_seed_relation_target, +) from sqlbuild.compiler.compile._helpers.config.namespace_validation import ( validate_preserved_logical_namespace, ) @@ -63,7 +71,7 @@ from sqlbuild.compiler.compile._helpers.sql_tests.identity import ( build_sql_test_case_fingerprint, ) -from sqlbuild.compiler.compile.constants import NOT_NULL_AUDIT_NAME, PRESERVE_TARGET_VALUE +from sqlbuild.compiler.compile.constants import NOT_NULL_AUDIT_NAME from sqlbuild.compiler.compile.exceptions import CompileInputError from sqlbuild.compiler.compile.main._scope_index_with_compile_usages import ( scope_index_with_compile_usages, @@ -71,6 +79,10 @@ from sqlbuild.compiler.compile.main.function_node_type import function_node_type from sqlbuild.compiler.compile.models import ( AnalysisCacheContext, + CompactAnalysisCacheCandidate, + CompactAnalysisCacheModel, + CompactAnalysisCachePlan, + CompactBatchPreparation, CompileAuditInput, CompiledAudit, CompiledDirectLogicSqlTestPayload, @@ -103,6 +115,7 @@ ) from sqlbuild.compiler.compile.types import ( AttachedAuditTargetKind, + CompactBatchResponseCallback, CompiledResourceType, DiagnosticPhase, DiagnosticSeverity, @@ -116,6 +129,7 @@ from sqlbuild.compiler.scopes.main._validate_scope_index import validate_scope_index from sqlbuild.compiler.scopes.models import ScopeIndex from sqlbuild.compiler.sql_analysis.constants import BINDING_UNKNOWN_TABLE_INTERNAL_CODE +from sqlbuild.compiler.sql_analysis.exceptions import SqlAnalysisBoundaryError from sqlbuild.compiler.sql_analysis.main._schema_validation import get_schema_validations from sqlbuild.compiler.sql_analysis.models import ( SqlBindingDiagnostic, @@ -133,7 +147,6 @@ SchemaAuditInstance, SchemaColumn, SchemaDynamicColumnFamily, - SchemaSeedEntry, SourceColumnEntry, SourceEntry, TargetConfig, @@ -161,6 +174,10 @@ class _DynamicContractAnalysisInputs: authoritative_column_types_by_table: dict[str, dict[str, str]] +_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT: int = 256 +_COMPACT_BATCH_ENTRY_REUSE_MIN_PERCENT: int = 10 + + def assemble_compiled_project( *, inputs: CompileProjectInputs, @@ -467,7 +484,7 @@ def _assemble_compiled_model( relative_path=model_input.model_file.relative_path, query_sql=model_input.query_sql, config=model_input.config, - destination=_build_model_relation_target(model_input=model_input, model_name=model_name), + destination=build_model_relation_target(model_input=model_input, model_name=model_name), references=model_input.references, schema_entry=model_input.schema_entry, inferred_columns=inferred_columns, @@ -516,15 +533,85 @@ def _analyze_model_sql_in_parallel( ) for model_input in model_inputs ) + request_cache_keys: tuple[str, ...] = tuple( + request.cache_key for request in requests if request.cache_key is not None + ) + compact_batch_plan: CompactAnalysisCachePlan | None = build_compact_analysis_cache_plan( + context=analysis_cache, + models=tuple( + CompactAnalysisCacheModel( + name=_model_name(request.model_input), + cache_key=request.cache_key, + upstream_names=_referenced_model_names( + model_input=request.model_input, + available_names=analyzed_model_names, + ), + ) + for request in requests + ), + min_model_count=_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT, + ) + compact_cached_analyses: dict[str, PolyglotAnalysisResult] = {} + compact_batch_hit_count: int = 0 + compact_candidate: CompactAnalysisCacheCandidate | None = ( + read_compact_analysis_cache_candidate( + plan=compact_batch_plan, + expected_count=len(requests), + min_model_count=_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT, + ) + if compact_batch_plan is not None + else None + ) + if compact_candidate is not None: + try: + compact_analyses: tuple[_ModelSqlAnalysis, ...] = _analyze_model_sql_requests( + requests=requests, + cached_analyses={}, + column_nullability_by_table=column_nullability_by_table, + column_types_by_table=column_types_by_table, + inference_profile=inference_profile, + allow_compact_analysis=allow_compact_analysis, + rich_type_inference=rich_type_inference, + cached_compact_batch=( + compact_candidate.preparation, + compact_candidate.response, + ), + ) + except SqlAnalysisBoundaryError: + pass + else: + compact_cached_analyses = { + request_cache_keys[index]: compact_analyses[index].polyglot_analysis + for index in compact_candidate.matching_indexes + } + compact_batch_hit_count = len(compact_candidate.matching_indexes) + if compact_batch_hit_count == len(requests): + record_analysis_cache_metrics( + batch_hits=compact_batch_hit_count, + entry_hits=0, + misses=0, + bypasses=0, + ) + compact_analyses = _complete_inferred_bindings( + requests=requests, + analyses=compact_analyses, + complete_binding_schemas=complete_binding_schemas, + inference_profile=inference_profile, + ) + return { + _model_name(model_input): analysis + for model_input, analysis in zip(model_inputs, compact_analyses, strict=True) + } + entry_cache_keys: tuple[str, ...] = tuple( + cache_key for cache_key in request_cache_keys if cache_key not in compact_cached_analyses + ) cached_analyses: dict[str, PolyglotAnalysisResult] previous_signatures: dict[str, str] cached_output_signatures_by_key: dict[str, str] cached_analyses, previous_signatures, cached_output_signatures_by_key = ( read_model_analyses( context=analysis_cache, - cache_keys=tuple( - request.cache_key for request in requests if request.cache_key is not None - ), + cache_keys=entry_cache_keys, model_names=tuple(_model_name(request.model_input) for request in requests), upstream_model_names_by_key={ request.cache_key: _referenced_model_names( @@ -533,11 +620,45 @@ def _analyze_model_sql_in_parallel( ) for request in requests if request.cache_key is not None + and request.cache_key not in compact_cached_analyses }, ) if analysis_cache is not None else ({}, {}, {}) ) + entry_cache_hit_count: int = sum( + request.cache_key is not None and request.cache_key in cached_analyses + for request in requests + ) + if ( + compact_batch_plan is not None + and compact_batch_hit_count == 0 + and entry_cache_hit_count > 0 + and entry_cache_hit_count * 100 < len(requests) * _COMPACT_BATCH_ENTRY_REUSE_MIN_PERCENT + ): + cached_analyses.clear() + cached_output_signatures_by_key.clear() + entry_cache_hit_count = 0 + cached_analyses.update(compact_cached_analyses) + cached_output_signatures_by_key.update( + { + cache_key: model_analysis_output_signature(analysis) + for cache_key, analysis in compact_cached_analyses.items() + } + ) + record_analysis_cache_metrics( + batch_hits=compact_batch_hit_count, + entry_hits=entry_cache_hit_count, + misses=( + sum( + request.cache_key is None or request.cache_key not in cached_analyses + for request in requests + ) + if analysis_cache is not None + else 0 + ), + bypasses=(len(requests) if analysis_cache is None else 0), + ) analyses: tuple[_ModelSqlAnalysis, ...] = _analyze_model_sql_requests( requests=requests, cached_analyses=cached_analyses, @@ -546,6 +667,14 @@ def _analyze_model_sql_in_parallel( inference_profile=inference_profile, allow_compact_analysis=allow_compact_analysis, rich_type_inference=rich_type_inference, + on_compact_response=( + compact_analysis_batch_response_writer( + plan=compact_batch_plan, + count=len(requests), + ) + if compact_batch_plan is not None and not cached_analyses + else None + ), ) analyses = _complete_inferred_bindings( requests=requests, @@ -689,6 +818,8 @@ def _analyze_model_sql_requests( inference_profile: ExpressionInferenceProfile, allow_compact_analysis: bool, rich_type_inference: bool, + cached_compact_batch: tuple[CompactBatchPreparation, object] | None = None, + on_compact_response: CompactBatchResponseCallback | None = None, ) -> tuple[_ModelSqlAnalysis, ...]: if allow_compact_analysis: uncached: tuple[tuple[int, _ModelSqlAnalysisRequest], ...] = tuple( @@ -711,6 +842,8 @@ def _analyze_model_sql_requests( _should_recover_cte_facts(request.model_input) for _, request in uncached ), rich_type_inference=rich_type_inference, + cached_batch=cached_compact_batch, + on_response=(on_compact_response if len(uncached) == len(requests) else None), ) ) prepared_by_index = { @@ -1364,7 +1497,7 @@ def _assemble_compiled_seed( target_config: TargetConfig | None, effective_vars: dict[str, object], ) -> CompiledSeed: - target: CompiledRelationLocation = _build_seed_relation_target( + target: CompiledRelationLocation = build_seed_relation_target( seed_entry=seed_input.schema_entry, defaults=defaults, target_config=target_config, @@ -1733,196 +1866,3 @@ def _resolve_test_name(test_input: CompileSqlTestInput) -> str: if test_input.case_name is not None: return f"{parent_name} [{test_input.case_name}]" return parent_name - - -def _build_model_relation_target( - *, model_input: CompileModelInput, model_name: str -) -> CompiledRelationLocation: - raw_database: object | None = model_input.config.values.get("database") - raw_schema: object | None = model_input.config.values.get("schema") - raw_alias: object | None = model_input.config.values.get("alias") - database: str | None = raw_database if isinstance(raw_database, str) else None - schema: str | None = raw_schema if isinstance(raw_schema, str) else None - name: str = raw_alias if isinstance(raw_alias, str) else model_name - return CompiledRelationLocation( - database=database, - schema=schema, - name=name, - qualified_name=None, - logical_schema=model_input.config.logical_schema, - logical_database=model_input.config.logical_database, - ) - - -def _build_seed_relation_target( - *, - seed_entry: SchemaSeedEntry, - defaults: DefaultsConfig, - target_config: TargetConfig | None, - effective_vars: dict[str, object], -) -> CompiledRelationLocation: - logical_database, logical_schema = _resolve_seed_logical_namespace( - defaults=defaults, - effective_vars=effective_vars, - ) - if seed_entry.database is not None: - logical_database: str | None = _expand_seed_target_value( - raw_value=seed_entry.database, - seed_name=seed_entry.name, - database=logical_database, - schema=logical_schema, - effective_vars=effective_vars, - context_label=f"seed '{seed_entry.name}' database", - ) - if seed_entry.schema is not None: - logical_schema: str | None = _expand_seed_target_value( - raw_value=seed_entry.schema, - seed_name=seed_entry.name, - database=logical_database, - schema=logical_schema, - effective_vars=effective_vars, - context_label=f"seed '{seed_entry.name}' schema", - ) - validate_preserved_logical_namespace( - resource_label=f"Seed '{seed_entry.name}'", - logical_database=logical_database, - logical_schema=logical_schema, - target_config=target_config, - ) - database, schema = _apply_seed_target_overrides( - logical_database=logical_database, - logical_schema=logical_schema, - target_config=target_config, - effective_vars=effective_vars, - ) - return CompiledRelationLocation( - database=database, - schema=schema, - name=seed_entry.name, - qualified_name=None, - logical_database=logical_database, - logical_schema=logical_schema, - ) - - -def _resolve_seed_logical_namespace( - *, - defaults: DefaultsConfig, - effective_vars: dict[str, object], -) -> tuple[str | None, str | None]: - database: str | None = _expand_seed_default_value( - raw_value=( - defaults.seed_database if defaults.seed_database is not None else defaults.database - ), - effective_vars=effective_vars, - context_label="default seed database", - ) - schema: str | None = _expand_seed_default_value( - raw_value=defaults.seed_schema if defaults.seed_schema is not None else defaults.schema, - effective_vars=effective_vars, - context_label="default seed schema", - ) - return database, schema - - -def _apply_seed_target_overrides( - *, - logical_database: str | None, - logical_schema: str | None, - target_config: TargetConfig | None, - effective_vars: dict[str, object], -) -> tuple[str | None, str | None]: - if target_config is None: - return logical_database, logical_schema - database: str | None = logical_database - schema: str | None = logical_schema - if target_config.database is not None and target_config.database != PRESERVE_TARGET_VALUE: - database = _expand_seed_environment_value( - raw_value=target_config.database, - effective_vars=effective_vars, - context_label="target database", - ) - if target_config.schema is not None and target_config.schema != PRESERVE_TARGET_VALUE: - schema = _expand_seed_environment_value( - raw_value=target_config.schema, - effective_vars=effective_vars, - context_label="target schema", - ) - return database, schema - - -def _expand_seed_default_value( - *, raw_value: str | None, effective_vars: dict[str, object], context_label: str -) -> str | None: - if raw_value is None: - return None - return _expand_seed_environment_value( - raw_value=raw_value, - effective_vars=effective_vars, - context_label=context_label, - ) - - -def _expand_seed_environment_value( - *, raw_value: str, effective_vars: dict[str, object], context_label: str -) -> str | None: - if raw_value == PRESERVE_TARGET_VALUE: - return None - return str( - expand_template_data( - value=raw_value, - variables=effective_vars, - context_values={}, - context_label=context_label, - allow_context=False, - preserve_context_tokens=False, - preserve_unknown_context=False, - ) - ) - - -def _expand_seed_target_value( - *, - raw_value: str, - seed_name: str, - database: str | None, - schema: str | None, - effective_vars: dict[str, object], - context_label: str, -) -> str | None: - if raw_value == PRESERVE_TARGET_VALUE: - return None - return str( - expand_template_data( - value=raw_value, - variables=effective_vars, - context_values={ - "model.name": seed_name, - "model.database": database, - "model.schema": schema, - "model.alias": seed_name, - "destination.database": database, - "destination.schema": schema, - "destination.table": seed_name, - "destination.qualified": _build_seed_destination_qualified_context( - database=database, - schema=schema, - name=seed_name, - ), - }, - context_label=context_label, - allow_context=True, - preserve_context_tokens=False, - preserve_unknown_context=False, - ) - ) - - -def _build_seed_destination_qualified_context( - *, database: str | None, schema: str | None, name: str -) -> str | None: - if database is not None and schema is not None: - return f"{database}.{schema}.{name}" - if schema is not None: - return f"{schema}.{name}" - return None diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/targets.py b/src/sqlbuild/compiler/compile/_helpers/assembly/targets.py new file mode 100644 index 0000000000..28bd9e7bed --- /dev/null +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/targets.py @@ -0,0 +1,210 @@ +"""Compiled model and seed relation target resolution.""" + +from __future__ import annotations + +from sqlbuild.compiler.compile._helpers.config.namespace_validation import ( + validate_preserved_logical_namespace, +) +from sqlbuild.compiler.compile._helpers.render.templating import expand_template_data +from sqlbuild.compiler.compile.constants import PRESERVE_TARGET_VALUE +from sqlbuild.compiler.compile.models import CompiledRelationLocation, CompileModelInput +from sqlbuild.spec.contracts.models import DefaultsConfig, SchemaSeedEntry, TargetConfig + + +def build_model_relation_target( + *, model_input: CompileModelInput, model_name: str +) -> CompiledRelationLocation: + """Resolve one model's configured physical and logical destination.""" + + raw_database: object | None = model_input.config.values.get("database") + raw_schema: object | None = model_input.config.values.get("schema") + raw_alias: object | None = model_input.config.values.get("alias") + database: str | None = raw_database if isinstance(raw_database, str) else None + schema: str | None = raw_schema if isinstance(raw_schema, str) else None + name: str = raw_alias if isinstance(raw_alias, str) else model_name + return CompiledRelationLocation( + database=database, + schema=schema, + name=name, + qualified_name=None, + logical_schema=model_input.config.logical_schema, + logical_database=model_input.config.logical_database, + ) + + +def build_seed_relation_target( + *, + seed_entry: SchemaSeedEntry, + defaults: DefaultsConfig, + target_config: TargetConfig | None, + effective_vars: dict[str, object], +) -> CompiledRelationLocation: + """Resolve one seed's target-aware physical and logical destination.""" + + logical_namespace: tuple[str | None, str | None] = _resolve_seed_logical_namespace( + defaults=defaults, + effective_vars=effective_vars, + ) + logical_database: str | None = logical_namespace[0] + logical_schema: str | None = logical_namespace[1] + if seed_entry.database is not None: + logical_database = _expand_seed_target_value( + raw_value=seed_entry.database, + seed_name=seed_entry.name, + database=logical_database, + schema=logical_schema, + effective_vars=effective_vars, + context_label=f"seed '{seed_entry.name}' database", + ) + if seed_entry.schema is not None: + logical_schema = _expand_seed_target_value( + raw_value=seed_entry.schema, + seed_name=seed_entry.name, + database=logical_database, + schema=logical_schema, + effective_vars=effective_vars, + context_label=f"seed '{seed_entry.name}' schema", + ) + validate_preserved_logical_namespace( + resource_label=f"Seed '{seed_entry.name}'", + logical_database=logical_database, + logical_schema=logical_schema, + target_config=target_config, + ) + database, schema = _apply_seed_target_overrides( + logical_database=logical_database, + logical_schema=logical_schema, + target_config=target_config, + effective_vars=effective_vars, + ) + return CompiledRelationLocation( + database=database, + schema=schema, + name=seed_entry.name, + qualified_name=None, + logical_database=logical_database, + logical_schema=logical_schema, + ) + + +def _resolve_seed_logical_namespace( + *, + defaults: DefaultsConfig, + effective_vars: dict[str, object], +) -> tuple[str | None, str | None]: + database: str | None = _expand_seed_default_value( + raw_value=( + defaults.seed_database if defaults.seed_database is not None else defaults.database + ), + effective_vars=effective_vars, + context_label="default seed database", + ) + schema: str | None = _expand_seed_default_value( + raw_value=defaults.seed_schema if defaults.seed_schema is not None else defaults.schema, + effective_vars=effective_vars, + context_label="default seed schema", + ) + return database, schema + + +def _apply_seed_target_overrides( + *, + logical_database: str | None, + logical_schema: str | None, + target_config: TargetConfig | None, + effective_vars: dict[str, object], +) -> tuple[str | None, str | None]: + if target_config is None: + return logical_database, logical_schema + database: str | None = logical_database + schema: str | None = logical_schema + if target_config.database is not None and target_config.database != PRESERVE_TARGET_VALUE: + database = _expand_seed_environment_value( + raw_value=target_config.database, + effective_vars=effective_vars, + context_label="target database", + ) + if target_config.schema is not None and target_config.schema != PRESERVE_TARGET_VALUE: + schema = _expand_seed_environment_value( + raw_value=target_config.schema, + effective_vars=effective_vars, + context_label="target schema", + ) + return database, schema + + +def _expand_seed_default_value( + *, raw_value: str | None, effective_vars: dict[str, object], context_label: str +) -> str | None: + if raw_value is None: + return None + return _expand_seed_environment_value( + raw_value=raw_value, + effective_vars=effective_vars, + context_label=context_label, + ) + + +def _expand_seed_environment_value( + *, raw_value: str, effective_vars: dict[str, object], context_label: str +) -> str | None: + if raw_value == PRESERVE_TARGET_VALUE: + return None + return str( + expand_template_data( + value=raw_value, + variables=effective_vars, + context_values={}, + context_label=context_label, + allow_context=False, + preserve_context_tokens=False, + preserve_unknown_context=False, + ) + ) + + +def _expand_seed_target_value( + *, + raw_value: str, + seed_name: str, + database: str | None, + schema: str | None, + effective_vars: dict[str, object], + context_label: str, +) -> str | None: + if raw_value == PRESERVE_TARGET_VALUE: + return None + return str( + expand_template_data( + value=raw_value, + variables=effective_vars, + context_values={ + "model.name": seed_name, + "model.database": database, + "model.schema": schema, + "model.alias": seed_name, + "destination.database": database, + "destination.schema": schema, + "destination.table": seed_name, + "destination.qualified": _build_seed_destination_qualified_context( + database=database, + schema=schema, + name=seed_name, + ), + }, + context_label=context_label, + allow_context=True, + preserve_context_tokens=False, + preserve_unknown_context=False, + ) + ) + + +def _build_seed_destination_qualified_context( + *, database: str | None, schema: str | None, name: str +) -> str | None: + if database is not None and schema is not None: + return f"{database}.{schema}.{name}" + if schema is not None: + return f"{schema}.{name}" + return None diff --git a/src/sqlbuild/compiler/compile/models.py b/src/sqlbuild/compiler/compile/models.py index 2b13e69cdd..7a627778d9 100644 --- a/src/sqlbuild/compiler/compile/models.py +++ b/src/sqlbuild/compiler/compile/models.py @@ -1236,6 +1236,34 @@ class CompactBatchPreparation: projections: tuple[dict[str, object], ...] +@dataclass(frozen=True) +class CompactAnalysisCacheModel: + """One ordered model identity and its project-model dependencies.""" + + name: str + cache_key: str | None + upstream_names: tuple[str, ...] + + +@dataclass(frozen=True) +class CompactAnalysisCachePlan: + """Current invocation identity for one compact analysis generation.""" + + context: AnalysisCacheContext + batch_key: str + cache_keys: tuple[str, ...] + reuse_keys: tuple[str, ...] + + +@dataclass(frozen=True) +class CompactAnalysisCacheCandidate: + """Validated persisted generation and positions safe for reuse.""" + + matching_indexes: tuple[int, ...] + preparation: CompactBatchPreparation + response: object + + @dataclass(frozen=True) class CompactProjectionCaches: """Shared identity caches used while decoding compact native facts.""" diff --git a/src/sqlbuild/compiler/compile/types.py b/src/sqlbuild/compiler/compile/types.py index 6650714bba..9c7a2d2a58 100644 --- a/src/sqlbuild/compiler/compile/types.py +++ b/src/sqlbuild/compiler/compile/types.py @@ -3,10 +3,13 @@ from __future__ import annotations from enum import StrEnum -from typing import Protocol +from typing import TYPE_CHECKING, Protocol from sqlbuild.sql_values.models import SqlValue +if TYPE_CHECKING: + from sqlbuild.compiler.compile.models import CompactBatchPreparation + class TypedSqlValueRenderer(Protocol): """Adapter rendering operations consumed by authored SQL expansion.""" @@ -22,6 +25,12 @@ def render_typed_array(self, *, value: SqlValue) -> str: ... def render_typed_object(self, *, value: SqlValue) -> str: ... +class CompactBatchResponseCallback(Protocol): + """Keyword-only callback for publishing one native compact response.""" + + def __call__(self, *, preparation: CompactBatchPreparation, response: object) -> None: ... + + class AttachedAuditTargetKind(StrEnum): MODEL = "model" SOURCE = "source" diff --git a/src/sqlbuild/compiler/planner/main/execution/sql_test_model_chain.py b/src/sqlbuild/compiler/planner/main/execution/sql_test_model_chain.py index 7e23a0ee0d..c55e4f8b96 100644 --- a/src/sqlbuild/compiler/planner/main/execution/sql_test_model_chain.py +++ b/src/sqlbuild/compiler/planner/main/execution/sql_test_model_chain.py @@ -1,15 +1,15 @@ -"""Public planner entrypoint for SQL-test model closure discovery.""" +"""Public planner entrypoint for batched SQL-test model closure discovery.""" -from sqlbuild.compiler.compile.models import CompiledModel, CompiledProject, CompiledSqlTest -from sqlbuild.compiler.planner._helpers.sql_tests.assembly import resolve_test_model_chain_names +from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledProject, CompiledSqlTest +from sqlbuild.compiler.planner._helpers.sql_tests.assembly import build_sql_test_planning_context -def sql_test_model_chain_names( - *, - test: CompiledSqlTest, - project: CompiledProject, - model_map: dict[str, CompiledModel] | None = None, -) -> tuple[str, ...]: - """Return the exact model closure a SQL test plan will expand.""" +def sql_test_model_chain_names_by_key( + *, project: CompiledProject, tests: tuple[CompiledSqlTest, ...] +) -> dict[CompiledObjectKey, tuple[str, ...]]: + """Return model closures using one shared project dependency index.""" - return resolve_test_model_chain_names(test=test, project=project, model_map=model_map) + return build_sql_test_planning_context( + project=project, + tests=tests, + ).chain_names_by_test_key diff --git a/src/sqlbuild/compiler/profiling/constants.py b/src/sqlbuild/compiler/profiling/constants.py index b1e90441ba..ba6a20cddc 100644 --- a/src/sqlbuild/compiler/profiling/constants.py +++ b/src/sqlbuild/compiler/profiling/constants.py @@ -1,6 +1,6 @@ """Compile profiling constants.""" -from sqlbuild.compiler.profiling.types import CompileTimingPhase +from sqlbuild.compiler.profiling.types import CompileMetric, CompileTimingPhase COMPILE_TIMING_PHASES: tuple[CompileTimingPhase, ...] = ( "attachment_ms", @@ -12,3 +12,10 @@ "physical_write_ms", "stale_traversal_ms", ) + +COMPILE_METRICS: tuple[CompileMetric, ...] = ( + "analysis_batch_cache_hits", + "analysis_entry_cache_hits", + "analysis_cache_misses", + "analysis_cache_bypasses", +) diff --git a/src/sqlbuild/compiler/profiling/main/_metric.py b/src/sqlbuild/compiler/profiling/main/_metric.py new file mode 100644 index 0000000000..17c7c6367a --- /dev/null +++ b/src/sqlbuild/compiler/profiling/main/_metric.py @@ -0,0 +1,13 @@ +"""Internal compile metric recording entry point.""" + +from sqlbuild.compiler.profiling.classes.context import CompileTimingContext +from sqlbuild.compiler.profiling.models import CompileTimingCollector +from sqlbuild.compiler.profiling.types import CompileMetric + + +def record_compile_metric(*, metric: CompileMetric, value: int) -> None: + """Record one non-negative metric when a compile collector is active.""" + + collector: CompileTimingCollector | None = CompileTimingContext.active.get() + if collector is not None: + collector.add_metric(metric=metric, value=max(0, value)) diff --git a/src/sqlbuild/compiler/profiling/models.py b/src/sqlbuild/compiler/profiling/models.py index bbee3c1a6b..fd6ad7e715 100644 --- a/src/sqlbuild/compiler/profiling/models.py +++ b/src/sqlbuild/compiler/profiling/models.py @@ -2,8 +2,8 @@ from dataclasses import dataclass, field -from sqlbuild.compiler.profiling.constants import COMPILE_TIMING_PHASES -from sqlbuild.compiler.profiling.types import CompileTimingPhase +from sqlbuild.compiler.profiling.constants import COMPILE_METRICS, COMPILE_TIMING_PHASES +from sqlbuild.compiler.profiling.types import CompileMetric, CompileTimingPhase @dataclass(frozen=True) @@ -11,15 +11,21 @@ class CompileTimingCollector: """Accumulate elapsed nanoseconds for nested compile operations.""" elapsed_ns: dict[CompileTimingPhase, int] = field(default_factory=dict) + metrics: dict[CompileMetric, int] = field(default_factory=dict) def add(self, *, phase: CompileTimingPhase, elapsed_ns: int) -> None: """Add one measured operation to a compile phase.""" self.elapsed_ns[phase] = self.elapsed_ns.get(phase, 0) + elapsed_ns + def add_metric(self, *, metric: CompileMetric, value: int) -> None: + """Add a non-negative compile metric to the current invocation.""" + + self.metrics[metric] = self.metrics.get(metric, 0) + value + def as_milliseconds(self) -> dict[str, int]: """Return every supported phase in stable presentation order.""" return { phase: self.elapsed_ns.get(phase, 0) // 1_000_000 for phase in COMPILE_TIMING_PHASES - } + } | {metric: self.metrics.get(metric, 0) for metric in COMPILE_METRICS} diff --git a/src/sqlbuild/compiler/profiling/types.py b/src/sqlbuild/compiler/profiling/types.py index 678e447474..35337a97a3 100644 --- a/src/sqlbuild/compiler/profiling/types.py +++ b/src/sqlbuild/compiler/profiling/types.py @@ -12,3 +12,10 @@ "physical_write_ms", "stale_traversal_ms", ] + +type CompileMetric = Literal[ + "analysis_batch_cache_hits", + "analysis_entry_cache_hits", + "analysis_cache_misses", + "analysis_cache_bypasses", +] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py index 31bcbd3834..b17f1277b8 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py @@ -78,6 +78,10 @@ def test_given_waffle_shop_when_running_compile_json_then_it_reports_offline_que "cache_publication_ms", "physical_write_ms", "stale_traversal_ms", + "analysis_batch_cache_hits", + "analysis_entry_cache_hits", + "analysis_cache_misses", + "analysis_cache_bypasses", "total_ms", } assert all(isinstance(value, int) and value >= 0 for value in timings.values()) 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 47f519e8d6..6b90899436 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 @@ -105,6 +105,27 @@ class FreshProcessCompilePerformanceGuardTestCase: expected_semantic_fingerprint: str +@dataclass(frozen=True) +class FreshProcessCompileCachePerformanceGuardTestCase: + description: str + model_count: int + source_count: int + seed_count: int + function_count: int + macro_count: int + test_count: int + audit_count: int + expected_cold_max_wall_seconds: float + expected_warm_max_wall_seconds: float + expected_edit_max_wall_seconds: float + expected_max_rss_bytes: int + expected_max_cache_bytes: int + expected_cold_fingerprint: str + expected_leaf_edit_fingerprint: str + expected_macro_edit_fingerprint: str + expected_project_config_fingerprint: str + + @dataclass(frozen=True) class NamespaceCompileTestCase: description: str diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py index eafe20a348..c2900457eb 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -23,6 +23,9 @@ semantic_compile_fingerprint, ) from sqlbuild.cli.commands.main.entrypoint.entry import main +from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( + FreshProcessCompileCachePerformanceGuardTestCase, +) _DBT_SHAPED_SQL_SIZE_PROFILE: tuple[tuple[float, int], ...] = ( (0.50, 1_800), @@ -74,6 +77,77 @@ class FreshProcessCompileBenchmarkResult(NamedTuple): payload: dict[str, object] +class FreshProcessCompileCacheBenchmarkResult(NamedTuple): + cold: FreshProcessCompileBenchmarkResult + warm: FreshProcessCompileBenchmarkResult + leaf_edit: FreshProcessCompileBenchmarkResult + after_leaf_edit: FreshProcessCompileBenchmarkResult + macro_edit: FreshProcessCompileBenchmarkResult + after_macro_edit: FreshProcessCompileBenchmarkResult + project_config_edit: FreshProcessCompileBenchmarkResult + after_project_config_edit: FreshProcessCompileBenchmarkResult + cache_bytes: int + + +def fresh_process_compile_cache_metrics( + measurement: FreshProcessCompileBenchmarkResult, +) -> tuple[int, int, int, int]: + """Return semantic analysis cache counters from one compile result.""" + + timings: object = measurement.payload["compile_timings"] + assert isinstance(timings, dict) + timings_by_name: dict[str, object] = {str(key): value for key, value in timings.items()} + + def metric(name: str) -> int: + value: object = timings_by_name.get(name) + assert type(value) is int + return value + + return ( + metric("analysis_batch_cache_hits"), + metric("analysis_entry_cache_hits"), + metric("analysis_cache_misses"), + metric("analysis_cache_bypasses"), + ) + + +def assert_complete_compile_cache_hit( + *, measurement: FreshProcessCompileBenchmarkResult, model_count: int +) -> None: + """Assert every model was served by one of the analysis cache layers.""" + + batch_hits, entry_hits, misses, bypasses = fresh_process_compile_cache_metrics(measurement) + assert batch_hits + entry_hits == model_count + assert misses == 0 + assert bypasses == 0 + + +def assert_successful_compile_cache_payload( + *, + measurement: FreshProcessCompileBenchmarkResult, + test_case: FreshProcessCompileCachePerformanceGuardTestCase, +) -> None: + """Assert exact representative resource counts and clean diagnostics.""" + + assert measurement.payload["has_errors"] is False + assert measurement.payload["diagnostics"] == [] + assert measurement.payload["summary"] == { + "models": test_case.model_count, + "selected_models": test_case.model_count, + "sources": test_case.source_count, + "seeds": test_case.seed_count, + "selected_seeds": test_case.seed_count, + "functions": test_case.function_count, + "selected_functions": test_case.function_count, + "audits": test_case.audit_count, + "tests": test_case.test_count, + "hooks": 2, + "execution_layers": 54, + "errors": 0, + "warnings": 0, + } + + class DbtShapedCompileBenchmarkResult(NamedTuple): cold_seconds: float warm_median_seconds: float @@ -345,9 +419,134 @@ def run_fresh_process_semantic_compile_benchmark( ) target_dir: Path = project_dir / "target" assert not target_dir.exists() - measurement_path: Path = project_dir.parent / f"fresh-process-{model_count}-measurement.txt" - output_path: Path = project_dir.parent / f"fresh-process-{model_count}.json" - stderr_path: Path = project_dir.parent / f"fresh-process-{model_count}.stderr" + result: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"fresh-process-{model_count}", + expected_max_wall_seconds=expected_max_wall_seconds, + compile_args=("--no-cache",), + ) + assert not (target_dir / "cache").exists() + return result + + +def run_fresh_process_compile_cache_benchmark( + *, + project_dir: Path, + model_count: int, + source_count: int, + seed_count: int, + function_count: int, + macro_count: int, + test_count: int, + audit_count: int, + expected_cold_max_wall_seconds: float, + expected_warm_max_wall_seconds: float, + expected_edit_max_wall_seconds: float, +) -> FreshProcessCompileCacheBenchmarkResult: + """Measure exact and incremental cache reuse across independent CLI processes.""" + + write_semantic_compile_project( + project_dir=project_dir, + model_count=model_count, + source_count=source_count, + seed_count=seed_count, + function_count=function_count, + macro_count=macro_count, + test_count=test_count, + audit_count=audit_count, + ) + target_dir: Path = project_dir / "target" + assert not target_dir.exists() + cold: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-cold-{model_count}", + expected_max_wall_seconds=expected_cold_max_wall_seconds, + compile_args=(), + ) + warm: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-warm-{model_count}", + expected_max_wall_seconds=expected_warm_max_wall_seconds, + compile_args=(), + ) + _append_benchmark_edit(_generated_model_path(project_dir, model_count - 1), "leaf model") + leaf_edit: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-leaf-edit-{model_count}", + expected_max_wall_seconds=expected_edit_max_wall_seconds, + compile_args=(), + ) + after_leaf_edit: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-after-leaf-edit-{model_count}", + expected_max_wall_seconds=expected_warm_max_wall_seconds, + compile_args=(), + ) + macro_path: Path = project_dir / "macros" / "macro_00000.py" + _replace_benchmark_text( + path=macro_path, + old='return f"({expression} + 1)"', + new='return f"({expression} + 1000)"', + ) + macro_edit: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-macro-edit-{model_count}", + expected_max_wall_seconds=expected_edit_max_wall_seconds, + compile_args=(), + ) + after_macro_edit: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-after-macro-edit-{model_count}", + expected_max_wall_seconds=expected_warm_max_wall_seconds, + compile_args=(), + ) + project_config_path: Path = project_dir / "sqlbuild_project.toml" + _replace_benchmark_text( + path=project_config_path, + old='benchmark_revision = "0"', + new='benchmark_revision = "1"', + ) + project_config_edit: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-project-config-edit-{model_count}", + expected_max_wall_seconds=expected_cold_max_wall_seconds, + compile_args=(), + ) + after_project_config_edit: FreshProcessCompileBenchmarkResult = ( + _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-after-project-config-edit-{model_count}", + expected_max_wall_seconds=expected_warm_max_wall_seconds, + compile_args=(), + ) + ) + cache_dir: Path = target_dir / "cache" / "compiler" + cache_bytes: int = sum(path.stat().st_size for path in cache_dir.rglob("*.json")) + sum( + path.stat().st_size for path in cache_dir.rglob("*.sqlite3") + ) + return FreshProcessCompileCacheBenchmarkResult( + cold=cold, + warm=warm, + leaf_edit=leaf_edit, + after_leaf_edit=after_leaf_edit, + macro_edit=macro_edit, + after_macro_edit=after_macro_edit, + project_config_edit=project_config_edit, + after_project_config_edit=after_project_config_edit, + cache_bytes=cache_bytes, + ) + + +def _run_fresh_process_compile_benchmark( + *, + project_dir: Path, + label: str, + expected_max_wall_seconds: float, + compile_args: tuple[str, ...], +) -> FreshProcessCompileBenchmarkResult: + measurement_path: Path = project_dir.parent / f"{label}-measurement.txt" + output_path: Path = project_dir.parent / f"{label}.json" + stderr_path: Path = project_dir.parent / f"{label}.stderr" command: list[str] = [ "/usr/bin/time", "--quiet", @@ -361,7 +560,7 @@ def run_fresh_process_semantic_compile_benchmark( "--no-color", "compile", "--json", - "--no-cache", + *compile_args, ] with ( output_path.open("wb") as output_file, @@ -381,11 +580,10 @@ def run_fresh_process_semantic_compile_benchmark( assert isinstance(payload_object, dict) payload: dict[str, object] = payload_object peak_rss_bytes: int = int(peak_rss_kib_text) * 1024 - compiled_dir: Path = target_dir / "compiled" + compiled_dir: Path = project_dir / "target" / "compiled" semantic_fingerprint: str = semantic_compile_fingerprint( payload=payload, compiled_dir=compiled_dir ) - assert not (target_dir / "cache").exists() return FreshProcessCompileBenchmarkResult( elapsed_seconds=elapsed_seconds, peak_rss_bytes=peak_rss_bytes, diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py new file mode 100644 index 0000000000..79db1575f6 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_fresh_process_compile_cache_performance.py @@ -0,0 +1,224 @@ +"""Fresh-process cache guards for representative semantic projects.""" + +from __future__ import annotations + +import logging +from pathlib import Path + +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._test_types import ( + FreshProcessCompileCachePerformanceGuardTestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import ( + FreshProcessCompileBenchmarkResult, + FreshProcessCompileCacheBenchmarkResult, + assert_complete_compile_cache_hit, + assert_successful_compile_cache_payload, + fresh_process_compile_cache_metrics, + run_fresh_process_compile_cache_benchmark, +) + +_LOGGER: logging.Logger = logging.getLogger(__name__) +_GIB: int = 1024 * 1024 * 1024 +_MIB: int = 1024 * 1024 + + +@pytest.mark.performance +@pytest.mark.cache_compile_performance +@pytest.mark.parametrize( + "test_case", + [ + FreshProcessCompileCachePerformanceGuardTestCase( + description="models_3000_fresh_process_compile_cache_stays_incremental", + model_count=3_000, + source_count=713, + seed_count=141, + function_count=71, + macro_count=37, + test_count=2_945, + audit_count=5_056, + expected_cold_max_wall_seconds=18.0, + expected_warm_max_wall_seconds=10.0, + expected_edit_max_wall_seconds=11.0, + expected_max_rss_bytes=2 * _GIB, + expected_max_cache_bytes=96 * _MIB, + expected_cold_fingerprint=( + "02af81fb7723d30ec5082ed06fb039e45acba3d60b2d8a30fcbccb5a990bb9fc" + ), + expected_leaf_edit_fingerprint=( + "eb9c3fce74888baad3818aa3f732e8786e8306b2a10c0ff82c15bb74a5a1620a" + ), + expected_macro_edit_fingerprint=( + "1de325ddf1ed816c797218def79b23a45e4c020c96e564b71a8ec3cc86112d50" + ), + expected_project_config_fingerprint=( + "a320a971740013ccfbd09870864892635d5f5fca6a34ede117f3100537df8cc3" + ), + ), + FreshProcessCompileCachePerformanceGuardTestCase( + description="models_5000_fresh_process_compile_cache_stays_incremental", + model_count=5_000, + source_count=1_189, + seed_count=236, + function_count=118, + macro_count=61, + test_count=4_908, + audit_count=8_427, + expected_cold_max_wall_seconds=28.0, + expected_warm_max_wall_seconds=15.0, + expected_edit_max_wall_seconds=16.0, + expected_max_rss_bytes=2 * _GIB, + expected_max_cache_bytes=160 * _MIB, + expected_cold_fingerprint=( + "0c9258635fe4f9d31ec2dd1c99dbc63f5c8551ca7fea494003c5ab1c3a2ad57c" + ), + expected_leaf_edit_fingerprint=( + "d9f1e6840338b4ec75a105666a37e255a6c0eedb1edbcb924cc9c07ec06a7666" + ), + expected_macro_edit_fingerprint=( + "68ef78100da39c45b9a28751abff295c2f01a246f4515935cb288a648629ed6a" + ), + expected_project_config_fingerprint=( + "539f538bdd4387d1aa01cc9e36f378e2a0e61854229d8a482a8cfae4d7c123b6" + ), + ), + FreshProcessCompileCachePerformanceGuardTestCase( + description="models_10000_fresh_process_compile_cache_stays_incremental", + model_count=10_000, + source_count=2_377, + seed_count=471, + function_count=236, + macro_count=123, + test_count=9_816, + audit_count=16_855, + expected_cold_max_wall_seconds=50.0, + expected_warm_max_wall_seconds=28.0, + expected_edit_max_wall_seconds=30.0, + expected_max_rss_bytes=2 * _GIB, + expected_max_cache_bytes=320 * _MIB, + expected_cold_fingerprint=( + "6887ddd2547137efd8fe67d13fb07142f6287c0911d152d69eca0d495310e02b" + ), + expected_leaf_edit_fingerprint=( + "6447fccd4a45b5cfe959780cd9c3577d19e97bf7ad53b31cd30c31e1bdba0f34" + ), + expected_macro_edit_fingerprint=( + "999eddf213ff4f95b4b10b11890fc852c610410b730c38c81721d466a6534a68" + ), + expected_project_config_fingerprint=( + "17ae2532010bef51a653859929c3dc26ad88fdd77725cc791a2102184c86cbbc" + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_semantic_project_when_compiling_across_processes_then_cache_is_incremental( + tmp_path: Path, + test_case: FreshProcessCompileCachePerformanceGuardTestCase, +) -> None: + assert_required_cgroup_memory_limit() + result: FreshProcessCompileCacheBenchmarkResult = run_fresh_process_compile_cache_benchmark( + project_dir=tmp_path / f"semantic_cache_{test_case.model_count}", + model_count=test_case.model_count, + source_count=test_case.source_count, + seed_count=test_case.seed_count, + function_count=test_case.function_count, + macro_count=test_case.macro_count, + test_count=test_case.test_count, + audit_count=test_case.audit_count, + expected_cold_max_wall_seconds=test_case.expected_cold_max_wall_seconds, + expected_warm_max_wall_seconds=test_case.expected_warm_max_wall_seconds, + expected_edit_max_wall_seconds=test_case.expected_edit_max_wall_seconds, + ) + measurements: dict[str, FreshProcessCompileBenchmarkResult] = { + "cold": result.cold, + "warm": result.warm, + "leaf_edit": result.leaf_edit, + "after_leaf_edit": result.after_leaf_edit, + "macro_edit": result.macro_edit, + "after_macro_edit": result.after_macro_edit, + "project_config_edit": result.project_config_edit, + "after_project_config_edit": result.after_project_config_edit, + } + for label, measurement in measurements.items(): + _LOGGER.info( + "fresh-process cache models=%d path=%s wall=%.3fs peak_rss_bytes=%d " + "fingerprint=%s metrics=%s", + test_case.model_count, + label, + measurement.elapsed_seconds, + measurement.peak_rss_bytes, + measurement.semantic_fingerprint, + fresh_process_compile_cache_metrics(measurement), + ) + assert_successful_compile_cache_payload(measurement=measurement, test_case=test_case) + assert measurement.peak_rss_bytes < test_case.expected_max_rss_bytes + + assert result.cold.elapsed_seconds < test_case.expected_cold_max_wall_seconds + for measurement in ( + result.warm, + result.after_leaf_edit, + result.after_macro_edit, + result.after_project_config_edit, + ): + assert measurement.elapsed_seconds < test_case.expected_warm_max_wall_seconds + for measurement in (result.leaf_edit, result.macro_edit): + assert measurement.elapsed_seconds < test_case.expected_edit_max_wall_seconds + assert result.project_config_edit.elapsed_seconds < test_case.expected_cold_max_wall_seconds + assert result.cache_bytes < test_case.expected_max_cache_bytes + + assert result.cold.semantic_fingerprint == test_case.expected_cold_fingerprint + assert result.warm.semantic_fingerprint == test_case.expected_cold_fingerprint + assert result.leaf_edit.semantic_fingerprint == test_case.expected_leaf_edit_fingerprint + assert result.after_leaf_edit.semantic_fingerprint == test_case.expected_leaf_edit_fingerprint + assert result.macro_edit.semantic_fingerprint == test_case.expected_macro_edit_fingerprint + assert result.after_macro_edit.semantic_fingerprint == test_case.expected_macro_edit_fingerprint + assert ( + result.project_config_edit.semantic_fingerprint + == test_case.expected_project_config_fingerprint + ) + assert ( + result.after_project_config_edit.semantic_fingerprint + == test_case.expected_project_config_fingerprint + ) + + assert fresh_process_compile_cache_metrics(result.cold) == (0, 0, test_case.model_count, 0) + assert fresh_process_compile_cache_metrics(result.warm) == (test_case.model_count, 0, 0, 0) + assert fresh_process_compile_cache_metrics(result.leaf_edit) == ( + test_case.model_count - 1, + 0, + 1, + 0, + ) + assert fresh_process_compile_cache_metrics(result.after_leaf_edit) == ( + test_case.model_count - 1, + 1, + 0, + 0, + ) + macro_batch_hits, macro_entry_hits, macro_misses, macro_bypasses = ( + fresh_process_compile_cache_metrics(result.macro_edit) + ) + assert macro_batch_hits + macro_entry_hits + macro_misses == test_case.model_count + assert macro_batch_hits > 0 + assert macro_entry_hits > 0 + assert macro_misses == 7 + assert macro_bypasses == 0 + assert_complete_compile_cache_hit( + measurement=result.after_macro_edit, + model_count=test_case.model_count, + ) + config_batch_hits, config_entry_hits, config_misses, config_bypasses = ( + fresh_process_compile_cache_metrics(result.project_config_edit) + ) + assert config_batch_hits + config_entry_hits + config_misses == test_case.model_count + assert config_misses > 0 + assert config_bypasses == 0 + assert_complete_compile_cache_hit( + measurement=result.after_project_config_edit, + model_count=test_case.model_count, + ) 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 beebcc4a4f..dfa3f3f307 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 @@ -28,6 +28,7 @@ PolyglotAnalysisResult, ) from sqlbuild.compiler.lineage.types import InferredNullability +from sqlbuild.compiler.profiling.main.collect import collect_compile_timings from sqlbuild.compiler.references.types import SqlReferenceKind from tests.unit.src.sqlbuild.compiler.compile._helpers._test_types import ( AnalysisCacheTestCase, @@ -102,6 +103,105 @@ def test_given_successful_analysis_when_compiling_again_then_reuses_identical_ca ) +@pytest.mark.parametrize( + "test_case", + (AnalysisCacheTestCase(description="exact compact batch reuse", expected_count=1),), + ids=lambda case: case.description, +) +def test_given_exact_compact_batch_when_entry_rows_are_absent_then_warm_compile_reuses_batch( + test_case: AnalysisCacheTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + 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 = tmp_path / "target" / "cache" / "compiler" / "v10" / "model-analysis.sqlite3" + with sqlite3.connect(cache_path) as connection: + batch_count: int = connection.execute( + "SELECT COUNT(*) FROM model_analysis_compact_batch" + ).fetchone()[0] + _ = connection.execute("DELETE FROM model_analysis") + + with collect_compile_timings() as metrics: + warm_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) + + assert batch_count == 1 + assert warm_project.models == cold_project.models + assert warm_project.diagnostics == cold_project.diagnostics + assert metrics.as_milliseconds()["analysis_batch_cache_hits"] == test_case.expected_count + assert metrics.as_milliseconds()["analysis_entry_cache_hits"] == 0 + assert metrics.as_milliseconds()["analysis_cache_misses"] == 0 + assert metrics.as_milliseconds()["analysis_cache_bypasses"] == 0 + + +@pytest.mark.parametrize( + "test_case", + (AnalysisCacheTestCase(description="one changed compact model", expected_count=4),), + ids=lambda case: case.description, +) +def test_given_one_changed_model_when_entry_rows_are_absent_then_compact_batch_reuses_unchanged( + test_case: AnalysisCacheTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + 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 = tmp_path / "target" / "cache" / "compiler" / "v10" / "model-analysis.sqlite3" + with sqlite3.connect(cache_path) as connection: + _ = connection.execute("DELETE FROM model_analysis") + (tmp_path / "models" / "unrelated.sql").write_text( + "MODEL ();\n\nSELECT 3 AS id\n", + encoding="utf-8", + ) + + changed_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) + + models_by_name: dict[str, CompiledModel] = { + model.name: model for model in changed_project.models + } + cold_models_by_name: dict[str, CompiledModel] = { + model.name: model for model in cold_project.models + } + assert models_by_name["root"] == cold_models_by_name["root"] + assert models_by_name["middle"] == cold_models_by_name["middle"] + assert models_by_name["leaf"] == cold_models_by_name["leaf"] + assert models_by_name["unrelated"].query_sql == "SELECT 3 AS id" + assert len(models_by_name) == test_case.expected_count + assert changed_project.diagnostics == () + + +@pytest.mark.parametrize( + "test_case", + (AnalysisCacheTestCase(description="corrupt compact batch fallback", expected_count=1),), + ids=lambda case: case.description, +) +def test_given_corrupt_compact_batch_when_entry_rows_are_valid_then_warm_compile_falls_back( + test_case: AnalysisCacheTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + 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 = tmp_path / "target" / "cache" / "compiler" / "v10" / "model-analysis.sqlite3" + with sqlite3.connect(cache_path) as connection: + _ = connection.execute( + "UPDATE model_analysis_compact_batch SET payload = ?", + (b'{"version":1,"response":"tampered"}',), + ) + + warm_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) + + assert warm_project.models == cold_project.models + assert len(warm_project.models) == test_case.expected_count + assert warm_project.diagnostics == cold_project.diagnostics + + @pytest.mark.parametrize( "test_case", (AnalysisCacheTestCase(description="changed SQL cache miss", expected_count=1),), @@ -139,7 +239,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 = tmp_path / "target" / "cache" / "compiler" / "v9" / "model-analysis.sqlite3" + cache_path: Path = tmp_path / "target" / "cache" / "compiler" / "v10" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: persisted_contents: str = connection.execute( "SELECT payload FROM model_analysis" @@ -162,7 +262,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"] == 9 + assert repaired_payload["v"] == 10 assert isinstance(repaired_payload["s"], str) assert analyzer.call_count == test_case.expected_count @@ -180,7 +280,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 = tmp_path / "target" / "cache" / "compiler" / "v9" / "model-analysis.sqlite3" + cache_path: Path = tmp_path / "target" / "cache" / "compiler" / "v10" / "model-analysis.sqlite3" with sqlite3.connect(cache_path) as connection: _ = connection.execute( "UPDATE model_analysis SET payload = ?", @@ -610,6 +710,71 @@ def test_given_selected_upstream_change_when_compiling_full_project_then_stale_c assert analyzer.call_count == test_case.expected_count +@pytest.mark.parametrize( + "test_case", + (AnalysisCacheTestCase(description="compact upstream interface change", expected_count=4),), + ids=lambda case: case.description, +) +def test_given_upstream_interface_change_when_compact_batch_exists_then_dependents_reanalyze( + test_case: AnalysisCacheTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + write_repo_files( + tmp_path, + { + **_SELECTION_REPO_FILES, + "models/middle.sql": 'MODEL ();\n\nSELECT * FROM __ref("root")\n', + "models/leaf.sql": 'MODEL ();\n\nSELECT * FROM __ref("middle")\n', + }, + ) + monkeypatch.setattr(assembly_project, "_COMPACT_BATCH_CACHE_MIN_MODEL_COUNT", 1) + _ = compile_project_with_cache(project_dir=tmp_path) + (tmp_path / "models" / "root.sql").write_text( + "MODEL ();\n\nSELECT 1 AS id, 2 AS extra\n", + encoding="utf-8", + ) + + with collect_compile_timings() as metrics: + changed_project: CompiledProject = compile_project_with_cache(project_dir=tmp_path) + expected_project: CompiledProject = compile_project_with_cache( + project_dir=tmp_path, + no_cache=True, + ) + + models_by_name: dict[str, CompiledModel] = { + model.name: model for model in changed_project.models + } + expected_models_by_name: dict[str, CompiledModel] = { + model.name: model for model in expected_project.models + } + assert ( + models_by_name["root"].inferred_columns == expected_models_by_name["root"].inferred_columns + ) + assert models_by_name["middle"].inferred_columns == ( + expected_models_by_name["middle"].inferred_columns + ) + assert ( + models_by_name["leaf"].inferred_columns == expected_models_by_name["leaf"].inferred_columns + ) + recorded: dict[str, int] = metrics.as_milliseconds() + assert ( + sum( + recorded[name] + for name in ( + "analysis_batch_cache_hits", + "analysis_entry_cache_hits", + "analysis_cache_misses", + ) + ) + == test_case.expected_count + ) + assert recorded["analysis_batch_cache_hits"] == 1 + assert recorded["analysis_entry_cache_hits"] == 2 + assert recorded["analysis_cache_misses"] == 1 + + @pytest.mark.parametrize( "test_case", (AnalysisCacheTestCase(description="selected upstream analysis", expected_count=2),), @@ -692,10 +857,16 @@ def test_given_compile_cache_bypass_when_compiling_twice_then_both_runs_analyze_ monkeypatch.setattr(assembly_project, "analyze_columns_and_lineage_with_polyglot", analyzer) _ = compile_project_with_cache(project_dir=tmp_path) - _ = compile_project_with_cache(project_dir=tmp_path) + with collect_compile_timings() as metrics: + _ = compile_project_with_cache(project_dir=tmp_path) assert analyzer.call_count == test_case.expected_count assert not tuple((tmp_path / "target").rglob("*.sqlite3")) + recorded: dict[str, int] = metrics.as_milliseconds() + assert recorded["analysis_batch_cache_hits"] == 0 + assert recorded["analysis_entry_cache_hits"] == 0 + assert recorded["analysis_cache_misses"] == 0 + assert recorded["analysis_cache_bypasses"] == 1 @pytest.mark.parametrize(