Skip to content

[SPARK-29883][SQL] Implement a helper method for aliasing bool_and() and bool_or() #26712

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
amanomer wants to merge 12 commits into from
Closed

[SPARK-29883][SQL] Implement a helper method for aliasing bool_and() and bool_or() #26712

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
dbf6b4f
Added MultiNamedExpression
amanomer Dec 2, 2019
3387eef
Removed unused import
amanomer Dec 2, 2019
4476e7e
Added expressionWithAlias
amanomer Dec 3, 2019
77ad53f
Updated try-catch
amanomer Dec 3, 2019
404d829
Updated TC ExplainSuite.scala
amanomer Dec 3, 2019
27f1a7f
TC update
amanomer Dec 3, 2019
2952898
Updated TC in group-by.sql.out
amanomer Dec 4, 2019
ab2e422
TC updated udf-group-by.sql
amanomer Dec 4, 2019
bb665e4
fixed TC
amanomer Dec 4, 2019
3c87222
validate expression
amanomer Dec 5, 2019
17c91f4
Handled for expressions that take more than 1 parameter
amanomer Dec 5, 2019
d07d261
Add meaningfull error message
amanomer Dec 6, 2019
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
28 changes: 23 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -313,11 +313,11 @@ object FunctionRegistry {
expression[CollectList]("collect_list"),
expression[CollectSet]("collect_set"),
expression[CountMinSketchAgg]("count_min_sketch"),
expression[BoolAnd]("every"),
expression[BoolAnd]("bool_and"),
expression[BoolOr]("any"),
expression[BoolOr]("some"),
expression[BoolOr]("bool_or"),
expressionWithAlias[BoolAnd]("every"),
expressionWithAlias[BoolAnd]("bool_and"),
expressionWithAlias[BoolOr]("any"),
expressionWithAlias[BoolOr]("some"),
expressionWithAlias[BoolOr]("bool_or"),

// string functions
expression[Ascii]("ascii"),
Expand Down Expand Up @@ -631,6 +631,24 @@ object FunctionRegistry {
(name, (expressionInfo[T](name), builder))
}

private def expressionWithAlias[T <: Expression](name: String)
(implicit tag: ClassTag[T]): (String, (ExpressionInfo, FunctionBuilder)) = {
val constructors = tag.runtimeClass.getConstructors
.filter(_.getParameterTypes.head == classOf[String])
assert(constructors.length == 1)
val builder = (expressions: Seq[Expression]) => {
Try(constructors.head.newInstance(name.toString, expressions.head).asInstanceOf[Expression])
Copy link
Contributor

@cloud-fan cloud-fan Dec 3, 2019

Choose a reason for hiding this comment

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

seems better to use the normal try catch?

try {
  constructors.head.newInstance(name.toString, expressions.head).asInstanceOf[Expression]
} catch {
  // the original comment ...
  case e => throw new AnalysisException(e.getCause.getMessage)
}

We can update def expression as well.

Copy link
Contributor Author

@amanomer amanomer Dec 3, 2019

Choose a reason for hiding this comment

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

Updated in latest commit.

match {
case Success(e) => e
case Failure(e) =>
// the exception is an invocation exception. To get a meaningful message, we need the
// cause.
throw new AnalysisException(e.getCause.getMessage)
}
}
(name, (expressionInfo[T](name), builder))
}

/**
* Creates a function registry lookup entry for cast aliases (SPARK-16730).
* For example, if name is "int", and dataType is IntegerType, this means int(x) would become
Expand Down
8 changes: 4 additions & 4 deletions .../src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/UnevaluableAggs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,8 @@ abstract class UnevaluableBooleanAggBase(arg: Expression)
false
""",
since = "3.0.0")
case class BoolAnd(arg: Expression) extends UnevaluableBooleanAggBase(arg) {
override def nodeName: String = "bool_and"
case class BoolAnd(funcName: String, arg: Expression) extends UnevaluableBooleanAggBase(arg) {
override def nodeName: String = funcName
}

@ExpressionDescription(
Expand All @@ -68,6 +68,6 @@ case class BoolAnd(arg: Expression) extends UnevaluableBooleanAggBase(arg) {
false
""",
since = "3.0.0")
case class BoolOr(arg: Expression) extends UnevaluableBooleanAggBase(arg) {
override def nodeName: String = "bool_or"
case class BoolOr(funcName: String, arg: Expression) extends UnevaluableBooleanAggBase(arg) {
override def nodeName: String = funcName
}
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,8 @@ object ReplaceExpressions extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
case e: RuntimeReplaceable => e.child
case CountIf(predicate) => Count(new NullIf(predicate, Literal.FalseLiteral))
case BoolOr(arg) => Max(arg)
case BoolAnd(arg) => Min(arg)
case BoolOr(_, arg) => Max(arg)
case BoolAnd(_, arg) => Min(arg)
}
}

Expand Down
4 changes: 2 additions & 2 deletions ...t/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,8 +144,8 @@ class ExpressionTypeCheckingSuite extends SparkFunSuite {
assertSuccess(Sum('stringField))
assertSuccess(Average('stringField))
assertSuccess(Min('arrayField))
assertSuccess(new BoolAnd('booleanField))
assertSuccess(new BoolOr('booleanField))
assertSuccess(new BoolAnd("bool_and", 'booleanField))
assertSuccess(new BoolOr("bool_or", 'booleanField))

assertError(Min('mapField), "min does not support ordering on type")
assertError(Max('mapField), "max does not support ordering on type")
Expand Down