Skip to content

[SPARK-9393] [SQL] Fix several error-handling bugs in ScriptTransform operator #7710

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

[SPARK-9393] [SQL] Fix several error-handling bugs in ScriptTransform operator #7710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
fa18d26
Add basic unit test for script transform with 'cat' command.
JoshRosen Jul 27, 2015
b43e4ec
Clean up nullability in ScriptTransformation
JoshRosen Jul 27, 2015
bd4c948
Skip launching of external command for empty partitions.
JoshRosen Jul 27, 2015
4ee36a2
Kill script transform subprocess when error occurs in input writer.
JoshRosen Jul 27, 2015
8b162b6
Add failing test which demonstrates exception masking issue
JoshRosen Jul 28, 2015
88278de
Split ScriptTransformation writer thread into own class.
JoshRosen Jul 28, 2015
b31258d
Rename iterator variables to disambiguate.
JoshRosen Jul 28, 2015
323bb2b
Fix error-swallowing bug
JoshRosen Jul 28, 2015
494cde0
Propagate TaskContext to writer thread
JoshRosen Jul 28, 2015
6a06a8c
Clean up handling of quotes in serde class name
JoshRosen Jul 28, 2015
983f200
Use unescapeSQLString instead of stripQuotes
JoshRosen Jul 28, 2015
16c44e2
Update some comments
JoshRosen Jul 28, 2015
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
27 changes: 16 additions & 11 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.test.TestSQLContext
import org.apache.spark.sql.{DataFrame, DataFrameHolder, Row}
import org.apache.spark.sql.{SQLContext, DataFrame, DataFrameHolder, Row}

