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 d2be7eaced [core] Introducing DeletionVector mechanism for
DataEvolution tables (#8380)
d2be7eaced is described below
commit d2be7eacedf7c001ed1022856a62ade089e954c4
Author: Faiz <[email protected]>
AuthorDate: Tue Jun 30 18:54:18 2026 +0800
[core] Introducing DeletionVector mechanism for DataEvolution tables (#8380)
---
.../DataEvolutionCompactCoordinator.java | 12 +
.../dataevolution/DataEvolutionCompactTask.java | 7 +-
.../ApplyDeletionFileRecordIterator.java | 10 +-
.../deletionvectors/ApplyDeletionVectorReader.java | 20 +-
.../append/AppendDeleteFileMaintainer.java | 14 +-
.../operation/AllPlaceholdersRecordReader.java | 149 ++++++
.../paimon/operation/BlobFallbackRecordReader.java | 251 ++++-----
.../operation/DataEvolutionFileStoreScan.java | 12 +
.../paimon/operation/DataEvolutionSplitRead.java | 206 ++++---
.../org/apache/paimon/schema/SchemaValidation.java | 3 -
.../org/apache/paimon/table/source/DataSplit.java | 24 +-
.../apache/paimon/utils/DataEvolutionUtils.java | 64 +++
.../DataEvolutionCompactCoordinatorTest.java | 36 ++
.../paimon/deletionvectors/DeletionVectorTest.java | 114 ++++
.../operation/AllPlaceholdersRecordReaderTest.java | 103 ++++
.../operation/BlobFallbackRecordReaderTest.java | 169 +++++-
.../table/DataEvolutionDeletionVectorTest.java | 590 +++++++++++++++++++++
.../table/source/DataSplitCompatibleTest.java | 25 +
.../paimon/utils/DataEvolutionUtilsTest.java | 93 ++++
19 files changed, 1669 insertions(+), 233 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
index 3c1496b851..d183a8ebb5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
@@ -64,6 +64,7 @@ public class DataEvolutionCompactCoordinator {
private static final int FILES_BATCH = 100_000;
private static final int BLOB_COMPACT_MIN_FILE_NUM = 2;
+ private final boolean deletionVectorsEnabled;
private final CompactScanner scanner;
private final CompactPlanner planner;
@@ -78,6 +79,13 @@ public class DataEvolutionCompactCoordinator {
boolean compactBlob,
boolean compactVector) {
CoreOptions options = table.coreOptions();
+ this.deletionVectorsEnabled = options.deletionVectorsEnabled();
+ if (deletionVectorsEnabled) {
+ this.scanner = null;
+ this.planner = null;
+ return;
+ }
+
long targetFileSize = options.targetFileSize(false);
long openFileCost = options.splitOpenFileCost();
long compactMinFileNum = options.compactionMinFileNum();
@@ -109,6 +117,10 @@ public class DataEvolutionCompactCoordinator {
}
public List<DataEvolutionCompactTask> plan() {
+ if (deletionVectorsEnabled) {
+ throw new EndOfScanException();
+ }
+
// scan files in snapshot
List<ManifestEntry> entries = scanner.scan();
if (!entries.isEmpty()) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
index f8bdc959c3..c2ee672844 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
@@ -96,6 +96,11 @@ public class DataEvolutionCompactTask extends
AppendCompactTask {
}
public CommitMessage doCompact(FileStoreTable table, String commitUser)
throws Exception {
+ CoreOptions options = table.coreOptions();
+ checkArgument(
+ !options.deletionVectorsEnabled(),
+ "Data evolution compaction does not support deletion
vectors.");
+
if (blobTask) {
return doCompactBlobFiles(table, commitUser);
}
@@ -104,8 +109,6 @@ public class DataEvolutionCompactTask extends
AppendCompactTask {
throw new UnsupportedOperationException("Vector-store task is not
supported");
}
- CoreOptions options = table.coreOptions();
-
Set<String> fieldsInDedicatedFile =
SetUtils.union(
fieldNamesInBlobFile(table.rowType(),
options.blobInlineField()),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
index 473b0fef53..39162c4783 100644
---
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionFileRecordIterator.java
@@ -26,15 +26,17 @@ import javax.annotation.Nullable;
import java.io.IOException;
-/** A {@link FileRecordIterator} wraps a {@link FileRecordIterator} and {@link
DeletionVector}. */
+/**
+ * A {@link FileRecordIterator} wraps a {@link FileRecordIterator} and {@link
DeletionVectorJudger}.
+ */
public class ApplyDeletionFileRecordIterator
implements FileRecordIterator<InternalRow>, DeletionFileRecordIterator
{
private final FileRecordIterator<InternalRow> iterator;
- private final DeletionVector deletionVector;
+ private final DeletionVectorJudger deletionVector;
public ApplyDeletionFileRecordIterator(
- FileRecordIterator<InternalRow> iterator, DeletionVector
deletionVector) {
+ FileRecordIterator<InternalRow> iterator, DeletionVectorJudger
deletionVector) {
this.iterator = iterator;
this.deletionVector = deletionVector;
}
@@ -45,7 +47,7 @@ public class ApplyDeletionFileRecordIterator
}
@Override
- public DeletionVector deletionVector() {
+ public DeletionVectorJudger deletionVector() {
return deletionVector;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java
index 2fc292e54d..7787ac4b5a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java
@@ -32,19 +32,33 @@ public class ApplyDeletionVectorReader implements
FileRecordReader<InternalRow>
private final FileRecordReader<InternalRow> reader;
- private final DeletionVector deletionVector;
+ private final DeletionVectorJudger deletionVector;
public ApplyDeletionVectorReader(
FileRecordReader<InternalRow> reader, DeletionVector
deletionVector) {
+ this(reader, deletionVector, 0L);
+ }
+
+ /**
+ * @param fileOffset offset from this reader's local returned position to
the deletion vector
+ * position. The wrapped judger is converted to reader-local positions
here, so both {@link
+ * ApplyDeletionFileRecordIterator#next()} and external consumers of
{@link
+ * DeletionFileRecordIterator#deletionVector()} use the same position
mapping.
+ */
+ public ApplyDeletionVectorReader(
+ FileRecordReader<InternalRow> reader, DeletionVector
deletionVector, long fileOffset) {
this.reader = reader;
- this.deletionVector = deletionVector;
+ this.deletionVector =
+ fileOffset == 0
+ ? deletionVector
+ : position -> deletionVector.isDeleted(fileOffset +
position);
}
public RecordReader<InternalRow> reader() {
return reader;
}
- public DeletionVector deletionVector() {
+ public DeletionVectorJudger deletionVector() {
return deletionVector;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
index 70e00bc4ad..93ea8a7676 100644
---
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java
@@ -37,7 +37,19 @@ import java.util.Set;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
-/** A {@link BaseAppendDeleteFileMaintainer} of unaware bucket append table. */
+/**
+ * A {@link BaseAppendDeleteFileMaintainer} for an unaware-bucket append
table. This class maintains
+ * a mapping from data file names to DeletionVectors. It has different
semantics for normal append
+ * tables and data evolution tables:
+ *
+ * <ul>
+ * <li>For append tables, each entry maps a file to its corresponding
deletion vector.
+ * <li>For data evolution tables, a row is logically composed of several
files covering the same
+ * row range. The key becomes the oldest file in the group (i.e., the
one with the smallest
+ * max_seq_num). For each file group, the oldest file is used to look up
the corresponding
+ * deletion vectors.
+ * </ul>
+ */
public class AppendDeleteFileMaintainer implements
BaseAppendDeleteFileMaintainer {
private final DeletionVectorsIndexFile dvIndexFile;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
new file mode 100644
index 0000000000..5a087439b0
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AllPlaceholdersRecordReader.java
@@ -0,0 +1,149 @@
+/*
+ * 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.paimon.operation;
+
+import org.apache.paimon.data.BlobPlaceholder;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+
+/** A {@link FileRecordReader} which emits blob placeholder rows for a row-id
range. */
+class AllPlaceholdersRecordReader implements FileRecordReader<InternalRow> {
+
+ private static final Path PLACEHOLDER_PATH = new Path("placeholder");
+
+ private final long firstRowId;
+ private final int fieldCount;
+ private final int blobIndex;
+ private final int rowIdIndex;
+ private final int seqNumIndex;
+ private final long sequenceNumber;
+ private final List<Range> selectedRanges;
+ private boolean returned;
+
+ AllPlaceholdersRecordReader(
+ long firstRowId,
+ long rowCount,
+ @Nullable List<Range> rowRanges,
+ RowType readRowType,
+ int blobIndex,
+ long sequenceNumber) {
+ this.firstRowId = firstRowId;
+ this.fieldCount = readRowType.getFieldCount();
+ this.blobIndex = blobIndex;
+ this.rowIdIndex =
readRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ this.seqNumIndex =
readRowType.getFieldIndex(SpecialFields.SEQUENCE_NUMBER.name());
+ this.sequenceNumber = sequenceNumber;
+ this.selectedRanges = selectedRanges(firstRowId, rowCount, rowRanges);
+ }
+
+ @Nullable
+ @Override
+ public FileRecordIterator<InternalRow> readBatch() {
+ if (returned || selectedRanges.isEmpty()) {
+ return null;
+ }
+ returned = true;
+ return new PlaceholderIterator();
+ }
+
+ @Override
+ public void close() {
+ // nothing to close
+ }
+
+ private List<Range> selectedRanges(
+ long firstRowId, long rowCount, @Nullable List<Range> rowRanges) {
+ if (rowCount <= 0) {
+ return Collections.emptyList();
+ }
+
+ List<Range> fullRange =
+ Collections.singletonList(new Range(firstRowId, firstRowId +
rowCount - 1));
+ if (rowRanges == null) {
+ return fullRange;
+ }
+
+ return Range.and(fullRange, Range.sortAndMergeOverlap(rowRanges));
+ }
+
+ private InternalRow placeholderRow(long rowId) {
+ GenericRow row = new GenericRow(fieldCount);
+ row.setField(blobIndex, BlobPlaceholder.INSTANCE);
+ if (rowIdIndex >= 0) {
+ row.setField(rowIdIndex, rowId);
+ }
+ if (seqNumIndex >= 0) {
+ row.setField(seqNumIndex, sequenceNumber);
+ }
+ return row;
+ }
+
+ /** Iterator to emit placeholders with row ranges pushed. */
+ private class PlaceholderIterator implements
FileRecordIterator<InternalRow> {
+
+ private int rangeIndex = 0;
+ private long nextRowId = selectedRanges.get(0).from;
+ private long returnedRowId = firstRowId - 1;
+
+ @Override
+ public long returnedPosition() {
+ return returnedRowId - firstRowId;
+ }
+
+ @Override
+ public Path filePath() {
+ return PLACEHOLDER_PATH;
+ }
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ while (rangeIndex < selectedRanges.size()) {
+ Range range = selectedRanges.get(rangeIndex);
+ if (nextRowId <= range.to) {
+ returnedRowId = nextRowId;
+ nextRowId++;
+ return placeholderRow(returnedRowId);
+ }
+
+ rangeIndex++;
+ if (rangeIndex < selectedRanges.size()) {
+ nextRowId = selectedRanges.get(rangeIndex).from;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void releaseBatch() {
+ // nothing to release
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
index 30f214afcf..3a81b4d791 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java
@@ -23,7 +23,11 @@ import org.apache.paimon.data.BlobPlaceholder;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.mergetree.compact.ConcatRecordReader;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.reader.ReaderSupplier;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.Range;
@@ -57,16 +61,22 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
private final List<RecordReader<InternalRow>> groupReaders = new
ArrayList<>();
private final int blobIndex;
private final int fieldCount;
+ private final int rowIdIndex;
+ private final int seqNumIndex;
private boolean returned;
BlobFallbackRecordReader(
List<DataFileMeta> files,
BlobFileReaderFactory readerFactory,
+ ReaderWrapper readerWrapper,
List<Range> rowRanges,
RowType readRowType,
- int blobIndex) {
+ int blobIndex)
+ throws IOException {
this.blobIndex = blobIndex;
this.fieldCount = readRowType.getFieldCount();
+ this.rowIdIndex =
readRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ this.seqNumIndex =
readRowType.getFieldIndex(SpecialFields.SEQUENCE_NUMBER.name());
checkArgument(!files.isEmpty(), "Blob bunch should not be empty.");
long firstRowId = Long.MAX_VALUE;
@@ -104,8 +114,10 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
groupReaders.add(
new ForceSingleBatchReader(
new BlobSequenceGroupRecordReader(
+ entry.getKey(),
groupFiles,
readerFactory,
+ readerWrapper,
rowRanges,
readRowType,
blobIndex,
@@ -146,6 +158,7 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
@Override
public InternalRow next() throws IOException {
InternalRow result = null;
+ long rowId = -1L;
// We should always move each iterator forward
// This may significantly increase memory usage and decrease
read efficiency
// if `blob-as-descriptor` is disabled and many non-null blobs
are updated
@@ -172,9 +185,12 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
if (result == null && !isPlaceHolder(row)) {
result = row;
}
+ if (rowIdIndex >= 0 && rowId < 0) {
+ rowId = row.getLong(rowIdIndex);
+ }
}
if (result == null) {
- result = nullBlobRow();
+ result = nullBlobRow(rowId);
}
return result;
}
@@ -188,8 +204,16 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
};
}
- private InternalRow nullBlobRow() {
- return new GenericRow(fieldCount);
+ private InternalRow nullBlobRow(long rowId) {
+ GenericRow row = new GenericRow(fieldCount);
+ if (rowIdIndex >= 0) {
+ row.setField(rowIdIndex, rowId);
+ }
+ // Set seq num as -1 to mark this row as an all-placeholder null
+ if (seqNumIndex >= 0) {
+ row.setField(seqNumIndex, -1L);
+ }
+ return row;
}
private boolean isPlaceHolder(InternalRow row) {
@@ -262,172 +286,105 @@ public class BlobFallbackRecordReader implements
RecordReader<InternalRow> {
*/
public static class BlobSequenceGroupRecordReader implements
RecordReader<InternalRow> {
- private final List<DataFileMeta> files;
- private final BlobFileReaderFactory readerFactory;
- // pushed row ranges
- private final List<Range> rowRanges;
- private final RowType readRowType;
- private final int blobIndex;
- private final long lastRowId;
-
- private RecordReader<InternalRow> currentReader;
- private DataFileMeta currentFile;
- private int nextFileIndex;
- private int nextRowRangeIndex;
- // expected next row id
- private long nextRowId;
-
- private InternalRow placeholderRow;
+ private final RecordReader<InternalRow> reader;
BlobSequenceGroupRecordReader(
+ long maxSeq,
List<DataFileMeta> files,
BlobFileReaderFactory readerFactory,
+ ReaderWrapper readerWrapper,
List<Range> rowRanges,
RowType readRowType,
int blobIndex,
long firstRowId,
- long lastRowId) {
- this.files = files;
- this.readerFactory = readerFactory;
- this.rowRanges = rowRanges == null ? null :
Range.sortAndMergeOverlap(rowRanges);
- this.readRowType = readRowType;
- this.blobIndex = blobIndex;
- this.lastRowId = lastRowId;
-
- this.nextFileIndex = 0;
- this.nextRowRangeIndex = 0;
- setNextRowId(firstRowId);
-
- this.placeholderRow = null;
- }
-
- @Nullable
- @Override
- public RecordIterator<InternalRow> readBatch() throws IOException {
- while (true) {
- if (currentReader != null) {
- RecordIterator<InternalRow> batch =
currentReader.readBatch();
- if (batch != null) {
- return batch;
- }
- // row ranges have been pushed to readers
- // directly set nextRowId as the lastRowId + 1
- setNextRowId(lastRowId(currentFile) + 1);
- closeCurrentFileReader();
- continue;
- }
-
- if (nextRowId > lastRowId) {
- return null;
- }
-
- // skip files whose ranges are before nextRowId
- while (nextFileIndex < files.size()
- && lastRowId(files.get(nextFileIndex)) < nextRowId) {
- nextFileIndex++;
- }
- if (nextFileIndex >= files.size()) {
- return placeHolderBatch(lastRowId);
- }
-
- DataFileMeta nextFile = files.get(nextFileIndex);
- if (nextFile.nonNullFirstRowId() > nextRowId) {
- return placeHolderBatch(nextFile.nonNullFirstRowId() - 1);
- }
-
- createReader(nextFile);
- }
- }
-
- /**
- * Set nextRowId and try to move to the next selected row id. So the
final nextRowId may be
- * greater than the input value.
- */
- private void setNextRowId(long nextRowId) {
- this.nextRowId = nextRowId;
- tryMoveToSelectedRow();
- }
-
- private void tryMoveToSelectedRow() {
- if (nextRowId > lastRowId || rowRanges == null) {
- return;
- }
-
- while (nextRowRangeIndex < rowRanges.size()) {
- Range range = rowRanges.get(nextRowRangeIndex);
- if (nextRowId >= range.from && nextRowId <= range.to) {
- // if nextRowId is within the range, do not need to move
- return;
- } else if (nextRowId < range.from) {
- // else if nextRowId < next range, move to next range's
`from`
- nextRowId = range.from;
- return;
- }
- // else nextRowId > range.to, try next range
- nextRowRangeIndex++;
- }
-
- // all ranges consumed, no need to read
- nextRowId = lastRowId + 1;
+ long lastRowId)
+ throws IOException {
+ this.reader =
+ ConcatRecordReader.create(
+ createReaders(
+ maxSeq,
+ files,
+ readerFactory,
+ readerWrapper,
+ rowRanges,
+ readRowType,
+ blobIndex,
+ firstRowId,
+ lastRowId));
}
- private RecordIterator<InternalRow> placeHolderBatch(long endRowId) {
- return new RecordIterator<InternalRow>() {
- long rowId;
-
- @Nullable
- @Override
- public InternalRow next() {
- rowId = nextRowId;
- if (rowId > endRowId) {
- return null;
- }
- setNextRowId(rowId + 1);
- return placeHolderRow();
- }
-
- @Override
- public void releaseBatch() {
- // nothing to release
+ private List<ReaderSupplier<InternalRow>> createReaders(
+ long maxSeq,
+ List<DataFileMeta> files,
+ BlobFileReaderFactory readerFactory,
+ ReaderWrapper readerWrapper,
+ List<Range> rowRanges,
+ RowType readRowType,
+ int blobIndex,
+ long firstRowId,
+ long lastRowId) {
+ List<ReaderSupplier<InternalRow>> suppliers = new ArrayList<>();
+ long nextRowId = firstRowId;
+ for (DataFileMeta file : files) {
+ Range fileRange = file.nonNullRowIdRange();
+ if (nextRowId < fileRange.from) {
+ long gapFirstRowId = nextRowId;
+ long gapRowCount = fileRange.from - gapFirstRowId;
+ Range gapRange = new Range(gapFirstRowId, fileRange.from -
1);
+ suppliers.add(
+ () ->
+ readerWrapper.wrap(
+ new AllPlaceholdersRecordReader(
+ gapFirstRowId,
+ gapRowCount,
+ rowRanges,
+ readRowType,
+ blobIndex,
+ maxSeq),
+ gapRange));
}
- };
- }
-
- private InternalRow placeHolderRow() {
- if (placeholderRow == null) {
- GenericRow row = new GenericRow(readRowType.getFieldCount());
- row.setField(blobIndex, BlobPlaceholder.INSTANCE);
- placeholderRow = row;
+ suppliers.add(() -> readerFactory.create(file));
+ nextRowId = fileRange.to + 1;
}
- return placeholderRow;
- }
-
- private long lastRowId(DataFileMeta file) {
- return file.nonNullFirstRowId() + file.rowCount() - 1;
- }
-
- private void closeCurrentFileReader() throws IOException {
- if (currentReader != null) {
- currentReader.close();
- currentReader = null;
+ if (nextRowId <= lastRowId) {
+ long gapFirstRowId = nextRowId;
+ long gapRowCount = lastRowId - gapFirstRowId + 1;
+ Range gapRange = new Range(gapFirstRowId, lastRowId);
+ suppliers.add(
+ () ->
+ readerWrapper.wrap(
+ new AllPlaceholdersRecordReader(
+ gapFirstRowId,
+ gapRowCount,
+ rowRanges,
+ readRowType,
+ blobIndex,
+ maxSeq),
+ gapRange));
}
- currentFile = null;
+ return suppliers;
}
- private void createReader(DataFileMeta nextFile) throws IOException {
- currentFile = nextFile;
- currentReader = readerFactory.create(nextFile);
- nextFileIndex++;
+ @Nullable
+ @Override
+ public RecordIterator<InternalRow> readBatch() throws IOException {
+ return reader.readBatch();
}
@Override
public void close() throws IOException {
- closeCurrentFileReader();
+ reader.close();
}
}
/** Factory to create readers. */
interface BlobFileReaderFactory {
- RecordReader<InternalRow> create(DataFileMeta file) throws IOException;
+ FileRecordReader<InternalRow> create(DataFileMeta file) throws
IOException;
+ }
+
+ /** Wraps placeholder readers with shared post-processing. e.g. Apply
Deletion Vectors. */
+ interface ReaderWrapper {
+ FileRecordReader<InternalRow> wrap(FileRecordReader<InternalRow>
reader, Range range)
+ throws IOException;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
index b5ea80b060..e097b3be53 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java
@@ -62,12 +62,14 @@ import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
/** {@link FileStoreScan} for data-evolution enabled table. */
public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan {
private boolean dropStats = false;
@Nullable private RowType readType;
+ private final boolean deletionVectorsEnabled;
// Cache file's physical field id set per (schemaId, writeCols) to avoid
recomputing during
// per-file column pruning in postFilterManifestEntries.
@@ -94,6 +96,7 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
false,
deletionVectorsEnabled,
true);
+ this.deletionVectorsEnabled = deletionVectorsEnabled;
}
@Override
@@ -140,6 +143,7 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
if (inputFilter != null
|| limit == null
|| limit <= 0
+ || deletionVectorsEnabled
|| !allContainsRowId(manifestFiles)) {
return super.readManifestEntries(manifestFiles, useSequential);
}
@@ -212,11 +216,16 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
* <p>When every file in the group lacks a requested column (e.g. an ADD
COLUMN projection over
* a row-disjoint pre-ALTER group), one file is kept as a row-count
representative so the reader
* can emit the right number of NULL-filled rows.
+ *
+ * <p>If Deletion-Vector is enabled, we always keep the oldest normal file
for each group as the
+ * anchor file to lookup corresponding Deletion Files.
*/
private List<ManifestEntry> pruneByReadType(List<ManifestEntry> group) {
if (readType == null || group.size() <= 1) {
return group;
}
+ ManifestEntry anchor =
+ deletionVectorsEnabled ? retrieveAnchorFile(group,
ManifestEntry::file) : null;
Set<Integer> readFieldIds = new HashSet<>();
for (DataField f : readType.getFields()) {
readFieldIds.add(f.id());
@@ -231,6 +240,9 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
}
}
}
+ if (anchor != null && !kept.contains(anchor)) {
+ kept.add(anchor);
+ }
// Group must contribute at least one file so the reader sees rowCount
and can NULL-fill
// missing columns for the projection's rows.
return kept.isEmpty() ? Collections.singletonList(group.get(0)) : kept;
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 31ff6cd614..1fd7f55c3a 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
@@ -23,6 +23,8 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.append.ForceSingleBatchReader;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
+import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatKey;
@@ -45,6 +47,7 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypeRoot;
@@ -78,6 +81,7 @@ import static java.util.Comparator.comparingLong;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.table.SpecialFields.rowTypeWithRowTracking;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -160,6 +164,13 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
private RecordReader<InternalRow> createReader(
DataSplit dataSplit, List<Range> rowRanges, RowType readRowType)
throws IOException {
List<DataFileMeta> files = dataSplit.dataFiles();
+
+ List<DeletionFile> deletionFiles =
dataSplit.deletionFiles().orElse(null);
+ DeletionVector.Factory deletionVectorFactory =
+ deletionFiles == null ||
deletionFiles.stream().allMatch(Objects::isNull)
+ ? null
+ : DeletionVector.factory(fileIO, files, deletionFiles);
+
BinaryRow partition = dataSplit.partition();
DataFilePathFactory dataFilePathFactory =
pathFactory.createDataFilePathFactory(partition,
dataSplit.bucket());
@@ -182,25 +193,33 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
if (needMergeFiles.size() == 1 ||
readRowType.getFields().isEmpty()) {
// No need to merge fields, just create a single file reader
suppliers.add(
- () ->
- createFileReader(
- partition,
- dataFilePathFactory,
- needMergeFiles.get(0),
- formatBuilder,
- rowRanges,
- readRowType));
+ () -> {
+ DeletionVectorWithRange deletionVector =
+ readDeletionVector(needMergeFiles,
deletionVectorFactory);
+ return createFileReader(
+ partition,
+ dataFilePathFactory,
+ needMergeFiles.get(0),
+ formatBuilder,
+ rowRanges,
+ readRowType,
+ deletionVector);
+ });
} else {
suppliers.add(
- () ->
- createUnionReader(
- needMergeFiles,
- partition,
- dataFilePathFactory,
- formatBuilder,
- rowRanges,
- readRowType));
+ () -> {
+ DeletionVectorWithRange deletionVector =
+ readDeletionVector(needMergeFiles,
deletionVectorFactory);
+ return createUnionReader(
+ needMergeFiles,
+ partition,
+ dataFilePathFactory,
+ formatBuilder,
+ rowRanges,
+ readRowType,
+ deletionVector);
+ });
}
}
@@ -222,7 +241,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataFilePathFactory dataFilePathFactory,
Builder formatBuilder,
List<Range> rowRanges,
- RowType readRowType)
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
List<FieldBunch> fieldsFiles =
splitFieldBunches(
@@ -318,7 +338,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
dataFilePathFactory,
formatReaderMapping,
rowRanges,
- partialReadRowType));
+ partialReadRowType,
+ deletionVector));
}
}
@@ -341,7 +362,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataFilePathFactory dataFilePathFactory,
FormatReaderMapping formatReaderMapping,
List<Range> rowRanges,
- RowType readRowType)
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
if (bunch instanceof DataBunch) {
// for data bunch, directly read the single file
@@ -351,11 +373,17 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
dataFilePathFactory,
formatReaderMapping,
rowRanges,
- readRowType);
+ readRowType,
+ deletionVector);
} else if (bunch instanceof VectorFileBunch) {
// for vector bunch, sequential read all data files and concat them
return sequentialReadFiles(
- bunch.files(), partition, dataFilePathFactory,
formatReaderMapping, rowRanges);
+ bunch.files(),
+ partition,
+ dataFilePathFactory,
+ formatReaderMapping,
+ rowRanges,
+ deletionVector);
} else if (bunch instanceof BlobFileBunch) {
// for blob bunch, fallback on placeholders
@@ -366,7 +394,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
partition,
dataFilePathFactory,
formatReaderMapping,
- rowRanges);
+ rowRanges,
+ deletionVector);
}
int blobIndex = findBlobFieldIndex(readRowType);
checkArgument(blobIndex >= 0, "Blob bunch read type should contain
a blob field.");
@@ -379,7 +408,9 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
dataFilePathFactory,
formatReaderMapping,
rowRanges,
- readRowType),
+ readRowType,
+ deletionVector),
+ (reader, range) -> applyDeletionVector(reader, range,
deletionVector),
rowRanges,
readRowType,
blobIndex);
@@ -393,30 +424,24 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
BinaryRow partition,
DataFilePathFactory dataFilePathFactory,
FormatReaderMapping formatReaderMapping,
- List<Range> rowRanges)
+ List<Range> rowRanges,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
List<ReaderSupplier<InternalRow>> readerSuppliers = new ArrayList<>();
for (DataFileMeta file : files) {
- RoaringBitmap32 selection = file.toFileSelection(rowRanges);
- FormatReaderContext formatReaderContext =
- new FormatReaderContext(
- fileIO, dataFilePathFactory.toPath(file),
file.fileSize(), selection);
readerSuppliers.add(
() ->
- new DataFileRecordReader(
+ createFileReader(
+ partition,
+ file,
+ formatReaderMapping,
+ rowRanges,
readRowType,
- formatReaderMapping.getReaderFactory(),
- formatReaderContext,
- coreOptions.scanIgnoreCorruptFile(),
- coreOptions.scanIgnoreLostFile(),
- formatReaderMapping.getIndexMapping(),
- formatReaderMapping.getCastMapping(),
- PartitionUtils.create(
-
formatReaderMapping.getPartitionPair(), partition),
- true,
- file.firstRowId(),
- file.maxSequenceNumber(),
- formatReaderMapping.getSystemFields()));
+ new FileReadTarget(
+
DataFilePathFactory.formatIdentifier(file.fileName()),
+ dataFilePathFactory.toPath(file),
+ file.fileSize()),
+ deletionVector));
}
return ConcatRecordReader.create(readerSuppliers);
}
@@ -436,7 +461,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataFileMeta file,
Builder formatBuilder,
List<Range> rowRanges,
- RowType readRowType)
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
FileReadTarget readTarget = readTarget(file, dataFilePathFactory,
rowRanges);
String formatIdentifier = readTarget.formatIdentifier;
@@ -452,7 +478,13 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
? schema
:
schemaFetcher.apply(schemaId)));
return createFileReader(
- partition, file, formatReaderMapping, rowRanges, readRowType,
readTarget);
+ partition,
+ file,
+ formatReaderMapping,
+ rowRanges,
+ readRowType,
+ readTarget,
+ deletionVector);
}
private FileRecordReader<InternalRow> createFileReader(
@@ -461,7 +493,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataFilePathFactory dataFilePathFactory,
FormatReaderMapping formatReaderMapping,
List<Range> rowRanges,
- RowType readRowType)
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
return createFileReader(
partition,
@@ -469,7 +502,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
formatReaderMapping,
rowRanges,
readRowType,
- readTarget(file, dataFilePathFactory, rowRanges));
+ readTarget(file, dataFilePathFactory, rowRanges),
+ deletionVector);
}
private FileRecordReader<InternalRow> createFileReader(
@@ -478,24 +512,67 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
FormatReaderMapping formatReaderMapping,
List<Range> rowRanges,
RowType readRowType,
- FileReadTarget readTarget)
+ FileReadTarget readTarget,
+ @Nullable DeletionVectorWithRange deletionVector)
throws IOException {
RoaringBitmap32 selection = file.toFileSelection(rowRanges);
FormatReaderContext formatReaderContext =
new FormatReaderContext(fileIO, readTarget.path,
readTarget.fileSize, selection);
- return new DataFileRecordReader(
- readRowType,
- formatReaderMapping.getReaderFactory(),
- formatReaderContext,
- coreOptions.scanIgnoreCorruptFile(),
- coreOptions.scanIgnoreLostFile(),
- formatReaderMapping.getIndexMapping(),
- formatReaderMapping.getCastMapping(),
- PartitionUtils.create(formatReaderMapping.getPartitionPair(),
partition),
- true,
- file.firstRowId(),
- file.maxSequenceNumber(),
- formatReaderMapping.getSystemFields());
+ FileRecordReader<InternalRow> fileRecordReader =
+ new DataFileRecordReader(
+ readRowType,
+ formatReaderMapping.getReaderFactory(),
+ formatReaderContext,
+ coreOptions.scanIgnoreCorruptFile(),
+ coreOptions.scanIgnoreLostFile(),
+ formatReaderMapping.getIndexMapping(),
+ formatReaderMapping.getCastMapping(),
+
PartitionUtils.create(formatReaderMapping.getPartitionPair(), partition),
+ true,
+ file.firstRowId(),
+ file.maxSequenceNumber(),
+ formatReaderMapping.getSystemFields());
+ return applyDeletionVector(fileRecordReader, file.nonNullRowIdRange(),
deletionVector);
+ }
+
+ private FileRecordReader<InternalRow> applyDeletionVector(
+ FileRecordReader<InternalRow> reader,
+ Range readerRange,
+ @Nullable DeletionVectorWithRange deletionVector) {
+ if (deletionVector == null || deletionVector.deletionVector.isEmpty())
{
+ return reader;
+ }
+
+ checkArgument(
+ deletionVector.range.from <= readerRange.from
+ && deletionVector.range.to >= readerRange.to,
+ "Deletion vector range %s should contain reader range %s.",
+ deletionVector.range,
+ readerRange);
+
+ return new ApplyDeletionVectorReader(
+ reader,
+ deletionVector.deletionVector,
+ // Convert anchor-range DV positions to this reader's local
returned positions.
+ readerRange.from - deletionVector.range.from);
+ }
+
+ @Nullable
+ private DeletionVectorWithRange readDeletionVector(
+ List<DataFileMeta> group, @Nullable DeletionVector.Factory
deletionVectorFactory)
+ throws IOException {
+ if (deletionVectorFactory == null) {
+ return null;
+ }
+
+ // pack row ranges and deletion vector
+ DataFileMeta anchor = retrieveAnchorFile(group, Function.identity());
+ Range range = anchor.nonNullRowIdRange();
+
+ return deletionVectorFactory
+ .create(anchor.fileName())
+ .map(dv -> new DeletionVectorWithRange(range, dv))
+ .orElse(null);
}
private FileReadTarget readTarget(
@@ -603,6 +680,17 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
}
}
+ private static class DeletionVectorWithRange {
+
+ private final Range range;
+ private final DeletionVector deletionVector;
+
+ private DeletionVectorWithRange(Range range, DeletionVector
deletionVector) {
+ this.range = range;
+ this.deletionVector = deletionVector;
+ }
+ }
+
@VisibleForTesting
public static List<FieldBunch> splitFieldBunches(
List<DataFileMeta> needMergeFiles, Function<DataFileMeta, RowType>
fileToRowType) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index b67c94ac81..491da71ff1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -894,9 +894,6 @@ public class SchemaValidation {
checkArgument(
rowTrackingEnabled,
"Data evolution config must enabled with
row-tracking.enabled");
- checkArgument(
- !options.deletionVectorsEnabled(),
- "Data evolution config must disabled with
deletion-vectors.enabled");
checkArgument(
!options.clusteringIncrementalEnabled(),
"Data evolution config must disabled with
clustering.incremental");
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
index 2ba4410f34..df31763a30 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
@@ -72,6 +72,11 @@ public class DataSplit implements Split {
@Nullable private Integer totalBuckets;
private List<DataFileMeta> dataFiles;
+
+ /**
+ * This list should have the same size as dataFiles. For data evolution
tables, only anchor
+ * files would have corresponding deletion file.
+ */
@Nullable private List<DeletionFile> dataDeletionFiles;
private boolean isStreaming = false;
@@ -141,7 +146,7 @@ public class DataSplit implements Split {
}
private boolean rawMergedRowCountAvailable() {
- return rawConvertible
+ return rawConvertible()
&& (dataDeletionFiles == null
|| dataDeletionFiles.stream()
.allMatch(f -> f == null || f.cardinality() !=
null));
@@ -168,6 +173,14 @@ public class DataSplit implements Split {
return false;
}
}
+
+ if (dataDeletionFiles != null) {
+ for (DeletionFile deletionFile : dataDeletionFiles) {
+ if (deletionFile != null && deletionFile.cardinality() ==
null) {
+ return false;
+ }
+ }
+ }
return true;
}
@@ -182,6 +195,13 @@ public class DataSplit implements Split {
}
sum += maxCount;
}
+ if (dataDeletionFiles != null) {
+ for (DeletionFile deletionFile : dataDeletionFiles) {
+ if (deletionFile != null) {
+ sum -= deletionFile.cardinality();
+ }
+ }
+ }
return sum;
}
@@ -245,7 +265,7 @@ public class DataSplit implements Split {
@Override
public Optional<List<RawFile>> convertToRawFiles() {
- if (rawConvertible) {
+ if (rawConvertible()) {
return Optional.of(
dataFiles.stream()
.map(f -> makeRawTableFile(bucketPath, f))
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
new file mode 100644
index 0000000000..4f6cb81cda
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java
@@ -0,0 +1,64 @@
+/*
+ * 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.paimon.utils;
+
+import org.apache.paimon.io.DataFileMeta;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.function.Function;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Util class for Deletion Vectors. */
+public class DataEvolutionUtils {
+
+ /**
+ * Retrieve the anchor file of a row range group. Always the oldest normal
file. Files are
+ * compared by (max_seq, fileName) pairs.
+ */
+ public static <T> T retrieveAnchorFile(
+ Collection<T> entries, Function<T, DataFileMeta> fileMetaFunc) {
+ T anchor = null;
+ DataFileMeta minMeta = null;
+
+ Comparator<DataFileMeta> fileComparator =
+ Comparator.comparingLong(DataFileMeta::maxSequenceNumber)
+ .thenComparing(DataFileMeta::fileName);
+
+ for (T entry : entries) {
+ DataFileMeta meta = fileMetaFunc.apply(entry);
+ if (isBlobFile(meta.fileName()) ||
isVectorStoreFile(meta.fileName())) {
+ continue;
+ }
+
+ if (minMeta == null || fileComparator.compare(meta, minMeta) < 0) {
+ minMeta = meta;
+ anchor = entry;
+ }
+ }
+
+ checkState(
+ anchor != null,
+ "Data-evolution deletion vectors should have a normal anchor
file in each row range group.");
+ return anchor;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
index 3c1ba1f796..6f900cf32c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
@@ -34,6 +34,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.stats.StatsTestUtils;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.EndOfScanException;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.DataField;
@@ -56,6 +57,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.LongFunction;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -554,6 +556,40 @@ public class DataEvolutionCompactCoordinatorTest {
.containsExactly(b1.file().fileName(), b2.file().fileName());
}
+ @Test
+ public void testPlanEndsScanWhenDeletionVectorsEnabled() {
+ Options options = new Options();
+ options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true);
+ options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
+ FileStoreTable table = mock(FileStoreTable.class);
+ when(table.coreOptions()).thenReturn(new CoreOptions(options));
+
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(table, false, false);
+
+
assertThatThrownBy(coordinator::plan).isInstanceOf(EndOfScanException.class);
+ }
+
+ @Test
+ public void testCompactTaskRejectsDeletionVectorEnabledTable() {
+ Options options = new Options();
+ options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
+ FileStoreTable table = mock(FileStoreTable.class);
+ when(table.coreOptions()).thenReturn(new CoreOptions(options));
+
+ DataEvolutionCompactTask task =
+ new DataEvolutionCompactTask(
+ BinaryRow.EMPTY_ROW,
+ Collections.singletonList(
+ createDataFileMeta("file1.parquet", 0L, 100L,
0, 1024)),
+ false);
+
+ assertThatThrownBy(() -> task.doCompact(table, "user"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Data evolution compaction does not support deletion
vectors.");
+ }
+
private ManifestEntry makeEntry(
String fileName, long firstRowId, long rowCount, long fileSize) {
return makeEntryWithSize(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
index a33f729963..1800f4c332 100644
---
a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/DeletionVectorTest.java
@@ -18,8 +18,17 @@
package org.apache.paimon.deletionvectors;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
+
import org.junit.jupiter.api.Test;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
import java.util.HashSet;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
@@ -29,6 +38,54 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link DeletionVector}. */
public class DeletionVectorTest {
+ @Test
+ public void testApplyDeletionFileRecordIteratorUsesProvidedJudger() throws
Exception {
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.checkedDelete(12);
+ deletionVector.checkedDelete(14);
+
+ ApplyDeletionFileRecordIterator iterator =
+ new ApplyDeletionFileRecordIterator(
+ new TestingFileRecordIterator(5),
+ position -> deletionVector.isDeleted(10 + position));
+
+ assertThat(iterator.deletionVector().isDeleted(2)).isTrue();
+ assertThat(iterator.deletionVector().isDeleted(4)).isTrue();
+ assertThat(iterator.deletionVector().isDeleted(12)).isFalse();
+
+ assertThat(iterator.next().getInt(0)).isEqualTo(0);
+ assertThat(iterator.next().getInt(0)).isEqualTo(1);
+ assertThat(iterator.next().getInt(0)).isEqualTo(3);
+ assertThat(iterator.next()).isNull();
+ }
+
+ @Test
+ public void testApplyDeletionVectorReaderUsesOffsetAwareJudger() throws
Exception {
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.checkedDelete(12);
+ deletionVector.checkedDelete(14);
+
+ ApplyDeletionVectorReader reader =
+ new ApplyDeletionVectorReader(
+ new TestingFileRecordReader(new
TestingFileRecordIterator(5)),
+ deletionVector,
+ 10);
+
+ assertThat(reader.deletionVector().isDeleted(2)).isTrue();
+ assertThat(reader.deletionVector().isDeleted(4)).isTrue();
+ assertThat(reader.deletionVector().isDeleted(12)).isFalse();
+
+ FileRecordIterator<InternalRow> batch = reader.readBatch();
+ assertThat(batch).isInstanceOf(ApplyDeletionFileRecordIterator.class);
+ ApplyDeletionFileRecordIterator iterator =
(ApplyDeletionFileRecordIterator) batch;
+
assertThat(iterator.deletionVector()).isSameAs(reader.deletionVector());
+
+ assertThat(iterator.next().getInt(0)).isEqualTo(0);
+ assertThat(iterator.next().getInt(0)).isEqualTo(1);
+ assertThat(iterator.next().getInt(0)).isEqualTo(3);
+ assertThat(iterator.next()).isNull();
+ }
+
@Test
public void testBitmapDeletionVector() {
HashSet<Integer> toDelete = new HashSet<>();
@@ -138,4 +195,61 @@ public class DeletionVectorTest {
assertThat(bitmap64DeletionVector.isDeleted(i)).isFalse();
}
}
+
+ private static class TestingFileRecordIterator implements
FileRecordIterator<InternalRow> {
+
+ private final int rows;
+ private int nextPosition;
+ private int returnedPosition = -1;
+
+ private TestingFileRecordIterator(int rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ public long returnedPosition() {
+ return returnedPosition;
+ }
+
+ @Override
+ public Path filePath() {
+ return new Path("/tmp/testing");
+ }
+
+ @Nullable
+ @Override
+ public InternalRow next() throws IOException {
+ if (nextPosition >= rows) {
+ return null;
+ }
+ returnedPosition = nextPosition++;
+ return GenericRow.of(returnedPosition);
+ }
+
+ @Override
+ public void releaseBatch() {}
+ }
+
+ private static class TestingFileRecordReader implements
FileRecordReader<InternalRow> {
+
+ private final FileRecordIterator<InternalRow> iterator;
+ private boolean returned;
+
+ private TestingFileRecordReader(FileRecordIterator<InternalRow>
iterator) {
+ this.iterator = iterator;
+ }
+
+ @Nullable
+ @Override
+ public FileRecordIterator<InternalRow> readBatch() {
+ if (returned) {
+ return null;
+ }
+ returned = true;
+ return iterator;
+ }
+
+ @Override
+ public void close() {}
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
new file mode 100644
index 0000000000..7fbf912a64
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/AllPlaceholdersRecordReaderTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.paimon.operation;
+
+import org.apache.paimon.data.BlobPlaceholder;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link AllPlaceholdersRecordReader}. */
+public class AllPlaceholdersRecordReaderTest {
+
+ private static final int BLOB_INDEX = 0;
+ private static final int ROW_ID_INDEX = 1;
+ private static final int SEQUENCE_NUMBER_INDEX = 2;
+ private static final long SEQUENCE_NUMBER = 100L;
+ private static final RowType READ_ROW_TYPE =
+ new RowType(
+ Arrays.asList(
+ new DataField(BLOB_INDEX, "blob_col",
DataTypes.BLOB()),
+ new DataField(
+ ROW_ID_INDEX, SpecialFields.ROW_ID.name(),
DataTypes.BIGINT()),
+ new DataField(
+ SEQUENCE_NUMBER_INDEX,
+ SpecialFields.SEQUENCE_NUMBER.name(),
+ DataTypes.BIGINT())));
+
+ @Test
+ public void testFullScan() throws Exception {
+ AllPlaceholdersRecordReader reader =
+ new AllPlaceholdersRecordReader(
+ 5L, 4L, null, READ_ROW_TYPE, BLOB_INDEX,
SEQUENCE_NUMBER);
+
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+ assertNextPlaceholder(iterator, 5L, 0L);
+ assertNextPlaceholder(iterator, 6L, 1L);
+ assertNextPlaceholder(iterator, 7L, 2L);
+ assertNextPlaceholder(iterator, 8L, 3L);
+ assertThat(iterator.next()).isNull();
+ iterator.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+
+ @Test
+ public void testRowRangePushed() throws Exception {
+ AllPlaceholdersRecordReader reader =
+ new AllPlaceholdersRecordReader(
+ 10L,
+ 10L,
+ Arrays.asList(new Range(8, 11), new Range(13, 14), new
Range(18, 22)),
+ READ_ROW_TYPE,
+ BLOB_INDEX,
+ SEQUENCE_NUMBER);
+
+ FileRecordIterator<InternalRow> iterator = reader.readBatch();
+ assertNextPlaceholder(iterator, 10L, 0L);
+ assertNextPlaceholder(iterator, 11L, 1L);
+ assertNextPlaceholder(iterator, 13L, 3L);
+ assertNextPlaceholder(iterator, 14L, 4L);
+ assertNextPlaceholder(iterator, 18L, 8L);
+ assertNextPlaceholder(iterator, 19L, 9L);
+ assertThat(iterator.next()).isNull();
+ iterator.releaseBatch();
+ assertThat(reader.readBatch()).isNull();
+ }
+
+ private static void assertNextPlaceholder(
+ FileRecordIterator<InternalRow> iterator, long rowId, long
returnedPosition)
+ throws Exception {
+ InternalRow row = iterator.next();
+ assertThat(row).isNotNull();
+ assertThat(row.getBlob(BLOB_INDEX)).isSameAs(BlobPlaceholder.INSTANCE);
+ assertThat(row.getLong(ROW_ID_INDEX)).isEqualTo(rowId);
+
assertThat(row.getLong(SEQUENCE_NUMBER_INDEX)).isEqualTo(SEQUENCE_NUMBER);
+ assertThat(iterator.returnedPosition()).isEqualTo(returnedPosition);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
index 596dbcd955..fc57cf9f2f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
@@ -23,9 +23,14 @@ import org.apache.paimon.data.BlobPlaceholder;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import
org.apache.paimon.operation.BlobFallbackRecordReader.BlobSequenceGroupRecordReader;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.reader.RecordReader.RecordIterator;
import org.apache.paimon.stats.SimpleStats;
@@ -58,6 +63,17 @@ public class BlobFallbackRecordReaderTest {
new DataField(1, SpecialFields.ROW_ID.name(),
DataTypes.BIGINT()),
new DataField(
2, SpecialFields.SEQUENCE_NUMBER.name(),
DataTypes.BIGINT())));
+ private static final RowType READ_ROW_TYPE_WITH_ROW_ID_ONLY =
+ new RowType(
+ Arrays.asList(
+ new DataField(BLOB_INDEX, BLOB_FIELD,
DataTypes.BLOB()),
+ new DataField(1, SpecialFields.ROW_ID.name(),
DataTypes.BIGINT())));
+ private static final RowType READ_ROW_TYPE_WITH_SEQUENCE_ONLY =
+ new RowType(
+ Arrays.asList(
+ new DataField(BLOB_INDEX, BLOB_FIELD,
DataTypes.BLOB()),
+ new DataField(
+ 1, SpecialFields.SEQUENCE_NUMBER.name(),
DataTypes.BIGINT())));
@Test
public void testBlobSequenceGroupReaderWithRowRanges() throws Exception {
@@ -144,6 +160,48 @@ public class BlobFallbackRecordReaderTest {
placeholderRows(newFile, 0, oldFile, 0));
assertThat(rows.rowIds).isEmpty();
+ assertThat(rows.nullBlobRowIds).containsExactly(0L);
+ assertThat(rows.nullBlobSequenceNumbers).containsExactly(-1L);
+ assertThat(rows.nullBlobRowCount).isEqualTo(1);
+ assertThat(rows.placeholderRowCount).isEqualTo(0);
+ }
+
+ @Test
+ public void
testBlobFallbackRecordReaderReturnsRowIdIfAllRowsArePlaceholders()
+ throws Exception {
+ DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2);
+ DataFileMeta oldFile = blobFile("old-placeholder-file", 0, 1, 1);
+
+ ReadResult rows =
+ readFallback(
+ Arrays.asList(newFile, oldFile),
+ null,
+ placeholderRows(newFile, 0, oldFile, 0),
+ READ_ROW_TYPE_WITH_ROW_ID_ONLY);
+
+ assertThat(rows.rowIds).isEmpty();
+ assertThat(rows.nullBlobRowIds).containsExactly(0L);
+ assertThat(rows.nullBlobSequenceNumbers).isEmpty();
+ assertThat(rows.nullBlobRowCount).isEqualTo(1);
+ assertThat(rows.placeholderRowCount).isEqualTo(0);
+ }
+
+ @Test
+ public void
testBlobFallbackRecordReaderReturnsSequenceIfAllRowsArePlaceholders()
+ throws Exception {
+ DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2);
+ DataFileMeta oldFile = blobFile("old-placeholder-file", 0, 1, 1);
+
+ ReadResult rows =
+ readFallback(
+ Arrays.asList(newFile, oldFile),
+ null,
+ placeholderRows(newFile, 0, oldFile, 0),
+ READ_ROW_TYPE_WITH_SEQUENCE_ONLY);
+
+ assertThat(rows.rowIds).isEmpty();
+ assertThat(rows.nullBlobRowIds).isEmpty();
+ assertThat(rows.nullBlobSequenceNumbers).containsExactly(-1L);
assertThat(rows.nullBlobRowCount).isEqualTo(1);
assertThat(rows.placeholderRowCount).isEqualTo(0);
}
@@ -168,16 +226,62 @@ public class BlobFallbackRecordReaderTest {
.containsExactly(2L, 2L, 2L, 2L, 2L, 1L, 1L, 1L, 1L, 1L, 1L,
1L, 1L, 2L, 2L);
}
+ @Test
+ public void
testBlobFallbackRecordReaderAppliesDeletionVectorToPlaceholderGaps()
+ throws Exception {
+ DataFileMeta oldFile = blobFile("old-file", 0, 6, 1);
+ DataFileMeta newFile = blobFile("new-file", 0, 3, 2);
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(4);
+
+ ReadResult rows =
+ ReadResult.read(
+ new BlobFallbackRecordReader(
+ Arrays.asList(newFile, oldFile),
+ file ->
+ new ApplyDeletionVectorReader(
+ oneRowPerBatchReader(
+ file,
+ fileRows(
+ file,
+ null,
+
placeholderRows(newFile, 1))),
+ deletionVector,
+ file.nonNullFirstRowId()),
+ (reader, range) ->
+ new ApplyDeletionVectorReader(
+ reader, deletionVector,
range.from),
+ null,
+ READ_ROW_TYPE,
+ BLOB_INDEX));
+
+ assertThat(rows.rowIds).containsExactly(0L, 1L, 2L, 3L, 5L);
+ assertThat(rows.sequenceNumbers).containsExactly(2L, 1L, 2L, 1L, 1L);
+ }
+
private static ReadResult readFallback(
List<DataFileMeta> files, List<Range> rowRanges, Set<String>
placeholderRows)
throws Exception {
+ return readFallback(files, rowRanges, placeholderRows, READ_ROW_TYPE);
+ }
+
+ private static ReadResult readFallback(
+ List<DataFileMeta> files,
+ List<Range> rowRanges,
+ Set<String> placeholderRows,
+ RowType readRowType)
+ throws Exception {
return ReadResult.read(
new BlobFallbackRecordReader(
files,
- file -> oneRowPerBatchReader(fileRows(file, rowRanges,
placeholderRows)),
+ file ->
+ oneRowPerBatchReader(
+ file, fileRows(file, rowRanges,
placeholderRows)),
+ (reader, range) -> reader,
rowRanges,
- READ_ROW_TYPE,
- BLOB_INDEX));
+ readRowType,
+ BLOB_INDEX),
+ readRowType);
}
private static ReadResult readSequenceGroup(
@@ -189,13 +293,16 @@ public class BlobFallbackRecordReaderTest {
throws Exception {
return ReadResult.read(
new BlobSequenceGroupRecordReader(
+ sequenceNumber,
files,
- file -> oneRowPerBatchReader(fileRows(file,
rowRanges)),
+ file -> oneRowPerBatchReader(file, fileRows(file,
rowRanges)),
+ (reader, range) -> reader,
rowRanges,
READ_ROW_TYPE,
BLOB_INDEX,
firstRowId,
- lastRowId));
+ lastRowId),
+ READ_ROW_TYPE);
}
private static DataFileMeta blobFile(
@@ -324,27 +431,40 @@ public class BlobFallbackRecordReaderTest {
return row;
}
- private static RecordReader<InternalRow>
oneRowPerBatchReader(List<InternalRow> rows) {
- return new RecordReader<InternalRow>() {
+ private static FileRecordReader<InternalRow> oneRowPerBatchReader(
+ DataFileMeta file, List<InternalRow> rows) {
+ return new FileRecordReader<InternalRow>() {
int index;
+ long returnedPosition = -1L;
@Override
- public RecordIterator<InternalRow> readBatch() {
+ public FileRecordIterator<InternalRow> readBatch() {
if (index >= rows.size()) {
return null;
}
InternalRow row = rows.get(index++);
- return new RecordIterator<InternalRow>() {
+ return new FileRecordIterator<InternalRow>() {
boolean returned;
+ @Override
+ public long returnedPosition() {
+ return returnedPosition;
+ }
+
+ @Override
+ public Path filePath() {
+ return new Path(file.fileName());
+ }
+
@Override
public InternalRow next() {
if (returned) {
return null;
}
returned = true;
+ returnedPosition = row.getLong(1) -
file.nonNullFirstRowId();
return row;
}
@@ -361,13 +481,20 @@ public class BlobFallbackRecordReaderTest {
private static class ReadResult {
final List<Long> rowIds = new ArrayList<>();
final List<Long> sequenceNumbers = new ArrayList<>();
+ final List<Long> nullBlobRowIds = new ArrayList<>();
+ final List<Long> nullBlobSequenceNumbers = new ArrayList<>();
final List<Integer> batchSizes = new ArrayList<>();
int placeholderRowCount;
int nullBlobRowCount;
static ReadResult read(RecordReader<InternalRow> reader) throws
Exception {
+ return read(reader, READ_ROW_TYPE);
+ }
+
+ static ReadResult read(RecordReader<InternalRow> reader, RowType
readRowType)
+ throws Exception {
try {
- ReadResult result = new ReadResult();
+ ReadResult result = new ReadResult(readRowType);
RecordIterator<InternalRow> batch;
while ((batch = reader.readBatch()) != null) {
int batchSize = 0;
@@ -385,14 +512,32 @@ public class BlobFallbackRecordReaderTest {
}
}
+ private final int rowIdIndex;
+ private final int seqNumIndex;
+
+ private ReadResult(RowType readRowType) {
+ this.rowIdIndex =
readRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ this.seqNumIndex =
readRowType.getFieldIndex(SpecialFields.SEQUENCE_NUMBER.name());
+ }
+
private void add(InternalRow row) {
if (row.isNullAt(BLOB_INDEX)) {
nullBlobRowCount++;
+ if (rowIdIndex >= 0) {
+ nullBlobRowIds.add(row.getLong(rowIdIndex));
+ }
+ if (seqNumIndex >= 0) {
+ nullBlobSequenceNumbers.add(row.getLong(seqNumIndex));
+ }
} else if (row.getBlob(BLOB_INDEX) == BlobPlaceholder.INSTANCE) {
placeholderRowCount++;
} else {
- rowIds.add(row.getLong(1));
- sequenceNumbers.add(row.getLong(2));
+ if (rowIdIndex >= 0) {
+ rowIds.add(row.getLong(rowIdIndex));
+ }
+ if (seqNumIndex >= 0) {
+ sequenceNumbers.add(row.getLong(seqNumIndex));
+ }
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
new file mode 100644
index 0000000000..3290f6b80d
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -0,0 +1,590 @@
+/*
+ * 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.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BlobData;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
+import org.apache.paimon.format.blob.BlobFileFormat;
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RangeHelper;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests filename-anchored deletion vectors for data evolution tables. */
+public class DataEvolutionDeletionVectorTest extends DataEvolutionTestBase {
+
+ private static final Range FULL_RANGE = new Range(0, 14);
+ private static final Range FIRST_RANGE = new Range(0, 4);
+ private static final List<DvSpec> DEFAULT_DV_SPECS =
+ Arrays.asList(
+ new DvSpec(new Range(0, 4), 1, 4),
+ new DvSpec(new Range(5, 9), 6),
+ new DvSpec(new Range(10, 14), 10, 12));
+
+ @Test
+ public void testReadAfterDeletionVectors() throws Exception {
+ // basic read
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ assertBaseFileLayout(table);
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ assertReadMatrix(getTableDefault(), "base");
+ }
+
+ @Test
+ public void testReadAfterUpdatingDeletionVectors() throws Exception {
+ // update DVs then read
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ assertBaseFileLayout(table);
+ commitDeletionVectors(
+ table,
+ Arrays.asList(new DvSpec(new Range(0, 4), 1), new DvSpec(new
Range(5, 9), 6)));
+ table = getTableDefault();
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ assertReadMatrix(getTableDefault(), "base");
+ }
+
+ @Test
+ public void testReadAfterAddingColumnAndDeletionVectors() throws Exception
{
+ // DVs with adding new columns.
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("f0", DataTypes.INT());
+ schemaBuilder.column("f1", DataTypes.STRING());
+ schemaBuilder.column("f3", DataTypes.BLOB());
+ schemaBuilder.option(CoreOptions.TARGET_FILE_SIZE.key(), "128 MB");
+ schemaBuilder.option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b");
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ catalog.createTable(identifier(), schemaBuilder.build(), true);
+
+ FileStoreTable table = getTableDefault();
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int rowId = batch * 5; rowId < batch * 5 + 5; rowId++) {
+ write.write(
+ GenericRow.of(
+ rowId,
+ BinaryString.fromString("name-" + rowId),
+ new BlobData(new byte[] {(byte) rowId})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+ assertBaseFileLayout(table);
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ catalog.alterTable(
+ identifier(),
+ SchemaChange.addColumn(
+ "f2", DataTypes.STRING(), null,
SchemaChange.Move.before("f2", "f3")),
+ false);
+ table = getTableDefault();
+ updateStructuredColumn(table);
+
+ assertReadMatrix(getTableDefault(), "updated");
+ }
+
+ @Test
+ public void testDataEvolutionDeletionFilesDoNotLeakAcrossSplits() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ assertBaseFileLayout(table);
+ updateStructuredColumn(table);
+ commitDeletionVectors(table, Collections.singletonList(new DvSpec(new
Range(0, 4), 1)));
+
+ table = getTableDefault();
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 B");
+ dynamicOptions.put(CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(), "1
B");
+ table = table.copy(dynamicOptions);
+
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ List<DataSplit> splits =
+ plan.splits().stream()
+ .map(DataEvolutionDeletionVectorTest::toDataSplit)
+ .sorted(Comparator.comparingLong(split ->
splitRowRange(split).from))
+ .collect(Collectors.toList());
+ assertThat(splits).hasSize(3);
+ assertDeletionFileRanges(splits.get(0), new Range(0, 4));
+ assertDeletionFileRanges(splits.get(1));
+ assertDeletionFileRanges(splits.get(2));
+ assertThat(splits.get(1).mergedRowCount()).hasValue(5L);
+ assertThat(splits.get(2).mergedRowCount()).hasValue(5L);
+
+ assertThat(readRows(readBuilder, plan))
+ .containsExactly(
+ "0|name-0|updated-0|0",
+ "2|name-2|updated-2|2",
+ "3|name-3|updated-3|3",
+ "4|name-4|updated-4|4",
+ "5|name-5|updated-5|5",
+ "6|name-6|updated-6|6",
+ "7|name-7|updated-7|7",
+ "8|name-8|updated-8|8",
+ "9|name-9|updated-9|9",
+ "10|name-10|updated-10|10",
+ "11|name-11|updated-11|11",
+ "12|name-12|updated-12|12",
+ "13|name-13|updated-13|13",
+ "14|name-14|updated-14|14");
+ }
+
+ @Test
+ public void testLimitPushDownWithHeavilyDeletedFirstRange() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ assertBaseFileLayout(table);
+ commitDeletionVectors(
+ table, Collections.singletonList(new DvSpec(new Range(0, 4),
1, 2, 3, 4)));
+
+ table = getTableDefault();
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 B");
+ dynamicOptions.put(CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(), "1
B");
+ table = table.copy(dynamicOptions);
+
+ ReadBuilder readBuilder = table.newReadBuilder().withLimit(2);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ List<DataSplit> splits =
+ plan.splits().stream()
+ .map(DataEvolutionDeletionVectorTest::toDataSplit)
+ .sorted(Comparator.comparingLong(split ->
splitRowRange(split).from))
+ .collect(Collectors.toList());
+
+ // Limit pushdown works at split level. If the first split's DV
cardinality is ignored,
+ // limit=2 would incorrectly keep only the first split even though it
has one visible row.
+ assertThat(splits).hasSize(2);
+ assertThat(splits.get(0).mergedRowCount()).hasValue(1L);
+ assertThat(splits.get(1).mergedRowCount()).hasValue(5L);
+ assertThat(readRows(table.newReadBuilder(), plan))
+ .containsExactly(
+ "0|name-0|base-0|0",
+ "5|name-5|base-5|5",
+ "6|name-6|base-6|6",
+ "7|name-7|base-7|7",
+ "8|name-8|base-8|8",
+ "9|name-9|base-9|9");
+ }
+
+ @Override
+ protected Schema schemaDefault() {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("f0", DataTypes.INT());
+ schemaBuilder.column("f1", DataTypes.STRING());
+ schemaBuilder.column("f2", DataTypes.STRING());
+ schemaBuilder.column("f3", DataTypes.BLOB());
+ schemaBuilder.option(CoreOptions.TARGET_FILE_SIZE.key(), "128 MB");
+ schemaBuilder.option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b");
+ schemaBuilder.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ schemaBuilder.option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ return schemaBuilder.build();
+ }
+
+ private void writeBaseRows(FileStoreTable table) throws Exception {
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int rowId = batch * 5; rowId < batch * 5 + 5; rowId++) {
+ write.write(
+ GenericRow.of(
+ rowId,
+ BinaryString.fromString("name-" + rowId),
+ BinaryString.fromString("base-" + rowId),
+ new BlobData(new byte[] {(byte) rowId})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+ }
+
+ private void updateStructuredColumn(FileStoreTable table) throws Exception
{
+ RowType writeType =
table.rowType().project(Collections.singletonList("f2"));
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ long firstRowId = batch * 5L;
+ for (int rowId = batch * 5; rowId < batch * 5 + 5; rowId++) {
+
write.write(GenericRow.of(BinaryString.fromString("updated-" + rowId)));
+ }
+ List<CommitMessage> commitables = write.prepareCommit();
+ setFirstRowId(commitables, firstRowId);
+ commit.commit(commitables);
+ }
+ }
+ }
+
+ private void commitDeletionVectors(FileStoreTable table, List<DvSpec>
deletionVectorSpecs)
+ throws Exception {
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ table.store().newIndexFileHandler(),
+ table.latestSnapshot().get(),
+ BinaryRow.EMPTY_ROW);
+ Map<Range, String> anchorFiles = anchorFilesByRange(table);
+
+ for (DvSpec spec : deletionVectorSpecs) {
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ for (long rowId : spec.deletedRowIds) {
+ deletionVector.delete(rowId - spec.range.from);
+ }
+ maintainer.notifyNewDeletionVector(anchorFiles.get(spec.range),
deletionVector);
+ }
+
+ List<IndexFileMeta> newIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ if (entry.kind() == FileKind.ADD) {
+ newIndexFiles.add(entry.indexFile());
+ } else if (entry.kind() == FileKind.DELETE) {
+ deletedIndexFiles.add(entry.indexFile());
+ }
+ }
+
+ commitDefault(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ UNAWARE_BUCKET,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ newIndexFiles,
+ deletedIndexFiles),
+ CompactIncrement.emptyIncrement())));
+ }
+
+ private Map<Range, String> anchorFilesByRange(FileStoreTable table) {
+ List<DataFileMeta> dataFiles =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ Map<Range, String> result = new HashMap<>();
+ for (List<DataFileMeta> group :
rangeHelper.mergeOverlappingRanges(dataFiles)) {
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
+ result.put(anchor.nonNullRowIdRange(), anchor.fileName());
+ }
+ return result;
+ }
+
+ private static void assertReadMatrix(FileStoreTable table, String
structuredValuePrefix)
+ throws Exception {
+ List<String> expectedRows = expectedRows(structuredValuePrefix,
FULL_RANGE);
+ List<String> expectedFirstRangeRows =
expectedRows(structuredValuePrefix, FIRST_RANGE);
+ List<String> expectedProjectedStrings =
+ expectedProjectedStrings(structuredValuePrefix, FULL_RANGE);
+ List<Integer> expectedBlobValues = expectedBlobValues(FULL_RANGE);
+
+
assertThat(readRows(table.newReadBuilder())).containsExactlyElementsOf(expectedRows);
+ assertThat(
+ readRows(
+ table.newReadBuilder()
+
.withRowRanges(Collections.singletonList(FULL_RANGE))))
+ .containsExactlyElementsOf(expectedRows);
+ assertThat(
+ readRows(
+ table.newReadBuilder()
+
.withRowRanges(Collections.singletonList(FIRST_RANGE))))
+ .containsExactlyElementsOf(expectedFirstRangeRows);
+ assertThat(
+ readRows(
+ table.newReadBuilder()
+
.withRowRanges(Collections.singletonList(new Range(4, 4)))))
+ .isEmpty();
+ assertThat(
+ readRows(
+ table.newReadBuilder()
+
.withRowRanges(Collections.singletonList(new Range(7, 7)))))
+ .containsExactly(expectedRow(structuredValuePrefix, 7));
+
+
assertThat(readProjectedStrings(table.newReadBuilder().withProjection(new int[]
{2})))
+ .containsExactlyElementsOf(expectedProjectedStrings);
+ assertThat(
+ readProjectedStrings(
+ table.newReadBuilder()
+ .withProjection(new int[] {2})
+
.withRowRanges(Collections.singletonList(FULL_RANGE))))
+ .containsExactlyElementsOf(expectedProjectedStrings);
+
assertThat(readProjectedBlobValues(table.newReadBuilder().withProjection(new
int[] {3})))
+ .containsExactlyElementsOf(expectedBlobValues);
+ assertThat(
+ readProjectedBlobValues(
+ table.newReadBuilder()
+ .withProjection(new int[] {3})
+
.withRowRanges(Collections.singletonList(FULL_RANGE))))
+ .containsExactlyElementsOf(expectedBlobValues);
+
+ DataSplit fullRangeSplit = planDataSplit(table, FULL_RANGE);
+ assertDeletionFileRanges(
+ fullRangeSplit, new Range(0, 4), new Range(5, 9), new
Range(10, 14));
+ assertThat(fullRangeSplit.mergedRowCount()).hasValue(10L);
+ assertThat(planDataSplit(table,
FIRST_RANGE).mergedRowCount()).hasValue(3L);
+ }
+
+ private static void assertDeletionFileRanges(DataSplit split, Range...
expectedRanges) {
+ List<DeletionFile> deletionFiles =
split.deletionFiles().orElse(Collections.emptyList());
+ assertThat(deletionFiles).hasSize(split.dataFiles().size());
+
+ Map<Range, DeletionFile> actual = new HashMap<>();
+ RangeHelper<DataFileMeta> rangeHelper = new
RangeHelper<>(DataFileMeta::nonNullRowIdRange);
+ for (List<DataFileMeta> group :
rangeHelper.mergeOverlappingRanges(split.dataFiles())) {
+ DataFileMeta anchor = retrieveAnchorFile(group, file -> file);
+ DeletionFile deletionFile =
deletionFiles.get(split.dataFiles().indexOf(anchor));
+ if (deletionFile != null) {
+ actual.put(anchor.nonNullRowIdRange(), deletionFile);
+ }
+ }
+
+ assertThat(deletionFiles.stream().filter(file -> file != null).count())
+ .isEqualTo((long) expectedRanges.length);
+ assertThat(actual.keySet()).containsExactlyInAnyOrder(expectedRanges);
+ }
+
+ private static List<String> expectedRows(String structuredValuePrefix,
Range range) {
+ List<String> rows = new ArrayList<>();
+ for (int rowId = (int) range.from; rowId <= range.to; rowId++) {
+ if (!isDeletedByDefaultDv(rowId)) {
+ rows.add(expectedRow(structuredValuePrefix, rowId));
+ }
+ }
+ return rows;
+ }
+
+ private static List<String> expectedProjectedStrings(
+ String structuredValuePrefix, Range range) {
+ List<String> rows = new ArrayList<>();
+ for (int rowId = (int) range.from; rowId <= range.to; rowId++) {
+ if (!isDeletedByDefaultDv(rowId)) {
+ rows.add(structuredValuePrefix + "-" + rowId);
+ }
+ }
+ return rows;
+ }
+
+ private static List<Integer> expectedBlobValues(Range range) {
+ List<Integer> rows = new ArrayList<>();
+ for (int rowId = (int) range.from; rowId <= range.to; rowId++) {
+ if (!isDeletedByDefaultDv(rowId)) {
+ rows.add(rowId);
+ }
+ }
+ return rows;
+ }
+
+ private static boolean isDeletedByDefaultDv(int rowId) {
+ for (DvSpec spec : DEFAULT_DV_SPECS) {
+ for (long deletedRowId : spec.deletedRowIds) {
+ if (deletedRowId == rowId) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private static String expectedRow(String structuredValuePrefix, int rowId)
{
+ return rowId + "|name-" + rowId + "|" + structuredValuePrefix + "-" +
rowId + "|" + rowId;
+ }
+
+ private static void assertBaseFileLayout(FileStoreTable table) {
+ assertRegularFileRowRanges(
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList()),
+ Arrays.asList(new Range(0, 4), new Range(5, 9), new Range(10,
14)));
+ assertFirstBlobFileRowRanges(
+ table, Arrays.asList(new Range(0, 0), new Range(1, 1), new
Range(2, 2)), 15);
+ }
+
+ private static DataSplit planDataSplit(FileStoreTable table, Range range) {
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withRowRanges(Collections.singletonList(range));
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ assertThat(plan.splits()).hasSize(1);
+ return toDataSplit(plan.splits().get(0));
+ }
+
+ private static List<String> readRows(ReadBuilder readBuilder,
TableScan.Plan plan)
+ throws IOException {
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> rows.add(formatRow(row)));
+ }
+
rows.sort(Comparator.comparingInt(DataEvolutionDeletionVectorTest::rowId));
+ return rows;
+ }
+
+ private static List<String> readRows(ReadBuilder readBuilder) throws
IOException {
+ return readRows(readBuilder, readBuilder.newScan().plan());
+ }
+
+ private static List<String> readProjectedStrings(ReadBuilder readBuilder)
throws IOException {
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row ->
rows.add(row.getString(0).toString()));
+ }
+
rows.sort(Comparator.comparingInt(DataEvolutionDeletionVectorTest::projectedRowId));
+ return rows;
+ }
+
+ private static List<Integer> readProjectedBlobValues(ReadBuilder
readBuilder)
+ throws IOException {
+ List<Integer> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row -> rows.add(row.getBlob(0).toData()[0]
& 0xFF));
+ }
+ Collections.sort(rows);
+ return rows;
+ }
+
+ private static int rowId(String row) {
+ return Integer.parseInt(row.substring(0, row.indexOf('|')));
+ }
+
+ private static int projectedRowId(String row) {
+ return Integer.parseInt(row.substring(row.lastIndexOf('-') + 1));
+ }
+
+ private static String formatRow(InternalRow row) {
+ return row.getInt(0)
+ + "|"
+ + row.getString(1)
+ + "|"
+ + row.getString(2)
+ + "|"
+ + (row.getBlob(3).toData()[0] & 0xFF);
+ }
+
+ private static DataSplit toDataSplit(Split split) {
+ if (split instanceof IndexedSplit) {
+ return ((IndexedSplit) split).dataSplit();
+ }
+ return (DataSplit) split;
+ }
+
+ private static void assertRegularFileRowRanges(
+ List<DataFileMeta> dataFiles, List<Range> expected) {
+ List<Range> actual =
+ dataFiles.stream()
+ .filter(DataEvolutionDeletionVectorTest::isNormalFile)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .sorted(Comparator.comparingLong(range -> range.from))
+ .collect(Collectors.toList());
+ assertThat(actual).isEqualTo(expected);
+ }
+
+ private static void assertFirstBlobFileRowRanges(
+ FileStoreTable table, List<Range> expectedFirstRanges, int
expectedCount) {
+ List<Range> actual =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file ->
BlobFileFormat.isBlobFile(file.fileName()))
+ .map(DataFileMeta::nonNullRowIdRange)
+ .sorted(Comparator.comparingLong(range -> range.from))
+ .collect(Collectors.toList());
+ assertThat(actual).hasSize(expectedCount);
+ assertThat(actual.subList(0,
expectedFirstRanges.size())).isEqualTo(expectedFirstRanges);
+ }
+
+ private static Range splitRowRange(DataSplit split) {
+ return split.dataFiles().stream()
+ .filter(DataEvolutionDeletionVectorTest::isNormalFile)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .min(Comparator.comparingLong(range -> range.from))
+ .get();
+ }
+
+ private static boolean isNormalFile(DataFileMeta file) {
+ return !BlobFileFormat.isBlobFile(file.fileName()) &&
!isVectorStoreFile(file.fileName());
+ }
+
+ private static class DvSpec {
+
+ private final Range range;
+ private final long[] deletedRowIds;
+
+ private DvSpec(Range range, long... deletedRowIds) {
+ this.range = range;
+ this.deletedRowIds = deletedRowIds;
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
index dcd888b7f1..a0c76537ab 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/DataSplitCompatibleTest.java
@@ -90,6 +90,31 @@ public class DataSplitCompatibleTest {
assertThat(split.mergedRowCount()).hasValue(5700L);
}
+ @Test
+ public void testDeletionFilesSerialize() throws Exception {
+ List<DataFileMeta> dataFiles =
+ Collections.singletonList(
+ newDataFile(10, SimpleStats.EMPTY_STATS,
null).assignFirstRowId(0));
+ List<DeletionFile> deletionFiles =
+ Collections.singletonList(new DeletionFile("p", 1, 2, 3L));
+ DataSplit split =
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(1)
+ .withBucketPath("my path")
+ .rawConvertible(true)
+ .withDataFiles(dataFiles)
+ .withDataDeletionFiles(deletionFiles)
+ .build();
+
+ DataSplit actual = InstantiationUtil.clone(split);
+
+ assertThat(actual.rawConvertible()).isTrue();
+ assertThat(actual.deletionFiles()).hasValue(deletionFiles);
+ assertThat(actual.mergedRowCount()).hasValue(7L);
+ }
+
@Test
public void testSplitMinMaxValue() {
Map<Long, List<DataField>> schemas = new HashMap<>();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
new file mode 100644
index 0000000000..6f51235362
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/DataEvolutionUtilsTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.paimon.utils;
+
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.function.Function;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link DataEvolutionUtils}. */
+public class DataEvolutionUtilsTest {
+
+ @Test
+ public void testRetrieveAnchorFileSkipsSpecialFiles() {
+ DataFileMeta blobFile = dataFile("blob-file.blob", 1);
+ DataFileMeta vectorFile = dataFile("data.vector.lance", 2);
+ DataFileMeta oldestNormalFile = dataFile("oldest-normal.parquet", 3);
+ DataFileMeta newestNormalFile = dataFile("newest-normal.parquet", 4);
+
+ assertThat(
+ DataEvolutionUtils.retrieveAnchorFile(
+ Arrays.asList(
+ blobFile, newestNormalFile,
vectorFile, oldestNormalFile),
+ Function.identity()))
+ .isSameAs(oldestNormalFile);
+ }
+
+ @Test
+ public void testRetrieveAnchorFileFailsWithoutNormalFile() {
+ assertThatThrownBy(
+ () ->
+ DataEvolutionUtils.retrieveAnchorFile(
+ Arrays.asList(
+ dataFile("blob-file.blob", 1),
+ dataFile("data.vector.lance",
2)),
+ Function.identity()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("normal anchor file");
+ }
+
+ @Test
+ public void testRetrieveAnchorFileTieBreaksWithFileName() {
+ DataFileMeta largerFileName = dataFile("normal-2.parquet", 1);
+ DataFileMeta smallerFileName = dataFile("normal-1.parquet", 1);
+
+ assertThat(
+ DataEvolutionUtils.retrieveAnchorFile(
+ Arrays.asList(largerFileName, smallerFileName),
+ Function.identity()))
+ .isSameAs(smallerFileName);
+ }
+
+ private static DataFileMeta dataFile(String fileName, long
maxSequenceNumber) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 1L,
+ 1L,
+ SimpleStats.EMPTY_STATS,
+ maxSequenceNumber,
+ maxSequenceNumber,
+ 1L,
+ Collections.emptyList(),
+ null,
+ null,
+ null,
+ null,
+ 0L,
+ Collections.emptyList());
+ }
+}
