This is an automated email from the ASF dual-hosted git repository.
leaves12138 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 437bcc3c84 [format] Fix RLE skipping at sparse row-range boundaries
(#9127)
437bcc3c84 is described below
commit 437bcc3c84a10b71b827a0a00c1661ccaad6e64f
Author: CrownChu <[email protected]>
AuthorDate: Mon Aug 10 11:04:56 2026 +0800
[format] Fix RLE skipping at sparse row-range boundaries (#9127)
---
.../operation/DataEvolutionSplitReadTest.java | 135 ++++++++++++++++++++-
.../parquet/reader/VectorizedRleValuesReader.java | 6 +-
2 files changed, 135 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
index 89316c249b..ac865ba2ab 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionSplitReadTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
@@ -320,9 +321,121 @@ class DataEvolutionSplitReadTest {
assertEquals(Arrays.asList(1000, 1032), actual);
}
+ @Test
+ public void testSparseRowIdReadUsesParquetRowsAtSelectedPositions() throws
Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ Path tableRoot = new Path(tempDir.toUri().toString(),
"sparse-parquet");
+ Options options = new Options();
+ options.set(CoreOptions.FILE_FORMAT, "parquet");
+ CoreOptions coreOptions = new CoreOptions(options);
+ FileStorePathFactory pathFactory =
+ new FileStorePathFactory(
+ tableRoot,
+ RowType.of(),
+ coreOptions.partitionDefaultName(),
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null);
+
+ Schema schema =
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.ARRAY(DataTypes.INT()))
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .build();
+ SchemaManager schemaManager = new SchemaManager(fileIO, tableRoot);
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ RowType rowType = tableSchema.logicalRowType();
+
+ Path bucketPath = pathFactory.bucketPath(EMPTY_ROW, 0);
+ fileIO.mkdirs(bucketPath);
+ String fileName = "data-0.parquet";
+ Path filePath = new Path(bucketPath, fileName);
+ Options parquetOptions = new Options();
+ parquetOptions.set("parquet.block.size", "65536");
+ parquetOptions.set("parquet.page.size", "4096");
+ parquetOptions.set("parquet.writer.version", "v2");
+ parquetOptions.set("parquet.page.size.row.check.min", "100");
+ writeNestedParquetFile(fileIO, filePath, rowType, 10_000,
parquetOptions);
+
+ DataFileMeta dataFile =
+ createFile(fileName, fileIO.getFileStatus(filePath).getLen(),
10L, 10_000, 1);
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withPartition(EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath(bucketPath.toString())
+ .withDataFiles(Collections.singletonList(dataFile))
+ .rawConvertible(false)
+ .build();
+
+ DataEvolutionSplitRead splitRead =
+ new DataEvolutionSplitRead(
+ fileIO, schemaManager, tableSchema, rowType,
coreOptions, pathFactory);
+ IndexedSplit indexedSplit =
+ new IndexedSplit(
+ dataSplit,
+ Arrays.asList(new Range(10L, 10L), new Range(4210L,
4210L)),
+ new float[] {1.0F, 0.5F});
+
+ List<String> actual = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
splitRead.createReader(indexedSplit)) {
+ reader.forEachRemaining(
+ row ->
+ actual.add(
+ String.format(
+ "%d:[%d,%d]",
+ row.getInt(0),
+ row.getArray(1).getInt(0),
+ row.getArray(1).getInt(1))));
+ }
+
+ assertEquals(Arrays.asList("1000:[0,1]", "5200:[4200,4201]"), actual);
+ }
+
+ private static void writeNestedParquetFile(
+ LocalFileIO fileIO, Path path, RowType rowType, int rowCount,
Options options)
+ throws IOException {
+ FileFormat format = FileFormat.fromIdentifier("parquet", options);
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ FormatWriter writer =
format.createWriterFactory(rowType).create(out, "zstd");
+ for (int i = 0; i < rowCount; i++) {
+ writer.addElement(GenericRow.of(1000 + i, new GenericArray(new
int[] {i, i + 1})));
+ }
+ writer.close();
+ }
+ }
+
private static void writeRowFile(LocalFileIO fileIO, Path path, RowType
rowType, int rowCount)
throws IOException {
- FileFormat format = FileFormat.fromIdentifier("row", new Options());
+ writeFormatFile(fileIO, path, rowType, rowCount, "row");
+ }
+
+ private static void writeFormatFile(
+ LocalFileIO fileIO, Path path, RowType rowType, int rowCount,
String formatIdentifier)
+ throws IOException {
+ writeFormatFile(fileIO, path, rowType, rowCount, formatIdentifier, new
Options());
+ }
+
+ private static void writeFormatFile(
+ LocalFileIO fileIO,
+ Path path,
+ RowType rowType,
+ int rowCount,
+ String formatIdentifier,
+ Options options)
+ throws IOException {
+ FileFormat format = FileFormat.fromIdentifier(formatIdentifier,
options);
try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
FormatWriter writer =
format.createWriterFactory(rowType).create(out, "zstd");
for (int i = 0; i < rowCount; i++) {
@@ -334,18 +447,34 @@ class DataEvolutionSplitReadTest {
private static DataFileMeta createFile(
String name, long firstRowId, long rowCount, long maxSequence) {
- return createFile(name, firstRowId, rowCount, maxSequence,
Collections.emptyList());
+ return createFile(name, 10000L, firstRowId, rowCount, maxSequence);
+ }
+
+ private static DataFileMeta createFile(
+ String name, long fileSize, long firstRowId, long rowCount, long
maxSequence) {
+ return createFile(
+ name, fileSize, firstRowId, rowCount, maxSequence,
Collections.emptyList());
+ }
+
+ private static DataFileMeta createFile(
+ String name,
+ long firstRowId,
+ long rowCount,
+ long maxSequence,
+ List<String> extraFiles) {
+ return createFile(name, 10000L, firstRowId, rowCount, maxSequence,
extraFiles);
}
private static DataFileMeta createFile(
String name,
+ long fileSize,
long firstRowId,
long rowCount,
long maxSequence,
List<String> extraFiles) {
return DataFileMeta.create(
name,
- 10000L,
+ fileSize,
(int) rowCount,
EMPTY_ROW,
EMPTY_ROW,
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedRleValuesReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedRleValuesReader.java
index 2570831a14..46c352da1d 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedRleValuesReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedRleValuesReader.java
@@ -235,7 +235,7 @@ public final class VectorizedRleValuesReader extends
ValuesReader
long rangeStart = state.currentRangeStart();
long rangeEnd = state.currentRangeEnd();
- if (rowId + n < rangeStart) {
+ if (rowId + n <= rangeStart) {
skipValues(n, state, valueReader, updater);
rowId += n;
leftInPage -= n;
@@ -315,7 +315,7 @@ public final class VectorizedRleValuesReader extends
ValuesReader
long rangeStart = state.currentRangeStart();
long rangeEnd = state.currentRangeEnd();
- if (rowId + n < rangeStart) {
+ if (rowId + n <= rangeStart) {
skipValues(n, state, valueReader, updater);
rowId += n;
leftInPage -= n;
@@ -471,7 +471,7 @@ public final class VectorizedRleValuesReader extends
ValuesReader
// current batch
int n = Math.min(leftInBatch,
valuesLeftInBlock);
- if (rowId + n < rangeStart) {
+ if (rowId + n <= rangeStart) {
// Need to skip all rows in [rowId, rowId
+ n)
defLevelProcessor.skipValues(n);
rowId += n;
