This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 3f744eafb130 feat(core): Pass the number of updates from workload
profiling through to merge handle (#18396)
3f744eafb130 is described below
commit 3f744eafb13088b09e08e013c01f61010b8898f5
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Aug 17 03:48:47 2026 -0700
feat(core): Pass the number of updates from workload profiling through to
merge handle (#18396)
* feat(core): Pass the number of updates from workload profiling through to
merge handle
* Add more test coverage
* fixup: adapt LSM merge-handle classes and raw-type test casts to
MergeContext
* Re-trigger CI
* review(18396): unify the update-count naming and finish the java-side
propagation
- MergeContext: add UNKNOWN_NUM_UPDATES, rename numIncomingUpdates to
numUpdates and
getRecordItr to getRecordIterator, and document that the count is a lower
bound of the
iterator size (small-file inserts flow through the iterator uncounted)
- JavaUpsertPartitioner: pass the profiled update count into UPDATE buckets
and 0 for
insert-only small-file buckets, mirroring the Spark partitioner
- BucketInfo: document the population scope (upsert partitioners only) and
why numUpdates
stays out of equals/hashCode
- clarify the empty-iterator MergeContext on the concat and compaction
constructors
* review(18396): dedupe the bucket-info tests, cover the factory fallback
and the e2e count propagation
- fold the BucketInfo numUpdates assertions into the existing
TestBucketTypeAndInfo and drop
TestBucketInfo plus the copy parked in TestUpsertPartitioner
- replace the tautological TestHoodieMergeHandleConstruction with
factory-instantiation tests:
custom handle via the new MergeContext signature, silent fallback for a
legacy
Iterator-constructor handle, and a loud failure with the fallback disabled
- add an end-to-end test that a real upsert carries the profiled update
count into the merge
handle through the partitioner, bucket info, and factory reflection
- make the concat-handle test discriminating: incoming records must
actually be written
- collapse the near-duplicate MergeContext tests into a parameterized one
* review(18396): expose getNumUpdates on the public interface and cover the
java-side propagation
- hoist getNumUpdates() onto the HoodieMergeHandle interface so custom
handles can read the
count without downcasting; document that it is a lower bound
- mirror the e2e propagation test on the java client, which had no coverage
for the
partitioner change
- pin that insert-overwrite drops the profiled count when rebuilding
buckets as INSERT
- document the 0-for-insert-only convention on BucketInfo
- test hygiene: MockitoExtension in the factory test, javadoc casing, note
the existing e2e
concat coverage
* review(18396): address the follow-up nits
- point the numUpdates comment at the public getter instead of the private
field
- comment the known-zero convention at the INSERT-bucket sites
* review(18396): replace the raw iterator casts with a shared typed helper
The record lists come out of the test data generators as raw
List<HoodieRecord>, so the cast
cannot be avoided upstream; centralize it in
HoodieClientTestUtils.toTypedRecordIterator with
a scoped suppression instead of repeating the raw cast at every call site.
---------
Co-authored-by: Lokesh Jain <[email protected]>
Co-authored-by: voon <[email protected]>
---
.../hudi/io/FileGroupReaderBasedMergeHandle.java | 14 +-
.../apache/hudi/io/HoodieAbstractMergeHandle.java | 10 +-
.../org/apache/hudi/io/HoodieConcatHandle.java | 9 +-
.../java/org/apache/hudi/io/HoodieMergeHandle.java | 9 +
.../apache/hudi/io/HoodieMergeHandleFactory.java | 7 +-
.../hudi/io/HoodieMergeHandleWithChangeLog.java | 5 +-
.../apache/hudi/io/HoodieSortedMergeHandle.java | 7 +-
.../io/HoodieSortedMergeHandleWithChangeLog.java | 5 +-
.../org/apache/hudi/io/HoodieWriteMergeHandle.java | 16 +-
.../io/LsmFileGroupReaderBasedMergeHandle.java | 8 +-
.../main/java/org/apache/hudi/io/MergeContext.java | 71 +++++++
.../action/commit/BaseCommitActionExecutor.java | 1 +
.../hudi/table/action/commit/BucketInfo.java | 15 +-
.../TestHoodieMergeHandleFactoryInstantiation.java | 212 +++++++++++++++++++++
.../java/org/apache/hudi/io/TestMergeContext.java | 51 +++++
.../table/action/commit/TestBucketTypeAndInfo.java | 21 +-
.../io/FlinkFileGroupReaderBasedMergeHandle.java | 2 +-
.../FlinkLsmFileGroupReaderBasedMergeHandle.java | 2 +-
.../java/org/apache/hudi/io/FlinkMergeHandle.java | 2 +-
.../commit/BaseFlinkCommitActionExecutor.java | 6 +-
.../delta/BaseFlinkDeltaCommitActionExecutor.java | 2 +-
.../FlinkUpsertDeltaCommitActionExecutor.java | 2 +-
.../commit/BaseJavaCommitActionExecutor.java | 13 +-
.../table/action/commit/JavaUpsertPartitioner.java | 12 +-
.../BaseJavaDeltaCommitActionExecutor.java | 6 +-
.../commit/TestJavaCopyOnWriteActionExecutor.java | 60 +++++-
.../SparkBootstrapCommitActionExecutor.java | 2 +-
.../commit/BaseSparkCommitActionExecutor.java | 15 +-
.../table/action/commit/UpsertPartitioner.java | 12 +-
.../BaseSparkDeltaCommitActionExecutor.java | 5 +-
.../hudi/client/TestUpdateSchemaEvolution.java | 6 +-
.../commit/TestCopyOnWriteActionExecutor.java | 59 +++++-
.../table/action/commit/TestUpsertPartitioner.java | 25 ++-
.../hudi/testutils/HoodieClientTestUtils.java | 10 +
.../java/org/apache/hudi/io/TestMergeHandle.java | 73 ++++++-
.../hudi/table/TestHoodieMergeOnReadTable.java | 3 +-
36 files changed, 694 insertions(+), 84 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
index 460ef5d2b4fc..84fe098ff4cc 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
@@ -109,16 +109,16 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O>
extends HoodieWriteMerg
* @param config instance of {@link HoodieWriteConfig} to use.
* @param instantTime instant time of the current commit.
* @param hoodieTable instance of {@link HoodieTable} being updated.
- * @param recordItr iterator of records to be merged with the file.
+ * @param mergeContext context carrying incoming data to merge and its
characteristics.
* @param partitionPath partition path of the base file.
* @param fileId file ID of the base file.
* @param taskContextSupplier instance of {@link TaskContextSupplier} to use.
* @param keyGeneratorOpt optional instance of {@link BaseKeyGenerator} to
use for extracting keys from records.
*/
public FileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr,
String partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
- this(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, getLatestBaseFile(hoodieTable, partitionPath, fileId),
keyGeneratorOpt);
+ this(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, getLatestBaseFile(hoodieTable, partitionPath,
fileId), keyGeneratorOpt);
}
/**
@@ -128,7 +128,7 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O>
extends HoodieWriteMerg
* @param config instance of {@link HoodieWriteConfig} to use.
* @param instantTime instant time of the current commit.
* @param hoodieTable instance of {@link HoodieTable} being updated.
- * @param recordItr iterator of records to be merged with the file.
+ * @param mergeContext context carrying incoming data to merge and its
characteristics.
* @param partitionPath partition path of the base file.
* @param fileId file ID of the base file.
* @param taskContextSupplier instance of {@link TaskContextSupplier} to use.
@@ -136,16 +136,16 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O>
extends HoodieWriteMerg
* @param keyGeneratorOpt optional instance of {@link BaseKeyGenerator} to
use for extracting keys from records.
*/
public FileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr,
String partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, HoodieBaseFile baseFile, Option<BaseKeyGenerator>
keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, baseFile, keyGeneratorOpt);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, baseFile, keyGeneratorOpt);
this.compactionOperation = Option.empty();
TypedProperties properties = config.getProps();
properties.putAll(hoodieTable.getMetaClient().getTableConfig().getProps());
this.maxInstantTime = instantTime;
initRecordType(hoodieTable.getConfig().getRecordMerger().getRecordType());
this.props = TypedProperties.copy(config.getProps());
- this.incomingRecordsItr = recordItr;
+ this.incomingRecordsItr = mergeContext.getRecordIterator();
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
index 614a9c77e0bd..54db1467a49f 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAbstractMergeHandle.java
@@ -46,6 +46,12 @@ import java.util.NoSuchElementException;
@Slf4j
public abstract class HoodieAbstractMergeHandle<T, I, K, O> extends
HoodieWriteHandle<T, I, K, O> implements HoodieMergeHandle<T, I, K, O> {
+ // The number of incoming update and delete records based on tagging; see the
+ // MergeContext#getNumUpdates() javadoc for the exact semantics (lower
bound, not iterator size).
+ // MergeContext.UNKNOWN_NUM_UPDATES on code paths without workload
profiling, e.g.,
+ // compaction and metadata table writes.
+ @Getter
+ protected long numUpdates = MergeContext.UNKNOWN_NUM_UPDATES;
protected Map<String, HoodieRecord<T>> keyToNewRecords;
protected StoragePath newFilePath;
@Getter
@@ -64,6 +70,7 @@ public abstract class HoodieAbstractMergeHandle<T, I, K, O>
extends HoodieWriteH
* @param config Hoodie writer configs.
* @param instantTime current instant time.
* @param hoodieTable an instance of {@link HoodieTable}
+ * @param mergeContext Context carrying incoming data to merge and its
characteristics.
* @param partitionPath Partition path of the upsert and insert records.
* @param fileId New file id of the target base file.
* @param taskContextSupplier Base task context supplier
@@ -71,9 +78,10 @@ public abstract class HoodieAbstractMergeHandle<T, I, K, O>
extends HoodieWriteH
* @param keyGeneratorOpt Optional instance of the {@link
org.apache.hudi.keygen.KeyGenerator} used.
*/
public HoodieAbstractMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId,
TaskContextSupplier taskContextSupplier,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId, TaskContextSupplier taskContextSupplier,
HoodieBaseFile baseFile,
Option<BaseKeyGenerator> keyGeneratorOpt, boolean preserveMetadata) {
super(config, instantTime, partitionPath, fileId, hoodieTable,
taskContextSupplier, preserveMetadata);
+ this.numUpdates = mergeContext.getNumUpdates();
this.baseFileToMerge = baseFile;
this.keyGeneratorOpt = keyGeneratorOpt;
initPartitionMetadataAndFilePaths(partitionPath);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
index 0e918b53cb72..a7a14cc4936e 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
@@ -72,10 +72,13 @@ public class HoodieConcatHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I,
private final Iterator<HoodieRecord<T>> recordItr;
public HoodieConcatHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr, String
partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, Collections.emptyIterator(),
partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
- this.recordItr = recordItr;
+ // The parent must not consume the incoming records into keyToNewRecords:
this handle
+ // concatenates them after the existing records instead of merging by key.
Hand the parent
+ // an empty iterator (keeping the update count) and keep the real iterator
locally.
+ super(config, instantTime, hoodieTable,
MergeContext.create(mergeContext.getNumUpdates(), Collections.emptyIterator()),
partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
+ this.recordItr = mergeContext.getRecordIterator();
}
public HoodieConcatHandle(HoodieWriteConfig config, String instantTime,
HoodieTable hoodieTable,
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
index cca9c93139ce..421c0f828bb0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
@@ -44,6 +44,15 @@ public interface HoodieMergeHandle<T, I, K, O> {
@PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
HoodieBaseFile baseFileForMerge();
+ /**
+ * @return the number of incoming update and delete records tagged to the
file group based on
+ * workload profiling, or {@link MergeContext#UNKNOWN_NUM_UPDATES}
if unknown. See
+ * {@link MergeContext} for the exact semantics; this is a lower
bound, not the
+ * incoming record count.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ long getNumUpdates();
+
@PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
void setPartitionFields(Option<String[]> partitionFields);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
index 6f7ef0bb8c96..488b637b504a 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
@@ -35,7 +35,6 @@ import org.apache.hudi.table.HoodieTable;
import lombok.extern.slf4j.Slf4j;
-import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
@@ -55,7 +54,7 @@ public class HoodieMergeHandleFactory {
HoodieWriteConfig writeConfig,
String instantTime,
HoodieTable<T, I, K, O> table,
- Iterator<HoodieRecord<T>> recordItr,
+ MergeContext<T> mergeContext,
String partitionPath,
String fileId,
TaskContextSupplier taskContextSupplier,
@@ -67,13 +66,13 @@ public class HoodieMergeHandleFactory {
log.info("Create HoodieMergeHandle implementation {} {}",
mergeHandleClasses.getLeft(), logContext);
Class<?>[] constructorParamTypes = new Class<?>[] {
- HoodieWriteConfig.class, String.class, HoodieTable.class,
Iterator.class,
+ HoodieWriteConfig.class, String.class, HoodieTable.class,
MergeContext.class,
String.class, String.class, TaskContextSupplier.class, Option.class
};
return instantiateMergeHandle(
isFallbackEnabled, mergeHandleClasses.getLeft(),
mergeHandleClasses.getRight(), logContext, constructorParamTypes,
- writeConfig, instantTime, table, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ writeConfig, instantTime, table, mergeContext, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
index 2ef6c3964413..e635294dbba0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
@@ -39,7 +39,6 @@ import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
import java.io.IOException;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -52,9 +51,9 @@ public class HoodieMergeHandleWithChangeLog<T, I, K, O>
extends HoodieWriteMerge
protected final HoodieCDCLogWriter<IndexedRecord> cdcLogger;
public HoodieMergeHandleWithChangeLog(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr,
String partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
this.cdcLogger = createCDCLogWriter(instantTime, config, hoodieTable,
partitionPath, taskContextSupplier);
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
index 7cc74c40afee..458567b1fd91 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
@@ -33,7 +33,6 @@ import org.apache.hudi.table.HoodieTable;
import javax.annotation.concurrent.NotThreadSafe;
import java.io.IOException;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
@@ -51,9 +50,9 @@ public class HoodieSortedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<
private final Queue<String> newRecordKeysSorted = new
PriorityQueue<>(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
public HoodieSortedMergeHandle(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, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
+ TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
newRecordKeysSorted.addAll(keyToNewRecords.keySet());
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
index 539441949bc7..48a32deceda6 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
@@ -30,7 +30,6 @@ import org.apache.hudi.table.HoodieTable;
import org.apache.avro.generic.IndexedRecord;
import java.io.IOException;
-import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
@@ -39,9 +38,9 @@ import java.util.Properties;
*/
public class HoodieSortedMergeHandleWithChangeLog<T, I, K, O> extends
HoodieMergeHandleWithChangeLog<T, I, K, O> {
public HoodieSortedMergeHandleWithChangeLog(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>>
recordItr, String partitionPath, String fileId,
+ MergeContext<T> mergeContext,
String partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
index f343aefe9a40..d997d0438453 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
@@ -114,17 +114,17 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
protected long insertRecordsWritten = 0;
public HoodieWriteMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr, String
partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
- this(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier,
+ this(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier,
getLatestBaseFile(hoodieTable, partitionPath, fileId),
keyGeneratorOpt);
}
public HoodieWriteMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>> recordItr, String
partitionPath, String fileId,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
TaskContextSupplier taskContextSupplier,
HoodieBaseFile baseFile, Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, partitionPath, fileId,
taskContextSupplier, baseFile, keyGeneratorOpt, false);
- populateIncomingRecordsMap(recordItr);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, baseFile, keyGeneratorOpt, false);
+ populateIncomingRecordsMap(mergeContext.getRecordIterator());
initMarkerFileAndFileWriter(fileId, partitionPath);
this.readerContext =
hoodieTable.getReaderContextFactoryForWrite().getContext();
this.orderingFields = ConfigUtils.getOrderingFields(config.getProps());
@@ -132,12 +132,16 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
/**
* Called by compactor code path.
+ *
+ * <p>The incoming records arrive as the {@code keyToNewRecords} map instead
of an iterator,
+ * so the {@link MergeContext} passed to the parent carries an empty
iterator and
+ * {@link MergeContext#UNKNOWN_NUM_UPDATES}; workload profiling does not run
on this path.
*/
public HoodieWriteMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
Map<String, HoodieRecord<T>> keyToNewRecords,
String partitionPath, String fileId,
HoodieBaseFile dataFileToBeMerged,
TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, partitionPath, fileId,
taskContextSupplier, dataFileToBeMerged, keyGeneratorOpt,
+ super(config, instantTime, hoodieTable,
MergeContext.create(Collections.emptyIterator()), partitionPath, fileId,
taskContextSupplier, dataFileToBeMerged, keyGeneratorOpt,
// preserveMetadata is disabled by default for MDT but enabled
otherwise
!HoodieTableMetadata.isMetadataTable(config.getBasePath()));
this.keyToNewRecords = keyToNewRecords;
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/LsmFileGroupReaderBasedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/LsmFileGroupReaderBasedMergeHandle.java
index 01c327de6374..5b7d0a5f0afa 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/LsmFileGroupReaderBasedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/LsmFileGroupReaderBasedMergeHandle.java
@@ -46,15 +46,15 @@ import java.util.stream.Stream;
public class LsmFileGroupReaderBasedMergeHandle<T, I, K, O> extends
FileGroupReaderBasedMergeHandle<T, I, K, O> {
public LsmFileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>>
recordItr, String partitionPath, String fileId,
+ MergeContext<T> mergeContext,
String partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
}
public LsmFileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
- Iterator<HoodieRecord<T>>
recordItr, String partitionPath, String fileId,
+ MergeContext<T> mergeContext,
String partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, HoodieBaseFile baseFile, Option<BaseKeyGenerator>
keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, baseFile, keyGeneratorOpt);
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, baseFile, keyGeneratorOpt);
}
public LsmFileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/MergeContext.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/MergeContext.java
new file mode 100644
index 000000000000..58765f62fe64
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/MergeContext.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.apache.hudi.io;
+
+import org.apache.hudi.common.model.HoodieRecord;
+
+import java.util.Iterator;
+
+/**
+ * Context for merge handle creation, carrying incoming data to merge and its
characteristics.
+ */
+public class MergeContext<T> {
+
+ /**
+ * Sentinel value indicating that the number of update records is unknown.
+ */
+ public static final long UNKNOWN_NUM_UPDATES = -1L;
+
+ /**
+ * The number of incoming update and delete records tagged to the file group
based on
+ * workload profiling, or {@link #UNKNOWN_NUM_UPDATES} if unknown.
+ *
+ * <p>This is a lower bound of the record count of {@link #recordIterator},
not its exact
+ * size: with small-file handling, inserts routed to the same file group
also flow through
+ * the iterator but are not counted here. Do not use this value to pre-size
a data structure
+ * that must hold all incoming records.
+ */
+ private final long numUpdates;
+
+ /**
+ * Iterator over the incoming records to be merged.
+ */
+ private final Iterator<HoodieRecord<T>> recordIterator;
+
+ private MergeContext(long numUpdates, Iterator<HoodieRecord<T>>
recordIterator) {
+ this.numUpdates = numUpdates;
+ this.recordIterator = recordIterator;
+ }
+
+ public static <T> MergeContext<T> create(long numUpdates,
Iterator<HoodieRecord<T>> recordIterator) {
+ return new MergeContext<>(numUpdates, recordIterator);
+ }
+
+ public static <T> MergeContext<T> create(Iterator<HoodieRecord<T>>
recordIterator) {
+ return new MergeContext<>(UNKNOWN_NUM_UPDATES, recordIterator);
+ }
+
+ public long getNumUpdates() {
+ return numUpdates;
+ }
+
+ public Iterator<HoodieRecord<T>> getRecordIterator() {
+ return recordIterator;
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
index c666615571ff..cda73b9252c1 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
@@ -280,6 +280,7 @@ public abstract class BaseCommitActionExecutor<T, I, K, O,
R>
Iterator<HoodieRecord<T>> recordItr) throws Exception;
protected abstract Iterator<List<WriteStatus>> handleUpdate(String
partitionPath, String fileId,
+ long numUpdates,
Iterator<HoodieRecord<T>> recordItr) throws IOException;
protected HoodieWriteMetadata<HoodieData<WriteStatus>>
executeClustering(HoodieClusteringPlan clusteringPlan) {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BucketInfo.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BucketInfo.java
index 38f8877f22fb..cb3552d28ef5 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BucketInfo.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BucketInfo.java
@@ -18,6 +18,8 @@
package org.apache.hudi.table.action.commit;
+import org.apache.hudi.io.MergeContext;
+
import lombok.AllArgsConstructor;
import lombok.Getter;
@@ -34,12 +36,23 @@ public class BucketInfo implements Serializable {
BucketType bucketType;
String fileIdPrefix;
String partitionPath;
+ // The number of update and delete records from input based on tagging.
Populated only by the
+ // Spark and Java upsert partitioners, which report 0 (known none) for
INSERT buckets and
+ // insert-only small-file buckets; other producers (e.g., bucket index,
insert overwrite,
+ // metadata table, Flink) leave it at MergeContext.UNKNOWN_NUM_UPDATES.
Intentionally excluded
+ // from equals/hashCode: bucket identity is (bucketType, fileIdPrefix,
partitionPath).
+ long numUpdates;
+
+ public BucketInfo(BucketType bucketType, String fileIdPrefix, String
partitionPath) {
+ this(bucketType, fileIdPrefix, partitionPath,
MergeContext.UNKNOWN_NUM_UPDATES);
+ }
@Override
public String toString() {
return "BucketInfo {" + "bucketType=" + bucketType + ", "
+ "fileIdPrefix=" + fileIdPrefix + ", "
- + "partitionPath=" + partitionPath
+ + "partitionPath=" + partitionPath + ", "
+ + "numUpdates=" + numUpdates
+ '}';
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieMergeHandleFactoryInstantiation.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieMergeHandleFactoryInstantiation.java
new file mode 100644
index 000000000000..2086884ecf0c
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestHoodieMergeHandleFactoryInstantiation.java
@@ -0,0 +1,212 @@
+/*
+ * 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 org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.engine.LocalTaskContextSupplier;
+import org.apache.hudi.common.engine.ReaderContextFactory;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.view.TableFileSystemView;
+import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.core.io.storage.HoodieFileWriter;
+import org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests merge handle instantiation through {@link HoodieMergeHandleFactory}:
the reflection-based
+ * constructor lookup against the {@link MergeContext} signature, and the
fallback behavior for
+ * custom merge handle implementations that do not expose that constructor.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class TestHoodieMergeHandleFactoryInstantiation extends
HoodieCommonTestHarness {
+
+ private static final String DEFAULT_PARTITION_PATH = "partition";
+ private static final String DEFAULT_FILE_ID = "fileId";
+ private static final String DEFAULT_INSTANT_TIME = "0000002";
+ private static final String NEXT_INSTANT_TIME = "0000005";
+ private static final String DEFAULT_FILE_NAME =
String.format("%s_0-0-0_%s.parquet", DEFAULT_FILE_ID, DEFAULT_INSTANT_TIME);
+
+ @Mock
+ private HoodieTable mockTable;
+
+ @Mock
+ private TableFileSystemView.BaseFileOnlyView mockFileSystemView;
+
+ @Mock
+ private HoodieBaseFile mockBaseFile;
+
+ private TaskContextSupplier taskContextSupplier;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ // Lenient strictness: the fallback-disabled test never instantiates a
handle, so the shared
+ // stubs below are legitimately unused there.
+ initPath();
+ initMetaClient();
+ taskContextSupplier = new LocalTaskContextSupplier();
+
+ when(mockTable.getMetaClient()).thenReturn(metaClient);
+ when(mockTable.getBaseFileOnlyView()).thenReturn(mockFileSystemView);
+ when(mockTable.getPartitionMetafileFormat()).thenReturn(Option.empty());
+
when(mockTable.getBaseFileExtension()).thenReturn(HoodieFileFormat.PARQUET.getFileExtension());
+ when(mockTable.shouldTrackSuccessRecords()).thenReturn(false);
+ when(mockTable.isMetadataTable()).thenReturn(false);
+ when(mockTable.getStorageConf()).thenReturn(metaClient.getStorageConf());
+ when(mockTable.getStorage()).thenReturn(metaClient.getStorage());
+ ReaderContextFactory mockReaderContextFactory =
mock(ReaderContextFactory.class);
+
when(mockReaderContextFactory.getContext()).thenReturn(mock(HoodieReaderContext.class));
+
when(mockTable.getReaderContextFactoryForWrite()).thenReturn(mockReaderContextFactory);
+ when(mockFileSystemView.getLatestBaseFile(DEFAULT_PARTITION_PATH,
DEFAULT_FILE_ID)).thenReturn(Option.of(mockBaseFile));
+ when(mockBaseFile.getFileName()).thenReturn(DEFAULT_FILE_NAME);
+ when(mockBaseFile.getFileId()).thenReturn(DEFAULT_FILE_ID);
+ when(mockBaseFile.getCommitTime()).thenReturn(DEFAULT_INSTANT_TIME);
+ }
+
+ @AfterEach
+ void clean() {
+ cleanMetaClient();
+ }
+
+ private HoodieWriteConfig buildWriteConfig(Map<String, String> overrides) {
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)
+ .withProps(overrides)
+ .build();
+ when(mockTable.getConfig()).thenReturn(writeConfig);
+ return writeConfig;
+ }
+
+ private HoodieMergeHandle createHandleViaFactory(HoodieWriteConfig
writeConfig, long numUpdates) {
+ try (MockedStatic<WriteMarkersFactory> mockedMarkers =
mockStatic(WriteMarkersFactory.class);
+ MockedStatic<HoodieFileWriterFactory> mockedWriterFactory =
mockStatic(HoodieFileWriterFactory.class)) {
+ WriteMarkers mockWriteMarkers = mock(WriteMarkers.class);
+ mockedMarkers.when(() -> WriteMarkersFactory.get(any(), any(),
anyString())).thenReturn(mockWriteMarkers);
+ HoodieFileWriter mockWriter = mock(HoodieFileWriter.class);
+ mockedWriterFactory.when(() -> HoodieFileWriterFactory.getFileWriter(
+ anyString(), any(), any(), any(), any(), any(),
any())).thenReturn(mockWriter);
+
+ MergeContext<HoodieRecord> mergeContext =
MergeContext.create(numUpdates, Collections.emptyIterator());
+ return HoodieMergeHandleFactory.create(
+ WriteOperationType.UPSERT, writeConfig, NEXT_INSTANT_TIME,
mockTable, mergeContext,
+ DEFAULT_PARTITION_PATH, DEFAULT_FILE_ID, taskContextSupplier,
Option.empty());
+ }
+ }
+
+ @Test
+ void testFactoryCreatesCustomHandleWithMergeContextConstructor() {
+ Map<String, String> overrides = new HashMap<>();
+ overrides.put(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
MergeContextBasedMergeHandle.class.getName());
+ HoodieWriteConfig writeConfig = buildWriteConfig(overrides);
+
+ long expectedNumUpdates = 42L;
+ HoodieMergeHandle handle = createHandleViaFactory(writeConfig,
expectedNumUpdates);
+
+ assertInstanceOf(MergeContextBasedMergeHandle.class, handle);
+ assertEquals(expectedNumUpdates, ((HoodieWriteMergeHandle)
handle).getNumUpdates(),
+ "The factory should thread numUpdates through MergeContext into the
handle");
+ }
+
+ @Test
+ void testFactoryFallsBackToDefaultHandleForLegacyIteratorConstructor() {
+ // A custom merge handle that only exposes the pre-1.2.0 Iterator-based
constructor is no
+ // longer instantiable after the switch to the MergeContext-based
constructor lookup. With
+ // hoodie.write.merge.handle.fallback enabled (the default), the factory
silently substitutes
+ // the default implementation. This test documents that behavior.
+ Map<String, String> overrides = new HashMap<>();
+ overrides.put(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
LegacyIteratorBasedMergeHandle.class.getName());
+ HoodieWriteConfig writeConfig = buildWriteConfig(overrides);
+
+ HoodieMergeHandle handle = createHandleViaFactory(writeConfig, 42L);
+
+ assertInstanceOf(FileGroupReaderBasedMergeHandle.class, handle,
+ "With fallback enabled, a legacy custom handle should be silently
replaced by the default");
+ }
+
+ @Test
+ void testFactoryThrowsForLegacyIteratorConstructorWhenFallbackDisabled() {
+ Map<String, String> overrides = new HashMap<>();
+ overrides.put(HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
LegacyIteratorBasedMergeHandle.class.getName());
+ overrides.put(HoodieWriteConfig.MERGE_HANDLE_PERFORM_FALLBACK.key(),
"false");
+ HoodieWriteConfig writeConfig = buildWriteConfig(overrides);
+
+ assertThrows(HoodieException.class, () ->
createHandleViaFactory(writeConfig, 42L),
+ "With fallback disabled, a legacy custom handle should fail loudly");
+ }
+
+ /**
+ * A custom merge handle exposing the current MergeContext-based constructor.
+ */
+ public static class MergeContextBasedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
+ public MergeContextBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
+ TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
+ }
+ }
+
+ /**
+ * A custom merge handle exposing only the legacy Iterator-based constructor
shape used before
+ * the MergeContext refactoring.
+ */
+ public static class LegacyIteratorBasedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
+ public LegacyIteratorBasedMergeHandle(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, MergeContext.create(recordItr),
partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
+ }
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestMergeContext.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestMergeContext.java
new file mode 100644
index 000000000000..ed6b6636e2ea
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/TestMergeContext.java
@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.hudi.io;
+
+import org.apache.hudi.common.model.HoodieRecord;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.Iterator;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class TestMergeContext {
+
+ @Test
+ void testCreateWithIteratorOnlyDefaultsToUnknown() {
+ Iterator<HoodieRecord<Object>> itr = Collections.emptyIterator();
+ MergeContext<Object> ctx = MergeContext.create(itr);
+ assertEquals(MergeContext.UNKNOWN_NUM_UPDATES, ctx.getNumUpdates());
+ assertSame(itr, ctx.getRecordIterator());
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {MergeContext.UNKNOWN_NUM_UPDATES, 0L, 100L})
+ void testCreateWithExplicitNumUpdates(long numUpdates) {
+ Iterator<HoodieRecord<Object>> itr = Collections.emptyIterator();
+ MergeContext<Object> ctx = MergeContext.create(numUpdates, itr);
+ assertEquals(numUpdates, ctx.getNumUpdates());
+ assertSame(itr, ctx.getRecordIterator());
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/commit/TestBucketTypeAndInfo.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/commit/TestBucketTypeAndInfo.java
index 537250c00e61..af5b3cb20f5e 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/commit/TestBucketTypeAndInfo.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/commit/TestBucketTypeAndInfo.java
@@ -18,6 +18,8 @@
package org.apache.hudi.table.action.commit;
+import org.apache.hudi.io.MergeContext;
+
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -43,10 +45,24 @@ public class TestBucketTypeAndInfo {
assertSame(BucketType.INSERT, info.getBucketType());
assertEquals("fileId-1", info.getFileIdPrefix());
assertEquals("2024/01/01", info.getPartitionPath());
+ assertEquals(MergeContext.UNKNOWN_NUM_UPDATES, info.getNumUpdates(),
+ "The three-arg constructor should default numUpdates to unknown");
+
+ BucketInfo infoWithNumUpdates = new BucketInfo(BucketType.UPDATE,
"fileId-1", "2024/01/01", 42L);
+ assertEquals(42L, infoWithNumUpdates.getNumUpdates());
+ }
+
+ @Test
+ void bucketInfoEqualsAndHashCodeIgnoreNumUpdates() {
+ BucketInfo a = new BucketInfo(BucketType.UPDATE, "f1", "p1", 10L);
+ BucketInfo differentNumUpdates = new BucketInfo(BucketType.UPDATE, "f1",
"p1", 20L);
+ assertEquals(a, differentNumUpdates,
+ "Bucket identity is (bucketType, fileIdPrefix, partitionPath);
numUpdates is metadata");
+ assertEquals(a.hashCode(), differentNumUpdates.hashCode());
}
@Test
- void bucketInfoEqualsAndHashCodeUseAllFields() {
+ void bucketInfoEqualsAndHashCodeUseIdentityFields() {
BucketInfo a = new BucketInfo(BucketType.UPDATE, "f1", "p1");
BucketInfo same = new BucketInfo(BucketType.UPDATE, "f1", "p1");
BucketInfo differentType = new BucketInfo(BucketType.INSERT, "f1", "p1");
@@ -64,9 +80,10 @@ public class TestBucketTypeAndInfo {
@Test
void bucketInfoToStringContainsFields() {
- String rendered = new BucketInfo(BucketType.INSERT, "fileId-9",
"part-9").toString();
+ String rendered = new BucketInfo(BucketType.INSERT, "fileId-9", "part-9",
42L).toString();
assertTrue(rendered.contains("INSERT"));
assertTrue(rendered.contains("fileId-9"));
assertTrue(rendered.contains("part-9"));
+ assertTrue(rendered.contains("numUpdates=42"));
}
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
index fa68e055cec5..b974b85a63d2 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
@@ -59,7 +59,7 @@ public class FlinkFileGroupReaderBasedMergeHandle<T, I, K, O>
public FlinkFileGroupReaderBasedMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
Iterator<HoodieRecord<T>>
recordItr, String partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, HoodieBaseFile hoodieBaseFile) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, hoodieBaseFile, Option.empty());
+ super(config, instantTime, hoodieTable, MergeContext.create(recordItr),
partitionPath, fileId, taskContextSupplier, hoodieBaseFile, Option.empty());
// delete invalid data files generated by task retry.
if (getAttemptId() > 0) {
deleteInvalidDataFile(getAttemptId() - 1);
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkLsmFileGroupReaderBasedMergeHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkLsmFileGroupReaderBasedMergeHandle.java
index 03652b0fed3e..559bae1ba0be 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkLsmFileGroupReaderBasedMergeHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkLsmFileGroupReaderBasedMergeHandle.java
@@ -53,7 +53,7 @@ public class FlinkLsmFileGroupReaderBasedMergeHandle<T, I, K,
O>
public FlinkLsmFileGroupReaderBasedMergeHandle(HoodieWriteConfig config,
String instantTime, HoodieTable<T, I, K, O> hoodieTable,
Iterator<HoodieRecord<T>>
recordItr, String partitionPath, String fileId,
TaskContextSupplier
taskContextSupplier, HoodieBaseFile hoodieBaseFile) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, hoodieBaseFile, Option.empty());
+ super(config, instantTime, hoodieTable, MergeContext.create(recordItr),
partitionPath, fileId, taskContextSupplier, hoodieBaseFile, Option.empty());
if (getAttemptId() > 0) {
deleteInvalidDataFile(getAttemptId() - 1);
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
index 3505a0aeea15..b8f8c90310f6 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
@@ -60,7 +60,7 @@ public class FlinkMergeHandle<T, I, K, O>
public FlinkMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
Iterator<HoodieRecord<T>> recordItr, String
partitionPath, String fileId,
TaskContextSupplier taskContextSupplier,
HoodieBaseFile baseFile) {
- super(config, instantTime, hoodieTable, recordItr, partitionPath, fileId,
taskContextSupplier, baseFile, Option.empty());
+ super(config, instantTime, hoodieTable, MergeContext.create(recordItr),
partitionPath, fileId, taskContextSupplier, baseFile, Option.empty());
// delete invalid data files generated by task retry.
if (getAttemptId() > 0) {
deleteInvalidDataFile(getAttemptId() - 1);
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
index c75aeeaaf9ad..f31d3f3b58c0 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
@@ -36,6 +36,7 @@ import org.apache.hudi.io.HoodieCreateHandle;
import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.io.HoodieWriteHandle;
import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.io.MergeContext;
import org.apache.hudi.io.MergeUtils;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
@@ -167,13 +168,13 @@ public abstract class BaseFlinkCommitActionExecutor<T>
extends
// and append instead of UPDATE.
return handleInsert(fileIdHint, recordItr);
} else if (this.writeHandle instanceof HoodieWriteMergeHandle) {
- return handleUpdate(partitionPath, fileIdHint, recordItr);
+ return handleUpdate(partitionPath, fileIdHint,
MergeContext.UNKNOWN_NUM_UPDATES, recordItr);
} else {
switch (bucketType) {
case INSERT:
return handleInsert(fileIdHint, recordItr);
case UPDATE:
- return handleUpdate(partitionPath, fileIdHint, recordItr);
+ return handleUpdate(partitionPath, fileIdHint,
MergeContext.UNKNOWN_NUM_UPDATES, recordItr);
default:
throw new AssertionError();
}
@@ -187,6 +188,7 @@ public abstract class BaseFlinkCommitActionExecutor<T>
extends
@Override
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId,
+ long numUpdates,
Iterator<HoodieRecord<T>>
recordItr)
throws IOException {
ValidationUtils.checkArgument(this.writeHandle instanceof
HoodieMergeHandle,
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java
index 45dd3f42ed65..ee371eb2502d 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java
@@ -51,7 +51,7 @@ public abstract class BaseFlinkDeltaCommitActionExecutor<T>
}
@Override
- public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId, Iterator<HoodieRecord<T>> recordItr) {
+ public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId, long numUpdates, Iterator<HoodieRecord<T>> recordItr) {
HoodieAppendHandle appendHandle = (HoodieAppendHandle) writeHandle;
appendHandle.doAppend();
List<WriteStatus> writeStatuses = appendHandle.close();
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java
index 239dc834c70e..fcc8dfd95f0c 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java
@@ -54,7 +54,7 @@ public class FlinkUpsertDeltaCommitActionExecutor<T> extends
BaseFlinkCommitActi
}
@Override
- public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId, Iterator<HoodieRecord<T>> recordItr) {
+ public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId, long numUpdates, Iterator<HoodieRecord<T>> recordItr) {
return handleWrite();
}
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
index dd3b62f7fb55..ab1b0beec3ed 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/BaseJavaCommitActionExecutor.java
@@ -39,6 +39,7 @@ import org.apache.hudi.execution.JavaLazyInsertIterable;
import org.apache.hudi.io.CreateHandleFactory;
import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.io.HoodieMergeHandleFactory;
+import org.apache.hudi.io.MergeContext;
import org.apache.hudi.io.MergeUtils;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.keygen.factory.HoodieAvroKeyGeneratorFactory;
@@ -214,7 +215,7 @@ public abstract class BaseJavaCommitActionExecutor<T>
extends
if (btype.equals(BucketType.INSERT)) {
return handleInsert(binfo.fileIdPrefix, recordItr);
} else if (btype.equals(BucketType.UPDATE)) {
- return handleUpdate(binfo.partitionPath, binfo.fileIdPrefix,
recordItr);
+ return handleUpdate(binfo.partitionPath, binfo.fileIdPrefix,
binfo.getNumUpdates(), recordItr);
} else {
throw new HoodieUpsertException("Unknown bucketType " + btype + " for
partition :" + partition);
}
@@ -232,6 +233,7 @@ public abstract class BaseJavaCommitActionExecutor<T>
extends
@Override
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId,
+ long numUpdates,
Iterator<HoodieRecord<T>>
recordItr)
throws IOException {
// This is needed since sometimes some buckets are never picked in
getPartition() and end up with 0 records
@@ -240,11 +242,13 @@ public abstract class BaseJavaCommitActionExecutor<T>
extends
return Collections.singletonList((List<WriteStatus>)
Collections.EMPTY_LIST).iterator();
}
// these are updates
- HoodieMergeHandle<?, ?, ?, ?> mergeHandle = getUpdateHandle(partitionPath,
fileId, recordItr);
+ HoodieMergeHandle<?, ?, ?, ?> mergeHandle = getUpdateHandle(partitionPath,
fileId, numUpdates, recordItr);
return MergeUtils.runMerge(mergeHandle, instantTime, fileId);
}
- protected HoodieMergeHandle<?, ?, ?, ?> getUpdateHandle(String
partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr) {
+ protected HoodieMergeHandle<?, ?, ?, ?> getUpdateHandle(String
partitionPath, String fileId,
+ long numUpdates,
+
Iterator<HoodieRecord<T>> recordItr) {
Option<BaseKeyGenerator> keyGeneratorOpt = Option.empty();
if (!config.populateMetaFields()) {
try {
@@ -254,7 +258,8 @@ public abstract class BaseJavaCommitActionExecutor<T>
extends
+ "columns are disabled. Please choose the right key generator if
you wish to disable meta fields.", e);
}
}
- return HoodieMergeHandleFactory.create(operationType, config, instantTime,
table, recordItr, partitionPath, fileId,
+ MergeContext<T> mergeContext = MergeContext.create(numUpdates, recordItr);
+ return HoodieMergeHandleFactory.create(operationType, config, instantTime,
table, mergeContext, partitionPath, fileId,
taskContextSupplier, keyGeneratorOpt);
}
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java
index 9b86e6ccb806..3f00578132d8 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java
@@ -104,7 +104,7 @@ public class JavaUpsertPartitioner<T> implements
Partitioner {
WorkloadStat outputWorkloadStats =
profile.getOutputPartitionPathStatMap().getOrDefault(partitionStat.getKey(),
new WorkloadStat());
for (Map.Entry<String, Pair<String, Long>> updateLocEntry :
partitionStat.getValue().getUpdateLocationToCount().entrySet()) {
- addUpdateBucket(partitionStat.getKey(), updateLocEntry.getKey());
+ addUpdateBucket(partitionStat.getKey(), updateLocEntry.getKey(),
updateLocEntry.getValue().getValue());
if (profile.hasOutputWorkLoadStats()) {
HoodieRecordLocation hoodieRecordLocation = new
HoodieRecordLocation(updateLocEntry.getValue().getKey(),
updateLocEntry.getKey());
outputWorkloadStats.addUpdates(hoodieRecordLocation,
updateLocEntry.getValue().getValue());
@@ -116,10 +116,10 @@ public class JavaUpsertPartitioner<T> implements
Partitioner {
}
}
- private int addUpdateBucket(String partitionPath, String fileIdHint) {
+ private int addUpdateBucket(String partitionPath, String fileIdHint, long
numUpdates) {
int bucket = totalBuckets;
updateLocationToBucket.put(fileIdHint, bucket);
- BucketInfo bucketInfo = new BucketInfo(BucketType.UPDATE, fileIdHint,
partitionPath);
+ BucketInfo bucketInfo = new BucketInfo(BucketType.UPDATE, fileIdHint,
partitionPath, numUpdates);
bucketInfoMap.put(totalBuckets, bucketInfo);
totalBuckets++;
return bucket;
@@ -161,7 +161,8 @@ public class JavaUpsertPartitioner<T> implements
Partitioner {
bucket =
updateLocationToBucket.get(smallFile.location.getFileId());
log.info("Assigning {} inserts to existing update bucket {}",
recordsToAppend, bucket);
} else {
- bucket = addUpdateBucket(partitionPath,
smallFile.location.getFileId());
+ // the bucket receives inserts only, so no tagged updates are
expected
+ bucket = addUpdateBucket(partitionPath,
smallFile.location.getFileId(), 0L);
log.info("Assigning {} inserts to new update bucket {}",
recordsToAppend, bucket);
}
if (profile.hasOutputWorkLoadStats()) {
@@ -189,7 +190,8 @@ public class JavaUpsertPartitioner<T> implements
Partitioner {
} else {
recordsPerBucket.add(totalUnassignedInserts - (insertBuckets -
1) * insertRecordsPerBucket);
}
- BucketInfo bucketInfo = new BucketInfo(BucketType.INSERT,
FSUtils.createNewFileIdPfx(), partitionPath);
+ // INSERT buckets carry no tagged updates, so the count is a known
zero rather than unknown
+ BucketInfo bucketInfo = new BucketInfo(BucketType.INSERT,
FSUtils.createNewFileIdPfx(), partitionPath, 0L);
bucketInfoMap.put(totalBuckets, bucketInfo);
if (profile.hasOutputWorkLoadStats()) {
outputWorkloadStats.addInserts(new
HoodieRecordLocation(HoodieWriteStat.NULL_COMMIT,
bucketInfo.getFileIdPrefix()), recordsPerBucket.get(recordsPerBucket.size() -
1));
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseJavaDeltaCommitActionExecutor.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseJavaDeltaCommitActionExecutor.java
index 6a205ca75371..7104a852c6b3 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseJavaDeltaCommitActionExecutor.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseJavaDeltaCommitActionExecutor.java
@@ -70,12 +70,14 @@ abstract class BaseJavaDeltaCommitActionExecutor<T> extends
BaseJavaCommitAction
}
@Override
- public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId, Iterator<HoodieRecord<T>> recordItr) throws IOException {
+ public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId,
+ long numUpdates,
+ Iterator<HoodieRecord<T>>
recordItr) throws IOException {
log.info("Merging updates for commit {} for file {}", instantTime, fileId);
if (!table.getIndex().canIndexLogFiles() && partitioner != null
&& partitioner.getSmallFileIds().contains(fileId)) {
log.info("Small file corrections for updates for commit {} for file {}",
instantTime, fileId);
- return super.handleUpdate(partitionPath, fileId, recordItr);
+ return super.handleUpdate(partitionPath, fileId, numUpdates, recordItr);
} else {
if (table.requireSortedRecords()) {
recordItr = sortRecordsByRecordKey(recordItr);
diff --git
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaCopyOnWriteActionExecutor.java
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaCopyOnWriteActionExecutor.java
index a0998cbdd031..2f43f910ef55 100644
---
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaCopyOnWriteActionExecutor.java
+++
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaCopyOnWriteActionExecutor.java
@@ -24,6 +24,7 @@ import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
@@ -41,6 +42,9 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.HoodieParquetInputFormat;
import org.apache.hudi.hadoop.utils.HoodieHiveUtils;
import org.apache.hudi.io.HoodieCreateHandle;
+import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.io.MergeContext;
+import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieJavaCopyOnWriteTable;
import org.apache.hudi.table.HoodieJavaTable;
@@ -67,6 +71,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
@@ -122,6 +127,59 @@ public class TestJavaCopyOnWriteActionExecutor extends
HoodieJavaClientTestHarne
.withSchema(SCHEMA.toString());
}
+ @Test
+ public void testUpsertPropagatesProfiledNumUpdatesToMergeHandle() throws
Exception {
+ // End-to-end check of the numUpdates plumbing for the Java client:
workload profile ->
+ // JavaUpsertPartitioner -> BucketInfo -> handleUpdate -> MergeContext ->
merge handle.
+ HoodieWriteConfig config = makeHoodieClientConfigBuilder()
+ .withProps(Collections.singletonMap(
+ HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
+ NumUpdatesRecordingMergeHandle.class.getName()))
+ .build();
+ int startInstant = 1;
+ String firstCommitTime = makeNewCommitTime(startInstant++, "%09d");
+ HoodieJavaWriteClient writeClient = getHoodieWriteClient(config);
+ WriteClientTestUtils.startCommitWithTime(writeClient, firstCommitTime);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
+ List<HoodieRecord> records = new ArrayList<>();
+ records.add(createSimpleRecord("aeb5b87a-1feh-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 12));
+ records.add(createSimpleRecord("aeb5b87b-1feu-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 100));
+ records.add(createSimpleRecord("aeb5b87c-1fej-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 15));
+ writeClient.commit(firstCommitTime, writeClient.insert(records,
firstCommitTime), Option.empty(), COMMIT_ACTION, Collections.emptyMap());
+
+ // Update two existing records and insert a new one. The tagged update
count for the single
+ // file group must be exactly 2, no matter how the new insert is routed.
+ List<HoodieRecord> secondBatch = Arrays.asList(
+ createSimpleRecord("aeb5b87a-1feh-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 15),
+ createSimpleRecord("aeb5b87b-1feu-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 101),
+ createSimpleRecord("aeb5b87d-1fej-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 51));
+ NumUpdatesRecordingMergeHandle.RECORDED_NUM_UPDATES.set(Long.MIN_VALUE);
+ String secondCommitTime = makeNewCommitTime(startInstant++, "%09d");
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ WriteClientTestUtils.startCommitWithTime(writeClient, secondCommitTime);
+ List<WriteStatus> statuses = writeClient.upsert(secondBatch,
secondCommitTime);
+ writeClient.commit(secondCommitTime, statuses, Option.empty(),
COMMIT_ACTION, Collections.emptyMap());
+
+ assertEquals(2L, NumUpdatesRecordingMergeHandle.RECORDED_NUM_UPDATES.get(),
+ "The merge handle must receive the tagged update count from workload
profiling");
+ }
+
+ /**
+ * A merge handle that records the numUpdates it receives, to assert the
end-to-end propagation
+ * from workload profiling.
+ */
+ public static class NumUpdatesRecordingMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
+ public static final AtomicLong RECORDED_NUM_UPDATES = new
AtomicLong(Long.MIN_VALUE);
+
+ public NumUpdatesRecordingMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
+ TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
+ RECORDED_NUM_UPDATES.set(mergeContext.getNumUpdates());
+ }
+ }
+
@Test
public void testUpdateRecords() throws Exception {
// Prepare the AvroParquetIO
@@ -413,7 +471,7 @@ public class TestJavaCopyOnWriteActionExecutor extends
HoodieJavaClientTestHarne
taskContextSupplier.reset();
final List<List<WriteStatus>> updateStatus = new ArrayList<>();
- newActionExecutor.handleUpdate(partitionPath, fileId, updates.iterator())
+ newActionExecutor.handleUpdate(partitionPath, fileId, updates.size(),
updates.iterator())
.forEachRemaining(x -> updateStatus.add((List<WriteStatus>)x));
assertEquals(updates.size() - numRecordsInPartition,
updateStatus.get(0).get(0).getTotalErrorRecords());
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
index d48341fa09b7..b66aaefa808b 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/bootstrap/SparkBootstrapCommitActionExecutor.java
@@ -326,7 +326,7 @@ public class SparkBootstrapCommitActionExecutor<T>
}
@Override
- protected Iterator<List<WriteStatus>> handleUpdate(String partitionPath,
String fileId, Iterator<HoodieRecord<T>> recordItr) {
+ protected Iterator<List<WriteStatus>> handleUpdate(String partitionPath,
String fileId, long numUpdates, Iterator<HoodieRecord<T>> recordItr) {
throw new UnsupportedOperationException("Should not called in bootstrap
code path");
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index 06079a511bf8..add4d33c4ed7 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -54,6 +54,7 @@ import org.apache.hudi.io.CreateHandleFactory;
import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.io.HoodieMergeHandleFactory;
import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.io.MergeContext;
import org.apache.hudi.io.MergeUtils;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory;
@@ -405,7 +406,7 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
if (btype.equals(BucketType.INSERT)) {
return handleInsert(binfo.fileIdPrefix, recordItr);
} else if (btype.equals(BucketType.UPDATE)) {
- return handleUpdate(binfo.partitionPath, binfo.fileIdPrefix,
recordItr);
+ return handleUpdate(binfo.partitionPath, binfo.fileIdPrefix,
binfo.getNumUpdates(), recordItr);
} else {
throw new HoodieUpsertException("Unknown bucketType " + btype + " for
partition :" + partition);
}
@@ -423,6 +424,7 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
@Override
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId,
+ long numUpdates,
Iterator<HoodieRecord<T>>
recordItr)
throws IOException {
// This is needed since sometimes some buckets are never picked in
getPartition() and end up with 0 records
@@ -438,13 +440,16 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
}
// these are updates
- HoodieMergeHandle mergeHandle = getUpdateHandle(partitionPath, fileId,
recordItr);
+ HoodieMergeHandle mergeHandle = getUpdateHandle(partitionPath, fileId,
numUpdates, recordItr);
return MergeUtils.runMerge(mergeHandle, instantTime, fileId);
}
- protected HoodieMergeHandle getUpdateHandle(String partitionPath, String
fileId, Iterator<HoodieRecord<T>> recordItr) {
- HoodieMergeHandle mergeHandle =
HoodieMergeHandleFactory.create(operationType, config, instantTime, table,
recordItr, partitionPath, fileId,
- taskContextSupplier, keyGeneratorOpt);
+ protected HoodieMergeHandle getUpdateHandle(String partitionPath, String
fileId,
+ long numUpdates,
Iterator<HoodieRecord<T>> recordItr) {
+ MergeContext<T> mergeContext = MergeContext.create(numUpdates, recordItr);
+ HoodieMergeHandle mergeHandle = HoodieMergeHandleFactory.create(
+ operationType, config, instantTime, table, mergeContext,
+ partitionPath, fileId, taskContextSupplier, keyGeneratorOpt);
if (mergeHandle.getOldFilePath() != null &&
mergeHandle.baseFileForMerge().getBootstrapBaseFile().isPresent()) {
Option<String[]> partitionFields =
table.getMetaClient().getTableConfig().getPartitionFields();
Object[] partitionValues =
SparkPartitionUtils.getPartitionFieldVals(table.getMetaClient().getTableConfig(),
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
index 858b590cda79..96d874b1e9a6 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
@@ -115,7 +115,7 @@ public class UpsertPartitioner<T> extends
SparkHoodiePartitioner<T> {
WorkloadStat outputWorkloadStats =
profile.getOutputPartitionPathStatMap().getOrDefault(partitionStat.getKey(),
new WorkloadStat());
for (Map.Entry<String, Pair<String, Long>> updateLocEntry :
partitionStat.getValue().getUpdateLocationToCount().entrySet()) {
- addUpdateBucket(partitionStat.getKey(), updateLocEntry.getKey());
+ addUpdateBucket(partitionStat.getKey(), updateLocEntry.getKey(),
updateLocEntry.getValue().getValue());
if (profile.hasOutputWorkLoadStats()) {
HoodieRecordLocation hoodieRecordLocation = new
HoodieRecordLocation(updateLocEntry.getValue().getKey(),
updateLocEntry.getKey());
outputWorkloadStats.addUpdates(hoodieRecordLocation,
updateLocEntry.getValue().getValue());
@@ -127,10 +127,10 @@ public class UpsertPartitioner<T> extends
SparkHoodiePartitioner<T> {
}
}
- private int addUpdateBucket(String partitionPath, String fileIdHint) {
+ private int addUpdateBucket(String partitionPath, String fileIdHint, long
numUpdates) {
int bucket = totalBuckets;
updateLocationToBucket.put(fileIdHint, bucket);
- BucketInfo bucketInfo = new BucketInfo(BucketType.UPDATE, fileIdHint,
partitionPath);
+ BucketInfo bucketInfo = new BucketInfo(BucketType.UPDATE, fileIdHint,
partitionPath, numUpdates);
bucketInfoMap.put(totalBuckets, bucketInfo);
totalBuckets++;
return bucket;
@@ -212,7 +212,8 @@ public class UpsertPartitioner<T> extends
SparkHoodiePartitioner<T> {
bucket =
updateLocationToBucket.get(smallFile.location.getFileId());
log.debug("Assigning {} inserts to existing update bucket {}",
recordsToAppend, bucket);
} else {
- bucket = addUpdateBucket(partitionPath,
smallFile.location.getFileId());
+ // the bucket receives inserts only, so no tagged updates are
expected
+ bucket = addUpdateBucket(partitionPath,
smallFile.location.getFileId(), 0L);
log.debug("Assigning {} inserts to new update bucket {}",
recordsToAppend, bucket);
}
if (profile.hasOutputWorkLoadStats()) {
@@ -245,7 +246,8 @@ public class UpsertPartitioner<T> extends
SparkHoodiePartitioner<T> {
} else {
recordsPerBucket.add(totalUnassignedInserts - (insertBuckets -
1) * insertRecordsPerBucket);
}
- BucketInfo bucketInfo = new BucketInfo(BucketType.INSERT,
FSUtils.createNewFileIdPfx(), partitionPath);
+ // INSERT buckets carry no tagged updates, so the count is a known
zero rather than unknown
+ BucketInfo bucketInfo = new BucketInfo(BucketType.INSERT,
FSUtils.createNewFileIdPfx(), partitionPath, 0L);
bucketInfoMap.put(totalBuckets, bucketInfo);
if (profile.hasOutputWorkLoadStats()) {
outputWorkloadStats.addInserts(new
HoodieRecordLocation(HoodieWriteStat.NULL_COMMIT,
bucketInfo.getFileIdPrefix()), recordsPerBucket.get(recordsPerBucket.size() -
1));
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
index eafb21a9c1c9..7728e2772d8b 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
@@ -70,12 +70,13 @@ public abstract class BaseSparkDeltaCommitActionExecutor<T>
@Override
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String
fileId,
- Iterator<HoodieRecord<T>> recordItr) throws IOException {
+ long numUpdates,
+ Iterator<HoodieRecord<T>>
recordItr) throws IOException {
log.info("Merging updates for commit {} for file {}", instantTime, fileId);
if (!table.getIndex().canIndexLogFiles() && mergeOnReadUpsertPartitioner
!= null
&& mergeOnReadUpsertPartitioner.getSmallFileIds().contains(fileId)) {
log.info("Small file corrections for updates for commit {} for file {}",
instantTime, fileId);
- return super.handleUpdate(partitionPath, fileId, recordItr);
+ return super.handleUpdate(partitionPath, fileId, numUpdates, recordItr);
} else {
HoodieAppendHandle<?, ?, ?, ?> appendHandle = new AppendHandleFactory()
.create(config, instantTime, table, partitionPath, fileId,
recordItr, taskContextSupplier);
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
index 4690ebb5805d..b6f7fcc1834e 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestUpdateSchemaEvolution.java
@@ -49,6 +49,7 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.io.CreateHandleFactory;
import org.apache.hudi.io.HoodieWriteHandle;
import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.io.MergeContext;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
@@ -81,6 +82,7 @@ import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME
import static
org.apache.hudi.common.testutils.HoodieTestUtils.createSimpleRecord;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.extractPartitionFromTimeField;
import static
org.apache.hudi.common.testutils.SchemaTestUtil.getSchemaFromResource;
+import static
org.apache.hudi.testutils.HoodieClientTestUtils.toTypedRecordIterator;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -159,7 +161,7 @@ public class TestUpdateSchemaEvolution extends
HoodieSparkClientTestHarness impl
jsc.parallelize(Arrays.asList(1)).map(x -> {
Executable executable = () -> {
HoodieWriteMergeHandle mergeHandle = new
HoodieWriteMergeHandle(updateTable.getConfig(), "101", updateTable,
- updateRecords.iterator(), updateRecords.get(0).getPartitionPath(),
insertResult.getFileId(), supplier, Option.empty());
+ MergeContext.create(updateRecords.size(),
toTypedRecordIterator(updateRecords)), updateRecords.get(0).getPartitionPath(),
insertResult.getFileId(), supplier, Option.empty());
List<GenericRecord> oldRecords =
HoodieIOFactory.getIOFactory(updateTable.getStorage())
.getFileFormatUtils(updateTable.getBaseFileFormat())
.readAvroRecords(updateTable.getStorage(),
@@ -337,7 +339,7 @@ public class TestUpdateSchemaEvolution extends
HoodieSparkClientTestHarness impl
HoodieSparkTable table = HoodieSparkTable.create(config, context);
List<String> mergedFilePaths = jsc.parallelize(Arrays.asList(1)).map(x -> {
HoodieWriteMergeHandle mergeHandle = new HoodieWriteMergeHandle(config,
"101", table,
- updateRecords.iterator(), updateRecords.get(0).getPartitionPath(),
insertResult.getFileId(), supplier, Option.empty());
+ MergeContext.create(toTypedRecordIterator(updateRecords)),
updateRecords.get(0).getPartitionPath(), insertResult.getFileId(), supplier,
Option.empty());
// `doMerge` is the only entry point into HoodieMergeHelper: it reads
the base file and feeds the handle
mergeHandle.doMerge();
return ((WriteStatus) mergeHandle.close().get(0)).getStat().getPath();
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
index c23c37b16a47..082e179f9249 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
@@ -50,6 +51,8 @@ import org.apache.hudi.hadoop.utils.HoodieHiveUtils;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.io.HoodieCreateHandle;
import org.apache.hudi.io.HoodieWriteMergeHandle;
+import org.apache.hudi.io.MergeContext;
+import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.keygen.KeyGeneratorInterface;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.storage.StoragePath;
@@ -89,6 +92,7 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -258,6 +262,43 @@ public class TestCopyOnWriteActionExecutor extends
HoodieClientTestBase implemen
assertEquals(4, writeStatus.getStat().getNumWrites());// 3 rewritten
records + 1 new record
}
+ @Test
+ public void testUpsertPropagatesProfiledNumUpdatesToMergeHandle() throws
Exception {
+ // End-to-end check of the numUpdates plumbing: workload profile ->
UpsertPartitioner ->
+ // BucketInfo -> handleUpdate -> MergeContext -> HoodieMergeHandleFactory
-> merge handle.
+ // Also exercises the factory reflection against the MergeContext
constructor signature for
+ // a custom merge handle class in a real upsert.
+ HoodieWriteConfig config = makeHoodieClientConfigBuilder()
+ .withProps(Collections.singletonMap(
+ HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key(),
+ NumUpdatesRecordingMergeHandle.class.getName()))
+ .build();
+ SparkRDDWriteClient writeClient = getHoodieWriteClient(config);
+ String firstCommitTime = writeClient.startCommit();
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
+ List<HoodieRecord> records = new ArrayList<>();
+ records.add(createSimpleRecord("aeb5b87a-1feh-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 12));
+ records.add(createSimpleRecord("aeb5b87b-1feu-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:20:41.415Z", 100));
+ records.add(createSimpleRecord("aeb5b87c-1fej-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 15));
+ JavaRDD<WriteStatus> writeStatuses =
writeClient.insert(jsc.parallelize(records, 1), firstCommitTime);
+ writeClient.commit(firstCommitTime, writeStatuses, Option.empty(),
COMMIT_ACTION, Collections.emptyMap(), Option.empty());
+
+ // Update two existing records and insert a new one. The tagged update
count for the single
+ // file group must be exactly 2, no matter how the new insert is routed.
+ List<HoodieRecord> secondBatch = Arrays.asList(
+ createSimpleRecord("aeb5b87a-1feh-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 15),
+ createSimpleRecord("aeb5b87b-1feu-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:20:41.415Z", 101),
+ createSimpleRecord("aeb5b87d-1fej-4edd-87b4-6ec96dc405a0",
"2016-01-31T03:16:41.415Z", 51));
+ NumUpdatesRecordingMergeHandle.RECORDED_NUM_UPDATES.set(Long.MIN_VALUE);
+ String secondCommitTime = writeClient.startCommit();
+ writeStatuses = writeClient.upsert(jsc.parallelize(secondBatch, 1),
secondCommitTime);
+ writeClient.commit(secondCommitTime, writeStatuses, Option.empty(),
COMMIT_ACTION, Collections.emptyMap(), Option.empty());
+
+ assertEquals(2L, NumUpdatesRecordingMergeHandle.RECORDED_NUM_UPDATES.get(),
+ "The merge handle must receive the tagged update count from workload
profiling");
+ }
+
private FileStatus[] getIncrementalFiles(String partitionPath, String
startCommitTime, int numCommitsToPull)
throws Exception {
// initialize parquet input format
@@ -462,7 +503,8 @@ public class TestCopyOnWriteActionExecutor extends
HoodieClientTestBase implemen
instantTime, context.parallelize(updates));
final List<List<WriteStatus>> updateStatus =
jsc.parallelize(Arrays.asList(1))
.map(x -> (Iterator<List<WriteStatus>>)
- newActionExecutor.handleUpdate(partitionPath, fileId,
updates.iterator()))
+ newActionExecutor.handleUpdate(
+ partitionPath, fileId, updates.size(), updates.iterator()))
.map(Transformations::flatten).collect();
assertEquals(updates.size() - numRecordsInPartition,
updateStatus.get(0).get(0).getTotalErrorRecords());
@@ -637,6 +679,21 @@ public class TestCopyOnWriteActionExecutor extends
HoodieClientTestBase implemen
assertTrue(metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().empty());
}
+ /**
+ * A merge handle that records the numUpdates it receives, to assert the
end-to-end propagation
+ * from workload profiling. Only valid in local-mode tests where the
executor shares the JVM.
+ */
+ public static class NumUpdatesRecordingMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
+ public static final AtomicLong RECORDED_NUM_UPDATES = new
AtomicLong(Long.MIN_VALUE);
+
+ public NumUpdatesRecordingMergeHandle(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ MergeContext<T> mergeContext, String
partitionPath, String fileId,
+ TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, mergeContext, partitionPath,
fileId, taskContextSupplier, keyGeneratorOpt);
+ RECORDED_NUM_UPDATES.set(mergeContext.getNumUpdates());
+ }
+ }
+
/**
* A key generator that deliberately does not extend {@link
org.apache.hudi.keygen.BaseKeyGenerator}, which is the
* only shape supported when the meta fields are turned off.
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
index 98e794c030e5..408a9569fb3f 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
@@ -36,6 +36,7 @@ import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.io.MergeContext;
import org.apache.hudi.table.HoodieSparkCopyOnWriteTable;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.WorkloadProfile;
@@ -197,10 +198,16 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
SparkBucketInfoGetter bucketInfoGetter =
partitioner.getSparkBucketInfoGetter();
assertEquals(BucketType.UPDATE,
bucketInfoGetter.getBucketInfo(0).bucketType,
"Bucket 0 is UPDATE");
+ assertEquals(100, bucketInfoGetter.getBucketInfo(0).getNumUpdates(),
+ "Bucket 0 should have 100 numUpdates");
assertEquals(BucketType.INSERT,
bucketInfoGetter.getBucketInfo(1).bucketType,
"Bucket 1 is INSERT");
+ assertEquals(0, bucketInfoGetter.getBucketInfo(1).getNumUpdates(),
+ "INSERT bucket 1 should have 0 numUpdates");
assertEquals(BucketType.INSERT,
bucketInfoGetter.getBucketInfo(2).bucketType,
"Bucket 2 is INSERT");
+ assertEquals(0, bucketInfoGetter.getBucketInfo(2).getNumUpdates(),
+ "INSERT bucket 2 should have 0 numUpdates");
assertEquals(3, insertBuckets.size(), "Total of 3 insert buckets");
Double[] weights = { 0.5, 0.25, 0.25};
@@ -215,6 +222,8 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
bucketInfoGetter = partitioner.getSparkBucketInfoGetter();
assertEquals(BucketType.UPDATE,
bucketInfoGetter.getBucketInfo(0).bucketType,
"Bucket 0 is UPDATE");
+ assertEquals(100, bucketInfoGetter.getBucketInfo(0).getNumUpdates(),
+ "Bucket 0 should have 100 numUpdates");
assertEquals(BucketType.INSERT,
bucketInfoGetter.getBucketInfo(1).bucketType,
"Bucket 1 is INSERT");
assertEquals(BucketType.INSERT,
bucketInfoGetter.getBucketInfo(2).bucketType,
@@ -262,6 +271,8 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
"Bucket 0 is UPDATE");
assertEquals("2", bucketInfoGetter.getBucketInfo(0).fileIdPrefix,
"Should be assigned to only file id not pending compaction which
is 2");
+ assertEquals(0, bucketInfoGetter.getBucketInfo(0).getNumUpdates(),
+ "Small file UPDATE bucket with only inserts should have 0 numUpdates");
}
@Test
@@ -340,6 +351,8 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
"Bucket 0 should be UPDATE");
assertEquals("fg1", bucketInfoGetter.getBucketInfo(0).fileIdPrefix,
"Insert should be assigned to fg1");
+ assertEquals(0, bucketInfoGetter.getBucketInfo(0).getNumUpdates(),
+ "Small file UPDATE bucket with only inserts should have 0 numUpdates");
}
@Test
@@ -392,8 +405,8 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
@Test
void testMapAndListBasedSparkBucketInfoGetter() {
- List<BucketInfo> bucketInfos = Arrays.asList(new
BucketInfo(BucketType.UPDATE, "bucket1", "partition1"),
- new BucketInfo(BucketType.UPDATE, "bucket2", "partition2"));
+ List<BucketInfo> bucketInfos = Arrays.asList(new
BucketInfo(BucketType.UPDATE, "bucket1", "partition1", 42),
+ new BucketInfo(BucketType.UPDATE, "bucket2", "partition2", 99));
Map<Integer, BucketInfo> bucketInfoMap = new HashMap<>();
bucketInfoMap.put(0, bucketInfos.get(0));
bucketInfoMap.put(1, bucketInfos.get(1));
@@ -401,14 +414,18 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
ListBasedSparkBucketInfoGetter listGetter = new
ListBasedSparkBucketInfoGetter(bucketInfos);
assertEquals(bucketInfos.get(0), getter.getBucketInfo(0));
assertEquals(bucketInfos.get(0), listGetter.getBucketInfo(0));
+ assertEquals(42, getter.getBucketInfo(0).getNumUpdates());
+ assertEquals(42, listGetter.getBucketInfo(0).getNumUpdates());
assertEquals(bucketInfos.get(1), getter.getBucketInfo(1));
assertEquals(bucketInfos.get(1), listGetter.getBucketInfo(1));
+ assertEquals(99, getter.getBucketInfo(1).getNumUpdates());
+ assertEquals(99, listGetter.getBucketInfo(1).getNumUpdates());
}
@Test
void testInsertOverwriteBucketInfoGetter() {
BucketInfo insertInfo = new BucketInfo(BucketType.INSERT, "bucket1",
"partition1");
- BucketInfo updateInfo = new BucketInfo(BucketType.UPDATE, "bucket2",
"partition2");
+ BucketInfo updateInfo = new BucketInfo(BucketType.UPDATE, "bucket2",
"partition2", 75);
Map<Integer, BucketInfo> map = new HashMap<>();
map.put(0, insertInfo);
map.put(1, updateInfo);
@@ -420,5 +437,7 @@ public class TestUpsertPartitioner extends
HoodieClientTestBase {
assertEquals(BucketType.INSERT, result.getBucketType());
assertEquals(updateInfo.getPartitionPath(), result.getPartitionPath());
assertNotEquals(updateInfo.getFileIdPrefix(), result.getFileIdPrefix());
+ assertEquals(MergeContext.UNKNOWN_NUM_UPDATES, result.getNumUpdates(),
+ "insert-overwrite rebuilds the bucket as INSERT and drops the profiled
count");
}
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
index 03f9bf8f6743..49a8ba058a1e 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestUtils.java
@@ -61,6 +61,7 @@ import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -335,4 +336,13 @@ public class HoodieClientTestUtils {
return false;
}
}
+
+ /**
+ * Adapts a raw-typed record list, as produced by the test data generators,
to the typed
+ * iterator expected by {@code MergeContext} without a raw cast at every
call site.
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public static <T> Iterator<HoodieRecord<T>>
toTypedRecordIterator(List<HoodieRecord> records) {
+ return (Iterator) records.iterator();
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
index 2e1cee6b9f14..b47e474ac3a1 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
@@ -88,11 +88,13 @@ import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static
org.apache.hudi.common.table.cdc.HoodieCDCUtils.schemaBySupplementalLoggingMode;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.HOODIE_SCHEMA;
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static
org.apache.hudi.testutils.HoodieClientTestUtils.toTypedRecordIterator;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -155,13 +157,16 @@ public class TestMergeHandle extends BaseTestHandle {
}
assertTrue(numDeletes > 0);
HoodieWriteMergeHandle mergeHandle;
+ MergeContext mergeContext = MergeContext.create(newRecords.size(),
toTypedRecordIterator(newRecords));
if (useFileGroupReader) {
- mergeHandle = new FileGroupReaderBasedMergeHandle(config, instantTime,
table, newRecords.iterator(), partitionPath, fileId,
+ mergeHandle = new FileGroupReaderBasedMergeHandle(config, instantTime,
table, mergeContext, partitionPath, fileId,
new LocalTaskContextSupplier(), Option.empty());
} else {
- mergeHandle = new HoodieWriteMergeHandle(config, instantTime, table,
newRecords.iterator(), partitionPath, fileId, new LocalTaskContextSupplier(),
+ mergeHandle = new HoodieWriteMergeHandle(config, instantTime, table,
mergeContext, partitionPath, fileId, new LocalTaskContextSupplier(),
new HoodieBaseFile(fileGroup.getAllBaseFiles().findFirst().get()),
Option.empty());
}
+ assertEquals(newRecords.size(), mergeHandle.getNumUpdates(),
+ "Merge handle should receive the correct number of incoming updates
from the caller");
mergeHandle.doMerge();
WriteStatus writeStatus = (WriteStatus) mergeHandle.close().get(0);
// verify stats after merge
@@ -240,7 +245,7 @@ public class TestMergeHandle extends BaseTestHandle {
// Merge the UTF-8-larger key in directly via HoodieSortedMergeHandle.
List<HoodieRecord> newRecords =
withRowKey(dataGenerator.generateInserts("001", 1), supplementaryKey,
partitionPath);
HoodieSortedMergeHandle mergeHandle = new HoodieSortedMergeHandle(
- config, "001", table, newRecords.iterator(), partitionPath, fileId,
new LocalTaskContextSupplier(), Option.empty());
+ config, "001", table,
MergeContext.create(toTypedRecordIterator(newRecords)), partitionPath, fileId,
new LocalTaskContextSupplier(), Option.empty());
mergeHandle.doMerge();
WriteStatus writeStatus = (WriteStatus) mergeHandle.close().get(0);
@@ -251,6 +256,57 @@ public class TestMergeHandle extends BaseTestHandle {
assertEquals(Arrays.asList(bmpHighKey, supplementaryKey),
actualKeysInOrder);
}
+ @Test
+ public void testConcatHandleWritesIncomingRecordsAndCarriesNumUpdates()
throws Exception {
+ // HoodieConcatHandle splits the MergeContext: the parent gets an empty
iterator (so records
+ // are not merged by key) while the handle-local iterator concatenates the
incoming records
+ // after the existing ones. This test fails if either half of the context
is dropped.
+ // The factory-selected concat path is covered end-to-end by
testHoodieConcatHandleOnDupInserts
+ // (HoodieWriterClientTestHarness); this is the only direct-construction
concat coverage and the
+ // only assertion on numUpdates carriage through the re-wrapped context.
+ // delete and recreate
+ metaClient.getStorage().deleteDirectory(metaClient.getBasePath());
+ Properties properties = new Properties();
+ properties.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+ properties.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
"partition_path");
+ properties.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(),
ORDERING_FIELD);
+ initMetaClient(getTableType(), properties);
+
+ HoodieWriteConfig config = getHoodieWriteConfigBuilder().build();
+ HoodieSparkTable.create(config, new HoodieLocalEngineContext(storageConf),
metaClient);
+
+ String partitionPath = HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[0];
+ HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator(new
String[] {partitionPath});
+
+ List<HoodieRecord> baseRecords = dataGenerator.generateInserts("000", 3);
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+ String instantTime = client.startCommit();
+ JavaRDD<WriteStatus> statuses = client.upsert(jsc.parallelize(baseRecords,
1), instantTime);
+ client.commit(instantTime, statuses, Option.empty(), COMMIT_ACTION,
Collections.emptyMap(), Option.empty());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkCopyOnWriteTable table = (HoodieSparkCopyOnWriteTable)
HoodieSparkCopyOnWriteTable.create(config, context, metaClient);
+ HoodieFileGroup fileGroup =
table.getFileSystemView().getAllFileGroups(partitionPath).collect(Collectors.toList()).get(0);
+ String fileId = fileGroup.getFileGroupId().getFileId();
+
+ List<HoodieRecord> newRecords = dataGenerator.generateInserts("001", 2);
+ HoodieConcatHandle concatHandle = new HoodieConcatHandle(
+ config, "001", table, MergeContext.create(newRecords.size(),
toTypedRecordIterator(newRecords)),
+ partitionPath, fileId, new LocalTaskContextSupplier(), Option.empty());
+ assertEquals(newRecords.size(), concatHandle.getNumUpdates(),
+ "The concat handle should carry numUpdates from the merge context");
+ concatHandle.doMerge();
+ WriteStatus writeStatus = (WriteStatus) concatHandle.close().get(0);
+
+ String fullPath = metaClient.getBasePath() + "/" +
writeStatus.getStat().getPath();
+ List<GenericRecord> actualRecords = new
ParquetUtils().readAvroRecords(metaClient.getStorage(), new
StoragePath(fullPath));
+ List<String> actualKeys = actualRecords.stream().map(r ->
r.get("_row_key").toString()).sorted().collect(Collectors.toList());
+ List<String> expectedKeys = Stream.concat(baseRecords.stream(),
newRecords.stream())
+ .map(HoodieRecord::getRecordKey).sorted().collect(Collectors.toList());
+ assertEquals(expectedKeys, actualKeys,
+ "Concat must write the existing records plus every incoming record");
+ }
+
@Test
void testWriteFailures() throws Exception {
// delete and recreate
@@ -290,8 +346,8 @@ public class TestMergeHandle extends BaseTestHandle {
// can use either inline Avro CDC or native CDC depending on the effective
log format, but both paths must retract
// CDC and index callback state when the final file write fails for a
record.
FileGroupReaderBasedMergeHandle fileGroupReaderBasedMergeHandle = new
FileGroupReaderBasedMergeHandle(
- config, instantTime, table, updates.iterator(), partitionPath, fileId,
new LocalTaskContextSupplier(),
- Option.empty()) {
+ config, instantTime, table, MergeContext.create(updates.size(),
toTypedRecordIterator(updates)), partitionPath, fileId,
+ new LocalTaskContextSupplier(), Option.empty()) {
@Override
protected void writeToFile(HoodieKey key, HoodieRecord record,
HoodieSchema schema, Properties props,
boolean shouldPreserveRecordMetadata) throws
IOException {
@@ -403,8 +459,11 @@ public class TestMergeHandle extends BaseTestHandle {
}).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
FileGroupReaderBasedMergeHandle fileGroupReaderBasedMergeHandle = new
FileGroupReaderBasedMergeHandle(
- config, instantTime, table,
inputAndExpectedDataSet.getRecordsToMerge().iterator(), partitionPath, fileId,
new LocalTaskContextSupplier(),
- Option.empty());
+ config, instantTime, table,
+
MergeContext.create(inputAndExpectedDataSet.getRecordsToMerge().size(),
toTypedRecordIterator(inputAndExpectedDataSet.getRecordsToMerge())),
+ partitionPath, fileId, new LocalTaskContextSupplier(), Option.empty());
+ assertEquals(inputAndExpectedDataSet.getRecordsToMerge().size(),
fileGroupReaderBasedMergeHandle.getNumUpdates(),
+ "Merge handle should receive the correct numUpdates");
fileGroupReaderBasedMergeHandle.doMerge();
List<WriteStatus> writeStatuses = fileGroupReaderBasedMergeHandle.close();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
index e7a14b3c7928..846136bacff5 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
@@ -905,7 +905,8 @@ public class TestHoodieMergeOnReadTable extends
SparkClientFunctionalTestHarness
actionExecutor.getUpsertPartitioner(new
WorkloadProfile(buildProfile(deleteRDD)));
final List<List<WriteStatus>> deleteStatus =
jsc().parallelize(Arrays.asList(1))
.map(x -> (Iterator<List<WriteStatus>>)
- actionExecutor.handleUpdate(partitionPath, fileId,
fewRecordsForDelete.iterator()))
+ actionExecutor.handleUpdate(
+ partitionPath, fileId, fewRecordsForDelete.size(),
fewRecordsForDelete.iterator()))
.map(Transformations::flatten).collect();
// Verify there are errors because records are from multiple partitions
(but handleUpdate is invoked for
