propagate meta/tags on columns to top-level and config (#11992)

Author: MichelleArk

.changes/unreleased/Fixes-20250910-163905.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Propagate column meta/tags from config to tests
+time: 2025-09-10T16:39:05.129379-04:00
+custom:
+    Author: michelleark
+    Issue: "11984"

core/dbt/parser/common.py

@@ -222,19 +222,30 @@ def _add(self, column: HasColumnProps) -> None:
         ):
             raise ParsingError(f"Invalid constraint type on column {column.name}")
 
+        # Merge meta and tags from column and config
+        column_config_meta = (
+            column.config["meta"] if isinstance(column.config.get("meta"), dict) else {}
+        )
+        column_config_tags = []
+        if "tags" in column.config:
+            if isinstance(column.config["tags"], list):
+                column_config_tags = column.config["tags"]
+            elif isinstance(column.config["tags"], str):
+                column_config_tags = [column.config["tags"]]
+
+        column_meta = {**column.meta, **column_config_meta}
+        column_tags = list(set(tags + column_config_tags))
         self.column_info[column.name] = ColumnInfo(
             name=column.name,
             description=column.description,
             data_type=column.data_type,
             constraints=[ColumnLevelConstraint.from_dict(c) for c in column.constraints],
-            meta=column.meta,
-            tags=tags,
+            meta=column_meta,
+            tags=column_tags,
             quote=quote,
             _extra=column.extra,
             granularity=granularity,
-            config=ColumnConfig(
-                meta=column.config.get("meta", {}), tags=column.config.get("tags", [])
-            ),
+            config=ColumnConfig(meta=column_meta, tags=column_tags),
         )
 
     @classmethod

core/dbt/parser/schema_generic_tests.py

@@ -383,7 +383,11 @@ def parse_test(
             should_quote = column.quote or (column.quote is None and target_block.quote_columns)
             if should_quote:
                 column_name = get_adapter(self.root_project).quote(column_name)
-            column_tags = column.tags
+
+            column_config_tags = column.config.get("tags", [])
+            if isinstance(column_config_tags, str):
+                column_config_tags = [column_config_tags]
+            column_tags = list(set(column.tags + column_config_tags))
 
         block = GenericTestBlock.from_test_block(
             src=target_block,

core/dbt/parser/sources.py

@@ -262,6 +262,11 @@ def parse_source_test(
         tags_sources = [target.source.tags, target.table.tags]
         if column is not None:
             tags_sources.append(column.tags)
+            if column_config_tags := column.config.get("tags", []):
+                if isinstance(column_config_tags, list):
+                    tags_sources.append(column_config_tags)
+                elif isinstance(column_config_tags, str):
+                    tags_sources.append([column_config_tags])
         tags = list(itertools.chain.from_iterable(tags_sources))
 
         generic_test_parser = self.get_generic_test_parser_for(target.package_name)