Repository: dbt-labs/dbt-core
Pull request: Add a configuration option `kms_key_name` to dbt_project.yml for BigQ…
…uery projects where a Cloud KMS key can be specified and used to encrypt models. The key is specified to bigquery in the ddl OPTIONS.

ex. 
```
models:
  fishtown_analytics:
  kms_key_name: 'projects/PROJECT_ID/locations/global/keyRings/test/cryptoKeys/quickstart'
    
    my_model:
      kms_key_name: none   # disables kms encryption
```

Fixes #1829

Problem statement:
Support KMS encryption for BigQuery
### Describe the feature
Allow for KMS encryption to be specified in project configuration. Ideally this would be done in dbt_projects.yml like:
```name: fishtown_analytics

models:
  fishtown_analytics:
  kms_key_name: 'some_kms_key'
  
    events:
      kms_key_name: 'some_other_key' 
```

but may also exist in a model's {{ config() }}

### Describe alternatives you've considered
It may be possible to do this as a post-processing step outside of dbt, or in a post-hook (although I am not aware of any way to specify KMS key in sql).

### Additional context
This feature is specific to BigQuery as it currently stands but may generalize to other databases.

### Who will this benefit?
Anyone who uses BigQuery and may need to encrypt their data with cloud KMS keys (as opposed to the default encryption BigQuery already performs).

Hints from issue discussion:
Really neat idea @kconvey!

It looks to me like a `kms_key_name` option can be provided to a `create table as` statement in BigQuery. That would be a good mechanism for dbt to use here I think.

It appears to me that the `kms_key_name` will also need to be supplied when _querying_ this data too. I think that might make it challenging for dbt to use different KMS keys for different models. What happens if you write a query like:
```
select *
from my_dataset.my_first_model
left join my_dataset.my_second_model using (id)
```

If `my_first_model` and `my_second_model` use different KMS keys, which key do you use?

I haven't use KMS on BQ before - if you have insight into how this works in practice, I'd love to hear it!
As a quick note on querying, the KMS key name is not needed when reading tables.  See https://cloud.google.com/bigquery/docs/customer-managed-encryption:

> No special arrangements are required to query a table protected by Cloud KMS. BigQuery stores the name of the key used to encrypt the table content and will use that key when a table protected by Cloud KMS is queried.

> All existing tools, the BigQuery console, and the bq command-line interface run the same way as with default-encrypted tables, as long as BigQuery has access to the Cloud KMS key used to encrypt the table content.

Commit message: Add a configuration option `kms_key_name` to dbt_project.yml for BigQuery
projects where a Cloud KMS key can be specified and used to encrypt
models. The key is specified to bigquery in the ddl OPTIONS.

Patch to review:
diff --git a/plugins/bigquery/dbt/adapters/bigquery/impl.py b/plugins/bigquery/dbt/adapters/bigquery/impl.py
index d25cc984e77..2aa878d5615 100644
--- a/plugins/bigquery/dbt/adapters/bigquery/impl.py
+++ b/plugins/bigquery/dbt/adapters/bigquery/impl.py
@@ -45,7 +45,8 @@ class BigQueryAdapter(BaseAdapter):
     Column = BigQueryColumn
     ConnectionManager = BigQueryConnectionManager
 
-    AdapterSpecificConfigs = frozenset({"cluster_by", "partition_by"})
+    AdapterSpecificConfigs = frozenset({"cluster_by", "partition_by",
+                                        "kms_key_name"})
 
     ###
     # Implementations of abstract methods
diff --git a/plugins/bigquery/dbt/include/bigquery/macros/adapters.sql b/plugins/bigquery/dbt/include/bigquery/macros/adapters.sql
index 719860ea40c..8e3d61c5d1c 100644
--- a/plugins/bigquery/dbt/include/bigquery/macros/adapters.sql
+++ b/plugins/bigquery/dbt/include/bigquery/macros/adapters.sql
@@ -26,7 +26,7 @@
 
 {%- endmacro -%}
 
-{% macro bigquery_table_options(persist_docs, temporary) %}
+{% macro bigquery_table_options(persist_docs, temporary, kms_key_name) %}
   {% set opts = {} %}
 
   {% set description = get_relation_comment(persist_docs, model) %}
@@ -36,24 +36,25 @@
   {% if temporary %}
     {% do opts.update({'expiration_timestamp': 'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)'}) %}
   {% endif %}
+  {% if kms_key_name %}
+    {% do opts.update({'kms_key_name': "'" ~ kms_key_name ~ "'"}) %}
+  {% endif %}
 
-  {% set options -%}
-    OPTIONS({% for opt_key, opt_val in opts.items() %}
-      {{ opt_key }}={{ opt_val }}{{ "," if not loop.last }}
-    {% endfor %})
-  {%- endset %}
-  {% do return(options) %}
+  OPTIONS({% for opt_key, opt_val in opts.items() %}
+    {{ opt_key }}={{ opt_val }}
+  {% endfor %})
 {%- endmacro -%}
 
 {% macro bigquery__create_table_as(temporary, relation, sql) -%}
   {%- set raw_partition_by = config.get('partition_by', none) -%}
   {%- set raw_cluster_by = config.get('cluster_by', none) -%}
   {%- set raw_persist_docs = config.get('persist_docs', {}) -%}
+  {%- set raw_kms_key_name = config.get('kms_key_name', none) -%}
 
   create or replace table {{ relation }}
   {{ partition_by(raw_partition_by) }}
   {{ cluster_by(raw_cluster_by) }}
-  {{ bigquery_table_options(persist_docs=raw_persist_docs, temporary=temporary) }}
+  {{ bigquery_table_options(persist_docs=raw_persist_docs, temporary=temporary, kms_key_name=raw_kms_key_name) }}
   as (
     {{ sql }}
   );


Code review comments:
diff_hunk: @@ -36,24 +36,25 @@
   {% if temporary %}
     {% do opts.update({'expiration_timestamp': 'TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 12 hour)'}) %}
   {% endif %}
+  {% if kms_key_name %}
+    {% do opts.update({'kms_key_name': "'" ~ kms_key_name ~ "'"}) %}
+  {% endif %}
 
-  {% set options -%}
-    OPTIONS({% for opt_key, opt_val in opts.items() %}
-      {{ opt_key }}={{ opt_val }}{{ "," if not loop.last }}
-    {% endfor %})
-  {%- endset %}
-  {% do return(options) %}
+  OPTIONS({% for opt_key, opt_val in opts.items() %}
path: plugins/bigquery/dbt/include/bigquery/macros/adapters.sql
text: @user1:
I think we want to keep some of the logic that was overwritten here. It looks to me like you may not have pulled `dev/louisa-may-alcott` before making this PR? I think this code was _just_ merged a couple of days ago.

The `{% set %}` block is used to better manage whitespace in the compiled SQL output. The `{{ "," if not loop.last }}` construct is used to ensure that these options are comma-delimited.

@author:
I was definitely behind. Thanks for catching!