Merge branch 'dev/wilt-chamberlain' into feature/rpc-tasks

.bumpversion.cfg

@@ -1,25 +1,27 @@
 [bumpversion]
-current_version = 0.13.0a1
+current_version = 0.13.0a2
 parse = (?P<major>\d+)
 	\.(?P<minor>\d+)
 	\.(?P<patch>\d+)
 	((?P<prerelease>[a-z]+)(?P<num>\d+))?
-serialize = 
+serialize =
 	{major}.{minor}.{patch}{prerelease}{num}
 	{major}.{minor}.{patch}
 commit = False
 tag = False
 
 [bumpversion:part:prerelease]
 first_value = a
-values = 
+values =
 	a
 	b
 	rc
 
 [bumpversion:part:num]
 first_value = 1
 
+[bumpversion:file:setup.py]
+
 [bumpversion:file:core/setup.py]
 
 [bumpversion:file:core/dbt/version.py]
@@ -31,4 +33,3 @@ first_value = 1
 [bumpversion:file:plugins/snowflake/setup.py]
 
 [bumpversion:file:plugins/bigquery/setup.py]
-

core/dbt/adapters/base/impl.py

@@ -744,9 +744,16 @@ def execute_macro(self, macro_name, manifest=None, project=None,
 
         macro = manifest.find_macro_by_name(macro_name, project)
         if macro is None:
+            if project is None:
+                package_name = 'any package'
+            else:
+                package_name = 'the "{}" package'.format(project)
+
+            # The import of dbt.context.runtime below shadows 'dbt'
+            import dbt.exceptions
             raise dbt.exceptions.RuntimeException(
-                'Could not find macro with name {} in project {}'
-                .format(macro_name, project)
+                'dbt could not find a macro with the name "{}" in {}'
+                .format(macro_name, package_name)
             )
 
         # This causes a reference cycle, as dbt.context.runtime.generate()

core/dbt/clients/registry.py

@@ -3,6 +3,7 @@
 import requests
 from dbt.exceptions import RegistryException
 from dbt.utils import memoized
+from dbt.logger import GLOBAL_LOGGER as logger
 import os
 
 if os.getenv('DBT_PACKAGE_HUB_URL'):
@@ -32,7 +33,10 @@ def wrapper(*args, **kwargs):
 @_wrap_exceptions
 def _get(path, registry_base_url=None):
     url = _get_url(path, registry_base_url)
+    logger.debug('Making package registry request: GET {}'.format(url))
     resp = requests.get(url)
+    logger.debug('Response from registry: GET {} {}'.format(url,
+                                                            resp.status_code))
     resp.raise_for_status()
     return resp.json()
 

core/dbt/compilation.py

@@ -36,6 +36,7 @@ def print_compile_stats(stats):
         NodeType.Macro: 'macros',
         NodeType.Operation: 'operations',
         NodeType.Seed: 'seed files',
+        NodeType.Source: 'sources',
     }
 
     results = {k: 0 for k in names.keys()}

core/dbt/contracts/results.py

@@ -261,8 +261,8 @@ class SourceFreshnessResult(NodeSerializable):
     def __init__(self, node, max_loaded_at, snapshotted_at,
                  age, status, thread_id, error=None,
                  timing=None, execution_time=0):
-        max_loaded_at = max_loaded_at.isoformat() + 'Z'
-        snapshotted_at = snapshotted_at.isoformat() + 'Z'
+        max_loaded_at = max_loaded_at.isoformat()
+        snapshotted_at = snapshotted_at.isoformat()
         if timing is None:
             timing = []
         super(SourceFreshnessResult, self).__init__(

core/dbt/main.py

@@ -21,6 +21,7 @@
 import dbt.task.generate as generate_task
 import dbt.task.serve as serve_task
 import dbt.task.freshness as freshness_task
+import dbt.task.run_operation as run_operation_task
 from dbt.task.rpc_server import RPCServerTask
 from dbt.adapters.factory import reset_adapters
 
@@ -653,6 +654,33 @@ def parse_args(args):
     _build_docs_serve_subparser(docs_subs, base_subparser)
     _build_source_snapshot_freshness_subparser(source_subs, base_subparser)
 
+    sub = subs.add_parser(
+        'run-operation',
+        parents=[base_subparser],
+        help="""
+            (beta) Run the named macro with any supplied arguments. This
+            subcommand is unstable and subject to change in a future release
+            of dbt. Please use it with caution"""
+    )
+    sub.add_argument(
+        '--macro',
+        required=True,
+        help="""
+            Specify the macro to invoke. dbt will call this macro with the
+            supplied arguments and then exit"""
+    )
+    sub.add_argument(
+        '--args',
+        type=str,
+        default='{}',
+        help="""
+            Supply arguments to the macro. This dictionary will be mapped
+            to the keyword arguments defined in the selected macro. This
+            argument should be a YAML string, eg. '{my_variable: my_value}'"""
+    )
+    sub.set_defaults(cls=run_operation_task.RunOperationTask,
+                     which='run-operation')
+
     if len(args) == 0:
         p.print_help()
         sys.exit(1)

core/dbt/node_runners.py

@@ -299,9 +299,9 @@ def print_start_line(self):
                                         self.num_nodes)
 
     def print_result_line(self, result):
