You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/07/06 01:51:47 UTC
[GitHub] [spark] turboFei commented on a change in pull request #29000: [SPARK-27194][SPARK-29302][SQL] Fix commit collision in dynamic partition overwrite mode
turboFei commented on a change in pull request #29000:
URL: https://github.com/apache/spark/pull/29000#discussion_r449948718
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala
##########
@@ -164,4 +167,45 @@ class PartitionedWriteSuite extends QueryTest with SharedSparkSession {
assert(e.getMessage.contains("Found duplicate column(s) b, b: `b`;"))
}
}
+
+ test("SPARK-27194 SPARK-29302: Fix commit collision in dynamic partition overwrite mode") {
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString,
+ SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key ->
+ classOf[ConstantJobIdCommitProtocol].getName) {
+ withTempDir { d =>
+ withTable("t") {
+ sql(
+ s"""
+ | create table t(c1 int, p1 int) using parquet partitioned by (p1)
+ | location '${d.getAbsolutePath}'
+ """.stripMargin)
+
+ // File commit protocol is ConstantJobIdCommitProtocol, whose jobId is 'jobId'.
+ val stagingDir = new File(d, ".spark-staging-jobId")
+ stagingDir.mkdirs()
+ val conflictTaskFile = new File(stagingDir, "part-00000-jobId-c000.snappy.parquet")
+ conflictTaskFile.createNewFile()
Review comment:
Sorry, the ut is wrong, I have fix it.
val stagingDir = new File(d, ".spark-staging-jobId")
stagingDir.mkdirs()
val stagingPartDir = new File(stagingDir, "p1=2")
stagingPartDir.mkdirs()
val conflictTaskFile = new File(stagingPartDir, "part-00000-jobId.c000.snappy.parquet")
conflictTaskFile.createNewFile()
I also recreated another PR, https://github.com/apache/spark/pull/28989,
In this Pr, I define a Spark staging output committer to leverage OutputCommitCoordinator
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/sources/PartitionedWriteSuite.scala
##########
@@ -164,4 +167,45 @@ class PartitionedWriteSuite extends QueryTest with SharedSparkSession {
assert(e.getMessage.contains("Found duplicate column(s) b, b: `b`;"))
}
}
+
+ test("SPARK-27194 SPARK-29302: Fix commit collision in dynamic partition overwrite mode") {
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString,
+ SQLConf.FILE_COMMIT_PROTOCOL_CLASS.key ->
+ classOf[ConstantJobIdCommitProtocol].getName) {
+ withTempDir { d =>
+ withTable("t") {
+ sql(
+ s"""
+ | create table t(c1 int, p1 int) using parquet partitioned by (p1)
+ | location '${d.getAbsolutePath}'
+ """.stripMargin)
+
+ // File commit protocol is ConstantJobIdCommitProtocol, whose jobId is 'jobId'.
+ val stagingDir = new File(d, ".spark-staging-jobId")
+ stagingDir.mkdirs()
+ val conflictTaskFile = new File(stagingDir, "part-00000-jobId-c000.snappy.parquet")
+ conflictTaskFile.createNewFile()
Review comment:
Sorry, the ut is wrong, I have fixed it.
val stagingDir = new File(d, ".spark-staging-jobId")
stagingDir.mkdirs()
val stagingPartDir = new File(stagingDir, "p1=2")
stagingPartDir.mkdirs()
val conflictTaskFile = new File(stagingPartDir, "part-00000-jobId.c000.snappy.parquet")
conflictTaskFile.createNewFile()
I also recreated another PR, https://github.com/apache/spark/pull/28989,
In this Pr, I define a Spark staging output committer to leverage OutputCommitCoordinator
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org