Skip to content

[SPARK-32914][SQL] Avoid constructing dataType multiple times #29790

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
wangyum wants to merge 5 commits into from
Closed

[SPARK-32914][SQL] Avoid constructing dataType multiple times #29790

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5755273
fix
wangyum Sep 22, 2020
649d3c2
Address comment
wangyum Sep 22, 2020
f2dc664
revert ScalarSubquery
wangyum Sep 22, 2020
6a8877d
Fix
wangyum Sep 28, 2020
37d0786
fix
wangyum Sep 29, 2020
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
10 changes: 6 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1048,6 +1048,11 @@ trait ComplexTypeMergingExpression extends Expression {
@transient
lazy val inputTypesForMerging: Seq[DataType] = children.map(_.dataType)

private lazy val internalDataType: DataType = {
Comment thread
cloud-fan marked this conversation as resolved.
Outdated

@cloud-fan cloud-fan Sep 29, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

can we put it right before the line of override def dataType: DataType?

dataTypeCheck
inputTypesForMerging.reduceLeft(TypeCoercion.findCommonTypeDifferentOnlyInNullFlags(_, _).get)
}

def dataTypeCheck: Unit = {
require(
inputTypesForMerging.nonEmpty,
Expand All @@ -1058,10 +1063,7 @@ trait ComplexTypeMergingExpression extends Expression {
s" The input types found are\n\t${inputTypesForMerging.mkString("\n\t")}")
}

override def dataType: DataType = {
dataTypeCheck
inputTypesForMerging.reduceLeft(TypeCoercion.findCommonTypeDifferentOnlyInNullFlags(_, _).get)
}
override def dataType: DataType = internalDataType
}

/**
Expand Down
4 changes: 3 additions & 1 deletion ...ain/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,10 +185,12 @@ case class ApproximatePercentile(
override def nullable: Boolean = true

// The result type is the same as the input type.
override def dataType: DataType = {
private lazy val internalDataType: DataType = {
if (returnPercentileArray) ArrayType(child.dataType, false) else child.dataType
}

override def dataType: DataType = internalDataType

override def prettyName: String =
getTagValue(FunctionRegistry.FUNC_ALIAS).getOrElse("percentile_approx")

Expand Down
13 changes: 10 additions & 3 deletions ...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,7 @@ case class MapEntries(child: Expression)

@transient private lazy val childDataType: MapType = child.dataType.asInstanceOf[MapType]

override def dataType: DataType = {
private lazy val internalDataType: DataType = {

@cloud-fan cloud-fan Sep 22, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

is it expensive? it just creates a few objects.

@wangyum wangyum Sep 22, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

This is to improve this case:
image

Benchmark code Before this PR(Milliseconds) After this PR(Milliseconds)
spark.range(100000000L).selectExpr("approx_count_distinct(map_entries(map(1, id)))").collect() 21787 15551

ArrayType(
StructType(
StructField("key", childDataType.keyType, false) ::
Expand All @@ -377,6 +377,8 @@ case class MapEntries(child: Expression)
false)
}

override def dataType: DataType = internalDataType

override protected def nullSafeEval(input: Any): Any = {
val childMap = input.asInstanceOf[MapData]
val keys = childMap.keyArray()
Expand Down Expand Up @@ -3498,13 +3500,16 @@ object ArrayUnion {
since = "2.4.0")
case class ArrayIntersect(left: Expression, right: Expression) extends ArrayBinaryLike
with ComplexTypeMergingExpression {
override def dataType: DataType = {

private lazy val internalDataType: DataType = {
dataTypeCheck
Comment thread
wangyum marked this conversation as resolved.
Comment thread
wangyum marked this conversation as resolved.
ArrayType(elementType,
left.dataType.asInstanceOf[ArrayType].containsNull &&
right.dataType.asInstanceOf[ArrayType].containsNull)
}

override def dataType: DataType = internalDataType

@transient lazy val evalIntersect: (ArrayData, ArrayData) => ArrayData = {
if (TypeUtils.typeWithProperEquals(elementType)) {
(array1, array2) =>
Expand Down Expand Up @@ -3741,11 +3746,13 @@ case class ArrayIntersect(left: Expression, right: Expression) extends ArrayBina
case class ArrayExcept(left: Expression, right: Expression) extends ArrayBinaryLike
with ComplexTypeMergingExpression {

override def dataType: DataType = {
private lazy val internalDataType: DataType = {
dataTypeCheck
Comment thread
wangyum marked this conversation as resolved.
left.dataType
}

override def dataType: DataType = internalDataType

@transient lazy val evalExcept: (ArrayData, ArrayData) => ArrayData = {
if (TypeUtils.typeWithProperEquals(elementType)) {
(array1, array2) =>
Expand Down
5 changes: 4 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/subquery.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,10 +225,13 @@ case class ScalarSubquery(
children: Seq[Expression] = Seq.empty,
exprId: ExprId = NamedExpression.newExprId)
extends SubqueryExpression(plan, children, exprId) with Unevaluable {
override def dataType: DataType = {

private lazy val internalDataType: DataType = {

@cloud-fan cloud-fan Sep 22, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

does this need to be a lazy val? seems a very cheap method.

@wangyum wangyum Sep 22, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I reverted this change because I did not find an expression to call this method many times.

assert(plan.schema.fields.nonEmpty, "Scalar subquery should have only one column")
plan.schema.fields.head.dataType
}

override def dataType: DataType = internalDataType
override def nullable: Boolean = true
override def withNewPlan(plan: LogicalPlan): ScalarSubquery = copy(plan = plan)
override def toString: String = s"scalar-subquery#${exprId.id} $conditionString"
Expand Down