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 b1a68a57d2 [core] Wait for global index in visibility callback (#8265)
b1a68a57d2 is described below
commit b1a68a57d2e28374f84fc0f52e3b7be86f19c7ff
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jun 17 20:09:44 2026 +0800
[core] Wait for global index in visibility callback (#8265)
Support waiting for newly committed row-tracking data to be covered by
existing global indexes when `visibility-callback.enabled` is enabled.
This extends the visibility callback beyond compaction so batch writes
do not return before global index visibility catches up.
---
docs/generated/core_configuration.html | 4 +-
.../main/java/org/apache/paimon/CoreOptions.java | 12 +-
.../java/org/apache/paimon/AbstractFileStore.java | 19 +-
.../metastore/GlobalIndexVisibilityChecker.java | 222 +++++++++++++++++++++
.../paimon/metastore/VisibilityWaitCallback.java | 44 +++-
.../metastore/VisibilityWaitCallbackTest.java | 214 ++++++++++++++++++++
6 files changed, 494 insertions(+), 21 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 45e47fdf23..c17b6a5007 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1678,13 +1678,13 @@ If the data size allocated for the sorting task is
uneven,which may lead to perf
<td><h5>visibility-callback.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Whether to enable the visibility wait callback that waits for
compaction to complete after commit. This is useful for primary key tables with
deletion vectors or postpone bucket mode to ensure data visibility, only used
for batch mode or bounded stream.</td>
+ <td>Whether to enable the visibility wait callback that waits for
compaction or global index build to complete after commit. This is useful for
primary key tables with deletion vectors or postpone bucket mode and
row-tracking tables with global indexes to ensure data visibility, only used
for batch mode or bounded stream.</td>
</tr>
<tr>
<td><h5>visibility-callback.timeout</h5></td>
<td style="word-wrap: break-word;">30 min</td>
<td>Duration</td>
- <td>The maximum time to wait for compaction to complete when
visibility callback is enabled. If the timeout is reached, an exception will be
thrown.</td>
+ <td>The maximum time to wait for compaction or global index build
to complete when visibility callback is enabled. If the timeout is reached, an
exception will be thrown.</td>
</tr>
<tr>
<td><h5>write-buffer-for-append</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 4a4b7b5f81..81bf1fe0e0 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2554,17 +2554,19 @@ public class CoreOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether to enable the visibility wait callback
that waits for compaction to complete "
- + "after commit. This is useful for
primary key tables with deletion vectors or "
- + "postpone bucket mode to ensure data
visibility, only used for batch mode or bounded stream.");
+ "Whether to enable the visibility wait callback
that waits for compaction or global "
+ + "index build to complete after commit.
This is useful for primary key tables "
+ + "with deletion vectors or postpone
bucket mode and row-tracking tables with "
+ + "global indexes to ensure data
visibility, only used for batch mode or bounded stream.");
public static final ConfigOption<Duration> VISIBILITY_CALLBACK_TIMEOUT =
key("visibility-callback.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(30))
.withDescription(
- "The maximum time to wait for compaction to
complete when visibility callback is enabled. "
- + "If the timeout is reached, an exception
will be thrown.");
+ "The maximum time to wait for compaction or global
index build to complete when "
+ + "visibility callback is enabled. If the
timeout is reached, an exception will "
+ + "be thrown.");
public static final ConfigOption<Duration>
VISIBILITY_CALLBACK_CHECK_INTERVAL =
key("visibility-callback.check-interval")
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index cc298823ff..524dd4962c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -434,17 +434,26 @@ abstract class AbstractFileStore<T> implements
FileStore<T> {
callbacks.add(new ChainTableOverwriteCommitCallback(table));
}
- if (options.visibilityCallbackEnabled() &&
!schema.primaryKeys().isEmpty()) {
- if (table.bucketMode() == BucketMode.POSTPONE_MODE
- || options.deletionVectorsEnabled()) {
- callbacks.add(new VisibilityWaitCallback(table));
- }
+ if (options.visibilityCallbackEnabled() &&
shouldWaitForVisibility(table)) {
+ callbacks.add(new VisibilityWaitCallback(table));
}
callbacks.addAll(CallbackUtils.loadCommitCallbacks(options, table));
return callbacks;
}
+ private boolean shouldWaitForVisibility(FileStoreTable table) {
+ if (options.rowTrackingEnabled()) {
+ return true;
+ }
+
+ if (schema.primaryKeys().isEmpty()) {
+ return false;
+ }
+
+ return table.bucketMode() == BucketMode.POSTPONE_MODE ||
options.deletionVectorsEnabled();
+ }
+
@Override
@Nullable
public PartitionExpire newPartitionExpire(String commitUser,
FileStoreTable table) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/metastore/GlobalIndexVisibilityChecker.java
b/paimon-core/src/main/java/org/apache/paimon/metastore/GlobalIndexVisibilityChecker.java
new file mode 100644
index 0000000000..596bf15e91
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/metastore/GlobalIndexVisibilityChecker.java
@@ -0,0 +1,222 @@
+/*
+ * 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.metastore;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.Range;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+
+/** Checks whether newly added data files are covered by existing global
indexes. */
+class GlobalIndexVisibilityChecker {
+
+ private final FileStoreTable table;
+ private final Map<BinaryRow, List<Range>> rowIdRangesByPartition;
+ private final Map<BinaryRow, Set<GlobalIndexIdentifier>>
globalIndexesByPartition;
+
+ private GlobalIndexVisibilityChecker(
+ FileStoreTable table,
+ Map<BinaryRow, List<Range>> rowIdRangesByPartition,
+ Map<BinaryRow, Set<GlobalIndexIdentifier>>
globalIndexesByPartition) {
+ this.table = table;
+ this.rowIdRangesByPartition = rowIdRangesByPartition;
+ this.globalIndexesByPartition = globalIndexesByPartition;
+ }
+
+ static GlobalIndexVisibilityChecker create(
+ FileStoreTable table, Snapshot snapshot, List<ManifestEntry>
deltaFiles) {
+ Map<BinaryRow, List<Range>> rowIdRangesByPartition =
+ collectRowIdRangesByPartition(deltaFiles);
+ Map<BinaryRow, Set<GlobalIndexIdentifier>> globalIndexesByPartition =
+ rowIdRangesByPartition.isEmpty()
+ ? new HashMap<>()
+ : collectGlobalIndexesByPartition(
+ table, snapshot,
rowIdRangesByPartition.keySet());
+ return new GlobalIndexVisibilityChecker(
+ table, rowIdRangesByPartition, globalIndexesByPartition);
+ }
+
+ boolean noNeedToWait() {
+ return globalIndexesByPartition.isEmpty();
+ }
+
+ boolean visibleIn(Snapshot snapshot) {
+ Map<BinaryRow, Map<GlobalIndexIdentifier, List<Range>>>
indexedRangesByPartition =
+ new HashMap<>();
+ for (IndexManifestEntry entry : scanGlobalIndexes(table, snapshot)) {
+ GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
+ if (globalIndex == null) {
+ continue;
+ }
+
+ Set<GlobalIndexIdentifier> identifiers =
+ globalIndexesByPartition.get(entry.partition());
+ if (identifiers == null) {
+ continue;
+ }
+
+ GlobalIndexIdentifier identifier = identifierOf(entry);
+ if (identifiers.contains(identifier)) {
+ indexedRangesByPartition
+ .computeIfAbsent(entry.partition().copy(), k -> new
HashMap<>())
+ .computeIfAbsent(identifier, k -> new ArrayList<>())
+ .add(globalIndex.rowRange());
+ }
+ }
+
+ for (Map.Entry<BinaryRow, Set<GlobalIndexIdentifier>> partitionIndexes
:
+ globalIndexesByPartition.entrySet()) {
+ BinaryRow partition = partitionIndexes.getKey();
+ List<Range> rowIdRanges = rowIdRangesByPartition.get(partition);
+ Map<GlobalIndexIdentifier, List<Range>> indexedRanges =
+ indexedRangesByPartition.get(partition);
+ for (GlobalIndexIdentifier identifier :
partitionIndexes.getValue()) {
+ List<Range> ranges =
+ Range.sortAndMergeOverlap(
+ indexedRanges == null ? null :
indexedRanges.get(identifier), true);
+ for (Range rowIdRange : rowIdRanges) {
+ if (!rowIdRange.exclude(ranges).isEmpty()) {
+ return false;
+ }
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private static Map<BinaryRow, List<Range>> collectRowIdRangesByPartition(
+ List<ManifestEntry> deltaFiles) {
+ Map<BinaryRow, List<Range>> rangesByPartition = new HashMap<>();
+ for (ManifestEntry entry : deltaFiles) {
+ if (shouldTrackGlobalIndex(entry)) {
+ rangesByPartition
+ .computeIfAbsent(entry.partition().copy(), k -> new
ArrayList<>())
+ .add(entry.file().nonNullRowIdRange());
+ }
+ }
+ for (Map.Entry<BinaryRow, List<Range>> entry :
rangesByPartition.entrySet()) {
+ entry.setValue(Range.sortAndMergeOverlap(entry.getValue(), true));
+ }
+ return rangesByPartition;
+ }
+
+ private static boolean shouldTrackGlobalIndex(ManifestEntry entry) {
+ if (!FileKind.ADD.equals(entry.kind())) {
+ return false;
+ }
+
+ DataFileMeta file = entry.file();
+ if (file.firstRowId() == null || file.rowCount() <= 0) {
+ return false;
+ }
+
+ Optional<FileSource> fileSource = file.fileSource();
+ if (!fileSource.isPresent() ||
!FileSource.APPEND.equals(fileSource.get())) {
+ return false;
+ }
+
+ return !isBlobFile(file.fileName());
+ }
+
+ private static Map<BinaryRow, Set<GlobalIndexIdentifier>>
collectGlobalIndexesByPartition(
+ FileStoreTable table, Snapshot snapshot, Set<BinaryRow>
partitionsToTrack) {
+ Map<BinaryRow, Set<GlobalIndexIdentifier>> indexesByPartition = new
HashMap<>();
+ for (IndexManifestEntry entry : scanGlobalIndexes(table, snapshot)) {
+ GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
+ if (globalIndex != null &&
partitionsToTrack.contains(entry.partition())) {
+ indexesByPartition
+ .computeIfAbsent(entry.partition().copy(), k -> new
HashSet<>())
+ .add(identifierOf(entry));
+ }
+ }
+ return indexesByPartition;
+ }
+
+ private static List<IndexManifestEntry> scanGlobalIndexes(
+ FileStoreTable table, Snapshot snapshot) {
+ return table.store()
+ .newIndexFileHandler()
+ .scan(snapshot, entry -> entry.indexFile().globalIndexMeta()
!= null);
+ }
+
+ private static GlobalIndexIdentifier identifierOf(IndexManifestEntry
entry) {
+ GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
+ return new GlobalIndexIdentifier(
+ entry.indexFile().indexType(),
+ globalIndex.indexFieldId(),
+ globalIndex.extraFieldIds());
+ }
+
+ private static class GlobalIndexIdentifier {
+
+ private final String indexType;
+ private final int indexFieldId;
+ private final int[] extraFieldIds;
+
+ private GlobalIndexIdentifier(String indexType, int indexFieldId,
int[] extraFieldIds) {
+ this.indexType = indexType;
+ this.indexFieldId = indexFieldId;
+ this.extraFieldIds =
+ extraFieldIds == null
+ ? null
+ : Arrays.copyOf(extraFieldIds,
extraFieldIds.length);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof GlobalIndexIdentifier)) {
+ return false;
+ }
+
+ GlobalIndexIdentifier that = (GlobalIndexIdentifier) o;
+ return indexFieldId == that.indexFieldId
+ && Objects.equals(indexType, that.indexType)
+ && Arrays.equals(extraFieldIds, that.extraFieldIds);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Objects.hash(indexType, indexFieldId);
+ result = 31 * result + Arrays.hashCode(extraFieldIds);
+ return result;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/metastore/VisibilityWaitCallback.java
b/paimon-core/src/main/java/org/apache/paimon/metastore/VisibilityWaitCallback.java
index 824c232367..562cd55a40 100644
---
a/paimon-core/src/main/java/org/apache/paimon/metastore/VisibilityWaitCallback.java
+++
b/paimon-core/src/main/java/org/apache/paimon/metastore/VisibilityWaitCallback.java
@@ -41,7 +41,7 @@ import java.util.concurrent.TimeoutException;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
-/** A {@link CommitCallback} to wait for compaction for visibility. */
+/** A {@link CommitCallback} to wait for compaction and global indexes for
visibility. */
public class VisibilityWaitCallback implements CommitCallback {
private static final Logger LOG =
LoggerFactory.getLogger(VisibilityWaitCallback.class);
@@ -68,6 +68,8 @@ public class VisibilityWaitCallback implements CommitCallback
{
Set<String> namesToTrack = new HashSet<>();
Set<BinaryRow> partitionsToTrack = new HashSet<>();
+ GlobalIndexVisibilityChecker globalIndexVisibilityChecker =
+ GlobalIndexVisibilityChecker.create(table, context.snapshot,
context.deltaFiles);
for (ManifestEntry entry : context.deltaFiles) {
if (shouldBeTracked(entry)) {
namesToTrack.add(entry.fileName());
@@ -75,12 +77,16 @@ public class VisibilityWaitCallback implements
CommitCallback {
}
}
- if (namesToTrack.isEmpty()) {
+ if (namesToTrack.isEmpty() &&
globalIndexVisibilityChecker.noNeedToWait()) {
return;
}
try {
- waitForCompaction(context.snapshot, namesToTrack,
partitionsToTrack);
+ waitForVisibility(
+ context.snapshot,
+ namesToTrack,
+ partitionsToTrack,
+ globalIndexVisibilityChecker);
} catch (InterruptedException | TimeoutException e) {
throw new RuntimeException(e);
}
@@ -91,25 +97,45 @@ public class VisibilityWaitCallback implements
CommitCallback {
// No-op for retry as the callback is idempotent
}
- private void waitForCompaction(
- Snapshot fromSnapshot, Set<String> namesToTrack, Set<BinaryRow>
partitionsToTrack)
+ private void waitForVisibility(
+ Snapshot fromSnapshot,
+ Set<String> namesToTrack,
+ Set<BinaryRow> partitionsToTrack,
+ GlobalIndexVisibilityChecker globalIndexVisibilityChecker)
throws InterruptedException, TimeoutException {
long startTime = System.currentTimeMillis();
+ boolean compactionDone = namesToTrack.isEmpty();
+ boolean globalIndexDone = globalIndexVisibilityChecker.noNeedToWait();
+ Snapshot checkedSnapshot = fromSnapshot;
while (System.currentTimeMillis() - startTime < timeout.toMillis()) {
Snapshot latest = table.snapshotManager().latestSnapshot();
checkNotNull(latest, "No latest snapshot");
- if (latest.id() > fromSnapshot.id()
+
+ if (!compactionDone
+ && latest.id() > checkedSnapshot.id()
&& !stillInLatest(latest, namesToTrack,
partitionsToTrack)) {
+ compactionDone = true;
+ }
+ if (!globalIndexDone &&
globalIndexVisibilityChecker.visibleIn(latest)) {
+ globalIndexDone = true;
+ }
+ if (compactionDone && globalIndexDone) {
return;
}
- fromSnapshot = latest;
+ checkedSnapshot = latest;
- LOG.info("Waiting for files of table {} to be compacted...",
table.fullName());
+ LOG.info(
+ "Waiting for visibility of table {}. Compaction done: {},
global index done: {}.",
+ table.fullName(),
+ compactionDone,
+ globalIndexDone);
//noinspection BusyWait
Thread.sleep(checkInterval.toMillis());
}
- throw new TimeoutException("Timeout waiting for files to be compacted
after " + timeout);
+ throw new TimeoutException(
+ "Timeout waiting for files to be compacted or global indexes
to be built after "
+ + timeout);
}
private boolean stillInLatest(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
new file mode 100644
index 0000000000..030b2af41c
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/metastore/VisibilityWaitCallbackTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.metastore;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.RowRangeIndex;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link VisibilityWaitCallback}. */
+public class VisibilityWaitCallbackTest extends TableTestBase {
+
+ @Override
+ protected Schema schemaDefault() {
+ return Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.STRING())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.VISIBILITY_CALLBACK_ENABLED.key(), "true")
+ .option(CoreOptions.VISIBILITY_CALLBACK_CHECK_INTERVAL.key(),
"100 ms")
+ .option(CoreOptions.VISIBILITY_CALLBACK_TIMEOUT.key(), "30 s")
+ .build();
+ }
+
+ @Test
+ public void testWaitForGlobalIndexBuildOfNewData() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeRows(table, 0, 3);
+ buildIndex(table, false);
+
+ long indexedSnapshotId = table.snapshotManager().latestSnapshot().id();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CompletableFuture<Void> writeFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ writeRows(getTableDefault(), 3, 2);
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ },
+ executor);
+
+ try {
+ waitUntil(() -> table.snapshotManager().latestSnapshot().id() >
indexedSnapshotId);
+ Thread.sleep(300L);
+ assertThat(writeFuture.isDone()).isFalse();
+
+ buildIndex(getTableDefault(), true);
+ writeFuture.get(10, TimeUnit.SECONDS);
+
+ BTreeGlobalIndexBuilder builder =
+ new
BTreeGlobalIndexBuilder(getTableDefault()).withIndexField("f1");
+ assertThat(builder.incrementalScan()).isNotPresent();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testDoNotWaitForGlobalIndexBuildOfUnindexedPartition() throws
Exception {
+ Identifier identifier = identifier("PartitionedTable");
+ catalog.createTable(identifier, partitionedSchema(), false);
+ FileStoreTable table = getTable(identifier);
+
+ writePartitionRows(table, "a", 0, 3);
+ buildPartitionIndex(table, "a");
+
+ writePartitionRows(getTable(identifier), "b", 3, 2);
+ }
+
+ private Schema partitionedSchema() {
+ return Schema.newBuilder()
+ .column("pt", DataTypes.STRING())
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.STRING())
+ .partitionKeys("pt")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .option(CoreOptions.VISIBILITY_CALLBACK_ENABLED.key(), "true")
+ .option(CoreOptions.VISIBILITY_CALLBACK_CHECK_INTERVAL.key(),
"100 ms")
+ .option(CoreOptions.VISIBILITY_CALLBACK_TIMEOUT.key(), "1 s")
+ .build();
+ }
+
+ private void writeRows(FileStoreTable table, int start, int count) throws
Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = start; i < start + count; i++) {
+ write.write(GenericRow.of(i, BinaryString.fromString("a" +
i)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void writePartitionRows(FileStoreTable table, String partition,
int start, int count)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = start; i < start + count; i++) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(partition),
+ i,
+ BinaryString.fromString("a" + i)));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void buildIndex(FileStoreTable table, boolean incremental) throws
Exception {
+ BTreeGlobalIndexBuilder builder = new
BTreeGlobalIndexBuilder(table).withIndexField("f1");
+ Optional<Pair<RowRangeIndex, List<DataSplit>>> scan =
+ incremental ? builder.incrementalScan() : builder.scan();
+ assertThat(scan).isPresent();
+
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (DataSplit dataSplit : scan.get().getRight()) {
+ commitMessages.addAll(builder.build(dataSplit, ioManager));
+ }
+
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(commitMessages);
+ }
+ }
+
+ private void buildPartitionIndex(FileStoreTable table, String partition)
throws Exception {
+ BTreeGlobalIndexBuilder builder =
+ new BTreeGlobalIndexBuilder(table)
+ .withIndexField("f1")
+ .withPartitionPredicate(partitionPredicate(table,
partition));
+ Optional<Pair<RowRangeIndex, List<DataSplit>>> scan = builder.scan();
+ assertThat(scan).isPresent();
+
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (DataSplit dataSplit : scan.get().getRight()) {
+ commitMessages.addAll(builder.build(dataSplit, ioManager));
+ }
+
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(commitMessages);
+ }
+ }
+
+ private PartitionPredicate partitionPredicate(FileStoreTable table, String
partition) {
+ RowType partType = table.rowType().project("pt");
+ Predicate predicate =
+ PartitionPredicate.createPartitionPredicate(
+ partType,
+ Collections.singletonMap("pt",
BinaryString.fromString(partition)));
+ return PartitionPredicate.fromPredicate(partType, predicate);
+ }
+
+ private void waitUntil(BooleanSupplier condition) throws Exception {
+ long deadline = System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(10);
+ while (System.currentTimeMillis() < deadline) {
+ if (condition.getAsBoolean()) {
+ return;
+ }
+ Thread.sleep(50L);
+ }
+ throw new AssertionError("Condition was not met before timeout.");
+ }
+}
