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 fe750e4ac7 [spark] Honor delete.force-produce-changelog on a
full-table DELETE (#10311)
fe750e4ac7 is described below
commit fe750e4ac7f1f7aff975ad9ce1c3be4372054b6d
Author: cxzl25 <[email protected]>
AuthorDate: Fri Oct 2 09:41:29 2026 +0800
[spark] Honor delete.force-produce-changelog on a full-table DELETE (#10311)
---
...OptimizeMetadataOnlyDeleteFromPaimonTable.scala | 9 +++----
.../paimon/spark/sql/DeleteFromTableTestBase.scala | 28 ++++++++++++++++++++++
2 files changed, 33 insertions(+), 4 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
index aa56627a73..a367fabbfe 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
@@ -59,7 +59,7 @@ case class OptimizeMetadataOnlyDeleteFromPaimonTable(spark:
SparkSession)
override def apply(plan: LogicalPlan): LogicalPlan = {
plan.transform {
case d @ DeleteFromPaimonTableCommand(r: DataSourceV2Relation, table,
condition) =>
- if (isTruncateTable(condition)) {
+ if (isTruncateTable(table, condition)) {
TruncatePaimonTableWithFilter(table, None)
} else if (isTruncatePartition(table, condition)) {
tryConvertToPartitionPredicate(r, table, condition) match {
@@ -73,11 +73,12 @@ case class OptimizeMetadataOnlyDeleteFromPaimonTable(spark:
SparkSession)
}
def isMetadataOnlyDelete(table: FileStoreTable, condition: Expression):
Boolean = {
- isTruncateTable(condition) || isTruncatePartition(table, condition)
+ isTruncateTable(table, condition) || isTruncatePartition(table, condition)
}
- private def isTruncateTable(condition: Expression): Boolean = {
- condition == null || condition == TrueLiteral
+ private def isTruncateTable(table: FileStoreTable, condition: Expression):
Boolean = {
+ !table.coreOptions().deleteForceProduceChangelog() &&
+ (condition == null || condition == TrueLiteral)
}
private def isTruncatePartition(table: FileStoreTable, condition:
Expression): Boolean = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
index fcbc0390c5..f051f1e61e 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTestBase.scala
@@ -424,6 +424,34 @@ abstract class DeleteFromTableTestBase extends
PaimonSparkTestBase {
.isEqualTo(3)
}
+ test("test delete producer changelog for full table delete") {
+ // 'delete.force-produce-changelog' = 'true' must force a row-level delete
and emit a '-D'
+ // changelog row for every deleted row, even when the DELETE has no WHERE
clause. A full-table
+ // DELETE must not be rewritten to the metadata-only truncate fast path,
which produces no
+ // changelog and silently drops every delete from a downstream CDC
consumer.
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, name STRING, dt STRING, hh STRING)
+ |TBLPROPERTIES ('primary-key' = 'id, dt, hh', 'merge-engine' =
'deduplicate', 'changelog-producer'='input',
'delete.force-produce-changelog'='true')
+ |PARTITIONED BY (dt, hh)
+ |""".stripMargin)
+
+ spark.sql(
+ "INSERT INTO T VALUES " +
+ "(1, 'a', '2023-10-01', '12')," +
+ "(2, 'b', '2023-10-01', '12')," +
+ "(3, 'c', '2023-10-02', '12')," +
+ "(4, 'd', '2023-10-02', '13')," +
+ "(5, 'e', '2023-10-02', '14')," +
+ "(6, 'f', '2023-10-02', '15')")
+
+ // full-table delete: every row must show up as a '-D' changelog entry
+ spark.sql("DELETE FROM T")
+ assertThat(spark.sql("SELECT * FROM
T").collectAsList().size()).isEqualTo(0)
+ assertThat(spark.sql("SELECT * FROM `T$audit_log` WHERE
rowkind='-D'").collectAsList().size())
+ .isEqualTo(6)
+ }
+
test("Paimon Delete: delete null partition with specified default partition
name") {
spark.sql(s"""
|CREATE TABLE T (a INT, dt STRING)
