Adding minimal write option when creating table with Delta format

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## New version
 - Fix session provisioning timeout and delay handling
+- Add write options for Delta format
 - Add on_schema_change possibility
 - Fix table materialization for Delta models
 - Change GlueColumn parent from base Column to SparkColumn

dbt/adapters/glue/impl.py

@@ -634,7 +634,7 @@ def delta_update_manifest(self, target_relation, custom_location, partition_by):
             self._update_additional_location(target_relation, location)
 
     @available
-    def delta_create_table(self, target_relation, request, primary_key, partition_key, custom_location):
+    def delta_create_table(self, target_relation, request, primary_key, partition_key, custom_location, delta_create_table_write_options=None):
         session, client = self.get_connection()
         logger.debug(request)
 
@@ -643,6 +643,11 @@ def delta_create_table(self, target_relation, request, primary_key, partition_ke
             location = f"{session.credentials.location}/{target_relation.schema}/{target_relation.name}"
         else:
             location = custom_location
+
+        options_string = ""
+        if delta_create_table_write_options:
+            for key, value in delta_create_table_write_options.items():
+                options_string += f'.option("{key}", "{value}")'
 
         create_table_query = f"""
 CREATE TABLE {table_name}
@@ -654,7 +659,7 @@ def delta_create_table(self, target_relation, request, primary_key, partition_ke
 custom_glue_code_for_dbt_adapter
 spark.sql("""
 {request}
-""").write.format("delta").mode("overwrite")'''
+""").write.format("delta").mode("overwrite"){options_string}'''
 
         write_data_footer = f'''.save("{location}")
 SqlWrapper2.execute("""select 1""")

dbt/include/glue/macros/materializations/incremental/incremental.sql

@@ -21,6 +21,7 @@
   {%- set custom_location = config.get('custom_location', default='empty') -%}
   {%- set expire_snapshots = config.get('iceberg_expire_snapshots', 'True') -%}
   {%- set table_properties = config.get('table_properties', default='empty') -%}
+  {%- set delta_create_table_write_options = config.get('write_options', default={}) -%}
 
   {% set target_relation = this %}
   {%- set existing_relation = load_relation(this) -%}
@@ -50,7 +51,7 @@
       {% endif %}
       {% if existing_relation_type is none %}
         {% if file_format == 'delta' %}
-            {{ adapter.delta_create_table(target_relation, sql, unique_key, partition_by, custom_location) }}
+            {{ adapter.delta_create_table(target_relation, sql, unique_key, partition_by, custom_location, delta_create_table_write_options) }}
             {% set build_sql = "select * from " + target_relation.schema + "." + target_relation.identifier + " limit 1 " %}
         {% elif file_format == 'iceberg' %}
             {{ adapter.iceberg_write(target_relation, sql, unique_key, partition_by, custom_location, strategy, table_properties) }}
@@ -61,7 +62,7 @@
       {% elif existing_relation_type == 'view' or should_full_refresh() %}
         {{ drop_relation(target_relation) }}
         {% if file_format == 'delta' %}
-            {{ adapter.delta_create_table(target_relation, sql, unique_key, partition_by, custom_location) }}
+            {{ adapter.delta_create_table(target_relation, sql, unique_key, partition_by, custom_location, delta_create_table_write_options) }}
             {% set build_sql = "select * from " + target_relation.schema + "." + target_relation.identifier + " limit 1 " %}
         {% elif file_format == 'iceberg' %}
             {{ adapter.iceberg_write(target_relation, sql, unique_key, partition_by, custom_location, strategy, table_properties) }}