Skip to content

Commit 20035da

Browse files
fresioASclaude
andcommitted
Perf: add opt-in project-index loading for lint
Add --use-project-index to targeted and full lint. The default path preserves the existing full context load. With both --use-project-index and --model, SQLMesh uses a persistent model-to-file dependency index to load, resolve, and validate only the selected models and their transitive upstream dependencies. A full indexed lint still loads and lints every model, but creates or refreshes the index for later targeted commands. Missing or expanded dependencies safely retry with a full indexed load. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Andreas Fredhøi <andreas.fredhoi@fresio.no>
1 parent e91c9b9 commit 20035da

6 files changed

Lines changed: 461 additions & 45 deletions

File tree

sqlmesh/cli/main.py

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,10 @@ def cli(
141141
if ctx.invoked_subcommand in SKIP_LOAD_COMMANDS:
142142
load = False
143143

144+
# Unlike the other commands above, lint can scope its own load for multi-project contexts.
145+
if ctx.invoked_subcommand == "lint":
146+
load = False
147+
144148
configs = load_configs(config, Context.CONFIG_TYPE, paths, dotenv_path=dotenv)
145149
log_limit = list(configs.values())[0].log_limit
146150

@@ -1208,6 +1212,11 @@ def environments(obj: Context) -> None:
12081212
multiple=True,
12091213
help="A model to lint. Multiple models can be linted. If no models are specified, every model will be linted.",
12101214
)
1215+
@click.option(
1216+
"--use-project-index",
1217+
is_flag=True,
1218+
help="Use the persistent project index. With --model, only the selected models and their upstream dependencies are loaded, resolved, and validated, so errors in unrelated models are not reported. Without --model, every model is still loaded and linted.",
1219+
)
12111220
@click.option(
12121221
"--local",
12131222
is_flag=True,
@@ -1220,9 +1229,18 @@ def environments(obj: Context) -> None:
12201229
def lint(
12211230
obj: Context,
12221231
models: t.Iterator[str],
1232+
use_project_index: bool,
12231233
) -> None:
12241234
"""Run the linter for the target model(s)."""
1225-
obj.lint_models(models)
1235+
obj.lint_models(
1236+
models,
1237+
use_project_index=use_project_index,
1238+
)
1239+
1240+
if not obj.models:
1241+
raise click.ClickException(
1242+
f"`{obj.path}` doesn't seem to have any models... cd into the proper directory or specify the path(s) with -p."
1243+
)
12261244

12271245

12281246
@cli.group(no_args_is_help=True)

sqlmesh/core/context.py

Lines changed: 110 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -418,6 +418,7 @@ def __init__(
418418
self._linters: t.Dict[str, Linter] = {}
419419
self._loaded: bool = False
420420
self._load_state: bool = load_state
421+
self._uncached_model_names: t.Set[str] = set()
421422
self._selector_cls = selector or NativeSelector
422423

423424
self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
@@ -641,11 +642,22 @@ def refresh(self) -> None:
641642
if any(loader.reload_needed() for loader in self._loaders):
642643
self.load()
643644

644-
def load(self, update_schemas: bool = True) -> GenericContext[C]:
645-
"""Load all files in the context's path."""
645+
def load(
646+
self,
647+
update_schemas: bool = True,
648+
model_fqns: t.Optional[t.Set[str]] = None,
649+
use_project_index: bool = False,
650+
) -> GenericContext[C]:
651+
"""Load files in the context's path, optionally scoped to specific models."""
646652
load_start_ts = time.perf_counter()
647653

648-
loaded_projects = [loader.load() for loader in self._loaders]
654+
loaded_projects = [
655+
loader.load(
656+
model_fqns=model_fqns,
657+
use_project_index=use_project_index,
658+
)
659+
for loader in self._loaders
660+
]
649661

650662
self.dag = DAG()
651663
self._standalone_audits.clear()
@@ -688,6 +700,27 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
688700
BUILTIN_RULES.union(project.user_rules), config.linter
689701
)
690702

703+
indexed_model_fqns = {
704+
fqn for project in loaded_projects for fqn in (project.indexed_model_fqns or set())
705+
}
706+
if model_fqns and (
707+
not model_fqns <= self._models.keys()
708+
or any(
709+
dependency in indexed_model_fqns and dependency not in self._models
710+
for model in self._models.values()
711+
for dependency in model.depends_on
712+
)
713+
):
714+
# A missing or stale index, a new model, or a dependency crossing project
715+
# boundaries requires a full load to preserve existing behavior.
716+
self.load(
717+
update_schemas=False,
718+
use_project_index=use_project_index,
719+
)
720+
if update_schemas:
721+
self._update_model_schemas_and_validate(model_fqns)
722+
return self
723+
691724
# Load environment statements from state for projects not in current load
692725
if self._load_state and any(self._projects):
693726
prod = self.state_reader.get_environment(c.PROD)
@@ -713,34 +746,13 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
713746
else:
714747
local_store[snapshot.name] = snapshot.node # type: ignore
715748

749+
self._uncached_model_names = uncached
750+
716751
for model in self._models.values():
717752
self.dag.add(model.fqn, model.depends_on)
718753

719754
if update_schemas:
720-
for fqn in self.dag:
721-
model = self._models.get(fqn) # type: ignore
722-
723-
if not model or fqn in uncached:
724-
continue
725-
726-
# make a copy of remote models that depend on local models or in the downstream chain
727-
# without this, a SELECT * FROM local will not propogate properly because the downstream
728-
# model will get mutated (schema changes) but the object is the same as the remote cache
729-
if any(dep in uncached for dep in model.depends_on):
730-
uncached.add(fqn)
731-
self._models.update({fqn: model.copy(update={"mapping_schema": {}})})
732-
continue
733-
734-
update_model_schemas(
735-
self.dag,
736-
models=self._models,
737-
cache_dir=self.cache_dir,
738-
)
739-
740-
models = self.models.values()
741-
for model in models:
742-
# The model definition can be validated correctly only after the schema is set.
743-
model.validate_definition()
755+
self._update_model_schemas_and_validate(model_fqns or None)
744756

745757
duplicates = set(self._models) & set(self._standalone_audits)
746758
if duplicates:
@@ -767,6 +779,53 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
767779
self._loaded = True
768780
return self
769781

782+
def _update_model_schemas_and_validate(self, model_fqns: t.Optional[t.Set[str]] = None) -> None:
783+
"""Updates the mapping schemas of the given models (all models by default) and validates their definitions.
784+
785+
Args:
786+
model_fqns: If provided, only these models and their transitive upstream
787+
dependencies are processed.
788+
"""
789+
if model_fqns is not None:
790+
model_fqns = {
791+
fqn for target in model_fqns for fqn in (target, *self.dag.upstream(target))
792+
}
793+
794+
uncached = set(self._uncached_model_names)
795+
796+
for fqn in self.dag:
797+
if model_fqns is not None and fqn not in model_fqns:
798+
continue
799+
800+
model = self._models.get(fqn)
801+
802+
if not model or fqn in uncached:
803+
continue
804+
805+
# make a copy of remote models that depend on local models or in the downstream chain
806+
# without this, a SELECT * FROM local will not propogate properly because the downstream
807+
# model will get mutated (schema changes) but the object is the same as the remote cache
808+
if any(dep in uncached for dep in model.depends_on):
809+
uncached.add(fqn)
810+
self._models.update({fqn: model.copy(update={"mapping_schema": {}})})
811+
continue
812+
813+
models = self._models
814+
if model_fqns is not None:
815+
models = UniqueKeyDict(
816+
"models", {fqn: model for fqn, model in self._models.items() if fqn in model_fqns}
817+
)
818+
819+
update_model_schemas(
820+
self.dag,
821+
models=models,
822+
cache_dir=self.cache_dir,
823+
)
824+
825+
for model in models.values():
826+
# The model definition can be validated correctly only after the schema is set.
827+
model.validate_definition()
828+
770829
@python_api_analytics
771830
def run(
772831
self,
@@ -3358,12 +3417,34 @@ def lint_models(
33583417
self,
33593418
models: t.Optional[t.Iterable[t.Union[str, Model]]] = None,
33603419
raise_on_error: bool = True,
3420+
use_project_index: bool = False,
33613421
) -> t.List[AnnotatedRuleViolation]:
3422+
input_models = list(models) if models is not None else []
3423+
3424+
if not self._loaded:
3425+
if input_models and use_project_index:
3426+
target_fqns = {
3427+
normalize_model_name(
3428+
model,
3429+
default_catalog=self.default_catalog,
3430+
dialect=self.default_dialect,
3431+
)
3432+
if isinstance(model, str)
3433+
else model.fqn
3434+
for model in input_models
3435+
}
3436+
self.load(
3437+
model_fqns=target_fqns,
3438+
use_project_index=True,
3439+
)
3440+
else:
3441+
self.load(use_project_index=use_project_index)
3442+
33623443
found_error = False
33633444

33643445
model_list = (
3365-
list(self.get_model(model, raise_if_missing=True) for model in models)
3366-
if models
3446+
list(self.get_model(model, raise_if_missing=True) for model in input_models)
3447+
if input_models
33673448
else self.models.values()
33683449
)
33693450
all_violations = []

0 commit comments

Comments
 (0)