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 de718276d6 [core] Support vector indexes on DV tables (#8930)
de718276d6 is described below
commit de718276d635850ff7fb7c134b38875c6370d550
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Aug 4 19:51:59 2026 +0800
[core] Support vector indexes on DV tables (#8930)
---
.../paimon/operation/FileStoreCommitImpl.java | 72 +++++++++-
.../source/AbstractDataEvolutionVectorRead.java | 24 +++-
.../paimon/operation/FileStoreCommitTest.java | 31 +++++
.../table/DataEvolutionDeletionVectorTest.java | 62 ++++++++-
.../table/source/VectorSearchBuilderTest.java | 50 +++++++
.../globalindex/GenericGlobalIndexBuilder.java | 7 -
...enericGlobalIndexBuilderDeletionVectorTest.java | 58 ++++++++
paimon-python/pypaimon/table/file_store_table.py | 14 +-
.../table/source/global_index_live_row_filter.py | 28 +++-
.../pypaimon/table/source/vector_search_read.py | 93 ++++++++-----
.../pypaimon/table/source/vector_search_scan.py | 13 +-
.../pypaimon/tests/table/file_store_table_test.py | 20 +++
.../pypaimon/tests/vector_search_filter_test.py | 154 ++++++++++++++++++++-
.../spark/read/SparkDataEvolutionVectorRead.java | 1 +
.../read/SparkDataEvolutionVectorReadTest.java | 54 +++++++-
15 files changed, 610 insertions(+), 71 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index bc22370c47..88d6083dec 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -340,7 +340,9 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
int generatedSnapshot = 0;
int attempts = 0;
- ManifestEntryChanges changes =
collectChanges(committable.fileCommittables());
+ List<CommitMessage> commitMessages = committable.fileCommittables();
+ ManifestEntryChanges changes = collectChanges(commitMessages);
+ Set<Pair<BinaryRow, Integer>> materializedBuckets =
materializedBuckets(commitMessages);
try {
List<SimpleFileEntry> appendSimpleEntries =
SimpleFileEntry.from(changes.appendTableFiles);
@@ -393,10 +395,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
|| !changes.compactIndexFiles.isEmpty()) {
attempts +=
tryCommit(
- CommitChangesProvider.provider(
- changes.compactTableFiles,
- changes.compactChangelog,
- changes.compactIndexFiles),
+ compactChangesProvider(changes,
materializedBuckets),
committable.identifier(),
committable.watermark(),
committable.properties(),
@@ -767,6 +766,69 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return changes;
}
+ private Set<Pair<BinaryRow, Integer>>
materializedBuckets(List<CommitMessage> commitMessages) {
+ if (!options.dataEvolutionEnabled() ||
!options.deletionVectorsEnabled()) {
+ return Collections.emptySet();
+ }
+
+ Set<Pair<BinaryRow, Integer>> result = new HashSet<>();
+ for (CommitMessage message : commitMessages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ if (commitMessage.compactIncrement().compactBefore().stream()
+ .noneMatch(
+ file ->
+ !isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ || commitMessage.compactIncrement().compactAfter().stream()
+ .anyMatch(file -> file.firstRowId() != null)) {
+ continue;
+ }
+ result.add(Pair.of(commitMessage.partition(),
commitMessage.bucket()));
+ }
+ return result;
+ }
+
+ @VisibleForTesting
+ CommitChangesProvider compactChangesProvider(
+ ManifestEntryChanges changes, Set<Pair<BinaryRow, Integer>>
materializedBuckets) {
+ if (materializedBuckets.isEmpty()) {
+ return CommitChangesProvider.provider(
+ changes.compactTableFiles, changes.compactChangelog,
changes.compactIndexFiles);
+ }
+
+ return latestSnapshot -> {
+ List<IndexManifestEntry> indexFiles =
+ changes.compactIndexFiles.stream()
+ // Replace global-index deletions prepared against
an older snapshot.
+ .filter(
+ entry ->
+ entry.kind() != FileKind.DELETE
+ ||
entry.indexFile().globalIndexMeta() == null
+ ||
!materializedBuckets.contains(
+ Pair.of(
+
entry.partition(),
+
entry.bucket())))
+ .collect(Collectors.toList());
+
+ // This provider is invoked again after every optimistic-commit
conflict. Scanning the
+ // latest snapshot here guarantees that an index committed
concurrently is either
+ // deleted by this attempt or makes this attempt retry and is
deleted by the next one.
+ if (latestSnapshot != null && latestSnapshot.indexManifest() !=
null) {
+ for (IndexManifestEntry entry :
+
indexManifestFile.read(latestSnapshot.indexManifest())) {
+ if (entry.indexFile().globalIndexMeta() != null
+ && materializedBuckets.contains(
+ Pair.of(entry.partition(),
entry.bucket()))) {
+ indexFiles.add(entry.toDeleteEntry());
+ }
+ }
+ }
+
+ return new CommitChanges(
+ changes.compactTableFiles, changes.compactChangelog,
indexFiles);
+ };
+ }
+
private int tryCommit(
CommitChangesProvider changesProvider,
long identifier,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
index 1f505cc128..75fbb2b5b7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalRow;
@@ -83,8 +84,8 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
protected final DataField vectorColumn;
protected final Map<String, String> options;
- /** Snapshot the plan was built against; pins live-row filtering to it. */
- @Nullable protected transient Snapshot planSnapshot;
+ /** Snapshot the plan was built against; pins filters and raw reads to it.
*/
+ @Nullable protected Snapshot planSnapshot;
private static final Comparator<long[]> WEAKEST_SCORE_FIRST =
Comparator.<long[]>comparingDouble(a -> Float.intBitsToFloat((int)
a[1]))
@@ -593,7 +594,7 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
}
private ReadBuilder newRawReadBuilder(RowType readType, boolean
includeFilter) {
- ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ ReadBuilder readBuilder =
rawReadTable().newReadBuilder().withReadType(readType);
if (partitionFilter != null) {
readBuilder.withPartitionFilter(partitionFilter);
}
@@ -603,6 +604,23 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
return readBuilder;
}
+ private FileStoreTable rawReadTable() {
+ if (planSnapshot == null) {
+ return table;
+ }
+
+ Map<String, String> pinOptions = new HashMap<>();
+ pinOptions.put(
+ CoreOptions.SCAN_MODE.key(),
CoreOptions.StartupMode.FROM_SNAPSHOT.toString());
+ pinOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
String.valueOf(planSnapshot.id()));
+ pinOptions.put(CoreOptions.SCAN_VERSION.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TAG_NAME.key(), null);
+ pinOptions.put(CoreOptions.SCAN_WATERMARK.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TIMESTAMP.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
+ return table.copyWithoutTimeTravel(pinOptions);
+ }
+
protected static void splitSearchSplits(
List<? extends VectorSearchSplit> splits,
List<IndexVectorSearchSplit> indexSplits,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index fb22df8c1f..6458f82fab 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -46,6 +46,7 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
+import org.apache.paimon.operation.commit.CommitChanges;
import org.apache.paimon.operation.commit.ConflictDetection;
import org.apache.paimon.operation.commit.ManifestEntryChanges;
import org.apache.paimon.operation.commit.RetryCommitResult;
@@ -912,6 +913,36 @@ public class FileStoreCommitTest {
assertThat(file).isEmpty();
}
+ @Test
+ public void
testMaterializedCompactionOnlyRefreshesGlobalIndexInSameBucket() throws
Exception {
+ TestFileStore store = createStore(false, 2);
+ BinaryRow partition = gen.getPartition(gen.next());
+ IndexManifestEntry materializedBucketDelete =
+ globalIndexDeleteEntry(partition, 0,
"materialized-bucket-index");
+ IndexManifestEntry otherBucketDelete =
+ globalIndexDeleteEntry(partition, 1, "other-bucket-index");
+ ManifestEntryChanges changes = new ManifestEntryChanges(2);
+ changes.compactIndexFiles.add(materializedBucketDelete);
+ changes.compactIndexFiles.add(otherBucketDelete);
+
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ CommitChanges refreshed =
+ commit.compactChangesProvider(
+ changes,
Collections.singleton(Pair.of(partition, 0)))
+ .provide(null);
+
+
assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete);
+ }
+ }
+
+ private static IndexManifestEntry globalIndexDeleteEntry(
+ BinaryRow partition, int bucket, String fileName) {
+ IndexFileMeta file =
+ new IndexFileMeta(
+ "btree", fileName, 1, 1, new GlobalIndexMeta(0, 0, 0,
null, null), null);
+ return new IndexManifestEntry(FileKind.DELETE, partition, bucket,
file);
+ }
+
@Test
public void testWriteStats() throws Exception {
TestFileStore store = createStore(false, 1,
CoreOptions.ChangelogProducer.NONE);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
index df0b646cc8..5b40fe4cb9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -37,6 +37,7 @@ import
org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
import org.apache.paimon.format.blob.BlobFileFormat;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.DeletionVectorMeta;
+import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
@@ -737,6 +738,31 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
"b|9|updated-9");
}
+ @Test
+ public void
testMaterializeCompactionDropsGlobalIndexCommittedAfterPreparation()
+ throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeBaseRows(table);
+ updateStructuredColumn(table);
+ commitDeletionVectors(table, DEFAULT_DV_SPECS);
+
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+
dynamicOptions.put(CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key(),
"true");
+ FileStoreTable compactTable = table.copy(dynamicOptions);
+ List<CommitMessage> compactMessages =
prepareDataEvolutionCompaction(compactTable, false);
+
+ String indexFile = "concurrent-global-index";
+ commitGlobalIndex(table, indexFile, FULL_RANGE);
+ assertThat(liveGlobalIndexFileNames(table)).containsExactly(indexFile);
+
+ commit(compactTable, compactMessages);
+
+ assertThat(liveGlobalIndexFileNames(table)).isEmpty();
+ assertThat(normalFileRowRanges(table)).containsExactly(new Range(15,
24));
+ }
+
@Test
public void
testMaterializeCompactionUsesRemainingSizeForLargeDeletedRange() throws
Exception {
createTableDefault();
@@ -1137,6 +1163,11 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
private void compactDataEvolutionTable(FileStoreTable table, boolean
compactBlob)
throws Exception {
+ commit(table, prepareDataEvolutionCompaction(table, compactBlob));
+ }
+
+ private List<CommitMessage> prepareDataEvolutionCompaction(
+ FileStoreTable table, boolean compactBlob) throws Exception {
Snapshot snapshot = table.latestSnapshot().get();
DataEvolutionCompactCoordinator coordinator =
new DataEvolutionCompactCoordinator(table, compactBlob, false,
snapshot);
@@ -1154,7 +1185,36 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
commitMessages.addAll(
new DataEvolutionCompactionCommitPreparation(table, snapshot)
.prepare(commitMessages));
- commit(table, commitMessages);
+ return commitMessages;
+ }
+
+ private void commitGlobalIndex(FileStoreTable table, String fileName,
Range rowRange)
+ throws Exception {
+ IndexFileMeta indexFile =
+ new IndexFileMeta(
+ "test-global-index",
+ fileName,
+ 1,
+ rowRange.count(),
+ new GlobalIndexMeta(rowRange.from, rowRange.to, 0,
null, null),
+ null);
+ commit(
+ table,
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ UNAWARE_BUCKET,
+ null,
+
DataIncrement.indexIncrement(Collections.singletonList(indexFile)),
+ CompactIncrement.emptyIncrement())));
+ }
+
+ private static List<String> liveGlobalIndexFileNames(FileStoreTable table)
{
+ return table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry -> entry.indexFile().globalIndexMeta() != null)
+ .map(entry -> entry.indexFile().fileName())
+ .sorted()
+ .collect(Collectors.toList());
}
private void commitDeletionVectors(FileStoreTable table, List<DvSpec>
deletionVectorSpecs)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
index 3713e0a38b..f287f1875e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
@@ -19,6 +19,10 @@
package org.apache.paimon.table.source;
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.BinaryRow;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
@@ -282,6 +286,52 @@ public class VectorSearchBuilderTest extends TableTestBase
{
assertThat(result.results()).contains(0L);
}
+ @Test
+ public void testRawFallbackPinsDataReadToPlanSnapshot() throws Exception {
+ catalog.createTable(
+ identifier("vector_raw_search_pinned_snapshot"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("vector_raw_search_pinned_snapshot"));
+
+ writeVectors(table, new float[][] {{0.0f, 0.0f}, {1.0f, 0.0f}});
+ VectorSearchBuilder builder =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {0.0f, 0.0f})
+ .withLimit(2)
+ .withVectorColumn(VECTOR_FIELD_NAME);
+ VectorScan.Plan plan = builder.newVectorScan().scan();
+
+ commitDeletionVectors(table, 0L);
+ Map<String, String> compactOptions = new HashMap<>();
+
compactOptions.put(CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key(),
"true");
+ FileStoreTable compactTable = table.copy(compactOptions);
+ Snapshot compactSnapshot = compactTable.latestSnapshot().get();
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(compactTable, false,
false, compactSnapshot);
+ List<DataEvolutionCompactTask> tasks = coordinator.plan();
+ assertThat(tasks)
+ .singleElement()
+ .extracting(DataEvolutionCompactTask::type)
+
.isEqualTo(DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION);
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataEvolutionCompactTask task : tasks) {
+ messages.add(task.doCompact(compactTable,
"test-vector-snapshot-pin"));
+ }
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(compactTable,
compactSnapshot)
+ .prepare(messages));
+ try (BatchTableCommit commit =
compactTable.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+
+ GlobalIndexResult result = builder.newVectorRead().read(plan);
+ assertThat(result.results()).containsExactlyInAnyOrder(0L, 1L);
+ }
+
@Test
public void testVectorLiveRowPlanningSkipsUnindexedDeletionVectors()
throws Exception {
catalog.createTable(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
index b20c513cd0..2872ac591d 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilder.java
@@ -67,13 +67,6 @@ public class GenericGlobalIndexBuilder implements
Serializable {
+ "but table '%s' has bucket = %d.",
table.name(),
table.coreOptions().bucket());
- checkArgument(
- !table.coreOptions().deletionVectorsEnabled(),
- "Generic global index does not support tables with deletion
vectors enabled. "
- + "Table '%s' has 'deletion-vectors.enabled' = true,
which may cause "
- + "deleted rows to be indexed.",
- table.name());
-
scanSnapshot = table.snapshotManager().latestSnapshot();
if (scanSnapshot == null) {
return Collections.emptyList();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java
new file mode 100644
index 0000000000..62de5eee68
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericGlobalIndexBuilderDeletionVectorTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.flink.globalindex;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests deletion-vector table support in {@link GenericGlobalIndexBuilder}.
*/
+class GenericGlobalIndexBuilderDeletionVectorTest {
+
+ private static FileStoreTable table(boolean deletionVectorsEnabled) {
+ FileStoreTable table = mock(FileStoreTable.class);
+ CoreOptions options = mock(CoreOptions.class);
+ when(options.bucket()).thenReturn(-1);
+
when(options.deletionVectorsEnabled()).thenReturn(deletionVectorsEnabled);
+ when(table.coreOptions()).thenReturn(options);
+ when(table.name()).thenReturn("T");
+ SnapshotManager snapshotManager = mock(SnapshotManager.class);
+ when(snapshotManager.latestSnapshot()).thenReturn(null);
+ when(table.snapshotManager()).thenReturn(snapshotManager);
+ return table;
+ }
+
+ @Test
+ void testGlobalIndexAllowedOnDeletionVectorTable() {
+ assertThatCode(() -> new GenericGlobalIndexBuilder(table(true)).scan())
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testGlobalIndexAllowedWithoutDeletionVectors() {
+ assertThatCode(() -> new
GenericGlobalIndexBuilder(table(false)).scan())
+ .doesNotThrowAnyException();
+ }
+}
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 4bc9b1c048..4181e3ae7c 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -503,6 +503,13 @@ class FileStoreTable(Table):
raise ValueError(f"Unsupported bucket mode: {bucket_mode}")
def copy(self, options: dict) -> 'FileStoreTable':
+ return self._copy(options, resolve_time_travel=True)
+
+ def copy_without_time_travel(self, options: dict) -> 'FileStoreTable':
+ """Copy this table while preserving its already resolved schema."""
+ return self._copy(options, resolve_time_travel=False)
+
+ def _copy(self, options: dict, resolve_time_travel: bool) ->
'FileStoreTable':
if CoreOptions.BUCKET.key() in options and
int(options.get(CoreOptions.BUCKET.key())) != self.options.bucket():
raise ValueError("Cannot change bucket number")
new_options = CoreOptions.copy(self.options).options.to_map()
@@ -514,9 +521,10 @@ class FileStoreTable(Table):
new_table_schema = self.table_schema.copy(new_options=new_options)
- time_travel_schema = self._try_time_travel(Options(new_options))
- if time_travel_schema is not None:
- new_table_schema = time_travel_schema
+ if resolve_time_travel:
+ time_travel_schema = self._try_time_travel(Options(new_options))
+ if time_travel_schema is not None:
+ new_table_schema = time_travel_schema
# Re-encode the branch into the identifier when the option changes, so
# current_branch() and any catalog-routed snapshot commit see the
diff --git
a/paimon-python/pypaimon/table/source/global_index_live_row_filter.py
b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py
index 4e7d2bfcdd..af96b85e5b 100644
--- a/paimon-python/pypaimon/table/source/global_index_live_row_filter.py
+++ b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py
@@ -19,6 +19,7 @@
from typing import Optional
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.deletionvectors.deletion_vector import DeletionVector
from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.split import DataSplit
@@ -26,8 +27,8 @@ from pypaimon.utils.range import Range
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
-def live_rows(table, partition_filter=None) -> Optional[RoaringBitmap64]:
- """Return current live global row ids for deletion-vector tables.
+def live_rows(table, partition_filter=None, snapshot=None) ->
Optional[RoaringBitmap64]:
+ """Return live global row ids at ``snapshot`` for deletion-vector tables.
``None`` means no live-row filter is needed. This keeps tables without
deletion vectors on the old zero-overhead path.
@@ -39,7 +40,8 @@ def live_rows(table, partition_filter=None) ->
Optional[RoaringBitmap64]:
or not deletion_vectors_enabled(False)):
return None
- read_builder = table.new_read_builder()
+ read_table = table_at_snapshot(table, snapshot)
+ read_builder = read_table.new_read_builder()
if partition_filter is not None:
read_builder = read_builder.with_partition_filter(partition_filter)
@@ -56,10 +58,28 @@ def live_rows(table, partition_filter=None) ->
Optional[RoaringBitmap64]:
# Phase 2: subtract each DV. Ranges are all unioned first (no re-add), and
# peak memory stays at one DV at a time.
for split in data_splits:
- _subtract_deleted_rows(table, rows, split)
+ _subtract_deleted_rows(read_table, rows, split)
return rows
+def table_at_snapshot(table, snapshot):
+ """Return a table pinned to ``snapshot``, clearing conflicting time
travel."""
+ if snapshot is None:
+ return table
+
+ pin_options = {
+ CoreOptions.SCAN_MODE.key(): "from-snapshot",
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id),
+ }
+ for option in (CoreOptions.SCAN_TAG_NAME,
+ CoreOptions.SCAN_WATERMARK,
+ CoreOptions.SCAN_TIMESTAMP,
+ CoreOptions.SCAN_TIMESTAMP_MILLIS):
+ if option.key() in table.table_schema.options:
+ pin_options[option.key()] = None
+ return table.copy_without_time_travel(pin_options)
+
+
def for_range(live_row_ids: Optional[RoaringBitmap64],
from_: int, to: int) -> Optional[RoaringBitmap64]:
if live_row_ids is None:
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index 84b0ae7c00..199dc42c93 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -41,11 +41,14 @@ class VectorSearchRead(ABC):
def read_plan(self, plan):
# type: (VectorSearchScanPlan) -> GlobalIndexResult
- return self.read(plan.splits())
+ return self._read(plan.splits(), plan.snapshot())
- @abstractmethod
def read(self, splits):
# type: (List[VectorSearchSplit]) -> GlobalIndexResult
+ return self._read(splits, None)
+
+ @abstractmethod
+ def _read(self, splits, snapshot):
pass
@@ -54,11 +57,14 @@ class BatchVectorSearchRead(ABC):
def read_batch_plan(self, plan):
# type: (VectorSearchScanPlan) -> List[GlobalIndexResult]
- return self.read_batch(plan.splits())
+ return self._read_batch(plan.splits(), plan.snapshot())
- @abstractmethod
def read_batch(self, splits):
# type: (List[VectorSearchSplit]) -> List[GlobalIndexResult]
+ return self._read_batch(splits, None)
+
+ @abstractmethod
+ def _read_batch(self, splits, snapshot):
pass
@@ -81,15 +87,15 @@ class AbstractVectorSearchReadImpl:
self._partition_filter = partition_filter
self._options = dict(options or {})
- def _pre_filters(self, splits):
+ def _pre_filters(self, splits, snapshot=None):
# type: (list) -> List[RoaringBitmap64]
"""Evaluate live-row/scalar filters and return one bitmap per index
split."""
if not splits:
return []
live_rows = global_index_live_row_filter.live_rows(
- self._table, self._partition_filter)
- matched_rows = self._scalar_matched_rows(splits)
+ self._table, self._partition_filter, snapshot)
+ matched_rows = self._scalar_matched_rows(splits, snapshot)
if live_rows is None and matched_rows is None:
return []
@@ -110,7 +116,7 @@ class AbstractVectorSearchReadImpl:
has_filter = True
return include_row_ids if has_filter else []
- def _scalar_matched_rows(self, splits):
+ def _scalar_matched_rows(self, splits, snapshot=None):
"""Evaluate scalar indexes and return matching global row ids."""
if self._filter is None:
return None
@@ -133,6 +139,7 @@ class AbstractVectorSearchReadImpl:
self._table,
index_files=scalar_files,
partition_filter=self._partition_filter,
+ snapshot=snapshot,
)
if scanner is None:
return RoaringBitmap64()
@@ -144,9 +151,9 @@ class AbstractVectorSearchReadImpl:
finally:
scanner.close()
- def _pre_filter(self, splits):
+ def _pre_filter(self, splits, snapshot=None):
# Backwards-compatible helper used by older tests/callers.
- pre_filters = self._pre_filters(splits)
+ pre_filters = self._pre_filters(splits, snapshot)
if not pre_filters:
return None
merged = RoaringBitmap64()
@@ -157,7 +164,7 @@ class AbstractVectorSearchReadImpl:
merged = RoaringBitmap64.or_(merged, bitmap)
return merged
- def _raw_pre_filter(self, splits):
+ def _raw_pre_filter(self, splits, snapshot=None):
if self._filter is None:
return None
raw_rows = _bitmap_of_ranges(_raw_row_ranges(splits))
@@ -180,6 +187,7 @@ class AbstractVectorSearchReadImpl:
self._table,
index_files=scalar_files,
partition_filter=self._partition_filter,
+ snapshot=snapshot,
)
if scanner is None:
return None
@@ -249,12 +257,12 @@ class AbstractVectorSearchReadImpl:
def _read_raw_search(self, raw_row_ranges, pre_filter, query_vector,
index_type=None, include_filter=True,
- score_candidates=None):
+ score_candidates=None, snapshot=None):
raw_row_ranges = _filtered_raw_row_ranges(raw_row_ranges, pre_filter)
if not raw_row_ranges:
return DictBasedScoredIndexResult({})
- table = self._read_raw_arrow(raw_row_ranges, include_filter)
+ table = self._read_raw_arrow(raw_row_ranges, include_filter, snapshot)
if table is None or table.num_rows == 0:
return DictBasedScoredIndexResult({})
@@ -278,17 +286,17 @@ class AbstractVectorSearchReadImpl:
)
return _scored_result(top_k_heap)
- def _read_raw_vectors(self, candidates, include_filter=True):
+ def _read_raw_vectors(self, candidates, include_filter=True,
snapshot=None):
return self._read_raw_candidate_vectors(
- candidates.to_range_list(), candidates, include_filter)
+ candidates.to_range_list(), candidates, include_filter, snapshot)
def _read_raw_candidate_vectors(self, raw_row_ranges, candidates,
- include_filter=True):
+ include_filter=True, snapshot=None):
raw_row_ranges = _filtered_raw_row_ranges(raw_row_ranges, None)
if not raw_row_ranges:
return {}
- table = self._read_raw_arrow(raw_row_ranges, include_filter)
+ table = self._read_raw_arrow(raw_row_ranges, include_filter, snapshot)
if table is None or table.num_rows == 0:
return {}
@@ -303,8 +311,10 @@ class AbstractVectorSearchReadImpl:
raw_vectors[row_id] = _to_vector_list(stored)
return raw_vectors
- def _read_raw_arrow(self, raw_row_ranges, include_filter):
- read_builder = self._table.new_read_builder()
+ def _read_raw_arrow(self, raw_row_ranges, include_filter, snapshot=None):
+ read_table = global_index_live_row_filter.table_at_snapshot(
+ self._table, snapshot)
+ read_builder = read_table.new_read_builder()
if self._partition_filter is not None:
read_builder = read_builder.with_partition_filter(
self._partition_filter)
@@ -331,17 +341,20 @@ class AbstractVectorSearchReadImpl:
)
return _scored_result(top_k_heap)
- def _read_raw_refine_search(self, candidates, query_vector,
index_type=None):
+ def _read_raw_refine_search(self, candidates, query_vector,
index_type=None,
+ snapshot=None):
return self._read_raw_candidate_search(
candidates.to_range_list(),
candidates,
query_vector,
index_type,
include_filter=False,
+ snapshot=snapshot,
)
def _read_raw_candidate_search(self, raw_row_ranges, candidates,
query_vector,
- index_type=None, include_filter=False):
+ index_type=None, include_filter=False,
+ snapshot=None):
return self._read_raw_search(
raw_row_ranges,
None,
@@ -349,6 +362,7 @@ class AbstractVectorSearchReadImpl:
index_type,
include_filter=include_filter,
score_candidates=candidates,
+ snapshot=snapshot,
)
def _raw_search_projection(self, include_filter):
@@ -390,7 +404,8 @@ class AbstractVectorSearchReadImpl:
return self._limit
return self._limit * refine_factor
- def _maybe_rerank_indexed_result(self, result, index_type, query_vector):
+ def _maybe_rerank_indexed_result(self, result, index_type, query_vector,
+ snapshot=None):
if (self._configured_refine_factor(index_type) == 0 or
result.results().is_empty()):
return result
@@ -399,9 +414,11 @@ class AbstractVectorSearchReadImpl:
candidates.results(),
query_vector,
index_type,
+ snapshot,
)
- def _maybe_rerank_indexed_results(self, results, index_type,
query_vectors):
+ def _maybe_rerank_indexed_results(self, results, index_type, query_vectors,
+ snapshot=None):
if self._configured_refine_factor(index_type) == 0:
return results
@@ -414,7 +431,8 @@ class AbstractVectorSearchReadImpl:
if union_candidates.is_empty():
return candidates
- raw_vectors = self._read_raw_vectors(union_candidates,
include_filter=False)
+ raw_vectors = self._read_raw_vectors(
+ union_candidates, include_filter=False, snapshot=snapshot)
metric = _raw_search_metric(
self._table, self._vector_column, self._options, index_type)
return [
@@ -458,8 +476,7 @@ class DataEvolutionVectorRead(AbstractVectorSearchReadImpl,
VectorSearchRead):
options=options)
self._query_vector = query_vector
- def read(self, splits):
- # type: (List[VectorSearchSplit]) -> GlobalIndexResult
+ def _read(self, splits, snapshot):
index_splits, raw_splits = _split_search_splits(splits)
if not index_splits and not raw_splits:
return GlobalIndexResult.create_empty()
@@ -467,20 +484,21 @@ class
DataEvolutionVectorRead(AbstractVectorSearchReadImpl, VectorSearchRead):
indexed = (
DictBasedScoredIndexResult({})
if not index_splits
- else self._read_indexed(index_splits, self._query_vector)
+ else self._read_indexed(index_splits, self._query_vector, snapshot)
)
raw_result = self._read_raw_search(
_raw_row_ranges(raw_splits),
- self._raw_pre_filter(raw_splits),
+ self._raw_pre_filter(raw_splits, snapshot),
self._query_vector,
_raw_search_index_type(raw_splits),
+ snapshot=snapshot,
)
return indexed.or_(raw_result).top_k(self._limit)
- def _read_indexed(self, splits, query_vector):
+ def _read_indexed(self, splits, query_vector, snapshot):
index_type = _vector_index_type(splits)
search_limit = self._indexed_search_limit(index_type)
- pre_filters = self._pre_filters(splits)
+ pre_filters = self._pre_filters(splits, snapshot)
futures = [
self._eval(
split.row_range_start, split.row_range_end,
@@ -504,7 +522,8 @@ class DataEvolutionVectorRead(AbstractVectorSearchReadImpl,
VectorSearchRead):
merged_scores[row_id] = score_getter(row_id)
indexed = DictBasedScoredIndexResult(merged_scores).top_k(search_limit)
- return self._maybe_rerank_indexed_result(indexed, index_type,
query_vector)
+ return self._maybe_rerank_indexed_result(
+ indexed, index_type, query_vector, snapshot)
class BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
@@ -519,8 +538,7 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
options=options)
self._query_vectors = list(query_vectors)
- def read_batch(self, splits):
- # type: (List[VectorSearchSplit]) -> List[GlobalIndexResult]
+ def _read_batch(self, splits, snapshot):
n = len(self._query_vectors)
index_splits, raw_splits = _split_search_splits(splits)
if not index_splits and not raw_splits:
@@ -530,7 +548,7 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
# passing that split's pre-filter. Each future returns n per-query
results.
index_type = _vector_index_type(index_splits)
search_limit = self._indexed_search_limit(index_type)
- pre_filters = self._pre_filters(index_splits)
+ pre_filters = self._pre_filters(index_splits, snapshot)
futures = [
self._eval_batch(
split.row_range_start, split.row_range_end,
@@ -561,16 +579,17 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
for i in range(n)
]
indexed_results = self._maybe_rerank_indexed_results(
- indexed_results, index_type, self._query_vectors)
+ indexed_results, index_type, self._query_vectors, snapshot)
# Each query: merge indexed results with the raw (brute-force)
fallback.
- raw_pre_filter = self._raw_pre_filter(raw_splits)
+ raw_pre_filter = self._raw_pre_filter(raw_splits, snapshot)
raw_ranges = _raw_row_ranges(raw_splits)
raw_index_type = _raw_search_index_type(raw_splits)
results = []
for i in range(n):
raw = self._read_raw_search(
- raw_ranges, raw_pre_filter, self._query_vectors[i],
raw_index_type)
+ raw_ranges, raw_pre_filter, self._query_vectors[i],
raw_index_type,
+ snapshot=snapshot)
results.append(indexed_results[i].or_(raw).top_k(self._limit))
return results
diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py
b/paimon-python/pypaimon/table/source/vector_search_scan.py
index 7e002064db..c97c173f23 100644
--- a/paimon-python/pypaimon/table/source/vector_search_scan.py
+++ b/paimon-python/pypaimon/table/source/vector_search_scan.py
@@ -25,7 +25,6 @@ from
pypaimon.globalindex.data_evolution_global_index_coverage import DataEvolut
from pypaimon.table.source.vector_search_split import (
IndexVectorSearchSplit,
RawVectorSearchSplit,
- VectorSearchSplit,
)
from pypaimon.utils.range import Range
@@ -33,14 +32,18 @@ from pypaimon.utils.range import Range
class VectorSearchScanPlan:
"""Plan of vector search scan."""
- def __init__(self, splits):
- # type: (List[VectorSearchSplit]) -> None
+ def __init__(self, splits, snapshot=None):
+ # type: (list, object) -> None
self._splits = splits
+ self._snapshot = snapshot
def splits(self):
- # type: () -> List[VectorSearchSplit]
+ # type: () -> list
return self._splits
+ def snapshot(self):
+ return self._snapshot
+
class VectorSearchScan(ABC):
"""Vector search scan to scan index files."""
@@ -213,7 +216,7 @@ class DataEvolutionVectorScan(VectorSearchScan):
)
)
- return VectorSearchScanPlan(splits)
+ return VectorSearchScanPlan(splits, snapshot)
def _has_intersection(ranges, row_range):
diff --git a/paimon-python/pypaimon/tests/table/file_store_table_test.py
b/paimon-python/pypaimon/tests/table/file_store_table_test.py
index 9690a550b2..0786984af8 100644
--- a/paimon-python/pypaimon/tests/table/file_store_table_test.py
+++ b/paimon-python/pypaimon/tests/table/file_store_table_test.py
@@ -19,6 +19,7 @@ import os
import shutil
import tempfile
import unittest
+from unittest import mock
import pyarrow as pa
@@ -84,6 +85,25 @@ class FileStoreTableTest(unittest.TestCase):
self.assertIn("Cannot change bucket number", str(context.exception))
+ def test_copy_without_time_travel_preserves_resolved_schema(self):
+ current_schema = self.table.table_schema
+ with mock.patch.object(
+ self.table,
+ "_try_time_travel",
+ side_effect=AssertionError("must not resolve time travel
again")):
+ copied_table = self.table.copy_without_time_travel({
+ CoreOptions.SCAN_MODE.key(): "from-snapshot",
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): "1",
+ })
+
+ self.assertEqual(current_schema.fields,
copied_table.table_schema.fields)
+ self.assertEqual(current_schema.id, copied_table.table_schema.id)
+ self.assertEqual(
+ "1",
+ copied_table.table_schema.options[
+ CoreOptions.SCAN_SNAPSHOT_ID.key()],
+ )
+
def test_consumer_manager(self):
"""Test that FileStoreTable has consumer_manager method."""
# Get consumer_manager
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 9b861a626c..8cd2d7e6a2 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -116,6 +116,12 @@ class _StubTable:
return _G()
return _P()
+ def copy(self, options):
+ return self
+
+ def copy_without_time_travel(self, options):
+ return self
+
def new_vector_search_builder(self):
from pypaimon.table.source.vector_search_builder import (
VectorSearchBuilderImpl,
@@ -289,6 +295,11 @@ def _install_raw_full_text_read_builder(table,
text_column_name, row_id_to_text,
def _patch_snapshot(testcase, entries, snapshot=None):
"""Stub IndexFileHandler.scan + snapshot resolution."""
+ if snapshot is None:
+ snapshot = types.SimpleNamespace(id=1)
+ elif not hasattr(snapshot, "id"):
+ snapshot.id = 1
+
mock.patch.stopall()
for attr in ("_scan_patch", "_travel_patch"):
patcher = getattr(testcase, attr, None)
@@ -309,7 +320,7 @@ def _patch_snapshot(testcase, entries, snapshot=None):
testcase._scan_patch.start()
testcase._travel_patch = mock.patch(
"pypaimon.snapshot.time_travel_util.TimeTravelUtil.try_travel_to_snapshot",
- return_value=snapshot if snapshot is not None else object())
+ return_value=snapshot)
testcase._travel_patch.start()
@@ -395,8 +406,13 @@ class GlobalIndexLiveRowFilterTest(unittest.TestCase):
class _Table:
options = _Options()
+ table_schema = _StubSchema()
file_io = object()
+ def copy_without_time_travel(self_inner, options):
+ calls["copy_options"] = options
+ return self_inner
+
def new_read_builder(self_inner):
calls["new_read_builder"] = True
return _Builder()
@@ -409,14 +425,18 @@ class GlobalIndexLiveRowFilterTest(unittest.TestCase):
return [1, 3]
partition_filter = object()
+ snapshot = types.SimpleNamespace(id=3)
+ table = _Table()
with mock.patch(
"pypaimon.table.source.global_index_live_row_filter."
"DeletionVector.read",
return_value=_DeletionVector()) as read:
rows = global_index_live_row_filter.live_rows(
- _Table(), partition_filter)
+ table, partition_filter, snapshot)
read.assert_called_once_with(_Table.file_io, deletion_file)
+ self.assertEqual("from-snapshot", calls["copy_options"]["scan.mode"])
+ self.assertEqual("3", calls["copy_options"]["scan.snapshot-id"])
self.assertIs(partition_filter, calls["partition_filter"])
self.assertTrue(calls["new_read_builder"])
self.assertTrue(calls["new_scan"])
@@ -938,7 +958,8 @@ class VectorSearchFilterTest(unittest.TestCase):
]
self.table = _StubTable(fields=[self.id_field, self.embedding_field],
entries=self.entries)
- _patch_snapshot(self, self.entries)
+ self.snapshot = types.SimpleNamespace(id=1, next_row_id=10)
+ _patch_snapshot(self, self.entries, self.snapshot)
def tearDown(self):
mock.patch.stopall()
@@ -1009,13 +1030,20 @@ class VectorSearchFilterTest(unittest.TestCase):
return _FakeReader()
with mock.patch(
+ "pypaimon.table.source.vector_search_read."
+ "global_index_live_row_filter.live_rows",
+ return_value=None) as live_rows_fn, \
+ mock.patch(
"pypaimon.globalindex.data_evolution_global_index_scanner.DataEvolutionGlobalIndexScanner.create",
- return_value=scanner), \
+ return_value=scanner) as scanner_factory, \
mock.patch(
"pypaimon.table.source.vector_search_read._create_vector_reader",
side_effect=_capture_create):
self._builder(filter_pred).new_vector_search_read().read_plan(scan_plan)
+ self.assertIs(self.snapshot, scan_plan.snapshot())
+ live_rows_fn.assert_called_once_with(self.table, None, self.snapshot)
+ self.assertIs(self.snapshot,
scanner_factory.call_args.kwargs["snapshot"])
# Pre-filter happened once with our filter.
self.assertEqual(1, scanner.scan.call_count)
self.assertIs(filter_pred, scanner.scan.call_args[0][0])
@@ -2753,6 +2781,124 @@ class VectorSearchManySplitsTest(unittest.TestCase):
self.assertEqual(["split"], calls["splits"])
self.assertEqual([5], sorted(list(result.results())))
+ def test_read_plan_pins_raw_search_to_planned_snapshot(self):
+ from pypaimon.table.source.vector_search_read import
DataEvolutionVectorRead
+ from pypaimon.table.source.vector_search_scan import
VectorSearchScanPlan
+ from pypaimon.table.source.vector_search_split import
RawVectorSearchSplit
+
+ snapshot = types.SimpleNamespace(id=7)
+ embedding_field = _field(1, "embedding", "FLOAT")
+ table = _StubTable(fields=[embedding_field], entries=[])
+ read_table = _StubTable(fields=[embedding_field], entries=[])
+ _install_raw_vector_read_builder(
+ read_table, "embedding", {0: [1.0]})
+ table.copy_without_time_travel = mock.Mock(return_value=read_table)
+ plan = VectorSearchScanPlan(
+ [RawVectorSearchSplit([Range(0, 0)], [], "ivf-flat")],
+ snapshot,
+ )
+
+ result = DataEvolutionVectorRead(
+ table,
+ limit=1,
+ vector_column=embedding_field,
+ query_vector=[1.0],
+ ).read_plan(plan)
+
+ table.copy_without_time_travel.assert_called_once_with({
+ CoreOptions.SCAN_MODE.key(): "from-snapshot",
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): "7",
+ })
+ self.assertEqual([0], sorted(list(result.results())))
+
+ def test_read_does_not_reuse_snapshot_from_previous_plan(self):
+ from pypaimon.table.source.vector_search_read import
DataEvolutionVectorRead
+ from pypaimon.table.source.vector_search_scan import
VectorSearchScanPlan
+ from pypaimon.table.source.vector_search_split import
RawVectorSearchSplit
+
+ snapshot = types.SimpleNamespace(id=7)
+ embedding_field = _field(1, "embedding", "FLOAT")
+ table = _StubTable(fields=[embedding_field], entries=[])
+ planned_table = _StubTable(fields=[embedding_field], entries=[])
+ _install_raw_vector_read_builder(table, "embedding", {1: [1.0]})
+ _install_raw_vector_read_builder(planned_table, "embedding", {0:
[1.0]})
+ table.copy_without_time_travel = mock.Mock(return_value=planned_table)
+ split = RawVectorSearchSplit([Range(0, 1)], [], "ivf-flat")
+ reader = DataEvolutionVectorRead(
+ table,
+ limit=1,
+ vector_column=embedding_field,
+ query_vector=[1.0],
+ )
+
+ reader.read_plan(VectorSearchScanPlan([split], snapshot))
+ table.copy_without_time_travel.reset_mock()
+ result = reader.read([split])
+
+ table.copy_without_time_travel.assert_not_called()
+ self.assertEqual([1], sorted(list(result.results())))
+
+ def test_read_batch_does_not_reuse_snapshot_from_previous_plan(self):
+ from pypaimon.table.source.vector_search_read import
BatchVectorSearchReadImpl
+ from pypaimon.table.source.vector_search_scan import
VectorSearchScanPlan
+ from pypaimon.table.source.vector_search_split import
RawVectorSearchSplit
+
+ snapshot = types.SimpleNamespace(id=7)
+ embedding_field = _field(1, "embedding", "FLOAT")
+ table = _StubTable(fields=[embedding_field], entries=[])
+ planned_table = _StubTable(fields=[embedding_field], entries=[])
+ _install_raw_vector_read_builder(table, "embedding", {1: [1.0]})
+ _install_raw_vector_read_builder(planned_table, "embedding", {0:
[1.0]})
+ table.copy_without_time_travel = mock.Mock(return_value=planned_table)
+ split = RawVectorSearchSplit([Range(0, 1)], [], "ivf-flat")
+ reader = BatchVectorSearchReadImpl(
+ table,
+ limit=1,
+ vector_column=embedding_field,
+ query_vectors=[[1.0]],
+ )
+
+ reader.read_batch_plan(VectorSearchScanPlan([split], snapshot))
+ table.copy_without_time_travel.reset_mock()
+ results = reader.read_batch([split])
+
+ table.copy_without_time_travel.assert_not_called()
+ self.assertEqual([1], sorted(list(results[0].results())))
+
+ def test_read_plan_clears_conflicting_time_travel_options(self):
+ from pypaimon.table.source.vector_search_read import
DataEvolutionVectorRead
+ from pypaimon.table.source.vector_search_scan import
VectorSearchScanPlan
+ from pypaimon.table.source.vector_search_split import
RawVectorSearchSplit
+
+ snapshot = types.SimpleNamespace(id=7)
+ embedding_field = _field(1, "embedding", "FLOAT")
+ table = _StubTable(fields=[embedding_field], entries=[])
+ table.table_schema.options = {
+ CoreOptions.SCAN_TAG_NAME.key(): "tag-1",
+ CoreOptions.SCAN_TIMESTAMP.key(): "2026-08-03 12:00:00",
+ }
+ read_table = _StubTable(fields=[embedding_field], entries=[])
+ _install_raw_vector_read_builder(
+ read_table, "embedding", {0: [1.0]})
+ table.copy_without_time_travel = mock.Mock(return_value=read_table)
+
+ DataEvolutionVectorRead(
+ table,
+ limit=1,
+ vector_column=embedding_field,
+ query_vector=[1.0],
+ ).read_plan(VectorSearchScanPlan(
+ [RawVectorSearchSplit([Range(0, 0)], [], "ivf-flat")],
+ snapshot,
+ ))
+
+ table.copy_without_time_travel.assert_called_once_with({
+ CoreOptions.SCAN_MODE.key(): "from-snapshot",
+ CoreOptions.SCAN_SNAPSHOT_ID.key(): "7",
+ CoreOptions.SCAN_TAG_NAME.key(): None,
+ CoreOptions.SCAN_TIMESTAMP.key(): None,
+ })
+
def tearDown(self):
mock.patch.stopall()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
index 491fa035ef..a65f7dcb82 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
@@ -71,6 +71,7 @@ public class SparkDataEvolutionVectorRead extends
DataEvolutionVectorRead {
@Override
public GlobalIndexResult read(VectorScan.Plan plan) {
+ this.planSnapshot = plan.snapshot();
List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
splitSearchSplits(plan.splits(), indexSplits, rawSplits);
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
index b89efd1afb..2479118c32 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.read;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
@@ -48,6 +49,7 @@ import org.apache.paimon.table.source.VectorSearchSplit;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import org.apache.paimon.utils.SerializableFunction;
@@ -76,19 +78,63 @@ public class SparkDataEvolutionVectorReadTest {
@TempDir java.nio.file.Path tempDir;
@Test
- public void testRawSearchUsesSparkPath() {
+ public void testRawSearchUsesSparkPath() throws Exception {
TestingSparkVectorRead read = new TestingSparkVectorRead();
+ Snapshot snapshot = snapshot(1L);
RawVectorSearchSplit rawSplit =
new RawVectorSearchSplit(
Collections.singletonList(new Range(42, 42)),
Collections.emptyList(),
null);
- VectorScan.Plan plan = () -> Collections.singletonList(rawSplit);
+ VectorScan.Plan plan =
+ new VectorScan.Plan() {
+ @Override
+ public List<VectorSearchSplit> splits() {
+ return Collections.singletonList(rawSplit);
+ }
+
+ @Override
+ public Snapshot snapshot() {
+ return snapshot;
+ }
+ };
GlobalIndexResult result = read.read(plan);
assertThat(read.rawSparkPathUsed).isTrue();
+ assertThat(read.plannedSnapshot()).isSameAs(snapshot);
assertThat(result.results().contains(42L)).isTrue();
+
+ byte[] serialized = InstantiationUtil.serializeObject(read);
+ TestingSparkVectorRead restored =
+ InstantiationUtil.deserializeObject(
+ serialized,
Thread.currentThread().getContextClassLoader());
+ assertThat(restored.plannedSnapshot().id()).isEqualTo(snapshot.id());
+ }
+
+ private static Snapshot snapshot(long id) {
+ return new Snapshot(
+ id,
+ 0L,
+ "base-manifest-list",
+ null,
+ "delta-manifest-list",
+ null,
+ null,
+ null,
+ null,
+ "user",
+ 0L,
+ Snapshot.CommitKind.APPEND,
+ 0L,
+ 0L,
+ 0L,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
}
@Test
@@ -177,6 +223,10 @@ public class SparkDataEvolutionVectorReadTest {
null);
}
+ private Snapshot plannedSnapshot() {
+ return planSnapshot;
+ }
+
@Override
protected GlobalIndexResult readSplits(List<? extends
VectorSearchSplit> splits) {
throw new AssertionError("Raw search should not fall back to local
vector read.");
