This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 4394513a8cb7 test(common): add log-format reader and scanner coverage
(#19223)
4394513a8cb7 is described below
commit 4394513a8cb7a96d03e301255d6da39310d86b47
Author: Y Ethan Guo <[email protected]>
AuthorDate: Sun Jul 19 10:09:30 2026 -0700
test(common): add log-format reader and scanner coverage (#19223)
* test(common): add log-format reader and scanner coverage
* test(common): fix scanner close/lifecycle and restrict BRAF tests to read
path
- Drop non-existent HoodieUnMergedLogRecordScanner.close() calls that
caused Azure CI to fail compilation of hudi-hadoop-common tests.
- Refactor TestBufferedRandomAccessFile to seed fixtures via Files.write
and only exercise the read path. Production only opens the class in
read mode ("r") (BitCaskDiskMap / LazyFileIterable). Exercising its
write path across buffer boundaries hangs in an infinite loop (dormant
writer-side bug in expandBufferToCapacityIfNeeded not addressed here).
---------
Co-authored-by: sivabalan <[email protected]>
Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
---
.../common/util/TestBufferedRandomAccessFile.java | 132 ++++++++++++++
.../common/functional/TestHoodieLogFormat.java | 202 +++++++++++++++++++++
2 files changed, 334 insertions(+)
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/util/TestBufferedRandomAccessFile.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestBufferedRandomAccessFile.java
new file mode 100644
index 000000000000..5dc1fab9c86d
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestBufferedRandomAccessFile.java
@@ -0,0 +1,132 @@
+/*
+ * 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.util;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Random;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Unit tests for {@link BufferedRandomAccessFile}, the buffered wrapper used
to
+ * back the log-file read path (via BitCaskDiskMap / LazyFileIterable). The
buffer
+ * capacity is clamped to a 64K minimum, so the payloads here are deliberately
+ * larger than that to force reads and seeks that cross buffer boundaries.
+ *
+ * <p>All fixtures are seeded via {@link Files#write} rather than the class's
own
+ * write path since Hudi opens this class only in read mode ("r") in
production.
+ */
+public class TestBufferedRandomAccessFile {
+
+ // Larger than the internal 64K minimum buffer so that access crosses buffer
boundaries.
+ private static final int PAYLOAD_SIZE = (1 << 16) * 3 + 517;
+
+ private static byte[] deterministicBytes(int size) {
+ byte[] bytes = new byte[size];
+ new Random(42).nextBytes(bytes);
+ return bytes;
+ }
+
+ private static File seedFile(Path dir, String name, byte[] contents) throws
Exception {
+ File file = new File(dir.toFile(), name);
+ Files.write(file.toPath(), contents);
+ return file;
+ }
+
+ @Test
+ public void testReadFullyAcrossBufferBoundaries(@TempDir Path tempDir)
throws Exception {
+ byte[] expected = deterministicBytes(PAYLOAD_SIZE);
+ File file = seedFile(tempDir, "read.bin", expected);
+
+ byte[] readBack = new byte[PAYLOAD_SIZE];
+ try (BufferedRandomAccessFile raf = new BufferedRandomAccessFile(file,
"r")) {
+ // readFully loops over read(...) so partial reads at buffer boundaries
are handled.
+ raf.readFully(readBack);
+ assertEquals(PAYLOAD_SIZE, raf.getFilePointer(), "Read should consume
the whole file");
+ assertEquals(-1, raf.read(), "Reading past EOF should return -1");
+ }
+ assertArrayEquals(expected, readBack, "Bytes read back should match bytes
written");
+ }
+
+ @Test
+ public void testSeekRandomAccessCrossingBoundaries(@TempDir Path tempDir)
throws Exception {
+ byte[] expected = deterministicBytes(PAYLOAD_SIZE);
+ File file = seedFile(tempDir, "seek.bin", expected);
+
+ try (BufferedRandomAccessFile raf = new BufferedRandomAccessFile(file,
"r")) {
+ // Probe positions that fall in the first, second, third and last
logical buffer blocks,
+ // out of order, so both forward and backward seeks are exercised.
+ int[] probes = {0, (1 << 16) + 7, (1 << 16) * 2 + 100, 5, PAYLOAD_SIZE -
1};
+ for (int pos : probes) {
+ raf.seek(pos);
+ assertEquals(pos, raf.getFilePointer(), "getFilePointer should reflect
the seek target");
+ int actual = raf.read();
+ assertEquals(expected[pos] & 0xFF, actual, "Byte at position " + pos +
" should match");
+ }
+
+ // A ranged read that starts mid-buffer and spans a boundary must return
the correct slice.
+ int start = (1 << 16) - 10;
+ int len = 40;
+ raf.seek(start);
+ byte[] slice = new byte[len];
+ raf.readFully(slice);
+ byte[] expectedSlice = new byte[len];
+ System.arraycopy(expected, start, expectedSlice, 0, len);
+ assertArrayEquals(expectedSlice, slice, "Ranged read across a buffer
boundary should match");
+ }
+ }
+
+ @Test
+ public void testReadIntoOffsetAndLength(@TempDir Path tempDir) throws
Exception {
+ byte[] expected = deterministicBytes(1024);
+ File file = seedFile(tempDir, "offset.bin", expected);
+
+ try (BufferedRandomAccessFile raf = new BufferedRandomAccessFile(file,
"r")) {
+ byte[] target = new byte[1024 + 8];
+ // Leave a 4-byte prefix untouched and read the payload into the middle
of the array.
+ raf.readFully(target, 4, 1024);
+ byte[] payload = new byte[1024];
+ System.arraycopy(target, 4, payload, 0, 1024);
+ assertArrayEquals(expected, payload, "readFully into an offset should
place bytes at that offset");
+ assertEquals(0, target[0], "Bytes before the offset must remain
untouched");
+ assertEquals(0, target[3], "Byte just before the offset must remain
untouched");
+ assertEquals(0, target[1028], "Byte just after the range must remain
untouched");
+ assertEquals(0, target[1031], "Last byte after the range must remain
untouched");
+ }
+ }
+
+ @Test
+ public void testLengthAndEofAfterFullRead(@TempDir Path tempDir) throws
Exception {
+ byte[] expected = deterministicBytes(PAYLOAD_SIZE);
+ File file = seedFile(tempDir, "length.bin", expected);
+
+ try (BufferedRandomAccessFile raf = new BufferedRandomAccessFile(file,
"r")) {
+ assertEquals(PAYLOAD_SIZE, raf.length(), "length should report the
on-disk size");
+ raf.seek(PAYLOAD_SIZE);
+ assertEquals(-1, raf.read(), "Reading at EOF should return -1");
+ assertEquals(PAYLOAD_SIZE, raf.getFilePointer(), "File pointer should
stay at EOF");
+ }
+ }
+}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
index f4b3732a17a9..ca7893266062 100755
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
@@ -47,6 +47,8 @@ import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormat.Reader;
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
+import org.apache.hudi.common.table.log.HoodieUnMergedLogRecordScanner;
+import org.apache.hudi.common.table.log.InstantRange;
import org.apache.hudi.common.table.log.TestLogReaderUtils;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
@@ -3039,6 +3041,206 @@ public class TestHoodieLogFormat extends
HoodieCommonTestHarness {
assertEquals(expectedPositions, new
HashSet<>(dataBlock.getRecordPositionList()));
}
+ @Test
+ public void testUnMergedLogRecordScannerCallbacksWithDeletes()
+ throws IOException, URISyntaxException, InterruptedException {
+ HoodieSchema schema =
HoodieSchemaUtils.addMetadataFields(getSimpleSchema());
+ HoodieLogFormat.Writer writer =
+ HoodieLogFormatWriter.builder().withParentPath(partitionPath)
+ .withFileExtension(HoodieLogFile.DELTA_EXTENSION)
+ .withLogFileId("test-fileid1")
+ .withInstantTime("100")
+ .withStorage(storage)
+ .build();
+
+ SchemaTestUtil testUtil = new SchemaTestUtil();
+ List<IndexedRecord> records = testUtil.generateHoodieTestRecords(0, 100);
+ List<IndexedRecord> copyOfRecords = records.stream()
+ .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record,
schema.toAvroSchema()))
+ .collect(Collectors.toList());
+ Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100");
+ header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
+ writer.appendBlock(getDataBlock(DEFAULT_DATA_BLOCK_TYPE, records, header));
+
+ // Delete the first 20 keys via a delete block.
+ List<String> allKeys = copyOfRecords.stream()
+ .map(r -> ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString())
+ .collect(Collectors.toList());
+ List<String> deletedKeys = new ArrayList<>(allKeys.subList(0, 20));
+ List<Pair<DeleteRecord, Long>> deleteRecordList = copyOfRecords.subList(0,
20).stream()
+ .map(r -> Pair.of(DeleteRecord.create(
+ ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString(),
+ ((GenericRecord)
r).get(HoodieRecord.PARTITION_PATH_METADATA_FIELD).toString()),
+ -1L))
+ .collect(Collectors.toList());
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100");
+ writer.appendBlock(new HoodieDeleteBlock(deleteRecordList, header));
+ writer.close();
+
+ FileCreateUtilsLegacy.createDeltaCommit(basePath, "100", storage);
+
+ List<String> insertedKeys = new ArrayList<>();
+ List<String> deletedKeysSeen = new ArrayList<>();
+ HoodieUnMergedLogRecordScanner scanner =
HoodieUnMergedLogRecordScanner.newBuilder()
+ .withStorage(storage)
+ .withBasePath(basePath)
+
.withLogFilePaths(Collections.singletonList(writer.getLogFile().getPath().toString()))
+ .withReaderSchema(schema)
+ .withLatestInstantTime("100")
+ .withReverseReader(false)
+ .withBufferSize(BUFFER_SIZE)
+ .withLogRecordScannerCallback(record ->
insertedKeys.add(record.getRecordKey()))
+ .withRecordDeletionCallback(key ->
deletedKeysSeen.add(key.getRecordKey()))
+ .build();
+ scanner.scan();
+
+ // The un-merged scanner streams every data record and every deleted key
through the callbacks
+ // without merging them, so all 100 inserts and all 20 deletes should be
observed exactly once.
+ assertEquals(100, insertedKeys.size(), "Callback should see every appended
data record");
+ assertEquals(new HashSet<>(allKeys), new HashSet<>(insertedKeys),
+ "Callback keys should match every appended record key");
+ assertEquals(20, deletedKeysSeen.size(), "Deletion callback should see
every deleted key");
+ assertEquals(new HashSet<>(deletedKeys), new HashSet<>(deletedKeysSeen),
+ "Deletion callback keys should match the delete block keys");
+ }
+
+ @Test
+ public void testUnMergedLogRecordScannerInstantRangeFiltering()
+ throws IOException, URISyntaxException, InterruptedException {
+ HoodieSchema schema =
HoodieSchemaUtils.addMetadataFields(getSimpleSchema());
+ HoodieLogFormat.Writer writer =
+ HoodieLogFormatWriter.builder().withParentPath(partitionPath)
+ .withFileExtension(HoodieLogFile.DELTA_EXTENSION)
+ .withLogFileId("test-fileid1")
+ .withInstantTime("100")
+ .withStorage(storage)
+ .build();
+
+ SchemaTestUtil testUtil = new SchemaTestUtil();
+ // First block belongs to instant 100, second block to instant 101.
+ List<IndexedRecord> recordsAt100 = testUtil.generateHoodieTestRecords(0,
60);
+ List<String> keysAt100 = recordsAt100.stream()
+ .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record,
schema.toAvroSchema()))
+ .map(r -> ((GenericRecord)
r).get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString())
+ .collect(Collectors.toList());
+ Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100");
+ header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
+ writer.appendBlock(getDataBlock(DEFAULT_DATA_BLOCK_TYPE, recordsAt100,
header));
+
+ List<IndexedRecord> recordsAt101 = testUtil.generateHoodieTestRecords(0,
40);
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "101");
+ writer.appendBlock(getDataBlock(DEFAULT_DATA_BLOCK_TYPE, recordsAt101,
header));
+ writer.close();
+
+ FileCreateUtilsLegacy.createDeltaCommit(basePath, "100", storage);
+ FileCreateUtilsLegacy.createDeltaCommit(basePath, "101", storage);
+
+ // A closed range of [100, 100] should keep only the first block and drop
the instant 101 block.
+ InstantRange instantRange = InstantRange.builder()
+ .startInstant("100")
+ .endInstant("100")
+ .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
+ .build();
+
+ List<String> seenKeys = new ArrayList<>();
+ HoodieUnMergedLogRecordScanner scanner =
HoodieUnMergedLogRecordScanner.newBuilder()
+ .withStorage(storage)
+ .withBasePath(basePath)
+
.withLogFilePaths(Collections.singletonList(writer.getLogFile().getPath().toString()))
+ .withReaderSchema(schema)
+ .withLatestInstantTime("101")
+ .withReverseReader(false)
+ .withBufferSize(BUFFER_SIZE)
+ .withInstantRange(Option.of(instantRange))
+ .withLogRecordScannerCallback(record ->
seenKeys.add(record.getRecordKey()))
+ .build();
+ scanner.scan();
+
+ assertEquals(60, seenKeys.size(), "Only the instant 100 block should pass
the instant range");
+ assertEquals(new HashSet<>(keysAt100), new HashSet<>(seenKeys),
+ "Records outside the instant range should be filtered out");
+ }
+
+ @Test
+ public void testLogFileReaderReadsPastCorruptBlock()
+ throws IOException, URISyntaxException, InterruptedException {
+ HoodieSchema schema = getSimpleSchema();
+ HoodieLogFormat.Writer writer =
+ HoodieLogFormatWriter.builder().withParentPath(partitionPath)
+ .withFileExtension(HoodieLogFile.DELTA_EXTENSION)
+ .withLogFileId("test-fileid1")
+ .withInstantTime("100")
+ .withStorage(storage)
+ .build();
+ List<IndexedRecord> records1 = SchemaTestUtil.generateTestRecords(0, 100);
+ List<IndexedRecord> copyOfRecords1 = records1.stream()
+ .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record,
schema.toAvroSchema()))
+ .collect(Collectors.toList());
+ Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100");
+ header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
+ writer.appendBlock(getDataBlock(DEFAULT_DATA_BLOCK_TYPE, records1,
header));
+ writer.close();
+
+ // Append a block whose declared length does not match its content,
mimicking a partial write.
+ FSDataOutputStream outputStream = (FSDataOutputStream)
storage.append(writer.getLogFile().getPath());
+ outputStream.write(HoodieLogFormat.MAGIC);
+ outputStream.writeLong(474);
+ outputStream.writeInt(HoodieLogBlockType.AVRO_DATA_BLOCK.ordinal());
+ outputStream.writeInt(HoodieLogFormat.INLINE_LOG_FORMAT_VERSION);
+ outputStream.writeLong(400);
+ outputStream.write(getUTF8Bytes("truncated-block-content"));
+ outputStream.flush();
+ outputStream.close();
+
+ // Append a valid trailing block so the reader has a real block to recover
to after the corruption.
+ HoodieLogFormat.Writer appendWriter =
+ HoodieLogFormatWriter.builder().withParentPath(partitionPath)
+ .withFileExtension(HoodieLogFile.DELTA_EXTENSION)
+ .withLogFileId("test-fileid1")
+ .withInstantTime("100")
+ .withStorage(storage)
+ .build();
+ ((HoodieLogFormatWriter) appendWriter).withOutputStream(
+ (FSDataOutputStream) storage.append(writer.getLogFile().getPath()));
+ List<IndexedRecord> records2 = SchemaTestUtil.generateTestRecords(0, 10);
+ List<IndexedRecord> copyOfRecords2 = records2.stream()
+ .map(record -> HoodieAvroUtils.rewriteRecord((GenericRecord) record,
schema.toAvroSchema()))
+ .collect(Collectors.toList());
+ appendWriter.appendBlock(getDataBlock(DEFAULT_DATA_BLOCK_TYPE, records2,
header));
+ appendWriter.close();
+
+ HoodieLogFile logFile = new
HoodieLogFile(appendWriter.getLogFile().getPath(),
+ storage.getPathInfo(appendWriter.getLogFile().getPath()).getLength());
+ try (HoodieLogFileReader reader =
+ new HoodieLogFileReader(storage, logFile,
SchemaTestUtil.getSimpleSchema(), BUFFER_SIZE)) {
+ // First a valid data block.
+ assertTrue(reader.hasNext(), "First data block should be available");
+ HoodieLogBlock firstBlock = reader.next();
+ assertEquals(HoodieLogBlockType.AVRO_DATA_BLOCK,
firstBlock.getBlockType(), "First block should be a data block");
+ List<IndexedRecord> firstRead = getRecords((HoodieDataBlock) firstBlock);
+ assertEquals(convertAvroToSerializableIndexedRecords(copyOfRecords1),
firstRead,
+ "First block contents should match the written records");
+
+ // The reader seeks past the bad magic/length and surfaces a corrupt
block.
+ assertTrue(reader.hasNext(), "Corrupt block should be surfaced");
+ HoodieLogBlock corruptBlock = reader.next();
+ assertEquals(HoodieLogBlockType.CORRUPT_BLOCK,
corruptBlock.getBlockType(), "Second block should be a corrupt block");
+
+ // The valid trailing block should still be readable after recovery.
+ assertTrue(reader.hasNext(), "Trailing data block should be available
after the corrupt block");
+ HoodieLogBlock lastBlock = reader.next();
+ assertEquals(HoodieLogBlockType.AVRO_DATA_BLOCK,
lastBlock.getBlockType(), "Third block should be a data block");
+ List<IndexedRecord> lastRead = getRecords((HoodieDataBlock) lastBlock);
+ assertEquals(convertAvroToSerializableIndexedRecords(copyOfRecords2),
lastRead,
+ "Trailing block contents should match the written records");
+
+ assertFalse(reader.hasNext(), "There should be no more blocks");
+ }
+ }
+
private static Stream<Arguments> testArguments() {
// Arg1: ExternalSpillableMap Type, Arg2: isDiskMapCompressionEnabled
return Stream.of(
