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 e462cf84d3 [spark] compact_chain_table should always use dynamic
partition overwrite (#8529)
e462cf84d3 is described below
commit e462cf84d3e9fed4653f3e3e3bb6ee9ef3328076
Author: Juntao Zhang <[email protected]>
AuthorDate: Fri Jul 10 08:36:58 2026 +0800
[spark] compact_chain_table should always use dynamic partition overwrite
(#8529)
---
.../procedure/CompactChainTableProcedure.java | 14 +-
.../procedure/CompactChainTableProcedureTest.scala | 165 ++++++++++++---------
2 files changed, 109 insertions(+), 70 deletions(-)
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
index 6594dabd38..66f7f31615 100644
---
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
@@ -44,6 +44,7 @@ import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -169,7 +170,14 @@ public class CompactChainTableProcedure extends
BaseProcedure {
ScanPlanHelper$.MODULE$.createNewScanPlan(
splits.toArray(new Split[0]), relation));
- PaimonSparkWriter writer = PaimonSparkWriter.apply(snapshotTable);
+ // Always enable dynamic partition overwrite for compact_chain_table
so that overwrite
+ // only affects the full partitions actually present in the compacted
DataFrame,
+ // preserving other groups that have no delta data under the same
partial partition.
+ PaimonSparkWriter writer =
+ PaimonSparkWriter.apply(
+ snapshotTable.copy(
+ Collections.singletonMap(
+
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "true")));
Map<String, String> targetPartition =
ParameterUtils.parseCommaSeparatedKeyValues(partitionStr);
for (Map.Entry<String, String> entry : targetPartition.entrySet()) {
@@ -190,12 +198,10 @@ public class CompactChainTableProcedure extends
BaseProcedure {
PartitionPredicate snapshotPartitionPredicate =
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
partition, snapshotTable, spark());
-
return !snapshotTable
.newScan()
.withPartitionFilter(snapshotPartitionPredicate)
- .plan()
- .splits()
+ .listPartitions()
.isEmpty();
}
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
index b7ffc9be5f..aef65f9f30 100644
---
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
@@ -305,72 +305,105 @@ class CompactChainTableProcedureTest extends
PaimonSparkTestBase {
}
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")
- )
- )
- }
+ Seq(true, false).foreach(
+ overwrite => {
+ withTable("chain_compact_t4") {
+ spark.sql(
+ s"""
+ |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 (
+ | 'dynamic-partition-overwrite' = '$overwrite',
+ | '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')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='FR', dt = '20250810', hour = '22') values (31, 1, '1')")
+ spark.sql(
+ "insert into `chain_compact_t4$branch_snapshot` partition
(region='CA', dt = '20250810', hour = '21') values (41, 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"),
+ Row(31, 1, "1", "FR", "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 snapshot_id,commit_kind from
`chain_compact_t4$branch_snapshot$snapshots`"),
+ Seq(
+ Row(1, "APPEND"),
+ Row(2, "APPEND"),
+ Row(3, "APPEND"),
+ Row(4, "APPEND"),
+ Row(5, "APPEND"),
+ Row(6, "APPEND"),
+ Row(7, "OVERWRITE")
+ )
+ )
+
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t4$branch_snapshot` where dt =
'20250810' and hour = '21'"),
+ Seq(
+ Row(2, 1, "1", "CN", "20250810", "21"),
+ Row(21, 1, "1", "UK", "20250810", "21"),
+ Row(41, 1, "1", "CA", "20250810", "21")
+ )
+ )
+
+ checkAnswer(
+ sql(
+ "SELECT * FROM `chain_compact_t4$branch_snapshot` where dt =
'20250810' and hour = '22'"),
+ Seq(
+ Row(2, 1, "1", "CN", "20250810", "22"),
+ Row(4, 1, "1", "CN", "20250810", "22"),
+ Row(11, 1, "1", "US", "20250810", "22"),
+ Row(21, 1, "1", "UK", "20250810", "22"),
+ Row(22, 1, "1", "UK", "20250810", "22"),
+ Row(31, 1, "1", "FR", "20250810", "22")
+ )
+ )
+ }
+ })
}
def setupChainTableBranches(tableName: String): Unit = {
