diff --git a/crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/planning.rs b/crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/planning.rs index 058724762e..404be12990 100644 --- a/crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/planning.rs +++ b/crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/planning.rs @@ -77,10 +77,19 @@ struct PlanBatchRequest { #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct ChainBatchRequest { - models: Vec, + models: Vec, tests: Vec, } +/// Chain ordering reads only declared dependencies, so chain requests omit model SQL. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ChainModelInput { + name: String, + #[serde(default)] + model_dependencies: Vec, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct ChainBatchResponse { @@ -529,7 +538,7 @@ pub(crate) fn resolve_chains_json(request_json: &str) -> Result ( model.name, ModelInputOwned { - query_sql: model.query_sql, + query_sql: String::new(), model_dependencies: model.model_dependencies, }, ) diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py index 0e76848f80..54418cdf95 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/cache.py @@ -56,6 +56,8 @@ _COMPACT_CONFIDENCE_CODES: dict[str, int] = { CompactLineageFacts.confidence(code).value: code for code in range(3) } +_COMPACT_TRANSFORM_VALUES: tuple[str, ...] = tuple(_COMPACT_TRANSFORM_CODES) +_COMPACT_CONFIDENCE_VALUES: tuple[str, ...] = tuple(_COMPACT_CONFIDENCE_CODES) _RESOURCE_TYPE_VALUES: frozenset[str] = frozenset(item.value for item in CompiledResourceType) _NULLABILITY_BY_VALUE: dict[str, InferredNullability] = { item.value: item for item in InferredNullability @@ -586,9 +588,23 @@ def write_model_analyses( ) -> None: """Transactionally persist deterministic analyses; cache failures never fail compilation.""" + signed_analyses: dict[int, PolyglotAnalysisResult] = { + id(analysis): analysis + for analysis in (*analyses_by_key.values(), *(latest_analyses_by_model or {}).values()) + } + output_signatures: dict[int, str] = { + key: model_analysis_output_signature(analysis) for key, analysis in signed_analyses.items() + } rows: list[tuple[str, str]] = [] for cache_key, analysis in analyses_by_key.items(): - contents: str = _analysis_contents(cache_key=cache_key, analysis=analysis) + try: + contents: str = _analysis_contents( + cache_key=cache_key, + analysis=analysis, + output_signature=output_signatures[id(analysis)], + ) + except orjson.JSONEncodeError: + continue if len(contents.encode()) <= _MAX_CACHE_ENTRY_BYTES: rows.append((cache_key, contents)) signature_rows: list[tuple[str, str, str, str]] = [ @@ -596,7 +612,7 @@ def write_model_analyses( context.shared_fingerprint, context.signature_namespace, model_name, - model_analysis_output_signature(analysis), + output_signatures[id(analysis)], ) for model_name, analysis in (latest_analyses_by_model or {}).items() ] @@ -793,12 +809,18 @@ def _referenced_analysis_tables( return tuple(sorted(names)) -def _analysis_payload(*, cache_key: str, analysis: PolyglotAnalysisResult) -> dict[str, object]: +def _analysis_payload( + *, cache_key: str, analysis: PolyglotAnalysisResult, output_signature: str | None = None +) -> dict[str, object]: return { "v": _ANALYSIS_CACHE_VERSION, "k": cache_key, "a": analysis.analysis_succeeded, - "s": model_analysis_output_signature(analysis), + "s": ( + model_analysis_output_signature(analysis) + if output_signature is None + else output_signature + ), "c": ( None if analysis.columns is None @@ -806,7 +828,7 @@ def _analysis_payload(*, cache_key: str, analysis: PolyglotAnalysisResult) -> di [column.name, column.type, column.nullability.value] for column in analysis.columns ] ), - "l": [_lineage_column_payload(column) for column in analysis.lineage_columns], + "l": _lineage_payload(analysis.lineage_columns), "h": analysis.has_star, "sr": analysis.star_resolved, "b": [ @@ -825,6 +847,39 @@ def _analysis_payload(*, cache_key: str, analysis: PolyglotAnalysisResult) -> di } +def _lineage_payload(lineage_columns: Sequence[CompiledLineageColumnFact]) -> list[list[object]]: + if isinstance(lineage_columns, CompactLineageFacts): + compact: list[list[object]] | None = _compact_lineage_payload(lineage_columns) + if compact is not None: + return compact + return [_lineage_column_payload(column) for column in lineage_columns] + + +def _compact_lineage_payload(lineage: CompactLineageFacts) -> list[list[object]] | None: + """Serialise indexed rows without projecting facts; None defers to the validating projection.""" + + pool: tuple[str, ...] = lineage.string_pool + payload: list[list[object]] = [] + for name_index, transform_code, confidence_code, sources in lineage.rows: + upstream_columns: list[list[str]] = [] + for resource_type_index, resource_name_index, column_index in sources: + resource_type: str = pool[resource_type_index] + if resource_type not in _RESOURCE_TYPE_VALUES: + return None + upstream_columns.append( + [resource_type, lineage.resource_name(resource_name_index), pool[column_index]] + ) + payload.append( + [ + pool[name_index], + _COMPACT_TRANSFORM_VALUES[transform_code], + _COMPACT_CONFIDENCE_VALUES[confidence_code], + upstream_columns, + ] + ) + return payload + + def _lineage_column_payload(column: CompiledLineageColumnFact) -> list[object]: return [ column.output_column, @@ -1054,13 +1109,15 @@ def _cache_database_path(*, context: AnalysisCacheContext) -> Path: return context.root / f"v{_ANALYSIS_CACHE_VERSION}" / _CACHE_DATABASE_NAME -def _analysis_contents(*, cache_key: str, analysis: PolyglotAnalysisResult) -> str: - serialized_payload: str = json.dumps( - _analysis_payload(cache_key=cache_key, analysis=analysis), - ensure_ascii=False, - separators=(",", ":"), - sort_keys=True, - ) +def _analysis_contents( + *, cache_key: str, analysis: PolyglotAnalysisResult, output_signature: str | None = None +) -> str: + serialized_payload: str = orjson.dumps( + _analysis_payload( + cache_key=cache_key, analysis=analysis, output_signature=output_signature + ), + option=orjson.OPT_SORT_KEYS, + ).decode() return _CACHE_ENTRY_SEPARATOR.join( ( _cache_entry_digest(cache_key=cache_key, serialized_payload=serialized_payload), diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py b/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py index 6ec89bc911..7a7613b533 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py @@ -157,6 +157,11 @@ r"(?:\s|--[^\r\n]*(?:\r?\n|$)|/\*.*?\*/)*\.", re.DOTALL, ) +_QUALIFIED_SCAN_CONSUMING_TOKENS: tuple[str, ...] = ('"', "`", "[", "--", "/*") +_PLAIN_IDENTIFIER_PATTERN: re.Pattern[str] = re.compile(r"[a-z_$][a-z0-9_$]*") +_IDENTIFIER_CHARACTERS: frozenset[str] = frozenset( + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_$" +) def _infer_columns_with_polyglot( @@ -944,6 +949,8 @@ def _replace_refs_with_stubs( def _qualified_reference_names(*, query_sql: str, reference_names: Iterable[str]) -> frozenset[str]: + """Return qualifier reference names, searching names directly when no token can hide one.""" + if SQL_QUALIFIER_SEPARATOR_TOKEN not in query_sql: return frozenset() names_by_normalized: dict[str, list[str]] = {} @@ -952,7 +959,42 @@ def _qualified_reference_names(*, query_sql: str, reference_names: Iterable[str] names_by_normalized.setdefault(name.casefold(), []).append(name) if not names_by_normalized: return frozenset() + if not query_sql.isascii() or any( + token in query_sql for token in _QUALIFIED_SCAN_CONSUMING_TOKENS + ): + return _scanned_qualified_reference_names( + query_sql=query_sql, names_by_normalized=names_by_normalized + ) + + folded_sql: str = query_sql.lower() + qualified: set[str] = set() + for normalized, names in names_by_normalized.items(): + if _PLAIN_IDENTIFIER_PATTERN.fullmatch(normalized) is not None and _is_qualifier( + folded_sql=folded_sql, name=normalized + ): + qualified.update(names) + return frozenset(qualified) + + +def _is_qualifier(*, folded_sql: str, name: str) -> bool: + length: int = len(folded_sql) + start: int = folded_sql.find(name) + while start != -1: + end: int = start + len(name) + if (start == 0 or folded_sql[start - 1] not in _IDENTIFIER_CHARACTERS) and ( + end == length or folded_sql[end] not in _IDENTIFIER_CHARACTERS + ): + while end < length and folded_sql[end].isspace(): + end += 1 + if end < length and folded_sql[end] == SQL_QUALIFIER_SEPARATOR_TOKEN: + return True + start = folded_sql.find(name, start + 1) + return False + +def _scanned_qualified_reference_names( + *, query_sql: str, names_by_normalized: dict[str, list[str]] +) -> frozenset[str]: qualified: set[str] = set() for match in _QUALIFIED_IDENTIFIER_PATTERN.finditer(query_sql): identifier: str = next(value for value in match.groups() if value is not None) diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/metadata_validation.py b/src/sqlbuild/compiler/compile/_helpers/assembly/metadata_validation.py index cc0569c749..f5682e2d82 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/metadata_validation.py +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/metadata_validation.py @@ -66,12 +66,20 @@ def get_semantic_metadata_diagnostics( if not project.settings.sql_analysis: return () shapes: dict[str, dict[str, str]] = semantic_shapes(project=project, profile=profile) + functions: dict[str, CompiledFunction] = { + function.name.casefold(): function for function in project.functions + } diagnostics: list[CompilerDiagnostic] = [] for model in project.models: if model.config.values.get("sql_analysis") is False or not model.binding_validated: continue diagnostics.extend( - _function_errors(model=model, project=project, shapes=shapes, profile=profile) + _function_errors( + model=model, + functions=functions, + shapes=shapes, + profile=profile, + ) ) shape: dict[str, str] | None = shapes.get(model.name) if shape is None: @@ -431,13 +439,10 @@ def _names(value: object) -> tuple[str, ...]: def _function_errors( *, model: CompiledModel, - project: CompiledProject, + functions: dict[str, CompiledFunction], shapes: dict[str, dict[str, str]], profile: ExpressionInferenceProfile, ) -> tuple[CompilerDiagnostic, ...]: - functions: dict[str, CompiledFunction] = { - function.name.casefold(): function for function in project.functions - } if not functions or not any( reference.ref_kind in {SqlReferenceKind.UDF, SqlReferenceKind.TABLE_FUNCTION} for reference in model.references @@ -454,13 +459,28 @@ def _function_errors( parsed: Any = module.parse_one(request.sql, dialect=profile.sql_analysis_dialect) except module.PolyglotError: return () + calls: list[tuple[Any, Any, dict[str, Any], CompiledFunction]] = list( + _declared_function_calls(root=parsed, functions=functions) + ) + column_selects: dict[int, Any] = { + id(select): select + for _, select, payload, function in calls + if select is not None and _resolves_argument_columns(payload=payload, function=function) + } + relations_by_select: dict[int, list[dict[str, Any]]] = { + key: _select_relations(select=select) for key, select in column_selects.items() + } + table_names: set[str] = set() + for relations in relations_by_select.values(): + table_names.update(map(_relation_table_name, relations)) + folded_shapes: dict[str, dict[str, list[str]]] = { + table_name: _folded_columns(shapes[table_name]) + for table_name in table_names + if table_name in shapes + } diagnostics: list[CompilerDiagnostic] = [] - for call, select in _function_scopes(parsed): - payload: dict[str, Any] = call.to_dict().get("function", {}) + for call, select, payload, function in calls: name: str = str(payload.get("name", "")).removeprefix("__sqlbuild_udf_") - function: CompiledFunction | None = functions.get(name.casefold()) - if function is None: - continue arguments: list[dict[str, Any]] = payload.get("args", []) if len(arguments) != len(function.arguments): diagnostics.append( @@ -490,7 +510,11 @@ def _function_errors( literal.get("literal_type"), "UNKNOWN" ) elif column: - actual = _argument_column_type(column=column, select=select, shapes=shapes) + actual = _argument_column_type( + column=column, + relations=None if select is None else relations_by_select[id(select)], + folded_shapes=folded_shapes, + ) if actual != _UNKNOWN_TYPE and _different_argument_families( left=actual, right=declaration.type, profile=profile ): @@ -510,6 +534,26 @@ def _function_errors( return tuple(diagnostics) +def _declared_function_calls( + *, root: Any, functions: dict[str, CompiledFunction] +) -> Iterator[tuple[Any, Any, dict[str, Any], CompiledFunction]]: + """Yield calls of declared functions with their nearest select and serialised payload.""" + + for call, select in _function_scopes(root): + payload: dict[str, Any] = call.to_dict().get("function", {}) + name: str = str(payload.get("name", "")).removeprefix("__sqlbuild_udf_") + function: CompiledFunction | None = functions.get(name.casefold()) + if function is not None: + yield call, select, payload, function + + +def _resolves_argument_columns(*, payload: dict[str, Any], function: CompiledFunction) -> bool: + arguments: list[dict[str, Any]] = payload.get("args", []) + return len(arguments) == len(function.arguments) and any( + not argument.get("literal") and argument.get("column") for argument in arguments + ) + + def _function_scopes(root: Any) -> Iterator[tuple[Any, Any]]: pending: list[tuple[Any, Any]] = [(root, None)] node: Any @@ -523,29 +567,48 @@ def _function_scopes(root: Any) -> Iterator[tuple[Any, Any]]: pending.extend((child, select) for child in reversed(node.children())) -def _argument_column_type( - *, column: dict[str, Any], select: Any, shapes: dict[str, dict[str, str]] -) -> str: - if select is None: - return _UNKNOWN_TYPE +def _select_relations(*, select: Any) -> list[dict[str, Any]]: payload: dict[str, Any] = select.to_dict().get("select", {}) relations: list[dict[str, Any]] = list((payload.get("from") or {}).get("expressions", [])) relations.extend(join.get("this", {}) for join in payload.get("joins", [])) + return relations + + +def _relation_table_name(relation: dict[str, Any]) -> str: + return str(relation.get("table", {}).get("name", {}).get("name", "")) + + +def _folded_columns(shape: dict[str, str]) -> dict[str, list[str]]: + """Index column types by case-folded name, preserving declaration order per name.""" + + folded: dict[str, list[str]] = {} + for key, value in shape.items(): + folded.setdefault(key.casefold(), []).append(value) + return folded + + +def _argument_column_type( + *, + column: dict[str, Any], + relations: list[dict[str, Any]] | None, + folded_shapes: dict[str, dict[str, list[str]]], +) -> str: + if relations is None: + return _UNKNOWN_TYPE qualifier: str | None = (column.get("table") or {}).get("name") name: str = str(column.get("name", {}).get("name", "")) + folded_name: str = name.casefold() candidates: list[str] = [] for relation in relations: table: dict[str, Any] = relation.get("table", {}) - table_name: str = str(table.get("name", {}).get("name", "")) + table_name: str = _relation_table_name(relation) alias: str = str((table.get("alias") or {}).get("name", table_name)) if qualifier is not None and qualifier.casefold() != alias.casefold(): continue - shape: dict[str, str] | None = shapes.get(table_name) + shape: dict[str, list[str]] | None = folded_shapes.get(table_name) if shape is None: return _UNKNOWN_TYPE - candidates.extend( - value for key, value in shape.items() if key.casefold() == name.casefold() - ) + candidates.extend(shape.get(folded_name, ())) return candidates[0] if len(candidates) == 1 else _UNKNOWN_TYPE diff --git a/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py b/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py index 24b990bd30..21ba613ae5 100644 --- a/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py +++ b/src/sqlbuild/compiler/compile/_helpers/sharing/binding.py @@ -213,8 +213,9 @@ def _shared_reference_names( if SQL_QUOTED_IDENTIFIER_DELIMITER in cleaned_sql: return None for relation, columns in binding_schema.items(): - if SQL_QUOTED_IDENTIFIER_DELIMITER in relation or any( - SQL_QUOTED_IDENTIFIER_DELIMITER in column for column in columns + if ( + SQL_QUOTED_IDENTIFIER_DELIMITER in relation + or SQL_QUOTED_IDENTIFIER_DELIMITER in "".join(columns) ): return None if _qualified_reference_names(query_sql=cleaned_sql, reference_names=names): diff --git a/src/sqlbuild/compiler/planner/_helpers/sql_tests/native_planning.py b/src/sqlbuild/compiler/planner/_helpers/sql_tests/native_planning.py index 93742a0b40..d4a74c3bc4 100644 --- a/src/sqlbuild/compiler/planner/_helpers/sql_tests/native_planning.py +++ b/src/sqlbuild/compiler/planner/_helpers/sql_tests/native_planning.py @@ -161,7 +161,7 @@ def resolve_sql_test_model_chains( if not tests: return () request: dict[str, object] = { - "models": _model_requests(project=project), + "models": _chain_model_requests(project=project), "tests": [_test_request(test=test) for test in tests], } try: @@ -426,30 +426,34 @@ def _test_request( } -def _model_requests( - *, project: CompiledProject, adapter: BaseAdapter | None = None -) -> list[dict[str, object]]: - requests: list[dict[str, object]] = [] - for model in project.models: - dependencies: list[str] = [] - for dependency in model.deps: - if dependency.resource_type == CompiledResourceType.MODEL: - dependencies.append(dependency.name) - query_sql: str = ( - model.query_sql - if adapter is None - else render_test_cursor_intrinsics( +def _model_requests(*, project: CompiledProject, adapter: BaseAdapter) -> list[dict[str, object]]: + return [ + { + "name": model.name, + "querySql": render_test_cursor_intrinsics( sql=model.query_sql, model=model, adapter=adapter, test=None - ) - ) - requests.append( - { - "name": model.name, - "querySql": query_sql, - "modelDependencies": dependencies, - } - ) - return requests + ), + "modelDependencies": _model_dependencies(model=model), + } + for model in project.models + ] + + +def _chain_model_requests(*, project: CompiledProject) -> list[dict[str, object]]: + """Chain ordering reads only declared dependencies, so model SQL is not sent.""" + + return [ + {"name": model.name, "modelDependencies": _model_dependencies(model=model)} + for model in project.models + ] + + +def _model_dependencies(*, model: CompiledModel) -> list[str]: + return [ + dependency.name + for dependency in model.deps + if dependency.resource_type == CompiledResourceType.MODEL + ] def _cte_request(*, cte: CompileSqlTestCte) -> dict[str, str]: 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 e2ae536ae9..dcf8d2f9ef 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 @@ -177,6 +177,8 @@ class FreshProcessCompileCachePerformanceGuardTestCase: expected_edit_max_wall_seconds: float expected_max_warm_to_cold_ratio: float expected_max_edit_to_cold_ratio: float + expected_max_cache_write_cpu_overhead_ratio: float + expected_max_cache_write_wall_overhead_ratio: float expected_max_rss_bytes: int expected_max_cache_bytes: int expected_cold_fingerprint: 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 237b3bc064..8faed82407 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/helpers.py @@ -133,6 +133,7 @@ def __repr__(self) -> str: class FreshProcessCompileCacheBenchmarkResult(NamedTuple): + cache_disabled: FreshProcessCompileBenchmarkResult cold: FreshProcessCompileBenchmarkResult warm: FreshProcessCompileBenchmarkResult leaf_edit: FreshProcessCompileBenchmarkResult @@ -516,6 +517,13 @@ def run_fresh_process_compile_cache_benchmark( ) target_dir: Path = project_dir / "target" assert not target_dir.exists() + cache_disabled: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( + project_dir=project_dir, + label=f"cache-disabled-cold-{model_count}", + expected_max_wall_seconds=expected_cold_max_wall_seconds, + compile_args=("--no-cache",), + ) + shutil.rmtree(target_dir) cold: FreshProcessCompileBenchmarkResult = _run_fresh_process_compile_benchmark( project_dir=project_dir, label=f"cache-cold-{model_count}", @@ -584,6 +592,7 @@ def run_fresh_process_compile_cache_benchmark( path.stat().st_size for path in cache_dir.rglob("*.sqlite3") ) return FreshProcessCompileCacheBenchmarkResult( + cache_disabled=cache_disabled, cold=cold, warm=warm, leaf_edit=leaf_edit, 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 index 506dc68ff5..d6029a242e 100644 --- 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 @@ -25,8 +25,11 @@ _LOGGER: logging.Logger = logging.getLogger(__name__) _GIB: int = 1024 * 1024 * 1024 _MIB: int = 1024 * 1024 -_MAX_WARM_TO_COLD_RATIO: float = 0.65 -_MAX_EDIT_TO_COLD_RATIO: float = 0.60 +_MAX_WARM_TO_COLD_RATIO: float = 0.70 +_MAX_EDIT_TO_COLD_RATIO: float = 0.70 +# Writing the cache during a cold compile must stay a small same-runner cost over --no-cache. +_MAX_CACHE_WRITE_CPU_OVERHEAD_RATIO: float = 1.10 +_MAX_CACHE_WRITE_WALL_OVERHEAD_RATIO: float = 1.15 @pytest.mark.performance @@ -48,6 +51,8 @@ expected_edit_max_wall_seconds=11.5, expected_max_warm_to_cold_ratio=_MAX_WARM_TO_COLD_RATIO, expected_max_edit_to_cold_ratio=_MAX_EDIT_TO_COLD_RATIO, + expected_max_cache_write_cpu_overhead_ratio=_MAX_CACHE_WRITE_CPU_OVERHEAD_RATIO, + expected_max_cache_write_wall_overhead_ratio=_MAX_CACHE_WRITE_WALL_OVERHEAD_RATIO, expected_max_rss_bytes=2 * _GIB, expected_max_cache_bytes=96 * _MIB, expected_cold_fingerprint=( @@ -80,6 +85,8 @@ expected_edit_max_wall_seconds=18.0, expected_max_warm_to_cold_ratio=_MAX_WARM_TO_COLD_RATIO, expected_max_edit_to_cold_ratio=_MAX_EDIT_TO_COLD_RATIO, + expected_max_cache_write_cpu_overhead_ratio=_MAX_CACHE_WRITE_CPU_OVERHEAD_RATIO, + expected_max_cache_write_wall_overhead_ratio=_MAX_CACHE_WRITE_WALL_OVERHEAD_RATIO, expected_max_rss_bytes=2 * _GIB, expected_max_cache_bytes=160 * _MIB, expected_cold_fingerprint=( @@ -112,6 +119,8 @@ expected_edit_max_wall_seconds=37.5, expected_max_warm_to_cold_ratio=_MAX_WARM_TO_COLD_RATIO, expected_max_edit_to_cold_ratio=_MAX_EDIT_TO_COLD_RATIO, + expected_max_cache_write_cpu_overhead_ratio=_MAX_CACHE_WRITE_CPU_OVERHEAD_RATIO, + expected_max_cache_write_wall_overhead_ratio=_MAX_CACHE_WRITE_WALL_OVERHEAD_RATIO, expected_max_rss_bytes=2 * _GIB, expected_max_cache_bytes=320 * _MIB, expected_cold_fingerprint=( @@ -154,6 +163,7 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in scoped_macros=test_case.scoped_macros, ) measurements: dict[str, FreshProcessCompileBenchmarkResult] = { + "cache_disabled": result.cache_disabled, "cold": result.cold, "warm": result.warm, "leaf_edit": result.leaf_edit, @@ -179,6 +189,18 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in assert measurement.peak_rss_bytes < test_case.expected_max_rss_bytes assert result.cold.elapsed_seconds < test_case.expected_cold_max_wall_seconds + cache_write_wall_ratio: float = ( + result.cold.elapsed_seconds / result.cache_disabled.elapsed_seconds + ) + cache_write_cpu_ratio: float = result.cold.cpu_seconds / result.cache_disabled.cpu_seconds + _LOGGER.info( + "fresh-process cache models=%d cache-write overhead wall_ratio=%.3f cpu_ratio=%.3f", + test_case.model_count, + cache_write_wall_ratio, + cache_write_cpu_ratio, + ) + assert cache_write_wall_ratio <= test_case.expected_max_cache_write_wall_overhead_ratio + assert cache_write_cpu_ratio <= test_case.expected_max_cache_write_cpu_overhead_ratio for measurement in ( result.warm, result.after_leaf_edit, @@ -199,6 +221,7 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in 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.cache_disabled.semantic_fingerprint == test_case.expected_cold_fingerprint 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 @@ -214,6 +237,12 @@ def test_given_semantic_project_when_compiling_across_processes_then_cache_is_in == test_case.expected_project_config_fingerprint ) + assert fresh_process_compile_cache_metrics(result.cache_disabled) == ( + 0, + 0, + 0, + test_case.model_count, + ) 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) == ( diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py index 807d7529bc..264e05aba9 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/_test_types.py @@ -380,3 +380,10 @@ class PythonSourceReadPlanTestCase: description: str target_config: str expected_schema: str + + +@dataclass(frozen=True) +class FunctionArgumentTypeCase: + description: str + projection: str + expected_codes: tuple[str, ...] = () diff --git a/tests/integration/src/sqlbuild/compiler/pipeline/test_semantic_triage.py b/tests/integration/src/sqlbuild/compiler/pipeline/test_semantic_triage.py index 2bdd90eef7..b96ee287a2 100644 --- a/tests/integration/src/sqlbuild/compiler/pipeline/test_semantic_triage.py +++ b/tests/integration/src/sqlbuild/compiler/pipeline/test_semantic_triage.py @@ -15,7 +15,10 @@ from sqlbuild.compiler.compile.models import CompiledProject, CompilerDiagnostic from sqlbuild.compiler.discovery.main.discover import discover_project_inputs from sqlbuild.compiler.pipeline.main.compiled_project import build_compiled_project -from tests.integration.src.sqlbuild.compiler.pipeline._test_types import SemanticTriageCase +from tests.integration.src.sqlbuild.compiler.pipeline._test_types import ( + FunctionArgumentTypeCase, + SemanticTriageCase, +) @pytest.mark.parametrize( @@ -274,3 +277,67 @@ def test_given_repeated_root_messages_when_recovering_then_preserves_each_output if __name__ == "__main__": pytest.main([__file__, "-vv"]) + + +@pytest.mark.parametrize( + "test_case", + [ + FunctionArgumentTypeCase( + "matching qualified argument", projection='__udf("add_one")(o.amount) AS total' + ), + FunctionArgumentTypeCase( + "mismatched argument through a join alias", + projection='__udf("add_one")(c.customer_name) AS total', + expected_codes=("B301",), + ), + FunctionArgumentTypeCase( + "mismatched argument matches the column case-insensitively", + projection='__udf("add_one")(O.STATUS) AS total', + expected_codes=("B301",), + ), + FunctionArgumentTypeCase( + "several calls in one select report each mismatch", + projection=( + '__udf("add_one")(o.amount) AS total, __udf("add_one")(status) AS a, ' + '__udf("add_one")(customer_name) AS b' + ), + expected_codes=("B301", "B301"), + ), + FunctionArgumentTypeCase( + "argument in a nested select uses the nested relations", + projection=( + '(SELECT __udf("add_one")(i.status) FROM __source("orders") i LIMIT 1) AS total' + ), + expected_codes=("B301",), + ), + ], + ids=lambda case: case.description, +) +def test_given_udf_argument_columns_when_compiling_then_checks_declared_argument_types( + test_case: FunctionArgumentTypeCase, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "duckdb"\n') + (tmp_path / "sources").mkdir() + (tmp_path / "models").mkdir() + (tmp_path / "functions/sql").mkdir(parents=True) + (tmp_path / "sources/raw.yml").write_text( + "sources:\n" + " - name: orders\n table: orders\n contract: enforced\n columns:\n" + " - name: id\n type: INTEGER\n" + " - name: customer_id\n type: INTEGER\n" + " - name: status\n type: VARCHAR\n" + " - name: amount\n type: DOUBLE\n" + " - name: customers\n table: customers\n contract: enforced\n columns:\n" + " - name: id\n type: INTEGER\n" + " - name: customer_name\n type: VARCHAR\n" + ) + (tmp_path / "functions/sql/add_one.sql").write_text( + "FUNCTION (\n arguments (input_value DOUBLE),\n returns DOUBLE,\n);\n\ninput_value + 1\n" + ) + (tmp_path / "models/order_totals.sql").write_text( + f"MODEL (materialized table);\n\nSELECT {test_case.projection}\n" + 'FROM __source("orders") o JOIN __source("customers") c ON o.customer_id = c.id\n' + ) + main(["--project-dir", str(tmp_path), "compile", "--no-cache", "--json"]) + result: dict[str, Any] = json.loads(capsys.readouterr().out) + assert tuple(item["code"] for item in result["diagnostics"]) == test_case.expected_codes diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py index 667e9f22b7..71a583db11 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py @@ -1049,3 +1049,19 @@ class TypedMacroArgumentDependencyTestCase: macro_file_contents: str sql: str expected_dependency_names: frozenset[str] + + +@dataclass(frozen=True) +class QualifiedReferenceScanTestCase: + description: str + query_sql: str + reference_names: tuple[str, ...] + expected_names: frozenset[str] + + +@dataclass(frozen=True) +class CompactLineageCacheEncodingTestCase: + description: str + output_column: str + resource_names: dict[int, str] + expected_lineage_columns: int = 2 diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/helpers.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/helpers.py index 74fa229840..632b2e2b39 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/helpers.py @@ -1,3 +1,4 @@ +import sqlite3 from collections.abc import Callable from dataclasses import replace from pathlib import Path @@ -21,6 +22,7 @@ from sqlbuild.compiler.compile.main._assemble_project import assemble_project from sqlbuild.compiler.compile.main._build_compile_inputs import build_compile_inputs from sqlbuild.compiler.compile.models import ( + AnalysisCacheContext, CompileAdapterContext, CompileAnalysisSelection, CompiledDirectLogicSqlTestPayload, @@ -441,3 +443,14 @@ def render_compile_diagnostics(*, project: CompiledProject) -> str: f"= help: {diagnostic.help}" for diagnostic in project.diagnostics ) + + +def stored_analysis_contents(*, context: AnalysisCacheContext, cache_key: str) -> str: + """Return the raw stored payload of one analysis cache entry.""" + + database_path: Path = next(context.root.rglob("model-analysis.sqlite3")) + with sqlite3.connect(database_path) as connection: + row: tuple[str] = connection.execute( + "SELECT payload FROM model_analysis WHERE cache_key = ?", (cache_key,) + ).fetchone() + return row[0] 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 922ed461ff..c5d8619aaa 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 @@ -41,9 +41,11 @@ from tests.unit.src.sqlbuild.compiler.compile._helpers._test_types import ( AnalysisCacheTestCase, CachedLineageRoundTripTestCase, + CompactLineageCacheEncodingTestCase, ) from tests.unit.src.sqlbuild.compiler.compile._helpers.helpers import ( compile_project_with_cache, + stored_analysis_contents, ) _CACHE_REPO_FILES: dict[str, str] = { @@ -540,6 +542,69 @@ def test_given_cached_lineage_when_reading_then_facts_round_trip_exactly( ) +@pytest.mark.parametrize( + "test_case", + ( + CompactLineageCacheEncodingTestCase( + description="canonical relation names", + output_column="order_id", + resource_names={}, + ), + CompactLineageCacheEncodingTestCase( + description="shared result projected onto member relation names", + output_column='caf\u00e9 "note"\n\t\x01\u2028', + resource_names={4: "orders_archive"}, + ), + ), + ids=lambda case: case.description, +) +def test_given_compact_lineage_when_writing_then_entry_matches_projected_fact_encoding( + test_case: CompactLineageCacheEncodingTestCase, + tmp_path: Path, +) -> None: + compact: CompactLineageFacts = CompactLineageFacts( + string_pool=("source", "raw_orders", test_case.output_column, "model", "orders"), + rows=( + (2, 1, 1, ((0, 1, 2),)), + (2, 2, 2, ((0, 1, 2), (3, 4, 2))), + ), + resource_names=test_case.resource_names, + ) + compact_analysis: PolyglotAnalysisResult = PolyglotAnalysisResult( + analysis_succeeded=True, lineage_columns=compact + ) + projected_analysis: PolyglotAnalysisResult = PolyglotAnalysisResult( + analysis_succeeded=True, lineage_columns=tuple(compact) + ) + compact_context: AnalysisCacheContext = AnalysisCacheContext( + root=tmp_path / "compact", shared_fingerprint="shared" + ) + projected_context: AnalysisCacheContext = AnalysisCacheContext( + root=tmp_path / "projected", shared_fingerprint="shared" + ) + cache_key: str = "c" * 64 + + write_model_analyses(context=compact_context, analyses_by_key={cache_key: compact_analysis}) + write_model_analyses(context=projected_context, analyses_by_key={cache_key: projected_analysis}) + + compact_contents: str = stored_analysis_contents(context=compact_context, cache_key=cache_key) + serialized_payload: str = compact_contents.partition("\n")[2] + assert compact_contents == stored_analysis_contents( + context=projected_context, cache_key=cache_key + ) + assert serialized_payload == json.dumps( + json.loads(serialized_payload), ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + analyses, _, _ = read_model_analyses( + context=compact_context, + cache_keys=(cache_key,), + model_names=(), + upstream_model_names_by_key={cache_key: ()}, + ) + assert tuple(analyses[cache_key].lineage_columns) == tuple(compact) + assert len(analyses[cache_key].lineage_columns) == test_case.expected_lineage_columns + + @pytest.mark.parametrize( "test_case", (AnalysisCacheTestCase(description="stable per-model digests", expected_count=1),), diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py index e489937fb9..93fa1ead40 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_sql_analysis_columns.py @@ -39,6 +39,7 @@ NativeTypeInferenceModeTestCase, PolyglotAnalysisTestCase, QualifiedReferenceAnalysisTestCase, + QualifiedReferenceScanTestCase, SubstitutePlaceholderDefaultsTestCase, UnexpectedAnalysisFailureTestCase, ) @@ -377,6 +378,98 @@ def test_given_sql_without_dots_when_finding_qualified_references_then_skips_sca assert scanner.finditer.call_count == test_case.expected_count +@pytest.mark.parametrize( + "test_case", + ( + QualifiedReferenceScanTestCase( + description="alias qualifier does not qualify the relation", + query_sql="SELECT input.id FROM orders AS input", + reference_names=("orders",), + expected_names=frozenset(), + ), + QualifiedReferenceScanTestCase( + description="relation qualifier matches case-insensitively", + query_sql="SELECT ORDERS . id FROM orders", + reference_names=("orders", "Orders"), + expected_names=frozenset({"orders", "Orders"}), + ), + QualifiedReferenceScanTestCase( + description="longer identifier sharing a prefix is not a qualifier", + query_sql="SELECT orders_archive.id, x.orders FROM orders_archive, orders AS x", + reference_names=("orders", "archive"), + expected_names=frozenset(), + ), + QualifiedReferenceScanTestCase( + description="dollar and digit identifier characters bound the name", + query_sql="SELECT $orders.id, orders$1.id, orders.id FROM t", + reference_names=("orders",), + expected_names=frozenset({"orders"}), + ), + QualifiedReferenceScanTestCase( + description="string literal text still counts like the regex scan", + query_sql="SELECT 'customers.id' AS note FROM orders", + reference_names=("orders", "customers"), + expected_names=frozenset({"customers"}), + ), + QualifiedReferenceScanTestCase( + description="comment between qualifier and separator", + query_sql="SELECT orders /* note */ .id -- customers.id\nFROM orders", + reference_names=("orders", "customers"), + expected_names=frozenset({"orders", "customers"}), + ), + QualifiedReferenceScanTestCase( + description="comment consumed by an earlier qualifier hides its contents", + query_sql="SELECT x /* customers. */ .id FROM orders AS x", + reference_names=("orders", "customers"), + expected_names=frozenset(), + ), + QualifiedReferenceScanTestCase( + description="quoted identifiers qualify by their unquoted name", + query_sql='SELECT "orders".id, `customers`.id, [products].id FROM t', + reference_names=("orders", "customers", "products"), + expected_names=frozenset({"orders", "customers", "products"}), + ), + QualifiedReferenceScanTestCase( + description="quoted identifier containing a dot is one name", + query_sql='SELECT "orders.v2".id FROM t', + reference_names=("orders",), + expected_names=frozenset(), + ), + QualifiedReferenceScanTestCase( + description="non-identifier reference names never qualify", + query_sql="SELECT order-lines.id FROM t", + reference_names=("order-lines", "lines"), + expected_names=frozenset({"lines"}), + ), + QualifiedReferenceScanTestCase( + description="non-ascii SQL uses the full scan", + query_sql="SELECT caf\u00e9.id, orders.id FROM t", + reference_names=("orders", "caf\u00e9"), + expected_names=frozenset({"orders"}), + ), + ), + ids=lambda case: case.description, +) +def test_given_reference_names_when_finding_qualified_references_then_matches_full_scan( + test_case: QualifiedReferenceScanTestCase, +) -> None: + names_by_normalized: dict[str, list[str]] = {} + for name in test_case.reference_names: + names_by_normalized.setdefault(name.casefold(), []).append(name) + + result: frozenset[str] = analysis_columns._qualified_reference_names( + query_sql=test_case.query_sql, reference_names=test_case.reference_names + ) + + assert result == test_case.expected_names + assert ( + analysis_columns._scanned_qualified_reference_names( + query_sql=test_case.query_sql, names_by_normalized=names_by_normalized + ) + == test_case.expected_names + ) + + @pytest.mark.parametrize( "test_case", [