Repository: dbt-labs/dbt-core
Pull request: Ensure that target_schema from snapshot config is promoted to node level
resolves #6745 

### Problem

Snapshot target_schema and target_database were being processed when the snapshot nodes were processed, but those config fields were not being promoted  to the node level schema and database if set in config in dbt_proejct.yml or a schema yaml file.

### Solution

Added code to the "update_parsed_node_relation_names" to promote target_schema and target_database for snapshot nodes.

### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me  
- [x] I have run this code in development and it appears to resolve the stated issue  
- [x] This PR includes tests, or tests are not required/relevant for this PR
- [x] This PR has no interface changes (e.g. macros, cli, logs, json artifacts, config files, adapter interface, etc) or this PR has already received feedback and approval from Product or DX


Problem statement:
[CT-1913] Snapshot target_schema & target_database used from dbt_project.yml
### Is this a new bug in dbt-core?

- [X] I believe this is a new bug in dbt-core
- [X] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

Profiles.yml
```
    dev:
      type: bigquery
      method: oauth
      project: gcp-project-test
      **dataset: holding**
      location: EU
```

snapshot config in dbt_project.yml 

```
snapshots:
    test_dbt_project
        +persist_docs:
            relation: true
            columns: true
        primary:
            snapshot_models:
                **+target_schema: snapshot_dataset**
```

schema.yml to add the descriptions to snapshots tables

```
snapshots:
    - name: customers
      config:
        enabled: true
      columns:
        - name: customer_id
           description: 'customer unique id'
```

snapshot folder structure:
```
-snapshots
  -primary
      -snapshot_models
         -customers.sql
         -orders.sql
         - schema.yml
```

When we add the schema.yml file under snapshots/primary/snapshots_models and run the below command

`dbt snapshot --select customers` then the model is getting created under holding dataset(profile.yml) instead of snapshot_dataset(dbt_project.yml)



-


### Expected Behavior

The snapshots should be created in the dataset defined in dbt_project.yml file. For now have removed the schema.yml file so that it can create the model's under dataset provided in dbt_project.yml file. This is stopping to add the descriptions to table columns

### Steps To Reproduce

details provide in descriptions

### Relevant log output

_No response_

### Environment

```markdown
- OS: Airflow 2.2 composer 2.2
- Python:3.8
- dbt: dbt bigquery 1.2.0
```


### Which database adapter are you using with dbt?

bigquery

### Additional Context

_No response_

Hints from issue discussion:
Thank you for reaching out about this @minhajpasha !

p.s. I added some ` ``` ` to your description to help format each or your YAML examples -- hope you don't mind.

I was able to reproduce what you described, and I agree this is a bug. It _might_ be related to https://github.com/dbt-labs/dbt-core/issues/4000

### Workaround

Removing your `snapshots/primary/snapshot_models/schema.yml` file altogether is an option!

But if you want to keep that file (for its column descriptions, etc), then you can use the following workaround.

Remove each `config` section from your `schema.yaml`:
```yaml
version: 2

snapshots:
  - name: customers
    columns:
      - name: customer_id
        description: 'customer unique id'
```

And add your specific `config` assignments within each snapshot instead:
```
{% snapshot customers %}

{{
    config(
      enabled=true,
      ...
    )
}}

...
{% endsnapshot %}
``` 

### How to reproduce
Here's the simplified setup that I used to reproduce the error using dbt-snowflake:

`dbt_project.yml`
```yaml
snapshots:
  +target_schema: dbeatty_snapshot_dataset
```

`snapshots/customers.sql`
```sql
{% snapshot customers %}

{{
    config(
      enabled=true,
      unique_key='id',
      strategy='check',
      check_cols='all',
    )
}}

select 0 as id 

{% endsnapshot %}
```

This will not work as intended (since it won't use configured the snapshot `target_schema` and will use the schema from profiles.yml instead):
`snapshots/schema.yml`
```yaml
version: 2

snapshots:
  - name: customers
    # Adding a config section here will break the snapshot `target_schema`!
    config:
      enabled: true
```

#### Run it
Run it and observe the (unintended!) schema it uses for snapshots:
```shell
dbt snapshot --select customers
```

But removing the `config` key will use the intended target schema for snapshots:
`snapshots/schema.yml`
```yaml
version: 2

