Merge pull request #2686 from fishtown-analytics/feature/override-cor…

…e-macros

include project macros in the manifest the adapter stores locally

CHANGELOG.md

@@ -3,6 +3,7 @@
 
 ### Features
 - Add support for impersonating a service account using `impersonate_service_account` in the BigQuery profile configuration ([#2677](https://github.com/fishtown-analytics/dbt/issues/2677)) ([docs](https://docs.getdbt.com/reference/warehouse-profiles/bigquery-profile#service-account-impersonation))
+- 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))
 
 
 ### Breaking changes

core/dbt/adapters/base/impl.py

@@ -158,7 +158,7 @@ def __init__(self, config):
         self.config = config
         self.cache = RelationsCache()
         self.connections = self.ConnectionManager(config)
-        self._internal_manifest_lazy: Optional[Manifest] = None
+        self._macro_manifest_lazy: Optional[Manifest] = None
 
     ###
     # Methods that pass through to the connection manager
@@ -239,24 +239,30 @@ def type(cls) -> str:
         return cls.ConnectionManager.TYPE
 
     @property
-    def _internal_manifest(self) -> Manifest:
-        if self._internal_manifest_lazy is None:
-            return self.load_internal_manifest()
-        return self._internal_manifest_lazy
+    def _macro_manifest(self) -> Manifest:
+        if self._macro_manifest_lazy is None:
+            return self.load_macro_manifest()
+        return self._macro_manifest_lazy
 
-    def check_internal_manifest(self) -> Optional[Manifest]:
+    def check_macro_manifest(self) -> Optional[Manifest]:
         """Return the internal manifest (used for executing macros) if it's
         been initialized, otherwise return None.
         """
-        return self._internal_manifest_lazy
+        return self._macro_manifest_lazy
 
-    def load_internal_manifest(self) -> Manifest:
-        if self._internal_manifest_lazy is None:
+    def load_macro_manifest(self) -> Manifest:
+        if self._macro_manifest_lazy is None:
             # avoid a circular import
-            from dbt.parser.manifest import load_internal_manifest
-            manifest = load_internal_manifest(self.config)
-            self._internal_manifest_lazy = manifest
-        return self._internal_manifest_lazy
+            from dbt.parser.manifest import load_macro_manifest
+            manifest = load_macro_manifest(
+                self.config, self.connections.set_query_header
+            )
+            self._macro_manifest_lazy = manifest
+        return self._macro_manifest_lazy
+
+    def clear_macro_manifest(self):
+        if self._macro_manifest_lazy is not None:
+            self._macro_manifest_lazy = None
 
     ###
     # Caching methods
