(hive) branch master updated: HIVE-27593: Iceberg: Keep Iceberg properties in sync with HMS (Butao Zhang, reviewed by Denys Kuzmenko)

[dk...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dkuzmenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git


The following commit(s) were added to refs/heads/master by this push:
     new a4aef817326 HIVE-27593: Iceberg: Keep Iceberg properties in sync with HMS (Butao Zhang, reviewed by Denys Kuzmenko)
a4aef817326 is described below

commit a4aef8173269d213eb13666baf1751bf110579c8
Author: Butao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Wed Nov 8 17:15:35 2023 +0800

    HIVE-27593: Iceberg: Keep Iceberg properties in sync with HMS (Butao Zhang, reviewed by Denys Kuzmenko)
    
    Closes #4573
---
 .../iceberg/mr/hive/HiveIcebergMetaHook.java       | 36 +++++++++++++----
 .../hive/TestHiveIcebergStorageHandlerNoScan.java  | 47 ++++++++++++++++++++++
 2 files changed, 76 insertions(+), 7 deletions(-)

diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
index f174bd7b2fe..540a3b8638c 100644
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
@@ -128,6 +128,10 @@ import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.iceberg.TableProperties.DELETE_MODE;
+import static org.apache.iceberg.TableProperties.MERGE_MODE;
+import static org.apache.iceberg.TableProperties.UPDATE_MODE;
+
 public class HiveIcebergMetaHook implements HiveMetaHook {
   private static final Logger LOG = LoggerFactory.getLogger(HiveIcebergMetaHook.class);
   public static final Map<String, String> COMMON_HMS_PROPERTIES = ImmutableMap.of(
@@ -428,7 +432,7 @@ public class HiveIcebergMetaHook implements HiveMetaHook {
       assertNotMigratedTable(hmsTable.getParameters(), "CHANGE COLUMN");
       handleChangeColumn(hmsTable);
     } else {
-      setDeleteModeOnTableProperties(icebergTable, hmsTable.getParameters());
+      setDeleteModeOnTableProperties(icebergTable, hmsTable.getParameters(), context);
       assertNotCrossTableMetadataLocationChange(hmsTable.getParameters(), context);
     }
 
@@ -741,7 +745,7 @@ public class HiveIcebergMetaHook implements HiveMetaHook {
     // Remove creation related properties
     PARAMETERS_TO_REMOVE.forEach(hmsParams::remove);
 
-    setDeleteModeOnTableProperties(null, hmsParams);
+    setDeleteModeOnTableProperties(null, hmsParams, null);
   }
 
   /**
@@ -981,14 +985,32 @@ public class HiveIcebergMetaHook implements HiveMetaHook {
   }
 
   // TODO: remove this if copy-on-write mode gets implemented in Hive
-  private static void setDeleteModeOnTableProperties(Table icebergTable, Map<String, String> newProps) {
+  private void setDeleteModeOnTableProperties(Table icebergTbl, Map<String, String> newProps,
+      EnvironmentContext context) {
     // Hive only supports merge-on-read delete mode, it will actually throw an error if DML operations are attempted on
     // tables that don't have this (the default is copy-on-write). We set this at table creation and v1->v2 conversion.
-    if ((icebergTable == null || ((BaseTable) icebergTable).operations().current().formatVersion() == 1) &&
+    if ((icebergTbl == null || ((BaseTable) icebergTbl).operations().current().formatVersion() == 1) &&
         IcebergTableUtil.isV2Table(newProps)) {
-      newProps.put(TableProperties.DELETE_MODE, HiveIcebergStorageHandler.MERGE_ON_READ);
-      newProps.put(TableProperties.UPDATE_MODE, HiveIcebergStorageHandler.MERGE_ON_READ);
-      newProps.put(TableProperties.MERGE_MODE, HiveIcebergStorageHandler.MERGE_ON_READ);
+      List<String> writeModeList = Arrays.asList(DELETE_MODE, UPDATE_MODE, MERGE_MODE);
+      writeModeList.stream()
+          .filter(writeMode -> catalogProperties.get(writeMode) == null)
+          .forEach(writeMode -> {
+            catalogProperties.put(writeMode, HiveIcebergStorageHandler.MERGE_ON_READ);
+            newProps.put(writeMode, HiveIcebergStorageHandler.MERGE_ON_READ);
+          });
+
+      if (context != null) {
+        Splitter splitter = Splitter.on(PROPERTIES_SEPARATOR);
+        Map<String, String> contextProperties = context.getProperties();
+        if (contextProperties.containsKey(SET_PROPERTIES)) {
+          String propValue = context.getProperties().get(SET_PROPERTIES);
+          String writeModeStr = writeModeList.stream().filter(writeMode ->
+              !splitter.splitToList(propValue).contains(writeMode)).collect(Collectors.joining("'"));
+          if (!writeModeStr.isEmpty()) {
+            contextProperties.put(SET_PROPERTIES, propValue + "'" + writeModeStr);
+          }
+        }
+      }
     }
   }
 
diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java
index dc17e472b9c..0baf210340f 100644
--- a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java
+++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java
@@ -1928,6 +1928,53 @@ public class TestHiveIcebergStorageHandlerNoScan {
     Assert.assertTrue(hmsTblLocation.getFileSystem(shell.getHiveConf()).exists(hmsTblLocation));
   }
 
+  @Test
+  public void testSnycProperties() throws TException, InterruptedException {
+    Assume.assumeTrue("This test is only for hive catalog", testTableType == TestTables.TestTableType.HIVE_CATALOG);
+
+    // Test create v2 iceberg table and check iceberg properties & hms properties
+    TableIdentifier identifier = TableIdentifier.of("default", "customers_v2");
+    shell.executeStatement("CREATE TABLE customers_v2 (id int, name string) Stored by Iceberg stored as ORC " +
+        "TBLPROPERTIES ('format-version'='2','write.delete.mode'='copy-on-write')");
+    org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
+    org.apache.hadoop.hive.metastore.api.Table hmsTable = shell.metastore().getTable("default", "customers_v2");
+    Map<String, String> icePros = icebergTable.properties();
+    Map<String, String> hmsProps = hmsTable.getParameters();
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), HiveIcebergStorageHandler.COPY_ON_WRITE);
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), HiveIcebergStorageHandler.MERGE_ON_READ);
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), HiveIcebergStorageHandler.MERGE_ON_READ);
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), hmsProps.get(TableProperties.DELETE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), hmsProps.get(TableProperties.UPDATE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), hmsProps.get(TableProperties.MERGE_MODE));
+
+    // Test create v1 iceberg table and check its properties before and after it upgrades to v2
+    identifier = TableIdentifier.of("default", "customers_v1");
+    shell.executeStatement("CREATE TABLE customers_v1 (id int, name string) Stored by Iceberg stored as ORC");
+    icebergTable = testTables.loadTable(identifier);
+    hmsTable = shell.metastore().getTable("default", "customers_v1");
+    icePros = icebergTable.properties();
+    hmsProps = hmsTable.getParameters();
+    // check v1 iceberg table properties
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), null);
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), null);
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), null);
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), hmsProps.get(TableProperties.DELETE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), hmsProps.get(TableProperties.UPDATE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), hmsProps.get(TableProperties.MERGE_MODE));
+    // check table properties after upgrading to v2
+    shell.executeStatement("ALTER TABLE customers_v1 SET TBLPROPERTIES ('format-version'='2')");
+    icebergTable = testTables.loadTable(identifier);
+    hmsTable = shell.metastore().getTable("default", "customers_v1");
+    icePros = icebergTable.properties();
+    hmsProps = hmsTable.getParameters();
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), HiveIcebergStorageHandler.MERGE_ON_READ);
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), HiveIcebergStorageHandler.MERGE_ON_READ);
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), HiveIcebergStorageHandler.MERGE_ON_READ);
+    Assert.assertEquals(icePros.get(TableProperties.DELETE_MODE), hmsProps.get(TableProperties.DELETE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.UPDATE_MODE), hmsProps.get(TableProperties.UPDATE_MODE));
+    Assert.assertEquals(icePros.get(TableProperties.MERGE_MODE), hmsProps.get(TableProperties.MERGE_MODE));
+  }
+
   private String getCurrentSnapshotForHiveCatalogTable(org.apache.iceberg.Table icebergTable) {
     return ((BaseMetastoreTableOperations) ((BaseTable) icebergTable).operations()).currentMetadataLocation();
   }