add parameter

Author: kukushking

awswrangler/athena/_write_iceberg.py

@@ -115,7 +115,13 @@ def _determine_differences(
 
     catalog_column_types = typing.cast(
         Dict[str, str],
-        catalog.get_table_types(database=database, table=table, catalog_id=catalog_id, boto3_session=boto3_session),
+        catalog.get_table_types(
+            database=database,
+            table=table,
+            catalog_id=catalog_id,
+            return_iceberg_current=True,
+            boto3_session=boto3_session,
+        ),
     )
 
     original_column_names = set(catalog_column_types)

awswrangler/catalog/_get.py

@@ -107,6 +107,7 @@ def get_table_types(
     database: str,
     table: str,
     catalog_id: str | None = None,
+    return_iceberg_current: bool = False,
     boto3_session: boto3.Session | None = None,
 ) -> dict[str, str] | None:
     """Get all columns and types from a table.
@@ -120,6 +121,9 @@ def get_table_types(
     catalog_id
         The ID of the Data Catalog from which to retrieve Databases.
         If ``None`` is provided, the AWS account ID is used by default.
+    return_iceberg_current
+            If True, returns only current iceberg fields (fields marked with iceberg.field.current: true).
+            Otherwise, returns the all fields. False by default (return all fields).
     boto3_session
         The default boto3 session will be used if **boto3_session** receive ``None``.
 
@@ -139,7 +143,10 @@ def get_table_types(
         response = client_glue.get_table(**_catalog_id(catalog_id=catalog_id, DatabaseName=database, Name=table))
     except client_glue.exceptions.EntityNotFoundException:
         return None
-    return _extract_dtypes_from_table_details(response=response)
+    return _extract_dtypes_from_table_details(
+        response=response,
+        return_iceberg_current=return_iceberg_current,
+    )
 
 
 def get_databases(

awswrangler/catalog/_utils.py

@@ -31,12 +31,16 @@ def _sanitize_name(name: str) -> str:
     return re.sub("[^A-Za-z0-9_]+", "_", name).lower()  # Replacing non alphanumeric characters by underscore
 
 
-def _extract_dtypes_from_table_details(response: "GetTableResponseTypeDef") -> dict[str, str]:
+def _extract_dtypes_from_table_details(
+    response: "GetTableResponseTypeDef",
+    return_iceberg_current: bool = False,
+) -> dict[str, str]:
     dtypes: dict[str, str] = {}
     for col in response["Table"]["StorageDescriptor"]["Columns"]:
-        # Do not return "hidden" iceberg columns
-        if col.get("Parameters", {}).get("iceberg.field.current") != "false":
+        # Only return current fields if flag is enabled
+        if not return_iceberg_current or col.get("Parameters", {}).get("iceberg.field.current") == "true":
             dtypes[col["Name"]] = col["Type"]
+    # Add partition keys as columns
     if "PartitionKeys" in response["Table"]:
         for par in response["Table"]["PartitionKeys"]:
             dtypes[par["Name"]] = par["Type"]