[state:modified] store unrendered_database and unrendered_schema on s…

…ource definition for state:modified comparisons (#10675)

.changes/unreleased/Features-20240926-140206.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: Use unrendered database and schema source properties during state:modified,
+  behind state_modified_compare_more_unrendered_values behavoiur flag
+time: 2024-09-26T14:02:06.732068+01:00
+custom:
+  Author: michelleark
+  Issue: "9573"

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

@@ -72,3 +72,5 @@ class SourceDefinition(ParsedSourceMandatory):
     vars: Dict[str, Any] = field(default_factory=dict)
     relation_name: Optional[str] = None
     created_at: float = field(default_factory=lambda: time.time())
+    unrendered_database: Optional[str] = None
+    unrendered_schema: Optional[str] = None

core/dbt/contracts/files.py

@@ -213,6 +213,8 @@ class SchemaSourceFile(BaseSourceFile):
     sop: List[SourceKey] = field(default_factory=list)
     env_vars: Dict[str, Any] = field(default_factory=dict)
     unrendered_configs: Dict[str, Any] = field(default_factory=dict)
+    unrendered_databases: Dict[str, Any] = field(default_factory=dict)
+    unrendered_schemas: Dict[str, Any] = field(default_factory=dict)
     vars: Dict[str, Any] = field(default_factory=dict)
     pp_dict: Optional[Dict[str, Any]] = None
     pp_test_index: Optional[Dict[str, Any]] = None
@@ -386,6 +388,30 @@ def delete_from_env_vars(self, yaml_key, name):
             if not self.env_vars[yaml_key]:
                 del self.env_vars[yaml_key]
 
+    def add_unrendered_database(self, yaml_key: str, name: str, unrendered_database: str) -> None:
+        if yaml_key not in self.unrendered_databases:
+            self.unrendered_databases[yaml_key] = {}
+
+        self.unrendered_databases[yaml_key][name] = unrendered_database
+
+    def get_unrendered_database(self, yaml_key: str, name: str) -> Optional[str]:
+        if yaml_key not in self.unrendered_databases:
+            return None
+
+        return self.unrendered_databases[yaml_key].get(name)
+
+    def add_unrendered_schema(self, yaml_key: str, name: str, unrendered_schema: str) -> None:
+        if yaml_key not in self.unrendered_schemas:
+            self.unrendered_schemas[yaml_key] = {}
+
+        self.unrendered_schemas[yaml_key][name] = unrendered_schema
+
+    def get_unrendered_schema(self, yaml_key: str, name: str) -> Optional[str]:
+        if yaml_key not in self.unrendered_schemas:
+            return None
+
+        return self.unrendered_schemas[yaml_key].get(name)
+
 
 @dataclass
 class FixtureSourceFile(BaseSourceFile):

core/dbt/contracts/graph/nodes.py

@@ -1236,12 +1236,16 @@ def resource_class(cls) -> Type[SourceDefinitionResource]:
         return SourceDefinitionResource
 
     def same_database_representation(self, other: "SourceDefinition") -> bool:
-        return (
-            self.database == other.database
-            and self.schema == other.schema
-            and self.identifier == other.identifier
-            and True
-        )
+
+        # preserve legacy behaviour -- use potentially rendered database
+        if get_flags().state_modified_compare_more_unrendered_values is False:
+            same_database = self.database == other.database
+            same_schema = self.schema == other.schema
+        else:
+            same_database = self.unrendered_database == other.unrendered_database
+            same_schema = self.unrendered_schema == other.unrendered_schema
+
+        return same_database and same_schema and self.identifier == other.identifier and True
 
     def same_quoting(self, other: "SourceDefinition") -> bool:
         return self.quoting == other.quoting

core/dbt/contracts/graph/unparsed.py

@@ -340,6 +340,8 @@ class UnparsedSourceDefinition(dbtClassMixin):
     tables: List[UnparsedSourceTableDefinition] = field(default_factory=list)
     tags: List[str] = field(default_factory=list)
     config: Dict[str, Any] = field(default_factory=dict)
+    unrendered_database: Optional[str] = None
+    unrendered_schema: Optional[str] = None
 
     @classmethod
     def validate(cls, data):

core/dbt/parser/schemas.py

@@ -386,6 +386,10 @@ def get_key_dicts(self) -> Iterable[Dict[str, Any]]:
                     if "v" in version:
                         unrendered_version_configs[version["v"]] = version.get("config", {})
 
+            # For sources
+            unrendered_database = entry.get("database", None)
+            unrendered_schema = entry.get("schema", None)
+
             # Render the data (except for tests, data_tests and descriptions).
             # See the SchemaYamlRenderer
             entry = self.render_entry(entry)
@@ -401,6 +405,11 @@ def get_key_dicts(self) -> Iterable[Dict[str, Any]]:
                     unrendered_version_config, self.key, entry["name"], version
                 )
 
