Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions docs/concepts/models/external_models.md
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ If SQLMesh does not have access to an external table's metadata, the table will

`create_external_models` solely queries SQL engine metadata and does not query external tables themselves.

By default, the command overwrites same-gateway entries with freshly fetched columns and types. Use `--mode sync` to preserve any hand-edited metadata (such as descriptions, audits, or start dates) while only updating columns from the database. Use `--mode sync_prune` to additionally remove entries that are no longer referenced by any model.

### Gateway-specific external models

In some use-cases such as [isolated systems with multiple gateways](../../guides/isolated_systems.md#multiple-gateways), there are external models that only exist on a certain gateway.
Expand Down Expand Up @@ -116,9 +118,9 @@ The file can be constructed by hand using a standard text editor or IDE.

Sometimes, SQLMesh cannot infer the structure of a model and you need to add it manually.

However, since `sqlmesh create_external_models` replaces the `external_models.yaml` file, any manual changes you made to that file will be overwritten.
By default, running `sqlmesh create_external_models` replaces the entries in `external_models.yaml` any manual changes to that file may be lost. To preserve hand-edited metadata such as descriptions and audits, pass `--mode sync`, which only updates columns from the database while keeping all other fields.

The solution is to create the manual model definition files in the `external_models/` directory, like so:
Alternatively, you can place manual model definitions in the `external_models/` directory, like so:

```
external_models.yaml
Expand Down
10 changes: 9 additions & 1 deletion docs/reference/cli.md
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,15 @@ Usage: sqlmesh create_external_models [OPTIONS]
Create a schema file containing external model schemas.

Options:
--help Show this message and exit.
--strict Raise an error if the external model is
missing in the database
--mode [overwrite|sync|sync_prune]
The mode for updating external models.
overwrite replaces all entries (default), sync
syncs columns while preserving metadata and
warns on stale entries, sync_prune also
removes stale entries.
--help Show this message and exit.
```

## create_test
Expand Down
2 changes: 1 addition & 1 deletion docs/reference/notebook.md
Original file line number Diff line number Diff line change
Expand Up @@ -304,7 +304,7 @@ Migrate SQLMesh to the current running version

#### create_external_models
```
%create_external_models
%create_external_models [--strict] [--mode {overwrite,sync,sync_prune}]

Create a schema file containing external model schemas.
```
Expand Down
8 changes: 8 additions & 0 deletions sqlmesh/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -977,6 +977,14 @@ def rollback(obj: Context) -> None:
is_flag=True,
help="Raise an error if the external model is missing in the database",
)
@click.option(
"--mode",
type=click.Choice(["overwrite", "sync", "sync_prune"], case_sensitive=False),
default="overwrite",
help="The mode for updating external models. overwrite replaces all entries (default), "
"sync syncs columns while preserving metadata and warns on stale entries, "
"sync_prune also removes stale entries.",
)
@click.pass_obj
@error_handler
@cli_analytics
Expand Down
17 changes: 15 additions & 2 deletions sqlmesh/core/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,10 @@
from sqlmesh.core.plan.definition import UserProvidedFlags
from sqlmesh.core.reference import ReferenceGraph
from sqlmesh.core.scheduler import Scheduler, CompletionStatus
from sqlmesh.core.schema_loader import create_external_models_file
from sqlmesh.core.schema_loader import (
CreateExternalModelsMode,
create_external_models_file,
)
from sqlmesh.core.selector import Selector, NativeSelector
from sqlmesh.core.snapshot import (
DeployabilityIndex,
Expand Down Expand Up @@ -2486,15 +2489,24 @@ def rollback(self) -> None:
self._new_state_sync().rollback()

@python_api_analytics
def create_external_models(self, strict: bool = False) -> None:
def create_external_models(
self,
strict: bool = False,
mode: t.Union[CreateExternalModelsMode, str] = CreateExternalModelsMode.default,
) -> None:
"""Create a file to document the schema of external models.

The external models file contains all columns and types of external models, allowing for more
robust lineage, validation, and optimizations.

Args:
strict: If True, raise an error if the external model is missing in the database.
mode: The mode for updating external models. "overwrite" replaces all entries (default),
"sync" syncs columns while preserving metadata and warns on stale entries,
"sync_prune" also removes stale entries.
"""
if isinstance(mode, str):
mode = CreateExternalModelsMode(mode)
if not self._models:
self.load(update_schemas=False)

Expand Down Expand Up @@ -2529,6 +2541,7 @@ def create_external_models(self, strict: bool = False) -> None:
max_workers=self.concurrent_tasks,
strict=strict,
all_models=self._models,
mode=mode,
)

@python_api_analytics
Expand Down
121 changes: 106 additions & 15 deletions sqlmesh/core/schema_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import typing as t
from concurrent.futures import ThreadPoolExecutor
from enum import Enum
from pathlib import Path

from sqlglot import exp
Expand All @@ -11,10 +12,34 @@
from sqlmesh.core.engine_adapter import EngineAdapter
from sqlmesh.core.model.definition import Model
from sqlmesh.core.state_sync import StateReader
from sqlmesh.utils import UniqueKeyDict, yaml
from sqlmesh.utils import UniqueKeyDict, classproperty, yaml
from sqlmesh.utils.errors import SQLMeshError


class CreateExternalModelsMode(str, Enum):
"""Determines how existing entries are handled when creating external models."""

OVERWRITE = "overwrite"
SYNC = "sync"
SYNC_PRUNE = "sync_prune"

