fix: add metadata_properties to _construct_parameters when update hive table

Author: kadai0308

pyiceberg/catalog/hive.py

@@ -211,11 +211,18 @@ def _construct_hive_storage_descriptor(
 DEFAULT_PROPERTIES = {TableProperties.PARQUET_COMPRESSION: TableProperties.PARQUET_COMPRESSION_DEFAULT}
 
 
-def _construct_parameters(metadata_location: str, previous_metadata_location: Optional[str] = None) -> Dict[str, Any]:
+def _construct_parameters(
+    metadata_location: str, previous_metadata_location: Optional[str] = None, metadata_properties: Optional[Properties] = None
+) -> Dict[str, Any]:
     properties = {PROP_EXTERNAL: "TRUE", PROP_TABLE_TYPE: "ICEBERG", PROP_METADATA_LOCATION: metadata_location}
     if previous_metadata_location:
         properties[PROP_PREVIOUS_METADATA_LOCATION] = previous_metadata_location
 
+    if metadata_properties:
+        for key, value in metadata_properties.items():
+            if key not in properties:
+                properties[key] = str(value)
+
     return properties
 
 
@@ -360,7 +367,7 @@ def _convert_iceberg_into_hive(self, table: Table) -> HiveTable:
                 property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
             ),
             tableType=EXTERNAL_TABLE,
-            parameters=_construct_parameters(table.metadata_location),
+            parameters=_construct_parameters(metadata_location=table.metadata_location, metadata_properties=table.properties),
         )
 
     def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None:
@@ -541,6 +548,7 @@ def commit_table(
                     hive_table.parameters = _construct_parameters(
                         metadata_location=updated_staged_table.metadata_location,
                         previous_metadata_location=current_table.metadata_location,
+                        metadata_properties=updated_staged_table.properties,
                     )
                     open_client.alter_table_with_environment_context(
                         dbname=database_name,

tests/integration/test_reads.py

@@ -48,7 +48,7 @@
     pyarrow_to_schema,
 )
 from pyiceberg.schema import Schema
-from pyiceberg.table import Table
+from pyiceberg.table import Table, update_table_metadata
 from pyiceberg.types import (
     BinaryType,
     BooleanType,
@@ -59,6 +59,7 @@
     TimestampType,
 )
 from pyiceberg.utils.concurrent import ExecutorFactory
+from pyiceberg.table.update import SetPropertiesUpdate, RemovePropertiesUpdate
 
 DEFAULT_PROPERTIES = {"write.parquet.compression-codec": "zstd"}
 
@@ -111,6 +112,23 @@ def test_table_properties(catalog: Catalog) -> None:
         table.transaction().set_properties(property_name=None).commit_transaction()
     assert "None type is not a supported value in properties: property_name" in str(exc_info.value)
 
+    if isinstance(catalog, HiveCatalog):
+        table.transaction().set_properties({"abc": "def", "p1": "123"}).commit_transaction()
+
+        hive_client: _HiveClient = _HiveClient(catalog.properties["uri"])
+
+        with hive_client as open_client:
+            hive_table = open_client.get_table(*TABLE_NAME)
+            assert hive_table.parameters.get("abc") == "def"
+            assert hive_table.parameters.get("p1") == "123"
+            assert hive_table.parameters.get("not_exist_parameter") is None
+
+        table.transaction().remove_properties("abc").commit_transaction()
+
+        with hive_client as open_client:
+            hive_table = open_client.get_table(*TABLE_NAME)
+            assert hive_table.parameters.get("abc") is None
+
 
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])