+            if unrendered_database:
+                schema_file.add_unrendered_database(self.key, entry["name"], unrendered_database)
+            if unrendered_schema:
+                schema_file.add_unrendered_schema(self.key, entry["name"], unrendered_schema)
+
             if self.schema_yaml_vars.env_vars:
                 self.schema_parser.manifest.env_vars.update(self.schema_yaml_vars.env_vars)
                 for env_var in self.schema_yaml_vars.env_vars.keys():
@@ -465,6 +474,15 @@ def parse(self) -> ParseResult:
                 source_file.source_patches.append(key)
             else:
                 source = self._target_from_dict(UnparsedSourceDefinition, data)
+                # Store unrendered_database and unrendered_schema for state:modified comparisons
+                if isinstance(self.yaml.file, SchemaSourceFile):
+                    source.unrendered_database = self.yaml.file.get_unrendered_database(
+                        "sources", source.name
+                    )
+                    source.unrendered_schema = self.yaml.file.get_unrendered_schema(
+                        "sources", source.name
+                    )
+
                 self.add_source_definitions(source)
         return ParseResult()
 

core/dbt/parser/sources.py

@@ -173,7 +173,9 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
         parsed_source = SourceDefinition(
             package_name=target.package_name,
             database=(source.database or default_database),
+            unrendered_database=source.unrendered_database,
             schema=(source.schema or source.name),
+            unrendered_schema=source.unrendered_schema,
             identifier=(table.identifier or table.name),
             path=target.path,
             original_file_path=target.original_file_path,

schemas/dbt/manifest/v12.json

@@ -8198,6 +8198,28 @@
           },
           "created_at": {
             "type": "number"
+          },
+          "unrendered_database": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null
+          },
+          "unrendered_schema": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null
           }
         },
         "additionalProperties": false,
@@ -18101,6 +18123,28 @@
                     },
                     "created_at": {
                       "type": "number"
+                    },
+                    "unrendered_database": {
+                      "anyOf": [
+                        {
+                          "type": "string"
+                        },
+                        {
+                          "type": "null"
+                        }
+                      ],
+                      "default": null
+                    },
+                    "unrendered_schema": {
+                      "anyOf": [
+                        {
+                          "type": "string"
+                        },
+                        {
+                          "type": "null"
+                        }
+                      ],
+                      "default": null
                     }
                   },
                   "additionalProperties": false,

tests/functional/artifacts/expected_manifest.py

@@ -814,6 +814,8 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "unique_id": "source.test.my_source.my_table",
                 "fqn": ["test", "my_source", "my_table"],
                 "unrendered_config": {},
+                "unrendered_database": None,
+                "unrendered_schema": "{{ var('test_schema') }}",
                 "vars": {"test_schema": ANY},
             },
         },
@@ -1327,6 +1329,8 @@ def expected_references_manifest(project):
                 "unique_id": "source.test.my_source.my_table",
                 "fqn": ["test", "my_source", "my_table"],
                 "unrendered_config": {},
+                "unrendered_database": None,
+                "unrendered_schema": "{{ var('test_schema') }}",
                 "vars": {"test_schema": ANY},
             },
         },

tests/functional/defer_state/fixtures.py

@@ -560,14 +560,6 @@
 
 """
 
-schema_source_with_env_var_as_property_yml = """
-sources:
-  - name: jaffle_shop
-    database: "{{ env_var('DBT_TEST_STATE_MODIFIED') }}"
-    tables:
-      - name: customers
-"""
-
 
 model_with_var_in_config_sql = """
 {{ config(materialized=var('DBT_TEST_STATE_MODIFIED')) }}
@@ -603,3 +595,63 @@
     config:
       materialized: "{{ ('view' if execute else 'table') }}"
 """
+
+schema_source_with_env_var_as_database_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "{{ env_var('DBT_TEST_STATE_MODIFIED') }}"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_env_var_as_schema_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "test"
+    schema: "{{ env_var('DBT_TEST_STATE_MODIFIED') }}"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_updated_env_var_as_schema_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "test"
+    schema: "updated"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_jinja_as_database_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "{{ ('foo' if execute else 'bar') }}"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_updated_jinja_as_database_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "{{ ('bar' if execute else 'foo') }}"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_jinja_as_schema_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "test"
+    schema: "{{ ('foo' if execute else 'bar') }}"
+    tables:
+      - name: customers
+"""
+
+schema_source_with_updated_jinja_as_schema_property_yml = """
+sources:
+  - name: jaffle_shop
+    database: "test"
+    schema: "{{ ('bar' if execute else 'foo') }}"
+    tables:
+      - name: customers
+"""

tests/functional/defer_state/test_modified_state_jinja.py

@@ -29,8 +29,7 @@ def project_config_update(self):
     def update_jinja_expression_in_config(self, project):
         pass
 
-    def test_change_target_jinja_if(self, project, dbt_profile_data, profiles_root):
-        # Generate ./state without changing target.name
+    def test_change_jinja_if(self, project):
         run_dbt(["run"])
         self.copy_state()
         # Model is table when execute = True