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_task 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
  )
  ])