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