You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2022/05/16 09:43:21 UTC
[spark] branch master updated: [SPARK-39096][SQL] Support MERGE commands with DEFAULT values
This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ad96695fbb2 [SPARK-39096][SQL] Support MERGE commands with DEFAULT values
ad96695fbb2 is described below
commit ad96695fbb23e300dd7061d5c29627a8a2f42fcd
Author: Daniel Tenedorio <da...@databricks.com>
AuthorDate: Mon May 16 17:42:56 2022 +0800
[SPARK-39096][SQL] Support MERGE commands with DEFAULT values
## What changes were proposed in this pull request?
DEFAULT columns are now supported in MERGE commands.
Example:
```
CREATE TABLE t1(
i STRING,
c CHAR(5),
d VARCHAR(5),
e STRING DEFAULT 'abc',
f BIGINT DEFAULT 42)
USING DELTA;
INSERT INTO t1 VALUES('1', '12345', '67890', 'def', 31);
CREATE TABLE t2(
i STRING,
c CHAR(5),
d VARCHAR(5),
e STRING DEFAULT 'abc',
f BIGINT DEFAULT 42)
USING DELTA;
INSERT INTO t2 VALUES('1', '77777', '77777', 'ghi', 20);
INSERT INTO t2 VALUES('2', '88888', '88888', 'jkl', 10);
INSERT INTO t2 VALUES('3', '99999', '99999', 'jkl', 10);
MERGE INTO t1 AS target
USING t2 AS source
ON source.i = target.i
WHEN MATCHED AND (target.c = '12345') THEN DELETE
WHEN NOT MATCHED AND (source.i = '2')
THEN INSERT (target.i, target.c, target.d, target.e, target.f)
VALUES (DEFAULT, DEFAULT, DEFAULT, DEFAULT, DEFAULT);
SELECT e, f FROM t1 WHERE e = 'abc';
> 'abc', 42
```
### Why are the changes needed?
This makes MERGE commands easier to use.
### Does this PR introduce _any_ user-facing change?
Yes.
## How was this patch tested?
This PR adds new unit test coverage.
Closes #36445 from dtenedor/default-merge.
Authored-by: Daniel Tenedorio <da...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 10 +-
.../catalyst/analysis/ResolveDefaultColumns.scala | 113 +++++++++---
.../catalyst/util/ResolveDefaultColumnsUtil.scala | 13 ++
.../spark/sql/errors/QueryCompilationErrors.scala | 14 ++
.../execution/command/PlanResolutionSuite.scala | 191 ++++++++++++++++++++-
5 files changed, 315 insertions(+), 26 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 20c1756ef4e..4dd2081c67f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -44,6 +44,7 @@ import org.apache.spark.sql.catalyst.trees.{AlwaysProcess, CurrentOrigin}
import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.util.{toPrettySQL, CharVarcharUtils}
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.TableChange.{After, ColumnPosition}
@@ -1552,7 +1553,14 @@ class Analyzer(override val catalogManager: CatalogManager)
private def resolveMergeExprOrFail(e: Expression, p: LogicalPlan): Expression = {
val resolved = resolveExpressionByPlanChildren(e, p)
- resolved.references.filter(!_.resolved).foreach { a =>
+ resolved.references.filter { attribute: Attribute =>
+ !attribute.resolved &&
+ // We exclude attribute references named "DEFAULT" from consideration since they are
+ // handled exclusively by the ResolveDefaultColumns analysis rule. That rule checks the
+ // MERGE command for such references and either replaces each one with a corresponding
+ // value, or returns a custom error message.
+ normalizeFieldName(attribute.name) != normalizeFieldName(CURRENT_DEFAULT_COLUMN_NAME)
+ }.foreach { a =>
// Note: This will throw error only on unresolved attribute issues,
// not other resolution errors like mismatched data types.
val cols = p.inputSet.toSeq.map(_.sql).mkString(", ")
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
index 80861962f06..792a0d8cf24 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
@@ -63,6 +63,8 @@ case class ResolveDefaultColumns(
resolveDefaultColumnsForInsertFromProject(i)
case u: UpdateTable =>
resolveDefaultColumnsForUpdate(u)
+ case m: MergeIntoTable =>
+ resolveDefaultColumnsForMerge(m)
}
}
@@ -164,13 +166,80 @@ case class ResolveDefaultColumns(
// For each assignment in the UPDATE command's SET clause with a DEFAULT column reference on
// the right-hand side, look up the corresponding expression from the above map.
val newAssignments: Option[Seq[Assignment]] =
- replaceExplicitDefaultValuesForUpdateAssignments(u.assignments, columnNamesToExpressions)
+ replaceExplicitDefaultValuesForUpdateAssignments(
+ u.assignments, CommandType.Update, columnNamesToExpressions)
newAssignments.map { n =>
u.copy(assignments = n)
}.getOrElse(u)
}.getOrElse(u)
}
+ /**
+ * Resolves DEFAULT column references for a MERGE INTO command.
+ */
+ private def resolveDefaultColumnsForMerge(m: MergeIntoTable): LogicalPlan = {
+ val schema: StructType = getSchemaForTargetTable(m.targetTable).getOrElse(return m)
+ // Return a more descriptive error message if the user tries to use a DEFAULT column reference
+ // inside an UPDATE command's WHERE clause; this is not allowed.
+ m.mergeCondition.foreach { c: Expression =>
+ if (c.find(isExplicitDefaultColumn).isDefined) {
+ throw QueryCompilationErrors.defaultReferencesNotAllowedInMergeCondition()
+ }
+ }
+ val defaultExpressions: Seq[Expression] = schema.fields.map {
+ case f if f.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY) =>
+ analyze(analyzer, f, "MERGE")
+ case _ => Literal(null)
+ }
+ val columnNamesToExpressions: Map[String, Expression] =
+ mapStructFieldNamesToExpressions(schema, defaultExpressions)
+ var replaced = false
+ val newMatchedActions: Seq[MergeAction] = m.matchedActions.map { action: MergeAction =>
+ replaceExplicitDefaultValuesInMergeAction(action, columnNamesToExpressions).map { r =>
+ replaced = true
+ r
+ }.getOrElse(action)
+ }
+ val newNotMatchedActions: Seq[MergeAction] = m.notMatchedActions.map { action: MergeAction =>
+ replaceExplicitDefaultValuesInMergeAction(action, columnNamesToExpressions).map { r =>
+ replaced = true
+ r
+ }.getOrElse(action)
+ }
+ if (replaced) {
+ m.copy(matchedActions = newMatchedActions,
+ notMatchedActions = newNotMatchedActions)
+ } else {
+ m
+ }
+ }
+
+ /**
+ * Replaces unresolved DEFAULT column references with corresponding values in one action of a
+ * MERGE INTO command.
+ */
+ private def replaceExplicitDefaultValuesInMergeAction(
+ action: MergeAction,
+ columnNamesToExpressions: Map[String, Expression]): Option[MergeAction] = {
+ action match {
+ case u: UpdateAction =>
+ val replaced: Option[Seq[Assignment]] =
+ replaceExplicitDefaultValuesForUpdateAssignments(
+ u.assignments, CommandType.Merge, columnNamesToExpressions)
+ replaced.map { r =>
+ Some(u.copy(assignments = r))
+ }.getOrElse(None)
+ case i: InsertAction =>
+ val replaced: Option[Seq[Assignment]] =
+ replaceExplicitDefaultValuesForUpdateAssignments(
+ i.assignments, CommandType.Merge, columnNamesToExpressions)
+ replaced.map { r =>
+ Some(i.copy(assignments = r))
+ }.getOrElse(None)
+ case _ => Some(action)
+ }
+ }
+
/**
* Regenerates user-specified columns of an InsertIntoStatement based on the names in the
* insertTableSchemaWithoutPartitionColumns field of this class.
@@ -335,7 +404,7 @@ case class ResolveDefaultColumns(
* Represents a type of command we are currently processing.
*/
private object CommandType extends Enumeration {
- val Insert, Update = Value
+ val Insert, Update, Merge = Value
}
/**
@@ -372,6 +441,9 @@ case class ResolveDefaultColumns(
case CommandType.Update =>
throw QueryCompilationErrors
.defaultReferencesNotAllowedInComplexExpressionsInUpdateSetClause()
+ case CommandType.Merge =>
+ throw QueryCompilationErrors
+ .defaultReferencesNotAllowedInComplexExpressionsInMergeInsertsOrUpdates()
}
case _ =>
None
@@ -405,17 +477,6 @@ case class ResolveDefaultColumns(
getStructFieldsForDefaultExpressions(nonUserSpecifiedFields)))
}
- /**
- * Normalizes a schema field name suitable for use in map lookups.
- */
- private def normalizeFieldName(str: String): String = {
- if (SQLConf.get.caseSensitiveAnalysis) {
- str
- } else {
- str.toLowerCase()
- }
- }
-
/**
* Returns a map of the names of fields in a schema to the fields themselves.
*/
@@ -442,17 +503,22 @@ case class ResolveDefaultColumns(
*/
private def getSchemaForTargetTable(table: LogicalPlan): Option[StructType] = {
// Check if the target table is already resolved. If so, return the computed schema.
- table match {
- case r: NamedRelation if r.schema.fields.nonEmpty => return Some(r.schema)
- case SubqueryAlias(_, r: NamedRelation) if r.schema.fields.nonEmpty => return Some (r.schema)
- case _ =>
+ // Note that we use 'collectFirst' to descend past any SubqueryAlias nodes that may be present.
+ val source: Option[LogicalPlan] = table.collectFirst {
+ case r: NamedRelation => r
+ case r: UnresolvedCatalogRelation => r
+ }
+ source.map { r =>
+ if (r.schema.fields.nonEmpty) {
+ return Some(r.schema)
+ }
}
// Lookup the relation from the catalog by name. This either succeeds or returns some "not
// found" error. In the latter cases, return out of this rule without changing anything and let
// the analyzer return a proper error message elsewhere.
- val tableName: TableIdentifier = table match {
- case r: UnresolvedRelation => TableIdentifier(r.name)
- case r: UnresolvedCatalogRelation => r.tableMeta.identifier
+ val tableName: TableIdentifier = source match {
+ case Some(r: UnresolvedRelation) => TableIdentifier(r.name)
+ case Some(r: UnresolvedCatalogRelation) => r.tableMeta.identifier
case _ => return None
}
val lookup: LogicalPlan = try {
@@ -464,17 +530,18 @@ case class ResolveDefaultColumns(
case SubqueryAlias(_, r: UnresolvedCatalogRelation) =>
Some(r.tableMeta.schema)
case SubqueryAlias(_, r: View) if r.isTempView =>
- Some(r.schema)
+ Some(r.desc.schema)
case _ => None
}
}
/**
* Replaces unresolved DEFAULT column references with corresponding values in a series of
- * assignments in an UPDATE command.
+ * assignments in an UPDATE assignment, either comprising an UPDATE command or as part of a MERGE.
*/
private def replaceExplicitDefaultValuesForUpdateAssignments(
assignments: Seq[Assignment],
+ command: CommandType.Value,
columnNamesToExpressions: Map[String, Expression]): Option[Seq[Assignment]] = {
var replaced = false
val newAssignments: Seq[Assignment] =
@@ -491,7 +558,7 @@ case class ResolveDefaultColumns(
replaceExplicitDefaultReferenceInExpression(
assignment.value,
defaultExpr,
- CommandType.Update,
+ command,
addAlias = false)
updated.map { e =>
replaced = true
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
index 6e23ba46396..d2963a60409 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
@@ -155,4 +155,17 @@ object ResolveDefaultColumns {
s"statement provided a value of incompatible type ${analyzed.dataType}")
}
}
+ /**
+ * Normalizes a schema field name suitable for use in looking up into maps keyed by schema field
+ * names.
+ * @param str the field name to normalize
+ * @return the normalized result
+ */
+ def normalizeFieldName(str: String): String = {
+ if (SQLConf.get.caseSensitiveAnalysis) {
+ str
+ } else {
+ str.toLowerCase()
+ }
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index d803cd23df6..912f65aa58c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -2431,4 +2431,18 @@ object QueryCompilationErrors extends QueryErrorsBase {
"Failed to execute UPDATE command because the WHERE clause contains a DEFAULT column " +
"reference; this is not allowed")
}
+
+ // Return a more descriptive error message if the user tries to use a DEFAULT column reference
+ // inside an UPDATE command's WHERE clause; this is not allowed.
+ def defaultReferencesNotAllowedInMergeCondition(): Throwable = {
+ new AnalysisException(
+ "Failed to execute MERGE command because the WHERE clause contains a DEFAULT column " +
+ "reference; this is not allowed")
+ }
+
+ def defaultReferencesNotAllowedInComplexExpressionsInMergeInsertsOrUpdates(): Throwable = {
+ new AnalysisException(
+ "Failed to execute MERGE INTO command because one of its INSERT or UPDATE assignments " +
+ "contains a DEFAULT column reference as part of another expression; this is not allowed")
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
index 7e86bd01912..05326e09f66 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.{AnalysisException, SaveMode}
import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AnalysisTest, Analyzer, EmptyFunctionRegistry, NoSuchTableException, ResolvedDBObjectName, ResolvedFieldName, ResolvedTable, ResolveSessionCatalog, UnresolvedAttribute, UnresolvedRelation, UnresolvedSubqueryColumnAliases, UnresolvedTable}
import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogTable, CatalogTableType, InMemoryCatalog, SessionCatalog}
-import org.apache.spark.sql.catalyst.expressions.{AnsiCast, AttributeReference, EqualTo, Expression, InSubquery, IntegerLiteral, ListQuery, Literal, StringLiteral}
+import org.apache.spark.sql.catalyst.expressions.{AnsiCast, AttributeReference, Cast, EqualTo, Expression, InSubquery, IntegerLiteral, ListQuery, Literal, StringLiteral}
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
import org.apache.spark.sql.catalyst.plans.logical.{AlterColumn, AnalysisOnlyCommand, AppendData, Assignment, CreateTable, CreateTableAsSelect, DeleteAction, DeleteFromTable, DescribeRelation, DropTable, InsertAction, LocalRelation, LogicalPlan, MergeIntoTable, OneRowRelation, Project, SetTableLocation, SetTableProperties, ShowTableProperties, SubqueryAlias, UnsetTableProperties, UpdateAction, UpdateTable}
@@ -114,6 +114,16 @@ class PlanResolutionSuite extends AnalysisTest {
t
}
+ private val tableWithColumnNamedDefault: Table = {
+ val t = mock(classOf[Table])
+ when(t.schema()).thenReturn(
+ new StructType()
+ .add("s", StringType)
+ .add("default", StringType))
+ when(t.partitioning()).thenReturn(Array.empty[Transform])
+ t
+ }
+
private val v1Table: V1Table = {
val t = mock(classOf[CatalogTable])
when(t.schema).thenReturn(new StructType()
@@ -151,6 +161,7 @@ class PlanResolutionSuite extends AnalysisTest {
case "charvarchar" => charVarcharTable
case "defaultvalues" => defaultValues
case "defaultvalues2" => defaultValues2
+ case "tablewithcolumnnameddefault" => tableWithColumnNamedDefault
case name => throw new NoSuchTableException(name)
}
})
@@ -1335,7 +1346,7 @@ class PlanResolutionSuite extends AnalysisTest {
}
}
- test("MERGE INTO TABLE") {
+ test("MERGE INTO TABLE - primary") {
def checkResolution(
target: LogicalPlan,
source: LogicalPlan,
@@ -1536,6 +1547,182 @@ class PlanResolutionSuite extends AnalysisTest {
case other => fail("Expect MergeIntoTable, but got:\n" + other.treeString)
}
+
+ // DEFAULT columns (implicit):
+ // All cases of the $target table lack any explicitly-defined DEFAULT columns. Therefore any
+ // DEFAULT column references in the below MERGE INTO command should resolve to literal NULL.
+ // This test case covers that behavior.
+ val sql6 =
+ s"""
+ |MERGE INTO $target AS target
+ |USING $source AS 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 = DEFAULT, target.i = target.i
+ |WHEN NOT MATCHED AND (source.s='insert')
+ | THEN INSERT (target.i, target.s) values (DEFAULT, DEFAULT)
+ """.stripMargin
+ parseAndResolve(sql6) match {
+ case m: MergeIntoTable =>
+ val source = m.sourceTable
+ val target = m.targetTable
+ val ti = target.output.find(_.name == "i").get.asInstanceOf[AttributeReference]
+ val si = source.output.find(_.name == "i").get.asInstanceOf[AttributeReference]
+ m.mergeCondition match {
+ case EqualTo(l: AttributeReference, r: AttributeReference) =>
+ assert(l.sameRef(ti) && r.sameRef(si))
+ case Literal(_, BooleanType) => // this is acceptable as a merge condition
+ case other => fail("unexpected merge condition " + other)
+ }
+ assert(m.matchedActions.length == 2)
+ val first = m.matchedActions(0)
+ first match {
+ case DeleteAction(Some(EqualTo(_: AttributeReference, StringLiteral("delete")))) =>
+ case other => fail("unexpected first matched action " + other)
+ }
+ val second = m.matchedActions(1)
+ second match {
+ case UpdateAction(Some(EqualTo(_: AttributeReference, StringLiteral("update"))),
+ Seq(
+ Assignment(_: AttributeReference, AnsiCast(Literal(null, _), StringType, _)),
+ Assignment(_: AttributeReference, _: AttributeReference))) =>
+ case other => fail("unexpected second matched action " + other)
+ }
+ assert(m.notMatchedActions.length == 1)
+ val negative = m.notMatchedActions(0)
+ negative match {
+ case InsertAction(Some(EqualTo(_: AttributeReference, StringLiteral("insert"))),
+ Seq(Assignment(i: AttributeReference, AnsiCast(Literal(null, _), IntegerType, _)),
+ Assignment(s: AttributeReference, AnsiCast(Literal(null, _), StringType, _)))) =>
+ assert(i.name == "i")
+ assert(s.name == "s")
+ case other => fail("unexpected not matched action " + other)
+ }
+
+ case other =>
+ fail("Expect MergeIntoTable, but got:\n" + other.treeString)
+ }
+
+ // DEFAULT column reference in the merge condition:
+ // This MERGE INTO command includes an ON clause with a DEFAULT column reference. This is
+ // invalid and returns an error message.
+ val mergeWithDefaultReferenceInMergeCondition =
+ s"""
+ |MERGE INTO testcat.tab AS target
+ |USING testcat.tab1 AS source
+ |ON target.i = DEFAULT
+ |WHEN MATCHED AND (target.s = 31) THEN DELETE
+ |WHEN MATCHED AND (target.s = 31)
+ | THEN UPDATE SET target.s = DEFAULT
+ |WHEN NOT MATCHED AND (source.s='insert')
+ | THEN INSERT (target.i, target.s) values (DEFAULT, DEFAULT)
+ """.stripMargin
+ assert(intercept[AnalysisException] {
+ parseAndResolve(mergeWithDefaultReferenceInMergeCondition)
+ }.getMessage.contains(
+ QueryCompilationErrors.defaultReferencesNotAllowedInMergeCondition().getMessage))
+
+ // DEFAULT column reference within a complex expression:
+ // This MERGE INTO command includes a WHEN MATCHED clause with a DEFAULT column reference as
+ // of a complex expression (DEFAULT + 1). This is invalid and returns an error message.
+ val mergeWithDefaultReferenceAsPartOfComplexExpression =
+ s"""
+ |MERGE INTO testcat.tab AS target
+ |USING testcat.tab1 AS source
+ |ON target.i = source.i
+ |WHEN MATCHED AND (target.s = 31) THEN DELETE
+ |WHEN MATCHED AND (target.s = 31)
+ | THEN UPDATE SET target.s = DEFAULT + 1
+ |WHEN NOT MATCHED AND (source.s='insert')
+ | THEN INSERT (target.i, target.s) values (DEFAULT, DEFAULT)
+ """.stripMargin
+ assert(intercept[AnalysisException] {
+ parseAndResolve(mergeWithDefaultReferenceAsPartOfComplexExpression)
+ }.getMessage.contains(
+ QueryCompilationErrors
+ .defaultReferencesNotAllowedInComplexExpressionsInMergeInsertsOrUpdates().getMessage))
+
+ // Ambiguous DEFAULT column reference when the table itself contains a column named
+ // "DEFAULT".
+ val mergeIntoTableWithColumnNamedDefault =
+ s"""
+ |MERGE INTO testcat.tablewithcolumnnameddefault AS target
+ |USING testcat.tab1 AS source
+ |ON default = source.i
+ |WHEN MATCHED AND (target.s = 32) THEN DELETE
+ |WHEN MATCHED AND (target.s = 32)
+ | THEN UPDATE SET target.s = DEFAULT
+ |WHEN NOT MATCHED AND (source.s='insert')
+ | THEN INSERT (target.s) values (DEFAULT)
+ """.stripMargin
+ parseAndResolve(mergeIntoTableWithColumnNamedDefault, withDefault = true) match {
+ case m: MergeIntoTable =>
+ val target = m.targetTable
+ val d = target.output.find(_.name == "default").get.asInstanceOf[AttributeReference]
+ m.mergeCondition match {
+ case EqualTo(Cast(l: AttributeReference, _, _, _), _: AttributeReference) =>
+ assert(l.sameRef(d))
+ case Literal(_, BooleanType) => // this is acceptable as a merge condition
+ case other =>
+ fail("unexpected merge condition " + other)
+ }
+ assert(m.matchedActions.length == 2)
+ assert(m.notMatchedActions.length == 1)
+
+ case other =>
+ fail("Expect MergeIntoTable, but got:\n" + other.treeString)
+ }
+ }
+
+ // DEFAULT columns (explicit):
+ // The defaultvalues table includes explicitly-defined DEFAULT columns in its schema. Therefore,
+ // DEFAULT column references in the below MERGE INTO command should resolve to the corresponding
+ // values. This test case covers that behavior.
+ val mergeDefaultWithExplicitDefaultColumns =
+ s"""
+ |MERGE INTO defaultvalues AS target
+ |USING testcat.tab1 AS source
+ |ON target.i = source.i
+ |WHEN MATCHED AND (target.s = 31) THEN DELETE
+ |WHEN MATCHED AND (target.s = 31)
+ | THEN UPDATE SET target.s = DEFAULT
+ |WHEN NOT MATCHED AND (source.s='insert')
+ | THEN INSERT (target.i, target.s) values (DEFAULT, DEFAULT)
+ """.stripMargin
+ parseAndResolve(mergeDefaultWithExplicitDefaultColumns, true) match {
+ case m: MergeIntoTable =>
+ val cond = m.mergeCondition
+ cond match {
+ case EqualTo(Cast(l: AttributeReference, IntegerType, _, _), r: AttributeReference) =>
+ assert(l.name == "i")
+ assert(r.name == "i")
+ case Literal(_, BooleanType) => // this is acceptable as a merge condition
+ case other => fail("unexpected merge condition " + other)
+ }
+ assert(m.matchedActions.length == 2)
+ val first = m.matchedActions(0)
+ first match {
+ case DeleteAction(Some(EqualTo(_: AttributeReference, Literal(31, IntegerType)))) =>
+ case other => fail("unexpected first matched action " + other)
+ }
+ val second = m.matchedActions(1)
+ second match {
+ case UpdateAction(Some(EqualTo(_: AttributeReference, Literal(31, IntegerType))),
+ Seq(Assignment(_: AttributeReference, Literal(42, IntegerType)))) =>
+ case other => fail("unexpected second matched action " + other)
+ }
+ assert(m.notMatchedActions.length == 1)
+ val negative = m.notMatchedActions(0)
+ negative match {
+ case InsertAction(Some(EqualTo(_: AttributeReference, StringLiteral("insert"))),
+ Seq(Assignment(_: AttributeReference, Literal(true, BooleanType)),
+ Assignment(_: AttributeReference, Literal(42, IntegerType)))) =>
+ case other => fail("unexpected not matched action " + other)
+ }
+
+ case other =>
+ fail("Expect MergeIntoTable, but got:\n" + other.treeString)
}
// no aliases
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org