demo: OperationRefResolver for SqlOperation

core/dbt/compilation.py

@@ -10,7 +10,7 @@ from dbt import flags
 from dbt.adapters.factory import get_adapter
 from dbt.clients import jinja
 from dbt.clients.system import make_directory
-from dbt.context.providers import generate_runtime_model_context
+from dbt.context.providers import generate_runtime_model_context, generate_runtime_sql_operation_context
 from dbt.contracts.graph.manifest import Manifest, UniqueID
 from dbt.contracts.graph.compiled import (
     COMPILED_TYPES,
@@ -19,6 +19,8 @@ from dbt.contracts.graph.compiled import (
     InjectedCTE,
     ManifestNode,
     NonSourceCompiledNode,
+    CompiledSqlNode,
+    CompiledRPCNode,
 )
 from dbt.contracts.graph.parsed import ParsedNode
 from dbt.exceptions import (
@@ -180,8 +182,12 @@ class Compiler:
         manifest: Manifest,
         extra_context: Dict[str, Any],
     ) -> Dict[str, Any]:
-
-        context = generate_runtime_model_context(node, self.config, manifest)
+        
+        if isinstance(node, CompiledSqlNode) or isinstance(node, CompiledRPCNode):
+            # or node.resource_type in ('SqlOperation', 'RPCCall'):
+            context = generate_runtime_sql_operation_context(node, self.config, manifest)
+        else:
+            context = generate_runtime_model_context(node, self.config, manifest)
         context.update(extra_context)
         if isinstance(node, CompiledGenericTestNode):
             # for test nodes, add a special keyword args value to the context

core/dbt/context/providers.py

@@ -32,6 +32,8 @@ from dbt.contracts.graph.compiled import (
     CompiledResource,
     CompiledSeedNode,
     ManifestNode,
+    CompiledSqlNode,
+    CompiledRPCNode,
 )
 from dbt.contracts.graph.parsed import (
     ParsedMacro,
@@ -470,11 +472,11 @@ class OperationRefResolver(RuntimeRefResolver):
         pass
 
     def create_relation(self, target_model: ManifestNode, name: str) -> RelationProxy:
-        if target_model.is_ephemeral_model:
+        if not hasattr(self.model, "set_cte") and target_model.is_ephemeral_model:
             # In operations, we can't ref() ephemeral nodes, because
             # ParsedMacros do not support set_cte
             raise_compiler_error(
-                "Operations can not ref() ephemeral nodes, but {} is ephemeral".format(
+                "Macros run as operations cannot ref() ephemeral nodes, but {} is ephemeral".format(
                     target_model.name
                 ),
                 self.model,
@@ -1286,6 +1288,15 @@ def generate_runtime_macro_context(
     return ctx.to_dict()
 
 
+def generate_runtime_sql_operation_context(
+    model: Union[CompiledSqlNode, CompiledRPCNode],
+    config: RuntimeConfig,
+    manifest: Manifest,
+) -> Dict[str, Any]:
+    ctx = ModelContext(model, config, manifest, OperationProvider(), None)
+    return ctx.to_dict()
+
+
 class ExposureRefResolver(BaseResolver):
     def __call__(self, *args) -> str:
         if len(args) not in (1, 2):