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 289010a422 [core] Introduce a vector read planner for data evolution
(#9877)
289010a422 is described below
commit 289010a4223138250c740a8157050cd3ebd46ea7
Author: zhoulii <[email protected]>
AuthorDate: Thu Sep 17 12:11:22 2026 +0800
[core] Introduce a vector read planner for data evolution (#9877)
---
.../paimon/operation/DataEvolutionSplitRead.java | 114 +++++++-
.../operation/DataEvolutionVectorReadPlanner.java | 183 +++++++++++++
.../apache/paimon/append/VectorStoreTableTest.java | 303 ++++++++++++++++++++-
3 files changed, 596 insertions(+), 4 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
index 919100ec1f..de0f8fd5ee 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
@@ -278,7 +278,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
createReader(dataSplit, rowRanges, info.actualReadType), info);
}
- private DataEvolutionFileReader createUnionReader(
+ private RecordReader<InternalRow> createUnionReader(
List<DataFileMeta> needMergeFiles,
BinaryRow partition,
DataFilePathFactory dataFilePathFactory,
@@ -286,6 +286,54 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
RowType readRowType,
@Nullable DeletionVectorWithRange deletionVector)
throws IOException {
+ List<DataEvolutionVectorReadPlanner.ReadRange> vectorRanges =
+ DataEvolutionVectorReadPlanner.plan(
+ needMergeFiles,
+ readRowType,
+ file ->
+ schemaFetcher
+ .apply(file.schemaId())
+ .dataFileSchema(file.writeCols())
+ .logicalRowType());
+ if (vectorRanges != null) {
+ List<FieldBunch> nonVectorBunches =
+ splitFieldBunches(
+ needMergeFiles.stream()
+ .filter(file ->
!isVectorStoreFile(file.fileName()))
+ .collect(Collectors.toList()),
+ file ->
schemaFetcher.apply(file.schemaId()).logicalRowType(),
+ rowRanges != null);
+ List<Range> selectedRanges = Range.sortAndMergeOverlap(rowRanges,
true);
+ List<ReaderSupplier<InternalRow>> suppliers = new ArrayList<>();
+ for (DataEvolutionVectorReadPlanner.ReadRange vectorRange :
vectorRanges) {
+ List<Range> ranges =
Collections.singletonList(vectorRange.range);
+ if (rowRanges != null) {
+ ranges = Range.and(ranges, selectedRanges);
+ }
+ if (ranges.isEmpty()) {
+ continue;
+ }
+ // Union readers have fixed field offsets. Apply the same
selection to every
+ // column reader so their rows remain aligned when a vector
provider changes.
+ List<Range> readRanges = ranges;
+ List<FieldBunch> bunches = new ArrayList<>(nonVectorBunches);
+ // Providers are newest-first within this range. Keep fields
from the same
+ // physical file together so the column planner opens that
file only once.
+ vectorRange.files.forEach(file -> bunches.add(new
DataBunch(file)));
+ suppliers.add(
+ () ->
+ createUnionReader(
+ bunches,
+ needMergeFiles,
+ partition,
+ dataFilePathFactory,
+ readRanges,
+ readRowType,
+ deletionVector));
+ }
+ return ConcatRecordReader.create(suppliers);
+ }
+
List<FieldBunch> fieldsFiles =
splitFieldBunches(
needMergeFiles,
@@ -298,6 +346,26 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
},
rowRanges != null);
+ return createUnionReader(
+ fieldsFiles,
+ needMergeFiles,
+ partition,
+ dataFilePathFactory,
+ rowRanges,
+ readRowType,
+ deletionVector);
+ }
+
+ private RecordReader<InternalRow> createUnionReader(
+ List<FieldBunch> fieldsFiles,
+ List<DataFileMeta> needMergeFiles,
+ BinaryRow partition,
+ DataFilePathFactory dataFilePathFactory,
+ List<Range> rowRanges,
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
+ throws IOException {
+
long rowCount = fieldsFiles.get(0).rowCount();
long firstRowId = bunchFirstRowId(fieldsFiles.get(0));
@@ -331,6 +399,21 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
DataEvolutionReadPlanner.DataEvolutionReadPlan plan =
new DataEvolutionReadPlanner(readRowType, bunchAvailTypes,
nestedFieldEnabled)
.plan();
+ if (plan.bunchReadFields.stream().allMatch(List::isEmpty)) {
+ // For example, a newly added vector column may cover only part of
the normal file's
+ // row range. Projecting only that column leaves no fields to
merge in uncovered ranges,
+ // but the selected rows must still be emitted.
+ // Read one bunch and let schema evolution fill the missing fields
with NULL.
+ return createMissingFieldsReader(
+ partition,
+ fieldsFiles.get(0),
+ bunchDataSchemas[0],
+ dataFilePathFactory,
+ formatBuilder,
+ rowRanges,
+ readRowType,
+ deletionVector);
+ }
// Build the per-bunch readers from the planned partial read row types.
for (int i = 0; i < numBunches; i++) {
@@ -379,6 +462,35 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
plan.rowOffsets, plan.fieldOffsets, fileRecordReaders);
}
+ private RecordReader<InternalRow> createMissingFieldsReader(
+ BinaryRow partition,
+ FieldBunch bunch,
+ TableSchema dataSchema,
+ DataFilePathFactory dataFilePathFactory,
+ Builder formatBuilder,
+ List<Range> rowRanges,
+ RowType readRowType,
+ @Nullable DeletionVectorWithRange deletionVector)
+ throws IOException {
+ DataFileMeta firstFile = bunch.files().get(0);
+ // Use the physical schema: the full table schema may declare columns
this file never wrote.
+ FormatReaderMapping mapping =
+ formatBuilder.build(
+ readTarget(firstFile, dataFilePathFactory,
rowRanges).formatIdentifier,
+ schema,
+ dataSchema,
+ readRowType.getFields(),
+ false);
+ return createFieldBunchReader(
+ partition,
+ bunch,
+ dataFilePathFactory,
+ mapping,
+ rowRanges,
+ readRowType,
+ deletionVector);
+ }
+
private boolean nestedFieldEnabledFor(List<DataFileMeta> files) {
if (coreOptions.dataEvolutionNestedFieldEnabled()) {
return true;
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
new file mode 100644
index 0000000000..1b77489cbf
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java
@@ -0,0 +1,183 @@
+/*
+ * 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.io.DataFilePathFactory;
+import org.apache.paimon.operation.DataEvolutionSplitRead.VectorStoreBunchKey;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+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. */
+ @Nullable
+ static List<ReadRange> plan(
+ List<DataFileMeta> files,
+ RowType readType,
+ Function<DataFileMeta, RowType> fileToRowType) {
+ Set<Integer> readIds =
+
readType.getFields().stream().map(DataField::id).collect(Collectors.toSet());
+ boolean vectorOnly = files.stream().allMatch(file ->
isVectorStoreFile(file.fileName()));
+ Map<Integer, VectorStoreBunchKey> fieldGroups = new HashMap<>();
+ Set<Integer> vectorReadIds = new HashSet<>();
+ List<Candidate> candidates = new ArrayList<>();
+ boolean overlappingGroups = false;
+ Range logicalRange = null;
+ long firstRowId = Long.MAX_VALUE;
+ long lastRowId = Long.MIN_VALUE;
+ for (DataFileMeta file : files) {
+ Range range = file.nonNullRowIdRange();
+ firstRowId = Math.min(firstRowId, range.from);
+ lastRowId = Math.max(lastRowId, range.to);
+ if (!isVectorStoreFile(file.fileName())) {
+ if (!isBlobFile(file.fileName())) {
+ logicalRange = range;
+ }
+ continue;
+ }
+ RowType rowType = fileToRowType.apply(file);
+ VectorStoreBunchKey key =
+ new VectorStoreBunchKey(
+ file.schemaId(),
+
DataFilePathFactory.formatIdentifier(file.fileName()),
+ file.writeCols(),
+ rowType);
+ Set<Integer> fieldIds = new HashSet<>();
+ for (DataField field : rowType.getFields()) {
+ // Match historical fields by id: a rename must not create a
different provider.
+ VectorStoreBunchKey previous =
fieldGroups.putIfAbsent(field.id(), key);
+ overlappingGroups |= previous != null && !previous.equals(key);
+ if (readIds.contains(field.id())) {
+ fieldIds.add(field.id());
+ vectorReadIds.add(field.id());
+ }
+ }
+ if (!fieldIds.isEmpty() || vectorOnly) {
+ candidates.add(new Candidate(file, fieldIds));
+ }
+ }
+
+ // Keep the sequential path for disjoint column groups, including
rolled vector files.
+ if (!overlappingGroups) {
+ return null;
+ }
+ if (logicalRange == null) {
+ logicalRange = new Range(firstRowId, lastRowId);
+ }
+
+ // 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.
+ TreeMap<Long, List<Candidate>> boundaries = new TreeMap<>();
+ boundaries.put(logicalRange.from, new ArrayList<>());
+ boundaries.put(logicalRange.to + 1, new ArrayList<>());
+ // Candidate ranges lie within the normal anchor, or the enclosing
range computed above.
+ for (Candidate candidate : candidates) {
+ Range range = candidate.file.nonNullRowIdRange();
+ boundaries.computeIfAbsent(range.from, ignored -> new
ArrayList<>()).add(candidate);
+ boundaries.computeIfAbsent(range.to + 1, ignored -> new
ArrayList<>()).add(candidate);
+ }
+
+ TreeSet<Candidate> active =
+ new TreeSet<>(
+ Comparator.<Candidate>comparingLong(c ->
c.file.maxSequenceNumber())
+ .reversed()
+ .thenComparing(c -> c.file.fileName()));
+ List<ReadRange> result = new ArrayList<>();
+ long start = logicalRange.from;
+ for (Map.Entry<Long, List<Candidate>> boundary :
boundaries.entrySet()) {
+ long end = boundary.getKey();
+ if (start < end) {
+ Set<Integer> assigned = new HashSet<>();
+ List<DataFileMeta> providers = new ArrayList<>();
+ for (Candidate candidate : active) {
+ // Presence in writeCols is what matters. An explicit NULL
in the newest
+ // file must overwrite the old value, just like any other
partial update.
+ if (assigned.addAll(candidate.fieldIds)) {
+ providers.add(candidate.file);
+ }
+ if (assigned.size() == vectorReadIds.size()) {
+ break;
+ }
+ }
+ if (providers.isEmpty() && vectorOnly && !active.isEmpty()) {
+ // Column pruning can remove the normal anchor. Retain a
row-count provider
+ // for ranges where the projected vector has not been
populated yet.
+ providers.add(active.first().file);
+ }
+ ReadRange previous = result.isEmpty() ? null :
result.get(result.size() - 1);
+ if (previous != null && previous.files.equals(providers)) {
+ previous.range = new Range(previous.range.from, end - 1);
+ } else {
+ result.add(new ReadRange(new Range(start, end - 1),
providers));
+ }
+ }
+ // Each file enters at its first row id and leaves just after its
last row id.
+ for (Candidate candidate : boundary.getValue()) {
+ if (!active.remove(candidate)) {
+ active.add(candidate);
+ }
+ }
+ start = end;
+ }
+ return result;
+ }
+
+ static class ReadRange {
+
+ Range range;
+ final List<DataFileMeta> files;
+
+ private ReadRange(Range range, List<DataFileMeta> files) {
+ this.range = range;
+ this.files = files;
+ }
+ }
+
+ private static class Candidate {
+
+ final DataFileMeta file;
+ final Set<Integer> fieldIds;
+
+ private Candidate(DataFileMeta file, Set<Integer> fieldIds) {
+ this.file = file;
+ this.fieldIds = fieldIds;
+ }
+ }
+}
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 c27f34d976..63d6618197 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
@@ -19,27 +19,51 @@
package org.apache.paimon.append;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BinaryVector;
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.DeletionVectorsIndexFile;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.DataEvolutionSplitRead;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.DataEvolutionTestBase;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
-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.BatchWriteBuilderImpl;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
+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.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 org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -48,16 +72,289 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for table with vector-store and data evolution. */
-public class VectorStoreTableTest extends TableTestBase {
+public class VectorStoreTableTest extends DataEvolutionTestBase {
private static final int VECTOR_DIM = 12;
private final AtomicInteger uniqueIdGen = new AtomicInteger(0);
private final Map<Integer, InternalRow> rowsWritten = new HashMap<>();
+ @Test
+ public void testPartialUpdateOfSharedVectorFile() throws Exception {
+ Schema schema =
+ vectorSchema("json")
+ .column("embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))
+ .build();
+ catalog.createTable(identifier(), schema, false);
+ BinaryVector original = vector(1);
+ BinaryVector updated = vector(2);
+ write(
+ getTableDefault(),
+ GenericRow.of(0, original, null),
+ GenericRow.of(1, original, original));
+
+ List<DataFileMeta> vectorFiles =
+ getTableDefault().store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file -> file.fileName().contains(".vector."))
+ .collect(Collectors.toList());
+ assertThat(vectorFiles.size()).isEqualTo(1);
+ assertThat(vectorFiles.get(0).writeCols())
+ .isEqualTo(Arrays.asList("embedding", "embedding_v2"));
+
+ updateVectors(
+ 0,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(updated),
+ GenericRow.of((Object) null));
+
+ List<InternalRow> actual = read(getTableDefault());
+ assertThat(actual).extracting(row -> row.getInt(0)).containsExactly(0,
1);
+ actual.forEach(
+ row ->
+ assertThat(row.getVector(1).toFloatArray())
+ .isEqualTo(original.toFloatArray()));
+
assertThat(actual.get(0).getVector(2).toFloatArray()).isEqualTo(updated.toFloatArray());
+ assertThat(actual.get(1).isNullAt(2)).isTrue();
+
+ List<InternalRow> projected = read(getTableDefault(), new int[] {2});
+ assertThat(projected.size()).isEqualTo(2);
+
assertThat(projected.get(0).getVector(0).toFloatArray()).isEqualTo(updated.toFloatArray());
+ assertThat(projected.get(1).isNullAt(0)).isTrue();
+
+ Table historical =
+ getTableDefault()
+
.copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"));
+ List<InternalRow> beforeUpdate = read(historical);
+ assertThat(beforeUpdate.size()).isEqualTo(2);
+ assertThat(beforeUpdate.get(0).isNullAt(2)).isTrue();
+ assertThat(beforeUpdate.get(1).getVector(2).toFloatArray())
+ .isEqualTo(original.toFloatArray());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "parquet"})
+ public void testPartialUpdateAcrossCompactedRanges(String format) throws
Exception {
+ catalog.createTable(
+ identifier(),
+ vectorSchema(format)
+ .column("embedding_v2", DataTypes.VECTOR(2,
DataTypes.FLOAT()))
+ .build(),
+ false);
+ write(
+ getTableDefault(),
+ GenericRow.of(0, vector(1), vector(10)),
+ GenericRow.of(1, vector(1), vector(10)));
+ 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(100)),
+ GenericRow.of(vector(100)));
+ updateVectors(
+ 2,
+ Arrays.asList("embedding", "embedding_v2"),
+ GenericRow.of(vector(2), vector(200)),
+ GenericRow.of(vector(2), vector(200)));
+ updateVectors(
+ 0,
+ Collections.singletonList("embedding_v2"),
+ GenericRow.of(vector(300)),
+ GenericRow.of(vector(300)));
+ compactVectorTable();
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.renameColumn("embedding_v2",
"query_embedding")),
+ false);
+
+ // The single-column file wins for rows 0-1, but the shared file wins
for rows 2-3.
+ // Sorting whole column groups by their maximum sequence would return
100 for rows 2-3.
+ List<InternalRow> rows = read(getTableDefault());
+ assertThat(rows).extracting(row -> row.getInt(0)).containsExactly(0,
1, 2, 3);
+ assertThat(rows)
+ .extracting(row -> row.getVector(1).toFloatArray()[0])
+ .containsExactly(1F, 1F, 2F, 2F);
+ assertThat(rows)
+ .extracting(row -> row.getVector(2).toFloatArray()[0])
+ .containsExactly(300F, 300F, 200F, 200F);
+ assertThat(read(getTableDefault(), new int[] {2}))
+ .extracting(row -> row.getVector(0).toFloatArray()[0])
+ .containsExactly(300F, 300F, 200F, 200F);
+ assertThat(read(getTableDefault(), new int[] {1}))
+ .extracting(row -> row.getVector(0).toFloatArray()[0])
+ .containsExactly(1F, 1F, 2F, 2F);
+
+ ReadBuilder builder =
+ getTableDefault()
+ .newReadBuilder()
+ .withProjection(new int[] {2})
+ .withRowRanges(Arrays.asList(new Range(1, 1), new
Range(2, 2)));
+ List<Float> selected = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ builder.newRead().createReader(builder.newScan().plan())) {
+ reader.forEachRemaining(row ->
selected.add(row.getVector(0).toFloatArray()[0]));
+ }
+ assertThat(selected).containsExactly(300F, 200F);
+ }
+
+ @Test
+ public void testPartialUpdateWithMissingVectorRange() throws Exception {
+ catalog.createTable(
+ identifier(),
+ vectorSchema("json")
+ // Keep the normal anchor when pruning columns to
preserve missing rows.
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .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);
+ assertThat(
+ readVectorValues(
+ builder.newRead()
+ .createReader(
+ new IndexedSplit(
+ split,
+ Arrays.asList(
+ new Range(1,
1), new Range(3, 3)),
+ null))))
+ .containsExactly(null, 200F);
+
+ // Delete a row from both the missing and populated vector ranges.
+ DeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(0);
+ 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(null, 200F);
+ assertThat(
+ readVectorValues(
+ builder.newRead()
+ .createReader(
+ new IndexedSplit(
+ deletedSplit,
+
Collections.singletonList(new Range(0, 2)),
+ null))))
+ .containsExactly((Float) null);
+ }
+
+ private List<Float> readVectorValues(RecordReader<InternalRow> reader)
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);
+ actual.add(row.isNullAt(0) ? null :
row.getVector(0).toFloatArray()[0]);
+ });
+ }
+ return actual;
+ }
+
+ private Schema.Builder vectorSchema(String format) {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("embedding", DataTypes.VECTOR(2, DataTypes.FLOAT()))
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.VECTOR_FILE_FORMAT.key(), format)
+ .option(CoreOptions.FILE_COMPRESSION.key(), "none")
+ .option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ }
+
+ private static BinaryVector vector(float value) {
+ return BinaryVector.fromPrimitiveArray(new float[] {value, 0});
+ }
+
+ private void updateVectors(long firstRowId, List<String> columns,
InternalRow... rows)
+ throws Exception {
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ ((BatchWriteBuilderImpl)
builder).rowIdCheckConflict(table.latestSnapshot().get().id());
+ try (BatchTableWrite writer =
+
builder.newWrite().withWriteType(table.rowType().project(columns));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (InternalRow row : rows) {
+ writer.write(row);
+ }
+ List<CommitMessage> messages = writer.prepareCommit();
+ setFirstRowId(messages, firstRowId);
+ commit.commit(messages);
+ }
+ }
+
+ private void compactVectorTable() throws Exception {
+ FileStoreTable table = getTableDefault();
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataEvolutionCompactTask task :
+ new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan()) {
+ messages.add(task.doCompact(table, commitUser));
+ }
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ List<DataFileMeta> normalFiles =
+ getTableDefault().store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file -> !file.fileName().contains(".vector."))
+ .collect(Collectors.toList());
+ assertThat(normalFiles).hasSize(1);
+ assertThat(normalFiles.get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 3));
+ }
+
@Test
public void testBasic() throws Exception {
int rowNum = RANDOM.nextInt(64) + 1;
