Skip to content

[SPARK-21845] [SQL] Make codegen fallback of expressions configurable #19062

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
gatorsmile wants to merge 3 commits into from
Closed

[SPARK-21845] [SQL] Make codegen fallback of expressions configurable #19062

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9f3a1cd
fix.
gatorsmile Aug 26, 2017
26dcbd6
fix
gatorsmile Aug 29, 2017
4f39a52
fix
gatorsmile Aug 29, 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
6 changes: 3 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -551,9 +551,9 @@ object SQLConf {
.intConf
.createWithDefault(100)

val WHOLESTAGE_FALLBACK = buildConf("spark.sql.codegen.fallback")
val CODEGEN_FALLBACK = buildConf("spark.sql.codegen.fallback")
.internal()
.doc("When true, whole stage codegen could be temporary disabled for the part of query that" +
.doc("When true, (whole stage) codegen could be temporary disabled for the part of query that" +
" fail to compile generated code")
.booleanConf
.createWithDefault(true)
Expand Down Expand Up @@ -1041,7 +1041,7 @@ class SQLConf extends Serializable with Logging {

def wholeStageMaxNumFields: Int = getConf(WHOLESTAGE_MAX_NUM_FIELDS)

def wholeStageFallback: Boolean = getConf(WHOLESTAGE_FALLBACK)
def codegenFallback: Boolean = getConf(CODEGEN_FALLBACK)

def maxCaseBranchesForCodegen: Int = getConf(MAX_CASES_BRANCHES)

Expand Down
6 changes: 4 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
@transient
final val sqlContext = SparkSession.getActiveSession.map(_.sqlContext).orNull

// whether we should fallback when hitting compilation errors caused by codegen
private val codeGenFallBack = sqlContext == null || sqlContext.conf.codegenFallback
Copy link
Member

@kiszk kiszk Aug 27, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

Is it better to add !Utils.isTesting && or to drop !Utils.isTesting && from WholeStageCodegenExec to make these conditions consistent?

Copy link
Member Author

@gatorsmile gatorsmile Aug 27, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

Originally, I did it like what you said. However, if using that approach, I need to remove the test case. Then, I think we might just keep using the codegenFallback for controlling it.

Copy link
Member

@kiszk kiszk Aug 27, 2017

Choose a reason for hiding this comment

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

I see


protected def sparkContext = sqlContext.sparkContext

// sqlContext will be null when we are being deserialized on the slaves. In this instance
Expand Down Expand Up @@ -370,8 +373,7 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
try {
GeneratePredicate.generate(expression, inputSchema)
} catch {
case e @ (_: JaninoRuntimeException | _: CompileException)
if sqlContext == null || sqlContext.conf.wholeStageFallback =>
Copy link
Member Author

@gatorsmile gatorsmile Aug 26, 2017

Choose a reason for hiding this comment

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

Because sqlContext is always null when running it in executors, and thus, this always return true.

Copy link
Member

@maropu maropu Aug 28, 2017

Choose a reason for hiding this comment

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

Better to put this comment in https://github.com/apache/spark/pull/19062/files#diff-b9f96d092fb3fea76bcf75e016799678R57?

Copy link
Member

@sameeragarwal sameeragarwal Aug 29, 2017

Choose a reason for hiding this comment

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

Removing the null check here makes sense although this means existing spark jobs that were previously switching to the non-codegen version even with sqlContext.conf.wholeStageFallback = false will now start failing at runtime (perhaps, rightly so). Might be worth calling this out in the 2.3 release notes and/or the migration guide.

case _ @ (_: JaninoRuntimeException | _: CompileException) if codeGenFallBack =>
genInterpretedPredicate(expression, inputSchema)
}
}
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -382,7 +382,7 @@ case class WholeStageCodegenExec(child: SparkPlan) extends UnaryExecNode with Co
try {
CodeGenerator.compile(cleanedSource)
} catch {
case e: Exception if !Utils.isTesting && sqlContext.conf.wholeStageFallback =>
case _: Exception if !Utils.isTesting && sqlContext.conf.codegenFallback =>
// We should already saw the error message
logWarning(s"Whole-stage codegen disabled for this plan:\n $treeString")
return child.execute()
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,7 +422,7 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSQLContext {
v
}
withSQLConf(
(SQLConf.WHOLESTAGE_FALLBACK.key, codegenFallback.toString),
(SQLConf.CODEGEN_FALLBACK.key, codegenFallback.toString),
(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key, wholeStage.toString)) {
val df = spark.range(0, 4, 1, 4).withColumn("c", c)
val rows = df.collect()
Expand Down
12 changes: 11 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2011,7 +2011,17 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {

val filter = (0 until N)
.foldLeft(lit(false))((e, index) => e.or(df.col(df.columns(index)) =!= "string"))
df.filter(filter).count

withSQLConf(SQLConf.CODEGEN_FALLBACK.key -> "true") {
df.filter(filter).count()
}

withSQLConf(SQLConf.CODEGEN_FALLBACK.key -> "false") {
val e = intercept[SparkException] {
df.filter(filter).count()
}.getMessage
assert(e.contains("grows beyond 64 KB"))
}
}

test("SPARK-20897: cached self-join should not fail") {
Expand Down
2 changes: 2 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/test/SharedSQLContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import org.scalatest.concurrent.Eventually

import org.apache.spark.{DebugFilesystem, SparkConf}
import org.apache.spark.sql.{SparkSession, SQLContext}
import org.apache.spark.sql.internal.SQLConf

/**
* Helper trait for SQL test suites where all tests share a single [[TestSparkSession]].
Expand All @@ -34,6 +35,7 @@ trait SharedSQLContext extends SQLTestUtils with BeforeAndAfterEach with Eventua
new SparkConf()
.set("spark.hadoop.fs.file.impl", classOf[DebugFilesystem].getName)
.set("spark.unsafe.exceptionOnMemoryLeak", "true")
.set(SQLConf.CODEGEN_FALLBACK.key, "false")
}

/**
Expand Down
1 change: 1 addition & 0 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ object TestHive
"TestSQLContext",
new SparkConf()
.set("spark.sql.test", "")
.set(SQLConf.CODEGEN_FALLBACK.key, "false")
Copy link
Member Author

@gatorsmile gatorsmile Aug 26, 2017

Choose a reason for hiding this comment

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

Turn it false to ensure it does not hide the actual bugs of our expression codegen that causes compilation falure.

.set("spark.sql.hive.metastore.barrierPrefixes",
"org.apache.spark.sql.hive.execution.PairSerDe")
.set("spark.sql.warehouse.dir", TestHiveContext.makeWarehouseDir().toURI.getPath)
Expand Down