[ARCTIC-1167][core][hive] fix Trash for restoring deleted files

core/src/main/java/com/netease/arctic/io/ArcticFileIO.java

@@ -18,6 +18,7 @@
 
 package com.netease.arctic.io;
 
+import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.iceberg.io.FileIO;
 
@@ -27,7 +28,7 @@
 /**
  * Arctic extension from {@link FileIO}, adding more operations.
  */
-public interface ArcticFileIO extends FileIO {
+public interface ArcticFileIO extends FileIO, Configurable {
 
   /**
    * Run the given action with login user.

core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java

@@ -31,8 +31,6 @@
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -43,7 +41,6 @@
  * Implementation of {@link ArcticFileIO} for hadoop file system with authentication.
  */
 public class ArcticHadoopFileIO extends HadoopFileIO implements ArcticFileIO {
-  private static final Logger LOG = LoggerFactory.getLogger(ArcticHadoopFileIO.class);
 
   private final TableMetaStore tableMetaStore;
 

core/src/main/java/com/netease/arctic/io/BasicTableTrashManager.java

@@ -101,6 +101,9 @@ public void moveFileToTrash(String path) {
       if (!arcticFileIO.exists(targetFileDir)) {
         arcticFileIO.mkdirs(targetFileDir);
       }
+      if (arcticFileIO.exists(targetFileLocation)) {
+        arcticFileIO.deleteFile(targetFileLocation);
+      }
       arcticFileIO.rename(path, targetFileLocation);
     } catch (Exception e) {
       LOG.error("{} failed to move file to trash, {}", tableIdentifier, path, e);

core/src/main/java/com/netease/arctic/io/RecoverableArcticFileIO.java

@@ -1,5 +1,6 @@
 package com.netease.arctic.io;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
@@ -144,4 +145,14 @@ private void moveToTrash(String filePath) {
     trashManager.moveFileToTrash(filePath);
     LOG.debug("Move file:{} to table trash", filePath);
   }
+
+  @Override
+  public void setConf(Configuration conf) {
+    fileIO.setConf(conf);
+  }
+
+  @Override
+  public Configuration getConf() {
+    return fileIO.getConf();
+  }
 }

core/src/main/java/com/netease/arctic/io/TableTrashManager.java

@@ -20,12 +20,13 @@
 
 import com.netease.arctic.table.TableIdentifier;
 
+import java.io.Serializable;
 import java.time.LocalDate;
 
 /**
  * Trash Manager for a table.
  */
-public interface TableTrashManager {
+public interface TableTrashManager extends Serializable {
   /**
    * Table identifier.
    * A TableTrashManager only handle files in this table's location.

core/src/test/java/com/netease/arctic/io/BasicTableTrashManagerTest.java

@@ -79,6 +79,20 @@ public void testDeleteAndRestore() {
     Assert.assertFalse(getArcticTable().io().exists(fileLocationInTrash));
   }
 
+  @Test
+  public void testMoveAndOverwrite() {
+    TableTrashManager tableTrashManager = TableTrashManagers.build(getArcticTable());
+
+    String relativeFilePath = "base/test/test1.parquet";
+    String path = createFile(getArcticTable().io(), getArcticTable().location() + File.separator + relativeFilePath);
+
+    tableTrashManager.moveFileToTrash(path);
+    Assert.assertTrue(tableTrashManager.fileExistInTrash(path));
+    createFile(getArcticTable().io(), getArcticTable().location() + File.separator + relativeFilePath);
+    tableTrashManager.moveFileToTrash(path);
+    Assert.assertTrue(tableTrashManager.fileExistInTrash(path));
+  }
+
   @Test
   public void testDeleteDirectory() {
     TableTrashManager tableTrashManager = TableTrashManagers.build(getArcticTable());

hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java

@@ -21,7 +21,6 @@
 import com.netease.arctic.hive.HMSClientPool;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.op.OverwriteHiveFiles;
-import com.netease.arctic.io.ArcticHadoopFileIO;
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
@@ -219,11 +218,11 @@ public static void syncHiveDataToArctic(ArcticTable table, HMSClientPool hiveCli
   }
 
   private static List<DataFile> listHivePartitionFiles(ArcticTable arcticTable, Map<String, String> partitionValueMap,
-      String partitionLocation) {
+                                                       String partitionLocation) {
     return arcticTable.io().doAs(() -> TableMigrationUtil.listPartition(partitionValueMap, partitionLocation,
         arcticTable.properties().getOrDefault(TableProperties.DEFAULT_FILE_FORMAT,
             TableProperties.DEFAULT_FILE_FORMAT_DEFAULT),
-        arcticTable.spec(), ((ArcticHadoopFileIO)arcticTable.io()).getTableMetaStore().getConfiguration(),
+        arcticTable.spec(), arcticTable.io().getConf(),
         MetricsConfig.fromProperties(arcticTable.properties()), NameMappingParser.fromJson(
             arcticTable.properties().get(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING))));
   }