add unrendered_schema to source definition

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

@@ -72,3 +72,4 @@ class SourceDefinition(ParsedSourceMandatory):
     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

@@ -214,6 +214,7 @@ class SchemaSourceFile(BaseSourceFile):
     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)
     pp_dict: Optional[Dict[str, Any]] = None
     pp_test_index: Optional[Dict[str, Any]] = None
 

core/dbt/contracts/graph/nodes.py

@@ -1226,15 +1226,12 @@ def same_database_representation(self, other: "SourceDefinition") -> bool:
         # 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 self.schema == other.schema
-            and self.identifier == other.identifier
-            and True
-        )
+        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

@@ -346,6 +346,7 @@ class UnparsedSourceDefinition(dbtClassMixin):
     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

@@ -397,6 +397,7 @@ def get_key_dicts(self) -> Iterable[Dict[str, Any]]:
 
             # 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
@@ -420,6 +421,13 @@ def get_key_dicts(self) -> Iterable[Dict[str, Any]]:
                     else schema_file.unrendered_databases[self.key]
                 )
                 schema_file.unrendered_databases[self.key][entry["name"]] = unrendered_database
+            if unrendered_schema:
+                schema_file.unrendered_schemas[self.key] = (
+                    {}
+                    if self.key not in schema_file.unrendered_schemas
+                    else schema_file.unrendered_schemas[self.key]
+                )
+                schema_file.unrendered_schemas[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)
@@ -487,6 +495,12 @@ def parse(self) -> ParseResult:
                         "sources", {}
                     ).get(source.name)
                     source.unrendered_database = unrendered_database
+
+                    unrendered_schema = self.yaml.file.unrendered_schemas.get("sources", {}).get(
+                        source.name
+                    )
+                    source.unrendered_schema = unrendered_schema
+
                 self.add_source_definitions(source)
         return ParseResult()
 

tests/functional/defer_state/fixtures.py

@@ -560,14 +560,23 @@
 
 """
 
-schema_source_with_env_var_as_property_yml = """
+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
+"""
+
 
 model_with_var_in_config_sql = """
 {{ config(materialized=var('DBT_TEST_STATE_MODIFIED')) }}

tests/functional/defer_state/test_modified_state_environment_vars.py

@@ -7,7 +7,8 @@
     model_with_env_var_in_config_sql,
     model_with_no_in_config_sql,
     schema_model_with_env_var_in_config_yml,
-    schema_source_with_env_var_as_property_yml,
+    schema_source_with_env_var_as_database_property_yml,
+    schema_source_with_env_var_as_schema_property_yml,
 )
 from tests.functional.defer_state.test_modified_state import BaseModifiedState
 
@@ -109,9 +110,17 @@ def models(self):
         }
 
 
+class TestSourceNodeWithEnvVarConfigInDatabase(BaseTestStateSelectionEnvVarConfig):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": schema_source_with_env_var_as_database_property_yml,
+        }
+
+
 class TestSourceNodeWithEnvVarConfigInSchema(BaseTestStateSelectionEnvVarConfig):
     @pytest.fixture(scope="class")
     def models(self):
         return {
-            "schema.yml": schema_source_with_env_var_as_property_yml,
+            "schema.yml": schema_source_with_env_var_as_schema_property_yml,
         }