You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "zhoulii (via GitHub)" <gi...@apache.org> on 2026/08/31 06:18:45 UTC
[PR] [spark] Support partition pruning for data evolution self-merge [paimon]
zhoulii opened a new pull request, #9489:
URL: https://github.com/apache/paimon/pull/9489
### Purpose
close #9488
### Tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9489:
URL: https://github.com/apache/paimon/pull/9489#discussion_r3893150942
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -150,25 +151,90 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
- if (!isPaimonTable(sourceTable)) {
- false
- } else if (
- !originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
- ) {
- false
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private def passthroughSourceRelation(plan: LogicalPlan): Option[DataSourceV2Relation] = {
+ EliminateSubqueryAliases(plan) match {
+ case relation: DataSourceV2Relation if relation.table.isInstanceOf[SparkTable] =>
+ Some(relation)
+ case Project(projectList, child) if isPassthroughProject(projectList, child) =>
+ passthroughSourceRelation(child)
+ case _ =>
+ None
+ }
+ }
+
+ private def isPassthroughProject(projectList: Seq[Expression], child: LogicalPlan): Boolean = {
+ val childAttributes = child.output ++ child.metadataOutput
+
+ def isChildAttribute(attr: AttributeReference): Boolean =
+ childAttributes.exists(_.exprId == attr.exprId)
+
+ projectList.forall {
+ case attr: AttributeReference => isChildAttribute(attr)
+ case alias: Alias =>
+ alias.child match {
+ case attr: AttributeReference =>
+ resolver(alias.name, attr.name) && isChildAttribute(attr)
+ case _ => false
+ }
+ case _ => false
+ }
+ }
+
+ private lazy val sameSourceAndTargetTable: Boolean =
+ passthroughSourceRelation(sourceTable)
+ .exists(sourceRelation => originalTargetRelation.name.equals(sourceRelation.name))
Review Comment:
[P1] Reject same-name sources with different scan semantics
This still treats a time-travel source as the current target because the relation name remains the table full name even when `SparkCatalog.loadTable(ident, version)` adds a time-travel scan option. I reproduced this by inserting `b = 10`, recording that snapshot, updating the current row to `b = 20`, and then merging from a passthrough `SELECT ... FROM target VERSION AS OF <oldSnapshot>` with `_ROW_ID` equality plus `target.dt = 'p1'`. The shortcut drops the source scan and leaves `b = 20`, whereas forcing the regular MERGE path correctly restores the old-snapshot value `b = 10`. Please require the source scan to be semantically identical to the current target scan; at minimum, any Paimon time-travel option should make this return false and fall back to regular MERGE. The Spark 4.0 copy needs the same guard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #9489:
URL: https://github.com/apache/paimon/pull/9489#issuecomment-5476864054
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "zhoulii (via GitHub)" <gi...@apache.org>.
zhoulii commented on code in PR #9489:
URL: https://github.com/apache/paimon/pull/9489#discussion_r3893415496
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -150,25 +151,90 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
- if (!isPaimonTable(sourceTable)) {
- false
- } else if (
- !originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
- ) {
- false
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private def passthroughSourceRelation(plan: LogicalPlan): Option[DataSourceV2Relation] = {
+ EliminateSubqueryAliases(plan) match {
+ case relation: DataSourceV2Relation if relation.table.isInstanceOf[SparkTable] =>
+ Some(relation)
+ case Project(projectList, child) if isPassthroughProject(projectList, child) =>
+ passthroughSourceRelation(child)
+ case _ =>
+ None
+ }
+ }
+
+ private def isPassthroughProject(projectList: Seq[Expression], child: LogicalPlan): Boolean = {
+ val childAttributes = child.output ++ child.metadataOutput
+
+ def isChildAttribute(attr: AttributeReference): Boolean =
+ childAttributes.exists(_.exprId == attr.exprId)
+
+ projectList.forall {
+ case attr: AttributeReference => isChildAttribute(attr)
+ case alias: Alias =>
+ alias.child match {
+ case attr: AttributeReference =>
+ resolver(alias.name, attr.name) && isChildAttribute(attr)
+ case _ => false
+ }
+ case _ => false
+ }
+ }
+
+ private lazy val sameSourceAndTargetTable: Boolean =
+ passthroughSourceRelation(sourceTable)
+ .exists(sourceRelation => originalTargetRelation.name.equals(sourceRelation.name))
Review Comment:
Thanks for catching this. Similar to the issue discussed in the comment above, this was also a pre-existing problem in the original `_ROW_ID` shortcut, rather than being introduced by this PR.
Following your suggestion, I made the minimal fix to disable the shortcut for Paimon time-travel sources, so they fall back to regular MERGE and preserve snapshot semantics. The same fix has been applied to Spark 4, with a regression test added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "zhoulii (via GitHub)" <gi...@apache.org>.
zhoulii commented on code in PR #9489:
URL: https://github.com/apache/paimon/pull/9489#discussion_r3892786210
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -150,25 +151,65 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private lazy val sameSourceAndTargetTable: Boolean = {
Review Comment:
Thanks for pointing this out. This issue actually existed in the original `_ROW_ID`-only self-merge shortcut as well: it could unwrap a computed `Project`, remove the source plan, and rewrite source attributes to target attributes by name, thereby losing expressions such as `b + 1 AS b`.
I fixed it by enabling the shortcut only for a direct Paimon relation or a verified passthrough projection. Computed projections and other transformed source plans now fall back to the regular MERGE path. A regression test has also been added for this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #9489:
URL: https://github.com/apache/paimon/pull/9489
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Support partition pruning for data evolution self-merge [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9489:
URL: https://github.com/apache/paimon/pull/9489#discussion_r3892339010
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala:
##########
@@ -150,25 +151,65 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private lazy val sameSourceAndTargetTable: Boolean = {
Review Comment:
[P1] Restrict the shortcut to semantically identical source scans
This name-only check can silently ignore source projections because `PaimonRelation.unapply` also accepts `Project(_, PaimonV2Relation(...))`. For example, `USING (SELECT _ROW_ID, b + 1 AS b FROM target) source` with `ON target._ROW_ID = source._ROW_ID AND target.dt = 'p1'` enters the new shortcut; the shortcut removes the source plan and maps `source.b` to `target.b` by name, so a value of 10 remains 10 instead of becoming 11. I reproduced this on the current head, while changing the residual condition to non-partition `target.id = 1` forces the regular merge path and correctly produces 11. Please only enable this shortcut for a direct scan of the same table, or for a verified passthrough projection; otherwise fall back to the regular merge. The Spark 4.0 copy needs the same guard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org