You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2021/09/17 13:55:17 UTC
[spark] branch master updated: [SPARK-36663][SQL] Support
number-only column names in ORC data sources
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new cd1b7e1 [SPARK-36663][SQL] Support number-only column names in ORC data sources
cd1b7e1 is described below
commit cd1b7e1777f75f9f576eed534b9e567288e0f4e1
Author: Kousuke Saruta <sa...@oss.nttdata.com>
AuthorDate: Fri Sep 17 21:54:36 2021 +0800
[SPARK-36663][SQL] Support number-only column names in ORC data sources
### What changes were proposed in this pull request?
This PR aims to support number-only column names in ORC data sources.
In the current master, with ORC datasource, we can write a DataFrame which contains such columns into ORC files.
```
spark.sql("SELECT 'a' as `1`").write.orc("/path/to/dest")
```
But reading the ORC files will fail.
```
spark.read.orc("/path/to/dest")
...
== SQL ==
struct<1:string>
-------^^^
at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:265)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:126)
at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseDataType(ParseDriver.scala:40)
at org.apache.spark.sql.execution.datasources.orc.OrcUtils$.org$apache$spark$sql$execution$datasources$orc$OrcUtils$$toCatalystSchema(OrcUtils.scala:91)
```
The cause of this is `OrcUtils.toCatalystSchema` fails to handle such column names.
In `OrcUtils.toCatalystSchema`, `CatalystSqlParser` is used to create a instance of `StructType` which represents a schema but `CatalystSqlParser.parseDataType` fails to parse if a column name (and nested field) consists of only numbers.
### Why are the changes needed?
For better usability.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New tests.
Closes #33915 from sarutak/fix-orc-schema-issue.
Authored-by: Kousuke Saruta <sa...@oss.nttdata.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/execution/datasources/orc/OrcUtils.scala | 39 ++++-
.../execution/datasources/orc/OrcSourceSuite.scala | 178 ++++++++++++++++++++-
2 files changed, 214 insertions(+), 3 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index 391ead8..ec57375 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -21,6 +21,7 @@ import java.nio.charset.StandardCharsets.UTF_8
import java.util.Locale
import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, Path}
@@ -85,10 +86,44 @@ object OrcUtils extends Logging {
}
private def toCatalystSchema(schema: TypeDescription): StructType = {
+ import TypeDescription.Category
+
+ def toCatalystType(orcType: TypeDescription): DataType = {
+ orcType.getCategory match {
+ case Category.STRUCT => toStructType(orcType)
+ case Category.LIST => toArrayType(orcType)
+ case Category.MAP => toMapType(orcType)
+ case _ => CatalystSqlParser.parseDataType(orcType.toString)
+ }
+ }
+
+ def toStructType(orcType: TypeDescription): StructType = {
+ val fieldNames = orcType.getFieldNames.asScala
+ val fieldTypes = orcType.getChildren.asScala
+ val fields = new ArrayBuffer[StructField]()
+ fieldNames.zip(fieldTypes).foreach {
+ case (fieldName, fieldType) =>
+ val catalystType = toCatalystType(fieldType)
+ fields += StructField(fieldName, catalystType)
+ }
+ StructType(fields.toSeq)
+ }
+
+ def toArrayType(orcType: TypeDescription): ArrayType = {
+ val elementType = orcType.getChildren.get(0)
+ ArrayType(toCatalystType(elementType))
+ }
+
+ def toMapType(orcType: TypeDescription): MapType = {
+ val Seq(keyType, valueType) = orcType.getChildren.asScala.toSeq
+ val catalystKeyType = toCatalystType(keyType)
+ val catalystValueType = toCatalystType(valueType)
+ MapType(catalystKeyType, catalystValueType)
+ }
+
// The Spark query engine has not completely supported CHAR/VARCHAR type yet, and here we
// replace the orc CHAR/VARCHAR with STRING type.
- CharVarcharUtils.replaceCharVarcharWithStringInSchema(
- CatalystSqlParser.parseDataType(schema.toString).asInstanceOf[StructType])
+ CharVarcharUtils.replaceCharVarcharWithStringInSchema(toStructType(schema))
}
def readSchema(sparkSession: SparkSession, files: Seq[FileStatus], options: Map[String, String])
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
index fb75ea1..c2f63cc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.{Row, SPARK_VERSION_METADATA_KEY}
import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, SchemaMergeUtils}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{LongType, StructField, StructType}
+import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
case class OrcData(intField: Int, stringField: String)
@@ -630,4 +630,180 @@ class OrcSourceSuite extends OrcSuite with SharedSparkSession {
checkAnswer(spark.sql("SELECT _col0, _col2.c1 FROM t1"), Seq(Row(1, "a")))
}
}
+
+ test("SPARK-36663: OrcUtils.toCatalystSchema should correctly handle " +
+ "a column name which consists of only numbers") {
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark.sql("SELECT 'a' as `1`, 'b' as `2`, 'c' as `3`").write.orc(path)
+ val df = spark.read.orc(path)
+ checkAnswer(df, Row("a", "b", "c"))
+ assert(df.schema.toArray ===
+ Array(
+ StructField("1", StringType),
+ StructField("2", StringType),
+ StructField("3", StringType)))
+ }
+
+ // test for struct in struct
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark.sql(
+ "SELECT 'a' as `10`, named_struct('20', 'b', '30', named_struct('40', 'c')) as `50`")
+ .write.orc(path)
+ val df = spark.read.orc(path)
+ checkAnswer(df, Row("a", Row("b", Row("c"))))
+ assert(df.schema.toArray === Array(
+ StructField("10", StringType),
+ StructField("50",
+ StructType(
+ StructField("20", StringType) ::
+ StructField("30",
+ StructType(
+ StructField("40", StringType) :: Nil)) :: Nil))))
+ }
+
+ // test for struct in array
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark.sql("SELECT array(array(named_struct('123', 'a'), named_struct('123', 'b'))) as `789`")
+ .write.orc(path)
+ val df = spark.read.orc(path)
+ checkAnswer(df, Row(Seq(Seq(Row("a"), Row("b")))))
+ assert(df.schema.toArray === Array(
+ StructField("789",
+ ArrayType(
+ ArrayType(
+ StructType(
+ StructField("123", StringType) :: Nil))))))
+ }
+
+ // test for struct in map
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark.sql(
+ """
+ |SELECT
+ | map(
+ | named_struct('123', 'a'),
+ | map(
+ | named_struct('456', 'b'),
+ | named_struct('789', 'c'))) as `012`""".stripMargin).write.orc(path)
+ val df = spark.read.orc(path)
+ checkAnswer(df, Row(Map(Row("a") -> Map(Row("b") -> Row("c")))))
+ assert(df.schema.toArray === Array(
+ StructField("012",
+ MapType(
+ StructType(
+ StructField("123", StringType) :: Nil),
+ MapType(
+ StructType(
+ StructField("456", StringType) :: Nil),
+ StructType(
+ StructField("789", StringType) :: Nil))))))
+ }
+
+ // test for deeply nested struct with complex types
+ withTempPath { dir =>
+ val path = dir.getAbsolutePath
+ spark.sql(
+ """
+ |SELECT
+ | named_struct('123',
+ | array(
+ | map(
+ | named_struct('456', 'a'),
+ | named_struct('789', 'b')))) as `1000`,
+ | named_struct('123',
+ | map(
+ | array(named_struct('456', 'a')),
+ | array(named_struct('789', 'b')))) as `2000`,
+ | array(
+ | named_struct('123',
+ | map(
+ | named_struct('456', 'a'),
+ | named_struct('789', 'b')))) as `3000`,
+ | array(
+ | map(
+ | named_struct('123', 'a'),
+ | named_struct('456', 'b'))) as `4000`,
+ | map(
+ | named_struct('123',
+ | array(
+ | named_struct('456', 'a'))),
+ | named_struct('789',
+ | array(
+ | named_struct('012', 'b')))) as `5000`,
+ | map(
+ | array(
+ | named_struct('123', 'a')),
+ | array(
+ | named_struct('456', 'b'))) as `6000`
+ """.stripMargin).write.orc(path)
+ val df = spark.read.orc(path)
+ checkAnswer(df, Row(
+ Row(Seq(Map(Row("a") -> Row("b")))),
+ Row(Map(Seq(Row("a")) -> Seq(Row("b")))),
+ Seq(Row(Map(Row("a") -> Row("b")))),
+ Seq(Map(Row("a") -> Row("b"))),
+ Map(Row(Seq(Row("a"))) -> Row(Seq(Row("b")))),
+ Map(Seq(Row("a")) -> Seq(Row("b")))))
+ assert(df.schema.toArray === Array(
+ StructField("1000",
+ StructType(
+ StructField("123",
+ ArrayType(
+ MapType(
+ StructType(
+ StructField("456", StringType) :: Nil),
+ StructType(
+ StructField("789", StringType) :: Nil)))) :: Nil)),
+ StructField("2000",
+ StructType(
+ StructField("123",
+ MapType(
+ ArrayType(
+ StructType(
+ StructField("456", StringType) :: Nil)),
+ ArrayType(
+ StructType(
+ StructField("789", StringType) :: Nil)))) :: Nil)),
+ StructField("3000",
+ ArrayType(
+ StructType(
+ StructField("123",
+ MapType(
+ StructType(
+ StructField("456", StringType) :: Nil),
+ StructType(
+ StructField("789", StringType) :: Nil))) :: Nil))),
+ StructField("4000",
+ ArrayType(
+ MapType(
+ StructType(
+ StructField("123", StringType) :: Nil),
+ StructType(
+ StructField("456", StringType) :: Nil)))),
+ StructField("5000",
+ MapType(
+ StructType(
+ StructField("123",
+ ArrayType(
+ StructType(
+ StructField("456", StringType) :: Nil))) :: Nil),
+ StructType(
+ StructField("789",
+ ArrayType(
+ StructType(
+ StructField("012", StringType) :: Nil))) :: Nil))),
+ StructField("6000",
+ MapType(
+ ArrayType(
+ StructType(
+ StructField("123", StringType) :: Nil)),
+ ArrayType(
+ StructType(
+ StructField("456", StringType) :: Nil))))))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org