Skip to content
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

Jump to bottom

[SPARK-16484][SQL] Update hll function type checks to also check for non-foldable inputs #41203

Closed
wants to merge 4 commits into from
Closed

[SPARK-16484][SQL] Update hll function type checks to also check for non-foldable inputs #41203

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
20d414a
Update hll function type checks to also check for non-foldable inputs
RyanBerti May 17, 2023
4d6dd31
Introduce ExpectsInputTypesAndFoldable, utilize it in datasketches fu…
RyanBerti May 17, 2023
87be383
Fix error message formatting
RyanBerti May 18, 2023
a169cec
Update sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expr…
RyanBerti May 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
43 changes: 43 additions & 0 deletions ...in/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala
View file
Open in desktop
Expand Up @@ -22,7 +22,10 @@ import org.apache.datasketches.hll.{HllSketch, TgtHllType, Union}
import org.apache.datasketches.memory.Memory

import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, ExpressionDescription, Literal}
import org.apache.spark.sql.catalyst.expressions.Cast.{toSQLExpr, toSQLType}
import org.apache.spark.sql.catalyst.trees.BinaryLike
import org.apache.spark.sql.types.{AbstractDataType, BinaryType, BooleanType, DataType, IntegerType, LongType, StringType, TypeCollection}
import org.apache.spark.unsafe.types.UTF8String
Expand Down Expand Up @@ -96,6 +99,26 @@ case class HllSketchAgg(
newRight: Expression): HllSketchAgg =
copy(left = newLeft, right = newRight)

// Overrides for ExpectsInputTypes

override def checkInputDataTypes(): TypeCheckResult = {
val defaultCheck = super.checkInputDataTypes()
if (defaultCheck.isFailure) {
defaultCheck
} else if (!right.foldable) {
DataTypeMismatch(
errorSubClass = "NON_FOLDABLE_INPUT",
messageParameters = Map(
"inputName" -> "lgConfigK",
"inputType" -> toSQLType(right.dataType),
"inputExpr" -> toSQLExpr(right)
)
)
} else {
TypeCheckSuccess
}
}

// Overrides for TypedImperativeAggregate

override def prettyName: String = "hll_sketch_agg"
Expand Down Expand Up @@ -265,6 +288,26 @@ case class HllUnionAgg(
override protected def withNewChildrenInternal(newLeft: Expression, newRight: Expression):
HllUnionAgg = copy(left = newLeft, right = newRight)

// Overrides for ExpectsInputTypes

override def checkInputDataTypes(): TypeCheckResult = {
Copy link
Contributor

Choose a reason for hiding this comment

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

this looks mostly duplicated with L104-120 above. Can we dedup into one place and just call the helper twice instead?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I think it's possible, but might be more valuable to create a new trait that extends ExpectsInputTypes that by default allows all of the types defined by inputTypes to be non-foldable, but provides a mechanism to override each type and define whether its foldable/non-foldable. Feels like this could be utilized by other functions too - thoughts?

Copy link
Contributor

@dtenedor dtenedor May 17, 2023 •
edited

Choose a reason for hiding this comment

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

Could we define e.g.

  private def checkInputDataTypesForHllFunction(
      defaultCheck: TypeCheckResult, right: Expression): TypeCheckResult = {
    if (defaultCheck.isFailure) {
      defaultCheck
    } else if (!right.foldable) {
      DataTypeMismatch(
        errorSubClass = "NON_FOLDABLE_INPUT",
        messageParameters = Map(
          "inputName" -> "lgConfigK",
          "inputType" -> toSQLType(right.dataType),
          "inputExpr" -> toSQLExpr(right)
        )
      )
    } else {
      TypeCheckSuccess
    }
  }

val defaultCheck = super.checkInputDataTypes()
if (defaultCheck.isFailure) {
defaultCheck
} else if (!right.foldable) {
DataTypeMismatch(
errorSubClass = "NON_FOLDABLE_INPUT",
messageParameters = Map(
"inputName" -> "allowDifferentLgConfigK",
"inputType" -> toSQLType(right.dataType),
"inputExpr" -> toSQLExpr(right)
)
)
} else {
TypeCheckSuccess
}
}

// Overrides for TypedImperativeAggregate

override def prettyName: String = "hll_union_agg"
Expand Down
21 changes: 21 additions & 0 deletions ...st/src/main/scala/org/apache/spark/sql/catalyst/expressions/datasketchesExpressions.scala
View file
Open in desktop
Expand Up @@ -20,6 +20,9 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.datasketches.hll.{HllSketch, TgtHllType, Union}
import org.apache.datasketches.memory.Memory

import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.Cast.{toSQLExpr, toSQLType}
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
import org.apache.spark.sql.types.{AbstractDataType, BinaryType, BooleanType, DataType, LongType}

Expand Down Expand Up @@ -91,6 +94,24 @@ case class HllUnion(first: Expression, second: Expression, third: Expression)
newFirst: Expression, newSecond: Expression, newThird: Expression):
HllUnion = copy(first = newFirst, second = newSecond, third = newThird)

override def checkInputDataTypes(): TypeCheckResult = {
val defaultCheck = super.checkInputDataTypes()
if (defaultCheck.isFailure) {
defaultCheck
} else if (!third.foldable) {
DataTypeMismatch(
errorSubClass = "NON_FOLDABLE_INPUT",
messageParameters = Map(
"inputName" -> "allowDifferentLgConfigK",
"inputType" -> toSQLType(third.dataType),
"inputExpr" -> toSQLExpr(third)
)
)
} else {
TypeCheckSuccess
}
}

override def prettyName: String = "hll_union"

override def inputTypes: Seq[AbstractDataType] = Seq(BinaryType, BinaryType, BooleanType)
Expand Down
43 changes: 43 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
View file
Open in desktop
Expand Up @@ -1876,6 +1876,49 @@ class DataFrameAggregateSuite extends QueryTest
checkAnswer(res, Nil)
}
assert(error7.toString contains "UnsupportedOperationException")

// validate specific args require foldable = true
val error8 = intercept[AnalysisException] {
val res = sql(
"""with cte as (
|select * from values
| (1, 12)
| as tab(col, logk)
|)
|select hll_sketch_agg(col, logk) from cte
|""".stripMargin
)
checkAnswer(res, Nil)
}
assert(error8.toString contains "NON_FOLDABLE_INPUT")

val error9 = intercept[AnalysisException] {
val res = sql(
"""with cte as (
|select * from values
| (binary(1), true)
| as tab(col, allow_different_lgconfigk)
|)
|select hll_union_agg(col, allow_different_lgconfigk) from cte
|""".stripMargin
)
checkAnswer(res, Nil)
}
assert(error9.toString contains "NON_FOLDABLE_INPUT")

val error10 = intercept[AnalysisException] {
val res = sql(
"""with cte as (
|select * from values
| (binary(1), binary(1), true)
| as tab(col1, col2, allow_different_lgconfigk)
|)
|select hll_union(col1, col2, allow_different_lgconfigk) from cte
|""".stripMargin
)
checkAnswer(res, Nil)
}
assert(error10.toString contains "NON_FOLDABLE_INPUT")
}
}

Expand Down