[WIP]Feature optimize support hive

ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift

@@ -63,7 +63,8 @@ enum JobType {
 
 enum OptimizeType {
     Minor,
-    Major
+    Major,
+    FullMajor
 }
 
 

ams/ams-server/pom.xml

@@ -74,24 +74,6 @@
                 </exclusion>
             </exclusions>
         </dependency>
-<!--        <dependency>-->
-<!--            <groupId>com.netease.arctic</groupId>-->
-<!--            <artifactId>arctic-hive</artifactId>-->
-<!--            <exclusions>-->
-<!--                <exclusion>-->
-<!--                    <groupId>org.apache.hadoop</groupId>-->
-<!--                    <artifactId>hadoop-client</artifactId>-->
-<!--                </exclusion>-->
-<!--                <exclusion>-->
-<!--                    <groupId>org.apache.hadoop</groupId>-->
-<!--                    <artifactId>hadoop-yarn-api</artifactId>-->
-<!--                </exclusion>-->
-<!--                <exclusion>-->
-<!--                    <artifactId>libthrift</artifactId>-->
-<!--                    <groupId>org.apache.thrift</groupId>-->
-<!--                </exclusion>-->
-<!--            </exclusions>-->
-<!--        </dependency>-->
 
         <dependency>
             <groupId>com.netease.arctic</groupId>
@@ -104,6 +86,13 @@
                 </exclusion>
             </exclusions>
         </dependency>
+
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_2.12</artifactId>
@@ -227,6 +216,14 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.netease.arctic</groupId>
             <artifactId>arctic-ams-api</artifactId>

ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java

@@ -174,6 +174,7 @@ private static void startMetaStoreThreads(
         startOptimizeCommit(conf.getInteger(ArcticMetaStoreConf.OPTIMIZE_COMMIT_THREAD_POOL_SIZE));
         startExpiredClean();
         startOrphanClean();
+        startSupportHiveSync();
         monitorOptimizerStatus();
         tableRuntimeDataExpire();
         AmsRestServer.startRestServer(httpPort);
@@ -245,6 +246,14 @@ private static void startOrphanClean() {
         TimeUnit.MILLISECONDS);
   }
 
+  private static void startSupportHiveSync() {
+    ThreadPool.getPool(ThreadPool.Type.HIVE_SYNC).scheduleWithFixedDelay(
+        ServiceContainer.getSupportHiveSyncService()::checkHiveSyncTasks,
+        3 * 1000L,
+        60 * 1000L,
+        TimeUnit.MILLISECONDS);
+  }
+
   private static void monitorOptimizerStatus() {
     OptimizeExecuteService.OptimizerMonitor monitor = new OptimizeExecuteService.OptimizerMonitor();
     ThreadPool.getPool(ThreadPool.Type.OPTIMIZER_MONITOR).scheduleWithFixedDelay(

ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ArcticMetaStoreConf.java

@@ -80,6 +80,12 @@ public class ArcticMetaStoreConf {
           .defaultValue(10)
           .withDescription("Number of threads in the thread pool.  " +
               "These will be used to execute all orphan file clean processes.");
+  public static final ConfigOption<Integer> SUPPORT_HIVE_SYNC_THREAD_POOL_SIZE =
+      ConfigOptions.key("arctic.ams.support.hive.sync.thread.pool-size")
+          .intType()
+          .defaultValue(10)
+          .withDescription("Number of threads in the thread pool.  " +
+              "These will be used to execute all support hive sync processes.");
   public static final ConfigOption<Integer> SYNC_FILE_INFO_CACHE_THREAD_POOL_SIZE =
       ConfigOptions.key("arctic.ams.file.sync.thread.pool-size")
           .intType()

ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizeTasksMapper.java

@@ -37,7 +37,7 @@ public interface OptimizeTasksMapper {
 
   @Select("select trace_id, optimize_type, catalog_name, db_name, table_name, `partition`," +
       " task_group, task_history_id, max_change_transaction_id, is_delete_pos_delete," +
-      " source_nodes, create_time, properties, queue_id, " +
+      " source_nodes, create_time, properties, queue_id," +
       " insert_file_size, delete_file_size, base_file_size, pos_delete_file_size," +
       " insert_files, delete_files, base_files, pos_delete_files" +
       " from " + TABLE_NAME)

ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTask.java

@@ -41,6 +41,7 @@ public class BaseOptimizeTask extends OptimizeTask {
   protected long createTime;
 
   private long maxChangeTransactionId = INVALID_TRANSACTION_ID;
+  @Deprecated
   private int isDeletePosDelete;
 
   public BaseOptimizeTask() {
@@ -70,7 +71,6 @@ public void setMaxChangeTransactionId(long maxChangeTransactionId) {
     this.maxChangeTransactionId = maxChangeTransactionId;
   }
 
-
   public String getPartition() {
     return partition;
   }
@@ -143,10 +143,12 @@ public void setDeleteFileCnt(int deleteFileCnt) {
     this.deleteFileCnt = deleteFileCnt;
   }
 
+  @Deprecated
   public int getIsDeletePosDelete() {
     return isDeletePosDelete;
   }
 
+  @Deprecated
   public void setIsDeletePosDelete(int isDeletePosDelete) {
     this.isDeletePosDelete = isDeletePosDelete;
   }

ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/FileTree.java

@@ -152,20 +152,20 @@ public void collectDeleteFiles(List<DataFile> collector) {
   }
 
   public void collectBaseFiles(List<DataFile> collector) {
-    collectBaseFiles(collector, false, 0);
+    collectBaseFiles(collector, false, Collections.emptyList());
   }
 
-  public void collectBaseFiles(List<DataFile> collector, boolean isFilterSmallFile, long smallFileSize) {
-    if (isFilterSmallFile) {
+  public void collectBaseFiles(List<DataFile> collector, boolean isMajor, List<DataFile> needOptimizeFiles) {
+    if (isMajor) {
       baseFiles = baseFiles.stream()
-          .filter(dataFile -> dataFile.fileSizeInBytes() < smallFileSize).collect(Collectors.toList());
+          .filter(needOptimizeFiles::contains).collect(Collectors.toList());
     }
     collector.addAll(baseFiles);
     if (left != null) {
-      left.collectBaseFiles(collector, isFilterSmallFile, smallFileSize);
+      left.collectBaseFiles(collector, isMajor, needOptimizeFiles);
     }
     if (right != null) {
-      right.collectBaseFiles(collector, isFilterSmallFile, smallFileSize);
+      right.collectBaseFiles(collector, isMajor, needOptimizeFiles);
     }
   }
 

ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TableOptimizeRuntime.java

@@ -36,7 +36,9 @@ public class TableOptimizeRuntime {
   private long currentChangeSnapshotId = INVALID_SNAPSHOT_ID;
   private TableOptimizeInfo.OptimizeStatus optimizeStatus = TableOptimizeInfo.OptimizeStatus.Idle;
   private long optimizeStatusStartTime = -1;
+
   private final Map<String, Long> latestMajorOptimizeTime = new HashMap<>();
+  private final Map<String, Long> latestFullMinorOptimizeTime = new HashMap<>();
   private final Map<String, Long> latestMinorOptimizeTime = new HashMap<>();
   private String latestTaskHistoryId;
   private volatile boolean isRunning;
@@ -77,6 +79,15 @@ public void putLatestMajorOptimizeTime(String partition, long time) {
     }
   }
 
+  public void putLatestFullMajorOptimizeTime(String partition, long time) {
+    Long oldValue = latestFullMinorOptimizeTime.putIfAbsent(partition, time);
+    if (oldValue != null) {
+      if (time > oldValue) {
+        latestFullMinorOptimizeTime.put(partition, time);
+      }
+    }
+  }
+
   public TableOptimizeInfo.OptimizeStatus getOptimizeStatus() {
     return optimizeStatus;
   }
@@ -99,6 +110,11 @@ public long getLatestMajorOptimizeTime(String partition) {
     return time == null ? -1 : time;
   }
 
+  public long getLatestFullMajorOptimizeTime(String partition) {
+    Long time = latestFullMinorOptimizeTime.get(partition);
+    return time == null ? -1 : time;
+  }
+
   public void putLatestMinorOptimizeTime(String partition, long time) {
     Long oldValue = latestMinorOptimizeTime.putIfAbsent(partition, time);
     if (oldValue != null) {
@@ -118,6 +134,9 @@ public Set<String> getPartitions() {
     if (MapUtils.isNotEmpty(latestMajorOptimizeTime)) {
       result.addAll(latestMajorOptimizeTime.keySet());
     }
+    if (MapUtils.isNotEmpty(latestFullMinorOptimizeTime)) {
+      result.addAll(latestFullMinorOptimizeTime.keySet());
+    }
     if (MapUtils.isNotEmpty(latestMinorOptimizeTime)) {
       result.addAll(latestMinorOptimizeTime.keySet());
     }
@@ -158,6 +177,7 @@ public String toString() {
         ", optimizeStatus=" + optimizeStatus +
         ", optimizeStatusStartTime=" + optimizeStatusStartTime +
         ", latestMajorOptimizeTime=" + latestMajorOptimizeTime +
+        ", latestFullMinorOptimizeTime=" + latestFullMinorOptimizeTime +
         ", latestMinorOptimizeTime=" + latestMinorOptimizeTime +
         ", latestTaskHistoryId='" + latestTaskHistoryId + '\'' +
         ", isRunning=" + isRunning +

ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TaskConfig.java

@@ -31,7 +31,8 @@ public class TaskConfig {
   private final String historyId;
   private final long createTime;
 
-  public TaskConfig(String partition, @Nullable Long maxTransactionId, String group, String historyId,
+  public TaskConfig(String partition, @Nullable Long maxTransactionId,
+                    String group, String historyId,
                     OptimizeType optimizeType, long createTime) {
     this.optimizeType = optimizeType;
     this.partition = partition;
@@ -49,6 +50,7 @@ public String getPartition() {
     return partition;
   }
 
+  @org.jetbrains.annotations.Nullable
   public Long getMaxTransactionId() {
     return maxTransactionId;
   }