[HUDI-3581] Reorganize some clazz for hudi flink

Author: danny0405

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/HoodieAsyncService.java

@@ -170,6 +170,7 @@ private void shutdownCallback(Function<Boolean, Boolean> callback) {
       if (null != callback) {
         callback.apply(null != error);
       }
+      this.started = false;
     });
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java

@@ -835,11 +835,18 @@ public void archive() {
    * Provides a new commit time for a write operation (insert/update/delete).
    */
   public String startCommit() {
+    HoodieTableMetaClient metaClient = createMetaClient(true);
+    return startCommit(metaClient.getCommitActionType(), metaClient);
+  }
+
+  /**
+   * Provides a new commit time for a write operation (insert/update/delete/insert_overwrite/insert_overwrite_table) with specified action.
+   */
+  public String startCommit(String actionType, HoodieTableMetaClient metaClient) {
     CleanerUtils.rollbackFailedWrites(config.getFailedWritesCleanPolicy(),
         HoodieTimeline.COMMIT_ACTION, () -> rollbackFailedWrites());
     String instantTime = HoodieActiveTimeline.createNewInstantTime();
-    HoodieTableMetaClient metaClient = createMetaClient(true);
-    startCommit(instantTime, metaClient.getCommitActionType(), metaClient);
+    startCommit(instantTime, actionType, metaClient);
     return instantTime;
   }
 

hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java

@@ -364,7 +364,7 @@ private void addEventToBuffer(WriteMetadataEvent event) {
   private void startInstant() {
     // put the assignment in front of metadata generation,
     // because the instant request from write task is asynchronous.
-    this.instant = this.writeClient.startCommit();
+    this.instant = this.writeClient.startCommit(tableState.commitAction, this.metaClient);
     this.metaClient.getActiveTimeline().transitionRequestedToInflight(tableState.commitAction, this.instant);
     this.ckpMetadata.startInstant(this.instant);
     LOG.info("Create instant [{}] for table [{}] with type [{}]", this.instant,

hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.table.format.cow;
 
 import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
 
 import org.apache.flink.api.common.io.FileInputFormat;
 import org.apache.flink.api.common.io.FilePathFilter;

hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java

@@ -22,8 +22,12 @@
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
 import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
 import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector;
+import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
 import org.apache.hudi.table.format.cow.vector.reader.ArrayColumnReader;
+import org.apache.hudi.table.format.cow.vector.reader.FixedLenBytesColumnReader;
+import org.apache.hudi.table.format.cow.vector.reader.Int64TimestampColumnReader;
 import org.apache.hudi.table.format.cow.vector.reader.MapColumnReader;
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
 import org.apache.hudi.table.format.cow.vector.reader.RowColumnReader;
 
 import org.apache.flink.core.fs.Path;

…ble/format/cow/ParquetDecimalVector.java …mat/cow/vector/ParquetDecimalVector.javahudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetDecimalVector.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetDecimalVector.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.table.format.cow;
+package org.apache.hudi.table.format.cow.vector;
 
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.vector.BytesColumnVector;

…ble/format/cow/AbstractColumnReader.java …/vector/reader/AbstractColumnReader.javahudi-flink/src/main/java/org/apache/hudi/table/format/cow/AbstractColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/AbstractColumnReader.java hudi-flink/src/main/java/org/apache/hudi/table/format/cow/AbstractColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/AbstractColumnReader.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.table.format.cow;
+package org.apache.hudi.table.format.cow.vector.reader;
 
 import org.apache.flink.formats.parquet.vector.ParquetDictionary;
 import org.apache.flink.formats.parquet.vector.reader.ColumnReader;

hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayColumnReader.java

@@ -18,8 +18,8 @@
 
 package org.apache.hudi.table.format.cow.vector.reader;
 
-import org.apache.hudi.table.format.cow.ParquetDecimalVector;
 import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
+import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
 
 import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
 import org.apache.flink.table.data.TimestampData;

…ormat/cow/FixedLenBytesColumnReader.java …or/reader/FixedLenBytesColumnReader.javahudi-flink/src/main/java/org/apache/hudi/table/format/cow/FixedLenBytesColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/FixedLenBytesColumnReader.java hudi-flink/src/main/java/org/apache/hudi/table/format/cow/FixedLenBytesColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/FixedLenBytesColumnReader.java

@@ -15,12 +15,11 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.table.format.cow;
+package org.apache.hudi.table.format.cow.vector.reader;
 
 import org.apache.flink.table.data.vector.writable.WritableBytesVector;
 import org.apache.flink.table.data.vector.writable.WritableColumnVector;
 import org.apache.flink.table.data.vector.writable.WritableIntVector;
-
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.column.page.PageReader;
 import org.apache.parquet.io.api.Binary;

…rmat/cow/Int64TimestampColumnReader.java …r/reader/Int64TimestampColumnReader.javahudi-flink/src/main/java/org/apache/hudi/table/format/cow/Int64TimestampColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/Int64TimestampColumnReader.java hudi-flink/src/main/java/org/apache/hudi/table/format/cow/Int64TimestampColumnReader.java renamed to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/vector/reader/Int64TimestampColumnReader.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.table.format.cow;
+package org.apache.hudi.table.format.cow.vector.reader;
 
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.data.vector.writable.WritableIntVector;