fix: properly quote reserved keyword columns in snapshot unique_key

dbt/include/exasol/macros/materializations/snapshot.sql

@@ -7,6 +7,22 @@
   {%- endif -%}
 {%- endmacro %}
 
+{#
+    Exasol-specific unique_key_fields to properly quote column names.
+    This handles reserved keywords like TIME that must be quoted.
+    Uses exasol__quote_column from strategies.sql for consistent quoting.
+#}
+{% macro exasol__unique_key_fields(unique_key) %}
+    {% if unique_key | is_list %}
+        {% for key in unique_key %}
+            {{ exasol__quote_column(key) }} as dbt_unique_key_{{ loop.index }}
+            {%- if not loop.last %} , {%- endif %}
+        {% endfor %}
+    {% else %}
+        {{ exasol__quote_column(unique_key) }} as dbt_unique_key
+    {% endif %}
+{% endmacro %}
+
 {% macro exasol__build_snapshot_table(strategy, sql) %}
     {% set columns = config.get('snapshot_table_column_names') or get_snapshot_table_column_names() %}
 
@@ -39,7 +55,7 @@
 
         select
             sd.*,
-            {{ unique_key_fields(strategy.unique_key) }}
+            {{ exasol__unique_key_fields(strategy.unique_key) }}
         from {{ target_relation | upper }} sd
         where
             {% if config.get('dbt_valid_to_current') %}
@@ -54,7 +70,7 @@
 
         select
             sq.*,
-            {{ unique_key_fields(strategy.unique_key) }},
+            {{ exasol__unique_key_fields(strategy.unique_key) }},
             {{ strategy.updated_at }} as {{ columns.dbt_updated_at }},
             {{ strategy.updated_at }} as {{ columns.dbt_valid_from }},
             {{ get_dbt_valid_to_current(strategy, columns) }},
@@ -67,7 +83,7 @@
 
         select
             sq.*,
-            {{ unique_key_fields(strategy.unique_key) }},
+            {{ exasol__unique_key_fields(strategy.unique_key) }},
             {{ strategy.updated_at }} as {{ columns.dbt_updated_at }},
             {{ strategy.updated_at }} as {{ columns.dbt_valid_from }},
             {{ strategy.updated_at }} as {{ columns.dbt_valid_to }}
@@ -81,7 +97,7 @@
 
         select
             sq.*,
-            {{ unique_key_fields(strategy.unique_key) }}
+            {{ exasol__unique_key_fields(strategy.unique_key) }}
         from snapshot_query sq
     ),
     {% endif %}
@@ -183,7 +199,7 @@
             {% endfor -%}
             {%- if strategy.unique_key | is_list -%}
                 {%- for key in strategy.unique_key -%}
-            snapshotted_data.{{ key }} as dbt_unique_key_{{ loop.index }},
+            snapshotted_data.{{ exasol__quote_column(key) }} as dbt_unique_key_{{ loop.index }},
                 {% endfor -%}
             {%- else -%}
             snapshotted_data.dbt_unique_key as dbt_unique_key,

dbt/include/exasol/macros/materializations/snapshot_merge.sql

@@ -21,8 +21,8 @@
             and DBT_INTERNAL_SOURCE.dbt_change_type in ('update', 'delete')
 
     when not matched
-        then insert ({{ insert_cols_csv | upper }})
-        values ({{ insert_cols_csv | upper}})
+        then insert ({{ insert_cols_csv }})
+        values ({{ insert_cols_csv }})
         where DBT_INTERNAL_SOURCE.dbt_change_type = 'insert'
     ;
-{% endmacro %}
+{% endmacro %}

dbt/include/exasol/macros/materializations/strategies.sql

@@ -4,6 +4,82 @@
     {% endfor %})
 {% endmacro %}
 
