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 f87fe0fb07 [core] Generalize source metadata for primary-key indexes
(#8595)
f87fe0fb07 is described below
commit f87fe0fb07ee644cfcc7e83892cf577c79eb545e
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 13 18:00:22 2026 +0800
[core] Generalize source metadata for primary-key indexes (#8595)
Prepare the primary-key index source metadata for reuse by additional
source-backed index families, such as BTree and Bitmap, without changing
the existing vector-index behavior.
---
.../PrimaryKeyIndexSourceFile.java} | 10 +-
.../PrimaryKeyIndexSourceMeta.java} | 50 +++++----
.../PrimaryKeyIndexSourcePolicy.java} | 8 +-
.../pkvector/BucketedVectorIndexMaintainer.java | 25 +++--
.../paimon/index/pkvector/PkVectorAnnLevels.java | 12 ++-
.../index/pkvector/PkVectorAnnSegmentFile.java | 18 ++--
.../index/pkvector/PkVectorAnnSegmentSearcher.java | 25 +++--
.../index/pkvector/PkVectorBucketIndexState.java | 10 +-
.../pkvector/PrimaryKeyVectorBucketSearch.java | 6 +-
.../paimon/table/source/PrimaryKeyVectorRead.java | 4 +-
.../index/pk/PrimaryKeyIndexSourceMetaTest.java | 115 +++++++++++++++++++++
.../BucketedVectorIndexMaintainerTest.java | 16 +--
.../index/pkvector/PkVectorAnnLevelsTest.java | 6 +-
.../index/pkvector/PkVectorAnnSegmentFileTest.java | 10 +-
.../pkvector/PkVectorBucketIndexStateTest.java | 6 +-
.../index/pkvector/PkVectorSourceMetaTest.java | 56 ----------
.../pkvector/PrimaryKeyVectorBucketSearchTest.java | 10 +-
.../table/source/PrimaryKeyVectorScanTest.java | 9 +-
.../coordinator/TableWriteCoordinatorTest.java | 10 +-
19 files changed, 253 insertions(+), 153 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceFile.java
similarity index 84%
rename from
paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
rename to
paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceFile.java
index 87429d5694..701a31f652 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceFile.java
@@ -16,19 +16,19 @@
* limitations under the License.
*/
-package org.apache.paimon.index.pkvector;
+package org.apache.paimon.index.pk;
import java.util.Objects;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Immutable source data-file identity captured when a vector segment is
built. */
-public final class PkVectorSourceFile {
+/** Immutable identity of a data file covered by a source-backed primary-key
index. */
+public final class PrimaryKeyIndexSourceFile {
private final String fileName;
private final long rowCount;
- public PkVectorSourceFile(String fileName, long rowCount) {
+ public PrimaryKeyIndexSourceFile(String fileName, long rowCount) {
this.fileName = Objects.requireNonNull(fileName);
this.rowCount = rowCount;
checkArgument(rowCount >= 0, "Source file row count must not be
negative.");
@@ -50,7 +50,7 @@ public final class PkVectorSourceFile {
if (o == null || getClass() != o.getClass()) {
return false;
}
- PkVectorSourceFile that = (PkVectorSourceFile) o;
+ PrimaryKeyIndexSourceFile that = (PrimaryKeyIndexSourceFile) o;
return rowCount == that.rowCount && Objects.equals(fileName,
that.fileName);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMeta.java
similarity index 59%
rename from
paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
rename to
paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMeta.java
index d0a6a8171c..2bc0169a9d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourceMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMeta.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.index.pkvector;
+package org.apache.paimon.index.pk;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
@@ -30,27 +30,39 @@ import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Ordered source data files for a primary-key vector index payload. */
-public final class PkVectorSourceMeta {
+/** Ordered source data files covered by a source-backed primary-key index
payload. */
+public final class PrimaryKeyIndexSourceMeta {
private static final int VERSION = 1;
- private final List<PkVectorSourceFile> sourceFiles;
+ private final List<PrimaryKeyIndexSourceFile> sourceFiles;
- public PkVectorSourceMeta(List<PkVectorSourceFile> sourceFiles) {
+ public PrimaryKeyIndexSourceMeta(List<PrimaryKeyIndexSourceFile>
sourceFiles) {
this.sourceFiles = Collections.unmodifiableList(new
ArrayList<>(sourceFiles));
- checkArgument(!this.sourceFiles.isEmpty(), "A vector index must
reference source files.");
+ checkArgument(!this.sourceFiles.isEmpty(), "An index must reference
source files.");
}
- public List<PkVectorSourceFile> sourceFiles() {
+ public PrimaryKeyIndexSourceMeta(PrimaryKeyIndexSourceFile sourceFile) {
+ this(Collections.singletonList(sourceFile));
+ }
+
+ public List<PrimaryKeyIndexSourceFile> sourceFiles() {
return sourceFiles;
}
- public static PkVectorSourceMeta fromIndexFile(IndexFileMeta indexFile) {
+ public PrimaryKeyIndexSourceFile sourceFile() {
+ checkArgument(
+ sourceFiles.size() == 1,
+ "Expected exactly one source file, but found %s.",
+ sourceFiles.size());
+ return sourceFiles.get(0);
+ }
+
+ public static PrimaryKeyIndexSourceMeta fromIndexFile(IndexFileMeta
indexFile) {
GlobalIndexMeta globalIndexMeta = indexFile.globalIndexMeta();
checkArgument(
globalIndexMeta != null && globalIndexMeta.sourceMeta() !=
null,
- "Vector index file %s has no source metadata.",
+ "Index file %s has no source metadata.",
indexFile.fileName());
return deserialize(globalIndexMeta.sourceMeta());
}
@@ -60,32 +72,32 @@ public final class PkVectorSourceMeta {
DataOutputSerializer output = new DataOutputSerializer(128);
output.writeInt(VERSION);
output.writeInt(sourceFiles.size());
- for (PkVectorSourceFile sourceFile : sourceFiles) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
output.writeUTF(sourceFile.fileName());
output.writeLong(sourceFile.rowCount());
}
return output.getCopyOfBuffer();
} catch (IOException e) {
- throw new RuntimeException("Failed to serialize vector source
metadata.", e);
+ throw new RuntimeException("Failed to serialize index source
metadata.", e);
}
}
- public static PkVectorSourceMeta deserialize(byte[] bytes) {
+ public static PrimaryKeyIndexSourceMeta deserialize(byte[] bytes) {
try {
DataInputDeserializer input = new DataInputDeserializer(bytes);
int version = input.readInt();
- checkArgument(version == VERSION, "Unsupported vector source
version: %s.", version);
+ checkArgument(version == VERSION, "Unsupported index source
version: %s.", version);
int sourceFileCount = input.readInt();
- checkArgument(sourceFileCount > 0, "A vector index must reference
source files.");
- List<PkVectorSourceFile> sourceFiles = new
ArrayList<>(sourceFileCount);
+ checkArgument(sourceFileCount > 0, "An index must reference source
files.");
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new
ArrayList<>(sourceFileCount);
for (int i = 0; i < sourceFileCount; i++) {
- sourceFiles.add(new PkVectorSourceFile(input.readUTF(),
input.readLong()));
+ sourceFiles.add(new PrimaryKeyIndexSourceFile(input.readUTF(),
input.readLong()));
}
checkArgument(
- input.available() == 0, "Unexpected trailing bytes in
vector source metadata.");
- return new PkVectorSourceMeta(sourceFiles);
+ input.available() == 0, "Unexpected trailing bytes in
index source metadata.");
+ return new PrimaryKeyIndexSourceMeta(sourceFiles);
} catch (IOException e) {
- throw new IllegalArgumentException("Failed to deserialize vector
source metadata.", e);
+ throw new IllegalArgumentException("Failed to deserialize index
source metadata.", e);
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourcePolicy.java
similarity index 85%
rename from
paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java
rename to
paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourcePolicy.java
index 9db854e11a..4e1c3bebf8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorSourcePolicy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourcePolicy.java
@@ -16,15 +16,15 @@
* limitations under the License.
*/
-package org.apache.paimon.index.pkvector;
+package org.apache.paimon.index.pk;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
-/** Selects data files whose complete rows can be used to build vector index
payloads. */
-public final class PkVectorSourcePolicy {
+/** Selects complete compacted data files for source-backed primary-key
indexes. */
+public final class PrimaryKeyIndexSourcePolicy {
- private PkVectorSourcePolicy() {}
+ private PrimaryKeyIndexSourcePolicy() {}
public static boolean shouldWrite(FileSource fileSource, int level) {
return fileSource == FileSource.COMPACT && level > 0;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
index e1fefc5479..3b106f79c8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
@@ -22,6 +22,9 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
@@ -95,7 +98,7 @@ public class BucketedVectorIndexMaintainer {
this.annSegments = new ArrayList<>(restoredState.annSegments());
this.activeSourceFiles = new LinkedHashMap<>();
for (DataFileMeta file : restoredDataFiles) {
- if (PkVectorSourcePolicy.shouldRead(file)) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
activeSourceFiles.put(file.fileName(), file);
}
}
@@ -123,7 +126,7 @@ public class BucketedVectorIndexMaintainer {
}
}
for (DataFileMeta file : compactIncrement.compactAfter()) {
- if (PkVectorSourcePolicy.shouldRead(file)) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
nextSources.put(file.fileName(), file);
}
}
@@ -294,7 +297,7 @@ public class BucketedVectorIndexMaintainer {
List<IndexFileMeta> retained = new ArrayList<>(annSegments);
retained.removeAll(build.inputSegments);
Set<String> covered = coveredSources(retained);
- for (PkVectorSourceFile source :
sourceMeta(build.segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source :
sourceMeta(build.segment).sourceFiles()) {
DataFileMeta activeFile = activeSourceFiles.get(source.fileName());
if (activeFile == null
|| activeFile.rowCount() != source.rowCount()
@@ -326,8 +329,8 @@ public class BucketedVectorIndexMaintainer {
try {
List<PkVectorAnnSegmentFile.Source> sources = new
ArrayList<>(files.size());
for (DataFileMeta file : files) {
- PkVectorSourceFile sourceFile =
- new PkVectorSourceFile(file.fileName(),
file.rowCount());
+ PrimaryKeyIndexSourceFile sourceFile =
+ new PrimaryKeyIndexSourceFile(file.fileName(),
file.rowCount());
sources.add(
PkVectorAnnSegmentFile.Source.lazy(
sourceFile, () ->
vectorReaderFactory.create(file)));
@@ -347,8 +350,8 @@ public class BucketedVectorIndexMaintainer {
if (file == null) {
continue;
}
- PkVectorSourceMeta sourceMeta = sourceMeta(entry.getValue());
- for (PkVectorSourceFile source : sourceMeta.sourceFiles()) {
+ PrimaryKeyIndexSourceMeta sourceMeta =
sourceMeta(entry.getValue());
+ for (PrimaryKeyIndexSourceFile source : sourceMeta.sourceFiles()) {
if (source.fileName().equals(entry.getKey())) {
checkArgument(
source.rowCount() == file.rowCount(),
@@ -436,7 +439,7 @@ public class BucketedVectorIndexMaintainer {
private static List<DataFileMeta> eligibleFiles(List<DataFileMeta> files) {
List<DataFileMeta> eligible = new ArrayList<>();
for (DataFileMeta file : files) {
- if (PkVectorSourcePolicy.shouldRead(file)) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
eligible.add(file);
}
}
@@ -455,15 +458,15 @@ public class BucketedVectorIndexMaintainer {
private static Set<String> coveredSources(List<IndexFileMeta> segments) {
Set<String> sources = new HashSet<>();
for (IndexFileMeta ann : segments) {
- for (PkVectorSourceFile source : sourceMeta(ann).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source :
sourceMeta(ann).sourceFiles()) {
sources.add(source.fileName());
}
}
return sources;
}
- private static PkVectorSourceMeta sourceMeta(IndexFileMeta ann) {
- return PkVectorSourceMeta.fromIndexFile(ann);
+ private static PrimaryKeyIndexSourceMeta sourceMeta(IndexFileMeta ann) {
+ return PrimaryKeyIndexSourceMeta.fromIndexFile(ann);
}
/** Vector index changes for Paimon's append snapshot followed by its
compact snapshot. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
index 04c3dd7209..bd028b529e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
@@ -19,6 +19,8 @@
package org.apache.paimon.index.pkvector;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.DataFileMeta;
import java.util.ArrayList;
@@ -88,7 +90,8 @@ final class PkVectorAnnLevels {
IndexFileMeta segment, Map<String, DataFileMeta>
activeSourceFiles) {
long totalRows = 0;
long staleRows = 0;
- for (PkVectorSourceFile source :
PkVectorSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source :
+
PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
totalRows = Math.addExact(totalRows, source.rowCount());
DataFileMeta active = activeSourceFiles.get(source.fileName());
if (active == null) {
@@ -107,8 +110,8 @@ final class PkVectorAnnLevels {
List<IndexFileMeta> inputSegments, Map<String, DataFileMeta>
activeSourceFiles) {
Map<String, DataFileMeta> selectedSources = new TreeMap<>();
for (IndexFileMeta segment : inputSegments) {
- for (PkVectorSourceFile source :
- PkVectorSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source :
+
PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
DataFileMeta active = activeSourceFiles.get(source.fileName());
if (active != null) {
checkArgument(
@@ -124,7 +127,8 @@ final class PkVectorAnnLevels {
private static long buildRowCount(IndexFileMeta segment) {
long rowCount = 0;
- for (PkVectorSourceFile source :
PkVectorSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source :
+
PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
rowCount = Math.addExact(rowCount, source.rowCount());
}
return rowCount;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
index 9a3306734e..ed59ac9cf3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFile.java
@@ -31,6 +31,8 @@ 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;
@@ -65,7 +67,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
throws IOException {
checkArgument(!sources.isEmpty(), "An ANN segment must reference
source files.");
long totalRowCount = 0;
- List<PkVectorSourceFile> sourceFiles = new ArrayList<>(sources.size());
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new
ArrayList<>(sources.size());
for (Source source : sources) {
totalRowCount = Math.addExact(totalRowCount,
source.sourceFile.rowCount());
sourceFiles.add(source.sourceFile);
@@ -159,7 +161,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
vectorField.id(),
null,
payloadMetadata,
- new
PkVectorSourceMeta(sourceFiles).serialize()),
+ new
PrimaryKeyIndexSourceMeta(sourceFiles).serialize()),
pathFactory.isExternalPath() ?
payloadPath.toString() : null);
success = true;
return segment;
@@ -173,8 +175,8 @@ public class PkVectorAnnSegmentFile extends IndexFile {
}
}
- private static PkVectorSourceFile sourceMetadata(DataFileMeta sourceFile) {
- return new PkVectorSourceFile(sourceFile.fileName(),
sourceFile.rowCount());
+ private static PrimaryKeyIndexSourceFile sourceMetadata(DataFileMeta
sourceFile) {
+ return new PrimaryKeyIndexSourceFile(sourceFile.fileName(),
sourceFile.rowCount());
}
private static String normalizeMetric(String metric) {
@@ -213,7 +215,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
/** One vector source used while building an ANN segment. */
public static class Source {
- private final PkVectorSourceFile sourceFile;
+ private final PrimaryKeyIndexSourceFile sourceFile;
@Nullable private final PkVectorReader vectors;
@Nullable private final ReaderFactory readerFactory;
private final LongPredicate excludedPosition;
@@ -228,7 +230,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
}
Source(
- PkVectorSourceFile sourceFile,
+ PrimaryKeyIndexSourceFile sourceFile,
PkVectorReader vectors,
LongPredicate excludedPosition) {
this.sourceFile = sourceFile;
@@ -238,7 +240,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
}
private Source(
- PkVectorSourceFile sourceFile,
+ PrimaryKeyIndexSourceFile sourceFile,
ReaderFactory readerFactory,
LongPredicate excludedPosition) {
this.sourceFile = sourceFile;
@@ -247,7 +249,7 @@ public class PkVectorAnnSegmentFile extends IndexFile {
this.excludedPosition = excludedPosition;
}
- static Source lazy(PkVectorSourceFile sourceFile, ReaderFactory
readerFactory) {
+ static Source lazy(PrimaryKeyIndexSourceFile sourceFile, ReaderFactory
readerFactory) {
return new Source(sourceFile, readerFactory, position -> false);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
index b59d6a6ae5..e7fd34bd41 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
@@ -28,6 +28,8 @@ import org.apache.paimon.globalindex.VectorGlobalIndexer;
import org.apache.paimon.globalindex.VectorSearchMetric;
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.apache.paimon.options.Options;
import org.apache.paimon.predicate.VectorSearch;
import org.apache.paimon.types.DataField;
@@ -82,7 +84,7 @@ public class PkVectorAnnSegmentSearcher {
public List<PkVectorSearchResult> search(
IndexFileMeta segment,
- PkVectorSourceMeta sourceMeta,
+ PrimaryKeyIndexSourceMeta sourceMeta,
float[] query,
int limit,
@Nullable DeletionVector deletionVector,
@@ -99,13 +101,13 @@ public class PkVectorAnnSegmentSearcher {
public List<PkVectorSearchResult> search(
IndexFileMeta segment,
- PkVectorSourceMeta sourceMeta,
+ PrimaryKeyIndexSourceMeta sourceMeta,
float[] query,
int limit,
Map<String, DeletionVector> deletionVectors,
Map<String, String> searchOptions) {
Set<String> activeSourceFiles = new HashSet<>();
- for (PkVectorSourceFile sourceFile : sourceMeta.sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceMeta.sourceFiles()) {
activeSourceFiles.add(sourceFile.fileName());
}
return search(
@@ -120,7 +122,7 @@ public class PkVectorAnnSegmentSearcher {
public List<PkVectorSearchResult> search(
IndexFileMeta segment,
- PkVectorSourceMeta sourceMeta,
+ PrimaryKeyIndexSourceMeta sourceMeta,
float[] query,
int limit,
Map<String, DeletionVector> deletionVectors,
@@ -207,11 +209,11 @@ public class PkVectorAnnSegmentSearcher {
@Nullable
private static RoaringNavigableMap64 liveRowPositions(
- List<PkVectorSourceFile> sourceFiles,
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
Set<String> activeSourceFiles,
Map<String, DeletionVector> deletionVectors) {
boolean allSourcesActive = true;
- for (PkVectorSourceFile sourceFile : sourceFiles) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
if (!activeSourceFiles.contains(sourceFile.fileName())) {
allSourcesActive = false;
break;
@@ -223,7 +225,7 @@ public class PkVectorAnnSegmentSearcher {
RoaringNavigableMap64 live = new RoaringNavigableMap64();
RoaringNavigableMap64 deleted = new RoaringNavigableMap64();
long fileOffset = 0;
- for (PkVectorSourceFile sourceFile : sourceFiles) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
boolean active = activeSourceFiles.contains(sourceFile.fileName());
if (active && sourceFile.rowCount() > 0) {
live.addRange(new Range(fileOffset, fileOffset +
sourceFile.rowCount() - 1));
@@ -240,17 +242,18 @@ public class PkVectorAnnSegmentSearcher {
return live;
}
- private static long totalRowCount(List<PkVectorSourceFile> sourceFiles) {
+ private static long totalRowCount(List<PrimaryKeyIndexSourceFile>
sourceFiles) {
long total = 0;
- for (PkVectorSourceFile sourceFile : sourceFiles) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
total = Math.addExact(total, sourceFile.rowCount());
}
return total;
}
- private static FilePosition filePosition(List<PkVectorSourceFile>
sourceFiles, long ordinal) {
+ private static FilePosition filePosition(
+ List<PrimaryKeyIndexSourceFile> sourceFiles, long ordinal) {
long fileOffset = 0;
- for (PkVectorSourceFile sourceFile : sourceFiles) {
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
long nextOffset = fileOffset + sourceFile.rowCount();
if (ordinal < nextOffset) {
return new FilePosition(sourceFile.fileName(), ordinal -
fileOffset);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java
index 12eb770b0e..313f59e732 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexState.java
@@ -19,6 +19,8 @@
package org.apache.paimon.index.pkvector;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -52,8 +54,8 @@ public final class PkVectorBucketIndexState {
payloadsByName.put(payload.fileName(), payload) == null,
"Active vector payload %s appears more than once in the
index manifest.",
payload.fileName());
- PkVectorSourceMeta sourceMeta = validatedSourceMeta(payload);
- for (PkVectorSourceFile sourceFile : sourceMeta.sourceFiles()) {
+ PrimaryKeyIndexSourceMeta sourceMeta =
validatedSourceMeta(payload);
+ for (PrimaryKeyIndexSourceFile sourceFile :
sourceMeta.sourceFiles()) {
IndexFileMeta previous =
annBySource.put(sourceFile.fileName(), payload);
checkArgument(
previous == null,
@@ -84,9 +86,9 @@ public final class PkVectorBucketIndexState {
return sourceFileToAnnSegment;
}
- private PkVectorSourceMeta validatedSourceMeta(IndexFileMeta payload) {
+ private PrimaryKeyIndexSourceMeta validatedSourceMeta(IndexFileMeta
payload) {
validateIdentity(payload);
- return PkVectorSourceMeta.fromIndexFile(payload);
+ return PrimaryKeyIndexSourceMeta.fromIndexFile(payload);
}
private void validateIdentity(IndexFileMeta payload) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearch.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearch.java
index 62bb835b8f..04d4a32755 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearch.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearch.java
@@ -21,6 +21,8 @@ package org.apache.paimon.index.pkvector;
import org.apache.paimon.CoreOptions.GlobalIndexSearchMode;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.DataFileMeta;
import javax.annotation.Nullable;
@@ -106,8 +108,8 @@ public class PrimaryKeyVectorBucketSearch {
Set<String> activeSourceFiles = new HashSet<>(filesByName.keySet());
Set<String> covered = new HashSet<>();
for (IndexFileMeta ann : state.annSegments()) {
- PkVectorSourceMeta sourceMeta =
PkVectorSourceMeta.fromIndexFile(ann);
- for (PkVectorSourceFile source : sourceMeta.sourceFiles()) {
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(ann);
+ for (PrimaryKeyIndexSourceFile source : sourceMeta.sourceFiles()) {
DataFileMeta file = filesByName.get(source.fileName());
if (file == null) {
continue;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java
index 785a93c5f4..6daf359148 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java
@@ -28,11 +28,11 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.VectorSearchMetric;
import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
import org.apache.paimon.index.pkvector.PkVectorAnnSegmentSearcher;
import org.apache.paimon.index.pkvector.PkVectorBucketIndexState;
import org.apache.paimon.index.pkvector.PkVectorDataFileReader;
import org.apache.paimon.index.pkvector.PkVectorSearchResult;
-import org.apache.paimon.index.pkvector.PkVectorSourcePolicy;
import org.apache.paimon.index.pkvector.PrimaryKeyVectorBucketSearch;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.KeyValueFileReaderFactory;
@@ -212,7 +212,7 @@ public class PrimaryKeyVectorRead implements VectorRead,
Serializable {
DataSplit dataSplit = split.dataSplit();
List<DataFileMeta> activeFiles =
dataSplit.dataFiles().stream()
- .filter(PkVectorSourcePolicy::shouldRead)
+ .filter(PrimaryKeyIndexSourcePolicy::shouldRead)
.collect(Collectors.toList());
PkVectorBucketIndexState state =
PkVectorBucketIndexState.fromActivePayloads(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMetaTest.java
new file mode 100644
index 0000000000..9cdcbfd2a5
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexSourceMetaTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.pk;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PrimaryKeyIndexSourceMeta}. */
+class PrimaryKeyIndexSourceMetaTest {
+
+ @Test
+ void testMultipleSourceRoundTrip() {
+ PrimaryKeyIndexSourceMeta metadata =
+ new PrimaryKeyIndexSourceMeta(
+ Arrays.asList(
+ new PrimaryKeyIndexSourceFile("data-1", 10),
+ new PrimaryKeyIndexSourceFile("data-2", 20)));
+
+ PrimaryKeyIndexSourceMeta restored =
+ PrimaryKeyIndexSourceMeta.deserialize(metadata.serialize());
+
+ assertThat(restored.sourceFiles()).isEqualTo(metadata.sourceFiles());
+ assertThatThrownBy(restored::sourceFile)
+ .hasMessageContaining("Expected exactly one source file");
+ }
+
+ @Test
+ void testSingleSourceRoundTrip() {
+ PrimaryKeyIndexSourceMeta metadata =
+ new PrimaryKeyIndexSourceMeta(new
PrimaryKeyIndexSourceFile("data-1", 0));
+
+ PrimaryKeyIndexSourceMeta restored =
+ PrimaryKeyIndexSourceMeta.deserialize(metadata.serialize());
+
+ assertThat(restored.sourceFile()).isEqualTo(metadata.sourceFile());
+ }
+
+ @Test
+ void testRejectsInvalidSourceFile() {
+ assertThatThrownBy(() -> new PrimaryKeyIndexSourceFile("data-1", -1))
+ .hasMessageContaining("row count must not be negative");
+ }
+
+ @Test
+ void testRejectsUnsupportedVersion() throws Exception {
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ output.writeInt(2);
+ output.writeInt(1);
+ output.writeUTF("data-1");
+ output.writeLong(1);
+
+ assertThatThrownBy(() ->
PrimaryKeyIndexSourceMeta.deserialize(output.getCopyOfBuffer()))
+ .hasMessageContaining("Unsupported index source version: 2");
+ }
+
+ @Test
+ void testRejectsTruncatedAndTrailingMetadata() throws Exception {
+ DataOutputSerializer truncated = new DataOutputSerializer(64);
+ truncated.writeInt(1);
+ truncated.writeInt(1);
+ truncated.writeUTF("data-1");
+ assertThatThrownBy(() ->
PrimaryKeyIndexSourceMeta.deserialize(truncated.getCopyOfBuffer()))
+ .hasMessageContaining("Failed to deserialize index source
metadata");
+
+ DataOutputSerializer trailing = new DataOutputSerializer(64);
+ trailing.writeInt(1);
+ trailing.writeInt(1);
+ trailing.writeUTF("data-1");
+ trailing.writeLong(1);
+ trailing.writeByte(42);
+ assertThatThrownBy(() ->
PrimaryKeyIndexSourceMeta.deserialize(trailing.getCopyOfBuffer()))
+ .hasMessageContaining("Unexpected trailing bytes in index
source metadata");
+ }
+
+ @Test
+ void testReadsFromIndexFile() {
+ PrimaryKeyIndexSourceMeta metadata =
+ new PrimaryKeyIndexSourceMeta(new
PrimaryKeyIndexSourceFile("data-1", 5));
+ IndexFileMeta indexFile =
+ new IndexFileMeta(
+ "btree",
+ "index-1",
+ 10,
+ 5,
+ new GlobalIndexMeta(0, 4, 1, null, null,
metadata.serialize()),
+ null);
+
+
assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(indexFile).sourceFile())
+ .isEqualTo(metadata.sourceFile());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
index d86e412366..2a6b0dc65e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
@@ -22,6 +22,8 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
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.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
@@ -189,8 +191,8 @@ class BucketedVectorIndexMaintainerTest {
assertThat(commit.appendIncrement()).isPresent();
IndexFileMeta segment =
commit.appendIncrement().get().newIndexFiles().get(0);
- assertThat(PkVectorSourceMeta.fromIndexFile(segment).sourceFiles())
- .extracting(PkVectorSourceFile::fileName)
+
assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles())
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
.containsExactly("data-2");
assertThat(annFile.exists(segment)).isTrue();
} finally {
@@ -253,8 +255,8 @@ class BucketedVectorIndexMaintainerTest {
assertThat(increment.newIndexFiles()).hasSize(1);
IndexFileMeta delta = increment.newIndexFiles().get(0);
assertThat(delta.indexType()).isEqualTo("test-vector-ann");
- assertThat(PkVectorSourceMeta.fromIndexFile(delta).sourceFiles())
- .extracting(PkVectorSourceFile::fileName)
+
assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(delta).sourceFiles())
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
.containsExactly("data-3");
assertThat(maintainer.segments()).containsExactly(initialAnn, delta);
}
@@ -361,8 +363,10 @@ class BucketedVectorIndexMaintainerTest {
commit.appendIncrement().get();
assertThat(increment.deletedIndexFiles()).containsExactlyInAnyOrder(ann1, ann2,
ann3);
assertThat(increment.newIndexFiles()).hasSize(1);
-
assertThat(PkVectorSourceMeta.fromIndexFile(increment.newIndexFiles().get(0)).sourceFiles())
- .extracting(PkVectorSourceFile::fileName)
+ assertThat(
+
PrimaryKeyIndexSourceMeta.fromIndexFile(increment.newIndexFiles().get(0))
+ .sourceFiles())
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
.containsExactly("data-1", "data-2", "data-3");
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
index fb7eb22d3f..14c8648fea 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
@@ -20,6 +20,8 @@ package org.apache.paimon.index.pkvector;
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.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.stats.SimpleStats;
@@ -101,11 +103,11 @@ class PkVectorAnnLevelsTest {
private static IndexFileMeta segment(String fileName, DataFileMeta...
sources) {
long rowCount =
Arrays.stream(sources).mapToLong(DataFileMeta::rowCount).sum();
byte[] sourceMeta =
- new PkVectorSourceMeta(
+ new PrimaryKeyIndexSourceMeta(
Arrays.stream(sources)
.map(
source ->
- new PkVectorSourceFile(
+ new
PrimaryKeyIndexSourceFile(
source.fileName(),
source.rowCount()))
.collect(java.util.stream.Collectors.toList()))
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
index 3178636075..76c9fa832c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentFileTest.java
@@ -23,6 +23,8 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
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.manifest.FileSource;
import org.apache.paimon.options.Options;
@@ -70,9 +72,9 @@ class PkVectorAnnSegmentFileTest {
assertThat(segment.indexType()).isEqualTo("test-vector-ann");
assertThat(segment.rowCount()).isEqualTo(1);
- PkVectorSourceMeta sourceMeta =
PkVectorSourceMeta.fromIndexFile(segment);
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(segment);
assertThat(sourceMeta.sourceFiles())
- .extracting(PkVectorSourceFile::fileName)
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
.containsExactly("data-1");
}
@@ -95,7 +97,7 @@ class PkVectorAnnSegmentFileTest {
"test-vector-ann");
assertThat(segment.globalIndexMeta().rowRangeStart()).isZero();
assertThat(segment.globalIndexMeta().rowRangeEnd()).isEqualTo(3);
- PkVectorSourceMeta sourceMeta =
PkVectorSourceMeta.fromIndexFile(segment);
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(segment);
BitmapDeletionVector data2Deletes = new BitmapDeletionVector();
data2Deletes.delete(0);
Map<String, org.apache.paimon.deletionvectors.DeletionVector>
deletionVectors =
@@ -153,7 +155,7 @@ class PkVectorAnnSegmentFileTest {
fileIO, annFile, vectorField(),
indexOptions(), "l2", executor)
.search(
segment,
- PkVectorSourceMeta.fromIndexFile(segment),
+
PrimaryKeyIndexSourceMeta.fromIndexFile(segment),
new float[] {0, 0},
3,
Collections.emptyMap(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java
index 6fd35521fe..be939cf8a0 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorBucketIndexStateTest.java
@@ -20,6 +20,8 @@ package org.apache.paimon.index.pkvector;
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;
@@ -81,9 +83,9 @@ class PkVectorBucketIndexStateTest {
private static IndexFileMeta segment(
String segmentFileName, String sourceFileName, String indexType) {
- PkVectorSourceFile sourceFile = new PkVectorSourceFile(sourceFileName,
10);
+ PrimaryKeyIndexSourceFile sourceFile = new
PrimaryKeyIndexSourceFile(sourceFileName, 10);
byte[] sourceMeta =
- new
PkVectorSourceMeta(Collections.singletonList(sourceFile)).serialize();
+ new
PrimaryKeyIndexSourceMeta(Collections.singletonList(sourceFile)).serialize();
return new IndexFileMeta(
indexType,
segmentFileName,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
deleted file mode 100644
index 9d8e967267..0000000000
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorSourceMetaTest.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.pkvector;
-
-import org.apache.paimon.io.DataOutputSerializer;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Tests for {@link PkVectorSourceMeta}. */
-class PkVectorSourceMetaTest {
-
- @Test
- void testRoundTrip() {
- PkVectorSourceMeta metadata =
- new PkVectorSourceMeta(
- Arrays.asList(
- new PkVectorSourceFile("data-1", 10),
- new PkVectorSourceFile("data-2", 20)));
-
- PkVectorSourceMeta restored =
PkVectorSourceMeta.deserialize(metadata.serialize());
-
- assertThat(restored.sourceFiles()).isEqualTo(metadata.sourceFiles());
- }
-
- @Test
- void testRejectsTruncatedSourceMetadata() throws Exception {
- DataOutputSerializer output = new DataOutputSerializer(128);
- output.writeInt(1);
- output.writeInt(1);
- output.writeUTF("data-1");
-
- assertThatThrownBy(() ->
PkVectorSourceMeta.deserialize(output.getCopyOfBuffer()))
- .hasMessageContaining("Failed to deserialize vector source
metadata");
- }
-}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearchTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearchTest.java
index fc52293be7..3437d94f24 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearchTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PrimaryKeyVectorBucketSearchTest.java
@@ -23,6 +23,8 @@ import org.apache.paimon.deletionvectors.BitmapDeletionVector;
import org.apache.paimon.deletionvectors.DeletionVector;
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.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.stats.SimpleStats;
@@ -111,7 +113,7 @@ class PrimaryKeyVectorBucketSearchTest {
Map<String, String> searchOptions =
Collections.singletonMap("nprobes", "8");
when(annSearcher.search(
org.mockito.ArgumentMatchers.eq(ann),
-
org.mockito.ArgumentMatchers.any(PkVectorSourceMeta.class),
+
org.mockito.ArgumentMatchers.any(PrimaryKeyIndexSourceMeta.class),
org.mockito.ArgumentMatchers.any(float[].class),
org.mockito.ArgumentMatchers.eq(2),
org.mockito.ArgumentMatchers.eq(deletionVectors),
@@ -161,7 +163,7 @@ class PrimaryKeyVectorBucketSearchTest {
Map<String, DeletionVector> deletionVectors = Collections.emptyMap();
when(annSearcher.search(
org.mockito.ArgumentMatchers.eq(ann),
-
org.mockito.ArgumentMatchers.any(PkVectorSourceMeta.class),
+
org.mockito.ArgumentMatchers.any(PrimaryKeyIndexSourceMeta.class),
org.mockito.ArgumentMatchers.any(float[].class),
org.mockito.ArgumentMatchers.eq(1),
org.mockito.ArgumentMatchers.eq(deletionVectors),
@@ -274,11 +276,11 @@ class PrimaryKeyVectorBucketSearchTest {
private static IndexFileMeta segment(String fileName, List<DataFileMeta>
sources) {
long rowCount =
sources.stream().mapToLong(DataFileMeta::rowCount).sum();
byte[] sourceMeta =
- new PkVectorSourceMeta(
+ new PrimaryKeyIndexSourceMeta(
sources.stream()
.map(
source ->
- new PkVectorSourceFile(
+ new
PrimaryKeyIndexSourceFile(
source.fileName(),
source.rowCount()))
.collect(java.util.stream.Collectors.toList()))
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 25b823139c..5ff206752f 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
@@ -25,8 +25,8 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.index.pkvector.PkVectorSourceFile;
-import org.apache.paimon.index.pkvector.PkVectorSourceMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
@@ -223,8 +223,9 @@ class PrimaryKeyVectorScanTest {
private static IndexFileMeta payloadFile(String indexType, int fieldId,
String fileName) {
byte[] sourceMeta =
- new PkVectorSourceMeta(
- Collections.singletonList(new
PkVectorSourceFile("data-1", 2)))
+ new PrimaryKeyIndexSourceMeta(
+ Collections.singletonList(
+ new
PrimaryKeyIndexSourceFile("data-1", 2)))
.serialize();
return new IndexFileMeta(
indexType,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
index 7909580876..4b01a3a1e6 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
@@ -26,8 +26,8 @@ import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.index.pkvector.PkVectorSourceFile;
-import org.apache.paimon.index.pkvector.PkVectorSourceMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.ManifestEntry;
@@ -99,9 +99,9 @@ class TableWriteCoordinatorTest extends TableTestBase {
FileStoreTable table = getTable(identifier);
write(table, GenericRow.of(1));
- PkVectorSourceMeta sourceMeta =
- new PkVectorSourceMeta(
- Collections.singletonList(new
PkVectorSourceFile("data", 1)));
+ PrimaryKeyIndexSourceMeta sourceMeta =
+ new PrimaryKeyIndexSourceMeta(
+ Collections.singletonList(new
PrimaryKeyIndexSourceFile("data", 1)));
IndexFileMeta ann =
new IndexFileMeta(
"test-vector-ann",
