From c13f1dda0a6c04f8c0884962c27682af536ec562 Mon Sep 17 00:00:00 2001
From: Cor <jczuurmond@protonmail.com>
Date: Wed, 2 Jun 2021 16:50:42 +0000
Subject: [PATCH] Add options clause to create table macro (#171)

* Add option clause macro

* Add option clause to create table macro

* Add test for options clause

* Add change log entry

* Add file format delta to test

* Change order of table expression

* Make options lower case

* Change order of table definitions

* Add options to spark config
---
 CHANGELOG.md                          |  4 ++++
 dbt/adapters/spark/impl.py            |  1 +
 dbt/include/spark/macros/adapters.sql | 11 +++++++++++
 test/unit/test_macros.py              |  8 ++++++++
 4 files changed, 24 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 002a0a1e7..5a2a966b2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,8 @@
 ## dbt next
 
+### Features
+- Allow setting table `OPTIONS` using `config` ([#171](https://github.com/fishtown-analytics/dbt-spark/pull/171))
+
 ### Fixes
 
 - Cast `table_owner` to string to avoid errors generating docs ([#158](https://github.com/fishtown-analytics/dbt-spark/pull/158), [#159](https://github.com/fishtown-analytics/dbt-spark/pull/159))
@@ -13,6 +16,7 @@
 - [@friendofasquid](https://github.com/friendofasquid) ([#159](https://github.com/fishtown-analytics/dbt-spark/pull/159))
 - [@franloza](https://github.com/franloza) ([#160](https://github.com/fishtown-analytics/dbt-spark/pull/160))
 - [@Fokko](https://github.com/Fokko) ([#165](https://github.com/fishtown-analytics/dbt-spark/pull/165))
+- [@JCZuurmond](https://github.com/JCZuurmond) ([#171](https://github.com/fishtown-analytics/dbt-spark/pull/171))
 
 ## dbt-spark 0.19.1 (Release TBD)
 
diff --git a/dbt/adapters/spark/impl.py b/dbt/adapters/spark/impl.py
index 93cfba687..b2060f8c9 100644
--- a/dbt/adapters/spark/impl.py
+++ b/dbt/adapters/spark/impl.py
@@ -36,6 +36,7 @@ class SparkConfig(AdapterConfig):
     partition_by: Optional[Union[List[str], str]] = None
     clustered_by: Optional[Union[List[str], str]] = None
     buckets: Optional[int] = None
+    options: Optional[Dict[str, str]] = None
 
 
 class SparkAdapter(SQLAdapter):
diff --git a/dbt/include/spark/macros/adapters.sql b/dbt/include/spark/macros/adapters.sql
index a45b0d1ae..8d095e9ce 100644
--- a/dbt/include/spark/macros/adapters.sql
+++ b/dbt/include/spark/macros/adapters.sql
@@ -13,6 +13,16 @@
   {%- endif %}
 {%- endmacro -%}
 
+{% macro options_clause() -%}
+  {%- set options = config.get('options') -%}
+  {%- if options is not none %}
+    options (
+      {%- for option in options -%}
+      {{ option }} "{{ options[option] }}" {% if not loop.last %}, {% endif %}
+      {%- endfor %}
+    )
+  {%- endif %}
+{%- endmacro -%}
 
 {% macro comment_clause() %}
   {%- set raw_persist_docs = config.get('persist_docs', {}) -%}
@@ -83,6 +93,7 @@
       create table {{ relation }}
     {% endif %}
     {{ file_format_clause() }}
+    {{ options_clause() }}
     {{ partition_cols(label="partitioned by") }}
     {{ clustered_cols(label="clustered by") }}
     {{ location_clause() }}
diff --git a/test/unit/test_macros.py b/test/unit/test_macros.py
index 5c5e3f8cf..151631e08 100644
--- a/test/unit/test_macros.py
+++ b/test/unit/test_macros.py
@@ -43,6 +43,14 @@ def test_macros_create_table_as_file_format(self):
         sql = self.__run_macro(template, 'spark__create_table_as', False, 'my_table', 'select 1').strip()
         self.assertEqual(sql, "create or replace table my_table using delta as select 1")
 
+    def test_macros_create_table_as_options(self):
+        template = self.__get_template('adapters.sql')
+
+        self.config['file_format'] = 'delta'
+        self.config['options'] = {"compression": "gzip"}
+        sql = self.__run_macro(template, 'spark__create_table_as', False, 'my_table', 'select 1').strip()
+        self.assertEqual(sql, 'create or replace table my_table using delta options (compression "gzip" ) as select 1')
+
     def test_macros_create_table_as_partition(self):
         template = self.__get_template('adapters.sql')