[#3885] Handle env_vars in partial parsing of SQL files

CHANGELOG.md

@@ -27,6 +27,7 @@ Contributors:
 - Wait for postgres docker container to be ready in `setup_db.sh`. ([#3876](https://github.com/dbt-labs/dbt-core/issues/3876), [#3908](https://github.com/dbt-labs/dbt-core/pull/3908))
 - Prefer macros defined in the project over the ones in a package by default ([#4106](https://github.com/dbt-labs/dbt-core/issues/4106), [#4114](https://github.com/dbt-labs/dbt-core/pull/4114)) 
 - Dependency updates ([#4079](https://github.com/dbt-labs/dbt-core/pull/4079)), ([#3532](https://github.com/dbt-labs/dbt-core/pull/3532)
+- Schedule partial parsing for SQL files with env_var changes ([#3885](https://github.com/dbt-labs/dbt-core/issues/3885), [#4101](https://github.com/dbt-labs/dbt-core/pull/4101))
 
 Contributors:
 - [@sungchun12](https://github.com/sungchun12) ([#4017](https://github.com/dbt-labs/dbt/pull/4017))

core/dbt/adapters/base/impl.py

@@ -968,10 +968,10 @@ def execute_macro(
                 'dbt could not find a macro with the name "{}" in {}'
                 .format(macro_name, package_name)
             )
-        # This causes a reference cycle, as generate_runtime_macro()
+        # This causes a reference cycle, as generate_runtime_macro_context()
         # ends up calling get_adapter, so the import has to be here.
-        from dbt.context.providers import generate_runtime_macro
-        macro_context = generate_runtime_macro(
+        from dbt.context.providers import generate_runtime_macro_context
+        macro_context = generate_runtime_macro_context(
             macro=macro,
             config=self.config,
             manifest=manifest,

core/dbt/compilation.py

@@ -9,7 +9,7 @@
 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
+from dbt.context.providers import generate_runtime_model_context
 from dbt.contracts.graph.manifest import Manifest, UniqueID
 from dbt.contracts.graph.compiled import (
     COMPILED_TYPES,
@@ -178,7 +178,7 @@ def _create_node_context(
         extra_context: Dict[str, Any],
     ) -> Dict[str, Any]:
 
-        context = generate_runtime_model(
+        context = generate_runtime_model_context(
             node, self.config, manifest
         )
         context.update(extra_context)

core/dbt/config/runtime.py

@@ -253,7 +253,7 @@ def _get_v2_config_paths(
     ) -> PathSet:
         for key, value in config.items():
             if isinstance(value, dict) and not key.startswith('+'):
-                self._get_v2_config_paths(value, path + (key,), paths)
+                self._get_config_paths(value, path + (key,), paths)
             else:
                 paths.add(path)
         return frozenset(paths)

core/dbt/context/base.py

@@ -160,9 +160,12 @@ def __call__(self, var_name, default=_VAR_NOTSET):
 
 
 class BaseContext(metaclass=ContextMeta):
+    # subclass is TargetContext
     def __init__(self, cli_vars):
         self._ctx = {}
         self.cli_vars = cli_vars
+        # Save the env_vars encountered using this
+        self.env_vars = {}
 
     def generate_builtins(self):
         builtins: Dict[str, Any] = {}
@@ -271,17 +274,21 @@ def var(self) -> Var:
         return Var(self._ctx, self.cli_vars)
 
     @contextmember
-    @staticmethod
-    def env_var(var: str, default: Optional[str] = None) -> str:
+    def env_var(self, var: str, default: Optional[str] = None) -> str:
         """The env_var() function. Return the environment variable named 'var'.
         If there is no such environment variable set, return the default.
 
         If the default is None, raise an exception for an undefined variable.
         """
+        return_value = None
         if var in os.environ:
-            return os.environ[var]
+            return_value = os.environ[var]
         elif default is not None:
-            return default
+            return_value = default
+
+        if return_value is not None:
+            self.env_vars[var] = return_value
+            return return_value
         else:
             msg = f"Env var required but not provided: '{var}'"
             undefined_error(msg)

core/dbt/context/configured.py

@@ -9,6 +9,7 @@
 
 
 class ConfiguredContext(TargetContext):
+    # subclasses are SchemaYamlContext, MacroResolvingContext, ManifestContext
     config: AdapterRequiredConfig
 
     def __init__(
@@ -64,6 +65,7 @@ def __call__(self, var_name, default=Var._VAR_NOTSET):
 
 
 class SchemaYamlContext(ConfiguredContext):
+    # subclass is DocsRuntimeContext
     def __init__(self, config, project_name: str):
         super().__init__(config)
         self._project_name = project_name
@@ -86,7 +88,7 @@ def var(self) -> ConfiguredVar:
         )
 
 
-def generate_schema_yml(
+def generate_schema_yml_context(
     config: AdapterRequiredConfig, project_name: str
 ) -> Dict[str, Any]:
     ctx = SchemaYamlContext(config, project_name)

core/dbt/context/docs.py

@@ -75,7 +75,7 @@ def doc(self, *args: str) -> str:
         return target_doc.block_contents
 
 
-def generate_runtime_docs(
+def generate_runtime_docs_context(
     config: RuntimeConfig,
     target: Any,
     manifest: Manifest,

core/dbt/context/manifest.py

@@ -17,6 +17,7 @@ class ManifestContext(ConfiguredContext):
     The given macros can override any previous context values, which will be
     available as if they were accessed relative to the package name.
     """
+    # subclasses are QueryHeaderContext and ProviderContext
     def __init__(
         self,
         config: AdapterRequiredConfig,

core/dbt/context/providers.py

@@ -11,7 +11,7 @@
     get_adapter, get_adapter_package_names, get_adapter_type_names
 )
 from dbt.clients import agate_helper
-from dbt.clients.jinja import get_rendered, MacroGenerator, MacroStack
+from dbt.clients.jinja import get_rendered, MacroGenerator, MacroStack, undefined_error
 from dbt.config import RuntimeConfig, Project
 from .base import contextmember, contextproperty, Var
 from .configured import FQNLookup
@@ -49,7 +49,7 @@
     wrapped_exports,
 )
 from dbt.config import IsFQNResource
-from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.logger import GLOBAL_LOGGER as logger, SECRET_ENV_PREFIX  # noqa
 from dbt.node_types import NodeType
 
 from dbt.utils import (
@@ -636,6 +636,7 @@ class OperationProvider(RuntimeProvider):
 
 # Base context collection, used for parsing configs.
 class ProviderContext(ManifestContext):
+    # subclasses are MacroContext, ModelContext, TestContext
     def __init__(
         self,
         model,
@@ -1161,6 +1162,30 @@ def adapter_macro(self, name: str, *args, **kwargs):
         )
         raise CompilationException(msg)
 
+    @contextmember
+    def env_var(self, var: str, default: Optional[str] = None) -> str:
+        """The env_var() function. Return the environment variable named 'var'.
+        If there is no such environment variable set, return the default.
+
+        If the default is None, raise an exception for an undefined variable.
+        """
+        return_value = None
+        if var in os.environ:
+            return_value = os.environ[var]
+        elif default is not None:
+            return_value = default
+
+        if return_value is not None:
+            # Save the env_var value in the manifest and the var name in the source_file
+            if not var.startswith(SECRET_ENV_PREFIX) and self.model:
+                self.manifest.env_vars[var] = return_value
+                source_file = self.manifest.files[self.model.file_id]
+                source_file.env_vars.append(var)
+            return return_value
+        else:
+            msg = f"Env var required but not provided: '{var}'"
+            undefined_error(msg)
+
 
 class MacroContext(ProviderContext):
     """Internally, macros can be executed like nodes, with some restrictions:
@@ -1262,7 +1287,7 @@ def this(self) -> Optional[RelationProxy]:
 
 
 # This is called by '_context_for', used in 'render_with_context'
-def generate_parser_model(
+def generate_parser_model_context(
     model: ManifestNode,
     config: RuntimeConfig,
     manifest: Manifest,
@@ -1279,7 +1304,7 @@ def generate_parser_model(
     return ctx.to_dict()
 
 
-def generate_generate_component_name_macro(
+def generate_generate_name_macro_context(
     macro: ParsedMacro,
     config: RuntimeConfig,
     manifest: Manifest,
@@ -1290,7 +1315,7 @@ def generate_generate_component_name_macro(
     return ctx.to_dict()
 
 
-def generate_runtime_model(
+def generate_runtime_model_context(
     model: ManifestNode,
     config: RuntimeConfig,
     manifest: Manifest,
@@ -1301,7 +1326,7 @@ def generate_runtime_model(
     return ctx.to_dict()
 
 
-def generate_runtime_macro(
+def generate_runtime_macro_context(
     macro: ParsedMacro,
     config: RuntimeConfig,
     manifest: Manifest,

core/dbt/context/target.py

@@ -8,6 +8,7 @@
 
 
 class TargetContext(BaseContext):
+    # subclass is ConfiguredContext
     def __init__(self, config: HasCredentials, cli_vars: Dict[str, Any]):
         super().__init__(cli_vars=cli_vars)
         self.config = config

core/dbt/contracts/files.py

@@ -190,6 +190,7 @@ class SourceFile(BaseSourceFile):
     nodes: List[str] = field(default_factory=list)
     docs: List[str] = field(default_factory=list)
     macros: List[str] = field(default_factory=list)
+    env_vars: List[str] = field(default_factory=list)
 
     @classmethod
     def big_seed(cls, path: FilePath) -> 'SourceFile':
@@ -230,6 +231,7 @@ class SchemaSourceFile(BaseSourceFile):
     # Patches are only against external sources. Sources can be
     # created too, but those are in 'sources'
     sop: List[SourceKey] = field(default_factory=list)
+    env_vars: Dict[str, Any] = field(default_factory=dict)
     pp_dict: Optional[Dict[str, Any]] = None
     pp_test_index: Optional[Dict[str, Any]] = None
 
@@ -252,7 +254,7 @@ def source_patches(self):
     def __post_serialize__(self, dct):
         dct = super().__post_serialize__(dct)
         # Remove partial parsing specific data
-        for key in ('pp_files', 'pp_test_index', 'pp_dict'):
+        for key in ('pp_test_index', 'pp_dict'):
             if key in dct:
                 del dct[key]
         return dct

core/dbt/contracts/graph/manifest.py

@@ -569,6 +569,7 @@ class Manifest(MacroMethods, DataClassMessagePackMixin, dbtClassMixin):
     state_check: ManifestStateCheck = field(default_factory=ManifestStateCheck)
     source_patches: MutableMapping[SourceKey, SourcePatch] = field(default_factory=dict)
     disabled: MutableMapping[str, List[CompileResultNode]] = field(default_factory=dict)
+    env_vars: MutableMapping[str, str] = field(default_factory=dict)
 
     _doc_lookup: Optional[DocLookup] = field(
         default=None, metadata={'serialize': lambda x: None, 'deserialize': lambda x: None}
@@ -1048,6 +1049,7 @@ def __reduce_ex__(self, protocol):
             self.state_check,
             self.source_patches,
             self.disabled,
+            self.env_vars,
             self._doc_lookup,
             self._source_lookup,
             self._ref_lookup,

core/dbt/contracts/graph/parsed.py

@@ -151,7 +151,7 @@ def patch(self, patch: 'ParsedNodePatch'):
         # Note: config should already be updated
         self.patch_path: Optional[str] = patch.file_id
         # update created_at so process_docs will run in partial parsing
-        self.created_at = int(time.time())
+        self.created_at = time.time()
         self.description = patch.description
         self.columns = patch.columns
         self.meta = patch.meta
@@ -193,7 +193,7 @@ class ParsedNodeDefaults(ParsedNodeMandatory):
     build_path: Optional[str] = None
     deferred: bool = False
     unrendered_config: Dict[str, Any] = field(default_factory=dict)
-    created_at: int = field(default_factory=lambda: int(time.time()))
+    created_at: float = field(default_factory=lambda: time.time())
     config_call_dict: Dict[str, Any] = field(default_factory=dict)
 
     def write_node(self, target_path: str, subdirectory: str, payload: str):
@@ -493,12 +493,12 @@ class ParsedMacro(UnparsedBaseNode, HasUniqueID):
     docs: Docs = field(default_factory=Docs)
     patch_path: Optional[str] = None
     arguments: List[MacroArgument] = field(default_factory=list)
-    created_at: int = field(default_factory=lambda: int(time.time()))
+    created_at: float = field(default_factory=lambda: time.time())
 
     def patch(self, patch: ParsedMacroPatch):
         self.patch_path: Optional[str] = patch.file_id
         self.description = patch.description
-        self.created_at = int(time.time())
+        self.created_at = time.time()
         self.meta = patch.meta
         self.docs = patch.docs
         self.arguments = patch.arguments
@@ -614,7 +614,7 @@ class ParsedSourceDefinition(
     patch_path: Optional[Path] = None
     unrendered_config: Dict[str, Any] = field(default_factory=dict)
     relation_name: Optional[str] = None
-    created_at: int = field(default_factory=lambda: int(time.time()))
+    created_at: float = field(default_factory=lambda: time.time())
 
     def same_database_representation(
         self, other: 'ParsedSourceDefinition'
@@ -725,7 +725,7 @@ class ParsedExposure(UnparsedBaseNode, HasUniqueID, HasFqn):
     depends_on: DependsOn = field(default_factory=DependsOn)
     refs: List[List[str]] = field(default_factory=list)
     sources: List[List[str]] = field(default_factory=list)
-    created_at: int = field(default_factory=lambda: int(time.time()))
+    created_at: float = field(default_factory=lambda: time.time())
 
     @property
     def depends_on_nodes(self):

core/dbt/parser/base.py

@@ -10,8 +10,8 @@
 from dbt import utils
 from dbt.clients.jinja import MacroGenerator
 from dbt.context.providers import (
-    generate_parser_model,
-    generate_generate_component_name_macro,
+    generate_parser_model_context,
+    generate_generate_name_macro_context,
 )
 from dbt.adapters.factory import get_adapter  # noqa: F401
 from dbt.clients.jinja import get_rendered
@@ -101,7 +101,7 @@ def __init__(
                 f'No macro with name generate_{component}_name found'
             )
 
-        root_context = generate_generate_component_name_macro(
+        root_context = generate_generate_name_macro_context(
             macro, config, manifest
         )
         self.updater = MacroGenerator(macro, root_context)
@@ -252,7 +252,7 @@ def _create_parsetime_node(
     def _context_for(
         self, parsed_node: IntermediateNode, config: ContextConfig
     ) -> Dict[str, Any]:
-        return generate_parser_model(
+        return generate_parser_model_context(
             parsed_node, self.root_project, self.manifest, config
         )