config.meta and config.tags propagation to top-level on sources and tables (#11839)

Author: MichelleArk

.changes/unreleased/Fixes-20250716-164535.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Propagate config.meta and config.tags to top-level on source nodes
+time: 2025-07-16T16:45:35.683199-04:00
+custom:
+    Author: michelleark
+    Issue: "11839"

core/dbt/artifacts/resources/v1/source_definition.py

@@ -10,7 +10,7 @@
     HasRelationMetadata,
     Quoting,
 )
-from dbt.artifacts.resources.v1.config import BaseConfig
+from dbt.artifacts.resources.v1.config import BaseConfig, MergeBehavior
 from dbt_common.contracts.config.properties import AdditionalPropertiesAllowed
 from dbt_common.contracts.util import Mergeable
 from dbt_common.exceptions import CompilationError
@@ -23,6 +23,8 @@ class SourceConfig(BaseConfig):
     freshness: Optional[FreshnessThreshold] = field(default_factory=FreshnessThreshold)
     loaded_at_field: Optional[str] = None
     loaded_at_query: Optional[str] = None
+    meta: Dict[str, Any] = field(default_factory=dict, metadata=MergeBehavior.Update.meta())
+    tags: List[str] = field(default_factory=list)
 
 
 @dataclass

core/dbt/parser/sources.py

@@ -132,13 +132,8 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
         source_description = source.description or ""
 
         quoting = source.quoting.merged(table.quoting)
-        # path = block.path.original_file_path
-        table_meta = table.meta or {}
-        source_meta = source.meta or {}
-        meta = {**source_meta, **table_meta}
-
-        # make sure we don't do duplicate tags from source + table
-        tags = sorted(set(itertools.chain(source.tags, table.tags)))
+        # Retain original source meta prior to merge with table meta
+        source_meta = {**source.meta, **source.config.get("meta", {})}
 
         config = self._generate_source_config(
             target=target,
@@ -176,15 +171,15 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             source_name=source.name,
             source_description=source_description,
             source_meta=source_meta,
-            meta=meta,
+            meta=config.meta,
             loader=source.loader,
             loaded_at_field=config.loaded_at_field,
             loaded_at_query=config.loaded_at_query,
             freshness=config.freshness,
             quoting=quoting,
             resource_type=NodeType.Source,
             fqn=target.fqn,
-            tags=tags,
+            tags=config.tags,
             config=config,
             unrendered_config=unrendered_config,
         )
