tdas commented on a change in pull request #27265: [SPARK-30555][SQL] MERGE
INTO insert action should only access columns from source table
URL: https://github.com/apache/spark/pull/27265#discussion_r369281271
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
##########
@@ -1255,46 +1246,91 @@ class PlanResolutionSuite extends AnalysisTest {
val target = pair._1
val source = pair._2
- val sql =
+ val sql1 =
s"""
|MERGE INTO $target
|USING $source
- |ON $target.i = $source.i
- |WHEN MATCHED AND ($target.s='delete') THEN DELETE
- |WHEN MATCHED AND ($target.s='update') THEN UPDATE SET $target.s =
$source.s
- |WHEN NOT MATCHED AND ($target.s='insert')
- |THEN INSERT ($target.i, $target.s) values ($source.i, $source.s)
+ |ON 1 = 1
+ |WHEN MATCHED THEN DELETE
+ |WHEN MATCHED THEN UPDATE SET s = 1
+ |WHEN NOT MATCHED AND (s = 'a') THEN INSERT (i) values (i)
""".stripMargin
- val parsed = parseAndResolve(sql)
-
- parsed match {
+ parseAndResolve(sql1) match {
case MergeIntoTable(
- _: DataSourceV2Relation,
- _: DataSourceV2Relation,
- EqualTo(l: UnresolvedAttribute, r: UnresolvedAttribute),
- Seq(DeleteAction(Some(EqualTo(dl: UnresolvedAttribute,
StringLiteral("delete")))),
- UpdateAction(Some(EqualTo(ul: UnresolvedAttribute,
StringLiteral("update"))),
- updateAssigns)),
- Seq(InsertAction(Some(EqualTo(il: UnresolvedAttribute,
StringLiteral("insert"))),
- insertAssigns))) =>
- assert(l.name == s"$target.i" && r.name == s"$source.i")
- assert(dl.name == s"$target.s")
- assert(ul.name == s"$target.s")
- assert(il.name == s"$target.s")
+ target: DataSourceV2Relation,
+ source: DataSourceV2Relation,
+ _,
+ Seq(DeleteAction(None), UpdateAction(None, updateAssigns)),
+ Seq(InsertAction(
+ Some(EqualTo(il: AttributeReference, StringLiteral("a"))),
+ insertAssigns))) =>
+ val ti = target.output.find(_.name == "i").get
+ val ts = target.output.find(_.name == "s").get
+ val si = source.output.find(_.name == "i").get
+ val ss = source.output.find(_.name == "s").get
+
+ // INSERT condition is resolved with source table only, so column
`s` is not ambiguous.
+ assert(il.sameRef(ss))
assert(updateAssigns.size == 1)
- assert(updateAssigns.head.key.isInstanceOf[UnresolvedAttribute] &&
- updateAssigns.head.key.asInstanceOf[UnresolvedAttribute].name ==
s"$target.s")
- assert(updateAssigns.head.value.isInstanceOf[UnresolvedAttribute] &&
- updateAssigns.head.value.asInstanceOf[UnresolvedAttribute].name
== s"$source.s")
- assert(insertAssigns.size == 2)
- assert(insertAssigns.head.key.isInstanceOf[UnresolvedAttribute] &&
- insertAssigns.head.key.asInstanceOf[UnresolvedAttribute].name ==
s"$target.i")
- assert(insertAssigns.head.value.isInstanceOf[UnresolvedAttribute] &&
- insertAssigns.head.value.asInstanceOf[UnresolvedAttribute].name
== s"$source.i")
-
- case _ => fail("Expect MergeIntoTable, but got:\n" + parsed.treeString)
+ // UPDATE key is resolved with target table only, so column `s` is
not ambiguous.
+
assert(updateAssigns.head.key.asInstanceOf[AttributeReference].sameRef(ts))
+ assert(insertAssigns.size == 1)
+ // INSERT key is resolved with target table only, so column `i` is
not ambiguous.
+
assert(insertAssigns.head.key.asInstanceOf[AttributeReference].sameRef(ti))
+ // INSERT value is resolved with source table only, so column `i` is
not ambiguous.
+
assert(insertAssigns.head.value.asInstanceOf[AttributeReference].sameRef(si))
+
+ case p => fail("Expect MergeIntoTable, but got:\n" + p.treeString)
}
+
+ val sql2 =
+ s"""
+ |MERGE INTO $target
+ |USING $source
+ |ON i = 1
+ |WHEN MATCHED THEN DELETE
+ """.stripMargin
+ // merge condition is resolved with both target and source tables, and
we can't
+ // resolve column `i` as it's ambiguous.
+ val e2 = intercept[AnalysisException](parseAndResolve(sql2))
+ assert(e2.message.contains("Reference 'i' is ambiguous"))
+
+ val sql3 =
+ s"""
+ |MERGE INTO $target
+ |USING $source
+ |ON 1 = 1
+ |WHEN MATCHED AND (s='delete') THEN DELETE
+ """.stripMargin
+ // delete condition is resolved with both target and source tables, and
we can't
+ // resolve column `s` as it's ambiguous.
+ val e3 = intercept[AnalysisException](parseAndResolve(sql3))
+ assert(e3.message.contains("Reference 's' is ambiguous"))
+
+ val sql4 =
+ s"""
+ |MERGE INTO $target
+ |USING $source
+ |ON 1 = 1
+ |WHEN MATCHED AND (s = 'a') THEN UPDATE SET i = 1
+ """.stripMargin
+ // update condition is resolved with both target and source tables, and
we can't
+ // resolve column `s` as it's ambiguous.
+ val e4 = intercept[AnalysisException](parseAndResolve(sql4))
+ assert(e4.message.contains("Reference 's' is ambiguous"))
+
+ val sql5 =
+ s"""
+ |MERGE INTO $target
+ |USING $source
+ |ON 1 = 1
+ |WHEN MATCHED THEN UPDATE SET s = s
+ """.stripMargin
+ // update value is resolved with both target and source tables, and we
can't
+ // resolve column `s` as it's ambiguous.
+ val e5 = intercept[AnalysisException](parseAndResolve(sql5))
+ assert(e5.message.contains("Reference 's' is ambiguous"))
}
val sql =
Review comment:
super nit: sql -> sql6
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
