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

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

Changes from 1 commit
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
Prev Previous commit
Next Next commit
Add failing test which demonstrates exception masking issue
  • Loading branch information
@JoshRosen
JoshRosen committed Jul 28, 2015
commit 8b162b6776f5cda19a0e80147e27a4fa53d1780f
33 changes: 31 additions & 2 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/ScriptTransformationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ package org.apache.spark.sql.hive.execution

import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.execution.{SparkPlan, SparkPlanTest}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
import org.apache.spark.sql.execution.{UnaryNode, SparkPlan, SparkPlanTest}
import org.apache.spark.sql.hive.test.TestHive
import org.apache.spark.sql.types.StringType

Expand Down Expand Up @@ -71,4 +73,31 @@ class ScriptTransformationSuite extends SparkPlanTest {
)(TestHive),
rowsDf.collect())
}

test("script transformation should not swallow errors from upstream pipelined operators") {
val rowsDf = Seq("a", "b", "c").map(Tuple1.apply).toDF("a")
val e = intercept[Exception] {
checkAnswer(
rowsDf,
(child: SparkPlan) => new ScriptTransformation(
input = Seq(rowsDf.col("a").expr),
script = "cat",
output = Seq(AttributeReference("a", StringType)()),
child = ExceptionInjectingOperator(child),
ioschema = serdeIOSchema
)(TestHive),
rowsDf.collect())
}
assert(e.getMessage === "intentional exception")
}
}

private case class ExceptionInjectingOperator(child: SparkPlan) extends UnaryNode {
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.

Note that it was important to test both with and without a SerDe, since an earlier version of my fix missed one of the cases.

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.

Also note that there's quite a bit of code redundancy in this test suite but I'm not super motivated to clean it up since I don't expect that it will ever need to change.

override protected def doExecute(): RDD[InternalRow] = {
child.execute().map { x =>
Thread.sleep(1000) // This sleep gives the external process time to start.
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 was necessary in order to trigger certain timing-related bugs. This shouldn't lead to test flakiness, though, but it might hypothetically lead to false-negatives if someone were to break this code again in the future. The idea here is to give the external child process time to start so that it has a chance to hang.

throw new Exception("intentional exception")
}
}
override def output: Seq[Attribute] = child.output
}