@classproperty
def default(cls) -> CreateExternalModelsMode:
return CreateExternalModelsMode.OVERWRITE

@property
def is_overwrite(self) -> bool:
return self == CreateExternalModelsMode.OVERWRITE

@property
def is_sync(self) -> bool:
return self == CreateExternalModelsMode.SYNC

@property
def is_sync_prune(self) -> bool:
return self == CreateExternalModelsMode.SYNC_PRUNE


def create_external_models_file(
path: Path,
models: UniqueKeyDict[str, Model],
Expand All @@ -25,6 +50,7 @@ def create_external_models_file(
max_workers: int = 1,
strict: bool = False,
all_models: t.Optional[t.Dict[str, Model]] = None,
mode: CreateExternalModelsMode = CreateExternalModelsMode.default,
) -> None:
"""Create or replace a YAML file with column and types of all columns in all external models.

Expand All @@ -40,16 +66,18 @@ def create_external_models_file(
all_models: FQN to model across all loaded repos. When provided, a dependency is only
classified as external if it is absent from this full set. This prevents cross-repo
internal models from being misclassified as external in multi-repo setups.
mode: The mode for updating external models. "overwrite" replaces all entries (default),
"sync" syncs columns while preserving metadata and warns on stale entries,
"sync_prune" also removes stale entries.
"""
known_models: t.Dict[str, Model] = all_models if all_models is not None else models
external_model_fqns = set()

for fqn, model in models.items():
if model.kind.is_external:
external_model_fqns.add(fqn)
for dep in model.depends_on:
if dep not in known_models:
external_model_fqns.add(dep)
external_model_fqns = {
dep
for model in models.values()
for dep in model.depends_on
if dep not in known_models or known_models[dep].kind.is_external
}

# Make sure we don't convert internal models into external ones.
existing_model_fqns = state_reader.nodes_exist(external_model_fqns, exclude_external=True)
Expand Down Expand Up @@ -79,15 +107,78 @@ def create_external_models_file(
if columns
]

# dont clobber existing entries from other gateways
entries_to_keep = (
[e for e in yaml.load(path) if e.get("gateway", None) != gateway]
if path.exists()
else []
if mode.is_overwrite:
# dont clobber existing entries from other gateways
entries_to_keep = (
[e for e in yaml.load(path) if e.get("gateway", None) != gateway]
if path.exists()
else []
)

with open(path, "w", encoding="utf-8") as file:
yaml.dump(entries_to_keep + schemas, file)
else:
_write_external_models_file_update(path, schemas, gateway, mode=mode)


def _write_external_models_file_update(
path: Path,
schemas: t.List[t.Dict[str, t.Any]],
gateway: t.Optional[str],
mode: CreateExternalModelsMode,
) -> None:
"""Write external models file for sync and sync_prune modes.

Preserves hand-edited metadata on existing entries, only syncing columns from the DB.
sync mode warns about stale entries but keeps them.
sync_prune mode removes stale entries.
"""
existing_entries = yaml.load(path) if path.exists() else []

other_gateway_entries: t.List[t.Dict[str, t.Any]] = []
same_gateway_entries: t.List[t.Dict[str, t.Any]] = []

for entry in existing_entries:
if entry.get("gateway", None) != gateway:
other_gateway_entries.append(entry)
else:
same_gateway_entries.append(entry)

same_gateway_by_name: t.Dict[str, t.Dict[str, t.Any]] = {}
for entry in same_gateway_entries:
same_gateway_by_name[entry["name"]] = entry

result = list(other_gateway_entries)
matched_in_schemas: t.Set[str] = {s["name"] for s in schemas}

for new_schema in schemas:
name = new_schema["name"]
if name in same_gateway_by_name:
existing = same_gateway_by_name[name]
existing["columns"] = new_schema["columns"]
result.append(existing)
else:
result.append(new_schema)

stale_count = 0
for entry in same_gateway_entries:
name = entry["name"]
if name not in matched_in_schemas:
stale_count += 1
if mode.is_sync:
get_console().log_warning(
f"External model '{name}' is no longer referenced but was preserved. "
"Use --mode sync_prune to automatically remove stale entries."
)
result.append(entry)

if stale_count and mode.is_sync_prune:
get_console().log_warning(
f"Removed {stale_count} stale external model(s) that are no longer referenced."
)

with open(path, "w", encoding="utf-8") as file:
yaml.dump(entries_to_keep + schemas, file)
with open(path, "w", encoding="utf-8") as file:
yaml.dump(result, file)


def get_columns(
Expand Down
10 changes: 9 additions & 1 deletion sqlmesh/magics.py
Original file line number Diff line number Diff line change
Expand Up @@ -738,12 +738,20 @@ def migrate(self, context: Context, line: str) -> None:
action="store_true",
help="Raise an error if the external model is missing in the database",
)
@argument(
"--mode",
choices=["overwrite", "sync", "sync_prune"],
default="overwrite",
help="The mode for updating external models. overwrite replaces all entries (default), "
"sync syncs columns while preserving metadata and warns on stale entries, "
"sync_prune also removes stale entries.",
)
@line_magic
@pass_sqlmesh_context
def create_external_models(self, context: Context, line: str) -> None:
"""Create a schema file containing external model schemas."""
args = parse_argstring(self.create_external_models, line)
context.create_external_models(strict=args.strict)
context.create_external_models(strict=args.strict, mode=args.mode)

@magic_arguments()
@argument(
Expand Down
Loading