Skip to content

[SPARK-24860][SQL] Support setting of partitionOverWriteMode in output options for writing DataFrame #21818

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

[SPARK-24860][SQL] Support setting of partitionOverWriteMode in output options for writing DataFrame #21818

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7dd2eab
support setting of partitionOverWriteMode in output options for writi…
koertkuipers Jul 19, 2018
7d0752b
fix jira issue for dynamic partition overwrite per source test
koertkuipers Jul 19, 2018
605a325
split up long logic for enableDynamicOverwrite
koertkuipers Jul 20, 2018
a4ebf9d
simplify test for SPARK-24860
koertkuipers Jul 20, 2018
03c0926
add precendence to doc for spark.sql.sources.partitionOverwriteMode
koertkuipers Jul 25, 2018
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
4 changes: 3 additions & 1 deletion 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 @@ -1335,7 +1335,9 @@ object SQLConf {
"overwriting. In dynamic mode, Spark doesn't delete partitions ahead, and only overwrite " +
"those partitions that have data written into it at runtime. By default we use static " +
"mode to keep the same behavior of Spark prior to 2.3. Note that this config doesn't " +
"affect Hive serde tables, as they are always overwritten with dynamic mode.")
"affect Hive serde tables, as they are always overwritten with dynamic mode. This can " +
"also be set as an output option for a data source using key partitionOverwriteMode, " +

@gatorsmile gatorsmile Jul 25, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Also need to explain the precedence between the option and sqlconf.

"e.g. dataframe.write.option(\"partitionOverwriteMode\", \"dynamic\").save(path)")
.stringConf
.transform(_.toUpperCase(Locale.ROOT))
.checkValues(PartitionOverwriteMode.values.map(_.toString))
Expand Down
9 changes: 7 additions & 2 deletions .../scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogTable, CatalogT
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
Expand Down Expand Up @@ -91,8 +92,12 @@ case class InsertIntoHadoopFsRelationCommand(

val pathExists = fs.exists(qualifiedOutputPath)

val enableDynamicOverwrite =
sparkSession.sessionState.conf.partitionOverwriteMode == PartitionOverwriteMode.DYNAMIC
val parameters = CaseInsensitiveMap(options)

val enableDynamicOverwrite = parameters.get("partitionOverwriteMode")
.map(mode => PartitionOverwriteMode.withName(mode.toUpperCase))
.getOrElse(sparkSession.sessionState.conf.partitionOverwriteMode) ==
PartitionOverwriteMode.DYNAMIC

@cloud-fan cloud-fan Jul 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: this is too long

val partitionOverwriteMode = parameters.get("partitionOverwriteMode")....
val enableDynamicOverwrite = partitionOverwriteMode == PartitionOverwriteMode.DYNAMIC

// This config only makes sense when we are overwriting a partitioned dataset with dynamic
// partition columns.
val dynamicPartitionOverwrite = enableDynamicOverwrite && mode == SaveMode.Overwrite &&
Expand Down
19 changes: 19 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,6 +486,25 @@ class InsertSuite extends DataSourceTest with SharedSQLContext {
}
}
}
test("SPARK-24860: dynamic partition overwrite specified per source without catalog table") {

@cloud-fan cloud-fan Jul 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: add a blank file before this new test case

withTempPath { path =>
Seq((1, 1, 1)).toDF("i", "part1", "part2")

@cloud-fan cloud-fan Jul 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

can we simplify the test? ideally we only need one partition column, and write some initial data to the table. then do an overwrite with partitionOverwriteMode=dynamic, and another overwrite with partitionOverwriteMode=static.

.write.partitionBy("part1", "part2")
.option("partitionOverwriteMode", "dynamic")
.parquet(path.getAbsolutePath)
checkAnswer(spark.read.parquet(path.getAbsolutePath), Row(1, 1, 1))

Seq((2, 1, 1)).toDF("i", "part1", "part2")
.write.partitionBy("part1", "part2").mode("overwrite")
.option("partitionOverwriteMode", "dynamic").parquet(path.getAbsolutePath)
checkAnswer(spark.read.parquet(path.getAbsolutePath), Row(2, 1, 1))

Seq((2, 2, 2)).toDF("i", "part1", "part2")
.write.partitionBy("part1", "part2").mode("overwrite")
.option("partitionOverwriteMode", "dynamic").parquet(path.getAbsolutePath)
checkAnswer(spark.read.parquet(path.getAbsolutePath), Row(2, 1, 1) :: Row(2, 2, 2) :: Nil)
}
}

test("SPARK-20236: dynamic partition overwrite") {
withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.DYNAMIC.toString) {
Expand Down