From 139592aa0094a914d3d78646befc0825d69dbe75 Mon Sep 17 00:00:00 2001 From: Benrishty <64180788+Benrishty@users.noreply.github.com> Date: Tue, 14 Jul 2026 12:57:15 -0400 Subject: [PATCH] fix: don't emit external models as multi-asset outputs (#65) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SQLMeshController.non_external_models_dag()` did not actually filter external models — its only guard was `if not model`, which skips FQNs that resolve to no model at all. External models resolve to a real `Model` (kind.is_external), so they passed through and `to_asset_outs()` emitted an `AssetOut` for each. External models are source tables the project READS, so materializing them as outputs is wrong and collides with any existing asset that already owns that key (e.g. a dlt asset), which surfaces as `DagsterInvalidDefinitionError: Duplicate asset key` (see #65, #53). Two changes make externals dependency-only: 1. `non_external_models_dag()` now skips `model.kind.is_external`, so externals are no longer emitted as outs (and drop out of `resource.py`'s materializable-model set, which is also correct). 2. `to_asset_outs()`'s dependency loop now routes external-model deps to the asset-dep path. This is required: an external dep resolves to a truthy `dep.model`, so with only change (1) it would land in `internal_asset_deps` without a matching out OR dep — which Dagster rejects ("specified as asset inputs, but not in internal_asset_deps"). Routing it through `create_asset_dep` keeps deps and internal_asset_deps consistent. Backward compatible: projects with no external models are unaffected (the sample project has none, so its existing to_asset_outs test is unchanged). Co-Authored-By: Claude --- dagster_sqlmesh/controller/base.py | 7 ++++++- dagster_sqlmesh/controller/dagster.py | 8 +++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/dagster_sqlmesh/controller/base.py b/dagster_sqlmesh/controller/base.py index b3291a0..890c4aa 100644 --- a/dagster_sqlmesh/controller/base.py +++ b/dagster_sqlmesh/controller/base.py @@ -367,7 +367,12 @@ def non_external_models_dag(self) -> t.Iterable[tuple[Model, set[str]]]: for model_fqn, deps in dag.graph.items(): logger.debug(f"model found: {model_fqn}") model = self.context.get_model(model_fqn) - if not model: + if not model or model.kind.is_external: + # External models are source tables the project READS, not models it + # materialises. They must never become multi-asset outputs (that produces + # a phantom asset and collides with any existing asset — e.g. a dlt asset — + # that already owns that key). They still surface as dependency edges via + # to_asset_outs' dep handling. continue yield (model, deps) diff --git a/dagster_sqlmesh/controller/dagster.py b/dagster_sqlmesh/controller/dagster.py index c0e0343..b6dc4f6 100644 --- a/dagster_sqlmesh/controller/dagster.py +++ b/dagster_sqlmesh/controller/dagster.py @@ -45,13 +45,19 @@ def to_asset_outs( asset_tags = translator.get_tags(context, model) for dep in model_deps: - if dep.model: + if dep.model and not dep.model.kind.is_external: dep_asset_key_str = translator.get_asset_key( context, dep.model.fqn ).to_user_string() internal_asset_deps.add(dep_asset_key_str) else: + # A dependency that is NOT a materialised model of this project — + # either an unmodelled table or an EXTERNAL model. Both are upstream + # asset DEPS, not outputs. (External models resolve to a truthy + # `dep.model`, so they must be routed here explicitly; otherwise they + # would be added to internal_asset_deps without a matching out or dep, + # which Dagster rejects once they are no longer emitted as outs.) table = translator.get_asset_key_str(dep.fqn) key = translator.get_asset_key( context, dep.fqn