leaves12138 commented on code in PR #8602:
URL: https://github.com/apache/paimon/pull/8602#discussion_r3571157055
##########
paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java:
##########
@@ -133,7 +318,70 @@ public BucketedPrimaryKeyIndexMaintainer create(
@Nullable List<IndexFileMeta> restoredPayloads,
ExecutorService executor,
@Nullable IOManager ioManager) {
- return create(partition, bucket, restoredDataFiles,
restoredPayloads, executor);
+ List<DataFileMeta> dataFiles =
+ 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);
Review Comment:
`restoredPayloads` is the full result of `scanSourceIndexes`, so it contains
the payloads for every configured source-backed definition. Passing that
unfiltered list into the vector factory makes `PkVectorBucketIndexState`
validate BTree/Bitmap payloads as vector segments and throw `Vector payload ...
has a different index type`. A table configured with both a vector index and a
scalar index therefore cannot restore a bucket writer after scalar payloads
have been committed. Please filter the restored payloads by the vector
field/index type before creating the vector maintainer, and add a mixed vector
+ scalar write/restore test.
##########
paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java:
##########
@@ -18,53 +18,172 @@
package org.apache.paimon.index.pk;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pksorted.BucketedSortedIndexMaintainer;
+import org.apache.paimon.index.pksorted.PkSortedDataFileReader;
+import org.apache.paimon.index.pksorted.PkSortedIndexBuilder;
+import org.apache.paimon.index.pksorted.PkSortedIndexFile;
import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
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.schema.TableSchema;
+import org.apache.paimon.types.DataField;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutorService;
-/** Coordinates source-backed primary-key indexes for one bucket. */
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Coordinates all source-backed primary-key indexes for one bucket. */
public final class BucketedPrimaryKeyIndexMaintainer {
- private final BucketedVectorIndexMaintainer vectorMaintainer;
+ @Nullable private final BucketedVectorIndexMaintainer vectorMaintainer;
+ private final List<BucketedSortedIndexMaintainer> sortedMaintainers;
- private BucketedPrimaryKeyIndexMaintainer(BucketedVectorIndexMaintainer
vectorMaintainer) {
+ private BucketedPrimaryKeyIndexMaintainer(
+ @Nullable BucketedVectorIndexMaintainer vectorMaintainer,
+ List<BucketedSortedIndexMaintainer> sortedMaintainers) {
this.vectorMaintainer = vectorMaintainer;
+ List<BucketedSortedIndexMaintainer> sorted = new
ArrayList<>(sortedMaintainers);
+
sorted.sort(Comparator.comparingInt(BucketedSortedIndexMaintainer::fieldId));
+ this.sortedMaintainers = Collections.unmodifiableList(sorted);
}
public static BucketedPrimaryKeyIndexMaintainer ofVector(
BucketedVectorIndexMaintainer vectorMaintainer) {
- return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer);
+ return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer,
Collections.emptyList());
+ }
+
+ public static BucketedPrimaryKeyIndexMaintainer of(
+ BucketedVectorIndexMaintainer vectorMaintainer,
+ List<BucketedSortedIndexMaintainer> sortedMaintainers) {
+ return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer,
sortedMaintainers);
+ }
+
+ public static BucketedPrimaryKeyIndexMaintainer ofSorted(
+ List<BucketedSortedIndexMaintainer> sortedMaintainers) {
+ return new BucketedPrimaryKeyIndexMaintainer(null, sortedMaintainers);
}
public void prepareCommit(
DataIncrement appendIncrement,
CompactIncrement compactIncrement,
boolean waitCompaction)
throws Exception {
- BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
- vectorMaintainer.prepareCommit(appendIncrement,
compactIncrement, waitCompaction);
- vectorCommit
- .appendIncrement()
+ if (vectorMaintainer != null) {
+ BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
+ vectorMaintainer.prepareCommit(
+ appendIncrement, compactIncrement, waitCompaction);
+ vectorCommit
+ .appendIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ appendIncrement.newIndexFiles(),
+
appendIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ vectorCommit
+ .compactIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ compactIncrement.newIndexFiles(),
+
compactIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ }
+
+ if (waitCompaction) {
+ prepareSortedBlocking(appendIncrement, compactIncrement);
Review Comment:
The coordinator publishes each maintainer's state and merges its files into
the caller increments before later maintainers have succeeded, but there is no
coordinator-level rollback. If an earlier vector/scalar definition completes
and a later blocking scalar `prepareCommit` is interrupted, the overall call
throws while the earlier definition remains covered and its generated payload
remains in the increment. Retrying or discarding that failed prepare can then
omit the payload or leave it orphaned. Individual maintainers only roll back
their own invocation. Please make this operation atomic across all definitions
(or defer publication until all definitions succeed) and cover an interruption
after the first of two scalar definitions completes.
##########
paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java:
##########
@@ -0,0 +1,281 @@
+/*
+ * 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.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+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.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link BucketedPrimaryKeyIndexMaintainer}. */
+class BucketedPrimaryKeyIndexMaintainerTest {
+
+ @TempDir java.nio.file.Path tempPath;
+ private final ExecutorService buildExecutor =
Executors.newFixedThreadPool(2);
+
+ @AfterEach
+ void shutdownExecutor() {
+ buildExecutor.shutdownNow();
+ }
+
+ @Test
+ void testDelegatesVectorLifecycle() {
+ BucketedVectorIndexMaintainer vector =
mock(BucketedVectorIndexMaintainer.class);
+ ExecutorService executor = mock(ExecutorService.class);
+ when(vector.buildNotCompleted()).thenReturn(true);
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ BucketedPrimaryKeyIndexMaintainer.ofVector(vector);
+
+ assertThat(maintainer.buildNotCompleted()).isTrue();
+ maintainer.withExecutor(executor);
+ maintainer.close();
+
+ verify(vector).withExecutor(executor);
+ verify(vector).close();
+ }
+
+ @Test
+ void testScalarFailureDoesNotSuppressOtherDefinitionsOrVector() throws
Exception {
+ DataFileMeta source = dataFile("data-1", 3);
+ IndexFileMeta vectorPayload =
+ new IndexFileMeta("vector", "vector", 1, 3, (GlobalIndexMeta)
null, null);
+ BucketedVectorIndexMaintainer vector =
mock(BucketedVectorIndexMaintainer.class);
+ BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
+ mock(BucketedVectorIndexMaintainer.VectorIndexCommit.class);
+ BucketedVectorIndexMaintainer.VectorIndexIncrement vectorIncrement =
+ mock(BucketedVectorIndexMaintainer.VectorIndexIncrement.class);
+ when(vector.prepareCommit(any(), any(),
eq(true))).thenReturn(vectorCommit);
+ when(vectorCommit.appendIncrement()).thenReturn(Optional.empty());
+
when(vectorCommit.compactIncrement()).thenReturn(Optional.of(vectorIncrement));
+
when(vectorIncrement.newIndexFiles()).thenReturn(Collections.singletonList(vectorPayload));
+
when(vectorIncrement.deletedIndexFiles()).thenReturn(Collections.emptyList());
+
+ List<String> buildOrder = Collections.synchronizedList(new
ArrayList<>());
+ BucketedSortedIndexMaintainer btree =
+ sortedMaintainer(
+ 7,
+ "btree",
+ source,
+ dataFile -> {
+ buildOrder.add("btree");
+ throw new IllegalStateException("expected BTree
failure");
+ });
+ IndexFileMeta bitmapPayload = payload("bitmap", source, 3, 8,
"bitmap");
+ BucketedSortedIndexMaintainer bitmap =
+ sortedMaintainer(
+ 8,
+ "bitmap",
+ source,
+ dataFile -> {
+ buildOrder.add("bitmap");
+ return Collections.singletonList(bitmapPayload);
+ });
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ BucketedPrimaryKeyIndexMaintainer.of(vector,
Arrays.asList(bitmap, btree));
+ CompactIncrement compactIncrement = compactAfter(source);
+
+ maintainer.prepareCommit(DataIncrement.emptyIncrement(),
compactIncrement, true);
+
+ assertThat(buildOrder).containsExactly("btree", "btree", "btree",
"bitmap");
+
assertThat(compactIncrement.newIndexFiles()).containsExactly(vectorPayload,
bitmapPayload);
+ assertThat(compactIncrement.deletedIndexFiles()).isEmpty();
+ assertThat(maintainer.buildNotCompleted()).isFalse();
+ }
+
+ @Test
+ void testScalarDefinitionsNeverBuildConcurrently() throws Exception {
+ DataFileMeta source = dataFile("data-1", 3);
+ AtomicInteger activeBuilds = new AtomicInteger();
+ AtomicInteger peakBuilds = new AtomicInteger();
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+ CountDownLatch firstCompleted = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ BucketedSortedIndexMaintainer first =
+ sortedMaintainer(
+ 7,
+ "btree",
+ source,
+ dataFile -> {
+ enterBuild(activeBuilds, peakBuilds);
+ firstStarted.countDown();
+ releaseFirst.await();
+ activeBuilds.decrementAndGet();
+ firstCompleted.countDown();
+ return Collections.singletonList(
+ payload("btree", source, 3, 7, "btree"));
+ });
+ BucketedSortedIndexMaintainer second =
+ sortedMaintainer(
+ 8,
+ "bitmap",
+ source,
+ dataFile -> {
+ enterBuild(activeBuilds, peakBuilds);
+ secondStarted.countDown();
+ activeBuilds.decrementAndGet();
+ return Collections.singletonList(
+ payload("bitmap", source, 3, 8, "bitmap"));
+ });
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+
BucketedPrimaryKeyIndexMaintainer.ofSorted(Arrays.asList(second, first));
+
+ maintainer.prepareCommit(DataIncrement.emptyIncrement(),
compactAfter(source), false);
+ assertThat(firstStarted.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(secondStarted.getCount()).isEqualTo(1);
+ assertThat(maintainer.buildNotCompleted()).isTrue();
+
+ releaseFirst.countDown();
+ assertThat(firstCompleted.await(5, TimeUnit.SECONDS)).isTrue();
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
CompactIncrement.emptyIncrement(), false);
+ assertThat(secondStarted.await(5, TimeUnit.SECONDS)).isTrue();
Review Comment:
This test is racy and fails consistently on my machine. `firstCompleted` is
counted down inside the callable immediately before it returns, so the
following non-blocking `prepareCommit` may still observe `Future.isDone() ==
false` and return without starting the second definition; the await here then
times out. Please wait/poll until the first future is actually consumed (for
example, retry non-blocking prepare until the second build starts) instead of
treating the in-callable latch as future completion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
