Repository: dbt-labs/dbt-core
Pull request: Create adapter.dispatch
resolves #2302 (the rest of it)


### Description
This is a follow-up to #2675.

This PR adds the `adapter.dispatch` described in #2302, and marked `adapter_macro` as deprecated. This is a pretty straightforward PR. The dispatch method happens on the "database wrapper" rather than the actual adapter, as adapters themselves don't know about manifests/namespacing. The adapter_macro context method now calls into dispatch.

I had to massage some of the error messaging, but I think it's ok.


A large part of this PR is just converting `adapter_macro(foo, *args, **kwargs)` calls into `adapter.dispatch(foo)(*args, **kwargs)`.

### Checklist
 - [X] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
 - [X] I have run this code in development and it appears to resolve the stated issue
 - [X] This PR includes tests, or tests are not required/relevant for this PR
 - [X] I have updated the `CHANGELOG.md` and added information about my change to the "dbt next" section.


Problem statement:
Port adapter_macro implementation into Python
### Describe the feature
Right now, the `adapter_macro` implementation is written in jinja. It doesn't really do any string templating, so it's a better candidate for Python code. Rewriting this macro in Python would let us add some more comprehensive unit tests, and be more flexible with the implementation.

In addition to moving this macro to Python, I think we should rename it to `dispatch`, and it should live on the adapter object. Example usage:

```
{% macro debug(args...) %}
  {% set macro = adapter.dispatch(macro_name='debug', packages=[...]) %}
  {% do return macro(args...) %}
{% endmacro %}
```

Here, `dispatch` takes two arguments:
1. The macro name to dispatch
2. A list of of packages/namespaces to search for the macro in

The macro resolution order should largely be the same as the current implementation with one notable difference. Instead of namespacing a macro in the `macro_name` argument (eg. `dbt_utils.debug`), we'd be able to supply a list of namespaces to search in. Package maintainers could use this functionality to allow 3rd-party packages to extend the adapter-specific implementations provided in the base package.

A quick example:
```
-- dbt-utils: macros/debug.sql

{% macro get_macro_namespaces() %}
  {% do return(var('namespaces') + ['dbt_utils'] if var('namespaces') is not none else ['dbt_utils'] %}
{% endmacro %}

{% macro debug() %}
  {% set macro = adapter.dispatch(macro_name='debug', packages=get_macro_namespaces()) %}
  {% do return(macro()) %}
{% endmacro %}

{% macro default__debug() %}
  {% do log("default impl", info=true) %}
{% endmacro %}
```

```
-- spark-utils: macros/debug.sql

{% macro spark__debug() %}
  {% do log("spark impl", info=true) %}
{% endmacro %}
```

1. There's a ton of opportunity for syntactic sugar here, so take this as a thing we could support, not a final spec!
2. We can probably do a better job than what's described in `get_macro_namespaces` above

Hints from issue discussion:

Commit message: pr feedback: better errors for dotted macro names

