Skip to content

[SPARK-24204][SQL] Verify a schema in Json/Orc/ParquetFileFormat #21389

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
maropu wants to merge 12 commits into from
Closed

[SPARK-24204][SQL] Verify a schema in Json/Orc/ParquetFileFormat #21389

Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
315c7e8
Fix
maropu May 8, 2018
731c873
Fix
maropu May 23, 2018
cf4cf2f
Fix
maropu May 23, 2018
3158e00
Fix
maropu May 28, 2018
0b25c4d
Merge with CSV.verifySChema
maropu May 28, 2018
927497d
Fix
maropu Jun 15, 2018
589479d
Add tests for CSV
maropu Jun 16, 2018
df1a67f
Keep backward-compatibility
maropu Jun 16, 2018
6303e49
Brush up code
maropu Jun 24, 2018
6301fb4
Spit test cases into pieces
maropu Jun 24, 2018
50e7b11
Review applied
maropu Jun 27, 2018
1cfc7b0
Fix
maropu Jun 27, 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
Prev Previous commit
Next Next commit
Review applied
  • Loading branch information
@maropu
maropu committed Jun 27, 2018
commit 50e7b113e976c669d981a9a93c10967162bec2c0
51 changes: 17 additions & 34 deletions sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,8 +269,8 @@ class FileBasedDataSourceSuite extends QueryTest with SharedSQLContext with Befo
withTempDir { dir =>
val tempDir = new File(dir, "files").getCanonicalPath

Seq("orc", "json").foreach { format =>
// write path
// write path
Seq("csv", "json", "parquet", "orc").foreach { format =>
var msg = intercept[AnalysisException] {
sql("select interval 1 days").write.format(format).mode("overwrite").save(tempDir)
}.getMessage
Expand All @@ -282,41 +282,11 @@ class FileBasedDataSourceSuite extends QueryTest with SharedSQLContext with Befo
}.getMessage
assert(msg.toLowerCase(Locale.ROOT)
.contains(s"$format data source does not support calendarinterval data type."))

// read path
// We expect the types below should be passed for backward-compatibility

// Interval type
var schema = StructType(StructField("a", CalendarIntervalType, true) :: Nil)
spark.range(1).write.format(format).mode("overwrite").save(tempDir)
spark.read.schema(schema).format(format).load(tempDir).collect()

// UDT having interval data
schema = StructType(StructField("a", new IntervalUDT(), true) :: Nil)
spark.range(1).write.format(format).mode("overwrite").save(tempDir)
spark.read.schema(schema).format(format).load(tempDir).collect()
}
}

withTempDir { dir =>
val tempDir = new File(dir, "files").getCanonicalPath

// read path
Seq("parquet", "csv").foreach { format =>
Copy link
Member

@gengliangwang gengliangwang Jun 27, 2018

Choose a reason for hiding this comment

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

Nit: we can put all the write path together to reduce duplicated code

Copy link
Member Author

@maropu maropu Jun 27, 2018

Choose a reason for hiding this comment

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

fixed

// write path
var msg = intercept[AnalysisException] {
sql("select interval 1 days").write.format(format).mode("overwrite").save(tempDir)
}.getMessage
assert(msg.contains("Cannot save interval data type into external storage."))

msg = intercept[UnsupportedOperationException] {
spark.udf.register("testType", () => new IntervalData())
sql("select testType()").write.format(format).mode("overwrite").save(tempDir)
}.getMessage
assert(msg.toLowerCase(Locale.ROOT)
.contains(s"$format data source does not support calendarinterval data type."))

// read path
msg = intercept[UnsupportedOperationException] {
var msg = intercept[UnsupportedOperationException] {
val schema = StructType(StructField("a", CalendarIntervalType, true) :: Nil)
spark.range(1).write.format(format).mode("overwrite").save(tempDir)
spark.read.schema(schema).format(format).load(tempDir).collect()
Expand All @@ -332,6 +302,19 @@ class FileBasedDataSourceSuite extends QueryTest with SharedSQLContext with Befo
assert(msg.toLowerCase(Locale.ROOT)
.contains(s"$format data source does not support calendarinterval data type."))
}

// We expect the types below should be passed for backward-compatibility
Seq("orc", "json").foreach { format =>
// Interval type
var schema = StructType(StructField("a", CalendarIntervalType, true) :: Nil)
spark.range(1).write.format(format).mode("overwrite").save(tempDir)
spark.read.schema(schema).format(format).load(tempDir).collect()

// UDT having interval data
schema = StructType(StructField("a", new IntervalUDT(), true) :: Nil)
spark.range(1).write.format(format).mode("overwrite").save(tempDir)
spark.read.schema(schema).format(format).load(tempDir).collect()
}
}
}

Expand Down