Repository: NetEase/arctic (java)

Issue:
[Bug]: Incorrect data synchronization caused the data of the arctic base table to be overwritten
### What happened?

The arctic base table was overwritten due to incorrect data synchronization after an empty hive table was upgraded to an arctic table. Because empty hive tables don't have full 'table.partition-properties' when upgraded, But we need 'hive-location' and 'transient-time' in 'table.partition-properties' to decide if we want to sync our data

### Affects Versions

0.4.0

### What engines are you seeing the problem on?

_No response_

### How to reproduce

_No response_

### Relevant log output

_No response_

### Anything else

_No response_

### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Pull request description:
[ARCTIC-1378] Incorrect data synchronization caused the data of the arctic base table to be overwritten
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
resolve #1378

## Brief change log

Set partition properties for all partitions even if there are no files under the partition when upgrading a Hive table to an Arctic table.

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? (not documented)


Patch:
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java
index 1474b2c4e..50e142b92 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.hive.utils;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.netease.arctic.hive.HMSClientPool;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.op.OverwriteHiveFiles;
@@ -221,7 +222,8 @@ public class HiveMetaSynchronizer {
     }
   }
 
-  private static boolean partitionHasModified(
+  @VisibleForTesting
+  static boolean partitionHasModified(
       UnkeyedTable arcticTable, Partition hivePartition,
       StructLike partitionData) {
     String hiveTransientTime = hivePartition.getParameters().get("transient_lastDdlTime");
@@ -248,7 +250,8 @@ public class HiveMetaSynchronizer {
     return false;
   }
 
-  private static boolean tableHasModified(UnkeyedTable arcticTable, Table table) {
+  @VisibleForTesting
+  static boolean tableHasModified(UnkeyedTable arcticTable, Table table) {
     String hiveTransientTime = table.getParameters().get("transient_lastDdlTime");
     StructLikeMap<Map<String, String>> structLikeMap = arcticTable.partitionProperty();
     String arcticTransientTime = null;
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
index e6e36a2a8..fc241465c 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
@@ -136,7 +136,7 @@ public class HivePartitionUtil {
    * @param tableIdentifier A table identifier
    * @return A List of Hive partition objects
    */
-  public List<Partition> getHiveAllPartitions(HMSClientPool hiveClient, TableIdentifier tableIdentifier) {
+  public static List<Partition> getHiveAllPartitions(HMSClientPool hiveClient, TableIdentifier tableIdentifier) {
     try {
       return hiveClient.run(client ->
           client.listPartitions(tableIdentifier.getDatabase(), tableIdentifier.getTableName(), Short.MAX_VALUE));
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
index 778d533bc..d4002ad8d 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
@@ -18,20 +18,27 @@
 
 package com.netease.arctic.hive.utils;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.netease.arctic.hive.HMSClientPool;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
 import com.netease.arctic.hive.table.SupportHive;
+import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.io.ArcticHadoopFileIO;
+import com.netease.arctic.op.UpdatePartitionProperties;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.SerDeInfo;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -136,6 +143,8 @@ public class UpgradeHiveTableUtil {
       }
     }
     HiveMetaSynchronizer.syncHiveDataToArctic(arcticTable, arcticHiveCatalog.getHMSClient());
+    hiveTable = HiveTableUtil.loadHmsTable(arcticHiveCatalog.getHMSClient(), tableIdentifier);
+    fillPartitionProperties(arcticTable, arcticHiveCatalog, hiveTable);
   }
 
   /**
@@ -172,4 +181,57 @@ public class UpgradeHiveTableUtil {
     }
     return isSupport.get();
   }
+
+  @VisibleForTesting
+  static void fillPartitionProperties(ArcticTable table, ArcticHiveCatalog arcticHiveCatalog, Table hiveTable) {
+    UnkeyedHiveTable baseTable;
+    if (table.isKeyedTable()) {
+      baseTable = (UnkeyedHiveTable) table.asKeyedTable().baseTable();
+    } else {
+      baseTable = (UnkeyedHiveTable) table.asUnkeyedTable();
+    }
+    UpdatePartitionProperties updatePartitionProperties =
+        baseTable.updatePartitionProperties(null);
+    if (table.spec().isUnpartitioned()) {
+      if (hasPartitionProperties(baseTable, false, null)) {
+        return;
+      }
+      updatePartitionProperties.set(
+          TablePropertyUtil.EMPTY_STRUCT,
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, baseTable.hiveLocation());
+      updatePartitionProperties.set(
+          TablePropertyUtil.EMPTY_STRUCT,
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+          hiveTable.getParameters().get("transient_lastDdlTime"));
+    } else {
+      List<Partition> partitions =
+          HivePartitionUtil.getHiveAllPartitions(arcticHiveCatalog.getHMSClient(), table.id());
+      partitions.forEach(partition -> {
+        StructLike partitionData = DataFiles.data(table.spec(), String.join("/", partition.getValues()));
+        if (hasPartitionProperties(baseTable, true, partitionData)) {
+          return;
+        }
+        updatePartitionProperties.set(
+            partitionData,
+            HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+            partition.getSd().getLocation());
+        updatePartitionProperties.set(
+            partitionData,
+            HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+            partition.getParameters().get("transient_lastDdlTime"));
+      });
+    }
+    updatePartitionProperties.commit();
+  }
+
+  private static boolean hasPartitionProperties(
+      UnkeyedHiveTable baseTable,
+      boolean isPartitioned,
+      StructLike partitionData) {
+    Map<String, String> partitionProperties = isPartitioned ? baseTable.partitionProperty().get(partitionData) :
+        baseTable.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT);
+    return partitionProperties != null &&
+        partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION) &&
+        partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME);
+  }
 }


Test patch:
diff --git a/hive/src/test/java/com/netease/arctic/hive/utils/TestUpgradeHiveTableUtil.java b/hive/src/test/java/com/netease/arctic/hive/utils/TestUpgradeHiveTableUtil.java
new file mode 100644
index 000000000..d6dbecad3
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/utils/TestUpgradeHiveTableUtil.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *  *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.utils;
+
+import com.netease.arctic.ams.api.TableFormat;
+import com.netease.arctic.catalog.CatalogTestBase;
+import com.netease.arctic.catalog.CatalogTestHelper;
+import com.netease.arctic.hive.HiveTableProperties;
+import com.netease.arctic.hive.TestHMS;
+import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
+import com.netease.arctic.hive.catalog.HiveCatalogTestHelper;
+import com.netease.arctic.hive.catalog.HiveTableTestHelper;
+import com.netease.arctic.hive.table.UnkeyedHiveTable;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.TablePropertyUtil;
+import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+@RunWith(Parameterized.class)
+public class TestUpgradeHiveTableUtil extends CatalogTestBase {
+
+  @ClassRule
+  public static TestHMS TEST_HMS = new TestHMS();
+
+  @Rule
+  public TemporaryFolder folder = new TemporaryFolder();
+
+  private Table hiveTable;
+  private TableIdentifier identifier;
+  private String db = "testUpgradeHiveDb";
+  private String table = "testUpgradeHiveTable";
+  private boolean isPartitioned;
+  private String[] partitionNames = {"name", HiveTableTestHelper.COLUMN_NAME_OP_DAY};
+  private String[] partitionValues = {"Bob", "2020-01-01"};
+
+  public TestUpgradeHiveTableUtil(
+      CatalogTestHelper catalogTestHelper, boolean isPartitioned) throws IOException {
+    super(catalogTestHelper);
+    folder.create();
+    this.isPartitioned = isPartitioned;
+  }
+
+  @Before
+  public void setUp() throws TException, IOException {
+    Database database = new Database();
+    database.setName(db);
+    TEST_HMS.getHiveClient().createDatabase(database);
+    PartitionSpec spec = isPartitioned ? PartitionSpec.builderFor(HiveTableTestHelper.HIVE_TABLE_SCHEMA)
+        .identity("name").identity(HiveTableTestHelper.COLUMN_NAME_OP_DAY).build() : PartitionSpec.unpartitioned();
+    hiveTable = createHiveTable(db, table, HiveTableTestHelper.HIVE_TABLE_SCHEMA, spec);
+    if (isPartitioned) {
+      createPartition();
+    }
+    identifier = TableIdentifier.of(getCatalog().name(), db, table);
+  }
+
+  @After
+  public void dropTable() throws TException {
+    getCatalog().dropTable(identifier, true);
+    TEST_HMS.getHiveClient().dropDatabase(db);
+  }
+
+  @Parameterized.Parameters(name = "{0}, {1}")
+  public static Object[] parameters() {
+    return new Object[][] {{new HiveCatalogTestHelper(TableFormat.MIXED_HIVE, TEST_HMS.getHiveConf()), true},
+                           {new HiveCatalogTestHelper(TableFormat.MIXED_HIVE, TEST_HMS.getHiveConf()), false}};
+  }
+
+  @Test
+  public void upgradeHiveTable() throws Exception {
+    UpgradeHiveTableUtil.upgradeHiveTable(
+        (ArcticHiveCatalog) getCatalog(),
+        identifier,
+        new ArrayList<>(),
+        new HashMap<>());
+    ArcticTable table = getCatalog().loadTable(identifier);
+    UnkeyedHiveTable baseTable = table.isKeyedTable() ?
+        (UnkeyedHiveTable) table.asKeyedTable().baseTable() :
+        (UnkeyedHiveTable) table.asUnkeyedTable();
+    if (table.spec().isPartitioned()) {
+      List<Partition> partitions =
+          HivePartitionUtil.getHiveAllPartitions(((ArcticHiveCatalog) getCatalog()).getHMSClient(), table.id());
+      for (Partition partition : partitions) {
+        StructLike partitionData = DataFiles.data(table.spec(), String.join("/", partition.getValues()));
+        Map<String, String> partitionProperties = baseTable.partitionProperty().get(partitionData);
+        Assert.assertTrue(partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+        Assert.assertTrue(partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME));
+        Assert.assertFalse(HiveMetaSynchronizer.partitionHasModified(baseTable, partition, partitionData));
+      }
+    } else {
+      Map<String, String> partitionProperties = baseTable.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT);
+      Assert.assertTrue(partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+      Assert.assertTrue(partitionProperties.containsKey(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME));
+      Assert.assertFalse(HiveMetaSynchronizer.tableHasModified(baseTable, hiveTable));
+    }
+  }
+
+  private Table createHiveTable(String db, String table, Schema schema, PartitionSpec spec) throws TException,
+      IOException {
+    Table hiveTable = newHiveTable(db, table, schema, spec);
+    hiveTable.setSd(HiveTableUtil.storageDescriptor(
+        schema,
+        spec,
+        folder.newFolder(db).getAbsolutePath(),
+        FileFormat.valueOf(TableProperties.DEFAULT_FILE_FORMAT_DEFAULT.toUpperCase(Locale.ENGLISH))));
+    TEST_HMS.getHiveClient().createTable(hiveTable);
+    return TEST_HMS.getHiveClient().getTable(db, table);
+  }
+
+  private void createPartition() throws TException {
+    List<String> partitions = Lists.newArrayList();
+    for (int i = 0; i < partitionNames.length; i++) {
+      partitions.add(String.join("=", partitionNames[i], partitionValues[i]));
+    }
+    Partition newPartition =
+        HivePartitionUtil.newPartition(hiveTable, partitions,
+            hiveTable.getSd().getLocation() + "/" + String.join("/", partitions), new ArrayList<>(),
+            (int) (System.currentTimeMillis() / 1000));
+    TEST_HMS.getHiveClient().add_partition(newPartition);
+  }
+
+  private org.apache.hadoop.hive.metastore.api.Table newHiveTable(
+      String db, String table, Schema schema, PartitionSpec partitionSpec) {
+    final long currentTimeMillis = System.currentTimeMillis();
+    org.apache.hadoop.hive.metastore.api.Table newTable = new org.apache.hadoop.hive.metastore.api.Table(
+        table,
+        db,
+        System.getProperty("user.name"),
+        (int) currentTimeMillis / 1000,
+        (int) currentTimeMillis / 1000,
+        Integer.MAX_VALUE,
+        null,
+        HiveSchemaUtil.hivePartitionFields(schema, partitionSpec),
+        new HashMap<>(),
+        null,
+        null,
+        TableType.EXTERNAL_TABLE.toString());
+
+    newTable.getParameters().put("EXTERNAL", "TRUE");
+    return newTable;
+  }
+}