add more tests

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d18c4c1d245..d96802a94cb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ### 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.
+
+### Features
+- Added a `dispatch` method to the context adapter and deprecated `adapter_macro`. ([#2302](https://github.com/fishtown-analytics/dbt/issues/2302), [#2679](https://github.com/fishtown-analytics/dbt/pull/2679))
 
 ## dbt 0.18.0b2 (July 30, 2020)
 
diff --git a/core/dbt/context/providers.py b/core/dbt/context/providers.py
index a515c16d78a..000f0e4caa3 100644
--- a/core/dbt/context/providers.py
+++ b/core/dbt/context/providers.py
@@ -6,16 +6,16 @@
 )
 from typing_extensions import Protocol
 
-
+from dbt import deprecations
 from dbt.adapters.base.column import Column
 from dbt.adapters.factory import get_adapter, get_adapter_package_names
 from dbt.clients import agate_helper
-from dbt.clients.jinja import get_rendered
+from dbt.clients.jinja import get_rendered, MacroGenerator
 from dbt.config import RuntimeConfig, Project
 from .base import contextmember, contextproperty, Var
 from .configured import FQNLookup
 from .context_config import ContextConfigType
-from .macros import MacroNamespaceBuilder
+from .macros import MacroNamespaceBuilder, MacroNamespace
 from .manifest import ManifestContext
 from dbt.contracts.graph.manifest import Manifest, Disabled
 from dbt.contracts.graph.compiled import (
@@ -82,9 +82,10 @@ class BaseDatabaseWrapper:
     Wrapper for runtime database interaction. Applies the runtime quote policy
     via a relation proxy.
     """
-    def __init__(self, adapter):
+    def __init__(self, adapter, namespace: MacroNamespace):
         self.adapter = adapter
         self.Relation = RelationProxy(adapter)
+        self.namespace = namespace
 
     def __getattr__(self, name):
         raise NotImplementedError('subclasses need to implement this')
@@ -99,6 +100,62 @@ def type(self):
     def commit(self):
         return self.adapter.commit_if_has_connection()
 
+    def _get_adapter_macro_prefixes(self) -> List[str]:
+        # a future version of this could have plugins automatically call fall
+        # back to their dependencies' dependencies by using
+        # `get_adapter_type_names` instead of `[self.config.credentials.type]`
+        search_prefixes = [self.adapter.type(), 'default']
+        return search_prefixes
+
+    def dispatch(
+        self, macro_name: str, packages: Optional[List[str]] = None
+    ) -> MacroGenerator:
+        search_packages: List[Optional[str]]
+
+        if '.' in macro_name:
+            suggest_package, suggest_macro_name = macro_name.split('.', 1)
+            msg = (
+                f'In adapter.dispatch, got a macro name of "{macro_name}", '
+                f'but "." is not a valid macro name component. Did you mean '
+                f'`adapter.dispatch("{suggest_macro_name}", '
+                f'packages=["{suggest_package}"])`?'
+            )
+            raise CompilationException(msg)
+
+        if packages is None:
+            search_packages = [None]
+        else:
+            search_packages = packages
+
+        attempts = []
+
+        for package_name in search_packages:
+            for prefix in self._get_adapter_macro_prefixes():
+                search_name = f'{prefix}__{macro_name}'
+                try:
+                    macro = self.namespace.get_from_package(
+                        package_name, search_name
+                    )
+                except CompilationException as exc:
+                    raise CompilationException(
+                        f'In dispatch: {exc.msg}',
+                    ) from exc
+
+                if package_name is None:
+                    attempts.append(search_name)
+                else:
+                    attempts.append(f'{package_name}.{search_name}')
+
+                if macro is not None:
+                    return macro
+
+        searched = ', '.join(repr(a) for a in attempts)
+        msg = (
+            f"In dispatch: No macro named '{macro_name}' found\n"
+            f"    Searched for: {searched}"
+        )
+        raise CompilationException(msg)
+
 
 class BaseResolver(metaclass=abc.ABCMeta):
     def __init__(self, db_wrapper, model, config, manifest):
@@ -574,7 +631,9 @@ def __init__(
         self.context_config: Optional[ContextConfigType] = context_config
         self.provider: Provider = provider
         self.adapter = get_adapter(self.config)
-        self.db_wrapper = self.provider.DatabaseWrapper(self.adapter)
+        self.db_wrapper = self.provider.DatabaseWrapper(
+            self.adapter, self.namespace
+        )
 
     def _get_namespace_builder(self):
         internal_packages = get_adapter_package_names(
@@ -1045,13 +1104,6 @@ def sql(self) -> Optional[str]:
     def sql_now(self) -> str:
         return self.adapter.date_function()
 
-    def _get_adapter_macro_prefixes(self) -> List[str]:
-        # a future version of this could have plugins automatically call fall
-        # back to their dependencies' dependencies by using
-        # `get_adapter_type_names` instead of `[self.config.credentials.type]`
-        search_prefixes = [self.config.credentials.type, 'default']
-        return search_prefixes
-
     @contextmember
     def adapter_macro(self, name: str, *args, **kwargs):
         """Find the most appropriate macro for the name, considering the
@@ -1096,38 +1148,23 @@ def adapter_macro(self, name: str, *args, **kwargs):
                ...
             {%- endmacro %}
         """
-        original_name: str = name
-        package_name: Optional[str] = None
+        deprecations.warn('adapter-macro', macro_name=name)
+        package_names: Optional[List[str]] = None
         if '.' in name:
             package_name, name = name.split('.', 1)
+            package_names = [package_name]
 
-        attempts = []
-
-        for prefix in self._get_adapter_macro_prefixes():
-            search_name = f'{prefix}__{name}'
-            try:
-                macro = self.namespace.get_from_package(
-                    package_name, search_name
-                )
-            except CompilationException as exc:
-                raise CompilationException(
-                    f'In adapter_macro: {exc.msg}, original name '
-                    f"'{original_name}'",
-                    node=self.model,
-                ) from exc
-            if package_name is None:
-                attempts.append(search_name)
-            else:
-                attempts.append(f'{package_name}.{search_name}')
-            if macro is not None:
-                return macro(*args, **kwargs)
-
-        searched = ', '.join(repr(a) for a in attempts)
-        raise_compiler_error(
-            f"In adapter_macro: No macro named '{name}' found\n"
-            f"    Original name: '{original_name}'\n"
-            f"    Searched for: {searched}"
-        )
+        try:
+            macro = self.db_wrapper.dispatch(
+                macro_name=name, packages=package_names
+            )
+        except CompilationException as exc:
+            raise CompilationException(
+                f'In adapter_macro: {exc.msg}\n'
+                f"    Original name: '{name}'",
+                node=self.model
+            ) from exc
+        return macro(*args, **kwargs)
 
 
 class MacroContext(ProviderContext):
diff --git a/core/dbt/deprecations.py b/core/dbt/deprecations.py
index 7ee620e5665..7fae52431c2 100644
--- a/core/dbt/deprecations.py
+++ b/core/dbt/deprecations.py
@@ -116,6 +116,15 @@ class ExecuteMacrosReleaseDeprecation(DBTDeprecation):
     '''
 
 
+class AdapterMacroDeprecation(DBTDeprecation):
+    _name = 'adapter-macro'
+    _description = '''\
+    The "adapter_macro" macro has been deprecated. Instead, use the
+    `adapter.dispatch` method to find a macro and call the result.
+    adapter_macro was called for: {macro_name}
+    '''
+
+
 _adapter_renamed_description = """\
 The adapter function `adapter.{old_name}` is deprecated and will be removed in
 a future release of dbt. Please use `adapter.{new_name}` instead.
@@ -160,6 +169,7 @@ def warn(name, *args, **kwargs):
     ModelsKeyNonModelDeprecation(),
     DbtProjectYamlDeprecation(),
     ExecuteMacrosReleaseDeprecation(),
+    AdapterMacroDeprecation(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {
diff --git a/core/dbt/include/global_project/macros/adapters/common.sql b/core/dbt/include/global_project/macros/adapters/common.sql
index 5e0860c5c82..bc6eeb5b24e 100644
--- a/core/dbt/include/global_project/macros/adapters/common.sql
+++ b/core/dbt/include/global_project/macros/adapters/common.sql
@@ -1,5 +1,5 @@
 {% macro get_columns_in_query(select_sql) -%}
-  {{ return(adapter_macro('get_columns_in_query', select_sql)) }}
+  {{ return(adapter.dispatch('get_columns_in_query')(select_sql)) }}
 {% endmacro %}
 
 {% macro default__get_columns_in_query(select_sql) %}
@@ -15,7 +15,7 @@
 {% endmacro %}
 
 {% macro create_schema(relation) -%}
-  {{ adapter_macro('create_schema', relation) }}
+  {{ adapter.dispatch('create_schema')(relation) }}
 {% endmacro %}
 
 {% macro default__create_schema(relation) -%}
@@ -25,7 +25,7 @@
 {% endmacro %}
 
 {% macro drop_schema(relation) -%}
-  {{ adapter_macro('drop_schema', relation) }}
+  {{ adapter.dispatch('drop_schema')(relation) }}
 {% endmacro %}
 
 {% macro default__drop_schema(relation) -%}
@@ -35,7 +35,7 @@
 {% endmacro %}
 
 {% macro create_table_as(temporary, relation, sql) -%}
-  {{ adapter_macro('create_table_as', temporary, relation, sql) }}
+  {{ adapter.dispatch('create_table_as')(temporary, relation, sql) }}
 {%- endmacro %}
 
 {% macro default__create_table_as(temporary, relation, sql) -%}
@@ -52,7 +52,7 @@
 {% endmacro %}
 
 {% macro create_view_as(relation, sql) -%}
-  {{ adapter_macro('create_view_as', relation, sql) }}
+  {{ adapter.dispatch('create_view_as')(relation, sql) }}
 {%- endmacro %}
 
 {% macro default__create_view_as(relation, sql) -%}
@@ -66,7 +66,7 @@
 
 
 {% macro get_catalog(information_schema, schemas) -%}
-  {{ return(adapter_macro('get_catalog', information_schema, schemas)) }}
+  {{ return(adapter.dispatch('get_catalog')(information_schema, schemas)) }}
 {%- endmacro %}
 
 {% macro default__get_catalog(information_schema, schemas) -%}
@@ -81,7 +81,7 @@
 
 
 {% macro get_columns_in_relation(relation) -%}
-  {{ return(adapter_macro('get_columns_in_relation', relation)) }}
+  {{ return(adapter.dispatch('get_columns_in_relation')(relation)) }}
 {% endmacro %}
 
 {% macro sql_convert_columns_in_relation(table) -%}
@@ -98,13 +98,13 @@
 {% endmacro %}
 
 {% macro alter_column_type(relation, column_name, new_column_type) -%}
-  {{ return(adapter_macro('alter_column_type', relation, column_name, new_column_type)) }}
+  {{ return(adapter.dispatch('alter_column_type')(relation, column_name, new_column_type)) }}
 {% endmacro %}
 
 
 
 {% macro alter_column_comment(relation, column_dict) -%}
-  {{ return(adapter_macro('alter_column_comment', relation, column_dict)) }}
+  {{ return(adapter.dispatch('alter_column_comment')(relation, column_dict)) }}
 {% endmacro %}
 
 {% macro default__alter_column_comment(relation, column_dict) -%}
@@ -113,7 +113,7 @@
 {% endmacro %}
 
 {% macro alter_relation_comment(relation, relation_comment) -%}
-  {{ return(adapter_macro('alter_relation_comment', relation, relation_comment)) }}
+  {{ return(adapter.dispatch('alter_relation_comment')(relation, relation_comment)) }}
 {% endmacro %}
 
 {% macro default__alter_relation_comment(relation, relation_comment) -%}
@@ -122,7 +122,7 @@
 {% endmacro %}
 
 {% macro persist_docs(relation, model, for_relation=true, for_columns=true) -%}
-  {{ return(adapter_macro('persist_docs', relation, model, for_relation, for_columns)) }}
+  {{ return(adapter.dispatch('persist_docs')(relation, model, for_relation, for_columns)) }}
 {% endmacro %}
 
 {% macro default__persist_docs(relation, model, for_relation, for_columns) -%}
@@ -157,7 +157,7 @@
 
 
 {% macro drop_relation(relation) -%}
-  {{ return(adapter_macro('drop_relation', relation)) }}
+  {{ return(adapter.dispatch('drop_relation')(relation)) }}
 {% endmacro %}
 
 
@@ -168,7 +168,7 @@
 {% endmacro %}
 
 {% macro truncate_relation(relation) -%}
-  {{ return(adapter_macro('truncate_relation', relation)) }}
+  {{ return(adapter.dispatch('truncate_relation')(relation)) }}
 {% endmacro %}
 
 
@@ -179,7 +179,7 @@
 {% endmacro %}
 
 {% macro rename_relation(from_relation, to_relation) -%}
-  {{ return(adapter_macro('rename_relation', from_relation, to_relation)) }}
+  {{ return(adapter.dispatch('rename_relation')(from_relation, to_relation)) }}
 {% endmacro %}
 
 {% macro default__rename_relation(from_relation, to_relation) -%}
@@ -191,7 +191,7 @@
 
 
 {% macro information_schema_name(database) %}
-  {{ return(adapter_macro('information_schema_name', database)) }}
+  {{ return(adapter.dispatch('information_schema_name')(database)) }}
 {% endmacro %}
 
 {% macro default__information_schema_name(database) -%}
@@ -204,7 +204,7 @@
 
 
 {% macro list_schemas(database) -%}
-  {{ return(adapter_macro('list_schemas', database)) }}
+  {{ return(adapter.dispatch('list_schemas')(database)) }}
 {% endmacro %}
 
 {% macro default__list_schemas(database) -%}
@@ -218,7 +218,7 @@
 
 
 {% macro check_schema_exists(information_schema, schema) -%}
-  {{ return(adapter_macro('check_schema_exists', information_schema, schema)) }}
+  {{ return(adapter.dispatch('check_schema_exists')(information_schema, schema)) }}
 {% endmacro %}
 
 {% macro default__check_schema_exists(information_schema, schema) -%}
@@ -233,7 +233,7 @@
 
 
 {% macro list_relations_without_caching(schema_relation) %}
-  {{ return(adapter_macro('list_relations_without_caching', schema_relation)) }}
+  {{ return(adapter.dispatch('list_relations_without_caching')(schema_relation)) }}
 {% endmacro %}
 
 
@@ -244,7 +244,7 @@
 
 
 {% macro current_timestamp() -%}
-  {{ adapter_macro('current_timestamp') }}
+  {{ adapter.dispatch('current_timestamp')() }}
 {%- endmacro %}
 
 
@@ -255,7 +255,7 @@
 
 
 {% macro collect_freshness(source, loaded_at_field, filter) %}
-  {{ return(adapter_macro('collect_freshness', source, loaded_at_field, filter))}}
+  {{ return(adapter.dispatch('collect_freshness')(source, loaded_at_field, filter))}}
 {% endmacro %}
 
 
@@ -273,7 +273,7 @@
 {% endmacro %}
 
 {% macro make_temp_relation(base_relation, suffix='__dbt_tmp') %}
-  {{ return(adapter_macro('make_temp_relation', base_relation, suffix))}}
+  {{ return(adapter.dispatch('make_temp_relation')(base_relation, suffix))}}
 {% endmacro %}
 
 {% macro default__make_temp_relation(base_relation, suffix) %}
diff --git a/core/dbt/include/global_project/macros/etc/get_custom_database.sql b/core/dbt/include/global_project/macros/etc/get_custom_database.sql
index bf12e366f5c..20eed34a2f7 100644
--- a/core/dbt/include/global_project/macros/etc/get_custom_database.sql
+++ b/core/dbt/include/global_project/macros/etc/get_custom_database.sql
@@ -14,7 +14,7 @@
 
 #}
 {% macro generate_database_name(custom_database_name=none, node=none) -%}
-    {% do return(adapter_macro('generate_database_name', custom_database_name, node)) %}
+    {% do return(adapter.dispatch('generate_database_name')(custom_database_name, node)) %}
 {%- endmacro %}
 
 {% macro default__generate_database_name(custom_database_name=none, node=none) -%}
diff --git a/core/dbt/include/global_project/macros/materializations/common/merge.sql b/core/dbt/include/global_project/macros/materializations/common/merge.sql
index 778fdf8ac72..b2f9c7f4d8e 100644
--- a/core/dbt/include/global_project/macros/materializations/common/merge.sql
+++ b/core/dbt/include/global_project/macros/materializations/common/merge.sql
@@ -1,17 +1,17 @@
 
 
 {% macro get_merge_sql(target, source, unique_key, dest_columns, predicates=none) -%}
-  {{ adapter_macro('get_merge_sql', target, source, unique_key, dest_columns, predicates) }}
+  {{ adapter.dispatch('get_merge_sql')(target, source, unique_key, dest_columns, predicates) }}
 {%- endmacro %}
 
 
 {% macro get_delete_insert_merge_sql(target, source, unique_key, dest_columns) -%}
-  {{ adapter_macro('get_delete_insert_merge_sql', target, source, unique_key, dest_columns) }}
+  {{ adapter.dispatch('get_delete_insert_merge_sql')(target, source, unique_key, dest_columns) }}
 {%- endmacro %}
 
 
 {% macro get_insert_overwrite_merge_sql(target, source, dest_columns, predicates, include_sql_header=false) -%}
-  {{ adapter_macro('get_insert_overwrite_merge_sql', target, source, dest_columns, predicates, include_sql_header) }}
+  {{ adapter.dispatch('get_insert_overwrite_merge_sql')(target, source, dest_columns, predicates, include_sql_header) }}
 {%- endmacro %}
 
 
@@ -97,7 +97,7 @@
     merge into {{ target }} as DBT_INTERNAL_DEST
         using {{ source }} as DBT_INTERNAL_SOURCE
         on FALSE
-    
+
     when not matched by source
         {% if predicates %} and {{ predicates | join(' and ') }} {% endif %}
         then delete
diff --git a/core/dbt/include/global_project/macros/materializations/seed/seed.sql b/core/dbt/include/global_project/macros/materializations/seed/seed.sql
index ad2221c359c..41482e06df3 100644
--- a/core/dbt/include/global_project/macros/materializations/seed/seed.sql
+++ b/core/dbt/include/global_project/macros/materializations/seed/seed.sql
@@ -1,14 +1,14 @@
 
 {% macro create_csv_table(model, agate_table) -%}
-  {{ adapter_macro('create_csv_table', model, agate_table) }}
+  {{ adapter.dispatch('create_csv_table')(model, agate_table) }}
 {%- endmacro %}
 
 {% macro reset_csv_table(model, full_refresh, old_relation, agate_table) -%}
-  {{ adapter_macro('reset_csv_table', model, full_refresh, old_relation, agate_table) }}
+  {{ adapter.dispatch('reset_csv_table')(model, full_refresh, old_relation, agate_table) }}
 {%- endmacro %}
 
 {% macro load_csv_rows(model, agate_table) -%}
-  {{ adapter_macro('load_csv_rows', model, agate_table) }}
+  {{ adapter.dispatch('load_csv_rows')(model, agate_table) }}
 {%- endmacro %}
 
 {% macro default__create_csv_table(model, agate_table) %}
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
index 0342ffa6a61..04d2112fe51 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
+++ b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot.sql
@@ -2,7 +2,7 @@
     Add new columns to the table if applicable
 #}
 {% macro create_columns(relation, columns) %}
-  {{ adapter_macro('create_columns', relation, columns) }}
+  {{ adapter.dispatch('create_columns')(relation, columns) }}
 {% endmacro %}
 
 {% macro default__create_columns(relation, columns) %}
@@ -15,7 +15,7 @@
 
 
 {% macro post_snapshot(staging_relation) %}
-  {{ adapter_macro('post_snapshot', staging_relation) }}
+  {{ adapter.dispatch('post_snapshot')(staging_relation) }}
 {% endmacro %}
 
 {% macro default__post_snapshot(staging_relation) %}
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
index 51a205e468c..e4370196db3 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
+++ b/core/dbt/include/global_project/macros/materializations/snapshot/snapshot_merge.sql
@@ -1,6 +1,6 @@
 
 {% macro snapshot_merge_sql(target, source, insert_cols) -%}
-  {{ adapter_macro('snapshot_merge_sql', target, source, insert_cols) }}
+  {{ adapter.dispatch('snapshot_merge_sql')(target, source, insert_cols) }}
 {%- endmacro %}
 
 
diff --git a/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql b/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
index d11d7c7c4ac..94915868061 100644
--- a/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
+++ b/core/dbt/include/global_project/macros/materializations/snapshot/strategies.sql
@@ -36,7 +36,7 @@
     Create SCD Hash SQL fields cross-db
 #}
 {% macro snapshot_hash_arguments(args) -%}
-  {{ adapter_macro('snapshot_hash_arguments', args) }}
+  {{ adapter.dispatch('snapshot_hash_arguments')(args) }}
 {%- endmacro %}
 
 
@@ -52,7 +52,7 @@
     Get the current time cross-db
 #}
 {% macro snapshot_get_time() -%}
-  {{ adapter_macro('snapshot_get_time') }}
+  {{ adapter.dispatch('snapshot_get_time')() }}
 {%- endmacro %}
 
 {% macro default__snapshot_get_time() -%}
@@ -92,7 +92,7 @@
 
 
 {% macro snapshot_string_as_time(timestamp) -%}
-    {{ adapter_macro('snapshot_string_as_time', timestamp) }}
+    {{ adapter.dispatch('snapshot_string_as_time')(timestamp) }}
 {%- endmacro %}
 
 
diff --git a/core/dbt/include/global_project/macros/materializations/view/create_or_replace_view.sql b/core/dbt/include/global_project/macros/materializations/view/create_or_replace_view.sql
index 1d59f4de54f..309ca6d632a 100644
--- a/core/dbt/include/global_project/macros/materializations/view/create_or_replace_view.sql
+++ b/core/dbt/include/global_project/macros/materializations/view/create_or_replace_view.sql
@@ -1,6 +1,6 @@
 
 {% macro handle_existing_table(full_refresh, old_relation) %}
-    {{ adapter_macro("dbt.handle_existing_table", full_refresh, old_relation) }}
+    {{ adapter.dispatch("handle_existing_table", packages=['dbt'])(full_refresh, old_relation) }}
 {% endmacro %}
 
 {% macro default__handle_existing_table(full_refresh, old_relation) %}
diff --git a/test/integration/004_simple_snapshot_test/macros/test_no_overlaps.sql b/test/integration/004_simple_snapshot_test/macros/test_no_overlaps.sql
index 136c6efe5c8..02f3e50368e 100644
--- a/test/integration/004_simple_snapshot_test/macros/test_no_overlaps.sql
+++ b/test/integration/004_simple_snapshot_test/macros/test_no_overlaps.sql
@@ -1,5 +1,5 @@
 {% macro get_snapshot_unique_id() -%}
-    {{ return(adapter_macro('get_snapshot_unique_id', )) }}
+    {{ return(adapter.dispatch('get_snapshot_unique_id')()) }}
 {%- endmacro %}
 
 {% macro default__get_snapshot_unique_id() -%}
diff --git a/test/integration/012_deprecation_tests/adapter-macro-macros/macros.sql b/test/integration/012_deprecation_tests/adapter-macro-macros/macros.sql
new file mode 100644
index 00000000000..b03ca28f17d
--- /dev/null
+++ b/test/integration/012_deprecation_tests/adapter-macro-macros/macros.sql
@@ -0,0 +1,17 @@
+{% macro some_macro(arg1, arg2) -%}
+    {{ adapter_macro('some_macro', arg1, arg2) }}
+{%- endmacro %}
+
+
+{% macro default__some_macro(arg1, arg2) %}
+    {% do exceptions.raise_compiler_error('not allowed') %}
+{% endmacro %}
+
+{% macro postgres__some_macro(arg1, arg2) -%}
+    {{ arg1 }}{{ arg2 }}
+{%- endmacro %}
+
+
+{% macro some_other_macro(arg1, arg2) -%}
+	{{ adapter_macro('test.some_macro', arg1, arg2) }}
+{%- endmacro %}
diff --git a/test/integration/012_deprecation_tests/adapter-macro-models-package/model.sql b/test/integration/012_deprecation_tests/adapter-macro-models-package/model.sql
new file mode 100644
index 00000000000..7b86017b484
--- /dev/null
+++ b/test/integration/012_deprecation_tests/adapter-macro-models-package/model.sql
@@ -0,0 +1,4 @@
+{% if some_other_macro('foo', 'bar') != 'foobar' %}
+  {% do exceptions.raise_compiler_error('invalid foobar') %}
+{% endif %}
+select 1 as id
diff --git a/test/integration/012_deprecation_tests/adapter-macro-models/model.sql b/test/integration/012_deprecation_tests/adapter-macro-models/model.sql
new file mode 100644
index 00000000000..37621d3d237
--- /dev/null
+++ b/test/integration/012_deprecation_tests/adapter-macro-models/model.sql
@@ -0,0 +1,4 @@
+{% if some_macro('foo', 'bar') != 'foobar' %}
+  {% do exceptions.raise_compiler_error('invalid foobar') %}
+{% endif %}
+select 1 as id
diff --git a/test/integration/012_deprecation_tests/test_deprecations.py b/test/integration/012_deprecation_tests/test_deprecations.py
index fa47d24bd39..c996c91f0a6 100644
--- a/test/integration/012_deprecation_tests/test_deprecations.py
+++ b/test/integration/012_deprecation_tests/test_deprecations.py
@@ -104,3 +104,77 @@ def test_postgres_project_deprecations(self):
         self.run_dbt(strict=False)
         expected = {'dbt-project-yaml-v1'}
         self.assertEqual(expected, deprecations.active_deprecations)
+
+
+class TestAdapterMacroDeprecation(BaseTestDeprecations):
+    @property
+    def models(self):
+        return self.dir('adapter-macro-models')
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': [self.dir('adapter-macro-macros')]
+        }
+
+    @use_profile('postgres')
+    def test_postgres_adapter_macro(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        self.run_dbt(strict=False)
+        expected = {'adapter-macro'}
+        self.assertEqual(expected, deprecations.active_deprecations)
+
+    @use_profile('postgres')
+    def test_postgres_adapter_macro_fail(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        with self.assertRaises(dbt.exceptions.CompilationException) as exc:
+            self.run_dbt(strict=True)
+        exc_str = ' '.join(str(exc.exception).split())  # flatten all whitespace
+        assert 'The "adapter_macro" macro has been deprecated' in exc_str
+
+    @use_profile('redshift')
+    def test_redshift_adapter_macro(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        # picked up the default -> error
+        with self.assertRaises(dbt.exceptions.CompilationException) as exc:
+            self.run_dbt(strict=False, expect_pass=False)
+        exc_str = ' '.join(str(exc.exception).split())  # flatten all whitespace
+        assert 'not allowed' in exc_str  # we saw the default macro
+
+
+class TestAdapterMacroDeprecationPackages(BaseTestDeprecations):
+    @property
+    def models(self):
+        return self.dir('adapter-macro-models-package')
+
+    @property
+    def project_config(self):
+        return {
+            'config-version': 2,
+            'macro-paths': [self.dir('adapter-macro-macros')]
+        }
+
+    @use_profile('postgres')
+    def test_postgres_adapter_macro_pkg(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        self.run_dbt(strict=False)
+        expected = {'adapter-macro'}
+        self.assertEqual(expected, deprecations.active_deprecations)
+
+    @use_profile('postgres')
+    def test_postgres_adapter_macro_pkg_fail(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        with self.assertRaises(dbt.exceptions.CompilationException) as exc:
+            self.run_dbt(strict=True)
+        exc_str = ' '.join(str(exc.exception).split())  # flatten all whitespace
+        assert 'The "adapter_macro" macro has been deprecated' in exc_str
+
+    @use_profile('redshift')
+    def test_redshift_adapter_macro_pkg(self):
+        self.assertEqual(deprecations.active_deprecations, set())
+        # picked up the default -> error
+        with self.assertRaises(dbt.exceptions.CompilationException) as exc:
+            self.run_dbt(strict=False, expect_pass=False)
+        exc_str = ' '.join(str(exc.exception).split())  # flatten all whitespace
+        assert 'not allowed' in exc_str  # we saw the default macro
diff --git a/test/integration/016_macro_tests/fail-missing-macro-models/model.sql b/test/integration/016_macro_tests/fail-missing-macro-models/model.sql
new file mode 100644
index 00000000000..6faebe5385c
--- /dev/null
+++ b/test/integration/016_macro_tests/fail-missing-macro-models/model.sql
@@ -0,0 +1,2 @@
+{{ dispatch_to_nowhere() }}
+select 1 as id
diff --git a/test/integration/016_macro_tests/macros/my_macros.sql b/test/integration/016_macro_tests/macros/my_macros.sql
index e929d0d96e2..d29007557d9 100644
--- a/test/integration/016_macro_tests/macros/my_macros.sql
+++ b/test/integration/016_macro_tests/macros/my_macros.sql
@@ -13,3 +13,10 @@
     {{ ref('table_model') }}
 
 {% endmacro %}
+
+
+{# there is no no default__dispatch_to_nowhere! #}
+{% macro dispatch_to_nowhere() %}
+	{% set macro = adapter.dispatch('dispatch_to_nowhere') %}
+	{{ macro() }}
+{% endmacro %}
diff --git a/test/integration/016_macro_tests/test_macros.py b/test/integration/016_macro_tests/test_macros.py
index c40f4270073..0b856f70f23 100644
--- a/test/integration/016_macro_tests/test_macros.py
+++ b/test/integration/016_macro_tests/test_macros.py
@@ -1,5 +1,8 @@
 from test.integration.base import DBTIntegrationTest, use_profile
 
+import dbt.exceptions
+import pytest
+
 
 class TestMacros(DBTIntegrationTest):
 
@@ -70,20 +73,11 @@ def project_config(self):
 
     @use_profile('postgres')
     def test_postgres_invalid_macro(self):
-
-        try:
-            self.run_dbt(["run"], expect_pass=False)
-            self.assertTrue(False,
-                            'compiling bad macro should raise a runtime error')
-
-        except RuntimeError:
-            pass
+        with pytest.raises(RuntimeError):
+            self.run_dbt(["run"])
 
 
-class TestMisusedMacros(DBTIntegrationTest):
-
-    def setUp(self):
-        DBTIntegrationTest.setUp(self)
+class TestAdapterMacroNoDestination(DBTIntegrationTest):
 
     @property
     def schema(self):
@@ -91,36 +85,11 @@ def schema(self):
 
     @property
     def models(self):
-        return "bad-models"
-
-    @property
-    def packages_config(self):
-        return {
-            'packages': [
-                {
-                    'git': 'https://github.com/fishtown-analytics/dbt-integration-project',
-                    'revision': 'dbt/0.17.0',
-                }
-            ]
-        }
+        return "fail-missing-macro-models"
 
-    @property
-    def project_config(self):
-        return {
-            'config-version': 2,
-            "macro-paths": ["macros"],
-        }
+    @use_profile('postgres')
+    def test_postgres_invalid_macro(self):
+        with pytest.raises(dbt.exceptions.CompilationException) as exc:
+            self.run_dbt(['run'])
 
-    # TODO: compilation no longer exists, so while the model calling this macro
-    # fails, it does not raise a runtime exception. change this test to verify
-    # that the model finished with ERROR state.
-    #
-    # @use_profile('postgres')
-    # def test_working_macros(self):
-    #     self.run_dbt(["deps"])
-
-    #     try:
-    #         self.run_dbt(["run"])
-    #         self.assertTrue(False, 'invoked a package macro from global scope')
-    #     except RuntimeError:
-    #         pass
+        assert "In dispatch: No macro named 'dispatch_to_nowhere' found" in str(exc.value)
diff --git a/test/integration/026_aliases_test/macros/cast.sql b/test/integration/026_aliases_test/macros/cast.sql
index ecc06ffa62a..3dc65d2d7df 100644
--- a/test/integration/026_aliases_test/macros/cast.sql
+++ b/test/integration/026_aliases_test/macros/cast.sql
@@ -1,7 +1,7 @@
 
 
 {% macro string_literal(s) -%}
-  {{ adapter_macro('test.string_literal', s) }}
+  {{ adapter.dispatch('string_literal', packages=['test'])(s) }}
 {%- endmacro %}
 
 {% macro default__string_literal(s) %}
diff --git a/test/integration/043_custom_aliases_test/macros-configs/macros.sql b/test/integration/043_custom_aliases_test/macros-configs/macros.sql
index 5b2587147e8..dd134ee3ed9 100644
--- a/test/integration/043_custom_aliases_test/macros-configs/macros.sql
+++ b/test/integration/043_custom_aliases_test/macros-configs/macros.sql
@@ -9,7 +9,7 @@
 {%- endmacro %}
 
 {% macro string_literal(s) -%}
-  {{ adapter_macro('test.string_literal', s) }}
+  {{ adapter.dispatch('string_literal', packages=['test'])(s) }}
 {%- endmacro %}
 
 {% macro default__string_literal(s) %}
diff --git a/test/integration/043_custom_aliases_test/macros/macros.sql b/test/integration/043_custom_aliases_test/macros/macros.sql
index 17fa7f1ca5e..a5347befa78 100644
--- a/test/integration/043_custom_aliases_test/macros/macros.sql
+++ b/test/integration/043_custom_aliases_test/macros/macros.sql
@@ -9,7 +9,7 @@
 
 
 {% macro string_literal(s) -%}
-  {{ adapter_macro('test.string_literal', s) }}
+  {{ adapter.dispatch('string_literal', packages=['test'])(s) }}
 {%- endmacro %}
 
 {% macro default__string_literal(s) %}
diff --git a/test/unit/test_context.py b/test/unit/test_context.py
index 2c63911b22d..aaa83d7159a 100644
--- a/test/unit/test_context.py
+++ b/test/unit/test_context.py
@@ -6,8 +6,8 @@
 
 import pytest
 
-# make sure 'postgres' is available
-from dbt.adapters import postgres  # noqa
+# make sure 'redshift' is available
+from dbt.adapters import postgres, redshift
 from dbt.adapters import factory
 from dbt.adapters.base import AdapterConfig
 from dbt.clients.jinja import MacroStack
@@ -18,7 +18,7 @@
 from dbt.context import base, target, configured, providers, docs, manifest, macros
 from dbt.node_types import NodeType
 import dbt.exceptions
-from .utils import profile_from_dict, config_from_parts_or_dicts
+from .utils import profile_from_dict, config_from_parts_or_dicts, inject_adapter, clear_plugin
 from .mock_adapter import adapter_factory
 
 
@@ -112,7 +112,8 @@ def setUp(self):
         self.mock_config = mock.MagicMock()
         adapter_class = adapter_factory()
         self.mock_adapter = adapter_class(self.mock_config)
-        self.wrapper = providers.ParseDatabaseWrapper(self.mock_adapter)
+        self.namespace = mock.MagicMock()
+        self.wrapper = providers.ParseDatabaseWrapper(self.mock_adapter, self.namespace)
         self.responder = self.mock_adapter.responder
 
     def test_unwrapped_method(self):
@@ -131,7 +132,8 @@ def setUp(self):
         self.mock_config.quoting = {'database': True, 'schema': True, 'identifier': True}
         adapter_class = adapter_factory()
         self.mock_adapter = adapter_class(self.mock_config)
-        self.wrapper = providers.RuntimeDatabaseWrapper(self.mock_adapter)
+        self.namespace = mock.MagicMock()
+        self.wrapper = providers.RuntimeDatabaseWrapper(self.mock_adapter, self.namespace)
         self.responder = self.mock_adapter.responder
 
     def test_unwrapped_method(self):
@@ -369,6 +371,39 @@ def manifest_fx(config):
     return mock_manifest(config)
 
 
+@pytest.fixture
+def manifest_extended(manifest_fx):
+    dbt_macro = mock_macro('default__some_macro', 'dbt')
+    # same namespace, same name, different pkg!
+    rs_macro = mock_macro('redshift__some_macro', 'dbt_redshift')
+    # same name, different package
+    package_default_macro = mock_macro('default__some_macro', 'root')
+    package_rs_macro = mock_macro('redshift__some_macro', 'root')
+    manifest_fx.macros[dbt_macro.unique_id] = dbt_macro
+    manifest_fx.macros[rs_macro.unique_id] = rs_macro
+    manifest_fx.macros[package_default_macro.unique_id] = package_default_macro
+    manifest_fx.macros[package_rs_macro.unique_id] = package_rs_macro
+    return manifest_fx
+
+
+@pytest.fixture
+def redshift_adapter(config, get_adapter):
+    adapter = redshift.RedshiftAdapter(config)
+    inject_adapter(adapter, redshift.Plugin)
+    get_adapter.return_value = adapter
+    yield adapter
+    clear_plugin(redshift.Plugin)
+
+
+@pytest.fixture
+def postgres_adapter(config, get_adapter):
+    adapter = postgres.PostgresAdapter(config)
+    inject_adapter(adapter, postgres.Plugin)
+    get_adapter.return_value = adapter
+    yield adapter
+    clear_plugin(postgres.Plugin)
+
+
 def test_query_header_context(config, manifest_fx):
     ctx = manifest.generate_query_header_context(
         config=config,
@@ -463,3 +498,62 @@ def test_macro_namespace(config, manifest_fx):
         assert result['root']['some_macro'].macro is package_macro
         assert result['some_macro'].macro is package_macro
 
+
+def test_resolve_specific(config, manifest_extended, redshift_adapter, get_include_paths):
+    rs_macro = manifest_extended.macros['macro.dbt_redshift.redshift__some_macro']
+    package_rs_macro = manifest_extended.macros['macro.root.redshift__some_macro']
+
+    ctx = providers.generate_runtime_model(
+        model=mock_model(),
+        config=config,
+        manifest=manifest_extended,
+    )
+
+    # macro_a exists, but default__macro_a and redshift__macro_a do not
+    with pytest.raises(dbt.exceptions.CompilationException):
+        ctx['adapter'].dispatch('macro_a').macro
+
+    assert ctx['adapter'].dispatch('some_macro').macro is package_rs_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['dbt']).macro is rs_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['root']).macro is package_rs_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['root', 'dbt']).macro is package_rs_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['dbt', 'root']).macro is rs_macro
+
+
+def test_resolve_default(config, manifest_extended, postgres_adapter, get_include_paths):
+    dbt_macro = manifest_extended.macros['macro.dbt.default__some_macro']
+    package_macro = manifest_extended.macros['macro.root.default__some_macro']
+
+    ctx = providers.generate_runtime_model(
+        model=mock_model(),
+        config=config,
+        manifest=manifest_extended,
+    )
+
+    # macro_a exists, but default__macro_a and redshift__macro_a do not
+    with pytest.raises(dbt.exceptions.CompilationException):
+        ctx['adapter'].dispatch('macro_a').macro
+
+    assert ctx['adapter'].dispatch('some_macro').macro is package_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['dbt']).macro is dbt_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['root']).macro is package_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['root', 'dbt']).macro is package_macro
+    assert ctx['adapter'].dispatch('some_macro', packages=['dbt', 'root']).macro is dbt_macro
+
+
+def test_resolve_errors(config, manifest_extended, redshift_adapter, get_include_paths):
+    ctx = providers.generate_runtime_model(
+        model=mock_model(),
+        config=config,
+        manifest=manifest_extended,
+    )
+    with pytest.raises(dbt.exceptions.CompilationException) as exc:
+        ctx['adapter'].dispatch('no_default_exists')
+    assert 'default__no_default_exists' in str(exc.value)
+    assert 'redshift__no_default_exists' in str(exc.value)
+
+    with pytest.raises(dbt.exceptions.CompilationException) as exc:
+        ctx['adapter'].dispatch('namespace.no_default_exists')
+    assert '"." is not a valid macro name component' in str(exc.value)
+    assert 'adapter.dispatch' in str(exc.value)
+    assert 'packages=["namespace"]' in str(exc.value)
diff --git a/test/unit/utils.py b/test/unit/utils.py
index 76971476db0..64208bee0f2 100644
--- a/test/unit/utils.py
+++ b/test/unit/utils.py
@@ -121,6 +121,13 @@ def inject_adapter(value, plugin):
     FACTORY.adapters[key] = value
 
 
+def clear_plugin(plugin):
+    from dbt.adapters.factory import FACTORY
+    key = plugin.adapter.type()
+    FACTORY.plugins.pop(key, None)
+    FACTORY.adapters.pop(key, None)
+
+
 class ContractTestCase(TestCase):
     ContractType = None
 


Code review comments:
diff_hunk: @@ -1096,38 +1148,23 @@ def adapter_macro(self, name: str, *args, **kwargs):
                ...
             {%- endmacro %}
         """
-        original_name: str = name
-        package_name: Optional[str] = None
+        deprecations.warn('adapter-macro', macro_name=name)
+        package_names: Optional[List[str]] = None
         if '.' in name:
             package_name, name = name.split('.', 1)
+            package_names = [package_name]
 
-        attempts = []
-
-        for prefix in self._get_adapter_macro_prefixes():
-            search_name = f'{prefix}__{name}'
-            try:
-                macro = self.namespace.get_from_package(
-                    package_name, search_name
-                )
-            except CompilationException as exc:
-                raise CompilationException(
-                    f'In adapter_macro: {exc.msg}, original name '
-                    f"'{original_name}'",
-                    node=self.model,
-                ) from exc
-            if package_name is None:
-                attempts.append(search_name)
-            else:
-                attempts.append(f'{package_name}.{search_name}')
-            if macro is not None:
-                return macro(*args, **kwargs)
-
-        searched = ', '.join(repr(a) for a in attempts)
-        raise_compiler_error(
-            f"In adapter_macro: No macro named '{name}' found\n"
-            f"    Original name: '{original_name}'\n"
-            f"    Searched for: {searched}"
-        )
+        try:
+            macro = self.db_wrapper.dispatch(
+                macro_name=name, packages=package_names
+            )
+        except CompilationException as exc:
+            raise CompilationException(
+                f'In adapter_macro: {exc.msg}\n'
+                f"    Original name: '{name}'",
path: core/dbt/context/providers.py
text: @user1:
Could we make this the full original macro name, including the package namespace if it included a `.`?