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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
