Description
Cosmos currently produces the Asset/Dataset URI for a dbt model through three different code paths, one per execution-mode family, and they don't all agree on how the URI is built:
- WATCHER / WATCHER_KUBERNETES - the producer calls
compute_model_outlet_uris(target/manifest.json, namespace) (cosmos/operators/_watcher/base.py), which goes through construct_dataset_uri, and pushes the result via XCom; the consumer sensor emits Asset(uri) via register_dataset (cosmos/operators/watcher.py). Built from the manifest, no runtime OpenLineage involved.
- LOCAL / VIRTUALENV -
get_datasets() builds the URI via construct_dataset_uri(output.namespace, output.name), but only from openlineage_events_completes populated by parsing dbt's OpenLineage artifacts after the run (cosmos/operators/local.py), gated by emit_datasets. If openlineage-integration-common isn't installed, or artifact parsing raises one of several caught exceptions, emission silently drops to zero datasets with no warning.
- AIRFLOW_ASYNC (BigQuery) -
_register_event hardcodes f"bigquery://{gcp_project}/{dataset}/{table_name}" (cosmos/operators/_asynchronous/bigquery.py), bypassing construct_dataset_uri entirely, so it uses a different URI scheme than every other mode.
Because URI construction isn't centralized, nothing guarantees that a URI computed from the manifest at parse time will match what a given execution mode actually emits at runtime for that same model.
Use case/motivation
This surfaced while designing automatic cross-DbtDag/DbtTaskGroup dependency resolution (#1321): a consumer DAG needs to compute its external upstream's Asset URI at parse time and trust that it matches what the producer emits at runtime. That holds for WATCHER (both sides derive from the manifest), holds in practice but not by guarantee for LOCAL/VIRTUALENV (OpenLineage-dependent, can silently emit nothing), and never holds for AIRFLOW_ASYNC (divergent URI scheme).
Unifying emission onto one manifest/construct_dataset_uri-based path, mode by mode, would make dataset-based scheduling correctness independent of execution mode and unblock features like #1321 that need parse-time URIs to be trustworthy.
Proposed approach (sketch, open for discussion)
- Make relation identity independent of
LoadMode. Today, relation identity (database/schema/alias) is only available when a manifest is loaded - dbt ls, via LoadMode.DBT_LS, never requests it. Cosmos's dbt ls invocation passes an explicit --output-keys name unique_id resource_type depends_on original_file_path tags config freshness fqn (cosmos/dbt/graph.py) that omits database/schema/alias/relation_name, and DbtNode has no fields for them. Extending the requested keys and adding matching DbtNode fields (plus updating the dbt-ls parser and the manifest-graph parser) would let Cosmos compute a relation-based Asset URI regardless of LoadMode, not only when a manifest is present - the prerequisite for building one shared URI-computation path that every execution mode can just consume. Two things need validating before treating this as simple:
- Whether
database/schema/alias/relation_name are valid --output-keys values across every dbt-core version Cosmos supports and on dbt Fusion - Fusion's default dbt ls --output json already returns a narrower key set than dbt-core's, which is why Cosmos already special-cases it via settings.pre_dbt_fusion (cosmos/dbt/graph.py).
- The
dbt ls cache. dbt_ls_cache_key_args / _calculate_dbt_ls_cache_current_version (cosmos/dbt/graph.py, cosmos/cache.py) hash the project folder contents plus select/exclude/vars/selector/partial-parse plus a few explicitly-configured env/Airflow vars - nothing in that hash reflects the Cosmos version or which --output-keys were requested. Shipping this change as-is means a user who upgrades Cosmos keeps serving their pre-upgrade cached dbt ls output (missing the new fields) until an unrelated cache-bust event happens. This needs either folding the output-keys list into dbt_ls_cache_key_args, or a one-time cache-schema-version bump.
- Route
_asynchronous/bigquery.py's _register_event through construct_dataset_uri instead of a hardcoded scheme.
- For LOCAL/VIRTUALENV, evaluate making emission unconditional on OpenLineage where relation identity is already known at parse time, or at minimum surface silent zero-emission as a warning instead of swallowing it at debug level.
- Audit all emitting paths against the Airflow 2-vs-3 URI standard (
settings.use_dataset_airflow3_uri_standard) for consistency.
Related issues
Are you willing to submit a PR?
Description
Cosmos currently produces the Asset/Dataset URI for a dbt model through three different code paths, one per execution-mode family, and they don't all agree on how the URI is built:
compute_model_outlet_uris(target/manifest.json, namespace)(cosmos/operators/_watcher/base.py), which goes throughconstruct_dataset_uri, and pushes the result via XCom; the consumer sensor emitsAsset(uri)viaregister_dataset(cosmos/operators/watcher.py). Built from the manifest, no runtime OpenLineage involved.get_datasets()builds the URI viaconstruct_dataset_uri(output.namespace, output.name), but only fromopenlineage_events_completespopulated by parsing dbt's OpenLineage artifacts after the run (cosmos/operators/local.py), gated byemit_datasets. Ifopenlineage-integration-commonisn't installed, or artifact parsing raises one of several caught exceptions, emission silently drops to zero datasets with no warning._register_eventhardcodesf"bigquery://{gcp_project}/{dataset}/{table_name}"(cosmos/operators/_asynchronous/bigquery.py), bypassingconstruct_dataset_urientirely, so it uses a different URI scheme than every other mode.Because URI construction isn't centralized, nothing guarantees that a URI computed from the manifest at parse time will match what a given execution mode actually emits at runtime for that same model.
Use case/motivation
This surfaced while designing automatic cross-
DbtDag/DbtTaskGroupdependency resolution (#1321): a consumer DAG needs to compute its external upstream's Asset URI at parse time and trust that it matches what the producer emits at runtime. That holds for WATCHER (both sides derive from the manifest), holds in practice but not by guarantee for LOCAL/VIRTUALENV (OpenLineage-dependent, can silently emit nothing), and never holds for AIRFLOW_ASYNC (divergent URI scheme).Unifying emission onto one manifest/
construct_dataset_uri-based path, mode by mode, would make dataset-based scheduling correctness independent of execution mode and unblock features like #1321 that need parse-time URIs to be trustworthy.Proposed approach (sketch, open for discussion)
LoadMode. Today, relation identity (database/schema/alias) is only available when a manifest is loaded -dbt ls, viaLoadMode.DBT_LS, never requests it. Cosmos'sdbt lsinvocation passes an explicit--output-keys name unique_id resource_type depends_on original_file_path tags config freshness fqn(cosmos/dbt/graph.py) that omitsdatabase/schema/alias/relation_name, andDbtNodehas no fields for them. Extending the requested keys and adding matchingDbtNodefields (plus updating the dbt-ls parser and the manifest-graph parser) would let Cosmos compute a relation-based Asset URI regardless ofLoadMode, not only when a manifest is present - the prerequisite for building one shared URI-computation path that every execution mode can just consume. Two things need validating before treating this as simple:database/schema/alias/relation_nameare valid--output-keysvalues across every dbt-core version Cosmos supports and on dbt Fusion - Fusion's defaultdbt ls --output jsonalready returns a narrower key set than dbt-core's, which is why Cosmos already special-cases it viasettings.pre_dbt_fusion(cosmos/dbt/graph.py).dbt lscache.dbt_ls_cache_key_args/_calculate_dbt_ls_cache_current_version(cosmos/dbt/graph.py,cosmos/cache.py) hash the project folder contents plusselect/exclude/vars/selector/partial-parse plus a few explicitly-configured env/Airflow vars - nothing in that hash reflects the Cosmos version or which--output-keyswere requested. Shipping this change as-is means a user who upgrades Cosmos keeps serving their pre-upgrade cacheddbt lsoutput (missing the new fields) until an unrelated cache-bust event happens. This needs either folding the output-keys list intodbt_ls_cache_key_args, or a one-time cache-schema-version bump._asynchronous/bigquery.py's_register_eventthroughconstruct_dataset_uriinstead of a hardcoded scheme.settings.use_dataset_airflow3_uri_standard) for consistency.Related issues
DbtDags that represent sub-parts of a dbt project #1321 - surfaced this gap during designExecutionMode.KUBERNETES#2329 -ExecutionMode.KUBERNETESdoesn't emit assets at all (adjacent gap, different mode)Are you willing to submit a PR?