merge source meta and table meta on SourceDefinition.meta

.changes/unreleased/Features-20240315-161209.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: SourceDefinition.meta represents source-level and table-level meta properties,
+  instead of only table-level
+time: 2024-03-15T16:12:09.789935-04:00
+custom:
+  Author: michelleark
+  Issue: "9766"

core/dbt/parser/sources.py

@@ -132,14 +132,15 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
         refs = ParserRef.from_target(table)
         unique_id = target.unique_id
         description = table.description or ""
-        meta = table.meta or {}
         source_description = source.description or ""
         loaded_at_field = table.loaded_at_field or source.loaded_at_field
 
         freshness = merge_freshness(source.freshness, table.freshness)
         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)))

tests/unit/test_parser.py

@@ -225,6 +225,23 @@ def assertEqualNodes(node_one, node_two):
         - name: my_table
 """
 
+
+MULTIPLE_TABLE_SOURCE_META = """
+sources:
+    - name: my_source
+      meta:
+        source_field: source_value
+        shared_field: shared_field_default
+      tables:
+        - name: my_table_shared_field_default
+          meta:
+            table_field: table_value
+        - name: my_table_shared_field_override
+          meta:
+            shared_field: shared_field_table_override
+            table_field: table_value
+"""
+
 SINGLE_TABLE_SOURCE_TESTS = """
 sources:
     - name: my_source
@@ -416,6 +433,41 @@ def test__parse_basic_source(self):
         assert src.resource_type == NodeType.Source
         assert src.fqn == ["snowplow", "my_source", "my_table"]
 
+    @mock.patch("dbt.parser.sources.get_adapter")
+    def test__parse_basic_source_meta(self, mock_get_adapter):
+        block = self.file_block_for(MULTIPLE_TABLE_SOURCE_META, "test_one.yml")
+        dct = yaml_from_file(block.file)
+        self.parser.parse_file(block, dct)
+        self.assert_has_manifest_lengths(self.parser.manifest, sources=2)
+
+        unpatched_src_default = self.parser.manifest.sources[
+            "source.snowplow.my_source.my_table_shared_field_default"
+        ]
+        src_default = self.source_patcher.parse_source(unpatched_src_default)
+        assert src_default.meta == {
+            "source_field": "source_value",
+            "shared_field": "shared_field_default",
+            "table_field": "table_value",
+        }
+        assert src_default.source_meta == {
+            "source_field": "source_value",
+            "shared_field": "shared_field_default",
+        }
+
+        unpatched_src_override = self.parser.manifest.sources[
+            "source.snowplow.my_source.my_table_shared_field_override"
+        ]
+        src_override = self.source_patcher.parse_source(unpatched_src_override)
+        assert src_override.meta == {
+            "source_field": "source_value",
+            "shared_field": "shared_field_table_override",
+            "table_field": "table_value",
+        }
+        assert src_override.source_meta == {
+            "source_field": "source_value",
+            "shared_field": "shared_field_default",
+        }
+
     def test__read_basic_source_tests(self):
         block = self.yaml_block_for(SINGLE_TABLE_SOURCE_TESTS, "test_one.yml")
         analysis_tests = AnalysisPatchParser(self.parser, block, "analyses").parse().test_blocks