Feature: adapter cte generation

CHANGELOG.md

@@ -3,11 +3,14 @@
 ### Breaking changes
 - `adapter_macro` is no longer a macro, instead it is a builtin context method. Any custom macros that intercepted it by going through `context['dbt']` will need to instead access it via `context['builtins']` ([#2302](https://github.com/fishtown-analytics/dbt/issues/2302), [#2673](https://github.com/fishtown-analytics/dbt/pull/2673))
 - `adapter_macro` is now deprecated. Use `adapter.dispatch` instead.
+- Data tests are now written as CTEs instead of subqueries. Adapter plugins for adapters that don't support CTEs may require modification. ([#2712](https://github.com/fishtown-analytics/dbt/pull/2712))
 
 
 ### Under the hood
 - Upgraded snowflake-connector-python dependency to 2.2.10 and enabled the SSO token cache ([#2613](https://github.com/fishtown-analytics/dbt/issues/2613), [#2689](https://github.com/fishtown-analytics/dbt/issues/2689), [#2698](https://github.com/fishtown-analytics/dbt/pull/2698))
 - Add deprecation warnings to anonymous usage tracking ([#2688](https://github.com/fishtown-analytics/dbt/issues/2688), [#2710](https://github.com/fishtown-analytics/dbt/issues/2710))
+- Data tests now behave like dbt CTEs ([#2609](https://github.com/fishtown-analytics/dbt/issues/2609), [#2712](https://github.com/fishtown-analytics/dbt/pull/2712))
+- Adapter plugins can now override the CTE prefix by overriding their `Relation` attribute with a class that has a custom `add_ephemeral_prefix` implementation. ([#2660](https://github.com/fishtown-analytics/dbt/issues/2660), [#2712](https://github.com/fishtown-analytics/dbt/pull/2712))
 
 ### Features
 - Add better retry support when using the BigQuery adapter ([#2694](https://github.com/fishtown-analytics/dbt/pull/2694), follow-up to [#1963](https://github.com/fishtown-analytics/dbt/pull/1963))
@@ -17,6 +20,7 @@
 - Macros in the current project can override internal dbt macros that are called through `execute_macros`. ([#2301](https://github.com/fishtown-analytics/dbt/issues/2301), [#2686](https://github.com/fishtown-analytics/dbt/pull/2686))
 - Add state:modified and state:new selectors ([#2641](https://github.com/fishtown-analytics/dbt/issues/2641), [#2695](https://github.com/fishtown-analytics/dbt/pull/2695))
 
+
 ### Fixes
 - Fix Redshift table size estimation; e.g. 44 GB tables are no longer reported as 44 KB. [#2702](https://github.com/fishtown-analytics/dbt/issues/2702)
 

core/dbt/adapters/base/impl.py

@@ -25,7 +25,9 @@
 )
 from dbt.clients.agate_helper import empty_table, merge_tables, table_from_rows
 from dbt.clients.jinja import MacroGenerator
-from dbt.contracts.graph.compiled import CompileResultNode, CompiledSeedNode
+from dbt.contracts.graph.compiled import (
+    CompileResultNode, CompiledSeedNode
+)
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.parsed import ParsedSeedNode
 from dbt.exceptions import warn_or_error
@@ -289,7 +291,10 @@ def _get_cache_schemas(self, manifest: Manifest) -> Set[BaseRelation]:
         return {
             self.Relation.create_from(self.config, node).without_identifier()
             for node in manifest.nodes.values()
-            if node.resource_type in NodeType.executable()
+            if (
+                node.resource_type in NodeType.executable() and
+                not node.is_ephemeral_model
+            )
         }
 
     def _get_catalog_schemas(self, manifest: Manifest) -> SchemaSearchMap:
@@ -1142,6 +1147,10 @@ def get_rows_different_sql(
 
         return sql
 
+    def get_compiler(self):
+        from dbt.compilation import Compiler
+        return Compiler(self.config)
+
 
 COLUMNS_EQUAL_SQL = '''
 with diff_count as (

core/dbt/adapters/base/relation.py

@@ -201,6 +201,23 @@ def create_from_source(
             **kwargs
         )
 
+    @staticmethod
+    def add_ephemeral_prefix(name: str):
+        return f'__dbt__CTE__{name}'
+
+    @classmethod
+    def create_ephemeral_from_node(
+        cls: Type[Self],
+        config: HasQuoting,
+        node: Union[ParsedNode, CompiledNode],
+    ) -> Self:
+        # Note that ephemeral models are based on the name.
+        identifier = cls.add_ephemeral_prefix(node.name)
+        return cls.create(
+            type=cls.CTE,
+            identifier=identifier,
+        ).quote(identifier=False)
+
     @classmethod
     def create_from_node(
         cls: Type[Self],

core/dbt/adapters/protocol.py

@@ -1,19 +1,23 @@
 from dataclasses import dataclass
 from typing import (
     Type, Hashable, Optional, ContextManager, List, Generic, TypeVar, ClassVar,
-    Tuple, Union
+    Tuple, Union, Dict, Any
 )
 from typing_extensions import Protocol
 
 import agate
 
 from dbt.contracts.connection import Connection, AdapterRequiredConfig
-from dbt.contracts.graph.compiled import CompiledNode
+from dbt.contracts.graph.compiled import (
+    CompiledNode, NonSourceNode, NonSourceCompiledNode
+)
 from dbt.contracts.graph.parsed import ParsedNode, ParsedSourceDefinition
 from dbt.contracts.graph.model_config import BaseConfig
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.relation import Policy, HasQuoting
 
+from dbt.graph import Graph
+
 
 @dataclass
 class AdapterConfig(BaseConfig):
@@ -45,6 +49,19 @@ def create_from(
         ...
 
 
+class CompilerProtocol(Protocol):
+    def compile(self, manifest: Manifest, write=True) -> Graph:
+        ...
+
+    def compile_node(
+        self,
+        node: NonSourceNode,
+        manifest: Manifest,
+        extra_context: Optional[Dict[str, Any]] = None,
+    ) -> NonSourceCompiledNode:
+        ...
+
+
 AdapterConfig_T = TypeVar(
     'AdapterConfig_T', bound=AdapterConfig
 )
@@ -57,11 +74,18 @@ def create_from(
 Column_T = TypeVar(
     'Column_T', bound=ColumnProtocol
 )
+Compiler_T = TypeVar('Compiler_T', bound=CompilerProtocol)
 
 
 class AdapterProtocol(
     Protocol,
-    Generic[AdapterConfig_T, ConnectionManager_T, Relation_T, Column_T]
+    Generic[
+        AdapterConfig_T,
+        ConnectionManager_T,
+        Relation_T,
+        Column_T,
+        Compiler_T,
+    ]
 ):
     AdapterSpecificConfigs: ClassVar[Type[AdapterConfig_T]]
     Column: ClassVar[Type[Column_T]]
@@ -132,3 +156,6 @@ def execute(
         self, sql: str, auto_begin: bool = False, fetch: bool = False
     ) -> Tuple[str, agate.Table]:
         ...
+
+    def get_compiler(self) -> Compiler_T:
+        ...