Fix table materialization for Delta models (#420)

* added delta_table_replace materialization

* fixed table materialization for delta format

* removed or replace without as clause

CHANGELOG.md

@@ -1,5 +1,6 @@
-## New versio
+## New version
 - Fix session provisioning timeout and delay handling
+- Fix table materialization for Delta models
 
 ## v1.8.1
 - Fix typo in README.md

dbt/adapters/glue/impl.py

@@ -187,11 +187,14 @@ def get_relation(self, database, schema, identifier):
                 DatabaseName=schema,
                 Name=identifier
             )
+            is_delta = response.get('Table').get("Parameters").get("spark.sql.sources.provider") == "delta"
+
             relations = self.Relation.create(
                 database=schema,
                 schema=schema,
                 identifier=identifier,
-                type=self.relation_type_map.get(response.get("Table", {}).get("TableType", "Table"))
+                type=self.relation_type_map.get(response.get("Table", {}).get("TableType", "Table")),
+                is_delta=is_delta
             )
             logger.debug(f"""schema : {schema}
                              identifier : {identifier}

dbt/include/glue/macros/adapters.sql

@@ -62,10 +62,18 @@
   {%- set file_format = config.get('file_format', validator=validation.any[basestring]) -%}
   {%- set table_properties = config.get('table_properties', default={}) -%}
 
+  {%- set create_statement_string -%}
+    {% if file_format in ['delta', 'iceberg'] -%}
+      create or replace table
+    {%- else -%}
+      create table
+    {% endif %}
+  {%- endset %}
+
   {% if temporary -%}
     {{ create_temporary_view(relation, sql) }}
   {%- else -%}
-    	create table {{ relation }}
+    	{{ create_statement_string }} {{ relation }}
     	{% set contract_config = config.get('contract') %}
       {% if contract_config.enforced %}
         {{ get_assert_columns_equivalent(sql) }}