-
Notifications
You must be signed in to change notification settings - Fork 29k
[SPARK-21603][SQL]The wholestage codegen will be much slower then that is closed when the function is too long #18810
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
Changes from 7 commits
ca9eff6
1b0ac5e
5582f00
7c185c6
7e84753
c4235dc
52da6b2
d0c753a
d3238e9
d44a2f8
ce544a5
08f5ddf
4fbe5f8
5c180ac
b83cd1c
6814047
8b32b54
32813b0
b879dbf
44ce894
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 |
|---|---|---|
|
|
@@ -355,6 +355,18 @@ class CodegenContext { | |
| */ | ||
| private val placeHolderToComments = new mutable.HashMap[String, String] | ||
|
|
||
| /** | ||
| * Returns if the length of codegen function is too long or not | ||
| * It will count the lines of every codegen function, if there is a function of length | ||
| * greater than spark.sql.codegen.MaxFunctionLength, it will return true. | ||
| */ | ||
| def existTooLongFunction(): Boolean = { | ||
|
Member
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. Adding a checking logics here, instead of returning
Contributor
Author
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. Ok, I have modified it, thanks.
Member
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. def existsTooLongGeneratedFunction: Boolean = {
classFunctions.values.exists { _.values.exists { code =>
val codeWithoutComments = CodeFormatter.stripExtraNewLinesAndComments(code)
codeWithoutComments.count(_ == '\n') > SQLConf.get.maxLinesPerFunction
}
}
}
Contributor
Author
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. Ok,updated,thanks. |
||
| classFunctions.exists { case (className, functions) => | ||
| functions.exists{ case (name, code) => | ||
| CodeFormatter.stripExtraNewLines(code).count(_ == '\n') > SQLConf.get.maxFunctionLength | ||
|
||
| } | ||
| } | ||
| } | ||
| /** | ||
|
Member
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. Add one more empty line
Contributor
Author
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. Ok, added, thanks |
||
| * Returns a term name that is unique within this instance of a `CodegenContext`. | ||
| */ | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -572,6 +572,13 @@ object SQLConf { | |
| "disable logging or -1 to apply no limit.") | ||
| .createWithDefault(1000) | ||
|
|
||
| val WHOLESTAGE_MAX_FUNCTION_LEN = buildConf("spark.sql.codegen.MaxFunctionLength") | ||
|
||
| .internal() | ||
| .doc("The maximum lines of a function that will be supported before" + | ||
| " deactivating whole-stage codegen.") | ||
|
||
| .intConf | ||
| .createWithDefault(1500) | ||
|
||
|
|
||
| val FILES_MAX_PARTITION_BYTES = buildConf("spark.sql.files.maxPartitionBytes") | ||
| .doc("The maximum number of bytes to pack into a single partition when reading files.") | ||
| .longConf | ||
|
|
@@ -1014,6 +1021,8 @@ class SQLConf extends Serializable with Logging { | |
|
|
||
| def loggingMaxLinesForCodegen: Int = getConf(CODEGEN_LOGGING_MAX_LINES) | ||
|
|
||
| def maxFunctionLength: Int = getConf(WHOLESTAGE_MAX_FUNCTION_LEN) | ||
|
|
||
| def tableRelationCacheSize: Int = | ||
| getConf(StaticSQLConf.FILESOURCE_TABLE_RELATION_CACHE_SIZE) | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -370,6 +370,15 @@ case class WholeStageCodegenExec(child: SparkPlan) extends UnaryExecNode with Co | |
|
|
||
| override def doExecute(): RDD[InternalRow] = { | ||
| val (ctx, cleanedSource) = doCodeGen() | ||
| val existLongFunction = ctx.existTooLongFunction | ||
| if (existLongFunction) { | ||
| logWarning(s"Found too long generated codes and JIT optimization might not work, " + | ||
| s"Whole-stage codegen disabled for this plan, " + | ||
| s"You can change the config spark.sql.codegen.MaxFunctionLength " + | ||
| s"to adjust the function length limit:\n " | ||
|
||
| + s"$treeString") | ||
|
Member
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. This could be very big. Please follow what did in #18658
Contributor
Author
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. @gatorsmile , thank you for review, the treeString not contains the code, it only contains the tree string of the Physical plan like below: |
||
| return child.execute() | ||
| } | ||
|
Member
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. We need to add a test in which we create a query with long generated function, and check if whole-stage codegen is disabled for it.
Contributor
Author
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. I think it can be tested by " max function length of wholestagecodegen" added in AggregateBenchmark.scala, thanks.
Member
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. AggregateBenchmark is more like a benchmark than a test. It won't run every time. We need a test to prevent regression brought by future change.
Contributor
Author
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. @viirya, it is hard to check if whole-stage codegen is disabled or not for me, would you like to give me some suggestion, thanks.
Member
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. We can check if there is a
Member
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. Ok. I'll take a look later. Thanks.
Member
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. There are multiple ways to verify it. One of the solution is using SQL metrics.
Member
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. I am fine about your proposed way, but needs to simplify it.
Member
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. @gatorsmile Do you mean
Member
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. Yes |
||
| // try to compile and fallback if it failed | ||
| try { | ||
| CodeGenerator.compile(cleanedSource) | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -301,6 +301,61 @@ class AggregateBenchmark extends BenchmarkBase { | |
| */ | ||
| } | ||
|
|
||
| ignore("max function length of wholestagecodegen") { | ||
| val N = 20 << 15 | ||
|
|
||
| val benchmark = new Benchmark("max function length of wholestagecodegen", N) | ||
| def f(): Unit = sparkSession.range(N) | ||
| .selectExpr( | ||
| "id", | ||
| "(id & 1023) as k1", | ||
| "cast(id & 1023 as double) as k2", | ||
| "cast(id & 1023 as int) as k3", | ||
| "case when id > 100 and id <= 200 then 1 else 0 end as v1", | ||
| "case when id > 200 and id <= 300 then 1 else 0 end as v2", | ||
| "case when id > 300 and id <= 400 then 1 else 0 end as v3", | ||
| "case when id > 400 and id <= 500 then 1 else 0 end as v4", | ||
| "case when id > 500 and id <= 600 then 1 else 0 end as v5", | ||
| "case when id > 600 and id <= 700 then 1 else 0 end as v6", | ||
| "case when id > 700 and id <= 800 then 1 else 0 end as v7", | ||
| "case when id > 800 and id <= 900 then 1 else 0 end as v8", | ||
| "case when id > 900 and id <= 1000 then 1 else 0 end as v9", | ||
| "case when id > 1000 and id <= 1100 then 1 else 0 end as v10", | ||
| "case when id > 1100 and id <= 1200 then 1 else 0 end as v11", | ||
| "case when id > 1200 and id <= 1300 then 1 else 0 end as v12", | ||
| "case when id > 1300 and id <= 1400 then 1 else 0 end as v13", | ||
| "case when id > 1400 and id <= 1500 then 1 else 0 end as v14", | ||
| "case when id > 1500 and id <= 1600 then 1 else 0 end as v15", | ||
| "case when id > 1600 and id <= 1700 then 1 else 0 end as v16", | ||
| "case when id > 1700 and id <= 1800 then 1 else 0 end as v17", | ||
| "case when id > 1800 and id <= 1900 then 1 else 0 end as v18") | ||
| .groupBy("k1", "k2", "k3") | ||
| .sum() | ||
| .collect() | ||
|
|
||
| benchmark.addCase(s"codegen = F") { iter => | ||
| sparkSession.conf.set("spark.sql.codegen.wholeStage", "false") | ||
| f() | ||
| } | ||
|
|
||
| benchmark.addCase(s"codegen = T") { iter => | ||
| sparkSession.conf.set("spark.sql.codegen.wholeStage", "true") | ||
| sparkSession.conf.set("spark.sql.codegen.MaxFunctionLength", "10000") | ||
|
||
| f() | ||
| } | ||
|
|
||
| benchmark.run() | ||
|
|
||
| /* | ||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_111-b14 on Windows 7 6.1 | ||
| Intel64 Family 6 Model 58 Stepping 9, GenuineIntel | ||
| max function length of wholestagecodegen: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ---------------------------------------------------------------------------------------------- | ||
| codegen = F 443 / 507 1.5 676.0 1.0X | ||
| codegen = T 3279 / 3283 0.2 5002.6 0.1X | ||
| */ | ||
| } | ||
|
|
||
|
|
||
| ignore("cube") { | ||
| val N = 5 << 20 | ||
|
|
||


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.
length is misleading. Here, it is just number of lines.
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.
Ok, I have modified it, thanks