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 1b623570d3 [core] Replay index file changes when restoring
MergeTreeWriter (#10260)
1b623570d3 is described below
commit 1b623570d35eed7819d1518dc8d6564330245931
Author: YangJie <[email protected]>
AuthorDate: Mon Oct 5 10:37:26 2026 -0400
[core] Replay index file changes when restoring MergeTreeWriter (#10260)
---
.../apache/paimon/mergetree/MergeTreeWriter.java | 28 +++++-
.../source/PrimaryKeySortedIndexReadTest.java | 108 +++++++++++++++++++++
2 files changed, 134 insertions(+), 2 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index 4f7b248870..2d58056b1e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -27,6 +27,7 @@ import org.apache.paimon.compact.CompactResult;
import org.apache.paimon.compression.CompressOptions;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
@@ -80,6 +81,10 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
private final LinkedHashMap<String, DataFileMeta> compactBefore;
private final LinkedHashSet<DataFileMeta> compactAfter;
private final LinkedHashSet<DataFileMeta> compactChangelog;
+ private final List<IndexFileMeta> newIndexFiles;
+ private final List<IndexFileMeta> deletedIndexFiles;
+ private final List<IndexFileMeta> compactNewIndexFiles;
+ private final List<IndexFileMeta> compactDeletedIndexFiles;
@Nullable private CompactDeletionFile compactDeletionFile;
@@ -123,16 +128,27 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
this.compactBefore = new LinkedHashMap<>();
this.compactAfter = new LinkedHashSet<>();
this.compactChangelog = new LinkedHashSet<>();
+ this.newIndexFiles = new ArrayList<>();
+ this.deletedIndexFiles = new ArrayList<>();
+ this.compactNewIndexFiles = new ArrayList<>();
+ this.compactDeletedIndexFiles = new ArrayList<>();
if (increment != null) {
newFiles.addAll(increment.newFilesIncrement().newFiles());
deletedFiles.addAll(increment.newFilesIncrement().deletedFiles());
newFilesChangelog.addAll(increment.newFilesIncrement().changelogFiles());
+ // index file changes must survive the restore replay too: a
payload accepted
+ // during checkpoint would otherwise never reach the manifest, and
replaced
+ // payloads would stay as zombie entries
+
newIndexFiles.addAll(increment.newFilesIncrement().newIndexFiles());
+
deletedIndexFiles.addAll(increment.newFilesIncrement().deletedIndexFiles());
increment
.compactIncrement()
.compactBefore()
.forEach(f -> compactBefore.put(f.fileName(), f));
compactAfter.addAll(increment.compactIncrement().compactAfter());
compactChangelog.addAll(increment.compactIncrement().changelogFiles());
+
compactNewIndexFiles.addAll(increment.compactIncrement().newIndexFiles());
+
compactDeletedIndexFiles.addAll(increment.compactIncrement().deletedIndexFiles());
updateCompactDeletionFile(increment.compactDeletionFile());
}
}
@@ -285,20 +301,28 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
new DataIncrement(
new ArrayList<>(newFiles),
new ArrayList<>(deletedFiles),
- new ArrayList<>(newFilesChangelog));
+ new ArrayList<>(newFilesChangelog),
+ new ArrayList<>(newIndexFiles),
+ new ArrayList<>(deletedIndexFiles));
CompactIncrement compactIncrement =
new CompactIncrement(
new ArrayList<>(compactBefore.values()),
new ArrayList<>(compactAfter),
- new ArrayList<>(compactChangelog));
+ new ArrayList<>(compactChangelog),
+ new ArrayList<>(compactNewIndexFiles),
+ new ArrayList<>(compactDeletedIndexFiles));
CompactDeletionFile drainDeletionFile = compactDeletionFile;
newFiles.clear();
deletedFiles.clear();
newFilesChangelog.clear();
+ newIndexFiles.clear();
+ deletedIndexFiles.clear();
compactBefore.clear();
compactAfter.clear();
compactChangelog.clear();
+ compactNewIndexFiles.clear();
+ compactDeletedIndexFiles.clear();
compactDeletionFile = null;
return new CommitIncrement(dataIncrement, compactIncrement,
drainDeletionFile);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
index 19f568152f..c76ba9ddb7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
@@ -27,20 +27,33 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.operation.FileStoreWrite;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.types.DataTypes;
+import
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors;
+
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -202,6 +215,101 @@ class PrimaryKeySortedIndexReadTest extends TableTestBase
{
assertThat(ids).containsExactlyInAnyOrderElementsOf(expectedIds);
}
+ @ParameterizedTest(name = "compaction committed before checkpoint: {0}")
+ @ValueSource(booleans = {true, false})
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ void testIndexAcceptedAtCheckpointSurvivesWriterRestore(boolean
compactionCommitted)
+ throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ BinaryString tag = BinaryString.fromString("tag");
+ write(table, ioManager, GenericRow.of(1, 10, tag), GenericRow.of(2,
20, tag));
+ write(table, ioManager, GenericRow.of(3, 10, tag));
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+ List<String> replacedPayloads =
+ sourcePayloads(table).stream()
+ .map(IndexFileMeta::fileName)
+ .collect(Collectors.toList());
+ assertThat(replacedPayloads).isNotEmpty();
+
+ // the compaction rewrites the indexed file, so the next index build
replaces its payload
+ ExecutorService compactExecutor =
MoreExecutors.newDirectExecutorService();
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(ioManager)
+ .withCompactExecutor(compactExecutor);
+ write.write(GenericRow.of(4, 10, tag));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ if (compactionCommitted) {
+ // the build is not waited for here, so the checkpoint below
accepts it
+ try (TableCommitImpl commit = table.newCommit(commitUser)) {
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+ }
+
+ // a CDC writer refreshing its schema runs checkpoint, close and
restore in process, as
+ // StoreSinkWriteImpl#replace does
+ List<? extends FileStoreWrite.State<?>> states = write.checkpoint();
+ write.close();
+ compactExecutor.shutdown();
+ catalog.alterTable(identifier(), SchemaChange.addColumn("extra",
DataTypes.INT()), false);
+ FileStoreTable newTable = getTableDefault();
+ TableWriteImpl<?> restored =
newTable.newWrite(commitUser).withIOManager(ioManager);
+ restored.restore((List) states);
+ try (TableCommitImpl commit = newTable.newCommit(commitUser)) {
+ commit.commit(2, restored.prepareCommit(false, 2));
+ }
+ assertThat(restored.prepareCommit(false, 3))
+ .flatExtracting(PrimaryKeySortedIndexReadTest::indexChanges)
+ .isEmpty();
+ restored.close();
+
+ List<IndexFileMeta> payloads = sourcePayloads(newTable);
+ assertThat(payloads)
+ .extracting(IndexFileMeta::fileName)
+ .doesNotContainAnyElementsOf(replacedPayloads);
+ assertThat(
+ payloads.stream()
+ .flatMap(
+ payload ->
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
+
.sourceFiles().stream())
+ .map(PrimaryKeyIndexSourceFile::fileName))
+ .containsExactlyInAnyOrderElementsOf(
+ newTable.store().newScan().plan().files().stream()
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toList()));
+
+ ReadBuilder readBuilder =
+ newTable.newReadBuilder()
+ .withFilter(new
PredicateBuilder(newTable.rowType()).equal(1, 10));
+ TableScan.Plan plan = readBuilder.newScan().plan();
+
assertThat(plan.splits()).isNotEmpty().allMatch(IndexedSplit.class::isInstance);
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ readBuilder.newRead().executeFilter().createReader(plan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactlyInAnyOrder(1, 3, 4);
+ }
+
+ private static List<IndexFileMeta> sourcePayloads(FileStoreTable table) {
+ return table.store()
+ .newIndexFileHandler()
+ .scanSourceIndexes(
+ table.snapshotManager().latestSnapshot(),
BinaryRow.EMPTY_ROW, 0);
+ }
+
+ private static List<IndexFileMeta> indexChanges(CommitMessage message) {
+ CommitMessageImpl impl = (CommitMessageImpl) message;
+ List<IndexFileMeta> changes = new ArrayList<>();
+ changes.addAll(impl.newFilesIncrement().newIndexFiles());
+ changes.addAll(impl.newFilesIncrement().deletedIndexFiles());
+ changes.addAll(impl.compactIncrement().newIndexFiles());
+ changes.addAll(impl.compactIncrement().deletedIndexFiles());
+ return changes;
+ }
+
@Test
void testReadWithMultiValueIndex() throws Exception {
Schema schema =