@@ -313,6 +308,13 @@ def _generate_source_config(self, target: UnpatchedSourceDefinition, rendered: b
         precedence_configs["loaded_at_field"] = precedence_loaded_at_field
         precedence_configs["loaded_at_query"] = precedence_loaded_at_query
 
+        # Handle merges across source, table, and config for meta and tags
+        precedence_meta = self.calculate_meta_from_raw_target(target)
+        precedence_configs["meta"] = precedence_meta
+
+        precedence_tags = self.calculate_tags_from_raw_target(target)
+        precedence_configs["tags"] = precedence_tags
+
         # Because freshness is a "object" config, the freshness from the dbt_project.yml and the freshness
         # from the schema file _won't_ get merged by this process. The result will be that the freshness will
         # come from the schema file if provided, and if not, it'll fall back to the dbt_project.yml freshness.
@@ -472,6 +474,34 @@ def calculate_loaded_at_field_query_from_raw_target(
 
         return loaded_at_field, loaded_at_query
 
+    def calculate_meta_from_raw_target(self, target: UnpatchedSourceDefinition) -> Dict[str, Any]:
+        source_meta = target.source.meta or {}
+        source_config_meta = target.source.config.get("meta", {})
+        source_config_meta = source_config_meta if isinstance(source_config_meta, dict) else {}
+
+        table_meta = target.table.meta or {}
+        table_config_meta = target.table.config.get("meta", {})
+        table_config_meta = table_config_meta if isinstance(table_config_meta, dict) else {}
+
+        return {**source_meta, **source_config_meta, **table_meta, **table_config_meta}
+
+    def calculate_tags_from_raw_target(self, target: UnpatchedSourceDefinition) -> List[str]:
+        source_tags = target.source.tags or []
+        source_config_tags = target.source.config.get("tags", [])
+        source_config_tags = (
+            source_config_tags if isinstance(source_config_tags, list) else [source_config_tags]
+        )
+
+        table_tags = target.table.tags or []
+        table_config_tags = target.table.config.get("tags", [])
+        table_config_tags = (
+            table_config_tags if isinstance(table_config_tags, list) else [table_config_tags]
+        )
+
+        return sorted(
+            set(itertools.chain(source_tags, source_config_tags, table_tags, table_config_tags))
+        )
+
 
 def merge_freshness_time_thresholds(
     base: Optional[Time], update: Optional[Time]

tests/functional/artifacts/expected_manifest.py

@@ -823,6 +823,8 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                     },
                     "loaded_at_query": None,
                     "loaded_at_field": None,
+                    "meta": {},
+                    "tags": [],
                 },
                 "quoting": {
                     "database": None,
@@ -859,7 +861,12 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "tags": [],
                 "unique_id": "source.test.my_source.my_table",
                 "fqn": ["test", "my_source", "my_table"],
-                "unrendered_config": {"loaded_at_field": None, "loaded_at_query": None},
+                "unrendered_config": {
+                    "loaded_at_field": None,
+                    "loaded_at_query": None,
+                    "meta": {},
+                    "tags": [],
+                },
                 "unrendered_database": None,
                 "unrendered_schema": "{{ var('test_schema') }}",
                 "doc_blocks": [],
@@ -1366,6 +1373,8 @@ def expected_references_manifest(project):
                     },
                     "loaded_at_field": None,
                     "loaded_at_query": None,
+                    "meta": {},
+                    "tags": [],
                 },
                 "quoting": {
                     "database": False,
@@ -1401,7 +1410,12 @@ def expected_references_manifest(project):
                 "tags": [],
                 "unique_id": "source.test.my_source.my_table",
                 "fqn": ["test", "my_source", "my_table"],
-                "unrendered_config": {"loaded_at_field": None, "loaded_at_query": None},
+                "unrendered_config": {
+                    "loaded_at_field": None,
+                    "loaded_at_query": None,
+                    "meta": {},
+                    "tags": [],
+                },
                 "unrendered_database": None,
                 "unrendered_schema": "{{ var('test_schema') }}",
                 "doc_blocks": ["doc.test.table_info"],

tests/functional/fixtures/happy_path_project/models/schema.yml

@@ -141,7 +141,8 @@ sources:
       event_time: column_name
       loaded_at_field: column_name
       meta:
-        test: 1
+        source_meta: 1
+      tags: ["source_tag"]
       freshness:
         warn_after:
           count: 1
@@ -187,14 +188,14 @@ sources:
             config:
               meta:
                 test: 1
-                tags: ["tags"]
+              tags: ["column_tag"]
         config:
           enabled: true
           event_time: column_name
           loaded_at_field: column_name
           meta:
-            test: 1
-          tags: ["tag"]
+            table_meta: 1
+          tags: ["table_tag"]
           freshness:
             warn_after:
               count: 1

tests/functional/list/test_list.py

@@ -642,8 +642,8 @@ def expect_source_output(self):
                         },
                         "filter": "column_name = 1",
                     },
-                    "meta": {"test": 1},
-                    "tags": ["tag"],
+                    "meta": {"source_meta": 1, "table_meta": 1},
+                    "tags": ["source_tag", "table_tag"],
                     "loaded_at_query": None,
                     "loaded_at_field": "column_name",
                 },
@@ -653,7 +653,7 @@ def expect_source_output(self):
                 "name": "my_table",
                 "source_name": "my_source",
                 "resource_type": "source",
-                "tags": [],
+                "tags": ["source_tag", "table_tag"],
             },
             "path": self.dir("models/schema.yml"),
         }

tests/unit/contracts/graph/test_nodes_parsed.py

@@ -1904,6 +1904,8 @@ def basic_parsed_source_definition_dict():
                 "warn_after": {},
                 "error_after": {},
             },
+            "tags": [],
+            "meta": {},
         },
         "unrendered_config": {},
         "doc_blocks": [],
@@ -1939,6 +1941,8 @@ def complex_parsed_source_definition_dict():
                 "warn_after": {},
                 "error_after": {},
             },
+            "tags": [],
+            "meta": {},
         },
         "freshness": {"warn_after": {"period": "hour", "count": 1}, "error_after": {}},
         "loaded_at_field": "loaded_at",