You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yh...@apache.org on 2016/06/18 03:13:08 UTC
spark git commit: [SPARK-16033][SQL] insertInto() can't be used
together with partitionBy()
Repository: spark
Updated Branches:
refs/heads/master ebb9a3b6f -> 10b671447
[SPARK-16033][SQL] insertInto() can't be used together with partitionBy()
## What changes were proposed in this pull request?
When inserting into an existing partitioned table, partitioning columns should always be determined by catalog metadata of the existing table to be inserted. Extra `partitionBy()` calls don't make sense, and mess up existing data because newly inserted data may have wrong partitioning directory layout.
## How was this patch tested?
New test case added in `InsertIntoHiveTableSuite`.
Author: Cheng Lian <li...@databricks.com>
Closes #13747 from liancheng/spark-16033-insert-into-without-partition-by.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/10b67144
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/10b67144
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/10b67144
Branch: refs/heads/master
Commit: 10b671447bc04af250cbd8a7ea86f2769147a78a
Parents: ebb9a3b
Author: Cheng Lian <li...@databricks.com>
Authored: Fri Jun 17 20:13:04 2016 -0700
Committer: Yin Huai <yh...@databricks.com>
Committed: Fri Jun 17 20:13:04 2016 -0700
----------------------------------------------------------------------
.../org/apache/spark/sql/DataFrameWriter.scala | 12 +++++--
.../sql/hive/InsertIntoHiveTableSuite.scala | 37 ++++++++++++++++++++
2 files changed, 46 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/10b67144/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 60a9d1f..e6fc974 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -21,8 +21,6 @@ import java.util.Properties
import scala.collection.JavaConverters._
-import org.apache.hadoop.fs.Path
-
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, Project}
@@ -243,7 +241,15 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
private def insertInto(tableIdent: TableIdentifier): Unit = {
assertNotBucketed("insertInto")
- val partitions = normalizedParCols.map(_.map(col => col -> (Option.empty[String])).toMap)
+
+ if (partitioningColumns.isDefined) {
+ throw new AnalysisException(
+ "insertInto() can't be used together with partitionBy(). " +
+ "Partition columns are defined by the table into which is being inserted."
+ )
+ }
+
+ val partitions = normalizedParCols.map(_.map(col => col -> Option.empty[String]).toMap)
val overwrite = mode == SaveMode.Overwrite
// A partitioned relation's schema can be different from the input logicalPlan, since
http://git-wip-us.apache.org/repos/asf/spark/blob/10b67144/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
----------------------------------------------------------------------
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
index 3bf45ce..b890b4b 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala
@@ -346,6 +346,43 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
}
}
+ private def testPartitionedHiveSerDeTable(testName: String)(f: String => Unit): Unit = {
+ test(s"Hive SerDe table - $testName") {
+ val hiveTable = "hive_table"
+
+ withTable(hiveTable) {
+ withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
+ sql(s"CREATE TABLE $hiveTable (a INT) PARTITIONED BY (b INT, c INT) STORED AS TEXTFILE")
+ f(hiveTable)
+ }
+ }
+ }
+ }
+
+ private def testPartitionedDataSourceTable(testName: String)(f: String => Unit): Unit = {
+ test(s"Data source table - $testName") {
+ val dsTable = "ds_table"
+
+ withTable(dsTable) {
+ sql(s"CREATE TABLE $dsTable (a INT, b INT, c INT) USING PARQUET PARTITIONED BY (b, c)")
+ f(dsTable)
+ }
+ }
+ }
+
+ private def testPartitionedTable(testName: String)(f: String => Unit): Unit = {
+ testPartitionedHiveSerDeTable(testName)(f)
+ testPartitionedDataSourceTable(testName)(f)
+ }
+
+ testPartitionedTable("partitionBy() can't be used together with insertInto()") { tableName =>
+ val cause = intercept[AnalysisException] {
+ Seq((1, 2, 3)).toDF("a", "b", "c").write.partitionBy("b", "c").insertInto(tableName)
+ }
+
+ assert(cause.getMessage.contains("insertInto() can't be used together with partitionBy()."))
+ }
+
test("InsertIntoTable#resolved should include dynamic partitions") {
withSQLConf(("hive.exec.dynamic.partition.mode", "nonstrict")) {
sql("CREATE TABLE partitioned (id bigint, data string) PARTITIONED BY (part string)")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org