Skip to content

[SPARK-20254][SQL] Remove unnecessary data conversion for Dataset with primitive array #17568

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
kiszk wants to merge 19 commits into from
Closed

[SPARK-20254][SQL] Remove unnecessary data conversion for Dataset with primitive array #17568

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
1ece803
initial commit
kiszk Apr 7, 2017
c042ff2
addressed review comment
kiszk Apr 8, 2017
264a18b
add an optimizer to remove MapObjects
kiszk Apr 10, 2017
82a1f2b
add test suites to check deserializer in DeserializeToObjectExec
kiszk Apr 10, 2017
cc3c7bd
eliminate no-op AssertNotNull
kiszk Apr 14, 2017
7d43d27
remove unnecessary Mapobjects by checking MapObjects and lambdaVariable
kiszk Apr 14, 2017
3bcee3f
fix scala style error
kiszk Apr 14, 2017
1b24be0
use simpler matchings
kiszk Apr 14, 2017
5174b5e
use non-e2e test suite
kiszk Apr 14, 2017
e608c45
rebase with master
kiszk Apr 15, 2017
2ee4030
fix test failure
kiszk Apr 15, 2017
1d6ab36
address review comments
kiszk Apr 16, 2017
0fd8c25
rebase with master
kiszk Apr 16, 2017
791aad9
address review comments
kiszk Apr 16, 2017
f695e50
address review comments
kiszk Apr 17, 2017
ce6927d
remove MapObject with LambdaVariable if it is for primitive type (non…
kiszk Apr 18, 2017
a6bedee
remove unused import
kiszk Apr 18, 2017
c0dca2b
address review comment
kiszk Apr 18, 2017
8de6915
address review comment
kiszk Apr 18, 2017
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
Prev Previous commit
Next Next commit
address review comments
  • Loading branch information
@kiszk
kiszk committed Apr 17, 2017
commit f695e50e38bd329db3b75951dd7af52fea3b3dde
14 changes: 3 additions & 11 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/objects.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,20 +104,12 @@ object CombineTypedFilters extends Rule[LogicalPlan] {
* 1. Mapobject(e) where e is lambdavariable(), which means types for input output
Copy link
Member

@viirya viirya Apr 18, 2017

Choose a reason for hiding this comment

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

This comment is obscure. Can we improve it a bit?

For example, MapObject(e) is confusing. Shall we clearly say the lambdaFunction of MapObject?

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

thanks, done

* are primitive types
* 2. no custom collection class specified
* representation of data item. For example back to back map operations.
* representation of data item. For example back to back map operations.
Copy link
Member

@viirya viirya Apr 18, 2017

Choose a reason for hiding this comment

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

Can we rephrase this comment too? It looks weird.

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

thanks, deleted

*/
object EliminateMapObjects extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
Copy link
Contributor

@cloud-fan cloud-fan Apr 18, 2017

Choose a reason for hiding this comment

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

shall we call plan.transformAllExpressions?

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

Thanks, it works. done

case _ @ DeserializeToObject(Invoke(
MapObjects(_, _, _, Cast(LambdaVariable(_, _, dataType, _), castDataType, _),
inputData, None),
funcName, returnType: ObjectType, arguments, propagateNull, returnNullable),
outputObjAttr, child) if dataType == castDataType =>
DeserializeToObject(Invoke(
inputData, funcName, returnType, arguments, propagateNull, returnNullable),
outputObjAttr, child)
case _ @ DeserializeToObject(Invoke(
MapObjects(_, _, _, LambdaVariable(_, _, dataType, _), inputData, None),
case DeserializeToObject(Invoke(
MapObjects(_, _, _, _ : LambdaVariable, inputData, None),
Copy link
Contributor

@cloud-fan cloud-fan Apr 17, 2017

Choose a reason for hiding this comment

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

can we just replace MapObjects with its child? Seems the only reason you match the whole DeserializeToObject is to make sure the returnType is object type, but that's guaranteed if the collectionClass is None.

Copy link
Member Author

@kiszk kiszk Apr 17, 2017

Choose a reason for hiding this comment

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

To replace MapObjects with its child is a type LogicalPlan => Expression while this method requires LogicalPlan => LogicalPlan.
Is it fine to replace Invoke(MapObject(..., inputData, None)...) with Invoke(inputData, ...)?

Copy link
Member Author

@kiszk kiszk Apr 17, 2017

Choose a reason for hiding this comment

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

@cloud-fan, I misunderstood. Both Expression and Invoke are not LogicalPlan.
I think that we have to replace some of arguments in DeserializeToObject.
What do you think?

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

@cloud-fan Unfortunately, this change caused two test failures.
"checkAnswer should compare map correctly" and "SPARK-18717: code generation works for both scala.collection.Map and scala.collection.imutable.Map" in DatasetSuite
I will check what's happen very soon.

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

When we run the following code, the following exception occurs. This is because UnsafeArrayData.copy(), which is unsupported, is called.

    val ds = Seq((1, Map(2 -> 3))).toDS.map(t => t)
    ds.collect.toSeq

Copy link
Member Author

@kiszk kiszk Apr 18, 2017

Choose a reason for hiding this comment

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

The solution is to use the following matching:

  def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
     case MapObjects(_, _, _, LambdaVariable(_, _, _, false), inputData, None) => inputData
  }

Previous one avoided the case such as Object[] -> Integer[]. Thus, the generated code incorrectly called UnsafeArrayData.array().

funcName, returnType: ObjectType, arguments, propagateNull, returnNullable),
outputObjAttr, child) =>
DeserializeToObject(Invoke(
Expand Down
32 changes: 12 additions & 20 deletions ...yst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/EliminateMapObjectsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,16 +28,12 @@ import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.types._

class EliminateMapObjectsSuite extends PlanTest {
class Optimize(addSimplifyCast: Boolean) extends RuleExecutor[LogicalPlan] {
val batches = if (addSimplifyCast) {
object Optimize extends RuleExecutor[LogicalPlan] {
val batches = {
Batch("EliminateMapObjects", FixedPoint(50),
NullPropagation(conf),
SimplifyCasts,
EliminateMapObjects) :: Nil
} else {
Batch("EliminateMapObjects", FixedPoint(50),
NullPropagation(conf),
EliminateMapObjects) :: Nil
}
}

Expand All @@ -48,23 +44,19 @@ class EliminateMapObjectsSuite extends PlanTest {
val intObjType = ObjectType(classOf[Array[Int]])
val intInput = LocalRelation('a.array(ArrayType(IntegerType, false)))
val intQuery = intInput.deserialize[Array[Int]].analyze
Seq(true, false).foreach { addSimplifyCast =>
val intOptimized = new Optimize(addSimplifyCast).execute(intQuery)
val intExpected = DeserializeToObject(
Invoke(intInput.output(0), "toIntArray", intObjType, Nil, true, false),
AttributeReference("obj", intObjType, true)(), intInput)
comparePlans(intOptimized, intExpected)
}
val intOptimized = Optimize.execute(intQuery)
val intExpected = DeserializeToObject(
Invoke(intInput.output(0), "toIntArray", intObjType, Nil, true, false),
AttributeReference("obj", intObjType, true)(), intInput)
comparePlans(intOptimized, intExpected)

val doubleObjType = ObjectType(classOf[Array[Double]])
val doubleInput = LocalRelation('a.array(ArrayType(DoubleType, false)))
val doubleQuery = doubleInput.deserialize[Array[Double]].analyze
Seq(true, false).foreach { addSimplifyCast =>
val doubleOptimized = new Optimize(addSimplifyCast).execute(doubleQuery)
val doubleExpected = DeserializeToObject(
Invoke(doubleInput.output(0), "toDoubleArray", doubleObjType, Nil, true, false),
AttributeReference("obj", doubleObjType, true)(), doubleInput)
comparePlans(doubleOptimized, doubleExpected)
}
val doubleOptimized = Optimize.execute(doubleQuery)
val doubleExpected = DeserializeToObject(
Invoke(doubleInput.output(0), "toDoubleArray", doubleObjType, Nil, true, false),
AttributeReference("obj", doubleObjType, true)(), doubleInput)
comparePlans(doubleOptimized, doubleExpected)
}
}
2 changes: 0 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/DatasetPrimitiveSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@ package org.apache.spark.sql
import scala.collection.immutable.Queue
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.sql.catalyst.expressions.objects.Invoke
import org.apache.spark.sql.execution.DeserializeToObjectExec
import org.apache.spark.sql.test.SharedSQLContext

case class IntClass(value: Int)
Expand Down