point at branch

* applied new integration tests to existing framework

* applied new integration tests to existing framework

* generalized tests for reusability in adapters; fixed drop index issue

* generalized tests for reusability in adapters; fixed drop index issue

* removed unnecessary overrides in tests

* adjusted import to allow for usage in adapters

* adjusted import to allow for usage in adapters

* removed fixture artifact

* generalized the materialized view fixture which will need to be specific to the adapter

* unskipped tests in the test runner package

* corrected test condition

* corrected test condition

* added missing initial build for the relation type swap tests

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 1.6.0b5
+current_version = 1.6.0b8
 parse = (?P<major>[\d]+) # major version number
 	\.(?P<minor>[\d]+) # minor version number
 	\.(?P<patch>[\d]+) # patch version number

.changes/1.6.0-b6.md

@@ -0,0 +1,6 @@
+## dbt-core 1.6.0-b6 - June 23, 2023
+
+### Fixes
+
+- Allow semantic model measure exprs to be defined with ints and bools in yaml ([#7865](https://github.com/dbt-labs/dbt-core/issues/7865))
+- Update `use_discrete_percentile` and `use_approximate_percentile` to be non optional and default to `False` ([#7866](https://github.com/dbt-labs/dbt-core/issues/7866))

.changes/1.6.0-b7.md

@@ -0,0 +1,25 @@
+## dbt-core 1.6.0-b7 - June 28, 2023
+
+### Features
+
+- Add merge as valid incremental strategy for postgres ([#1880](https://github.com/dbt-labs/dbt-core/issues/1880))
+- Handle external model nodes in state:modified ([#7563](https://github.com/dbt-labs/dbt-core/issues/7563))
+- Add invocation_command to flags ([#6051](https://github.com/dbt-labs/dbt-core/issues/6051))
+- Add thread_id context var ([#7941](https://github.com/dbt-labs/dbt-core/issues/7941))
+- Add partial parsing support for semantic models ([#7897](https://github.com/dbt-labs/dbt-core/issues/7897))
+- Add restrict-access to dbt_project.yml ([#7713](https://github.com/dbt-labs/dbt-core/issues/7713))
+- allow setting enabled and depends_on_nodes from ModelNodeArgs ([#7506](https://github.com/dbt-labs/dbt-core/issues/7506))
+
+### Fixes
+
+- Raise better error message when dispatching a package that is not installed ([#5801](https://github.com/dbt-labs/dbt-core/issues/5801))
+- add access to ModelNodeArgs for external node building ([#7890](https://github.com/dbt-labs/dbt-core/issues/7890))
+
+### Under the Hood
+
+- Update mashumaro to 3.8.1 ([#7950](https://github.com/dbt-labs/dbt-core/issues/7950))
+- Refactor: entry point for cross-project ref ([#7954](https://github.com/dbt-labs/dbt-core/issues/7954))
+
+### Contributors
+- [@NiallRees](https://github.com/NiallRees) ([#6051](https://github.com/dbt-labs/dbt-core/issues/6051), [#7941](https://github.com/dbt-labs/dbt-core/issues/7941))
+- [@rainermensing](https://github.com/rainermensing) ([#1880](https://github.com/dbt-labs/dbt-core/issues/1880))

.changes/1.6.0-b8.md

@@ -0,0 +1,30 @@
+## dbt-core 1.6.0-b8 - June 30, 2023
+
+### Features
+
+- This change adds new selector methods to the state selector. Namely, state:unmodified and state:old. ([#7564](https://github.com/dbt-labs/dbt-core/issues/7564))
+- dbt clone ([#7258](https://github.com/dbt-labs/dbt-core/issues/7258))
+- Support '_'-delimited fqn matching for versioned models and matching on Path.stem for path selection ([#7639](https://github.com/dbt-labs/dbt-core/issues/7639))
+- Store time_spline table configuration in semantic manifest ([#7938](https://github.com/dbt-labs/dbt-core/issues/7938))
+
+### Fixes
+
+- Fix CTE insertion position when the model uses WITH RECURSIVE ([#7350](https://github.com/dbt-labs/dbt-core/issues/7350))
+- Unified to UTC ([#7664](https://github.com/dbt-labs/dbt-core/issues/7664))
+- Remove limitation on use of sqlparse 0.4.4 ([#7515](https://github.com/dbt-labs/dbt-core/issues/7515))
+- Move project_root contextvar into events.contextvars ([#7937](https://github.com/dbt-labs/dbt-core/issues/7937))
+- Fix typo in ModelNodeArgs ([#7991](https://github.com/dbt-labs/dbt-core/issues/7991))
+- Allow on_schema_change = fail for contracted incremental models ([#7975](https://github.com/dbt-labs/dbt-core/issues/7975))
+
+### Docs
+
+- add note before running integration tests ([dbt-docs/#nothing](https://github.com/dbt-labs/dbt-docs/issues/nothing))
+
+### Under the Hood
+
+- Populate metric input measures ([#7884](https://github.com/dbt-labs/dbt-core/issues/7884))
+
+### Contributors
+- [@d-kaneshiro](https://github.com/d-kaneshiro) ([#7664](https://github.com/dbt-labs/dbt-core/issues/7664), [#nothing](https://github.com/dbt-labs/dbt-core/issues/nothing))
+- [@trouze](https://github.com/trouze) ([#7564](https://github.com/dbt-labs/dbt-core/issues/7564))
+- [@willbryant](https://github.com/willbryant) ([#7350](https://github.com/dbt-labs/dbt-core/issues/7350))

.changes/1.6.0/Docs-20230525-103918.yaml

@@ -0,0 +1,6 @@
+kind: Docs
+body: add note before running integration tests
+time: 2023-05-25T10:39:18.75138+09:00
+custom:
+  Author: d-kaneshiro
+  Issue: nothing

.changes/1.6.0/Features-20230211-163236.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add merge as valid incremental strategy for postgres
+time: 2023-02-11T16:32:36.2260502+01:00
+custom:
+  Author: rainermensing
+  Issue: "1880"

.changes/1.6.0/Features-20230603-141625.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: This change adds new selector methods to the state selector. Namely, state:unmodified
+  and state:old.
+time: 2023-06-03T14:16:25.249884-05:00
+custom:
+  Author: trouze
+  Issue: "7564"

.changes/1.6.0/Features-20230616-104849.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: dbt clone
+time: 2023-06-16T10:48:49.079961-05:00
+custom:
+  Author: jtcohen6 aranke McKnight-42
+  Issue: "7258"

.changes/1.6.0/Features-20230622-142002.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Handle external model nodes in state:modified
+time: 2023-06-22T14:20:02.104809-04:00
+custom:
+  Author: michelleark
+  Issue: "7563"

.changes/1.6.0/Features-20230623-111254.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add invocation_command to flags
+time: 2023-06-23T11:12:54.523157-07:00
+custom:
+  Author: NiallRees
+  Issue: "6051"

.changes/1.6.0/Features-20230623-173357.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add thread_id context var
+time: 2023-06-23T17:33:57.412102-07:00
+custom:
+  Author: NiallRees
+  Issue: "7941"

.changes/1.6.0/Features-20230627-123123.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add partial parsing support for semantic models
+time: 2023-06-27T12:31:23.953018-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "7897"

.changes/1.6.0/Features-20230627-132749.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add restrict-access to dbt_project.yml
+time: 2023-06-27T13:27:49.114257-04:00
+custom:
+  Author: michelleark
+  Issue: "7713"

.changes/1.6.0/Features-20230627-152207.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: allow setting enabled and depends_on_nodes from ModelNodeArgs
+time: 2023-06-27T15:22:07.313639-04:00
+custom:
+  Author: michelleark
+  Issue: "7506"

.changes/1.6.0/Features-20230629-135758.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support '_'-delimited fqn matching for versioned models and matching on Path.stem for path selection
+time: 2023-06-29T13:57:58.38283-04:00
+custom:
+  Author: michelleark
+  Issue: "7639"

.changes/1.6.0/Features-20230629-172524.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Store time_spline table configuration in semantic manifest
+time: 2023-06-29T17:25:24.265366-04:00
+custom:
+  Author: gshank
+  Issue: "7938"

.changes/1.6.0/Fixes-20220909-164413.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Raise better error message when dispatching a package that is not installed
+time: 2022-09-09T16:44:13.382685-06:00
+custom:
+  Author: "dbeatty10"
+  Issue: "5801"
+  PR: "5804"

.changes/1.6.0/Fixes-20230420-123221.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix CTE insertion position when the model uses WITH RECURSIVE
+time: 2023-04-20T12:32:21.432848+12:00
+custom:
+  Author: willbryant
+  Issue: "7350"

.changes/1.6.0/Fixes-20230525-091955.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Unified to UTC
+time: 2023-05-25T09:19:55.865281+09:00
+custom:
+  Author: d-kaneshiro
+  Issue: "7664"

.changes/1.6.0/Fixes-20230614-134933.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Allow semantic model measure exprs to be defined with ints and bools in yaml
+time: 2023-06-14T13:49:33.544552-07:00
+custom:
+  Author: QMalcolm
+  Issue: "7865"

.changes/1.6.0/Fixes-20230614-145954.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Update `use_discrete_percentile` and `use_approximate_percentile` to be non
+  optional and default to `False`
+time: 2023-06-14T14:59:54.042341-07:00
+custom:
+  Author: QMalcolm
+  Issue: "7866"

.changes/1.6.0/Fixes-20230621-185452.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Remove limitation on use of sqlparse 0.4.4
+time: 2023-06-21T18:54:52.246578-04:00
+custom:
+  Author: gshank
+  Issue: "7515"

.changes/1.6.0/Fixes-20230626-115838.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Move project_root contextvar into events.contextvars
+time: 2023-06-26T11:58:38.965299-04:00
+custom:
+  Author: gshank
+  Issue: "7937"

.changes/1.6.0/Fixes-20230627-152528.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: add access to ModelNodeArgs for external node building
+time: 2023-06-27T15:25:28.488767-04:00
+custom:
+  Author: michelleark
+  Issue: "7890"

.changes/1.6.0/Fixes-20230629-120321.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix typo in ModelNodeArgs
+time: 2023-06-29T12:03:21.179283-04:00
+custom:
+  Author: michelleark
+  Issue: "7991"

.changes/1.6.0/Fixes-20230629-221615.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Allow on_schema_change = fail for contracted incremental models
+time: 2023-06-29T22:16:15.34895-04:00
+custom:
+  Author: michelleark
+  Issue: "7975"

.changes/1.6.0/Under the Hood-20230612-070827.yaml

@@ -1,5 +1,5 @@
 kind: Under the Hood
-body: Rm space from NodeType strings
+body: Replace space with underscore in NodeType strings
 time: 2023-06-12T07:08:27.276551-04:00
 custom:
   Author: jtcohen6

.changes/1.6.0/Under the Hood-20230626-122339.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Update mashumaro to 3.8.1
+time: 2023-06-26T12:23:39.095215-04:00
+custom:
+  Author: gshank
+  Issue: "7950"

.changes/1.6.0/Under the Hood-20230627-222328.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: 'Refactor: entry point for cross-project ref'
+time: 2023-06-27T22:23:28.73069-04:00
+custom:
+  Author: michelleark
+  Issue: "7954"

.changes/1.6.0/Under the Hood-20230628-142006.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Populate metric input measures
+time: 2023-06-28T14:20:06.022738-07:00
+custom:
+  Author: QMalcolm
+  Issue: "7884"

.changes/unreleased/Dependencies-20230707-104052.yaml

@@ -0,0 +1,6 @@
+kind: Dependencies
+body: Pin click>=8.1.1,<8.1.4
+time: 2023-07-07T10:40:52.565603-05:00
+custom:
+  Author: emmyoop
+  PR: "8050"

.changes/unreleased/Dependencies-20230712-165219.yaml

@@ -0,0 +1,6 @@
+kind: Dependencies
+body: Bump `dbt-semantic-interfaces` to `~=0.1.0rc1`
+time: 2023-07-12T16:52:19.748953-07:00
+custom:
+  Author: QMalcolm
+  PR: "8082"

.changes/unreleased/Docs-20230702-130158.yaml

@@ -0,0 +1,6 @@
+kind: Docs
+body: Fix broken links in `CONTRIBUTING.md`.
+time: 2023-07-02T13:01:58.622569-04:00
+custom:
+  Author: gem7318
+  Issue: "8018"

.changes/unreleased/Features-20230629-175712.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add validate_sql method to BaseAdapter with implementation for SQLAdapter
+time: 2023-06-29T17:57:12.599313-07:00
+custom:
+  Author: tlento
+  Issue: "7839"

.changes/unreleased/Features-20230707-112838.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support validation of metrics and semantic models.
+time: 2023-07-07T11:28:38.760462-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "7969"

.changes/unreleased/Features-20230712-123724.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Begin populating `depends_on` of metric nodes
+time: 2023-07-12T12:37:24.01449-07:00
+custom:
+  Author: QMalcolm gshank
+  Issue: "7854"

.changes/unreleased/Features-20230714-144909.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Enumerate supported materialized view features for dbt-postgres
+time: 2023-07-14T14:49:09.680123-04:00
+custom:
+  Author: mikealfare
+  Issue: "6911"

.changes/unreleased/Fixes-20230320-153225.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: add negative part_number arg for split part macro
+time: 2023-03-20T15:32:25.5932-05:00
+custom:
+  Author: dave-connors-3
+  Issue: "7915"

.changes/unreleased/Fixes-20230615-102639.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix accidental propagation of log messages to root logger.
+time: 2023-06-15T10:26:39.139421-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "7872"

.changes/unreleased/Fixes-20230615-110538.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Fixed an issue which blocked debug logging to stdout with --log-level debug,
+  unless --debug was also used.
+time: 2023-06-15T11:05:38.053387-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "7872"

.changes/unreleased/Fixes-20230623-092933.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix query comment tests
+time: 2023-06-23T09:29:33.225649+02:00
+custom:
+  Author: damian3031
+  Issue: "7845"

.changes/unreleased/Fixes-20230628-144125.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Inline query emit proper error message
+time: 2023-06-28T14:41:25.699046-07:00
+custom:
+  Author: ChenyuLInx
+  Issue: "7940"

.changes/unreleased/Fixes-20230704-114752.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Nicer error message if model with enforced contract is missing 'columns' specification
+time: 2023-07-04T11:47:52.976527+02:00
+custom:
+  Author: jtcohen6
+  Issue: "7943"

.changes/unreleased/Fixes-20230705-192039.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: include 'v' in ModelNodeArgs.unique_id
+time: 2023-07-05T19:20:39.581291-04:00
+custom:
+  Author: michelleark
+  Issue: "8039"

.changes/unreleased/Fixes-20230710-172547.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix fail-fast behavior (including retry)
+time: 2023-07-10T17:25:47.912129-05:00
+custom:
+  Author: aranke
+  Issue: "7785"

.changes/unreleased/Fixes-20230711-083213.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Remove `create_metric` as a `SemanticModel.Measure` property because it currently
+  doesn't do anything
+time: 2023-07-11T08:32:13.158779-07:00
+custom:
+  Author: QMalcolm
+  Issue: "8064"

.changes/unreleased/Fixes-20230711-141432.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Remove `VOLUME` declaration within Dockerfile
+time: 2023-07-11T14:14:32.117718-06:00
+custom:
+  Author: alexrosenfeld10
+  Issue: "4784"

.changes/unreleased/Fixes-20230711-145311.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix Dockerfile.test
+time: 2023-07-11T14:53:11.454863-06:00
+custom:
+  Author: dbeatty10
+  Issue: "7352"

.changes/unreleased/Fixes-20230711-171901.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Detect breaking contract changes to versioned models
+time: 2023-07-11T17:19:01.120379-07:00
+custom:
+  Author: michelleark
+  Issue: "8030"

.changes/unreleased/Fixes-20230711-234737.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Update DryRunMethod test classes ValidateSqlMethod naming
+time: 2023-07-11T23:47:37.9525-04:00
+custom:
+  Author: tlento
+  Issue: "7839"

.changes/unreleased/Fixes-20230712-172037.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix typo in `NonAdditiveDimension` implementation
+time: 2023-07-12T17:20:37.089887-07:00
+custom:
+  Author: QMalcolm
+  Issue: "8088"

.changes/unreleased/Under the Hood-20230706-175507.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add option to specify partial parse file
+time: 2023-07-06T17:55:07.525287-07:00
+custom:
+  Author: ChenyuLInx
+  Issue: "7911"

.changes/unreleased/Under the Hood-20230712-170559.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Add semantic_models to resource counts
+time: 2023-07-12T17:05:59.497596+02:00
+custom:
+  Author: jtcohen6
+  Issue: "8077"

.github/pull_request_template.md

@@ -1,23 +1,35 @@
-resolves #
+resolves #  
+[docs](https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose) dbt-labs/docs.getdbt.com/#  
 
 <!---
   Include the number of the issue addressed by this PR above if applicable.
   PRs for code changes without an associated issue *will not be merged*.
   See CONTRIBUTING.md for more information.
+
+  Include the number of the docs issue that was opened for this PR. If
+  this change has no user-facing implications, "N/A" suffices instead. New
+  docs tickets can be created by clicking the link above or by going to
+  https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose.
 -->
 
-### Description
+### Problem
 
 <!---
-  Describe the Pull Request here. Add any references and info to help reviewers
-  understand your changes. Include any tradeoffs you considered.
+  Describe the problem this PR is solving. What is the application state
+  before this PR is merged?
+-->
+
+### Solution
+
+<!---
+  Describe the way this PR solves the above problem. Add as much detail as you
+  can to help reviewers understand your changes. Include any alternatives and
+  tradeoffs you considered.
 -->
 
 ### Checklist
 
-- [ ] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me
-- [ ] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
-- [ ] I have run this code in development and it appears to resolve the stated issue
+- [ ] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me  
+- [ ] I have run this code in development and it appears to resolve the stated issue  
 - [ ] This PR includes tests, or tests are not required/relevant for this PR
-- [ ] I have [opened an issue to add/update docs](https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose), or docs changes are not required/relevant for this PR
-- [ ] I have run `changie new` to [create a changelog entry](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md#adding-a-changelog-entry)
+- [ ] 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

.github/workflows/cut-release-branch.yml

@@ -31,7 +31,7 @@ permissions:
 jobs:
   cut_branch:
     name: "Cut branch and clean up main for dbt-core"
-    uses: dbt-labs/actions/.github/workflows/cut-release-branch.yml@main
+    uses: dbt-labs/actions/.github/workflows/cut-release-branch.yml@er/fix-latest-cutter
     with:
       version_to_bump_main: ${{ inputs.version_to_bump_main }}
       new_branch_name: ${{ inputs.new_branch_name }}

.github/workflows/main.yml

@@ -131,6 +131,11 @@ jobs:
       DBT_TEST_USER_1: dbt_test_user_1
       DBT_TEST_USER_2: dbt_test_user_2
       DBT_TEST_USER_3: dbt_test_user_3
+      DD_CIVISIBILITY_AGENTLESS_ENABLED: true
+      DD_API_KEY: ${{ secrets.DATADOG_API_KEY }}
+      DD_SITE: datadoghq.com
+      DD_ENV: ci
+      DD_SERVICE: ${{ github.event.repository.name }}
 
     steps:
       - name: Check out the repository
@@ -161,7 +166,7 @@ jobs:
           tox --version
 
       - name: Run tests
-        run: tox
+        run: tox -- --ddtrace
 
       - name: Get current date
         if: always()

.github/workflows/test-repeater.yml

@@ -0,0 +1,155 @@
+# **what?**
+# This workflow will test all test(s) at the input path given number of times to determine if it's flaky or not.  You can test with any supported OS/Python combination.
+# This is batched in 10 to allow more test iterations faster.
+
+# **why?**
+# Testing if a test is flaky and if a previously flaky test has been fixed.  This allows easy testing on supported python versions and OS combinations.
+
+# **when?**
+# This is triggered manually from dbt-core.
+
+name: Flaky Tester
+
+on:
+  workflow_dispatch:
+    inputs:
+      branch:
+        description: 'Branch to check out'
+        type: string
+        required: true
+        default: 'main'
+      test_path:
+        description: 'Path to single test to run (ex: tests/functional/retry/test_retry.py::TestRetry::test_fail_fast)'
+        type: string
+        required: true
+        default: 'tests/functional/...'
+      python_version:
+        description: 'Version of Python to Test Against'
+        type: choice
+        options:
+          - '3.8'
+          - '3.9'
+          - '3.10'
+          - '3.11'
+      os:
+        description: 'OS to run test in'
+        type: choice
+        options:
+          - 'ubuntu-latest'
+          - 'macos-latest'
+          - 'windows-latest'
+      num_runs_per_batch:
+        description: 'Max number of times to run the test per batch.  We always run 10 batches.'
+        type: number
+        required: true
+        default: '50'
+
+permissions: read-all
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  debug:
+    runs-on: ubuntu-latest
+    steps:
+      - name: "[DEBUG] Output Inputs"
+        run: |
+          echo "Branch: ${{ inputs.branch }}"
+          echo "test_path: ${{ inputs.test_path }}"
+          echo "python_version: ${{ inputs.python_version }}"
+          echo "os: ${{ inputs.os }}"
+          echo "num_runs_per_batch: ${{ inputs.num_runs_per_batch }}"
+
+  pytest:
+    runs-on: ${{ inputs.os }}
+    strategy:
+      # run all batches, even if one fails.  This informs how flaky the test may be.
+      fail-fast: false
+      # using a matrix to speed up the jobs since the matrix will run in parallel when runners are available
+      matrix:
+        batch: ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
+    env:
+      PYTEST_ADDOPTS: "-v --color=yes -n4 --csv integration_results.csv"
+      DBT_TEST_USER_1: dbt_test_user_1
+      DBT_TEST_USER_2: dbt_test_user_2
+      DBT_TEST_USER_3: dbt_test_user_3
+      DD_CIVISIBILITY_AGENTLESS_ENABLED: true
+      DD_API_KEY: ${{ secrets.DATADOG_API_KEY }}
+      DD_SITE: datadoghq.com
+      DD_ENV: ci
+      DD_SERVICE: ${{ github.event.repository.name }}
+
+    steps:
+      - name: "Checkout code"
+        uses: actions/checkout@v3
+        with:
+          ref: ${{ inputs.branch }}
+
+      - name: "Setup Python"
+        uses: actions/setup-python@v4
+        with:
+          python-version: "${{ inputs.python_version }}"
+
+      - name: "Setup Dev Environment"
+        run: make dev
+
+      - name: "Set up postgres (linux)"
+        if: inputs.os == 'ubuntu-latest'
+        run: make setup-db
+
+        # mac and windows don't use make due to limitations with docker with those runners in GitHub
+      - name: "Set up postgres (macos)"
+        if: inputs.os == 'macos-latest'
+        uses: ./.github/actions/setup-postgres-macos
+
+      - name: "Set up postgres (windows)"
+        if: inputs.os == 'windows-latest'
+        uses: ./.github/actions/setup-postgres-windows
+
+      - name: "Test Command"
+        id: command
+        run: |
+          test_command="python -m pytest ${{ inputs.test_path }}"
+          echo "test_command=$test_command" >> $GITHUB_OUTPUT
+
+      - name: "Run test ${{ inputs.num_runs_per_batch }} times"
+        id: pytest
+        run: |
+          set +e
+          for ((i=1; i<=${{ inputs.num_runs_per_batch }}; i++))
+          do
+            echo "Running pytest iteration $i..."
+            python -m pytest --ddtrace ${{ inputs.test_path }}
+            exit_code=$?
+
+            if [[ $exit_code -eq 0 ]]; then
+              success=$((success + 1))
+              echo "Iteration $i: Success"
+            else
+              failure=$((failure + 1))
+              echo "Iteration $i: Failure"
+            fi
+
+            echo
+            echo "==========================="
+            echo "Successful runs: $success"
+            echo "Failed runs: $failure"
+            echo "==========================="
+            echo
+          done
+
+          echo "failure=$failure" >> $GITHUB_OUTPUT
+
+      - name: "Success and Failure Summary: ${{ inputs.os }}/Python ${{ inputs.python_version }}"
+        run: |
+          echo "Batch: ${{ matrix.batch }}"
+          echo "Successful runs: ${{ steps.pytest.outputs.success }}"
+          echo "Failed runs: ${{ steps.pytest.outputs.failure }}"
+
+      - name: "Error for Failures"
+        if: ${{ steps.pytest.outputs.failure }}
+        run: |
+            echo "Batch ${{ matrix.batch }} failed ${{ steps.pytest.outputs.failure }} of ${{ inputs.num_runs_per_batch }} tests"
+            exit 1

.gitignore

@@ -29,6 +29,8 @@ var/
 .mypy_cache/
 .dmypy.json
 logs/
+.user.yml
+profiles.yml
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

CHANGELOG.md

@@ -5,6 +5,71 @@
 - "Breaking changes" listed under a version may require action from end users or external maintainers when upgrading to that version.
 - Do not edit this file directly. This file is auto-generated using [changie](https://github.com/miniscruff/changie). For details on how to document a change, see [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md#adding-changelog-entry)
 
+## dbt-core 1.6.0-b8 - June 30, 2023
+
+### Features
+
+- This change adds new selector methods to the state selector. Namely, state:unmodified and state:old. ([#7564](https://github.com/dbt-labs/dbt-core/issues/7564))
+- dbt clone ([#7258](https://github.com/dbt-labs/dbt-core/issues/7258))
+- Support '_'-delimited fqn matching for versioned models and matching on Path.stem for path selection ([#7639](https://github.com/dbt-labs/dbt-core/issues/7639))
+- Store time_spline table configuration in semantic manifest ([#7938](https://github.com/dbt-labs/dbt-core/issues/7938))
+
+### Fixes
+
+- Fix CTE insertion position when the model uses WITH RECURSIVE ([#7350](https://github.com/dbt-labs/dbt-core/issues/7350))
+- Unified to UTC ([#7664](https://github.com/dbt-labs/dbt-core/issues/7664))
+- Remove limitation on use of sqlparse 0.4.4 ([#7515](https://github.com/dbt-labs/dbt-core/issues/7515))
+- Move project_root contextvar into events.contextvars ([#7937](https://github.com/dbt-labs/dbt-core/issues/7937))
+- Fix typo in ModelNodeArgs ([#7991](https://github.com/dbt-labs/dbt-core/issues/7991))
+- Allow on_schema_change = fail for contracted incremental models ([#7975](https://github.com/dbt-labs/dbt-core/issues/7975))
+
+### Docs
+
+- add note before running integration tests ([dbt-docs/#nothing](https://github.com/dbt-labs/dbt-docs/issues/nothing))
+
+### Under the Hood
+
+- Populate metric input measures ([#7884](https://github.com/dbt-labs/dbt-core/issues/7884))
+
+### Contributors
+- [@d-kaneshiro](https://github.com/d-kaneshiro) ([#7664](https://github.com/dbt-labs/dbt-core/issues/7664), [#nothing](https://github.com/dbt-labs/dbt-core/issues/nothing))
+- [@trouze](https://github.com/trouze) ([#7564](https://github.com/dbt-labs/dbt-core/issues/7564))
+- [@willbryant](https://github.com/willbryant) ([#7350](https://github.com/dbt-labs/dbt-core/issues/7350))
+
+
+## dbt-core 1.6.0-b7 - June 28, 2023
+
+### Features
+
+- Add merge as valid incremental strategy for postgres ([#1880](https://github.com/dbt-labs/dbt-core/issues/1880))
+- Handle external model nodes in state:modified ([#7563](https://github.com/dbt-labs/dbt-core/issues/7563))
+- Add invocation_command to flags ([#6051](https://github.com/dbt-labs/dbt-core/issues/6051))
+- Add thread_id context var ([#7941](https://github.com/dbt-labs/dbt-core/issues/7941))
+- Add partial parsing support for semantic models ([#7897](https://github.com/dbt-labs/dbt-core/issues/7897))
+- Add restrict-access to dbt_project.yml ([#7713](https://github.com/dbt-labs/dbt-core/issues/7713))
+- allow setting enabled and depends_on_nodes from ModelNodeArgs ([#7506](https://github.com/dbt-labs/dbt-core/issues/7506))
+
+### Fixes
+
+- Raise better error message when dispatching a package that is not installed ([#5801](https://github.com/dbt-labs/dbt-core/issues/5801))
+- add access to ModelNodeArgs for external node building ([#7890](https://github.com/dbt-labs/dbt-core/issues/7890))
+
+### Under the Hood
+
+- Update mashumaro to 3.8.1 ([#7950](https://github.com/dbt-labs/dbt-core/issues/7950))
+- Refactor: entry point for cross-project ref ([#7954](https://github.com/dbt-labs/dbt-core/issues/7954))
+
+### Contributors
+- [@NiallRees](https://github.com/NiallRees) ([#6051](https://github.com/dbt-labs/dbt-core/issues/6051), [#7941](https://github.com/dbt-labs/dbt-core/issues/7941))
+- [@rainermensing](https://github.com/rainermensing) ([#1880](https://github.com/dbt-labs/dbt-core/issues/1880))
+
+## dbt-core 1.6.0-b6 - June 23, 2023
+
+### Fixes
+
+- Allow semantic model measure exprs to be defined with ints and bools in yaml ([#7865](https://github.com/dbt-labs/dbt-core/issues/7865))
+- Update `use_discrete_percentile` and `use_approximate_percentile` to be non optional and default to `False` ([#7866](https://github.com/dbt-labs/dbt-core/issues/7866))
+
 ## dbt-core 1.6.0-b5 - June 22, 2023
 
 ### Features
@@ -34,8 +99,6 @@
 
 - Bump mypy from 0.981 to 1.0.1 ([#7027](https://github.com/dbt-labs/dbt-core/pull/7027))
 
-
-
 ## dbt-core 1.6.0-b4 - June 13, 2023
 
 ### Fixes

CONTRIBUTING.md

@@ -5,10 +5,10 @@
 1. [About this document](#about-this-document)
 2. [Getting the code](#getting-the-code)
 3. [Setting up an environment](#setting-up-an-environment)
-4. [Running `dbt` in development](#running-dbt-core-in-development)
+4. [Running dbt-core in development](#running-dbt-core-in-development)
 5. [Testing dbt-core](#testing)
 6. [Debugging](#debugging)
-7. [Adding a changelog entry](#adding-a-changelog-entry)
+7. [Adding or modifying a changelog entry](#adding-or-modifying-a-changelog-entry)
 8. [Submitting a Pull Request](#submitting-a-pull-request)
 
 ## About this document
@@ -113,7 +113,7 @@ When installed in this way, any changes you make to your local copy of the sourc
 
 With your virtualenv activated, the `dbt` script should point back to the source code you've cloned on your machine. You can verify this by running `which dbt`. This command should show you a path to an executable in your virtualenv.
 
-Configure your [profile](https://docs.getdbt.com/docs/configure-your-profile) as necessary to connect to your target databases. It may be a good idea to add a new profile pointing to a local Postgres instance, or a specific test sandbox within your data warehouse if appropriate.
+Configure your [profile](https://docs.getdbt.com/docs/configure-your-profile) as necessary to connect to your target databases. It may be a good idea to add a new profile pointing to a local Postgres instance, or a specific test sandbox within your data warehouse if appropriate. Make sure to create a profile before running integration tests.
 
 ## Testing
 
@@ -174,9 +174,7 @@ Finally, you can also run a specific test or group of tests using [`pytest`](htt
 python3 -m pytest tests/unit/test_graph.py
 # run a specific unit test
 python3 -m pytest tests/unit/test_graph.py::GraphTest::test__dependency_list
-# run specific Postgres integration tests (old way)
-python3 -m pytest -m profile_postgres test/integration/074_postgres_unlogged_table_tests
-# run specific Postgres integration tests (new way)
+# run specific Postgres functional tests
 python3 -m pytest tests/functional/sources
 ```
 
@@ -186,8 +184,7 @@ python3 -m pytest tests/functional/sources
 
 Here are some general rules for adding tests:
 * unit tests (`tests/unit`) don’t need to access a database; "pure Python" tests should be written as unit tests
-* functional tests (`test/integration` & `tests/functional`) cover anything that interacts with a database, namely adapter
-* *everything in* `test/*` *is being steadily migrated to* `tests/*`
+* functional tests (`tests/functional`) cover anything that interacts with a database, namely adapter
 
 ## Debugging
 

Dockerfile.test

@@ -3,13 +3,13 @@
 #  See `/docker` for a generic and production-ready docker file
 ##
 
-FROM ubuntu:23.10
+FROM ubuntu:22.04
 
 ENV DEBIAN_FRONTEND noninteractive
 
 RUN apt-get update \
     && apt-get install -y --no-install-recommends \
-    software-properties-common \
+    software-properties-common gpg-agent \
     && add-apt-repository ppa:git-core/ppa -y \
     && apt-get dist-upgrade -y \
     && apt-get install -y --no-install-recommends \
@@ -30,8 +30,8 @@ RUN apt-get update \
     unixodbc-dev \
     && add-apt-repository ppa:deadsnakes/ppa \
     && apt-get install -y \
-    python \
-    python-dev \
+    python-is-python3 \
+    python-dev-is-python3 \
     python3-pip \
     python3.8 \
     python3.8-dev \

core/dbt/adapters/base/impl.py

@@ -289,6 +289,17 @@ class BaseAdapter(metaclass=AdapterMeta):
         """
         return self.connections.execute(sql=sql, auto_begin=auto_begin, fetch=fetch, limit=limit)
 
+    def validate_sql(self, sql: str) -> AdapterResponse:
+        """Submit the given SQL to the engine for validation, but not execution.
+
+        This should throw an appropriate exception if the input SQL is invalid, although
+        in practice that will generally be handled by delegating to an existing method
+        for execution and allowing the error handler to take care of the rest.
+
+        :param str sql: The sql to validate
+        """
+        raise NotImplementedError("`validate_sql` is not implemented for this adapter!")
+
     @available.parse(lambda *a, **k: [])
     def get_column_schema_from_query(self, sql: str) -> List[BaseColumn]:
         """Get a list of the Columns with names and data types from the given sql."""
@@ -785,7 +796,6 @@ class BaseAdapter(metaclass=AdapterMeta):
         schema: str,
         identifier: str,
     ) -> List[BaseRelation]:
-
         matches = []
 
         search = self._make_match_kwargs(database, schema, identifier)
@@ -1063,7 +1073,6 @@ class BaseAdapter(metaclass=AdapterMeta):
         schemas: Set[str],
         manifest: Manifest,
     ) -> agate.Table:
-
         kwargs = {"information_schema": information_schema, "schemas": schemas}
         table = self.execute_macro(
             GET_CATALOG_MACRO_NAME,
@@ -1453,7 +1462,6 @@ join diff_count using (id)
 def catch_as_completed(
     futures,  # typing: List[Future[agate.Table]]
 ) -> Tuple[agate.Table, List[Exception]]:
-
     # catalogs: agate.Table = agate.Table(rows=[])
     tables: List[agate.Table] = []
     exceptions: List[Exception] = []

core/dbt/adapters/sql/connections.py

@@ -52,7 +52,6 @@ class SQLConnectionManager(BaseConnectionManager):
         bindings: Optional[Any] = None,
         abridge_sql_log: bool = False,
     ) -> Tuple[Connection, Any]:
-
         connection = self.get_thread_connection()
         if auto_begin and connection.transaction_open is False:
             self.begin()

core/dbt/adapters/sql/impl.py

@@ -1,7 +1,7 @@
 import agate
 from typing import Any, Optional, Tuple, Type, List
 
-from dbt.contracts.connection import Connection
+from dbt.contracts.connection import Connection, AdapterResponse
 from dbt.exceptions import RelationTypeNullError
 from dbt.adapters.base import BaseAdapter, available
 from dbt.adapters.cache import _make_ref_key_dict
@@ -22,6 +22,7 @@ RENAME_RELATION_MACRO_NAME = "rename_relation"
 TRUNCATE_RELATION_MACRO_NAME = "truncate_relation"
 DROP_RELATION_MACRO_NAME = "drop_relation"
 ALTER_COLUMN_TYPE_MACRO_NAME = "alter_column_type"
+VALIDATE_SQL_MACRO_NAME = "validate_sql"
 
 
 class SQLAdapter(BaseAdapter):
@@ -218,6 +219,34 @@ class SQLAdapter(BaseAdapter):
         results = self.execute_macro(CHECK_SCHEMA_EXISTS_MACRO_NAME, kwargs=kwargs)
         return results[0][0] > 0
 
+    def validate_sql(self, sql: str) -> AdapterResponse:
+        """Submit the given SQL to the engine for validation, but not execution.
+
+        By default we simply prefix the query with the explain keyword and allow the
+        exceptions thrown by the underlying engine on invalid SQL inputs to bubble up
+        to the exception handler. For adjustments to the explain statement - such as
+        for adapters that have different mechanisms for hinting at query validation
+        or dry-run - callers may be able to override the validate_sql_query macro with
+        the addition of an <adapter>__validate_sql implementation.
+
+        :param sql str: The sql to validate
+        """
+        kwargs = {
+            "sql": sql,
+        }
+        result = self.execute_macro(VALIDATE_SQL_MACRO_NAME, kwargs=kwargs)
+        # The statement macro always returns an AdapterResponse in the output AttrDict's
+        # `response` property, and we preserve the full payload in case we want to
+        # return fetched output for engines where explain plans are emitted as columnar
+        # results. Any macro override that deviates from this behavior may encounter an
+        # assertion error in the runtime.
+        adapter_response = result.response  # type: ignore[attr-defined]
+        assert isinstance(adapter_response, AdapterResponse), (
+            f"Expected AdapterResponse from validate_sql macro execution, "
+            f"got {type(adapter_response)}."
+        )
+        return adapter_response
+
     # This is for use in the test suite
     def run_sql_for_tests(self, sql, fetch, conn):
         cursor = conn.handle.cursor()

core/dbt/cli/flags.py

@@ -206,6 +206,9 @@ class Flags:
             profiles_dir = getattr(self, "PROFILES_DIR", None)
             user_config = read_user_config(profiles_dir) if profiles_dir else None
 
+        # Add entire invocation command to flags
+        object.__setattr__(self, "INVOCATION_COMMAND", "dbt " + " ".join(sys.argv[1:]))
+
         # Overwrite default assignments with user config if available.
         if user_config:
             param_assigned_from_default_copy = params_assigned_from_default.copy()
@@ -373,6 +376,7 @@ def command_args(command: CliCommand) -> ArgsList:
     CMD_DICT: Dict[CliCommand, ClickCommand] = {
         CliCommand.BUILD: cli.build,
         CliCommand.CLEAN: cli.clean,
+        CliCommand.CLONE: cli.clone,
         CliCommand.COMPILE: cli.compile,
         CliCommand.DOCS_GENERATE: cli.docs_generate,
         CliCommand.DOCS_SERVE: cli.docs_serve,

core/dbt/cli/main.py

@@ -23,6 +23,7 @@ from dbt.contracts.results import (
 from dbt.events.base_types import EventMsg
 from dbt.task.build import BuildTask
 from dbt.task.clean import CleanTask
+from dbt.task.clone import CloneTask
 from dbt.task.compile import CompileTask
 from dbt.task.debug import DebugTask
 from dbt.task.deps import DepsTask
@@ -76,7 +77,6 @@ class dbtRunner:
             dbt_ctx.obj = {
                 "manifest": self.manifest,
                 "callbacks": self.callbacks,
-                "_publications": kwargs.get("publications"),
             }
 
             for key, value in kwargs.items():
@@ -139,6 +139,7 @@ class dbtRunner:
 @p.log_path
 @p.macro_debugging
 @p.partial_parse
+@p.partial_parse_file_path
 @p.populate_cache
 @p.print
 @p.printer_width
@@ -578,7 +579,7 @@ def run(ctx, **kwargs):
     return results, success
 
 
-# dbt run
+# dbt retry
 @cli.command("retry")
 @click.pass_context
 @p.project_dir
@@ -608,6 +609,43 @@ def retry(ctx, **kwargs):
     return results, success
 
 
+# dbt clone
+@cli.command("clone")
+@click.pass_context
+@p.defer_state
+@p.exclude
+@p.full_refresh
+@p.profile
+@p.profiles_dir
+@p.project_dir
+@p.resource_type
+@p.select
+@p.selector
+@p.state  # required
+@p.target
+@p.target_path
+@p.threads
+@p.vars
+@p.version_check
+@requires.preflight
+@requires.profile
+@requires.project
+@requires.runtime_config
+@requires.manifest
+@requires.postflight
+def clone(ctx, **kwargs):
+    """Create clones of selected nodes based on their location in the manifest provided to --state."""
+    task = CloneTask(
+        ctx.obj["flags"],
+        ctx.obj["runtime_config"],
+        ctx.obj["manifest"],
+    )
+
+    results = task.run()
+    success = task.interpret_results(results)
+    return results, success
+
+
 # dbt run operation
 @cli.command("run-operation")
 @click.pass_context

core/dbt/cli/params.py

@@ -239,6 +239,15 @@ partial_parse = click.option(
     default=True,
 )
 
+partial_parse_file_path = click.option(
+    "--partial-parse-file-path",
+    envvar="DBT_PARTIAL_PARSE_FILE_PATH",
+    help="Internal flag for path to partial_parse.manifest file.",
+    default=None,
+    hidden=True,
+    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+)
+
 populate_cache = click.option(
     "--populate-cache/--no-populate-cache",
     envvar="DBT_POPULATE_CACHE",

core/dbt/cli/requires.py

@@ -23,6 +23,7 @@ from dbt.parser.manifest import ManifestLoader, write_manifest
 from dbt.profiler import profiler
 from dbt.tracking import active_user, initialize_from_flags, track_run
 from dbt.utils import cast_dict_to_dict_of_strings
+from dbt.plugins import set_up_plugin_manager, get_plugin_manager
 
 from click import Context
 from functools import update_wrapper
@@ -160,6 +161,9 @@ def project(func):
         )
         ctx.obj["project"] = project
 
+        # Plugins
+        set_up_plugin_manager(project_name=project.project_name)
+
         if dbt.tracking.active_user is not None:
             project_id = None if project is None else project.hashed_name()
 
@@ -242,12 +246,15 @@ def manifest(*args0, write=True, write_perf_info=False):
                 manifest = ManifestLoader.get_full_manifest(
                     runtime_config,
                     write_perf_info=write_perf_info,
-                    publications=ctx.obj.get("_publications"),
                 )
 
                 ctx.obj["manifest"] = manifest
                 if write and ctx.obj["flags"].write_json:
-                    write_manifest(manifest, ctx.obj["runtime_config"].project_target_path)
+                    write_manifest(manifest, runtime_config.project_target_path)
+                    pm = get_plugin_manager(runtime_config.project_name)
+                    plugin_artifacts = pm.get_manifest_artifacts(manifest)
+                    for path, plugin_artifact in plugin_artifacts.items():
+                        plugin_artifact.write(path)
 
             return func(*args, **kwargs)
 

core/dbt/cli/types.py

@@ -8,6 +8,7 @@ class Command(Enum):
     BUILD = "build"
     CLEAN = "clean"
     COMPILE = "compile"
+    CLONE = "clone"
     DOCS_GENERATE = "generate"
     DOCS_SERVE = "serve"
     DEBUG = "debug"

core/dbt/compilation.py

@@ -48,9 +48,10 @@ def print_compile_stats(stats):
         NodeType.Analysis: "analysis",
         NodeType.Macro: "macro",
         NodeType.Operation: "operation",
-        NodeType.Seed: "seed file",
+        NodeType.Seed: "seed",
         NodeType.Source: "source",
         NodeType.Exposure: "exposure",
+        NodeType.SemanticModel: "semantic model",
         NodeType.Metric: "metric",
         NodeType.Group: "group",
     }
@@ -63,7 +64,8 @@ def print_compile_stats(stats):
         resource_counts = {k.pluralize(): v for k, v in results.items()}
         dbt.tracking.track_resource_counts(resource_counts)
 
-    stat_line = ", ".join([pluralize(ct, names.get(t)) for t, ct in results.items() if t in names])
+    # do not include resource types that are not actually defined in the project
+    stat_line = ", ".join([pluralize(ct, names.get(t)) for t, ct in stats.items() if t in names])
 
     fire_event(FoundStats(stat_line=stat_line))
 
@@ -82,16 +84,16 @@ def _generate_stats(manifest: Manifest):
         if _node_enabled(node):
             stats[node.resource_type] += 1
 
-    for source in manifest.sources.values():
-        stats[source.resource_type] += 1
-    for exposure in manifest.exposures.values():
-        stats[exposure.resource_type] += 1
-    for metric in manifest.metrics.values():
-        stats[metric.resource_type] += 1
-    for macro in manifest.macros.values():
-        stats[macro.resource_type] += 1
-    for group in manifest.groups.values():
-        stats[group.resource_type] += 1
+    # Disabled nodes don't appear in the following collections, so we don't check.
+    stats[NodeType.Source] += len(manifest.sources)
+    stats[NodeType.Exposure] += len(manifest.exposures)
+    stats[NodeType.Metric] += len(manifest.metrics)
+    stats[NodeType.Macro] += len(manifest.macros)
+    stats[NodeType.Group] += len(manifest.groups)
+    stats[NodeType.SemanticModel] += len(manifest.semantic_models)
+
+    # TODO: should we be counting dimensions + entities?
+
     return stats
 
 
@@ -173,6 +175,8 @@ class Linker:
                 self.dependency(node.unique_id, (manifest.sources[dependency].unique_id))
             elif dependency in manifest.metrics:
                 self.dependency(node.unique_id, (manifest.metrics[dependency].unique_id))
+            elif dependency in manifest.semantic_models:
+                self.dependency(node.unique_id, (manifest.semantic_models[dependency].unique_id))
             else:
                 raise GraphDependencyNotFoundError(node, dependency)
 
@@ -181,7 +185,6 @@ class Linker:
             self.add_node(source.unique_id)
         for semantic_model in manifest.semantic_models.values():
             self.add_node(semantic_model.unique_id)
-
         for node in manifest.nodes.values():
             self.link_node(node, manifest)
         for exposure in manifest.exposures.values():
@@ -301,62 +304,6 @@ class Compiler:
         relation_cls = adapter.Relation
         return relation_cls.add_ephemeral_prefix(name)
 
-    def _inject_ctes_into_sql(self, sql: str, ctes: List[InjectedCTE]) -> str:
-        """
-        `ctes` is a list of InjectedCTEs like:
-
-            [
-                InjectedCTE(
-                    id="cte_id_1",
-                    sql="__dbt__cte__ephemeral as (select * from table)",
-                ),
-                InjectedCTE(
-                    id="cte_id_2",
-                    sql="__dbt__cte__events as (select id, type from events)",
-                ),
-            ]
-
-        Given `sql` like:
-
-          "with internal_cte as (select * from sessions)
-           select * from internal_cte"
-
-        This will spit out:
-
-          "with __dbt__cte__ephemeral as (select * from table),
-                __dbt__cte__events as (select id, type from events),
-                with internal_cte as (select * from sessions)
-           select * from internal_cte"
-
-        (Whitespace enhanced for readability.)
-        """
-        if len(ctes) == 0:
-            return sql
-
-        parsed_stmts = sqlparse.parse(sql)
-        parsed = parsed_stmts[0]
-
-        with_stmt = None
-        for token in parsed.tokens:
-            if token.is_keyword and token.normalized == "WITH":
-                with_stmt = token
-                break
-
-        if with_stmt is None:
-            # no with stmt, add one, and inject CTEs right at the beginning
-            first_token = parsed.token_first()
-            with_stmt = sqlparse.sql.Token(sqlparse.tokens.Keyword, "with")
-            parsed.insert_before(first_token, with_stmt)
-        else:
-            # stmt exists, add a comma (which will come after injected CTEs)
-            trailing_comma = sqlparse.sql.Token(sqlparse.tokens.Punctuation, ",")
-            parsed.insert_after(with_stmt, trailing_comma)
-
-        token = sqlparse.sql.Token(sqlparse.tokens.Keyword, ", ".join(c.sql for c in ctes))
-        parsed.insert_after(with_stmt, token)
-
-        return str(parsed)
-
     def _recursively_prepend_ctes(
         self,
         model: ManifestSQLNode,
@@ -431,7 +378,7 @@ class Compiler:
 
             _add_prepended_cte(prepended_ctes, InjectedCTE(id=cte.id, sql=sql))
 
-        injected_sql = self._inject_ctes_into_sql(
+        injected_sql = inject_ctes_into_sql(
             model.compiled_code,
             prepended_ctes,
         )
@@ -582,3 +529,74 @@ class Compiler:
         if write:
             self._write_node(node)
         return node
+
+
+def inject_ctes_into_sql(sql: str, ctes: List[InjectedCTE]) -> str:
+    """
+    `ctes` is a list of InjectedCTEs like:
+
+        [
+            InjectedCTE(
+                id="cte_id_1",
+                sql="__dbt__cte__ephemeral as (select * from table)",
+            ),
+            InjectedCTE(
+                id="cte_id_2",
+                sql="__dbt__cte__events as (select id, type from events)",
+            ),
+        ]
+
+    Given `sql` like:
+
+      "with internal_cte as (select * from sessions)
+       select * from internal_cte"
+
+    This will spit out:
+
+      "with __dbt__cte__ephemeral as (select * from table),
+            __dbt__cte__events as (select id, type from events),
+            internal_cte as (select * from sessions)
+       select * from internal_cte"
+
+    (Whitespace enhanced for readability.)
+    """
+    if len(ctes) == 0:
+        return sql
+
+    parsed_stmts = sqlparse.parse(sql)
+    parsed = parsed_stmts[0]
+
+    with_stmt = None
+    for token in parsed.tokens:
+        if token.is_keyword and token.normalized == "WITH":
+            with_stmt = token
+        elif token.is_keyword and token.normalized == "RECURSIVE" and with_stmt is not None:
+            with_stmt = token
+            break
+        elif not token.is_whitespace and with_stmt is not None:
+            break
+
+    if with_stmt is None:
+        # no with stmt, add one, and inject CTEs right at the beginning
+        # [original_sql]
+        first_token = parsed.token_first()
+        with_token = sqlparse.sql.Token(sqlparse.tokens.Keyword, "with")
+        parsed.insert_before(first_token, with_token)
+        # [with][original_sql]
+        injected_ctes = ", ".join(c.sql for c in ctes) + " "
+        injected_ctes_token = sqlparse.sql.Token(sqlparse.tokens.Keyword, injected_ctes)
+        parsed.insert_after(with_token, injected_ctes_token)
+        # [with][joined_ctes][original_sql]
+    else:
+        # with stmt exists so we don't need to add one, but we do need to add a comma
+        # between the injected ctes and the original sql
+        # [with][original_sql]
+        injected_ctes = ", ".join(c.sql for c in ctes)
+        injected_ctes_token = sqlparse.sql.Token(sqlparse.tokens.Keyword, injected_ctes)
+        parsed.insert_after(with_stmt, injected_ctes_token)
+        # [with][joined_ctes][original_sql]
+        comma_token = sqlparse.sql.Token(sqlparse.tokens.Punctuation, ", ")
+        parsed.insert_after(injected_ctes_token, comma_token)
+        # [with][joined_ctes][, ][original_sql]
+
+    return str(parsed)

core/dbt/config/project.py

@@ -39,7 +39,6 @@ from dbt.contracts.project import (
     SemverString,
 )
 from dbt.contracts.project import PackageConfig, ProjectPackageMetadata
-from dbt.contracts.publication import ProjectDependencies
 from dbt.dataclass_schema import ValidationError
 from .renderer import DbtProjectYamlRenderer, PackageRenderer
 from .selectors import (
@@ -115,16 +114,13 @@ def package_and_project_data_from_root(project_root):
 
     packages_specified_path = PACKAGES_FILE_NAME
     packages_dict = {}
-    dependent_projects_dict = {}
     if "packages" in dependencies_yml_dict:
         packages_dict["packages"] = dependencies_yml_dict["packages"]
         packages_specified_path = DEPENDENCIES_FILE_NAME
     else:  # don't check for "packages" here so we capture invalid keys in packages.yml
         packages_dict = packages_yml_dict
-    if "projects" in dependencies_yml_dict:
-        dependent_projects_dict["projects"] = dependencies_yml_dict["projects"]
 
-    return packages_dict, dependent_projects_dict, packages_specified_path
+    return packages_dict, packages_specified_path
 
 
 def package_config_from_data(packages_data: Dict[str, Any]) -> PackageConfig:
@@ -139,21 +135,6 @@ def package_config_from_data(packages_data: Dict[str, Any]) -> PackageConfig:
     return packages
 
 
-def dependent_project_config_from_data(
-    dependent_projects_data: Dict[str, Any]
-) -> ProjectDependencies:
-    if not dependent_projects_data:
-        dependent_projects_data = {"projects": []}
-
-    try:
-        ProjectDependencies.validate(dependent_projects_data)
-        dependent_projects = ProjectDependencies.from_dict(dependent_projects_data)
-    except ValidationError as e:
-        msg = f"Malformed dependencies.yml: {e}"
-        raise DbtProjectError(msg)
-    return dependent_projects
-
-
 def _parse_versions(versions: Union[List[str], str]) -> List[VersionSpecifier]:
     """Parse multiple versions as read from disk. The versions value may be any
     one of:
@@ -278,9 +259,6 @@ def _get_required_version(
 class RenderComponents:
     project_dict: Dict[str, Any] = field(metadata=dict(description="The project dictionary"))
     packages_dict: Dict[str, Any] = field(metadata=dict(description="The packages dictionary"))
-    dependent_projects_dict: Dict[str, Any] = field(
-        metadata=dict(description="The dependent projects dictionary")
-    )
     selectors_dict: Dict[str, Any] = field(metadata=dict(description="The selectors dictionary"))
 
 
@@ -321,15 +299,11 @@ class PartialProject(RenderComponents):
         rendered_packages = renderer.render_packages(
             self.packages_dict, self.packages_specified_path
         )
-        rendered_dependent_projects = renderer.render_dependent_projects(
-            self.dependent_projects_dict
-        )
         rendered_selectors = renderer.render_selectors(self.selectors_dict)
 
         return RenderComponents(
             project_dict=rendered_project,
             packages_dict=rendered_packages,
-            dependent_projects_dict=rendered_dependent_projects,
             selectors_dict=rendered_selectors,
         )
 
@@ -376,7 +350,6 @@ class PartialProject(RenderComponents):
         unrendered = RenderComponents(
             project_dict=self.project_dict,
             packages_dict=self.packages_dict,
-            dependent_projects_dict=self.dependent_projects_dict,
             selectors_dict=self.selectors_dict,
         )
         dbt_version = _get_required_version(
@@ -478,9 +451,6 @@ class PartialProject(RenderComponents):
         query_comment = _query_comment_from_cfg(cfg.query_comment)
 
         packages: PackageConfig = package_config_from_data(rendered.packages_dict)
-        dependent_projects: ProjectDependencies = dependent_project_config_from_data(
-            rendered.dependent_projects_dict
-        )
         selectors = selector_config_from_data(rendered.selectors_dict)
         manifest_selectors: Dict[str, Any] = {}
         if rendered.selectors_dict and rendered.selectors_dict["selectors"]:
@@ -516,7 +486,6 @@ class PartialProject(RenderComponents):
             snapshots=snapshots,
             dbt_version=dbt_version,
             packages=packages,
-            dependent_projects=dependent_projects,
             manifest_selectors=manifest_selectors,
             selectors=selectors,
             query_comment=query_comment,
@@ -528,6 +497,7 @@ class PartialProject(RenderComponents):
             config_version=cfg.config_version,
             unrendered=unrendered,
             project_env_vars=project_env_vars,
+            restrict_access=cfg.restrict_access,
         )
         # sanity check - this means an internal issue
         project.validate()
@@ -539,7 +509,6 @@ class PartialProject(RenderComponents):
         project_root: str,
         project_dict: Dict[str, Any],
         packages_dict: Dict[str, Any],
-        dependent_projects_dict: Dict[str, Any],
         selectors_dict: Dict[str, Any],
         *,
         verify_version: bool = False,
@@ -556,7 +525,6 @@ class PartialProject(RenderComponents):
             project_root=project_root,
             project_dict=project_dict,
             packages_dict=packages_dict,
-            dependent_projects_dict=dependent_projects_dict,
             selectors_dict=selectors_dict,
             verify_version=verify_version,
             packages_specified_path=packages_specified_path,
@@ -570,7 +538,6 @@ class PartialProject(RenderComponents):
         project_dict = load_raw_project(project_root)
         (
             packages_dict,
-            dependent_projects_dict,
             packages_specified_path,
         ) = package_and_project_data_from_root(project_root)
         selectors_dict = selector_data_from_root(project_root)
@@ -579,7 +546,6 @@ class PartialProject(RenderComponents):
             project_dict=project_dict,
             selectors_dict=selectors_dict,
             packages_dict=packages_dict,
-            dependent_projects_dict=dependent_projects_dict,
             verify_version=verify_version,
             packages_specified_path=packages_specified_path,
         )
@@ -636,13 +602,13 @@ class Project:
     vars: VarProvider
     dbt_version: List[VersionSpecifier]
     packages: PackageConfig
-    dependent_projects: ProjectDependencies
     manifest_selectors: Dict[str, Any]
     selectors: SelectorConfig
     query_comment: QueryComment
     config_version: int
     unrendered: RenderComponents
     project_env_vars: Dict[str, Any]
+    restrict_access: bool
 
     @property
     def all_source_paths(self) -> List[str]:
@@ -711,6 +677,7 @@ class Project:
                 "vars": self.vars.to_dict(),
                 "require-dbt-version": [v.to_version_string() for v in self.dbt_version],
                 "config-version": self.config_version,
+                "restrict-access": self.restrict_access,
             }
         )
         if self.query_comment:
@@ -727,13 +694,6 @@ class Project:
         except ValidationError as e:
             raise ProjectContractBrokenError(e) from e
 
-    @classmethod
-    def partial_load(cls, project_root: str, *, verify_version: bool = False) -> PartialProject:
-        return PartialProject.from_project_root(
-            project_root,
-            verify_version=verify_version,
-        )
-
     # Called by:
     # RtConfig.load_dependencies => RtConfig.load_projects => RtConfig.new_project => Project.from_project_root
     # RtConfig.from_args => RtConfig.collect_parts => load_project => Project.from_project_root

core/dbt/config/renderer.py

@@ -142,10 +142,6 @@ class DbtProjectYamlRenderer(BaseRenderer):
         else:
             return package_renderer.render_data(packages)
 
-    def render_dependent_projects(self, dependent_projects: Dict[str, Any]):
-        """This is a no-op to maintain regularity in the interfaces. We don't render dependencies.yml."""
-        return dependent_projects
-
     def render_selectors(self, selectors: Dict[str, Any]):
         return self.render_data(selectors)
 

core/dbt/config/runtime.py

@@ -161,7 +161,6 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
             snapshots=project.snapshots,
             dbt_version=project.dbt_version,
             packages=project.packages,
-            dependent_projects=project.dependent_projects,
             manifest_selectors=project.manifest_selectors,
             selectors=project.selectors,
             query_comment=project.query_comment,
@@ -173,6 +172,7 @@ class RuntimeConfig(Project, Profile, AdapterRequiredConfig):
             config_version=project.config_version,
             unrendered=project.unrendered,
             project_env_vars=project.project_env_vars,
+            restrict_access=project.restrict_access,
             profile_env_vars=profile.profile_env_vars,
             profile_name=profile.profile_name,
             target_name=profile.target_name,

core/dbt/context/base.py

@@ -1,6 +1,7 @@
 import json
 import os
 from typing import Any, Dict, NoReturn, Optional, Mapping, Iterable, Set, List
+import threading
 
 from dbt.flags import get_flags
 import dbt.flags as flags_module
@@ -596,6 +597,11 @@ class BaseContext(metaclass=ContextMeta):
         """
         return get_invocation_id()
 
+    @contextproperty
+    def thread_id(self) -> str:
+        """thread_id outputs an ID for the current thread (useful for auditing)"""
+        return threading.current_thread().name
+
     @contextproperty
     def modules(self) -> Dict[str, Any]:
         """The `modules` variable in the Jinja context contains useful Python

core/dbt/context/providers.py

@@ -133,6 +133,25 @@ class BaseDatabaseWrapper:
         search_prefixes = get_adapter_type_names(self._adapter.type()) + ["default"]
         return search_prefixes
 
+    def _get_search_packages(self, namespace: Optional[str] = None) -> List[Optional[str]]:
+        search_packages: List[Optional[str]] = [None]
+
+        if namespace is None:
+            search_packages = [None]
+        elif isinstance(namespace, str):
+            macro_search_order = self._adapter.config.get_macro_search_order(namespace)
+            if macro_search_order:
+                search_packages = macro_search_order
+            elif not macro_search_order and namespace in self._adapter.config.dependencies:
+                search_packages = [self.config.project_name, namespace]
+        else:
+            raise CompilationError(
+                f"In adapter.dispatch, got a {type(namespace)} macro_namespace argument "
+                f'("{namespace}"), but macro_namespace should be None or a string.'
+            )
+
+        return search_packages
+
     def dispatch(
         self,
         macro_name: str,
@@ -154,20 +173,7 @@ class BaseDatabaseWrapper:
         if packages is not None:
             raise MacroDispatchArgError(macro_name)
 
-        namespace = macro_namespace
-
-        if namespace is None:
-            search_packages = [None]
-        elif isinstance(namespace, str):
-            search_packages = self._adapter.config.get_macro_search_order(namespace)
-            if not search_packages and namespace in self._adapter.config.dependencies:
-                search_packages = [self.config.project_name, namespace]
-        else:
-            # Not a string and not None so must be a list
-            raise CompilationError(
-                f"In adapter.dispatch, got a list macro_namespace argument "
-                f'("{macro_namespace}"), but macro_namespace should be None or a string.'
-            )
+        search_packages = self._get_search_packages(macro_namespace)
 
         attempts = []
 
@@ -191,7 +197,7 @@ class BaseDatabaseWrapper:
                     return macro
 
         searched = ", ".join(repr(a) for a in attempts)
-        msg = f"In dispatch: No macro named '{macro_name}' found\n    Searched for: {searched}"
+        msg = f"In dispatch: No macro named '{macro_name}' found within namespace: '{macro_namespace}'\n    Searched for: {searched}"
         raise CompilationError(msg)
 
 
@@ -500,19 +506,24 @@ class RuntimeRefResolver(BaseRefResolver):
                 target_version=target_version,
                 disabled=isinstance(target_model, Disabled),
             )
-        elif (
-            target_model.resource_type == NodeType.Model
-            and target_model.access == AccessType.Private
-            # don't raise this reference error for ad hoc 'preview' queries
-            and self.model.resource_type != NodeType.SqlOperation
-            and self.model.resource_type != NodeType.RPCCall  # TODO: rm
+        elif self.manifest.is_invalid_private_ref(
+            self.model, target_model, self.config.dependencies
         ):
-            if not self.model.group or self.model.group != target_model.group:
-                raise DbtReferenceError(
-                    unique_id=self.model.unique_id,
-                    ref_unique_id=target_model.unique_id,
-                    group=cast_to_str(target_model.group),
-                )
+            raise DbtReferenceError(
+                unique_id=self.model.unique_id,
+                ref_unique_id=target_model.unique_id,
+                access=AccessType.Private,
+                scope=cast_to_str(target_model.group),
+            )
+        elif self.manifest.is_invalid_protected_ref(
+            self.model, target_model, self.config.dependencies
+        ):
+            raise DbtReferenceError(
+                unique_id=self.model.unique_id,
+                ref_unique_id=target_model.unique_id,
+                access=AccessType.Protected,
+                scope=target_model.package_name,
+            )
 
         self.validate(target_model, target_name, target_package, target_version)
         return self.create_relation(target_model)
@@ -1367,20 +1378,30 @@ class ModelContext(ProviderContext):
     @contextproperty
     def sql(self) -> Optional[str]:
         # only doing this in sql model for backward compatible
-        if (
-            getattr(self.model, "extra_ctes_injected", None)
-            and self.model.language == ModelLanguage.sql  # type: ignore[union-attr]
-        ):
-            # TODO CT-211
-            return self.model.compiled_code  # type: ignore[union-attr]
-        return None
+        if self.model.language == ModelLanguage.sql:  # type: ignore[union-attr]
+            # If the model is deferred and the adapter doesn't support zero-copy cloning, then select * from the prod
+            # relation
+            if getattr(self.model, "defer_relation", None):
+                # TODO https://github.com/dbt-labs/dbt-core/issues/7976
+                return f"select * from {self.model.defer_relation.relation_name or str(self.defer_relation)}"  # type: ignore[union-attr]
+            elif getattr(self.model, "extra_ctes_injected", None):
+                # TODO CT-211
+                return self.model.compiled_code  # type: ignore[union-attr]
+            else:
+                return None
+        else:
+            return None
 
     @contextproperty
     def compiled_code(self) -> Optional[str]:
-        if getattr(self.model, "extra_ctes_injected", None):
+        if getattr(self.model, "defer_relation", None):
+            # TODO https://github.com/dbt-labs/dbt-core/issues/7976
+            return f"select * from {self.model.defer_relation.relation_name or str(self.defer_relation)}"  # type: ignore[union-attr]
+        elif getattr(self.model, "extra_ctes_injected", None):
             # TODO CT-211
             return self.model.compiled_code  # type: ignore[union-attr]
-        return None
+        else:
+            return None
 
     @contextproperty
     def database(self) -> str:
@@ -1425,6 +1446,20 @@ class ModelContext(ProviderContext):
             return None
         return self.db_wrapper.Relation.create_from(self.config, self.model)
 
+    @contextproperty
+    def defer_relation(self) -> Optional[RelationProxy]:
+        """
+        For commands which add information about this node's corresponding
+        production version (via a --state artifact), access the Relation
+        object for that stateful other
+        """
+        if getattr(self.model, "defer_relation", None):
+            return self.db_wrapper.Relation.create_from_node(
+                self.config, self.model.defer_relation  # type: ignore
+            )
+        else:
+            return None
+
 
 # This is called by '_context_for', used in 'render_with_context'
 def generate_parser_model_context(

core/dbt/contracts/graph/manifest.py

@@ -1,9 +1,11 @@
 import enum
+from collections import defaultdict
 from dataclasses import dataclass, field
 from itertools import chain, islice
 from mashumaro.mixins.msgpack import DataClassMessagePackMixin
 from multiprocessing.synchronize import Lock
 from typing import (
+    DefaultDict,
     Dict,
     List,
     Optional,
@@ -22,8 +24,6 @@ from typing import (
 from typing_extensions import Protocol
 from uuid import UUID
 
-from dbt.contracts.publication import PublicationConfig
-
 from dbt.contracts.graph.nodes import (
     BaseNode,
     Documentation,
@@ -35,7 +35,7 @@ from dbt.contracts.graph.nodes import (
     ManifestNode,
     Metric,
     ModelNode,
-    RelationalNode,
+    DeferRelation,
     ResultNode,
     SemanticModel,
     SourceDefinition,
@@ -57,13 +57,11 @@ from dbt.helper_types import PathSet
 from dbt.events.functions import fire_event
 from dbt.events.types import MergedFromState, UnpinnedRefNewVersionAvailable
 from dbt.events.contextvars import get_node_info
-from dbt.node_types import NodeType
+from dbt.node_types import NodeType, AccessType
 from dbt.flags import get_flags, MP_CONTEXT
 from dbt import tracking
 import dbt.utils
-from dbt_semantic_interfaces.implementations.metric import PydanticMetric
-from dbt_semantic_interfaces.implementations.semantic_manifest import PydanticSemanticManifest
-from dbt_semantic_interfaces.implementations.semantic_model import PydanticSemanticModel
+
 
 NodeEdgeMap = Dict[str, List[str]]
 PackageName = str
@@ -301,6 +299,49 @@ class MetricLookup(dbtClassMixin):
         return manifest.metrics[unique_id]
 
 
+class SemanticModelByMeasureLookup(dbtClassMixin):
+    """Lookup utility for finding SemanticModel by measure
+
+    This is possible because measure names are supposed to be unique across
+    the semantic models in a manifest.
+    """
+
+    def __init__(self, manifest: "Manifest"):
+        self.storage: DefaultDict[str, Dict[PackageName, UniqueID]] = defaultdict(dict)
+        self.populate(manifest)
+
+    def get_unique_id(self, search_name: str, package: Optional[PackageName]):
+        return find_unique_id_for_package(self.storage, search_name, package)
+
+    def find(
+        self, search_name: str, package: Optional[PackageName], manifest: "Manifest"
+    ) -> Optional[SemanticModel]:
+        """Tries to find a SemanticModel based on a measure name"""
+        unique_id = self.get_unique_id(search_name, package)
+        if unique_id is not None:
+            return self.perform_lookup(unique_id, manifest)
+        return None
+
+    def add(self, semantic_model: SemanticModel):
+        """Sets all measures for a SemanticModel as paths to the SemanticModel's `unique_id`"""
+        for measure in semantic_model.measures:
+            self.storage[measure.name][semantic_model.package_name] = semantic_model.unique_id
+
+    def populate(self, manifest: "Manifest"):
+        """Populate storage with all the measure + package paths to the Manifest's SemanticModels"""
+        for semantic_model in manifest.semantic_models.values():
+            self.add(semantic_model=semantic_model)
+
+    def perform_lookup(self, unique_id: UniqueID, manifest: "Manifest") -> SemanticModel:
+        """Tries to get a SemanticModel from the Manifest"""
+        semantic_model = manifest.semantic_models.get(unique_id)
+        if semantic_model is None:
+            raise dbt.exceptions.DbtInternalError(
+                f"Semantic model `{unique_id}` found in cache but not found in manifest"
+            )
+        return semantic_model
+
+
 # This handles both models/seeds/snapshots and sources/metrics/exposures
 class DisabledLookup(dbtClassMixin):
     def __init__(self, manifest: "Manifest"):
@@ -700,7 +741,6 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
     source_patches: MutableMapping[SourceKey, SourcePatch] = field(default_factory=dict)
     disabled: MutableMapping[str, List[GraphMemberNode]] = field(default_factory=dict)
     env_vars: MutableMapping[str, str] = field(default_factory=dict)
-    publications: MutableMapping[str, PublicationConfig] = field(default_factory=dict)
     semantic_models: MutableMapping[str, SemanticModel] = field(default_factory=dict)
 
     _doc_lookup: Optional[DocLookup] = field(
@@ -715,6 +755,9 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
     _metric_lookup: Optional[MetricLookup] = field(
         default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}
     )
+    _semantic_model_by_measure_lookup: Optional[SemanticModelByMeasureLookup] = field(
+        default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}
+    )
     _disabled_lookup: Optional[DisabledLookup] = field(
         default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}
     )
@@ -851,7 +894,6 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             disabled={k: _deepcopy(v) for k, v in self.disabled.items()},
             files={k: _deepcopy(v) for k, v in self.files.items()},
             state_check=_deepcopy(self.state_check),
-            publications={k: _deepcopy(v) for k, v in self.publications.items()},
             semantic_models={k: _deepcopy(v) for k, v in self.semantic_models.items()},
         )
         copy.build_flat_graph()
@@ -966,6 +1008,13 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             self._metric_lookup = MetricLookup(self)
         return self._metric_lookup
 
+    @property
+    def semantic_model_by_measure_lookup(self) -> SemanticModelByMeasureLookup:
+        """Gets (and creates if necessary) the lookup utility for getting SemanticModels by measures"""
+        if self._semantic_model_by_measure_lookup is None:
+            self._semantic_model_by_measure_lookup = SemanticModelByMeasureLookup(self)
+        return self._semantic_model_by_measure_lookup
+
     def rebuild_ref_lookup(self):
         self._ref_lookup = RefableLookup(self)
 
@@ -984,20 +1033,6 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             self._analysis_lookup = AnalysisLookup(self)
         return self._analysis_lookup
 
-    @property
-    def pydantic_semantic_manifest(self) -> PydanticSemanticManifest:
-        pydantic_semantic_manifest = PydanticSemanticManifest(metrics=[], semantic_models=[])
-
-        for semantic_model in self.semantic_models.values():
-            pydantic_semantic_manifest.semantic_models.append(
-                PydanticSemanticModel.parse_obj(semantic_model.to_dict())
-            )
-
-        for metric in self.metrics.values():
-            pydantic_semantic_manifest.metrics.append(PydanticMetric.parse_obj(metric.to_dict()))
-
-        return pydantic_semantic_manifest
-
     @property
     def external_node_unique_ids(self):
         return [node.unique_id for node in self.nodes.values() if node.is_external_node]
@@ -1107,6 +1142,25 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             return Disabled(disabled[0])
         return None
 
+    def resolve_semantic_model_for_measure(
+        self,
+        target_measure_name: str,
+        current_project: str,
+        node_package: str,
+        target_package: Optional[str] = None,
+    ) -> Optional[SemanticModel]:
+        """Tries to find the SemanticModel that a measure belongs to"""
+        candidates = _packages_to_search(current_project, node_package, target_package)
+
+        for pkg in candidates:
+            semantic_model = self.semantic_model_by_measure_lookup.find(
+                target_measure_name, pkg, self
+            )
+            if semantic_model is not None:
+                return semantic_model
+
+        return None
+
     # Called by DocsRuntimeContext.doc
     def resolve_doc(
         self,
@@ -1127,6 +1181,50 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
                 return result
         return None
 
+    def is_invalid_private_ref(
+        self, node: GraphMemberNode, target_model: MaybeNonSource, dependencies: Optional[Mapping]
+    ) -> bool:
+        dependencies = dependencies or {}
+        if not isinstance(target_model, ModelNode):
+            return False
+
+        is_private_ref = (
+            target_model.access == AccessType.Private
+            # don't raise this reference error for ad hoc 'preview' queries
+            and node.resource_type != NodeType.SqlOperation
+            and node.resource_type != NodeType.RPCCall  # TODO: rm
+        )
+        target_dependency = dependencies.get(target_model.package_name)
+        restrict_package_access = target_dependency.restrict_access if target_dependency else False
+
+        # TODO: SemanticModel and SourceDefinition do not have group, and so should not be able to make _any_ private ref.
+        return is_private_ref and (
+            not hasattr(node, "group")
+            or not node.group
+            or node.group != target_model.group
+            or restrict_package_access
+        )
+
+    def is_invalid_protected_ref(
+        self, node: GraphMemberNode, target_model: MaybeNonSource, dependencies: Optional[Mapping]
+    ) -> bool:
+        dependencies = dependencies or {}
+        if not isinstance(target_model, ModelNode):
+            return False
+
+        is_protected_ref = (
+            target_model.access == AccessType.Protected
+            # don't raise this reference error for ad hoc 'preview' queries
+            and node.resource_type != NodeType.SqlOperation
+            and node.resource_type != NodeType.RPCCall  # TODO: rm
+        )
+        target_dependency = dependencies.get(target_model.package_name)
+        restrict_package_access = target_dependency.restrict_access if target_dependency else False
+
+        return is_protected_ref and (
+            node.package_name != target_model.package_name and restrict_package_access
+        )
+
     # Called by RunTask.defer_to_manifest
     def merge_from_artifact(
         self,
@@ -1178,8 +1276,10 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
         for unique_id, node in other.nodes.items():
             current = self.nodes.get(unique_id)
             if current and (node.resource_type in refables and not node.is_ephemeral):
-                state_relation = RelationalNode(node.database, node.schema, node.alias)
-                self.nodes[unique_id] = current.replace(state_relation=state_relation)
+                defer_relation = DeferRelation(
+                    node.database, node.schema, node.alias, node.relation_name
+                )
+                self.nodes[unique_id] = current.replace(defer_relation=defer_relation)
 
     # Methods that were formerly in ParseResult
 
@@ -1297,12 +1397,12 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
             self.source_patches,
             self.disabled,
             self.env_vars,
-            self.publications,
             self.semantic_models,
             self._doc_lookup,
             self._source_lookup,
             self._ref_lookup,
             self._metric_lookup,
+            self._semantic_model_by_measure_lookup,
             self._disabled_lookup,
             self._analysis_lookup,
         )
@@ -1401,8 +1501,8 @@ class WritableManifest(ArtifactMixin):
         for unique_id, node in dct["nodes"].items():
             if "config_call_dict" in node:
                 del node["config_call_dict"]
-            if "state_relation" in node:
-                del node["state_relation"]
+            if "defer_relation" in node:
+                del node["defer_relation"]
         return dct
 
 

core/dbt/contracts/graph/model_config.py

@@ -386,6 +386,11 @@ class BaseConfig(AdditionalPropertiesAllowed, Replaceable):
         return self.from_dict(dct)
 
 
+@dataclass
+class SemanticModelConfig(BaseConfig):
+    enabled: bool = True
+
+
 @dataclass
 class MetricConfig(BaseConfig):
     enabled: bool = True
@@ -495,12 +500,12 @@ class NodeConfig(NodeAndTestConfig):
         if (
             self.contract.enforced
             and self.materialized == "incremental"
-            and self.on_schema_change != "append_new_columns"
+            and self.on_schema_change not in ("append_new_columns", "fail")
         ):
             raise ValidationError(
                 f"Invalid value for on_schema_change: {self.on_schema_change}. Models "
                 "materialized as incremental with contracts enabled must set "
-                "on_schema_change to 'append_new_columns'"
+                "on_schema_change to 'append_new_columns' or 'fail'"
             )
 
     @classmethod
@@ -550,6 +555,8 @@ class SeedConfig(NodeConfig):
 
 @dataclass
 class TestConfig(NodeAndTestConfig):
+    __test__ = False
+
     # this is repeated because of a different default
     schema: Optional[str] = field(
         default="dbt_test__audit",

core/dbt/contracts/graph/node_args.py

@@ -1,8 +1,9 @@
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Optional
+from typing import Optional, List
 
 from dbt.contracts.graph.unparsed import NodeVersion
+from dbt.node_types import NodeType, AccessType
 
 
 @dataclass
@@ -16,4 +17,15 @@ class ModelNodeArgs:
     version: Optional[NodeVersion] = None
     latest_version: Optional[NodeVersion] = None
     deprecation_date: Optional[datetime] = None
+    access: Optional[str] = AccessType.Protected.value
     generated_at: datetime = field(default_factory=datetime.utcnow)
+    depends_on_nodes: List[str] = field(default_factory=list)
+    enabled: bool = True
+
+    @property
+    def unique_id(self) -> str:
+        unique_id = f"{NodeType.Model}.{self.package_name}.{self.name}"
+        if self.version:
+            unique_id = f"{unique_id}.v{self.version}"
+
+        return unique_id

core/dbt/contracts/graph/nodes.py

@@ -45,16 +45,19 @@ from dbt.events.types import (
     SeedExceedsLimitAndPathChanged,
     SeedExceedsLimitChecksumChanged,
 )
-from dbt.events.contextvars import set_contextvars
+from dbt.events.contextvars import set_log_contextvars
 from dbt.flags import get_flags
 from dbt.node_types import ModelLanguage, NodeType, AccessType
+from dbt_semantic_interfaces.call_parameter_sets import FilterCallParameterSets
 from dbt_semantic_interfaces.references import (
     MeasureReference,
     LinkableElementReference,
     SemanticModelReference,
+    TimeDimensionReference,
 )
 from dbt_semantic_interfaces.references import MetricReference as DSIMetricReference
 from dbt_semantic_interfaces.type_enums import MetricType, TimeGranularity
+from dbt_semantic_interfaces.parsing.where_filter_parser import WhereFilterParser
 
 from .model_config import (
     NodeConfig,
@@ -65,6 +68,7 @@ from .model_config import (
     ExposureConfig,
     EmptySnapshotConfig,
     SnapshotConfig,
+    SemanticModelConfig,
 )
 
 
@@ -258,8 +262,9 @@ class MacroDependsOn(dbtClassMixin, Replaceable):
 
 
 @dataclass
-class RelationalNode(HasRelationMetadata):
+class DeferRelation(HasRelationMetadata):
     alias: str
+    relation_name: Optional[str]
 
     @property
     def identifier(self):
@@ -275,17 +280,6 @@ class DependsOn(MacroDependsOn):
             self.nodes.append(value)
 
 
-@dataclass
-class StateRelation(dbtClassMixin):
-    alias: str
-    database: Optional[str]
-    schema: str
-
-    @property
-    def identifier(self):
-        return self.alias
-
-
 @dataclass
 class ParsedNodeMandatory(GraphNode, HasRelationMetadata, Replaceable):
     alias: str
@@ -327,7 +321,7 @@ class NodeInfoMixin:
     def update_event_status(self, **kwargs):
         for k, v in kwargs.items():
             self._event_status[k] = v
-        set_contextvars(node_info=self.node_info)
+        set_log_contextvars(node_info=self.node_info)
 
     def clear_event_status(self):
         self._event_status = dict()
@@ -577,11 +571,18 @@ class ModelNode(CompiledNode):
     version: Optional[NodeVersion] = None
     latest_version: Optional[NodeVersion] = None
     deprecation_date: Optional[datetime] = None
-    state_relation: Optional[StateRelation] = None
+    defer_relation: Optional[DeferRelation] = None
 
     @classmethod
     def from_args(cls, args: ModelNodeArgs) -> "ModelNode":
-        unique_id = f"{NodeType.Model}.{args.package_name}.{args.name}"
+        unique_id = args.unique_id
+
+        # build unrendered config -- for usage in ParsedNode.same_contents
+        unrendered_config = {}
+        unrendered_config["alias"] = args.identifier
+        unrendered_config["schema"] = args.schema
+        if args.database:
+            unrendered_config["database"] = args.database
 
         return cls(
             resource_type=NodeType.Model,
@@ -597,8 +598,12 @@ class ModelNode(CompiledNode):
             alias=args.identifier,
             deprecation_date=args.deprecation_date,
             checksum=FileHash.from_contents(f"{unique_id},{args.generated_at}"),
+            access=AccessType(args.access),
             original_file_path="",
             path="",
+            unrendered_config=unrendered_config,
+            depends_on=DependsOn(nodes=args.depends_on_nodes),
+            config=NodeConfig(enabled=args.enabled),
         )
 
     @property
@@ -624,6 +629,11 @@ class ModelNode(CompiledNode):
         # We don't need to construct the checksum if the model does not
         # have contract enforced, because it won't be used.
         # This needs to be executed after contract config is set
+
+        # Avoid rebuilding the checksum if it has already been set.
+        if self.contract.checksum is not None:
+            return
+
         if self.contract.enforced is True:
             contract_state = ""
             # We need to sort the columns so that order doesn't matter
@@ -785,7 +795,7 @@ class SeedNode(ParsedNode):  # No SQLDefaults!
     # and we need the root_path to load the seed later
     root_path: Optional[str] = None
     depends_on: MacroDependsOn = field(default_factory=MacroDependsOn)
-    state_relation: Optional[StateRelation] = None
+    defer_relation: Optional[DeferRelation] = None
 
     def same_seeds(self, other: "SeedNode") -> bool:
         # for seeds, we check the hashes. If the hashes are different types,
@@ -922,6 +932,8 @@ class SingularTestNode(TestShouldStoreFailures, CompiledNode):
 
 @dataclass
 class TestMetadata(dbtClassMixin, Replaceable):
+    __test__ = False
+
     name: str
     # kwargs are the args that are left in the test builder after
     # removing configs. They are set from the test builder when
@@ -980,7 +992,7 @@ class IntermediateSnapshotNode(CompiledNode):
 class SnapshotNode(CompiledNode):
     resource_type: NodeType = field(metadata={"restrict": [NodeType.Snapshot]})
     config: SnapshotConfig
-    state_relation: Optional[StateRelation] = None
+    defer_relation: Optional[DeferRelation] = None
 
 
 # ====================================
@@ -1307,6 +1319,10 @@ class Exposure(GraphNode):
 class WhereFilter(dbtClassMixin):
     where_sql_template: str
 
+    @property
+    def call_parameter_sets(self) -> FilterCallParameterSets:
+        return WhereFilterParser.parse_call_parameter_sets(self.where_sql_template)
+
 
 @dataclass
 class MetricInputMeasure(dbtClassMixin):
@@ -1481,6 +1497,7 @@ class SemanticModel(GraphNode):
     depends_on: DependsOn = field(default_factory=DependsOn)
     refs: List[RefArgs] = field(default_factory=list)
     created_at: float = field(default_factory=lambda: time.time())
+    config: SemanticModelConfig = field(default_factory=SemanticModelConfig)
 
     @property
     def entity_references(self) -> List[LinkableElementReference]:
@@ -1539,6 +1556,29 @@ class SemanticModel(GraphNode):
     def depends_on_macros(self):
         return self.depends_on.macros
 
+    def checked_agg_time_dimension_for_measure(
+        self, measure_reference: MeasureReference
+    ) -> TimeDimensionReference:
+        measure: Optional[Measure] = None
+        for measure in self.measures:
+            if measure.reference == measure_reference:
+                measure = measure
+
+        assert (
+            measure is not None
+        ), f"No measure with name ({measure_reference.element_name}) in semantic_model with name ({self.name})"
+
+        if self.defaults is not None:
+            default_agg_time_dimesion = self.defaults.agg_time_dimension
+
+        agg_time_dimension_name = measure.agg_time_dimension or default_agg_time_dimesion
+        assert agg_time_dimension_name is not None, (
+            f"Aggregation time dimension for measure {measure.name} is not set! This should either be set directly on "
+            f"the measure specification in the model, or else defaulted to the primary time dimension in the data "
+            f"source containing the measure."
+        )
+        return TimeDimensionReference(element_name=agg_time_dimension_name)
+
 
 # ====================================
 # Patches

core/dbt/contracts/graph/semantic_manifest.py

@@ -0,0 +1,95 @@
+from dbt_semantic_interfaces.implementations.metric import PydanticMetric
+from dbt_semantic_interfaces.implementations.project_configuration import (
+    PydanticProjectConfiguration,
+)
+from dbt_semantic_interfaces.implementations.semantic_manifest import PydanticSemanticManifest
+from dbt_semantic_interfaces.implementations.semantic_model import PydanticSemanticModel
+from dbt_semantic_interfaces.implementations.time_spine_table_configuration import (
+    PydanticTimeSpineTableConfiguration,
+)
+from dbt_semantic_interfaces.type_enums import TimeGranularity
+from dbt_semantic_interfaces.validations.semantic_manifest_validator import (
+    SemanticManifestValidator,
+)
+
+from dbt.clients.system import write_file
+from dbt.events.base_types import EventLevel
+from dbt.events.functions import fire_event
+from dbt.events.types import SemanticValidationFailure
+from dbt.exceptions import ParsingError
+
+
+class SemanticManifest:
+    def __init__(self, manifest):
+        self.manifest = manifest
+
+    def validate(self) -> bool:
+
+        # TODO: Enforce this check.
+        # if self.manifest.metrics and not self.manifest.semantic_models:
+        #    fire_event(
+        #        SemanticValidationFailure(
+        #            msg="Metrics require semantic models, but none were found."
+        #        ),
+        #        EventLevel.ERROR,
+        #    )
+        #    return False
+
+        if not self.manifest.metrics or not self.manifest.semantic_models:
+            return True
+
+        semantic_manifest = self._get_pydantic_semantic_manifest()
+        validator = SemanticManifestValidator[PydanticSemanticManifest]()
+        validation_results = validator.validate_semantic_manifest(semantic_manifest)
+
+        for warning in validation_results.warnings:
+            fire_event(SemanticValidationFailure(msg=warning.message))
+
+        for error in validation_results.errors:
+            fire_event(SemanticValidationFailure(msg=error.message), EventLevel.ERROR)
+
+        return not validation_results.errors
+
+    def write_json_to_file(self, file_path: str):
+        semantic_manifest = self._get_pydantic_semantic_manifest()
+        json = semantic_manifest.json()
+        write_file(file_path, json)
+
+    def _get_pydantic_semantic_manifest(self) -> PydanticSemanticManifest:
+        project_config = PydanticProjectConfiguration(
+            time_spine_table_configurations=[],
+        )
+        pydantic_semantic_manifest = PydanticSemanticManifest(
+            metrics=[], semantic_models=[], project_configuration=project_config
+        )
+
+        for semantic_model in self.manifest.semantic_models.values():
+            pydantic_semantic_manifest.semantic_models.append(
+                PydanticSemanticModel.parse_obj(semantic_model.to_dict())
+            )
+
+        for metric in self.manifest.metrics.values():
+            pydantic_semantic_manifest.metrics.append(PydanticMetric.parse_obj(metric.to_dict()))
+
+        # Look for time-spine table model and create time spine table configuration
+        if self.manifest.semantic_models:
+            # Get model for time_spine_table
+            time_spine_model_name = "metricflow_time_spine"
+            model = self.manifest.ref_lookup.find(time_spine_model_name, None, None, self.manifest)
+            if not model:
+                raise ParsingError(
+                    "The semantic layer requires a 'metricflow_time_spine' model in the project, but none was found. "
+                    "Guidance on creating this model can be found on our docs site ("
+                    "https://docs.getdbt.com/docs/build/metricflow-time-spine) "
+                )
+            # Create time_spine_table_config, set it in project_config, and add to semantic manifest
+            time_spine_table_config = PydanticTimeSpineTableConfiguration(
+                location=model.relation_name,
+                column_name="date_day",
+                grain=TimeGranularity.DAY,
+            )
+            pydantic_semantic_manifest.project_configuration.time_spine_table_configurations = [
+                time_spine_table_config
+            ]
+
+        return pydantic_semantic_manifest

core/dbt/contracts/graph/semantic_models.py

@@ -120,15 +120,15 @@ class Entity(dbtClassMixin):
 @dataclass
 class MeasureAggregationParameters(dbtClassMixin):
     percentile: Optional[float] = None
-    use_discrete_percentile: Optional[bool] = None
-    use_approximate_percentile: Optional[bool] = None
+    use_discrete_percentile: bool = False
+    use_approximate_percentile: bool = False
 
 
 @dataclass
 class NonAdditiveDimension(dbtClassMixin):
     name: str
     window_choice: AggregationType
-    window_grouples: List[str]
+    window_groupings: List[str]
 
 
 @dataclass
@@ -142,13 +142,6 @@ class Measure(dbtClassMixin):
     non_additive_dimension: Optional[NonAdditiveDimension] = None
     agg_time_dimension: Optional[str] = None
 
-    @property
-    def checked_agg_time_dimension(self) -> TimeDimensionReference:
-        if self.agg_time_dimension is not None:
-            return TimeDimensionReference(element_name=self.agg_time_dimension)
-        else:
-            raise Exception("Measure is missing agg_time_dimension!")
-
     @property
     def reference(self) -> MeasureReference:
         return MeasureReference(element_name=self.name)

core/dbt/contracts/graph/unparsed.py

@@ -618,7 +618,7 @@ class UnparsedMetricTypeParams(dbtClassMixin):
     measure: Optional[Union[UnparsedMetricInputMeasure, str]] = None
     numerator: Optional[Union[UnparsedMetricInput, str]] = None
     denominator: Optional[Union[UnparsedMetricInput, str]] = None
-    expr: Optional[str] = None
+    expr: Optional[Union[str, bool]] = None
     window: Optional[str] = None
     grain_to_date: Optional[str] = None  # str is really a TimeGranularity Enum
     metrics: Optional[List[Union[UnparsedMetricInput, str]]] = None
@@ -689,7 +689,7 @@ class UnparsedEntity(dbtClassMixin):
 class UnparsedNonAdditiveDimension(dbtClassMixin):
     name: str
     window_choice: str  # AggregationType enum
-    window_grouples: List[str]
+    window_groupings: List[str]
 
 
 @dataclass
@@ -697,8 +697,7 @@ class UnparsedMeasure(dbtClassMixin):
     name: str
     agg: str  # actually an enum
     description: Optional[str] = None
-    create_metric: bool = False
-    expr: Optional[str] = None
+    expr: Optional[Union[str, bool, int]] = None
     agg_params: Optional[MeasureAggregationParameters] = None
     non_additive_dimension: Optional[UnparsedNonAdditiveDimension] = None
     agg_time_dimension: Optional[str] = None

core/dbt/contracts/project.py

@@ -223,6 +223,7 @@ class Project(HyphenatedDbtClassMixin, Replaceable):
     )
     packages: List[PackageSpec] = field(default_factory=list)
     query_comment: Optional[Union[QueryComment, NoValue, str]] = field(default_factory=NoValue)
+    restrict_access: bool = False
 
     @classmethod
     def validate(cls, data):

core/dbt/contracts/publication.py

@@ -1,88 +0,0 @@
-from typing import Any, Dict, List, Optional
-from datetime import datetime
-
-
-from dataclasses import dataclass, field
-
-from dbt.contracts.util import (
-    AdditionalPropertiesMixin,
-    ArtifactMixin,
-    BaseArtifactMetadata,
-    schema_version,
-)
-from dbt.contracts.graph.unparsed import NodeVersion
-from dbt.dataclass_schema import dbtClassMixin, ExtensibleDbtClassMixin
-
-
-@dataclass
-class ProjectDependency(AdditionalPropertiesMixin, ExtensibleDbtClassMixin):
-    name: str
-    _extra: Dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass
-class ProjectDependencies(dbtClassMixin):
-    projects: List[ProjectDependency] = field(default_factory=list)
-
-
-@dataclass
-class PublicationMetadata(BaseArtifactMetadata):
-    dbt_schema_version: str = field(
-        default_factory=lambda: str(PublicationArtifact.dbt_schema_version)
-    )
-    adapter_type: Optional[str] = None
-    quoting: Dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass
-class PublicModel(dbtClassMixin):
-    """Used to represent cross-project models"""
-
-    name: str
-    package_name: str
-    unique_id: str
-    relation_name: str
-    identifier: str
-    schema: str
-    database: Optional[str] = None
-    version: Optional[NodeVersion] = None
-    latest_version: Optional[NodeVersion] = None
-    # list of model unique_ids
-    public_node_dependencies: List[str] = field(default_factory=list)
-    generated_at: datetime = field(default_factory=datetime.utcnow)
-    deprecation_date: Optional[datetime] = None
-
-
-@dataclass
-class PublicationMandatory:
-    project_name: str
-
-
-@dataclass
-@schema_version("publication", 1)
-class PublicationArtifact(ArtifactMixin, PublicationMandatory):
-    public_models: Dict[str, PublicModel] = field(default_factory=dict)
-    metadata: PublicationMetadata = field(default_factory=PublicationMetadata)
-    # list of project name strings
-    dependencies: List[str] = field(default_factory=list)
-
-
-@dataclass
-class PublicationConfig(ArtifactMixin, PublicationMandatory):
-    """This is for the part of the publication artifact which is stored in
-    the internal manifest. The public_nodes are stored separately in the manifest,
-    and just the unique_ids of the public models are stored here."""
-
-    metadata: PublicationMetadata = field(default_factory=PublicationMetadata)
-    # list of project name strings
-    dependencies: List[str] = field(default_factory=list)
-    public_node_ids: List[str] = field(default_factory=list)
-
-    @classmethod
-    def from_publication(cls, publication: PublicationArtifact):
-        return cls(
-            project_name=publication.project_name,
-            metadata=publication.metadata,
-            dependencies=publication.dependencies,
-            public_node_ids=list(publication.public_models.keys()),
-        )

core/dbt/contracts/results.py

@@ -1,3 +1,5 @@
+import threading
+
 from dbt.contracts.graph.unparsed import FreshnessThreshold
 from dbt.contracts.graph.nodes import SourceDefinition, ResultNode
 from dbt.contracts.util import (
@@ -161,6 +163,20 @@ class RunResult(NodeResult):
     def skipped(self):
         return self.status == RunStatus.Skipped
 
+    @classmethod
+    def from_node(cls, node: ResultNode, status: RunStatus, message: Optional[str]):
+        thread_id = threading.current_thread().name
+        return RunResult(
+            status=status,
+            thread_id=thread_id,
+            execution_time=0,
+            timing=[],
+            message=message,
+            node=node,
+            adapter_response={},
+            failures=None,
+        )
+
 
 @dataclass
 class ExecutionResult(dbtClassMixin):

core/dbt/events/contextvars.py

@@ -5,48 +5,65 @@ from typing import Any, Generator, Mapping, Dict
 
 
 LOG_PREFIX = "log_"
-LOG_PREFIX_LEN = len(LOG_PREFIX)
+TASK_PREFIX = "task_"
 
-_log_context_vars: Dict[str, contextvars.ContextVar] = {}
+_context_vars: Dict[str, contextvars.ContextVar] = {}
 
 
-def get_contextvars() -> Dict[str, Any]:
+def get_contextvars(prefix: str) -> Dict[str, Any]:
     rv = {}
     ctx = contextvars.copy_context()
 
+    prefix_len = len(prefix)
     for k in ctx:
-        if k.name.startswith(LOG_PREFIX) and ctx[k] is not Ellipsis:
-            rv[k.name[LOG_PREFIX_LEN:]] = ctx[k]
+        if k.name.startswith(prefix) and ctx[k] is not Ellipsis:
+            rv[k.name[prefix_len:]] = ctx[k]
 
     return rv
 
 
 def get_node_info():
-    cvars = get_contextvars()
+    cvars = get_contextvars(LOG_PREFIX)
     if "node_info" in cvars:
         return cvars["node_info"]
     else:
         return {}
 
 
-def clear_contextvars() -> None:
+def get_project_root():
+    cvars = get_contextvars(TASK_PREFIX)
+    if "project_root" in cvars:
+        return cvars["project_root"]
+    else:
+        return None
+
+
+def clear_contextvars(prefix: str) -> None:
     ctx = contextvars.copy_context()
     for k in ctx:
-        if k.name.startswith(LOG_PREFIX):
+        if k.name.startswith(prefix):
             k.set(Ellipsis)
 
 
+def set_log_contextvars(**kwargs: Any) -> Mapping[str, contextvars.Token]:
+    return set_contextvars(LOG_PREFIX, **kwargs)
+
+
+def set_task_contextvars(**kwargs: Any) -> Mapping[str, contextvars.Token]:
+    return set_contextvars(TASK_PREFIX, **kwargs)
+
+
 # put keys and values into context. Returns the contextvar.Token mapping
 # Save and pass to reset_contextvars
-def set_contextvars(**kwargs: Any) -> Mapping[str, contextvars.Token]:
+def set_contextvars(prefix: str, **kwargs: Any) -> Mapping[str, contextvars.Token]:
     cvar_tokens = {}
     for k, v in kwargs.items():
-        log_key = f"{LOG_PREFIX}{k}"
+        log_key = f"{prefix}{k}"
         try:
-            var = _log_context_vars[log_key]
+            var = _context_vars[log_key]
         except KeyError:
             var = contextvars.ContextVar(log_key, default=Ellipsis)
-            _log_context_vars[log_key] = var
+            _context_vars[log_key] = var
 
         cvar_tokens[k] = var.set(v)
 
@@ -54,30 +71,44 @@ def set_contextvars(**kwargs: Any) -> Mapping[str, contextvars.Token]:
 
 
 # reset by Tokens
-def reset_contextvars(**kwargs: contextvars.Token) -> None:
+def reset_contextvars(prefix: str, **kwargs: contextvars.Token) -> None:
     for k, v in kwargs.items():
-        log_key = f"{LOG_PREFIX}{k}"
-        var = _log_context_vars[log_key]
+        log_key = f"{prefix}{k}"
+        var = _context_vars[log_key]
         var.reset(v)
 
 
 # remove from contextvars
-def unset_contextvars(*keys: str) -> None:
+def unset_contextvars(prefix: str, *keys: str) -> None:
     for k in keys:
-        if k in _log_context_vars:
-            log_key = f"{LOG_PREFIX}{k}"
-            _log_context_vars[log_key].set(Ellipsis)
+        if k in _context_vars:
+            log_key = f"{prefix}{k}"
+            _context_vars[log_key].set(Ellipsis)
 
 
 # Context manager or decorator to set and unset the context vars
 @contextlib.contextmanager
 def log_contextvars(**kwargs: Any) -> Generator[None, None, None]:
-    context = get_contextvars()
+    context = get_contextvars(LOG_PREFIX)
     saved = {k: context[k] for k in context.keys() & kwargs.keys()}
 
-    set_contextvars(**kwargs)
+    set_contextvars(LOG_PREFIX, **kwargs)
     try:
         yield
     finally:
-        unset_contextvars(*kwargs.keys())
-        set_contextvars(**saved)
+        unset_contextvars(LOG_PREFIX, *kwargs.keys())
+        set_contextvars(LOG_PREFIX, **saved)
+
+
+# Context manager for earlier in task.run
+@contextlib.contextmanager
+def task_contextvars(**kwargs: Any) -> Generator[None, None, None]:
+    context = get_contextvars(TASK_PREFIX)
+    saved = {k: context[k] for k in context.keys() & kwargs.keys()}
+
+    set_contextvars(TASK_PREFIX, **kwargs)
+    try:
+        yield
+    finally:
+        unset_contextvars(TASK_PREFIX, *kwargs.keys())
+        set_contextvars(TASK_PREFIX, **saved)

core/dbt/events/eventmgr.py

@@ -110,6 +110,7 @@ class _Logger:
         log.setLevel(_log_level_map[self.level])
         handler.setFormatter(logging.Formatter(fmt="%(message)s"))
         log.handlers.clear()
+        log.propagate = False
         log.addHandler(handler)
         return log
 

core/dbt/events/functions.py

@@ -39,14 +39,18 @@ def setup_event_logger(flags, callbacks: List[Callable[[EventMsg], None]] = [])
     else:
         if flags.LOG_LEVEL != "none":
             line_format = _line_format_from_str(flags.LOG_FORMAT, LineFormat.PlainText)
-            log_level = EventLevel.DEBUG if flags.DEBUG else EventLevel(flags.LOG_LEVEL)
+            log_level = (
+                EventLevel.ERROR
+                if flags.QUIET
+                else EventLevel.DEBUG
+                if flags.DEBUG
+                else EventLevel(flags.LOG_LEVEL)
+            )
             console_config = _get_stdout_config(
                 line_format,
-                flags.DEBUG,
                 flags.USE_COLORS,
                 log_level,
                 flags.LOG_CACHE_EVENTS,
-                flags.QUIET,
             )
             EVENT_MANAGER.add_logger(console_config)
 
@@ -81,11 +85,9 @@ def _line_format_from_str(format_str: str, default: LineFormat) -> LineFormat:
 
 def _get_stdout_config(
     line_format: LineFormat,
-    debug: bool,
     use_colors: bool,
     level: EventLevel,
     log_cache_events: bool,
-    quiet: bool,
 ) -> LoggerConfig:
 
     return LoggerConfig(
@@ -97,8 +99,6 @@ def _get_stdout_config(
         filter=partial(
             _stdout_filter,
             log_cache_events,
-            debug,
-            quiet,
             line_format,
         ),
         output_stream=sys.stdout,
@@ -107,16 +107,11 @@ def _get_stdout_config(
 
 def _stdout_filter(
     log_cache_events: bool,
-    debug_mode: bool,
-    quiet_mode: bool,
     line_format: LineFormat,
     msg: EventMsg,
 ) -> bool:
-    return (
-        (msg.info.name not in ["CacheAction", "CacheDumpGraph"] or log_cache_events)
-        and (EventLevel(msg.info.level) != EventLevel.DEBUG or debug_mode)
-        and (EventLevel(msg.info.level) == EventLevel.ERROR or not quiet_mode)
-        and not (line_format == LineFormat.Json and type(msg.data) == Formatting)
+    return (msg.info.name not in ["CacheAction", "CacheDumpGraph"] or log_cache_events) and not (
+        line_format == LineFormat.Json and type(msg.data) == Formatting
     )
 
 
@@ -147,11 +142,9 @@ def _get_logbook_log_config(
 ) -> LoggerConfig:
     config = _get_stdout_config(
         LineFormat.PlainText,
-        debug,
         use_colors,
-        EventLevel.DEBUG if debug else EventLevel.INFO,
+        EventLevel.ERROR if quiet else EventLevel.DEBUG if debug else EventLevel.INFO,
         log_cache_events,
-        quiet,
     )
     config.name = "logbook_log"
     config.filter = (
@@ -183,7 +176,7 @@ EVENT_MANAGER: EventManager = EventManager()
 EVENT_MANAGER.add_logger(
     _get_logbook_log_config(False, True, False, False)  # type: ignore
     if ENABLE_LEGACY_LOGGER
-    else _get_stdout_config(LineFormat.PlainText, False, True, EventLevel.INFO, False, False)
+    else _get_stdout_config(LineFormat.PlainText, True, EventLevel.INFO, False)
 )
 
 # This global, and the following two functions for capturing stdout logs are

core/dbt/events/types.proto

@@ -1227,6 +1227,27 @@ message UnsupportedConstraintMaterializationMsg {
     UnsupportedConstraintMaterialization data = 2;
 }
 
+// I069
+message ParseInlineNodeError{
+    NodeInfo node_info = 1;
+    string exc = 2;
+}
+
+message ParseInlineNodeErrorMsg {
+    EventInfo info = 1;
+    ParseInlineNodeError data = 2;
+}
+
+// I070
+message SemanticValidationFailure {
+    string msg = 2;
+}
+
+message SemanticValidationFailureMsg {
+    EventInfo info = 1;
+    SemanticValidationFailure data = 2;
+}
+
 
 // M - Deps generation
 
@@ -1517,18 +1538,6 @@ message NoNodesForSelectionCriteriaMsg {
     NoNodesForSelectionCriteria data = 2;
 }
 
-// P - Artifacts
-
-// P001
-message PublicationArtifactAvailable {
-    google.protobuf.Struct pub_artifact = 1;
-}
-
-message PublicationArtifactAvailableMsg {
-    EventInfo info = 1;
-    PublicationArtifactAvailable data = 2;
-}
-
 // Q - Node execution
 
 // Q001

core/dbt/events/types.py

@@ -1216,6 +1216,22 @@ class UnsupportedConstraintMaterialization(WarnLevel):
         return line_wrap_message(warning_tag(msg))
 
 
+class ParseInlineNodeError(ErrorLevel):
+    def code(self):
+        return "I069"
+
+    def message(self) -> str:
+        return "Error while parsing node: " + self.node_info.node_name + "\n" + self.exc
+
+
+class SemanticValidationFailure(WarnLevel):
+    def code(self):
+        return "I070"
+
+    def message(self) -> str:
+        return self.msg
+
+
 # =======================================================
 # M - Deps generation
 # =======================================================
@@ -1468,19 +1484,6 @@ class NoNodesForSelectionCriteria(WarnLevel):
 # =======================================================
 
 
-class PublicationArtifactAvailable(DebugLevel):
-    def code(self):
-        return "P001"
-
-    def message(self) -> str:
-        return "Publication artifact available"
-
-
-# =======================================================
-# Q - Node execution
-# =======================================================
-
-
 class RunningOperationCaughtError(ErrorLevel):
     def code(self):
         return "Q001"

core/dbt/exceptions.py

@@ -7,7 +7,7 @@ from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 
 from dbt.dataclass_schema import ValidationError
 from dbt.events.helpers import env_secrets, scrub_secrets
-from dbt.node_types import NodeType
+from dbt.node_types import NodeType, AccessType
 from dbt.ui import line_wrap_message
 
 import dbt.dataclass_schema
@@ -297,6 +297,11 @@ class ParsingError(DbtRuntimeError):
         return "Parsing"
 
 
+class dbtPluginError(DbtRuntimeError):
+    CODE = 10020
+    MESSAGE = "Plugin Error"
+
+
 # TODO: this isn't raised in the core codebase.  Is it raised elsewhere?
 class JSONValidationError(DbtValidationError):
     def __init__(self, typename, errors):
@@ -405,19 +410,6 @@ class DbtProfileError(DbtConfigError):
     pass
 
 
-class PublicationConfigNotFound(DbtConfigError):
-    def __init__(self, project=None, file_name=None):
-        self.project = project
-        msg = self.message()
-        super().__init__(msg, project=project)
-
-    def message(self):
-        return (
-            f"A dependency on project {self.project} was specified, "
-            f"but a publication for {self.project} was not found."
-        )
-
-
 class SemverError(Exception):
     def __init__(self, msg: Optional[str] = None):
         self.msg = msg
@@ -910,19 +902,6 @@ class SecretEnvVarLocationError(ParsingError):
         return msg
 
 
-class ProjectDependencyCycleError(ParsingError):
-    def __init__(self, pub_project_name, project_name):
-        self.pub_project_name = pub_project_name
-        self.project_name = project_name
-        super().__init__(msg=self.get_message())
-
-    def get_message(self) -> str:
-        return (
-            f"A project dependency cycle has been detected. The current project {self.project_name} "
-            f"depends on {self.pub_project_name} which also depends on the current project."
-        )
-
-
 class MacroArgTypeError(CompilationError):
     def __init__(self, method_name: str, arg_name: str, got_value: Any, expected_type):
         self.method_name = method_name
@@ -1240,16 +1219,18 @@ class SnapshopConfigError(ParsingError):
 
 
 class DbtReferenceError(ParsingError):
-    def __init__(self, unique_id: str, ref_unique_id: str, group: str):
+    def __init__(self, unique_id: str, ref_unique_id: str, access: AccessType, scope: str):
         self.unique_id = unique_id
         self.ref_unique_id = ref_unique_id
-        self.group = group
+        self.access = access
+        self.scope = scope
+        self.scope_type = "group" if self.access == AccessType.Private else "package"
         super().__init__(msg=self.get_message())
 
     def get_message(self) -> str:
         return (
             f"Node {self.unique_id} attempted to reference node {self.ref_unique_id}, "
-            f"which is not allowed because the referenced node is private to the {self.group} group."
+            f"which is not allowed because the referenced node is {self.access} to the '{self.scope}' {self.scope_type}."
         )
 
 
@@ -2349,6 +2330,11 @@ class ContractError(CompilationError):
         return table_from_data_flat(mismatches_sorted, column_names)
 
     def get_message(self) -> str:
+        if not self.yaml_columns:
+            return (
+                "This model has an enforced contract, and its 'columns' specification is missing"
+            )
+
         table: agate.Table = self.get_mismatches()
         # Hack to get Agate table output as string
         output = io.StringIO()

core/dbt/flags.py

@@ -87,6 +87,7 @@ def get_flag_dict():
         "introspect",
         "target_path",
         "log_path",
+        "invocation_command",
     }
     return {key: getattr(GLOBAL_FLAGS, key.upper(), None) for key in flag_attr}
 

core/dbt/graph/selector.py

@@ -168,6 +168,8 @@ class NodeSelector(MethodManager):
         elif unique_id in self.manifest.metrics:
             metric = self.manifest.metrics[unique_id]
             return metric.config.enabled
+        elif unique_id in self.manifest.semantic_models:
+            return True
         node = self.manifest.nodes[unique_id]
 
         if self.include_empty_nodes:
@@ -191,6 +193,8 @@ class NodeSelector(MethodManager):
             node = self.manifest.exposures[unique_id]
         elif unique_id in self.manifest.metrics:
             node = self.manifest.metrics[unique_id]
+        elif unique_id in self.manifest.semantic_models:
+            node = self.manifest.semantic_models[unique_id]
         else:
             raise DbtInternalError(f"Node {unique_id} not found in the manifest!")
         return self.node_is_match(node)

core/dbt/graph/selector_methods.py

@@ -26,7 +26,7 @@ from dbt.exceptions import (
     DbtRuntimeError,
 )
 from dbt.node_types import NodeType
-from dbt.task.contextvars import cv_project_root
+from dbt.events.contextvars import get_project_root
 
 
 SELECTOR_GLOB = "*"
@@ -61,8 +61,8 @@ def is_selected_node(fqn: List[str], node_selector: str, is_versioned: bool) ->
         flat_node_selector = node_selector.split(".")
         if fqn[-2] == node_selector:
             return True
-        # If this is a versioned model, then the last two segments should be allowed to exactly match
-        elif fqn[-2:] == flat_node_selector[-2:]:
+        # If this is a versioned model, then the last two segments should be allowed to exactly match on either the '.' or '_' delimiter
+        elif "_".join(fqn[-2:]) == "_".join(flat_node_selector[-2:]):
             return True
     else:
         if fqn[-1] == node_selector:
@@ -326,7 +326,11 @@ class PathSelectorMethod(SelectorMethod):
     def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
         """Yields nodes from included that match the given path."""
         # get project root from contextvar
-        root = Path(cv_project_root.get())
+        project_root = get_project_root()
+        if project_root:
+            root = Path(project_root)
+        else:
+            root = Path.cwd()
         paths = set(p.relative_to(root) for p in root.glob(selector))
         for node, real_node in self.all_nodes(included_nodes):
             ofp = Path(real_node.original_file_path)
@@ -347,6 +351,8 @@ class FileSelectorMethod(SelectorMethod):
         for node, real_node in self.all_nodes(included_nodes):
             if fnmatch(Path(real_node.original_file_path).name, selector):
                 yield node
+            elif fnmatch(Path(real_node.original_file_path).stem, selector):
+                yield node
 
 
 class PackageSelectorMethod(SelectorMethod):
@@ -431,6 +437,8 @@ class ResourceTypeSelectorMethod(SelectorMethod):
 
 
 class TestNameSelectorMethod(SelectorMethod):
+    __test__ = False
+
     def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
         for node, real_node in self.parsed_nodes(included_nodes):
             if real_node.resource_type == NodeType.Test and hasattr(real_node, "test_metadata"):
@@ -439,6 +447,8 @@ class TestNameSelectorMethod(SelectorMethod):
 
 
 class TestTypeSelectorMethod(SelectorMethod):
+    __test__ = False
+
     def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[UniqueId]:
         search_type: Type
         # continue supporting 'schema' + 'data' for backwards compatibility
@@ -538,6 +548,11 @@ class StateSelectorMethod(SelectorMethod):
         upstream_macro_change = self.check_macros_modified(new)
         return different_contents or upstream_macro_change
 
+    def check_unmodified_content(
+        self, old: Optional[SelectorTarget], new: SelectorTarget, adapter_type: str
+    ) -> bool:
+        return not self.check_modified_content(old, new, adapter_type)
+
     def check_modified_macros(self, old, new: SelectorTarget) -> bool:
         return self.check_macros_modified(new)
 
@@ -582,8 +597,10 @@ class StateSelectorMethod(SelectorMethod):
         state_checks = {
             # it's new if there is no old version
             "new": lambda old, new: old is None,
+            "old": lambda old, new: old is not None,
             # use methods defined above to compare properties of old + new
             "modified": self.check_modified_content,
+            "unmodified": self.check_unmodified_content,
             "modified.body": self.check_modified_factory("same_body"),
             "modified.configs": self.check_modified_factory("same_config"),
             "modified.persisted_descriptions": self.check_modified_factory(
@@ -615,7 +632,11 @@ class StateSelectorMethod(SelectorMethod):
                 previous_node = manifest.metrics[node]
 
             keyword_args = {}
-            if checker.__name__ in ["same_contract", "check_modified_content"]:
+            if checker.__name__ in [
+                "same_contract",
+                "check_modified_content",
+                "check_unmodified_content",
+            ]:
                 keyword_args["adapter_type"] = adapter_type  # type: ignore
 
             if checker(previous_node, real_node, **keyword_args):  # type: ignore

core/dbt/include/global_project/macros/adapters/validate_sql.sql

@@ -0,0 +1,10 @@
+{% macro validate_sql(sql) -%}
+  {{ return(adapter.dispatch('validate_sql', 'dbt')(sql)) }}
+{% endmacro %}
+
+{% macro default__validate_sql(sql) -%}
+  {% call statement('validate_sql') -%}
+    explain {{ sql }}
+  {% endcall %}
+  {{ return(load_result('validate_sql')) }}
+{% endmacro %}

core/dbt/include/global_project/macros/materializations/models/clone/can_clone_table.sql

@@ -0,0 +1,7 @@
+{% macro can_clone_table() %}
+    {{ return(adapter.dispatch('can_clone_table', 'dbt')()) }}
+{% endmacro %}
+
+{% macro default__can_clone_table() %}
+    {{ return(False) }}
+{% endmacro %}

core/dbt/include/global_project/macros/materializations/models/clone/clone.sql

@@ -0,0 +1,62 @@
+{%- materialization clone, default -%}
+
+  {%- set relations = {'relations': []} -%}
+
+  {%- if not defer_relation -%}
+      -- nothing to do
+      {{ log("No relation found in state manifest for " ~ model.unique_id, info=True) }}
+      {{ return(relations) }}
+  {%- endif -%}
+
+  {%- set existing_relation = load_cached_relation(this) -%}
+
+  {%- if existing_relation and not flags.FULL_REFRESH -%}
+      -- noop!
+      {{ log("Relation " ~ existing_relation ~ " already exists", info=True) }}
+      {{ return(relations) }}
+  {%- endif -%}
+
+  {%- set other_existing_relation = load_cached_relation(defer_relation) -%}
+
+  -- If this is a database that can do zero-copy cloning of tables, and the other relation is a table, then this will be a table
+  -- Otherwise, this will be a view
+
+  {% set can_clone_table = can_clone_table() %}
+
+  {%- if other_existing_relation and other_existing_relation.type == 'table' and can_clone_table -%}
+
+      {%- set target_relation = this.incorporate(type='table') -%}
+      {% if existing_relation is not none and not existing_relation.is_table %}
+        {{ log("Dropping relation " ~ existing_relation ~ " because it is of type " ~ existing_relation.type) }}
+        {{ drop_relation_if_exists(existing_relation) }}
+      {% endif %}
+
+      -- as a general rule, data platforms that can clone tables can also do atomic 'create or replace'
+      {% call statement('main') %}
+          {{ create_or_replace_clone(target_relation, defer_relation) }}
+      {% endcall %}
+
+      {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %}
+      {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %}
+      {% do persist_docs(target_relation, model) %}
+
+      {{ return({'relations': [target_relation]}) }}
+
+  {%- else -%}
+
+      {%- set target_relation = this.incorporate(type='view') -%}
+
+      -- reuse the view materialization
+      -- TODO: support actual dispatch for materialization macros
+      -- Tracking ticket: https://github.com/dbt-labs/dbt-core/issues/7799
+      {% set search_name = "materialization_view_" ~ adapter.type() %}
+      {% if not search_name in context %}
+          {% set search_name = "materialization_view_default" %}
+      {% endif %}
+      {% set materialization_macro = context[search_name] %}
+      {% set relations = materialization_macro() %}
+      {{ return(relations) }}
+
+  {%- endif -%}
+
+{%- endmaterialization -%}

core/dbt/include/global_project/macros/materializations/models/clone/create_or_replace_clone.sql

@@ -0,0 +1,7 @@
+{% macro create_or_replace_clone(this_relation, defer_relation) %}
+    {{ return(adapter.dispatch('create_or_replace_clone', 'dbt')(this_relation, defer_relation)) }}
+{% endmacro %}
+
+{% macro default__create_or_replace_clone(this_relation, defer_relation) %}
+    create or replace table {{ this_relation }} clone {{ defer_relation }}
+{% endmacro %}

core/dbt/include/global_project/macros/materializations/models/table/columns_spec_ddl.sql

@@ -33,10 +33,17 @@
   If any differences in name, data_type or number of columns exist between the two schemas, raises a compiler error
 #}
 {% macro assert_columns_equivalent(sql) %}
+
+  {#-- First ensure the user has defined 'columns' in yaml specification --#}
+  {%- set user_defined_columns = model['columns'] -%}
+  {%- if not user_defined_columns -%}
+      {{ exceptions.raise_contract_error([], []) }}
+  {%- endif -%}
+
   {#-- Obtain the column schema provided by sql file. #}
   {%- set sql_file_provided_columns = get_column_schema_from_query(sql, config.get('sql_header', none)) -%}
   {#--Obtain the column schema provided by the schema file by generating an 'empty schema' query from the model's columns. #}
-  {%- set schema_file_provided_columns = get_column_schema_from_query(get_empty_schema_sql(model['columns'])) -%}
+  {%- set schema_file_provided_columns = get_column_schema_from_query(get_empty_schema_sql(user_defined_columns)) -%}
 
   {#-- create dictionaries with name and formatted data type and strings for exception #}
   {%- set sql_columns = format_columns(sql_file_provided_columns) -%}