Skip to content

[SPARK-42074][SQL] Enable KryoSerializer in TPCDSQueryBenchmark to enforce SQL class registration #39584

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
dongjoon-hyun wants to merge 1 commit into from
Closed

[SPARK-42074][SQL] Enable KryoSerializer in TPCDSQueryBenchmark to enforce SQL class registration #39584

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
32 changes: 32 additions & 0 deletions core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ import org.apache.spark.scheduler.{CompressedMapStatus, HighlyCompressedMapStatu
import org.apache.spark.storage._
import org.apache.spark.util.{BoundedPriorityQueue, ByteBufferInputStream, SerializableConfiguration, SerializableJobConf, Utils}
import org.apache.spark.util.collection.CompactBuffer
import org.apache.spark.util.io.ChunkedByteBuffer

/**
* A Spark serializer that uses the <a href="https://code.google.com/p/kryo/">
Expand Down Expand Up @@ -220,6 +221,7 @@ class KryoSerializer(conf: SparkConf)
kryo.register(Utils.classForName("scala.collection.immutable.$colon$colon"))
kryo.register(Utils.classForName("scala.collection.immutable.Map$EmptyMap$"))
kryo.register(Utils.classForName("scala.math.Ordering$Reverse"))
kryo.register(Utils.classForName("scala.reflect.ClassTag$GenericClassTag"))
kryo.register(classOf[ArrayBuffer[Any]])
kryo.register(classOf[Array[Array[Byte]]])

Expand Down Expand Up @@ -466,9 +468,11 @@ private[serializer] object KryoSerializer {
// Commonly used classes.
private val toRegister: Seq[Class[_]] = Seq(
ByteBuffer.allocate(1).getClass,
classOf[Array[ByteBuffer]],
classOf[StorageLevel],
classOf[CompressedMapStatus],
classOf[HighlyCompressedMapStatus],
classOf[ChunkedByteBuffer],
classOf[CompactBuffer[_]],
classOf[BlockManagerId],
classOf[Array[Boolean]],
Expand Down Expand Up @@ -503,9 +507,37 @@ private[serializer] object KryoSerializer {
// SQL / ML / MLlib classes once and then re-use that filtered list in newInstance() calls.
private lazy val loadableSparkClasses: Seq[Class[_]] = {
Seq(
"org.apache.spark.sql.catalyst.expressions.BoundReference",
"org.apache.spark.sql.catalyst.expressions.SortOrder",
"[Lorg.apache.spark.sql.catalyst.expressions.SortOrder;",
"org.apache.spark.sql.catalyst.InternalRow",
"org.apache.spark.sql.catalyst.InternalRow$",
"[Lorg.apache.spark.sql.catalyst.InternalRow;",
"org.apache.spark.sql.catalyst.expressions.UnsafeRow",
"org.apache.spark.sql.catalyst.expressions.UnsafeArrayData",
"org.apache.spark.sql.catalyst.expressions.UnsafeMapData",
"org.apache.spark.sql.catalyst.expressions.codegen.LazilyGeneratedOrdering",
"org.apache.spark.sql.catalyst.expressions.Ascending$",
"org.apache.spark.sql.catalyst.expressions.NullsFirst$",
"org.apache.spark.sql.catalyst.trees.Origin",
"org.apache.spark.sql.types.IntegerType",
"org.apache.spark.sql.types.IntegerType$",
"org.apache.spark.sql.types.LongType$",
"org.apache.spark.sql.types.Metadata",
"org.apache.spark.sql.types.StringType$",
"org.apache.spark.sql.types.StructField",
"[Lorg.apache.spark.sql.types.StructField;",
"org.apache.spark.sql.types.StructType",
"[Lorg.apache.spark.sql.types.StructType;",
"org.apache.spark.sql.types.DateType$",
"org.apache.spark.sql.types.DecimalType",
"org.apache.spark.sql.types.Decimal$DecimalAsIfIntegral$",
"org.apache.spark.sql.types.Decimal$DecimalIsFractional$",
"org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTaskResult",
"org.apache.spark.sql.execution.joins.EmptyHashedRelation$",
"org.apache.spark.sql.execution.joins.LongHashedRelation",
"org.apache.spark.sql.execution.joins.LongToUnsafeRowMap",
"org.apache.spark.sql.execution.joins.UnsafeHashedRelation",

"org.apache.spark.ml.attribute.Attribute",
"org.apache.spark.ml.attribute.AttributeGroup",
Expand Down
2 changes: 2 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/TPCDSQueryBenchmark.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@ object TPCDSQueryBenchmark extends SqlBasedBenchmark with Logging {
.set("spark.executor.memory", "3g")
.set("spark.sql.autoBroadcastJoinThreshold", (20 * 1024 * 1024).toString)
.set("spark.sql.crossJoin.enabled", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.set("spark.kryo.registrationRequired", "true")

@dongjoon-hyun dongjoon-hyun Jan 15, 2023

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 will cause failure in TPCDS SF=1 GitHub Action job if some PR missed the class registration.


SparkSession.builder.config(conf).getOrCreate()
}
Expand Down