You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2020/05/05 06:16:24 UTC
[spark] branch branch-3.0 updated: [SPARK-31467][SQL][TEST]
Refactor the sql tests to prevent TableAlreadyExistsException
This is an automated email from the ASF dual-hosted git repository.
yamamuro pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new edc0cb9 [SPARK-31467][SQL][TEST] Refactor the sql tests to prevent TableAlreadyExistsException
edc0cb9 is described below
commit edc0cb97477bfc37a69ee1fa4e94c52bf84fc5b9
Author: turbofei <fw...@ebay.com>
AuthorDate: Tue May 5 15:14:33 2020 +0900
[SPARK-31467][SQL][TEST] Refactor the sql tests to prevent TableAlreadyExistsException
### What changes were proposed in this pull request?
If we add UT in hive/SQLQuerySuite or other sql test suites and use table named `test`.
We may meet TableAlreadyExistsException.
```
org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException: Table or view 'test' already exists in database 'default'
```
The reason is that, there is some tests that does not clean up the tables/views.
In this PR, I add `withTempViews` for these tests.
### Why are the changes needed?
To fix the TableAlreadyExistsException issue when adding an UT, which uses table named `test` or others, in some sql test suites, such as hive/SQLQuerySuite.
### Does this PR introduce any user-facing change?
No.
### How was this patch tested?
Existed UT.
Closes #28239 from turboFei/SPARK-31467.
Authored-by: turbofei <fw...@ebay.com>
Signed-off-by: Takeshi Yamamuro <ya...@apache.org>
(cherry picked from commit 8d1f7d2a4ae08135df56cc32462b29703be46c42)
Signed-off-by: Takeshi Yamamuro <ya...@apache.org>
---
.../org/apache/spark/sql/CachedTableSuite.scala | 238 ++--
.../apache/spark/sql/ColumnExpressionSuite.scala | 40 +-
.../org/apache/spark/sql/DataFrameSuite.scala | 87 +-
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 122 +-
.../scala/org/apache/spark/sql/JoinSuite.scala | 270 ++--
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 72 +-
.../scala/org/apache/spark/sql/SQLQuerySuite.scala | 895 +++++++------
.../spark/sql/ScalaReflectionRelationSuite.scala | 96 +-
.../scala/org/apache/spark/sql/SubquerySuite.scala | 40 +-
.../apache/spark/sql/UserDefinedTypeSuite.scala | 14 +-
.../sql/execution/SQLWindowFunctionSuite.scala | 433 ++++---
.../columnar/InMemoryColumnarQuerySuite.scala | 46 +-
.../sql/execution/datasources/json/JsonSuite.scala | 1367 ++++++++++----------
.../sql/execution/joins/BroadcastJoinSuite.scala | 37 +-
.../sql/execution/metric/SQLMetricsSuite.scala | 45 +-
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 14 +-
.../org/apache/spark/sql/sources/InsertSuite.scala | 16 +-
.../apache/spark/sql/sources/SaveLoadSuite.scala | 36 +-
.../apache/spark/sql/streaming/StreamSuite.scala | 16 +-
.../sql/streaming/continuous/ContinuousSuite.scala | 22 +-
.../sql/hive/execution/AggregationQuerySuite.scala | 78 +-
.../spark/sql/hive/execution/HiveDDLSuite.scala | 54 +-
.../spark/sql/hive/execution/HiveQuerySuite.scala | 72 +-
.../spark/sql/hive/execution/SQLQuerySuite.scala | 535 ++++----
24 files changed, 2445 insertions(+), 2200 deletions(-)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index 9baaaa9..20f2a7f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -141,27 +141,31 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
}
test("uncaching temp table") {
- testData.select("key").createOrReplaceTempView("tempTable1")
- testData.select("key").createOrReplaceTempView("tempTable2")
- spark.catalog.cacheTable("tempTable1")
+ withTempView("tempTable1", "tempTable2") {
+ testData.select("key").createOrReplaceTempView("tempTable1")
+ testData.select("key").createOrReplaceTempView("tempTable2")
+ spark.catalog.cacheTable("tempTable1")
- assertCached(sql("SELECT COUNT(*) FROM tempTable1"))
- assertCached(sql("SELECT COUNT(*) FROM tempTable2"))
+ assertCached(sql("SELECT COUNT(*) FROM tempTable1"))
+ assertCached(sql("SELECT COUNT(*) FROM tempTable2"))
- // Is this valid?
- uncacheTable("tempTable2")
+ // Is this valid?
+ uncacheTable("tempTable2")
- // Should this be cached?
- assertCached(sql("SELECT COUNT(*) FROM tempTable1"), 0)
+ // Should this be cached?
+ assertCached(sql("SELECT COUNT(*) FROM tempTable1"), 0)
+ }
}
test("too big for memory") {
- val data = "*" * 1000
- sparkContext.parallelize(1 to 200000, 1).map(_ => BigData(data)).toDF()
- .createOrReplaceTempView("bigData")
- spark.table("bigData").persist(StorageLevel.MEMORY_AND_DISK)
- assert(spark.table("bigData").count() === 200000L)
- spark.table("bigData").unpersist(blocking = true)
+ withTempView("bigData") {
+ val data = "*" * 1000
+ sparkContext.parallelize(1 to 200000, 1).map(_ => BigData(data)).toDF()
+ .createOrReplaceTempView("bigData")
+ spark.table("bigData").persist(StorageLevel.MEMORY_AND_DISK)
+ assert(spark.table("bigData").count() === 200000L)
+ spark.table("bigData").unpersist(blocking = true)
+ }
}
test("calling .cache() should use in-memory columnar caching") {
@@ -225,12 +229,14 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
}
test("SELECT star from cached table") {
- sql("SELECT * FROM testData").createOrReplaceTempView("selectStar")
- spark.catalog.cacheTable("selectStar")
- checkAnswer(
- sql("SELECT * FROM selectStar WHERE key = 1"),
- Seq(Row(1, "1")))
- uncacheTable("selectStar")
+ withTempView("selectStar") {
+ sql("SELECT * FROM testData").createOrReplaceTempView("selectStar")
+ spark.catalog.cacheTable("selectStar")
+ checkAnswer(
+ sql("SELECT * FROM selectStar WHERE key = 1"),
+ Seq(Row(1, "1")))
+ uncacheTable("selectStar")
+ }
}
test("Self-join cached") {
@@ -375,102 +381,112 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
}
test("Drops temporary table") {
- testData.select("key").createOrReplaceTempView("t1")
- spark.table("t1")
- spark.catalog.dropTempView("t1")
- intercept[AnalysisException](spark.table("t1"))
+ withTempView("t1") {
+ testData.select("key").createOrReplaceTempView("t1")
+ spark.table("t1")
+ spark.catalog.dropTempView("t1")
+ intercept[AnalysisException](spark.table("t1"))
+ }
}
test("Drops cached temporary table") {
- testData.select("key").createOrReplaceTempView("t1")
- testData.select("key").createOrReplaceTempView("t2")
- spark.catalog.cacheTable("t1")
+ withTempView("t1", "t2") {
+ testData.select("key").createOrReplaceTempView("t1")
+ testData.select("key").createOrReplaceTempView("t2")
+ spark.catalog.cacheTable("t1")
- assert(spark.catalog.isCached("t1"))
- assert(spark.catalog.isCached("t2"))
+ assert(spark.catalog.isCached("t1"))
+ assert(spark.catalog.isCached("t2"))
- spark.catalog.dropTempView("t1")
- intercept[AnalysisException](spark.table("t1"))
- assert(!spark.catalog.isCached("t2"))
+ spark.catalog.dropTempView("t1")
+ intercept[AnalysisException](spark.table("t1"))
+ assert(!spark.catalog.isCached("t2"))
+ }
}
test("Clear all cache") {
- sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
- sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
- spark.catalog.cacheTable("t1")
- spark.catalog.cacheTable("t2")
- spark.catalog.clearCache()
- assert(spark.sharedState.cacheManager.isEmpty)
-
- sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
- sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
- spark.catalog.cacheTable("t1")
- spark.catalog.cacheTable("t2")
- sql("Clear CACHE")
- assert(spark.sharedState.cacheManager.isEmpty)
+ withTempView("t1", "t2") {
+ sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
+ sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
+ spark.catalog.cacheTable("t1")
+ spark.catalog.cacheTable("t2")
+ spark.catalog.clearCache()
+ assert(spark.sharedState.cacheManager.isEmpty)
+
+ sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
+ sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
+ spark.catalog.cacheTable("t1")
+ spark.catalog.cacheTable("t2")
+ sql("Clear CACHE")
+ assert(spark.sharedState.cacheManager.isEmpty)
+ }
}
test("Ensure accumulators to be cleared after GC when uncacheTable") {
- sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
- sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
+ withTempView("t1", "t2") {
+ sql("SELECT key FROM testData LIMIT 10").createOrReplaceTempView("t1")
+ sql("SELECT key FROM testData LIMIT 5").createOrReplaceTempView("t2")
- spark.catalog.cacheTable("t1")
- spark.catalog.cacheTable("t2")
+ spark.catalog.cacheTable("t1")
+ spark.catalog.cacheTable("t2")
- sql("SELECT * FROM t1").count()
- sql("SELECT * FROM t2").count()
- sql("SELECT * FROM t1").count()
- sql("SELECT * FROM t2").count()
+ sql("SELECT * FROM t1").count()
+ sql("SELECT * FROM t2").count()
+ sql("SELECT * FROM t1").count()
+ sql("SELECT * FROM t2").count()
+
+ val toBeCleanedAccIds = new HashSet[Long]
+
+ val accId1 = spark.table("t1").queryExecution.withCachedData.collect {
+ case i: InMemoryRelation => i.cacheBuilder.sizeInBytesStats.id
+ }.head
+ toBeCleanedAccIds += accId1
+
+ val accId2 = spark.table("t1").queryExecution.withCachedData.collect {
+ case i: InMemoryRelation => i.cacheBuilder.sizeInBytesStats.id
+ }.head
+ toBeCleanedAccIds += accId2
+
+ val cleanerListener = new CleanerListener {
+ def rddCleaned(rddId: Int): Unit = {}
+ def shuffleCleaned(shuffleId: Int): Unit = {}
+ def broadcastCleaned(broadcastId: Long): Unit = {}
+ def accumCleaned(accId: Long): Unit = {
+ toBeCleanedAccIds.synchronized { toBeCleanedAccIds -= accId }
+ }
+ def checkpointCleaned(rddId: Long): Unit = {}
+ }
+ spark.sparkContext.cleaner.get.attachListener(cleanerListener)
- val toBeCleanedAccIds = new HashSet[Long]
+ uncacheTable("t1")
+ uncacheTable("t2")
- val accId1 = spark.table("t1").queryExecution.withCachedData.collect {
- case i: InMemoryRelation => i.cacheBuilder.sizeInBytesStats.id
- }.head
- toBeCleanedAccIds += accId1
+ System.gc()
- val accId2 = spark.table("t1").queryExecution.withCachedData.collect {
- case i: InMemoryRelation => i.cacheBuilder.sizeInBytesStats.id
- }.head
- toBeCleanedAccIds += accId2
-
- val cleanerListener = new CleanerListener {
- def rddCleaned(rddId: Int): Unit = {}
- def shuffleCleaned(shuffleId: Int): Unit = {}
- def broadcastCleaned(broadcastId: Long): Unit = {}
- def accumCleaned(accId: Long): Unit = {
- toBeCleanedAccIds.synchronized { toBeCleanedAccIds -= accId }
+ eventually(timeout(10.seconds)) {
+ assert(toBeCleanedAccIds.synchronized { toBeCleanedAccIds.isEmpty },
+ "batchStats accumulators should be cleared after GC when uncacheTable")
}
- def checkpointCleaned(rddId: Long): Unit = {}
- }
- spark.sparkContext.cleaner.get.attachListener(cleanerListener)
-
- uncacheTable("t1")
- uncacheTable("t2")
-
- System.gc()
- eventually(timeout(10.seconds)) {
- assert(toBeCleanedAccIds.synchronized { toBeCleanedAccIds.isEmpty },
- "batchStats accumulators should be cleared after GC when uncacheTable")
+ assert(AccumulatorContext.get(accId1).isEmpty)
+ assert(AccumulatorContext.get(accId2).isEmpty)
}
-
- assert(AccumulatorContext.get(accId1).isEmpty)
- assert(AccumulatorContext.get(accId2).isEmpty)
}
test("SPARK-10327 Cache Table is not working while subquery has alias in its project list") {
- sparkContext.parallelize((1, 1) :: (2, 2) :: Nil)
- .toDF("key", "value").selectExpr("key", "value", "key+1").createOrReplaceTempView("abc")
- spark.catalog.cacheTable("abc")
-
- val sparkPlan = sql(
- """select a.key, b.key, c.key from
- |abc a join abc b on a.key=b.key
- |join abc c on a.key=c.key""".stripMargin).queryExecution.sparkPlan
-
- assert(sparkPlan.collect { case e: InMemoryTableScanExec => e }.size === 3)
- assert(sparkPlan.collect { case e: RDDScanExec => e }.size === 0)
+ withTempView("abc") {
+ sparkContext.parallelize((1, 1) :: (2, 2) :: Nil)
+ .toDF("key", "value").selectExpr("key", "value", "key+1").createOrReplaceTempView("abc")
+ spark.catalog.cacheTable("abc")
+
+ val sparkPlan = sql(
+ """select a.key, b.key, c.key from
+ |abc a join abc b on a.key=b.key
+ |join abc c on a.key=c.key""".stripMargin).queryExecution.sparkPlan
+
+ assert(sparkPlan.collect { case e: InMemoryTableScanExec => e }.size === 3)
+ assert(sparkPlan.collect { case e: RDDScanExec => e }.size === 0)
+ }
}
/**
@@ -628,26 +644,30 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
}
test("SPARK-15870 DataFrame can't execute after uncacheTable") {
- val selectStar = sql("SELECT * FROM testData WHERE key = 1")
- selectStar.createOrReplaceTempView("selectStar")
+ withTempView("selectStar") {
+ val selectStar = sql("SELECT * FROM testData WHERE key = 1")
+ selectStar.createOrReplaceTempView("selectStar")
- spark.catalog.cacheTable("selectStar")
- checkAnswer(
- selectStar,
- Seq(Row(1, "1")))
+ spark.catalog.cacheTable("selectStar")
+ checkAnswer(
+ selectStar,
+ Seq(Row(1, "1")))
- uncacheTable("selectStar")
- checkAnswer(
- selectStar,
- Seq(Row(1, "1")))
+ uncacheTable("selectStar")
+ checkAnswer(
+ selectStar,
+ Seq(Row(1, "1")))
+ }
}
test("SPARK-15915 Logical plans should use canonicalized plan when override sameResult") {
- val localRelation = Seq(1, 2, 3).toDF()
- localRelation.createOrReplaceTempView("localRelation")
+ withTempView("localRelation") {
+ val localRelation = Seq(1, 2, 3).toDF()
+ localRelation.createOrReplaceTempView("localRelation")
- spark.catalog.cacheTable("localRelation")
- assert(getNumInMemoryRelations(localRelation) == 1)
+ spark.catalog.cacheTable("localRelation")
+ assert(getNumInMemoryRelations(localRelation) == 1)
+ }
}
test("SPARK-19093 Caching in side subquery") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
index 4bf19532..fa06484 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ColumnExpressionSuite.scala
@@ -260,26 +260,28 @@ class ColumnExpressionSuite extends QueryTest with SharedSparkSession {
}
test("nanvl") {
- val testData = spark.createDataFrame(sparkContext.parallelize(
- Row(null, 3.0, Double.NaN, Double.PositiveInfinity, 1.0f, 4) :: Nil),
- StructType(Seq(StructField("a", DoubleType), StructField("b", DoubleType),
- StructField("c", DoubleType), StructField("d", DoubleType),
- StructField("e", FloatType), StructField("f", IntegerType))))
+ withTempView("t") {
+ val testData = spark.createDataFrame(sparkContext.parallelize(
+ Row(null, 3.0, Double.NaN, Double.PositiveInfinity, 1.0f, 4) :: Nil),
+ StructType(Seq(StructField("a", DoubleType), StructField("b", DoubleType),
+ StructField("c", DoubleType), StructField("d", DoubleType),
+ StructField("e", FloatType), StructField("f", IntegerType))))
- checkAnswer(
- testData.select(
- nanvl($"a", lit(5)), nanvl($"b", lit(10)), nanvl(lit(10), $"b"),
- nanvl($"c", lit(null).cast(DoubleType)), nanvl($"d", lit(10)),
- nanvl($"b", $"e"), nanvl($"e", $"f")),
- Row(null, 3.0, 10.0, null, Double.PositiveInfinity, 3.0, 1.0)
- )
- testData.createOrReplaceTempView("t")
- checkAnswer(
- sql(
- "select nanvl(a, 5), nanvl(b, 10), nanvl(10, b), nanvl(c, null), nanvl(d, 10), " +
- " nanvl(b, e), nanvl(e, f) from t"),
- Row(null, 3.0, 10.0, null, Double.PositiveInfinity, 3.0, 1.0)
- )
+ checkAnswer(
+ testData.select(
+ nanvl($"a", lit(5)), nanvl($"b", lit(10)), nanvl(lit(10), $"b"),
+ nanvl($"c", lit(null).cast(DoubleType)), nanvl($"d", lit(10)),
+ nanvl($"b", $"e"), nanvl($"e", $"f")),
+ Row(null, 3.0, 10.0, null, Double.PositiveInfinity, 3.0, 1.0)
+ )
+ testData.createOrReplaceTempView("t")
+ checkAnswer(
+ sql(
+ "select nanvl(a, 5), nanvl(b, 10), nanvl(10, b), nanvl(c, null), nanvl(d, 10), " +
+ " nanvl(b, e), nanvl(e, f) from t"),
+ Row(null, 3.0, 10.0, null, Double.PositiveInfinity, 3.0, 1.0)
+ )
+ }
}
test("===") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
index f20e684..4e91a7c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
@@ -1384,47 +1384,48 @@ class DataFrameSuite extends QueryTest
test("SPARK-6941: Better error message for inserting into RDD-based Table") {
withTempDir { dir =>
+ withTempView("parquet_base", "json_base", "rdd_base", "indirect_ds", "one_row") {
+ val tempParquetFile = new File(dir, "tmp_parquet")
+ val tempJsonFile = new File(dir, "tmp_json")
+
+ val df = Seq(Tuple1(1)).toDF()
+ val insertion = Seq(Tuple1(2)).toDF("col")
+
+ // pass case: parquet table (HadoopFsRelation)
+ df.write.mode(SaveMode.Overwrite).parquet(tempParquetFile.getCanonicalPath)
+ val pdf = spark.read.parquet(tempParquetFile.getCanonicalPath)
+ pdf.createOrReplaceTempView("parquet_base")
+
+ insertion.write.insertInto("parquet_base")
+
+ // pass case: json table (InsertableRelation)
+ df.write.mode(SaveMode.Overwrite).json(tempJsonFile.getCanonicalPath)
+ val jdf = spark.read.json(tempJsonFile.getCanonicalPath)
+ jdf.createOrReplaceTempView("json_base")
+ insertion.write.mode(SaveMode.Overwrite).insertInto("json_base")
+
+ // error cases: insert into an RDD
+ df.createOrReplaceTempView("rdd_base")
+ val e1 = intercept[AnalysisException] {
+ insertion.write.insertInto("rdd_base")
+ }
+ assert(e1.getMessage.contains("Inserting into an RDD-based table is not allowed."))
- val tempParquetFile = new File(dir, "tmp_parquet")
- val tempJsonFile = new File(dir, "tmp_json")
-
- val df = Seq(Tuple1(1)).toDF()
- val insertion = Seq(Tuple1(2)).toDF("col")
-
- // pass case: parquet table (HadoopFsRelation)
- df.write.mode(SaveMode.Overwrite).parquet(tempParquetFile.getCanonicalPath)
- val pdf = spark.read.parquet(tempParquetFile.getCanonicalPath)
- pdf.createOrReplaceTempView("parquet_base")
-
- insertion.write.insertInto("parquet_base")
-
- // pass case: json table (InsertableRelation)
- df.write.mode(SaveMode.Overwrite).json(tempJsonFile.getCanonicalPath)
- val jdf = spark.read.json(tempJsonFile.getCanonicalPath)
- jdf.createOrReplaceTempView("json_base")
- insertion.write.mode(SaveMode.Overwrite).insertInto("json_base")
-
- // error cases: insert into an RDD
- df.createOrReplaceTempView("rdd_base")
- val e1 = intercept[AnalysisException] {
- insertion.write.insertInto("rdd_base")
- }
- assert(e1.getMessage.contains("Inserting into an RDD-based table is not allowed."))
-
- // error case: insert into a logical plan that is not a LeafNode
- val indirectDS = pdf.select("_1").filter($"_1" > 5)
- indirectDS.createOrReplaceTempView("indirect_ds")
- val e2 = intercept[AnalysisException] {
- insertion.write.insertInto("indirect_ds")
- }
- assert(e2.getMessage.contains("Inserting into an RDD-based table is not allowed."))
+ // error case: insert into a logical plan that is not a LeafNode
+ val indirectDS = pdf.select("_1").filter($"_1" > 5)
+ indirectDS.createOrReplaceTempView("indirect_ds")
+ val e2 = intercept[AnalysisException] {
+ insertion.write.insertInto("indirect_ds")
+ }
+ assert(e2.getMessage.contains("Inserting into an RDD-based table is not allowed."))
- // error case: insert into an OneRowRelation
- Dataset.ofRows(spark, OneRowRelation()).createOrReplaceTempView("one_row")
- val e3 = intercept[AnalysisException] {
- insertion.write.insertInto("one_row")
+ // error case: insert into an OneRowRelation
+ Dataset.ofRows(spark, OneRowRelation()).createOrReplaceTempView("one_row")
+ val e3 = intercept[AnalysisException] {
+ insertion.write.insertInto("one_row")
+ }
+ assert(e3.getMessage.contains("Inserting into an RDD-based table is not allowed."))
}
- assert(e3.getMessage.contains("Inserting into an RDD-based table is not allowed."))
}
}
@@ -1796,13 +1797,17 @@ class DataFrameSuite extends QueryTest
val df = Seq("foo", "bar").map(Tuple1.apply).toDF("col")
// invalid table names
Seq("11111", "t~", "#$@sum", "table!#").foreach { name =>
- val m = intercept[AnalysisException](df.createOrReplaceTempView(name)).getMessage
- assert(m.contains(s"Invalid view name: $name"))
+ withTempView(name) {
+ val m = intercept[AnalysisException](df.createOrReplaceTempView(name)).getMessage
+ assert(m.contains(s"Invalid view name: $name"))
+ }
}
// valid table names
Seq("table1", "`11111`", "`t~`", "`#$@sum`", "`table!#`").foreach { name =>
- df.createOrReplaceTempView(name)
+ withTempView(name) {
+ df.createOrReplaceTempView(name)
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
index f72ccaa..09ce43e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
@@ -61,26 +61,28 @@ class DataFrameWindowFunctionsSuite extends QueryTest
}
test("rank functions in unspecific window") {
- val df = Seq((1, "1"), (2, "2"), (1, "2"), (2, "2")).toDF("key", "value")
- df.createOrReplaceTempView("window_table")
- checkAnswer(
- df.select(
- $"key",
- max("key").over(Window.partitionBy("value").orderBy("key")),
- min("key").over(Window.partitionBy("value").orderBy("key")),
- mean("key").over(Window.partitionBy("value").orderBy("key")),
- count("key").over(Window.partitionBy("value").orderBy("key")),
- sum("key").over(Window.partitionBy("value").orderBy("key")),
- ntile(2).over(Window.partitionBy("value").orderBy("key")),
- row_number().over(Window.partitionBy("value").orderBy("key")),
- dense_rank().over(Window.partitionBy("value").orderBy("key")),
- rank().over(Window.partitionBy("value").orderBy("key")),
- cume_dist().over(Window.partitionBy("value").orderBy("key")),
- percent_rank().over(Window.partitionBy("value").orderBy("key"))),
- Row(1, 1, 1, 1.0d, 1, 1, 1, 1, 1, 1, 1.0d, 0.0d) ::
- Row(1, 1, 1, 1.0d, 1, 1, 1, 1, 1, 1, 1.0d / 3.0d, 0.0d) ::
- Row(2, 2, 1, 5.0d / 3.0d, 3, 5, 1, 2, 2, 2, 1.0d, 0.5d) ::
- Row(2, 2, 1, 5.0d / 3.0d, 3, 5, 2, 3, 2, 2, 1.0d, 0.5d) :: Nil)
+ withTempView("window_table") {
+ val df = Seq((1, "1"), (2, "2"), (1, "2"), (2, "2")).toDF("key", "value")
+ df.createOrReplaceTempView("window_table")
+ checkAnswer(
+ df.select(
+ $"key",
+ max("key").over(Window.partitionBy("value").orderBy("key")),
+ min("key").over(Window.partitionBy("value").orderBy("key")),
+ mean("key").over(Window.partitionBy("value").orderBy("key")),
+ count("key").over(Window.partitionBy("value").orderBy("key")),
+ sum("key").over(Window.partitionBy("value").orderBy("key")),
+ ntile(2).over(Window.partitionBy("value").orderBy("key")),
+ row_number().over(Window.partitionBy("value").orderBy("key")),
+ dense_rank().over(Window.partitionBy("value").orderBy("key")),
+ rank().over(Window.partitionBy("value").orderBy("key")),
+ cume_dist().over(Window.partitionBy("value").orderBy("key")),
+ percent_rank().over(Window.partitionBy("value").orderBy("key"))),
+ Row(1, 1, 1, 1.0d, 1, 1, 1, 1, 1, 1, 1.0d, 0.0d) ::
+ Row(1, 1, 1, 1.0d, 1, 1, 1, 1, 1, 1, 1.0d / 3.0d, 0.0d) ::
+ Row(2, 2, 1, 5.0d / 3.0d, 3, 5, 1, 2, 2, 2, 1.0d, 0.5d) ::
+ Row(2, 2, 1, 5.0d / 3.0d, 3, 5, 2, 3, 2, 2, 1.0d, 0.5d) :: Nil)
+ }
}
test("window function should fail if order by clause is not specified") {
@@ -348,15 +350,17 @@ class DataFrameWindowFunctionsSuite extends QueryTest
}
test("SPARK-16195 empty over spec") {
- val df = Seq(("a", 1), ("a", 1), ("a", 2), ("b", 2)).
- toDF("key", "value")
- df.createOrReplaceTempView("window_table")
- checkAnswer(
- df.select($"key", $"value", sum($"value").over(), avg($"value").over()),
- Seq(Row("a", 1, 6, 1.5), Row("a", 1, 6, 1.5), Row("a", 2, 6, 1.5), Row("b", 2, 6, 1.5)))
- checkAnswer(
- sql("select key, value, sum(value) over(), avg(value) over() from window_table"),
- Seq(Row("a", 1, 6, 1.5), Row("a", 1, 6, 1.5), Row("a", 2, 6, 1.5), Row("b", 2, 6, 1.5)))
+ withTempView("window_table") {
+ val df = Seq(("a", 1), ("a", 1), ("a", 2), ("b", 2)).
+ toDF("key", "value")
+ df.createOrReplaceTempView("window_table")
+ checkAnswer(
+ df.select($"key", $"value", sum($"value").over(), avg($"value").over()),
+ Seq(Row("a", 1, 6, 1.5), Row("a", 1, 6, 1.5), Row("a", 2, 6, 1.5), Row("b", 2, 6, 1.5)))
+ checkAnswer(
+ sql("select key, value, sum(value) over(), avg(value) over() from window_table"),
+ Seq(Row("a", 1, 6, 1.5), Row("a", 1, 6, 1.5), Row("a", 2, 6, 1.5), Row("b", 2, 6, 1.5)))
+ }
}
test("window function with udaf") {
@@ -548,37 +552,41 @@ class DataFrameWindowFunctionsSuite extends QueryTest
}
test("aggregation and rows between with unbounded + predicate pushdown") {
- val df = Seq((1, "1"), (2, "2"), (2, "3"), (1, "3"), (3, "2"), (4, "3")).toDF("key", "value")
- df.createOrReplaceTempView("window_table")
- val selectList = Seq($"key", $"value",
- last("key").over(
- Window.partitionBy($"value").orderBy($"key").rowsBetween(0, Long.MaxValue)),
- last("key").over(
- Window.partitionBy($"value").orderBy($"key").rowsBetween(Long.MinValue, 0)),
- last("key").over(Window.partitionBy($"value").orderBy($"key").rowsBetween(-1, 1)))
-
- checkAnswer(
- df.select(selectList: _*).where($"value" < "3"),
- Seq(Row(1, "1", 1, 1, 1), Row(2, "2", 3, 2, 3), Row(3, "2", 3, 3, 3)))
+ withTempView("window_table") {
+ val df = Seq((1, "1"), (2, "2"), (2, "3"), (1, "3"), (3, "2"), (4, "3")).toDF("key", "value")
+ df.createOrReplaceTempView("window_table")
+ val selectList = Seq($"key", $"value",
+ last("key").over(
+ Window.partitionBy($"value").orderBy($"key").rowsBetween(0, Long.MaxValue)),
+ last("key").over(
+ Window.partitionBy($"value").orderBy($"key").rowsBetween(Long.MinValue, 0)),
+ last("key").over(Window.partitionBy($"value").orderBy($"key").rowsBetween(-1, 1)))
+
+ checkAnswer(
+ df.select(selectList: _*).where($"value" < "3"),
+ Seq(Row(1, "1", 1, 1, 1), Row(2, "2", 3, 2, 3), Row(3, "2", 3, 3, 3)))
+ }
}
test("aggregation and range between with unbounded + predicate pushdown") {
- val df = Seq((5, "1"), (5, "2"), (4, "2"), (6, "2"), (3, "1"), (2, "2")).toDF("key", "value")
- df.createOrReplaceTempView("window_table")
- val selectList = Seq($"key", $"value",
- last("value").over(
- Window.partitionBy($"value").orderBy($"key").rangeBetween(-2, -1)).equalTo("2")
- .as("last_v"),
- avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(Long.MinValue, 1))
- .as("avg_key1"),
- avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(0, Long.MaxValue))
- .as("avg_key2"),
- avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(-1, 1))
- .as("avg_key3"))
-
- checkAnswer(
- df.select(selectList: _*).where($"value" < 2),
- Seq(Row(3, "1", null, 3.0, 4.0, 3.0), Row(5, "1", false, 4.0, 5.0, 5.0)))
+ withTempView("window_table") {
+ val df = Seq((5, "1"), (5, "2"), (4, "2"), (6, "2"), (3, "1"), (2, "2")).toDF("key", "value")
+ df.createOrReplaceTempView("window_table")
+ val selectList = Seq($"key", $"value",
+ last("value").over(
+ Window.partitionBy($"value").orderBy($"key").rangeBetween(-2, -1)).equalTo("2")
+ .as("last_v"),
+ avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(Long.MinValue, 1))
+ .as("avg_key1"),
+ avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(0, Long.MaxValue))
+ .as("avg_key2"),
+ avg("key").over(Window.partitionBy("value").orderBy("key").rangeBetween(-1, 1))
+ .as("avg_key3"))
+
+ checkAnswer(
+ df.select(selectList: _*).where($"value" < 2),
+ Seq(Row(3, "1", null, 3.0, 4.0, 3.0), Row(5, "1", false, 4.0, 5.0, 5.0)))
+ }
}
test("Window spill with less than the inMemoryThreshold") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
index f45bd95..fe6775c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
@@ -401,94 +401,96 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
}
test("full outer join") {
- upperCaseData.where('N <= 4).createOrReplaceTempView("`left`")
- upperCaseData.where('N >= 3).createOrReplaceTempView("`right`")
+ withTempView("`left`", "`right`") {
+ upperCaseData.where('N <= 4).createOrReplaceTempView("`left`")
+ upperCaseData.where('N >= 3).createOrReplaceTempView("`right`")
- val left = UnresolvedRelation(TableIdentifier("left"))
- val right = UnresolvedRelation(TableIdentifier("right"))
+ val left = UnresolvedRelation(TableIdentifier("left"))
+ val right = UnresolvedRelation(TableIdentifier("right"))
- checkAnswer(
- left.join(right, $"left.N" === $"right.N", "full"),
- Row(1, "A", null, null) ::
- Row(2, "B", null, null) ::
- Row(3, "C", 3, "C") ::
- Row(4, "D", 4, "D") ::
- Row(null, null, 5, "E") ::
- Row(null, null, 6, "F") :: Nil)
+ checkAnswer(
+ left.join(right, $"left.N" === $"right.N", "full"),
+ Row(1, "A", null, null) ::
+ Row(2, "B", null, null) ::
+ Row(3, "C", 3, "C") ::
+ Row(4, "D", 4, "D") ::
+ Row(null, null, 5, "E") ::
+ Row(null, null, 6, "F") :: Nil)
- checkAnswer(
- left.join(right, ($"left.N" === $"right.N") && ($"left.N" =!= 3), "full"),
- Row(1, "A", null, null) ::
- Row(2, "B", null, null) ::
- Row(3, "C", null, null) ::
- Row(null, null, 3, "C") ::
- Row(4, "D", 4, "D") ::
- Row(null, null, 5, "E") ::
- Row(null, null, 6, "F") :: Nil)
+ checkAnswer(
+ left.join(right, ($"left.N" === $"right.N") && ($"left.N" =!= 3), "full"),
+ Row(1, "A", null, null) ::
+ Row(2, "B", null, null) ::
+ Row(3, "C", null, null) ::
+ Row(null, null, 3, "C") ::
+ Row(4, "D", 4, "D") ::
+ Row(null, null, 5, "E") ::
+ Row(null, null, 6, "F") :: Nil)
- checkAnswer(
- left.join(right, ($"left.N" === $"right.N") && ($"right.N" =!= 3), "full"),
- Row(1, "A", null, null) ::
- Row(2, "B", null, null) ::
- Row(3, "C", null, null) ::
- Row(null, null, 3, "C") ::
- Row(4, "D", 4, "D") ::
- Row(null, null, 5, "E") ::
- Row(null, null, 6, "F") :: Nil)
-
- // Make sure we are UnknownPartitioning as the outputPartitioning for the outer join
- // operator.
- checkAnswer(
- sql(
- """
- |SELECT l.a, count(*)
- |FROM allNulls l FULL OUTER JOIN upperCaseData r ON (l.a = r.N)
- |GROUP BY l.a
- """.
- stripMargin),
- Row(null, 10))
+ checkAnswer(
+ left.join(right, ($"left.N" === $"right.N") && ($"right.N" =!= 3), "full"),
+ Row(1, "A", null, null) ::
+ Row(2, "B", null, null) ::
+ Row(3, "C", null, null) ::
+ Row(null, null, 3, "C") ::
+ Row(4, "D", 4, "D") ::
+ Row(null, null, 5, "E") ::
+ Row(null, null, 6, "F") :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT r.N, count(*)
+ // Make sure we are UnknownPartitioning as the outputPartitioning for the outer join
+ // operator.
+ checkAnswer(
+ sql(
+ """
+ |SELECT l.a, count(*)
|FROM allNulls l FULL OUTER JOIN upperCaseData r ON (l.a = r.N)
- |GROUP BY r.N
- """.stripMargin),
- Row
+ |GROUP BY l.a
+ """.
+ stripMargin),
+ Row(null, 10))
+
+ checkAnswer(
+ sql(
+ """
+ |SELECT r.N, count(*)
+ |FROM allNulls l FULL OUTER JOIN upperCaseData r ON (l.a = r.N)
+ |GROUP BY r.N
+ """.stripMargin),
+ Row
(1, 1) ::
- Row(2, 1) ::
- Row(3, 1) ::
- Row(4, 1) ::
- Row(5, 1) ::
- Row(6, 1) ::
- Row(null, 4) :: Nil)
+ Row(2, 1) ::
+ Row(3, 1) ::
+ Row(4, 1) ::
+ Row(5, 1) ::
+ Row(6, 1) ::
+ Row(null, 4) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT l.N, count(*)
- |FROM upperCaseData l FULL OUTER JOIN allNulls r ON (l.N = r.a)
- |GROUP BY l.N
- """.stripMargin),
- Row(1
- , 1) ::
- Row(2, 1) ::
- Row(3, 1) ::
- Row(4, 1) ::
- Row(5, 1) ::
- Row(6, 1) ::
- Row(null, 4) :: Nil)
+ checkAnswer(
+ sql(
+ """
+ |SELECT l.N, count(*)
+ |FROM upperCaseData l FULL OUTER JOIN allNulls r ON (l.N = r.a)
+ |GROUP BY l.N
+ """.stripMargin),
+ Row(1
+ , 1) ::
+ Row(2, 1) ::
+ Row(3, 1) ::
+ Row(4, 1) ::
+ Row(5, 1) ::
+ Row(6, 1) ::
+ Row(null, 4) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT r.a, count(*)
- |FROM upperCaseData l FULL OUTER JOIN allNulls r ON (l.N = r.a)
- |GROUP BY r.a
- """.
- stripMargin),
- Row(null, 10))
+ checkAnswer(
+ sql(
+ """
+ |SELECT r.a, count(*)
+ |FROM upperCaseData l FULL OUTER JOIN allNulls r ON (l.N = r.a)
+ |GROUP BY r.a
+ """.
+ stripMargin),
+ Row(null, 10))
+ }
}
test("broadcasted existence join operator selection") {
@@ -614,63 +616,65 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
}
test("cross join detection") {
- testData.createOrReplaceTempView("A")
- testData.createOrReplaceTempView("B")
- testData2.createOrReplaceTempView("C")
- testData3.createOrReplaceTempView("D")
- upperCaseData.where('N >= 3).createOrReplaceTempView("`right`")
- val cartesianQueries = Seq(
- /** The following should error out since there is no explicit cross join */
- "SELECT * FROM testData inner join testData2",
- "SELECT * FROM testData left outer join testData2",
- "SELECT * FROM testData right outer join testData2",
- "SELECT * FROM testData full outer join testData2",
- "SELECT * FROM testData, testData2",
- "SELECT * FROM testData, testData2 where testData.key = 1 and testData2.a = 22",
- /** The following should fail because after reordering there are cartesian products */
- "select * from (A join B on (A.key = B.key)) join D on (A.key=D.a) join C",
- "select * from ((A join B on (A.key = B.key)) join C) join D on (A.key = D.a)",
- /** Cartesian product involving C, which is not involved in a CROSS join */
- "select * from ((A join B on (A.key = B.key)) cross join D) join C on (A.key = D.a)");
-
- def checkCartesianDetection(query: String): Unit = {
- val e = intercept[Exception] {
- checkAnswer(sql(query), Nil);
+ withTempView("A", "B", "C", "D") {
+ testData.createOrReplaceTempView("A")
+ testData.createOrReplaceTempView("B")
+ testData2.createOrReplaceTempView("C")
+ testData3.createOrReplaceTempView("D")
+ upperCaseData.where('N >= 3).createOrReplaceTempView("`right`")
+ val cartesianQueries = Seq(
+ /** The following should error out since there is no explicit cross join */
+ "SELECT * FROM testData inner join testData2",
+ "SELECT * FROM testData left outer join testData2",
+ "SELECT * FROM testData right outer join testData2",
+ "SELECT * FROM testData full outer join testData2",
+ "SELECT * FROM testData, testData2",
+ "SELECT * FROM testData, testData2 where testData.key = 1 and testData2.a = 22",
+ /** The following should fail because after reordering there are cartesian products */
+ "select * from (A join B on (A.key = B.key)) join D on (A.key=D.a) join C",
+ "select * from ((A join B on (A.key = B.key)) join C) join D on (A.key = D.a)",
+ /** Cartesian product involving C, which is not involved in a CROSS join */
+ "select * from ((A join B on (A.key = B.key)) cross join D) join C on (A.key = D.a)");
+
+ def checkCartesianDetection(query: String): Unit = {
+ val e = intercept[Exception] {
+ checkAnswer(sql(query), Nil);
+ }
+ assert(e.getMessage.contains("Detected implicit cartesian product"))
}
- assert(e.getMessage.contains("Detected implicit cartesian product"))
- }
- withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "false") {
- cartesianQueries.foreach(checkCartesianDetection)
- }
+ withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "false") {
+ cartesianQueries.foreach(checkCartesianDetection)
+ }
- // Check that left_semi, left_anti, existence joins without conditions do not throw
- // an exception if cross joins are disabled
- withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "false") {
- checkAnswer(
- sql("SELECT * FROM testData3 LEFT SEMI JOIN testData2"),
- Row(1, null) :: Row (2, 2) :: Nil)
- checkAnswer(
- sql("SELECT * FROM testData3 LEFT ANTI JOIN testData2"),
- Nil)
- checkAnswer(
- sql(
- """
- |SELECT a FROM testData3
- |WHERE
- | EXISTS (SELECT * FROM testData)
- |OR
- | EXISTS (SELECT * FROM testData2)""".stripMargin),
- Row(1) :: Row(2) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT key FROM testData
- |WHERE
- | key IN (SELECT a FROM testData2)
- |OR
- | key IN (SELECT a FROM testData3)""".stripMargin),
- Row(1) :: Row(2) :: Row(3) :: Nil)
+ // Check that left_semi, left_anti, existence joins without conditions do not throw
+ // an exception if cross joins are disabled
+ withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "false") {
+ checkAnswer(
+ sql("SELECT * FROM testData3 LEFT SEMI JOIN testData2"),
+ Row(1, null) :: Row (2, 2) :: Nil)
+ checkAnswer(
+ sql("SELECT * FROM testData3 LEFT ANTI JOIN testData2"),
+ Nil)
+ checkAnswer(
+ sql(
+ """
+ |SELECT a FROM testData3
+ |WHERE
+ | EXISTS (SELECT * FROM testData)
+ |OR
+ | EXISTS (SELECT * FROM testData2)""".stripMargin),
+ Row(1) :: Row(2) :: Nil)
+ checkAnswer(
+ sql(
+ """
+ |SELECT key FROM testData
+ |WHERE
+ | key IN (SELECT a FROM testData2)
+ |OR
+ | key IN (SELECT a FROM testData3)""".stripMargin),
+ Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index b989b5d..a4ed7cc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -429,57 +429,69 @@ class JsonFunctionsSuite extends QueryTest with SharedSparkSession {
}
test("from_json - array of arrays") {
- val jsonDF = Seq("[[1], [2, 3], [4, 5, 6]]").toDF("a")
- val schema = new ArrayType(ArrayType(IntegerType, false), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("[[1], [2, 3], [4, 5, 6]]").toDF("a")
+ val schema = new ArrayType(ArrayType(IntegerType, false), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select json[0][0], json[1][1], json[2][2] from jsonTable"),
- Seq(Row(1, 3, 6)))
+ checkAnswer(
+ sql("select json[0][0], json[1][1], json[2][2] from jsonTable"),
+ Seq(Row(1, 3, 6)))
+ }
}
test("from_json - array of arrays - malformed row") {
- val jsonDF = Seq("[[1], [2, 3], 4, 5, 6]]").toDF("a")
- val schema = new ArrayType(ArrayType(IntegerType, false), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("[[1], [2, 3], 4, 5, 6]]").toDF("a")
+ val schema = new ArrayType(ArrayType(IntegerType, false), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(sql("select json[0] from jsonTable"), Seq(Row(null)))
+ checkAnswer(sql("select json[0] from jsonTable"), Seq(Row(null)))
+ }
}
test("from_json - array of structs") {
- val jsonDF = Seq("""[{"a":1}, {"a":2}, {"a":3}]""").toDF("a")
- val schema = new ArrayType(new StructType().add("a", IntegerType), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("""[{"a":1}, {"a":2}, {"a":3}]""").toDF("a")
+ val schema = new ArrayType(new StructType().add("a", IntegerType), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select json[0], json[1], json[2] from jsonTable"),
- Seq(Row(Row(1), Row(2), Row(3))))
+ checkAnswer(
+ sql("select json[0], json[1], json[2] from jsonTable"),
+ Seq(Row(Row(1), Row(2), Row(3))))
+ }
}
test("from_json - array of structs - malformed row") {
- val jsonDF = Seq("""[{"a":1}, {"a:2}, {"a":3}]""").toDF("a")
- val schema = new ArrayType(new StructType().add("a", IntegerType), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("""[{"a":1}, {"a:2}, {"a":3}]""").toDF("a")
+ val schema = new ArrayType(new StructType().add("a", IntegerType), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(sql("select json[0], json[1]from jsonTable"), Seq(Row(null, null)))
+ checkAnswer(sql("select json[0], json[1]from jsonTable"), Seq(Row(null, null)))
+ }
}
test("from_json - array of maps") {
- val jsonDF = Seq("""[{"a":1}, {"b":2}]""").toDF("a")
- val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("""[{"a":1}, {"b":2}]""").toDF("a")
+ val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("""select json[0], json[1] from jsonTable"""),
- Seq(Row(Map("a" -> 1), Map("b" -> 2))))
+ checkAnswer(
+ sql("""select json[0], json[1] from jsonTable"""),
+ Seq(Row(Map("a" -> 1), Map("b" -> 2))))
+ }
}
test("from_json - array of maps - malformed row") {
- val jsonDF = Seq("""[{"a":1} "b":2}]""").toDF("a")
- val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
- jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = Seq("""[{"a":1} "b":2}]""").toDF("a")
+ val schema = new ArrayType(MapType(StringType, IntegerType, false), false)
+ jsonDF.select(from_json($"a", schema) as "json").createOrReplaceTempView("jsonTable")
- checkAnswer(sql("""select json[0] from jsonTable"""), Seq(Row(null)))
+ checkAnswer(sql("""select json[0] from jsonTable"""), Seq(Row(null)))
+ }
}
test("to_json - array of primitive types") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 58a40b0..d336f52 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -50,13 +50,15 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
setupTestData()
test("SPARK-8010: promote numeric to string") {
- val df = Seq((1, 1)).toDF("key", "value")
- df.createOrReplaceTempView("src")
- val queryCaseWhen = sql("select case when true then 1.0 else '1' end from src ")
- val queryCoalesce = sql("select coalesce(null, 1, '1') from src ")
+ withTempView("src") {
+ val df = Seq((1, 1)).toDF("key", "value")
+ df.createOrReplaceTempView("src")
+ val queryCaseWhen = sql("select case when true then 1.0 else '1' end from src ")
+ val queryCoalesce = sql("select coalesce(null, 1, '1') from src ")
- checkAnswer(queryCaseWhen, Row("1.0") :: Nil)
- checkAnswer(queryCoalesce, Row("1") :: Nil)
+ checkAnswer(queryCaseWhen, Row("1.0") :: Nil)
+ checkAnswer(queryCoalesce, Row("1") :: Nil)
+ }
}
test("show functions") {
@@ -124,31 +126,35 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("SPARK-6743: no columns from cache") {
- Seq(
- (83, 0, 38),
- (26, 0, 79),
- (43, 81, 24)
- ).toDF("a", "b", "c").createOrReplaceTempView("cachedData")
+ withTempView("cachedData") {
+ Seq(
+ (83, 0, 38),
+ (26, 0, 79),
+ (43, 81, 24)
+ ).toDF("a", "b", "c").createOrReplaceTempView("cachedData")
- spark.catalog.cacheTable("cachedData")
- withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "true") {
- checkAnswer(
- sql("SELECT t1.b FROM cachedData, cachedData t1 GROUP BY t1.b"),
- Row(0) :: Row(81) :: Nil)
+ spark.catalog.cacheTable("cachedData")
+ withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "true") {
+ checkAnswer(
+ sql("SELECT t1.b FROM cachedData, cachedData t1 GROUP BY t1.b"),
+ Row(0) :: Row(81) :: Nil)
+ }
}
}
test("self join with aliases") {
- Seq(1, 2, 3).map(i => (i, i.toString)).toDF("int", "str").createOrReplaceTempView("df")
+ withTempView("df") {
+ Seq(1, 2, 3).map(i => (i, i.toString)).toDF("int", "str").createOrReplaceTempView("df")
- checkAnswer(
- sql(
- """
- |SELECT x.str, COUNT(*)
- |FROM df x JOIN df y ON x.str = y.str
- |GROUP BY x.str
+ checkAnswer(
+ sql(
+ """
+ |SELECT x.str, COUNT(*)
+ |FROM df x JOIN df y ON x.str = y.str
+ |GROUP BY x.str
""".stripMargin),
- Row("1", 1) :: Row("2", 1) :: Row("3", 1) :: Nil)
+ Row("1", 1) :: Row("2", 1) :: Row("3", 1) :: Nil)
+ }
}
test("support table.star") {
@@ -162,6 +168,7 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("self join with alias in agg") {
+ withTempView("df") {
Seq(1, 2, 3)
.map(i => (i, i.toString))
.toDF("int", "str")
@@ -169,14 +176,15 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
.agg($"str", count("str").as("strCount"))
.createOrReplaceTempView("df")
- checkAnswer(
- sql(
- """
- |SELECT x.str, SUM(x.strCount)
- |FROM df x JOIN df y ON x.str = y.str
- |GROUP BY x.str
+ checkAnswer(
+ sql(
+ """
+ |SELECT x.str, SUM(x.strCount)
+ |FROM df x JOIN df y ON x.str = y.str
+ |GROUP BY x.str
""".stripMargin),
- Row("1", 1) :: Row("2", 1) :: Row("3", 1) :: Nil)
+ Row("1", 1) :: Row("2", 1) :: Row("3", 1) :: Nil)
+ }
}
test("SPARK-8668 expr function") {
@@ -223,41 +231,47 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("grouping on nested fields") {
- spark.read
- .json(Seq("""{"nested": {"attribute": 1}, "value": 2}""").toDS())
- .createOrReplaceTempView("rows")
+ withTempView("rows") {
+ spark.read
+ .json(Seq("""{"nested": {"attribute": 1}, "value": 2}""").toDS())
+ .createOrReplaceTempView("rows")
- checkAnswer(
- sql(
- """
- |select attribute, sum(cnt)
- |from (
- | select nested.attribute, count(*) as cnt
- | from rows
- | group by nested.attribute) a
- |group by attribute
+ checkAnswer(
+ sql(
+ """
+ |select attribute, sum(cnt)
+ |from (
+ | select nested.attribute, count(*) as cnt
+ | from rows
+ | group by nested.attribute) a
+ |group by attribute
""".stripMargin),
- Row(1, 1) :: Nil)
+ Row(1, 1) :: Nil)
+ }
}
test("SPARK-6201 IN type conversion") {
- spark.read
- .json(Seq("{\"a\": \"1\"}}", "{\"a\": \"2\"}}", "{\"a\": \"3\"}}").toDS())
- .createOrReplaceTempView("d")
+ withTempView("d") {
+ spark.read
+ .json(Seq("{\"a\": \"1\"}}", "{\"a\": \"2\"}}", "{\"a\": \"3\"}}").toDS())
+ .createOrReplaceTempView("d")
- checkAnswer(
- sql("select * from d where d.a in (1,2)"),
- Seq(Row("1"), Row("2")))
+ checkAnswer(
+ sql("select * from d where d.a in (1,2)"),
+ Seq(Row("1"), Row("2")))
+ }
}
test("SPARK-11226 Skip empty line in json file") {
- spark.read
- .json(Seq("{\"a\": \"1\"}}", "{\"a\": \"2\"}}", "{\"a\": \"3\"}}", "").toDS())
- .createOrReplaceTempView("d")
+ withTempView("d") {
+ spark.read
+ .json(Seq("{\"a\": \"1\"}}", "{\"a\": \"2\"}}", "{\"a\": \"3\"}}", "").toDS())
+ .createOrReplaceTempView("d")
- checkAnswer(
- sql("select count(1) from d"),
- Seq(Row(3)))
+ checkAnswer(
+ sql("select count(1) from d"),
+ Seq(Row(3)))
+ }
}
test("SPARK-8828 sum should return null if all input values are null") {
@@ -419,40 +433,42 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("SPARK-3173 Timestamp support in the parser") {
- (0 to 3).map(i => Tuple1(new Timestamp(i))).toDF("time").createOrReplaceTempView("timestamps")
+ withTempView("timestamps") {
+ (0 to 3).map(i => Tuple1(new Timestamp(i))).toDF("time").createOrReplaceTempView("timestamps")
- checkAnswer(sql(
- "SELECT time FROM timestamps WHERE time='1969-12-31 16:00:00.0'"),
- Row(Timestamp.valueOf("1969-12-31 16:00:00")))
+ checkAnswer(sql(
+ "SELECT time FROM timestamps WHERE time='1969-12-31 16:00:00.0'"),
+ Row(Timestamp.valueOf("1969-12-31 16:00:00")))
- checkAnswer(sql(
- "SELECT time FROM timestamps WHERE time=CAST('1969-12-31 16:00:00.001' AS TIMESTAMP)"),
- Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
+ checkAnswer(sql(
+ "SELECT time FROM timestamps WHERE time=CAST('1969-12-31 16:00:00.001' AS TIMESTAMP)"),
+ Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
- checkAnswer(sql(
- "SELECT time FROM timestamps WHERE time='1969-12-31 16:00:00.001'"),
- Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
+ checkAnswer(sql(
+ "SELECT time FROM timestamps WHERE time='1969-12-31 16:00:00.001'"),
+ Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
- checkAnswer(sql(
- "SELECT time FROM timestamps WHERE '1969-12-31 16:00:00.001'=time"),
- Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
+ checkAnswer(sql(
+ "SELECT time FROM timestamps WHERE '1969-12-31 16:00:00.001'=time"),
+ Row(Timestamp.valueOf("1969-12-31 16:00:00.001")))
- checkAnswer(sql(
- """SELECT time FROM timestamps WHERE time<'1969-12-31 16:00:00.003'
+ checkAnswer(sql(
+ """SELECT time FROM timestamps WHERE time<'1969-12-31 16:00:00.003'
AND time>'1969-12-31 16:00:00.001'"""),
- Row(Timestamp.valueOf("1969-12-31 16:00:00.002")))
+ Row(Timestamp.valueOf("1969-12-31 16:00:00.002")))
- checkAnswer(sql(
- """
- |SELECT time FROM timestamps
- |WHERE time IN ('1969-12-31 16:00:00.001','1969-12-31 16:00:00.002')
+ checkAnswer(sql(
+ """
+ |SELECT time FROM timestamps
+ |WHERE time IN ('1969-12-31 16:00:00.001','1969-12-31 16:00:00.002')
""".stripMargin),
- Seq(Row(Timestamp.valueOf("1969-12-31 16:00:00.001")),
- Row(Timestamp.valueOf("1969-12-31 16:00:00.002"))))
+ Seq(Row(Timestamp.valueOf("1969-12-31 16:00:00.001")),
+ Row(Timestamp.valueOf("1969-12-31 16:00:00.002"))))
- checkAnswer(sql(
- "SELECT time FROM timestamps WHERE time='123'"),
- Nil)
+ checkAnswer(sql(
+ "SELECT time FROM timestamps WHERE time='123'"),
+ Nil)
+ }
}
test("left semi greater than predicate") {
@@ -781,20 +797,22 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("SPARK-3349 partitioning after limit") {
- sql("SELECT DISTINCT n FROM lowerCaseData ORDER BY n DESC")
- .limit(2)
- .createOrReplaceTempView("subset1")
- sql("SELECT DISTINCT n FROM lowerCaseData ORDER BY n ASC")
- .limit(2)
- .createOrReplaceTempView("subset2")
- checkAnswer(
- sql("SELECT * FROM lowerCaseData INNER JOIN subset1 ON subset1.n = lowerCaseData.n"),
- Row(3, "c", 3) ::
- Row(4, "d", 4) :: Nil)
- checkAnswer(
- sql("SELECT * FROM lowerCaseData INNER JOIN subset2 ON subset2.n = lowerCaseData.n"),
- Row(1, "a", 1) ::
- Row(2, "b", 2) :: Nil)
+ withTempView("subset1", "subset2") {
+ sql("SELECT DISTINCT n FROM lowerCaseData ORDER BY n DESC")
+ .limit(2)
+ .createOrReplaceTempView("subset1")
+ sql("SELECT DISTINCT n FROM lowerCaseData ORDER BY n ASC")
+ .limit(2)
+ .createOrReplaceTempView("subset2")
+ checkAnswer(
+ sql("SELECT * FROM lowerCaseData INNER JOIN subset1 ON subset1.n = lowerCaseData.n"),
+ Row(3, "c", 3) ::
+ Row(4, "d", 4) :: Nil)
+ checkAnswer(
+ sql("SELECT * FROM lowerCaseData INNER JOIN subset2 ON subset2.n = lowerCaseData.n"),
+ Row(1, "a", 1) ::
+ Row(2, "b", 2) :: Nil)
+ }
}
test("mixed-case keywords") {
@@ -1037,84 +1055,87 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("apply schema") {
- val schema1 = StructType(
- StructField("f1", IntegerType, false) ::
- StructField("f2", StringType, false) ::
- StructField("f3", BooleanType, false) ::
- StructField("f4", IntegerType, true) :: Nil)
-
- val rowRDD1 = unparsedStrings.map { r =>
- val values = r.split(",").map(_.trim)
- val v4 = try values(3).toInt catch {
- case _: NumberFormatException => null
+ withTempView("applySchema1", "applySchema2", "applySchema3") {
+ val schema1 = StructType(
+ StructField("f1", IntegerType, false) ::
+ StructField("f2", StringType, false) ::
+ StructField("f3", BooleanType, false) ::
+ StructField("f4", IntegerType, true) :: Nil)
+
+ val rowRDD1 = unparsedStrings.map { r =>
+ val values = r.split(",").map(_.trim)
+ val v4 = try values(3).toInt catch {
+ case _: NumberFormatException => null
+ }
+ Row(values(0).toInt, values(1), values(2).toBoolean, v4)
}
- Row(values(0).toInt, values(1), values(2).toBoolean, v4)
- }
-
- val df1 = spark.createDataFrame(rowRDD1, schema1)
- df1.createOrReplaceTempView("applySchema1")
- checkAnswer(
- sql("SELECT * FROM applySchema1"),
- Row(1, "A1", true, null) ::
- Row(2, "B2", false, null) ::
- Row(3, "C3", true, null) ::
- Row(4, "D4", true, 2147483644) :: Nil)
- checkAnswer(
- sql("SELECT f1, f4 FROM applySchema1"),
- Row(1, null) ::
- Row(2, null) ::
- Row(3, null) ::
- Row(4, 2147483644) :: Nil)
-
- val schema2 = StructType(
- StructField("f1", StructType(
- StructField("f11", IntegerType, false) ::
- StructField("f12", BooleanType, false) :: Nil), false) ::
- StructField("f2", MapType(StringType, IntegerType, true), false) :: Nil)
+ val df1 = spark.createDataFrame(rowRDD1, schema1)
+ df1.createOrReplaceTempView("applySchema1")
+ checkAnswer(
+ sql("SELECT * FROM applySchema1"),
+ Row(1, "A1", true, null) ::
+ Row(2, "B2", false, null) ::
+ Row(3, "C3", true, null) ::
+ Row(4, "D4", true, 2147483644) :: Nil)
- val rowRDD2 = unparsedStrings.map { r =>
- val values = r.split(",").map(_.trim)
- val v4 = try values(3).toInt catch {
- case _: NumberFormatException => null
+ checkAnswer(
+ sql("SELECT f1, f4 FROM applySchema1"),
+ Row(1, null) ::
+ Row(2, null) ::
+ Row(3, null) ::
+ Row(4, 2147483644) :: Nil)
+
+ val schema2 = StructType(
+ StructField("f1", StructType(
+ StructField("f11", IntegerType, false) ::
+ StructField("f12", BooleanType, false) :: Nil), false) ::
+ StructField("f2", MapType(StringType, IntegerType, true), false) :: Nil)
+
+ val rowRDD2 = unparsedStrings.map { r =>
+ val values = r.split(",").map(_.trim)
+ val v4 = try values(3).toInt catch {
+ case _: NumberFormatException => null
+ }
+ Row(Row(values(0).toInt, values(2).toBoolean), Map(values(1) -> v4))
}
- Row(Row(values(0).toInt, values(2).toBoolean), Map(values(1) -> v4))
- }
-
- val df2 = spark.createDataFrame(rowRDD2, schema2)
- df2.createOrReplaceTempView("applySchema2")
- checkAnswer(
- sql("SELECT * FROM applySchema2"),
- Row(Row(1, true), Map("A1" -> null)) ::
- Row(Row(2, false), Map("B2" -> null)) ::
- Row(Row(3, true), Map("C3" -> null)) ::
- Row(Row(4, true), Map("D4" -> 2147483644)) :: Nil)
- checkAnswer(
- sql("SELECT f1.f11, f2['D4'] FROM applySchema2"),
- Row(1, null) ::
- Row(2, null) ::
- Row(3, null) ::
- Row(4, 2147483644) :: Nil)
+ val df2 = spark.createDataFrame(rowRDD2, schema2)
+ df2.createOrReplaceTempView("applySchema2")
+ checkAnswer(
+ sql("SELECT * FROM applySchema2"),
+ Row(Row(1, true), Map("A1" -> null)) ::
+ Row(Row(2, false), Map("B2" -> null)) ::
+ Row(Row(3, true), Map("C3" -> null)) ::
+ Row(Row(4, true), Map("D4" -> 2147483644)) :: Nil)
- // The value of a MapType column can be a mutable map.
- val rowRDD3 = unparsedStrings.map { r =>
- val values = r.split(",").map(_.trim)
- val v4 = try values(3).toInt catch {
- case _: NumberFormatException => null
+ checkAnswer(
+ sql("SELECT f1.f11, f2['D4'] FROM applySchema2"),
+ Row(1, null) ::
+ Row(2, null) ::
+ Row(3, null) ::
+ Row(4, 2147483644) :: Nil)
+
+ // The value of a MapType column can be a mutable map.
+ val rowRDD3 = unparsedStrings.map { r =>
+ val values = r.split(",").map(_.trim)
+ val v4 = try values(3).toInt catch {
+ case _: NumberFormatException => null
+ }
+ Row(Row(values(0).toInt, values(2).toBoolean),
+ scala.collection.mutable.Map(values(1) -> v4))
}
- Row(Row(values(0).toInt, values(2).toBoolean), scala.collection.mutable.Map(values(1) -> v4))
- }
- val df3 = spark.createDataFrame(rowRDD3, schema2)
- df3.createOrReplaceTempView("applySchema3")
+ val df3 = spark.createDataFrame(rowRDD3, schema2)
+ df3.createOrReplaceTempView("applySchema3")
- checkAnswer(
- sql("SELECT f1.f11, f2['D4'] FROM applySchema3"),
- Row(1, null) ::
- Row(2, null) ::
- Row(3, null) ::
- Row(4, 2147483644) :: Nil)
+ checkAnswer(
+ sql("SELECT f1.f11, f2['D4'] FROM applySchema3"),
+ Row(1, null) ::
+ Row(2, null) ::
+ Row(3, null) ::
+ Row(4, 2147483644) :: Nil)
+ }
}
test("SPARK-3423 BETWEEN") {
@@ -1166,28 +1187,30 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("metadata is propagated correctly") {
- val person: DataFrame = sql("SELECT * FROM person")
- val schema = person.schema
- val docKey = "doc"
- val docValue = "first name"
- val metadata = new MetadataBuilder()
- .putString(docKey, docValue)
- .build()
- val schemaWithMeta = new StructType(Array(
- schema("id"), schema("name").copy(metadata = metadata), schema("age")))
- val personWithMeta = spark.createDataFrame(person.rdd, schemaWithMeta)
- def validateMetadata(rdd: DataFrame): Unit = {
- assert(rdd.schema("name").metadata.getString(docKey) == docValue)
- }
- personWithMeta.createOrReplaceTempView("personWithMeta")
- validateMetadata(personWithMeta.select($"name"))
- validateMetadata(personWithMeta.select($"name"))
- validateMetadata(personWithMeta.select($"id", $"name"))
- validateMetadata(sql("SELECT * FROM personWithMeta"))
- validateMetadata(sql("SELECT id, name FROM personWithMeta"))
- validateMetadata(sql("SELECT * FROM personWithMeta JOIN salary ON id = personId"))
- validateMetadata(sql(
- "SELECT name, salary FROM personWithMeta JOIN salary ON id = personId"))
+ withTempView("personWithMeta") {
+ val person: DataFrame = sql("SELECT * FROM person")
+ val schema = person.schema
+ val docKey = "doc"
+ val docValue = "first name"
+ val metadata = new MetadataBuilder()
+ .putString(docKey, docValue)
+ .build()
+ val schemaWithMeta = new StructType(Array(
+ schema("id"), schema("name").copy(metadata = metadata), schema("age")))
+ val personWithMeta = spark.createDataFrame(person.rdd, schemaWithMeta)
+ def validateMetadata(rdd: DataFrame): Unit = {
+ assert(rdd.schema("name").metadata.getString(docKey) == docValue)
+ }
+ personWithMeta.createOrReplaceTempView("personWithMeta")
+ validateMetadata(personWithMeta.select($"name"))
+ validateMetadata(personWithMeta.select($"name"))
+ validateMetadata(personWithMeta.select($"id", $"name"))
+ validateMetadata(sql("SELECT * FROM personWithMeta"))
+ validateMetadata(sql("SELECT id, name FROM personWithMeta"))
+ validateMetadata(sql("SELECT * FROM personWithMeta JOIN salary ON id = personId"))
+ validateMetadata(sql(
+ "SELECT name, salary FROM personWithMeta JOIN salary ON id = personId"))
+ }
}
test("SPARK-3371 Renaming a function expression with group by gives error") {
@@ -1229,10 +1252,12 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("SPARK-3483 Special chars in column names") {
- val data = Seq("""{"key?number1": "value1", "key.number2": "value2"}""").toDS()
- spark.read.json(data).createOrReplaceTempView("records")
- withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
- sql("SELECT `key?number1`, `key.number2` FROM records")
+ withTempView("records") {
+ val data = Seq("""{"key?number1": "value1", "key.number2": "value2"}""").toDS()
+ spark.read.json(data).createOrReplaceTempView("records")
+ withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
+ sql("SELECT `key?number1`, `key.number2` FROM records")
+ }
}
}
@@ -1299,138 +1324,152 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("Supporting relational operator '<=>' in Spark SQL") {
- val nullCheckData1 = TestData(1, "1") :: TestData(2, null) :: Nil
- val rdd1 = sparkContext.parallelize((0 to 1).map(i => nullCheckData1(i)))
- rdd1.toDF().createOrReplaceTempView("nulldata1")
- val nullCheckData2 = TestData(1, "1") :: TestData(2, null) :: Nil
- val rdd2 = sparkContext.parallelize((0 to 1).map(i => nullCheckData2(i)))
- rdd2.toDF().createOrReplaceTempView("nulldata2")
- checkAnswer(sql("SELECT nulldata1.key FROM nulldata1 join " +
- "nulldata2 on nulldata1.value <=> nulldata2.value"),
+ withTempView("nulldata1", "nulldata2") {
+ val nullCheckData1 = TestData(1, "1") :: TestData(2, null) :: Nil
+ val rdd1 = sparkContext.parallelize((0 to 1).map(i => nullCheckData1(i)))
+ rdd1.toDF().createOrReplaceTempView("nulldata1")
+ val nullCheckData2 = TestData(1, "1") :: TestData(2, null) :: Nil
+ val rdd2 = sparkContext.parallelize((0 to 1).map(i => nullCheckData2(i)))
+ rdd2.toDF().createOrReplaceTempView("nulldata2")
+ checkAnswer(sql("SELECT nulldata1.key FROM nulldata1 join " +
+ "nulldata2 on nulldata1.value <=> nulldata2.value"),
(1 to 2).map(i => Row(i)))
+ }
}
test("Multi-column COUNT(DISTINCT ...)") {
- val data = TestData(1, "val_1") :: TestData(2, "val_2") :: Nil
- val rdd = sparkContext.parallelize((0 to 1).map(i => data(i)))
- rdd.toDF().createOrReplaceTempView("distinctData")
- checkAnswer(sql("SELECT COUNT(DISTINCT key,value) FROM distinctData"), Row(2))
+ withTempView("distinctData") {
+ val data = TestData(1, "val_1") :: TestData(2, "val_2") :: Nil
+ val rdd = sparkContext.parallelize((0 to 1).map(i => data(i)))
+ rdd.toDF().createOrReplaceTempView("distinctData")
+ checkAnswer(sql("SELECT COUNT(DISTINCT key,value) FROM distinctData"), Row(2))
+ }
}
test("SPARK-4699 case sensitivity SQL query") {
- withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
- val data = TestData(1, "val_1") :: TestData(2, "val_2") :: Nil
- val rdd = sparkContext.parallelize((0 to 1).map(i => data(i)))
- rdd.toDF().createOrReplaceTempView("testTable1")
- checkAnswer(sql("SELECT VALUE FROM TESTTABLE1 where KEY = 1"), Row("val_1"))
+ withTempView("testTable1") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val data = TestData(1, "val_1") :: TestData(2, "val_2") :: Nil
+ val rdd = sparkContext.parallelize((0 to 1).map(i => data(i)))
+ rdd.toDF().createOrReplaceTempView("testTable1")
+ checkAnswer(sql("SELECT VALUE FROM TESTTABLE1 where KEY = 1"), Row("val_1"))
+ }
}
}
test("SPARK-6145: ORDER BY test for nested fields") {
- spark.read
- .json(Seq("""{"a": {"b": 1, "a": {"a": 1}}, "c": [{"d": 1}]}""").toDS())
- .createOrReplaceTempView("nestedOrder")
+ withTempView("nestedOrder") {
+ spark.read
+ .json(Seq("""{"a": {"b": 1, "a": {"a": 1}}, "c": [{"d": 1}]}""").toDS())
+ .createOrReplaceTempView("nestedOrder")
- checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY a.b"), Row(1))
- checkAnswer(sql("SELECT a.b FROM nestedOrder ORDER BY a.b"), Row(1))
- checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY a.a.a"), Row(1))
- checkAnswer(sql("SELECT a.a.a FROM nestedOrder ORDER BY a.a.a"), Row(1))
- checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY c[0].d"), Row(1))
- checkAnswer(sql("SELECT c[0].d FROM nestedOrder ORDER BY c[0].d"), Row(1))
+ checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY a.b"), Row(1))
+ checkAnswer(sql("SELECT a.b FROM nestedOrder ORDER BY a.b"), Row(1))
+ checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY a.a.a"), Row(1))
+ checkAnswer(sql("SELECT a.a.a FROM nestedOrder ORDER BY a.a.a"), Row(1))
+ checkAnswer(sql("SELECT 1 FROM nestedOrder ORDER BY c[0].d"), Row(1))
+ checkAnswer(sql("SELECT c[0].d FROM nestedOrder ORDER BY c[0].d"), Row(1))
+ }
}
test("SPARK-6145: special cases") {
- spark.read
- .json(Seq("""{"a": {"b": [1]}, "b": [{"a": 1}], "_c0": {"a": 1}}""").toDS())
- .createOrReplaceTempView("t")
+ withTempView("t") {
+ spark.read
+ .json(Seq("""{"a": {"b": [1]}, "b": [{"a": 1}], "_c0": {"a": 1}}""").toDS())
+ .createOrReplaceTempView("t")
- checkAnswer(sql("SELECT a.b[0] FROM t ORDER BY _c0.a"), Row(1))
- checkAnswer(sql("SELECT b[0].a FROM t ORDER BY _c0.a"), Row(1))
+ checkAnswer(sql("SELECT a.b[0] FROM t ORDER BY _c0.a"), Row(1))
+ checkAnswer(sql("SELECT b[0].a FROM t ORDER BY _c0.a"), Row(1))
+ }
}
test("SPARK-6898: complete support for special chars in column names") {
- spark.read
- .json(Seq("""{"a": {"c.b": 1}, "b.$q": [{"a@!.q": 1}], "q.w": {"w.i&": [1]}}""").toDS())
- .createOrReplaceTempView("t")
+ withTempView("t") {
+ spark.read
+ .json(Seq("""{"a": {"c.b": 1}, "b.$q": [{"a@!.q": 1}], "q.w": {"w.i&": [1]}}""").toDS())
+ .createOrReplaceTempView("t")
- withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
- checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
+ withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
+ checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
+ }
}
}
test("SPARK-6583 order by aggregated function") {
- Seq("1" -> 3, "1" -> 4, "2" -> 7, "2" -> 8, "3" -> 5, "3" -> 6, "4" -> 1, "4" -> 2)
- .toDF("a", "b").createOrReplaceTempView("orderByData")
+ withTempView("orderByData") {
+ Seq("1" -> 3, "1" -> 4, "2" -> 7, "2" -> 8, "3" -> 5, "3" -> 6, "4" -> 1, "4" -> 2)
+ .toDF("a", "b").createOrReplaceTempView("orderByData")
- checkAnswer(
- sql(
- """
- |SELECT a
- |FROM orderByData
- |GROUP BY a
- |ORDER BY sum(b)
+ checkAnswer(
+ sql(
+ """
+ |SELECT a
+ |FROM orderByData
+ |GROUP BY a
+ |ORDER BY sum(b)
""".stripMargin),
- Row("4") :: Row("1") :: Row("3") :: Row("2") :: Nil)
+ Row("4") :: Row("1") :: Row("3") :: Row("2") :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT sum(b)
- |FROM orderByData
- |GROUP BY a
- |ORDER BY sum(b)
+ checkAnswer(
+ sql(
+ """
+ |SELECT sum(b)
+ |FROM orderByData
+ |GROUP BY a
+ |ORDER BY sum(b)
""".stripMargin),
- Row(3) :: Row(7) :: Row(11) :: Row(15) :: Nil)
+ Row(3) :: Row(7) :: Row(11) :: Row(15) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT sum(b)
- |FROM orderByData
- |GROUP BY a
- |ORDER BY sum(b), max(b)
+ checkAnswer(
+ sql(
+ """
+ |SELECT sum(b)
+ |FROM orderByData
+ |GROUP BY a
+ |ORDER BY sum(b), max(b)
""".stripMargin),
- Row(3) :: Row(7) :: Row(11) :: Row(15) :: Nil)
+ Row(3) :: Row(7) :: Row(11) :: Row(15) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT a, sum(b)
- |FROM orderByData
- |GROUP BY a
- |ORDER BY sum(b)
+ checkAnswer(
+ sql(
+ """
+ |SELECT a, sum(b)
+ |FROM orderByData
+ |GROUP BY a
+ |ORDER BY sum(b)
""".stripMargin),
- Row("4", 3) :: Row("1", 7) :: Row("3", 11) :: Row("2", 15) :: Nil)
+ Row("4", 3) :: Row("1", 7) :: Row("3", 11) :: Row("2", 15) :: Nil)
- checkAnswer(
- sql(
- """
+ checkAnswer(
+ sql(
+ """
|SELECT a, sum(b)
|FROM orderByData
|GROUP BY a
|ORDER BY sum(b) + 1
""".stripMargin),
- Row("4", 3) :: Row("1", 7) :: Row("3", 11) :: Row("2", 15) :: Nil)
+ Row("4", 3) :: Row("1", 7) :: Row("3", 11) :: Row("2", 15) :: Nil)
- checkAnswer(
- sql(
- """
+ checkAnswer(
+ sql(
+ """
|SELECT count(*)
|FROM orderByData
|GROUP BY a
|ORDER BY count(*)
""".stripMargin),
- Row(2) :: Row(2) :: Row(2) :: Row(2) :: Nil)
+ Row(2) :: Row(2) :: Row(2) :: Row(2) :: Nil)
- checkAnswer(
- sql(
- """
+ checkAnswer(
+ sql(
+ """
|SELECT a
|FROM orderByData
|GROUP BY a
|ORDER BY a, count(*), sum(b)
""".stripMargin),
- Row("1") :: Row("2") :: Row("3") :: Row("4") :: Nil)
+ Row("1") :: Row("2") :: Row("3") :: Row("4") :: Nil)
+ }
}
test("SPARK-7952: fix the equality check between boolean and numeric types") {
@@ -1742,137 +1781,141 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("Struct Star Expansion") {
- val structDf = testData2.select("a", "b").as("record")
+ withTempView("structTable", "nestedStructTable", "specialCharacterTable", "nameConflict") {
+ val structDf = testData2.select("a", "b").as("record")
- checkAnswer(
- structDf.select($"record.a", $"record.b"),
- Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
+ checkAnswer(
+ structDf.select($"record.a", $"record.b"),
+ Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
- checkAnswer(
- structDf.select($"record.*"),
- Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
+ checkAnswer(
+ structDf.select($"record.*"),
+ Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
- checkAnswer(
- structDf.select($"record.*", $"record.*"),
- Row(1, 1, 1, 1) :: Row(1, 2, 1, 2) :: Row(2, 1, 2, 1) :: Row(2, 2, 2, 2) ::
- Row(3, 1, 3, 1) :: Row(3, 2, 3, 2) :: Nil)
+ checkAnswer(
+ structDf.select($"record.*", $"record.*"),
+ Row(1, 1, 1, 1) :: Row(1, 2, 1, 2) :: Row(2, 1, 2, 1) :: Row(2, 2, 2, 2) ::
+ Row(3, 1, 3, 1) :: Row(3, 2, 3, 2) :: Nil)
- checkAnswer(
- sql("select struct(a, b) as r1, struct(b, a) as r2 from testData2").select($"r1.*", $"r2.*"),
- Row(1, 1, 1, 1) :: Row(1, 2, 2, 1) :: Row(2, 1, 1, 2) :: Row(2, 2, 2, 2) ::
- Row(3, 1, 1, 3) :: Row(3, 2, 2, 3) :: Nil)
+ checkAnswer(
+ sql("select struct(a, b) as r1, struct(b, a) as r2 from testData2")
+ .select($"r1.*", $"r2.*"),
+ Row(1, 1, 1, 1) :: Row(1, 2, 2, 1) :: Row(2, 1, 1, 2) :: Row(2, 2, 2, 2) ::
+ Row(3, 1, 1, 3) :: Row(3, 2, 2, 3) :: Nil)
- // Try with a temporary view
- sql("select struct(a, b) as record from testData2").createOrReplaceTempView("structTable")
- checkAnswer(
- sql("SELECT record.* FROM structTable"),
- Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
+ // Try with a temporary view
+ sql("select struct(a, b) as record from testData2").createOrReplaceTempView("structTable")
+ checkAnswer(
+ sql("SELECT record.* FROM structTable"),
+ Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
- checkAnswer(sql(
- """
- | SELECT min(struct(record.*)) FROM
- | (select struct(a,b) as record from testData2) tmp
+ checkAnswer(sql(
+ """
+ | SELECT min(struct(record.*)) FROM
+ | (select struct(a,b) as record from testData2) tmp
""".stripMargin),
- Row(Row(1, 1)) :: Nil)
+ Row(Row(1, 1)) :: Nil)
- // Try with an alias on the select list
- checkAnswer(sql(
- """
- | SELECT max(struct(record.*)) as r FROM
- | (select struct(a,b) as record from testData2) tmp
+ // Try with an alias on the select list
+ checkAnswer(sql(
+ """
+ | SELECT max(struct(record.*)) as r FROM
+ | (select struct(a,b) as record from testData2) tmp
""".stripMargin).select($"r.*"),
- Row(3, 2) :: Nil)
+ Row(3, 2) :: Nil)
- // With GROUP BY
- checkAnswer(sql(
- """
- | SELECT min(struct(record.*)) FROM
- | (select a as a, struct(a,b) as record from testData2) tmp
- | GROUP BY a
+ // With GROUP BY
+ checkAnswer(sql(
+ """
+ | SELECT min(struct(record.*)) FROM
+ | (select a as a, struct(a,b) as record from testData2) tmp
+ | GROUP BY a
""".stripMargin),
- Row(Row(1, 1)) :: Row(Row(2, 1)) :: Row(Row(3, 1)) :: Nil)
+ Row(Row(1, 1)) :: Row(Row(2, 1)) :: Row(Row(3, 1)) :: Nil)
- // With GROUP BY and alias
- checkAnswer(sql(
- """
- | SELECT max(struct(record.*)) as r FROM
- | (select a as a, struct(a,b) as record from testData2) tmp
- | GROUP BY a
+ // With GROUP BY and alias
+ checkAnswer(sql(
+ """
+ | SELECT max(struct(record.*)) as r FROM
+ | (select a as a, struct(a,b) as record from testData2) tmp
+ | GROUP BY a
""".stripMargin).select($"r.*"),
- Row(1, 2) :: Row(2, 2) :: Row(3, 2) :: Nil)
+ Row(1, 2) :: Row(2, 2) :: Row(3, 2) :: Nil)
- // With GROUP BY and alias and additional fields in the struct
- checkAnswer(sql(
- """
- | SELECT max(struct(a, record.*, b)) as r FROM
- | (select a as a, b as b, struct(a,b) as record from testData2) tmp
- | GROUP BY a
+ // With GROUP BY and alias and additional fields in the struct
+ checkAnswer(sql(
+ """
+ | SELECT max(struct(a, record.*, b)) as r FROM
+ | (select a as a, b as b, struct(a,b) as record from testData2) tmp
+ | GROUP BY a
""".stripMargin).select($"r.*"),
- Row(1, 1, 2, 2) :: Row(2, 2, 2, 2) :: Row(3, 3, 2, 2) :: Nil)
+ Row(1, 1, 2, 2) :: Row(2, 2, 2, 2) :: Row(3, 3, 2, 2) :: Nil)
- // Create a data set that contains nested structs.
- val nestedStructData = sql(
- """
- | SELECT struct(r1, r2) as record FROM
- | (SELECT struct(a, b) as r1, struct(b, a) as r2 FROM testData2) tmp
+ // Create a data set that contains nested structs.
+ val nestedStructData = sql(
+ """
+ | SELECT struct(r1, r2) as record FROM
+ | (SELECT struct(a, b) as r1, struct(b, a) as r2 FROM testData2) tmp
""".stripMargin)
- checkAnswer(nestedStructData.select($"record.*"),
- Row(Row(1, 1), Row(1, 1)) :: Row(Row(1, 2), Row(2, 1)) :: Row(Row(2, 1), Row(1, 2)) ::
- Row(Row(2, 2), Row(2, 2)) :: Row(Row(3, 1), Row(1, 3)) :: Row(Row(3, 2), Row(2, 3)) :: Nil)
- checkAnswer(nestedStructData.select($"record.r1"),
- Row(Row(1, 1)) :: Row(Row(1, 2)) :: Row(Row(2, 1)) :: Row(Row(2, 2)) ::
- Row(Row(3, 1)) :: Row(Row(3, 2)) :: Nil)
- checkAnswer(
- nestedStructData.select($"record.r1.*"),
- Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
-
- // Try with a temporary view
- withTempView("nestedStructTable") {
- nestedStructData.createOrReplaceTempView("nestedStructTable")
+ checkAnswer(nestedStructData.select($"record.*"),
+ Row(Row(1, 1), Row(1, 1)) :: Row(Row(1, 2), Row(2, 1)) :: Row(Row(2, 1), Row(1, 2)) ::
+ Row(Row(2, 2), Row(2, 2)) :: Row(Row(3, 1), Row(1, 3)) :: Row(Row(3, 2), Row(2, 3)) ::
+ Nil)
+ checkAnswer(nestedStructData.select($"record.r1"),
+ Row(Row(1, 1)) :: Row(Row(1, 2)) :: Row(Row(2, 1)) :: Row(Row(2, 2)) ::
+ Row(Row(3, 1)) :: Row(Row(3, 2)) :: Nil)
checkAnswer(
- sql("SELECT record.* FROM nestedStructTable"),
- nestedStructData.select($"record.*"))
- checkAnswer(
- sql("SELECT record.r1 FROM nestedStructTable"),
- nestedStructData.select($"record.r1"))
- checkAnswer(
- sql("SELECT record.r1.* FROM nestedStructTable"),
- nestedStructData.select($"record.r1.*"))
-
- // Try resolving something not there.
- assert(intercept[AnalysisException](sql("SELECT abc.* FROM nestedStructTable"))
- .getMessage.contains("cannot resolve"))
- }
+ nestedStructData.select($"record.r1.*"),
+ Row(1, 1) :: Row(1, 2) :: Row(2, 1) :: Row(2, 2) :: Row(3, 1) :: Row(3, 2) :: Nil)
- // Create paths with unusual characters
- withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
- val specialCharacterPath = sql(
- """
- | SELECT struct(`col$.a_`, `a.b.c.`) as `r&&b.c` FROM
- | (SELECT struct(a, b) as `col$.a_`, struct(b, a) as `a.b.c.` FROM testData2) tmp
- """.stripMargin)
- withTempView("specialCharacterTable") {
- specialCharacterPath.createOrReplaceTempView("specialCharacterTable")
+ // Try with a temporary view
+ withTempView("nestedStructTable") {
+ nestedStructData.createOrReplaceTempView("nestedStructTable")
checkAnswer(
- specialCharacterPath.select($"`r&&b.c`.*"),
+ sql("SELECT record.* FROM nestedStructTable"),
nestedStructData.select($"record.*"))
checkAnswer(
- sql(
- "SELECT `r&&b.c`.`col$.a_` FROM specialCharacterTable"),
- nestedStructData.select($"record.r1"))
- checkAnswer(
- sql("SELECT `r&&b.c`.`a.b.c.` FROM specialCharacterTable"),
- nestedStructData.select($"record.r2"))
+ sql("SELECT record.r1 FROM nestedStructTable"),
+ nestedStructData.select($"record.r1"))
checkAnswer(
- sql("SELECT `r&&b.c`.`col$.a_`.* FROM specialCharacterTable"),
+ sql("SELECT record.r1.* FROM nestedStructTable"),
nestedStructData.select($"record.r1.*"))
+
+ // Try resolving something not there.
+ assert(intercept[AnalysisException](sql("SELECT abc.* FROM nestedStructTable"))
+ .getMessage.contains("cannot resolve"))
}
- }
- // Try star expanding a scalar. This should fail.
- assert(intercept[AnalysisException](sql("select a.* from testData2")).getMessage.contains(
- "Can only star expand struct data types."))
+ // Create paths with unusual characters
+ withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
+ val specialCharacterPath = sql(
+ """
+ | SELECT struct(`col$.a_`, `a.b.c.`) as `r&&b.c` FROM
+ | (SELECT struct(a, b) as `col$.a_`, struct(b, a) as `a.b.c.` FROM testData2) tmp
+ """.stripMargin)
+ withTempView("specialCharacterTable") {
+ specialCharacterPath.createOrReplaceTempView("specialCharacterTable")
+ checkAnswer(
+ specialCharacterPath.select($"`r&&b.c`.*"),
+ nestedStructData.select($"record.*"))
+ checkAnswer(
+ sql(
+ "SELECT `r&&b.c`.`col$.a_` FROM specialCharacterTable"),
+ nestedStructData.select($"record.r1"))
+ checkAnswer(
+ sql("SELECT `r&&b.c`.`a.b.c.` FROM specialCharacterTable"),
+ nestedStructData.select($"record.r2"))
+ checkAnswer(
+ sql("SELECT `r&&b.c`.`col$.a_`.* FROM specialCharacterTable"),
+ nestedStructData.select($"record.r1.*"))
+ }
+ }
+
+ // Try star expanding a scalar. This should fail.
+ assert(intercept[AnalysisException](sql("select a.* from testData2")).getMessage.contains(
+ "Can only star expand struct data types."))
+ }
}
test("Struct Star Expansion - Name conflict") {
@@ -3129,38 +3172,40 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("string date comparison") {
- spark.range(1).selectExpr("date '2000-01-01' as d").createOrReplaceTempView("t1")
- val result = Date.valueOf("2000-01-01")
- checkAnswer(sql("select * from t1 where d < '2000'"), Nil)
- checkAnswer(sql("select * from t1 where d < '2001'"), Row(result))
- checkAnswer(sql("select * from t1 where d < '2000-01'"), Nil)
- checkAnswer(sql("select * from t1 where d < '2000-01-01'"), Nil)
- checkAnswer(sql("select * from t1 where d < '2000-1-1'"), Nil)
- checkAnswer(sql("select * from t1 where d <= '2000-1-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d <= '1999-12-30'"), Nil)
- checkAnswer(sql("select * from t1 where d = '2000-1-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d = '2000-01-01'"), Row(result))
- checkAnswer(sql("select * from t1 where d = '2000-1-02'"), Nil)
- checkAnswer(sql("select * from t1 where d > '2000-01-01'"), Nil)
- checkAnswer(sql("select * from t1 where d > '1999'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-1-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-1-01'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-01-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-01-01'"), Row(result))
- checkAnswer(sql("select * from t1 where d >= '2000-01-02'"), Nil)
- checkAnswer(sql("select * from t1 where '2000' >= d"), Row(result))
- checkAnswer(sql("select * from t1 where d > '2000-13'"), Nil)
-
- withSQLConf(SQLConf.LEGACY_CAST_DATETIME_TO_STRING.key -> "true") {
+ withTempView("t1") {
+ spark.range(1).selectExpr("date '2000-01-01' as d").createOrReplaceTempView("t1")
+ val result = Date.valueOf("2000-01-01")
checkAnswer(sql("select * from t1 where d < '2000'"), Nil)
checkAnswer(sql("select * from t1 where d < '2001'"), Row(result))
- checkAnswer(sql("select * from t1 where d < '2000-1-1'"), Row(result))
- checkAnswer(sql("select * from t1 where d <= '1999'"), Nil)
+ checkAnswer(sql("select * from t1 where d < '2000-01'"), Nil)
+ checkAnswer(sql("select * from t1 where d < '2000-01-01'"), Nil)
+ checkAnswer(sql("select * from t1 where d < '2000-1-1'"), Nil)
+ checkAnswer(sql("select * from t1 where d <= '2000-1-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d <= '1999-12-30'"), Nil)
+ checkAnswer(sql("select * from t1 where d = '2000-1-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d = '2000-01-01'"), Row(result))
+ checkAnswer(sql("select * from t1 where d = '2000-1-02'"), Nil)
+ checkAnswer(sql("select * from t1 where d > '2000-01-01'"), Nil)
+ checkAnswer(sql("select * from t1 where d > '1999'"), Row(result))
checkAnswer(sql("select * from t1 where d >= '2000'"), Row(result))
- checkAnswer(sql("select * from t1 where d > '1999-13'"), Row(result))
- checkAnswer(sql("select to_date('2000-01-01') > '1'"), Row(true))
+ checkAnswer(sql("select * from t1 where d >= '2000-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d >= '2000-1-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d >= '2000-1-01'"), Row(result))
+ checkAnswer(sql("select * from t1 where d >= '2000-01-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d >= '2000-01-01'"), Row(result))
+ checkAnswer(sql("select * from t1 where d >= '2000-01-02'"), Nil)
+ checkAnswer(sql("select * from t1 where '2000' >= d"), Row(result))
+ checkAnswer(sql("select * from t1 where d > '2000-13'"), Nil)
+
+ withSQLConf(SQLConf.LEGACY_CAST_DATETIME_TO_STRING.key -> "true") {
+ checkAnswer(sql("select * from t1 where d < '2000'"), Nil)
+ checkAnswer(sql("select * from t1 where d < '2001'"), Row(result))
+ checkAnswer(sql("select * from t1 where d < '2000-1-1'"), Row(result))
+ checkAnswer(sql("select * from t1 where d <= '1999'"), Nil)
+ checkAnswer(sql("select * from t1 where d >= '2000'"), Row(result))
+ checkAnswer(sql("select * from t1 where d > '1999-13'"), Row(result))
+ checkAnswer(sql("select to_date('2000-01-01') > '1'"), Row(true))
+ }
}
}
@@ -3210,28 +3255,30 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
test("SPARK-28156: self-join should not miss cached view") {
withTable("table1") {
withView("table1_vw") {
- val df = Seq.tabulate(5) { x => (x, x + 1, x + 2, x + 3) }.toDF("a", "b", "c", "d")
- df.write.mode("overwrite").format("orc").saveAsTable("table1")
- sql("drop view if exists table1_vw")
- sql("create view table1_vw as select * from table1")
-
- val cachedView = sql("select a, b, c, d from table1_vw")
-
- cachedView.createOrReplaceTempView("cachedview")
- cachedView.persist()
-
- val queryDf = sql(
- s"""select leftside.a, leftside.b
- |from cachedview leftside
- |join cachedview rightside
- |on leftside.a = rightside.a
+ withTempView("cachedview") {
+ val df = Seq.tabulate(5) { x => (x, x + 1, x + 2, x + 3) }.toDF("a", "b", "c", "d")
+ df.write.mode("overwrite").format("orc").saveAsTable("table1")
+ sql("drop view if exists table1_vw")
+ sql("create view table1_vw as select * from table1")
+
+ val cachedView = sql("select a, b, c, d from table1_vw")
+
+ cachedView.createOrReplaceTempView("cachedview")
+ cachedView.persist()
+
+ val queryDf = sql(
+ s"""select leftside.a, leftside.b
+ |from cachedview leftside
+ |join cachedview rightside
+ |on leftside.a = rightside.a
""".stripMargin)
- val inMemoryTableScan = collect(queryDf.queryExecution.executedPlan) {
- case i: InMemoryTableScanExec => i
+ val inMemoryTableScan = collect(queryDf.queryExecution.executedPlan) {
+ case i: InMemoryTableScanExec => i
+ }
+ assert(inMemoryTableScan.size == 2)
+ checkAnswer(queryDf, Row(0, 1) :: Row(1, 2) :: Row(2, 3) :: Row(3, 4) :: Row(4, 5) :: Nil)
}
- assert(inMemoryTableScan.size == 2)
- checkAnswer(queryDf, Row(0, 1) :: Row(1, 2) :: Row(2, 3) :: Row(3, 4) :: Row(4, 5) :: Nil)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ScalaReflectionRelationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ScalaReflectionRelationSuite.scala
index d51a461..66c6fbe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ScalaReflectionRelationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ScalaReflectionRelationSuite.scala
@@ -83,69 +83,79 @@ class ScalaReflectionRelationSuite extends SparkFunSuite with SharedSparkSession
private case class InvalidInJava2(`0`: Int)
test("query case class RDD") {
- val data = ReflectData("a", 1, 1L, 1.toFloat, 1.toDouble, 1.toShort, 1.toByte, true,
- new java.math.BigDecimal(1), Date.valueOf("1970-01-01"), new Timestamp(12345), Seq(1, 2, 3),
- new java.math.BigInteger("1"), scala.math.BigInt(1))
- Seq(data).toDF().createOrReplaceTempView("reflectData")
-
- assert(sql("SELECT * FROM reflectData").collect().head ===
- Row("a", 1, 1L, 1.toFloat, 1.toDouble, 1.toShort, 1.toByte, true,
- new java.math.BigDecimal(1), Date.valueOf("1970-01-01"),
- new Timestamp(12345), Seq(1, 2, 3), new java.math.BigDecimal(1),
- new java.math.BigDecimal(1)))
+ withTempView("reflectData") {
+ val data = ReflectData("a", 1, 1L, 1.toFloat, 1.toDouble, 1.toShort, 1.toByte, true,
+ new java.math.BigDecimal(1), Date.valueOf("1970-01-01"), new Timestamp(12345), Seq(1, 2, 3),
+ new java.math.BigInteger("1"), scala.math.BigInt(1))
+ Seq(data).toDF().createOrReplaceTempView("reflectData")
+
+ assert(sql("SELECT * FROM reflectData").collect().head ===
+ Row("a", 1, 1L, 1.toFloat, 1.toDouble, 1.toShort, 1.toByte, true,
+ new java.math.BigDecimal(1), Date.valueOf("1970-01-01"),
+ new Timestamp(12345), Seq(1, 2, 3), new java.math.BigDecimal(1),
+ new java.math.BigDecimal(1)))
+ }
}
test("query case class RDD with nulls") {
- val data = NullReflectData(null, null, null, null, null, null, null)
- Seq(data).toDF().createOrReplaceTempView("reflectNullData")
+ withTempView("reflectNullData") {
+ val data = NullReflectData(null, null, null, null, null, null, null)
+ Seq(data).toDF().createOrReplaceTempView("reflectNullData")
- assert(sql("SELECT * FROM reflectNullData").collect().head ===
- Row.fromSeq(Seq.fill(7)(null)))
+ assert(sql("SELECT * FROM reflectNullData").collect().head ===
+ Row.fromSeq(Seq.fill(7)(null)))
+ }
}
test("query case class RDD with Nones") {
- val data = OptionalReflectData(None, None, None, None, None, None, None)
- Seq(data).toDF().createOrReplaceTempView("reflectOptionalData")
+ withTempView("reflectOptionalData") {
+ val data = OptionalReflectData(None, None, None, None, None, None, None)
+ Seq(data).toDF().createOrReplaceTempView("reflectOptionalData")
- assert(sql("SELECT * FROM reflectOptionalData").collect().head ===
- Row.fromSeq(Seq.fill(7)(null)))
+ assert(sql("SELECT * FROM reflectOptionalData").collect().head ===
+ Row.fromSeq(Seq.fill(7)(null)))
+ }
}
// Equality is broken for Arrays, so we test that separately.
test("query binary data") {
- Seq(ReflectBinary(Array[Byte](1))).toDF().createOrReplaceTempView("reflectBinary")
+ withTempView("reflectBinary") {
+ Seq(ReflectBinary(Array[Byte](1))).toDF().createOrReplaceTempView("reflectBinary")
- val result = sql("SELECT data FROM reflectBinary")
- .collect().head(0).asInstanceOf[Array[Byte]]
- assert(result.toSeq === Seq[Byte](1))
+ val result = sql("SELECT data FROM reflectBinary")
+ .collect().head(0).asInstanceOf[Array[Byte]]
+ assert(result.toSeq === Seq[Byte](1))
+ }
}
test("query complex data") {
- val data = ComplexReflectData(
- Seq(1, 2, 3),
- Seq(Some(1), Some(2), None),
- Map(1 -> 10L, 2 -> 20L),
- Map(1 -> Some(10L), 2 -> Some(20L), 3 -> None),
- Data(
- Seq(10, 20, 30),
- Seq(Some(10), Some(20), None),
- Map(10 -> 100L, 20 -> 200L),
- Map(10 -> Some(100L), 20 -> Some(200L), 30 -> None),
- Nested(None, "abc")))
-
- Seq(data).toDF().createOrReplaceTempView("reflectComplexData")
- assert(sql("SELECT * FROM reflectComplexData").collect().head ===
- Row(
+ withTempView("reflectComplexData") {
+ val data = ComplexReflectData(
Seq(1, 2, 3),
- Seq(1, 2, null),
+ Seq(Some(1), Some(2), None),
Map(1 -> 10L, 2 -> 20L),
- Map(1 -> 10L, 2 -> 20L, 3 -> null),
- Row(
+ Map(1 -> Some(10L), 2 -> Some(20L), 3 -> None),
+ Data(
Seq(10, 20, 30),
- Seq(10, 20, null),
+ Seq(Some(10), Some(20), None),
Map(10 -> 100L, 20 -> 200L),
- Map(10 -> 100L, 20 -> 200L, 30 -> null),
- Row(null, "abc"))))
+ Map(10 -> Some(100L), 20 -> Some(200L), 30 -> None),
+ Nested(None, "abc")))
+
+ Seq(data).toDF().createOrReplaceTempView("reflectComplexData")
+ assert(sql("SELECT * FROM reflectComplexData").collect().head ===
+ Row(
+ Seq(1, 2, 3),
+ Seq(1, 2, null),
+ Map(1 -> 10L, 2 -> 20L),
+ Map(1 -> 10L, 2 -> 20L, 3 -> null),
+ Row(
+ Seq(10, 20, 30),
+ Seq(10, 20, null),
+ Map(10 -> 100L, 20 -> 200L),
+ Map(10 -> 100L, 20 -> 200L, 30 -> null),
+ Row(null, "abc"))))
+ }
}
test("better error message when use java reserved keyword as field name") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
index 760b03d..141f127 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -153,29 +153,31 @@ class SubquerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
test("uncorrelated scalar subquery on a DataFrame generated query") {
- val df = Seq((1, "one"), (2, "two"), (3, "three")).toDF("key", "value")
- df.createOrReplaceTempView("subqueryData")
+ withTempView("subqueryData") {
+ val df = Seq((1, "one"), (2, "two"), (3, "three")).toDF("key", "value")
+ df.createOrReplaceTempView("subqueryData")
- checkAnswer(
- sql("select (select key from subqueryData where key > 2 order by key limit 1) + 1"),
- Array(Row(4))
- )
+ checkAnswer(
+ sql("select (select key from subqueryData where key > 2 order by key limit 1) + 1"),
+ Array(Row(4))
+ )
- checkAnswer(
- sql("select -(select max(key) from subqueryData)"),
- Array(Row(-3))
- )
+ checkAnswer(
+ sql("select -(select max(key) from subqueryData)"),
+ Array(Row(-3))
+ )
- checkAnswer(
- sql("select (select value from subqueryData limit 0)"),
- Array(Row(null))
- )
+ checkAnswer(
+ sql("select (select value from subqueryData limit 0)"),
+ Array(Row(null))
+ )
- checkAnswer(
- sql("select (select min(value) from subqueryData" +
- " where key = (select max(key) from subqueryData) - 1)"),
- Array(Row("two"))
- )
+ checkAnswer(
+ sql("select (select min(value) from subqueryData" +
+ " where key = (select max(key) from subqueryData) - 1)"),
+ Array(Row("two"))
+ )
+ }
}
test("SPARK-15677: Queries against local relations with scalar subquery in Select list") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
index 157610f5..ed8ab1c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala
@@ -150,12 +150,14 @@ class UserDefinedTypeSuite extends QueryTest with SharedSparkSession with Parque
}
test("UDTs and UDFs") {
- spark.udf.register("testType",
- (d: TestUDT.MyDenseVector) => d.isInstanceOf[TestUDT.MyDenseVector])
- pointsRDD.createOrReplaceTempView("points")
- checkAnswer(
- sql("SELECT testType(features) from points"),
- Seq(Row(true), Row(true)))
+ withTempView("points") {
+ spark.udf.register("testType",
+ (d: TestUDT.MyDenseVector) => d.isInstanceOf[TestUDT.MyDenseVector])
+ pointsRDD.createOrReplaceTempView("points")
+ checkAnswer(
+ sql("SELECT testType(features) from points"),
+ Seq(Row(true), Row(true)))
+ }
}
testStandardAndLegacyModes("UDTs with Parquet") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
index 7aabf4d..67ec102 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
@@ -41,67 +41,70 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 10)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql(
- """
- |select area, sum(product), sum(sum(product)) over (partition by area)
- |from windowData group by month, area
- """.stripMargin),
- Seq(
- ("a", 5, 11),
- ("a", 6, 11),
- ("b", 7, 15),
- ("b", 8, 15),
- ("c", 9, 19),
- ("c", 10, 19)
- ).map(i => Row(i._1, i._2, i._3)))
-
- checkAnswer(
- sql(
- """
- |select area, sum(product) - 1, sum(sum(product)) over (partition by area)
- |from windowData group by month, area
- """.stripMargin),
- Seq(
- ("a", 4, 11),
- ("a", 5, 11),
- ("b", 6, 15),
- ("b", 7, 15),
- ("c", 8, 19),
- ("c", 9, 19)
- ).map(i => Row(i._1, i._2, i._3)))
-
- checkAnswer(
- sql(
- """
- |select area, sum(product), sum(product) / sum(sum(product)) over (partition by area)
- |from windowData group by month, area
- """.stripMargin),
- Seq(
- ("a", 5, 5d/11),
- ("a", 6, 6d/11),
- ("b", 7, 7d/15),
- ("b", 8, 8d/15),
- ("c", 10, 10d/19),
- ("c", 9, 9d/19)
- ).map(i => Row(i._1, i._2, i._3)))
-
- checkAnswer(
- sql(
- """
- |select area, sum(product), sum(product) / sum(sum(product) - 1) over (partition by area)
- |from windowData group by month, area
- """.stripMargin),
- Seq(
- ("a", 5, 5d/9),
- ("a", 6, 6d/9),
- ("b", 7, 7d/13),
- ("b", 8, 8d/13),
- ("c", 10, 10d/17),
- ("c", 9, 9d/17)
- ).map(i => Row(i._1, i._2, i._3)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product), sum(sum(product)) over (partition by area)
+ |from windowData group by month, area
+ """.stripMargin),
+ Seq(
+ ("a", 5, 11),
+ ("a", 6, 11),
+ ("b", 7, 15),
+ ("b", 8, 15),
+ ("c", 9, 19),
+ ("c", 10, 19)
+ ).map(i => Row(i._1, i._2, i._3)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product) - 1, sum(sum(product)) over (partition by area)
+ |from windowData group by month, area
+ """.stripMargin),
+ Seq(
+ ("a", 4, 11),
+ ("a", 5, 11),
+ ("b", 6, 15),
+ ("b", 7, 15),
+ ("c", 8, 19),
+ ("c", 9, 19)
+ ).map(i => Row(i._1, i._2, i._3)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product), sum(product) / sum(sum(product)) over (partition by area)
+ |from windowData group by month, area
+ """.stripMargin),
+ Seq(
+ ("a", 5, 5d/11),
+ ("a", 6, 6d/11),
+ ("b", 7, 7d/15),
+ ("b", 8, 8d/15),
+ ("c", 10, 10d/19),
+ ("c", 9, 9d/19)
+ ).map(i => Row(i._1, i._2, i._3)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product), sum(product) / sum(sum(product) - 1) over
+ |(partition by area)
+ |from windowData group by month, area
+ """.stripMargin),
+ Seq(
+ ("a", 5, 5d/9),
+ ("a", 6, 6d/9),
+ ("b", 7, 7d/13),
+ ("b", 8, 8d/13),
+ ("c", 10, 10d/17),
+ ("c", 9, 9d/17)
+ ).map(i => Row(i._1, i._2, i._3)))
+ }
}
test("window function: refer column in inner select block") {
@@ -113,22 +116,24 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 10)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql(
- """
- |select area, rank() over (partition by area order by tmp.month) + tmp.tmp1 as c1
- |from (select month, area, product, 1 as tmp1 from windowData) tmp
- """.stripMargin),
- Seq(
- ("a", 2),
- ("a", 3),
- ("b", 2),
- ("b", 3),
- ("c", 2),
- ("c", 3)
- ).map(i => Row(i._1, i._2)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql(
+ """
+ |select area, rank() over (partition by area order by tmp.month) + tmp.tmp1 as c1
+ |from (select month, area, product, 1 as tmp1 from windowData) tmp
+ """.stripMargin),
+ Seq(
+ ("a", 2),
+ ("a", 3),
+ ("b", 2),
+ ("b", 3),
+ ("c", 2),
+ ("c", 3)
+ ).map(i => Row(i._1, i._2)))
+ }
}
test("window function: partition and order expressions") {
@@ -140,38 +145,40 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 10)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql(
- """
- |select month, area, product, sum(product + 1) over (partition by 1 order by 2)
- |from windowData
- """.stripMargin),
- Seq(
- (1, "a", 5, 51),
- (2, "a", 6, 51),
- (3, "b", 7, 51),
- (4, "b", 8, 51),
- (5, "c", 9, 51),
- (6, "c", 10, 51)
- ).map(i => Row(i._1, i._2, i._3, i._4)))
-
- checkAnswer(
- sql(
- """
- |select month, area, product, sum(product)
- |over (partition by month % 2 order by 10 - product)
- |from windowData
- """.stripMargin),
- Seq(
- (1, "a", 5, 21),
- (2, "a", 6, 24),
- (3, "b", 7, 16),
- (4, "b", 8, 18),
- (5, "c", 9, 9),
- (6, "c", 10, 10)
- ).map(i => Row(i._1, i._2, i._3, i._4)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql(
+ """
+ |select month, area, product, sum(product + 1) over (partition by 1 order by 2)
+ |from windowData
+ """.stripMargin),
+ Seq(
+ (1, "a", 5, 51),
+ (2, "a", 6, 51),
+ (3, "b", 7, 51),
+ (4, "b", 8, 51),
+ (5, "c", 9, 51),
+ (6, "c", 10, 51)
+ ).map(i => Row(i._1, i._2, i._3, i._4)))
+
+ checkAnswer(
+ sql(
+ """
+ |select month, area, product, sum(product)
+ |over (partition by month % 2 order by 10 - product)
+ |from windowData
+ """.stripMargin),
+ Seq(
+ (1, "a", 5, 21),
+ (2, "a", 6, 24),
+ (3, "b", 7, 16),
+ (4, "b", 8, 18),
+ (5, "c", 9, 9),
+ (6, "c", 10, 10)
+ ).map(i => Row(i._1, i._2, i._3, i._4)))
+ }
}
test("window function: distinct should not be silently ignored") {
@@ -183,16 +190,18 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 10)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- val e = intercept[AnalysisException] {
- sql(
- """
- |select month, area, product, sum(distinct product + 1) over (partition by 1 order by 2)
- |from windowData
- """.stripMargin)
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ val e = intercept[AnalysisException] {
+ sql(
+ """
+ |select month, area, product, sum(distinct product + 1) over (partition by 1 order by 2)
+ |from windowData
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains("Distinct window functions are not supported"))
}
- assert(e.getMessage.contains("Distinct window functions are not supported"))
}
test("window function: expressions in arguments of a window functions") {
@@ -204,23 +213,25 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 10)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql(
- """
- |select month, area, month % 2,
- |lag(product, 1 + 1, product) over (partition by month % 2 order by area)
- |from windowData
- """.stripMargin),
- Seq(
- (1, "a", 1, 5),
- (2, "a", 0, 6),
- (3, "b", 1, 7),
- (4, "b", 0, 8),
- (5, "c", 1, 5),
- (6, "c", 0, 6)
- ).map(i => Row(i._1, i._2, i._3, i._4)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql(
+ """
+ |select month, area, month % 2,
+ |lag(product, 1 + 1, product) over (partition by month % 2 order by area)
+ |from windowData
+ """.stripMargin),
+ Seq(
+ (1, "a", 1, 5),
+ (2, "a", 0, 6),
+ (3, "b", 1, 7),
+ (4, "b", 0, 8),
+ (5, "c", 1, 5),
+ (6, "c", 0, 6)
+ ).map(i => Row(i._1, i._2, i._3, i._4)))
+ }
}
@@ -233,63 +244,65 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 11)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql("select month, product, sum(product + 1) over() from windowData order by area"),
- Seq(
- (2, 6, 57),
- (3, 7, 57),
- (4, 8, 57),
- (5, 9, 57),
- (6, 11, 57),
- (1, 10, 57)
- ).map(i => Row(i._1, i._2, i._3)))
-
- checkAnswer(
- sql(
- """
- |select area, rank() over (partition by area order by tmp.month) + tmp.tmp1 as c1
- |from (select month, area, product as p, 1 as tmp1 from windowData) tmp order by p
- """.stripMargin),
- Seq(
- ("a", 2),
- ("b", 2),
- ("b", 3),
- ("c", 2),
- ("d", 2),
- ("c", 3)
- ).map(i => Row(i._1, i._2)))
-
- checkAnswer(
- sql(
- """
- |select area, rank() over (partition by area order by month) as c1
- |from windowData group by product, area, month order by product, area
- """.stripMargin),
- Seq(
- ("a", 1),
- ("b", 1),
- ("b", 2),
- ("c", 1),
- ("d", 1),
- ("c", 2)
- ).map(i => Row(i._1, i._2)))
-
- checkAnswer(
- sql(
- """
- |select area, sum(product) / sum(sum(product)) over (partition by area) as c1
- |from windowData group by area, month order by month, c1
- """.stripMargin),
- Seq(
- ("d", 1.0),
- ("a", 1.0),
- ("b", 0.4666666666666667),
- ("b", 0.5333333333333333),
- ("c", 0.45),
- ("c", 0.55)
- ).map(i => Row(i._1, i._2)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql("select month, product, sum(product + 1) over() from windowData order by area"),
+ Seq(
+ (2, 6, 57),
+ (3, 7, 57),
+ (4, 8, 57),
+ (5, 9, 57),
+ (6, 11, 57),
+ (1, 10, 57)
+ ).map(i => Row(i._1, i._2, i._3)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, rank() over (partition by area order by tmp.month) + tmp.tmp1 as c1
+ |from (select month, area, product as p, 1 as tmp1 from windowData) tmp order by p
+ """.stripMargin),
+ Seq(
+ ("a", 2),
+ ("b", 2),
+ ("b", 3),
+ ("c", 2),
+ ("d", 2),
+ ("c", 3)
+ ).map(i => Row(i._1, i._2)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, rank() over (partition by area order by month) as c1
+ |from windowData group by product, area, month order by product, area
+ """.stripMargin),
+ Seq(
+ ("a", 1),
+ ("b", 1),
+ ("b", 2),
+ ("c", 1),
+ ("d", 1),
+ ("c", 2)
+ ).map(i => Row(i._1, i._2)))
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product) / sum(sum(product)) over (partition by area) as c1
+ |from windowData group by area, month order by month, c1
+ """.stripMargin),
+ Seq(
+ ("d", 1.0),
+ ("a", 1.0),
+ ("b", 0.4666666666666667),
+ ("b", 0.5333333333333333),
+ ("c", 0.45),
+ ("c", 0.55)
+ ).map(i => Row(i._1, i._2)))
+ }
}
// todo: fix this test case by reimplementing the function ResolveAggregateFunctions
@@ -302,23 +315,25 @@ class SQLWindowFunctionSuite extends QueryTest with SharedSparkSession {
WindowData(5, "c", 9),
WindowData(6, "c", 11)
)
- sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
-
- checkAnswer(
- sql(
- """
- |select area, sum(product) over () as c from windowData
- |where product > 3 group by area, product
- |having avg(month) > 0 order by avg(month), product
- """.stripMargin),
- Seq(
- ("a", 51),
- ("b", 51),
- ("b", 51),
- ("c", 51),
- ("c", 51),
- ("d", 51)
- ).map(i => Row(i._1, i._2)))
+ withTempView("windowData") {
+ sparkContext.parallelize(data).toDF().createOrReplaceTempView("windowData")
+
+ checkAnswer(
+ sql(
+ """
+ |select area, sum(product) over () as c from windowData
+ |where product > 3 group by area, product
+ |having avg(month) > 0 order by avg(month), product
+ """.stripMargin),
+ Seq(
+ ("a", 51),
+ ("b", 51),
+ ("b", 51),
+ ("c", 51),
+ ("c", 51),
+ ("d", 51)
+ ).map(i => Row(i._1, i._2)))
+ }
}
test("window function: multiple window expressions in a single expression") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
index 77047f3..18f29f7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
@@ -126,13 +126,15 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSparkSession {
}
test("default size avoids broadcast") {
- // TODO: Improve this test when we have better statistics
- sparkContext.parallelize(1 to 10).map(i => TestData(i, i.toString))
- .toDF().createOrReplaceTempView("sizeTst")
- spark.catalog.cacheTable("sizeTst")
- assert(
- spark.table("sizeTst").queryExecution.analyzed.stats.sizeInBytes >
- spark.conf.get(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD))
+ withTempView("sizeTst") {
+ // TODO: Improve this test when we have better statistics
+ sparkContext.parallelize(1 to 10).map(i => TestData(i, i.toString))
+ .toDF().createOrReplaceTempView("sizeTst")
+ spark.catalog.cacheTable("sizeTst")
+ assert(
+ spark.table("sizeTst").queryExecution.analyzed.stats.sizeInBytes >
+ spark.conf.get(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD))
+ }
}
test("projection") {
@@ -187,18 +189,20 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSparkSession {
}
test("SPARK-2729 regression: timestamp data type") {
- val timestamps = (0 to 3).map(i => Tuple1(new Timestamp(i))).toDF("time")
- timestamps.createOrReplaceTempView("timestamps")
+ withTempView("timestamps") {
+ val timestamps = (0 to 3).map(i => Tuple1(new Timestamp(i))).toDF("time")
+ timestamps.createOrReplaceTempView("timestamps")
- checkAnswer(
- sql("SELECT time FROM timestamps"),
- timestamps.collect().toSeq)
+ checkAnswer(
+ sql("SELECT time FROM timestamps"),
+ timestamps.collect().toSeq)
- spark.catalog.cacheTable("timestamps")
+ spark.catalog.cacheTable("timestamps")
- checkAnswer(
- sql("SELECT time FROM timestamps"),
- timestamps.collect().toSeq)
+ checkAnswer(
+ sql("SELECT time FROM timestamps"),
+ timestamps.collect().toSeq)
+ }
}
test("SPARK-3320 regression: batched column buffer building should work with empty partitions") {
@@ -229,10 +233,12 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSparkSession {
assert(df.schema.head.dataType === DecimalType(15, 10))
- df.cache().createOrReplaceTempView("test_fixed_decimal")
- checkAnswer(
- sql("SELECT * FROM test_fixed_decimal"),
- (1 to 10).map(i => Row(Decimal(i, 15, 10).toJavaBigDecimal)))
+ withTempView("test_fixed_decimal") {
+ df.cache().createOrReplaceTempView("test_fixed_decimal")
+ checkAnswer(
+ sql("SELECT * FROM test_fixed_decimal"),
+ (1 to 10).map(i => Row(Decimal(i, 15, 10).toJavaBigDecimal)))
+ }
}
test("test different data types") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index be95f33..999eadb 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -249,495 +249,523 @@ abstract class JsonSuite extends QueryTest with SharedSparkSession with TestJson
}
test("Complex field and type inferring with null in sampling") {
- val jsonDF = spark.read.json(jsonNullStruct)
- val expectedSchema = StructType(
- StructField("headers", StructType(
- StructField("Charset", StringType, true) ::
- StructField("Host", StringType, true) :: Nil)
- , true) ::
- StructField("ip", StringType, true) ::
- StructField("nullstr", StringType, true):: Nil)
-
- assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(jsonNullStruct)
+ val expectedSchema = StructType(
+ StructField("headers", StructType(
+ StructField("Charset", StringType, true) ::
+ StructField("Host", StringType, true) :: Nil)
+ , true) ::
+ StructField("ip", StringType, true) ::
+ StructField("nullstr", StringType, true):: Nil)
+
+ assert(expectedSchema === jsonDF.schema)
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select nullstr, headers.Host from jsonTable"),
- Seq(Row("", "1.abc.com"), Row("", null), Row("", null), Row(null, null))
- )
+ checkAnswer(
+ sql("select nullstr, headers.Host from jsonTable"),
+ Seq(Row("", "1.abc.com"), Row("", null), Row("", null), Row(null, null))
+ )
+ }
}
test("Primitive field and type inferring") {
- val jsonDF = spark.read.json(primitiveFieldAndType)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(primitiveFieldAndType)
- val expectedSchema = StructType(
- StructField("bigInteger", DecimalType(20, 0), true) ::
- StructField("boolean", BooleanType, true) ::
- StructField("double", DoubleType, true) ::
- StructField("integer", LongType, true) ::
- StructField("long", LongType, true) ::
- StructField("null", StringType, true) ::
- StructField("string", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("bigInteger", DecimalType(20, 0), true) ::
+ StructField("boolean", BooleanType, true) ::
+ StructField("double", DoubleType, true) ::
+ StructField("integer", LongType, true) ::
+ StructField("long", LongType, true) ::
+ StructField("null", StringType, true) ::
+ StructField("string", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row(new java.math.BigDecimal("92233720368547758070"),
- true,
- 1.7976931348623157,
- 10,
- 21474836470L,
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row(new java.math.BigDecimal("92233720368547758070"),
+ true,
+ 1.7976931348623157,
+ 10,
+ 21474836470L,
+ null,
+ "this is a simple string.")
+ )
+ }
}
test("Complex field and type inferring") {
- val jsonDF = spark.read.json(complexFieldAndType1)
-
- val expectedSchema = StructType(
- StructField("arrayOfArray1", ArrayType(ArrayType(StringType, true), true), true) ::
- StructField("arrayOfArray2", ArrayType(ArrayType(DoubleType, true), true), true) ::
- StructField("arrayOfBigInteger", ArrayType(DecimalType(21, 0), true), true) ::
- StructField("arrayOfBoolean", ArrayType(BooleanType, true), true) ::
- StructField("arrayOfDouble", ArrayType(DoubleType, true), true) ::
- StructField("arrayOfInteger", ArrayType(LongType, true), true) ::
- StructField("arrayOfLong", ArrayType(LongType, true), true) ::
- StructField("arrayOfNull", ArrayType(StringType, true), true) ::
- StructField("arrayOfString", ArrayType(StringType, true), true) ::
- StructField("arrayOfStruct", ArrayType(
- StructType(
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(complexFieldAndType1)
+
+ val expectedSchema = StructType(
+ StructField("arrayOfArray1", ArrayType(ArrayType(StringType, true), true), true) ::
+ StructField("arrayOfArray2", ArrayType(ArrayType(DoubleType, true), true), true) ::
+ StructField("arrayOfBigInteger", ArrayType(DecimalType(21, 0), true), true) ::
+ StructField("arrayOfBoolean", ArrayType(BooleanType, true), true) ::
+ StructField("arrayOfDouble", ArrayType(DoubleType, true), true) ::
+ StructField("arrayOfInteger", ArrayType(LongType, true), true) ::
+ StructField("arrayOfLong", ArrayType(LongType, true), true) ::
+ StructField("arrayOfNull", ArrayType(StringType, true), true) ::
+ StructField("arrayOfString", ArrayType(StringType, true), true) ::
+ StructField("arrayOfStruct", ArrayType(
+ StructType(
+ StructField("field1", BooleanType, true) ::
+ StructField("field2", StringType, true) ::
+ StructField("field3", StringType, true) :: Nil), true), true) ::
+ StructField("struct", StructType(
StructField("field1", BooleanType, true) ::
- StructField("field2", StringType, true) ::
- StructField("field3", StringType, true) :: Nil), true), true) ::
- StructField("struct", StructType(
- StructField("field1", BooleanType, true) ::
- StructField("field2", DecimalType(20, 0), true) :: Nil), true) ::
- StructField("structWithArrayFields", StructType(
- StructField("field1", ArrayType(LongType, true), true) ::
- StructField("field2", ArrayType(StringType, true), true) :: Nil), true) :: Nil)
+ StructField("field2", DecimalType(20, 0), true) :: Nil), true) ::
+ StructField("structWithArrayFields", StructType(
+ StructField("field1", ArrayType(LongType, true), true) ::
+ StructField("field2", ArrayType(StringType, true), true) :: Nil), true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- // Access elements of a primitive array.
- checkAnswer(
- sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from jsonTable"),
- Row("str1", "str2", null)
- )
+ // Access elements of a primitive array.
+ checkAnswer(
+ sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from jsonTable"),
+ Row("str1", "str2", null)
+ )
- // Access an array of null values.
- checkAnswer(
- sql("select arrayOfNull from jsonTable"),
- Row(Seq(null, null, null, null))
- )
+ // Access an array of null values.
+ checkAnswer(
+ sql("select arrayOfNull from jsonTable"),
+ Row(Seq(null, null, null, null))
+ )
- // Access elements of a BigInteger array (we use DecimalType internally).
- checkAnswer(
- sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] from jsonTable"),
- Row(new java.math.BigDecimal("922337203685477580700"),
- new java.math.BigDecimal("-922337203685477580800"), null)
- )
+ // Access elements of a BigInteger array (we use DecimalType internally).
+ checkAnswer(
+ sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] from " +
+ "jsonTable"),
+ Row(new java.math.BigDecimal("922337203685477580700"),
+ new java.math.BigDecimal("-922337203685477580800"), null)
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray1[0], arrayOfArray1[1] from jsonTable"),
- Row(Seq("1", "2", "3"), Seq("str1", "str2"))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray1[0], arrayOfArray1[1] from jsonTable"),
+ Row(Seq("1", "2", "3"), Seq("str1", "str2"))
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray2[0], arrayOfArray2[1] from jsonTable"),
- Row(Seq(1.0, 2.0, 3.0), Seq(1.1, 2.1, 3.1))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray2[0], arrayOfArray2[1] from jsonTable"),
+ Row(Seq(1.0, 2.0, 3.0), Seq(1.1, 2.1, 3.1))
+ )
- // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
- checkAnswer(
- sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from jsonTable"),
- Row("str2", 2.1)
- )
+ // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
+ checkAnswer(
+ sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from jsonTable"),
+ Row("str2", 2.1)
+ )
- // Access elements of an array of structs.
- checkAnswer(
- sql("select arrayOfStruct[0], arrayOfStruct[1], arrayOfStruct[2], arrayOfStruct[3] " +
- "from jsonTable"),
- Row(
- Row(true, "str1", null),
- Row(false, null, null),
- Row(null, null, null),
- null)
- )
+ // Access elements of an array of structs.
+ checkAnswer(
+ sql("select arrayOfStruct[0], arrayOfStruct[1], arrayOfStruct[2], arrayOfStruct[3] " +
+ "from jsonTable"),
+ Row(
+ Row(true, "str1", null),
+ Row(false, null, null),
+ Row(null, null, null),
+ null)
+ )
- // Access a struct and fields inside of it.
- checkAnswer(
- sql("select struct, struct.field1, struct.field2 from jsonTable"),
- Row(
- Row(true, new java.math.BigDecimal("92233720368547758070")),
- true,
- new java.math.BigDecimal("92233720368547758070")) :: Nil
- )
+ // Access a struct and fields inside of it.
+ checkAnswer(
+ sql("select struct, struct.field1, struct.field2 from jsonTable"),
+ Row(
+ Row(true, new java.math.BigDecimal("92233720368547758070")),
+ true,
+ new java.math.BigDecimal("92233720368547758070")) :: Nil
+ )
- // Access an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1, structWithArrayFields.field2 from jsonTable"),
- Row(Seq(4, 5, 6), Seq("str1", "str2"))
- )
+ // Access an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1, structWithArrayFields.field2 from jsonTable"),
+ Row(Seq(4, 5, 6), Seq("str1", "str2"))
+ )
- // Access elements of an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] from jsonTable"),
- Row(5, null)
- )
+ // Access elements of an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] from " +
+ "jsonTable"),
+ Row(5, null)
+ )
+ }
}
test("GetField operation on complex data type") {
- val jsonDF = spark.read.json(complexFieldAndType1)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(complexFieldAndType1)
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select arrayOfStruct[0].field1, arrayOfStruct[0].field2 from jsonTable"),
- Row(true, "str1")
- )
+ checkAnswer(
+ sql("select arrayOfStruct[0].field1, arrayOfStruct[0].field2 from jsonTable"),
+ Row(true, "str1")
+ )
- // Getting all values of a specific field from an array of structs.
- checkAnswer(
- sql("select arrayOfStruct.field1, arrayOfStruct.field2 from jsonTable"),
- Row(Seq(true, false, null), Seq("str1", null, null))
- )
+ // Getting all values of a specific field from an array of structs.
+ checkAnswer(
+ sql("select arrayOfStruct.field1, arrayOfStruct.field2 from jsonTable"),
+ Row(Seq(true, false, null), Seq("str1", null, null))
+ )
+ }
}
test("Type conflict in primitive field values") {
- val jsonDF = spark.read.json(primitiveFieldValueTypeConflict)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(primitiveFieldValueTypeConflict)
- val expectedSchema = StructType(
- StructField("num_bool", StringType, true) ::
- StructField("num_num_1", LongType, true) ::
- StructField("num_num_2", DoubleType, true) ::
- StructField("num_num_3", DoubleType, true) ::
- StructField("num_str", StringType, true) ::
- StructField("str_bool", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("num_bool", StringType, true) ::
+ StructField("num_num_1", LongType, true) ::
+ StructField("num_num_2", DoubleType, true) ::
+ StructField("num_num_3", DoubleType, true) ::
+ StructField("num_str", StringType, true) ::
+ StructField("str_bool", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row("true", 11L, null, 1.1, "13.1", "str1") ::
- Row("12", null, 21474836470.9, null, null, "true") ::
- Row("false", 21474836470L, 92233720368547758070d, 100, "str1", "false") ::
- Row(null, 21474836570L, 1.1, 21474836470L, "92233720368547758070", null) :: Nil
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row("true", 11L, null, 1.1, "13.1", "str1") ::
+ Row("12", null, 21474836470.9, null, null, "true") ::
+ Row("false", 21474836470L, 92233720368547758070d, 100, "str1", "false") ::
+ Row(null, 21474836570L, 1.1, 21474836470L, "92233720368547758070", null) :: Nil
+ )
- // Number and Boolean conflict: resolve the type as number in this query.
- checkAnswer(
- sql("select num_bool - 10 from jsonTable where num_bool > 11"),
- Row(2)
- )
+ // Number and Boolean conflict: resolve the type as number in this query.
+ checkAnswer(
+ sql("select num_bool - 10 from jsonTable where num_bool > 11"),
+ Row(2)
+ )
- // Widening to LongType
- checkAnswer(
- sql("select num_num_1 - 100 from jsonTable where num_num_1 > 11"),
- Row(21474836370L) :: Row(21474836470L) :: Nil
- )
+ // Widening to LongType
+ checkAnswer(
+ sql("select num_num_1 - 100 from jsonTable where num_num_1 > 11"),
+ Row(21474836370L) :: Row(21474836470L) :: Nil
+ )
- checkAnswer(
- sql("select num_num_1 - 100 from jsonTable where num_num_1 > 10"),
- Row(-89) :: Row(21474836370L) :: Row(21474836470L) :: Nil
- )
+ checkAnswer(
+ sql("select num_num_1 - 100 from jsonTable where num_num_1 > 10"),
+ Row(-89) :: Row(21474836370L) :: Row(21474836470L) :: Nil
+ )
- // Widening to DecimalType
- checkAnswer(
- sql("select num_num_2 + 1.3 from jsonTable where num_num_2 > 1.1"),
- Row(21474836472.2) ::
- Row(92233720368547758071.3) :: Nil
- )
+ // Widening to DecimalType
+ checkAnswer(
+ sql("select num_num_2 + 1.3 from jsonTable where num_num_2 > 1.1"),
+ Row(21474836472.2) ::
+ Row(92233720368547758071.3) :: Nil
+ )
- // Widening to Double
- checkAnswer(
- sql("select num_num_3 + 1.2 from jsonTable where num_num_3 > 1.1"),
- Row(101.2) :: Row(21474836471.2) :: Nil
- )
+ // Widening to Double
+ checkAnswer(
+ sql("select num_num_3 + 1.2 from jsonTable where num_num_3 > 1.1"),
+ Row(101.2) :: Row(21474836471.2) :: Nil
+ )
- // Number and String conflict: resolve the type as number in this query.
- checkAnswer(
- sql("select num_str + 1.2 from jsonTable where num_str > 14d"),
- Row(92233720368547758071.2)
- )
+ // Number and String conflict: resolve the type as number in this query.
+ checkAnswer(
+ sql("select num_str + 1.2 from jsonTable where num_str > 14d"),
+ Row(92233720368547758071.2)
+ )
- // Number and String conflict: resolve the type as number in this query.
- checkAnswer(
- sql("select num_str + 1.2 from jsonTable where num_str >= 92233720368547758060"),
- Row(new java.math.BigDecimal("92233720368547758071.2").doubleValue)
- )
+ // Number and String conflict: resolve the type as number in this query.
+ checkAnswer(
+ sql("select num_str + 1.2 from jsonTable where num_str >= 92233720368547758060"),
+ Row(new java.math.BigDecimal("92233720368547758071.2").doubleValue)
+ )
- // String and Boolean conflict: resolve the type as string.
- checkAnswer(
- sql("select * from jsonTable where str_bool = 'str1'"),
- Row("true", 11L, null, 1.1, "13.1", "str1")
- )
+ // String and Boolean conflict: resolve the type as string.
+ checkAnswer(
+ sql("select * from jsonTable where str_bool = 'str1'"),
+ Row("true", 11L, null, 1.1, "13.1", "str1")
+ )
+ }
}
test("Type conflict in complex field values") {
- val jsonDF = spark.read.json(complexFieldValueTypeConflict)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(complexFieldValueTypeConflict)
- val expectedSchema = StructType(
- StructField("array", ArrayType(LongType, true), true) ::
- StructField("num_struct", StringType, true) ::
- StructField("str_array", StringType, true) ::
- StructField("struct", StructType(
- StructField("field", StringType, true) :: Nil), true) ::
- StructField("struct_array", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("array", ArrayType(LongType, true), true) ::
+ StructField("num_struct", StringType, true) ::
+ StructField("str_array", StringType, true) ::
+ StructField("struct", StructType(
+ StructField("field", StringType, true) :: Nil), true) ::
+ StructField("struct_array", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row(Seq(), "11", "[1,2,3]", Row(null), "[]") ::
- Row(null, """{"field":false}""", null, null, "{}") ::
- Row(Seq(4, 5, 6), null, "str", Row(null), "[7,8,9]") ::
- Row(Seq(7), "{}", """["str1","str2",33]""", Row("str"), """{"field":true}""") :: Nil
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row(Seq(), "11", "[1,2,3]", Row(null), "[]") ::
+ Row(null, """{"field":false}""", null, null, "{}") ::
+ Row(Seq(4, 5, 6), null, "str", Row(null), "[7,8,9]") ::
+ Row(Seq(7), "{}", """["str1","str2",33]""", Row("str"), """{"field":true}""") :: Nil
+ )
+ }
}
test("Type conflict in array elements") {
- val jsonDF = spark.read.json(arrayElementTypeConflict)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(arrayElementTypeConflict)
- val expectedSchema = StructType(
- StructField("array1", ArrayType(StringType, true), true) ::
- StructField("array2", ArrayType(StructType(
- StructField("field", LongType, true) :: Nil), true), true) ::
- StructField("array3", ArrayType(StringType, true), true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("array1", ArrayType(StringType, true), true) ::
+ StructField("array2", ArrayType(StructType(
+ StructField("field", LongType, true) :: Nil), true), true) ::
+ StructField("array3", ArrayType(StringType, true), true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row(Seq("1", "1.1", "true", null, "[]", "{}", "[2,3,4]",
- """{"field":"str"}"""), Seq(Row(214748364700L), Row(1)), null) ::
- Row(null, null, Seq("""{"field":"str"}""", """{"field":1}""")) ::
- Row(null, null, Seq("1", "2", "3")) :: Nil
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row(Seq("1", "1.1", "true", null, "[]", "{}", "[2,3,4]",
+ """{"field":"str"}"""), Seq(Row(214748364700L), Row(1)), null) ::
+ Row(null, null, Seq("""{"field":"str"}""", """{"field":1}""")) ::
+ Row(null, null, Seq("1", "2", "3")) :: Nil
+ )
- // Treat an element as a number.
- checkAnswer(
- sql("select array1[0] + 1 from jsonTable where array1 is not null"),
- Row(2)
- )
+ // Treat an element as a number.
+ checkAnswer(
+ sql("select array1[0] + 1 from jsonTable where array1 is not null"),
+ Row(2)
+ )
+ }
}
test("Handling missing fields") {
- val jsonDF = spark.read.json(missingFields)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(missingFields)
- val expectedSchema = StructType(
- StructField("a", BooleanType, true) ::
- StructField("b", LongType, true) ::
- StructField("c", ArrayType(LongType, true), true) ::
- StructField("d", StructType(
- StructField("field", BooleanType, true) :: Nil), true) ::
- StructField("e", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("a", BooleanType, true) ::
+ StructField("b", LongType, true) ::
+ StructField("c", ArrayType(LongType, true), true) ::
+ StructField("d", StructType(
+ StructField("field", BooleanType, true) :: Nil), true) ::
+ StructField("e", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
+ }
}
test("Loading a JSON dataset from a text file") {
- val dir = Utils.createTempDir()
- dir.delete()
- val path = dir.getCanonicalPath
- primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
- val jsonDF = spark.read.json(path)
+ withTempView("jsonTable") {
+ val dir = Utils.createTempDir()
+ dir.delete()
+ val path = dir.getCanonicalPath
+ primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
+ val jsonDF = spark.read.json(path)
- val expectedSchema = StructType(
- StructField("bigInteger", DecimalType(20, 0), true) ::
- StructField("boolean", BooleanType, true) ::
- StructField("double", DoubleType, true) ::
- StructField("integer", LongType, true) ::
- StructField("long", LongType, true) ::
- StructField("null", StringType, true) ::
- StructField("string", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("bigInteger", DecimalType(20, 0), true) ::
+ StructField("boolean", BooleanType, true) ::
+ StructField("double", DoubleType, true) ::
+ StructField("integer", LongType, true) ::
+ StructField("long", LongType, true) ::
+ StructField("null", StringType, true) ::
+ StructField("string", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row(new java.math.BigDecimal("92233720368547758070"),
- true,
- 1.7976931348623157,
- 10,
- 21474836470L,
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row(new java.math.BigDecimal("92233720368547758070"),
+ true,
+ 1.7976931348623157,
+ 10,
+ 21474836470L,
+ null,
+ "this is a simple string.")
+ )
+ }
}
test("Loading a JSON dataset primitivesAsString returns schema with primitive types as strings") {
- val dir = Utils.createTempDir()
- dir.delete()
- val path = dir.getCanonicalPath
- primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
- val jsonDF = spark.read.option("primitivesAsString", "true").json(path)
-
- val expectedSchema = StructType(
- StructField("bigInteger", StringType, true) ::
- StructField("boolean", StringType, true) ::
- StructField("double", StringType, true) ::
- StructField("integer", StringType, true) ::
- StructField("long", StringType, true) ::
- StructField("null", StringType, true) ::
- StructField("string", StringType, true) :: Nil)
+ withTempView("jsonTable") {
+ val dir = Utils.createTempDir()
+ dir.delete()
+ val path = dir.getCanonicalPath
+ primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
+ val jsonDF = spark.read.option("primitivesAsString", "true").json(path)
+
+ val expectedSchema = StructType(
+ StructField("bigInteger", StringType, true) ::
+ StructField("boolean", StringType, true) ::
+ StructField("double", StringType, true) ::
+ StructField("integer", StringType, true) ::
+ StructField("long", StringType, true) ::
+ StructField("null", StringType, true) ::
+ StructField("string", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row("92233720368547758070",
- "true",
- "1.7976931348623157",
- "10",
- "21474836470",
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row("92233720368547758070",
+ "true",
+ "1.7976931348623157",
+ "10",
+ "21474836470",
+ null,
+ "this is a simple string.")
+ )
+ }
}
test("Loading a JSON dataset primitivesAsString returns complex fields as strings") {
- val jsonDF = spark.read.option("primitivesAsString", "true").json(complexFieldAndType1)
-
- val expectedSchema = StructType(
- StructField("arrayOfArray1", ArrayType(ArrayType(StringType, true), true), true) ::
- StructField("arrayOfArray2", ArrayType(ArrayType(StringType, true), true), true) ::
- StructField("arrayOfBigInteger", ArrayType(StringType, true), true) ::
- StructField("arrayOfBoolean", ArrayType(StringType, true), true) ::
- StructField("arrayOfDouble", ArrayType(StringType, true), true) ::
- StructField("arrayOfInteger", ArrayType(StringType, true), true) ::
- StructField("arrayOfLong", ArrayType(StringType, true), true) ::
- StructField("arrayOfNull", ArrayType(StringType, true), true) ::
- StructField("arrayOfString", ArrayType(StringType, true), true) ::
- StructField("arrayOfStruct", ArrayType(
- StructType(
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.option("primitivesAsString", "true").json(complexFieldAndType1)
+
+ val expectedSchema = StructType(
+ StructField("arrayOfArray1", ArrayType(ArrayType(StringType, true), true), true) ::
+ StructField("arrayOfArray2", ArrayType(ArrayType(StringType, true), true), true) ::
+ StructField("arrayOfBigInteger", ArrayType(StringType, true), true) ::
+ StructField("arrayOfBoolean", ArrayType(StringType, true), true) ::
+ StructField("arrayOfDouble", ArrayType(StringType, true), true) ::
+ StructField("arrayOfInteger", ArrayType(StringType, true), true) ::
+ StructField("arrayOfLong", ArrayType(StringType, true), true) ::
+ StructField("arrayOfNull", ArrayType(StringType, true), true) ::
+ StructField("arrayOfString", ArrayType(StringType, true), true) ::
+ StructField("arrayOfStruct", ArrayType(
+ StructType(
+ StructField("field1", StringType, true) ::
+ StructField("field2", StringType, true) ::
+ StructField("field3", StringType, true) :: Nil), true), true) ::
+ StructField("struct", StructType(
StructField("field1", StringType, true) ::
- StructField("field2", StringType, true) ::
- StructField("field3", StringType, true) :: Nil), true), true) ::
- StructField("struct", StructType(
- StructField("field1", StringType, true) ::
- StructField("field2", StringType, true) :: Nil), true) ::
- StructField("structWithArrayFields", StructType(
- StructField("field1", ArrayType(StringType, true), true) ::
- StructField("field2", ArrayType(StringType, true), true) :: Nil), true) :: Nil)
+ StructField("field2", StringType, true) :: Nil), true) ::
+ StructField("structWithArrayFields", StructType(
+ StructField("field1", ArrayType(StringType, true), true) ::
+ StructField("field2", ArrayType(StringType, true), true) :: Nil), true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- // Access elements of a primitive array.
- checkAnswer(
- sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from jsonTable"),
- Row("str1", "str2", null)
- )
+ // Access elements of a primitive array.
+ checkAnswer(
+ sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from jsonTable"),
+ Row("str1", "str2", null)
+ )
- // Access an array of null values.
- checkAnswer(
- sql("select arrayOfNull from jsonTable"),
- Row(Seq(null, null, null, null))
- )
+ // Access an array of null values.
+ checkAnswer(
+ sql("select arrayOfNull from jsonTable"),
+ Row(Seq(null, null, null, null))
+ )
- // Access elements of a BigInteger array (we use DecimalType internally).
- checkAnswer(
- sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] from jsonTable"),
- Row("922337203685477580700", "-922337203685477580800", null)
- )
+ // Access elements of a BigInteger array (we use DecimalType internally).
+ checkAnswer(
+ sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] from " +
+ "jsonTable"),
+ Row("922337203685477580700", "-922337203685477580800", null)
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray1[0], arrayOfArray1[1] from jsonTable"),
- Row(Seq("1", "2", "3"), Seq("str1", "str2"))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray1[0], arrayOfArray1[1] from jsonTable"),
+ Row(Seq("1", "2", "3"), Seq("str1", "str2"))
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray2[0], arrayOfArray2[1] from jsonTable"),
- Row(Seq("1", "2", "3"), Seq("1.1", "2.1", "3.1"))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray2[0], arrayOfArray2[1] from jsonTable"),
+ Row(Seq("1", "2", "3"), Seq("1.1", "2.1", "3.1"))
+ )
- // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
- checkAnswer(
- sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from jsonTable"),
- Row("str2", "2.1")
- )
+ // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
+ checkAnswer(
+ sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from jsonTable"),
+ Row("str2", "2.1")
+ )
- // Access elements of an array of structs.
- checkAnswer(
- sql("select arrayOfStruct[0], arrayOfStruct[1], arrayOfStruct[2], arrayOfStruct[3] " +
- "from jsonTable"),
- Row(
- Row("true", "str1", null),
- Row("false", null, null),
- Row(null, null, null),
- null)
- )
+ // Access elements of an array of structs.
+ checkAnswer(
+ sql("select arrayOfStruct[0], arrayOfStruct[1], arrayOfStruct[2], arrayOfStruct[3] " +
+ "from jsonTable"),
+ Row(
+ Row("true", "str1", null),
+ Row("false", null, null),
+ Row(null, null, null),
+ null)
+ )
- // Access a struct and fields inside of it.
- checkAnswer(
- sql("select struct, struct.field1, struct.field2 from jsonTable"),
- Row(
- Row("true", "92233720368547758070"),
- "true",
- "92233720368547758070") :: Nil
- )
+ // Access a struct and fields inside of it.
+ checkAnswer(
+ sql("select struct, struct.field1, struct.field2 from jsonTable"),
+ Row(
+ Row("true", "92233720368547758070"),
+ "true",
+ "92233720368547758070") :: Nil
+ )
- // Access an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1, structWithArrayFields.field2 from jsonTable"),
- Row(Seq("4", "5", "6"), Seq("str1", "str2"))
- )
+ // Access an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1, structWithArrayFields.field2 from jsonTable"),
+ Row(Seq("4", "5", "6"), Seq("str1", "str2"))
+ )
- // Access elements of an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] from jsonTable"),
- Row("5", null)
- )
+ // Access elements of an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] from " +
+ "jsonTable"),
+ Row("5", null)
+ )
+ }
}
test("Loading a JSON dataset prefersDecimal returns schema with float types as BigDecimal") {
- val jsonDF = spark.read.option("prefersDecimal", "true").json(primitiveFieldAndType)
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.option("prefersDecimal", "true").json(primitiveFieldAndType)
- val expectedSchema = StructType(
- StructField("bigInteger", DecimalType(20, 0), true) ::
- StructField("boolean", BooleanType, true) ::
- StructField("double", DecimalType(17, 16), true) ::
- StructField("integer", LongType, true) ::
- StructField("long", LongType, true) ::
- StructField("null", StringType, true) ::
- StructField("string", StringType, true) :: Nil)
+ val expectedSchema = StructType(
+ StructField("bigInteger", DecimalType(20, 0), true) ::
+ StructField("boolean", BooleanType, true) ::
+ StructField("double", DecimalType(17, 16), true) ::
+ StructField("integer", LongType, true) ::
+ StructField("long", LongType, true) ::
+ StructField("null", StringType, true) ::
+ StructField("string", StringType, true) :: Nil)
- assert(expectedSchema === jsonDF.schema)
+ assert(expectedSchema === jsonDF.schema)
- jsonDF.createOrReplaceTempView("jsonTable")
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select * from jsonTable"),
- Row(BigDecimal("92233720368547758070"),
- true,
- BigDecimal("1.7976931348623157"),
- 10,
- 21474836470L,
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable"),
+ Row(BigDecimal("92233720368547758070"),
+ true,
+ BigDecimal("1.7976931348623157"),
+ 10,
+ 21474836470L,
+ null,
+ "this is a simple string.")
+ )
+ }
}
test("Find compatible types even if inferred DecimalType is not capable of other IntegralType") {
@@ -834,171 +862,182 @@ abstract class JsonSuite extends QueryTest with SharedSparkSession with TestJson
}
test("Applying schemas") {
- val dir = Utils.createTempDir()
- dir.delete()
- val path = dir.getCanonicalPath
- primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
+ withTempView("jsonTable1", "jsonTable2") {
+ val dir = Utils.createTempDir()
+ dir.delete()
+ val path = dir.getCanonicalPath
+ primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).write.text(path)
- val schema = StructType(
- StructField("bigInteger", DecimalType.SYSTEM_DEFAULT, true) ::
- StructField("boolean", BooleanType, true) ::
- StructField("double", DoubleType, true) ::
- StructField("integer", IntegerType, true) ::
- StructField("long", LongType, true) ::
- StructField("null", StringType, true) ::
- StructField("string", StringType, true) :: Nil)
+ val schema = StructType(
+ StructField("bigInteger", DecimalType.SYSTEM_DEFAULT, true) ::
+ StructField("boolean", BooleanType, true) ::
+ StructField("double", DoubleType, true) ::
+ StructField("integer", IntegerType, true) ::
+ StructField("long", LongType, true) ::
+ StructField("null", StringType, true) ::
+ StructField("string", StringType, true) :: Nil)
- val jsonDF1 = spark.read.schema(schema).json(path)
+ val jsonDF1 = spark.read.schema(schema).json(path)
- assert(schema === jsonDF1.schema)
+ assert(schema === jsonDF1.schema)
- jsonDF1.createOrReplaceTempView("jsonTable1")
+ jsonDF1.createOrReplaceTempView("jsonTable1")
- checkAnswer(
- sql("select * from jsonTable1"),
- Row(new java.math.BigDecimal("92233720368547758070"),
- true,
- 1.7976931348623157,
- 10,
- 21474836470L,
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable1"),
+ Row(new java.math.BigDecimal("92233720368547758070"),
+ true,
+ 1.7976931348623157,
+ 10,
+ 21474836470L,
+ null,
+ "this is a simple string.")
+ )
- val jsonDF2 = spark.read.schema(schema).json(primitiveFieldAndType)
+ val jsonDF2 = spark.read.schema(schema).json(primitiveFieldAndType)
- assert(schema === jsonDF2.schema)
+ assert(schema === jsonDF2.schema)
- jsonDF2.createOrReplaceTempView("jsonTable2")
+ jsonDF2.createOrReplaceTempView("jsonTable2")
- checkAnswer(
- sql("select * from jsonTable2"),
- Row(new java.math.BigDecimal("92233720368547758070"),
- true,
- 1.7976931348623157,
- 10,
- 21474836470L,
- null,
- "this is a simple string.")
- )
+ checkAnswer(
+ sql("select * from jsonTable2"),
+ Row(new java.math.BigDecimal("92233720368547758070"),
+ true,
+ 1.7976931348623157,
+ 10,
+ 21474836470L,
+ null,
+ "this is a simple string.")
+ )
+ }
}
test("Applying schemas with MapType") {
- val schemaWithSimpleMap = StructType(
- StructField("map", MapType(StringType, IntegerType, true), false) :: Nil)
- val jsonWithSimpleMap = spark.read.schema(schemaWithSimpleMap).json(mapType1)
+ withTempView("jsonWithSimpleMap", "jsonWithComplexMap") {
+ val schemaWithSimpleMap = StructType(
+ StructField("map", MapType(StringType, IntegerType, true), false) :: Nil)
+ val jsonWithSimpleMap = spark.read.schema(schemaWithSimpleMap).json(mapType1)
- jsonWithSimpleMap.createOrReplaceTempView("jsonWithSimpleMap")
+ jsonWithSimpleMap.createOrReplaceTempView("jsonWithSimpleMap")
- checkAnswer(
- sql("select `map` from jsonWithSimpleMap"),
- Row(Map("a" -> 1)) ::
- Row(Map("b" -> 2)) ::
- Row(Map("c" -> 3)) ::
- Row(Map("c" -> 1, "d" -> 4)) ::
- Row(Map("e" -> null)) :: Nil
- )
-
- withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
checkAnswer(
- sql("select `map`['c'] from jsonWithSimpleMap"),
- Row(null) ::
- Row(null) ::
- Row(3) ::
- Row(1) ::
- Row(null) :: Nil
+ sql("select `map` from jsonWithSimpleMap"),
+ Row(Map("a" -> 1)) ::
+ Row(Map("b" -> 2)) ::
+ Row(Map("c" -> 3)) ::
+ Row(Map("c" -> 1, "d" -> 4)) ::
+ Row(Map("e" -> null)) :: Nil
)
- }
- val innerStruct = StructType(
- StructField("field1", ArrayType(IntegerType, true), true) ::
- StructField("field2", IntegerType, true) :: Nil)
- val schemaWithComplexMap = StructType(
- StructField("map", MapType(StringType, innerStruct, true), false) :: Nil)
+ withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
+ checkAnswer(
+ sql("select `map`['c'] from jsonWithSimpleMap"),
+ Row(null) ::
+ Row(null) ::
+ Row(3) ::
+ Row(1) ::
+ Row(null) :: Nil
+ )
+ }
- val jsonWithComplexMap = spark.read.schema(schemaWithComplexMap).json(mapType2)
+ val innerStruct = StructType(
+ StructField("field1", ArrayType(IntegerType, true), true) ::
+ StructField("field2", IntegerType, true) :: Nil)
+ val schemaWithComplexMap = StructType(
+ StructField("map", MapType(StringType, innerStruct, true), false) :: Nil)
- jsonWithComplexMap.createOrReplaceTempView("jsonWithComplexMap")
+ val jsonWithComplexMap = spark.read.schema(schemaWithComplexMap).json(mapType2)
- checkAnswer(
- sql("select `map` from jsonWithComplexMap"),
- Row(Map("a" -> Row(Seq(1, 2, 3, null), null))) ::
- Row(Map("b" -> Row(null, 2))) ::
- Row(Map("c" -> Row(Seq(), 4))) ::
- Row(Map("c" -> Row(null, 3), "d" -> Row(Seq(null), null))) ::
- Row(Map("e" -> null)) ::
- Row(Map("f" -> Row(null, null))) :: Nil
- )
+ jsonWithComplexMap.createOrReplaceTempView("jsonWithComplexMap")
- withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
checkAnswer(
- sql("select `map`['a'].field1, `map`['c'].field2 from jsonWithComplexMap"),
- Row(Seq(1, 2, 3, null), null) ::
- Row(null, null) ::
- Row(null, 4) ::
- Row(null, 3) ::
- Row(null, null) ::
- Row(null, null) :: Nil
+ sql("select `map` from jsonWithComplexMap"),
+ Row(Map("a" -> Row(Seq(1, 2, 3, null), null))) ::
+ Row(Map("b" -> Row(null, 2))) ::
+ Row(Map("c" -> Row(Seq(), 4))) ::
+ Row(Map("c" -> Row(null, 3), "d" -> Row(Seq(null), null))) ::
+ Row(Map("e" -> null)) ::
+ Row(Map("f" -> Row(null, null))) :: Nil
)
+
+ withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
+ checkAnswer(
+ sql("select `map`['a'].field1, `map`['c'].field2 from jsonWithComplexMap"),
+ Row(Seq(1, 2, 3, null), null) ::
+ Row(null, null) ::
+ Row(null, 4) ::
+ Row(null, 3) ::
+ Row(null, null) ::
+ Row(null, null) :: Nil
+ )
+ }
}
}
test("SPARK-2096 Correctly parse dot notations") {
- val jsonDF = spark.read.json(complexFieldAndType2)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(complexFieldAndType2)
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql("select arrayOfStruct[0].field1, arrayOfStruct[0].field2 from jsonTable"),
- Row(true, "str1")
- )
- checkAnswer(
- sql(
- """
- |select complexArrayOfStruct[0].field1[1].inner2[0], complexArrayOfStruct[1].field2[0][1]
- |from jsonTable
- """.stripMargin),
- Row("str2", 6)
- )
+ checkAnswer(
+ sql("select arrayOfStruct[0].field1, arrayOfStruct[0].field2 from jsonTable"),
+ Row(true, "str1")
+ )
+ checkAnswer(
+ sql(
+ """
+ |select complexArrayOfStruct[0].field1[1].inner2[0],
+ |complexArrayOfStruct[1].field2[0][1]
+ |from jsonTable
+ """.stripMargin),
+ Row("str2", 6)
+ )
+ }
}
test("SPARK-3390 Complex arrays") {
- val jsonDF = spark.read.json(complexFieldAndType2)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(complexFieldAndType2)
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql(
- """
- |select arrayOfArray1[0][0][0], arrayOfArray1[1][0][1], arrayOfArray1[1][1][0]
- |from jsonTable
- """.stripMargin),
- Row(5, 7, 8)
- )
- checkAnswer(
- sql(
- """
- |select arrayOfArray2[0][0][0].inner1, arrayOfArray2[1][0],
- |arrayOfArray2[1][1][1].inner2[0], arrayOfArray2[2][0][0].inner3[0][0].inner4
- |from jsonTable
- """.stripMargin),
- Row("str1", Nil, "str4", 2)
- )
+ checkAnswer(
+ sql(
+ """
+ |select arrayOfArray1[0][0][0], arrayOfArray1[1][0][1], arrayOfArray1[1][1][0]
+ |from jsonTable
+ """.stripMargin),
+ Row(5, 7, 8)
+ )
+ checkAnswer(
+ sql(
+ """
+ |select arrayOfArray2[0][0][0].inner1, arrayOfArray2[1][0],
+ |arrayOfArray2[1][1][1].inner2[0], arrayOfArray2[2][0][0].inner3[0][0].inner4
+ |from jsonTable
+ """.stripMargin),
+ Row("str1", Nil, "str4", 2)
+ )
+ }
}
test("SPARK-3308 Read top level JSON arrays") {
- val jsonDF = spark.read.json(jsonArray)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(jsonArray)
+ jsonDF.createOrReplaceTempView("jsonTable")
- checkAnswer(
- sql(
- """
- |select a, b, c
- |from jsonTable
- """.stripMargin),
- Row("str_a_1", null, null) ::
- Row("str_a_2", null, null) ::
- Row(null, "str_b_3", null) ::
- Row("str_a_4", "str_b_4", "str_c_4") :: Nil
- )
+ checkAnswer(
+ sql(
+ """
+ |select a, b, c
+ |from jsonTable
+ """.stripMargin),
+ Row("str_a_1", null, null) ::
+ Row("str_a_2", null, null) ::
+ Row(null, "str_b_3", null) ::
+ Row("str_a_4", "str_b_4", "str_c_4") :: Nil
+ )
+ }
}
test("Corrupt records: FAILFAST mode") {
@@ -1140,158 +1179,162 @@ abstract class JsonSuite extends QueryTest with SharedSparkSession with TestJson
}
test("SPARK-4068: nulls in arrays") {
- val jsonDF = spark.read.json(nullsInArrays)
- jsonDF.createOrReplaceTempView("jsonTable")
+ withTempView("jsonTable") {
+ val jsonDF = spark.read.json(nullsInArrays)
+ jsonDF.createOrReplaceTempView("jsonTable")
- val schema = StructType(
- StructField("field1",
- ArrayType(ArrayType(ArrayType(ArrayType(StringType, true), true), true), true), true) ::
- StructField("field2",
- ArrayType(ArrayType(
- StructType(StructField("Test", LongType, true) :: Nil), true), true), true) ::
- StructField("field3",
- ArrayType(ArrayType(
- StructType(StructField("Test", StringType, true) :: Nil), true), true), true) ::
- StructField("field4",
- ArrayType(ArrayType(ArrayType(LongType, true), true), true), true) :: Nil)
+ val schema = StructType(
+ StructField("field1",
+ ArrayType(ArrayType(ArrayType(ArrayType(StringType, true), true), true), true), true) ::
+ StructField("field2",
+ ArrayType(ArrayType(
+ StructType(StructField("Test", LongType, true) :: Nil), true), true), true) ::
+ StructField("field3",
+ ArrayType(ArrayType(
+ StructType(StructField("Test", StringType, true) :: Nil), true), true), true) ::
+ StructField("field4",
+ ArrayType(ArrayType(ArrayType(LongType, true), true), true), true) :: Nil)
- assert(schema === jsonDF.schema)
+ assert(schema === jsonDF.schema)
- checkAnswer(
- sql(
- """
- |SELECT field1, field2, field3, field4
- |FROM jsonTable
- """.stripMargin),
- Row(Seq(Seq(null), Seq(Seq(Seq("Test")))), null, null, null) ::
- Row(null, Seq(null, Seq(Row(1))), null, null) ::
- Row(null, null, Seq(Seq(null), Seq(Row("2"))), null) ::
- Row(null, null, null, Seq(Seq(null, Seq(1, 2, 3)))) :: Nil
- )
+ checkAnswer(
+ sql(
+ """
+ |SELECT field1, field2, field3, field4
+ |FROM jsonTable
+ """.stripMargin),
+ Row(Seq(Seq(null), Seq(Seq(Seq("Test")))), null, null, null) ::
+ Row(null, Seq(null, Seq(Row(1))), null, null) ::
+ Row(null, null, Seq(Seq(null), Seq(Row("2"))), null) ::
+ Row(null, null, null, Seq(Seq(null, Seq(1, 2, 3)))) :: Nil
+ )
+ }
}
test("SPARK-4228 DataFrame to JSON") {
- val schema1 = StructType(
- StructField("f1", IntegerType, false) ::
- StructField("f2", StringType, false) ::
- StructField("f3", BooleanType, false) ::
- StructField("f4", ArrayType(StringType), nullable = true) ::
- StructField("f5", IntegerType, true) :: Nil)
-
- val rowRDD1 = unparsedStrings.map { r =>
- val values = r.split(",").map(_.trim)
- val v5 = try values(3).toInt catch {
- case _: NumberFormatException => null
+ withTempView("applySchema1", "applySchema2", "primitiveTable", "complexTable") {
+ val schema1 = StructType(
+ StructField("f1", IntegerType, false) ::
+ StructField("f2", StringType, false) ::
+ StructField("f3", BooleanType, false) ::
+ StructField("f4", ArrayType(StringType), nullable = true) ::
+ StructField("f5", IntegerType, true) :: Nil)
+
+ val rowRDD1 = unparsedStrings.map { r =>
+ val values = r.split(",").map(_.trim)
+ val v5 = try values(3).toInt catch {
+ case _: NumberFormatException => null
+ }
+ Row(values(0).toInt, values(1), values(2).toBoolean, r.split(",").toList, v5)
}
- Row(values(0).toInt, values(1), values(2).toBoolean, r.split(",").toList, v5)
- }
-
- val df1 = spark.createDataFrame(rowRDD1, schema1)
- df1.createOrReplaceTempView("applySchema1")
- val df2 = df1.toDF
- val result = df2.toJSON.collect()
- // scalastyle:off
- assert(result(0) === "{\"f1\":1,\"f2\":\"A1\",\"f3\":true,\"f4\":[\"1\",\" A1\",\" true\",\" null\"]}")
- assert(result(3) === "{\"f1\":4,\"f2\":\"D4\",\"f3\":true,\"f4\":[\"4\",\" D4\",\" true\",\" 2147483644\"],\"f5\":2147483644}")
- // scalastyle:on
- val schema2 = StructType(
- StructField("f1", StructType(
- StructField("f11", IntegerType, false) ::
- StructField("f12", BooleanType, false) :: Nil), false) ::
- StructField("f2", MapType(StringType, IntegerType, true), false) :: Nil)
-
- val rowRDD2 = unparsedStrings.map { r =>
- val values = r.split(",").map(_.trim)
- val v4 = try values(3).toInt catch {
- case _: NumberFormatException => null
+ val df1 = spark.createDataFrame(rowRDD1, schema1)
+ df1.createOrReplaceTempView("applySchema1")
+ val df2 = df1.toDF
+ val result = df2.toJSON.collect()
+ // scalastyle:off
+ assert(result(0) === "{\"f1\":1,\"f2\":\"A1\",\"f3\":true,\"f4\":[\"1\",\" A1\",\" true\",\" null\"]}")
+ assert(result(3) === "{\"f1\":4,\"f2\":\"D4\",\"f3\":true,\"f4\":[\"4\",\" D4\",\" true\",\" 2147483644\"],\"f5\":2147483644}")
+ // scalastyle:on
+
+ val schema2 = StructType(
+ StructField("f1", StructType(
+ StructField("f11", IntegerType, false) ::
+ StructField("f12", BooleanType, false) :: Nil), false) ::
+ StructField("f2", MapType(StringType, IntegerType, true), false) :: Nil)
+
+ val rowRDD2 = unparsedStrings.map { r =>
+ val values = r.split(",").map(_.trim)
+ val v4 = try values(3).toInt catch {
+ case _: NumberFormatException => null
+ }
+ Row(Row(values(0).toInt, values(2).toBoolean), Map(values(1) -> v4))
}
- Row(Row(values(0).toInt, values(2).toBoolean), Map(values(1) -> v4))
- }
- val df3 = spark.createDataFrame(rowRDD2, schema2)
- df3.createOrReplaceTempView("applySchema2")
- val df4 = df3.toDF
- val result2 = df4.toJSON.collect()
+ val df3 = spark.createDataFrame(rowRDD2, schema2)
+ df3.createOrReplaceTempView("applySchema2")
+ val df4 = df3.toDF
+ val result2 = df4.toJSON.collect()
- assert(result2(1) === "{\"f1\":{\"f11\":2,\"f12\":false},\"f2\":{\"B2\":null}}")
- assert(result2(3) === "{\"f1\":{\"f11\":4,\"f12\":true},\"f2\":{\"D4\":2147483644}}")
+ assert(result2(1) === "{\"f1\":{\"f11\":2,\"f12\":false},\"f2\":{\"B2\":null}}")
+ assert(result2(3) === "{\"f1\":{\"f11\":4,\"f12\":true},\"f2\":{\"D4\":2147483644}}")
- val jsonDF = spark.read.json(primitiveFieldAndType)
- val primTable = spark.read.json(jsonDF.toJSON)
- primTable.createOrReplaceTempView("primitiveTable")
- checkAnswer(
- sql("select * from primitiveTable"),
- Row(new java.math.BigDecimal("92233720368547758070"),
- true,
- 1.7976931348623157,
- 10,
- 21474836470L,
- "this is a simple string.")
+ val jsonDF = spark.read.json(primitiveFieldAndType)
+ val primTable = spark.read.json(jsonDF.toJSON)
+ primTable.createOrReplaceTempView("primitiveTable")
+ checkAnswer(
+ sql("select * from primitiveTable"),
+ Row(new java.math.BigDecimal("92233720368547758070"),
+ true,
+ 1.7976931348623157,
+ 10,
+ 21474836470L,
+ "this is a simple string.")
)
- val complexJsonDF = spark.read.json(complexFieldAndType1)
- val compTable = spark.read.json(complexJsonDF.toJSON)
- compTable.createOrReplaceTempView("complexTable")
- // Access elements of a primitive array.
- checkAnswer(
- sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from complexTable"),
- Row("str1", "str2", null)
- )
+ val complexJsonDF = spark.read.json(complexFieldAndType1)
+ val compTable = spark.read.json(complexJsonDF.toJSON)
+ compTable.createOrReplaceTempView("complexTable")
+ // Access elements of a primitive array.
+ checkAnswer(
+ sql("select arrayOfString[0], arrayOfString[1], arrayOfString[2] from complexTable"),
+ Row("str1", "str2", null)
+ )
- // Access an array of null values.
- checkAnswer(
- sql("select arrayOfNull from complexTable"),
- Row(Seq(null, null, null, null))
- )
+ // Access an array of null values.
+ checkAnswer(
+ sql("select arrayOfNull from complexTable"),
+ Row(Seq(null, null, null, null))
+ )
- // Access elements of a BigInteger array (we use DecimalType internally).
- checkAnswer(
- sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] " +
- " from complexTable"),
- Row(new java.math.BigDecimal("922337203685477580700"),
- new java.math.BigDecimal("-922337203685477580800"), null)
- )
+ // Access elements of a BigInteger array (we use DecimalType internally).
+ checkAnswer(
+ sql("select arrayOfBigInteger[0], arrayOfBigInteger[1], arrayOfBigInteger[2] " +
+ " from complexTable"),
+ Row(new java.math.BigDecimal("922337203685477580700"),
+ new java.math.BigDecimal("-922337203685477580800"), null)
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray1[0], arrayOfArray1[1] from complexTable"),
- Row(Seq("1", "2", "3"), Seq("str1", "str2"))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray1[0], arrayOfArray1[1] from complexTable"),
+ Row(Seq("1", "2", "3"), Seq("str1", "str2"))
+ )
- // Access elements of an array of arrays.
- checkAnswer(
- sql("select arrayOfArray2[0], arrayOfArray2[1] from complexTable"),
- Row(Seq(1.0, 2.0, 3.0), Seq(1.1, 2.1, 3.1))
- )
+ // Access elements of an array of arrays.
+ checkAnswer(
+ sql("select arrayOfArray2[0], arrayOfArray2[1] from complexTable"),
+ Row(Seq(1.0, 2.0, 3.0), Seq(1.1, 2.1, 3.1))
+ )
- // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
- checkAnswer(
- sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from complexTable"),
- Row("str2", 2.1)
- )
+ // Access elements of an array inside a filed with the type of ArrayType(ArrayType).
+ checkAnswer(
+ sql("select arrayOfArray1[1][1], arrayOfArray2[1][1] from complexTable"),
+ Row("str2", 2.1)
+ )
- // Access a struct and fields inside of it.
- checkAnswer(
- sql("select struct, struct.field1, struct.field2 from complexTable"),
- Row(
- Row(true, new java.math.BigDecimal("92233720368547758070")),
- true,
- new java.math.BigDecimal("92233720368547758070")) :: Nil
- )
+ // Access a struct and fields inside of it.
+ checkAnswer(
+ sql("select struct, struct.field1, struct.field2 from complexTable"),
+ Row(
+ Row(true, new java.math.BigDecimal("92233720368547758070")),
+ true,
+ new java.math.BigDecimal("92233720368547758070")) :: Nil
+ )
- // Access an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1, structWithArrayFields.field2 from complexTable"),
- Row(Seq(4, 5, 6), Seq("str1", "str2"))
- )
+ // Access an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1, structWithArrayFields.field2 from complexTable"),
+ Row(Seq(4, 5, 6), Seq("str1", "str2"))
+ )
- // Access elements of an array field of a struct.
- checkAnswer(
- sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] " +
- "from complexTable"),
- Row(5, null)
- )
+ // Access elements of an array field of a struct.
+ checkAnswer(
+ sql("select structWithArrayFields.field1[1], structWithArrayFields.field2[3] " +
+ "from complexTable"),
+ Row(5, null)
+ )
+ }
}
test("Dataset toJSON doesn't construct rdd") {
@@ -1372,20 +1415,21 @@ abstract class JsonSuite extends QueryTest with SharedSparkSession with TestJson
}
withTempPath(root => {
- val d1 = new File(root, "d1=1")
- // root/dt=1/col1=abc
- val p1_col1 = makePartition(
- sparkContext.parallelize(2 to 5).map(i => s"""{"a": 1, "b": "str$i"}"""),
- d1,
- "col1",
- "abc")
-
- // root/dt=1/col1=abd
- val p2 = makePartition(
- sparkContext.parallelize(6 to 10).map(i => s"""{"a": 1, "b": "str$i"}"""),
- d1,
- "col1",
- "abd")
+ withTempView("test_myjson_with_part") {
+ val d1 = new File(root, "d1=1")
+ // root/dt=1/col1=abc
+ val p1_col1 = makePartition(
+ sparkContext.parallelize(2 to 5).map(i => s"""{"a": 1, "b": "str$i"}"""),
+ d1,
+ "col1",
+ "abc")
+
+ // root/dt=1/col1=abd
+ val p2 = makePartition(
+ sparkContext.parallelize(6 to 10).map(i => s"""{"a": 1, "b": "str$i"}"""),
+ d1,
+ "col1",
+ "abd")
spark.read.json(root.getAbsolutePath).createOrReplaceTempView("test_myjson_with_part")
checkAnswer(sql(
@@ -1394,6 +1438,7 @@ abstract class JsonSuite extends QueryTest with SharedSparkSession with TestJson
"SELECT count(a) FROM test_myjson_with_part where d1 = 1 and col1='abd'"), Row(5))
checkAnswer(sql(
"SELECT count(a) FROM test_myjson_with_part where d1 = 1"), Row(9))
+ }
})
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
index 64ecf5d..1be9308 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/joins/BroadcastJoinSuite.scala
@@ -207,24 +207,25 @@ abstract class BroadcastJoinSuiteBase extends QueryTest with SQLTestUtils
test("broadcast hint in SQL") {
import org.apache.spark.sql.catalyst.plans.logical.Join
-
- spark.range(10).createOrReplaceTempView("t")
- spark.range(10).createOrReplaceTempView("u")
-
- for (name <- Seq("BROADCAST", "BROADCASTJOIN", "MAPJOIN")) {
- val plan1 = sql(s"SELECT /*+ $name(t) */ * FROM t JOIN u ON t.id = u.id").queryExecution
- .optimizedPlan
- val plan2 = sql(s"SELECT /*+ $name(u) */ * FROM t JOIN u ON t.id = u.id").queryExecution
- .optimizedPlan
- val plan3 = sql(s"SELECT /*+ $name(v) */ * FROM t JOIN u ON t.id = u.id").queryExecution
- .optimizedPlan
-
- assert(plan1.asInstanceOf[Join].hint.leftHint.get.strategy.contains(BROADCAST))
- assert(plan1.asInstanceOf[Join].hint.rightHint.isEmpty)
- assert(plan2.asInstanceOf[Join].hint.leftHint.isEmpty)
- assert(plan2.asInstanceOf[Join].hint.rightHint.get.strategy.contains(BROADCAST))
- assert(plan3.asInstanceOf[Join].hint.leftHint.isEmpty)
- assert(plan3.asInstanceOf[Join].hint.rightHint.isEmpty)
+ withTempView("t", "u") {
+ spark.range(10).createOrReplaceTempView("t")
+ spark.range(10).createOrReplaceTempView("u")
+
+ for (name <- Seq("BROADCAST", "BROADCASTJOIN", "MAPJOIN")) {
+ val plan1 = sql(s"SELECT /*+ $name(t) */ * FROM t JOIN u ON t.id = u.id").queryExecution
+ .optimizedPlan
+ val plan2 = sql(s"SELECT /*+ $name(u) */ * FROM t JOIN u ON t.id = u.id").queryExecution
+ .optimizedPlan
+ val plan3 = sql(s"SELECT /*+ $name(v) */ * FROM t JOIN u ON t.id = u.id").queryExecution
+ .optimizedPlan
+
+ assert(plan1.asInstanceOf[Join].hint.leftHint.get.strategy.contains(BROADCAST))
+ assert(plan1.asInstanceOf[Join].hint.rightHint.isEmpty)
+ assert(plan2.asInstanceOf[Join].hint.leftHint.isEmpty)
+ assert(plan2.asInstanceOf[Join].hint.rightHint.get.strategy.contains(BROADCAST))
+ assert(plan3.asInstanceOf[Join].hint.leftHint.isEmpty)
+ assert(plan3.asInstanceOf[Join].hint.rightHint.isEmpty)
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
index 1c4a818..00c0315 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
@@ -463,27 +463,30 @@ class SQLMetricsSuite extends SharedSparkSession with SQLMetricsTestUtils
// TODO: test file source V2 as well when its statistics is correctly computed.
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") {
withTempDir { tempDir =>
- val dir = new File(tempDir, "pqS").getCanonicalPath
-
- spark.range(10).write.parquet(dir)
- spark.read.parquet(dir).createOrReplaceTempView("pqS")
-
- // The executed plan looks like:
- // Exchange RoundRobinPartitioning(2)
- // +- BroadcastNestedLoopJoin BuildLeft, Cross
- // :- BroadcastExchange IdentityBroadcastMode
- // : +- Exchange RoundRobinPartitioning(3)
- // : +- *Range (0, 30, step=1, splits=2)
- // +- *FileScan parquet [id#465L] Batched: true, Format: Parquet, Location: ...(ignored)
- val res3 = InputOutputMetricsHelper.run(
- spark.range(30).repartition(3).crossJoin(sql("select * from pqS")).repartition(2).toDF()
- )
- // The query above is executed in the following stages:
- // 1. range(30) => (30, 0, 30)
- // 2. sql("select * from pqS") => (0, 30, 0)
- // 3. crossJoin(...) of 1. and 2. => (10, 0, 300)
- // 4. shuffle & return results => (0, 300, 0)
- assert(res3 === (30L, 0L, 30L) :: (0L, 30L, 0L) :: (10L, 0L, 300L) :: (0L, 300L, 0L) :: Nil)
+ withTempView("pqS") {
+ val dir = new File(tempDir, "pqS").getCanonicalPath
+
+ spark.range(10).write.parquet(dir)
+ spark.read.parquet(dir).createOrReplaceTempView("pqS")
+
+ // The executed plan looks like:
+ // Exchange RoundRobinPartitioning(2)
+ // +- BroadcastNestedLoopJoin BuildLeft, Cross
+ // :- BroadcastExchange IdentityBroadcastMode
+ // : +- Exchange RoundRobinPartitioning(3)
+ // : +- *Range (0, 30, step=1, splits=2)
+ // +- *FileScan parquet [id#465L] Batched: true, Format: Parquet, Location: ...(ignored)
+ val res3 = InputOutputMetricsHelper.run(
+ spark.range(30).repartition(3).crossJoin(sql("select * from pqS")).repartition(2).toDF()
+ )
+ // The query above is executed in the following stages:
+ // 1. range(30) => (30, 0, 30)
+ // 2. sql("select * from pqS") => (0, 30, 0)
+ // 3. crossJoin(...) of 1. and 2. => (10, 0, 300)
+ // 4. shuffle & return results => (0, 300, 0)
+ assert(res3 === (30L, 0L, 30L) :: (0L, 30L, 0L) :: (10L, 0L, 300L) :: (0L, 300L, 0L) ::
+ Nil)
+ }
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index fd691f5..7c723f08 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -635,12 +635,14 @@ class JDBCSuite extends QueryTest
}
test("test DATE types in cache") {
- val rows = spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new Properties()).collect()
- spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new Properties())
- .cache().createOrReplaceTempView("mycached_date")
- val cachedRows = sql("select * from mycached_date").collect()
- assert(rows(0).getAs[java.sql.Date](1) === java.sql.Date.valueOf("1996-01-01"))
- assert(cachedRows(0).getAs[java.sql.Date](1) === java.sql.Date.valueOf("1996-01-01"))
+ withTempView("mycached_date") {
+ val rows = spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new Properties()).collect()
+ spark.read.jdbc(urlWithUserAndPass, "TEST.TIMETYPES", new Properties())
+ .cache().createOrReplaceTempView("mycached_date")
+ val cachedRows = sql("select * from mycached_date").collect()
+ assert(rows(0).getAs[java.sql.Date](1) === java.sql.Date.valueOf("1996-01-01"))
+ assert(cachedRows(0).getAs[java.sql.Date](1) === java.sql.Date.valueOf("1996-01-01"))
+ }
}
test("test types for null value") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
index 2d66637..bb762d2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -523,15 +523,17 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
test("new partitions should be added to catalog after writing to catalog table") {
val table = "partitioned_catalog_table"
+ val tempTable = "partitioned_catalog_temp_table"
val numParts = 210
withTable(table) {
- val df = (1 to numParts).map(i => (i, i)).toDF("part", "col1")
- val tempTable = "partitioned_catalog_temp_table"
- df.createOrReplaceTempView(tempTable)
- sql(s"CREATE TABLE $table (part Int, col1 Int) USING parquet PARTITIONED BY (part)")
- sql(s"INSERT INTO TABLE $table SELECT * from $tempTable")
- val partitions = spark.sessionState.catalog.listPartitionNames(TableIdentifier(table))
- assert(partitions.size == numParts)
+ withTempView(tempTable) {
+ val df = (1 to numParts).map(i => (i, i)).toDF("part", "col1")
+ df.createOrReplaceTempView(tempTable)
+ sql(s"CREATE TABLE $table (part Int, col1 Int) USING parquet PARTITIONED BY (part)")
+ sql(s"INSERT INTO TABLE $table SELECT * from $tempTable")
+ val partitions = spark.sessionState.catalog.listPartitionNames(TableIdentifier(table))
+ assert(partitions.size == numParts)
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/SaveLoadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/SaveLoadSuite.scala
index ce3ec7f..1e0d9bd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/SaveLoadSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/SaveLoadSuite.scala
@@ -103,30 +103,32 @@ class SaveLoadSuite extends DataSourceTest with SharedSparkSession with BeforeAn
}
test("save and save again") {
- df.write.json(path.toString)
-
- val message = intercept[AnalysisException] {
+ withTempView("jsonTable2") {
df.write.json(path.toString)
- }.getMessage
- assert(
- message.contains("already exists"),
- "We should complain that the path already exists.")
+ val message = intercept[AnalysisException] {
+ df.write.json(path.toString)
+ }.getMessage
- if (path.exists()) Utils.deleteRecursively(path)
+ assert(
+ message.contains("already exists"),
+ "We should complain that the path already exists.")
- df.write.json(path.toString)
- checkLoad()
+ if (path.exists()) Utils.deleteRecursively(path)
- df.write.mode(SaveMode.Overwrite).json(path.toString)
- checkLoad()
+ df.write.json(path.toString)
+ checkLoad()
- // verify the append mode
- df.write.mode(SaveMode.Append).json(path.toString)
- val df2 = df.union(df)
- df2.createOrReplaceTempView("jsonTable2")
+ df.write.mode(SaveMode.Overwrite).json(path.toString)
+ checkLoad()
- checkLoad(df2, "jsonTable2")
+ // verify the append mode
+ df.write.mode(SaveMode.Append).json(path.toString)
+ val df2 = df.union(df)
+ df2.createOrReplaceTempView("jsonTable2")
+
+ checkLoad(df2, "jsonTable2")
+ }
}
test("SPARK-23459: Improve error message when specified unknown column in partition columns") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
index 5bcc4e9..d8d997e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
@@ -193,13 +193,15 @@ class StreamSuite extends StreamTest {
}
test("sql queries") {
- val inputData = MemoryStream[Int]
- inputData.toDF().createOrReplaceTempView("stream")
- val evens = sql("SELECT * FROM stream WHERE value % 2 = 0")
-
- testStream(evens)(
- AddData(inputData, 1, 2, 3, 4),
- CheckAnswer(2, 4))
+ withTempView("stream") {
+ val inputData = MemoryStream[Int]
+ inputData.toDF().createOrReplaceTempView("stream")
+ val evens = sql("SELECT * FROM stream WHERE value % 2 = 0")
+
+ testStream(evens)(
+ AddData(inputData, 1, 2, 3, 4),
+ CheckAnswer(2, 4))
+ }
}
test("DataFrame reuse") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
index 8599ceb..0d17f2e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
@@ -181,17 +181,19 @@ class ContinuousSuite extends ContinuousSuiteBase {
}
test("subquery alias") {
- val input = ContinuousMemoryStream[Int]
- input.toDF().createOrReplaceTempView("memory")
- val test = spark.sql("select value from memory where value > 2")
+ withTempView("memory") {
+ val input = ContinuousMemoryStream[Int]
+ input.toDF().createOrReplaceTempView("memory")
+ val test = spark.sql("select value from memory where value > 2")
- testStream(test)(
- AddData(input, 0, 1),
- CheckAnswer(),
- StopStream,
- AddData(input, 2, 3, 4),
- StartStream(),
- CheckAnswer(3, 4))
+ testStream(test)(
+ AddData(input, 0, 1),
+ CheckAnswer(),
+ StopStream,
+ AddData(input, 2, 3, 4),
+ StartStream(),
+ CheckAnswer(3, 4))
+ }
}
test("repeatedly restart") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
index ce40a65..fac9812 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
@@ -204,11 +204,13 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te
}
test("group by function") {
- Seq((1, 2)).toDF("a", "b").createOrReplaceTempView("data")
+ withTempView("data") {
+ Seq((1, 2)).toDF("a", "b").createOrReplaceTempView("data")
- checkAnswer(
- sql("SELECT floor(a) AS a, collect_set(b) FROM data GROUP BY floor(a) ORDER BY a"),
- Row(1, Array(2)) :: Nil)
+ checkAnswer(
+ sql("SELECT floor(a) AS a, collect_set(b) FROM data GROUP BY floor(a) ORDER BY a"),
+ Row(1, Array(2)) :: Nil)
+ }
}
test("empty table") {
@@ -800,43 +802,45 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te
(5, 8, 17),
(6, 2, 11)).toDF("a", "b", "c")
- covar_tab.createOrReplaceTempView("covar_tab")
+ withTempView("covar_tab") {
+ covar_tab.createOrReplaceTempView("covar_tab")
- checkAnswer(
- spark.sql(
- """
- |SELECT corr(b, c) FROM covar_tab WHERE a < 1
- """.stripMargin),
- Row(null) :: Nil)
+ checkAnswer(
+ spark.sql(
+ """
+ |SELECT corr(b, c) FROM covar_tab WHERE a < 1
+ """.stripMargin),
+ Row(null) :: Nil)
- checkAnswer(
- spark.sql(
- """
- |SELECT corr(b, c) FROM covar_tab WHERE a < 3
- """.stripMargin),
- Row(null) :: Nil)
+ checkAnswer(
+ spark.sql(
+ """
+ |SELECT corr(b, c) FROM covar_tab WHERE a < 3
+ """.stripMargin),
+ Row(null) :: Nil)
- checkAnswer(
- spark.sql(
- """
- |SELECT corr(b, c) FROM covar_tab WHERE a = 3
- """.stripMargin),
- Row(Double.NaN) :: Nil)
+ checkAnswer(
+ spark.sql(
+ """
+ |SELECT corr(b, c) FROM covar_tab WHERE a = 3
+ """.stripMargin),
+ Row(Double.NaN) :: Nil)
- checkAnswer(
- spark.sql(
- """
- |SELECT a, corr(b, c) FROM covar_tab GROUP BY a ORDER BY a
- """.stripMargin),
- Row(1, null) ::
- Row(2, null) ::
- Row(3, Double.NaN) ::
- Row(4, Double.NaN) ::
- Row(5, Double.NaN) ::
- Row(6, Double.NaN) :: Nil)
-
- val corr7 = spark.sql("SELECT corr(b, c) FROM covar_tab").collect()(0).getDouble(0)
- assert(math.abs(corr7 - 0.6633880657639323) < 1e-12)
+ checkAnswer(
+ spark.sql(
+ """
+ |SELECT a, corr(b, c) FROM covar_tab GROUP BY a ORDER BY a
+ """.stripMargin),
+ Row(1, null) ::
+ Row(2, null) ::
+ Row(3, Double.NaN) ::
+ Row(4, Double.NaN) ::
+ Row(5, Double.NaN) ::
+ Row(6, Double.NaN) :: Nil)
+
+ val corr7 = spark.sql("SELECT corr(b, c) FROM covar_tab").collect()(0).getDouble(0)
+ assert(math.abs(corr7 - 0.6633880657639323) < 1e-12)
+ }
}
test("covariance: covar_pop and covar_samp") {
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 08d2506..61e1fef 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
@@ -2315,37 +2315,39 @@ class HiveDDLSuite
implicit val _sqlContext = spark.sqlContext
- Seq((1, "one"), (2, "two"), (4, "four")).toDF("number", "word").createOrReplaceTempView("t1")
- // Make a table and ensure it will be broadcast.
- sql("""CREATE TABLE smallTable(word string, number int)
- |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
- |STORED AS TEXTFILE
- """.stripMargin)
+ withTempView("t1") {
+ Seq((1, "one"), (2, "two"), (4, "four")).toDF("number", "word").createOrReplaceTempView("t1")
+ // Make a table and ensure it will be broadcast.
+ sql("""CREATE TABLE smallTable(word string, number int)
+ |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ |STORED AS TEXTFILE
+ """.stripMargin)
- sql(
- """INSERT INTO smallTable
- |SELECT word, number from t1
- """.stripMargin)
+ sql(
+ """INSERT INTO smallTable
+ |SELECT word, number from t1
+ """.stripMargin)
- val inputData = MemoryStream[Int]
- val joined = inputData.toDS().toDF()
- .join(spark.table("smallTable"), $"value" === $"number")
+ val inputData = MemoryStream[Int]
+ val joined = inputData.toDS().toDF()
+ .join(spark.table("smallTable"), $"value" === $"number")
- val sq = joined.writeStream
- .format("memory")
- .queryName("t2")
- .start()
- try {
- inputData.addData(1, 2)
+ val sq = joined.writeStream
+ .format("memory")
+ .queryName("t2")
+ .start()
+ try {
+ inputData.addData(1, 2)
- sq.processAllAvailable()
+ sq.processAllAvailable()
- checkAnswer(
- spark.table("t2"),
- Seq(Row(1, "one", 1), Row(2, "two", 2))
- )
- } finally {
- sq.stop()
+ checkAnswer(
+ spark.table("t2"),
+ Seq(Row(1, "one", 1), Row(2, "two", 2))
+ )
+ } finally {
+ sq.stop()
+ }
}
}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
index 5a83650..1e89db2 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
@@ -698,15 +698,17 @@ class HiveQuerySuite extends HiveComparisonTest with SQLTestUtils with BeforeAnd
"SELECT srcalias.KEY, SRCALIAS.value FROM sRc SrCAlias WHERE SrCAlias.kEy < 15")
test("case sensitivity: created temporary view") {
- val testData =
- TestHive.sparkContext.parallelize(
- TestData(1, "str1") ::
- TestData(2, "str2") :: Nil)
- testData.toDF().createOrReplaceTempView("REGisteredTABle")
-
- assertResult(Array(Row(2, "str2"))) {
- sql("SELECT tablealias.A, TABLEALIAS.b FROM reGisteredTABle TableAlias " +
- "WHERE TableAliaS.a > 1").collect()
+ withTempView("REGisteredTABle") {
+ val testData =
+ TestHive.sparkContext.parallelize(
+ TestData(1, "str1") ::
+ TestData(2, "str2") :: Nil)
+ testData.toDF().createOrReplaceTempView("REGisteredTABle")
+
+ assertResult(Array(Row(2, "str2"))) {
+ sql("SELECT tablealias.A, TABLEALIAS.b FROM reGisteredTABle TableAlias " +
+ "WHERE TableAliaS.a > 1").collect()
+ }
}
}
@@ -725,16 +727,18 @@ class HiveQuerySuite extends HiveComparisonTest with SQLTestUtils with BeforeAnd
}
test("SPARK-2180: HAVING support in GROUP BY clauses (positive)") {
- val fixture = List(("foo", 2), ("bar", 1), ("foo", 4), ("bar", 3))
- .zipWithIndex.map {case ((value, attr), key) => HavingRow(key, value, attr)}
- TestHive.sparkContext.parallelize(fixture).toDF().createOrReplaceTempView("having_test")
- val results =
- sql("SELECT value, max(attr) AS attr FROM having_test GROUP BY value HAVING attr > 3")
- .collect()
- .map(x => (x.getString(0), x.getInt(1)))
+ withTempView("having_test") {
+ val fixture = List(("foo", 2), ("bar", 1), ("foo", 4), ("bar", 3))
+ .zipWithIndex.map {case ((value, attr), key) => HavingRow(key, value, attr)}
+ TestHive.sparkContext.parallelize(fixture).toDF().createOrReplaceTempView("having_test")
+ val results =
+ sql("SELECT value, max(attr) AS attr FROM having_test GROUP BY value HAVING attr > 3")
+ .collect()
+ .map(x => (x.getString(0), x.getInt(1)))
- assert(results === Array(("foo", 4)))
- TestHive.reset()
+ assert(results === Array(("foo", 4)))
+ TestHive.reset()
+ }
}
test("SPARK-2180: HAVING with non-boolean clause raises no exceptions") {
@@ -966,22 +970,24 @@ class HiveQuerySuite extends HiveComparisonTest with SQLTestUtils with BeforeAnd
}
test("SPARK-3414 regression: should store analyzed logical plan when creating a temporary view") {
- sparkContext.makeRDD(Seq.empty[LogEntry]).toDF().createOrReplaceTempView("rawLogs")
- sparkContext.makeRDD(Seq.empty[LogFile]).toDF().createOrReplaceTempView("logFiles")
+ withTempView("rawLogs", "logFiles", "boom") {
+ sparkContext.makeRDD(Seq.empty[LogEntry]).toDF().createOrReplaceTempView("rawLogs")
+ sparkContext.makeRDD(Seq.empty[LogFile]).toDF().createOrReplaceTempView("logFiles")
- sql(
- """
- SELECT name, message
- FROM rawLogs
- JOIN (
- SELECT name
- FROM logFiles
- ) files
- ON rawLogs.filename = files.name
- """).createOrReplaceTempView("boom")
-
- // This should be successfully analyzed
- sql("SELECT * FROM boom").queryExecution.analyzed
+ sql(
+ """
+ SELECT name, message
+ FROM rawLogs
+ JOIN (
+ SELECT name
+ FROM logFiles
+ ) files
+ ON rawLogs.filename = files.name
+ """).createOrReplaceTempView("boom")
+
+ // This should be successfully analyzed
+ sql("SELECT * FROM boom").queryExecution.analyzed
+ }
}
test("SPARK-3810: PreprocessTableInsertion static partitioning support") {
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
index e93f585..79c6ade 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
@@ -89,24 +89,28 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("script") {
- assume(TestUtils.testCommandAvailable("/bin/bash"))
- assume(TestUtils.testCommandAvailable("echo | sed"))
- val scriptFilePath = getTestResourcePath("test_script.sh")
- val df = Seq(("x1", "y1", "z1"), ("x2", "y2", "z2")).toDF("c1", "c2", "c3")
- df.createOrReplaceTempView("script_table")
- val query1 = sql(
- s"""
- |SELECT col1 FROM (from(SELECT c1, c2, c3 FROM script_table) tempt_table
- |REDUCE c1, c2, c3 USING 'bash $scriptFilePath' AS
- |(col1 STRING, col2 STRING)) script_test_table""".stripMargin)
- checkAnswer(query1, Row("x1_y1") :: Row("x2_y2") :: Nil)
+ withTempView("script_table") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ assume(TestUtils.testCommandAvailable("echo | sed"))
+ val scriptFilePath = getTestResourcePath("test_script.sh")
+ val df = Seq(("x1", "y1", "z1"), ("x2", "y2", "z2")).toDF("c1", "c2", "c3")
+ df.createOrReplaceTempView("script_table")
+ val query1 = sql(
+ s"""
+ |SELECT col1 FROM (from(SELECT c1, c2, c3 FROM script_table) tempt_table
+ |REDUCE c1, c2, c3 USING 'bash $scriptFilePath' AS
+ |(col1 STRING, col2 STRING)) script_test_table""".stripMargin)
+ checkAnswer(query1, Row("x1_y1") :: Row("x2_y2") :: Nil)
+ }
}
test("SPARK-6835: udtf in lateral view") {
- val df = Seq((1, 1)).toDF("c1", "c2")
- df.createOrReplaceTempView("table1")
- val query = sql("SELECT c1, v FROM table1 LATERAL VIEW stack(3, 1, c1 + 1, c1 + 2) d AS v")
- checkAnswer(query, Row(1, 1) :: Row(1, 2) :: Row(1, 3) :: Nil)
+ withTempView("table1") {
+ val df = Seq((1, 1)).toDF("c1", "c2")
+ df.createOrReplaceTempView("table1")
+ val query = sql("SELECT c1, v FROM table1 LATERAL VIEW stack(3, 1, c1 + 1, c1 + 2) d AS v")
+ checkAnswer(query, Row(1, 1) :: Row(1, 2) :: Row(1, 3) :: Nil)
+ }
}
test("SPARK-13651: generator outputs shouldn't be resolved from its child's output") {
@@ -120,70 +124,72 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("SPARK-6851: Self-joined converted parquet tables") {
- val orders = Seq(
- Order(1, "Atlas", "MTB", 234, "2015-01-07", "John D", "Pacifica", "CA", 20151),
- Order(3, "Swift", "MTB", 285, "2015-01-17", "John S", "Redwood City", "CA", 20151),
- Order(4, "Atlas", "Hybrid", 303, "2015-01-23", "Jones S", "San Mateo", "CA", 20151),
- Order(7, "Next", "MTB", 356, "2015-01-04", "Jane D", "Daly City", "CA", 20151),
- Order(10, "Next", "YFlikr", 187, "2015-01-09", "John D", "Fremont", "CA", 20151),
- Order(11, "Swift", "YFlikr", 187, "2015-01-23", "John D", "Hayward", "CA", 20151),
- Order(2, "Next", "Hybrid", 324, "2015-02-03", "Jane D", "Daly City", "CA", 20152),
- Order(5, "Next", "Street", 187, "2015-02-08", "John D", "Fremont", "CA", 20152),
- Order(6, "Atlas", "Street", 154, "2015-02-09", "John D", "Pacifica", "CA", 20152),
- Order(8, "Swift", "Hybrid", 485, "2015-02-19", "John S", "Redwood City", "CA", 20152),
- Order(9, "Atlas", "Split", 303, "2015-02-28", "Jones S", "San Mateo", "CA", 20152))
-
- val orderUpdates = Seq(
- Order(1, "Atlas", "MTB", 434, "2015-01-07", "John D", "Pacifica", "CA", 20151),
- Order(11, "Swift", "YFlikr", 137, "2015-01-23", "John D", "Hayward", "CA", 20151))
-
- orders.toDF.createOrReplaceTempView("orders1")
- orderUpdates.toDF.createOrReplaceTempView("orderupdates1")
-
- withTable("orders", "orderupdates") {
- sql(
- """CREATE TABLE orders(
- | id INT,
- | make String,
- | type String,
- | price INT,
- | pdate String,
- | customer String,
- | city String)
- |PARTITIONED BY (state STRING, month INT)
- |STORED AS PARQUET
- """.stripMargin)
+ withTempView("orders1", "orderupdates1") {
+ val orders = Seq(
+ Order(1, "Atlas", "MTB", 234, "2015-01-07", "John D", "Pacifica", "CA", 20151),
+ Order(3, "Swift", "MTB", 285, "2015-01-17", "John S", "Redwood City", "CA", 20151),
+ Order(4, "Atlas", "Hybrid", 303, "2015-01-23", "Jones S", "San Mateo", "CA", 20151),
+ Order(7, "Next", "MTB", 356, "2015-01-04", "Jane D", "Daly City", "CA", 20151),
+ Order(10, "Next", "YFlikr", 187, "2015-01-09", "John D", "Fremont", "CA", 20151),
+ Order(11, "Swift", "YFlikr", 187, "2015-01-23", "John D", "Hayward", "CA", 20151),
+ Order(2, "Next", "Hybrid", 324, "2015-02-03", "Jane D", "Daly City", "CA", 20152),
+ Order(5, "Next", "Street", 187, "2015-02-08", "John D", "Fremont", "CA", 20152),
+ Order(6, "Atlas", "Street", 154, "2015-02-09", "John D", "Pacifica", "CA", 20152),
+ Order(8, "Swift", "Hybrid", 485, "2015-02-19", "John S", "Redwood City", "CA", 20152),
+ Order(9, "Atlas", "Split", 303, "2015-02-28", "Jones S", "San Mateo", "CA", 20152))
+
+ val orderUpdates = Seq(
+ Order(1, "Atlas", "MTB", 434, "2015-01-07", "John D", "Pacifica", "CA", 20151),
+ Order(11, "Swift", "YFlikr", 137, "2015-01-23", "John D", "Hayward", "CA", 20151))
+
+ orders.toDF.createOrReplaceTempView("orders1")
+ orderUpdates.toDF.createOrReplaceTempView("orderupdates1")
+
+ withTable("orders", "orderupdates") {
+ sql(
+ """CREATE TABLE orders(
+ | id INT,
+ | make String,
+ | type String,
+ | price INT,
+ | pdate String,
+ | customer String,
+ | city String)
+ |PARTITIONED BY (state STRING, month INT)
+ |STORED AS PARQUET
+ """.stripMargin)
- sql(
- """CREATE TABLE orderupdates(
- | id INT,
- | make String,
- | type String,
- | price INT,
- | pdate String,
- | customer String,
- | city String)
- |PARTITIONED BY (state STRING, month INT)
- |STORED AS PARQUET
- """.stripMargin)
+ sql(
+ """CREATE TABLE orderupdates(
+ | id INT,
+ | make String,
+ | type String,
+ | price INT,
+ | pdate String,
+ | customer String,
+ | city String)
+ |PARTITIONED BY (state STRING, month INT)
+ |STORED AS PARQUET
+ """.stripMargin)
- sql("set hive.exec.dynamic.partition.mode=nonstrict")
- sql("INSERT INTO TABLE orders PARTITION(state, month) SELECT * FROM orders1")
- sql("INSERT INTO TABLE orderupdates PARTITION(state, month) SELECT * FROM orderupdates1")
+ sql("set hive.exec.dynamic.partition.mode=nonstrict")
+ sql("INSERT INTO TABLE orders PARTITION(state, month) SELECT * FROM orders1")
+ sql("INSERT INTO TABLE orderupdates PARTITION(state, month) SELECT * FROM orderupdates1")
- checkAnswer(
- sql(
- """
- |select orders.state, orders.month
- |from orders
- |join (
- | select distinct orders.state,orders.month
- | from orders
- | join orderupdates
- | on orderupdates.id = orders.id) ao
- | on ao.state = orders.state and ao.month = orders.month
- """.stripMargin),
- (1 to 6).map(_ => Row("CA", 20151)))
+ checkAnswer(
+ sql(
+ """
+ |select orders.state, orders.month
+ |from orders
+ |join (
+ | select distinct orders.state,orders.month
+ | from orders
+ | join orderupdates
+ | on orderupdates.id = orders.id) ao
+ | on ao.state = orders.state and ao.month = orders.month
+ """.stripMargin),
+ (1 to 6).map(_ => Row("CA", 20151)))
+ }
}
}
@@ -336,71 +342,76 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("SPARK-5371: union with null and sum") {
- val df = Seq((1, 1)).toDF("c1", "c2")
- df.createOrReplaceTempView("table1")
-
- val query = sql(
- """
- |SELECT
- | MIN(c1),
- | MIN(c2)
- |FROM (
- | SELECT
- | SUM(c1) c1,
- | NULL c2
- | FROM table1
- | UNION ALL
- | SELECT
- | NULL c1,
- | SUM(c2) c2
- | FROM table1
- |) a
- """.stripMargin)
- checkAnswer(query, Row(1, 1) :: Nil)
- }
-
- test("CTAS with WITH clause") {
+ withTempView("table1") {
+ val df = Seq((1, 1)).toDF("c1", "c2")
+ df.createOrReplaceTempView("table1")
- val df = Seq((1, 1)).toDF("c1", "c2")
- df.createOrReplaceTempView("table1")
- withTable("with_table1") {
- sql(
+ val query = sql(
"""
- |CREATE TABLE with_table1 AS
- |WITH T AS (
- | SELECT *
+ |SELECT
+ | MIN(c1),
+ | MIN(c2)
+ |FROM (
+ | SELECT
+ | SUM(c1) c1,
+ | NULL c2
+ | FROM table1
+ | UNION ALL
+ | SELECT
+ | NULL c1,
+ | SUM(c2) c2
| FROM table1
- |)
- |SELECT *
- |FROM T
+ |) a
""".stripMargin)
- val query = sql("SELECT * FROM with_table1")
checkAnswer(query, Row(1, 1) :: Nil)
}
}
+ test("CTAS with WITH clause") {
+ withTempView("table1") {
+ val df = Seq((1, 1)).toDF("c1", "c2")
+ df.createOrReplaceTempView("table1")
+ withTable("with_table1") {
+ sql(
+ """
+ |CREATE TABLE with_table1 AS
+ |WITH T AS (
+ | SELECT *
+ | FROM table1
+ |)
+ |SELECT *
+ |FROM T
+ """.stripMargin)
+ val query = sql("SELECT * FROM with_table1")
+ checkAnswer(query, Row(1, 1) :: Nil)
+ }
+ }
+ }
+
test("explode nested Field") {
- Seq(NestedArray1(NestedArray2(Seq(1, 2, 3)))).toDF.createOrReplaceTempView("nestedArray")
- checkAnswer(
- sql("SELECT ints FROM nestedArray LATERAL VIEW explode(a.b) a AS ints"),
- Row(1) :: Row(2) :: Row(3) :: Nil)
+ withTempView("nestedArray") {
+ Seq(NestedArray1(NestedArray2(Seq(1, 2, 3)))).toDF.createOrReplaceTempView("nestedArray")
+ checkAnswer(
+ sql("SELECT ints FROM nestedArray LATERAL VIEW explode(a.b) a AS ints"),
+ Row(1) :: Row(2) :: Row(3) :: Nil)
- checkAnswer(
- sql("SELECT `ints` FROM nestedArray LATERAL VIEW explode(a.b) `a` AS `ints`"),
- Row(1) :: Row(2) :: Row(3) :: Nil)
+ checkAnswer(
+ sql("SELECT `ints` FROM nestedArray LATERAL VIEW explode(a.b) `a` AS `ints`"),
+ Row(1) :: Row(2) :: Row(3) :: Nil)
- checkAnswer(
- sql("SELECT `a`.`ints` FROM nestedArray LATERAL VIEW explode(a.b) `a` AS `ints`"),
- Row(1) :: Row(2) :: Row(3) :: Nil)
+ checkAnswer(
+ sql("SELECT `a`.`ints` FROM nestedArray LATERAL VIEW explode(a.b) `a` AS `ints`"),
+ Row(1) :: Row(2) :: Row(3) :: Nil)
- checkAnswer(
- sql(
- """
- |SELECT `weird``tab`.`weird``col`
- |FROM nestedArray
- |LATERAL VIEW explode(a.b) `weird``tab` AS `weird``col`
- """.stripMargin),
- Row(1) :: Row(2) :: Row(3) :: Nil)
+ checkAnswer(
+ sql(
+ """
+ |SELECT `weird``tab`.`weird``col`
+ |FROM nestedArray
+ |LATERAL VIEW explode(a.b) `weird``tab` AS `weird``col`
+ """.stripMargin),
+ Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
}
test("SPARK-4512 Fix attribute reference resolution error when using SORT BY") {
@@ -742,20 +753,22 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("double nested data") {
- withTable("test_ctas_1234") {
- sparkContext.parallelize(Nested1(Nested2(Nested3(1))) :: Nil)
- .toDF().createOrReplaceTempView("nested")
- checkAnswer(
- sql("SELECT f1.f2.f3 FROM nested"),
- Row(1))
+ withTempView("nested") {
+ withTable("test_ctas_1234") {
+ sparkContext.parallelize(Nested1(Nested2(Nested3(1))) :: Nil)
+ .toDF().createOrReplaceTempView("nested")
+ checkAnswer(
+ sql("SELECT f1.f2.f3 FROM nested"),
+ Row(1))
- sql("CREATE TABLE test_ctas_1234 AS SELECT * from nested")
- checkAnswer(
- sql("SELECT * FROM test_ctas_1234"),
- sql("SELECT * FROM nested").collect().toSeq)
+ sql("CREATE TABLE test_ctas_1234 AS SELECT * from nested")
+ checkAnswer(
+ sql("SELECT * FROM test_ctas_1234"),
+ sql("SELECT * FROM nested").collect().toSeq)
- intercept[AnalysisException] {
- sql("CREATE TABLE test_ctas_1234 AS SELECT * from notexists").collect()
+ intercept[AnalysisException] {
+ sql("CREATE TABLE test_ctas_1234 AS SELECT * from notexists").collect()
+ }
}
}
}
@@ -832,13 +845,15 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("SPARK-4963 DataFrame sample on mutable row return wrong result") {
- sql("SELECT * FROM src WHERE key % 2 = 0")
- .sample(withReplacement = false, fraction = 0.3)
- .createOrReplaceTempView("sampled")
- (1 to 10).foreach { i =>
- checkAnswer(
- sql("SELECT * FROM sampled WHERE key % 2 = 1"),
- Seq.empty[Row])
+ withTempView("sampled") {
+ sql("SELECT * FROM src WHERE key % 2 = 0")
+ .sample(withReplacement = false, fraction = 0.3)
+ .createOrReplaceTempView("sampled")
+ (1 to 10).foreach { i =>
+ checkAnswer(
+ sql("SELECT * FROM sampled WHERE key % 2 = 1"),
+ Seq.empty[Row])
+ }
}
}
@@ -899,33 +914,38 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("resolve udtf in projection #1") {
- val ds = (1 to 5).map(i => s"""{"a":[$i, ${i + 1}]}""").toDS()
- read.json(ds).createOrReplaceTempView("data")
- val df = sql("SELECT explode(a) AS val FROM data")
- val col = df("val")
+ withTempView("data") {
+ val ds = (1 to 5).map(i => s"""{"a":[$i, ${i + 1}]}""").toDS()
+ read.json(ds).createOrReplaceTempView("data")
+ sql("SELECT explode(a) AS val FROM data")
+ }
}
test("resolve udtf in projection #2") {
- val ds = (1 to 2).map(i => s"""{"a":[$i, ${i + 1}]}""").toDS()
- read.json(ds).createOrReplaceTempView("data")
- checkAnswer(sql("SELECT explode(map(1, 1)) FROM data LIMIT 1"), Row(1, 1) :: Nil)
- checkAnswer(sql("SELECT explode(map(1, 1)) as (k1, k2) FROM data LIMIT 1"), Row(1, 1) :: Nil)
- intercept[AnalysisException] {
- sql("SELECT explode(map(1, 1)) as k1 FROM data LIMIT 1")
- }
+ withTempView("data") {
+ val ds = (1 to 2).map(i => s"""{"a":[$i, ${i + 1}]}""").toDS()
+ read.json(ds).createOrReplaceTempView("data")
+ checkAnswer(sql("SELECT explode(map(1, 1)) FROM data LIMIT 1"), Row(1, 1) :: Nil)
+ checkAnswer(sql("SELECT explode(map(1, 1)) as (k1, k2) FROM data LIMIT 1"), Row(1, 1) :: Nil)
+ intercept[AnalysisException] {
+ sql("SELECT explode(map(1, 1)) as k1 FROM data LIMIT 1")
+ }
- intercept[AnalysisException] {
- sql("SELECT explode(map(1, 1)) as (k1, k2, k3) FROM data LIMIT 1")
+ intercept[AnalysisException] {
+ sql("SELECT explode(map(1, 1)) as (k1, k2, k3) FROM data LIMIT 1")
+ }
}
}
// TGF with non-TGF in project is allowed in Spark SQL, but not in Hive
test("TGF with non-TGF in projection") {
- val ds = Seq("""{"a": "1", "b":"1"}""").toDS()
- read.json(ds).createOrReplaceTempView("data")
- checkAnswer(
- sql("SELECT explode(map(a, b)) as (k1, k2), a, b FROM data"),
- Row("1", "1", "1", "1") :: Nil)
+ withTempView("data") {
+ val ds = Seq("""{"a": "1", "b":"1"}""").toDS()
+ read.json(ds).createOrReplaceTempView("data")
+ checkAnswer(
+ sql("SELECT explode(map(a, b)) as (k1, k2), a, b FROM data"),
+ Row("1", "1", "1", "1") :: Nil)
+ }
}
test("logical.Project should not be resolved if it contains aggregates or generators") {
@@ -975,47 +995,58 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("SPARK-5203 union with different decimal precision") {
- Seq.empty[(java.math.BigDecimal, java.math.BigDecimal)]
- .toDF("d1", "d2")
- .select($"d1".cast(DecimalType(10, 5)).as("d"))
- .createOrReplaceTempView("dn")
+ withTempView("dn") {
+ Seq.empty[(java.math.BigDecimal, java.math.BigDecimal)]
+ .toDF("d1", "d2")
+ .select($"d1".cast(DecimalType(10, 5)).as("d"))
+ .createOrReplaceTempView("dn")
- sql("select d from dn union all select d * 2 from dn")
- .queryExecution.analyzed
+ sql("select d from dn union all select d * 2 from dn")
+ .queryExecution.analyzed
+ }
}
test("Star Expansion - script transform") {
- assume(TestUtils.testCommandAvailable("/bin/bash"))
- val data = (1 to 100000).map { i => (i, i, i) }
- data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
- assert(100000 === sql("SELECT TRANSFORM (*) USING 'cat' FROM script_trans").count())
+ withTempView("script_trans") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val data = (1 to 100000).map { i => (i, i, i) }
+ data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
+ assert(100000 === sql("SELECT TRANSFORM (*) USING 'cat' FROM script_trans").count())
+ }
}
test("test script transform for stdout") {
- assume(TestUtils.testCommandAvailable("/bin/bash"))
- val data = (1 to 100000).map { i => (i, i, i) }
- data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
- assert(100000 ===
- sql("SELECT TRANSFORM (d1, d2, d3) USING 'cat' AS (a,b,c) FROM script_trans").count())
+ withTempView("script_trans") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val data = (1 to 100000).map { i => (i, i, i) }
+ data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
+ assert(100000 ===
+ sql("SELECT TRANSFORM (d1, d2, d3) USING 'cat' AS (a,b,c) FROM script_trans").count())
+ }
}
test("test script transform for stderr") {
- assume(TestUtils.testCommandAvailable("/bin/bash"))
- val data = (1 to 100000).map { i => (i, i, i) }
- data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
- assert(0 ===
- sql("SELECT TRANSFORM (d1, d2, d3) USING 'cat 1>&2' AS (a,b,c) FROM script_trans").count())
+ withTempView("script_trans") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val data = (1 to 100000).map { i => (i, i, i) }
+ data.toDF("d1", "d2", "d3").createOrReplaceTempView("script_trans")
+ assert(0 ===
+ sql("SELECT TRANSFORM (d1, d2, d3) USING 'cat 1>&2' AS (a,b,c) FROM script_trans").count())
+ }
}
test("test script transform data type") {
- assume(TestUtils.testCommandAvailable("/bin/bash"))
- val data = (1 to 5).map { i => (i, i) }
- data.toDF("key", "value").createOrReplaceTempView("test")
- checkAnswer(
- sql("""FROM
- |(FROM test SELECT TRANSFORM(key, value) USING 'cat' AS (`thing1` int, thing2 string)) t
- |SELECT thing1 + 1
- """.stripMargin), (2 to 6).map(i => Row(i)))
+ withTempView("test") {
+ assume(TestUtils.testCommandAvailable("/bin/bash"))
+ val data = (1 to 5).map { i => (i, i) }
+ data.toDF("key", "value").createOrReplaceTempView("test")
+ checkAnswer(
+ sql(
+ """FROM
+ |(FROM test SELECT TRANSFORM(key, value) USING 'cat' AS (`thing1` int, thing2 string)) t
+ |SELECT thing1 + 1
+ """.stripMargin), (2 to 6).map(i => Row(i)))
+ }
}
test("Sorting columns are not in Generate") {
@@ -1046,27 +1077,31 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("test case key when") {
- (1 to 5).map(i => (i, i.toString)).toDF("k", "v").createOrReplaceTempView("t")
- checkAnswer(
- sql("SELECT CASE k WHEN 2 THEN 22 WHEN 4 THEN 44 ELSE 0 END, v FROM t"),
- Row(0, "1") :: Row(22, "2") :: Row(0, "3") :: Row(44, "4") :: Row(0, "5") :: Nil)
+ withTempView("t") {
+ (1 to 5).map(i => (i, i.toString)).toDF("k", "v").createOrReplaceTempView("t")
+ checkAnswer(
+ sql("SELECT CASE k WHEN 2 THEN 22 WHEN 4 THEN 44 ELSE 0 END, v FROM t"),
+ Row(0, "1") :: Row(22, "2") :: Row(0, "3") :: Row(44, "4") :: Row(0, "5") :: Nil)
+ }
}
test("SPARK-7269 Check analysis failed in case in-sensitive") {
- Seq(1, 2, 3).map { i =>
- (i.toString, i.toString)
- }.toDF("key", "value").createOrReplaceTempView("df_analysis")
- sql("SELECT kEy from df_analysis group by key").collect()
- sql("SELECT kEy+3 from df_analysis group by key+3").collect()
- sql("SELECT kEy+3, a.kEy, A.kEy from df_analysis A group by key").collect()
- sql("SELECT cast(kEy+1 as Int) from df_analysis A group by cast(key+1 as int)").collect()
- sql("SELECT cast(kEy+1 as Int) from df_analysis A group by key+1").collect()
- sql("SELECT 2 from df_analysis A group by key+1").collect()
- intercept[AnalysisException] {
- sql("SELECT kEy+1 from df_analysis group by key+3")
- }
- intercept[AnalysisException] {
- sql("SELECT cast(key+2 as Int) from df_analysis A group by cast(key+1 as int)")
+ withTempView("df_analysis") {
+ Seq(1, 2, 3).map { i =>
+ (i.toString, i.toString)
+ }.toDF("key", "value").createOrReplaceTempView("df_analysis")
+ sql("SELECT kEy from df_analysis group by key").collect()
+ sql("SELECT kEy+3 from df_analysis group by key+3").collect()
+ sql("SELECT kEy+3, a.kEy, A.kEy from df_analysis A group by key").collect()
+ sql("SELECT cast(kEy+1 as Int) from df_analysis A group by cast(key+1 as int)").collect()
+ sql("SELECT cast(kEy+1 as Int) from df_analysis A group by key+1").collect()
+ sql("SELECT 2 from df_analysis A group by key+1").collect()
+ intercept[AnalysisException] {
+ sql("SELECT kEy+1 from df_analysis group by key+3")
+ }
+ intercept[AnalysisException] {
+ sql("SELECT cast(key+2 as Int) from df_analysis A group by cast(key+1 as int)")
+ }
}
}
@@ -1179,10 +1214,12 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
}
test("SPARK-9371: fix the support for special chars in column names for hive context") {
- val ds = Seq("""{"a": {"c.b": 1}, "b.$q": [{"a@!.q": 1}], "q.w": {"w.i&": [1]}}""").toDS()
- read.json(ds).createOrReplaceTempView("t")
+ withTempView("t") {
+ val ds = Seq("""{"a": {"c.b": 1}, "b.$q": [{"a@!.q": 1}], "q.w": {"w.i&": [1]}}""").toDS()
+ read.json(ds).createOrReplaceTempView("t")
- checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
+ checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
+ }
}
test("specifying database name for a temporary view is not allowed") {
@@ -1237,43 +1274,47 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
ignore("SPARK-10310: " +
"script transformation using default input/output SerDe and record reader/writer") {
- spark
- .range(5)
- .selectExpr("id AS a", "id AS b")
- .createOrReplaceTempView("test")
+ withTempView("test") {
+ spark
+ .range(5)
+ .selectExpr("id AS a", "id AS b")
+ .createOrReplaceTempView("test")
- val scriptFilePath = getTestResourcePath("data")
- checkAnswer(
- sql(
- s"""FROM(
- | FROM test SELECT TRANSFORM(a, b)
- | USING 'python $scriptFilePath/scripts/test_transform.py "\t"'
- | AS (c STRING, d STRING)
- |) t
- |SELECT c
- """.stripMargin),
- (0 until 5).map(i => Row(i + "#")))
+ val scriptFilePath = getTestResourcePath("data")
+ checkAnswer(
+ sql(
+ s"""FROM(
+ | FROM test SELECT TRANSFORM(a, b)
+ | USING 'python $scriptFilePath/scripts/test_transform.py "\t"'
+ | AS (c STRING, d STRING)
+ |) t
+ |SELECT c
+ """.stripMargin),
+ (0 until 5).map(i => Row(i + "#")))
+ }
}
ignore("SPARK-10310: script transformation using LazySimpleSerDe") {
- spark
- .range(5)
- .selectExpr("id AS a", "id AS b")
- .createOrReplaceTempView("test")
-
- val scriptFilePath = getTestResourcePath("data")
- val df = sql(
- s"""FROM test
- |SELECT TRANSFORM(a, b)
- |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
- |WITH SERDEPROPERTIES('field.delim' = '|')
- |USING 'python $scriptFilePath/scripts/test_transform.py "|"'
- |AS (c STRING, d STRING)
- |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
- |WITH SERDEPROPERTIES('field.delim' = '|')
- """.stripMargin)
+ withTempView("test") {
+ spark
+ .range(5)
+ .selectExpr("id AS a", "id AS b")
+ .createOrReplaceTempView("test")
+
+ val scriptFilePath = getTestResourcePath("data")
+ val df = sql(
+ s"""FROM test
+ |SELECT TRANSFORM(a, b)
+ |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ |WITH SERDEPROPERTIES('field.delim' = '|')
+ |USING 'python $scriptFilePath/scripts/test_transform.py "|"'
+ |AS (c STRING, d STRING)
+ |ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ |WITH SERDEPROPERTIES('field.delim' = '|')
+ """.stripMargin)
- checkAnswer(df, (0 until 5).map(i => Row(i + "#", i + "#")))
+ checkAnswer(df, (0 until 5).map(i => Row(i + "#", i + "#")))
+ }
}
test("SPARK-10741: Sort on Aggregate using parquet") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org