fix: Prune old table version for incremental models (#533)

Author: svdimchenko

dbt/adapters/athena/impl.py

@@ -889,9 +889,7 @@ def _get_glue_table_versions_to_expire(self, relation: AthenaRelation, to_keep:
         return table_versions_ordered[int(to_keep) :]
 
     @available
-    def expire_glue_table_versions(
-        self, relation: AthenaRelation, to_keep: int, delete_s3: bool
-    ) -> List[TableVersionTypeDef]:
+    def expire_glue_table_versions(self, relation: AthenaRelation, to_keep: int, delete_s3: bool) -> List[str]:
         conn = self.connections.get_thread_connection()
         creds = conn.credentials
         client = conn.handle
@@ -924,11 +922,9 @@ def expire_glue_table_versions(
                 LOGGER.debug(f"Deleted version {version} of table {relation.render()} ")
                 if delete_s3:
                     self.delete_from_s3(location)
+                    LOGGER.debug(f"{location} was deleted")
             except Exception as err:
                 LOGGER.debug(f"There was an error when expiring table version {version} with error: {err}")
-
-            LOGGER.debug(f"{location} was deleted")
-
         return deleted_versions
 
     @available

dbt/include/athena/macros/adapters/persist_docs.sql

@@ -1,12 +1,14 @@
-{% macro athena__persist_docs(relation, model, for_relation, for_columns) -%}
+{% macro athena__persist_docs(relation, model, for_relation=true, for_columns=true) -%}
   {% set persist_relation_docs = for_relation and config.persist_relation_docs() and model.description %}
   {% set persist_column_docs = for_columns and config.persist_column_docs() and model.columns %}
-  {% set skip_archive_table_version = not is_incremental() %}
   {% if persist_relation_docs or persist_column_docs %}
-    {% do adapter.persist_docs_to_glue(relation,
-                                       model,
-                                       persist_relation_docs,
-                                       persist_column_docs,
-                                       skip_archive_table_version=skip_archive_table_version) %}}
+    {% do adapter.persist_docs_to_glue(
+            relation=relation,
+            model=model,
+            persist_relation_docs=persist_relation_docs,
+            persist_column_docs=persist_column_docs,
+            skip_archive_table_version=true
+         )
+    %}}
   {% endif %}
 {% endmacro %}

dbt/include/athena/macros/materializations/models/incremental/incremental.sql

@@ -4,7 +4,7 @@
   {% set table_type = config.get('table_type', default='hive') | lower %}
   {% set strategy = validate_get_incremental_strategy(raw_strategy, table_type) %}
   {% set on_schema_change = incremental_validate_on_schema_change(config.get('on_schema_change'), default='ignore') %}
-
+  {% set versions_to_keep = config.get('versions_to_keep', 1) | as_number %}
   {% set lf_tags_config = config.get('lf_tags_config') %}
   {% set lf_grants = config.get('lf_grants') %}
   {% set partitioned_by = config.get('partitioned_by') %}
@@ -125,6 +125,8 @@
 
   {% do persist_docs(target_relation, model) %}
 
+  {% do adapter.expire_glue_table_versions(target_relation, versions_to_keep, False) %}
+
   {{ return({'relations': [target_relation]}) }}
 
 {%- endmaterialization %}

dbt/include/athena/macros/materializations/models/view/view.sql

@@ -8,7 +8,7 @@
 
     {% set to_return = create_or_replace_view(run_outside_transaction_hooks=False) %}
 
-    {% do adapter.expire_glue_table_versions(target_relation, versions_to_keep, True) %}
+    {% do adapter.expire_glue_table_versions(target_relation, versions_to_keep, False) %}
 
     {% set target_relation = this.incorporate(type='view') %}
     {% do persist_docs(target_relation, model) %}