diff --git a/docs/guides/multi_engine.md b/docs/guides/multi_engine.md index f2ccd31394..b4f179044e 100644 --- a/docs/guides/multi_engine.md +++ b/docs/guides/multi_engine.md @@ -11,9 +11,13 @@ SQLMesh enables this decoupling by supporting multiple engine adapters within a Configuring your project to use multiple engines follows a simple process: - Include all required [gateway connections](../reference/configuration.md#connection) in your configuration. -- Specify the `gateway` to be used for execution in the `MODEL` DDL. +- Set `model_defaults.gateway` to the gateway most models should use, and override individual models + with `gateway` in the `MODEL` DDL when needed. -If no gateway is explicitly defined for a model, the [default_gateway](../reference/configuration.md#default-gateway) of the project is used. +If no gateway is explicitly defined for a model, SQLMesh uses the project's +[`model_defaults.gateway`](../reference/model_configuration.md#model-defaults), when configured, +and otherwise uses its [default_gateway](../reference/configuration.md#default-gateway). This lets +all managed models in a project use a gateway without repeating it in every model definition. By default, virtual layer views are created in the `default_gateway`. This approach requires that all engines can read from and write to the same shared catalog, so a view in the `default_gateway` can access a table in another gateway. diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md index f5dd0eadf0..47dcaaced1 100644 --- a/docs/reference/model_configuration.md +++ b/docs/reference/model_configuration.md @@ -193,10 +193,15 @@ The SQLMesh project-level `model_defaults` key supports the following options, d - allow_partials - enabled - interval_unit +- gateway - pre_statements (described [here](../concepts/models/sql_models.md#pre--and-post-statements)) - post_statements (described [here](../concepts/models/sql_models.md#pre--and-post-statements)) - on_virtual_update (described [here](../concepts/models/sql_models.md#on-virtual-update-statements)) +The `gateway` default applies to managed SQL, Python, and seed models. It does not apply to +external models because an external model's `gateway` selects a gateway-specific source +definition. Set that gateway explicitly in `external_models.yaml` when needed. + ### Model Naming diff --git a/sqlmesh/core/config/model.py b/sqlmesh/core/config/model.py index 326d0baf6f..e7af572cc7 100644 --- a/sqlmesh/core/config/model.py +++ b/sqlmesh/core/config/model.py @@ -50,6 +50,7 @@ class ModelDefaultsConfig(BaseConfig): pre_statements: The list of SQL statements that get executed before a model runs. post_statements: The list of SQL statements that get executed before a model runs. on_virtual_update: The list of SQL statements to be executed after the virtual update. + gateway: The gateway used by models that do not specify one explicitly. """ @@ -76,6 +77,7 @@ class ModelDefaultsConfig(BaseConfig): pre_statements: t.Optional[t.List[t.Union[str, exp.Expr]]] = None post_statements: t.Optional[t.List[t.Union[str, exp.Expr]]] = None on_virtual_update: t.Optional[t.List[t.Union[str, exp.Expr]]] = None + gateway: t.Optional[str] = None _model_kind_validator = model_kind_validator _on_destructive_change_validator = on_destructive_change_validator diff --git a/sqlmesh/core/model/decorator.py b/sqlmesh/core/model/decorator.py index 54a43e0080..304c07276c 100644 --- a/sqlmesh/core/model/decorator.py +++ b/sqlmesh/core/model/decorator.py @@ -125,8 +125,12 @@ def models( blueprints = blueprints[0] + gateway = self.kwargs.get("gateway") + if isinstance(gateway, str) and gateway.lstrip().startswith("@"): + gateway = parse_one(gateway, dialect=dialect) + return create_models_from_blueprints( - gateway=self.kwargs.get("gateway"), + gateway=gateway, blueprints=blueprints, get_variables=get_variables, loader=self.model, diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py index c3569f9bd3..e8e122dece 100644 --- a/sqlmesh/core/model/definition.py +++ b/sqlmesh/core/model/definition.py @@ -2071,9 +2071,14 @@ def create_models_from_blueprints( loader_kwargs["default_catalog"] = original_default_catalog blueprint_variables = _extract_blueprint_variables(blueprint, path) - if gateway: + gateway_name: t.Optional[str] + if isinstance(gateway, str): + # Python decorator gateway names are literals, not SQL expressions. In particular, + # parsing a gateway such as "secondary-gw" as SQL would interpret it as subtraction. + gateway_name = gateway.lower() + elif gateway: rendered_gateway = render_expression( - expression=exp.maybe_parse(gateway, dialect=dialect), + expression=gateway, module_path=module_path, macros=loader_kwargs.get("macros"), jinja_macros=loader_kwargs.get("jinja_macros"), @@ -2082,7 +2087,11 @@ def create_models_from_blueprints( default_catalog=loader_kwargs.get("default_catalog"), blueprint_variables=blueprint_variables, ) - gateway_name = rendered_gateway[0].name if rendered_gateway else None + gateway_name = rendered_gateway[0].name.lower() if rendered_gateway else None + elif configured_gateway := (loader_kwargs.get("defaults") or {}).get("gateway"): + # Config gateway names are literals, not SQL expressions. In particular, parsing a + # gateway such as "secondary-gw" as SQL would interpret it as subtraction. + gateway_name = configured_gateway.lower() else: gateway_name = None @@ -2600,6 +2609,11 @@ def _create_model( kwargs["kind"] = create_model_kind(raw_kind, dialect, defaults or {}) defaults = {k: v for k, v in (defaults or {}).items() if k in klass.all_fields()} + if issubclass(klass, ExternalModel): + # An external model's gateway selects a gateway-specific source definition in + # external_models.yaml, so it must remain explicit rather than inheriting the + # gateway used to execute managed models in the project. + defaults.pop("gateway", None) if not issubclass(klass, SqlModel): defaults.pop("optimize_query", None) diff --git a/tests/core/integration/test_multi_repo.py b/tests/core/integration/test_multi_repo.py index 0b38231664..035c8bfda8 100644 --- a/tests/core/integration/test_multi_repo.py +++ b/tests/core/integration/test_multi_repo.py @@ -120,6 +120,71 @@ def test_multi(mocker): ] +def test_multi_repo_model_default_gateways(tmp_path: Path) -> None: + """Each project routes its models using its own default gateway.""" + repo_one = tmp_path / "repo_one" + repo_two = tmp_path / "repo_two" + (repo_one / "models").mkdir(parents=True) + (repo_two / "models").mkdir(parents=True) + + (repo_one / "models" / "default.sql").write_text( + "MODEL (name analytics.repo_one_default, kind FULL); SELECT @owner AS owner" + ) + (repo_one / "models" / "override.sql").write_text( + "MODEL (name analytics.explicit_override, kind FULL, gateway 'repo-two'); " + "SELECT @owner AS owner" + ) + (repo_two / "models" / "default.sql").write_text( + "MODEL (name analytics.repo_two_default, kind FULL); SELECT @owner AS owner" + ) + + def make_config(project: str, default_gateway: str) -> Config: + return Config( + project=project, + gateways={ + "repo-one": GatewayConfig( + connection=DuckDBConnectionConfig(database=str(tmp_path / "repo_one.duckdb")), + variables={"owner": "repo-one-variable"}, + ), + "repo-two": GatewayConfig( + connection=DuckDBConnectionConfig(database=str(tmp_path / "repo_two.duckdb")), + variables={"owner": "repo-two-variable"}, + ), + }, + default_gateway=default_gateway, + model_defaults=ModelDefaultsConfig(dialect="duckdb", gateway=default_gateway), + variables={"owner": "global-variable"}, + ) + + context = Context( + paths=[repo_one, repo_two], + config={ + repo_one: make_config("repo_one", "repo-one"), + repo_two: make_config("repo_two", "repo-two"), + }, + gateway="repo-one", + ) + + repo_one_model = context.get_model("repo_one.analytics.repo_one_default") + repo_two_model = context.get_model("repo_two.analytics.repo_two_default") + override_model = context.get_model("repo_two.analytics.explicit_override") + + assert repo_one_model.gateway == "repo-one" + assert repo_one_model.catalog == "repo_one" + assert repo_one_model.project == "repo_one" + assert context.render(repo_one_model.fqn).sql() == ("SELECT 'repo-one-variable' AS \"owner\"") + + assert repo_two_model.gateway == "repo-two" + assert repo_two_model.catalog == "repo_two" + assert repo_two_model.project == "repo_two" + assert context.render(repo_two_model.fqn).sql() == ("SELECT 'repo-two-variable' AS \"owner\"") + + assert override_model.gateway == "repo-two" + assert override_model.catalog == "repo_two" + assert override_model.project == "repo_one" + assert context.render(override_model.fqn).sql() == ("SELECT 'repo-two-variable' AS \"owner\"") + + @use_terminal_console def test_multi_repo_single_project_environment_statements_update(copy_to_temp_path): paths = copy_to_temp_path("examples/multi") diff --git a/tests/core/test_config.py b/tests/core/test_config.py index cbd4349999..0da5b6e22f 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -966,6 +966,27 @@ def test_gateway_model_defaults(tmp_path): assert ctx.config.model_defaults == expected +def test_model_defaults_gateway_from_yaml(tmp_path: Path) -> None: + config_path = tmp_path / "config.yaml" + config_path.write_text( + """ +gateways: + project_gateway: + connection: + type: duckdb + +model_defaults: + dialect: duckdb + gateway: project_gateway +""", + encoding="utf-8", + ) + + config = load_config_from_paths(Config, project_paths=[config_path]) + + assert config.model_defaults.gateway == "project_gateway" + + def test_model_defaults_cron_tz(tmp_path): """Test that cron_tz can be set in model_defaults.""" import zoneinfo diff --git a/tests/core/test_model.py b/tests/core/test_model.py index 1f3cde265b..2f8ff49f7d 100644 --- a/tests/core/test_model.py +++ b/tests/core/test_model.py @@ -5039,6 +5039,62 @@ def python_model_prop(context, **kwargs): assert m.interval_unit == IntervalUnit.QUARTER_HOUR +def test_explicit_hyphenated_gateway_python_model() -> None: + @model( + name="model_schema.python_explicit_gateway", + kind="full", + gateway="secondary-gw", + columns={"some_col": "int"}, + ) + def python_explicit_gateway(context, **kwargs): + yield {"some_col": 1} + + requested_variable_gateways: t.List[t.Optional[str]] = [] + + def get_variables(gateway: t.Optional[str]) -> t.Dict[str, str]: + requested_variable_gateways.append(gateway) + return {} + + loaded_models = model.get_registry()["model_schema.python_explicit_gateway"].models( + get_variables=get_variables, + module_path=Path("."), + path=Path("."), + dialect="duckdb", + defaults=ModelDefaultsConfig().dict(), + default_catalog="default_db", + default_catalog_per_gateway={"secondary-gw": "secondary_db"}, + ) + + assert len(loaded_models) == 1 + assert loaded_models[0].gateway == "secondary-gw" + assert loaded_models[0].catalog == "secondary_db" + assert requested_variable_gateways == ["secondary-gw"] + + +def test_model_defaults_gateway_python_model() -> None: + @model( + name="model_schema.python_gateway_default", + kind="full", + columns={"some_col": "int"}, + ) + def python_gateway_default(context, **kwargs): + yield {"some_col": 1} + + loaded_models = model.get_registry()["model_schema.python_gateway_default"].models( + get_variables=lambda gateway: {}, + module_path=Path("."), + path=Path("."), + dialect="duckdb", + defaults=ModelDefaultsConfig(gateway="python_gateway").dict(), + default_catalog="default_db", + default_catalog_per_gateway={"python_gateway": "python_db"}, + ) + + assert len(loaded_models) == 1 + assert loaded_models[0].gateway == "python_gateway" + assert loaded_models[0].catalog == "python_db" + + def test_model_defaults_macros(make_snapshot): model_defaults = ModelDefaultsConfig( table_format="@IF(@gateway = 'dev', 'iceberg', NULL)", @@ -12982,6 +13038,99 @@ def test_default_catalog_still_applied_to_supported_gateway(): assert model.catalog == "other_db", f"Expected catalog 'other_db', got: {model.catalog}" +@pytest.mark.parametrize( + ("model_gateway", "expected_gateway", "expected_catalog"), + [ + (None, "secondary-gw", "secondary_db"), + ("default_gw", "default_gw", "example_catalog"), + ], +) +def test_model_defaults_gateway( + model_gateway: t.Optional[str], expected_gateway: str, expected_catalog: str +) -> None: + """A project-level gateway default controls loading unless the model overrides it.""" + gateway_property = f"gateway {model_gateway}," if model_gateway else "" + expressions = d.parse( + f""" + MODEL ( + name my_schema.my_model, + kind FULL, + {gateway_property} + ); + + SELECT 1 AS id + """, + default_dialect="duckdb", + ) + requested_variable_gateways: t.List[t.Optional[str]] = [] + + def get_variables(gateway: t.Optional[str]) -> t.Dict[str, str]: + requested_variable_gateways.append(gateway) + return {} + + models = load_sql_based_models( + expressions, + get_variables=get_variables, + defaults=ModelDefaultsConfig(gateway="secondary-gw").dict(), + dialect="duckdb", + default_catalog_per_gateway={ + "default_gw": "example_catalog", + "secondary-gw": "secondary_db", + }, + default_catalog="example_catalog", + ) + + assert len(models) == 1 + assert models[0].gateway == expected_gateway + assert models[0].catalog == expected_catalog + assert requested_variable_gateways == [expected_gateway] + + +def test_model_defaults_gateway_with_blueprints() -> None: + expressions = d.parse( + """ + MODEL ( + name model_@suffix.my_model, + kind FULL, + blueprints ( + (suffix := one), + (suffix := two), + ), + ); + + SELECT 1 AS id + """, + default_dialect="duckdb", + ) + + models = load_sql_based_models( + expressions, + get_variables=lambda gateway: {}, + defaults=ModelDefaultsConfig(gateway="other_duckdb").dict(), + dialect="duckdb", + default_catalog_per_gateway={"other_duckdb": "other_db"}, + default_catalog="example_catalog", + ) + + assert {model.gateway for model in models} == {"other_duckdb"} + assert {model.catalog for model in models} == {"other_db"} + + +def test_external_model_does_not_inherit_model_defaults_gateway() -> None: + default_external_model = create_external_model( + "source_schema.default_source", + defaults=ModelDefaultsConfig(gateway="managed_gateway").dict(), + ) + explicit_external_model = create_external_model( + "source_schema.explicit_source", + defaults=ModelDefaultsConfig(gateway="managed_gateway").dict(), + gateway="source_gateway", + ) + + assert default_external_model.gateway is None + assert explicit_external_model.gateway == "source_gateway" + + def test_no_gateway_uses_global_default_catalog(): """ Control test: when a model does NOT specify a gateway, the global