Support for COPY INTO

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Enable Pandas and Pandas-on-Spark DataFrames for dbt python models ([dbt-labs/dbt-spark#469](https://github.com/dbt-labs/dbt-spark/pull/469), [#181](https://github.com/databricks/dbt-databricks/pull/181))
 - Implement testing for a test for various Python models ([#189](https://github.com/databricks/dbt-databricks/pull/189))
 - Implement testing for `type_boolean` in Databricks ([dbt-labs/dbt-spark#471](https://github.com/dbt-labs/dbt-spark/pull/471), [#188](https://github.com/databricks/dbt-databricks/pull/188))
+- Add a macro to support [COPY INTO](https://docs.databricks.com/spark/latest/spark-sql/language-manual/delta-copy-into.html) ([#190](https://github.com/databricks/dbt-databricks/pull/190))
 
 ### Under the hood
 - Apply "Initial refactoring of incremental materialization" ([#148](https://github.com/databricks/dbt-databricks/pull/148))

dbt/include/databricks/macros/copy_into.sql

@@ -0,0 +1,76 @@
+{% macro databricks_copy_into(
+  target_table,
+  source,
+  file_format,
+  expression_list=none,
+  source_credential=none,
+  source_encryption=none,
+  validate=none,
+  files=none,
+  format_options=none,
+  copy_options=none) -%}
+
+  {% set target_relation_exists, target_relation = get_or_create_relation(
+        database=target.database,
+        schema=target.schema,
+        identifier=target_table,
+        type='table') -%}
+
+  {%- set source_clause -%}
+    {%- if expression_list -%}
+      select {{ expression_list }} from '{{ source }}'
+    {%- else -%}
+      '{{ source }}'
+    {%- endif -%}
+    {%- if source_credential or source_encryption %}
+      WITH (
+	{% if source_credential -%}
+	  credential (
+	    {%- for name in source_credential -%}
+	      '{{ name }}' = '{{ source_credential[name] }}' {%- if not loop.last %}, {% endif -%}
+	    {%- endfor -%}
+	  )
+	{%- endif %}
+        {% if source_encryption -%}
+	  encryption (
+	    {%- for name in source_encryption -%}
+	      '{{ name }}' = '{{ source_encryption[name] }}' {%- if not loop.last %}, {% endif -%}
+	    {%- endfor -%}
+	  )
+	{%- endif %}
+      )
+    {%- endif -%}
+  {%- endset -%}
+
+  {% set query %}
+    copy into {{ target_relation }}
+    from {{ source_clause }}
+    fileformat = {{ file_format }}
+    {% if validate -%} validate {{ validate }} {%- endif %}
+    {% if files -%}
+      files = (
+        {%- for file in files -%}
+          '{{ file }}' {%- if not loop.last %}, {% endif -%}
+        {%- endfor -%}
+      )
+    {%- endif %}
+    {% if format_options -%}
+      format_options (
+        {%- for key in format_options -%}
+          '{{ key }}' = '{{ format_options[key] }}' {%- if not loop.last %}, {% endif -%}
+        {%- endfor -%}
+      )
+    {%- endif %}
+    {% if copy_options -%}
+      copy_options (
+        {%- for key in copy_options -%}
+          '{{ key }}' = '{{ copy_options[key] }}' {%- if not loop.last %}, {% endif -%}
+        {%- endfor -%}
+      )
+    {%- endif %}
+  {% endset %}
+
+  {% do log("Running COPY INTO" ~ query, info=True) %}
+  {% do run_query(query) %}
+
+{% endmacro %}

tests/integration/copy_into/models/expected_target.sql

@@ -0,0 +1,7 @@
+{{config(materialized='table')}}
+
+select * from values
+	(0, 'Zero', '2022-01-01'),
+	(1, 'Alice', '2022-01-01'),
+	(2, 'Bob', '2022-01-02')
+	as t(id, name, date)

tests/integration/copy_into/models/schema.yml

@@ -0,0 +1,15 @@
+version: 2
+
+seeds:
+  - name: source
+    config:
+      column_types:
+        id: int
+        name: string
+        date: string
+  - name: target
+    config:
+      column_types:
+        id: int
+        name: string
+        date: string

tests/integration/copy_into/models/source.sql

@@ -0,0 +1,6 @@
+{{config(materialized='table', file_format='parquet')}}
+
+select * from values
+	(1, 'Alice', '2022-01-01'),
+	(2, 'Bob', '2022-01-02')
+	t(id, name, date)

tests/integration/copy_into/models/target.sql

@@ -0,0 +1,4 @@
+{{config(materialized='table')}}
+
+select * from values
+	(0, 'Zero', '2022-01-01') as t(id, name, date)

tests/integration/copy_into/test_copy_into.py

@@ -0,0 +1,50 @@
+from tests.integration.base import DBTIntegrationTest, use_profile
+
+
+args = """
+target_table: target
+source: {source_path}
+file_format: parquet
+format_options:
+  mergeSchema: 'true'
+copy_options:
+  mergeSchema: 'true'
+"""
+
+
+class TestCopyInto(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return "copy_into"
+
+    @property
+    def models(self):
+        return "models"
+
+    def test_copy_into(self):
+        self.run_dbt(["run"])
+        # Get the location of the source table.
+        rows = self.run_sql("describe table extended {database_schema}.source", fetch="all")
+        path = None
+        for row in rows:
+            if row.col_name == "Location":
+                path = row.data_type
+        if path is None:
+            raise Exception("No location found for the source table")
+        self.run_dbt(
+            [
+                "run-operation",
+                "databricks_copy_into",
+                "--args",
+                args.format(source_path=path),
+            ]
+        )
+        self.assertTablesEqual("target", "expected_target")
+
+    @use_profile("databricks_cluster")
+    def test_databricks_cluster(self):
+        self.test_copy_into()
+
+    @use_profile("databricks_sql_endpoint")
+    def test_databricks_sql_endpoint(self):
+        self.test_copy_into()