Skip to content

[core] Prepare PK vector index lifecycle for multiple columns #10688

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tzphh wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[core] Prepare PK vector index lifecycle for multiple columns #10688

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 77 additions & 48 deletions paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@
import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.IOUtils;

import javax.annotation.Nullable;

Expand All @@ -55,16 +57,16 @@
/** Coordinates all source-backed primary-key indexes for one bucket. */
public final class BucketedPrimaryKeyIndexMaintainer {

@Nullable private final BucketedVectorIndexMaintainer vectorMaintainer;
private final List<BucketedVectorIndexMaintainer> vectorMaintainers;

@Akash3121 Akash3121 Oct 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renaming this field breaks PrimaryKeyIndexWriteTest.testCreatesCoordinatorForMultipleScalarDefinitionsAndVector, which still calls readField(..., "vectorMaintainer") at line 142. Both Core JDK 8 and JDK 11 CI fail with NoSuchFieldException. Please read vectorMaintainers, assert that the list has one entry, and use its first maintainer for the existing deletion-vector-factory assertion.

@tzphh tzphh Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 0479ef5. The test now reads vectorMaintainers, asserts that it contains one maintainer, and uses that maintainer for the existing deletion-vector factory assertion. All 19 focused Core tests passed locally on this CR1 head, including the previously failing test (JDK 17).

I also verified the combined head e787eb7, which includes this fix, on JDK 8 (1.8.0_512) and JDK 11 (11.0.27). All 139 focused Core tests passed on each runtime, including this regression test.

@Nullable private final BucketedFullTextIndexMaintainer fullTextMaintainer;
private final List<BucketedSortedIndexMaintainer> sortedMaintainers;
private int nextSortedMaintainerIndex;

private BucketedPrimaryKeyIndexMaintainer(
@Nullable BucketedVectorIndexMaintainer vectorMaintainer,
List<BucketedVectorIndexMaintainer> vectorMaintainers,
@Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
this.vectorMaintainer = vectorMaintainer;
this.vectorMaintainers = Collections.unmodifiableList(new ArrayList<>(vectorMaintainers));
this.fullTextMaintainer = fullTextMaintainer;
List<BucketedSortedIndexMaintainer> sorted = new ArrayList<>(sortedMaintainers);
sorted.sort(Comparator.comparingInt(BucketedSortedIndexMaintainer::fieldId));
Expand All @@ -74,47 +76,60 @@ private BucketedPrimaryKeyIndexMaintainer(
public static BucketedPrimaryKeyIndexMaintainer ofVector(
BucketedVectorIndexMaintainer vectorMaintainer) {
return new BucketedPrimaryKeyIndexMaintainer(
vectorMaintainer, null, Collections.emptyList());
Collections.singletonList(vectorMaintainer), null, Collections.emptyList());
}

public static BucketedPrimaryKeyIndexMaintainer ofFullText(
BucketedFullTextIndexMaintainer fullTextMaintainer) {
return new BucketedPrimaryKeyIndexMaintainer(
null, fullTextMaintainer, Collections.emptyList());
Collections.emptyList(), fullTextMaintainer, Collections.emptyList());
}

public static BucketedPrimaryKeyIndexMaintainer of(
BucketedVectorIndexMaintainer vectorMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer, null, sortedMaintainers);
return of(vectorMaintainer, null, sortedMaintainers);
}

public static BucketedPrimaryKeyIndexMaintainer of(
@Nullable BucketedVectorIndexMaintainer vectorMaintainer,
@Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
return new BucketedPrimaryKeyIndexMaintainer(
vectorMaintainer, fullTextMaintainer, sortedMaintainers);
vectorMaintainer == null
? Collections.emptyList()
: Collections.singletonList(vectorMaintainer),
fullTextMaintainer,
sortedMaintainers);
}

public static BucketedPrimaryKeyIndexMaintainer ofVectors(
List<BucketedVectorIndexMaintainer> vectorMaintainers,
@Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
return new BucketedPrimaryKeyIndexMaintainer(
vectorMaintainers, fullTextMaintainer, sortedMaintainers);
}

