Merge branch 'master' into rollbackfirstwhenautoclustering

Author: yuezhang

docker/demo/config/test-suite/compact-test.properties

@@ -0,0 +1,50 @@
+
+#  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.
+
+hoodie.insert.shuffle.parallelism=100
+hoodie.upsert.shuffle.parallelism=100
+hoodie.bulkinsert.shuffle.parallelism=100
+
+hoodie.deltastreamer.source.test.num_partitions=100
+hoodie.deltastreamer.source.test.datagen.use_rocksdb_for_storing_existing_keys=false
+hoodie.deltastreamer.source.test.max_unique_records=100000000
+hoodie.embed.timeline.server=false
+hoodie.deltastreamer.source.input.selector=org.apache.hudi.integ.testsuite.helpers.DFSTestSuitePathSelector
+
+hoodie.deltastreamer.source.input.selector=org.apache.hudi.integ.testsuite.helpers.DFSTestSuitePathSelector
+hoodie.datasource.hive_sync.skip_ro_suffix=true
+
+hoodie.datasource.write.recordkey.field=_row_key
+hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.TimestampBasedKeyGenerator
+hoodie.datasource.write.partitionpath.field=timestamp
+
+hoodie.compact.inline.max.delta.commits=2
+
+hoodie.deltastreamer.source.dfs.root=/user/hive/warehouse/hudi-integ-test-suite/input
+hoodie.deltastreamer.schemaprovider.target.schema.file=file:/var/hoodie/ws/docker/demo/config/test-suite/source.avsc
+hoodie.deltastreamer.schemaprovider.source.schema.file=file:/var/hoodie/ws/docker/demo/config/test-suite/source.avsc
+hoodie.deltastreamer.keygen.timebased.timestamp.type=UNIX_TIMESTAMP
+hoodie.deltastreamer.keygen.timebased.output.dateformat=yyyy/MM/dd
+
+hoodie.datasource.hive_sync.jdbcurl=jdbc:hive2://hiveserver:10000/
+hoodie.datasource.hive_sync.mode=jdbc
+hoodie.datasource.hive_sync.database=testdb
+hoodie.datasource.hive_sync.table=table1
+hoodie.datasource.hive_sync.assume_date_partitioning=false
+hoodie.datasource.hive_sync.partition_fields=_hoodie_partition_path
+hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor
+

docker/demo/config/test-suite/mor-async-compact.yaml

@@ -0,0 +1,126 @@
+# 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.
+
+# Use compact-test.properties for this yaml file.
+dag_name: mor-async-compact.yaml
+dag_rounds: 4
+dag_intermittent_delay_mins: 0
+dag_content:
+  first_insert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      repeat_count: 1
+      num_records_insert: 10000
+    type: InsertNode
+    deps: none
+  first_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: first_insert
+  second_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: first_upsert
+  third_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: second_upsert
+  first_validate:
+    config:
+      delete_input_data: false
+    type: ValidateDatasetNode
+    deps: third_upsert
+  first_schedule_compact:
+    config:
+    type: ScheduleCompactNode
+    deps: first_validate
+  fourth_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: first_schedule_compact
+  fifth_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: fourth_upsert
+  second_insert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      repeat_count: 1
+      num_records_insert: 10000
+    type: InsertNode
+    deps: fifth_upsert
+  sixth_upsert:
+    config:
+      record_size: 1000
+      num_partitions_insert: 1
+      num_records_insert: 300
+      repeat_count: 1
+      num_records_upsert: 2000
+      num_partitions_upsert: 1
+    type: UpsertNode
+    deps: second_insert
+  third_validate:
+    config:
+      delete_input_data: false
+    type: ValidateDatasetNode
+    deps: sixth_upsert
+  first_compact:
+    config:
+    type: CompactNode
+    deps: third_validate
+  first_delete:
+    config:
+      num_partitions_delete: 1
+      num_records_delete: 500
+    type: DeleteNode
+    deps: first_compact
+  fifth_validate:
+    config:
+      delete_input_data: false
+    type: ValidateDatasetNode
+    deps: first_delete

docker/demo/config/test-suite/test.properties

@@ -25,10 +25,6 @@ hoodie.deltastreamer.source.test.max_unique_records=100000000
 hoodie.embed.timeline.server=false
 hoodie.deltastreamer.source.input.selector=org.apache.hudi.integ.testsuite.helpers.DFSTestSuitePathSelector
 
