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 29b42cbae382 refactor(reader): seal the shared merge path in
KeyBasedFileGroupRecordBuffer (#19415)
29b42cbae382 is described below
commit 29b42cbae382a27a157d46c1c10da382ac7c9cac
Author: Ranga Reddy <[email protected]>
AuthorDate: Fri Jul 31 15:37:05 2026 +0530
refactor(reader): seal the shared merge path in
KeyBasedFileGroupRecordBuffer (#19415)
* refactor(reader): seal the shared merge path in
KeyBasedFileGroupRecordBuffer
Mark processNextDataRecord and isPartialMergingEnabled as final so that
subclasses cannot replace the merge-then-store behavior shared by every
buffer in this hierarchy. Subclasses still choose the identifier a record
is buffered under, and still specialize block processing and base-record
advancement.
Adds a reflection-based regression test so the modifiers cannot be dropped
silently.
Closes #16920
* docs(reader): scope the sealed-method claim to what final actually
guarantees
Review feedback: the test name, javadoc and failure message claimed every
buffer funnels records through a single merge-then-put path. That is not
true
and the test did not check it. PositionBasedFileGroupRecordBuffer writes to
the protected records map directly when it re-keys entries in
fallbackToKeyBasedBuffer and when it overwrites delete markers under
COMMIT_TIME_ORDERING, and subclasses can still override block processing.
Narrow the claim instead of inventing the invariant: those two paths must
not
merge, so routing them through processNextDataRecord would change behaviour.
Renamed the test to sealedMethodsCannotBeOverridden, reworded its javadoc
and
message to the exact guarantee (the two methods keep their final modifier),
and corrected the same overstatement in the production javadoc, including
that
a subclass may still set enablePartialMerging while processing a data block.
No behaviour change.
---
.../read/buffer/KeyBasedFileGroupRecordBuffer.java | 20 ++++++++++++--
.../buffer/TestKeyBasedFileGroupRecordBuffer.java | 31 ++++++++++++++++++++++
2 files changed, 49 insertions(+), 2 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
index f40e5206e946..0238a7ed107c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/KeyBasedFileGroupRecordBuffer.java
@@ -98,8 +98,19 @@ public class KeyBasedFileGroupRecordBuffer<T> extends
FileGroupRecordBuffer<T> {
}
}
+ /**
+ * Merges the incoming record with whatever is already buffered under {@code
recordKey} and stores the
+ * result. Subclasses choose the identifier a record is buffered under (a
record key here, a record
+ * position in {@link PositionBasedFileGroupRecordBuffer}), but cannot
replace this merge-then-store
+ * step itself, hence the {@code final}.
+ *
+ * <p>This is not the only way {@code records} is mutated. The map is {@code
protected}, and
+ * {@link PositionBasedFileGroupRecordBuffer} writes to it directly where
merging would be wrong: when
+ * re-keying already-merged entries in its key-based fallback, and when
overwriting delete markers under
+ * commit-time ordering.
+ */
@Override
- public void processNextDataRecord(BufferedRecord<T> record, Serializable
recordKey) throws IOException {
+ public final void processNextDataRecord(BufferedRecord<T> record,
Serializable recordKey) throws IOException {
BufferedRecord<T> existingRecord = records.get(recordKey);
totalLogRecords++;
bufferedRecordMerger.deltaMerge(record,
existingRecord).ifPresent(bufferedRecord ->
@@ -139,7 +150,12 @@ public class KeyBasedFileGroupRecordBuffer<T> extends
FileGroupRecordBuffer<T> {
return hasNextLogRecord();
}
- public boolean isPartialMergingEnabled() {
+ /**
+ * Whether partial merging has been switched on for this buffer. Subclasses
cannot replace this accessor,
+ * hence the {@code final}; they may still set the underlying {@code
enablePartialMerging} flag while
+ * processing a data block, as {@link PositionBasedFileGroupRecordBuffer}
does.
+ */
+ public final boolean isPartialMergingEnabled() {
return enablePartialMerging;
}
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
index 560e0020228b..b2138b9a495c 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestKeyBasedFileGroupRecordBuffer.java
@@ -31,6 +31,7 @@ import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
+import org.apache.hudi.common.table.read.BufferedRecord;
import org.apache.hudi.common.table.read.FileGroupReaderSchemaHandler;
import org.apache.hudi.common.table.read.HoodieReadStats;
import org.apache.hudi.common.util.Option;
@@ -42,6 +43,9 @@ import org.apache.avro.generic.IndexedRecord;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -50,6 +54,7 @@ import java.util.stream.Stream;
import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY;
import static
org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -74,6 +79,32 @@ class TestKeyBasedFileGroupRecordBuffer extends
BaseTestFileGroupRecordBuffer {
private final IndexedRecord testRecord6DeleteByCustomMarker =
createTestRecord("6", 3, 2L);
private final IndexedRecord testRecord7 = createTestRecord("7", 1, 5L);
+ /**
+ * Asserts that {@code processNextDataRecord} and {@code
isPartialMergingEnabled} keep their
+ * {@code final} modifier, so a subclass cannot substitute its own
implementation of either.
+ *
+ * <p>That is the entire guarantee, and it is deliberately narrow. It does
<em>not</em> mean all buffer
+ * mutations go through {@code processNextDataRecord}: {@code records} and
{@code enablePartialMerging}
+ * are {@code protected}, and {@code PositionBasedFileGroupRecordBuffer}
legitimately writes to
+ * {@code records} directly when it re-keys entries in its key-based
fallback and when it overwrites
+ * delete markers under commit-time ordering, as well as overriding block
processing. Those paths must
+ * not merge, so routing them through this method would be wrong.
+ *
+ * <p>The modifier itself is compiler-enforced; this test exists only to
catch it being dropped.
+ */
+ @Test
+ void sealedMethodsCannotBeOverridden() throws NoSuchMethodException {
+ assertMethodIsFinal(KeyBasedFileGroupRecordBuffer.class
+ .getDeclaredMethod("processNextDataRecord", BufferedRecord.class,
Serializable.class));
+
assertMethodIsFinal(KeyBasedFileGroupRecordBuffer.class.getDeclaredMethod("isPartialMergingEnabled"));
+ }
+
+ private static void assertMethodIsFinal(Method method) {
+ assertTrue(Modifier.isFinal(method.getModifiers()),
+ () -> String.format("%s#%s must remain final so subclasses cannot
override it",
+ method.getDeclaringClass().getSimpleName(), method.getName()));
+ }
+
@Test
void readWithEventTimeOrdering() throws IOException {
HoodieReadStats readStats = new HoodieReadStats();
