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 c96f22b792 [spark] Support compact_chain_table procedure (#7313)
c96f22b792 is described below
commit c96f22b79210644f4f891996802f6dc0e8ead75d
Author: Juntao Zhang <[email protected]>
AuthorDate: Tue Jun 23 19:08:06 2026 +0800
[spark] Support compact_chain_table procedure (#7313)
---
docs/docs/primary-key-table/chain-table.md | 63 +++-
docs/docs/spark/procedures.md | 12 +
.../apache/paimon/table/ChainGroupReadTable.java | 55 +--
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../procedure/CompactChainTableProcedure.java | 220 ++++++++++++
.../paimon/spark/utils/SparkProcedureUtils.java | 31 ++
.../procedure/CompactChainTableProcedureTest.scala | 394 +++++++++++++++++++++
7 files changed, 738 insertions(+), 39 deletions(-)
diff --git a/docs/docs/primary-key-table/chain-table.md
b/docs/docs/primary-key-table/chain-table.md
index a861fb12e6..0cb2984b4c 100644
--- a/docs/docs/primary-key-table/chain-table.md
+++ b/docs/docs/primary-key-table/chain-table.md
@@ -130,7 +130,6 @@ Notice that:
- Chain table is only supported for primary key table, which means you should
define `bucket` and `bucket-key` for the table.
- Chain table should ensure that the schema of each branch is consistent.
- Both Spark and Flink batch read/write are supported. Flink streaming
read/write is not supported.
-- Chain compact is not supported for now, and it will be supported later.
- Deletion vector is not supported for chain table.
After creating a chain table, you can read and write data in the following
ways.
@@ -153,12 +152,12 @@ select t1, t2, t3 from default.t where date = '20250811'
```
you will get the following result:
```text
-+---+----+-----+
-| t1| t2| t3|
-+---+----+-----+
-| 1 | 1| 1 |
-| 2 | 1| 1 |
-+---+----+-----+
++---+----+-----+
+| t1| t2| t3|
++---+----+-----+
+| 1 | 1| 1 |
+| 2 | 1| 1 |
++---+----+-----+
```
- Incremental Query: Read the incremental partition from t$branch_delta
@@ -167,11 +166,11 @@ select t1, t2, t3 from `default`.`t$branch_delta` where
date = '20250811'
```
you will get the following result:
```text
-+---+----+-----+
-| t1| t2| t3|
-+---+----+-----+
-| 2 | 1| 1 |
-+---+----+-----+
++---+----+-----+
+| t1| t2| t3|
++---+----+-----+
+| 2 | 1| 1 |
++---+----+-----+
```
- Hybrid Query: Read both full and incremental data simultaneously.
@@ -182,13 +181,39 @@ select t1, t2, t3 from `default`.`t$branch_delta` where
date = '20250811'
```
you will get the following result:
```text
-+---+----+-----+
-| t1| t2| t3|
-+---+----+-----+
-| 1 | 1| 1 |
-| 2 | 1| 1 |
-| 2 | 1| 1 |
-+---+----+-----+
++---+----+-----+
+| t1| t2| t3|
++---+----+-----+
+| 1 | 1| 1 |
+| 2 | 1| 1 |
+| 2 | 1| 1 |
++---+----+-----+
+```
+
+- Chain Table Compaction: Merge data from snapshot and delta branches into the
snapshot branch.
+ This is useful for periodically compacting incremental data into full
snapshots.
+ You can use the `compact_chain_table` procedure to merge a specific
partition:
+
+```sql
+CALL sys.compact_chain_table(table => 'default.t', partition =>
'date="20250811"');
+```
+
+After compaction, the data in the snapshot branch will contain the merged
result from both snapshot
+and delta branches, and subsequent queries will benefit from direct snapshot
access without
+merge-on-read overhead.
+
+```sql
+select t1, t2, t3 from `default`.`t$branch_snapshot` where date = '20250811';
+```
+
+you will get the following result:
+```text
++---+----+-----+
+| t1| t2| t3|
++---+----+-----+
+| 1 | 1| 1 |
+| 2 | 1| 1 |
++---+----+-----+
```
## Group Partition
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index b264c49fc2..ed55504d07 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -79,6 +79,18 @@ This section introduce all available spark procedures about
paimon.
CALL sys.compact_database(including_databases => 'db1', options =>
'target-file-size=128m')
</td>
</tr>
+ <tr>
+ <td>compact_chain_table</td>
+ <td>
+ To compact chain table by merging snapshot and delta branches into
the snapshot branch. Arguments:
+ <li>table: The target chain table identifier. Cannot be empty.</li>
+ <li>partition: Partition specification format (e.g.,
'dt="20250810",hour="22"'). Cannot be empty.</li>
+ <li>overwrite: Whether to overwrite if the partition already
exists in the snapshot branch. Default is false. Optional.</li>
+ </td>
+ <td>
+ CALL sys.compact_chain_table(table => 'default.T', partition =>
'dt="20250810",hour="22"')<br/><br/>
+ </td>
+ </tr>
<tr>
<td>expire_snapshots</td>
<td>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index 75ba5f4097..95a591ebb4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -128,6 +128,7 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
private final ChainPartitionProjector partitionProjector;
private Predicate dataPredicate;
private Filter<Integer> bucketFilter;
+ protected boolean preloadTargetSnapshot = true;
public ChainTableBatchScan(
TableSchema tableSchema, ChainGroupReadTable
chainGroupReadTable) {
@@ -210,6 +211,11 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
return this;
}
+ public FallbackReadScan skipPreloadTargetSnapshot() {
+ this.preloadTargetSnapshot = false;
+ return this;
+ }
+
/**
* Builds a plan for chain tables.
*
@@ -237,26 +243,7 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
public Plan plan() {
List<Split> splits = new ArrayList<>();
PredicateBuilder builder = new
PredicateBuilder(tableSchema.logicalPartitionType());
- for (Split split : mainScan.plan().splits()) {
- DataSplit dataSplit = (DataSplit) split;
- HashMap<String, String> fileBucketPathMapping = new
HashMap<>();
- HashMap<String, String> fileBranchMapping = new HashMap<>();
- for (DataFileMeta file : dataSplit.dataFiles()) {
- fileBucketPathMapping.put(file.fileName(), ((DataSplit)
split).bucketPath());
- fileBranchMapping.put(file.fileName(),
options.scanFallbackSnapshotBranch());
- }
- splits.add(
- new ChainSplit(
- dataSplit.partition(),
- dataSplit.dataFiles(),
- fileBranchMapping,
- fileBucketPathMapping));
- }
-
- Set<BinaryRow> snapshotPartitions =
- new HashSet<>(
- newChainPartitionListingScan(true,
getMainPartitionPredicate())
- .listPartitions());
+ Set<BinaryRow> snapshotPartitions =
preloadTargetSnapshotSplits(splits);
DataTableScan deltaPartitionScan =
newChainPartitionListingScan(false,
getFallbackPartitionPredicate());
@@ -453,6 +440,34 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
return scan;
}
+ private Set<BinaryRow> preloadTargetSnapshotSplits(List<Split> splits)
{
+ Set<BinaryRow> snapshotPartitions = new HashSet<>();
+ if (!preloadTargetSnapshot) {
+ return snapshotPartitions;
+ }
+
+ for (Split split : mainScan.plan().splits()) {
+ DataSplit dataSplit = (DataSplit) split;
+ HashMap<String, String> fileBucketPathMapping = new
HashMap<>();
+ HashMap<String, String> fileBranchMapping = new HashMap<>();
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ fileBucketPathMapping.put(file.fileName(), ((DataSplit)
split).bucketPath());
+ fileBranchMapping.put(file.fileName(),
options.scanFallbackSnapshotBranch());
+ }
+ splits.add(
+ new ChainSplit(
+ dataSplit.partition(),
+ dataSplit.dataFiles(),
+ fileBranchMapping,
+ fileBucketPathMapping));
+ }
+
+ snapshotPartitions.addAll(
+ newChainPartitionListingScan(true,
getMainPartitionPredicate())
+ .listPartitions());
+ return snapshotPartitions;
+ }
+
private DataTableScan newFilteredScan(boolean snapshot) {
DataTableScan scan =
snapshot
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index ad0f2a93b7..6a2203b9cb 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -21,6 +21,7 @@ package org.apache.paimon.spark;
import org.apache.paimon.spark.procedure.AlterFunctionProcedure;
import org.apache.paimon.spark.procedure.AlterViewDialectProcedure;
import org.apache.paimon.spark.procedure.ClearConsumersProcedure;
+import org.apache.paimon.spark.procedure.CompactChainTableProcedure;
import org.apache.paimon.spark.procedure.CompactDatabaseProcedure;
import org.apache.paimon.spark.procedure.CompactManifestProcedure;
import org.apache.paimon.spark.procedure.CompactProcedure;
@@ -103,6 +104,7 @@ public class SparkProcedures {
procedureBuilders.put("rename_branch", RenameBranchProcedure::builder);
procedureBuilders.put("compact", CompactProcedure::builder);
procedureBuilders.put("compact_database",
CompactDatabaseProcedure::builder);
+ procedureBuilders.put("compact_chain_table",
CompactChainTableProcedure::builder);
procedureBuilders.put("rescale", RescaleProcedure::builder);
procedureBuilders.put("migrate_database",
MigrateDatabaseProcedure::builder);
procedureBuilders.put("migrate_table", MigrateTableProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
new file mode 100644
index 0000000000..ad992cf505
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
@@ -0,0 +1,220 @@
+/*
+ * 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.spark.procedure;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.spark.commands.PaimonSparkWriter;
+import org.apache.paimon.spark.util.ScanPlanHelper$;
+import org.apache.paimon.spark.utils.SparkProcedureUtils;
+import org.apache.paimon.table.ChainGroupReadTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.ParameterUtils;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.PaimonUtils;
+import org.apache.spark.sql.Row;
+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.functions;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Procedure to compact chain table. Usage:
+ *
+ * <pre><code>
+ * -- Compact chain table, overwrite default is false
+ * CALL sys.compact_chain_table(table => 'db.table', partition =>
'dt="20250810",hour="22"', [overwrite => true])
+ * </code></pre>
+ */
+public class CompactChainTableProcedure extends BaseProcedure {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CompactChainTableProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("partition", StringType),
+ ProcedureParameter.optional("overwrite", BooleanType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", BooleanType, false,
Metadata.empty())
+ });
+
+ protected CompactChainTableProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier tableIdent = toIdentifier(args.getString(0),
PARAMETERS[0].name());
+ String partitionStr = args.getString(1);
+ boolean overwrite = !args.isNullAt(2) && args.getBoolean(2);
+ checkArgument(StringUtils.isNotEmpty(partitionStr), "Partition string
cannot be empty");
+ checkArgument(
+ partitionStr.split(";").length == 1,
+ "compact_chain_table only supports a single partition, but
multiple partitions were provided: %s",
+ partitionStr);
+
+ return modifyPaimonTable(
+ tableIdent,
+ t -> {
+ checkArgument(
+ new CoreOptions(t.options()).isChainTable(),
+ "compact_chain_table only supports chain table");
+ checkArgument(
+ t instanceof FallbackReadFileStoreTable,
+ "Table %s is not a chain table",
+ tableIdent);
+ FallbackReadFileStoreTable table =
(FallbackReadFileStoreTable) t;
+ checkArgument(
+ table.other() instanceof ChainGroupReadTable,
+ "Table %s is not a chain table",
+ tableIdent);
+ DataSourceV2Relation relation = createRelation(tableIdent);
+ boolean success =
+ execute(
+ (ChainGroupReadTable) table.other(),
+ relation,
+ partitionStr,
+ overwrite);
+ return new InternalRow[] {newInternalRow(success)};
+ });
+ }
+
+ private boolean execute(
+ ChainGroupReadTable table,
+ DataSourceV2Relation relation,
+ String partitionStr,
+ boolean overwrite) {
+ String partition = SparkProcedureUtils.toWhere(partitionStr);
+ FileStoreTable snapshotTable = table.wrapped();
+
+ ChainGroupReadTable.ChainTableBatchScan scan =
+ (ChainGroupReadTable.ChainTableBatchScan) table.newScan();
+ PartitionPredicate partitionPredicate =
+ SparkProcedureUtils.convertToPartitionPredicate(
+ partition, table.schema().logicalPartitionType(),
spark(), relation);
+
+ // Check if target partition already exists in snapshot branch
+ boolean partitionExists = checkPartitionExists(snapshotTable,
partition, relation);
+ if (partitionExists) {
+ if (overwrite) {
+
scan.skipPreloadTargetSnapshot().withPartitionFilter(partitionPredicate);
+ LOG.info("Found existing partition {}, will overwrite it.",
partition);
+ } else {
+ LOG.info(
+ "Partition {} already exists in snapshot branch,
skipping compaction.",
+ partitionStr);
+ return false;
+ }
+ } else {
+ scan.withPartitionFilter(partitionPredicate);
+ }
+
+ List<Split> splits = scan.plan().splits();
+ if (splits.isEmpty()) {
+ LOG.warn(
+ "Table {} partition {} has no data to compact, skipping.",
table, partitionStr);
+ return false;
+ }
+
+ Dataset<Row> datasetForWrite =
+ PaimonUtils.createDataset(
+ spark(),
+ ScanPlanHelper$.MODULE$.createNewScanPlan(
+ splits.toArray(new Split[0]), relation));
+
+ PaimonSparkWriter writer = PaimonSparkWriter.apply(snapshotTable);
+ Map<String, String> targetPartition =
+ ParameterUtils.parseCommaSeparatedKeyValues(partitionStr);
+ for (Map.Entry<String, String> entry : targetPartition.entrySet()) {
+ datasetForWrite =
+ datasetForWrite.withColumn(entry.getKey(),
functions.expr(entry.getValue()));
+ }
+ if (partitionExists) {
+ Map<String, String> staticPartition =
+ SparkProcedureUtils.parseStaticPartition(spark(),
targetPartition);
+ writer.writeBuilder().withOverwrite(staticPartition);
+ }
+ writer.commit(writer.write(datasetForWrite));
+ LOG.info("Successfully compacted partition {} to snapshot branch.",
partitionStr);
+ return true;
+ }
+
+ private boolean checkPartitionExists(
+ FileStoreTable snapshotTable, String partition,
DataSourceV2Relation relation) {
+ PartitionPredicate snapshotPartitionPredicate =
+ SparkProcedureUtils.convertToPartitionPredicate(
+ partition,
+ snapshotTable.schema().logicalPartitionType(),
+ spark(),
+ relation);
+
+ return !snapshotTable
+ .newScan()
+ .withPartitionFilter(snapshotPartitionPredicate)
+ .plan()
+ .splits()
+ .isEmpty();
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<CompactChainTableProcedure>() {
+ @Override
+ public CompactChainTableProcedure doBuild() {
+ return new CompactChainTableProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "Compact chain table by merging snapshot + delta into target
snapshot.";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
index e4bc86b7a5..59f93f92fb 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/SparkProcedureUtils.java
@@ -27,6 +27,7 @@ import org.apache.paimon.utils.StringUtils;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.expressions.Literal;
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
import org.slf4j.Logger;
@@ -34,6 +35,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -107,4 +109,33 @@ public class SparkProcedureUtils {
.reduce((a, b) -> a + " OR " + b)
.orElse(null);
}
+
+ /**
+ * Parse partition spec values by evaluating them as Spark SQL literal
expressions. This strips
+ * quotes from string literals and validates that values are non-null
literals.
+ *
+ * @param spark the Spark session
+ * @param partitionSpec the partition spec with raw values (e.g., {"date":
"\"20260225\""})
+ * @return the static partition map with unquoted literal values (e.g.,
{"date": "20260225"})
+ */
+ public static Map<String, String> parseStaticPartition(
+ SparkSession spark, Map<String, String> partitionSpec) {
+ Map<String, String> staticPartition = new HashMap<>();
+ for (Map.Entry<String, String> entry : partitionSpec.entrySet()) {
+ Expression expr;
+ try {
+ expr =
spark.sessionState().sqlParser().parseExpression(entry.getValue());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ checkArgument(
+ expr instanceof Literal,
+ "Partition value must be a literal expression, but got:
%s",
+ entry.getValue());
+ Object value = ((Literal) expr).value();
+ checkArgument(value != null, "Partition value cannot be null");
+ staticPartition.put(entry.getKey(), value.toString());
+ }
+ return staticPartition;
+ }
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactChainTableProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactChainTableProcedureTest.scala
new file mode 100644
index 0000000000..b7ffc9be5f
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactChainTableProcedureTest.scala
@@ -0,0 +1,394 @@
+/*
+ * 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.spark.procedure
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.streaming.StreamTest
+
+/** Test compact_chain_table procedure. See [[CompactChainTableProcedure]]. */
+class CompactChainTableProcedureTest extends PaimonSparkTestBase {
+
+ test("Paimon Procedure: compact_chain_table - basic test") {
+ withTable("chain_compact_t1") {
+ // Create chain table
+ spark.sql("""
+ |CREATE TABLE IF NOT EXISTS chain_compact_t1 (
+ | `t1` BIGINT COMMENT 't1',
+ | `t2` BIGINT COMMENT 't2',
+ | `t3` STRING COMMENT 't3'
+ | ) PARTITIONED BY (`date` STRING COMMENT 'date')
+ |TBLPROPERTIES (
+ | 'chain-table.enabled' = 'true',
+ | 'primary-key' = 'date,t1',
+ | 'sequence.field' = 't2',
+ | 'bucket-key' = 't1',
+ | 'bucket' = '1',
+ | 'partition.timestamp-pattern' = '$date',
+ | 'partition.timestamp-formatter' = 'yyyyMMdd'
+ |)
+ |""".stripMargin)
+
+ // Create branches
+ setupChainTableBranches("chain_compact_t1")
+
+ // Compact empty partition
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t1`',
partition => 'date=\"20260224\"')"),
+ Row(false) :: Nil)
+
+ // Insert snapshot data
+ spark.sql(
+ "insert into `chain_compact_t1$branch_snapshot` partition (date =
'20260222') values (0, 1, '0')")
+ spark.sql(
+ "insert into `chain_compact_t1$branch_snapshot` partition (date =
'20260223') values (1, 1, '1')")
+
+ // Insert delta data
+ spark.sql(
+ "insert into `chain_compact_t1$branch_delta` partition (date =
'20260224') values (2, 2, '2')")
+
+ // Before compaction: verify chain read shows snapshot + delta
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t1` where date = '20260224'"),
+ Seq(Row(1, 1, "1", "20260224"), Row(2, 2, "2", "20260224"))
+ )
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t1`',
partition => 'date=\"20260224\"')"),
+ Row(true) :: Nil)
+
+ // After compaction: verify snapshot branch has merged data
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t1` where date = '20260224'"),
+ Seq(Row(1, 1, "1", "20260224"), Row(2, 2, "2", "20260224"))
+ )
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t1$branch_snapshot` where date =
'20260223'"),
+ Seq(Row(1, 1, "1", "20260223"))
+ )
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t1$branch_snapshot` where date =
'20260224'"),
+ Seq(Row(1, 1, "1", "20260224"), Row(2, 2, "2", "20260224"))
+ )
+ }
+ }
+
+ test("Paimon Procedure: compact_chain_table - overwrite test") {
+ withTable("chain_compact_t2") {
+ // Create chain table
+ spark.sql("""
+ |CREATE TABLE IF NOT EXISTS chain_compact_t2 (
+ | `t1` BIGINT COMMENT 't1',
+ | `t2` BIGINT COMMENT 't2',
+ | `t3` STRING COMMENT 't3'
+ | ) PARTITIONED BY (`date` STRING COMMENT 'date')
+ |TBLPROPERTIES (
+ | 'dynamic-partition-overwrite' = 'false',
+ | 'chain-table.enabled' = 'true',
+ | 'primary-key' = 'date,t1',
+ | 'sequence.field' = 't2',
+ | 'bucket-key' = 't1',
+ | 'bucket' = '1',
+ | 'partition.timestamp-pattern' = '$date',
+ | 'partition.timestamp-formatter' = 'yyyyMMdd'
+ | )
+ |""".stripMargin)
+
+ // Create branches
+ setupChainTableBranches("chain_compact_t2")
+
+ spark.sql("insert into `chain_compact_t2` partition (date = '20260222')
values (1, 1, '1')")
+
+ // Insert snapshot data
+ spark.sql(
+ "insert into `chain_compact_t2$branch_snapshot` partition (date =
'20260223') values (2, 1, '2')")
+ spark.sql(
+ "insert into `chain_compact_t2$branch_snapshot` partition (date =
'20260224') values (3, 1, '3')")
+ spark.sql(
+ "insert into `chain_compact_t2$branch_snapshot` partition (date =
'20260225') values (3, 2, '3-1')")
+
+ // Insert delta data
+ spark.sql(
+ "insert into `chain_compact_t2$branch_delta` partition (date =
'20260225') values (4, 2, '4')")
+
+ // First call should fail because partition exists
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t2`',
partition => 'date=\"20260225\"')"),
+ Row(false) :: Nil)
+
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t2` where date = '20260225'"),
+ Seq(Row(3, 2, "3-1", "20260225"))
+ )
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t2`',
partition => 'date=\"20260225\"', overwrite => true)"),
+ Row(true) :: Nil)
+
+ // Verify snapshot branch now has the data
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t2$branch_snapshot` where date =
'20260225'"),
+ Seq(Row(3, 1, "3", "20260225"), Row(4, 2, "4", "20260225"))
+ )
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t2$branch_snapshot` where date !=
'20260225'"),
+ Seq(Row(2, 1, "2", "20260223"), Row(3, 1, "3", "20260224"))
+ )
+
+ // Check snapshots commit_kind
+ checkAnswer(
+ sql("select snapshot_id, commit_kind from
`chain_compact_t2$branch_snapshot$snapshots`"),
+ Seq(Row(1, "APPEND"), Row(2, "APPEND"), Row(3, "APPEND"), Row(4,
"OVERWRITE"))
+ )
+ }
+ }
+
+ test("Paimon Procedure: compact_chain_table - test multiple partitions") {
+ withTable("chain_compact_t3") {
+ spark.sql("""
+ |CREATE TABLE IF NOT EXISTS chain_compact_t3 (
+ | `t1` BIGINT COMMENT 't1',
+ | `t2` BIGINT COMMENT 't2',
+ | `t3` STRING COMMENT 't3'
+ | ) PARTITIONED BY (`dt` STRING COMMENT 'dt', `hour` STRING
COMMENT 'hour')
+ |TBLPROPERTIES (
+ | 'chain-table.enabled' = 'true',
+ | 'primary-key' = 'dt,hour,t1',
+ | 'sequence.field' = 't2',
+ | 'bucket-key' = 't1',
+ | 'bucket' = '2',
+ | 'partition.timestamp-pattern' = '$dt $hour:00:00',
+ | 'partition.timestamp-formatter' = 'yyyyMMdd HH:mm:ss'
+ | )
+ |""".stripMargin)
+
+ // Create branches
+ setupChainTableBranches("chain_compact_t3")
+
+ // Write snapshot branch data
+ spark.sql(
+ "insert into `chain_compact_t3$branch_snapshot` partition (dt =
'20250810', hour = '20') values (0, 1, '0')")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_snapshot` partition (dt =
'20250810', hour = '22') values (1, 1, '1'),(2, 1, '1')")
+
+ // Write delta branch data
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250810', hour = '21') values (1, 1, '1'),(2, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250810', hour = '22') values (1, 2, '1-1' ),(3, 1, '1' )")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250810', hour = '23') values (2, 2, '1-1' ),(4, 1, '1' )")
+
+ // Compact partition 20250810/hour=23
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t3`',
partition => 'dt=\"20250810\",hour=\"23\"')"),
+ Row(true) :: Nil
+ )
+
+ // Verify chain read still works correctly
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t3` where dt = '20250810' and hour =
'23'"),
+ Seq(
+ Row(1, 1, "1", "20250810", "23"),
+ Row(2, 2, "1-1", "20250810", "23"),
+ Row(4, 1, "1", "20250810", "23")
+ )
+ )
+
+ spark.sql(
+ "insert into `chain_compact_t3$branch_snapshot` partition (dt =
'20250811', hour = '00') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4,
2, '1-1')")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_snapshot` partition (dt =
'20250811', hour = '02') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4,
2, '1-1'), (5, 1, '1' ), (6, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250811', hour = '00') values (3, 2, '1-1' ),(4, 2, '1-1')")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250811', hour = '01') values (5, 1, '1' ),(6, 1, '1' )")
+ spark.sql(
+ "insert into `chain_compact_t3$branch_delta` partition (dt =
'20250811', hour = '02') values (5, 2, '1-1' ),(6, 2, '1-1' )")
+
+ // Compact partition 20250811/hour=02 without overwrite
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t3`',
partition => 'dt=\"20250811\",hour=\"02\"')"),
+ Row(false) :: Nil
+ )
+
+ // Verify data is unchanged after skip
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t3` where dt = '20250811' and hour =
'02'"),
+ Seq(
+ Row(1, 2, "1-1", "20250811", "02"),
+ Row(2, 2, "1-1", "20250811", "02"),
+ Row(3, 2, "1-1", "20250811", "02"),
+ Row(4, 2, "1-1", "20250811", "02"),
+ Row(5, 1, "1", "20250811", "02"),
+ Row(6, 1, "1", "20250811", "02")
+ )
+ )
+
+ // Compact with overwrite to test overwrite path
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t3`',
partition => 'dt=\"20250811\",hour=\"02\"', overwrite => true)"),
+ Row(true) :: Nil
+ )
+ checkAnswer(
+ sql("select snapshot_id,commit_kind from
`chain_compact_t3$branch_snapshot$snapshots`"),
+ Seq(
+ Row(1, "APPEND"),
+ Row(2, "APPEND"),
+ Row(3, "APPEND"),
+ Row(4, "APPEND"),
+ Row(5, "APPEND"),
+ Row(6, "OVERWRITE")
+ )
+ )
+
+ // Check all snapshot partition data
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t3` where dt = '20250811' and hour =
'02'"),
+ Seq(
+ Row(1, 2, "1-1", "20250811", "02"),
+ Row(2, 2, "1-1", "20250811", "02"),
+ Row(3, 2, "1-1", "20250811", "02"),
+ Row(4, 2, "1-1", "20250811", "02"),
+ Row(5, 2, "1-1", "20250811", "02"),
+ Row(6, 2, "1-1", "20250811", "02")
+ )
+ )
+ checkAnswer(
+ sql("SELECT * FROM `chain_compact_t3` where dt = '20250811' and hour =
'00'"),
+ Seq(
+ Row(1, 2, "1-1", "20250811", "00"),
+ Row(2, 2, "1-1", "20250811", "00"),
+ Row(3, 2, "1-1", "20250811", "00"),
+ Row(4, 2, "1-1", "20250811", "00")
+ )
+ )
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t3$branch_snapshot` where dt =
'20250810' and hour = '23'"),
+ Seq(
+ Row(1, 1, "1", "20250810", "23"),
+ Row(2, 2, "1-1", "20250810", "23"),
+ Row(4, 1, "1", "20250810", "23")
+ )
+ )
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t3$branch_snapshot` where dt =
'20250810' and hour = '22'"),
+ Seq(Row(1, 1, "1", "20250810", "22"), Row(2, 1, "1", "20250810", "22"))
+ )
+ }
+ }
+
+ test("Paimon Procedure: compact_chain_table - test multiple partitions with
group") {
+ withTable("chain_compact_t4") {
+ spark.sql(
+ """
+ |CREATE TABLE IF NOT EXISTS chain_compact_t4 (
+ | `t1` BIGINT COMMENT 't1',
+ | `t2` BIGINT COMMENT 't2',
+ | `t3` STRING COMMENT 't3'
+ | ) PARTITIONED BY (`region` STRING COMMENT 'region', `dt` STRING
COMMENT 'dt', `hour` STRING COMMENT 'hour')
+ |TBLPROPERTIES (
+ | 'chain-table.enabled' = 'true',
+ | 'primary-key' = 'region,dt,hour,t1',
+ | 'sequence.field' = 't2',
+ | 'bucket-key' = 't1',
+ | 'bucket' = '1',
+ | 'partition.timestamp-pattern' = '$dt $hour:00:00',
+ | 'partition.timestamp-formatter' = 'yyyyMMdd HH:mm:ss',
+ | 'merge-engine' = 'deduplicate',
+ | 'chain-table.chain-partition-keys' = 'dt,hour'
+ | )
+ |""".stripMargin)
+
+ // Create branches
+ setupChainTableBranches("chain_compact_t4")
+
+ // Write snapshot branch data
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='CN', dt = '20250810', hour = '20') values (1, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='CN', dt = '20250810', hour = '21') values (2, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='CN', dt = '20250810', hour = '22') values (3, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='UK', dt = '20250810', hour = '21') values (21, 1, '1')")
+
+ // Write delta branch data
+ spark.sql(
+ "insert into `chain_compact_t4$branch_delta` partition (region='CN',
dt = '20250810', hour = '22') values (4, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_delta` partition (region='US',
dt = '20250810', hour = '22') values (11, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_delta` partition (region='UK',
dt = '20250810', hour = '22') values (22, 1, '1' )")
+
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t4$branch_snapshot` where dt =
'20250810' and hour = '22'"),
+ Seq(
+ Row(3, 1, "1", "CN", "20250810", "22")
+ )
+ )
+ checkAnswer(
+ spark.sql(
+ "CALL sys.compact_chain_table(table => '`chain_compact_t4`',
partition => 'dt=\"20250810\", hour=\"22\"', overwrite => true)"),
+ Row(true) :: Nil
+ )
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t4$branch_snapshot` where dt =
'20250810' and hour = '22' order by region"),
+ Seq(
+ Row(2, 1, "1", "CN", "20250810", "22"),
+ Row(4, 1, "1", "CN", "20250810", "22"),
+ Row(21, 1, "1", "UK", "20250810", "22"),
+ Row(22, 1, "1", "UK", "20250810", "22"),
+ Row(11, 1, "1", "US", "20250810", "22")
+ )
+ )
+ }
+ }
+
+ def setupChainTableBranches(tableName: String): Unit = {
+ spark.sql(s"CALL sys.create_branch('$tableName', 'snapshot');")
+ spark.sql(s"CALL sys.create_branch('$tableName', 'delta');")
+
+ // Set branch properties
+ spark.sql(
+ s"ALTER TABLE $tableName SET tblproperties (" +
+ "'scan.fallback-snapshot-branch' = 'snapshot', " +
+ "'scan.fallback-delta-branch' = 'delta')")
+ spark.sql(
+ s"ALTER TABLE `$tableName$$branch_snapshot` SET tblproperties (" +
+ "'scan.fallback-snapshot-branch' = 'snapshot'," +
+ "'scan.fallback-delta-branch' = 'delta')")
+ spark.sql(
+ s"ALTER TABLE `$tableName$$branch_delta` SET tblproperties (" +
+ "'scan.fallback-snapshot-branch' = 'snapshot'," +
+ "'scan.fallback-delta-branch' = 'delta')")
+ }
+}