-hoodie.insert.shuffle.parallelism=100
-hoodie.upsert.shuffle.parallelism=100
-hoodie.bulkinsert.shuffle.parallelism=100
-
 hoodie.deltastreamer.source.input.selector=org.apache.hudi.integ.testsuite.helpers.DFSTestSuitePathSelector
 hoodie.datasource.hive_sync.skip_ro_suffix=true
 

hudi-cli/src/main/java/org/apache/hudi/cli/commands/FileSystemViewCommand.java

@@ -119,7 +119,7 @@ public String showAllFileSlices(
 
   @CliCommand(value = "show fsview latest", help = "Show latest file-system view")
   public String showLatestFileSlices(
-      @CliOption(key = {"partitionPath"}, help = "A valid paritition path", mandatory = true) String partition,
+      @CliOption(key = {"partitionPath"}, help = "A valid partition path", mandatory = true) String partition,
       @CliOption(key = {"baseFileOnly"}, help = "Only display base file view",
           unspecifiedDefaultValue = "false") boolean baseFileOnly,
       @CliOption(key = {"maxInstant"}, help = "File-Slices upto this instant are displayed",

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java

@@ -148,7 +148,7 @@ public class HoodieHBaseIndexConfig extends HoodieConfig {
       .defaultValue(false)
       .withDocumentation("Only applies if index type is HBASE. "
           + "When an already existing record is upserted to a new partition compared to whats in storage, "
-          + "this config when set, will delete old record in old paritition "
+          + "this config when set, will delete old record in old partition "
           + "and will insert it as new record in new partition.");
 
   public static final ConfigProperty<Boolean> ROLLBACK_SYNC_ENABLE = ConfigProperty

…/hudi/io/storage/HoodieConcatHandle.java …g/apache/hudi/io/HoodieConcatHandle.javahudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieConcatHandle.java renamed to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieConcatHandle.java renamed to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java

@@ -16,15 +16,16 @@
  * limitations under the License.
  */
 
-package org.apache.hudi.io.storage;
+package org.apache.hudi.io;
 
 import org.apache.hudi.common.engine.TaskContextSupplier;
 import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.io.HoodieMergeHandle;
 import org.apache.hudi.keygen.BaseKeyGenerator;
 import org.apache.hudi.keygen.KeyGenUtils;
 import org.apache.hudi.table.HoodieTable;
@@ -34,6 +35,7 @@
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 
@@ -44,38 +46,44 @@
  * Simplified Logic:
  * For every existing record
  *     Write the record as is
- * For all incoming records, write to file as is.
+ * For all incoming records, write to file as is, without de-duplicating based on the record key.
  *
  * Illustration with simple data.
  * Incoming data:
- *     rec1_2, rec4_2, rec5_1, rec6_1
+ *     rec1_2, rec1_3, rec4_2, rec5_1, rec6_1
  * Existing data:
  *     rec1_1, rec2_1, rec3_1, rec4_1
  *
  * For every existing record, write to storage as is.
  *    => rec1_1, rec2_1, rec3_1 and rec4_1 is written to storage
  * Write all records from incoming set to storage
- *    => rec1_2, rec4_2, rec5_1 and rec6_1
+ *    => rec1_2, rec1_3, rec4_2, rec5_1 and rec6_1
  *
  * Final snapshot in storage
- * rec1_1, rec2_1, rec3_1, rec4_1, rec1_2, rec4_2, rec5_1, rec6_1
+ * rec1_1, rec2_1, rec3_1, rec4_1, rec1_2, rec1_3, rec4_2, rec5_1, rec6_1
  *
  * Users should ensure there are no duplicates when "insert" operation is used and if the respective config is enabled. So, above scenario should not
  * happen and every batch should have new records to be inserted. Above example is for illustration purposes only.
  */
 public class HoodieConcatHandle<T extends HoodieRecordPayload, I, K, O> extends HoodieMergeHandle<T, I, K, O> {
 
   private static final Logger LOG = LogManager.getLogger(HoodieConcatHandle.class);
+  // a representation of incoming records that tolerates duplicate keys
+  private final Iterator<HoodieRecord<T>> recordItr;
 
-  public HoodieConcatHandle(HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable, Iterator recordItr,
-                            String partitionPath, String fileId, TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
-    super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
+  public HoodieConcatHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+                            Iterator<HoodieRecord<T>> recordItr, String partitionPath, String fileId,
+                            TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
+    super(config, instantTime, hoodieTable, Collections.emptyIterator(), partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
+    this.recordItr = recordItr;
   }
 
-  public HoodieConcatHandle(HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable, Map keyToNewRecords, String partitionPath, String fileId,
-      HoodieBaseFile dataFileToBeMerged, TaskContextSupplier taskContextSupplier) {
-    super(config, instantTime, hoodieTable, keyToNewRecords, partitionPath, fileId, dataFileToBeMerged, taskContextSupplier,
+  public HoodieConcatHandle(HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable,
+                            Map<String, HoodieRecord<T>> keyToNewRecords, String partitionPath, String fileId,
+                            HoodieBaseFile dataFileToBeMerged, TaskContextSupplier taskContextSupplier) {
+    super(config, instantTime, hoodieTable, Collections.emptyMap(), partitionPath, fileId, dataFileToBeMerged, taskContextSupplier,
         Option.empty());
+    this.recordItr = keyToNewRecords.values().iterator();
   }
 
   /**
@@ -94,4 +102,17 @@ public void write(GenericRecord oldRecord) {
     }
     recordsWritten++;
   }
+
+  @Override
+  protected void writeIncomingRecords() throws IOException {
+    while (recordItr.hasNext()) {
+      HoodieRecord<T> record = recordItr.next();
+      if (needsUpdateLocation()) {
+        record.unseal();
+        record.setNewLocation(new HoodieRecordLocation(instantTime, fileId));
+        record.seal();
+      }
+      writeInsertRecord(record);
+    }
+  }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java

@@ -257,6 +257,18 @@ private boolean writeUpdateRecord(HoodieRecord<T> hoodieRecord, Option<IndexedRe
     return writeRecord(hoodieRecord, indexedRecord);
   }
 
+  protected void writeInsertRecord(HoodieRecord<T> hoodieRecord) throws IOException {
+    Schema schema = useWriterSchema ? tableSchemaWithMetaFields : tableSchema;
+    Option<IndexedRecord> insertRecord = hoodieRecord.getData().getInsertValue(schema, config.getProps());
+    // just skip the ignored record
+    if (insertRecord.isPresent() && insertRecord.get().equals(IGNORE_RECORD)) {
+      return;
+    }
+    if (writeRecord(hoodieRecord, insertRecord)) {
+      insertRecordsWritten++;
+    }
+  }
+
   protected boolean writeRecord(HoodieRecord<T> hoodieRecord, Option<IndexedRecord> indexedRecord) {
     Option recordMetadata = hoodieRecord.getData().getMetadata();
     if (!partitionPath.equals(hoodieRecord.getPartitionPath())) {
@@ -340,28 +352,28 @@ public void write(GenericRecord oldRecord) {
     }
   }
 
+  protected void writeIncomingRecords() throws IOException {
+    // write out any pending records (this can happen when inserts are turned into updates)
+    Iterator<HoodieRecord<T>> newRecordsItr = (keyToNewRecords instanceof ExternalSpillableMap)
+        ? ((ExternalSpillableMap)keyToNewRecords).iterator() : keyToNewRecords.values().iterator();
+    while (newRecordsItr.hasNext()) {
+      HoodieRecord<T> hoodieRecord = newRecordsItr.next();
+      if (!writtenRecordKeys.contains(hoodieRecord.getRecordKey())) {
+        writeInsertRecord(hoodieRecord);
+      }
+    }
+  }
+
   @Override
   public List<WriteStatus> close() {
     try {
-      // write out any pending records (this can happen when inserts are turned into updates)
-      Iterator<HoodieRecord<T>> newRecordsItr = (keyToNewRecords instanceof ExternalSpillableMap)
-          ? ((ExternalSpillableMap)keyToNewRecords).iterator() : keyToNewRecords.values().iterator();
-      while (newRecordsItr.hasNext()) {
-        HoodieRecord<T> hoodieRecord = newRecordsItr.next();
-        if (!writtenRecordKeys.contains(hoodieRecord.getRecordKey())) {
-          Schema schema = useWriterSchema ? tableSchemaWithMetaFields : tableSchema;
-          Option<IndexedRecord> insertRecord =
-              hoodieRecord.getData().getInsertValue(schema, config.getProps());
-          // just skip the ignore record
-          if (insertRecord.isPresent() && insertRecord.get().equals(IGNORE_RECORD)) {
-            continue;
-          }
-          writeRecord(hoodieRecord, insertRecord);
-          insertRecordsWritten++;
-        }
-      }
+      writeIncomingRecords();
 
-      ((ExternalSpillableMap) keyToNewRecords).close();
+      if (keyToNewRecords instanceof ExternalSpillableMap) {
+        ((ExternalSpillableMap) keyToNewRecords).close();
+      } else {
+        keyToNewRecords.clear();
+      }
       writtenRecordKeys.clear();
 
       if (fileWriter != null) {