Skip to content

Commit ee6ea4f

Browse files
authored
perf(rules): run custom rules across host processes (#928)
* perf(rules): split custom rules across host processes and start them first Custom-rule invocations are dealt round-robin by subject across up to one host process per available core, then merged back into single-host order so findings, read sets, untracked and uncacheable rules, observed digests, and the custom-rule cache are identical to one host. Any host failure reruns the whole plan in one host so the reported error is exactly the single-host error. Rule evaluation now launches the custom hosts and native built-in rules before preparing SQL rule inputs, and computes custom rule payloads and implementation fingerprints once instead of up to three times. * perf(rules): prepare speculative lint bodies in one native batch The early SQL lint expanded and prepared each body with its own native call, handing the GIL back and forth with graph building thousands of times. Bodies are now expanded first and prepared in one GIL-free native call; a body the batch cannot prepare is prepared again on its own so failures surface exactly as before, in file order relative to expansion failures. * fix(rules): keep split custom-rule hosts exact, bounded, and cancellable Rules that change module-level state on any host are evaluated again in one host over all of their subjects, replacing their split results, so cross-subject state such as a duplicate detector reports exactly what one host reports. The host count now honours cgroup v1 and v2 CPU quotas and is capped at eight. The first failing host writes a per-run cancellation marker. The other hosts check it between invocations, outside any Rule's guarded scope, and exit with a distinct cancelled status, and hosts that start late exit at once; no process is ever signalled. One invocation that is already running still ends or reaches its own timeout. One host then reruns the plan to report the single-host error.
1 parent b749d9a commit ee6ea4f

30 files changed

Lines changed: 2005 additions & 325 deletions

File tree

‎crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,31 @@ fn prepare_lint_sql(
117117
.map_err(value_error)
118118
}
119119

120+
/// Prepare many bodies in one detached call; `false` marks a body to prepare again individually.
121+
#[pyfunction]
122+
fn prepare_lint_sql_batch(
123+
py: Python<'_>,
124+
requests: Vec<LintPreparationRequest>,
125+
) -> PyResult<Vec<(bool, Option<crate::sql_lint::types::PreparedSql>)>> {
126+
py.compiler_detach(|| {
127+
Ok(requests
128+
.iter()
129+
.map(|request| {
130+
crate::bindings::_helpers::panics::catch_compiler_panic(|| {
131+
crate::sql_lint::main::preparation::prepare(
132+
&request.expanded,
133+
&request.before_expansion,
134+
&request.prior_sites,
135+
&request.dialect,
136+
)
137+
})
138+
.map_or((false, None), |prepared| (true, prepared))
139+
})
140+
.collect())
141+
})
142+
.map_err(value_error)
143+
}
144+
120145
#[pyfunction]
121146
fn lint_backtick_identifiers(dialect: &str) -> PyResult<bool> {
122147
compiler_guard(|| {
@@ -431,6 +456,7 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
431456
module.add_function(wrap_pyfunction!(run_custom_host_json, module)?)?;
432457
module.add_function(wrap_pyfunction!(lint_sql_json, module)?)?;
433458
module.add_function(wrap_pyfunction!(prepare_lint_sql, module)?)?;
459+
module.add_function(wrap_pyfunction!(prepare_lint_sql_batch, module)?)?;
434460
module.add(
435461
"NativeCompilerError",
436462
module

‎src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/rules/custom-rules/testing-and-determinism.md‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,14 @@
66
77
Online: https://sqlbuild.com/docs/concepts/rules/custom-rules/testing-and-determinism/
88

9+
## Contents
10+
11+
- Test Rules
12+
- Parallel evaluation
13+
- Helpers
14+
- Deterministic inputs
15+
- Cache granularity
16+
917
## Test Rules
1018

1119
Use the public harness to exercise discovery, compilation, and Rules evaluation:
@@ -39,6 +47,19 @@ the next. Compute findings from the subject and `ctx` alone.
3947

4048
Repository pytest files remain outside the SQLBuild project's SQL `tests/` directory.
4149

50+
## Parallel evaluation
51+
52+
On large projects, SQLBuild splits custom Rule evaluation across several host processes, at most
53+
one per available CPU and never more than eight. Each process evaluates a share of the subjects,
54+
and SQLBuild merges the findings back into the order a single process reports them.
55+
56+
A Rule that changes module-level state while it runs, for example by adding each model to a set of
57+
already-seen names, could report different findings when its subjects are split. SQLBuild detects
58+
the state change and evaluates that Rule again in one process over all of its subjects, so its
59+
findings and cache entries match a single-process run. The result stays correct, but the Rule no
60+
longer runs in parallel and costs an extra host process. Keep each subject's findings independent
61+
of the others so Rules can run in parallel.
62+
4263
## Helpers
4364

4465
Custom Rules can import repository-owned helpers under `rules/` and these deterministic standard

‎src/sqlbuild/_native.pyi‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,9 @@ def lint_sql_batch_json(request_json: str) -> str: ...
2323
def prepare_lint_sql(
2424
request: LintPreparationRequest,
2525
) -> tuple[str, list[tuple[str, int, int, int, int, str]], list[str]] | None: ...
26+
def prepare_lint_sql_batch(
27+
requests: list[LintPreparationRequest],
28+
) -> list[tuple[bool, tuple[str, list[tuple[str, int, int, int, int, str]], list[str]] | None]]: ...
2629
def lint_backtick_identifiers(dialect: str) -> bool: ...
2730
def load_config_json(project_dir: str) -> str: ...
2831
def selected_codes_json(request_json: str) -> str: ...

‎src/sqlbuild/lint/_helpers/expansion.py‎

Lines changed: 203 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
from __future__ import annotations
44

55
import re
6+
from dataclasses import replace
67
from pathlib import Path
78

89
import sqlbuild._native as _native
@@ -23,15 +24,32 @@
2324
from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
2425
from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
2526
from sqlbuild.compiler.scopes.types import DeclarationKind
27+
from sqlbuild.lint._helpers.headers import lint_body_ranges, lint_file_role
28+
from sqlbuild.lint._helpers.native import external_identifiers_for_headers
29+
from sqlbuild.lint._helpers.native_sql import check_lint_not_stopped
2630
from sqlbuild.lint._helpers.sqlbuild_tokens import (
2731
neutralize_context_interpolation,
2832
neutralize_generic_audit_parameters,
2933
neutralize_interpolation,
3034
sentinel_spans,
3135
)
32-
from sqlbuild.lint.constants import RELATION_IDENTITY_TEMPLATE, TEMPLATE_INTERPOLATION_START
36+
from sqlbuild.lint.constants import (
37+
HEADER_KIND_SCENARIO,
38+
HEADER_KIND_TEST,
39+
RELATION_IDENTITY_TEMPLATE,
40+
TEMPLATE_INTERPOLATION_START,
41+
)
3342
from sqlbuild.lint.exceptions import ProjectCompileError
34-
from sqlbuild.lint.models import InterpolationSite, LintBody, LintFileRole
43+
from sqlbuild.lint.models import (
44+
ExpandedLintFile,
45+
HeaderSpan,
46+
InterpolationSite,
47+
LintBody,
48+
LintFileRole,
49+
PendingLintBody,
50+
PreparedLintFiles,
51+
)
52+
from sqlbuild.lint.types import NativePreparedSql
3553
from sqlbuild.spec.contracts.main.resolve_effective_adapter_name import (
3654
resolve_effective_adapter_name,
3755
)
@@ -146,6 +164,38 @@ def prepare_lint_body(
146164
) -> LintBody:
147165
"""Expand one authored body and neutralize whatever interpolation remains."""
148166

167+
pending: PendingLintBody = expand_lint_body(
168+
role=role,
169+
file_path=file_path,
170+
contents=contents,
171+
body_range=body_range,
172+
context=context,
173+
dialect=dialect,
174+
external_identifiers=external_identifiers,
175+
allows_ceremonial_select=allows_ceremonial_select,
176+
allows_dynamic_output_star=allows_dynamic_output_star,
177+
compiled_expansion=compiled_expansion,
178+
)
179+
return finish_lint_body(
180+
pending=pending, prepared=_native.prepare_lint_sql(pending.request), dialect=dialect
181+
)
182+
183+
184+
def expand_lint_body(
185+
*,
186+
role: LintFileRole,
187+
file_path: Path,
188+
contents: str,
189+
body_range: tuple[int, int],
190+
context: SqlExpansionContext,
191+
dialect: str,
192+
external_identifiers: tuple[str, ...] = (),
193+
allows_ceremonial_select: bool = False,
194+
allows_dynamic_output_star: bool = False,
195+
compiled_expansion: CompiledSqlExpansion | None = None,
196+
) -> PendingLintBody:
197+
"""Expand one authored body like compile does, leaving its native preparation pending."""
198+
149199
body_start: int
150200
body_end: int
151201
body_start, body_end = body_range
@@ -184,50 +234,84 @@ def prepare_lint_body(
184234
raise ProjectCompileError(
185235
f"{file_path} could not be expanded, so its SQL cannot be linted: {error}"
186236
) from error
237+
return PendingLintBody(
238+
file_path=file_path,
239+
body_start=body_start,
240+
body_end=body_end,
241+
expansion_input=expansion_input,
242+
expanded=expanded,
243+
expansion_passes=expansion_passes,
244+
pre_expansion_sites=pre_expansion_sites,
245+
external_identifiers=external_identifiers,
246+
allows_ceremonial_select=allows_ceremonial_select,
247+
allows_dynamic_output_star=allows_dynamic_output_star,
248+
allows_empty_fixture_star=False,
249+
request={
250+
"expanded": expanded,
251+
"before_expansion": expansion_input,
252+
"prior_sites": [site.neutralized_start for site in pre_expansion_sites],
253+
"dialect": dialect,
254+
},
255+
)
256+
257+
258+
def prepare_lint_sql_batch(
259+
pending: tuple[PendingLintBody, ...],
260+
) -> tuple[NativePreparedSql | None, ...]:
261+
"""Prepare pending bodies in one GIL-free native call; failures re-raise per body, in order."""
262+
263+
if not pending:
264+
return ()
265+
results: list[tuple[bool, NativePreparedSql | None]] = _native.prepare_lint_sql_batch(
266+
[body.request for body in pending]
267+
)
268+
return tuple(
269+
prepared if succeeded else _native.prepare_lint_sql(body.request)
270+
for body, (succeeded, prepared) in zip(pending, results, strict=True)
271+
)
272+
273+
274+
def finish_lint_body(
275+
*, pending: PendingLintBody, prepared: NativePreparedSql | None, dialect: str
276+
) -> LintBody:
277+
"""Build the lint body from one body's native preparation, or the Python fallback."""
278+
279+
expanded: str = pending.expanded
187280
neutralized: str
188281
sites: tuple[InterpolationSite, ...]
189-
native_prepared: tuple[str, list[tuple[str, int, int, int, int, str]], list[str]] | None = (
190-
_native.prepare_lint_sql(
191-
{
192-
"expanded": expanded,
193-
"before_expansion": expansion_input,
194-
"prior_sites": [site.neutralized_start for site in pre_expansion_sites],
195-
"dialect": dialect,
196-
}
197-
)
198-
)
199-
if native_prepared is None:
282+
if prepared is None:
200283
neutralized, sites = neutralize_interpolation(body=expanded, dialect=dialect)
201284
externally_referenced_ctes: tuple[str, ...] = _externally_referenced_ctes(
202285
expanded=expanded,
203286
interpolation_sites=sites,
204-
pre_expansion_body=expansion_input,
205-
pre_expansion_sites=pre_expansion_sites,
287+
pre_expansion_body=pending.expansion_input,
288+
pre_expansion_sites=pending.pre_expansion_sites,
206289
)
207290
else:
208-
neutralized = native_prepared[0]
209-
sites = tuple(InterpolationSite(*site) for site in native_prepared[1])
210-
externally_referenced_ctes = tuple(native_prepared[2])
291+
neutralized = prepared[0]
292+
sites = tuple(InterpolationSite(*site) for site in prepared[1])
293+
externally_referenced_ctes = tuple(prepared[2])
211294
dependency_identifiers: tuple[str, ...] = tuple(
212295
site.sentinel
213296
for site in sites
214297
if _DEPENDENCY_INTRINSIC_PATTERN.match(site.original_text) is not None
215298
)
216299
return LintBody(
217-
file_path=file_path,
218-
body_start=body_start,
219-
body_end=body_end,
300+
file_path=pending.file_path,
301+
body_start=pending.body_start,
302+
body_end=pending.body_end,
220303
lint_text=neutralized,
221304
passes=(
222-
sentinel_spans(sites=pre_expansion_sites),
223-
*expansion_passes,
305+
sentinel_spans(sites=pending.pre_expansion_sites),
306+
*pending.expansion_passes,
224307
sentinel_spans(sites=sites),
225308
),
226-
external_identifiers=external_identifiers,
309+
external_identifiers=pending.external_identifiers,
227310
dependency_identifiers=dependency_identifiers,
228311
externally_referenced_ctes=externally_referenced_ctes,
229-
allows_ceremonial_select=allows_ceremonial_select,
230-
allows_dynamic_output_star=allows_dynamic_output_star,
312+
allows_ceremonial_select=pending.allows_ceremonial_select,
313+
allows_dynamic_output_star=pending.allows_dynamic_output_star,
314+
allows_empty_fixture_star=pending.allows_empty_fixture_star,
231315
dependency_relations=tuple(
232316
(
233317
site.sentinel,
@@ -249,3 +333,96 @@ def _is_sql_hook_path(*, role: LintFileRole) -> bool:
249333
return role.in_hook_directory or (
250334
role.in_project and role.declaration_kind is DeclarationKind.SQL_HOOK
251335
)
336+
337+
338+
def expand_file_bodies(
339+
*,
340+
file_path: Path,
341+
contents: str,
342+
headers: tuple[HeaderSpan, ...],
343+
context: SqlExpansionContext,
344+
dialect: str,
345+
project_dir: Path,
346+
relative_path: Path,
347+
dynamic_output_paths: frozenset[Path],
348+
compiled_expansions: dict[Path, CompiledSqlExpansion] | None,
349+
) -> ExpandedLintFile:
350+
"""Expand one file's bodies up to the first body that cannot be expanded."""
351+
352+
check_lint_not_stopped()
353+
bodies: list[PendingLintBody] = []
354+
role: LintFileRole = lint_file_role(
355+
file_path=file_path, project_dir=project_dir, relative_path=relative_path
356+
)
357+
allows_dynamic_output_star: bool = (
358+
bool(dynamic_output_paths) and file_path.resolve() in dynamic_output_paths
359+
)
360+
compiled_expansion: CompiledSqlExpansion | None = (compiled_expansions or {}).get(file_path)
361+
external_identifiers: tuple[str, ...] = external_identifiers_for_headers(
362+
contents=contents, headers=headers
363+
)
364+
allows_ceremonial_select: bool = any(
365+
header.kind in {HEADER_KIND_TEST, HEADER_KIND_SCENARIO} for header in headers
366+
)
367+
allows_empty_fixture_star: bool = any(header.kind == HEADER_KIND_TEST for header in headers)
368+
body_start: int
369+
body_end: int
370+
try:
371+
for body_start, body_end in lint_body_ranges(
372+
contents=contents,
373+
headers=headers,
374+
file_path=file_path,
375+
project_dir=project_dir,
376+
role=role,
377+
):
378+
bodies.append(
379+
replace(
380+
expand_lint_body(
381+
role=role,
382+
file_path=file_path,
383+
contents=contents,
384+
body_range=(body_start, body_end),
385+
context=context,
386+
dialect=dialect,
387+
external_identifiers=external_identifiers,
388+
allows_ceremonial_select=allows_ceremonial_select,
389+
allows_dynamic_output_star=allows_dynamic_output_star,
390+
compiled_expansion=compiled_expansion,
391+
),
392+
allows_empty_fixture_star=allows_empty_fixture_star,
393+
)
394+
)
395+
except ProjectCompileError as error:
396+
return ExpandedLintFile(file_path=file_path, bodies=tuple(bodies), failure=error)
397+
return ExpandedLintFile(file_path=file_path, bodies=tuple(bodies), failure=None)
398+
399+
400+
def prepare_expanded_files(
401+
*, expanded_files: tuple[ExpandedLintFile, ...], dialect: str, skip_unexpandable: bool
402+
) -> PreparedLintFiles:
403+
"""Prepare every expanded body natively at once, then report failures in file order."""
404+
405+
check_lint_not_stopped()
406+
pending: list[PendingLintBody] = []
407+
for expanded in expanded_files:
408+
pending.extend(expanded.bodies)
409+
prepared: tuple[NativePreparedSql | None, ...] = prepare_lint_sql_batch(tuple(pending))
410+
bodies: list[LintBody] = []
411+
unexpandable: dict[Path, str] = {}
412+
position: int = 0
413+
for expanded in expanded_files:
414+
file_bodies: list[LintBody] = []
415+
for body in expanded.bodies:
416+
file_bodies.append(
417+
finish_lint_body(pending=body, prepared=prepared[position], dialect=dialect)
418+
)
419+
position += 1
420+
if expanded.failure is None:
421+
bodies.extend(file_bodies)
422+
elif skip_unexpandable:
423+
unexpandable[expanded.file_path] = str(expanded.failure).removeprefix(
424+
f"{expanded.file_path} "
425+
)
426+
else:
427+
raise expanded.failure
428+
return PreparedLintFiles(bodies=tuple(bodies), unexpandable=unexpandable)

0 commit comments

Comments
 (0)