-
Notifications
You must be signed in to change notification settings - Fork 28.1k
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.
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
[SPARK-19008][SQL] Improve performance of Dataset.map by eliminating boxing/unboxing #17172
Changes from 1 commit
d8b5f8d
a885907
65fa05a
1f9ece4
dfbce2a
8ee91af
89c896e
1fb2933
35ba2c6
4c88c03
200cec7
b25b191
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -17,6 +17,8 @@ | |
|
||
package org.apache.spark.sql.execution | ||
|
||
import com.sun.org.apache.xalan.internal.xsltc.compiler.util.VoidType | ||
|
||
import scala.language.existentials | ||
|
||
import org.apache.spark.api.java.function.MapFunction | ||
|
@@ -32,7 +34,7 @@ import org.apache.spark.sql.catalyst.plans.physical._ | |
import org.apache.spark.sql.Row | ||
import org.apache.spark.sql.catalyst.plans.logical.LogicalKeyedState | ||
import org.apache.spark.sql.execution.streaming.KeyedStateImpl | ||
import org.apache.spark.sql.types.{DataType, ObjectType, StructType} | ||
import org.apache.spark.sql.types._ | ||
|
||
|
||
/** | ||
|
@@ -217,9 +219,33 @@ case class MapElementsExec( | |
} | ||
|
||
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = { | ||
val inType = if (child.output.length == 1) child.output(0).dataType else NullType | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. These two are only needed inside the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good catch. I simplified their scope. |
||
val outType = outputObjAttr.dataType | ||
val (funcClass, methodName) = func match { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. let's put this thing in a util so that There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sure. Now, can generate a call to a specialized method for |
||
case m: MapFunction[_, _] => classOf[MapFunction[_, _]] -> "call" | ||
case _ => classOf[Any => Any] -> "apply" | ||
case _ => | ||
(inType, outType) match { | ||
// if a pair of an argument and return types is one of specific types | ||
// whose specialized method (apply$mc..$sp) is generated by scalac, | ||
// Catalyst generated a direct method call to the specialized method. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can you link to some official document or blogpost? |
||
case (IntegerType, IntegerType) => classOf[Int => Int] -> "apply$mcII$sp" | ||
case (IntegerType, LongType) => classOf[Int => Long] -> "apply$mcJI$sp" | ||
case (IntegerType, FloatType) => classOf[Int => Float] -> "apply$mcFI$sp" | ||
case (IntegerType, DoubleType) => classOf[Int => Double] -> "apply$mcDI$sp" | ||
case (LongType, IntegerType) => classOf[Long => Int] -> "apply$mcIJ$sp" | ||
case (LongType, LongType) => classOf[Long => Long] -> "apply$mcJJ$sp" | ||
case (LongType, FloatType) => classOf[Long => Float] -> "apply$mcFJ$sp" | ||
case (LongType, DoubleType) => classOf[Long => Double] -> "apply$mcDJ$sp" | ||
case (FloatType, IntegerType) => classOf[Float => Int] -> "apply$mcIF$sp" | ||
case (FloatType, LongType) => classOf[Float => Long] -> "apply$mcJF$sp" | ||
case (FloatType, FloatType) => classOf[Float => Float] -> "apply$mcFF$sp" | ||
case (FloatType, DoubleType) => classOf[Float => Double] -> "apply$mcDF$sp" | ||
case (DoubleType, IntegerType) => classOf[Double => Int] -> "apply$mcID$sp" | ||
case (DoubleType, LongType) => classOf[Double => Long] -> "apply$mcJD$sp" | ||
case (DoubleType, FloatType) => classOf[Double => Float] -> "apply$mcFD$sp" | ||
case (DoubleType, DoubleType) => classOf[Double => Double] -> "apply$mcDD$sp" | ||
case _ => classOf[Any => Any] -> "apply" | ||
} | ||
} | ||
val funcObj = Literal.create(func, ObjectType(funcClass)) | ||
val callFunc = Invoke(funcObj, methodName, outputObjAttr.dataType, child.output) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this a mistaken import? I don't see it used in the change and can't imagine we'd be invoking Xalan here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, that is why the first try was failed. It was unintentionally imported during my debugging.