This is an automated email from the ASF dual-hosted git repository.
maxgekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 02ae919a6ae [SPARK-41021][SQL][TESTS] Test some subclasses of error
class DATATYPE_MISMATCH
02ae919a6ae is described below
commit 02ae919a6ae4f37ef859188db629ab564d6d42f5
Author: panbingkun <pbk1...@gmail.com>
AuthorDate: Mon Nov 7 12:19:21 2022 +0300
[SPARK-41021][SQL][TESTS] Test some subclasses of error class
DATATYPE_MISMATCH
### What changes were proposed in this pull request?
This PR aims to add new UT for some error classes, include:
1. DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES
2. DATATYPE_MISMATCH.CANNOT_CONVERT_TO_JSON
3. DATATYPE_MISMATCH.FRAME_LESS_OFFSET_WITHOUT_FOLDABLE
4. DATATYPE_MISMATCH.MAP_FROM_ENTRIES_WRONG_TYPE
5. DATATYPE_MISMATCH.NON_STRING_TYPE
6. DATATYPE_MISMATCH.NULL_TYPE
7. DATATYPE_MISMATCH.SPECIFIED_WINDOW_FRAME_DIFF_TYPES
8. DATATYPE_MISMATCH.SPECIFIED_WINDOW_FRAME_UNACCEPTED_TYPE (already exists)
https://github.com/apache/spark/blob/7009ef0510dae444c72e7513357e681b08379603/sql/core/src/test/resources/sql-tests/results/window.sql.out#L106
10. DATATYPE_MISMATCH.SPECIFIED_WINDOW_FRAME_WITHOUT_FOLDABLE
11. DATATYPE_MISMATCH.UNSPECIFIED_FRAME
### Why are the changes needed?
The changes improve the error framework.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass GA.
Closes #38520 from panbingkun/DATATYPE_MISMATCH_test.
Authored-by: panbingkun <pbk1...@gmail.com>
Signed-off-by: Max Gekk <max.g...@gmail.com>
---
.../analysis/ExpressionTypeCheckingSuite.scala | 48 ++++++++
.../expressions/CollectionExpressionsSuite.scala | 24 ++++
.../expressions/JsonExpressionsSuite.scala | 23 ++++
.../apache/spark/sql/DataFrameFunctionsSuite.scala | 137 ++++++++++++++++++---
.../spark/sql/DataFrameWindowFramesSuite.scala | 45 ++++++-
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 44 ++++++-
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 32 +++++
7 files changed, 333 insertions(+), 20 deletions(-)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala
index f656131c8e7..b192f12d569 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ExpressionTypeCheckingSuite.scala
@@ -765,4 +765,52 @@ class ExpressionTypeCheckingSuite extends SparkFunSuite
with SQLHelper with Quer
)
)
}
+
+ test("check types for Lag") {
+ val lag = Lag(Literal(1), NonFoldableLiteral(10), Literal(null), true)
+ assert(lag.checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "FRAME_LESS_OFFSET_WITHOUT_FOLDABLE",
+ messageParameters = Map("offset" -> "\"(- nonfoldableliteral())\"")
+ ))
+ }
+
+ test("check types for SpecifiedWindowFrame") {
+ val swf1 = SpecifiedWindowFrame(RangeFrame, Literal(10.0),
Literal(2147483648L))
+ assert(swf1.checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "SPECIFIED_WINDOW_FRAME_DIFF_TYPES",
+ messageParameters = Map(
+ "lower" -> "\"10.0\"",
+ "upper" -> "\"2147483648\"",
+ "lowerType" -> "\"DOUBLE\"",
+ "upperType" -> "\"BIGINT\""
+ )
+ )
+ )
+
+ val swf2 = SpecifiedWindowFrame(RangeFrame, NonFoldableLiteral(10.0),
Literal(2147483648L))
+ assert(swf2.checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "SPECIFIED_WINDOW_FRAME_WITHOUT_FOLDABLE",
+ messageParameters = Map(
+ "location" -> "lower",
+ "expression" -> "\"nonfoldableliteral()\""
+ )
+ )
+ )
+ }
+
+ test("check types for WindowSpecDefinition") {
+ val wsd = WindowSpecDefinition(
+ UnresolvedAttribute("a") :: Nil,
+ SortOrder(UnresolvedAttribute("b"), Ascending) :: Nil,
+ UnspecifiedFrame)
+ assert(wsd.checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "UNSPECIFIED_FRAME",
+ messageParameters = Map.empty
+ )
+ )
+ }
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
index 9839b784e60..676fb615e48 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CollectionExpressionsSuite.scala
@@ -28,6 +28,7 @@ import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.util.{DateTimeTestUtils, DateTimeUtils}
import
org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{outstandingZoneIds, LA,
UTC}
import org.apache.spark.sql.catalyst.util.IntervalUtils._
@@ -99,6 +100,17 @@ class CollectionExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper
}
test("MapContainsKey") {
+ val left = Literal.create(Map("a" -> "1", "b" -> "2"), MapType(StringType,
StringType))
+ val right = Literal.create(null, NullType)
+ assert(MapContainsKey(left, right).checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "NULL_TYPE",
+ messageParameters = Map("functionName" -> "`map_contains_key`")
+ )
+ )
+ }
+
+ test("ArrayContains") {
val m0 = Literal.create(Map("a" -> "1", "b" -> "2"), MapType(StringType,
StringType))
val m1 = Literal.create(null, MapType(StringType, StringType))
checkEvaluation(ArrayContains(MapKeys(m0), Literal("a")), true)
@@ -346,6 +358,18 @@ class CollectionExpressionsSuite extends SparkFunSuite
with ExpressionEvalHelper
assert(errorSubClass === "INVALID_MAP_KEY_TYPE")
assert(messageParameters === Map("keyType" -> "\"MAP<INT, INT>\""))
}
+
+ // accepts only arrays of pair structs
+ val mapWrongType = MapFromEntries(Literal(1))
+ assert(mapWrongType.checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "MAP_FROM_ENTRIES_WRONG_TYPE",
+ messageParameters = Map(
+ "functionName" -> "`map_from_entries`",
+ "childExpr" -> "\"1\"",
+ "childType" -> "\"INT\"")
+ )
+ )
}
test("Sort Array") {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index ab6da87287b..6aebc6b31d1 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -25,6 +25,7 @@ import org.scalatest.exceptions.TestFailedException
import org.apache.spark.{SparkException, SparkFunSuite}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.plans.PlanTestBase
import org.apache.spark.sql.catalyst.util._
@@ -406,6 +407,15 @@ class JsonExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper with
InternalRow(UTF8String.fromString("1"), null,
UTF8String.fromString("1")))
}
+ test("json_tuple - all arguments must be strings") {
+ assert(JsonTuple(Seq(Literal(888), Literal(999))).checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "NON_STRING_TYPE",
+ messageParameters = Map("funcName" -> "`json_tuple`")
+ )
+ )
+ }
+
test("from_json escaping") {
val schema = StructType(StructField("\"quote", IntegerType) :: Nil)
GenerateUnsafeProjection.generate(
@@ -581,6 +591,19 @@ class JsonExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper with
)
}
+ test("to_json - struct: unable to convert column of ObjectType to JSON") {
+ val schema = StructType(StructField("a",
ObjectType(classOf[java.lang.Integer])) :: Nil)
+ val structData = Literal.create(create_row(Integer.valueOf(1)), schema)
+ assert(StructsToJson(Map.empty, structData).checkInputDataTypes() ==
+ DataTypeMismatch(
+ errorSubClass = "CANNOT_CONVERT_TO_JSON",
+ messageParameters = Map(
+ "name" -> "`a`",
+ "type" -> "\"JAVA.LANG.INTEGER\"")
+ )
+ )
+ }
+
test("to_json - array") {
val inputSchema = ArrayType(StructType(StructField("a", IntegerType) ::
Nil))
val input = new GenericArrayData(InternalRow(1) :: InternalRow(2) :: Nil)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
index fe843c236c3..589946e77bc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
@@ -958,6 +958,24 @@ class DataFrameFunctionsSuite extends QueryTest with
SharedSparkSession {
testNonPrimitiveType()
}
+ test("map_contains_key function") {
+ val df = Seq(1, 2).toDF("a")
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr("map_contains_key(a, null)").collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.NULL_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"map_contains_key(a, NULL)\"",
+ "functionName" -> "`map_contains_key`"),
+ context = ExpectedContext(
+ fragment = "map_contains_key(a, null)",
+ start = 0,
+ stop = 24
+ )
+ )
+ }
+
test("map_concat function") {
val df1 = Seq(
(Map[Int, Int](1 -> 100, 2 -> 200), Map[Int, Int](3 -> 300, 4 -> 400)),
@@ -1128,6 +1146,20 @@ class DataFrameFunctionsSuite extends QueryTest with
SharedSparkSession {
// Test with cached relation, the Project will be evaluated with codegen
sdf.cache()
testNonPrimitiveType()
+
+ val wrongTypeDF = Seq(1, 2).toDF("a")
+ checkError(
+ exception = intercept[AnalysisException] {
+ wrongTypeDF.select(map_from_entries($"a"))
+ },
+ errorClass = "DATATYPE_MISMATCH.MAP_FROM_ENTRIES_WRONG_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"map_from_entries(a)\"",
+ "functionName" -> "`map_from_entries`",
+ "childExpr" -> "\"a\"",
+ "childType" -> "\"INT\""
+ )
+ )
}
test("array contains function") {
@@ -1825,28 +1857,97 @@ class DataFrameFunctionsSuite extends QueryTest with
SharedSparkSession {
checkAnswer(df5.selectExpr("array_union(a, b)"), ans5)
val df6 = Seq((null, Array("a"))).toDF("a", "b")
- assert(intercept[AnalysisException] {
- df6.select(array_union($"a", $"b"))
- }.getMessage.contains("data type mismatch"))
- assert(intercept[AnalysisException] {
- df6.selectExpr("array_union(a, b)")
- }.getMessage.contains("data type mismatch"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ df6.select(array_union($"a", $"b"))
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"VOID\"",
+ "rightType" -> "\"ARRAY<STRING>\""))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ df6.selectExpr("array_union(a, b)")
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"VOID\"",
+ "rightType" -> "\"ARRAY<STRING>\""),
+ context = ExpectedContext(
+ fragment = "array_union(a, b)",
+ start = 0,
+ stop = 16
+ )
+ )
val df7 = Seq((null, null)).toDF("a", "b")
- assert(intercept[AnalysisException] {
- df7.select(array_union($"a", $"b"))
- }.getMessage.contains("data type mismatch"))
- assert(intercept[AnalysisException] {
- df7.selectExpr("array_union(a, b)")
- }.getMessage.contains("data type mismatch"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ df7.select(array_union($"a", $"b"))
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"VOID\"",
+ "rightType" -> "\"VOID\"")
+ )
+ checkError(
+ exception = intercept[AnalysisException] {
+ df7.selectExpr("array_union(a, b)")
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"VOID\"",
+ "rightType" -> "\"VOID\""),
+ context = ExpectedContext(
+ fragment = "array_union(a, b)",
+ start = 0,
+ stop = 16
+ )
+ )
val df8 = Seq((Array(Array(1)), Array("a"))).toDF("a", "b")
- assert(intercept[AnalysisException] {
- df8.select(array_union($"a", $"b"))
- }.getMessage.contains("data type mismatch"))
- assert(intercept[AnalysisException] {
- df8.selectExpr("array_union(a, b)")
- }.getMessage.contains("data type mismatch"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ df8.select(array_union($"a", $"b"))
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"ARRAY<ARRAY<INT>>\"",
+ "rightType" -> "\"ARRAY<STRING>\"")
+ )
+ checkError(
+ exception = intercept[AnalysisException] {
+ df8.selectExpr("array_union(a, b)")
+ },
+ errorClass = "DATATYPE_MISMATCH.BINARY_ARRAY_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"array_union(a, b)\"",
+ "functionName" -> "`array_union`",
+ "arrayType" -> "\"ARRAY\"",
+ "leftType" -> "\"ARRAY<ARRAY<INT>>\"",
+ "rightType" -> "\"ARRAY<STRING>\""),
+ context = ExpectedContext(
+ fragment = "array_union(a, b)",
+ start = 0,
+ stop = 16
+ )
+ )
}
test("concat function - arrays") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFramesSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFramesSuite.scala
index 8616dc550b2..d68b1dde96e 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFramesSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFramesSuite.scala
@@ -17,9 +17,11 @@
package org.apache.spark.sql
-import org.apache.spark.sql.expressions.Window
+import org.apache.spark.sql.catalyst.expressions.{Ascending, Literal,
NonFoldableLiteral, RangeFrame, SortOrder, SpecifiedWindowFrame}
+import org.apache.spark.sql.expressions.{Window, WindowSpec}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.CalendarIntervalType
/**
* Window frame testing for DataFrame API.
@@ -419,4 +421,45 @@ class DataFrameWindowFramesSuite extends QueryTest with
SharedSparkSession {
lag("i", 1).over(Window.partitionBy("n").orderBy("i").rowsBetween(-1,
-1))),
res)
}
+
+ test("Window frame bounds lower and upper do not have the same type") {
+ val df = Seq((1L, "1"), (1L, "1")).toDF("key", "value")
+ val windowSpec = new WindowSpec(
+ Seq(Column("value").expr),
+ Seq(SortOrder(Column("key").expr, Ascending)),
+ SpecifiedWindowFrame(RangeFrame, Literal.create(null,
CalendarIntervalType), Literal(2))
+ )
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.select($"key", count("key").over(windowSpec)).collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.SPECIFIED_WINDOW_FRAME_DIFF_TYPES",
+ parameters = Map(
+ "sqlExpr" -> "\"RANGE BETWEEN NULL FOLLOWING AND 2 FOLLOWING\"",
+ "lower" -> "\"NULL\"",
+ "upper" -> "\"2\"",
+ "lowerType" -> "\"INTERVAL\"",
+ "upperType" -> "\"BIGINT\""
+ )
+ )
+ }
+
+ test("Window frame lower bound is not a literal") {
+ val df = Seq((1L, "1"), (1L, "1")).toDF("key", "value")
+ val windowSpec = new WindowSpec(
+ Seq(Column("value").expr),
+ Seq(SortOrder(Column("key").expr, Ascending)),
+ SpecifiedWindowFrame(RangeFrame, NonFoldableLiteral(1), Literal(2))
+ )
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.select($"key", count("key").over(windowSpec)).collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.SPECIFIED_WINDOW_FRAME_WITHOUT_FOLDABLE",
+ parameters = Map(
+ "sqlExpr" -> "\"RANGE BETWEEN nonfoldableliteral() FOLLOWING AND 2
FOLLOWING\"",
+ "location" -> "lower",
+ "expression" -> "\"nonfoldableliteral()\"")
+ )
+ }
}
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 fee33d55b51..5a543547708 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql
import org.scalatest.matchers.must.Matchers.the
import org.apache.spark.TestUtils.{assertNotSpilled, assertSpilled}
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
Expression, Lag, Literal, NonFoldableLiteral}
import org.apache.spark.sql.catalyst.optimizer.TransposeWindow
import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
@@ -846,6 +846,48 @@ class DataFrameWindowFunctionsSuite extends QueryTest
"v", "z", null, "v", "z", "y", "va")))
}
+ test("lag - Offset expression <offset> must be a literal") {
+ val nullStr: String = null
+ val df = Seq(
+ ("a", 0, nullStr),
+ ("a", 1, "x"),
+ ("b", 2, nullStr),
+ ("c", 3, nullStr),
+ ("a", 4, "y"),
+ ("b", 5, nullStr),
+ ("a", 6, "z"),
+ ("a", 7, "v"),
+ ("a", 8, nullStr)).
+ toDF("key", "order", "value")
+ val window = Window.orderBy($"order")
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.select(
+ $"key",
+ $"order",
+ $"value",
+ lead($"value", 1).over(window),
+ lead($"value", 2).over(window),
+ lead($"value", 0, null, true).over(window),
+ lead($"value", 1, null, true).over(window),
+ lead($"value", 2, null, true).over(window),
+ lead($"value", 3, null, true).over(window),
+ lead(concat($"value", $"key"), 1, null, true).over(window),
+ Column(Lag($"value".expr, NonFoldableLiteral(1), Literal(null),
true)).over(window),
+ lag($"value", 2).over(window),
+ lag($"value", 0, null, true).over(window),
+ lag($"value", 1, null, true).over(window),
+ lag($"value", 2, null, true).over(window),
+ lag($"value", 3, null, true).over(window),
+ lag(concat($"value", $"key"), 1, null,
true).over(window)).orderBy($"order").collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.FRAME_LESS_OFFSET_WITHOUT_FOLDABLE",
+ parameters = Map(
+ "sqlExpr" -> "\"lag(value, nonfoldableliteral(), NULL)\"",
+ "offset" -> "\"(- nonfoldableliteral())\"")
+ )
+ }
+
test("SPARK-12989 ExtractWindowExpressions treats alias as regular
attribute") {
val src = Seq((0, 3, 5)).toDF("a", "b", "c")
.withColumn("Data", struct("a", "b"))
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 e3c61fe1b25..0a84e9000c3 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
@@ -24,6 +24,8 @@ import java.util.Locale
import collection.JavaConverters._
import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Literal, StructsToJson}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -95,6 +97,18 @@ class JsonFunctionsSuite extends QueryTest with
SharedSparkSession {
checkAnswer(
df.selectExpr("key", "json_tuple(jstring, 'f1', 'f2', 'f3', 'f4',
'f5')"),
expected)
+
+ val nonStringDF = Seq(1, 2).toDF("a")
+ checkError(
+ exception = intercept[AnalysisException] {
+ nonStringDF.select(json_tuple($"a", "1")).collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.NON_STRING_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"json_tuple(a, 1)\"",
+ "funcName" -> "`json_tuple`"
+ )
+ )
}
test("json_tuple filter and group") {
@@ -1097,4 +1111,22 @@ class JsonFunctionsSuite extends QueryTest with
SharedSparkSession {
.selectExpr("value['key']")
checkAnswer(fromJsonDF, Row(localDT))
}
+
+ test("to_json: unable to convert column of ObjectType to JSON") {
+ val df = Seq(1).toDF("a")
+ val schema = StructType(StructField("b",
ObjectType(classOf[java.lang.Integer])) :: Nil)
+ val row = InternalRow.fromSeq(Seq(Integer.valueOf(1)))
+ val structData = Literal.create(row, schema)
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.select($"a").withColumn("c", Column(StructsToJson(Map.empty,
structData))).collect()
+ },
+ errorClass = "DATATYPE_MISMATCH.CANNOT_CONVERT_TO_JSON",
+ parameters = Map(
+ "sqlExpr" -> "\"to_json(NAMED_STRUCT('b', 1))\"",
+ "name" -> "`b`",
+ "type" -> "\"JAVA.LANG.INTEGER\""
+ )
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
