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 31a3e0a293 [core] Add physical search and full-text archive primitives
(#8649)
31a3e0a293 is described below
commit 31a3e0a2931a8cb9627eb6c95daf126d90215fff
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 14:31:51 2026 +0800
[core] Add physical search and full-text archive primitives (#8649)
Introduce the reusable physical-position search model and immutable
full-text archive primitives needed by primary-key table indexes. This
isolates the storage and ranking foundation before exposing primary-key
full-text configuration or query APIs.
---
.../org/apache/paimon/index/IndexFileHandler.java | 5 +
.../pkfulltext/PkFullTextBucketIndexState.java | 113 +++++++++
.../index/pkfulltext/PkFullTextDataFileReader.java | 108 ++++++++
.../index/pkfulltext/PkFullTextIndexBuilder.java | 70 ++++++
.../index/pkfulltext/PkFullTextIndexFile.java | 233 +++++++++++++++++
.../operation/PrimaryKeyIndexedSplitRead.java | 4 +-
...der.java => PrimaryKeyIndexPositionReader.java} | 6 +-
...ctorResult.java => PrimaryKeyScoredResult.java} | 187 +++++++-------
.../table/source/PrimaryKeySearchPosition.java | 139 +++++++++++
.../table/source/PrimaryKeySearchRanker.java | 183 ++++++++++++++
.../source/PrimaryKeyVectorPositionReader.java | 76 +-----
.../table/source/PrimaryKeyVectorResult.java | 197 +++------------
.../paimon/table/source/PrimaryKeyVectorScan.java | 41 ++-
.../table/source/VectorSearchBuilderImpl.java | 12 +-
.../pkfulltext/PkFullTextBucketIndexStateTest.java | 102 ++++++++
.../pkfulltext/PkFullTextDataFileReaderTest.java | 155 ++++++++++++
.../index/pkfulltext/PkFullTextIndexFileTest.java | 278 +++++++++++++++++++++
.../operation/PrimaryKeyIndexedSplitReadTest.java | 4 +-
.../table/source/PrimaryKeyScoredResultTest.java | 96 +++++++
.../table/source/PrimaryKeySearchRankerTest.java | 103 ++++++++
.../source/PrimaryKeyVectorPositionReaderTest.java | 3 +-
.../table/source/PrimaryKeyVectorResultTest.java | 6 +-
.../table/source/PrimaryKeyVectorScanTest.java | 6 +-
23 files changed, 1783 insertions(+), 344 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index 9fafedb7aa..8fa6554886 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -25,6 +25,7 @@ import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.deletionvectors.DeletionVectorsIndexFile;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
import org.apache.paimon.index.pksorted.PkSortedIndexFile;
import org.apache.paimon.index.pkvector.PkVectorAnnSegmentFile;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -90,6 +91,10 @@ public class IndexFileHandler {
return new PkVectorAnnSegmentFile(fileIO, pathFactories.get(partition,
bucket));
}
+ public PkFullTextIndexFile pkFullTextIndex(BinaryRow partition, int
bucket) {
+ return new PkFullTextIndexFile(fileIO, pathFactories.get(partition,
bucket));
+ }
+
public PkSortedIndexFile pkSortedIndex(BinaryRow partition, int bucket) {
return new PkSortedIndexFile(fileIO, pathFactories.get(partition,
bucket));
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexState.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexState.java
new file mode 100644
index 0000000000..01daa9f5e3
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexState.java
@@ -0,0 +1,113 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Immutable source-aligned primary-key full-text state for one bucket and
definition. */
+public final class PkFullTextBucketIndexState {
+
+ private final int textFieldId;
+ private final List<IndexFileMeta> currentPayloads;
+ private final List<IndexFileMeta> stalePayloads;
+ private final Map<String, IndexFileMeta> payloadBySourceFile;
+
+ public static PkFullTextBucketIndexState fromActivePayloads(
+ int textFieldId, List<IndexFileMeta> activePayloads) {
+ return new PkFullTextBucketIndexState(textFieldId, activePayloads);
+ }
+
+ public PkFullTextBucketIndexState(int textFieldId, List<IndexFileMeta>
activePayloads) {
+ this.textFieldId = textFieldId;
+
+ List<IndexFileMeta> current = new ArrayList<>();
+ List<IndexFileMeta> stale = new ArrayList<>();
+ Map<String, IndexFileMeta> bySource = new LinkedHashMap<>();
+ Set<String> payloadNames = new HashSet<>();
+ for (IndexFileMeta payload : activePayloads) {
+ GlobalIndexMeta globalMeta = payload.globalIndexMeta();
+ if (!PkFullTextIndexFile.INDEX_TYPE.equals(payload.indexType()) ||
globalMeta == null) {
+ continue;
+ }
+ if (globalMeta.indexFieldId() != textFieldId) {
+ if (globalMeta.sourceMeta() != null) {
+ stale.add(payload);
+ }
+ continue;
+ }
+ checkArgument(
+ payloadNames.add(payload.fileName()),
+ "Active full-text payload %s appears more than once in the
index manifest.",
+ payload.fileName());
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(payload);
+ long sourceRowCount = 0;
+ for (PrimaryKeyIndexSourceFile source : sourceMeta.sourceFiles()) {
+ sourceRowCount = Math.addExact(sourceRowCount,
source.rowCount());
+ }
+ checkArgument(
+ payload.rowCount() == sourceRowCount
+ && globalMeta.rowRangeStart() == 0
+ && globalMeta.rowRangeEnd() == sourceRowCount - 1,
+ "Full-text archive %s row metadata does not match its
source files.",
+ payload.fileName());
+ for (PrimaryKeyIndexSourceFile source : sourceMeta.sourceFiles()) {
+ IndexFileMeta previous = bySource.put(source.fileName(),
payload);
+ checkArgument(
+ previous == null,
+ "Source data file %s is covered by both full-text
archives %s and %s.",
+ source.fileName(),
+ previous == null ? "" : previous.fileName(),
+ payload.fileName());
+ }
+ current.add(payload);
+ }
+ this.currentPayloads = Collections.unmodifiableList(current);
+ this.stalePayloads = Collections.unmodifiableList(stale);
+ this.payloadBySourceFile = Collections.unmodifiableMap(bySource);
+ }
+
+ public int textFieldId() {
+ return textFieldId;
+ }
+
+ public List<IndexFileMeta> currentPayloads() {
+ return currentPayloads;
+ }
+
+ public List<IndexFileMeta> stalePayloads() {
+ return stalePayloads;
+ }
+
+ public Map<String, IndexFileMeta> payloadBySourceFile() {
+ return payloadBySourceFile;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReader.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReader.java
new file mode 100644
index 0000000000..a076fb817b
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReader.java
@@ -0,0 +1,108 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Reads one projected text value from every physical row of a data file. */
+public class PkFullTextDataFileReader implements Closeable {
+
+ private final DataFileMeta dataFile;
+ private final RecordReaderIterator<KeyValue> iterator;
+ private long rowsRead;
+
+ public PkFullTextDataFileReader(KeyValueFileReaderFactory readerFactory,
DataFileMeta dataFile)
+ throws IOException {
+ this.dataFile = dataFile;
+ this.iterator = new
RecordReaderIterator<>(readerFactory.createRecordReader(dataFile));
+ }
+
+ public long rowCount() {
+ return dataFile.rowCount();
+ }
+
+ @Nullable
+ public BinaryString readNextText() {
+ checkArgument(
+ rowsRead < rowCount(), "Read past data file %s row count.",
dataFile.fileName());
+ checkArgument(
+ iterator.hasNext(),
+ "Data file %s ended before its declared row count.",
+ dataFile.fileName());
+ KeyValue keyValue = iterator.next();
+ rowsRead++;
+ if (rowsRead == rowCount()) {
+ checkArgument(
+ !iterator.hasNext(),
+ "Data file %s contains more rows than declared.",
+ dataFile.fileName());
+ }
+ InternalRow value = keyValue.value();
+ return value.isNullAt(0) ? null : value.getString(0);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ iterator.close();
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException("Failed to close full-text reader for " +
dataFile.fileName(), e);
+ }
+ }
+
+ /** Bucket-scoped factory with text-only value projection and no
deletion-vector filtering. */
+ public static class Factory {
+
+ private final KeyValueFileReaderFactory readerFactory;
+
+ public Factory(
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder,
+ BinaryRow partition,
+ int bucket,
+ DataField textField) {
+ this.readerFactory =
+ readerFactoryBuilder
+ .copyWithoutProjection()
+ .withReadKeyType(RowType.of())
+ .withReadValueType(RowType.of(textField))
+ .buildWithoutDeletionVector(partition, bucket);
+ }
+
+ public PkFullTextDataFileReader create(DataFileMeta dataFile) throws
IOException {
+ return new PkFullTextDataFileReader(readerFactory, dataFile);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexBuilder.java
new file mode 100644
index 0000000000..f5718205b9
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexBuilder.java
@@ -0,0 +1,70 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Opens projected data-file readers and builds an ordered-source full-text
archive. */
+public class PkFullTextIndexBuilder {
+
+ private final PkFullTextIndexFile indexFile;
+ private final PkFullTextDataFileReader.Factory readerFactory;
+ private final DataField textField;
+ private final Options indexOptions;
+
+ public PkFullTextIndexBuilder(
+ PkFullTextIndexFile indexFile,
+ PkFullTextDataFileReader.Factory readerFactory,
+ DataField textField,
+ Options indexOptions) {
+ this.indexFile = indexFile;
+ this.readerFactory = readerFactory;
+ this.textField = textField;
+ this.indexOptions = indexOptions;
+ }
+
+ public IndexFileMeta build(DataFileMeta sourceFile) throws IOException {
+ PkFullTextDataFileReader reader = readerFactory.create(sourceFile);
+ return indexFile.build(sourceFile, reader, textField, indexOptions);
+ }
+
+ public IndexFileMeta build(List<DataFileMeta> sourceFiles) throws
IOException {
+ List<PkFullTextIndexFile.Source> sources = new
ArrayList<>(sourceFiles.size());
+ try {
+ for (DataFileMeta sourceFile : sourceFiles) {
+ sources.add(
+ new PkFullTextIndexFile.Source(
+ sourceFile, readerFactory.create(sourceFile)));
+ }
+ } catch (IOException | RuntimeException | Error failure) {
+ for (PkFullTextIndexFile.Source source : sources) {
+ source.close();
+ }
+ throw failure;
+ }
+ return indexFile.build(sources, textField, indexOptions);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFile.java
new file mode 100644
index 0000000000..361b8f00d0
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFile.java
@@ -0,0 +1,233 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFile;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.IOUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builds one immutable full-text archive from ordered compact primary-key
data files. */
+public class PkFullTextIndexFile extends IndexFile {
+
+ public static final String INDEX_TYPE = "full-text";
+
+ public PkFullTextIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
+ super(fileIO, pathFactory);
+ }
+
+ public IndexFileMeta build(
+ DataFileMeta sourceFile,
+ PkFullTextDataFileReader textReader,
+ DataField textField,
+ Options indexOptions)
+ throws IOException {
+ GlobalIndexer indexer;
+ try {
+ indexer = GlobalIndexer.create(INDEX_TYPE, textField,
indexOptions);
+ } catch (RuntimeException | Error failure) {
+ IOUtils.closeQuietly(textReader);
+ throw failure;
+ }
+ return build(
+ Collections.singletonList(new Source(sourceFile, textReader)),
textField, indexer);
+ }
+
+ public IndexFileMeta build(List<Source> sources, DataField textField,
Options indexOptions)
+ throws IOException {
+ GlobalIndexer indexer;
+ try {
+ indexer = GlobalIndexer.create(INDEX_TYPE, textField,
indexOptions);
+ } catch (RuntimeException | Error failure) {
+ closeSources(sources);
+ throw failure;
+ }
+ return build(sources, textField, indexer);
+ }
+
+ IndexFileMeta build(
+ DataFileMeta sourceFile,
+ PkFullTextDataFileReader textReader,
+ DataField textField,
+ GlobalIndexer indexer)
+ throws IOException {
+ return build(
+ Collections.singletonList(new Source(sourceFile, textReader)),
textField, indexer);
+ }
+
+ IndexFileMeta build(List<Source> sources, DataField textField,
GlobalIndexer indexer)
+ throws IOException {
+ checkArgument(!sources.isEmpty(), "A full-text archive must reference
source files.");
+ long totalRowCount = 0;
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new
ArrayList<>(sources.size());
+ for (Source source : sources) {
+ checkArgument(
+ source.sourceFile.rowCount() > 0, "Full-text source file
must contain rows.");
+ totalRowCount = Math.addExact(totalRowCount,
source.sourceFile.rowCount());
+ sourceFiles.add(
+ new PrimaryKeyIndexSourceFile(
+ source.sourceFile.fileName(),
source.sourceFile.rowCount()));
+ }
+
+ ArchiveFileWriter fileWriter = new ArchiveFileWriter();
+ GlobalIndexWriter writer = null;
+ boolean success = false;
+ try {
+ writer = indexer.createWriter(fileWriter);
+ checkArgument(
+ writer instanceof GlobalIndexSingleColumnWriter,
+ "Full-text indexer must create a single-column writer.");
+ GlobalIndexSingleColumnWriter singleColumnWriter =
+ (GlobalIndexSingleColumnWriter) writer;
+ long sourceOffset = 0;
+ for (Source source : sources) {
+ checkArgument(
+ source.textReader.rowCount() ==
source.sourceFile.rowCount(),
+ "Text row count %s does not match source file %s row
count %s.",
+ source.textReader.rowCount(),
+ source.sourceFile.fileName(),
+ source.sourceFile.rowCount());
+ for (long rowPosition = 0;
+ rowPosition < source.sourceFile.rowCount();
+ rowPosition++) {
+ singleColumnWriter.write(
+ source.textReader.readNextText(), sourceOffset +
rowPosition);
+ }
+ sourceOffset += source.sourceFile.rowCount();
+ }
+ List<ResultEntry> results = writer.finish();
+ checkArgument(
+ results.size() == 1,
+ "Full-text build must produce exactly one archive, but
produced %s.",
+ results.size());
+ checkArgument(
+ fileWriter.createdFiles.size() == 1,
+ "Full-text build must allocate exactly one archive, but
allocated %s.",
+ fileWriter.createdFiles.size());
+ ResultEntry result = results.get(0);
+ checkArgument(
+ result.rowCount() == totalRowCount,
+ "Full-text archive row count %s does not match source row
count %s.",
+ result.rowCount(),
+ totalRowCount);
+ Path archivePath = fileWriter.path(result.fileName());
+ byte[] archiveMetadata = result.meta() == null ? new byte[0] :
result.meta();
+ IndexFileMeta archive =
+ new IndexFileMeta(
+ INDEX_TYPE,
+ result.fileName(),
+ fileIO.getFileSize(archivePath),
+ result.rowCount(),
+ new GlobalIndexMeta(
+ 0,
+ totalRowCount - 1,
+ textField.id(),
+ null,
+ archiveMetadata,
+ new
PrimaryKeyIndexSourceMeta(sourceFiles).serialize()),
+ pathFactory.isExternalPath() ?
archivePath.toString() : null);
+ success = true;
+ return archive;
+ } finally {
+ for (Source source : sources) {
+ IOUtils.closeQuietly(source.textReader);
+ }
+ if (writer instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) writer);
+ }
+ if (!success) {
+ fileWriter.deleteCreatedFiles();
+ }
+ }
+ }
+
+ private static void closeSources(List<Source> sources) {
+ for (Source source : sources) {
+ IOUtils.closeQuietly(source.textReader);
+ }
+ }
+
+ /** One data-file source whose row positions are appended to an archive. */
+ public static class Source {
+
+ private final DataFileMeta sourceFile;
+ private final PkFullTextDataFileReader textReader;
+
+ public Source(DataFileMeta sourceFile, PkFullTextDataFileReader
textReader) {
+ this.sourceFile = sourceFile;
+ this.textReader = textReader;
+ }
+
+ void close() {
+ IOUtils.closeQuietly(textReader);
+ }
+ }
+
+ private class ArchiveFileWriter implements GlobalIndexFileWriter {
+
+ private final Map<String, Path> createdFiles = new HashMap<>();
+
+ @Override
+ public String newFileName(String prefix) {
+ Path path = pathFactory.newPath();
+ createdFiles.put(path.getName(), path);
+ return path.getName();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) throws
IOException {
+ return fileIO.newOutputStream(path(fileName), false);
+ }
+
+ private Path path(String fileName) {
+ Path path = createdFiles.get(fileName);
+ checkArgument(path != null, "Full-text archive %s was not
allocated.", fileName);
+ return path;
+ }
+
+ private void deleteCreatedFiles() {
+ for (Path path : createdFiles.values()) {
+ fileIO.deleteQuietly(path);
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitRead.java
index bb8d82e32c..519d2c4a94 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitRead.java
@@ -26,7 +26,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.PrimaryKeyVectorPositionReader;
+import org.apache.paimon.table.source.PrimaryKeyIndexPositionReader;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
@@ -116,6 +116,6 @@ public class PrimaryKeyIndexedSplitRead implements
SplitRead<InternalRow> {
IntFunction<Float> scoreGetter =
scoreByPosition == null ? position -> Float.NaN :
scoreByPosition::get;
FileRecordReader<InternalRow> reader =
rawRead.createFileReader(dataSplit, rowPositions);
- return new PrimaryKeyVectorPositionReader(reader, rowPositions,
scoreGetter);
+ return new PrimaryKeyIndexPositionReader(reader, rowPositions,
scoreGetter);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyIndexPositionReader.java
similarity index 94%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyIndexPositionReader.java
index 52f7423a92..cf49df828f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyIndexPositionReader.java
@@ -32,8 +32,8 @@ import java.util.function.IntFunction;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Reads selected physical file positions and exposes their vector-search
scores. */
-public class PrimaryKeyVectorPositionReader implements
ScoreRecordReader<InternalRow> {
+/** Reads selected primary-key physical file positions and exposes their
search scores. */
+public class PrimaryKeyIndexPositionReader implements
ScoreRecordReader<InternalRow> {
private final FileRecordReader<InternalRow> reader;
private final RoaringBitmap32 rowPositions;
@@ -41,7 +41,7 @@ public class PrimaryKeyVectorPositionReader implements
ScoreRecordReader<Interna
private final int lastPosition;
private boolean exhausted;
- public PrimaryKeyVectorPositionReader(
+ public PrimaryKeyIndexPositionReader(
FileRecordReader<InternalRow> reader,
RoaringBitmap32 rowPositions,
IntFunction<Float> scoreGetter) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyScoredResult.java
similarity index 51%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyScoredResult.java
index 3eba24b30c..187e2ae430 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyScoredResult.java
@@ -20,6 +20,8 @@ package org.apache.paimon.table.source;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.ScoreGetter;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
@@ -33,67 +35,90 @@ import java.util.Map;
import java.util.Objects;
import java.util.TreeMap;
-import static org.apache.paimon.globalindex.VectorSearchMetric.normalize;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Snapshot-scoped vector result addressed by physical primary-key table row
positions. */
-public class PrimaryKeyVectorResult implements GlobalIndexSplitResult {
-
- private final PrimaryKeyVectorScan.Plan plan;
- private final List<PrimaryKeyVectorRead.Candidate> candidates;
- private final String metric;
-
- PrimaryKeyVectorResult(
- PrimaryKeyVectorScan.Plan plan,
- List<PrimaryKeyVectorRead.Candidate> candidates,
- String metric) {
- this.plan = plan;
- this.candidates = Collections.unmodifiableList(new
ArrayList<>(candidates));
- this.metric = normalize(metric);
+/** Snapshot-scoped scored search result addressed by physical primary-key row
positions. */
+public class PrimaryKeyScoredResult implements GlobalIndexSplitResult,
ScoredGlobalIndexResult {
+
+ private final long snapshotId;
+ private final List<PrimaryKeySearchPosition> positions;
+ private final List<IndexedSplit> splits;
+
+ public PrimaryKeyScoredResult(
+ long snapshotId,
+ List<DataSplit> sourceSplits,
+ List<PrimaryKeySearchPosition> positions) {
+ this.snapshotId = snapshotId;
+ this.positions = Collections.unmodifiableList(new
ArrayList<>(positions));
+ this.splits =
+ Collections.unmodifiableList(
+ materializeSplits(snapshotId, sourceSplits,
positions));
}
@Override
public long snapshotId() {
- return plan.snapshotId();
+ return snapshotId;
+ }
+
+ public List<PrimaryKeySearchPosition> positions() {
+ return positions;
}
@Override
public List<IndexedSplit> splits() {
- Map<FileKey, BucketVectorSearchSplit> sourceSplits = sourceSplits();
- Map<FileKey, TreeMap<Integer, Float>> selectedByFile = new
LinkedHashMap<>();
- for (PrimaryKeyVectorRead.Candidate candidate : candidates) {
- FileKey key =
- new FileKey(
- candidate.partition(), candidate.bucket(),
candidate.dataFileName());
- BucketVectorSearchSplit bucketSplit = sourceSplits.get(key);
+ return splits;
+ }
+
+ @Override
+ public RoaringNavigableMap64 results() {
+ throw new UnsupportedOperationException(
+ "Primary-key scored results use physical file positions, not
global row ids.");
+ }
+
+ @Override
+ public ScoreGetter scoreGetter() {
+ throw new UnsupportedOperationException(
+ "Primary-key scored results attach scores to physical file
positions, not global row ids.");
+ }
+
+ private static List<IndexedSplit> materializeSplits(
+ long snapshotId,
+ List<DataSplit> sourceSplits,
+ List<PrimaryKeySearchPosition> positions) {
+ Map<FileKey, SourceFile> sources = sourceFiles(snapshotId,
sourceSplits);
+ TreeMap<PrimaryKeySearchPosition, PrimaryKeySearchPosition>
orderedPositions =
+ new TreeMap<>();
+ for (PrimaryKeySearchPosition position : positions) {
checkArgument(
- bucketSplit != null,
- "Primary-key vector candidate references unknown data file
%s in bucket %s.",
- candidate.dataFileName(),
- candidate.bucket());
- DataFileMeta dataFile = dataFile(bucketSplit.dataSplit(),
candidate.dataFileName());
+ orderedPositions.put(position, position) == null,
+ "Primary-key search result contains duplicate physical
position %s.",
+ position);
+ }
+
+ Map<FileKey, TreeMap<Integer, Float>> selectedByFile = new
LinkedHashMap<>();
+ for (PrimaryKeySearchPosition position : orderedPositions.values()) {
+ FileKey key = FileKey.from(position);
+ SourceFile sourceFile = sources.get(key);
checkArgument(
- candidate.rowPosition() >= 0
- && candidate.rowPosition() < dataFile.rowCount()
- && candidate.rowPosition() <= Integer.MAX_VALUE,
- "Primary-key vector row position %s is outside data file
%s.",
- candidate.rowPosition(),
- candidate.dataFileName());
- TreeMap<Integer, Float> selected =
- selectedByFile.computeIfAbsent(key, ignored -> new
TreeMap<>());
+ sourceFile != null,
+ "Primary-key search position references unknown data file
%s in bucket %s.",
+ position.dataFileName(),
+ position.bucket());
checkArgument(
- selected.put((int) candidate.rowPosition(),
score(candidate.distance()))
- == null,
- "Primary-key vector candidate contains duplicate row
position %s for data file %s.",
- candidate.rowPosition(),
- candidate.dataFileName());
+ position.rowPosition() < sourceFile.dataFile.rowCount()
+ && position.rowPosition() <= Integer.MAX_VALUE,
+ "Primary-key search row position %s is outside data file
%s.",
+ position.rowPosition(),
+ position.dataFileName());
+ selectedByFile
+ .computeIfAbsent(key, ignored -> new TreeMap<>())
+ .put((int) position.rowPosition(), position.score());
}
List<IndexedSplit> result = new ArrayList<>(selectedByFile.size());
for (Map.Entry<FileKey, TreeMap<Integer, Float>> entry :
selectedByFile.entrySet()) {
- BucketVectorSearchSplit bucketSplit =
sourceSplits.get(entry.getKey());
- DataSplit source = bucketSplit.dataSplit();
- int fileIndex = fileIndex(source, entry.getKey().dataFileName);
+ SourceFile sourceFile = sources.get(entry.getKey());
+ DataSplit source = sourceFile.split;
DataSplit.Builder builder =
DataSplit.builder()
.withSnapshot(source.snapshotId())
@@ -101,40 +126,36 @@ public class PrimaryKeyVectorResult implements
GlobalIndexSplitResult {
.withBucket(source.bucket())
.withBucketPath(source.bucketPath())
.withTotalBuckets(source.totalBuckets())
- .withDataFiles(
-
Collections.singletonList(source.dataFiles().get(fileIndex)))
+
.withDataFiles(Collections.singletonList(sourceFile.dataFile))
.isStreaming(false)
.rawConvertible(false);
if (source.deletionFiles().isPresent()) {
builder.withDataDeletionFiles(
-
Collections.singletonList(source.deletionFiles().get().get(fileIndex)));
+ Collections.singletonList(
+
source.deletionFiles().get().get(sourceFile.fileIndex)));
}
result.add(
new IndexedSplit(
builder.build(), ranges(entry.getValue()),
scores(entry.getValue())));
}
- return Collections.unmodifiableList(result);
- }
-
- @Override
- public RoaringNavigableMap64 results() {
- throw new UnsupportedOperationException(
- "Primary-key vector results use physical file positions, not
global row ids.");
+ return result;
}
- private Map<FileKey, BucketVectorSearchSplit> sourceSplits() {
- Map<FileKey, BucketVectorSearchSplit> result = new HashMap<>();
- for (VectorSearchSplit searchSplit : plan.splits()) {
- BucketVectorSearchSplit bucketSplit = (BucketVectorSearchSplit)
searchSplit;
- for (DataFileMeta dataFile : bucketSplit.dataSplit().dataFiles()) {
- FileKey key =
- new FileKey(
- bucketSplit.dataSplit().partition(),
- bucketSplit.dataSplit().bucket(),
- dataFile.fileName());
+ private static Map<FileKey, SourceFile> sourceFiles(
+ long snapshotId, List<DataSplit> sourceSplits) {
+ Map<FileKey, SourceFile> result = new HashMap<>();
+ for (DataSplit source : sourceSplits) {
+ checkArgument(
+ source.snapshotId() == snapshotId,
+ "Primary-key source split snapshot %s does not match
result snapshot %s.",
+ source.snapshotId(),
+ snapshotId);
+ for (int i = 0; i < source.dataFiles().size(); i++) {
+ DataFileMeta dataFile = source.dataFiles().get(i);
+ FileKey key = new FileKey(source.partition(), source.bucket(),
dataFile.fileName());
checkArgument(
- result.put(key, bucketSplit) == null,
- "Data file %s appears more than once in primary-key
vector plan.",
+ result.put(key, new SourceFile(source, dataFile, i))
== null,
+ "Data file %s appears more than once in primary-key
search sources.",
dataFile.fileName());
}
}
@@ -169,29 +190,17 @@ public class PrimaryKeyVectorResult implements
GlobalIndexSplitResult {
return result;
}
- private static DataFileMeta dataFile(DataSplit split, String dataFileName)
{
- return split.dataFiles().get(fileIndex(split, dataFileName));
- }
+ private static class SourceFile {
- private static int fileIndex(DataSplit split, String dataFileName) {
- for (int i = 0; i < split.dataFiles().size(); i++) {
- if (dataFileName.equals(split.dataFiles().get(i).fileName())) {
- return i;
- }
- }
- throw new IllegalArgumentException(
- "Data file " + dataFileName + " does not exist in vector
bucket split.");
- }
+ private final DataSplit split;
+ private final DataFileMeta dataFile;
+ private final int fileIndex;
- private float score(float distance) {
- if ("l2".equals(metric)) {
- return 1F / (1F + distance);
- } else if ("cosine".equals(metric)) {
- return 1F - distance;
- } else if ("inner_product".equals(metric)) {
- return -distance;
+ private SourceFile(DataSplit split, DataFileMeta dataFile, int
fileIndex) {
+ this.split = split;
+ this.dataFile = dataFile;
+ this.fileIndex = fileIndex;
}
- throw new IllegalArgumentException("Unsupported primary-key vector
metric: " + metric);
}
private static class FileKey {
@@ -206,12 +215,16 @@ public class PrimaryKeyVectorResult implements
GlobalIndexSplitResult {
this.dataFileName = dataFileName;
}
+ private static FileKey from(PrimaryKeySearchPosition position) {
+ return new FileKey(position.partition(), position.bucket(),
position.dataFileName());
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (o == null || getClass() != o.getClass()) {
+ if (!(o instanceof FileKey)) {
return false;
}
FileKey fileKey = (FileKey) o;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchPosition.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchPosition.java
new file mode 100644
index 0000000000..f2d7e09705
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchPosition.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** A scored physical row position in a primary-key table snapshot. */
+public final class PrimaryKeySearchPosition
+ implements Comparable<PrimaryKeySearchPosition>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final BinaryRow partition;
+ private final int bucket;
+ private final String dataFileName;
+ private final long rowPosition;
+ private final float score;
+
+ public PrimaryKeySearchPosition(
+ BinaryRow partition, int bucket, String dataFileName, long
rowPosition, float score) {
+ checkArgument(rowPosition >= 0, "Row position must not be negative:
%s.", rowPosition);
+ checkArgument(
+ !Float.isNaN(score) && !Float.isInfinite(score),
+ "Search score must be finite: %s.",
+ score);
+ this.partition = Objects.requireNonNull(partition, "partition").copy();
+ this.bucket = bucket;
+ this.dataFileName = Objects.requireNonNull(dataFileName,
"dataFileName");
+ this.rowPosition = rowPosition;
+ this.score = score;
+ }
+
+ public BinaryRow partition() {
+ return partition;
+ }
+
+ public int bucket() {
+ return bucket;
+ }
+
+ public String dataFileName() {
+ return dataFileName;
+ }
+
+ public long rowPosition() {
+ return rowPosition;
+ }
+
+ public float score() {
+ return score;
+ }
+
+ public PrimaryKeySearchPosition withScore(float newScore) {
+ return new PrimaryKeySearchPosition(partition, bucket, dataFileName,
rowPosition, newScore);
+ }
+
+ @Override
+ public int compareTo(PrimaryKeySearchPosition other) {
+ int comparison = compareBytes(partition.toBytes(),
other.partition.toBytes());
+ if (comparison != 0) {
+ return comparison;
+ }
+ comparison = Integer.compare(bucket, other.bucket);
+ if (comparison != 0) {
+ return comparison;
+ }
+ comparison = dataFileName.compareTo(other.dataFileName);
+ return comparison != 0 ? comparison : Long.compare(rowPosition,
other.rowPosition);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof PrimaryKeySearchPosition)) {
+ return false;
+ }
+ PrimaryKeySearchPosition that = (PrimaryKeySearchPosition) o;
+ return bucket == that.bucket
+ && rowPosition == that.rowPosition
+ && partition.equals(that.partition)
+ && dataFileName.equals(that.dataFileName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partition, bucket, dataFileName, rowPosition);
+ }
+
+ @Override
+ public String toString() {
+ return "PrimaryKeySearchPosition{"
+ + "partition="
+ + partition
+ + ", bucket="
+ + bucket
+ + ", dataFileName='"
+ + dataFileName
+ + '\''
+ + ", rowPosition="
+ + rowPosition
+ + ", score="
+ + score
+ + '}';
+ }
+
+ private static int compareBytes(byte[] left, byte[] right) {
+ int count = Math.min(left.length, right.length);
+ for (int i = 0; i < count; i++) {
+ int comparison = Integer.compare(left[i] & 0xFF, right[i] & 0xFF);
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(left.length, right.length);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
new file mode 100644
index 0000000000..779a64a400
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.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.table.source;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Deterministic reciprocal-rank fusion for physical primary-key search
positions. */
+public final class PrimaryKeySearchRanker {
+
+ public static final int DEFAULT_RRF_K = 60;
+
+ private static final Comparator<PrimaryKeySearchPosition> LOCAL_BEST_FIRST
=
+ (left, right) -> {
+ int score = Float.compare(right.score(), left.score());
+ return score != 0 ? score : left.compareTo(right);
+ };
+
+ private PrimaryKeySearchRanker() {}
+
+ public static List<PrimaryKeySearchPosition> rrf(
+ List<List<PrimaryKeySearchPosition>> rankings, int limit) {
+ List<Ranking> weighted = new ArrayList<>(rankings.size());
+ for (List<PrimaryKeySearchPosition> ranking : rankings) {
+ weighted.add(new Ranking(ranking, 1D));
+ }
+ return weightedRrf(weighted, limit);
+ }
+
+ public static List<PrimaryKeySearchPosition> weightedRrf(List<Ranking>
rankings, int limit) {
+ checkArgument(limit > 0, "RRF result limit must be positive: %s.",
limit);
+ Map<PrimaryKeySearchPosition, Double> fusedScores = new HashMap<>();
+ for (Ranking ranking : rankings) {
+ addRanking(fusedScores, ranking);
+ }
+
+ return topK(fusedScores, limit);
+ }
+
+ /** Fuses heterogeneous route scores after independently normalizing each
route to [0, 1]. */
+ public static List<PrimaryKeySearchPosition> weightedScore(List<Ranking>
rankings, int limit) {
+ checkArgument(limit > 0, "Weighted-score result limit must be
positive: %s.", limit);
+ Map<PrimaryKeySearchPosition, Double> fusedScores = new HashMap<>();
+ for (Ranking ranking : rankings) {
+ Set<PrimaryKeySearchPosition> unique = new HashSet<>();
+ float min = Float.POSITIVE_INFINITY;
+ float max = Float.NEGATIVE_INFINITY;
+ for (PrimaryKeySearchPosition position : ranking.positions) {
+ checkArgument(
+ unique.add(position),
+ "One weighted-score ranking contains duplicate
physical position %s.",
+ position);
+ min = Math.min(min, position.score());
+ max = Math.max(max, position.score());
+ }
+ float range = max - min;
+ for (PrimaryKeySearchPosition position : ranking.positions) {
+ double normalized = range > 0F ? (position.score() - min) /
range : 1D;
+ fusedScores.merge(position, ranking.weight * normalized,
Double::sum);
+ }
+ }
+ return topK(fusedScores, limit);
+ }
+
+ /** Fuses routes using weighted reciprocal rank without the RRF smoothing
constant. */
+ public static List<PrimaryKeySearchPosition> weightedMrr(List<Ranking>
rankings, int limit) {
+ checkArgument(limit > 0, "MRR result limit must be positive: %s.",
limit);
+ Map<PrimaryKeySearchPosition, Double> fusedScores = new HashMap<>();
+ for (Ranking ranking : rankings) {
+ List<PrimaryKeySearchPosition> sorted = new
ArrayList<>(ranking.positions);
+ sorted.sort(LOCAL_BEST_FIRST);
+ Set<PrimaryKeySearchPosition> unique = new HashSet<>();
+ for (int i = 0; i < sorted.size(); i++) {
+ PrimaryKeySearchPosition position = sorted.get(i);
+ checkArgument(
+ unique.add(position),
+ "One MRR ranking contains duplicate physical position
%s.",
+ position);
+ fusedScores.merge(position, ranking.weight / (i + 1D),
Double::sum);
+ }
+ }
+ return topK(fusedScores, limit);
+ }
+
+ private static List<PrimaryKeySearchPosition> topK(
+ Map<PrimaryKeySearchPosition, Double> fusedScores, int limit) {
+
+ Comparator<PrimaryKeySearchPosition> bestFirst =
+ (left, right) -> {
+ int score = Float.compare(right.score(), left.score());
+ return score != 0 ? score : left.compareTo(right);
+ };
+ PriorityQueue<PrimaryKeySearchPosition> topK =
+ new PriorityQueue<>(limit, bestFirst.reversed());
+ for (Map.Entry<PrimaryKeySearchPosition, Double> entry :
fusedScores.entrySet()) {
+ PrimaryKeySearchPosition position =
+ entry.getKey().withScore(entry.getValue().floatValue());
+ if (topK.size() < limit) {
+ topK.add(position);
+ } else if (bestFirst.compare(position, topK.peek()) < 0) {
+ topK.poll();
+ topK.add(position);
+ }
+ }
+ List<PrimaryKeySearchPosition> result = new ArrayList<>(topK);
+ result.sort(bestFirst);
+ return Collections.unmodifiableList(result);
+ }
+
+ private static void addRanking(
+ Map<PrimaryKeySearchPosition, Double> fusedScores, Ranking
ranking) {
+ List<PrimaryKeySearchPosition> sorted = new
ArrayList<>(ranking.positions);
+ sorted.sort(LOCAL_BEST_FIRST);
+ Set<PrimaryKeySearchPosition> unique = new HashSet<>();
+ int rank = 0;
+ float previousScore = Float.NaN;
+ for (int i = 0; i < sorted.size(); i++) {
+ PrimaryKeySearchPosition position = sorted.get(i);
+ checkArgument(
+ unique.add(position),
+ "One RRF ranking contains duplicate physical position %s.",
+ position);
+ if (i == 0 || Float.compare(position.score(), previousScore) != 0)
{
+ rank = i + 1;
+ previousScore = position.score();
+ }
+ double contribution = ranking.weight / (DEFAULT_RRF_K + rank);
+ fusedScores.merge(position, contribution, Double::sum);
+ }
+ }
+
+ /** One locally scored ranking and its optional route weight. */
+ public static class Ranking implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<PrimaryKeySearchPosition> positions;
+ private final double weight;
+
+ public Ranking(List<PrimaryKeySearchPosition> positions, double
weight) {
+ checkArgument(
+ weight > 0D && !Double.isNaN(weight) &&
!Double.isInfinite(weight),
+ "Search route weight must be finite and positive: %s.",
+ weight);
+ this.positions = Collections.unmodifiableList(new
ArrayList<>(positions));
+ this.weight = weight;
+ }
+
+ public List<PrimaryKeySearchPosition> positions() {
+ return positions;
+ }
+
+ public double weight() {
+ return weight;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
index 52f7423a92..de6eb0785f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReader.java
@@ -19,88 +19,18 @@
package org.apache.paimon.table.source;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.reader.FileRecordIterator;
import org.apache.paimon.reader.FileRecordReader;
-import org.apache.paimon.reader.ScoreRecordIterator;
-import org.apache.paimon.reader.ScoreRecordReader;
import org.apache.paimon.utils.RoaringBitmap32;
-import javax.annotation.Nullable;
-
-import java.io.IOException;
import java.util.function.IntFunction;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
-/** Reads selected physical file positions and exposes their vector-search
scores. */
-public class PrimaryKeyVectorPositionReader implements
ScoreRecordReader<InternalRow> {
-
- private final FileRecordReader<InternalRow> reader;
- private final RoaringBitmap32 rowPositions;
- private final IntFunction<Float> scoreGetter;
- private final int lastPosition;
- private boolean exhausted;
+/** Compatibility adapter for the generic primary-key index position reader. */
+public class PrimaryKeyVectorPositionReader extends
PrimaryKeyIndexPositionReader {
public PrimaryKeyVectorPositionReader(
FileRecordReader<InternalRow> reader,
RoaringBitmap32 rowPositions,
IntFunction<Float> scoreGetter) {
- checkArgument(!rowPositions.isEmpty(), "Selected row positions must
not be empty.");
- this.reader = reader;
- this.rowPositions = rowPositions.clone();
- this.scoreGetter = scoreGetter;
- this.lastPosition = rowPositions.last();
- }
-
- @Nullable
- @Override
- public ScoreRecordIterator<InternalRow> readBatch() throws IOException {
- if (exhausted) {
- return null;
- }
- FileRecordIterator<InternalRow> batch = reader.readBatch();
- if (batch == null) {
- return null;
- }
- FileRecordIterator<InternalRow> selected =
batch.selection(rowPositions);
- return new ScoreRecordIterator<InternalRow>() {
-
- private long returnedPosition = -1;
- private float returnedScore = Float.NaN;
-
- @Override
- public float returnedScore() {
- return returnedScore;
- }
-
- @Override
- public long returnedRowId() {
- return returnedPosition;
- }
-
- @Nullable
- @Override
- public InternalRow next() throws IOException {
- InternalRow row = selected.next();
- if (row != null) {
- returnedPosition = selected.returnedPosition();
- returnedScore = scoreGetter.apply((int) returnedPosition);
- if (returnedPosition >= lastPosition) {
- exhausted = true;
- }
- }
- return row;
- }
-
- @Override
- public void releaseBatch() {
- selected.releaseBatch();
- }
- };
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
+ super(reader, rowPositions, scoreGetter);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
index 3eba24b30c..c86fbf8226 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorResult.java
@@ -18,172 +18,65 @@
package org.apache.paimon.table.source;
-import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.IndexedSplit;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.TreeMap;
import static org.apache.paimon.globalindex.VectorSearchMetric.normalize;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Snapshot-scoped vector result addressed by physical primary-key table row
positions. */
+/** Compatibility result adapting primary-key vector candidates to generic
physical positions. */
public class PrimaryKeyVectorResult implements GlobalIndexSplitResult {
- private final PrimaryKeyVectorScan.Plan plan;
- private final List<PrimaryKeyVectorRead.Candidate> candidates;
- private final String metric;
+ private final PrimaryKeyScoredResult delegate;
PrimaryKeyVectorResult(
PrimaryKeyVectorScan.Plan plan,
List<PrimaryKeyVectorRead.Candidate> candidates,
String metric) {
- this.plan = plan;
- this.candidates = Collections.unmodifiableList(new
ArrayList<>(candidates));
- this.metric = normalize(metric);
- }
-
- @Override
- public long snapshotId() {
- return plan.snapshotId();
- }
-
- @Override
- public List<IndexedSplit> splits() {
- Map<FileKey, BucketVectorSearchSplit> sourceSplits = sourceSplits();
- Map<FileKey, TreeMap<Integer, Float>> selectedByFile = new
LinkedHashMap<>();
- for (PrimaryKeyVectorRead.Candidate candidate : candidates) {
- FileKey key =
- new FileKey(
- candidate.partition(), candidate.bucket(),
candidate.dataFileName());
- BucketVectorSearchSplit bucketSplit = sourceSplits.get(key);
- checkArgument(
- bucketSplit != null,
- "Primary-key vector candidate references unknown data file
%s in bucket %s.",
- candidate.dataFileName(),
- candidate.bucket());
- DataFileMeta dataFile = dataFile(bucketSplit.dataSplit(),
candidate.dataFileName());
- checkArgument(
- candidate.rowPosition() >= 0
- && candidate.rowPosition() < dataFile.rowCount()
- && candidate.rowPosition() <= Integer.MAX_VALUE,
- "Primary-key vector row position %s is outside data file
%s.",
- candidate.rowPosition(),
- candidate.dataFileName());
- TreeMap<Integer, Float> selected =
- selectedByFile.computeIfAbsent(key, ignored -> new
TreeMap<>());
- checkArgument(
- selected.put((int) candidate.rowPosition(),
score(candidate.distance()))
- == null,
- "Primary-key vector candidate contains duplicate row
position %s for data file %s.",
- candidate.rowPosition(),
- candidate.dataFileName());
+ String normalizedMetric = normalize(metric);
+ List<DataSplit> sourceSplits = new ArrayList<>(plan.splits().size());
+ for (VectorSearchSplit split : plan.splits()) {
+ sourceSplits.add(((BucketVectorSearchSplit) split).dataSplit());
}
-
- List<IndexedSplit> result = new ArrayList<>(selectedByFile.size());
- for (Map.Entry<FileKey, TreeMap<Integer, Float>> entry :
selectedByFile.entrySet()) {
- BucketVectorSearchSplit bucketSplit =
sourceSplits.get(entry.getKey());
- DataSplit source = bucketSplit.dataSplit();
- int fileIndex = fileIndex(source, entry.getKey().dataFileName);
- DataSplit.Builder builder =
- DataSplit.builder()
- .withSnapshot(source.snapshotId())
- .withPartition(source.partition())
- .withBucket(source.bucket())
- .withBucketPath(source.bucketPath())
- .withTotalBuckets(source.totalBuckets())
- .withDataFiles(
-
Collections.singletonList(source.dataFiles().get(fileIndex)))
- .isStreaming(false)
- .rawConvertible(false);
- if (source.deletionFiles().isPresent()) {
- builder.withDataDeletionFiles(
-
Collections.singletonList(source.deletionFiles().get().get(fileIndex)));
- }
- result.add(
- new IndexedSplit(
- builder.build(), ranges(entry.getValue()),
scores(entry.getValue())));
+ List<PrimaryKeySearchPosition> positions = new
ArrayList<>(candidates.size());
+ for (PrimaryKeyVectorRead.Candidate candidate : candidates) {
+ positions.add(
+ new PrimaryKeySearchPosition(
+ candidate.partition(),
+ candidate.bucket(),
+ candidate.dataFileName(),
+ candidate.rowPosition(),
+ score(normalizedMetric, candidate.distance())));
}
- return Collections.unmodifiableList(result);
+ this.delegate = new PrimaryKeyScoredResult(plan.snapshotId(),
sourceSplits, positions);
}
@Override
- public RoaringNavigableMap64 results() {
- throw new UnsupportedOperationException(
- "Primary-key vector results use physical file positions, not
global row ids.");
+ public long snapshotId() {
+ return delegate.snapshotId();
}
- private Map<FileKey, BucketVectorSearchSplit> sourceSplits() {
- Map<FileKey, BucketVectorSearchSplit> result = new HashMap<>();
- for (VectorSearchSplit searchSplit : plan.splits()) {
- BucketVectorSearchSplit bucketSplit = (BucketVectorSearchSplit)
searchSplit;
- for (DataFileMeta dataFile : bucketSplit.dataSplit().dataFiles()) {
- FileKey key =
- new FileKey(
- bucketSplit.dataSplit().partition(),
- bucketSplit.dataSplit().bucket(),
- dataFile.fileName());
- checkArgument(
- result.put(key, bucketSplit) == null,
- "Data file %s appears more than once in primary-key
vector plan.",
- dataFile.fileName());
- }
- }
- return result;
+ public List<PrimaryKeySearchPosition> positions() {
+ return delegate.positions();
}
- private static List<Range> ranges(TreeMap<Integer, Float> selected) {
- List<Range> result = new ArrayList<>();
- long from = -1;
- long to = -1;
- for (int position : selected.keySet()) {
- if (from < 0) {
- from = position;
- } else if (position != to + 1) {
- result.add(new Range(from, to));
- from = position;
- }
- to = position;
- }
- if (from >= 0) {
- result.add(new Range(from, to));
- }
- return result;
+ PrimaryKeyScoredResult scoredResult() {
+ return delegate;
}
- private static float[] scores(TreeMap<Integer, Float> selected) {
- float[] result = new float[selected.size()];
- int index = 0;
- for (float score : selected.values()) {
- result[index++] = score;
- }
- return result;
- }
-
- private static DataFileMeta dataFile(DataSplit split, String dataFileName)
{
- return split.dataFiles().get(fileIndex(split, dataFileName));
+ @Override
+ public List<IndexedSplit> splits() {
+ return delegate.splits();
}
- private static int fileIndex(DataSplit split, String dataFileName) {
- for (int i = 0; i < split.dataFiles().size(); i++) {
- if (dataFileName.equals(split.dataFiles().get(i).fileName())) {
- return i;
- }
- }
- throw new IllegalArgumentException(
- "Data file " + dataFileName + " does not exist in vector
bucket split.");
+ @Override
+ public RoaringNavigableMap64 results() {
+ return delegate.results();
}
- private float score(float distance) {
+ private static float score(String metric, float distance) {
if ("l2".equals(metric)) {
return 1F / (1F + distance);
} else if ("cosine".equals(metric)) {
@@ -193,36 +86,4 @@ public class PrimaryKeyVectorResult implements
GlobalIndexSplitResult {
}
throw new IllegalArgumentException("Unsupported primary-key vector
metric: " + metric);
}
-
- private static class FileKey {
-
- private final BinaryRow partition;
- private final int bucket;
- private final String dataFileName;
-
- private FileKey(BinaryRow partition, int bucket, String dataFileName) {
- this.partition = partition.copy();
- this.bucket = bucket;
- this.dataFileName = dataFileName;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- FileKey fileKey = (FileKey) o;
- return bucket == fileKey.bucket
- && partition.equals(fileKey.partition)
- && dataFileName.equals(fileKey.dataFileName);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(partition, bucket, dataFileName);
- }
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorScan.java
index e4bcb15d3c..cc3ef15874 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorScan.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.BinaryRow;
import org.apache.paimon.globalindex.IndexedSplit;
@@ -53,13 +54,14 @@ public class PrimaryKeyVectorScan implements VectorScan {
private final String indexType;
@Nullable private final PartitionPredicate partitionFilter;
@Nullable private final Predicate filter;
+ @Nullable private final Snapshot pinnedSnapshot;
public PrimaryKeyVectorScan(
FileStoreTable table,
int vectorFieldId,
String indexType,
@Nullable PartitionPredicate partitionFilter) {
- this(table, vectorFieldId, indexType, partitionFilter, null);
+ this(table, vectorFieldId, indexType, partitionFilter, null, null);
}
public PrimaryKeyVectorScan(
@@ -68,11 +70,22 @@ public class PrimaryKeyVectorScan implements VectorScan {
String indexType,
@Nullable PartitionPredicate partitionFilter,
@Nullable Predicate filter) {
+ this(table, vectorFieldId, indexType, partitionFilter, filter, null);
+ }
+
+ PrimaryKeyVectorScan(
+ FileStoreTable table,
+ int vectorFieldId,
+ String indexType,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
+ @Nullable Snapshot pinnedSnapshot) {
this.table = table;
this.vectorFieldId = vectorFieldId;
this.indexType = indexType;
this.partitionFilter = partitionFilter;
this.filter = filter;
+ this.pinnedSnapshot = pinnedSnapshot;
}
@Override
@@ -85,8 +98,9 @@ public class PrimaryKeyVectorScan implements VectorScan {
|| (table.coreOptions().deletionVectorsEnabled()
&&
!table.coreOptions().deletionVectorsMergeOnRead()),
"Primary-key vector pre-filter requires deletion vectors
without merge-on-read.");
- SnapshotReader snapshotReader = table.newSnapshotReader().keepStats();
- DataTableScan dataScan = table.newScan(ignored -> snapshotReader);
+ FileStoreTable scanTable = scanTable();
+ SnapshotReader snapshotReader =
scanTable.newSnapshotReader().keepStats();
+ DataTableScan dataScan = scanTable.newScan(ignored -> snapshotReader);
checkArgument(
dataScan instanceof PrimaryKeyBatchScan,
"Primary-key vector search requires a primary-key batch
scan.");
@@ -108,8 +122,16 @@ public class PrimaryKeyVectorScan implements VectorScan {
if (snapshotPlan.snapshotId() == null) {
return new Plan(0, Collections.emptyList());
}
- Snapshot snapshot =
snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId());
+ Snapshot snapshot =
+ pinnedSnapshot == null
+ ?
snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId())
+ : pinnedSnapshot;
checkArgument(snapshot != null, "Primary-key vector snapshot does not
exist.");
+ checkArgument(
+ snapshot.id() == snapshotPlan.snapshotId(),
+ "Primary-key vector plan snapshot %s does not match pinned
snapshot %s.",
+ snapshotPlan.snapshotId(),
+ snapshot.id());
IndexFileHandler indexFileHandler = snapshotReader.indexFileHandler();
checkArgument(indexFileHandler != null, "Primary-key vector index
handler is unavailable.");
@@ -127,6 +149,17 @@ public class PrimaryKeyVectorScan implements VectorScan {
return plan(snapshot.id(), snapshotPlan.splits(), vectorIndexEntries);
}
+ private FileStoreTable scanTable() {
+ if (pinnedSnapshot == null) {
+ return table;
+ }
+ return (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(pinnedSnapshot.id())));
+ }
+
static Plan plan(
long snapshotId,
List<? extends Split> dataSplits,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
index e37f4e9d43..4b59d43812 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -26,6 +27,8 @@ import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Pair;
+import javax.annotation.Nullable;
+
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -48,6 +51,7 @@ public class VectorSearchBuilderImpl implements
VectorSearchBuilder {
protected DataField vectorColumn;
protected float[] vector;
protected Map<String, String> options = new HashMap<>();
+ @Nullable private Snapshot pinnedSnapshot;
public VectorSearchBuilderImpl(InnerTable table) {
this.table = (FileStoreTable) table;
@@ -130,7 +134,8 @@ public class VectorSearchBuilderImpl implements
VectorSearchBuilder {
vectorColumn.id(),
table.coreOptions().primaryKeyVectorIndexType(vectorColumn.name()),
partitionFilter,
- filter);
+ filter,
+ pinnedSnapshot);
}
return new DataEvolutionVectorScan(table, partitionFilter, filter,
vectorColumn, options);
}
@@ -149,4 +154,9 @@ public class VectorSearchBuilderImpl implements
VectorSearchBuilder {
return vectorColumn != null
&&
table.coreOptions().primaryKeyVectorIndexColumns().contains(vectorColumn.name());
}
+
+ VectorSearchBuilderImpl withSnapshot(Snapshot snapshot) {
+ this.pinnedSnapshot = snapshot;
+ return this;
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexStateTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexStateTest.java
new file mode 100644
index 0000000000..a2a988c2a9
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextBucketIndexStateTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests restored file-aligned primary-key full-text state. */
+class PkFullTextBucketIndexStateTest {
+
+ @Test
+ void testClassifiesMatchingFieldAndRetiresOtherFields() {
+ IndexFileMeta current = payload("current", "data", 7);
+ IndexFileMeta alsoCurrent = payload("also-current", "other-data", 7);
+ IndexFileMeta otherField = payload("other", "third-data", 8);
+
+ PkFullTextBucketIndexState state =
+ PkFullTextBucketIndexState.fromActivePayloads(
+ 7, Arrays.asList(current, alsoCurrent, otherField));
+
+ assertThat(state.currentPayloads()).containsExactly(current,
alsoCurrent);
+ assertThat(state.stalePayloads()).containsExactly(otherField);
+ assertThat(state.payloadBySourceFile()).containsEntry("data", current);
+ }
+
+ @Test
+ void testRejectsDuplicateCurrentCoverage() {
+ assertThatThrownBy(
+ () ->
+ PkFullTextBucketIndexState.fromActivePayloads(
+ 7,
+ Arrays.asList(
+ payload("first", "data", 7),
+ payload("second", "data", 7))))
+ .hasMessageContaining("data")
+ .hasMessageContaining("covered by both full-text archives");
+ }
+
+ @Test
+ void testMapsEverySourceOfMultiSourceArchive() {
+ PrimaryKeyIndexSourceMeta sourceMeta =
+ new PrimaryKeyIndexSourceMeta(
+ Arrays.asList(
+ new PrimaryKeyIndexSourceFile("data-1", 1),
+ new PrimaryKeyIndexSourceFile("data-2", 1)));
+ IndexFileMeta payload = payload("multi", 7, 2, sourceMeta);
+
+ PkFullTextBucketIndexState state =
+ PkFullTextBucketIndexState.fromActivePayloads(
+ 7, Collections.singletonList(payload));
+
+ assertThat(state.currentPayloads()).containsExactly(payload);
+ assertThat(state.payloadBySourceFile())
+ .containsEntry("data-1", payload)
+ .containsEntry("data-2", payload);
+ }
+
+ private static IndexFileMeta payload(String payloadName, String
sourceName, int fieldId) {
+ return payload(
+ payloadName,
+ fieldId,
+ 1,
+ new PrimaryKeyIndexSourceMeta(new
PrimaryKeyIndexSourceFile(sourceName, 1)));
+ }
+
+ private static IndexFileMeta payload(
+ String payloadName, int fieldId, long rowCount,
PrimaryKeyIndexSourceMeta sourceMeta) {
+ return new IndexFileMeta(
+ "full-text",
+ payloadName,
+ 100,
+ rowCount,
+ new GlobalIndexMeta(0, rowCount - 1, fieldId, null, null,
sourceMeta.serialize()),
+ null);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReaderTest.java
new file mode 100644
index 0000000000..47a6303302
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextDataFileReaderTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.format.FlushingFileFormat;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
+import org.apache.paimon.io.KeyValueFileWriterFactory;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.KeyValueFieldsExtractor;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.options.MemorySize.VALUE_128_MB;
+import static
org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests reading physical text positions from compact data files. */
+class PkFullTextDataFileReaderTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testReadsProjectedTextAndPreservesNullPositions() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ DataField keyField = new DataField(0, "id", DataTypes.INT());
+ DataField textField = new DataField(1, "content", DataTypes.STRING());
+ DataField payloadField = new DataField(2, "payload", DataTypes.INT());
+ RowType keyType = RowType.of(keyField);
+ RowType valueType = RowType.of(textField, payloadField);
+ List<DataField> fields = Arrays.asList(keyField, textField,
payloadField);
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ fields,
+ 2,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyMap(),
+ "");
+ SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
+ assertThat(schemaManager.commit(schema)).isTrue();
+ FileStorePathFactory pathFactory = createNonPartFactory(tablePath);
+ FlushingFileFormat format = new FlushingFileFormat("avro");
+ CoreOptions options = new CoreOptions(new Options());
+
+ KeyValueFileWriterFactory writerFactory =
+ KeyValueFileWriterFactory.builder(
+ fileIO,
+ schema.id(),
+ keyType,
+ valueType,
+ format,
+ ignored -> pathFactory,
+ VALUE_128_MB.getBytes())
+ .build(BinaryRow.EMPTY_ROW, 0, options);
+ RollingFileWriter<KeyValue, DataFileMeta> writer =
+ writerFactory.createRollingMergeTreeFileWriter(1,
FileSource.COMPACT);
+ try {
+ writer.write(keyValue(1, "first", 10));
+ writer.write(keyValue(2, null, 20));
+ writer.write(keyValue(3, "third", 30));
+ } finally {
+ writer.close();
+ }
+ DataFileMeta dataFile = writer.result().get(0);
+
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder =
+ KeyValueFileReaderFactory.builder(
+ fileIO,
+ schemaManager,
+ schema,
+ keyType,
+ valueType,
+ ignored -> format,
+ pathFactory,
+ fieldsExtractor(keyType, valueType),
+ options);
+ PkFullTextDataFileReader reader =
+ new PkFullTextDataFileReader.Factory(
+ readerFactoryBuilder, BinaryRow.EMPTY_ROW, 0,
textField)
+ .create(dataFile);
+ try {
+ assertThat(reader.rowCount()).isEqualTo(3);
+ assertThat(reader.readNextText()).hasToString("first");
+ assertThat(reader.readNextText()).isNull();
+ assertThat(reader.readNextText()).hasToString("third");
+
assertThatThrownBy(reader::readNextText).hasMessageContaining("Read past data
file");
+ } finally {
+ reader.close();
+ }
+ }
+
+ private static KeyValue keyValue(int key, String text, int payload) {
+ return new KeyValue()
+ .replace(
+ GenericRow.of(key),
+ RowKind.INSERT,
+ GenericRow.of(
+ text == null ? null :
BinaryString.fromString(text), payload));
+ }
+
+ private static KeyValueFieldsExtractor fieldsExtractor(RowType keyType,
RowType valueType) {
+ return new KeyValueFieldsExtractor() {
+ @Override
+ public List<DataField> keyFields(TableSchema schema) {
+ return keyType.getFields();
+ }
+
+ @Override
+ public List<DataField> valueFields(TableSchema schema) {
+ return valueType.getFields();
+ }
+ };
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
new file mode 100644
index 0000000000..f7b3d82690
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
@@ -0,0 +1,278 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests building one file-aligned primary-key full-text archive. */
+class PkFullTextIndexFileTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @Test
+ void testWritesEveryPhysicalOrdinalAndV1SourceMetadata() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ PkFullTextDataFileReader reader = mock(PkFullTextDataFileReader.class);
+ when(reader.rowCount()).thenReturn(3L);
+ when(reader.readNextText())
+ .thenReturn(
+ BinaryString.fromString("first"), null,
BinaryString.fromString("third"));
+ RecordingIndexer indexer = new RecordingIndexer();
+
+ IndexFileMeta archive =
+ new PkFullTextIndexFile(fileIO, pathFactory())
+ .build(
+ dataFile("data-1", 3),
+ reader,
+ new DataField(7, "content",
DataTypes.STRING()),
+ indexer);
+
+ assertThat(indexer.rowIds).containsExactly(0L, 1L, 2L);
+ assertThat(indexer.values)
+ .containsExactly(
+ BinaryString.fromString("first"), null,
BinaryString.fromString("third"));
+ assertThat(archive.indexType()).isEqualTo("full-text");
+ assertThat(archive.rowCount()).isEqualTo(3);
+ assertThat(archive.globalIndexMeta().rowRangeStart()).isZero();
+ assertThat(archive.globalIndexMeta().rowRangeEnd()).isEqualTo(2);
+ assertThat(archive.globalIndexMeta().indexFieldId()).isEqualTo(7);
+ DataInputDeserializer sourceInput =
+ new
DataInputDeserializer(archive.globalIndexMeta().sourceMeta());
+ assertThat(sourceInput.readInt()).isEqualTo(1);
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(archive);
+ assertThat(sourceMeta.sourceFile().fileName()).isEqualTo("data-1");
+ assertThat(sourceMeta.sourceFile().rowCount()).isEqualTo(3);
+ }
+
+ @Test
+ void testBuildsMultiSourceArchiveWithContinuousRowIds() throws Exception {
+ PkFullTextDataFileReader first = mock(PkFullTextDataFileReader.class);
+ when(first.rowCount()).thenReturn(2L);
+ when(first.readNextText())
+ .thenReturn(BinaryString.fromString("first"),
BinaryString.fromString("second"));
+ PkFullTextDataFileReader second = mock(PkFullTextDataFileReader.class);
+ when(second.rowCount()).thenReturn(1L);
+
when(second.readNextText()).thenReturn(BinaryString.fromString("third"));
+ RecordingIndexer indexer = new RecordingIndexer();
+
+ IndexFileMeta archive =
+ new PkFullTextIndexFile(LocalFileIO.create(), pathFactory())
+ .build(
+ Arrays.asList(
+ new PkFullTextIndexFile.Source(
+ dataFile("data-1", 2), first),
+ new PkFullTextIndexFile.Source(
+ dataFile("data-2", 1),
second)),
+ new DataField(7, "content",
DataTypes.STRING()),
+ indexer);
+
+ assertThat(indexer.rowIds).containsExactly(0L, 1L, 2L);
+ assertThat(indexer.values)
+ .containsExactly(
+ BinaryString.fromString("first"),
+ BinaryString.fromString("second"),
+ BinaryString.fromString("third"));
+ assertThat(archive.rowCount()).isEqualTo(3);
+ assertThat(archive.globalIndexMeta().rowRangeEnd()).isEqualTo(2);
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(archive);
+ assertThat(sourceMeta.sourceFiles())
+ .extracting(source -> source.fileName() + ":" +
source.rowCount())
+ .containsExactly("data-1:2", "data-2:1");
+ }
+
+ @Test
+ void testBuilderOpensTheFileReaderAndDelegatesEffectiveOptions() throws
Exception {
+ DataFileMeta source = dataFile("data-1", 3);
+ DataField textField = new DataField(7, "content", DataTypes.STRING());
+ Options options = new Options();
+ PkFullTextIndexFile indexFile = mock(PkFullTextIndexFile.class);
+ PkFullTextDataFileReader.Factory readerFactory =
+ mock(PkFullTextDataFileReader.Factory.class);
+ PkFullTextDataFileReader reader = mock(PkFullTextDataFileReader.class);
+ IndexFileMeta expected = mock(IndexFileMeta.class);
+ when(readerFactory.create(source)).thenReturn(reader);
+ when(indexFile.build(source, reader, textField,
options)).thenReturn(expected);
+
+ IndexFileMeta actual =
+ new PkFullTextIndexBuilder(indexFile, readerFactory,
textField, options)
+ .build(source);
+
+ assertThat(actual).isSameAs(expected);
+ verify(readerFactory).create(source);
+ verify(indexFile).build(source, reader, textField, options);
+ }
+
+ @Test
+ void testBuilderDelegatesOrderedSources() throws Exception {
+ DataFileMeta first = dataFile("data-1", 2);
+ DataFileMeta second = dataFile("data-2", 1);
+ DataField textField = new DataField(7, "content", DataTypes.STRING());
+ Options options = new Options();
+ PkFullTextIndexFile indexFile = mock(PkFullTextIndexFile.class);
+ PkFullTextDataFileReader.Factory readerFactory =
+ mock(PkFullTextDataFileReader.Factory.class);
+
when(readerFactory.create(first)).thenReturn(mock(PkFullTextDataFileReader.class));
+
when(readerFactory.create(second)).thenReturn(mock(PkFullTextDataFileReader.class));
+ IndexFileMeta expected = mock(IndexFileMeta.class);
+ when(indexFile.build(anyList(), same(textField),
same(options))).thenReturn(expected);
+
+ IndexFileMeta actual =
+ new PkFullTextIndexBuilder(indexFile, readerFactory,
textField, options)
+ .build(Arrays.asList(first, second));
+
+ assertThat(actual).isSameAs(expected);
+ verify(readerFactory).create(first);
+ verify(readerFactory).create(second);
+ verify(indexFile).build(anyList(), same(textField), same(options));
+ }
+
+ @Test
+ void testClosesReaderWhenIndexerCreationFails() throws Exception {
+ PkFullTextDataFileReader reader = mock(PkFullTextDataFileReader.class);
+
+ assertThatThrownBy(
+ () ->
+ new PkFullTextIndexFile(LocalFileIO.create(),
pathFactory())
+ .build(
+ dataFile("data-1", 1),
+ reader,
+ new DataField(7, "content",
DataTypes.STRING()),
+ new Options()))
+ .isInstanceOf(RuntimeException.class);
+
+ verify(reader).close();
+ }
+
+ private static DataFileMeta dataFile(String fileName, long rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 1,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private IndexPathFactory pathFactory() {
+ Path directory = new Path(tempPath.toUri());
+ return new IndexPathFactory() {
+ @Override
+ public Path toPath(String fileName) {
+ return new Path(directory, fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ return new Path(directory, UUID.randomUUID().toString());
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+ };
+ }
+
+ private static class RecordingIndexer implements GlobalIndexer {
+
+ private final List<Object> values = new ArrayList<>();
+ private final List<Long> rowIds = new ArrayList<>();
+
+ @Override
+ public GlobalIndexWriter createWriter(GlobalIndexFileWriter
fileWriter) {
+ return new GlobalIndexSingleColumnWriter() {
+ @Override
+ public void write(Object key, long relativeRowId) {
+ values.add(key);
+ rowIds.add(relativeRowId);
+ }
+
+ @Override
+ public List<ResultEntry> finish() {
+ try {
+ String fileName = fileWriter.newFileName("full-text");
+ try (PositionOutputStream out =
fileWriter.newOutputStream(fileName)) {
+ out.write(1);
+ }
+ return Collections.singletonList(
+ new ResultEntry(fileName, rowIds.size(), new
byte[] {2}));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ }
+
+ @Override
+ public GlobalIndexReader createReader(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ ExecutorService executor) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitReadTest.java
index 3d4ef2413f..29266ee7a0 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexedSplitReadTest.java
@@ -27,7 +27,7 @@ import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.PrimaryKeyVectorPositionReader;
+import org.apache.paimon.table.source.PrimaryKeyIndexPositionReader;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringBitmap32;
@@ -63,7 +63,7 @@ class PrimaryKeyIndexedSplitReadTest {
RecordReader<InternalRow> reader =
new PrimaryKeyIndexedSplitRead(rawRead).createReader(split);
- assertThat(reader).isInstanceOf(PrimaryKeyVectorPositionReader.class);
+ assertThat(reader).isInstanceOf(PrimaryKeyIndexPositionReader.class);
ArgumentCaptor<RoaringBitmap32> positions =
ArgumentCaptor.forClass(RoaringBitmap32.class);
verify(rawRead).createFileReader(eq(dataSplit), positions.capture());
assertThat(positions.getValue()).isEqualTo(RoaringBitmap32.bitmapOf(1,
3));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyScoredResultTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyScoredResultTest.java
new file mode 100644
index 0000000000..86669b56be
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyScoredResultTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for snapshot-scoped scored physical primary-key results. */
+class PrimaryKeyScoredResultTest {
+
+ @Test
+ void testMaterializesDeterministicFileSplitsAndAlignedScores() {
+ DataFileMeta fileB = dataFile("file-b");
+ DataFileMeta fileA = dataFile("file-a");
+ DeletionFile deletionB = new DeletionFile("dv-b", 10, 20, 1L);
+ DeletionFile deletionA = new DeletionFile("dv-a", 10, 20, 1L);
+ DataSplit source =
+ DataSplit.builder()
+ .withSnapshot(7)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(2)
+ .withBucketPath("bucket-2")
+ .withTotalBuckets(3)
+ .withDataFiles(Arrays.asList(fileB, fileA))
+ .withDataDeletionFiles(Arrays.asList(deletionB,
deletionA))
+ .build();
+ List<PrimaryKeySearchPosition> positions =
+ Arrays.asList(
+ new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 2,
"file-b", 4, 0.25F),
+ new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 2,
"file-a", 3, 0.5F),
+ new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 2,
"file-a", 1, 0.75F));
+
+ PrimaryKeyScoredResult result =
+ new PrimaryKeyScoredResult(7,
Collections.singletonList(source), positions);
+
+ assertThat(result.snapshotId()).isEqualTo(7);
+ assertThat(result.positions()).containsExactlyElementsOf(positions);
+ assertThat(result.splits()).hasSize(2);
+ IndexedSplit first = result.splits().get(0);
+ assertThat(first.dataSplit().dataFiles()).containsExactly(fileA);
+
assertThat(first.dataSplit().deletionFiles().get()).containsExactly(deletionA);
+ assertThat(first.rowRanges()).containsExactly(new Range(1, 1), new
Range(3, 3));
+ assertThat(first.scores()).containsExactly(0.75F, 0.5F);
+ IndexedSplit second = result.splits().get(1);
+ assertThat(second.dataSplit().dataFiles()).containsExactly(fileB);
+
assertThat(second.dataSplit().deletionFiles().get()).containsExactly(deletionB);
+ assertThat(second.rowRanges()).containsExactly(new Range(4, 4));
+ assertThat(second.scores()).containsExactly(0.25F);
+ }
+
+ private static DataFileMeta dataFile(String fileName) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ 5,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 1,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySearchRankerTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySearchRankerTest.java
new file mode 100644
index 0000000000..aef7fbf003
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySearchRankerTest.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for deterministic reciprocal-rank fusion of physical search
positions. */
+class PrimaryKeySearchRankerTest {
+
+ @Test
+ void testFusesDuplicatePositionsAndAssignsTheSameRankToLocalTies() {
+ PrimaryKeySearchPosition a = position("a", 0, 10F);
+ PrimaryKeySearchPosition b = position("b", 0, 10F);
+ PrimaryKeySearchPosition c = position("c", 0, 5F);
+ List<PrimaryKeySearchPosition> fused =
+ PrimaryKeySearchRanker.rrf(
+ Arrays.asList(
+ Arrays.asList(c, b, a),
+ Arrays.asList(a.withScore(1F),
c.withScore(8F))),
+ 3);
+
+ assertThat(fused)
+ .extracting(PrimaryKeySearchPosition::dataFileName)
+ .containsExactly("a", "c", "b");
+ assertThat(fused.get(0).score())
+ .isCloseTo((float) (1D / 61D + 1D / 62D), within(0.000001F));
+ assertThat(fused.get(1).score())
+ .isCloseTo((float) (1D / 63D + 1D / 61D), within(0.000001F));
+ assertThat(fused.get(2).score()).isCloseTo((float) (1D / 61D),
within(0.000001F));
+ }
+
+ @Test
+ void testUsesRouteWeightsAndDeterministicPhysicalTieBreaking() {
+ PrimaryKeySearchPosition a = position("a", 0, 1F);
+ PrimaryKeySearchPosition b = position("b", 0, 1F);
+ List<PrimaryKeySearchPosition> fused =
+ PrimaryKeySearchRanker.weightedRrf(
+ Arrays.asList(
+ new PrimaryKeySearchRanker.Ranking(
+ Collections.singletonList(a), 2D),
+ new PrimaryKeySearchRanker.Ranking(
+ Collections.singletonList(b), 2D)),
+ 1);
+
+ assertThat(fused).hasSize(1);
+ assertThat(fused.get(0).dataFileName()).isEqualTo("a");
+ assertThat(fused.get(0).score()).isCloseTo((float) (2D / 61D),
within(0.000001F));
+ }
+
+ @Test
+ void testWeightedScoreNormalizesEachPhysicalRoute() {
+ PrimaryKeySearchPosition a = position("a", 0, 0F);
+ PrimaryKeySearchPosition b = position("b", 0, 10F);
+ PrimaryKeySearchPosition c = position("c", 0, 0F);
+
+ List<PrimaryKeySearchPosition> fused =
+ PrimaryKeySearchRanker.weightedScore(
+ Arrays.asList(
+ new
PrimaryKeySearchRanker.Ranking(Arrays.asList(a, b), 1D),
+ new PrimaryKeySearchRanker.Ranking(
+ Arrays.asList(a.withScore(100F), c),
2D)),
+ 3);
+
+ assertThat(fused)
+ .extracting(PrimaryKeySearchPosition::dataFileName)
+ .containsExactly("a", "b", "c");
+
assertThat(fused).extracting(PrimaryKeySearchPosition::score).containsExactly(2F,
1F, 0F);
+ }
+
+ private static PrimaryKeySearchPosition position(
+ String dataFileName, long rowPosition, float score) {
+ return new PrimaryKeySearchPosition(
+ BinaryRow.EMPTY_ROW, 0, dataFileName, rowPosition, score);
+ }
+
+ private static org.assertj.core.data.Offset<Float> within(float value) {
+ return org.assertj.core.data.Offset.offset(value);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReaderTest.java
index c99f1d0c41..cd2f4bd65c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReaderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorPositionReaderTest.java
@@ -97,8 +97,7 @@ class PrimaryKeyVectorPositionReaderTest {
() -> rawRead,
mock(TableSchema.class));
- assertThat(tableRead.createReader(split))
- .isInstanceOf(PrimaryKeyVectorPositionReader.class);
+
assertThat(tableRead.createReader(split)).isInstanceOf(PrimaryKeyIndexPositionReader.class);
verify(rawRead, never()).createReader(split);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorResultTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorResultTest.java
index 849fd76ba7..8769fff1a7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorResultTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorResultTest.java
@@ -52,9 +52,13 @@ class PrimaryKeyVectorResultTest {
new PrimaryKeyVectorRead.Candidate(
BinaryRow.EMPTY_ROW, 0, "data-1", 2, 4F));
- List<IndexedSplit> splits = new PrimaryKeyVectorResult(plan,
candidates, "l2").splits();
+ PrimaryKeyVectorResult result = new PrimaryKeyVectorResult(plan,
candidates, "l2");
+ List<IndexedSplit> splits = result.splits();
assertThat(splits).hasSize(1);
+ assertThat(result.positions())
+ .extracting(PrimaryKeySearchPosition::score)
+ .containsExactly(1F / (1F + 0.1F), 1F / (1F + 1F), 1F / (1F +
4F));
IndexedSplit split = splits.get(0);
assertThat(split.dataSplit().snapshotId()).isEqualTo(11);
assertThat(split.dataSplit().dataFiles()).containsExactly(dataFile);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
index 5de2923507..64b1731fc4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyVectorScanTest.java
@@ -109,6 +109,8 @@ class PrimaryKeyVectorScanTest {
when(snapshot.id()).thenReturn(11L);
when(table.coreOptions()).thenReturn(coreOptions);
when(table.latestSnapshot()).thenReturn(Optional.of(snapshot));
+
when(table.copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
"11")))
+ .thenReturn(table);
SnapshotReader snapshotReader = mock(SnapshotReader.class,
RETURNS_SELF);
SnapshotReader.Plan snapshotPlan = mock(SnapshotReader.Plan.class,
CALLS_REAL_METHODS);
@@ -139,7 +141,8 @@ class PrimaryKeyVectorScanTest {
when(snapshotReader.indexFileHandler()).thenReturn(indexFileHandler);
configureBatchScan(table, snapshotReader, snapshot);
- PrimaryKeyVectorScan.Plan plan = new PrimaryKeyVectorScan(table, 7,
"ivf-pq", null).scan();
+ PrimaryKeyVectorScan.Plan plan =
+ new PrimaryKeyVectorScan(table, 7, "ivf-pq", null, null,
snapshot).scan();
assertThat(plan.snapshotId()).isEqualTo(11);
assertThat(plan.splits()).hasSize(1);
@@ -147,6 +150,7 @@ class PrimaryKeyVectorScanTest {
assertThat(bucketSplit.payloadFiles())
.extracting(IndexFileMeta::fileName)
.containsExactly("ann-match");
+
verify(table).copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
"11"));
}
@Test