+{#
+    Helper macro to quote a column name for Exasol.
+    - If already quoted (starts and ends with "), pass through as-is (user explicitly specified case)
+    - Otherwise, uppercase and quote (standard Exasol behavior for unquoted identifiers)
+
+    This handles the common case where:
+    - Regular columns are created without quotes and stored as UPPERCASE
+    - Reserved keyword columns should be specified with quotes in the config: '"time"'
+#}
+{% macro exasol__quote_column(col) -%}
+    {%- if col.startswith('"') and col.endswith('"') -%}
+        {{- col -}}
+    {%- else -%}
+        {{- adapter.quote(col | upper) -}}
+    {%- endif -%}
+{%- endmacro %}
+
+{#
+    Build properly quoted scd_args for Exasol snapshots.
+    Handles both single and composite unique_key, plus the updated_at column.
+#}
+{% macro exasol__build_scd_args(primary_key, updated_at) %}
+    {% set quoted_args = [] %}
+    {% if primary_key is string %}
+        {% do quoted_args.append(exasol__quote_column(primary_key)) %}
+    {% else %}
+        {% for pk in primary_key %}
+            {% do quoted_args.append(exasol__quote_column(pk)) %}
+        {% endfor %}
+    {% endif %}
+    {% do quoted_args.append(exasol__quote_column(updated_at)) %}
+    {{ return(quoted_args) }}
+{% endmacro %}
+
+{#
+    IMPORTANT: Global override of snapshot_timestamp_strategy.
+
+    This macro does NOT use the exasol__ prefix because dbt's strategy_dispatch()
+    function does not use adapter.dispatch() - it directly calls the strategy macro
+    by name. Therefore, we must override the global macro to intercept it for Exasol.
+
+    WARNING: In multi-adapter environments where multiple adapter packages are loaded,
+    this global override may cause conflicts. This is a known limitation of dbt's
+    snapshot strategy dispatch mechanism. For Exasol-only projects, this works correctly.
+
+    Purpose: Properly quote column names (unique_key, updated_at) to handle SQL
+    reserved keywords like TIME, DATE, USER, etc.
+#}
+{% macro snapshot_timestamp_strategy(node, snapshotted_rel, current_rel, model_config, target_exists) %}
+    {% set primary_key = config.get('unique_key') %}
+    {% set updated_at = config.get('updated_at') %}
+    {% set hard_deletes = adapter.get_hard_deletes_behavior(config) %}
+    {% set invalidate_hard_deletes = hard_deletes == 'invalidate' %}
+    {% set columns = config.get("snapshot_table_column_names") or get_snapshot_table_column_names() %}
+
+    {# Quote the updated_at column for use in row_changed expression #}
+    {% set quoted_updated_at = exasol__quote_column(updated_at) %}
+
+    {% set row_changed_expr -%}
+        ({{ snapshotted_rel }}.{{ columns.dbt_valid_from }} < {{ current_rel }}.{{ quoted_updated_at }})
+    {%- endset %}
+
+    {# Build properly quoted scd_args #}
+    {% set scd_args = exasol__build_scd_args(primary_key, updated_at) %}
+    {% set scd_id_expr = snapshot_hash_arguments(scd_args) %}
+
+    {% do return({
+        "unique_key": primary_key,
+        "updated_at": quoted_updated_at,
+        "row_changed": row_changed_expr,
+        "scd_id": scd_id_expr,
+        "invalidate_hard_deletes": invalidate_hard_deletes,
+        "hard_deletes": hard_deletes
+    }) %}
+{% endmacro %}
+
 {% macro exasol__snapshot_check_all_get_existing_columns(node, target_exists) -%}
     {%- set query_columns = get_columns_in_query(node['injected_sql']) -%}
     {%- if not target_exists -%}
@@ -28,6 +104,10 @@
     {{ return([ns.column_added, intersection]) }}
 {%- endmacro %}
 
+{#
+    Exasol-specific snapshot_check_strategy with proper quoting for reserved keywords.
+    Uses exasol__ prefix for proper dispatch.
+#}
 {% macro exasol__snapshot_check_strategy(node, snapshotted_rel, current_rel, config, target_exists) %}
     {% set check_cols_config = config['check_cols'] %}
     {% set primary_key = config['unique_key'] %}
@@ -54,22 +134,34 @@
         {% do exceptions.raise_compiler_error("Invalid value for 'check_cols': " ~ check_cols_config) %}
     {% endif %}
 
+    {# Quote check_cols for row_changed expression #}
     {%- set row_changed_expr -%}
     (
     {%- if column_added -%}
         TRUE
     {%- else -%}
     {%- for col in check_cols -%}
-        {{ snapshotted_rel }}.{{ col }} != {{ current_rel }}.{{ col }}
+        {% set quoted_col = exasol__quote_column(col) %}
+        {{ snapshotted_rel }}.{{ quoted_col }} != {{ current_rel }}.{{ quoted_col }}
         or
-        ({{ snapshotted_rel }}.{{ col }} is null) != ({{ current_rel }}.{{ col }} is null)
+        ({{ snapshotted_rel }}.{{ quoted_col }} is null) != ({{ current_rel }}.{{ quoted_col }} is null)
         {%- if not loop.last %} or {% endif -%}
     {%- endfor -%}
     {%- endif -%}
     )
     {%- endset %}
 
-    {% set scd_id_expr = snapshot_hash_arguments([primary_key, updated_at]) %}
+    {# Build properly quoted scd_args - for check strategy, updated_at is a timestamp literal, not a column #}
+    {% set quoted_args = [] %}
+    {% if primary_key is string %}
+        {% do quoted_args.append(exasol__quote_column(primary_key)) %}
+    {% else %}
+        {% for pk in primary_key %}
+            {% do quoted_args.append(exasol__quote_column(pk)) %}
+        {% endfor %}
+    {% endif %}
+    {% do quoted_args.append(updated_at) %}
+    {% set scd_id_expr = snapshot_hash_arguments(quoted_args) %}
 
     {% do return({
         "unique_key": primary_key,