public static BucketedPrimaryKeyIndexMaintainer ofSorted(
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
return new BucketedPrimaryKeyIndexMaintainer(null, null, sortedMaintainers);
return new BucketedPrimaryKeyIndexMaintainer(
Collections.emptyList(), null, sortedMaintainers);
}

public synchronized void prepareCommit(
DataIncrement appendIncrement,
CompactIncrement compactIncrement,
boolean waitCompaction)
throws Exception {
BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit = null;
List<BucketedVectorIndexMaintainer.VectorIndexCommit> vectorCommits = new ArrayList<>();
BucketedFullTextIndexMaintainer.FullTextIndexCommit fullTextCommit = null;
List<BucketedSortedIndexMaintainer.SortedIndexCommit> sortedCommits = new ArrayList<>();
try {
if (vectorMaintainer != null) {
vectorCommit =
vectorMaintainer.prepareCommit(
appendIncrement, compactIncrement, waitCompaction);
for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
vectorCommits.add(
maintainer.prepareCommit(
appendIncrement, compactIncrement, waitCompaction));
}
if (fullTextMaintainer != null) {
fullTextCommit =
Expand All @@ -128,8 +143,8 @@ public synchronized void prepareCommit(
prepareSortedNonBlocking(appendIncrement, compactIncrement, sortedCommits);
}

if (vectorCommit != null) {
mergeVectorCommit(appendIncrement, compactIncrement, vectorCommit);
for (BucketedVectorIndexMaintainer.VectorIndexCommit commit : vectorCommits) {
mergeVectorCommit(appendIncrement, compactIncrement, commit);
}
if (fullTextCommit != null) {
mergeFullTextCommit(appendIncrement, compactIncrement, fullTextCommit);
Expand All @@ -144,8 +159,8 @@ public synchronized void prepareCommit(
if (fullTextCommit != null) {
fullTextCommit.abort(failure);
}
if (vectorCommit != null) {
vectorCommit.abort(failure);
for (int i = vectorCommits.size() - 1; i >= 0; i--) {
vectorCommits.get(i).abort(failure);
}
if (failure instanceof Exception) {
throw (Exception) failure;
Expand Down Expand Up @@ -287,8 +302,10 @@ private static void applyIndexIncrement(
}

public boolean buildNotCompleted() {
if (vectorMaintainer != null && vectorMaintainer.buildNotCompleted()) {
return true;
for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
if (maintainer.buildNotCompleted()) {
return true;
}
}
if (fullTextMaintainer != null && fullTextMaintainer.buildNotCompleted()) {
return true;
Expand All @@ -297,8 +314,11 @@ public boolean buildNotCompleted() {
}

public void withExecutor(ExecutorService executor) {
if (vectorMaintainer != null) {
vectorMaintainer.withExecutor(executor);
checkArgument(
!buildNotCompleted(),
"Cannot replace executor during primary-key index maintenance.");
for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
maintainer.withExecutor(executor);
}
if (fullTextMaintainer != null) {
fullTextMaintainer.withExecutor(executor);
Expand All @@ -309,39 +329,48 @@ public void withExecutor(ExecutorService executor) {
}

public void close() {
if (vectorMaintainer != null) {
vectorMaintainer.close();
List<AutoCloseable> closeables = new ArrayList<>();
for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
closeables.add(maintainer::close);
}
if (fullTextMaintainer != null) {
fullTextMaintainer.close();
closeables.add(fullTextMaintainer::close);
}
for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
maintainer.close();
closeables.add(maintainer::close);
}
try {
IOUtils.closeAll(closeables);
} catch (Exception e) {
ExceptionUtils.rethrow(e);
}
}

/** Factory to restore all configured source-backed indexes for a bucket. */
public static final class Factory {

private final IndexFileHandler handler;
@Nullable private final BucketedVectorIndexMaintainer.Factory vectorFactory;
private final List<BucketedVectorIndexMaintainer.Factory> vectorFactories;
@Nullable private final FullTextDefinitionFactory fullTextFactory;
private final List<SortedDefinitionFactory> sortedFactories;

private Factory(
IndexFileHandler handler,
@Nullable BucketedVectorIndexMaintainer.Factory vectorFactory,
List<BucketedVectorIndexMaintainer.Factory> vectorFactories,
@Nullable FullTextDefinitionFactory fullTextFactory,
List<SortedDefinitionFactory> sortedFactories) {
this.handler = handler;
this.vectorFactory = vectorFactory;
this.vectorFactories = Collections.unmodifiableList(new ArrayList<>(vectorFactories));
this.fullTextFactory = fullTextFactory;
this.sortedFactories = Collections.unmodifiableList(new ArrayList<>(sortedFactories));
}

public static Factory ofVector(BucketedVectorIndexMaintainer.Factory vectorFactory) {
return new Factory(
vectorFactory.indexFileHandler(), vectorFactory, null, Collections.emptyList());
vectorFactory.indexFileHandler(),
Collections.singletonList(vectorFactory),
null,
Collections.emptyList());
}

public static Factory create(
Expand All @@ -350,7 +379,7 @@ public static Factory create(
TableSchema schema) {
CoreOptions coreOptions = new CoreOptions(schema.options());
Map<String, DataField> fields = schema.nameToFieldMap();
BucketedVectorIndexMaintainer.Factory vectorFactory = null;
List<BucketedVectorIndexMaintainer.Factory> vectorFactories = new ArrayList<>();
FullTextDefinitionFactory fullTextFactory = null;
List<SortedDefinitionFactory> sortedFactories = new ArrayList<>();
for (PrimaryKeyIndexDefinition definition :
Expand All @@ -362,18 +391,15 @@ public static Factory create(
definition.column());
switch (definition.family()) {
case VECTOR:
checkArgument(
vectorFactory == null,
"Only one primary-key vector index is supported.");
vectorFactory =
vectorFactories.add(
new BucketedVectorIndexMaintainer.Factory(
handler,
readerFactoryBuilder,
field,
coreOptions.primaryKeyVectorDistanceMetric(
definition.column()),
definition.indexType())
.withIndexOptions(definition.options());
.withIndexOptions(definition.options()));
break;
case BTREE:
case BITMAP:
Expand Down Expand Up @@ -409,9 +435,11 @@ public static Factory create(
}
}
checkArgument(
vectorFactory != null || fullTextFactory != null || !sortedFactories.isEmpty(),
!vectorFactories.isEmpty()
|| fullTextFactory != null
|| !sortedFactories.isEmpty(),
"No primary-key index definition is configured.");
return new Factory(handler, vectorFactory, fullTextFactory, sortedFactories);
return new Factory(handler, vectorFactories, fullTextFactory, sortedFactories);
}

public IndexFileHandler indexFileHandler() {
Expand Down Expand Up @@ -456,16 +484,17 @@ public BucketedPrimaryKeyIndexMaintainer create(
restoredDataFiles == null ? Collections.emptyList() : restoredDataFiles;
List<IndexFileMeta> payloads =
restoredPayloads == null ? Collections.emptyList() : restoredPayloads;
BucketedVectorIndexMaintainer vector =
vectorFactory == null
? null
: vectorFactory.create(
partition,
bucket,
dataFiles,
payloads,
executor,
deletionVectorFactory);
List<BucketedVectorIndexMaintainer> vectors = new ArrayList<>();
for (BucketedVectorIndexMaintainer.Factory factory : vectorFactories) {
vectors.add(
factory.create(
partition,
bucket,
dataFiles,
payloads,
executor,
deletionVectorFactory));
}
BucketedFullTextIndexMaintainer fullText =
fullTextFactory == null
? null
Expand All @@ -478,7 +507,7 @@ public BucketedPrimaryKeyIndexMaintainer create(
handler, partition, bucket, dataFiles, payloads, executor,
ioManager));
}
return new BucketedPrimaryKeyIndexMaintainer(vector, fullText, sorted);
return new BucketedPrimaryKeyIndexMaintainer(vectors, fullText, sorted);
}

private static final class FullTextDefinitionFactory {
Expand Down
Loading
Loading