This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 5a8189aa54 [core] Consider deletion vectors when selecting row
sidecars (#10792)
5a8189aa54 is described below
commit 5a8189aa54cbaf641db626728eaa447753b5b8e0
Author: sanshi <[email protected]>
AuthorDate: Mon Oct 26 14:24:48 2026 +0800
[core] Consider deletion vectors when selecting row sidecars (#10792)
---
.../paimon/operation/DataEvolutionSplitRead.java | 264 +++++++++++++-------
.../operation/DataEvolutionSplitReadTest.java | 265 +++++++++++++++++----
2 files changed, 387 insertions(+), 142 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
index 953f73f935..7eb5026e10 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
@@ -28,6 +28,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.PartitionInfo;
import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fileindex.FileIndexResult;
@@ -83,6 +84,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -543,7 +545,12 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FieldBunch bunch = fieldsFiles.get(i);
DataFileMeta firstFile = bunch.files().get(0);
FileReadTarget readTarget =
- readTarget(firstFile, dataFilePathFactory, rowRanges,
groupSelection);
+ readTarget(
+ firstFile,
+ dataFilePathFactory,
+ rowRanges,
+ groupSelection,
+ deletionVector);
String formatIdentifier = readTarget.formatIdentifier;
Builder targetFormatBuilder =
readTarget.isPositionalRowFormat()
@@ -600,7 +607,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
throws IOException {
DataFileMeta firstFile = bunch.files().get(0);
FileReadTarget readTarget =
- readTarget(firstFile, dataFilePathFactory, rowRanges,
groupSelection);
+ readTarget(
+ firstFile, dataFilePathFactory, rowRanges,
groupSelection, deletionVector);
Builder formatBuilder =
readTarget.isPositionalRowFormat()
? positionalRowFormatBuilder(readRowType, null,
nestedFieldEnabled)
@@ -754,20 +762,23 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
List<ReaderSupplier<InternalRow>> readerSuppliers = new ArrayList<>();
for (DataFileMeta file : files) {
readerSuppliers.add(
- () ->
- createFileReader(
- partition,
- file,
- formatReaderMapping,
- rowRanges,
- readRowType,
- new FileReadTarget(
-
DataFilePathFactory.formatIdentifier(file.fileName()),
- dataFilePathFactory.toPath(file),
- file.fileSize()),
- deletionVector,
- groupSelection,
- null));
+ () -> {
+ RoaringBitmap32 selection = fileSelection(file,
rowRanges, groupSelection);
+ return createFileReader(
+ partition,
+ file,
+ formatReaderMapping,
+ rowRanges,
+ readRowType,
+ new FileReadTarget(
+
DataFilePathFactory.formatIdentifier(file.fileName()),
+ dataFilePathFactory.toPath(file),
+ file.fileSize(),
+ selection),
+ deletionVector,
+ groupSelection,
+ null);
+ });
}
return ConcatRecordReader.create(readerSuppliers);
}
@@ -845,7 +856,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
FileReadTarget readTarget =
- readTarget(file, dataFilePathFactory, rowRanges,
fileIndexResult);
+ readTarget(file, dataFilePathFactory, rowRanges,
fileIndexResult, deletionVector);
String formatIdentifier = readTarget.formatIdentifier;
boolean positionalRowFormat = readTarget.isPositionalRowFormat();
Builder targetFormatBuilder =
@@ -915,7 +926,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
formatReaderMapping,
rowRanges,
readRowType,
- readTarget(file, dataFilePathFactory, rowRanges,
groupSelection),
+ readTarget(file, dataFilePathFactory, rowRanges,
groupSelection, deletionVector),
deletionVector,
groupSelection,
metadataCache);
@@ -932,20 +943,15 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
@Nullable FileIndexResult fileIndexResult,
@Nullable FileMetadataCache metadataCache)
throws IOException {
- RoaringBitmap32 selection = file.toFileSelection(rowRanges);
+ RoaringBitmap32 selection = readTarget.selection;
BitmapIndexResult bitmapIndexResult =
fileIndexResult instanceof BitmapIndexResult
? (BitmapIndexResult) fileIndexResult
: null;
- if (bitmapIndexResult != null) {
- RoaringBitmap32 indexSelection = bitmapIndexResult.get();
- selection =
- selection == null
- ? indexSelection.clone()
- : RoaringBitmap32.and(selection, indexSelection);
- if (selection.isEmpty()) {
- return new EmptyFileRecordReader<>();
- }
+ if (selection != null && selection.isEmpty()) {
+ // Data-file targets never include DV-only filtering in their
selection, so a file
+ // fully deleted by its DV still reaches applyDeletionVector below.
+ return new EmptyFileRecordReader<>();
}
FormatReaderContext formatReaderContext =
@@ -1530,66 +1536,57 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataFileMeta file,
DataFilePathFactory dataFilePathFactory,
List<Range> rowRanges,
- @Nullable FileIndexResult fileIndexResult)
+ @Nullable FileIndexResult fileIndexResult,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
+ RoaringBitmap32 selection = fileSelection(file, rowRanges,
fileIndexResult);
String rowSidecar = rowSidecarFileName(file);
- if (rowSidecar != null
- && shouldReadRowSidecar(
- file,
- rowRanges,
- fileIndexResult,
- coreOptions.dataEvolutionRowSidecarMaxSelectedRows(),
-
coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) {
- Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar, file);
- try {
- return new FileReadTarget(
- ROW_SIDECAR_FORMAT, rowPath,
fileIO.getFileStatus(rowPath).getLen());
- } catch (IOException e) {
- if (!coreOptions.scanIgnoreLostFile()) {
- throw e;
+ if (rowSidecar != null) {
+ RoaringBitmap32 rowSidecarSelection =
+ sidecarSelection(
+ file,
+ selection,
+ deletionVector == null ? null :
deletionVector.deletionVector,
+ deletionVector == null ? null :
deletionVector.range);
+ if (shouldReadRowSidecar(
+ file,
+ rowSidecarSelection,
+ coreOptions.dataEvolutionRowSidecarMaxSelectedRows(),
+ coreOptions.dataEvolutionRowSidecarMaxSelectionRatio())) {
+ Path rowPath = dataFilePathFactory.toAlignedPath(rowSidecar,
file);
+ try {
+ return new FileReadTarget(
+ ROW_SIDECAR_FORMAT,
+ rowPath,
+ fileIO.getFileStatus(rowPath).getLen(),
+ rowSidecarSelection);
+ } catch (IOException e) {
+ if (!coreOptions.scanIgnoreLostFile()) {
+ throw e;
+ }
}
}
}
return new FileReadTarget(
DataFilePathFactory.formatIdentifier(file.fileName()),
dataFilePathFactory.toPath(file),
- file.fileSize());
- }
-
- @Nullable
- @VisibleForTesting
- static String rowSidecarFileName(DataFileMeta file) {
- List<String> rowFiles =
- file.extraFiles().stream()
- .filter(DataEvolutionSplitRead::isRowSidecarFile)
- .collect(Collectors.toList());
- return rowFiles.size() == 1 ? rowFiles.get(0) : null;
- }
-
- @VisibleForTesting
- static boolean shouldReadRowSidecar(DataFileMeta file, @Nullable
List<Range> rowRanges) {
- return shouldReadRowSidecar(
- file,
- rowRanges,
- null,
-
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTED_ROWS.defaultValue(),
-
CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_MAX_SELECTION_RATIO.defaultValue());
+ file.fileSize(),
+ selection);
}
@VisibleForTesting
static boolean shouldReadRowSidecar(
DataFileMeta file,
- @Nullable List<Range> rowRanges,
+ @Nullable RoaringBitmap32 selection,
long maxSelectedRows,
double maxSelectionRatio) {
- return shouldReadRowSidecar(file, rowRanges, null, maxSelectedRows,
maxSelectionRatio);
+ long selectedRowCount = selection == null ? file.rowCount() :
selection.getCardinality();
+ return shouldReadRowSidecar(file, selectedRowCount, maxSelectedRows,
maxSelectionRatio);
}
- @VisibleForTesting
- static boolean shouldReadRowSidecar(
+ private static boolean shouldReadRowSidecar(
DataFileMeta file,
- @Nullable List<Range> rowRanges,
- @Nullable FileIndexResult fileIndexResult,
+ long selectedRowCount,
long maxSelectedRows,
double maxSelectionRatio) {
if (file.rowCount() <= 0
@@ -1599,7 +1596,6 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return false;
}
- long selectedRowCount = selectedRowCount(file, rowRanges,
fileIndexResult);
if (selectedRowCount <= 0 || selectedRowCount >= file.rowCount()) {
return false;
}
@@ -1608,33 +1604,115 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return selectedRowCount <= maxSelectedRows && selectionRatio <=
maxSelectionRatio;
}
+ @Nullable
@VisibleForTesting
- static long selectedRowCount(DataFileMeta file, List<Range> rowRanges) {
- Range fileRange = file.nonNullRowIdRange();
- List<Range> intersections = new ArrayList<>();
- for (Range range : rowRanges) {
- Range intersection = Range.intersection(fileRange, range);
- if (intersection != null) {
- intersections.add(intersection);
- }
- }
- return Range.sortAndMergeOverlap(intersections, true).stream()
- .mapToLong(Range::count)
- .sum();
+ static String rowSidecarFileName(DataFileMeta file) {
+ List<String> rowFiles =
+ file.extraFiles().stream()
+ .filter(DataEvolutionSplitRead::isRowSidecarFile)
+ .collect(Collectors.toList());
+ return rowFiles.size() == 1 ? rowFiles.get(0) : null;
}
- private static long selectedRowCount(
+ /**
+ * Builds the rows read from one physical file in file-local positions.
Null means every row.
+ */
+ @Nullable
+ @VisibleForTesting
+ static RoaringBitmap32 fileSelection(
DataFileMeta file,
@Nullable List<Range> rowRanges,
@Nullable FileIndexResult fileIndexResult) {
+ RoaringBitmap32 selection = file.toFileSelection(rowRanges);
if (fileIndexResult instanceof BitmapIndexResult) {
RoaringBitmap32 indexSelection = ((BitmapIndexResult)
fileIndexResult).get();
- RoaringBitmap32 rangeSelection = file.toFileSelection(rowRanges);
- return rangeSelection == null
- ? indexSelection.getCardinality()
- : RoaringBitmap32.and(rangeSelection,
indexSelection).getCardinality();
+ selection =
+ selection == null
+ ? indexSelection.clone()
+ : RoaringBitmap32.and(selection, indexSelection);
+ }
+ return selection;
+ }
+
+ @Nullable
+ @VisibleForTesting
+ static RoaringBitmap32 sidecarSelection(
+ DataFileMeta file,
+ @Nullable RoaringBitmap32 selection,
+ @Nullable DeletionVector deletionVector,
+ @Nullable Range deletionVectorRange) {
+ if (deletionVector == null
+ || deletionVector.isEmpty()
+ || (selection != null && selection.isEmpty())) {
+ return selection;
}
- return isNullOrEmpty(rowRanges) ? file.rowCount() :
selectedRowCount(file, rowRanges);
+ if (file.rowCount() > RoaringBitmap32.MAX_VALUE) {
+ // A 32-bit selection cannot represent all positions in this file.
Keep the existing
+ // range/index selection and let ApplyDeletionVectorReader enforce
the DV. This
+ // conservatively ignores the DV when deciding whether a row
sidecar is selective.
+ return selection;
+ }
+ return liveSelection(file, selection, deletionVector,
deletionVectorRange);
+ }
+
+ private static RoaringBitmap32 liveSelection(
+ DataFileMeta file,
+ @Nullable RoaringBitmap32 selection,
+ DeletionVector deletionVector,
+ @Nullable Range deletionVectorRange) {
+ long rangeStart =
+ deletionVectorRange == null ? file.nonNullFirstRowId() :
deletionVectorRange.from;
+ long offset = file.nonNullFirstRowId() - rangeStart;
+
+ if (offset == 0 && deletionVector instanceof BitmapDeletionVector) {
+ RoaringBitmap32 deleted = ((BitmapDeletionVector)
deletionVector).get();
+ if (!deleted.intersects(0, file.rowCount())) {
+ return selection;
+ }
+ return RoaringBitmap32.andNot(
+ selection == null
+ ? RoaringBitmap32.bitmapOfRange(0, file.rowCount())
+ : selection,
+ deleted);
+ }
+
+ if (selection != null && selection.getCardinality() <=
deletionVector.getCardinality()) {
+ RoaringBitmap32 live = new RoaringBitmap32();
+ Iterator<Integer> iterator = selection.iterator();
+ while (iterator.hasNext()) {
+ int localPosition = iterator.next();
+ long deletionPosition = localPosition + offset;
+ if (deletionPosition < 0
+ || (deletionVector instanceof BitmapDeletionVector
+ && deletionPosition >
RoaringBitmap32.MAX_VALUE)
+ || !deletionVector.isDeleted(deletionPosition)) {
+ live.add(localPosition);
+ }
+ }
+ return live;
+ }
+
+ RoaringBitmap32 deleted = deletedPositions(file, deletionVector,
offset);
+ if (deleted.isEmpty()) {
+ return selection;
+ }
+ return RoaringBitmap32.andNot(
+ selection == null ? RoaringBitmap32.bitmapOfRange(0,
file.rowCount()) : selection,
+ deleted);
+ }
+
+ private static RoaringBitmap32 deletedPositions(
+ DataFileMeta file, DeletionVector deletionVector, long offset) {
+ long rowCount = file.rowCount();
+ RoaringBitmap32 deleted = new RoaringBitmap32();
+ deletionVector.forEachDeletedPosition(
+ position -> {
+ long localPosition = position - offset;
+ if (localPosition >= 0 && localPosition < rowCount) {
+ deleted.add((int) localPosition);
+ }
+ });
+ return deleted;
}
private static boolean isRowSidecarFile(String fileName) {
@@ -1650,11 +1728,17 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
private final String formatIdentifier;
private final Path path;
private final long fileSize;
+ @Nullable private final RoaringBitmap32 selection;
- private FileReadTarget(String formatIdentifier, Path path, long
fileSize) {
+ private FileReadTarget(
+ String formatIdentifier,
+ Path path,
+ long fileSize,
+ @Nullable RoaringBitmap32 selection) {
this.formatIdentifier = formatIdentifier;
this.path = path;
this.fileSize = fileSize;
+ this.selection = selection;
}
/**
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 2a1c5c4472..8e29490970 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
@@ -23,15 +23,21 @@ 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.deletionvectors.Bitmap64DeletionVector;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.FileSystemSchemaManager;
@@ -39,6 +45,7 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -47,12 +54,16 @@ import org.apache.paimon.utils.RoaringBitmap32;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -195,16 +206,6 @@ class DataEvolutionSplitReadTest {
assertNull(DataEvolutionSplitRead.rowSidecarFileName(ambiguousSidecar));
}
- @Test
- public void testShouldReadRowSidecarForSparseRowSelection() {
- DataFileMeta file =
- createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
-
- assertTrue(
- DataEvolutionSplitRead.shouldReadRowSidecar(
- file, Arrays.asList(new Range(10L, 10L), new
Range(42L, 42L))));
- }
-
@Test
public void testShouldReadRowSidecarRequiresSmallCountAndLowRatio() {
DataFileMeta smallFile =
@@ -215,42 +216,37 @@ class DataEvolutionSplitReadTest {
assertTrue(
DataEvolutionSplitRead.shouldReadRowSidecar(
- largeFile, Collections.singletonList(new Range(10L,
4105L))));
+ largeFile, RoaringBitmap32.bitmapOfRange(0L, 4096L),
4096L, 0.05d));
assertFalse(
DataEvolutionSplitRead.shouldReadRowSidecar(
- smallFile, Collections.singletonList(new Range(10L,
15L))));
+ smallFile, RoaringBitmap32.bitmapOfRange(0L, 6L),
4096L, 0.05d));
assertFalse(
DataEvolutionSplitRead.shouldReadRowSidecar(
- largeFile, Collections.singletonList(new Range(10L,
4106L))));
+ largeFile, RoaringBitmap32.bitmapOfRange(0L, 4097L),
4096L, 0.05d));
assertTrue(
DataEvolutionSplitRead.shouldReadRowSidecar(
- smallFile, Collections.singletonList(new Range(10L,
15L)), 64L, 0.25d));
+ smallFile, RoaringBitmap32.bitmapOfRange(0L, 6L), 64L,
0.25d));
}
@Test
- public void testShouldReadRowSidecarUsesFinalBitmapSelection() {
+ public void testDeletionVectorOnlyAffectsSidecarSelection() {
DataFileMeta file =
createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
- BitmapIndexResult bitmap = new BitmapIndexResult(() ->
RoaringBitmap32.bitmapOf(0, 5, 20));
- BitmapIndexResult singleRowBitmap =
- new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(50));
- List<Range> rowRanges = Collections.singletonList(new Range(10L, 19L));
-
- assertTrue(DataEvolutionSplitRead.shouldReadRowSidecar(file,
rowRanges, bitmap, 2L, 0.02d));
- assertTrue(
- DataEvolutionSplitRead.shouldReadRowSidecar(
- file, null, singleRowBitmap, 1L, 0.01d));
- assertTrue(
- DataEvolutionSplitRead.shouldReadRowSidecar(
- file,
- Collections.singletonList(new Range(10L, 109L)),
- singleRowBitmap,
- 1L,
- 0.01d));
- assertFalse(
- DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges,
bitmap, 1L, 0.02d));
- assertFalse(
- DataEvolutionSplitRead.shouldReadRowSidecar(file, rowRanges,
bitmap, 2L, 0.01d));
+ BitmapIndexResult bitmap =
+ new BitmapIndexResult(() -> RoaringBitmap32.bitmapOf(1, 2, 4,
6, 8, 9));
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(14L); // file-local position 4 for anchor range
starting at 0
+ deletionVector.delete(18L); // file-local position 8
+ deletionVector.delete(150L); // outside this file
+
+ RoaringBitmap32 fileSelection =
+ DataEvolutionSplitRead.fileSelection(
+ file, Collections.singletonList(new Range(12L, 18L)),
bitmap);
+ assertEquals(RoaringBitmap32.bitmapOf(2, 4, 6, 8), fileSelection);
+ assertEquals(
+ RoaringBitmap32.bitmapOf(2, 6),
+ DataEvolutionSplitRead.sidecarSelection(
+ file, fileSelection, deletionVector, new Range(0L,
199L)));
}
@Test
@@ -258,32 +254,23 @@ class DataEvolutionSplitReadTest {
DataFileMeta file =
createFile("file1.parquet", 10L, 100, 1,
Collections.singletonList("file1.row"));
- assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null));
- assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file,
Collections.emptyList()));
+ assertFalse(DataEvolutionSplitRead.shouldReadRowSidecar(file, null,
4096L, 0.05d));
assertFalse(
DataEvolutionSplitRead.shouldReadRowSidecar(
- file, Collections.singletonList(new Range(10L,
109L))));
+ file, new RoaringBitmap32(), 4096L, 0.05d));
assertFalse(
DataEvolutionSplitRead.shouldReadRowSidecar(
createFile("file2.parquet", 10L, 100, 1),
- Collections.singletonList(new Range(10L, 10L))));
+ RoaringBitmap32.bitmapOf(0),
+ 4096L,
+ 0.05d));
assertFalse(
DataEvolutionSplitRead.shouldReadRowSidecar(
createFile(
"file3.blob", 10L, 100, 1,
Collections.singletonList("file3.row")),
- Collections.singletonList(new Range(10L, 10L))));
- }
-
- @Test
- public void testSelectedRowCountMergesOverlappingRanges() {
- DataFileMeta file = createFile("file1.parquet", 10L, 100, 1);
-
- assertEquals(
- 12,
- DataEvolutionSplitRead.selectedRowCount(
- file,
- Arrays.asList(
- new Range(5L, 12L), new Range(12L, 15L), new
Range(20L, 25L))));
+ RoaringBitmap32.bitmapOf(0),
+ 4096L,
+ 0.05d));
}
@Test
@@ -432,6 +419,180 @@ class DataEvolutionSplitReadTest {
assertEquals(Arrays.asList("1000:[0,1]", "5200:[4200,4201]"), actual);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testDvSelectionNarrowsRowSidecarRead(boolean bitmap64) throws
Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ Path tableRoot = new Path(tempDir.toUri().toString(), "dv-sidecar-" +
bitmap64);
+ CoreOptions coreOptions = new CoreOptions(new 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.STRING())
+ .build();
+ SchemaManager schemaManager = new FileSystemSchemaManager(fileIO,
tableRoot);
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ RowType rowType = tableSchema.logicalRowType();
+
+ Path bucketPath = pathFactory.bucketPath(EMPTY_ROW, 0);
+ fileIO.mkdirs(bucketPath);
+ int rowCount = 60_000;
+ int liveRowCount = 500;
+ String rowSidecarName = "data-0.row";
+ Path rowSidecarPath = new Path(bucketPath, rowSidecarName);
+ Options rowFormatOptions = new Options();
+ rowFormatOptions.set(CoreOptions.FILE_BLOCK_SIZE,
MemorySize.ofKibiBytes(4));
+ writeFormatFile(fileIO, rowSidecarPath, rowType, rowCount, "row",
rowFormatOptions);
+
+ DeletionVector deletionVector =
+ bitmap64 ? new Bitmap64DeletionVector() : new
BitmapDeletionVector();
+ for (int i = 0; i < rowCount - liveRowCount; i++) {
+ deletionVector.delete(i);
+ }
+ Path dvPath = new Path(bucketPath, "data-0.parquet.dv");
+ long dvSize;
+ try (PositionOutputStream out = fileIO.newOutputStream(dvPath, false);
+ DataOutputStream dataOutput = new DataOutputStream(out)) {
+ dvSize = deletionVector.serializeTo(dataOutput);
+ }
+
+ DataFileMeta dataFile =
+ createFile(
+ "data-0.parquet",
+ 0L,
+ rowCount,
+ 1,
+ Collections.singletonList(rowSidecarName));
+ DataSplit split =
+ DataSplit.builder()
+ .withPartition(EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath(bucketPath.toString())
+ .withDataFiles(Collections.singletonList(dataFile))
+ .withDataDeletionFiles(
+ Collections.singletonList(
+ new DeletionFile(
+ dvPath.toString(),
+ 0,
+ dvSize,
+ (long) (rowCount -
liveRowCount))))
+ .rawConvertible(false)
+ .build();
+
+ CountingFileIO countingFileIO = new CountingFileIO(fileIO,
rowSidecarPath);
+ DataEvolutionSplitRead splitRead =
+ new DataEvolutionSplitRead(
+ countingFileIO,
+ schemaManager,
+ tableSchema,
+ rowType,
+ coreOptions,
+ pathFactory);
+
+ List<Integer> actual = new ArrayList<>();
+ try (RecordReader<InternalRow> reader = splitRead.createReader(split))
{
+ reader.forEachRemaining(row -> actual.add(row.getInt(0)));
+ }
+
+ assertEquals(liveRowCount, actual.size());
+ assertEquals(1000 + rowCount - liveRowCount, (int) actual.get(0));
+ assertEquals(1000 + rowCount - 1, (int) actual.get(actual.size() - 1));
+ long sidecarSize = fileIO.getFileStatus(rowSidecarPath).getLen();
+
+ CountingFileIO fullScanFileIO = new CountingFileIO(fileIO,
rowSidecarPath);
+ RecordReader<InternalRow> fullScanReader =
+ FileFormat.fromIdentifier("row", rowFormatOptions)
+ .createReaderFactory(rowType, rowType, null)
+ .createReader(
+ new FormatReaderContext(
+ fullScanFileIO, rowSidecarPath,
sidecarSize, null, null));
+ fullScanReader.forEachRemaining(ignored -> {});
+
+ assertTrue(
+ countingFileIO.bytesRead() < fullScanFileIO.bytesRead(),
+ "Selection read "
+ + countingFileIO.bytesRead()
+ + " sidecar bytes; full scan read "
+ + fullScanFileIO.bytesRead()
+ + ".");
+ }
+
+ private static final class CountingFileIO extends LocalFileIO {
+
+ private final LocalFileIO fileIO;
+ private final Path countedPath;
+ private final AtomicLong bytesRead = new AtomicLong();
+
+ private CountingFileIO(LocalFileIO fileIO, Path countedPath) {
+ this.fileIO = fileIO;
+ this.countedPath = countedPath;
+ }
+
+ private long bytesRead() {
+ return bytesRead.get();
+ }
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ SeekableInputStream input = fileIO.newInputStream(path);
+ if (!countedPath.equals(path)) {
+ return input;
+ }
+ return new SeekableInputStream() {
+ @Override
+ public void seek(long desired) throws IOException {
+ input.seek(desired);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return input.getPos();
+ }
+
+ @Override
+ public int read(byte[] buffer, int offset, int length) throws
IOException {
+ int read = input.read(buffer, offset, length);
+ if (read > 0) {
+ bytesRead.addAndGet(read);
+ }
+ return read;
+ }
+
+ @Override
+ public int read() throws IOException {
+ int read = input.read();
+ if (read >= 0) {
+ bytesRead.incrementAndGet();
+ }
+ return read;
+ }
+
+ @Override
+ public void close() throws IOException {
+ input.close();
+ }
+ };
+ }
+ }
+
private static void writeNestedParquetFile(
LocalFileIO fileIO, Path path, RowType rowType, int rowCount,
Options options)
throws IOException {
