perf(compile): cut cache-write overhead and fresh compile preparation cost

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<ModelInput>,
+    models: Vec<ChainModelInput>,
     tests: Vec<TestInput>,
 }
 
+/// 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<String>,
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct ChainBatchResponse {
@@ -529,7 +538,7 @@ pub(crate) fn resolve_chains_json(request_json: &str) -> Result<String, String>
             (
                 model.name,
                 ModelInputOwned {
-                    query_sql: model.query_sql,
+                    query_sql: String::new(),
                     model_dependencies: model.model_dependencies,
                 },
             )

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,17 +588,31 @@ 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]] = [
         (
             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,20 +809,26 @@ 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
             else [
                 [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),

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)

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