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 6594dabd38a5..66f7f31615ce 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.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -169,7 +170,14 @@ private boolean execute( 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 targetPartition = ParameterUtils.parseCommaSeparatedKeyValues(partitionStr); for (Map.Entry entry : targetPartition.entrySet()) { @@ -190,12 +198,10 @@ private boolean checkPartitionExists(FileStoreTable snapshotTable, String partit 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 b7ffc9be5f21..aef65f9f30bb 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 = {