Skip to content

[spark] compact_chain_table should always use dynamic partition overwrite #8529

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Jul 10, 2026
+109 −70
Merged

[spark] compact_chain_table should always use dynamic partition overwrite #8529

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions ...rk-common/src/main/java/org/apache/paimon/spark/procedure/CompactChainTableProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;

Expand Down Expand Up @@ -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<String, String> targetPartition =
ParameterUtils.parseCommaSeparatedKeyValues(partitionStr);
for (Map.Entry<String, String> entry : targetPartition.entrySet()) {
Expand All @@ -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();
}

Expand Down
165 changes: 99 additions & 66 deletions ...-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactChainTableProcedureTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down
Loading