diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java index 585c9a7b6c8a..44ff1c4be569 100644 --- a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java +++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java @@ -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; @@ -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 vectorMaintainers; @Nullable private final BucketedFullTextIndexMaintainer fullTextMaintainer; private final List sortedMaintainers; private int nextSortedMaintainerIndex; private BucketedPrimaryKeyIndexMaintainer( - @Nullable BucketedVectorIndexMaintainer vectorMaintainer, + List vectorMaintainers, @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer, List sortedMaintainers) { - this.vectorMaintainer = vectorMaintainer; + this.vectorMaintainers = Collections.unmodifiableList(new ArrayList<>(vectorMaintainers)); this.fullTextMaintainer = fullTextMaintainer; List sorted = new ArrayList<>(sortedMaintainers); sorted.sort(Comparator.comparingInt(BucketedSortedIndexMaintainer::fieldId)); @@ -74,19 +76,19 @@ 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 sortedMaintainers) { - return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer, null, sortedMaintainers); + return of(vectorMaintainer, null, sortedMaintainers); } public static BucketedPrimaryKeyIndexMaintainer of( @@ -94,12 +96,25 @@ public static BucketedPrimaryKeyIndexMaintainer of( @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer, List sortedMaintainers) { return new BucketedPrimaryKeyIndexMaintainer( - vectorMaintainer, fullTextMaintainer, sortedMaintainers); + vectorMaintainer == null + ? Collections.emptyList() + : Collections.singletonList(vectorMaintainer), + fullTextMaintainer, + sortedMaintainers); + } + + public static BucketedPrimaryKeyIndexMaintainer ofVectors( + List vectorMaintainers, + @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer, + List sortedMaintainers) { + return new BucketedPrimaryKeyIndexMaintainer( + vectorMaintainers, fullTextMaintainer, sortedMaintainers); } public static BucketedPrimaryKeyIndexMaintainer ofSorted( List sortedMaintainers) { - return new BucketedPrimaryKeyIndexMaintainer(null, null, sortedMaintainers); + return new BucketedPrimaryKeyIndexMaintainer( + Collections.emptyList(), null, sortedMaintainers); } public synchronized void prepareCommit( @@ -107,14 +122,14 @@ public synchronized void prepareCommit( CompactIncrement compactIncrement, boolean waitCompaction) throws Exception { - BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit = null; + List vectorCommits = new ArrayList<>(); BucketedFullTextIndexMaintainer.FullTextIndexCommit fullTextCommit = null; List 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 = @@ -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); @@ -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; @@ -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; @@ -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); @@ -309,14 +329,20 @@ public void withExecutor(ExecutorService executor) { } public void close() { - if (vectorMaintainer != null) { - vectorMaintainer.close(); + List 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); } } @@ -324,24 +350,27 @@ public void close() { public static final class Factory { private final IndexFileHandler handler; - @Nullable private final BucketedVectorIndexMaintainer.Factory vectorFactory; + private final List vectorFactories; @Nullable private final FullTextDefinitionFactory fullTextFactory; private final List sortedFactories; private Factory( IndexFileHandler handler, - @Nullable BucketedVectorIndexMaintainer.Factory vectorFactory, + List vectorFactories, @Nullable FullTextDefinitionFactory fullTextFactory, List 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( @@ -350,7 +379,7 @@ public static Factory create( TableSchema schema) { CoreOptions coreOptions = new CoreOptions(schema.options()); Map fields = schema.nameToFieldMap(); - BucketedVectorIndexMaintainer.Factory vectorFactory = null; + List vectorFactories = new ArrayList<>(); FullTextDefinitionFactory fullTextFactory = null; List sortedFactories = new ArrayList<>(); for (PrimaryKeyIndexDefinition definition : @@ -362,10 +391,7 @@ 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, @@ -373,7 +399,7 @@ public static Factory create( coreOptions.primaryKeyVectorDistanceMetric( definition.column()), definition.indexType()) - .withIndexOptions(definition.options()); + .withIndexOptions(definition.options())); break; case BTREE: case BITMAP: @@ -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() { @@ -456,16 +484,17 @@ public BucketedPrimaryKeyIndexMaintainer create( restoredDataFiles == null ? Collections.emptyList() : restoredDataFiles; List payloads = restoredPayloads == null ? Collections.emptyList() : restoredPayloads; - BucketedVectorIndexMaintainer vector = - vectorFactory == null - ? null - : vectorFactory.create( - partition, - bucket, - dataFiles, - payloads, - executor, - deletionVectorFactory); + List vectors = new ArrayList<>(); + for (BucketedVectorIndexMaintainer.Factory factory : vectorFactories) { + vectors.add( + factory.create( + partition, + bucket, + dataFiles, + payloads, + executor, + deletionVectorFactory)); + } BucketedFullTextIndexMaintainer fullText = fullTextFactory == null ? null @@ -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 { diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java index 531f440930c8..b029b42c492f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java @@ -31,11 +31,13 @@ import org.apache.paimon.index.pksorted.BucketedSortedIndexMaintainer; import org.apache.paimon.index.pksorted.PkSortedIndexFile; import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer; +import org.apache.paimon.index.pkvector.PkVectorAnnSegmentFile; import org.apache.paimon.io.CompactIncrement; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataIncrement; import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.options.Options; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.types.DataField; @@ -44,6 +46,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.InOrder; import java.util.ArrayList; import java.util.Arrays; @@ -64,8 +67,12 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Answers.RETURNS_SELF; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -82,18 +89,121 @@ void shutdownExecutor() { @Test void testDelegatesVectorLifecycle() { - BucketedVectorIndexMaintainer vector = mock(BucketedVectorIndexMaintainer.class); + BucketedVectorIndexMaintainer first = mock(BucketedVectorIndexMaintainer.class); + BucketedVectorIndexMaintainer second = mock(BucketedVectorIndexMaintainer.class); ExecutorService executor = mock(ExecutorService.class); - when(vector.buildNotCompleted()).thenReturn(true); + when(second.buildNotCompleted()).thenReturn(true); BucketedPrimaryKeyIndexMaintainer maintainer = - BucketedPrimaryKeyIndexMaintainer.ofVector(vector); + BucketedPrimaryKeyIndexMaintainer.ofVectors( + Arrays.asList(first, second), null, Collections.emptyList()); assertThat(maintainer.buildNotCompleted()).isTrue(); + assertThatThrownBy(() -> maintainer.withExecutor(executor)) + .hasMessageContaining("Cannot replace executor"); + verify(first, never()).withExecutor(any()); + verify(second, never()).withExecutor(any()); + + when(second.buildNotCompleted()).thenReturn(false); + assertThat(maintainer.buildNotCompleted()).isFalse(); maintainer.withExecutor(executor); maintainer.close(); - verify(vector).withExecutor(executor); - verify(vector).close(); + verify(first).withExecutor(executor); + verify(second).withExecutor(executor); + verify(first).close(); + verify(second).close(); + } + + @Test + void testClosesAllIndexesWhenVectorCloseFails() { + BucketedVectorIndexMaintainer first = mock(BucketedVectorIndexMaintainer.class); + BucketedVectorIndexMaintainer second = mock(BucketedVectorIndexMaintainer.class); + RuntimeException failure = new RuntimeException("first close failed"); + RuntimeException suppressed = new RuntimeException("second close failed"); + doThrow(failure).when(first).close(); + doThrow(suppressed).when(second).close(); + BucketedFullTextIndexMaintainer fullText = mock(BucketedFullTextIndexMaintainer.class); + BucketedSortedIndexMaintainer sorted = mock(BucketedSortedIndexMaintainer.class); + BucketedPrimaryKeyIndexMaintainer maintainer = + BucketedPrimaryKeyIndexMaintainer.ofVectors( + Arrays.asList(first, second), fullText, Collections.singletonList(sorted)); + + assertThatThrownBy(maintainer::close).isSameAs(failure).hasSuppressedException(suppressed); + verify(second).close(); + verify(fullText).close(); + verify(sorted).close(); + } + + @Test + void testVectorFailureRollsBackEarlierIndexAndAllowsRetry() throws Exception { + DataFileMeta source = dataFile("data-1", 3); + IndexFileMeta firstPayload = payload("first", source, 3, 2, "vector"); + IndexFileMeta secondPayload = payload("second", source, 3, 1, "vector"); + PkVectorAnnSegmentFile firstFile = mock(PkVectorAnnSegmentFile.class); + PkVectorAnnSegmentFile secondFile = mock(PkVectorAnnSegmentFile.class); + when(firstFile.build(anyList(), any(), any(), eq("l2"), eq("vector"))) + .thenReturn(firstPayload); + when(secondFile.build(anyList(), any(), any(), eq("l2"), eq("vector"))) + .thenThrow(new IllegalStateException("second vector failed")) + .thenReturn(secondPayload); + BucketedVectorIndexMaintainer first = vectorMaintainer(2, firstFile); + BucketedVectorIndexMaintainer second = vectorMaintainer(1, secondFile); + BucketedPrimaryKeyIndexMaintainer maintainer = + BucketedPrimaryKeyIndexMaintainer.ofVectors( + Arrays.asList(first, second), null, Collections.emptyList()); + DataIncrement append = DataIncrement.emptyIncrement(); + CompactIncrement failed = compactAfter(source); + + assertThatThrownBy(() -> maintainer.prepareCommit(append, failed, true)) + .hasMessage("second vector failed"); + assertThat(append.newIndexFiles()).isEmpty(); + assertThat(append.deletedIndexFiles()).isEmpty(); + assertThat(failed.newIndexFiles()).isEmpty(); + assertThat(failed.deletedIndexFiles()).isEmpty(); + assertThat(first.segments()).isEmpty(); + assertThat(second.segments()).isEmpty(); + assertThat(maintainer.buildNotCompleted()).isFalse(); + verify(firstFile).delete(firstPayload); + + CompactIncrement retry = compactAfter(source); + maintainer.prepareCommit(DataIncrement.emptyIncrement(), retry, true); + + assertThat(retry.newIndexFiles()).containsExactly(firstPayload, secondPayload); + assertThat(first.segments()).containsExactly(firstPayload); + assertThat(second.segments()).containsExactly(secondPayload); + maintainer.close(); + } + + @Test + void testFullTextFailureAbortsAllPreparedVectorsInReverseOrder() throws Exception { + BucketedVectorIndexMaintainer first = mock(BucketedVectorIndexMaintainer.class); + BucketedVectorIndexMaintainer second = mock(BucketedVectorIndexMaintainer.class); + BucketedVectorIndexMaintainer.VectorIndexCommit firstCommit = + mock(BucketedVectorIndexMaintainer.VectorIndexCommit.class); + BucketedVectorIndexMaintainer.VectorIndexCommit secondCommit = + mock(BucketedVectorIndexMaintainer.VectorIndexCommit.class); + when(first.prepareCommit(any(), any(), eq(true))).thenReturn(firstCommit); + when(second.prepareCommit(any(), any(), eq(true))).thenReturn(secondCommit); + BucketedFullTextIndexMaintainer fullText = mock(BucketedFullTextIndexMaintainer.class); + IllegalStateException failure = new IllegalStateException("full-text failed"); + when(fullText.prepareCommit(any(), any(), eq(true))).thenThrow(failure); + BucketedPrimaryKeyIndexMaintainer maintainer = + BucketedPrimaryKeyIndexMaintainer.ofVectors( + Arrays.asList(first, second), fullText, Collections.emptyList()); + DataIncrement append = DataIncrement.emptyIncrement(); + CompactIncrement compact = CompactIncrement.emptyIncrement(); + + assertThatThrownBy(() -> maintainer.prepareCommit(append, compact, true)).isSameAs(failure); + InOrder order = inOrder(first, second, fullText, firstCommit, secondCommit); + order.verify(first).prepareCommit(append, compact, true); + order.verify(second).prepareCommit(append, compact, true); + order.verify(fullText).prepareCommit(append, compact, true); + order.verify(secondCommit).abort(failure); + order.verify(firstCommit).abort(failure); + verify(firstCommit, never()).appendIncrement(); + verify(secondCommit, never()).appendIncrement(); + assertThat(append.newIndexFiles()).isEmpty(); + assertThat(compact.newIndexFiles()).isEmpty(); } @Test @@ -119,6 +229,7 @@ void testDelegatesFullTextLifecycleAndMergesCommit() throws Exception { assertThat(compactIncrement.newIndexFiles()).containsExactly(payload); assertThat(maintainer.buildNotCompleted()).isTrue(); + when(fullText.buildNotCompleted()).thenReturn(false); maintainer.withExecutor(buildExecutor); maintainer.close(); verify(fullText).withExecutor(buildExecutor); @@ -399,6 +510,28 @@ void testBlockingPrepareRollsBackAllDefinitionsAfterInterruption() throws Except .containsExactly("btree-2", "bitmap-2"); } + private BucketedVectorIndexMaintainer vectorMaintainer( + int fieldId, PkVectorAnnSegmentFile indexFile) { + IndexFileHandler handler = mock(IndexFileHandler.class); + when(handler.pkVectorAnnSegment(any(), eq(0))).thenReturn(indexFile); + return new BucketedVectorIndexMaintainer.Factory( + handler, + mock(KeyValueFileReaderFactory.Builder.class, RETURNS_SELF), + new DataField( + fieldId, + "vector_" + fieldId, + DataTypes.VECTOR(2, DataTypes.FLOAT())), + "l2", + "vector") + .withIndexOptions(new Options()) + .create( + BinaryRow.EMPTY_ROW, + 0, + Collections.emptyList(), + Collections.emptyList(), + buildExecutor); + } + private BucketedSortedIndexMaintainer sortedMaintainer( int fieldId, String indexType, diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java index 1c5e4587ffc4..994d59ef817f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java @@ -138,9 +138,10 @@ void testCreatesCoordinatorForMultipleScalarDefinitionsAndVector() throws Except AbstractFileStoreWrite.WriterContainer container = write.createWriterContainer(generator.getPartition(record), 1); - Object vectorMaintainer = - readField(container.primaryKeyIndexMaintainer, "vectorMaintainer"); - assertThat(vectorMaintainer).isNotNull(); + List vectorMaintainers = + (List) readField(container.primaryKeyIndexMaintainer, "vectorMaintainers"); + assertThat(vectorMaintainers).hasSize(1); + Object vectorMaintainer = vectorMaintainers.get(0); assertThat((List) readField(container.primaryKeyIndexMaintainer, "sortedMaintainers")) .hasSize(4); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java index 9b2aa3dcb1eb..34e6cbad340e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java @@ -37,6 +37,7 @@ import javax.annotation.Nullable; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -71,15 +72,19 @@ void testRestoreRebindsVectorIndexExecutor() throws Exception { BucketedVectorIndexMaintainer maintainer = mock(BucketedVectorIndexMaintainer.class); when(maintainer.prepareCommit(same(dataIncrement), same(compactIncrement), eq(true))) .thenReturn(vectorCommit); + BucketedVectorIndexMaintainer secondMaintainer = mock(BucketedVectorIndexMaintainer.class); + when(secondMaintainer.prepareCommit(same(dataIncrement), same(compactIncrement), eq(true))) + .thenReturn(vectorCommit); TestingFileStoreWrite write = new TestingFileStoreWrite(); - write.install(writer, maintainer); + write.install(writer, maintainer, secondMaintainer); List> states = write.checkpoint(); TestingFileStoreWrite restored = new TestingFileStoreWrite(mock(RecordWriter.class)); restored.restore(states); verify(maintainer).withExecutor(any(ExecutorService.class)); + verify(secondMaintainer).withExecutor(any(ExecutorService.class)); } @Test @@ -160,7 +165,7 @@ private TestingFileStoreWrite(@Nullable RecordWriter restoredWriter) { } private void install( - RecordWriter writer, BucketedVectorIndexMaintainer vectorMaintainer) { + RecordWriter writer, BucketedVectorIndexMaintainer... vectorMaintainers) { writers.computeIfAbsent(EMPTY_ROW, ignored -> new HashMap<>()) .put( 0, @@ -169,7 +174,10 @@ private void install( 1, null, null, - BucketedPrimaryKeyIndexMaintainer.ofVector(vectorMaintainer), + BucketedPrimaryKeyIndexMaintainer.ofVectors( + Arrays.asList(vectorMaintainers), + null, + Collections.emptyList()), null)); }