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 dbaa953a92 [core] Keep anchor normal file when DE read-type pruning
leaves only dedicated files (#9966)
dbaa953a92 is described below
commit dbaa953a92b1940a97953cb292308263c1fef6c8
Author: YangJie <[email protected]>
AuthorDate: Tue Oct 13 01:44:48 2026 -0400
[core] Keep anchor normal file when DE read-type pruning leaves only
dedicated files (#9966)
---
.../operation/DataEvolutionFileStoreScan.java | 18 +-
.../paimon/operation/DataEvolutionSplitRead.java | 3 +-
.../operation/DataEvolutionVectorReadPlanner.java | 100 ++++++-
.../org/apache/paimon/append/BlobUpdateTest.java | 98 ++++++-
.../paimon/append/MultipleBlobTableTest.java | 136 ++++++++++
.../apache/paimon/append/VectorStoreTableTest.java | 220 +++++++++++++++-
.../operation/DataEvolutionFileStoreScanTest.java | 262 +++++++++++++++++++
.../DataEvolutionVectorReadPlannerTest.java | 290 +++++++++++++++++++++
8 files changed, 1117 insertions(+), 10 deletions(-)
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 c8ad122f5d..442ad4c70d 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
@@ -225,7 +225,9 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
* 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.
+ * anchor file to lookup corresponding Deletion Files. Without deletion
vectors, the anchor is
+ * still kept when the kept files are all blob/vector-store files, which
may cover only part of
+ * the group's row-id range.
*/
private List<ManifestEntry> pruneByReadType(List<ManifestEntry> group) {
if (readType == null || group.size() <= 1) {
@@ -281,6 +283,16 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
if (anchor != null && !kept.contains(anchor)) {
kept.add(anchor);
}
+ // Blob and vector-store files may each cover only a sub-range of
their group, and the
+ // reader would derive the group's range from them and drop the other
rows, so keep the
+ // anchor. An incremental scan can see a group with no normal file at
all, which has no
+ // anchor to keep.
+ if (anchor == null
+ && !kept.isEmpty()
+ && kept.stream().allMatch(e -> isDedicatedFile(e.file()))
+ && !group.stream().allMatch(e -> isDedicatedFile(e.file()))) {
+ kept.add(retrieveAnchorFile(group, ManifestEntry::file));
+ }
// Group must contribute at least one file so the reader sees rowCount
and can NULL-fill
// missing columns for the projection's rows. The representative must
be a full-range
// normal file: a blob or vector-store file covers only a sub-range of
the group's row
@@ -290,6 +302,10 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
: kept;
}
+ private static boolean isDedicatedFile(DataFileMeta file) {
+ return isBlobFile(file.fileName()) ||
isVectorStoreFile(file.fileName());
+ }
+
private Set<Integer> fileFieldIdsForEntry(ManifestEntry entry) {
return fileFieldIdsCache.computeIfAbsent(
Pair.of(entry.file().schemaId(), entry.file().writeCols()),
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 ae2fc03d54..b4d58af88b 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
@@ -346,7 +346,8 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
schemaFetcher
.apply(file.schemaId())
.dataFileSchema(file.writeCols())
- .logicalRowType());
+ .logicalRowType(),
+ rowRanges);
if (vectorRanges != null) {
List<FieldBunch> nonVectorBunches =
splitFieldBunches(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java
index 1b77489cbf..599b29caa6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java
@@ -45,12 +45,16 @@ import static
org.apache.paimon.types.VectorType.isVectorStoreFile;
/** Plans ranges with fixed vector field providers before positional column
merging. */
class DataEvolutionVectorReadPlanner {
- /** Returns null when the existing sequential column-group readers
suffice. */
+ /**
+ * Returns null when the existing sequential column-group readers suffice.
{@code rowRanges} are
+ * the pushed-down row ranges, or null when every row is read.
+ */
@Nullable
static List<ReadRange> plan(
List<DataFileMeta> files,
RowType readType,
- Function<DataFileMeta, RowType> fileToRowType) {
+ Function<DataFileMeta, RowType> fileToRowType,
+ @Nullable List<Range> rowRanges) {
Set<Integer> readIds =
readType.getFields().stream().map(DataField::id).collect(Collectors.toSet());
boolean vectorOnly = files.stream().allMatch(file ->
isVectorStoreFile(file.fileName()));
@@ -93,14 +97,33 @@ class DataEvolutionVectorReadPlanner {
}
}
- // Keep the sequential path for disjoint column groups, including
rolled vector files.
- if (!overlappingGroups) {
- return null;
- }
if (logicalRange == null) {
logicalRange = new Range(firstRowId, lastRowId);
}
+ // The sequential path reads each vector column as one
VectorFileBunch, which must yield
+ // exactly the group's rows. Otherwise plan explicit ranges, which
NULL-fill uncovered rows
+ // and take every row from its latest file. Rolled files that tile the
group stay
+ // sequential.
+ boolean partialVector = false;
+ for (int fieldId : vectorReadIds) {
+ List<DataFileMeta> fieldFiles = new ArrayList<>();
+ for (Candidate candidate : candidates) {
+ if (candidate.fieldIds.contains(fieldId)) {
+ fieldFiles.add(candidate.file);
+ }
+ }
+ if (!readsSequentially(fieldFiles, logicalRange, rowRanges)) {
+ partialVector = true;
+ break;
+ }
+ }
+
+ // Keep the sequential path for disjoint column groups that each span
the whole range.
+ if (!overlappingGroups && !partialVector) {
+ return null;
+ }
+
// Resolve the original files before VectorFileBunch can discard older
overlapping files.
// A bunch may contain different sequences in adjacent ranges, so
sorting whole bunches
// by their maximum sequence cannot establish the latest provider for
every row.
@@ -159,6 +182,71 @@ class DataEvolutionVectorReadPlanner {
return result;
}
+ /**
+ * Whether a VectorFileBunch built from {@code files} yields exactly the
rows of {@code target},
+ * each from its latest file. The bunch takes files in first-row-id order,
newest first, and
+ * skips a file older than the one kept just before it, without comparing
it to later kept
+ * files. With row ranges pushed down, the bunch accepts gaps and every
reader reads only the
+ * selected rows, so a gap matters only if it contains a selected row.
+ */
+ private static boolean readsSequentially(
+ List<DataFileMeta> files, Range target, @Nullable List<Range>
rowRanges) {
+ files.sort(
+ Comparator.comparingLong(DataFileMeta::nonNullFirstRowId)
+ .thenComparing(
+
Comparator.comparingLong(DataFileMeta::maxSequenceNumber)
+ .reversed()));
+ List<DataFileMeta> kept = new ArrayList<>();
+ List<DataFileMeta> skipped = new ArrayList<>();
+ long cursor = target.from;
+ long latestSequence = Long.MIN_VALUE;
+ for (DataFileMeta file : files) {
+ Range range = file.nonNullRowIdRange();
+ if (range.from < cursor) {
+ if (file.maxSequenceNumber() < latestSequence) {
+ skipped.add(file);
+ continue;
+ }
+ return false;
+ }
+ if (range.from > cursor && selects(cursor, range.from - 1,
rowRanges)) {
+ return false;
+ }
+ kept.add(file);
+ cursor = range.to + 1;
+ latestSequence = file.maxSequenceNumber();
+ }
+ if (cursor > target.to + 1 || selects(cursor, target.to, rowRanges)) {
+ return false;
+ }
+ for (DataFileMeta older : skipped) {
+ Range olderRange = older.nonNullRowIdRange();
+ for (DataFileMeta file : kept) {
+ Range range = file.nonNullRowIdRange();
+ if (Range.intersect(olderRange.from, olderRange.to,
range.from, range.to)
+ && file.maxSequenceNumber() <=
older.maxSequenceNumber()) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean selects(long from, long to, @Nullable List<Range>
rowRanges) {
+ if (from > to) {
+ return false;
+ }
+ if (rowRanges == null) {
+ return true;
+ }
+ for (Range range : rowRanges) {
+ if (Range.intersect(from, to, range.from, range.to)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
static class ReadRange {
Range range;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java
index 9d1f05aa65..04d38c5003 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java
@@ -45,6 +45,10 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
+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.ReadBuilder;
@@ -171,7 +175,13 @@ public class BlobUpdateTest extends TableTestBase {
TableScan.Plan plan = readBuilder.newScan().plan();
assertThat(plan.splits().size()).isEqualTo(1);
DataSplit dataSplit = ((IndexedSplit)
plan.splits().get(0)).dataSplit();
- assertThat(dataSplit.dataFiles().size()).isEqualTo(3);
+ // The three blob files that intersect the pushed-down rows, plus the
normal anchor.
+ assertThat(dataSplit.dataFiles().size()).isEqualTo(4);
+ assertThat(
+ dataSplit.dataFiles().stream()
+ .filter(file ->
!BlobFileFormat.isBlobFile(file.fileName()))
+ .count())
+ .isEqualTo(1L);
RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan);
List<byte[]> actual = new ArrayList<>();
@@ -182,6 +192,92 @@ public class BlobUpdateTest extends TableTestBase {
assertThat(actual.get(1)).isEqualTo(updated9);
}
+ @Test
+ public void testProjectBlobColumnBackfilledForSubRangeKeepsAllRows()
throws Exception {
+ // A data-evolution table whose blob column is backfilled for only a
sub-range of a group:
+ // the normal file writes f0/f1 over row ids [0, 9], while the blob
file writes f2 over
+ // only [0, 4]. Projecting f2 alone (no filter, no row-range pushdown)
keeps just the blob
+ // file, which spans [0, 4]; without the anchor the reader derives the
group range from that
+ // sub-range and drops rows [5, 9]. See #9965.
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ RowType normalType = table.rowType().project(Arrays.asList("f0",
"f1"));
+ RowType blobType =
table.rowType().project(Collections.singletonList("f2"));
+
+ // normal file writes f0/f1 over row ids [0, 9]
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(normalType)) {
+ for (int i = 0; i < 10; i++) {
+ write.write(GenericRow.of(i, BinaryString.fromString("row-" +
i)));
+ }
+ BatchTableCommit commit = builder.newCommit();
+ commit.commit(write.prepareCommit());
+ }
+
+ // blob file writes f2 for only the first five rows [0, 4]
+ long firstRowId =
getTableDefault().snapshotManager().latestSnapshot().nextRowId() - 10;
+ List<byte[]> blobs = new ArrayList<>();
+ builder = getTableDefault().newBatchWriteBuilder();
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(blobType)) {
+ for (int i = 0; i < 5; i++) {
+ byte[] bytes = ("blob-" + i).getBytes();
+ blobs.add(bytes);
+ write.write(GenericRow.of(new BlobData(bytes)));
+ }
+ BatchTableCommit commit = builder.newCommit();
+ List<CommitMessage> commitables = write.prepareCommit();
+ reassignFirstRowId(commitables, firstRowId);
+ commit.commit(commitables);
+ }
+
+ FileStoreTable readTable = getTableDefault();
+ assertBlobFileRowIdRanges(readTable, Collections.singletonList(new
Range(0L, 4L)));
+
+ ReadBuilder readBuilder =
+ readTable
+ .newReadBuilder()
+
.withReadType(readTable.rowType().project(Collections.singletonList("f2")));
+ RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan());
+
+ List<byte[]> actual = new ArrayList<>();
+ reader.forEachRemaining(
+ row -> actual.add(row.isNullAt(0) ? null :
row.getBlob(0).toData()));
+
+ assertThat(actual.size()).isEqualTo(10);
+ for (int i = 0; i < 5; i++) {
+ assertThat(actual.get(i)).isEqualTo(blobs.get(i));
+ }
+ for (int i = 5; i < 10; i++) {
+ assertThat(actual.get(i)).isNull();
+ }
+
+ // Row-range pushdown across the end of the blob file: rows [5, 6] are
still NULL.
+ readBuilder.withRowRanges(Collections.singletonList(new Range(3L,
6L)));
+ List<byte[]> ranged = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(
+ row -> ranged.add(row.isNullAt(0) ? null :
row.getBlob(0).toData()));
+ assertThat(ranged.size()).isEqualTo(4);
+ assertThat(ranged.get(0)).isEqualTo(blobs.get(3));
+ assertThat(ranged.get(1)).isEqualTo(blobs.get(4));
+ assertThat(ranged.get(2)).isNull();
+ assertThat(ranged.get(3)).isNull();
+ }
+
+ private void reassignFirstRowId(List<CommitMessage> commitables, long
firstRowId) {
+ for (CommitMessage message : commitables) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ List<DataFileMeta> newFiles = new
ArrayList<>(impl.newFilesIncrement().newFiles());
+ impl.newFilesIncrement().newFiles().clear();
+ for (DataFileMeta file : newFiles) {
+
impl.newFilesIncrement().newFiles().add(file.assignFirstRowId(firstRowId));
+ }
+ }
+ }
+
/**
* This test manually simulates the compacted layout described in
BlobSequenceGroupRecordReader,
* scaled down to row id [0, 9].
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/MultipleBlobTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/MultipleBlobTableTest.java
index a8fd2cc328..5c1617f83e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/MultipleBlobTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/MultipleBlobTableTest.java
@@ -32,7 +32,15 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
+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.ReadBuilder;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -198,6 +206,134 @@ public class MultipleBlobTableTest extends TableTestBase {
new BlobData(blobBytes2));
}
+ @Test
+ public void testStreamScanReadsBlobOnlyDelta() throws Exception {
+ createTableDefault();
+ RowType rowType = getTableDefault().rowType();
+
+ // normal file writes f0/f1 over row ids [0, 1]
+ BatchWriteBuilder builder = getTableDefault().newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+
.withWriteType(rowType.project(Arrays.asList("f0", "f1")));
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(0, BinaryString.fromString("a")));
+ write.write(GenericRow.of(1, BinaryString.fromString("b")));
+ commit.commit(write.prepareCommit());
+ }
+
+ FileStoreTable table = getTableDefault();
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withReadType(rowType.project(Arrays.asList("f2", "f3")));
+ StreamTableScan streamScan = readBuilder.newStreamScan();
+ assertThat(readBlobs(readBuilder,
streamScan.plan()).size()).isEqualTo(2);
+
+ // One commit updates both blob columns over [0, 1], so its delta
holds only two .blob
+ // files; their normal file is in the previous snapshot.
+ builder = getTableDefault().newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+
.withWriteType(rowType.project(Arrays.asList("f2", "f3")));
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(new BlobData(blobBytes1), new
BlobData(blobBytes2)));
+ write.write(GenericRow.of(new BlobData(blobBytes2), new
BlobData(blobBytes1)));
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, 0L);
+ commit.commit(messages);
+ }
+
+ TableScan.Plan deltaPlan = streamScan.plan();
+ assertThat(
+ deltaPlan.splits().stream()
+ .flatMap(split -> ((DataSplit)
split).dataFiles().stream())
+ .allMatch(file -> isBlobFile(file.fileName())))
+ .isTrue();
+ List<byte[][]> delta = readBlobs(readBuilder, deltaPlan);
+ assertThat(delta.size()).isEqualTo(2);
+ assertThat(delta.get(0)[0]).isEqualTo(blobBytes1);
+ assertThat(delta.get(0)[1]).isEqualTo(blobBytes2);
+ assertThat(delta.get(1)[0]).isEqualTo(blobBytes2);
+ assertThat(delta.get(1)[1]).isEqualTo(blobBytes1);
+ }
+
+ @Test
+ public void testProjectTwoBlobColumnsCoveringDifferentSubRanges() throws
Exception {
+ // normal [0, 9], f2 over [0, 6] and f3 over [3, 9]: together the blob
files cover the
+ // group, but each blob column covers only part of it.
+ createTableDefault();
+ RowType rowType = getTableDefault().rowType();
+ BatchWriteBuilder builder = getTableDefault().newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+
.withWriteType(rowType.project(Arrays.asList("f0", "f1")));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < 10; i++) {
+ write.write(GenericRow.of(i, BinaryString.fromString("row-" +
i)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ writeBlobColumn("f2", blobBytes1, 7, 0L);
+ writeBlobColumn("f3", blobBytes2, 7, 3L);
+
+ ReadBuilder readBuilder =
+ getTableDefault()
+ .newReadBuilder()
+ .withReadType(rowType.project(Arrays.asList("f2",
"f3")));
+ List<byte[][]> rows = readBlobs(readBuilder,
readBuilder.newScan().plan());
+ assertThat(rows.size()).isEqualTo(10);
+ for (int i = 0; i < 10; i++) {
+ assertThat(rows.get(i)[0]).isEqualTo(i <= 6 ? blobBytes1 : null);
+ assertThat(rows.get(i)[1]).isEqualTo(i >= 3 ? blobBytes2 : null);
+ }
+ }
+
+ private void writeBlobColumn(String column, byte[] bytes, int count, long
firstRowId)
+ throws Exception {
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+ table.rowType()
+
.project(Collections.singletonList(column)));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < count; i++) {
+ write.write(GenericRow.of(new BlobData(bytes)));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, firstRowId);
+ commit.commit(messages);
+ }
+ }
+
+ private static List<byte[][]> readBlobs(ReadBuilder readBuilder,
TableScan.Plan plan)
+ throws Exception {
+ List<byte[][]> rows = new ArrayList<>();
+ readBuilder
+ .newRead()
+ .createReader(plan)
+ .forEachRemaining(
+ row ->
+ rows.add(
+ new byte[][] {
+ row.isNullAt(0) ? null :
row.getBlob(0).toData(),
+ row.isNullAt(1) ? null :
row.getBlob(1).toData()
+ }));
+ return rows;
+ }
+
+ private static void assignFirstRowId(List<CommitMessage> messages, long
firstRowId) {
+ for (CommitMessage message : messages) {
+ List<DataFileMeta> newFiles =
+ ((CommitMessageImpl)
message).newFilesIncrement().newFiles();
+ List<DataFileMeta> written = new ArrayList<>(newFiles);
+ newFiles.clear();
+ for (DataFileMeta file : written) {
+ newFiles.add(file.assignFirstRowId(firstRowId));
+ }
+ }
+ }
+
@Test
public void testProjectOnlyTwoBlobFields() throws Exception {
createTableDefault();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
index 3ecd11bf33..ea6d242f40 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/VectorStoreTableTest.java
@@ -288,13 +288,231 @@ public class VectorStoreTableTest extends
DataEvolutionTestBase {
.containsExactly((Float) null);
}
+ @Test
+ public void
testPartialUpdateWithMissingVectorRangeWithoutDeletionVectors() throws
Exception {
+ // Same partial-vector layout as
testPartialUpdateWithMissingVectorRange but without
+ // deletion vectors: embedding_v2 is populated only for rows [2, 3],
so projecting it must
+ // still emit rows [0, 1] as NULL.
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ write(getTableDefault(), GenericRow.of(0, vector(1)), GenericRow.of(1,
vector(1)));
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.addColumn(
+ "embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))),
+ false);
+ write(
+ getTableDefault(),
+ GenericRow.of(2, vector(1), vector(10)),
+ GenericRow.of(3, vector(1), vector(10)));
+ updateVectors(
+ 2,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(200)),
+ GenericRow.of(vector(200)));
+ compactVectorTable();
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {2});
+ List<Split> splits = builder.newScan().plan().splits();
+ assertThat(splits).hasSize(1);
+ DataSplit split = (DataSplit) splits.get(0);
+ assertThat(readVectorValues(builder.newRead().createReader(split)))
+ .containsExactly(null, null, 200F, 200F);
+ }
+
+ @Test
+ public void testNewerVectorUpdateOverPartOfTheColumn() throws Exception {
+ // embedding is written for rows [0, 3], then updated for [0, 1] only.
Together the two
+ // files cover the group, but the newer one overlaps the older, so the
sequential reader
+ // would keep just [0, 1].
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ write(
+ getTableDefault(),
+ GenericRow.of(0, vector(1)),
+ GenericRow.of(1, vector(1)),
+ GenericRow.of(2, vector(1)),
+ GenericRow.of(3, vector(1)));
+ updateVectors(
+ 0,
+ Collections.singletonList("embedding"),
+ GenericRow.of(vector(100)),
+ GenericRow.of(vector(100)));
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {1});
+
assertThat(readVectorValues(builder.newRead().createReader(builder.newScan().plan())))
+ .containsExactly(100F, 100F, 1F, 1F);
+ }
+
+ @Test
+ public void testNewerVectorUpdateOverPartOfTheColumnWithDeletionVectors()
throws Exception {
+ catalog.createTable(
+ identifier(),
+ vectorSchema("json")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .build(),
+ false);
+ write(
+ getTableDefault(),
+ GenericRow.of(0, vector(10)),
+ GenericRow.of(1, vector(11)),
+ GenericRow.of(2, vector(12)),
+ GenericRow.of(3, vector(13)));
+ updateVectors(
+ 0,
+ Collections.singletonList("embedding"),
+ GenericRow.of(vector(100)),
+ GenericRow.of(vector(101)));
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {1});
+ List<Split> splits = builder.newScan().plan().splits();
+ assertThat(splits).hasSize(1);
+ DataSplit split = (DataSplit) splits.get(0);
+
+ // Delete one row from the updated range and one from the original
range.
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(1);
+ deletionVector.delete(2);
+ DeletionVectorsIndexFile indexFile =
+ getTableDefault()
+ .store()
+ .newIndexFileHandler()
+ .dvIndex(split.partition(), split.bucket());
+ String anchor = retrieveAnchorFile(split.dataFiles(),
Function.identity()).fileName();
+ Map<String, DeletionFile> deletionFiles =
+ indexFile.toDeletionFiles(
+ Collections.singletonList(
+ indexFile.writeSingleFile(
+ Collections.singletonMap(anchor,
deletionVector))));
+ DataSplit deletedSplit =
+ DataSplit.builder()
+ .withPartition(split.partition())
+ .withBucket(split.bucket())
+ .withBucketPath(split.bucketPath())
+ .withDataFiles(split.dataFiles())
+ .withDataDeletionFiles(
+ split.dataFiles().stream()
+ .map(file ->
deletionFiles.get(file.fileName()))
+ .collect(Collectors.toList()))
+ .build();
+
assertThat(readVectorValues(builder.newRead().createReader(deletedSplit)))
+ .containsExactly(100F, 13F);
+ }
+
+ @Test
+ public void testOlderVectorUpdateNewerThanTheNextKeptFile() throws
Exception {
+ // [0, 9] s1, then [5, 9] s2, [2, 7] s3 and [0, 4] s4. The newest file
and [5, 9] tile the
+ // group, but rows [5, 7] must come from [2, 7], which is older than
[0, 4] only.
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ InternalRow[] rows = new InternalRow[10];
+ for (int i = 0; i < 10; i++) {
+ rows[i] = GenericRow.of(i, vector(1));
+ }
+ write(getTableDefault(), rows);
+ updateVectors(5, Collections.singletonList("embedding"), vectorRows(5,
2));
+ updateVectors(2, Collections.singletonList("embedding"), vectorRows(6,
3));
+ updateVectors(0, Collections.singletonList("embedding"), vectorRows(5,
4));
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {1});
+
assertThat(readVectorValues(builder.newRead().createReader(builder.newScan().plan()),
10))
+ .containsExactly(4F, 4F, 4F, 4F, 4F, 3F, 3F, 3F, 2F, 2F);
+ }
+
+ @Test
+ public void testVectorColumnBackfilledByTwoUpdatesLeavesTailNull() throws
Exception {
+ // embedding_v2 is backfilled for [0, 4] and [5, 7] by two updates;
[8, 9] stay NULL.
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ InternalRow[] rows = new InternalRow[10];
+ for (int i = 0; i < 10; i++) {
+ rows[i] = GenericRow.of(i, vector(1));
+ }
+ write(getTableDefault(), rows);
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.addColumn(
+ "embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))),
+ false);
+ updateVectors(0, Collections.singletonList("embedding_v2"),
vectorRows(5, 200));
+ updateVectors(5, Collections.singletonList("embedding_v2"),
vectorRows(3, 300));
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {2});
+
assertThat(readVectorValues(builder.newRead().createReader(builder.newScan().plan()),
10))
+ .containsExactly(200F, 200F, 200F, 200F, 200F, 300F, 300F,
300F, null, null);
+
+ // [0, 4] has no selected row and is pruned by the scan; [8, 9] still
needs NULL-filling.
+ builder.withRowRanges(Collections.singletonList(new Range(6, 9)));
+
assertThat(readVectorValues(builder.newRead().createReader(builder.newScan().plan())))
+ .containsExactly(300F, 300F, null, null);
+
+ // Every selected row is backfilled.
+ builder.withRowRanges(Arrays.asList(new Range(1, 1), new Range(6, 6)));
+
assertThat(readVectorValues(builder.newRead().createReader(builder.newScan().plan())))
+ .containsExactly(200F, 300F);
+ }
+
+ @Test
+ public void testVectorColumnWithAGapUnderRowRanges() throws Exception {
+ // embedding_v2 is backfilled for [0, 2] and [6, 9]; rows [3, 5] stay
NULL.
+ catalog.createTable(identifier(), vectorSchema("json").build(), false);
+ InternalRow[] rows = new InternalRow[10];
+ for (int i = 0; i < 10; i++) {
+ rows[i] = GenericRow.of(i, vector(1));
+ }
+ write(getTableDefault(), rows);
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.addColumn(
+ "embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))),
+ false);
+ updateVectors(0, Collections.singletonList("embedding_v2"),
vectorRows(3, 200));
+ updateVectors(6, Collections.singletonList("embedding_v2"),
vectorRows(4, 300));
+
+ ReadBuilder builder =
getTableDefault().newReadBuilder().withProjection(new int[] {0, 2});
+ // No selected row is in the gap.
+ builder.withRowRanges(Arrays.asList(new Range(1, 1), new Range(7, 7)));
+ assertThat(readIdAndVector(builder)).containsExactly("1=200.0",
"7=300.0");
+ // Row 4 is selected and must be NULL-filled.
+ builder.withRowRanges(Arrays.asList(new Range(1, 1), new Range(4, 4),
new Range(7, 7)));
+ assertThat(readIdAndVector(builder)).containsExactly("1=200.0",
"4=null", "7=300.0");
+ }
+
+ private static List<String> readIdAndVector(ReadBuilder builder) throws
IOException {
+ List<String> actual = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ builder.newRead().createReader(builder.newScan().plan())) {
+ reader.forEachRemaining(
+ row ->
+ actual.add(
+ row.getInt(0)
+ + "="
+ + (row.isNullAt(1)
+ ? null
+ :
row.getVector(1).toFloatArray()[0])));
+ }
+ return actual;
+ }
+
+ private static InternalRow[] vectorRows(int count, float value) {
+ InternalRow[] rows = new InternalRow[count];
+ for (int i = 0; i < count; i++) {
+ rows[i] = GenericRow.of(vector(value));
+ }
+ return rows;
+ }
+
private List<Float> readVectorValues(RecordReader<InternalRow> reader)
throws IOException {
+ return readVectorValues(reader, 4);
+ }
+
+ private List<Float> readVectorValues(RecordReader<InternalRow> reader, int
maxRows)
+ throws IOException {
List<Float> actual = new ArrayList<>();
try (RecordReader<InternalRow> closeable = reader) {
closeable.forEachRemaining(
row -> {
// Fail promptly if an all-missing projection produces
unbounded NULL rows.
- assertThat(actual.size()).isLessThan(4);
+ assertThat(actual.size()).isLessThan(maxRows);
actual.add(row.isNullAt(0) ? null :
row.getVector(0).toFloatArray()[0]);
});
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
index a891ff683a..f571003d2d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java
@@ -129,6 +129,268 @@ public class DataEvolutionFileStoreScanTest {
.containsExactly("data-older.parquet");
}
+ @Test
+ public void testReadTypePruningKeepsAnchorWhenOnlyDedicatedFilesRemain() {
+ // SELECT only a blob column: the anchor normal file does not write
it, so without
+ // this fix the kept list would hold the blob file alone and the
reader would derive
+ // the logical row range from the blob file's sub-range, dropping the
rows outside
+ Schema schema = createSchema("v", "b");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ // anchor normal file covers the whole group range [0, 9] but does not
write "b"
+ ManifestEntry anchorFile =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-anchor.parquet",
+ 0L,
+ new String[] {"v"},
+ new String[] {"v"},
+ null,
+ 1L,
+ 0L,
+ 10L);
+ // blob file covers only row ids [0, 2]
+ ManifestEntry blob =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b-0.blob",
+ 0L,
+ new String[] {"b"},
+ new String[] {"b"},
+ null,
+ 5L,
+ 0L,
+ 3L);
+
+ RowType readType = DataTypes.ROW(DataTypes.FIELD(1, "b",
DataTypes.STRING()));
+
+ List<ManifestEntry> pruned =
+ DataEvolutionFileStoreScan.pruneByReadType(
+ Arrays.asList(anchorFile, blob),
+ readType,
+ Collections.emptySet(),
+ false,
+ entry ->
fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+ assertThat(pruned)
+ .extracting(e -> e.file().fileName())
+ .containsExactlyInAnyOrder("data-b-0.blob",
"data-anchor.parquet");
+ }
+
+ @Test
+ public void testReadTypePruningKeepsAnchorOnlyWhenNoNormalFileRemains() {
+ // projection references the blob column and a normal column that a
newer full-range
+ // normal file writes: kept already contains a normal file, so the
anchor must not
+ // be pulled in just because the group has dedicated files
+ Schema schema = createSchema("u", "v", "b");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ // anchor writes only "u", which the query does not reference
+ ManifestEntry anchorFile =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-anchor.parquet",
+ 0L,
+ new String[] {"u"},
+ new String[] {"u"},
+ null,
+ 1L,
+ 0L,
+ 10L);
+ ManifestEntry newer =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-newer.parquet",
+ 0L,
+ new String[] {"v"},
+ new String[] {"v"},
+ null,
+ 5L,
+ 0L,
+ 10L);
+ ManifestEntry blob =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b-0.blob",
+ 0L,
+ new String[] {"b"},
+ new String[] {"b"},
+ null,
+ 5L,
+ 0L,
+ 3L);
+ RowType readType =
+ DataTypes.ROW(
+ DataTypes.FIELD(1, "v", DataTypes.STRING()),
+ DataTypes.FIELD(2, "b", DataTypes.INT()));
+
+ List<ManifestEntry> pruned =
+ DataEvolutionFileStoreScan.pruneByReadType(
+ Arrays.asList(anchorFile, newer, blob),
+ readType,
+ Collections.emptySet(),
+ false,
+ entry ->
fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+ assertThat(pruned)
+ .extracting(e -> e.file().fileName())
+ .containsExactlyInAnyOrder("data-newer.parquet",
"data-b-0.blob");
+ }
+
+ @Test
+ public void
testReadTypePruningKeepsAnchorEvenWhenDedicatedFilesCoverTheGroup() {
+ // Two rolled blob files cover the whole group [0, 9]; the anchor is
kept anyway, since only
+ // its metadata is needed and the reader opens no bunch without
requested fields.
+ Schema schema = createSchema("v", "b");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ ManifestEntry anchorFile =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-anchor.parquet",
+ 0L,
+ new String[] {"v"},
+ new String[] {"v"},
+ null,
+ 1L,
+ 0L,
+ 10L);
+ ManifestEntry blob0 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b-0.blob",
+ 0L,
+ new String[] {"b"},
+ new String[] {"b"},
+ null,
+ 5L,
+ 0L,
+ 5L);
+ ManifestEntry blob5 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b-5.blob",
+ 0L,
+ new String[] {"b"},
+ new String[] {"b"},
+ null,
+ 5L,
+ 5L,
+ 5L);
+ RowType readType = DataTypes.ROW(DataTypes.FIELD(1, "b",
DataTypes.STRING()));
+
+ List<ManifestEntry> pruned =
+ DataEvolutionFileStoreScan.pruneByReadType(
+ Arrays.asList(anchorFile, blob0, blob5),
+ readType,
+ Collections.emptySet(),
+ false,
+ entry ->
fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+ assertThat(pruned)
+ .extracting(e -> e.file().fileName())
+ .containsExactlyInAnyOrder("data-b-0.blob", "data-b-5.blob",
"data-anchor.parquet");
+ }
+
+ @Test
+ public void testReadTypePruningKeepsAnchorForMultiColumnBlobs() {
+ // SELECT two blob columns covering different sub-ranges: b1 over [0,
6] and b2 over [3, 9].
+ // Their union spans the whole group, but each column is read as its
own field bunch, so
+ // the anchor is needed for b1's gap at [7, 9].
+ Schema schema = createSchema("v", "b1", "b2");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ ManifestEntry anchorFile =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-anchor.parquet",
+ 0L,
+ new String[] {"v"},
+ new String[] {"v"},
+ null,
+ 1L,
+ 0L,
+ 10L);
+ ManifestEntry b1 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b1-0.blob",
+ 0L,
+ new String[] {"b1"},
+ new String[] {"b1"},
+ null,
+ 5L,
+ 0L,
+ 7L);
+ ManifestEntry b2 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b2-3.blob",
+ 0L,
+ new String[] {"b2"},
+ new String[] {"b2"},
+ null,
+ 5L,
+ 3L,
+ 7L);
+ RowType readType =
+ DataTypes.ROW(
+ DataTypes.FIELD(1, "b1", DataTypes.STRING()),
+ DataTypes.FIELD(2, "b2", DataTypes.STRING()));
+
+ List<ManifestEntry> pruned =
+ DataEvolutionFileStoreScan.pruneByReadType(
+ Arrays.asList(anchorFile, b1, b2),
+ readType,
+ Collections.emptySet(),
+ false,
+ entry ->
fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+ assertThat(pruned)
+ .extracting(e -> e.file().fileName())
+ .containsExactlyInAnyOrder(
+ "data-b1-0.blob", "data-b2-3.blob",
"data-anchor.parquet");
+ }
+
+ @Test
+ public void testReadTypePruningKeepsDedicatedOnlyGroup() {
+ // An incremental (DELTA) scan only sees the files a snapshot added,
so a group can hold two
+ // blob updates over [0, 1] while its normal file stays in an earlier
snapshot.
+ Schema schema = createSchema("v", "b1", "b2");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ ManifestEntry b1 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b1-0.blob",
+ 0L,
+ new String[] {"b1"},
+ new String[] {"b1"},
+ null,
+ 5L,
+ 0L,
+ 2L);
+ ManifestEntry b2 =
+ createManifestEntryWithDifferentColsAndFileName(
+ "data-b2-0.blob",
+ 0L,
+ new String[] {"b2"},
+ new String[] {"b2"},
+ null,
+ 5L,
+ 0L,
+ 2L);
+ RowType readType =
+ DataTypes.ROW(
+ DataTypes.FIELD(1, "b1", DataTypes.STRING()),
+ DataTypes.FIELD(2, "b2", DataTypes.STRING()));
+
+ List<ManifestEntry> pruned =
+ DataEvolutionFileStoreScan.pruneByReadType(
+ Arrays.asList(b1, b2),
+ readType,
+ Collections.emptySet(),
+ false,
+ entry ->
fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+ assertThat(pruned)
+ .extracting(e -> e.file().fileName())
+ .containsExactlyInAnyOrder("data-b1-0.blob", "data-b2-0.blob");
+ }
+
@Test
public void testEvolutionStatsSingleFile() {
Schema schema = createSchema("f0", "f1");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionVectorReadPlannerTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionVectorReadPlannerTest.java
new file mode 100644
index 0000000000..ea33b7035c
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionVectorReadPlannerTest.java
@@ -0,0 +1,290 @@
+/*
+ * 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.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+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 java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link DataEvolutionVectorReadPlanner}. */
+public class DataEvolutionVectorReadPlannerTest {
+
+ private static final RowType NORMAL_TYPE = RowType.of(new DataField(0,
"id", DataTypes.INT()));
+ private static final RowType VECTOR_TYPE =
+ RowType.of(new DataField(1, "embedding", DataTypes.BYTES()));
+ private static final RowType E1_TYPE = RowType.of(new DataField(2, "e1",
DataTypes.BYTES()));
+ private static final RowType E2_TYPE = RowType.of(new DataField(3, "e2",
DataTypes.BYTES()));
+
+ @Test
+ public void testRolledVectorFilesCoveringTheGroupKeepSequentialRead() {
+ // A rolled vector column: two files that together cover the normal
file's [0, 9].
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 5L, 2L),
+ file("data-2.vector.lance", "embedding", 5L, 5L, 2L));
+
+ assertThat(DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null))
+ .isNull();
+ }
+
+ @Test
+ public void testVectorColumnCoveringPartOfTheGroupPlansRanges() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 5L, 2L));
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 4), new Range(5, 9));
+ assertThat(ranges.get(0).files)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-1.vector.lance");
+ assertThat(ranges.get(1).files).isEmpty();
+ }
+
+ @Test
+ public void testNewerVectorFileOverPartOfTheColumnPlansRanges() {
+ // [0, 9] then a newer update of [0, 4]: the sequential bunch would
keep only [0, 4].
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 10L, 2L),
+ file("data-2.vector.lance", "embedding", 0L, 5L, 3L));
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 4), new Range(5, 9));
+ assertThat(ranges.get(0).files)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-2.vector.lance");
+ assertThat(ranges.get(1).files)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-1.vector.lance");
+ }
+
+ @Test
+ public void testNewerVectorFileOverlappingTheTailPlansRanges() {
+ // [0, 6] then a newer update of [3, 9]: the sequential bunch would
reject the overlap.
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 7L, 2L),
+ file("data-2.vector.lance", "embedding", 3L, 7L, 3L));
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 2), new Range(3, 9));
+ }
+
+ @Test
+ public void testSkippedFileNewerThanTheNextKeptFilePlansRanges() {
+ // [0, 4] s4 and [5, 9] s2 tile the group, but [2, 7] s3 is newer than
[5, 9] at [5, 7].
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 10L, 1L),
+ file("data-2.vector.lance", "embedding", 5L, 5L, 2L),
+ file("data-3.vector.lance", "embedding", 2L, 6L, 3L),
+ file("data-4.vector.lance", "embedding", 0L, 5L, 4L));
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 4), new Range(5, 7), new
Range(8, 9));
+ assertThat(ranges.get(1).files)
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-3.vector.lance");
+ }
+
+ @Test
+ public void testFullRewriteRolledOverAnOlderFileKeepsSequentialRead() {
+ // A newer full rewrite rolled into [0, 6] and [7, 9] fully shadows
the older [0, 9].
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 10L, 1L),
+ file("data-2.vector.lance", "embedding", 0L, 7L, 2L),
+ file("data-3.vector.lance", "embedding", 7L, 3L, 2L));
+
+ assertThat(DataEvolutionVectorReadPlanner.plan(files, VECTOR_TYPE,
this::rowType, null))
+ .isNull();
+ }
+
+ @Test
+ public void testCoverageIsJudgedPerVectorColumn() {
+ // e1 covers [0, 9] and e2 only [0, 4]: their union spans the group,
but e2's rows [5, 9]
+ // still need NULL-filling.
+ RowType readType = RowType.of(E1_TYPE.getFields().get(0),
E2_TYPE.getFields().get(0));
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "e1", 0L, 10L, 3L),
+ file("data-2.vector.lance", "e2", 0L, 5L, 2L));
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(files, readType,
this::rowType, null);
+
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 4), new Range(5, 9));
+ }
+
+ @Test
+ public void testRolledVectorFilesPrunedByRowRangesKeepSequentialRead() {
+ // Rolled [0, 3], [4, 6], [7, 9]; the scan drops [4, 6] because no
selected row is in it.
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 4L, 1L),
+ file("data-3.vector.lance", "embedding", 7L, 3L, 1L));
+ List<Range> rowRanges = Arrays.asList(new Range(1, 1), new Range(8,
8));
+
+ assertThat(
+ DataEvolutionVectorReadPlanner.plan(
+ files, VECTOR_TYPE, this::rowType, rowRanges))
+ .isNull();
+ }
+
+ @Test
+ public void testSelectedRowsOutsideTheVectorColumnPlanRanges() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 5L, 2L));
+
+ // No selected row falls outside [0, 4], so the sequential bunch reads
every one of them.
+ assertThat(
+ DataEvolutionVectorReadPlanner.plan(
+ files,
+ VECTOR_TYPE,
+ this::rowType,
+ Collections.singletonList(new Range(1, 1))))
+ .isNull();
+
+ List<DataEvolutionVectorReadPlanner.ReadRange> ranges =
+ DataEvolutionVectorReadPlanner.plan(
+ files,
+ VECTOR_TYPE,
+ this::rowType,
+ Arrays.asList(new Range(1, 1), new Range(7, 7)));
+ assertThat(ranges).isNotNull();
+ assertThat(ranges)
+ .extracting(r -> r.range)
+ .containsExactly(new Range(0, 4), new Range(5, 9));
+ assertThat(ranges.get(1).files).isEmpty();
+ }
+
+ @Test
+ public void testSelectedRowBeforeTheVectorColumnPlansRanges() {
+ // embedding covers only [5, 9].
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 5L, 5L, 2L));
+
+ assertThat(plan(files, new Range(5, 5), new Range(7, 7))).isNull();
+ assertThat(plan(files, new Range(1, 1), new Range(7, 7))).isNotNull();
+ assertThat(plan(files, new Range(4, 4), new Range(7, 7))).isNotNull();
+ }
+
+ @Test
+ public void testSelectedRowInAGapOrTheTailPlansRanges() {
+ // embedding covers [0, 2] and [6, 8]; [3, 5] and [9, 9] are uncovered.
+ List<DataFileMeta> files =
+ Arrays.asList(
+ file("data-0.parquet", "id", 0L, 10L, 1L),
+ file("data-1.vector.lance", "embedding", 0L, 3L, 2L),
+ file("data-2.vector.lance", "embedding", 6L, 3L, 3L));
+
+ assertThat(plan(files, new Range(1, 1), new Range(7, 7))).isNull();
+ assertThat(plan(files, new Range(1, 1), new Range(3, 3))).isNotNull();
+ assertThat(plan(files, new Range(1, 1), new Range(5, 5))).isNotNull();
+ assertThat(plan(files, new Range(1, 1), new Range(9, 9))).isNotNull();
+ }
+
+ private List<DataEvolutionVectorReadPlanner.ReadRange> plan(
+ List<DataFileMeta> files, Range... rowRanges) {
+ return DataEvolutionVectorReadPlanner.plan(
+ files, VECTOR_TYPE, this::rowType, Arrays.asList(rowRanges));
+ }
+
+ private RowType rowType(DataFileMeta file) {
+ if (file.writeCols().contains("embedding")) {
+ return VECTOR_TYPE;
+ } else if (file.writeCols().contains("e1")) {
+ return E1_TYPE;
+ } else if (file.writeCols().contains("e2")) {
+ return E2_TYPE;
+ }
+ return NORMAL_TYPE;
+ }
+
+ private static DataFileMeta file(
+ String fileName, String column, long firstRowId, long rowCount,
long sequence) {
+ return DataFileMeta.create(
+ fileName,
+ 100L,
+ rowCount,
+ row(0),
+ row(0),
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ sequence,
+ sequence,
+ 0L,
+ 0,
+ null,
+ null,
+ FileSource.APPEND,
+ null,
+ firstRowId,
+ Collections.singletonList(column));
+ }
+}