@@ -941,7 +947,7 @@ def execute_macro(
             context_override = {}
 
         if manifest is None:
-            manifest = self._internal_manifest
+            manifest = self._macro_manifest
 
         macro = manifest.find_macro_by_name(
             macro_name, self.config.project_name, project

core/dbt/config/runtime.py

@@ -355,6 +355,9 @@ def load_dependencies(self) -> Mapping[str, 'RuntimeConfig']:
             self.dependencies = all_projects
         return self.dependencies
 
+    def clear_dependencies(self):
+        self.dependencies = None
+
     def load_projects(
         self, paths: Iterable[Path]
     ) -> Iterator[Tuple[str, 'RuntimeConfig']]:

core/dbt/parser/manifest.py

@@ -11,8 +11,6 @@
 from dbt import deprecations
 from dbt.adapters.factory import (
     get_relation_class_by_name,
-    get_adapter_package_names,
-    get_include_paths,
 )
 from dbt.helper_types import PathSet
 from dbt.logger import GLOBAL_LOGGER as logger, DbtProcessState
@@ -123,28 +121,6 @@ def __init__(
         )
         self._loaded_file_cache: Dict[str, FileBlock] = {}
 
-    def _load_macros(
-        self,
-        old_results: Optional[ParseResult],
-        internal_manifest: Optional[Manifest] = None,
-    ) -> None:
-        projects = self.all_projects
-        if internal_manifest is not None:
-            # skip internal packages
-            packages = get_adapter_package_names(
-                self.root_project.credentials.type
-            )
-            projects = {
-                k: v for k, v in self.all_projects.items() if k not in packages
-            }
-            self.results.macros.update(internal_manifest.macros)
-            self.results.files.update(internal_manifest.files)
-
-        for project in projects.values():
-            parser = MacroParser(self.results, project)
-            for path in parser.search():
-                self.parse_with_cache(path, parser, old_results)
-
     def parse_with_cache(
         self,
         path: FilePath,
@@ -201,25 +177,26 @@ def parse_project(
 
     def load_only_macros(self) -> Manifest:
         old_results = self.read_parse_results()
-        self._load_macros(old_results, internal_manifest=None)
+
+        for project in self.all_projects.values():
+            parser = MacroParser(self.results, project)
+            for path in parser.search():
+                self.parse_with_cache(path, parser, old_results)
+
         # make a manifest with just the macros to get the context
         macro_manifest = Manifest.from_macros(
             macros=self.results.macros,
             files=self.results.files
         )
+        self.macro_hook(macro_manifest)
         return macro_manifest
 
-    def load(self, internal_manifest: Optional[Manifest] = None):
+    def load(self, macro_manifest: Manifest):
         old_results = self.read_parse_results()
         if old_results is not None:
             logger.debug('Got an acceptable cached parse result')
-        self._load_macros(old_results, internal_manifest=internal_manifest)
-        # make a manifest with just the macros to get the context
-        macro_manifest = Manifest.from_macros(
-            macros=self.results.macros,
-            files=self.results.files
-        )
-        self.macro_hook(macro_manifest)
+        self.results.macros.update(macro_manifest.macros)
+        self.results.files.update(macro_manifest.files)
 
         for project in self.all_projects.values():
             # parse a single project
@@ -352,7 +329,7 @@ def create_manifest(self) -> Manifest:
     def load_all(
         cls,
         root_config: RuntimeConfig,
-        internal_manifest: Optional[Manifest],
+        macro_manifest: Manifest,
         macro_hook: Callable[[Manifest], Any],
     ) -> Manifest:
         with PARSING_STATE:
@@ -367,18 +344,22 @@ def load_all(
                     project_names=''.join(v1_configs)
                 )
             loader = cls(root_config, projects, macro_hook)
-            loader.load(internal_manifest=internal_manifest)
+            loader.load(macro_manifest=macro_manifest)
             loader.write_parse_results()
             manifest = loader.create_manifest()
             _check_manifest(manifest, root_config)
             manifest.build_flat_graph()
             return manifest
 
     @classmethod
-    def load_internal(cls, root_config: RuntimeConfig) -> Manifest:
+    def load_macros(
+        cls,
+        root_config: RuntimeConfig,
+        macro_hook: Callable[[Manifest], Any],
+    ) -> Manifest:
         with PARSING_STATE:
-            projects = load_internal_projects(root_config)
-            loader = cls(root_config, projects)
+            projects = root_config.load_dependencies()
+            loader = cls(root_config, projects, macro_hook)
             return loader.load_only_macros()
 
 
@@ -681,18 +662,16 @@ def process_node(
     _process_docs_for_node(ctx, node)
 
 
-def load_internal_projects(config):
-    project_paths = get_include_paths(config.credentials.type)
-    return dict(_load_projects(config, project_paths))
-
-
-def load_internal_manifest(config: RuntimeConfig) -> Manifest:
-    return ManifestLoader.load_internal(config)
+def load_macro_manifest(
+    config: RuntimeConfig,
+    macro_hook: Callable[[Manifest], Any],
+) -> Manifest:
+    return ManifestLoader.load_macros(config, macro_hook)
 
 
 def load_manifest(
     config: RuntimeConfig,
-    internal_manifest: Optional[Manifest],
+    macro_manifest: Manifest,
     macro_hook: Callable[[Manifest], Any],
 ) -> Manifest:
-    return ManifestLoader.load_all(config, internal_manifest, macro_hook)
+    return ManifestLoader.load_all(config, macro_manifest, macro_hook)

core/dbt/perf_utils.py

@@ -7,17 +7,26 @@
 from dbt.config import RuntimeConfig
 
 
-def get_full_manifest(config: RuntimeConfig) -> Manifest:
+def get_full_manifest(
+    config: RuntimeConfig,
+    *,
+    reset: bool = False,
+) -> Manifest:
     """Load the full manifest, using the adapter's internal manifest if it
     exists to skip parsing internal (dbt + plugins) macros a second time.
 
     Also, make sure that we force-laod the adapter's manifest, so it gets
     attached to the adapter for any methods that need it.
     """
     adapter = get_adapter(config)  # type: ignore
-    internal: Manifest = adapter.load_internal_manifest()
+    if reset:
+        config.clear_dependencies()
+        adapter.clear_macro_manifest()
 
-    def set_header(manifest: Manifest) -> None:
-        adapter.connections.set_query_header(manifest)
+    internal: Manifest = adapter.load_macro_manifest()
 
-    return load_manifest(config, internal, set_header)
+    return load_manifest(
+        config,
+        internal,
+        adapter.connections.set_query_header,
+    )

core/dbt/rpc/task_manager.py

@@ -8,6 +8,7 @@
 
 import dbt.exceptions
 import dbt.flags as flags
+from dbt.adapters.factory import reset_adapters, register_adapter
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.rpc import (
     LastParse,
@@ -126,6 +127,8 @@ def reload_manifest(self) -> bool:
     def reload_config(self):
         config = self.config.from_args(self.args)
         self.config = config
+        reset_adapters()
+        register_adapter(config)
         return config
 
     def add_request(self, request_handler: TaskHandlerProtocol):
@@ -184,7 +187,7 @@ def set_parsing(self) -> bool:
         return True
 
     def parse_manifest(self) -> None:
-        self.manifest = get_full_manifest(self.config)
+        self.manifest = get_full_manifest(self.config, reset=True)
 
     def set_compile_exception(self, exc, logs=List[LogMessage]) -> None:
         assert self.last_parse.state == ManifestStatus.Compiling, \
@@ -227,6 +230,7 @@ def get_handler(
             return None
 
         task = self.rpc_task(method)
+
         return task
 
     def task_table(self) -> List[TaskRow]:

core/dbt/task/rpc/cli.py

@@ -104,7 +104,9 @@ def handle_request(self) -> Result:
         if dumped != self.args.vars:
             self.real_task.args.vars = dumped
             if isinstance(self.real_task, RemoteManifestMethod):
-                self.real_task.manifest = get_full_manifest(self.config)
+                self.real_task.manifest = get_full_manifest(
+                    self.config, reset=True
+                )
 
         # we parsed args from the cli, so we're set on that front
         return self.real_task.handle_request()

core/dbt/utils.py

@@ -320,7 +320,7 @@ def default(self, obj):
         if hasattr(obj, 'to_dict'):
             # if we have a to_dict we should try to serialize the result of
             # that!
-            obj = obj.to_dict()
+            return obj.to_dict()
         return super().default(obj)
 
 

test/integration/016_macro_tests/override-get-columns-macros/macros.sql

@@ -0,0 +1,3 @@
+{% macro get_columns_in_relation(relation) %}
+	{{ return('a string') }}
+{% endmacro %}

test/integration/016_macro_tests/override-get-columns-models/model.sql

@@ -0,0 +1,5 @@
+{% set result = adapter.get_columns_in_relation(this) %}
+{% if execute and result != 'a string' %}
+  {% do exceptions.raise_compiler_error('overriding get_columns_in_relation failed') %}
+{% endif %}
+select 1 as id

test/integration/016_macro_tests/test_macros.py

@@ -93,3 +93,27 @@ def test_postgres_invalid_macro(self):
             self.run_dbt(['run'])
 
         assert "In dispatch: No macro named 'dispatch_to_nowhere' found" in str(exc.value)
+
+
+class TestMacroOverrideBuiltin(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "test_macros_016"
+
+    @property
+    def models(self):
+        return 'override-get-columns-models'
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': ['override-get-columns-macros'],
+        }
+
+
+    @use_profile('postgres')
+    def test_postgres_overrides(self):
+        # the first time, the model doesn't exist
+        self.run_dbt()
+        self.run_dbt()

test/unit/test_graph.py

@@ -104,7 +104,7 @@ def _mock_parse_result(config, all_projects):
         self.mock_source_file = self.load_source_file_patcher.start()
         self.mock_source_file.side_effect = lambda path: [n for n in self.mock_models if n.path == path][0]
 
-        self.internal_manifest = Manifest.from_macros(macros={
+        self.macro_manifest = Manifest.from_macros(macros={
             n.unique_id: n for n in generate_name_macros('test_models_compile')
         })
 
@@ -161,7 +161,7 @@ def use_models(self, models):
 
     def load_manifest(self, config):
         loader = dbt.parser.manifest.ManifestLoader(config, {config.project_name: config})
-        loader.load(internal_manifest=self.internal_manifest)
+        loader.load(macro_manifest=self.macro_manifest)
         return loader.create_manifest()
 
     def test__single_model(self):
@@ -319,7 +319,7 @@ def test__partial_parse(self):
         config = self.get_config()
 
         loader = dbt.parser.manifest.ManifestLoader(config, {config.project_name: config})
-        loader.load(internal_manifest=self.internal_manifest)
+        loader.load(macro_manifest=self.macro_manifest)
         loader.create_manifest()
         results = loader.results