From e52499ea9c32326b399b50bf0e3f26278da3feb2 Mon Sep 17 00:00:00 2001
From: windpiger <songjun@outlook.com>
Date: Mon, 6 Mar 2017 22:36:43 -0800
Subject: [PATCH] [SPARK-19832][SQL] DynamicPartitionWriteTask get
partitionPath should escape the partition name
## What changes were proposed in this pull request?
Currently in DynamicPartitionWriteTask, when we get the paritionPath of a parition, we just escape the partition value, not escape the partition name.
this will cause some problems for some special partition name situation, for example :
1) if the partition name contains '%' etc, there will be two partition path created in the filesytem, one is for escaped path like '/path/a%25b=1', another is for unescaped path like '/path/a%b=1'.
and the data inserted stored in unescaped path, while the show partitions table will return 'a%25b=1' which the partition name is escaped. So here it is not consist. And I think the data should be stored in the escaped path in filesystem, which Hive2.0.0 also have the same action.
2) if the partition name contains ':', there will throw exception that new Path("/path","a:b"), this is illegal which has a colon in the relative path.
```
java.lang.IllegalArgumentException: java.net.URISyntaxException: Relative path in absolute URI: a:b
at org.apache.hadoop.fs.Path.initialize(Path.java:205)
at org.apache.hadoop.fs.Path.<init>(Path.java:171)
at org.apache.hadoop.fs.Path.<init>(Path.java:88)
... 48 elided
Caused by: java.net.URISyntaxException: Relative path in absolute URI: a:b
at java.net.URI.checkPath(URI.java:1823)
at java.net.URI.<init>(URI.java:745)
at org.apache.hadoop.fs.Path.initialize(Path.java:202)
... 50 more
```
## How was this patch tested?
unit test added
Author: windpiger <songjun@outlook.com>
Closes #17173 from windpiger/fixDatasourceSpecialCharPartitionName.
---
.../datasources/FileFormatWriter.scala | 2 +-
.../sql/execution/command/DDLSuite.scala | 23 ++++++++++++
.../sql/hive/execution/HiveDDLSuite.scala | 35 ++++++++++++++++++-
3 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index 950e5ca0d6..30a09a9ad3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -341,7 +341,7 @@ object FileFormatWriter extends Logging {
Seq(Cast(c, StringType, Option(desc.timeZoneId))),
Seq(StringType))
val str = If(IsNull(c), Literal(ExternalCatalogUtils.DEFAULT_PARTITION_NAME), escaped)
- val partitionName = Literal(c.name + "=") :: str :: Nil
+ val partitionName = Literal(ExternalCatalogUtils.escapePathName(c.name) + "=") :: str :: Nil
if (i == 0) partitionName else Literal(Path.SEPARATOR) :: partitionName
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index 6ffa58bcd9..b2199fdf90 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -1995,6 +1995,29 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
}
}
+ Seq("a b", "a:b", "a%b", "a,b").foreach { specialChars =>
+ test(s"data source table:partition column name containing $specialChars") {
+ withTable("t") {
+ withTempDir { dir =>
+ spark.sql(
+ s"""
+ |CREATE TABLE t(a string, `$specialChars` string)
+ |USING parquet
+ |PARTITIONED BY(`$specialChars`)
+ |LOCATION '$dir'
+ """.stripMargin)
+
+ assert(dir.listFiles().isEmpty)
+ spark.sql(s"INSERT INTO TABLE t PARTITION(`$specialChars`=2) SELECT 1")
+ val partEscaped = s"${ExternalCatalogUtils.escapePathName(specialChars)}=2"
+ val partFile = new File(dir, partEscaped)
+ assert(partFile.listFiles().length >= 1)
+ checkAnswer(spark.table("t"), Row("1", "2") :: Nil)
+ }
+ }
+ }
+ }
+
Seq("a b", "a:b", "a%b").foreach { specialChars =>
test(s"location uri contains $specialChars for datasource table") {
withTable("t", "t1") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index e956c9abae..df2c1cee94 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -27,7 +27,7 @@ import org.scalatest.BeforeAndAfterEach
import org.apache.spark.SparkException
import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, TableAlreadyExistsException}
-import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTableType, CatalogUtils}
+import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, CatalogTable, CatalogTableType, CatalogUtils, ExternalCatalogUtils}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.hive.HiveExternalCatalog
@@ -1690,6 +1690,39 @@ class HiveDDLSuite
}
}
+ Seq("parquet", "hive").foreach { datasource =>
+ Seq("a b", "a:b", "a%b", "a,b").foreach { specialChars =>
+ test(s"partition column name of $datasource table containing $specialChars") {
+ withTable("t") {
+ withTempDir { dir =>
+ spark.sql(
+ s"""
+ |CREATE TABLE t(a string, `$specialChars` string)
+ |USING $datasource
+ |PARTITIONED BY(`$specialChars`)
+ |LOCATION '$dir'
+ """.stripMargin)
+
+ assert(dir.listFiles().isEmpty)
+ spark.sql(s"INSERT INTO TABLE t PARTITION(`$specialChars`=2) SELECT 1")
+ val partEscaped = s"${ExternalCatalogUtils.escapePathName(specialChars)}=2"
+ val partFile = new File(dir, partEscaped)
+ assert(partFile.listFiles().length >= 1)
+ checkAnswer(spark.table("t"), Row("1", "2") :: Nil)
+
+ withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
+ spark.sql(s"INSERT INTO TABLE t PARTITION(`$specialChars`) SELECT 3, 4")
+ val partEscaped1 = s"${ExternalCatalogUtils.escapePathName(specialChars)}=4"
+ val partFile1 = new File(dir, partEscaped1)
+ assert(partFile1.listFiles().length >= 1)
+ checkAnswer(spark.table("t"), Row("1", "2") :: Row("3", "4") :: Nil)
+ }
+ }
+ }
+ }
+ }
+ }
+
Seq("a b", "a:b", "a%b").foreach { specialChars =>
test(s"datasource table: location uri contains $specialChars") {
withTable("t", "t1") {
--
GitLab