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 1da2ac6155 [core] Share rewrite options for compaction and clustering
(#10685)
1da2ac6155 is described below
commit 1da2ac615551c20dc452c2a5cbc6e5688fd8d162
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 13:29:54 2026 +0800
[core] Share rewrite options for compaction and clustering (#10685)
---
.../DataEvolutionBlobCompactTask.java | 3 +-
.../dataevolution/DataEvolutionCompactTask.java | 5 +--
.../org/apache/paimon/compact/CompactOptions.java | 44 ++++++++++++++++++++++
.../apache/paimon/flink/action/CompactAction.java | 2 +
.../flink/compact/IncrementalClusterCompact.java | 2 -
.../paimon/spark/procedure/CompactProcedure.java | 32 +++-------------
6 files changed, 56 insertions(+), 32 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java
index 54af1361aa..33154ff3d9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java
@@ -49,6 +49,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import static org.apache.paimon.compact.CompactOptions.REWRITE_OPTIONS;
import static org.apache.paimon.types.BlobType.fieldNamesInBlobFile;
import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -78,7 +79,7 @@ public class DataEvolutionBlobCompactTask extends
DataEvolutionCompactTask {
RowType blobWriteType = new
RowType(Collections.singletonList(blobField));
- FileStoreTable readTable = table.copy(BLOB_COMPACT_READ_OPTIONS);
+ FileStoreTable readTable = table.copy(REWRITE_OPTIONS);
AppendOnlyFileStore store = (AppendOnlyFileStore) readTable.store();
DataFilePathFactory pathFactory =
store.pathFactory().createDataFilePathFactory(partition, 0);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
index 6f24a41bcc..54e63932e4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
@@ -36,6 +36,7 @@ import java.util.Map;
import java.util.Objects;
import static java.util.Comparator.comparingLong;
+import static org.apache.paimon.compact.CompactOptions.REWRITE_OPTIONS;
import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -43,8 +44,6 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public abstract class DataEvolutionCompactTask {
protected static final Map<String, String> DYNAMIC_WRITE_OPTIONS =
dynamicWriteOptions();
- protected static final Map<String, String> BLOB_COMPACT_READ_OPTIONS =
- Collections.singletonMap(CoreOptions.BLOB_AS_DESCRIPTOR.key(),
"true");
protected final BinaryRow partition;
protected final List<DataFileMeta> compactBefore;
@@ -58,7 +57,7 @@ public abstract class DataEvolutionCompactTask {
}
private static Map<String, String> dynamicWriteOptions() {
- Map<String, String> options = new HashMap<>();
+ Map<String, String> options = new HashMap<>(REWRITE_OPTIONS);
options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "99999 G");
// Compaction is size-based; the write-time row limit must not roll
its output.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compact/CompactOptions.java
b/paimon-core/src/main/java/org/apache/paimon/compact/CompactOptions.java
new file mode 100644
index 0000000000..8ece8e5759
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactOptions.java
@@ -0,0 +1,44 @@
+/*
+ * 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.compact;
+
+import org.apache.paimon.CoreOptions;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Runtime options for physical compaction and clustering rewrites. */
+public final class CompactOptions {
+
+ public static final Map<String, String> REWRITE_OPTIONS = rewriteOptions();
+
+ private CompactOptions() {}
+
+ private static Map<String, String> rewriteOptions() {
+ Map<String, String> options = new HashMap<>();
+ // Preserve stored references, including views of upstream values that
are currently null.
+ options.put(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true");
+ options.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(), "false");
+ // Keep the layout produced by the rewrite instead of applying
ingestion optimizations.
+ options.put(CoreOptions.PARTITION_SINK_STRATEGY.key(), "none");
+ options.put(CoreOptions.CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE.key(),
"false");
+ return Collections.unmodifiableMap(options);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 94480deb1b..bef70b2627 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -75,6 +75,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import static org.apache.paimon.compact.CompactOptions.REWRITE_OPTIONS;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Table compact action for Flink. */
@@ -106,6 +107,7 @@ public class CompactAction extends TableActionBase {
}
HashMap<String, String> dynamicOptions = new HashMap<>(tableConf);
dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "false");
+ dynamicOptions.putAll(REWRITE_OPTIONS);
table = table.copy(dynamicOptions);
}
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 6fa55b1c4a..3a986dca07 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
@@ -107,8 +107,6 @@ public class IncrementalClusterCompact {
this.env = env;
if (table.coreOptions().dataEvolutionEnabled()) {
Map<String, String> rewriteOptions = new HashMap<>();
- // Rewrite stored references, including views whose upstream value
is currently null.
- rewriteOptions.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(),
"false");
// Sorted outputs must reach the clustering committer without
another rewrite or commit.
rewriteOptions.put(PRECOMMIT_COMPACT.key(), "false");
rewriteOptions.put(SINK_COORDINATOR_COMMIT_ENABLED.key(), "false");
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 5cdac96c4b..c69c789361 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
@@ -38,7 +38,6 @@ import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.operation.BaseAppendFileStoreWrite;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionValuesTimeExpireStrategy;
-import org.apache.paimon.spark.SparkTable;
import org.apache.paimon.spark.SparkUtils;
import
org.apache.paimon.spark.commands.DataEvolutionCompactMergeConflictRewriter;
import org.apache.paimon.spark.commands.PaimonSparkWriter;
@@ -77,7 +76,6 @@ import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
-import org.apache.spark.sql.paimon.shims.SparkShimLoader;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
@@ -109,6 +107,7 @@ import java.util.stream.Collectors;
import scala.collection.JavaConverters;
import static org.apache.paimon.CoreOptions.createCommitUser;
+import static org.apache.paimon.compact.CompactOptions.REWRITE_OPTIONS;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static
org.apache.paimon.spark.utils.SparkProcedureUtils.readParallelism;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
@@ -237,7 +236,9 @@ public class CompactProcedure extends BaseProcedure {
ProcedureUtils.putIfNotEmpty(
dynamicOptions, CoreOptions.WRITE_ONLY.key(),
"false");
ProcedureUtils.putAllOptions(dynamicOptions, options);
+ dynamicOptions.putAll(REWRITE_OPTIONS);
table = table.copy(dynamicOptions);
+ relation = createRelation(tableIdent,
sparkTable.copy(table));
if (coreOptions.clusteringIncrementalEnabled()
&& (!OrderType.NONE.name().equals(sortType))) {
throw new IllegalArgumentException(
@@ -881,27 +882,6 @@ public class CompactProcedure extends BaseProcedure {
Map<BinaryRow, Pair<List<DataSplit>, CommitMessage>> partitionSplits =
incrementalClusterManager.toSplitsAndRewriteDvFiles(compactUnits);
- final FileStoreTable clusterTable;
- final DataSourceV2Relation clusterRelation;
- if (table.coreOptions().dataEvolutionEnabled()) {
- Map<String, String> clusterOptions = new HashMap<>();
- // Preserve inline references and the layout already produced by
the sorter.
- clusterOptions.put(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true");
- clusterOptions.put(CoreOptions.BLOB_VIEW_RESOLVE_ENABLED.key(),
"false");
- clusterOptions.put(CoreOptions.PARTITION_SINK_STRATEGY.key(),
"none");
-
clusterOptions.put(CoreOptions.CLUSTERING_INCREMENTAL_OPTIMIZE_WRITE.key(),
"false");
- clusterTable = table.copy(clusterOptions);
- clusterRelation =
- SparkShimLoader.shim()
- .copyDataSourceV2Relation(
- relation,
- ((SparkTable)
relation.table()).copy(clusterTable),
- relation.output());
- } else {
- clusterTable = table;
- clusterRelation = relation;
- }
-
// sort in partition
TableSorter sorter =
TableSorter.getSorter(
@@ -929,7 +909,7 @@ public class CompactProcedure extends BaseProcedure {
spark(),
ScanPlanHelper$.MODULE$.createNewScanPlan(
splits.toArray(new
DataSplit[0]),
- clusterRelation));
+ relation));
// Use sortLocal() for LOCAL_SORT, sort()
for GLOBAL_SORT
if (mode ==
CoreOptions.ClusteringIncrementalMode.LOCAL_SORT) {
return sorter.sortLocal(dataset);
@@ -941,7 +921,7 @@ public class CompactProcedure extends BaseProcedure {
.orElse(null);
// set to write only to prevent invoking compaction
// do not use overwrite, we don't need to overwrite the whole partition
- PaimonSparkWriter writer =
PaimonSparkWriter.apply(clusterTable).writeOnly();
+ PaimonSparkWriter writer = PaimonSparkWriter.apply(table).writeOnly();
List<CommitMessage> commitMessages =
datasetForWrite == null
? Collections.emptyList()
@@ -1009,7 +989,7 @@ public class CompactProcedure extends BaseProcedure {
try (TableCommitImpl commit =
table.newCommit(commitUser).materializeDvRowIdCheck()) {
clusterMessages.addAll(
DataEvolutionRewriteExecutor.executeTasks(
- clusterTable,
+ table,
compactionTasks,
new JavaSparkContext(spark().sparkContext()),
spark(),
