This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 7a7544a3f1980ab033bfd7356fedcd300b11a96e 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;