import scala.language.implicitConversions
import scala.reflect.runtime.universe.TypeTag
Expand All @@ -33,11 +33,13 @@ import scala.util.control.NonFatal
*/
class SparkPlanTest extends SparkFunSuite {

protected def sqlContext: SQLContext = TestSQLContext
Copy link
Contributor Author

@JoshRosen JoshRosen Jul 28, 2015

Choose a reason for hiding this comment

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

The changes here in SparkPlanTest are to allow it to be used with TestHive without leading to the dreaded "multiple active SparkContexts in the same JVM" error.


/**
* Creates a DataFrame from a local Seq of Product.
*/
implicit def localSeqToDataFrameHolder[A <: Product : TypeTag](data: Seq[A]): DataFrameHolder = {
TestSQLContext.implicits.localSeqToDataFrameHolder(data)
sqlContext.implicits.localSeqToDataFrameHolder(data)
}

/**
Expand Down Expand Up @@ -98,7 +100,7 @@ class SparkPlanTest extends SparkFunSuite {
planFunction: Seq[SparkPlan] => SparkPlan,
expectedAnswer: Seq[Row],
sortAnswers: Boolean = true): Unit = {
SparkPlanTest.checkAnswer(input, planFunction, expectedAnswer, sortAnswers) match {
SparkPlanTest.checkAnswer(input, planFunction, expectedAnswer, sortAnswers, sqlContext) match {
Copy link
Contributor

@davies davies Jul 28, 2015

Choose a reason for hiding this comment

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

If sqlContext is already a member of SparkPlanTest, is it possible that we don't need to pass it to checkAnswer?

Copy link
Contributor Author

@JoshRosen JoshRosen Jul 28, 2015

Choose a reason for hiding this comment

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

This is a call on the SparkPlanTest companion object, which doesn't have that field.

Copy link
Contributor

@marmbrus marmbrus Jul 28, 2015

Choose a reason for hiding this comment

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

This is minor, and does not have to be addressed in this PR, but why did we make that an object? It seems like this just complicates function calls since now we can't access fields like the sqlContext.

Copy link
Contributor Author

@JoshRosen JoshRosen Jul 28, 2015

Choose a reason for hiding this comment

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

I think that the idea was to allow re-use of some of the SparkPlanTest functions without having to mix them into the test suite; I agree that this could maybe be simplified by making both SparkPlanTest and QueryTest into traits / mixins so that you can easily use both in the same test suite.

case Some(errorMessage) => fail(errorMessage)
case None =>
}
Expand All @@ -121,7 +123,8 @@ class SparkPlanTest extends SparkFunSuite {
planFunction: SparkPlan => SparkPlan,
expectedPlanFunction: SparkPlan => SparkPlan,
sortAnswers: Boolean = true): Unit = {
SparkPlanTest.checkAnswer(input, planFunction, expectedPlanFunction, sortAnswers) match {
SparkPlanTest.checkAnswer(
input, planFunction, expectedPlanFunction, sortAnswers, sqlContext) match {
case Some(errorMessage) => fail(errorMessage)
case None =>
}
Expand All @@ -147,13 +150,14 @@ object SparkPlanTest {
input: DataFrame,
planFunction: SparkPlan => SparkPlan,
expectedPlanFunction: SparkPlan => SparkPlan,
sortAnswers: Boolean): Option[String] = {
sortAnswers: Boolean,
sqlContext: SQLContext): Option[String] = {

val outputPlan = planFunction(input.queryExecution.sparkPlan)
val expectedOutputPlan = expectedPlanFunction(input.queryExecution.sparkPlan)

val expectedAnswer: Seq[Row] = try {
executePlan(expectedOutputPlan)
executePlan(expectedOutputPlan, sqlContext)
} catch {
case NonFatal(e) =>
val errorMessage =
Expand All @@ -168,7 +172,7 @@ object SparkPlanTest {
}

val actualAnswer: Seq[Row] = try {
executePlan(outputPlan)
executePlan(outputPlan, sqlContext)
} catch {
case NonFatal(e) =>
val errorMessage =
Expand Down Expand Up @@ -207,12 +211,13 @@ object SparkPlanTest {
input: Seq[DataFrame],
planFunction: Seq[SparkPlan] => SparkPlan,
expectedAnswer: Seq[Row],
sortAnswers: Boolean): Option[String] = {
sortAnswers: Boolean,
sqlContext: SQLContext): Option[String] = {

val outputPlan = planFunction(input.map(_.queryExecution.sparkPlan))

val sparkAnswer: Seq[Row] = try {
executePlan(outputPlan)
executePlan(outputPlan, sqlContext)
} catch {
case NonFatal(e) =>
val errorMessage =
Expand Down Expand Up @@ -275,10 +280,10 @@ object SparkPlanTest {
}
}

private def executePlan(outputPlan: SparkPlan): Seq[Row] = {
private def executePlan(outputPlan: SparkPlan, sqlContext: SQLContext): Seq[Row] = {
// A very simple resolver to make writing tests easier. In contrast to the real resolver
// this is always case sensitive and does not try to handle scoping or complex type resolution.
val resolvedPlan = TestSQLContext.prepareForExecution.execute(
val resolvedPlan = sqlContext.prepareForExecution.execute(
outputPlan transform {
case plan: SparkPlan =>
val inputMap = plan.children.flatMap(_.output).map(a => (a.name, a)).toMap
Expand Down
10 changes: 5 additions & 5 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -874,15 +874,15 @@ https://cwiki.apache.org/confluence/display/Hive/Enhanced+Aggregation%2C+Cube%2C
}

def matchSerDe(clause: Seq[ASTNode])
: (Seq[(String, String)], String, Seq[(String, String)]) = clause match {
: (Seq[(String, String)], Option[String], Seq[(String, String)]) = clause match {
case Token("TOK_SERDEPROPS", propsClause) :: Nil =>
val rowFormat = propsClause.map {
case Token(name, Token(value, Nil) :: Nil) => (name, value)
}
(rowFormat, "", Nil)
(rowFormat, None, Nil)
Copy link
Contributor Author

@JoshRosen JoshRosen Jul 28, 2015

Choose a reason for hiding this comment

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

I thought it was confusing to use an empty string to represent a missing value, hence this change.

Copy link
Contributor

@davies davies Jul 28, 2015

Choose a reason for hiding this comment

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

Could we use null here without changing the type?

Copy link
Contributor Author

@JoshRosen JoshRosen Jul 28, 2015

Choose a reason for hiding this comment

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

We could but I feel that's a bit less clear and more error-prone.

Copy link
Contributor

@marmbrus marmbrus Jul 28, 2015

Choose a reason for hiding this comment

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

+1 to this change.


case Token("TOK_SERDENAME", Token(serdeClass, Nil) :: Nil) :: Nil =>
(Nil, serdeClass, Nil)
(Nil, Some(BaseSemanticAnalyzer.unescapeSQLString(serdeClass)), Nil)

case Token("TOK_SERDENAME", Token(serdeClass, Nil) ::
Token("TOK_TABLEPROPERTIES",
Expand All @@ -891,9 +891,9 @@ https://cwiki.apache.org/confluence/display/Hive/Enhanced+Aggregation%2C+Cube%2C
case Token("TOK_TABLEPROPERTY", Token(name, Nil) :: Token(value, Nil) :: Nil) =>
(name, value)
}
(Nil, serdeClass, serdeProps)
(Nil, Some(BaseSemanticAnalyzer.unescapeSQLString(serdeClass)), serdeProps)

case Nil => (Nil, "", Nil)
case Nil => (Nil, None, Nil)
}

val (inRowFormat, inSerdeClass, inSerdeProps) = matchSerDe(inputSerdeClause)
Expand Down
Loading