fix: Remove non-current iceberg columns from catalog

dbt/adapters/athena/impl.py

@@ -548,8 +548,7 @@ def _s3_path_exists(self, s3_bucket: str, s3_prefix: str) -> bool:
         response = s3_client.list_objects_v2(Bucket=s3_bucket, Prefix=s3_prefix)
         return True if "Contents" in response else False
 
-    @staticmethod
-    def _get_one_table_for_catalog(table: TableTypeDef, database: str) -> List[Dict[str, Any]]:
+    def _get_one_table_for_catalog(self, table: TableTypeDef, database: str) -> List[Dict[str, Any]]:
         table_catalog = {
             "table_database": database,
             "table_schema": table["DatabaseName"],
@@ -568,6 +567,7 @@ def _get_one_table_for_catalog(table: TableTypeDef, database: str) -> List[Dict[
                 },
             }
             for idx, col in enumerate(table["StorageDescriptor"]["Columns"] + table.get("PartitionKeys", []))
+            if self._is_current_column(col)
         ]
 
     @staticmethod
@@ -1098,9 +1098,7 @@ def _is_current_column(col: ColumnTypeDef) -> bool:
         """
         Check if a column is explicitly set as not current. If not, it is considered as current.
         """
-        if col.get("Parameters", {}).get("iceberg.field.current") == "false":
-            return False
-        return True
+        return bool(col.get("Parameters", {}).get("iceberg.field.current") != "false")
 
     @available
     def get_columns_in_relation(self, relation: AthenaRelation) -> List[AthenaColumn]:

tests/functional/adapter/test_docs.py

@@ -15,6 +15,17 @@
 select 1 as id
 """
 
+iceberg_model_sql = """
+{{
+    config(
+        materialized="table",
+        table_type="iceberg",
+        post_hook="alter table model drop column to_drop"
+    )
+}}
+select 1 as id, 'to_drop' as to_drop
+"""
+
 override_macros_sql = """
 {% macro get_catalog_relations(information_schema, relations) %}
     {{ return(adapter.get_catalog_by_relations(information_schema, relations)) }}
@@ -88,14 +99,37 @@ def models(self):
     def macros(self):
         return {"override_macros_sql.sql": override_macros_sql}
 
-    def test_generate_docs(
-        self,
-        project,
-    ):
+    def test_generate_docs(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 1
+
+        docs_generate = run_dbt(["--warn-error", "docs", "generate"])
+        assert len(docs_generate._compile_results.results) == 1
+        assert docs_generate._compile_results.results[0].status == RunStatus.Success
+        assert docs_generate.errors is None
+
+
+class TestDocsGenerateIcebergNonCurrentColumn:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"model.sql": iceberg_model_sql}
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"override_macros_sql.sql": override_macros_sql}
+
+    def test_generate_docs(self, project):
         results = run_dbt(["run"])
         assert len(results) == 1
 
         docs_generate = run_dbt(["--warn-error", "docs", "generate"])
         assert len(docs_generate._compile_results.results) == 1
         assert docs_generate._compile_results.results[0].status == RunStatus.Success
         assert docs_generate.errors is None
+
+        catalog_path = os.path.join(project.project_root, "target", "catalog.json")
+        assert os.path.exists(catalog_path)
+        catalog = get_artifact(catalog_path)
+        columns = catalog["nodes"]["model.test.model"]["columns"]
+        assert "to_drop" not in columns
+        assert "id" in columns