Skip to content

[SPARK-32030][SPARK-32127][SQL] Support unlimited MATCHED and NOT MATCHED clauses in MERGE INTO #28875

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

Closed
xy-xin wants to merge 5 commits into from
Closed

[SPARK-32030][SPARK-32127][SQL] Support unlimited MATCHED and NOT MATCHED clauses in MERGE INTO #28875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e18a7a5
[SPARK-32030][SQL] Support unlimited MATCHED and NOT MATCHED clauses …
Jun 19, 2020
a6ac363
Preserve the order of the MATCHED and NOT MATCHED clauses
Jun 23, 2020
1d39c92
Optimize the code
Jun 23, 2020
ab97e31
Update according to review comments
Jun 23, 2020
d5edef3
Update accordding to comments
Jun 28, 2020
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
30 changes: 11 additions & 19 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -411,12 +411,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging

val mergeCondition = expression(ctx.mergeCondition)

val matchedClauses = ctx.matchedClause()
if (matchedClauses.size() > 2) {
throw new ParseException("There should be at most 2 'WHEN MATCHED' clauses.",
matchedClauses.get(2))
}
val matchedActions = matchedClauses.asScala.map {
val matchedActions = ctx.matchedClause().asScala.map {
clause => {
if (clause.matchedAction().DELETE() != null) {
DeleteAction(Option(clause.matchedCond).map(expression))
Expand All @@ -435,12 +430,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
}
}
}
val notMatchedClauses = ctx.notMatchedClause()
if (notMatchedClauses.size() > 1) {
throw new ParseException("There should be at most 1 'WHEN NOT MATCHED' clause.",
notMatchedClauses.get(1))
}
val notMatchedActions = notMatchedClauses.asScala.map {
val notMatchedActions = ctx.notMatchedClause().asScala.map {
clause => {
if (clause.notMatchedAction().INSERT() != null) {
val condition = Option(clause.notMatchedCond).map(expression)
Expand Down Expand Up @@ -468,13 +458,15 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
throw new ParseException("There must be at least one WHEN clause in a MERGE statement", ctx)
}
// children being empty means that the condition is not set
if (matchedActions.length == 2 && matchedActions.head.children.isEmpty) {
throw new ParseException("When there are 2 MATCHED clauses in a MERGE statement, " +
"the first MATCHED clause must have a condition", ctx)
}
if (matchedActions.groupBy(_.getClass).mapValues(_.size).exists(_._2 > 1)) {
throw new ParseException(
"UPDATE and DELETE can appear at most once in MATCHED clauses in a MERGE statement", ctx)
val matchedActionSize = matchedActions.length
if (matchedActionSize >= 2 && !matchedActions.init.forall(_.condition.nonEmpty)) {
Copy link
Contributor

@cloud-fan cloud-fan Jun 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can still write .children.isEmpty, then we don't need to change v2Commands.scala

Copy link
Author

@xy-xin xy-xin Jun 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think so, because the children of InsertAction and UpdateAction actually include condition and assignments. There may be cases where there're assignments and condition being ignored but children is nonEmpty.

Copy link
Contributor

@cloud-fan cloud-fan Jun 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

then this is an existing bug?

Copy link
Author

@xy-xin xy-xin Jun 28, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it was a bug.

Copy link
Contributor

@cloud-fan cloud-fan Jun 29, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you send a new PR against branch 3.0 to fix this bug?

Copy link
Author

@xy-xin xy-xin Jun 29, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Submitted a pr at #28943.

throw new ParseException("When there are more than one MATCHED clauses in a MERGE " +
"statement, only the last MATCHED clause can omit the condition.", ctx)
}
val notMatchedActionSize = notMatchedActions.length
if (notMatchedActionSize >= 2 && !notMatchedActions.init.forall(_.condition.nonEmpty)) {
throw new ParseException("When there are more than one NOT MATCHED clauses in a MERGE " +
"statement, only the last NOT MATCHED clause can omit the condition.", ctx)
}

MergeIntoTable(
Expand Down
10 changes: 5 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,25 +346,25 @@ case class MergeIntoTable(
override def children: Seq[LogicalPlan] = Seq(targetTable, sourceTable)
}

sealed abstract class MergeAction(
condition: Option[Expression]) extends Expression with Unevaluable {
sealed abstract class MergeAction extends Expression with Unevaluable {
def condition: Option[Expression]
override def foldable: Boolean = false
override def nullable: Boolean = false
override def dataType: DataType = throw new UnresolvedException(this, "nullable")
override def children: Seq[Expression] = condition.toSeq
}

case class DeleteAction(condition: Option[Expression]) extends MergeAction(condition)
case class DeleteAction(condition: Option[Expression]) extends MergeAction

case class UpdateAction(
condition: Option[Expression],
assignments: Seq[Assignment]) extends MergeAction(condition) {
assignments: Seq[Assignment]) extends MergeAction {
override def children: Seq[Expression] = condition.toSeq ++ assignments
}

case class InsertAction(
condition: Option[Expression],
assignments: Seq[Assignment]) extends MergeAction(condition) {
assignments: Seq[Assignment]) extends MergeAction {
override def children: Seq[Expression] = condition.toSeq ++ assignments
}

Expand Down
90 changes: 43 additions & 47 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1134,58 +1134,74 @@ class DDLParserSuite extends AnalysisTest {
}
}

test("merge into table: at most two matched clauses") {
val exc = intercept[ParseException] {
parsePlan(
"""
|MERGE INTO testcat1.ns1.ns2.tbl AS target
|USING testcat2.ns1.ns2.tbl AS source
|ON target.col1 = source.col1
|WHEN MATCHED AND (target.col2='delete') THEN DELETE
|WHEN MATCHED AND (target.col2='update1') THEN UPDATE SET target.col2 = source.col2
|WHEN MATCHED AND (target.col2='update2') THEN UPDATE SET target.col2 = source.col2
|WHEN NOT MATCHED AND (target.col2='insert')
|THEN INSERT (target.col1, target.col2) values (source.col1, source.col2)
""".stripMargin)
}

assert(exc.getMessage.contains("There should be at most 2 'WHEN MATCHED' clauses."))
test("merge into table: multi matched and not matched clauses") {
parseCompare(
"""
|MERGE INTO testcat1.ns1.ns2.tbl AS target
|USING testcat2.ns1.ns2.tbl AS source
|ON target.col1 = source.col1
|WHEN MATCHED AND (target.col2='delete') THEN DELETE
|WHEN MATCHED AND (target.col2='update1') THEN UPDATE SET target.col2 = 1
|WHEN MATCHED AND (target.col2='update2') THEN UPDATE SET target.col2 = 2
|WHEN NOT MATCHED AND (target.col2='insert1')
|THEN INSERT (target.col1, target.col2) values (source.col1, 1)
|WHEN NOT MATCHED AND (target.col2='insert2')
|THEN INSERT (target.col1, target.col2) values (source.col1, 2)
""".stripMargin,
MergeIntoTable(
SubqueryAlias("target", UnresolvedRelation(Seq("testcat1", "ns1", "ns2", "tbl"))),
SubqueryAlias("source", UnresolvedRelation(Seq("testcat2", "ns1", "ns2", "tbl"))),
EqualTo(UnresolvedAttribute("target.col1"), UnresolvedAttribute("source.col1")),
Seq(DeleteAction(Some(EqualTo(UnresolvedAttribute("target.col2"), Literal("delete")))),
UpdateAction(Some(EqualTo(UnresolvedAttribute("target.col2"), Literal("update1"))),
Seq(Assignment(UnresolvedAttribute("target.col2"), Literal(1)))),
UpdateAction(Some(EqualTo(UnresolvedAttribute("target.col2"), Literal("update2"))),
Seq(Assignment(UnresolvedAttribute("target.col2"), Literal(2))))),
Seq(InsertAction(Some(EqualTo(UnresolvedAttribute("target.col2"), Literal("insert1"))),
Seq(Assignment(UnresolvedAttribute("target.col1"), UnresolvedAttribute("source.col1")),
Assignment(UnresolvedAttribute("target.col2"), Literal(1)))),
InsertAction(Some(EqualTo(UnresolvedAttribute("target.col2"), Literal("insert2"))),
Seq(Assignment(UnresolvedAttribute("target.col1"), UnresolvedAttribute("source.col1")),
Assignment(UnresolvedAttribute("target.col2"), Literal(2)))))))
}

test("merge into table: at most one not matched clause") {
test("merge into table: only the last matched clause can omit the condition") {
Copy link
Contributor

@cloud-fan cloud-fan Jun 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shall we test the same thing for not matched clause?

Copy link
Author

@xy-xin xy-xin Jun 28, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done.

val exc = intercept[ParseException] {
parsePlan(
"""
|MERGE INTO testcat1.ns1.ns2.tbl AS target
|USING testcat2.ns1.ns2.tbl AS source
|ON target.col1 = source.col1
|WHEN MATCHED AND (target.col2='delete') THEN DELETE
|WHEN MATCHED AND (target.col2='update1') THEN UPDATE SET target.col2 = source.col2
|WHEN NOT MATCHED AND (target.col2='insert1')
|THEN INSERT (target.col1, target.col2) values (source.col1, source.col2)
|WHEN NOT MATCHED AND (target.col2='insert2')
|WHEN MATCHED AND (target.col2 == 'update1') THEN UPDATE SET target.col2 = 1
|WHEN MATCHED THEN UPDATE SET target.col2 = 2
|WHEN MATCHED THEN DELETE
|WHEN NOT MATCHED AND (target.col2='insert')
|THEN INSERT (target.col1, target.col2) values (source.col1, source.col2)
""".stripMargin)
}

assert(exc.getMessage.contains("There should be at most 1 'WHEN NOT MATCHED' clause."))
assert(exc.getMessage.contains("only the last MATCHED clause can omit the condition"))
}

test("merge into table: the first matched clause must have a condition if there's a second") {
test("merge into table: only the last not matched clause can omit the condition") {
val exc = intercept[ParseException] {
parsePlan(
"""
|MERGE INTO testcat1.ns1.ns2.tbl AS target
|USING testcat2.ns1.ns2.tbl AS source
|ON target.col1 = source.col1
|WHEN MATCHED AND (target.col2 == 'update') THEN UPDATE SET target.col2 = source.col2
|WHEN MATCHED THEN DELETE
|WHEN MATCHED THEN UPDATE SET target.col2 = source.col2
|WHEN NOT MATCHED AND (target.col2='insert')
|WHEN NOT MATCHED AND (target.col2='insert1')
|THEN INSERT (target.col1, target.col2) values (source.col1, 1)
|WHEN NOT MATCHED
|THEN INSERT (target.col1, target.col2) values (source.col1, 2)
|WHEN NOT MATCHED
|THEN INSERT (target.col1, target.col2) values (source.col1, source.col2)
""".stripMargin)
}

assert(exc.getMessage.contains("the first MATCHED clause must have a condition"))
assert(exc.getMessage.contains("only the last NOT MATCHED clause can omit the condition"))
}

test("merge into table: there must be a when (not) matched condition") {
Expand All @@ -1201,26 +1217,6 @@ class DDLParserSuite extends AnalysisTest {
assert(exc.getMessage.contains("There must be at least one WHEN clause in a MERGE statement"))
}

test("merge into table: there can be only a single use DELETE or UPDATE") {
Seq("UPDATE SET *", "DELETE").foreach { op =>
val exc = intercept[ParseException] {
parsePlan(
s"""
|MERGE INTO testcat1.ns1.ns2.tbl AS target
|USING testcat2.ns1.ns2.tbl AS source
|ON target.col1 = source.col1
|WHEN MATCHED AND (target.col2='delete') THEN $op
|WHEN MATCHED THEN $op
|WHEN NOT MATCHED AND (target.col2='insert')
|THEN INSERT (target.col1, target.col2) values (source.col1, source.col2)
""".stripMargin)
}

assert(exc.getMessage.contains(
"UPDATE and DELETE can appear at most once in MATCHED clauses"))
}
}

test("show tables") {
comparePlans(
parsePlan("SHOW TABLES"),
Expand Down