-        schema_name = self.node.schema
+        description = self.describe_node()
         dbt.ui.printer.print_model_result_line(result,
-                                               schema_name,
+                                               description,
                                                self.node_index,
                                                self.num_nodes)
 
@@ -398,7 +398,8 @@ def execute(self, compiled_node, manifest):
         freshness = self.adapter.calculate_freshness(
             relation,
             compiled_node.loaded_at_field,
-            manifest=manifest
+            manifest=manifest,
+            connection_name=compiled_node.unique_id
         )
         status = self._calculate_status(
             compiled_node.freshness,

core/dbt/task/run_operation.py

@@ -0,0 +1,40 @@
+from dbt.logger import GLOBAL_LOGGER as logger
+
+from dbt.task.base import BaseTask
+from dbt.adapters.factory import get_adapter
+from dbt.loader import GraphLoader
+
+import dbt
+import dbt.utils
+import dbt.exceptions
+
+
+class RunOperationTask(BaseTask):
+    def _get_macro_parts(self):
+        macro_name = self.args.macro
+        if '.' in macro_name:
+            package_name, macro_name = macro_name.split(".", 1)
+        else:
+            package_name = None
+
+        return package_name, macro_name
+
+    def _get_kwargs(self):
+        return dbt.utils.parse_cli_vars(self.args.args)
+
+    def run(self):
+        manifest = GraphLoader.load_all(self.config)
+        adapter = get_adapter(self.config)
+
+        package_name, macro_name = self._get_macro_parts()
+        macro_kwargs = self._get_kwargs()
+
+        res = adapter.execute_macro(
+            macro_name,
+            project=package_name,
+            kwargs=macro_kwargs,
+            manifest=manifest,
+            connection_name="macro_{}".format(macro_name)
+        )
+
+        return res

core/dbt/task/runnable.py

@@ -129,7 +129,7 @@ def _submit(self, pool, args, callback):
 
         This does still go through the callback path for result collection.
         """
-        if self.config.args.single_threaded or True:
+        if self.config.args.single_threaded:
             callback(self.call_runner(*args))
         else:
             pool.apply_async(self.call_runner, args=args, callback=callback)

core/dbt/ui/printer.py

@@ -146,17 +146,11 @@ def print_test_result_line(result, schema_name, index, total):
         result.execution_time)
 
 
-def print_model_result_line(result, schema_name, index, total):
-    model = result.node
-
+def print_model_result_line(result, description, index, total):
     info, status = get_printable_result(result, 'created', 'creating')
 
     print_fancy_output_line(
-        "{info} {model_type} model {schema}.{relation}".format(
-            info=info,
-            model_type=get_materialization(model),
-            schema=schema_name,
-            relation=model.get('alias')),
+        "{info} {description}".format(info=info, description=description),
         status,
         index,
         total,

core/dbt/version.py

@@ -57,5 +57,5 @@ def get_version_information():
                 .format(version_msg))
 
 
-__version__ = '0.13.0a1'
+__version__ = '0.13.0a2'
 installed = get_installed_version()

core/setup.py

@@ -9,7 +9,7 @@ def read(fname):
 
 
 package_name = "dbt-core"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """dbt (data build tool) is a command line tool that helps \
 analysts and engineers transform data in their warehouse more effectively"""
 

plugins/bigquery/setup.py

@@ -3,7 +3,7 @@
 from distutils.core import setup
 
 package_name = "dbt-bigquery"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """The bigquery adapter plugin for dbt (data build tool)"""
 
 

plugins/postgres/setup.py

@@ -3,7 +3,7 @@
 from distutils.core import setup
 
 package_name = "dbt-postgres"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """The postgres adpter plugin for dbt (data build tool)"""
 
 setup(

plugins/redshift/setup.py

@@ -3,7 +3,7 @@
 from distutils.core import setup
 
 package_name = "dbt-redshift"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """The redshift adapter plugin for dbt (data build tool)"""
 
 

plugins/snowflake/dbt/include/snowflake/macros/adapters.sql

@@ -81,3 +81,10 @@
 {% macro snowflake__current_timestamp() -%}
   convert_timezone('UTC', current_timestamp())
 {%- endmacro %}
+
+
+{% macro snowflake__rename_relation(from_relation, to_relation) -%}
+  {% call statement('rename_relation') -%}
+    alter table {{ from_relation }} rename to {{ to_relation }}
+  {%- endcall %}
+{% endmacro %}

plugins/snowflake/setup.py

@@ -3,7 +3,7 @@
 from distutils.core import setup
 
 package_name = "dbt-snowflake"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """The snowflake adapter plugin for dbt (data build tool)"""
 
 

setup.py

@@ -9,7 +9,7 @@ def read(fname):
 
 
 package_name = "dbt"
-package_version = "0.13.0a1"
+package_version = "0.13.0a2"
 description = """dbt (data build tool) is a command line tool that helps \
 analysts and engineers transform data in their warehouse more effectively"""
 

test/integration/024_custom_schema_test/models/view_3.sql

@@ -1,5 +1,5 @@
 
-{{ config(schema='test') }}
+{{ config(schema='test', materialized='table') }}
 
 
 with v1 as (

test/integration/024_custom_schema_test/test_custom_schema.py

@@ -1,5 +1,5 @@
 from nose.plugins.attrib import attr
-from test.integration.base import DBTIntegrationTest
+from test.integration.base import DBTIntegrationTest, use_profile
 
 
 class TestCustomSchema(DBTIntegrationTest):
@@ -85,6 +85,42 @@ def test__postgres__custom_schema_with_prefix(self):
         self.assertTablesEqual("agg","view_3", schema, xf_schema)
 
 
+class TestCustomProjectSchemaWithPrefixSnowflake(DBTIntegrationTest):
+
+    @property
+    def schema(self):
+        return "custom_schema_024"
+
+    @property
+    def models(self):
+        return "test/integration/024_custom_schema_test/models"
+
+    @property
+    def project_config(self):
+        return {
+            "models": {
+                "schema": "dbt_test"
+            }
+        }
+
+    @use_profile('snowflake')
+    def test__snowflake__custom_schema_with_prefix(self):
+        self.use_default_project()
+        self.run_sql_file("test/integration/024_custom_schema_test/seed.sql")
+
+        results = self.run_dbt()
+        self.assertEqual(len(results), 3)
+
+        schema = self.unique_schema().upper()
+        v1_schema = "{}_DBT_TEST".format(schema)
+        v2_schema = "{}_CUSTOM".format(schema)
+        xf_schema = "{}_TEST".format(schema)
+
+        self.assertTablesEqual("SEED","VIEW_1", schema, v1_schema)
+        self.assertTablesEqual("SEED","VIEW_2", schema, v2_schema)
+        self.assertTablesEqual("AGG","VIEW_3", schema, xf_schema)
+
+
 class TestCustomSchemaWithCustomMacro(DBTIntegrationTest):
 
     @property

test/integration/042_sources_test/test_sources.py

@@ -121,7 +121,7 @@ def setUp(self):
         self.maxDiff = None
         self._id = 100
         # this is the db initial value
-        self.last_inserted_time = "2016-09-19T14:45:51+00:00Z"
+        self.last_inserted_time = "2016-09-19T14:45:51+00:00"
 
     # test_source.test_table should have a loaded_at field of `updated_at`
     # and a freshness of warn_after: 10 hours, error_after: 18 hours
@@ -146,7 +146,7 @@ def _set_updated_at_to(self, delta):
                 'source': self.adapter.quote('source'),
             }
         )
-        self.last_inserted_time = insert_time.strftime("%Y-%m-%dT%H:%M:%S+00:00Z")
+        self.last_inserted_time = insert_time.strftime("%Y-%m-%dT%H:%M:%S+00:00")
 
     def _assert_freshness_results(self, path, state):
         self.assertTrue(os.path.exists(path))
@@ -162,7 +162,7 @@ def _assert_freshness_results(self, path, state):
 
         last_inserted_time = self.last_inserted_time
         if last_inserted_time is None:
-            last_inserted_time = "2016-09-19T14:45:51+00:00Z"
+            last_inserted_time = "2016-09-19T14:45:51+00:00"
 
         self.assertEqual(data['sources'], {
             'source.test.test_source.test_table': {

test/unit/test_snowflake_adapter.py

@@ -125,7 +125,7 @@ def test_quoting_on_rename(self):
         )
         self.mock_execute.assert_has_calls([
             mock.call(
-                'alter table "test_database"."test_schema".table_a rename to table_b',
+                'alter table "test_database"."test_schema".table_a rename to "test_database"."test_schema".table_b',
                 None
             )
         ])