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 cf2566cd06c4 feat: Support insert-order record positions for native
log files (#19200)
cf2566cd06c4 is described below
commit cf2566cd06c447ca86f5fbe45f9f3a949cd77ca2
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Jul 7 11:14:39 2026 +0800
feat: Support insert-order record positions for native log files (#19200)
---
.../hudi/io/cdc/HoodieNativeLogFormatWriter.java | 24 ++----
.../io/cdc/TestHoodieNativeLogFormatWriter.java | 26 +++----
.../hudi/common/table/log/LogReaderUtils.java | 87 ++++++++++++++++++++++
.../common/table/log/block/HoodieLogBlock.java | 24 +++++-
.../table/log/block/HoodieNativeLogDataBlock.java | 5 ++
.../log/block/HoodieNativeLogDeleteBlock.java | 5 ++
.../buffer/PositionBasedFileGroupRecordBuffer.java | 20 +----
.../common/functional/TestHoodieLogFormat.java | 4 +-
.../hudi/common/table/log/TestLogReaderUtils.java | 32 +++++++-
.../hudi/functional/TestMORDataSourceStorage.scala | 21 +++---
10 files changed, 178 insertions(+), 70 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
index 09c3608a5365..ed6a0b764be0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/cdc/HoodieNativeLogFormatWriter.java
@@ -24,7 +24,6 @@ import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemas;
import org.apache.hudi.common.table.HoodieTableVersion;
@@ -48,11 +47,9 @@ import org.apache.hudi.storage.StoragePath;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
@@ -237,23 +234,12 @@ public class HoodieNativeLogFormatWriter extends
HoodieLogFormat.Writer {
return header;
}
- Set<Long> positionSet = new HashSet<>(recordPositions.size());
- long previousPosition = Long.MIN_VALUE;
- for (Long position : recordPositions) {
- // RECORD_POSITIONS is encoded as a bitmap and decoded in ascending
order. Native log records
- // are read back in file write order, so only publish positions when
both orders match.
- if (!HoodieRecordLocation.isPositionValid(position) || position <=
previousPosition) {
- Map<HeaderMetadataType, String> updatedHeader = new HashMap<>(header);
-
updatedHeader.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS);
- return updatedHeader;
- }
- previousPosition = position;
- positionSet.add(position);
- }
-
Map<HeaderMetadataType, String> updatedHeader = new HashMap<>(header);
- if (positionSet.size() == recordPositions.size()) {
- updatedHeader.put(HeaderMetadataType.RECORD_POSITIONS,
LogReaderUtils.encodePositions(positionSet));
+ Option<String> encodedRecordPositions =
LogReaderUtils.encodePositionsAsLongList(recordPositions);
+ if (encodedRecordPositions.isPresent()) {
+ updatedHeader.put(HeaderMetadataType.RECORD_POSITIONS,
encodedRecordPositions.get());
+ } else {
+
updatedHeader.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS);
}
return updatedHeader;
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
index 7c38d7520d20..66ead8df3079 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/io/cdc/TestHoodieNativeLogFormatWriter.java
@@ -42,7 +42,6 @@ import org.apache.hudi.storage.StoragePathInfo;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
-import org.roaringbitmap.longlong.Roaring64NavigableMap;
import java.util.ArrayList;
import java.util.Arrays;
@@ -52,7 +51,6 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
@@ -70,15 +68,16 @@ public class TestHoodieNativeLogFormatWriter {
assertEquals("001",
parsedHeader.get(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
assertEquals(Arrays.asList(2L, 7L),
-
toList(LogReaderUtils.decodeRecordPositionsHeader(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS))));
+
LogReaderUtils.decodeRecordPositionsLongList(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS)));
}
@Test
- public void testSkipsOutOfOrderRecordPositions() throws Exception {
+ public void testAddsOutOfOrderRecordPositionsToDataLogFooter() throws
Exception {
Map<HeaderMetadataType, String> parsedHeader =
writeDataLogFooterWithPositions(7L, 2L);
-
assertFalse(parsedHeader.containsKey(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
- assertFalse(parsedHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS));
+ assertEquals("001",
parsedHeader.get(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
+ assertEquals(Arrays.asList(7L, 2L),
+
LogReaderUtils.decodeRecordPositionsLongList(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS)));
}
@Test
@@ -90,11 +89,12 @@ public class TestHoodieNativeLogFormatWriter {
}
@Test
- public void testSkipsDuplicateRecordPositions() throws Exception {
+ public void testAddsDuplicateRecordPositionsToDataLogFooter() throws
Exception {
Map<HeaderMetadataType, String> parsedHeader =
writeDataLogFooterWithPositions(7L, 7L);
-
assertNull(parsedHeader.get(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
- assertFalse(parsedHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS));
+ assertEquals("001",
parsedHeader.get(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
+ assertEquals(Arrays.asList(7L, 7L),
+
LogReaderUtils.decodeRecordPositionsLongList(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS)));
}
@Test
@@ -103,7 +103,7 @@ public class TestHoodieNativeLogFormatWriter {
assertEquals("001",
parsedHeader.get(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS));
assertEquals(Arrays.asList(7L),
-
toList(LogReaderUtils.decodeRecordPositionsHeader(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS))));
+
LogReaderUtils.decodeRecordPositionsLongList(parsedHeader.get(HeaderMetadataType.RECORD_POSITIONS)));
}
@Test
@@ -263,10 +263,4 @@ public class TestHoodieNativeLogFormatWriter {
when(record.getOrderingValue(eq(schema), any(),
any())).thenReturn(OrderingValues.getDefault());
return record;
}
-
- private static List<Long> toList(Roaring64NavigableMap positions) {
- List<Long> values = new ArrayList<>();
- positions.iterator().forEachRemaining(values::add);
- return values;
- }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
index 501b36c25b87..fdc69b9f13c9 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/LogReaderUtils.java
@@ -21,12 +21,14 @@ package org.apache.hudi.common.table.log;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.AbstractTableFileSystemView;
import org.apache.hudi.common.util.Base64CodecUtil;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.roaringbitmap.longlong.Roaring64NavigableMap;
@@ -35,6 +37,7 @@ import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -152,6 +155,28 @@ public class LogReaderUtils {
return Base64CodecUtil.encode(baos.toByteArray());
}
+ /**
+ * Encodes the record positions as an integer count followed by unsigned
varlong positions,
+ * preserving the input order.
+ *
+ * @param positions record positions in file read order.
+ * @return Base64-encoded bytes containing the list size followed by varlong
positions,
+ * or {@link Option#empty()} if any position is invalid.
+ * @throws IOException upon I/O error.
+ */
+ public static Option<String> encodePositionsAsLongList(List<Long> positions)
throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream(Integer.BYTES +
positions.size());
+ DataOutputStream dos = new DataOutputStream(baos);
+ dos.writeInt(positions.size());
+ for (Long position : positions) {
+ if (!HoodieRecordLocation.isPositionValid(position)) {
+ return Option.empty();
+ }
+ writeUnsignedVarLong(baos, position);
+ }
+ return Option.of(Base64CodecUtil.encode(baos.toByteArray()));
+ }
+
/**
* Decodes the {@link HeaderMetadataType#RECORD_POSITIONS} block header into
record positions.
*
@@ -168,4 +193,66 @@ public class LogReaderUtils {
positionBitmap.deserializePortable(dis);
return positionBitmap;
}
+
+ /**
+ * Decodes native-log record positions written as an ordered unsigned
varlong list.
+ *
+ * @param content Base64-encoded bytes containing the list size followed by
varlong positions.
+ * @return record positions in file read order.
+ * @throws IOException upon I/O error or invalid payload shape.
+ */
+ public static List<Long> decodeRecordPositionsLongList(String content)
throws IOException {
+ ByteArrayInputStream bais = new
ByteArrayInputStream(Base64CodecUtil.decode(content));
+ DataInputStream dis = new DataInputStream(bais);
+ int size = dis.readInt();
+ if (size < 0) {
+ throw new IOException("Invalid negative record position count " + size);
+ }
+ List<Long> positions = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ long position = readUnsignedVarLong(bais);
+ if (!HoodieRecordLocation.isPositionValid(position)) {
+ throw new IOException("Invalid record position " + position);
+ }
+ positions.add(position);
+ }
+ if (bais.available() != 0) {
+ throw new IOException("Invalid record positions payload with " +
bais.available() + " trailing bytes");
+ }
+ return positions;
+ }
+
+ /**
+ * Writes an unsigned long by splitting it into 7-bit groups from low bits
to high bits.
+ * The most significant bit of each byte is the continuation flag: {@code 1}
means more
+ * bytes follow and {@code 0} marks the final byte.
+ */
+ private static void writeUnsignedVarLong(ByteArrayOutputStream out, long
value) {
+ while ((value & 0xFFFFFFFFFFFFFF80L) != 0) {
+ out.write(((int) value & 0x7F) | 0x80);
+ value >>>= 7;
+ }
+ out.write((int) value & 0x7F);
+ }
+
+ /**
+ * Reads an unsigned long written by {@link
#writeUnsignedVarLong(ByteArrayOutputStream, long)}
+ * by rebuilding the value from 7-bit groups in low-to-high order.
+ */
+ private static long readUnsignedVarLong(ByteArrayInputStream in) throws
IOException {
+ long value = 0;
+ int shift = 0;
+ while (shift < Long.SIZE) {
+ int currentByte = in.read();
+ if (currentByte < 0) {
+ throw new EOFException("Unexpected EOF while reading record position");
+ }
+ value |= (long) (currentByte & 0x7F) << shift;
+ if ((currentByte & 0x80) == 0) {
+ return value;
+ }
+ shift += 7;
+ }
+ throw new IOException("Invalid unsigned varlong for record position");
+ }
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java
index 3e63daa98522..c6cb16d10dee 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieLogBlock.java
@@ -44,6 +44,8 @@ import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -142,13 +144,33 @@ public abstract class HoodieLogBlock {
* {@link Roaring64NavigableMap} bitmap.
* @throws IOException upon I/O error.
*/
- public Roaring64NavigableMap getRecordPositions() throws IOException {
+ private Roaring64NavigableMap getRecordPositions() throws IOException {
if (!logBlockHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS)) {
return new Roaring64NavigableMap();
}
return
LogReaderUtils.decodeRecordPositionsHeader(logBlockHeader.get(HeaderMetadataType.RECORD_POSITIONS));
}
+ /**
+ * @return record positions aligned with the block record iterator order.
+ * @throws IOException upon I/O error.
+ */
+ public List<Long> getRecordPositionList() throws IOException {
+ List<Long> positions = new ArrayList<>();
+ getRecordPositions().iterator().forEachRemaining(positions::add);
+ return positions;
+ }
+
+ /**
+ * Decodes ordered record positions stored for native log blocks.
+ */
+ protected static List<Long>
decodeOrderedRecordPositionList(Map<HeaderMetadataType, String> logBlockHeader)
throws IOException {
+ if (!logBlockHeader.containsKey(HeaderMetadataType.RECORD_POSITIONS)) {
+ return Collections.emptyList();
+ }
+ return
LogReaderUtils.decodeRecordPositionsLongList(logBlockHeader.get(HeaderMetadataType.RECORD_POSITIONS));
+ }
+
/**
* @return base file instant time of the record positions if the record
positions are enabled
* in the log block; {@code null} otherwise.
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDataBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDataBlock.java
index 673707e3f920..c4b0ed294028 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDataBlock.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDataBlock.java
@@ -69,6 +69,11 @@ public class HoodieNativeLogDataBlock extends
HoodieDataBlock {
return HoodieLogBlockType.PARQUET_DATA_BLOCK;
}
+ @Override
+ public List<Long> getRecordPositionList() throws IOException {
+ return decodeOrderedRecordPositionList(getLogBlockHeader());
+ }
+
@Override
protected <T> ClosableIterator<HoodieRecord<T>>
readRecordsFromBlockPayload(HoodieRecord.HoodieRecordType type) throws
IOException {
StoragePath path = logFile.getPath();
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDeleteBlock.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDeleteBlock.java
index ddaeab29760e..031bbe609929 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDeleteBlock.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieNativeLogDeleteBlock.java
@@ -75,6 +75,11 @@ public class HoodieNativeLogDeleteBlock extends
HoodieDeleteBlock {
this.props = props == null ? new Properties() : props;
}
+ @Override
+ public List<Long> getRecordPositionList() throws IOException {
+ return decodeOrderedRecordPositionList(getLogBlockHeader());
+ }
+
@Override
public DeleteRecord[] getRecordsToDelete() {
if (recordsToDelete == null) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
index 2ca86d6ad7a5..d3c49801e891 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/PositionBasedFileGroupRecordBuffer.java
@@ -42,13 +42,11 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieKeyException;
import lombok.extern.slf4j.Slf4j;
-import org.roaringbitmap.longlong.Roaring64NavigableMap;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
@@ -170,7 +168,7 @@ public class PositionBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupReco
records.remove(position);
} else {
//if it's a delete record and the key is null, then we need to still
use positions
- //this happens when we read the positions using
logBlock.getRecordPositions()
+ //this happens when we read the positions using
logBlock.getRecordPositionList()
//instead of reading the delete records themselves
needToDoHybridStrategy = true;
}
@@ -296,8 +294,6 @@ public class PositionBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupReco
*/
protected static List<Long> extractRecordPositions(HoodieLogBlock logBlock,
String
baseFileInstantTime) throws IOException {
- List<Long> blockPositions = new ArrayList<>();
-
String blockBaseFileInstantTime =
logBlock.getBaseFileInstantTimeOfPositions();
if (StringUtils.isNullOrEmpty(blockBaseFileInstantTime) ||
!baseFileInstantTime.equals(blockBaseFileInstantTime)) {
log.debug("The record positions cannot be used because the base file
instant time "
@@ -306,19 +302,9 @@ public class PositionBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupReco
blockBaseFileInstantTime, baseFileInstantTime);
return null;
}
- Roaring64NavigableMap positions = logBlock.getRecordPositions();
- if (positions == null || positions.isEmpty()) {
- log.info("No record position info is found when attempting to do
position based merge.");
- return null;
- }
-
- Iterator<Long> iterator = positions.iterator();
- while (iterator.hasNext()) {
- blockPositions.add(iterator.next());
- }
-
+ List<Long> blockPositions = logBlock.getRecordPositionList();
if (blockPositions.isEmpty()) {
- log.info("No positions are extracted.");
+ log.info("No record position info is found when attempting to do
position based merge.");
return null;
}
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 6efc1efbd498..f4b3732a17a9 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
@@ -3034,9 +3034,9 @@ public class TestHoodieLogFormat extends
HoodieCommonTestHarness {
String expectedInstantTime = hasPositions ? "001" : null;
Set<Long> expectedPositions = hasPositions ? new HashSet<>(positions) :
Collections.emptySet();
assertEquals(expectedInstantTime,
deleteBlock.getBaseFileInstantTimeOfPositions());
- TestLogReaderUtils.assertPositionEquals(expectedPositions,
deleteBlock.getRecordPositions());
+ assertEquals(expectedPositions, new
HashSet<>(deleteBlock.getRecordPositionList()));
assertEquals(expectedInstantTime,
dataBlock.getBaseFileInstantTimeOfPositions());
- TestLogReaderUtils.assertPositionEquals(expectedPositions,
dataBlock.getRecordPositions());
+ assertEquals(expectedPositions, new
HashSet<>(dataBlock.getRecordPositionList()));
}
private static Stream<Arguments> testArguments() {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
index 6e65ef3c656e..edb7ec42638c 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
@@ -19,6 +19,9 @@
package org.apache.hudi.common.table.log;
+import org.apache.hudi.common.util.Base64CodecUtil;
+import org.apache.hudi.common.util.Option;
+
import org.junit.jupiter.api.Test;
import org.roaringbitmap.longlong.Roaring64NavigableMap;
@@ -34,6 +37,7 @@ import java.util.stream.Collectors;
import static
org.apache.hudi.common.testutils.HoodieTestTable.readLastLineFromResourceFile;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests for {@link LogReaderUtils}
@@ -67,6 +71,22 @@ public class TestLogReaderUtils {
assertPositionEquals(expectedPositions, roaring64NavigableMap);
}
+ @Test
+ public void testEncodeAndDecodePositionsAsOrderedLongList() throws
IOException {
+ List<Long> positions = Arrays.asList(7L, 2L, 7L, 128L, 16_384L);
+
+ Option<String> content =
LogReaderUtils.encodePositionsAsLongList(positions);
+
+ assertTrue(content.isPresent());
+ assertEquals(positions,
LogReaderUtils.decodeRecordPositionsLongList(content.get()));
+ assertTrue(Base64CodecUtil.decode(content.get()).length < Integer.BYTES +
positions.size() * Long.BYTES);
+ }
+
+ @Test
+ public void testSkipEncodingInvalidPositionInOrderedLongList() throws
IOException {
+ assertFalse(LogReaderUtils.encodePositionsAsLongList(Arrays.asList(7L,
-1L)).isPresent());
+ }
+
public static Set<Long> generatePositions() {
Random random = new Random(0x2023);
Set<Long> positions = new HashSet<>();
@@ -81,12 +101,16 @@ public class TestLogReaderUtils {
Roaring64NavigableMap
roaring64NavigableMap) {
List<Long> sortedExpectedPositions =
expectedPositions.stream().sorted().collect(Collectors.toList());
+ assertPositionListEquals(sortedExpectedPositions,
roaring64NavigableMap.iterator());
+ }
+
+ private static void assertPositionListEquals(List<Long>
sortedExpectedPositions,
+ Iterator<Long> actualIterator) {
Iterator<Long> expectedIterator = sortedExpectedPositions.iterator();
- Iterator<Long> iterator = roaring64NavigableMap.iterator();
- while (expectedIterator.hasNext() && iterator.hasNext()) {
- assertEquals(expectedIterator.next(), iterator.next());
+ while (expectedIterator.hasNext() && actualIterator.hasNext()) {
+ assertEquals(expectedIterator.next(), actualIterator.next());
}
assertFalse(expectedIterator.hasNext());
- assertFalse(iterator.hasNext());
+ assertFalse(actualIterator.hasNext());
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
index 2b74a7bcb0cd..ce67270b0181 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala
@@ -362,17 +362,16 @@ class TestMORDataSourceStorage extends
SparkClientFunctionalTestHarness {
var numBlocks = 0
while (logFormatReader.hasNext) {
val logBlock = logFormatReader.next()
- // todo(https://github.com/apache/hudi/issues/19178), re-enable the
follow test after the issue solved.
- // val recordPositions = logBlock.getRecordPositions
- //
- // assertEquals(shouldContainRecordPosition,
!recordPositions.isEmpty)
- // if (shouldContainRecordPosition) {
- // val baseFile = fsv.getLatestBaseFile("", filename.getFileId)
- // assertTrue(baseFile.isPresent)
- // assertEquals(
- // shouldBaseFileInstantTimeMatch,
- //
baseFile.get().getCommitTime.equals(logBlock.getBaseFileInstantTimeOfPositions))
- // }
+ val recordPositions = logBlock.getRecordPositionList
+
+ assertEquals(shouldContainRecordPosition, !recordPositions.isEmpty)
+ if (shouldContainRecordPosition) {
+ val baseFile = fsv.getLatestBaseFile("", filename.getFileId)
+ assertTrue(baseFile.isPresent)
+ assertEquals(
+ shouldBaseFileInstantTimeMatch,
+
baseFile.get().getCommitTime.equals(logBlock.getBaseFileInstantTimeOfPositions))
+ }
numBlocks += 1
}
assertTrue(numBlocks > 0)
