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 c638a4d427 [core] Extract the shared incremental clustering rewriter
(#10679)
c638a4d427 is described below
commit c638a4d42756c84c14d1e503f23f024d05463930
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:39:08 2026 +0800
[core] Extract the shared incremental clustering rewriter (#10679)
---
.../append/cluster/HistoryPartitionCluster.java | 15 ++-
.../append/cluster/IncrementalClusterManager.java | 101 +++--------------
.../append/cluster/IncrementalClusterRewriter.java | 123 +++++++++++++++++++++
.../cluster/HistoryPartitionClusterTest.java | 10 +-
.../cluster/IncrementalClusterManagerTest.java | 44 +++++---
.../flink/compact/IncrementalClusterCompact.java | 4 +-
.../paimon/spark/procedure/CompactProcedure.java | 3 +-
7 files changed, 190 insertions(+), 110 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
index 8acbb8e778..d7d5081366 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/HistoryPartitionCluster.java
@@ -44,9 +44,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.function.Function;
import java.util.stream.Collectors;
-import static
org.apache.paimon.append.cluster.IncrementalClusterManager.constructPartitionLevels;
import static
org.apache.paimon.append.cluster.IncrementalClusterManager.logForPartitionLevel;
/** Handle historical partition for full clustering. */
@@ -62,6 +62,7 @@ public class HistoryPartitionCluster {
private final Duration historyPartitionIdleTime;
private final int historyPartitionLimit;
private final int maxLevel;
+ private final Function<List<DataFileMeta>, List<LevelSortedRun>>
levelConstructor;
public HistoryPartitionCluster(
FileStoreTable table,
@@ -70,7 +71,8 @@ public class HistoryPartitionCluster {
InternalRowPartitionComputer partitionComputer,
PartitionPredicate specifiedPartitions,
Duration historyPartitionIdleTime,
- int historyPartitionLimit) {
+ int historyPartitionLimit,
+ Function<List<DataFileMeta>, List<LevelSortedRun>>
levelConstructor) {
this.table = table;
this.snapshot = snapshot;
this.incrementalClusterStrategy = incrementalClusterStrategy;
@@ -79,6 +81,7 @@ public class HistoryPartitionCluster {
this.historyPartitionIdleTime = historyPartitionIdleTime;
this.historyPartitionLimit = historyPartitionLimit;
this.maxLevel = table.coreOptions().numLevels() - 1;
+ this.levelConstructor = levelConstructor;
}
@Nullable
@@ -87,7 +90,8 @@ public class HistoryPartitionCluster {
Snapshot snapshot,
IncrementalClusterStrategy incrementalClusterStrategy,
InternalRowPartitionComputer partitionComputer,
- @Nullable PartitionPredicate specifiedPartitions) {
+ @Nullable PartitionPredicate specifiedPartitions,
+ Function<List<DataFileMeta>, List<LevelSortedRun>>
levelConstructor) {
if (table.schema().partitionKeys().isEmpty()) {
return null;
}
@@ -108,7 +112,8 @@ public class HistoryPartitionCluster {
partitionComputer,
specifiedPartitions,
idleTime,
- limit);
+ limit,
+ levelConstructor);
}
public Map<BinaryRow, CompactUnit> pickForHistoryPartitions() {
@@ -163,7 +168,7 @@ public class HistoryPartitionCluster {
.collect(
Collectors.toMap(
Map.Entry::getKey,
- entry ->
constructPartitionLevels(entry.getValue())));
+ entry ->
levelConstructor.apply(entry.getValue())));
}
private Map<BinaryRow, List<DataFileMeta>> filterPartitions(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
index f5642f6648..2fa240e028 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterManager.java
@@ -25,21 +25,13 @@ import org.apache.paimon.compact.CompactUnit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
-import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.io.DataIncrement;
-import org.apache.paimon.manifest.FileKind;
-import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.mergetree.LevelSortedRun;
-import org.apache.paimon.mergetree.SortedRun;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
-import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.SplitGenerator;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.utils.InternalRowPartitionComputer;
@@ -51,8 +43,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -71,18 +61,17 @@ public class IncrementalClusterManager {
private final InternalRowPartitionComputer partitionComputer;
private final Snapshot snapshot;
private final SnapshotReader snapshotReader;
+ private final IncrementalClusterRewriter rewriter;
private final IncrementalClusterStrategy incrementalClusterStrategy;
private final CoreOptions.OrderType clusterCurve;
private final List<String> clusterKeys;
private final int numLevels;
private final @Nullable HistoryPartitionCluster historyPartitionCluster;
- public IncrementalClusterManager(FileStoreTable table) {
- this(table, null);
- }
-
public IncrementalClusterManager(
- FileStoreTable table, @Nullable PartitionPredicate
specifiedPartitions) {
+ FileStoreTable table,
+ @Nullable PartitionPredicate specifiedPartitions,
+ Snapshot snapshot) {
checkArgument(
table.bucketMode() == BucketMode.BUCKET_UNAWARE,
"only append unaware-bucket table support incremental
clustering.");
@@ -99,7 +88,8 @@ public class IncrementalClusterManager {
table.store().partitionType(),
table.partitionKeys().toArray(new String[0]),
table.coreOptions().legacyPartitionName());
- this.snapshot = table.snapshotManager().latestSnapshot();
+ this.snapshot = snapshot;
+ this.rewriter = IncrementalClusterRewriter.create(table, snapshot);
this.snapshotReader =
table.newSnapshotReader()
.dropStats()
@@ -120,7 +110,8 @@ public class IncrementalClusterManager {
snapshot,
incrementalClusterStrategy,
partitionComputer,
- specifiedPartitions);
+ specifiedPartitions,
+ rewriter::constructPartitionLevels);
}
public Map<BinaryRow, CompactUnit> createCompactUnits(boolean
fullCompaction) {
@@ -177,33 +168,7 @@ public class IncrementalClusterManager {
.collect(
Collectors.toMap(
Map.Entry::getKey,
- entry ->
constructPartitionLevels(entry.getValue())));
- }
-
- public static List<LevelSortedRun>
constructPartitionLevels(List<DataFileMeta> partitionFiles) {
- List<LevelSortedRun> partitionLevels = new ArrayList<>();
- Map<Integer, List<DataFileMeta>> levelMap =
-
partitionFiles.stream().collect(Collectors.groupingBy(DataFileMeta::level));
-
- for (Map.Entry<Integer, List<DataFileMeta>> entry :
levelMap.entrySet()) {
- int level = entry.getKey();
- if (level == 0) {
- for (DataFileMeta level0File : entry.getValue()) {
- partitionLevels.add(
- new LevelSortedRun(level,
SortedRun.fromSingle(level0File)));
- }
- } else {
- // don't need to guarantee that the files within the same
sorted run are
- // non-overlapping here, so we call SortedRun.fromSorted() to
avoid sorting and
- // validation
- partitionLevels.add(
- new LevelSortedRun(level,
SortedRun.fromSorted(entry.getValue())));
- }
- }
-
- // sort by level
- partitionLevels.sort(Comparator.comparing(LevelSortedRun::level));
- return partitionLevels;
+ entry ->
rewriter.constructPartitionLevels(entry.getValue())));
}
public Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>>
toSplitsAndRewriteDvFiles(
@@ -235,48 +200,14 @@ public class IncrementalClusterManager {
String bucketPath =
snapshotReader.pathFactory().bucketPath(partition,
0).toString();
- builder.withDataFiles(dataFiles)
- .rawConvertible(splitGroup.rawConvertible)
- .withBucketPath(bucketPath);
-
- if (dvMaintainer != null) {
- List<DeletionFile> dataDeletionFiles = new ArrayList<>();
- for (DataFileMeta file : dataFiles) {
- DeletionFile deletionFile =
-
dvMaintainer.notifyRemovedDeletionVector(file.fileName());
- dataDeletionFiles.add(deletionFile);
- }
- builder.withDataDeletionFiles(dataDeletionFiles);
- }
+ builder.withDataFiles(dataFiles).withBucketPath(bucketPath);
+
+ rewriter.configureSplit(builder, splitGroup, dvMaintainer);
splits.add(builder.build());
}
- // generate delete dv index meta
- CommitMessage dvCommitMessage = null;
- if (dvMaintainer != null) {
- List<IndexFileMeta> newIndexFiles = new ArrayList<>();
- List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
- List<IndexManifestEntry> indexEntries = dvMaintainer.persist();
- for (IndexManifestEntry entry : indexEntries) {
- if (entry.kind() == FileKind.ADD) {
- newIndexFiles.add(entry.indexFile());
- } else {
- deletedIndexFiles.add(entry.indexFile());
- }
- }
- dvCommitMessage =
- new CommitMessageImpl(
- dvMaintainer.getPartition(),
- 0,
- table.coreOptions().bucket(),
- DataIncrement.emptyIncrement(),
- new CompactIncrement(
- Collections.emptyList(),
- Collections.emptyList(),
- Collections.emptyList(),
- newIndexFiles,
- deletedIndexFiles));
- }
+ CommitMessage dvCommitMessage =
+ rewriter.finishPartition(dvMaintainer,
table.coreOptions().bucket());
result.put(partition, Pair.of(splits, dvCommitMessage));
}
@@ -284,6 +215,10 @@ public class IncrementalClusterManager {
return result;
}
+ public List<CommitMessage> prepareCommitMessages(List<CommitMessage>
messages) {
+ return rewriter.prepareCommitMessages(messages);
+ }
+
public static List<DataFileMeta> upgrade(
List<DataFileMeta> filesAfterCluster, int outputLevel) {
return filesAfterCluster.stream()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
new file mode 100644
index 0000000000..91fd178567
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/cluster/IncrementalClusterRewriter.java
@@ -0,0 +1,123 @@
+/*
+ * 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.cluster;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.mergetree.LevelSortedRun;
+import org.apache.paimon.mergetree.SortedRun;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.SplitGenerator;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Table-specific file and deletion-vector semantics for incremental
clustering. */
+public class IncrementalClusterRewriter {
+
+ public static IncrementalClusterRewriter create(FileStoreTable table,
Snapshot snapshot) {
+ return new IncrementalClusterRewriter();
+ }
+
+ List<LevelSortedRun> constructPartitionLevels(List<DataFileMeta>
partitionFiles) {
+ List<LevelSortedRun> partitionLevels = new ArrayList<>();
+ Map<Integer, List<DataFileMeta>> levelMap =
+
partitionFiles.stream().collect(Collectors.groupingBy(DataFileMeta::level));
+
+ for (Map.Entry<Integer, List<DataFileMeta>> entry :
levelMap.entrySet()) {
+ int level = entry.getKey();
+ if (level == 0) {
+ for (DataFileMeta level0File : entry.getValue()) {
+ partitionLevels.add(
+ new LevelSortedRun(level,
SortedRun.fromSingle(level0File)));
+ }
+ } else {
+ // don't need to guarantee that the files within the same
sorted run are
+ // non-overlapping here, so we call SortedRun.fromSorted() to
avoid sorting and
+ // validation
+ partitionLevels.add(
+ new LevelSortedRun(level,
SortedRun.fromSorted(entry.getValue())));
+ }
+ }
+
+ // sort by level
+ partitionLevels.sort(Comparator.comparing(LevelSortedRun::level));
+ return partitionLevels;
+ }
+
+ void configureSplit(
+ DataSplit.Builder builder,
+ SplitGenerator.SplitGroup splitGroup,
+ @Nullable AppendDeleteFileMaintainer dvMaintainer) {
+ builder.rawConvertible(splitGroup.rawConvertible);
+ if (dvMaintainer != null) {
+ builder.withDataDeletionFiles(
+ splitGroup.files.stream()
+ .map(file ->
dvMaintainer.notifyRemovedDeletionVector(file.fileName()))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ @Nullable
+ CommitMessage finishPartition(
+ @Nullable AppendDeleteFileMaintainer dvMaintainer, int
totalBuckets) {
+ if (dvMaintainer == null) {
+ return null;
+ }
+ List<IndexFileMeta> newIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ for (IndexManifestEntry entry : dvMaintainer.persist()) {
+ if (entry.kind() == FileKind.ADD) {
+ newIndexFiles.add(entry.indexFile());
+ } else {
+ deletedIndexFiles.add(entry.indexFile());
+ }
+ }
+ return new CommitMessageImpl(
+ dvMaintainer.getPartition(),
+ 0,
+ totalBuckets,
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ newIndexFiles,
+ deletedIndexFiles));
+ }
+
+ public List<CommitMessage> prepareCommitMessages(List<CommitMessage>
messages) {
+ return messages;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
index 2d64663f9a..566d70e20d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/HistoryPartitionClusterTest.java
@@ -79,7 +79,8 @@ public class HistoryPartitionClusterTest {
table,
PartitionPredicate.fromMultiple(
RowType.of(DataTypes.INT()),
-
Lists.newArrayList(BinaryRow.singleColumn("pt1"))))
+
Lists.newArrayList(BinaryRow.singleColumn("pt1"))),
+ table.snapshotManager().latestSnapshot())
.historyPartitionCluster();
Map<BinaryRow, List<LevelSortedRun>> partitionLevels =
historyPartitionCluster.constructLevelsForHistoryPartitions();
@@ -92,14 +93,17 @@ public class HistoryPartitionClusterTest {
table,
PartitionPredicate.fromMultiple(
RowType.of(DataTypes.INT()),
-
Lists.newArrayList(BinaryRow.singleColumn("pt3"))))
+
Lists.newArrayList(BinaryRow.singleColumn("pt3"))),
+ table.snapshotManager().latestSnapshot())
.historyPartitionCluster();
partitionLevels =
historyPartitionCluster.constructLevelsForHistoryPartitions();
assertThat(partitionLevels.size()).isEqualTo(1);
assertThat(partitionLevels.get(BinaryRow.singleColumn("pt1"))).isNotEmpty();
// test not specify partition and disable history partition auto
clustering
- historyPartitionCluster = new
IncrementalClusterManager(table).historyPartitionCluster();
+ historyPartitionCluster =
+ new IncrementalClusterManager(table, null,
table.snapshotManager().latestSnapshot())
+ .historyPartitionCluster();
assertThat(historyPartitionCluster).isNull();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
index 464c3c7703..2997ec5a9e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/cluster/IncrementalClusterManagerTest.java
@@ -64,19 +64,17 @@ public class IncrementalClusterManagerTest {
options.put(CoreOptions.BUCKET.key(), "-1");
options.put(CoreOptions.CLUSTERING_INCREMENTAL.key(), "false");
FileStoreTable table = createTable(options, Collections.emptyList());
- assertThatThrownBy(() -> new IncrementalClusterManager(table))
+ assertThatThrownBy(
+ () ->
+ new IncrementalClusterManager(
+ table, null,
table.snapshotManager().latestSnapshot()))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining(
"Only support incremental clustering when
'clustering.incremental' is true.");
}
@Test
- public void testConstructPartitionLevels() throws Exception {
- // Create a valid table for IncrementalClusterManager
- Map<String, String> options = new HashMap<>();
- FileStoreTable table = createTable(options, Collections.emptyList());
- IncrementalClusterManager incrementalClusterManager = new
IncrementalClusterManager(table);
-
+ public void testConstructPartitionLevels() {
// Create test files with different levels
List<DataFileMeta> partitionFiles = new ArrayList<>();
@@ -98,7 +96,7 @@ public class IncrementalClusterManagerTest {
// Call the method under test
List<LevelSortedRun> result =
-
incrementalClusterManager.constructPartitionLevels(partitionFiles);
+ new
IncrementalClusterRewriter().constructPartitionLevels(partitionFiles);
// Verify the results
assertThat(result).hasSize(4); // 2 level-0 runs + 1 level-1 run + 1
level-2 run
@@ -152,7 +150,9 @@ public class IncrementalClusterManagerTest {
// Test default mode is GLOBAL_SORT
Map<String, String> options = new HashMap<>();
FileStoreTable table = createTable(options, Collections.emptyList());
- IncrementalClusterManager manager = new
IncrementalClusterManager(table);
+ IncrementalClusterManager manager =
+ new IncrementalClusterManager(
+ table, null, table.snapshotManager().latestSnapshot());
assertThat(manager.clusteringIncrementalMode())
.isEqualTo(CoreOptions.ClusteringIncrementalMode.GLOBAL_SORT);
}
@@ -163,7 +163,9 @@ public class IncrementalClusterManagerTest {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.CLUSTERING_INCREMENTAL_MODE.key(),
"local-sort");
FileStoreTable table = createTable(options, Collections.emptyList());
- IncrementalClusterManager manager = new
IncrementalClusterManager(table);
+ IncrementalClusterManager manager =
+ new IncrementalClusterManager(
+ table, null, table.snapshotManager().latestSnapshot());
assertThat(manager.clusteringIncrementalMode())
.isEqualTo(CoreOptions.ClusteringIncrementalMode.LOCAL_SORT);
}
@@ -174,7 +176,9 @@ public class IncrementalClusterManagerTest {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.CLUSTERING_INCREMENTAL_MODE.key(),
"global-sort");
FileStoreTable table = createTable(options, Collections.emptyList());
- IncrementalClusterManager manager = new
IncrementalClusterManager(table);
+ IncrementalClusterManager manager =
+ new IncrementalClusterManager(
+ table, null, table.snapshotManager().latestSnapshot());
assertThat(manager.clusteringIncrementalMode())
.isEqualTo(CoreOptions.ClusteringIncrementalMode.GLOBAL_SORT);
}
@@ -211,7 +215,8 @@ public class IncrementalClusterManagerTest {
table,
PartitionPredicate.fromMultiple(
RowType.of(DataTypes.INT()),
-
Lists.newArrayList(BinaryRow.singleColumn("pt3"))));
+
Lists.newArrayList(BinaryRow.singleColumn("pt3"))),
+ table.snapshotManager().latestSnapshot());
Map<BinaryRow, CompactUnit> partitionLevels =
incrementalClusterManager.createCompactUnits(true);
assertThat(partitionLevels.size()).isEqualTo(2);
@@ -219,21 +224,26 @@ public class IncrementalClusterManagerTest {
assertThat(partitionLevels.get(BinaryRow.singleColumn("pt3"))).isNotNull();
// test don't specify partition and enable history partition auto
clustering
- incrementalClusterManager = new IncrementalClusterManager(table);
+ incrementalClusterManager =
+ new IncrementalClusterManager(
+ table, null, table.snapshotManager().latestSnapshot());
partitionLevels = incrementalClusterManager.createCompactUnits(true);
assertThat(partitionLevels.size()).isEqualTo(4);
// test specify partition and disable history partition auto clustering
SchemaChange schemaChange =
SchemaChange.removeOption(CoreOptions.CLUSTERING_HISTORY_PARTITION_IDLE_TIME.key());
+ FileStoreTable tableWithoutHistoryClustering =
+ table.copy(
+ table.schemaManager()
+
.commitChanges(Collections.singletonList(schemaChange)));
incrementalClusterManager =
new IncrementalClusterManager(
- table.copy(
- table.schemaManager()
-
.commitChanges(Collections.singletonList(schemaChange))),
+ tableWithoutHistoryClustering,
PartitionPredicate.fromMultiple(
RowType.of(DataTypes.INT()),
-
Lists.newArrayList(BinaryRow.singleColumn("pt3"))));
+
Lists.newArrayList(BinaryRow.singleColumn("pt3"))),
+
tableWithoutHistoryClustering.snapshotManager().latestSnapshot());
partitionLevels = incrementalClusterManager.createCompactUnits(true);
assertThat(partitionLevels.size()).isEqualTo(1);
assertThat(partitionLevels.get(BinaryRow.singleColumn("pt3"))).isNotNull();
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 a4f25486e1..0367adce8c 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
@@ -88,7 +88,9 @@ public class IncrementalClusterCompact {
@Nullable Boolean fullCompaction) {
this.env = env;
this.table = table;
- this.clusterManager = new IncrementalClusterManager(table,
partitionPredicate);
+ this.clusterManager =
+ new IncrementalClusterManager(
+ table, partitionPredicate,
table.snapshotManager().latestSnapshot());
Options options = new Options(table.options());
this.partitionComputer = table.store().partitionComputer();
this.commitUser = CoreOptions.createCommitUser(options);
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 218dc6d0a9..7acca09d85 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
@@ -844,7 +844,8 @@ public class CompactProcedure extends BaseProcedure {
boolean fullCompaction,
DataSourceV2Relation relation) {
IncrementalClusterManager incrementalClusterManager =
- new IncrementalClusterManager(table, partitionPredicate);
+ new IncrementalClusterManager(
+ table, partitionPredicate,
table.snapshotManager().latestSnapshot());
Map<BinaryRow, CompactUnit> compactUnits =
incrementalClusterManager.createCompactUnits(fullCompaction);
LOG.info(