snapshots:
  - name: customers
```

#### Run it again
Run it again and observe the intended schema it uses for snapshots:
```shell
dbt snapshot --select customers
```
@dbeatty10 - I hope everything is going well for you. Thanks for providing the workaround. Will try and update you 
This bug also applies when defining a "meta" property in the snapshot yml (so not just config).
I forked jaffle_shop and made some changes to illustrate the issue, but the example is identical to the steps to reproduce as given by @dbeatty10 

https://github.com/honeyp0t/jaffle_shop
@honeyp0t -- thanks for explaining that this extends beyond `config` to `meta` also and showing within that marvelous repo 🏆 
I've been pulling my hair over an issue with setting the snapshot database as well. Is this related?


dbt_project.yml
```
snapshots:
  +target_database: "{{ 'raw' if target.name == 'prod' else target.database }}"
```
snapshot_name.sql (with simply hard coded database)
```
{% snapshot snapshot_name %}

    {{
        config(
            target_database='raw',
```
My dbt snapshot runs would ignore these settings and use the database from profiles.yml, just like the schema issue noted above.
After many hours of hair pulling and probing everywhere, this is what I found.


dbt_project.yml
```
snapshots:
  +database: 'raw'
```
This works.

snapshot_name.sql
```
{% snapshot snapshot_name %}

    {{
        config(
            database='raw',
```
This also works.

I upgraded from v1.0 to v1.3 and I believe it happened between these versions.

@dbeatty10 Thanks for the great reproduction case above!

One thing I noticed is that the `target_schema` config is appropriately resolved, _however_, the node-level `schema` attribute is overwritten with `target.schema`. We have some special logic, just for snapshots, to always overwrite `node.schema` with `node.config.target_schema`, in [`transform`](https://github.com/dbt-labs/dbt-core/blob/4c63b630de1e9cf3c19d48dcb3ca728458ace5f6/core/dbt/parser/snapshots.py#L58-L69) &rarr; [`set_snapshot_attributes`](https://github.com/dbt-labs/dbt-core/blob/4c63b630de1e9cf3c19d48dcb3ca728458ace5f6/core/dbt/parser/snapshots.py#L28-L44).

https://github.com/dbt-labs/dbt-core/blob/4c63b630de1e9cf3c19d48dcb3ca728458ace5f6/core/dbt/parser/snapshots.py#L38-L40

That logic all happens correctly, _but then_ we call [`patch_node_config`](https://github.com/dbt-labs/dbt-core/blob/4c63b630de1e9cf3c19d48dcb3ca728458ace5f6/core/dbt/parser/schemas.py#L810-L821) &rarr; [`update_parsed_node_config`](https://github.com/dbt-labs/dbt-core/blob/4c63b630de1e9cf3c19d48dcb3ca728458ace5f6/core/dbt/parser/base.py#L264-L336) _again_, re-resolve configs _again_, and call `update_parsed_node_relation_names`, which re-renders all our `generate_x_name` macros, and re-replaces the snapshot's `node.schema` with the returned value based on `node.config.schema`.

So, the idea here is: We need to make sure that, for snapshots, `set_snapshot_attributes` always gets called _after_ `patch_node_config`. I think that's enough to go on for now; queuing this one up for estimation.
We also just hit this bug as we're adding `meta` information to our models and snapshots for tagging personal data. Running dbt 1.3.2 at the moment. Would love to see a bug fix in dbt 1.4/1.5 🤞 🙂 
Per #7946: This is apparently a regression from v1.2 &rarr; v1.3. At this point, we'd be most likely to fix for the next minor version, and (if the fix is precise enough) backport for inclusion in the previous version's next patch

Commit message: Update test

Patch to review:
diff --git a/.changes/unreleased/Fixes-20230717-160652.yaml b/.changes/unreleased/Fixes-20230717-160652.yaml
new file mode 100644
index 00000000000..8c63a584db9
--- /dev/null
+++ b/.changes/unreleased/Fixes-20230717-160652.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Copy target_schema from config into snapshot node
+time: 2023-07-17T16:06:52.957724-04:00
+custom:
+  Author: gshank
+  Issue: "6745"
diff --git a/core/dbt/parser/base.py b/core/dbt/parser/base.py
index 2e8a4663a3e..0624d519f10 100644
--- a/core/dbt/parser/base.py
+++ b/core/dbt/parser/base.py
@@ -102,8 +102,7 @@ def __init__(self, config: RuntimeConfig, manifest: Manifest, component: str) ->
         self.package_updaters = package_updaters
         self.component = component
 
-    def __call__(self, parsed_node: Any, config_dict: Dict[str, Any]) -> None:
-        override = config_dict.get(self.component)
+    def __call__(self, parsed_node: Any, override: Optional[str]) -> None:
         if parsed_node.package_name in self.package_updaters:
             new_value = self.package_updaters[parsed_node.package_name](override, parsed_node)
         else:
@@ -280,9 +279,18 @@ def update_parsed_node_config_dict(
     def update_parsed_node_relation_names(
         self, parsed_node: IntermediateNode, config_dict: Dict[str, Any]
     ) -> None:
-        self._update_node_database(parsed_node, config_dict)
-        self._update_node_schema(parsed_node, config_dict)
-        self._update_node_alias(parsed_node, config_dict)
+
+        if parsed_node.resource_type == "snapshot":
+            if "target_database" in config_dict and config_dict["target_database"]:
+                parsed_node.database = config_dict["target_database"]
+            if "target_schema" in config_dict and config_dict["target_schema"]:
+                parsed_node.schema = config_dict["target_schema"]
+        else:
+            # These call the RelationUpdate callable to go through generate_name macros
+            self._update_node_database(parsed_node, config_dict.get("database"))
+            self._update_node_schema(parsed_node, config_dict.get("schema"))
+            self._update_node_alias(parsed_node, config_dict.get("alias"))
+
         self._update_node_relation_name(parsed_node)
 
     def update_parsed_node_config(
@@ -349,7 +357,7 @@ def update_parsed_node_config(
         # do this once before we parse the node database/schema/alias, so
         # parsed_node.config is what it would be if they did nothing
         self.update_parsed_node_config_dict(parsed_node, config_dict)
-        # This updates the node database/schema/alias
+        # This updates the node database/schema/alias/relation_name
         self.update_parsed_node_relation_names(parsed_node, config_dict)
 
         # tests don't have hooks
diff --git a/tests/functional/simple_snapshot/fixtures.py b/tests/functional/simple_snapshot/fixtures.py
index 633a6be4efa..6b3ecc2b101 100644
--- a/tests/functional/simple_snapshot/fixtures.py
+++ b/tests/functional/simple_snapshot/fixtures.py
@@ -96,6 +96,18 @@
         owner: 'a_owner'
 """
 
+models__schema_with_target_schema_yml = """
+version: 2
+snapshots:
+  - name: snapshot_actual
+    tests:
+      - mutually_exclusive_ranges
+    config:
+      meta:
+        owner: 'a_owner'
+      target_schema: schema_from_schema_yml
+"""
+
 models__ref_snapshot_sql = """
 select * from {{ ref('snapshot_actual') }}
 """
@@ -281,6 +293,26 @@
 {% endsnapshot %}
 """
 
+snapshots_pg__snapshot_no_target_schema_sql = """
+{% snapshot snapshot_actual %}
+
+    {{
+        config(
+            target_database=var('target_database', database),
+            unique_key='id || ' ~ "'-'" ~ ' || first_name',
+            strategy='timestamp',
+            updated_at='updated_at',
+        )
+    }}
+
+    {% if var('invalidate_hard_deletes', 'false') | as_bool %}
+        {{ config(invalidate_hard_deletes=True) }}
+    {% endif %}
+
+    select * from {{target.database}}.{{target.schema}}.seed
+
+{% endsnapshot %}
+"""
 
 models_slow__gen_sql = """
 
diff --git a/tests/functional/simple_snapshot/test_basic_snapshot.py b/tests/functional/simple_snapshot/test_basic_snapshot.py
index d6aec68ab6c..ff4799f10ab 100644
--- a/tests/functional/simple_snapshot/test_basic_snapshot.py
+++ b/tests/functional/simple_snapshot/test_basic_snapshot.py
@@ -2,13 +2,15 @@
 from datetime import datetime
 import pytz
 import pytest
-from dbt.tests.util import run_dbt, check_relations_equal, relation_from_name
+from dbt.tests.util import run_dbt, check_relations_equal, relation_from_name, write_file
 from tests.functional.simple_snapshot.fixtures import (
     models__schema_yml,
+    models__schema_with_target_schema_yml,
     models__ref_snapshot_sql,
     seeds__seed_newcol_csv,
     seeds__seed_csv,
     snapshots_pg__snapshot_sql,
+    snapshots_pg__snapshot_no_target_schema_sql,
     macros__test_no_overlaps_sql,
     macros_custom_snapshot__custom_sql,
     snapshots_pg_custom_namespaced__snapshot_sql,
@@ -123,6 +125,41 @@ def test_basic_ref(self, project):
         ref_setup(project, num_snapshot_models=1)
 
 
+class TestBasicTargetSchemaConfig(Basic):
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshots_pg__snapshot_no_target_schema_sql}
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self, unique_schema):
+        return {
+            "snapshots": {
+                "test": {
+                    "target_schema": unique_schema + "_alt",
+                }
+            }
+        }
+
+    def test_target_schema(self, project):
+        manifest = run_dbt(["parse"])
+        assert len(manifest.nodes) == 5
+        # ensure that the schema in the snapshot node is the same as target_schema
+        snapshot_id = "snapshot.test.snapshot_actual"
+        snapshot_node = manifest.nodes[snapshot_id]
+        assert snapshot_node.schema == f"{project.test_schema}_alt"
+        assert (
+            snapshot_node.relation_name
+            == f'"{project.database}"."{project.test_schema}_alt"."snapshot_actual"'
+        )
+        assert snapshot_node.meta == {"owner": "a_owner"}
+
+        # write out schema.yml file and check again
+        write_file(models__schema_with_target_schema_yml, "models", "schema.yml")
+        manifest = run_dbt(["parse"])
+        snapshot_node = manifest.nodes[snapshot_id]
+        assert snapshot_node.schema == "schema_from_schema_yml"
+
+
 class CustomNamespace:
     @pytest.fixture(scope="class")
     def snapshots(self):


Code review comments:
diff_hunk: @@ -280,9 +279,18 @@ def update_parsed_node_config_dict(
     def update_parsed_node_relation_names(
         self, parsed_node: IntermediateNode, config_dict: Dict[str, Any]
     ) -> None:
-        self._update_node_database(parsed_node, config_dict)
-        self._update_node_schema(parsed_node, config_dict)
-        self._update_node_alias(parsed_node, config_dict)
+
+        if parsed_node.resource_type == "snapshot":
+            if "target_database" in config_dict and config_dict["target_database"]:
+                parsed_node.database = config_dict["target_database"]
+            if "target_schema" in config_dict and config_dict["target_schema"]:
+                parsed_node.schema = config_dict["target_schema"]
path: core/dbt/parser/base.py
text: @user1:
For my own understanding: Previously we were calling also `self._update_node_[database|schema|alias]` on snapshots nodes. Were those no-ops and are no longer necessary?

@author:
This feels like it has holes in it, but I was attempting to mimic the logic in "set_snapshot_attributes" in core/dbt/parser/snapshots.py.  Which doesn't actually use the database set in configs, and always just sets "target_schema" (which is required). It also doesn't go through the generate_name code (so that customers have to call the generate_name code directly in their models if they want it).

diff_hunk: @@ -280,9 +279,18 @@ def update_parsed_node_config_dict(
     def update_parsed_node_relation_names(
         self, parsed_node: IntermediateNode, config_dict: Dict[str, Any]
     ) -> None:
-        self._update_node_database(parsed_node, config_dict)
-        self._update_node_schema(parsed_node, config_dict)
-        self._update_node_alias(parsed_node, config_dict)
+
+        if parsed_node.resource_type == "snapshot":
path: core/dbt/parser/base.py
text: @user1:
small nit: let's use the `NodeType.Snapshot` enum here instead if possible

@author:
Done