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 44c5113b8b [core] Extract injectable data evolution commit preparation
(#10686)
44c5113b8b is described below
commit 44c5113b8b37e0310175df7a2582eface500d45c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 14:15:26 2026 +0800
[core] Extract injectable data evolution commit preparation (#10686)
---
.../ClusteringCommitChangesRewriter.java | 59 +++
.../DataEvolutionCommitChangesRewriter.java | 125 ++++++
...writer.java => DataEvolutionIndexRewriter.java} | 25 +-
...ializeDeletionVectorsCommitChangesRewriter.java | 42 ++
.../apache/paimon/operation/FileStoreCommit.java | 4 +
.../paimon/operation/FileStoreCommitImpl.java | 167 ++------
.../operation/commit/CommitChangesRewriter.java | 69 ++++
.../commit/ReassignCompactChangesProvider.java | 21 +
.../apache/paimon/table/sink/TableCommitImpl.java | 6 +
.../dataevolution/DataEvolutionClusterTest.java | 52 ++-
.../operation/FileStoreCommitRewriterTest.java | 453 +++++++++++++++++++++
.../paimon/operation/FileStoreCommitTest.java | 32 +-
.../table/DataEvolutionDeletionVectorTest.java | 135 +++++-
.../flink/compact/IncrementalClusterCompact.java | 5 +-
.../flink/sink/DataEvolutionClusterSink.java | 49 +++
...DataEvolutionDeletionVectorMaterializeSink.java | 3 +
.../paimon/spark/procedure/CompactProcedure.java | 2 +
.../MaterializeDeletionVectorsProcedure.java | 3 +
18 files changed, 1090 insertions(+), 162 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringCommitChangesRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringCommitChangesRewriter.java
new file mode 100644
index 0000000000..932362bb29
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringCommitChangesRewriter.java
@@ -0,0 +1,59 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.operation.commit.CommitChanges;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Remaps scalar indexes after clustering outputs receive their final row
IDs. */
+public final class ClusteringCommitChangesRewriter extends
DataEvolutionCommitChangesRewriter {
+
+ public ClusteringCommitChangesRewriter(FileStoreTable table) {
+ super(table);
+ }
+
+ @Override
+ public Attempt createAttempt(CommitKind kind, CommitChanges changes) {
+ if (kind != CommitKind.COMPACT) {
+ return Attempt.IDENTITY;
+ }
+
+ // Side files remain attached to committed data. Only outputs awaiting
their first row-ID
+ // assignment carry a permutation for this commit; later metadata
rewrites must ignore it.
+ Set<String> mappedClusteringOutputs =
+ changes.tableFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .filter(entry -> entry.file().firstRowId() == null)
+ .filter(
+ entry ->
+ entry.file()
+ .extraFiles()
+ .contains(
+
ClusteringRowIdMapping.fileName(
+
entry.file().fileName())))
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toSet());
+ return indexRewriteAttempt(changes, mappedClusteringOutputs);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCommitChangesRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCommitChangesRewriter.java
new file mode 100644
index 0000000000..3adcc63ba5
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCommitChangesRewriter.java
@@ -0,0 +1,125 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.IndexManifestFile;
+import org.apache.paimon.operation.commit.CommitChanges;
+import org.apache.paimon.operation.commit.CommitChangesProvider;
+import org.apache.paimon.operation.commit.CommitChangesRewriter;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.Pair;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
+
+/** Maintains index changes for operations which replace data-evolution row
IDs. */
+public abstract class DataEvolutionCommitChangesRewriter implements
CommitChangesRewriter {
+
+ private final FileStoreTable table;
+ private final IndexManifestFile indexManifestFile;
+
+ protected DataEvolutionCommitChangesRewriter(FileStoreTable table) {
+ this.table = table;
+ this.indexManifestFile =
table.store().indexManifestFileFactory().create();
+ }
+
+ @Override
+ public CommitChangesProvider compactChangesProvider(
+ CommitChanges changes, List<CommitMessage> messages) {
+ Set<Pair<BinaryRow, Integer>> materializedBuckets = new HashSet<>();
+ for (CommitMessage message : messages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ if (!materializedRowIdRanges(
+ commitMessage.compactIncrement().compactBefore(),
+ commitMessage.compactIncrement().compactAfter())
+ .isEmpty()) {
+ materializedBuckets.add(Pair.of(commitMessage.partition(),
commitMessage.bucket()));
+ }
+ }
+ if (materializedBuckets.isEmpty()) {
+ return CommitChangesRewriter.super.compactChangesProvider(changes,
messages);
+ }
+
+ return latestSnapshot -> {
+ List<IndexManifestEntry> indexFiles =
+ changes.indexFiles.stream()
+ // Replace global-index deletions prepared against
an older snapshot.
+ .filter(
+ entry ->
+ entry.kind() != FileKind.DELETE
+ ||
entry.indexFile().globalIndexMeta() == null
+ ||
!materializedBuckets.contains(
+ Pair.of(
+
entry.partition(),
+
entry.bucket())))
+ .collect(Collectors.toList());
+
+ // Each optimistic retry must include concurrently created indexes
before assigning
+ // final row IDs to this operation's outputs.
+ if (latestSnapshot != null && latestSnapshot.indexManifest() !=
null) {
+ for (IndexManifestEntry entry :
+
indexManifestFile.read(latestSnapshot.indexManifest())) {
+ if (entry.indexFile().globalIndexMeta() != null
+ && materializedBuckets.contains(
+ Pair.of(entry.partition(),
entry.bucket()))) {
+ indexFiles.add(entry.toDeleteEntry());
+ }
+ }
+ }
+
+ return new CommitChanges(changes.tableFiles,
changes.changelogFiles, indexFiles);
+ };
+ }
+
+ protected Attempt indexRewriteAttempt(CommitChanges changes, Set<String>
mappedOutputs) {
+ // Capture this before assignment makes newly allocated outputs
indistinguishable from
+ // ordinary compaction outputs which preserve their existing row IDs.
+ Set<String> rowIdPreservingOutputs =
+ changes.tableFiles.stream()
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .filter(entry -> entry.file().firstRowId() != null)
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toSet());
+ return (newSnapshotId, assignedFiles, indexFiles, ioManager) -> {
+ if (assignedFiles.stream().noneMatch(entry -> entry.kind() ==
FileKind.DELETE)) {
+ return indexFiles;
+ }
+ return DataEvolutionIndexRewriter.rewrite(
+ table.fileIO(),
+ table.store().pathFactory(),
+ table.schemaManager(),
+ table.coreOptions(),
+ ioManager,
+ assignedFiles,
+ indexFiles,
+ mappedOutputs,
+ rowIdPreservingOutputs,
+ newSnapshotId);
+ };
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionIndexRewriter.java
similarity index 95%
rename from
paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
rename to
paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionIndexRewriter.java
index 36d360dac4..d9ec534abc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/ClusteringIndexRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionIndexRewriter.java
@@ -22,8 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.globalindex.DataEvolutionGlobalIndexRefreshPlanner;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.GlobalIndexFileReadWrite;
@@ -71,14 +69,14 @@ import java.util.stream.Collectors;
import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Rewrites scalar index postings after clustering output files receive their
final row IDs. */
-public final class ClusteringIndexRewriter {
+/** Maintains scalar index coverage when data evolution replaces row IDs. */
+public final class DataEvolutionIndexRewriter {
// The current metadata protocol represents coverage with contiguous
ranges. Bound both
// tiny-file fan-out and repeated source scans until sparse coverage has
its own protocol.
private static final int MAX_COVERAGE_FRAGMENTS = 64;
- private ClusteringIndexRewriter() {}
+ private DataEvolutionIndexRewriter() {}
public static List<IndexManifestEntry> rewrite(
FileIO fileIO,
@@ -90,8 +88,7 @@ public final class ClusteringIndexRewriter {
List<IndexManifestEntry> indexFiles,
Set<String> mappedClusteringOutputs,
Set<String> rowIdPreservingOutputs,
- long newSnapshotId,
- List<Path> createdFiles)
+ long newSnapshotId)
throws IOException {
if (indexFiles.stream()
.noneMatch(
@@ -284,7 +281,7 @@ public final class ClusteringIndexRewriter {
GlobalIndexer.create(type, fields,
Options.fromMap(options.toMap()));
IndexPathFactory indexPath =
pathFactory.globalIndexFileFactory();
GlobalIndexFileReadWrite indexIO =
- trackingIndexIO(fileIO, indexPath, createdFiles);
+ new GlobalIndexFileReadWrite(fileIO, indexPath);
for (IndexManifestEntry source : sources) {
GlobalIndexMeta meta =
source.indexFile().globalIndexMeta();
List<Range> fragments = new ArrayList<>();
@@ -382,18 +379,6 @@ public final class ClusteringIndexRewriter {
fragments.add(fragment);
}
- private static GlobalIndexFileReadWrite trackingIndexIO(
- FileIO fileIO, IndexPathFactory indexPath, List<Path>
createdFiles) {
- return new GlobalIndexFileReadWrite(fileIO, indexPath) {
- @Override
- public PositionOutputStream newOutputStream(String fileName)
throws IOException {
- // Track before opening: failed writes can leave a partial
file without a result.
- createdFiles.add(indexPath.toPath(fileName));
- return super.newOutputStream(fileName);
- }
- };
- }
-
private static void readMapping(
FileIO fileIO,
DataFilePathFactory pathFactory,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/MaterializeDeletionVectorsCommitChangesRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/MaterializeDeletionVectorsCommitChangesRewriter.java
new file mode 100644
index 0000000000..6f2d3502c5
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/MaterializeDeletionVectorsCommitChangesRewriter.java
@@ -0,0 +1,42 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.operation.commit.CommitChanges;
+import org.apache.paimon.table.FileStoreTable;
+
+import java.util.Collections;
+
+/** Maintains index coverage for deletion-vector materialization. */
+public final class MaterializeDeletionVectorsCommitChangesRewriter
+ extends DataEvolutionCommitChangesRewriter {
+
+ public MaterializeDeletionVectorsCommitChangesRewriter(FileStoreTable
table) {
+ super(table);
+ }
+
+ @Override
+ public Attempt createAttempt(CommitKind kind, CommitChanges changes) {
+ if (kind != CommitKind.COMPACT) {
+ return Attempt.IDENTITY;
+ }
+ return indexRewriteAttempt(changes, Collections.emptySet());
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index 050a345a0e..ca4f96527f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.operation.commit.CommitChangesRewriter;
import org.apache.paimon.operation.metrics.CommitMetrics;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.CommitMessage;
@@ -36,6 +37,9 @@ public interface FileStoreCommit extends AutoCloseable {
FileStoreCommit withIOManager(IOManager ioManager);
+ /** Injects metadata preparation into the existing commit attempt
lifecycle. */
+ FileStoreCommit withCommitChangesRewriter(CommitChangesRewriter rewriter);
+
FileStoreCommit ignoreEmptyCommit(boolean ignoreEmptyCommit);
FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire);
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 70a4f2ac95..464f2ec3d6 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
@@ -22,8 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.append.dataevolution.ClusteringIndexRewriter;
-import org.apache.paimon.append.dataevolution.ClusteringRowIdMapping;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
@@ -47,6 +45,7 @@ import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.manifest.SimpleFileEntry;
import org.apache.paimon.operation.commit.CommitChanges;
import org.apache.paimon.operation.commit.CommitChangesProvider;
+import org.apache.paimon.operation.commit.CommitChangesRewriter;
import org.apache.paimon.operation.commit.CommitCleaner;
import org.apache.paimon.operation.commit.CommitResult;
import org.apache.paimon.operation.commit.CommitRollback;
@@ -118,7 +117,6 @@ import static
org.apache.paimon.operation.commit.ManifestEntryChanges.changedPar
import static
org.apache.paimon.operation.commit.RowTrackingCommitUtils.assignRowTracking;
import static
org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
import static
org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
-import static
org.apache.paimon.utils.DataEvolutionUtils.materializedRowIdRanges;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -170,6 +168,8 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
private final ConflictDetection conflictDetection;
private final CommitCleaner commitCleaner;
+ @Nullable private CommitChangesRewriter commitChangesRewriter;
+
private boolean ignoreEmptyCommit;
private CommitMetrics commitMetrics;
private boolean appendCommitCheckConflict = false;
@@ -253,6 +253,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
}
+ @Override
+ public FileStoreCommitImpl withCommitChangesRewriter(CommitChangesRewriter
rewriter) {
+ this.commitChangesRewriter = checkNotNull(rewriter);
+ return this;
+ }
+
@Override
public FileStoreCommit withIOManager(IOManager ioManager) {
this.ioManager = ioManager;
@@ -348,7 +354,6 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
List<CommitMessage> commitMessages = committable.fileCommittables();
configureRowIdCheckFromMessages(commitMessages);
ManifestEntryChanges changes = collectChanges(commitMessages);
- Set<Pair<BinaryRow, Integer>> materializedBuckets =
materializedBuckets(commitMessages);
try {
List<SimpleFileEntry> appendSimpleEntries =
SimpleFileEntry.from(changes.appendTableFiles);
@@ -406,7 +411,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
changes.compactTableFiles,
changes.compactChangelog,
changes.compactIndexFiles),
- materializedBuckets),
+ commitMessages),
committable.identifier(),
committable.watermark(),
committable.properties(),
@@ -595,7 +600,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
changes.compactTableFiles,
emptyList(),
changes.compactIndexFiles),
- Collections.emptySet()),
+ Collections.emptyList()),
committable.identifier(),
committable.watermark(),
committable.properties(),
@@ -836,70 +841,18 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
}
}
- private Set<Pair<BinaryRow, Integer>>
materializedBuckets(List<CommitMessage> commitMessages) {
- if (!options.dataEvolutionEnabled()) {
- return Collections.emptySet();
- }
-
- Set<Pair<BinaryRow, Integer>> result = new HashSet<>();
- for (CommitMessage message : commitMessages) {
- CommitMessageImpl commitMessage = (CommitMessageImpl) message;
- if (!materializedRowIdRanges(
- commitMessage.compactIncrement().compactBefore(),
- commitMessage.compactIncrement().compactAfter())
- .isEmpty()) {
- result.add(Pair.of(commitMessage.partition(),
commitMessage.bucket()));
- }
- }
- return result;
- }
-
- @VisibleForTesting
- CommitChangesProvider compactChangesProvider(
- CommitChanges changes, Set<Pair<BinaryRow, Integer>>
materializedBuckets) {
- if (materializedBuckets.isEmpty()) {
- Long lastSafeSnapshot =
options.commitLastSafeSnapshot().orElse(null);
- if (options.dataEvolutionEnabled()
- &&
!conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT)
- && lastSafeSnapshot != null
- && lastSafeSnapshot >= 0) {
- return new ReassignCompactChangesProvider(
- fileIO, pathFactory, snapshotManager,
lastSafeSnapshot, changes);
- }
- return CommitChangesProvider.provider(
- changes.tableFiles, changes.changelogFiles,
changes.indexFiles);
- }
-
- return latestSnapshot -> {
- List<IndexManifestEntry> indexFiles =
- changes.indexFiles.stream()
- // Replace global-index deletions prepared against
an older snapshot.
- .filter(
- entry ->
- entry.kind() != FileKind.DELETE
- ||
entry.indexFile().globalIndexMeta() == null
- ||
!materializedBuckets.contains(
- Pair.of(
-
entry.partition(),
-
entry.bucket())))
- .collect(Collectors.toList());
-
- // This provider is invoked again after every optimistic-commit
conflict. Scanning the
- // latest snapshot includes concurrently created indexes in this
attempt's changes.
- // Clustering can then remap their postings after the final row
IDs have been assigned.
- if (latestSnapshot != null && latestSnapshot.indexManifest() !=
null) {
- for (IndexManifestEntry entry :
-
indexManifestFile.read(latestSnapshot.indexManifest())) {
- if (entry.indexFile().globalIndexMeta() != null
- && materializedBuckets.contains(
- Pair.of(entry.partition(),
entry.bucket()))) {
- indexFiles.add(entry.toDeleteEntry());
- }
- }
- }
-
- return new CommitChanges(changes.tableFiles,
changes.changelogFiles, indexFiles);
- };
+ private CommitChangesProvider compactChangesProvider(
+ CommitChanges changes, List<CommitMessage> messages) {
+ if (commitChangesRewriter != null) {
+ return commitChangesRewriter.compactChangesProvider(changes,
messages);
+ }
+ return ReassignCompactChangesProvider.create(
+ fileIO,
+ pathFactory,
+ snapshotManager,
+ options,
+ changes,
+
conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT));
}
private int tryCommit(
@@ -1150,31 +1103,15 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
List<ManifestEntry> deltaFiles = changes.tableFiles;
List<ManifestEntry> changelogFiles = changes.changelogFiles;
List<IndexManifestEntry> indexFiles = changes.indexFiles;
- // Side files remain attached to committed data. Only outputs awaiting
their first row-ID
- // assignment carry a permutation for this commit; later metadata
rewrites must ignore it.
- Set<String> mappedClusteringOutputs =
- commitKind == CommitKind.COMPACT &&
options.dataEvolutionEnabled()
- ? deltaFiles.stream()
- .filter(entry -> entry.kind() == FileKind.ADD)
- .filter(entry -> entry.file().firstRowId() ==
null)
- .filter(
- entry ->
- entry.file()
- .extraFiles()
- .contains(
-
ClusteringRowIdMapping.fileName(
-
entry.file().fileName())))
- .map(entry -> entry.file().fileName())
- .collect(Collectors.toSet())
- : Collections.emptySet();
- Set<String> rowIdPreservingOutputs =
- commitKind == CommitKind.COMPACT &&
options.dataEvolutionEnabled()
- ? deltaFiles.stream()
- .filter(entry -> entry.kind() == FileKind.ADD)
- .filter(entry -> entry.file().firstRowId() !=
null)
- .map(entry -> entry.file().fileName())
- .collect(Collectors.toSet())
- : Collections.emptySet();
+ CommitChangesRewriter.Attempt rewriteAttempt =
+ commitChangesRewriter == null
+ ? CommitChangesRewriter.Attempt.IDENTITY
+ : commitChangesRewriter.createAttempt(
+ commitKind,
+ new CommitChanges(
+
Collections.unmodifiableList(changes.tableFiles),
+
Collections.unmodifiableList(changes.changelogFiles),
+
Collections.unmodifiableList(changes.indexFiles)));
Set<Long> rebasedReassignments =
changesProvider.rebasedReassignments();
long newSnapshotId = Snapshot.FIRST_SNAPSHOT_ID;
@@ -1283,7 +1220,6 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
List<ManifestFileMeta> mergeAfterManifests = new ArrayList<>();
boolean skipManifestMerge = false;
long nextRowIdStart = firstRowIdStart;
- List<Path> clusteringIndexOutputs = new ArrayList<>();
try {
long previousTotalRecordCount = 0L;
Long currentWatermark = watermark;
@@ -1349,23 +1285,13 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
deltaFiles = assigned.assignedEntries;
}
- if (commitKind == CommitKind.COMPACT
- && options.dataEvolutionEnabled()
- && deltaFiles.stream().anyMatch(entry -> entry.kind() ==
FileKind.DELETE)) {
- indexFiles =
- ClusteringIndexRewriter.rewrite(
- fileIO,
- pathFactory,
- schemaManager,
- options,
- ioManager,
- deltaFiles,
- indexFiles,
- mappedClusteringOutputs,
- rowIdPreservingOutputs,
- newSnapshotId,
- clusteringIndexOutputs);
- }
+ indexFiles =
+ checkNotNull(
+ rewriteAttempt.rewriteIndexFiles(
+ newSnapshotId,
+ Collections.unmodifiableList(deltaFiles),
+ Collections.unmodifiableList(indexFiles),
+ ioManager));
if (options.snapshotSequenceOrdering()) {
deltaFiles = stampSequenceWithSnapshotId(newSnapshotId,
commitKind, deltaFiles);
@@ -1447,7 +1373,6 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
operation);
} catch (Throwable e) {
// fails when preparing for commit, we should clean up
- clusteringIndexOutputs.forEach(fileIO::deleteQuietly);
commitCleaner.cleanUpReuseTmpManifests(
deltaManifestList, changelogManifestList,
oldIndexManifest, indexManifest);
commitCleaner.cleanUpNoReuseTmpManifests(
@@ -1464,16 +1389,10 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
final List<SimpleFileEntry> finalBaseFiles = baseDataFiles;
final List<ManifestEntry> finalDeltaFiles = deltaFiles;
final List<IndexManifestEntry> finalIndexFiles = indexFiles;
- try {
- commitPreCallbacks.forEach(
- callback ->
- callback.call(
- finalBaseFiles, finalDeltaFiles,
finalIndexFiles, newSnapshot));
- } catch (RuntimeException | Error failure) {
- // No snapshot publication has been attempted, so these new
payloads are unreferenced.
- clusteringIndexOutputs.forEach(fileIO::deleteQuietly);
- throw failure;
- }
+ commitPreCallbacks.forEach(
+ callback ->
+ callback.call(
+ finalBaseFiles, finalDeltaFiles,
finalIndexFiles, newSnapshot));
try {
success = commitSnapshotImpl(latestSnapshot, newSnapshot,
deltaPartitionEntries);
} catch (Exception e) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesRewriter.java
new file mode 100644
index 0000000000..00f31f2d74
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitChangesRewriter.java
@@ -0,0 +1,69 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot.CommitKind;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.List;
+
+/** Prepares commit metadata while reusing the committer's conflict and
publication lifecycle. */
+public interface CommitChangesRewriter {
+
+ CommitChangesRewriter IDENTITY = new CommitChangesRewriter() {};
+
+ /** Creates one provider whose state is retained across optimistic retries
of this compact. */
+ default CommitChangesProvider compactChangesProvider(
+ CommitChanges changes, List<CommitMessage> messages) {
+ return CommitChangesProvider.provider(
+ changes.tableFiles, changes.changelogFiles,
changes.indexFiles);
+ }
+
+ /**
+ * Creates isolated state for one attempt, after resolving its latest
changes and before row-ID
+ * assignment. The changes are read-only; this method must not create
files.
+ */
+ default Attempt createAttempt(CommitKind kind, CommitChanges changes) {
+ return Attempt.IDENTITY;
+ }
+
+ /** Index-only preparation after conflict checking and final row-ID
assignment. */
+ @FunctionalInterface
+ interface Attempt {
+
+ Attempt IDENTITY = (snapshotId, files, indexes, ioManager) -> indexes;
+
+ /**
+ * Returns index changes for this candidate snapshot. Inputs are
read-only. Generated index
+ * files are retained on failure; unreferenced files can be removed by
orphan-file cleanup.
+ */
+ List<IndexManifestEntry> rewriteIndexFiles(
+ long newSnapshotId,
+ List<ManifestEntry> assignedFiles,
+ List<IndexManifestEntry> indexFiles,
+ @Nullable IOManager ioManager)
+ throws IOException;
+ }
+}
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/ReassignCompactChangesProvider.java
index 88d3dcf906..0a66a6b488 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/ReassignCompactChangesProvider.java
@@ -18,6 +18,7 @@
package org.apache.paimon.operation.commit;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.SerializationAssignment;
import org.apache.paimon.fs.FileIO;
@@ -67,6 +68,26 @@ public final class ReassignCompactChangesProvider implements
CommitChangesProvid
this.changes = changes;
}
+ /** Selects the ordinary compaction provider, allowing reuse across
concurrent reassignment. */
+ public static CommitChangesProvider create(
+ FileIO fileIO,
+ FileStorePathFactory paths,
+ SnapshotManager snapshots,
+ CoreOptions options,
+ CommitChanges changes,
+ boolean checkRowIdsFromSnapshot) {
+ Long lastSafeSnapshot = options.commitLastSafeSnapshot().orElse(null);
+ if (options.dataEvolutionEnabled()
+ && !checkRowIdsFromSnapshot
+ && lastSafeSnapshot != null
+ && lastSafeSnapshot >= 0) {
+ return new ReassignCompactChangesProvider(
+ fileIO, paths, snapshots, lastSafeSnapshot, changes);
+ }
+ return CommitChangesProvider.provider(
+ changes.tableFiles, changes.changelogFiles,
changes.indexFiles);
+ }
+
@Override
public CommitChanges provide(@Nullable Snapshot latest) {
checkState(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 23f1132fd7..ee5fcd2eb9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -31,6 +31,7 @@ import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.operation.PartitionExpire;
+import org.apache.paimon.operation.commit.CommitChangesRewriter;
import org.apache.paimon.operation.metrics.CommitMetrics;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.tag.Tag;
@@ -181,6 +182,11 @@ public class TableCommitImpl implements InnerTableCommit {
return this;
}
+ public TableCommitImpl withCommitChangesRewriter(CommitChangesRewriter
rewriter) {
+ commit.withCommitChangesRewriter(rewriter);
+ return this;
+ }
+
@Override
public TableCommitImpl withOperation(Snapshot.Operation operation) {
commit.withOperation(operation);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
index adb355dfc9..9070a9a05e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionClusterTest.java
@@ -544,6 +544,35 @@ class DataEvolutionClusterTest extends TableTestBase {
"a|50|y|9"));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testIndexRemappingRequiresClusteringRewriter(boolean
materializeDeletionVectors)
+ throws Exception {
+ FileStoreTable table = createTable(false);
+ append(table, "a", 30, 10, 20);
+ commit(table, prepareGlobalIndex(table, table.latestSnapshot().get(),
"id"));
+ Snapshot snapshot = table.latestSnapshot().get();
+ List<CommitMessage> messages = prepareMappedCluster(table, snapshot,
true);
+ // An ordinary caller explicitly invalidates its old indexes. Only a
clustering committer
+ // should replace those deletions with remapped postings from the
permutation sidecar.
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ messages =
+ IncrementalClusterRewriter.create(table,
snapshot).prepareCommitMessages(messages);
+ try (TableCommitImpl commit =
table.newCommit("test-scoped-index-rewriter")) {
+ commit.materializeDvRowIdCheck();
+ if (materializeDeletionVectors) {
+ commit.withCommitChangesRewriter(
+ new
MaterializeDeletionVectorsCommitChangesRewriter(table));
+ }
+ commit.commit(messages);
+ }
+
+ assertThat(globalIndexNames(table)).isEmpty();
+ assertThat(readRows(table))
+ .containsExactly("a|10|base-10|3", "a|20|base-20|4",
"a|30|base-30|5");
+ }
+
@ParameterizedTest
@ValueSource(strings = {"btree", "bitmap"})
void testClusterPreservesExactPartialIndexCoverage(String indexType)
throws Exception {
@@ -707,8 +736,10 @@ class DataEvolutionClusterTest extends TableTestBase {
assertMinorClusterResult(table, deletionVectors, untouchedRows,
untouchedFiles, 21L, 4);
}
- @Test
- void testMinorClusterRemapsIndexWhenAllSelectedRowsAreDeleted() throws
Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFullyDeletedRewritePreservesIndexForUnchangedRows(boolean
clustering)
+ throws Exception {
FileStoreTable table = createMinorClusterTable(true);
List<String> untouchedRows = rowsInRange(table, new Range(12, 14));
List<String> untouchedFiles = filesInRange(table, new Range(12, 14));
@@ -726,7 +757,18 @@ class DataEvolutionClusterTest extends TableTestBase {
.compactIncrement()
.compactAfter())
.isEmpty());
- commitCluster(table, snapshot, messages);
+ if (clustering) {
+ commitCluster(table, snapshot, messages);
+ } else {
+ try (TableCommitImpl commit =
table.newCommit("test-materialize-deleted-rows")) {
+ commit.materializeDvRowIdCheck()
+ .withCommitChangesRewriter(
+ new
MaterializeDeletionVectorsCommitChangesRewriter(table))
+ .commit(
+ IncrementalClusterRewriter.create(table,
snapshot)
+ .prepareCommitMessages(messages));
+ }
+ }
PredicateBuilder predicates = new PredicateBuilder(table.rowType());
int id = table.rowType().getFieldIndex("id");
@@ -2077,7 +2119,9 @@ class DataEvolutionClusterTest extends TableTestBase {
private void commitChecked(FileStoreTable table, List<CommitMessage>
messages)
throws Exception {
try (TableCommitImpl commit = table.newCommit("test-cluster")) {
- commit.materializeDvRowIdCheck().commit(messages);
+ commit.materializeDvRowIdCheck()
+ .withCommitChangesRewriter(new
ClusteringCommitChangesRewriter(table))
+ .commit(messages);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitRewriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitRewriterTest.java
new file mode 100644
index 0000000000..8880bb9887
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitRewriterTest.java
@@ -0,0 +1,453 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.AppendOnlyFileStore;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.RenamingSnapshotCommit;
+import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.commit.CommitChanges;
+import org.apache.paimon.operation.commit.CommitChangesRewriter;
+import org.apache.paimon.operation.commit.ConflictDetection;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.CommitPreCallback;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Commit-attempt isolation and index payload retention for injected index
rewriters. */
+class FileStoreCommitRewriterTest extends TableTestBase {
+
+ @Test
+ void testRetryCreatesAttemptFromLatestSnapshotAndFinalRowIds() throws
Exception {
+ FileStoreTable table = createTable();
+ write(table, GenericRow.of(1));
+ ManifestCommittable pending = prepareAppend(table, 10, 20);
+ List<Long> snapshots = new ArrayList<>();
+ List<Long> firstRowIds = new ArrayList<>();
+ List<Path> outputs = new ArrayList<>();
+ List<CommitChangesRewriter.Attempt> attempts = new ArrayList<>();
+ AtomicInteger publications = new AtomicInteger();
+ SnapshotCommit publisher =
+ new DelegatingSnapshotCommit(table) {
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ if (publications.incrementAndGet() == 1) {
+ write(table, GenericRow.of(2), GenericRow.of(3));
+ assertThat(delegate.commit(baseUuid, snapshot,
branch, statistics))
+ .isFalse();
+ return false;
+ }
+ return delegate.commit(baseUuid, snapshot, branch,
statistics);
+ }
+ };
+ CommitChangesRewriter rewriter =
+ new CommitChangesRewriter() {
+ @Override
+ public Attempt createAttempt(Snapshot.CommitKind kind,
CommitChanges changes) {
+ assertThat(kind).isEqualTo(Snapshot.CommitKind.APPEND);
+ assertThat(changes.tableFiles).hasSize(1);
+
assertThat(changes.tableFiles.get(0).file().firstRowId()).isNull();
+ assertReadOnly(changes.tableFiles);
+ assertReadOnly(changes.changelogFiles);
+ assertReadOnly(changes.indexFiles);
+ AtomicInteger calls = new AtomicInteger();
+ Attempt attempt =
+ (snapshotId, assigned, indexes, ioManager) -> {
+
assertThat(calls.incrementAndGet()).isEqualTo(1);
+ snapshots.add(snapshotId);
+ assertReadOnly(assigned);
+ assertReadOnly(indexes);
+
firstRowIds.add(assigned.get(0).file().firstRowId());
+ return writeIndex(table, assigned.get(0),
outputs);
+ };
+ attempts.add(attempt);
+ return attempt;
+ }
+ };
+ try (FileStoreCommit commit =
+ newCommit(table, publisher, Collections.emptyList())
+ .withCommitChangesRewriter(rewriter)) {
+ commit.commit(pending, false);
+ }
+
+ assertThat(publications).hasValue(2);
+ assertThat(snapshots).containsExactly(2L, 3L);
+ assertThat(firstRowIds).containsExactly(1L, 3L);
+ assertThat(attempts).hasSize(2);
+ assertThat(attempts.get(0)).isNotSameAs(attempts.get(1));
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(5L);
+ assertThat(read(table))
+ .extracting(row -> row.getInt(0))
+ .containsExactlyInAnyOrder(1, 2, 3, 10, 20);
+ List<IndexManifestEntry> indexes =
table.store().newIndexFileHandler().scanEntries();
+ assertThat(indexes).hasSize(1);
+
assertThat(indexes.get(0).indexFile().fileName()).isEqualTo(outputs.get(1).getName());
+ assertThat(indexes.get(0).indexFile().globalIndexMeta().rowRange())
+ .isEqualTo(new Range(3, 4));
+ // A false publication response is uncertain, even when this test's
publisher knows it lost.
+ for (Path output : outputs) {
+ assertThat(table.fileIO().exists(output)).isTrue();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testPreparationFailureRetainsOrphanIndex(boolean fatalError) throws
Exception {
+ FileStoreTable table = createTable();
+ ManifestCommittable pending = prepareAppend(table, 10);
+ List<Path> outputs = new ArrayList<>();
+ Throwable failure =
+ fatalError
+ ? new AssertionError("rewrite failed")
+ : new IOException("rewrite failed");
+ AtomicInteger publications = new AtomicInteger();
+ CommitChangesRewriter rewriter =
+ new CommitChangesRewriter() {
+ @Override
+ public Attempt createAttempt(Snapshot.CommitKind kind,
CommitChanges changes) {
+ return (snapshotId, assigned, indexes, ioManager) -> {
+ writeIndex(table, assigned.get(0), outputs);
+ if (fatalError) {
+ throw (AssertionError) failure;
+ }
+ throw (IOException) failure;
+ };
+ }
+ };
+ try (FileStoreCommit commit =
+ newCommit(table, countingPublisher(table, publications),
Collections.emptyList())
+ .withCommitChangesRewriter(rewriter)) {
+ assertThatThrownBy(() -> commit.commit(pending, false))
+ .hasMessageContaining("preparing snapshot #1")
+ .hasCauseReference(failure);
+ }
+ assertThat(outputs).hasSize(1);
+
assertThat(table.fileIO().readFileUtf8(outputs.get(0))).isEqualTo("index
payload");
+ assertThat(publications).hasValue(0);
+ assertThat(table.latestSnapshot()).isEmpty();
+ }
+
+ @Test
+ void testManifestPreparationFailureRetainsOrphanIndex() throws Exception {
+ FileStoreTable table = createTable();
+ ManifestCommittable pending = prepareAppend(table, 10);
+ List<Path> outputs = new ArrayList<>();
+ AtomicInteger publications = new AtomicInteger();
+ CommitChangesRewriter rewriter =
+ new CommitChangesRewriter() {
+ @Override
+ public Attempt createAttempt(Snapshot.CommitKind kind,
CommitChanges changes) {
+ return (snapshotId, assigned, indexes, ioManager) -> {
+ IndexManifestEntry index =
+ writeIndex(table, assigned.get(0),
outputs).get(0);
+ // Fail after the extension returns, while merging
the index manifest.
+ return
Collections.singletonList(index.toDeleteEntry());
+ };
+ }
+ };
+ try (FileStoreCommit commit =
+ newCommit(table, countingPublisher(table, publications),
Collections.emptyList())
+ .withCommitChangesRewriter(rewriter)) {
+ assertThatThrownBy(() -> commit.commit(pending, false))
+ .hasMessageContaining("preparing snapshot #1")
+ .hasRootCauseInstanceOf(IllegalStateException.class)
+ .hasStackTraceContaining("which does not exist.");
+ }
+ assertThat(outputs).hasSize(1);
+
assertThat(table.fileIO().readFileUtf8(outputs.get(0))).isEqualTo("index
payload");
+ assertThat(publications).hasValue(0);
+ assertThat(table.latestSnapshot()).isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testPreCallbackFailureRetainsOrphanIndex(boolean fatalError) throws
Exception {
+ FileStoreTable table = createTable();
+ ManifestCommittable pending = prepareAppend(table, 10);
+ List<Path> outputs = new ArrayList<>();
+ Throwable failure =
+ fatalError
+ ? new AssertionError("validation failed")
+ : new IllegalStateException("validation failed");
+ AtomicInteger publications = new AtomicInteger();
+ CommitPreCallback callback =
+ new CommitPreCallback() {
+ @Override
+ public void call(
+ List<SimpleFileEntry> base,
+ List<ManifestEntry> delta,
+ List<IndexManifestEntry> indexes,
+ Snapshot snapshot) {
+ assertThat(indexes).hasSize(1);
+ if (fatalError) {
+ throw (AssertionError) failure;
+ }
+ throw (IllegalStateException) failure;
+ }
+
+ @Override
+ public void close() {}
+ };
+ try (FileStoreCommit commit =
+ newCommit(
+ table,
+ countingPublisher(table, publications),
+ Collections.singletonList(callback))
+ .withCommitChangesRewriter(indexRewriter(table,
outputs))) {
+ assertThatThrownBy(() -> commit.commit(pending,
false)).isSameAs(failure);
+ }
+ assertThat(outputs).hasSize(1);
+
assertThat(table.fileIO().readFileUtf8(outputs.get(0))).isEqualTo("index
payload");
+ assertThat(publications).hasValue(0);
+ assertThat(table.latestSnapshot()).isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testUncertainPublicationRetainsVisibleIndex(boolean
publicationThrows) throws Exception {
+ FileStoreTable table = createTable();
+ ManifestCommittable pending = prepareAppend(table, 10, 20);
+ List<Path> outputs = new ArrayList<>();
+ AtomicInteger attempts = new AtomicInteger();
+ AtomicInteger publications = new AtomicInteger();
+ SnapshotCommit publisher =
+ new DelegatingSnapshotCommit(table) {
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ publications.incrementAndGet();
+ assertThat(delegate.commit(baseUuid, snapshot, branch,
statistics))
+ .isTrue();
+ if (publicationThrows) {
+ throw new IOException("publication response lost");
+ }
+ return false;
+ }
+ };
+ CommitChangesRewriter rewriter =
+ new CommitChangesRewriter() {
+ @Override
+ public Attempt createAttempt(Snapshot.CommitKind kind,
CommitChanges changes) {
+ attempts.incrementAndGet();
+ return (snapshotId, assigned, indexes, ioManager) ->
+ writeIndex(table, assigned.get(0), outputs);
+ }
+ };
+ try (FileStoreCommit commit =
+ newCommit(table, publisher, Collections.emptyList())
+ .withCommitChangesRewriter(rewriter)) {
+ commit.commit(pending, false);
+ }
+
+ assertThat(publications).hasValue(1);
+ assertThat(attempts).hasValue(1);
+ // Reconciliation must happen before constructing another attempt or
rewriting another
+ // index.
+ assertThat(outputs).hasSize(1);
+
assertThat(table.fileIO().readFileUtf8(outputs.get(0))).isEqualTo("index
payload");
+ assertThat(table.latestSnapshot().get().id()).isEqualTo(1L);
+ assertThat(table.latestSnapshot().get().nextRowId()).isEqualTo(2L);
+ assertThat(read(table)).extracting(row ->
row.getInt(0)).containsExactlyInAnyOrder(10, 20);
+ List<IndexManifestEntry> indexes =
table.store().newIndexFileHandler().scanEntries();
+ assertThat(indexes).hasSize(1);
+
assertThat(indexes.get(0).indexFile().fileName()).isEqualTo(outputs.get(0).getName());
+ assertThat(indexes.get(0).indexFile().globalIndexMeta().rowRange())
+ .isEqualTo(new Range(0, 1));
+ }
+
+ private FileStoreTable createTable() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.WRITE_ONLY.key(), "true")
+ .option(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "0ms")
+ .option(CoreOptions.COMMIT_MAX_RETRY_WAIT.key(), "0ms")
+ .build(),
+ false);
+ return getTableDefault();
+ }
+
+ private ManifestCommittable prepareAppend(FileStoreTable table, int...
values)
+ throws Exception {
+ ManifestCommittable pending = new ManifestCommittable(17L);
+ try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) {
+ for (int value : values) {
+ write.write(GenericRow.of(value));
+ }
+ write.prepareCommit().forEach(pending::addFileCommittable);
+ }
+ return pending;
+ }
+
+ private CommitChangesRewriter indexRewriter(FileStoreTable table,
List<Path> outputs) {
+ return new CommitChangesRewriter() {
+ @Override
+ public Attempt createAttempt(Snapshot.CommitKind kind,
CommitChanges changes) {
+ return (snapshotId, assigned, indexes, ioManager) ->
+ writeIndex(table, assigned.get(0), outputs);
+ }
+ };
+ }
+
+ private List<IndexManifestEntry> writeIndex(
+ FileStoreTable table, ManifestEntry assigned, List<Path> outputs)
throws IOException {
+ Path path =
table.store().pathFactory().globalIndexFileFactory().newPath();
+ outputs.add(path);
+ table.fileIO().writeFile(path, "index payload", false);
+ IndexFileMeta index =
+ new IndexFileMeta(
+ "test-index",
+ path.getName(),
+ 13,
+ assigned.file().rowCount(),
+ new GlobalIndexMeta(
+ assigned.file().nonNullFirstRowId(),
+ assigned.file().nonNullRowIdRange().to,
+ 0,
+ null,
+ null),
+ null);
+ return Collections.singletonList(
+ new IndexManifestEntry(
+ FileKind.ADD, assigned.partition(), assigned.bucket(),
index));
+ }
+
+ private SnapshotCommit countingPublisher(FileStoreTable table,
AtomicInteger publications) {
+ return new DelegatingSnapshotCommit(table) {
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ publications.incrementAndGet();
+ return delegate.commit(baseUuid, snapshot, branch, statistics);
+ }
+ };
+ }
+
+ private FileStoreCommitImpl newCommit(
+ FileStoreTable table, SnapshotCommit publisher,
List<CommitPreCallback> callbacks) {
+ AppendOnlyFileStore store = (AppendOnlyFileStore) table.store();
+ CoreOptions options = store.options();
+ return new FileStoreCommitImpl(
+ publisher,
+ table.fileIO(),
+ new FileSystemSchemaManager(table.fileIO(), table.location()),
+ "rewriter-test",
+ "rewriter-test-user",
+ store.partitionType(),
+ options,
+ store.pathFactory(),
+ store.snapshotManager(),
+ store.manifestFileFactory(),
+ store.manifestListFactory(),
+ store.indexManifestFileFactory(),
+ store::newScan,
+ store.newStatsFileHandler(),
+ store.bucketMode(),
+ callbacks,
+ Collections.emptyList(),
+ scanner ->
+ ConflictDetection.create(
+ "rewriter-test",
+ "rewriter-test-user",
+ store.partitionType(),
+ store.pathFactory(),
+ store.newKeyComparator(),
+ store.bucketMode(),
+ options.deletionVectorsEnabled(),
+ false,
+ false,
+ options.pkClusteringOverride(),
+ store.newIndexFileHandler(),
+ store.snapshotManager(),
+ scanner),
+ null);
+ }
+
+ private static void assertReadOnly(List<?> entries) {
+
assertThatThrownBy(entries::clear).isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ private static class DelegatingSnapshotCommit extends SnapshotCommit {
+
+ protected final SnapshotCommit delegate;
+
+ private DelegatingSnapshotCommit(FileStoreTable table) {
+ super(Lock.empty());
+ delegate = new RenamingSnapshotCommit(table.snapshotManager(),
Lock.empty());
+ }
+
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ return delegate.commit(baseUuid, snapshot, branch, statistics);
+ }
+
+ @Override
+ protected void closeImpl() throws Exception {
+ delegate.close();
+ }
+ }
+}
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 4c9291b25b..110d81faaa 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
@@ -24,6 +24,7 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.TestAppendFileStore;
import org.apache.paimon.TestFileStore;
import org.apache.paimon.TestKeyValueGenerator;
+import
org.apache.paimon.append.dataevolution.MaterializeDeletionVectorsCommitChangesRewriter;
import org.apache.paimon.catalog.RenamingSnapshotCommit;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
@@ -1021,16 +1022,27 @@ public class FileStoreCommitTest {
otherBucketDelete,
nonGlobalDelete));
- try (FileStoreCommitImpl commit = store.newCommit()) {
- CommitChanges refreshed =
- commit.compactChangesProvider(
- changes,
Collections.singleton(Pair.of(partition, 0)))
- .provide(null);
-
- // Global-index removals from earlier preparation are replaced for
the entire bucket,
- // even if their ranges do not overlap the materialized input.
-
assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete,
nonGlobalDelete);
- }
+ CommitMessage message =
+ new CommitMessageImpl(
+ partition,
+ 0,
+ 2,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(
+ addFile(partition, 0, 2,
0).file().assignFirstRowId(0)),
+ Collections.emptyList(),
+ Collections.emptyList()));
+ MaterializeDeletionVectorsCommitChangesRewriter rewriter =
+ new MaterializeDeletionVectorsCommitChangesRewriter(
+ FileStoreTableFactory.create(store.fileIO(),
store.options().path()));
+ CommitChanges refreshed =
+ rewriter.compactChangesProvider(changes,
Collections.singletonList(message))
+ .provide(null);
+
+ // Global-index removals from earlier preparation are replaced for the
entire bucket,
+ // even if their ranges do not overlap the materialized input.
+ assertThat(refreshed.indexFiles).containsExactly(otherBucketDelete,
nonGlobalDelete);
}
private static IndexManifestEntry globalIndexDeleteEntry(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
index 1ec1343ab1..2620f51746 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -26,6 +26,9 @@ import
org.apache.paimon.append.dataevolution.DataEvolutionCompactTaskSerializer
import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import
org.apache.paimon.append.dataevolution.DataEvolutionDeletionVectorMaterializeCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassigner;
+import
org.apache.paimon.append.dataevolution.MaterializeDeletionVectorsCommitChangesRewriter;
+import org.apache.paimon.catalog.RenamingSnapshotCommit;
+import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
@@ -39,7 +42,9 @@ import org.apache.paimon.format.blob.BlobFileFormat;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
@@ -50,7 +55,11 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
+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;
@@ -71,6 +80,7 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RangeHelper;
+import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -87,6 +97,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
@@ -895,6 +906,125 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
.doesNotContainAnyElementsOf(oldAnchorFiles);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"btree", "bitmap"})
+ public void testMaterializeRetryRefreshesConcurrentScalarIndexes(String
indexType)
+ throws Exception {
+ Map<String, String> retryOptions = new HashMap<>();
+ retryOptions.put(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "0ms");
+ retryOptions.put(CoreOptions.COMMIT_MAX_RETRY_WAIT.key(), "0ms");
+ FileStoreTable original =
+
createPartitionedReassignTable("materialize_retry_index_table", false)
+ .copy(retryOptions);
+ writePartitionRows(original, "a", 0, 1, 2);
+ writePartitionRows(original, "a", 3, 4, 5);
+ commitDeletionVectors(
+ original,
+ partition(original, "a"),
+ Collections.singletonList(new DvSpec(new Range(0, 2), 1, 2)));
+ Snapshot planningSnapshot = original.latestSnapshot().get();
+ List<CommitMessage> materialized =
+ prepareMaterializeDeletionVectors(original, planningSnapshot,
null);
+ AtomicInteger attempts = new AtomicInteger();
+ List<String> untouchedIndexes = new ArrayList<>();
+ CatalogEnvironment environment =
+ new CatalogEnvironment(null, null, null, null, null, null,
false, false) {
+ @Override
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options,
String commitUser) {
+ SnapshotCommit delegate = new
RenamingSnapshotCommit(manager, Lock.empty());
+ return new SnapshotCommit(Lock.empty()) {
+ @Override
+ protected boolean commitImpl(
+ String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ int attempt = attempts.incrementAndGet();
+ if (attempt == 1) {
+ List<CommitMessage> indexes = new
ArrayList<>();
+ for (DataSplit split :
+ original.newSnapshotReader()
+
.withSnapshot(planningSnapshot)
+ .read()
+ .dataSplits()) {
+ for (DataFileMeta file :
split.dataFiles()) {
+ indexes.addAll(
+
SortedGlobalIndexTestUtils.buildIndex(
+ original,
+ indexType,
+ "f0",
+
split.filterDataFile(file::equals)
+ .get(),
+
planningSnapshot.id()));
+ }
+ }
+
DataEvolutionDeletionVectorTest.commit(original, indexes);
+
assertThat(liveGlobalIndexFileNames(original)).hasSize(2);
+ for (IndexManifestEntry entry :
+
original.store().newIndexFileHandler().scanEntries()) {
+ GlobalIndexMeta meta =
entry.indexFile().globalIndexMeta();
+ if (meta != null
+ && meta.rowRange().equals(new
Range(3, 5))) {
+
untouchedIndexes.add(entry.indexFile().fileName());
+ }
+ }
+ assertThat(untouchedIndexes).hasSize(1);
+ // Move the next available row ID after
the first candidate has
+ // already assigned its output. The retry
must assign it again.
+ writePartitionRows(original, "a", 6);
+ }
+ boolean committed =
+ delegate.commit(baseUuid, snapshot,
branch, statistics);
+ assertThat(committed).isEqualTo(attempt == 2);
+ return committed;
+ }
+
+ @Override
+ protected void closeImpl() throws Exception {
+ delegate.close();
+ }
+ };
+ }
+ };
+ FileStoreTable committingTable =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(),
original.schema(), environment);
+ commitMaterializeDeletionVectors(
+ committingTable, planningSnapshot, materialized,
"test-materialize-index-retry");
+
+ assertThat(attempts).hasValue(2);
+
assertThat(original.latestSnapshot().get().id()).isEqualTo(planningSnapshot.id()
+ 3);
+ assertThat(original.latestSnapshot().get().nextRowId()).isEqualTo(8L);
+ assertThat(readPartitionedRowsWithRowIds(original))
+ .containsExactly("a|3|3", "a|4|4", "a|5|5", "a|6|6", "a|0|7");
+ assertThat(readPartitionedRows(original))
+ .containsExactlyInAnyOrder(
+ "a|0|base-0", "a|3|base-3", "a|4|base-4",
"a|5|base-5", "a|6|base-6");
+ assertThat(liveDeletionVectorDataFileNames(original)).isEmpty();
+
assertThat(liveGlobalIndexFileNames(original)).containsExactlyElementsOf(untouchedIndexes);
+ assertThat(
+
original.store().newIndexFileHandler().scanEntries().stream()
+ .filter(entry ->
entry.indexFile().globalIndexMeta() != null)
+ .map(entry ->
entry.indexFile().globalIndexMeta().rowRange()))
+ .containsExactly(new Range(3, 5));
+ Predicate predicate = new
PredicateBuilder(original.rowType()).equal(1, 4);
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.create(
+ original, PartitionPredicate.ALWAYS_TRUE,
predicate)
+ .get()) {
+ assertThat(scanner.scan(predicate).get().results().toRangeList())
+ .containsExactly(new Range(4, 4));
+ assertThat(
+ scanner.scan(new
PredicateBuilder(original.rowType()).equal(1, 1))
+ .get()
+ .results()
+ .toRangeList())
+ .isEmpty();
+ }
+ }
+
@Test
public void testStaleMaterializeRejectedAfterConcurrentUpdate() throws
Exception {
createTableDefault();
@@ -1492,7 +1622,10 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
((CommitMessageImpl) message)
.withCheckFromSnapshot(snapshot.id()))
.collect(Collectors.toList());
- commit.materializeDvRowIdCheck().commit(checkedMessages);
+ commit.materializeDvRowIdCheck()
+ .withCommitChangesRewriter(
+ new
MaterializeDeletionVectorsCommitChangesRewriter(table))
+ .commit(checkedMessages);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
index 3a986dca07..45afd6c040 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/IncrementalClusterCompact.java
@@ -35,10 +35,10 @@ import
org.apache.paimon.flink.cluster.IncrementalClusterSplitSource;
import
org.apache.paimon.flink.cluster.RewriteIncrementalClusterCommittableOperator;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.sink.DataEvolutionClusterSink;
import org.apache.paimon.flink.sink.DataEvolutionCommitPreparationOperator;
import org.apache.paimon.flink.sink.DataEvolutionCompactionTaskTypeInfo;
import org.apache.paimon.flink.sink.DataEvolutionCompactionWorkerOperator;
-import org.apache.paimon.flink.sink.DataEvolutionDeletionVectorMaterializeSink;
import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.flink.sink.RowAppendTableSink;
import org.apache.paimon.flink.sorter.SortOperator;
@@ -405,8 +405,7 @@ public class IncrementalClusterCompact {
DataEvolutionCommitPreparationOperator.Factory.forClustering(
table, snapshot))
.forceNonParallel();
- new DataEvolutionDeletionVectorMaterializeSink(table, snapshot)
- .doCommit(dataStream, commitUser);
+ new DataEvolutionClusterSink(table).doCommit(dataStream,
commitUser);
} else {
sink.doCommit(dataStream, commitUser);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionClusterSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionClusterSink.java
new file mode 100644
index 0000000000..a53fa4abba
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionClusterSink.java
@@ -0,0 +1,49 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.paimon.append.dataevolution.ClusteringCommitChangesRewriter;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
+
+/** Sink which commits data-evolution clustering and accompanying
row-preserving compaction. */
+public class DataEvolutionClusterSink extends RowAppendTableSink {
+
+ private static final long serialVersionUID = 1L;
+
+ public DataEvolutionClusterSink(FileStoreTable table) {
+ super(table, null, null);
+ }
+
+ @Override
+ protected Committer.Factory<Committable, ManifestCommittable>
createCommitterFactory() {
+ return context -> {
+ TableCommitImpl commit = table.newCommit(context.commitUser());
+ commit.materializeDvRowIdCheck();
+ commit.withCommitChangesRewriter(new
ClusteringCommitChangesRewriter(table));
+ return new StoreCommitter(table, commit, context);
+ };
+ }
+
+ @Override
+ protected CommittableStateManager<ManifestCommittable>
createCommittableStateManager() {
+ return new NoopCommittableStateManager();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionDeletionVectorMaterializeSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionDeletionVectorMaterializeSink.java
index 3ca51a69d5..553989ac38 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionDeletionVectorMaterializeSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DataEvolutionDeletionVectorMaterializeSink.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import
org.apache.paimon.append.dataevolution.MaterializeDeletionVectorsCommitChangesRewriter;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.TableCommitImpl;
@@ -75,6 +76,8 @@ public class DataEvolutionDeletionVectorMaterializeSink
return context -> {
TableCommitImpl commit = table.newCommit(context.commitUser());
commit.materializeDvRowIdCheck();
+ commit.withCommitChangesRewriter(
+ new
MaterializeDeletionVectorsCommitChangesRewriter(table));
return new StoreCommitter(table, commit, context);
};
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index c69c789361..fbed108b36 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -25,6 +25,7 @@ import org.apache.paimon.append.AppendCompactCoordinator;
import org.apache.paimon.append.AppendCompactTask;
import org.apache.paimon.append.cluster.DataEvolutionClusterCompaction;
import org.apache.paimon.append.cluster.IncrementalClusterManager;
+import org.apache.paimon.append.dataevolution.ClusteringCommitChangesRewriter;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
import org.apache.paimon.compact.CompactUnit;
@@ -987,6 +988,7 @@ public class CompactProcedure extends BaseProcedure {
if (table.coreOptions().dataEvolutionEnabled()) {
String commitUser =
createCommitUser(table.coreOptions().toConfiguration());
try (TableCommitImpl commit =
table.newCommit(commitUser).materializeDvRowIdCheck()) {
+ commit.withCommitChangesRewriter(new
ClusteringCommitChangesRewriter(table));
clusterMessages.addAll(
DataEvolutionRewriteExecutor.executeTasks(
table,
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MaterializeDeletionVectorsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MaterializeDeletionVectorsProcedure.java
index 7377d8fb6e..c742a673bf 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MaterializeDeletionVectorsProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MaterializeDeletionVectorsProcedure.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
import
org.apache.paimon.append.dataevolution.DataEvolutionDeletionVectorMaterializeCoordinator;
+import
org.apache.paimon.append.dataevolution.MaterializeDeletionVectorsCommitChangesRewriter;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.spark.utils.SparkProcedureUtils;
import org.apache.paimon.table.BucketMode;
@@ -176,6 +177,8 @@ public class MaterializeDeletionVectorsProcedure extends
BaseProcedure {
@Override
public void configure(TableCommitImpl commit) {
commit.materializeDvRowIdCheck();
+ commit.withCommitChangesRewriter(
+ new
MaterializeDeletionVectorsCommitChangesRewriter(table));
}
@Override
