[SPARK-32308][SQL] Move by-name resolution logic of unionByName from API code to analysis phase #29107

viirya · 2020-07-14T16:50:54Z

What changes were proposed in this pull request?

Currently the by-name resolution logic of unionByName is put in API code. This patch moves the logic to analysis phase.
See #28996 (comment).

Why are the changes needed?

Logically we should do resolution in analysis phase. This refactoring cleans up API method and makes consistent resolution.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Unit tests.

SparkQA · 2020-07-14T20:32:13Z

Test build #125845 has finished for PR 29107 at commit 93c5ea1.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-07-14T20:49:40Z

Thank you, @viirya .
cc @cloud-fan

SparkQA · 2020-07-15T00:40:05Z

Test build #125853 has finished for PR 29107 at commit 8e4867a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

maropu · 2020-07-15T07:22:50Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

+      case Union(children, byName, allowMissingCol)
+          if byName =>
+        val union = children.reduceLeft { (left: LogicalPlan, right: LogicalPlan) =>
+          unionTwoSides(left, right, allowMissingCol)


Just a question; the process of unionTwoSides is a type-coercion one? Seems like column resolution rather than type coercion.

Seems not like typical type coercion case, but put here based on #28996 (comment).

If looks not proper after rethinking, we can also move to other rule or create another rule.

yea maybe a new analyzer rule.

maropu · 2020-07-18T02:56:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+    case e if !e.childrenResolved => e
+
+    case Union(children, byName, allowMissingCol)
+      if byName =>


nit: case Union(children, byName, allowMissingCol) if byName =>?

maropu · 2020-07-18T03:13:08Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala


+          case Union(_, byName, allowMissingCol) if byName || allowMissingCol =>
+            failAnalysis("Union should not be with true `byName` or " +
+              "`allowMissingCol` flags after analysis phase.")


Just a question; users can see this error message? That's the case of an analyzer bug?

Usually not. This mainly prevents we accidentally create a Union with byName or allowMissingCol after ResolveUnion rule.

And, yes, prevent a unexpected bug during analysis.

SparkQA · 2020-07-18T04:33:05Z

Test build #126077 has finished for PR 29107 at commit c23898e.

This patch fails PySpark pip packaging tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2020-07-18T05:07:06Z

retest this please

SparkQA · 2020-07-18T07:05:01Z

Test build #126093 has finished for PR 29107 at commit c23898e.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2020-07-18T15:37:05Z

retest this please

SparkQA · 2020-07-18T20:14:23Z

Test build #126112 has finished for PR 29107 at commit c23898e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

mukulmurthy · 2020-07-20T20:17:56Z

I just linked the JIRA to SPARK-29358 so it's easier to track the lineage.

Is there already another follow up ticket to make this work for nested structs?

viirya · 2020-07-20T21:02:53Z

I just linked the JIRA to SPARK-29358 so it's easier to track the lineage.

Is there already another follow up ticket to make this work for nested structs?

Thanks. I don't see an existing follow up ticket for nested structs.

mukulmurthy · 2020-07-20T22:13:05Z

Awesome, I just created https://issues.apache.org/jira/browse/SPARK-32376.

We've implemented a utility to do this before; I'll sync with my team and see if it's easy to port the code over and open a PR for that unless you're already planning on tackling it. Even if so, we're happy to share our test cases.

viirya · 2020-07-20T22:52:08Z

Thanks @mukulmurthy. I've not worked on it yet. Feel free to open a PR if you are ready to port the code.

cloud-fan · 2020-07-21T11:49:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+ * run before `TypeCoercion`, because `TypeCoercion` should be run on correctly resolved
+ * column by name.
+ */
+object ResolveUnion extends Rule[LogicalPlan] {


Can we put it in a new file?

cloud-fan · 2020-07-21T11:50:36Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+
+/**
+ * Resolves different children of Union to a common set of columns. Note that this must be
+ * run before `TypeCoercion`, because `TypeCoercion` should be run on correctly resolved


It's fragile to rely on rule order. How about we skip the type coercion rule if the union is by name and the name match is not done yet?

cloud-fan · 2020-07-21T11:52:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

+    } else {
+      left
+    }
+    Union(leftChild, rightChild)


can we avoid creating intermediate unions?

nvm, it was the behavior before and seems hard to get rid of it.

cloud-fan · 2020-07-21T11:58:37Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

          s.children.forall(_.output.length == s.children.head.output.length) && !s.resolved =>
        val newChildren: Seq[LogicalPlan] = buildNewChildrenWithWiderTypes(s.children)
-        s.makeCopy(Array(newChildren))
+        Union(newChildren, s.byName, s.allowMissingCol)


nit: s.copy(children = newChildren)

cloud-fan · 2020-07-21T11:58:55Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

-    case LocalLimit(exp, Union(children)) =>
-      LocalLimit(exp, Union(children.map(maybePushLocalLimit(exp, _))))
+    case LocalLimit(exp, Union(children, byName, allowMissingCol)) =>
+      LocalLimit(exp, Union(children.map(maybePushLocalLimit(exp, _)), byName, allowMissingCol))


nit: use copy would be better, as it preserves the tree node tag

cloud-fan · 2020-07-21T11:59:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

          Project(projectList.map(pushToRight(_, rewrites)), child)
        }
-        Union(newFirstChild +: newOtherChildren)
+        Union(newFirstChild +: newOtherChildren, byName, allowMissingCol)


cloud-fan · 2020-07-21T12:00:29Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

      }
    }
-    Union(flattened.toSeq)
+    Union(flattened, topByName, topAllowMissingCol)


ditto: use copy

cloud-fan · 2020-07-21T12:03:30Z

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnionSuite.scala

+    val expected2 = Union(table1 :: projected2 :: Nil)
+    comparePlans(analyzed2, expected2)
+
+    // By name + Allow missing column


by position?

cloud-fan · 2020-07-21T12:05:20Z

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnionSuite.scala

+    // By name + Allow missing column
+    val union3 = Union(union1 :: union2 :: Nil)
+    val analyzed3 = analyzer.execute(union3)
+    val expected3 = Union(expected1 :: expected2 :: Nil)


does this test prove anything? union1 and union2 have exactly the same schema.

Oh, let me change it.

cloud-fan · 2020-07-21T12:06:51Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

      sparkSession.sessionState.conf.caseSensitiveAnalysis)
    SchemaUtils.checkColumnNameDuplication(
      rightOutputAttrs.map(_.name),
      "in the right attributes",


shall we move the check to the analyzer rule as well?

maropu

LGTM except for the minor comments.

maropu · 2020-07-22T00:52:44Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnion.scala

+          Alias(Literal(null, lattr.dataType), lattr.name)()
+        } else {
+          throw new AnalysisException(
+            s"""Cannot resolve column name "${lattr.name}" among """ +


How about making it consistent (using ` for wrapping a column name)?

throw new AnalysisException( s"Cannot resolve column name `${lattr.name}` among " + s"(${rightOutputAttrs.map(_.name).mkString(", ")})")

https://github.com/apache/spark/pull/29107/files#diff-1d14ac233eac6f233c027dba0bdf871dR341

maropu · 2020-07-22T00:53:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveUnion.scala

+    case e if !e.childrenResolved => e
+
+    case Union(children, byName, allowMissingCol) if byName =>
+      val union = children.reduceLeft { (left: LogicalPlan, right: LogicalPlan) =>


nit: { (left: LogicalPlan, right: LogicalPlan) => -> { (left, right) =>

maropu · 2020-07-22T00:54:45Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

      case s: Union if s.childrenResolved &&
-          s.children.forall(_.output.length == s.children.head.output.length) && !s.resolved =>
+          s.children.forall(_.output.length == s.children.head.output.length) && !s.resolved &&
+          !s.byName =>


nit: Since s.byName is constant, how about evaluating it first?

case s @ Union(_, false, _) if s.childrenResolved && !s.byName && s.children.forall(_.output.length == s.children.head.output.length) && !s.resolved =>

SparkQA · 2020-07-22T01:38:57Z

Test build #126277 has finished for PR 29107 at commit 1839987.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-07-22T07:05:02Z

Test build #126294 has finished for PR 29107 at commit eca8fc6.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2020-07-22T07:13:12Z

retest this please

cloud-fan · 2020-07-22T08:40:04Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

      case e @ Except(left, right, _) if !e.duplicateResolved =>
        e.copy(right = dedupRight(left, right))
-      case u @ Union(children) if !u.duplicateResolved =>
+      case u @ Union(children, _, _) if !u.duplicateResolved =>


we should do this only when the by-name resolution is done?

duplicateResolved checks attribute sets from children that look at exprId actually. By-name resolution is only for attribute name. If you think it is safer, I can add it here.

I think it's safer to add it, to explicitly define the rule order (by-name resolution should happen before this rule)

...catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala

cloud-fan · 2020-07-22T08:47:24Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

          case Tail(limitExpr, _) => checkLimitLikeClause("tail", limitExpr)

+          case Union(_, byName, allowMissingCol) if byName || allowMissingCol =>
+            failAnalysis("Union should not be with true `byName` or " +


To be safe, shall we also override Union.resolved to include this?

I add !(byName || allowMissingCol) to the condition at Union.resolved now. Then do we still need to check them at CheckAnalysis like above? Or just follow other nodes with special check?

Since this is not something users can hit(only bug can trigger this), maybe we can just remove this check here.

cloud-fan · 2020-07-22T08:49:00Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

      case logical.GlobalLimit(IntegerLiteral(limit), child) =>
        execution.GlobalLimitExec(limit, planLater(child)) :: Nil
-      case logical.Union(unionChildren) =>
+      case logical.Union(unionChildren, _, _) =>


this pattern appears many times (Union(children, _, _)), and actually we can rewrite to

case union: logical.Union => execution.UnionExec(union.children.map(planLater)) :: Nil

cloud-fan · 2020-07-22T08:51:28Z

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala

      df1.unionByName(df2)
    }.getMessage
-    assert(errMsg.contains("""Cannot resolve column name "b" among (a, c, d)"""))
+    assert(errMsg.contains("""Cannot resolve column name `b` among (a, c, d)"""))


Shall we keep the previous error message? In general we should only use backticks when necessary(has special chars)

ok. let me revert to previous one.

SparkQA · 2020-07-22T12:54:04Z

Test build #126311 has finished for PR 29107 at commit eca8fc6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-07-22T22:01:03Z

Test build #126348 has finished for PR 29107 at commit 0381f5d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-07-23T00:19:26Z

Test build #126355 has finished for PR 29107 at commit 2ab990e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-07-23T04:47:29Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

        e.copy(right = dedupRight(left, right))
-      case u @ Union(children) if !u.duplicateResolved =>
+      // Only after we finish by-name resolution for Union
+      case u: logical.Union if !u.duplicateResolved && !u.byName =>


nit: logical.Union -> Union.

!u.duplicateResolved && !u.byName -> !u.byName && !u.duplicateResolved

cloud-fan · 2020-07-23T04:49:05Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

          case (l, r) => l.dataType.sameType(r.dataType)
        })
-    children.length > 1 && childrenResolved && allChildrenCompatible
+    children.length > 1 && childrenResolved && allChildrenCompatible && !(byName || allowMissingCol)


nit: !(byName || allowMissingCol) should be checked before childrenResolved, as it's cheaper.

cloud-fan

LGTM except a few minor comments

SparkQA · 2020-07-23T07:05:01Z

Test build #126388 has finished for PR 29107 at commit 2a9e1e4.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-07-23T14:40:13Z

retest this please

SparkQA · 2020-07-23T20:25:17Z

Test build #126417 has finished for PR 29107 at commit 2a9e1e4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-07-24T04:33:12Z

thanks, merging to master!

viirya · 2020-07-24T05:07:42Z

Thanks for review.

viirya added 3 commits July 14, 2020 09:36

Move by-name resolution logic of Union to analysis phase.

3d5c099

Merge remote-tracking branch 'upstream/master' into move-union-by-name

94087b8

Fix merging conflict.

93c5ea1

probot-autolabeler bot added the SQL label Jul 14, 2020

Fix test.

8e4867a

maropu reviewed Jul 15, 2020

View reviewed changes

viirya added 2 commits July 17, 2020 16:33

Move to individual analysis rule.

9ddd70f

Remove unnecessary change.

c23898e

maropu reviewed Jul 18, 2020

View reviewed changes

cloud-fan reviewed Jul 21, 2020

View reviewed changes

viirya added 2 commits July 21, 2020 13:27

Address comments.

c827eb2

Remove unused import.

1839987

maropu approved these changes Jul 22, 2020

View reviewed changes

Address comments.

eca8fc6

cloud-fan reviewed Jul 22, 2020

View reviewed changes

...catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jul 22, 2020

View reviewed changes

viirya added 2 commits July 22, 2020 09:28

Address comments.

0381f5d

For review comments.

2ab990e

cloud-fan reviewed Jul 23, 2020

View reviewed changes

cloud-fan approved these changes Jul 23, 2020

View reviewed changes

Change condition order.

2a9e1e4

cloud-fan closed this in 84efa04 Jul 24, 2020

viirya deleted the move-union-by-name branch December 27, 2023 18:28

[SPARK-32308][SQL] Move by-name resolution logic of unionByName from API code to analysis phase #29107

[SPARK-32308][SQL] Move by-name resolution logic of unionByName from API code to analysis phase #29107

Uh oh!

Conversation

viirya commented Jul 14, 2020

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Jul 14, 2020

Uh oh!

dongjoon-hyun commented Jul 14, 2020

Uh oh!

SparkQA commented Jul 15, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 18, 2020

Uh oh!

viirya commented Jul 18, 2020

Uh oh!

SparkQA commented Jul 18, 2020

Uh oh!

viirya commented Jul 18, 2020

Uh oh!

SparkQA commented Jul 18, 2020

Uh oh!

mukulmurthy commented Jul 20, 2020

Uh oh!

viirya commented Jul 20, 2020

Uh oh!

mukulmurthy commented Jul 20, 2020

Uh oh!

viirya commented Jul 20, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 21, 2020 •

edited

Loading