You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2025/07/03 11:11:08 UTC
(hudi) branch master updated: [HUDI-9564] Introduce BufferedRecordMerger API in FileGroupRecordBuffer (#13511)
This is an automated email from the ASF dual-hosted git repository.
danny0405 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 8aa815ffd28e [HUDI-9564] Introduce BufferedRecordMerger API in FileGroupRecordBuffer (#13511)
8aa815ffd28e is described below
commit 8aa815ffd28e9dc7758fab3bde040ab1e2fcb37e
Author: Shuo Cheng <nj...@gmail.com>
AuthorDate: Thu Jul 3 19:10:59 2025 +0800
[HUDI-9564] Introduce BufferedRecordMerger API in FileGroupRecordBuffer (#13511)
---
.../common/table/read/BufferedRecordMerger.java | 67 ++++
.../table/read/BufferedRecordMergerFactory.java | 438 +++++++++++++++++++++
.../common/table/read/FileGroupRecordBuffer.java | 275 +------------
.../table/read/KeyBasedFileGroupRecordBuffer.java | 4 +-
.../read/PositionBasedFileGroupRecordBuffer.java | 4 +-
...er.java => TestCustomBufferedRecordMerger.java} | 4 +-
6 files changed, 519 insertions(+), 273 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
new file mode 100644
index 000000000000..18ac9c12a953
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMerger.java
@@ -0,0 +1,67 @@
+/*
+ * 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.common.table.read;
+
+import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+
+import java.io.IOException;
+
+/**
+ * A {@link BufferedRecord} merger that covers three core merging scenarios within {@link FileGroupRecordBuffer}:
+ *
+ * <ol>
+ * <li>log & log merging;</li>
+ * <li>log & delete record merging;</li>
+ * <li>base & log merging.</li>
+ * </ol>
+ */
+public interface BufferedRecordMerger<T> {
+
+ /**
+ * Merges incoming record from log file with existing record in record buffer.
+ *
+ * @param newRecord Incoming record from log file
+ * @param existingRecord Existing record in record buffer
+ *
+ * @return The merged record.
+ */
+ Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException;
+
+ /**
+ * Merges incoming delete record from log file with existing record in record buffer.
+ *
+ * @param deleteRecord Incoming delete record from log file
+ * @param existingRecord Existing record in record buffer
+ *
+ * @return The merged record.
+ */
+ Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord);
+
+ /**
+ * Merges newer record from log file with old record from base file.
+ *
+ * @param olderRecord Older record from base file
+ * @param newerRecord Newer record from log file
+ *
+ * @return The merged record.
+ */
+ Pair<Boolean, T> finalMerge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException;
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
new file mode 100644
index 000000000000..676fd03a7e07
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/BufferedRecordMergerFactory.java
@@ -0,0 +1,438 @@
+/*
+ * 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.common.table.read;
+
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
+import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
+import org.apache.hudi.common.util.HoodieRecordUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Pair;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Factory to create a {@link BufferedRecordMerger}.
+ */
+public class BufferedRecordMergerFactory {
+ public static <T> BufferedRecordMerger<T> create(HoodieReaderContext<T> readerContext,
+ RecordMergeMode recordMergeMode,
+ boolean enablePartialMerging,
+ Option<HoodieRecordMerger> recordMerger,
+ Option<String> orderingFieldName,
+ Option<String> payloadClass,
+ Schema readerSchema,
+ TypedProperties props) {
+ if (enablePartialMerging) {
+ BufferedRecordMerger<T> deleteRecordMerger = create(
+ readerContext, recordMergeMode, false, recordMerger, orderingFieldName, payloadClass, readerSchema, props);
+ return new PartialUpdateBufferedRecordMerger<>(readerContext, recordMerger, deleteRecordMerger, readerSchema, props);
+ }
+ switch (recordMergeMode) {
+ case COMMIT_TIME_ORDERING:
+ return new CommitTimeBufferedRecordMerger<>();
+ case EVENT_TIME_ORDERING:
+ return new EventTimeBufferedRecordMerger<>();
+ default:
+ if (payloadClass.isPresent()) {
+ return new CustomPayloadBufferedRecordMerger<>(readerContext, recordMerger, orderingFieldName, payloadClass.get(), readerSchema, props);
+ } else {
+ return new CustomBufferedRecordMerger<>(readerContext, recordMerger, readerSchema, props);
+ }
+ }
+ }
+
+ /**
+ * An implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on {@code COMMIT_TIME_ORDERING} merge mode.
+ */
+ private static class CommitTimeBufferedRecordMerger<T> implements BufferedRecordMerger<T> {
+ @Override
+ public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) {
+ return Option.of(newRecord);
+ }
+
+ @Override
+ public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
+ return Option.of(deleteRecord);
+ }
+
+ @Override
+ public Pair<Boolean, T> finalMerge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) {
+ return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
+ }
+ }
+
+ /**
+ * An implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on {@code EVENT_TIME_ORDERING} merge mode.
+ */
+ private static class EventTimeBufferedRecordMerger<T> implements BufferedRecordMerger<T> {
+ @Override
+ public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) {
+ if (existingRecord == null || shouldKeepNewerRecord(existingRecord, newRecord)) {
+ return Option.of(newRecord);
+ }
+ return Option.empty();
+ }
+
+ @Override
+ public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
+ return deltaMergeDeleteRecord(deleteRecord, existingRecord);
+ }
+
+ @Override
+ public Pair<Boolean, T> finalMerge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) {
+ if (newerRecord.isCommitTimeOrderingDelete()) {
+ return Pair.of(true, newerRecord.getRecord());
+ }
+ Comparable newOrderingValue = newerRecord.getOrderingValue();
+ Comparable oldOrderingValue = olderRecord.getOrderingValue();
+ if (!olderRecord.isCommitTimeOrderingDelete()
+ && oldOrderingValue.compareTo(newOrderingValue) > 0) {
+ return Pair.of(olderRecord.isDelete(), olderRecord.getRecord());
+ }
+ return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
+ }
+ }
+
+ /**
+ * An implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on partial update merging.
+ */
+ private static class PartialUpdateBufferedRecordMerger<T> implements BufferedRecordMerger<T> {
+ private final HoodieReaderContext<T> readerContext;
+ private final Option<HoodieRecordMerger> recordMerger;
+ private final BufferedRecordMerger<T> deleteRecordMerger;
+ private final Schema readerSchema;
+ private final TypedProperties props;
+
+ public PartialUpdateBufferedRecordMerger(
+ HoodieReaderContext<T> readerContext,
+ Option<HoodieRecordMerger> recordMerger,
+ BufferedRecordMerger<T> deleteRecordMerger,
+ Schema readerSchema,
+ TypedProperties props) {
+ this.readerContext = readerContext;
+ this.recordMerger = recordMerger;
+ this.deleteRecordMerger = deleteRecordMerger;
+ this.readerSchema = readerSchema;
+ this.props = props;
+ }
+
+ @Override
+ public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException {
+ if (existingRecord == null) {
+ return Option.of(newRecord);
+ }
+ // TODO(HUDI-7843): decouple the merging logic from the merger
+ // and use the record merge mode to control how to merge partial updates
+ // Merge and store the combined record
+ Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = recordMerger.get().partialMerge(
+ readerContext.constructHoodieRecord(existingRecord),
+ readerContext.getSchemaFromBufferRecord(existingRecord),
+ readerContext.constructHoodieRecord(newRecord),
+ readerContext.getSchemaFromBufferRecord(newRecord),
+ readerSchema,
+ props);
+ if (!combinedRecordAndSchemaOpt.isPresent()) {
+ return Option.empty();
+ }
+ Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
+ HoodieRecord<T> combinedRecord = combinedRecordAndSchema.getLeft();
+
+ // If pre-combine returns existing record, no need to update it
+ if (combinedRecord.getData() != existingRecord.getRecord()) {
+ return Option.of(BufferedRecord.forRecordWithContext(combinedRecord, combinedRecordAndSchema.getRight(), readerContext, props));
+ }
+ return Option.empty();
+ }
+
+ @Override
+ public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
+ return this.deleteRecordMerger.deltaMerge(deleteRecord, existingRecord);
+ }
+
+ @Override
+ public Pair<Boolean, T> finalMerge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
+ // TODO(HUDI-7843): decouple the merging logic from the merger
+ // and use the record merge mode to control how to merge partial updates
+ Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().partialMerge(
+ readerContext.constructHoodieRecord(olderRecord), readerContext.getSchemaFromBufferRecord(olderRecord),
+ readerContext.constructHoodieRecord(newerRecord), readerContext.getSchemaFromBufferRecord(newerRecord),
+ readerSchema, props);
+
+ if (mergedRecord.isPresent()
+ && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
+ HoodieRecord hoodieRecord = mergedRecord.get().getLeft();
+ if (!mergedRecord.get().getRight().equals(readerSchema)) {
+ T data = (T) hoodieRecord.rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData();
+ return Pair.of(false, data);
+ }
+ return Pair.of(false, (T) hoodieRecord.getData());
+ }
+ return Pair.of(true, null);
+ }
+ }
+
+ /**
+ * An implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on {@code CUSTOM} merge mode.
+ */
+ private static class CustomBufferedRecordMerger<T> extends BaseCustomMerger<T> {
+ public CustomBufferedRecordMerger(
+ HoodieReaderContext<T> readerContext,
+ Option<HoodieRecordMerger> recordMerger,
+ Schema readerSchema,
+ TypedProperties props) {
+ super(readerContext, recordMerger, readerSchema, props);
+ }
+
+ @Override
+ public Option<BufferedRecord<T>> deltaMergeNonDeleteRecord(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException {
+ Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = recordMerger.get().merge(
+ readerContext.constructHoodieRecord(existingRecord),
+ readerContext.getSchemaFromBufferRecord(existingRecord),
+ readerContext.constructHoodieRecord(newRecord),
+ readerContext.getSchemaFromBufferRecord(newRecord),
+ props);
+
+ if (!combinedRecordAndSchemaOpt.isPresent()) {
+ return Option.empty();
+ }
+
+ Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
+ HoodieRecord<T> combinedRecord = combinedRecordAndSchema.getLeft();
+
+ // If pre-combine returns existing record, no need to update it
+ if (combinedRecord.getData() != existingRecord.getRecord()) {
+ return Option.of(BufferedRecord.forRecordWithContext(combinedRecord, combinedRecordAndSchema.getRight(), readerContext, props));
+ }
+ return Option.empty();
+ }
+
+ @Override
+ public Pair<Boolean, T> mergeNonDeleteRecord(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
+ Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().merge(
+ readerContext.constructHoodieRecord(olderRecord), readerContext.getSchemaFromBufferRecord(olderRecord),
+ readerContext.constructHoodieRecord(newerRecord), readerContext.getSchemaFromBufferRecord(newerRecord), props);
+ if (mergedRecord.isPresent()
+ && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
+ HoodieRecord hoodieRecord = mergedRecord.get().getLeft();
+ if (!mergedRecord.get().getRight().equals(readerSchema)) {
+ return Pair.of(false, (T) hoodieRecord.rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData());
+ }
+ return Pair.of(false, (T) hoodieRecord.getData());
+ }
+ return Pair.of(true, null);
+ }
+ }
+
+ /**
+ * An implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on {@code CUSTOM} merge mode and a given record payload class.
+ */
+ private static class CustomPayloadBufferedRecordMerger<T> extends BaseCustomMerger<T> {
+ private final Option<String> orderingFieldName;
+ private final String payloadClass;
+
+ public CustomPayloadBufferedRecordMerger(
+ HoodieReaderContext<T> readerContext,
+ Option<HoodieRecordMerger> recordMerger,
+ Option<String> orderingFieldName,
+ String payloadClass,
+ Schema readerSchema,
+ TypedProperties props) {
+ super(readerContext, recordMerger, readerSchema, props);
+ this.orderingFieldName = orderingFieldName;
+ this.payloadClass = payloadClass;
+ }
+
+ @Override
+ public Option<BufferedRecord<T>> deltaMergeNonDeleteRecord(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException {
+ Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = getMergedRecord(existingRecord, newRecord);
+ if (combinedRecordAndSchemaOpt.isPresent()) {
+ T combinedRecordData = readerContext.convertAvroRecord((IndexedRecord) combinedRecordAndSchemaOpt.get().getLeft().getData());
+ // If pre-combine does not return existing record, update it
+ if (combinedRecordData != existingRecord.getRecord()) {
+ Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
+ return Option.of(BufferedRecord.forRecordWithContext(combinedRecordData, combinedRecordAndSchema.getRight(), readerContext, orderingFieldName, false));
+ }
+ }
+ return Option.empty();
+ }
+
+ @Override
+ public Pair<Boolean, T> mergeNonDeleteRecord(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
+ Option<Pair<HoodieRecord, Schema>> mergedRecord =
+ getMergedRecord(olderRecord, newerRecord);
+ if (mergedRecord.isPresent()
+ && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
+ IndexedRecord indexedRecord;
+ if (!mergedRecord.get().getRight().equals(readerSchema)) {
+ indexedRecord = (IndexedRecord) mergedRecord.get().getLeft().rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData();
+ } else {
+ indexedRecord = (IndexedRecord) mergedRecord.get().getLeft().getData();
+ }
+ return Pair.of(false, readerContext.convertAvroRecord(indexedRecord));
+ }
+ return Pair.of(true, null);
+ }
+
+ private Option<Pair<HoodieRecord, Schema>> getMergedRecord(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
+ ValidationUtils.checkArgument(!Objects.equals(payloadClass, OverwriteWithLatestAvroPayload.class.getCanonicalName())
+ && !Objects.equals(payloadClass, DefaultHoodieRecordPayload.class.getCanonicalName()));
+ HoodieRecord oldHoodieRecord = constructHoodieAvroRecord(readerContext, olderRecord);
+ HoodieRecord newHoodieRecord = constructHoodieAvroRecord(readerContext, newerRecord);
+ Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().merge(
+ oldHoodieRecord, getSchemaForAvroPayloadMerge(oldHoodieRecord, olderRecord),
+ newHoodieRecord, getSchemaForAvroPayloadMerge(newHoodieRecord, newerRecord), props);
+ return mergedRecord;
+ }
+
+ private HoodieRecord constructHoodieAvroRecord(HoodieReaderContext<T> readerContext, BufferedRecord<T> bufferedRecord) {
+ GenericRecord record = null;
+ if (!bufferedRecord.isDelete()) {
+ Schema recordSchema = readerContext.getSchemaFromBufferRecord(bufferedRecord);
+ record = readerContext.convertToAvroRecord(bufferedRecord.getRecord(), recordSchema);
+ }
+ HoodieKey hoodieKey = new HoodieKey(bufferedRecord.getRecordKey(), null);
+ return new HoodieAvroRecord<>(hoodieKey,
+ HoodieRecordUtils.loadPayload(payloadClass, record, bufferedRecord.getOrderingValue()), null);
+ }
+
+ private Schema getSchemaForAvroPayloadMerge(HoodieRecord record, BufferedRecord<T> bufferedRecord) throws IOException {
+ if (record.isDelete(readerSchema, props)) {
+ return readerSchema;
+ }
+ return readerContext.getSchemaFromBufferRecord(bufferedRecord);
+ }
+ }
+
+ /**
+ * A base implementation of {@link BufferedRecordMerger} which merges {@link BufferedRecord}s
+ * based on {@code CUSTOM} merge mode.
+ */
+ private abstract static class BaseCustomMerger<T> implements BufferedRecordMerger<T> {
+ protected final HoodieReaderContext<T> readerContext;
+ protected final Option<HoodieRecordMerger> recordMerger;
+ protected final Schema readerSchema;
+ protected final TypedProperties props;
+
+ public BaseCustomMerger(
+ HoodieReaderContext<T> readerContext,
+ Option<HoodieRecordMerger> recordMerger,
+ Schema readerSchema,
+ TypedProperties props) {
+ this.readerContext = readerContext;
+ this.recordMerger = recordMerger;
+ this.readerSchema = readerSchema;
+ this.props = props;
+ }
+
+ @Override
+ public Option<BufferedRecord<T>> deltaMerge(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException {
+ if (existingRecord == null) {
+ return Option.of(newRecord);
+ }
+ if (existingRecord.isDelete() || newRecord.isDelete()) {
+ if (shouldKeepNewerRecord(existingRecord, newRecord)) {
+ // IMPORTANT:
+ // this is needed when the fallback HoodieAvroRecordMerger got used, the merger would
+ // return Option.empty when the old payload data is empty(a delete) and ignores its ordering value directly.
+ return Option.of(newRecord);
+ } else {
+ return Option.empty();
+ }
+ }
+ return deltaMergeNonDeleteRecord(newRecord, existingRecord);
+ }
+
+ public abstract Option<BufferedRecord<T>> deltaMergeNonDeleteRecord(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord) throws IOException;
+
+ @Override
+ public Option<DeleteRecord> deltaMerge(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
+ return deltaMergeDeleteRecord(deleteRecord, existingRecord);
+ }
+
+ @Override
+ public Pair<Boolean, T> finalMerge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
+ if (olderRecord.isDelete() || newerRecord.isDelete()) {
+ if (shouldKeepNewerRecord(olderRecord, newerRecord)) {
+ // IMPORTANT:
+ // this is needed when the fallback HoodieAvroRecordMerger got used, the merger would
+ // return Option.empty when the new payload data is empty(a delete) and ignores its ordering value directly.
+ return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
+ } else {
+ return Pair.of(olderRecord.isDelete(), olderRecord.getRecord());
+ }
+ }
+ return mergeNonDeleteRecord(olderRecord, newerRecord);
+ }
+
+ public abstract Pair<Boolean, T> mergeNonDeleteRecord(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException;
+ }
+
+ // -------------------------------------------------------------------------
+ // Utilities
+ // -------------------------------------------------------------------------
+
+ private static <T> Option<DeleteRecord> deltaMergeDeleteRecord(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
+ if (existingRecord == null) {
+ return Option.of(deleteRecord);
+ }
+ if (existingRecord.isCommitTimeOrderingDelete()) {
+ return Option.empty();
+ }
+ Comparable existingOrderingVal = existingRecord.getOrderingValue();
+ Comparable deleteOrderingVal = deleteRecord.getOrderingValue();
+ // Checks the ordering value does not equal to 0
+ // because we use 0 as the default value which means natural order
+ boolean chooseExisting = !deleteOrderingVal.equals(0)
+ && ReflectionUtils.isSameClass(existingOrderingVal, deleteOrderingVal)
+ && existingOrderingVal.compareTo(deleteOrderingVal) > 0;
+ if (chooseExisting) {
+ // The DELETE message is obsolete if the old message has greater orderingVal.
+ return Option.empty();
+ }
+ return Option.of(deleteRecord);
+ }
+
+ private static <T> boolean shouldKeepNewerRecord(BufferedRecord<T> oldRecord, BufferedRecord<T> newRecord) {
+ if (newRecord.isCommitTimeOrderingDelete()) {
+ // handle records coming from DELETE statements(the orderingVal is constant 0)
+ return true;
+ }
+ return newRecord.getOrderingValue().compareTo(oldRecord.getOrderingValue()) >= 0;
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupRecordBuffer.java
index 2fd2f32e3153..4ee26efa1604 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupRecordBuffer.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/FileGroupRecordBuffer.java
@@ -23,26 +23,18 @@ import org.apache.hudi.avro.AvroSchemaCache;
import org.apache.hudi.common.config.RecordMergeMode;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieReaderContext;
-import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
import org.apache.hudi.common.model.DeleteRecord;
-import org.apache.hudi.common.model.HoodieAvroRecord;
-import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieOperation;
import org.apache.hudi.common.model.HoodiePayloadProps;
-import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordMerger;
-import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.KeySpec;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.FileIOUtils;
-import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.InternalSchemaCache;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.CloseableMappingIterator;
import org.apache.hudi.common.util.collection.ExternalSpillableMap;
@@ -53,15 +45,12 @@ import org.apache.hudi.internal.schema.action.InternalSchemaMerger;
import org.apache.hudi.internal.schema.convert.AvroInternalSchemaConverter;
import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.generic.IndexedRecord;
import java.io.IOException;
import java.io.Serializable;
import java.util.Iterator;
import java.util.Locale;
import java.util.Map;
-import java.util.Objects;
import java.util.function.Function;
import static org.apache.hudi.common.config.HoodieCommonConfig.DISK_MAP_BITCASK_COMPRESSION_ENABLED;
@@ -70,7 +59,6 @@ import static org.apache.hudi.common.config.HoodieMemoryConfig.MAX_MEMORY_FOR_ME
import static org.apache.hudi.common.config.HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH;
import static org.apache.hudi.common.model.HoodieRecord.DEFAULT_ORDERING_VALUE;
import static org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
-import static org.apache.hudi.common.model.HoodieRecord.OPERATION_METADATA_FIELD;
import static org.apache.hudi.common.model.HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID;
import static org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType.INSTANT_TIME;
@@ -93,6 +81,7 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
protected boolean enablePartialMerging = false;
protected InternalSchema internalSchema;
protected HoodieTableMetaClient hoodieTableMetaClient;
+ protected BufferedRecordMerger<T> bufferedRecordMerger;
private long totalLogRecords = 0;
protected FileGroupRecordBuffer(HoodieReaderContext<T> readerContext,
@@ -141,6 +130,8 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
readerContext.getSchemaHandler().getCustomDeleteMarkerKeyValue().isPresent();
this.shouldCheckBuiltInDeleteMarker =
readerContext.getSchemaHandler().hasBuiltInDelete();
+ this.bufferedRecordMerger = BufferedRecordMergerFactory.create(
+ readerContext, recordMergeMode, enablePartialMerging, recordMerger, orderingFieldName, payloadClass, readerSchema, props);
}
/**
@@ -243,92 +234,7 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
protected Option<BufferedRecord<T>> doProcessNextDataRecord(BufferedRecord<T> newRecord, BufferedRecord<T> existingRecord)
throws IOException {
totalLogRecords++;
- if (existingRecord != null) {
- if (enablePartialMerging) {
- // TODO(HUDI-7843): decouple the merging logic from the merger
- // and use the record merge mode to control how to merge partial updates
- // Merge and store the combined record
- Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = recordMerger.get().partialMerge(
- readerContext.constructHoodieRecord(existingRecord),
- readerContext.getSchemaFromBufferRecord(existingRecord),
- readerContext.constructHoodieRecord(newRecord),
- readerContext.getSchemaFromBufferRecord(newRecord),
- readerSchema,
- props);
- if (!combinedRecordAndSchemaOpt.isPresent()) {
- return Option.empty();
- }
- Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
- HoodieRecord<T> combinedRecord = combinedRecordAndSchema.getLeft();
-
- // If pre-combine returns existing record, no need to update it
- if (combinedRecord.getData() != existingRecord.getRecord()) {
- return Option.of(BufferedRecord.forRecordWithContext(combinedRecord, combinedRecordAndSchema.getRight(), readerContext, props));
- }
- return Option.empty();
- } else {
- switch (recordMergeMode) {
- case COMMIT_TIME_ORDERING:
- return Option.of(newRecord);
- case EVENT_TIME_ORDERING:
- if (shouldKeepNewerRecord(existingRecord, newRecord)) {
- return Option.of(newRecord);
- }
- return Option.empty();
- case CUSTOM:
- default:
- if (existingRecord.isDelete() || newRecord.isDelete()) {
- if (shouldKeepNewerRecord(existingRecord, newRecord)) {
- // IMPORTANT:
- // this is needed when the fallback HoodieAvroRecordMerger got used, the merger would
- // return Option.empty when the old payload data is empty(a delete) and ignores its ordering value directly.
- return Option.of(newRecord);
- } else {
- return Option.empty();
- }
- }
- // Merge and store the combined record
- if (payloadClass.isPresent()) {
- Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = getMergedRecord(existingRecord, newRecord);
- if (combinedRecordAndSchemaOpt.isPresent()) {
- T combinedRecordData = readerContext.convertAvroRecord((IndexedRecord) combinedRecordAndSchemaOpt.get().getLeft().getData());
- // If pre-combine does not return existing record, update it
- if (combinedRecordData != existingRecord.getRecord()) {
- Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
- return Option.of(BufferedRecord.forRecordWithContext(combinedRecordData, combinedRecordAndSchema.getRight(), readerContext, orderingFieldName, false));
- }
- }
- return Option.empty();
- } else {
- Option<Pair<HoodieRecord, Schema>> combinedRecordAndSchemaOpt = recordMerger.get().merge(
- readerContext.constructHoodieRecord(existingRecord),
- readerContext.getSchemaFromBufferRecord(existingRecord),
- readerContext.constructHoodieRecord(newRecord),
- readerContext.getSchemaFromBufferRecord(newRecord),
- props);
-
- if (!combinedRecordAndSchemaOpt.isPresent()) {
- return Option.empty();
- }
-
- Pair<HoodieRecord, Schema> combinedRecordAndSchema = combinedRecordAndSchemaOpt.get();
- HoodieRecord<T> combinedRecord = combinedRecordAndSchema.getLeft();
-
- // If pre-combine returns existing record, no need to update it
- if (combinedRecord.getData() != existingRecord.getRecord()) {
- return Option.of(BufferedRecord.forRecordWithContext(combinedRecord, combinedRecordAndSchema.getRight(), readerContext, props));
- }
- return Option.empty();
- }
- }
- }
- } else {
- // Put the record as is
- // NOTE: Record have to be cloned here to make sure if it holds low-level engine-specific
- // payload pointing into a shared, mutable (underlying) buffer we get a clean copy of
- // it since these records will be put into records(Map).
- return Option.of(newRecord);
- }
+ return bufferedRecordMerger.deltaMerge(newRecord, existingRecord);
}
/**
@@ -341,31 +247,7 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
*/
protected Option<DeleteRecord> doProcessNextDeletedRecord(DeleteRecord deleteRecord, BufferedRecord<T> existingRecord) {
totalLogRecords++;
- if (existingRecord != null) {
- switch (recordMergeMode) {
- case COMMIT_TIME_ORDERING:
- return Option.of(deleteRecord);
- case EVENT_TIME_ORDERING:
- case CUSTOM:
- default:
- if (existingRecord.isCommitTimeOrderingDelete()) {
- return Option.empty();
- }
- Comparable existingOrderingVal = existingRecord.getOrderingValue();
- Comparable deleteOrderingVal = deleteRecord.getOrderingValue();
- // Checks the ordering value does not equal to 0
- // because we use 0 as the default value which means natural order
- boolean chooseExisting = !deleteOrderingVal.equals(0)
- && ReflectionUtils.isSameClass(existingOrderingVal, deleteOrderingVal)
- && existingOrderingVal.compareTo(deleteOrderingVal) > 0;
- if (chooseExisting) {
- // The DELETE message is obsolete if the old message has greater orderingVal.
- return Option.empty();
- }
- }
- }
- // Do delete.
- return Option.of(deleteRecord);
+ return bufferedRecordMerger.deltaMerge(deleteRecord, existingRecord);
}
/**
@@ -425,138 +307,7 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
* @throws IOException
*/
protected Pair<Boolean, T> merge(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
- if (enablePartialMerging) {
- // TODO(HUDI-7843): decouple the merging logic from the merger
- // and use the record merge mode to control how to merge partial updates
- Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().partialMerge(
- readerContext.constructHoodieRecord(olderRecord), readerContext.getSchemaFromBufferRecord(olderRecord),
- readerContext.constructHoodieRecord(newerRecord), readerContext.getSchemaFromBufferRecord(newerRecord),
- readerSchema, props);
-
- if (mergedRecord.isPresent()
- && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
- HoodieRecord hoodieRecord = mergedRecord.get().getLeft();
- if (!mergedRecord.get().getRight().equals(readerSchema)) {
- T data = (T) hoodieRecord.rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData();
- return Pair.of(false, data);
- }
- return Pair.of(false, (T) hoodieRecord.getData());
- }
- return Pair.of(true, null);
- } else {
- switch (recordMergeMode) {
- case COMMIT_TIME_ORDERING:
- return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
- case EVENT_TIME_ORDERING:
- if (newerRecord.isCommitTimeOrderingDelete()) {
- return Pair.of(true, newerRecord.getRecord());
- }
- Comparable newOrderingValue = newerRecord.getOrderingValue();
- Comparable oldOrderingValue = olderRecord.getOrderingValue();
- if (!olderRecord.isCommitTimeOrderingDelete()
- && oldOrderingValue.compareTo(newOrderingValue) > 0) {
- return Pair.of(olderRecord.isDelete(), olderRecord.getRecord());
- }
- return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
- case CUSTOM:
- default:
- if (payloadClass.isPresent()) {
- if (olderRecord.isDelete() || newerRecord.isDelete()) {
- if (shouldKeepNewerRecord(olderRecord, newerRecord)) {
- // IMPORTANT:
- // this is needed when the fallback HoodieAvroRecordMerger got used, the merger would
- // return Option.empty when the new payload data is empty(a delete) and ignores its ordering value directly.
- return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
- } else {
- return Pair.of(olderRecord.isDelete(), olderRecord.getRecord());
- }
- }
- Option<Pair<HoodieRecord, Schema>> mergedRecord =
- getMergedRecord(olderRecord, newerRecord);
- if (mergedRecord.isPresent()
- && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
- IndexedRecord indexedRecord;
- if (!mergedRecord.get().getRight().equals(readerSchema)) {
- indexedRecord = (IndexedRecord) mergedRecord.get().getLeft().rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData();
- } else {
- indexedRecord = (IndexedRecord) mergedRecord.get().getLeft().getData();
- }
- return Pair.of(false, readerContext.convertAvroRecord(indexedRecord));
- }
- return Pair.of(true, null);
- } else {
- if (olderRecord.isDelete() || newerRecord.isDelete()) {
- if (shouldKeepNewerRecord(olderRecord, newerRecord)) {
- // IMPORTANT:
- // this is needed when the fallback HoodieAvroRecordMerger got used, the merger would
- // return Option.empty when the new payload data is empty(a delete) and ignores its ordering value directly.
- return Pair.of(newerRecord.isDelete(), newerRecord.getRecord());
- } else {
- return Pair.of(olderRecord.isDelete(), olderRecord.getRecord());
- }
- }
- Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().merge(
- readerContext.constructHoodieRecord(olderRecord), readerContext.getSchemaFromBufferRecord(olderRecord),
- readerContext.constructHoodieRecord(newerRecord), readerContext.getSchemaFromBufferRecord(newerRecord), props);
- if (mergedRecord.isPresent()
- && !mergedRecord.get().getLeft().isDelete(mergedRecord.get().getRight(), props)) {
- HoodieRecord hoodieRecord = mergedRecord.get().getLeft();
- if (!mergedRecord.get().getRight().equals(readerSchema)) {
- return Pair.of(false, (T) hoodieRecord.rewriteRecordWithNewSchema(mergedRecord.get().getRight(), null, readerSchema).getData());
- }
- return Pair.of(false, (T) hoodieRecord.getData());
- }
- return Pair.of(true, null);
- }
- }
- }
- }
-
- /**
- * Decides whether to keep the incoming record with ordering value comparison.
- */
- private boolean shouldKeepNewerRecord(BufferedRecord<T> oldRecord, BufferedRecord<T> newRecord) {
- if (newRecord.isCommitTimeOrderingDelete()) {
- // handle records coming from DELETE statements(the orderingVal is constant 0)
- return true;
- }
- return newRecord.getOrderingValue().compareTo(oldRecord.getOrderingValue()) >= 0;
- }
-
- private Option<Pair<HoodieRecord, Schema>> getMergedRecord(BufferedRecord<T> olderRecord, BufferedRecord<T> newerRecord) throws IOException {
- ValidationUtils.checkArgument(!Objects.equals(payloadClass, OverwriteWithLatestAvroPayload.class.getCanonicalName())
- && !Objects.equals(payloadClass, DefaultHoodieRecordPayload.class.getCanonicalName()));
- HoodieRecord oldHoodieRecord = constructHoodieAvroRecord(readerContext, olderRecord);
- HoodieRecord newHoodieRecord = constructHoodieAvroRecord(readerContext, newerRecord);
- Option<Pair<HoodieRecord, Schema>> mergedRecord = recordMerger.get().merge(
- oldHoodieRecord, getSchemaForAvroPayloadMerge(oldHoodieRecord, olderRecord),
- newHoodieRecord, getSchemaForAvroPayloadMerge(newHoodieRecord, newerRecord), props);
- return mergedRecord;
- }
-
- /**
- * Constructs a new {@link HoodieAvroRecord} for payload based merging
- *
- * @param readerContext reader context
- * @param bufferedRecord buffered record
- * @return A new instance of {@link HoodieRecord}.
- */
- private HoodieRecord constructHoodieAvroRecord(HoodieReaderContext<T> readerContext, BufferedRecord<T> bufferedRecord) {
- GenericRecord record = null;
- if (!bufferedRecord.isDelete()) {
- Schema recordSchema = readerContext.getSchemaFromBufferRecord(bufferedRecord);
- record = readerContext.convertToAvroRecord(bufferedRecord.getRecord(), recordSchema);
- }
- HoodieKey hoodieKey = new HoodieKey(bufferedRecord.getRecordKey(), null);
- return new HoodieAvroRecord<>(hoodieKey,
- HoodieRecordUtils.loadPayload(payloadClass.get(), record, bufferedRecord.getOrderingValue()), null);
- }
-
- private Schema getSchemaForAvroPayloadMerge(HoodieRecord record, BufferedRecord<T> bufferedRecord) throws IOException {
- if (record.isDelete(readerSchema, props)) {
- return readerSchema;
- }
- return readerContext.getSchemaFromBufferRecord(bufferedRecord);
+ return bufferedRecordMerger.finalMerge(olderRecord, newerRecord);
}
protected boolean hasNextBaseRecord(T baseRecord, BufferedRecord<T> logRecordInfo) throws IOException {
@@ -639,18 +390,4 @@ public abstract class FileGroupRecordBuffer<T> implements HoodieFileGroupRecordB
? DEFAULT_ORDERING_VALUE
: readerContext.convertValueToEngineType(deleteRecord.getOrderingValue());
}
-
- private boolean isDeleteRecord(Option<T> record, Schema schema) {
- if (record.isEmpty()) {
- return true;
- }
-
- Object operation = readerContext.getValue(record.get(), schema, OPERATION_METADATA_FIELD);
- if (operation != null && HoodieOperation.isDeleteRecord(operation.toString())) {
- return true;
- }
-
- Object deleteMarker = readerContext.getValue(record.get(), schema, HOODIE_IS_DELETED_FIELD);
- return deleteMarker instanceof Boolean && (boolean) deleteMarker;
- }
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java
index 5bbce2ad2567..e7738bab6897 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/KeyBasedFileGroupRecordBuffer.java
@@ -67,10 +67,12 @@ public class KeyBasedFileGroupRecordBuffer<T> extends FileGroupRecordBuffer<T> {
public void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecOpt) throws IOException {
Pair<ClosableIterator<T>, Schema> recordsIteratorSchemaPair =
getRecordsIterator(dataBlock, keySpecOpt);
- if (dataBlock.containsPartialUpdates()) {
+ if (dataBlock.containsPartialUpdates() && !enablePartialMerging) {
// When a data block contains partial updates, subsequent record merging must always use
// partial merging.
enablePartialMerging = true;
+ bufferedRecordMerger = BufferedRecordMergerFactory.create(
+ readerContext, recordMergeMode, true, recordMerger, orderingFieldName, payloadClass, readerSchema, props);
}
Schema schema = AvroSchemaCache.intern(recordsIteratorSchemaPair.getRight());
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java b/hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java
index 152b7107b9b5..c08c9bc7788b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/read/PositionBasedFileGroupRecordBuffer.java
@@ -106,10 +106,12 @@ public class PositionBasedFileGroupRecordBuffer<T> extends KeyBasedFileGroupReco
isFullKey = keySpecOpt.get().isFullKey();
}
- if (dataBlock.containsPartialUpdates()) {
+ if (dataBlock.containsPartialUpdates() && !enablePartialMerging) {
// When a data block contains partial updates, subsequent record merging must always use
// partial merging.
enablePartialMerging = true;
+ bufferedRecordMerger = BufferedRecordMergerFactory.create(
+ readerContext, recordMergeMode, true, recordMerger, orderingFieldName, payloadClass, readerSchema, props);
}
Pair<Function<T, T>, Schema> schemaTransformerWithEvolvedSchema = getSchemaTransformerWithEvolvedSchema(dataBlock);
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomMerger.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomBufferedRecordMerger.java
similarity index 98%
rename from hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomMerger.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomBufferedRecordMerger.java
index 8a8d0f363ff7..a8eaec69251a 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomMerger.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestCustomBufferedRecordMerger.java
@@ -60,7 +60,7 @@ import static org.apache.hudi.common.testutils.reader.DataGenerationPlan.Operati
import static org.apache.hudi.common.testutils.reader.HoodieFileSliceTestUtils.ROW_KEY;
import static org.junit.jupiter.api.Assertions.assertEquals;
-public class TestCustomMerger extends HoodieFileGroupReaderTestHarness {
+public class TestCustomBufferedRecordMerger extends HoodieFileGroupReaderTestHarness {
@Override
protected Properties getMetaProps() {
@@ -122,7 +122,7 @@ public class TestCustomMerger extends HoodieFileGroupReaderTestHarness {
@BeforeEach
public void initialize() throws Exception {
- setTableName(TestCustomMerger.class.getName());
+ setTableName(TestCustomBufferedRecordMerger.class.getName());
initPath(tableName);
initMetaClient();
initTestDataGenerator(new String[]{PARTITION_PATH});