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 5c82868d50 [core] Advance index builds across row ID reassignment
(#10848)
5c82868d50 is described below
commit 5c82868d506d233ef922135fc618f8e684d43c80
Author: YeJunHao <[email protected]>
AuthorDate: Thu Oct 29 15:14:18 2026 +0800
[core] Advance index builds across row ID reassignment (#10848)
---
.../paimon/operation/FileStoreCommitImpl.java | 32 +-
...ider.java => DataEvolutionChangesProvider.java} | 6 +-
.../paimon/operation/commit/StrictModeChecker.java | 3 +-
.../dataevolution/CompactReassignReuseTest.java | 330 +++++++++++++++++++--
.../paimon/operation/FileStoreCommitTest.java | 6 +-
5 files changed, 332 insertions(+), 45 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index a460a5234a..3354d4ab1a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -50,8 +50,8 @@ import org.apache.paimon.operation.commit.CommitResult;
import org.apache.paimon.operation.commit.CommitRollback;
import org.apache.paimon.operation.commit.CommitScanner;
import org.apache.paimon.operation.commit.ConflictDetection;
+import org.apache.paimon.operation.commit.DataEvolutionChangesProvider;
import org.apache.paimon.operation.commit.ManifestEntryChanges;
-import org.apache.paimon.operation.commit.ReassignCompactChangesProvider;
import org.apache.paimon.operation.commit.RetryCommitResult;
import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
import
org.apache.paimon.operation.commit.RetryCommitResult.ManifestMergeResult;
@@ -381,10 +381,13 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
attempts +=
tryCommit(
- CommitChangesProvider.provider(
- changes.appendTableFiles,
- changes.appendChangelog,
- changes.appendIndexFiles),
+ dataEvolutionChangesProvider(
+ new CommitChanges(
+ changes.appendTableFiles,
+ changes.appendChangelog,
+ changes.appendIndexFiles),
+ commitKind,
+ Collections.emptySet()),
committable.identifier(),
committable.watermark(),
committable.properties(),
@@ -400,11 +403,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
|| !changes.compactIndexFiles.isEmpty()) {
attempts +=
tryCommit(
- compactChangesProvider(
+ dataEvolutionChangesProvider(
new CommitChanges(
changes.compactTableFiles,
changes.compactChangelog,
changes.compactIndexFiles),
+ CommitKind.COMPACT,
materializedBuckets),
committable.identifier(),
committable.watermark(),
@@ -589,11 +593,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
if (withCompact) {
attempts +=
tryCommit(
- compactChangesProvider(
+ dataEvolutionChangesProvider(
new CommitChanges(
changes.compactTableFiles,
emptyList(),
changes.compactIndexFiles),
+ CommitKind.COMPACT,
Collections.emptySet()),
committable.identifier(),
committable.watermark(),
@@ -858,15 +863,22 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
@VisibleForTesting
- CommitChangesProvider compactChangesProvider(
- CommitChanges changes, Set<Pair<BinaryRow, Integer>>
materializedBuckets) {
+ CommitChangesProvider dataEvolutionChangesProvider(
+ CommitChanges changes,
+ CommitKind commitKind,
+ Set<Pair<BinaryRow, Integer>> materializedBuckets) {
if (materializedBuckets.isEmpty()) {
Long lastSafeSnapshot =
options.commitLastSafeSnapshot().orElse(null);
if (options.dataEvolutionEnabled()
+ && (commitKind == CommitKind.COMPACT
+ || (commitKind == CommitKind.APPEND
+ && changes.tableFiles.isEmpty()
+ && changes.changelogFiles.isEmpty()
+ && !changes.indexFiles.isEmpty()))
&&
!conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT)
&& lastSafeSnapshot != null
&& lastSafeSnapshot >= 0) {
- return new ReassignCompactChangesProvider(
+ return new DataEvolutionChangesProvider(
fileIO, pathFactory, snapshotManager,
lastSafeSnapshot, changes);
}
return CommitChangesProvider.provider(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java
similarity index 97%
rename from
paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java
rename to
paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java
index 88d3dcf906..620326f69e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ReassignCompactChangesProvider.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionChangesProvider.java
@@ -44,8 +44,8 @@ import java.util.Set;
import static org.apache.paimon.utils.Preconditions.checkState;
-/** Reuses immutable compact outputs after verified, metadata-only row-id
reassignments. */
-public final class ReassignCompactChangesProvider implements
CommitChangesProvider {
+/** Reuses immutable outputs after verified, metadata-only row-id
reassignments. */
+public final class DataEvolutionChangesProvider implements
CommitChangesProvider {
private final FileIO fileIO;
private final FileStorePathFactory paths;
@@ -54,7 +54,7 @@ public final class ReassignCompactChangesProvider implements
CommitChangesProvid
private long checkedSnapshot;
private CommitChanges changes;
- public ReassignCompactChangesProvider(
+ public DataEvolutionChangesProvider(
FileIO fileIO,
FileStorePathFactory paths,
SnapshotManager snapshots,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java
index 092c2d331b..cfb700982b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java
@@ -71,7 +71,8 @@ public class StrictModeChecker {
Set<Long> rebasedReassignments) {
Set<BinaryRow> newPartitions = new HashSet<>(newChangedPartitions);
for (long id = lastSafeSnapshot + 1; id < newSnapshotId; id++) {
- if (newCommitKind == CommitKind.COMPACT &&
rebasedReassignments.contains(id)) {
+ if ((newCommitKind == CommitKind.COMPACT || newCommitKind ==
CommitKind.APPEND)
+ && rebasedReassignments.contains(id)) {
continue;
}
Snapshot snapshot = snapshotManager.snapshot(id);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
index 62d5a1e62a..6346f25d03 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
@@ -33,6 +33,7 @@ import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexScanner;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
+import org.apache.paimon.index.DataEvolutionIndexSourceMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
@@ -42,7 +43,7 @@ import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.operation.Lock;
import org.apache.paimon.operation.commit.CommitChanges;
-import org.apache.paimon.operation.commit.ReassignCompactChangesProvider;
+import org.apache.paimon.operation.commit.DataEvolutionChangesProvider;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -82,6 +83,7 @@ import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -158,19 +160,44 @@ public class CompactReassignReuseTest extends
TableTestBase {
return message;
}
- private CommitMessageImpl compactIndex(FileStoreTable table) throws
Exception {
+ private List<CommitMessage> buildIndex(FileStoreTable table, String field,
String partition)
+ throws Exception {
+ return buildIndex(table, "btree", field, partition);
+ }
+
+ private List<CommitMessage> buildIndex(
+ FileStoreTable table, String indexType, String field, String
partition)
+ throws Exception {
+ return buildIndex(table, indexType, Collections.singletonList(field),
partition);
+ }
+
+ private List<CommitMessage> buildIndex(
+ FileStoreTable table, String indexType, List<String> fields,
String partition)
+ throws Exception {
List<CommitMessage> builds = new ArrayList<>();
for (DataSplit split :
GlobalIndexBuilderUtils.splitByContiguousRowRange(
- new SortedGlobalIndexScanner(table, "btree")
-
.withIndexFields(Collections.singletonList("id"))
+ new SortedGlobalIndexScanner(table, indexType)
+ .withIndexFields(fields)
.scan()
.get()
.entries())) {
- builds.addAll(SortedGlobalIndexTestUtils.buildIndex(table,
"btree", "id", split, 4));
+ if (partition == null ||
split.partition().getString(0).toString().equals(partition)) {
+ builds.addAll(
+ SortedGlobalIndexTestUtils.buildIndex(
+ table,
+ indexType,
+ fields,
+ split,
+ table.snapshotManager().latestSnapshotId()));
+ }
}
+ return builds;
+ }
+
+ private CommitMessageImpl compactIndex(FileStoreTable table) throws
Exception {
try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
- commit.commit(builds);
+ commit.commit(buildIndex(table, "id", null));
}
Snapshot snapshot = table.snapshotManager().latestSnapshot();
List<IndexManifestEntry> inputs =
@@ -257,6 +284,148 @@ public class CompactReassignReuseTest extends
TableTestBase {
Arrays.stream(expected).boxed().collect(Collectors.toList()));
}
+ @ParameterizedTest
+ @CsvSource({
+ "btree,false,after-scan", "btree,true,after-scan",
+ "bitmap,false,after-scan", "bitmap,true,after-scan",
+ "composite,false,after-scan", "composite,true,after-scan",
+ "btree,false,before-scan", "btree,true,before-scan",
+ "bitmap,false,before-scan", "bitmap,true,before-scan",
+ "composite,false,before-scan", "composite,true,before-scan",
+ "btree,false,index-first", "btree,true,index-first",
+ "bitmap,false,index-first", "bitmap,true,index-first",
+ "composite,false,index-first", "composite,true,index-first"
+ })
+ void testIndexBuildAcrossReassignment(String indexType, boolean strict,
String order)
+ throws Exception {
+ FileStoreTable table = prepare();
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(buildIndex(table, "pt", null));
+ }
+ long lastSafeSnapshot = table.snapshotManager().latestSnapshotId();
+ boolean reassignBeforeScan = order.equals("before-scan");
+ if (reassignBeforeScan) {
+ assertThat(new
DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue();
+ }
+ long scanSnapshot = table.snapshotManager().latestSnapshotId();
+ boolean composite = indexType.equals("composite");
+ List<CommitMessage> messages =
+ buildIndex(
+ table,
+ composite ? "btree" : indexType,
+ composite ? Arrays.asList("id", "pt") :
Collections.singletonList("id"),
+ null);
+ List<String> files =
+ messages.stream()
+ .flatMap(
+ m ->
+ ((CommitMessageImpl) m)
+
.newFilesIncrement().newIndexFiles().stream())
+ .map(IndexFileMeta::fileName)
+ .collect(Collectors.toList());
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(),
Boolean.toString(strict));
+ options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(),
Long.toString(lastSafeSnapshot));
+ FileStoreTable committingTable = table.copy(options);
+ AtomicBoolean indexCommitted = new AtomicBoolean();
+ Runnable commitIndex =
+ () -> {
+ if (indexCommitted.compareAndSet(false, true)) {
+ try (BatchTableCommit commit =
+
committingTable.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ boolean indexFirst = order.equals("index-first");
+ if (indexFirst) {
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(table, null,
commitIndex).reassign();
+ assertThat(result.reassigned).isTrue();
+ assertThat(result.previousSnapshotId).isEqualTo(scanSnapshot + 1);
+ } else {
+ if (!reassignBeforeScan) {
+ assertThat(new
DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue();
+ }
+ commitIndex.run();
+ }
+ assertThat(indexCommitted).isTrue();
+ assertThat(table.snapshotManager().latestSnapshot().commitKind())
+ .isEqualTo(indexFirst ? Snapshot.CommitKind.OVERWRITE :
Snapshot.CommitKind.APPEND);
+ List<IndexManifestEntry> current =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .filter(e ->
e.indexFile().globalIndexMeta().indexFieldId() == 1)
+ .collect(Collectors.toList());
+ assertThat(current)
+ .extracting(e -> e.indexFile().fileName())
+ .containsExactlyInAnyOrderElementsOf(files);
+ for (IndexManifestEntry entry : current) {
+ assertThat(
+
DataEvolutionIndexSourceMeta.fromIndexFile(entry.indexFile())
+ .scanSnapshotId())
+ .isEqualTo(scanSnapshot);
+ }
+ long[] rowIds = {4, 5, 2, 6};
+ for (int id = 0; id < rowIds.length; id++) {
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate predicate = builder.equal(1, id);
+ if (composite) {
+ predicate =
+ PredicateBuilder.and(
+ predicate,
+ builder.equal(0, BinaryString.fromString(id ==
2 ? "b" : "a")));
+ }
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.create(table, null,
predicate).get()) {
+ GlobalIndexResult result = scanner.scan(predicate).get();
+ assertThat(result.results().getLongCardinality()).isEqualTo(1);
+ assertThat(result.results().contains(rowIds[id])).isTrue();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testIndexBuildKeepsExistingColumnUpdateBehavior(boolean reassign)
throws Exception {
+ FileStoreTable table = prepare();
+ long scanSnapshot = table.snapshotManager().latestSnapshotId();
+ List<CommitMessage> messages = buildIndex(table, "id", null);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+
.withWriteType(table.rowType().project(Arrays.asList("pt", "id")));
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(BinaryString.fromString("a"), 100));
+ List<CommitMessage> updates = write.prepareCommit();
+ for (CommitMessage update : updates) {
+ ((CommitMessageImpl) update)
+ .newFilesIncrement()
+ .newFiles()
+ .replaceAll(file -> file.assignFirstRowId(0L));
+ }
+ commit.commit(updates);
+ }
+ if (reassign) {
+ new DataEvolutionRowIdReassigner(table).reassign();
+ }
+ long snapshot = table.snapshotManager().latestSnapshotId();
+ try (BatchTableCommit commit =
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(),
+ Long.toString(scanSnapshot)))
+ .newBatchWriteBuilder()
+ .newCommit()) {
+ commit.commit(messages);
+ }
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot + 1);
+ assertThat(read(table))
+ .extracting(row -> row.getInt(1))
+ .containsExactlyInAnyOrder(100, 1, 2, 3);
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void testReuseNormalAndBlobAcrossRepeatedReassignments(boolean blob)
throws Exception {
@@ -291,12 +460,32 @@ public class CompactReassignReuseTest extends
TableTestBase {
}
@ParameterizedTest
- @ValueSource(booleans = {false, true})
- void testIndexReplacementKeepsPhysicalFilesAndPointLookup(boolean
loseSuccessfulResponse)
- throws Exception {
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void testIndexReplacementKeepsPhysicalFilesAndPointLookup(
+ boolean indexBuild, boolean loseSuccessfulResponse) throws
Exception {
FileStoreTable table = prepare();
CommitMessageImpl message = compactIndex(table);
List<IndexFileMeta> outputs =
message.compactIncrement().newIndexFiles();
+ List<String> replacedFiles =
+ message.compactIncrement().deletedIndexFiles().stream()
+ .map(IndexFileMeta::fileName)
+ .collect(Collectors.toList());
+ long scanSnapshot = lastSafeSnapshots.get(message);
+ if (indexBuild) {
+ message =
+ new CommitMessageImpl(
+ message.partition(),
+ message.bucket(),
+ message.totalBuckets(),
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ outputs,
+
message.compactIncrement().deletedIndexFiles()),
+ CompactIncrement.emptyIncrement());
+ lastSafeSnapshots.put(message, scanSnapshot);
+ }
assertThat(new
DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue();
long reassignedSnapshot = table.snapshotManager().latestSnapshotId();
AtomicInteger attempts = new AtomicInteger();
@@ -316,7 +505,10 @@ public class CompactReassignReuseTest extends
TableTestBase {
throws Exception {
boolean committed =
delegate.commit(baseUuid, snapshot,
branch, statistics);
- if (snapshot.commitKind() ==
Snapshot.CommitKind.COMPACT) {
+ if (snapshot.commitKind()
+ == (indexBuild
+ ? Snapshot.CommitKind.APPEND
+ :
Snapshot.CommitKind.COMPACT)) {
attempts.incrementAndGet();
return committed &&
!loseSuccessfulResponse;
}
@@ -333,6 +525,16 @@ public class CompactReassignReuseTest extends
TableTestBase {
FileStoreTable committingTable =
FileStoreTableFactory.create(
table.fileIO(), table.location(), table.schema(),
environment);
+ if (indexBuild) {
+ // A replaced index's source baseline predates this refresh's
planning snapshot.
+ assertThat(
+ table.fileIO()
+ .delete(
+
table.snapshotManager().snapshotPath(scanSnapshot),
+ false))
+ .isTrue();
+ table.snapshotManager().invalidateCache();
+ }
commit(committingTable, message, false);
assertThat(attempts).hasValue(1);
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(reassignedSnapshot
+ 1);
@@ -343,7 +545,8 @@ public class CompactReassignReuseTest extends TableTestBase
{
.read(table.snapshotManager().latestSnapshot().indexManifest());
assertThat(current.stream().map(e -> e.indexFile().fileName()))
.containsAll(
-
outputs.stream().map(IndexFileMeta::fileName).collect(Collectors.toList()));
+
outputs.stream().map(IndexFileMeta::fileName).collect(Collectors.toList()))
+ .doesNotContainAnyElementsOf(replacedFiles);
Predicate predicate = new PredicateBuilder(table.rowType()).equal(1,
1);
try (DataEvolutionGlobalIndexScanner scanner =
DataEvolutionGlobalIndexScanner.create(table, null,
predicate).get()) {
@@ -544,10 +747,15 @@ public class CompactReassignReuseTest extends
TableTestBase {
checkRows(table, 0, 1, 2, 3);
}
- @Test
- void testMissingPlanFailsClosed() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMissingPlanFailsClosed(boolean indexBuild) throws Exception {
FileStoreTable table = prepare();
- CommitMessageImpl message = compact(table, false);
+ CommitMessageImpl message =
+ indexBuild
+ ? (CommitMessageImpl) buildIndex(table, "id",
"a").get(0)
+ : compact(table, false);
+ lastSafeSnapshots.put(message,
table.snapshotManager().latestSnapshotId());
new DataEvolutionRowIdReassigner(table).reassign();
Snapshot snapshot = table.snapshotManager().latestSnapshot();
Path plan =
@@ -561,10 +769,15 @@ public class CompactReassignReuseTest extends
TableTestBase {
checkRows(table, 0, 1, 2, 3);
}
- @Test
- void testCorruptPlanFailsClosed() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCorruptPlanFailsClosed(boolean indexBuild) throws Exception {
FileStoreTable table = prepare();
- CommitMessageImpl message = compact(table, false);
+ CommitMessageImpl message =
+ indexBuild
+ ? (CommitMessageImpl) buildIndex(table, "id",
"a").get(0)
+ : compact(table, false);
+ lastSafeSnapshots.put(message,
table.snapshotManager().latestSnapshotId());
new DataEvolutionRowIdReassigner(table).reassign();
Snapshot snapshot = table.snapshotManager().latestSnapshot();
Path path =
@@ -579,20 +792,56 @@ public class CompactReassignReuseTest extends
TableTestBase {
checkRows(table, 0, 1, 2, 3);
}
- @Test
- void testNoBoundaryKeepsExistingConflictCheck() throws Exception {
+ @ParameterizedTest
+ @CsvSource({"false,", "true,", "false,-1", "true,-1"})
+ void testNoBoundaryKeepsExistingConflictCheck(boolean indexBuild, Long
lastSafeSnapshot)
+ throws Exception {
FileStoreTable table = prepare();
- CommitMessageImpl message = compact(table, false);
+ CommitMessageImpl message =
+ indexBuild
+ ? (CommitMessageImpl) buildIndex(table, "id",
"a").get(0)
+ : compact(table, false);
new DataEvolutionRowIdReassigner(table).reassign();
long snapshot = table.snapshotManager().latestSnapshotId();
- try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), "false");
+ if (lastSafeSnapshot != null) {
+ options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(),
lastSafeSnapshot.toString());
+ }
+ try (BatchTableCommit commit =
table.copy(options).newBatchWriteBuilder().newCommit()) {
assertThatThrownBy(() ->
commit.commit(Collections.singletonList(message)))
- .hasStackTraceContaining("conflict");
+ .hasStackTraceContaining(
+ indexBuild
+ ? "Global index row ID existence conflict"
+ : "Row ID existence conflict");
}
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshot);
checkRows(table, 0, 1, 2, 3);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"snapshot", "manifest-list"})
+ void testIndexBuildMissingMetadataFailsClosed(String missing) throws
Exception {
+ FileStoreTable table = prepare();
+ CommitMessageImpl message = (CommitMessageImpl) buildIndex(table,
"id", "a").get(0);
+ long boundary = table.snapshotManager().latestSnapshotId();
+ lastSafeSnapshots.put(message, boundary);
+ write(table, "b", 4, false);
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ boolean snapshotMissing = missing.equals("snapshot");
+ Path path =
+ snapshotMissing
+ ? table.snapshotManager().snapshotPath(boundary + 1)
+ :
table.store().pathFactory().toManifestListPath(latest.baseManifestList());
+ assertThat(table.fileIO().delete(path, false)).isTrue();
+ assertThatThrownBy(() -> commit(table, message, false))
+ .hasStackTraceContaining(
+ snapshotMissing ? "missing snapshot" :
latest.baseManifestList());
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(latest.id());
+ assertThat(table.snapshotManager().snapshotExists(latest.id() +
1)).isFalse();
+ }
+
@Test
void testExpiredBoundaryAndLegacyOption() throws Exception {
FileStoreTable table = prepare();
@@ -606,10 +855,15 @@ public class CompactReassignReuseTest extends
TableTestBase {
}
@ParameterizedTest
- @ValueSource(booleans = {false, true})
- void testReassignDuringCommitRetries(boolean loseSuccessfulResponse)
throws Exception {
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void testReassignDuringCommitRetries(boolean indexBuild, boolean
loseSuccessfulResponse)
+ throws Exception {
FileStoreTable original = prepare();
- CommitMessageImpl message = compact(original, false);
+ CommitMessageImpl message =
+ indexBuild
+ ? (CommitMessageImpl) buildIndex(original, "id",
"a").get(0)
+ : compact(original, false);
+ lastSafeSnapshots.put(message,
original.snapshotManager().latestSnapshotId());
AtomicInteger attempts = new AtomicInteger();
CatalogEnvironment environment =
new CatalogEnvironment(null, null, null, null, null, null,
false, false) {
@@ -625,7 +879,10 @@ public class CompactReassignReuseTest extends
TableTestBase {
String branch,
List<PartitionStatistics> statistics)
throws Exception {
- if (snapshot.commitKind() !=
Snapshot.CommitKind.COMPACT) {
+ if (snapshot.commitKind()
+ != (indexBuild
+ ? Snapshot.CommitKind.APPEND
+ :
Snapshot.CommitKind.COMPACT)) {
return delegate.commit(baseUuid, snapshot,
branch, statistics);
}
int attempt = attempts.getAndIncrement();
@@ -663,7 +920,22 @@ public class CompactReassignReuseTest extends
TableTestBase {
assertThat(attempts).hasValue(3);
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(9);
checkRows(table, 0, 1, 2, 3, 4, 5);
-
assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId()).isZero();
+ if (indexBuild) {
+ IndexFileMeta built =
message.newFilesIncrement().newIndexFiles().get(0);
+ assertThat(built.globalIndexMeta().rowRange()).isEqualTo(new
Range(0, 1));
+ assertThat(table.store().newIndexFileHandler().scanEntries())
+ .singleElement()
+ .satisfies(
+ entry -> {
+ assertThat(entry.indexFile().fileName())
+ .isEqualTo(built.fileName());
+
assertThat(entry.indexFile().globalIndexMeta().rowRange())
+ .isEqualTo(new Range(9, 10));
+ });
+ } else {
+
assertThat(message.compactIncrement().compactAfter().get(0).nonNullFirstRowId())
+ .isZero();
+ }
}
@Test
@@ -682,8 +954,8 @@ public class CompactReassignReuseTest extends TableTestBase
{
Collections.emptyList(),
Collections.emptyList(),
Collections.singletonList(unsupported));
- ReassignCompactChangesProvider provider =
- new ReassignCompactChangesProvider(
+ DataEvolutionChangesProvider provider =
+ new DataEvolutionChangesProvider(
table.fileIO(),
table.store().pathFactory(),
table.snapshotManager(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 99fd1f6532..b85329fd90 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1010,8 +1010,10 @@ public class FileStoreCommitTest {
try (FileStoreCommitImpl commit = store.newCommit()) {
CommitChanges refreshed =
- commit.compactChangesProvider(
- changes,
Collections.singleton(Pair.of(partition, 0)))
+ commit.dataEvolutionChangesProvider(
+ changes,
+ Snapshot.CommitKind.COMPACT,
+ Collections.singleton(Pair.of(partition,
0)))
.provide(null);
assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete);
