[SPARK-26370][SQL] Fix resolution of higher-order function for the same identifier.

[ueshin]

What changes were proposed in this pull request?

When using a higher-order function with the same variable name as the existing columns in Filter or something which uses Analyzer.resolveExpressionBottomUp during the resolution, e.g.,:

val df = Seq(
  (Seq(1, 9, 8, 7), 1, 2),
  (Seq(5, 9, 7), 2, 2),
  (Seq.empty, 3, 2),
  (null, 4, 2)
).toDF("i", "x", "d")

checkAnswer(df.filter("exists(i, x -> x % d == 0)"),
  Seq(Row(Seq(1, 9, 8, 7), 1, 2)))
checkAnswer(df.select("x").filter("exists(i, x -> x % d == 0)"),
  Seq(Row(1)))

the following exception happens:

java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.BoundReference cannot be cast to org.apache.spark.sql.catalyst.expressions.NamedExpression
  at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237)
  at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
  at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
  at scala.collection.TraversableLike.map(TraversableLike.scala:237)
  at scala.collection.TraversableLike.map$(TraversableLike.scala:230)
  at scala.collection.AbstractTraversable.map(Traversable.scala:108)
  at org.apache.spark.sql.catalyst.expressions.HigherOrderFunction.$anonfun$functionsForEval$1(higherOrderFunctions.scala:147)
  at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237)
  at scala.collection.immutable.List.foreach(List.scala:392)
  at scala.collection.TraversableLike.map(TraversableLike.scala:237)
  at scala.collection.TraversableLike.map$(TraversableLike.scala:230)
  at scala.collection.immutable.List.map(List.scala:298)
  at org.apache.spark.sql.catalyst.expressions.HigherOrderFunction.functionsForEval(higherOrderFunctions.scala:145)
  at org.apache.spark.sql.catalyst.expressions.HigherOrderFunction.functionsForEval$(higherOrderFunctions.scala:145)
  at org.apache.spark.sql.catalyst.expressions.ArrayExists.functionsForEval$lzycompute(higherOrderFunctions.scala:369)
  at org.apache.spark.sql.catalyst.expressions.ArrayExists.functionsForEval(higherOrderFunctions.scala:369)
  at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.functionForEval(higherOrderFunctions.scala:176)
  at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.functionForEval$(higherOrderFunctions.scala:176)
  at org.apache.spark.sql.catalyst.expressions.ArrayExists.functionForEval(higherOrderFunctions.scala:369)
  at org.apache.spark.sql.catalyst.expressions.ArrayExists.nullSafeEval(higherOrderFunctions.scala:387)
  at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.eval(higherOrderFunctions.scala:190)
  at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.eval$(higherOrderFunctions.scala:185)
  at org.apache.spark.sql.catalyst.expressions.ArrayExists.eval(higherOrderFunctions.scala:369)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificPredicate.eval(Unknown Source)
  at org.apache.spark.sql.execution.FilterExec.$anonfun$doExecute$3(basicPhysicalOperators.scala:216)
  at org.apache.spark.sql.execution.FilterExec.$anonfun$doExecute$3$adapted(basicPhysicalOperators.scala:215)

...

because the UnresolvedAttributes in LambdaFunction are unexpectedly resolved by the rule.

This pr modified to use a placeholder UnresolvedNamedLambdaVariable to prevent unexpected resolution.

How was this patch tested?

Added a test and modified some tests.

[ueshin]

cc @hvanhovell @cloud-fan @rednaxelafx

[SparkQA]

Test build #100135 has finished for PR 23320 at commit e63599d.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class UnresolvedNamedLambdaVariable(nameParts: Seq[String])

[ueshin]

Jenkins, retest this please.

[rednaxelafx]

Thank you so much for fixing this quickly! LGTM.

The overall implementation strategy looks nice. The basic idea is to use the new UnresolvedNamedLambdaVariable as a precedence marker, so that symbols in LambdaFunctions take higher precedence in resolving to the lambda argument than resolving to an existing attribute, right?

[rednaxelafx]

Can we provide something like toString to give better explain output? It looks like this node prints as just unresolvednamedlambdavariable() without the nameParts right now

[ueshin]

Added toString and sql.

[hvanhovell]

LGTM

[cloud-fan]

does a lambda variable name can have multiple name parts?

[rednaxelafx]

In this implementation, this placeholder/wrapper type isn't just used to represent lambda variables, but also for all UnresolvedAttributes inside of a LambdaFunction, so it needs to hold enough information to fallback to/reconstruct an UnresolvedAttribute the the name turned out not to be a lambda variable.

[cloud-fan]

ah i see!

[SparkQA]

Test build #100137 has finished for PR 23320 at commit e63599d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class UnresolvedNamedLambdaVariable(nameParts: Seq[String])

[SparkQA]

Test build #100144 has finished for PR 23320 at commit 48233a5.

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

[cloud-fan]

thanks, merging to master/2.4!