Repository: dbt-labs/dbt-core
Pull request: BigQuery copy materialization enhancement
resolves #3570

### Description

1. Add implicit usage of `copy_materialization='table'` if it is not found in config (simplification for end-user)
2. Align `adapter.table_copy` implementation with [BigQuery documentation](https://cloud.google.com/bigquery/docs/managing-tables#copying_multiple_source_tables) (ability to use several source tables to be copied into single target table at a time)
3. Change integration test to check (1) and to test (2) for two source tables (reference by `ref` and `source`)

### 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 (custom Docker image has been built from changed code and process of table(s) copying has been checked several times for different scenarios)
 - [ 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:
BigQuery 'copy' materialization doesn't work
Trying to check model:
```sql
{{ config(
    materialized='copy'
)}}

select * from {{ source('int', 'Order_delta') }}
```
 and receive following error:
 ```
Running with dbt=0.20.0
Found 2 models, 0 tests, 0 snapshots, 0 analyses, 166 macros, 0 operations, 0 seed files, 7 sources, 0 exposures

14:22:42 | Concurrency: 1 threads (target='poc')
14:22:42 | 
14:22:42 | 1 of 1 START copy model poc_dbt.OrderDedupe.......................... [RUN]
14:22:42 | 1 of 1 ERROR creating copy model poc_dbt.OrderDedupe................. [ERROR in 0.03s]
14:22:42 | 
14:22:42 | Finished running 1 copy model in 2.79s.

Completed with 1 error and 0 warnings:

Compilation Error in model OrderDedupe (models/int/OrderDedupe.sql)
  Copy table materialization must be "copy" or "table", but config.get('copy_materialization', 'table') was None

  > in macro materialization_copy_bigquery (macros/materializations/copy.sql)
  > called by model OrderDedupe (models/int/OrderDedupe.sql)

Done. PASS=0 WARN=0 ERROR=1 SKIP=0 TOTAL=1
```

There is no documentation for this materialization but it exists in the code.
Is there any way to get it work?

Hints from issue discussion:
@xemuliam You're right, this is an undocumented feature in dbt today, added by https://github.com/dbt-labs/dbt/pull/2709 last year. It's been useful, here or there, to have access to the `adapter.copy_table()` method—playing around with potential approaches to schema migration, or blue/green deployment. Frankly, I haven't yet documented this feature because I'm not sure of the exact use cases in which to recommend it.

In any case, this works just fine if you specify the `copy_materialization` config explicitly:

```sql
{{ config(
    materialized='copy',
    copy_materialization='table'
)}}

select * from {{ source('int', 'Order_delta') }}
```

The two options available are `incremental` and `table`, and these map to append-only and overwrite-only copy jobs respectively:

https://github.com/dbt-labs/dbt/blob/f460d275baa4df8cf8713f27e198a3577d6ea71c/plugins/bigquery/dbt/adapters/bigquery/impl.py#L433-L442

The issue is that `copy_materialization` is `None` if not specified, when the logic in the materialization sure seems like it's meant to be `table`. I was able to fix this by switching the logic from:

https://github.com/dbt-labs/dbt/blob/f460d275baa4df8cf8713f27e198a3577d6ea71c/plugins/bigquery/dbt/include/bigquery/macros/materializations/copy.sql#L27-L30

To:
```jinja
  {%- set result_str = adapter.copy_table(
      src,
      destination,
      config.get('copy_materialization', default = 'table')) -%}
```
Though, better form would have us setting that config at the top of the materialization, as we do in other materializations.

I'd welcome a PR for that fix, as well as a PR for docs.getdbt.com that offers basic documentation for this feature paired with a few potential uses. I'm going to call this a good first issue :)
@jtcohen6 
Thanks a lot for given explanation.
Defifnitely this should be fixed to be aligned with Google implementation (data replace is default).

However there are two different issues in addition to above one :)
1. Ideally it would be to parse `ref` and `source` in final SQL only. In other words kt would be great to be able to do smth. like that (currently it is impossible case. Parser tries to get `ref` and `source` from the whole file):
```sql
{{ config(
    database=target.database,
    schema='internal',
    alias='Order',
    materialized='copy',
    copy_materialization='incremental',
    post_hook="drop table if exists {{ source('int', 'Order_delta') }}"
)}}

{% call statement('order-dedupe-delta', fetch_result=False) %}
-- do something using `ref` or `source`
{%- endcall -%}

select * from {{ ref('OrderDedupe_delta') }}
```
2. BigQuery table copy API allows to use several source tables to be copied into one target table at a time. I hope it is a good idea to alignt BigQuery API and DBT implementation (currently target table could be copied from single source table only)
https://cloud.google.com/bigquery/docs/managing-tables#copying_multiple_source_tables
@jtcohen6 
Regarding use cases...
BigQuery table copy API is really useful when you need to avoid extra time and costs for exact copy of table(s). The effect is similar as from `create or replace table aaa as select * from bbb`. I've tested that on table with more than 60TB of data inside. `create or replace` statement hasn't been finished within 6 hours and query has failed. Copy API has been finished within 40-70 seconds and costed zero!
![image](https://user-images.githubusercontent.com/20856221/126455698-7e435df7-f318-45df-ab5a-4c6f84a03531.png)

Additional side-effect is data availability. Imagine you need to work with target table which is really useful for business and there is a lot of attention to it. You can't split table operations e.g. trucncate table (or delete some data from it) and re-calculate data. In this case business will see lack of data within some period of time. Thus you have two options:
- re-create table by `create or replace table` statement. It is costly and slowly. But data will be available cos actually there are two stages under this statement: select and commit data into temp table and (attention!) copy table using copy API with full rewrite of target table under the hood
- make a copy of target table using copy API, do all reqired operations (partial deletion, delta calculations etc.) on this copy and filnally make a copy of that temp table into target table with full rewrite or append (depends on your delta calculation approach)

Separate part is blu-green deploymants and other sync processes. Table copy API is extremely useful in those cases also.

Commit message: Split long code ilnes

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 27252b6fcec..3c7400fb462 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,11 @@
 ## dbt 0.21.0 (Release TBD)
 
+### Features
+
+- Enhance BigQuery copy materialization ([#3570](https://github.com/dbt-labs/dbt/issues/3570), [#3606](https://github.com/dbt-labs/dbt/pull/3606)):
+  - to simplify config (default usage of `copy_materialization='table'` if is is not found in global or local config)
+  - to let copy several source tables into single target table at a time. ([Google doc reference](https://cloud.google.com/bigquery/docs/managing-tables#copying_multiple_source_tables))
+
 ### Under the hood
 
 - Use GitHub Actions for CI ([#3688](https://github.com/dbt-labs/dbt/issues/3688), [#3669](https://github.com/dbt-labs/dbt/pull/3669))
diff --git a/plugins/bigquery/dbt/adapters/bigquery/connections.py b/plugins/bigquery/dbt/adapters/bigquery/connections.py
index 2c90297cdc8..9abaf5bc8d9 100644
--- a/plugins/bigquery/dbt/adapters/bigquery/connections.py
+++ b/plugins/bigquery/dbt/adapters/bigquery/connections.py
@@ -458,26 +458,32 @@ def copy_bq_table(self, source, destination, write_disposition):
         conn = self.get_thread_connection()
         client = conn.handle
 
-        source_ref = self.table_ref(
-            source.database, source.schema, source.table, conn)
+        if type(source) is not list:
+            source = [source]
+
+        source_ref_array = [self.table_ref(
+            src_table.database, src_table.schema, src_table.table, conn)
+            for src_table in source]
         destination_ref = self.table_ref(
             destination.database, destination.schema, destination.table, conn)
 
         logger.debug(
-            'Copying table "{}" to "{}" with disposition: "{}"',
-            source_ref.path, destination_ref.path, write_disposition)
+            'Copying table(s) "{}" to "{}" with disposition: "{}"',
+            ', '.join(source_ref.path for source_ref in source_ref_array),
+            destination_ref.path, write_disposition)
 
         def copy_and_results():
             job_config = google.cloud.bigquery.CopyJobConfig(
                 write_disposition=write_disposition)
             copy_job = client.copy_table(
-                source_ref, destination_ref, job_config=job_config)
+                source_ref_array, destination_ref, job_config=job_config)
             iterator = copy_job.result(timeout=self.get_timeout(conn))
             return copy_job, iterator
 
         self._retry_and_handle(
             msg='copy table "{}" to "{}"'.format(
-                source_ref.path, destination_ref.path),
+                ', '.join(source_ref.path for source_ref in source_ref_array),
+                destination_ref.path),
             conn=conn, fn=copy_and_results)
 
     @staticmethod
diff --git a/plugins/bigquery/dbt/include/bigquery/macros/materializations/copy.sql b/plugins/bigquery/dbt/include/bigquery/macros/materializations/copy.sql
index 0f24b50bcb8..0236f927588 100644
--- a/plugins/bigquery/dbt/include/bigquery/macros/materializations/copy.sql
+++ b/plugins/bigquery/dbt/include/bigquery/macros/materializations/copy.sql
@@ -3,31 +3,24 @@
   {# Setup #}
   {{ run_hooks(pre_hooks) }}
 
-  {# there should be exactly one ref or exactly one source #}
   {% set destination = this.incorporate(type='table') %}
 
-  {% set dependency_type = none %}
-  {% if (model.refs | length) == 1 and (model.sources | length) == 0 %}
-    {% set dependency_type = 'ref' %}
-  {% elif (model.refs | length) == 0 and (model.sources | length) == 1 %}
-    {% set dependency_type = 'source' %}
-  {% else %}
-    {% set msg %}
-        Expected exactly one ref or exactly one source, instead got {{ model.refs | length }} models and {{ model.sources | length }} sources.
-    {% endset %}
-    {% do exceptions.raise_compiler_error(msg) %}
-  {% endif %}
-
-  {% if dependency_type == 'ref' %}
-    {% set src =  ref(*model.refs[0]) %}
-  {% else %}
-    {% set src =  source(*model.sources[0]) %}
-  {% endif %}
 
+  {# Cycle over ref() and source() to create source tables array #}
+  {% set source_array = [] %}
+  {% for ref_table in model.refs %}
+    {{ source_array.append(ref(*ref_table)) }}
+  {% endfor %}
+
+  {% for src_table in model.sources %}
+    {{ source_array.append(source(*src_table)) }}
+  {% endfor %}
+
+  {# Call adapter's copy_table function #}
   {%- set result_str = adapter.copy_table(
-      src,
+      source_array,
       destination,
-      config.get('copy_materialization', 'table')) -%}
+      config.get('copy_materialization', default = 'table')) -%}
 
   {{ store_result('main', response=result_str) }}
 
diff --git a/test/integration/022_bigquery_test/copy-models/copy_as_several_tables.sql b/test/integration/022_bigquery_test/copy-models/copy_as_several_tables.sql
new file mode 100644
index 00000000000..2cdf88e2496
--- /dev/null
+++ b/test/integration/022_bigquery_test/copy-models/copy_as_several_tables.sql
@@ -0,0 +1 @@
+select * from {{ ref('original') }}, {{ source('test_copy_several_tables', 'original') }}
\ No newline at end of file
diff --git a/test/integration/022_bigquery_test/copy-models/copy_as_table.sql b/test/integration/022_bigquery_test/copy-models/copy_as_table.sql
index 4dda068ac3a..3f2fe2550af 100644
--- a/test/integration/022_bigquery_test/copy-models/copy_as_table.sql
+++ b/test/integration/022_bigquery_test/copy-models/copy_as_table.sql
@@ -1,2 +1 @@
-{{ config(copy_materialization='table') }}
-{{ ref('original') }}
\ No newline at end of file
+{{ ref('original') }}
diff --git a/test/integration/022_bigquery_test/copy-models/schema.yml b/test/integration/022_bigquery_test/copy-models/schema.yml
new file mode 100644
index 00000000000..69efa7632bf
--- /dev/null
+++ b/test/integration/022_bigquery_test/copy-models/schema.yml
@@ -0,0 +1,6 @@
+version: 2
+sources:
+  - name: test_copy_several_tables
+    schema: "{{ target.schema) }}"
+    tables:
+      - name: original
diff --git a/test/integration/022_bigquery_test/test_bigquery_copy_models.py b/test/integration/022_bigquery_test/test_bigquery_copy_models.py
index 00028357faf..5c30c32c583 100644
--- a/test/integration/022_bigquery_test/test_bigquery_copy_models.py
+++ b/test/integration/022_bigquery_test/test_bigquery_copy_models.py
@@ -27,6 +27,8 @@ def project_config(self):
                     materialized: table
                 copy_as_table:
                     materialized: copy
+                copy_as_several_tables:
+                    materialized: copy
                 copy_as_incremental:
                     materialized: copy
         '''))
diff --git a/test/unit/test_bigquery_adapter.py b/test/unit/test_bigquery_adapter.py
index 147b8cba0df..710941b61d9 100644
--- a/test/unit/test_bigquery_adapter.py
+++ b/test/unit/test_bigquery_adapter.py
@@ -617,7 +617,7 @@ def test_copy_bq_table_appends(self):
             write_disposition=dbt.adapters.bigquery.impl.WRITE_APPEND)
         args, kwargs = self.mock_client.copy_table.call_args
         self.mock_client.copy_table.assert_called_once_with(
-            self._table_ref('project', 'dataset', 'table1', None),
+            [self._table_ref('project', 'dataset', 'table1', None)],
             self._table_ref('project', 'dataset', 'table2', None),
             job_config=ANY)
         args, kwargs = self.mock_client.copy_table.call_args
@@ -630,7 +630,7 @@ def test_copy_bq_table_truncates(self):
             write_disposition=dbt.adapters.bigquery.impl.WRITE_TRUNCATE)
         args, kwargs = self.mock_client.copy_table.call_args
         self.mock_client.copy_table.assert_called_once_with(
-            self._table_ref('project', 'dataset', 'table1', None),
+            [self._table_ref('project', 'dataset', 'table1', None)],
             self._table_ref('project', 'dataset', 'table2', None),
             job_config=ANY)
         args, kwargs = self.mock_client.copy_table.call_args


Code review comments:
diff_hunk: @@ -0,0 +1,6 @@
+version: 2
+sources:
+  - name: test_copy_several_tables
+    schema: "{{ target.schema) }}"
path: test/integration/022_bigquery_test/copy-models/schema.yml
text: @user1:
```suggestion
    schema: "{{ target.schema }}"
```

@user1:
After I fix this locally, and run `test_bigquery_copy_models.py`, I see `copy_as_several_tables` failing with:
```
Database Error in model copy_as_several_tables (copy-models/copy_as_several_tables.sql)
  Table dbt-test-env:test16303148312675888629_bigquery_test_022.original appears multiple times in source tables.
```
Do you have a sense of what that error indicates?

@author:
Fixed