You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2021/08/17 12:11:23 UTC
[spark] branch branch-3.1 updated: [SPARK-36379][SQL][3.1] Null at
root level of a JSON array should not fail w/ permissive mode
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 32d127d [SPARK-36379][SQL][3.1] Null at root level of a JSON array should not fail w/ permissive mode
32d127d is described below
commit 32d127de4a4a628276e659bd6a5d572c625ed565
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Tue Aug 17 21:10:44 2021 +0900
[SPARK-36379][SQL][3.1] Null at root level of a JSON array should not fail w/ permissive mode
This PR backports https://github.com/apache/spark/pull/33608 to branch-3.1
-----------------------------------------------------------------------------
### What changes were proposed in this pull request?
This PR proposes to fail properly so JSON parser can proceed and parse the input with the permissive mode.
Previously, we passed `null`s as are, the root `InternalRow`s became `null`s, and it causes the query fails even with permissive mode on.
Now, we fail explicitly if `null` is passed when the input array contains `null`.
Note that this is consistent with non-array JSON input:
**Permissive mode:**
```scala
spark.read.json(Seq("""{"a": "str"}""", """null""").toDS).collect()
```
```
res0: Array[org.apache.spark.sql.Row] = Array([str], [null])
```
**Failfast mode**:
```scala
spark.read.option("mode", "failfast").json(Seq("""{"a": "str"}""", """null""").toDS).collect()
```
```
org.apache.spark.SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:70)
at org.apache.spark.sql.DataFrameReader.$anonfun$json$7(DataFrameReader.scala:540)
at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)
```
### Why are the changes needed?
To make the permissive mode to proceed and parse without throwing an exception.
### Does this PR introduce _any_ user-facing change?
**Permissive mode:**
```scala
spark.read.json(Seq("""[{"a": "str"}, null]""").toDS).collect()
```
Before:
```
java.lang.NullPointerException
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
```
After:
```
res0: Array[org.apache.spark.sql.Row] = Array([null])
```
NOTE that this behaviour is consistent when JSON object is malformed:
```scala
spark.read.schema("a int").json(Seq("""[{"a": 123}, {123123}, {"a": 123}]""").toDS).collect()
```
```
res0: Array[org.apache.spark.sql.Row] = Array([null])
```
Since we're parsing _one_ JSON array, related records all fail together.
**Failfast mode:**
```scala
spark.read.option("mode", "failfast").json(Seq("""[{"a": "str"}, null]""").toDS).collect()
```
Before:
```
java.lang.NullPointerException
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
```
After:
```
org.apache.spark.SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:70)
at org.apache.spark.sql.DataFrameReader.$anonfun$json$7(DataFrameReader.scala:540)
at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:484)
```
### How was this patch tested?
Manually tested, and unit test was added.
Closes #33762 from HyukjinKwon/cherry-pick-SPARK-36379.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../org/apache/spark/sql/catalyst/json/JacksonParser.scala | 9 ++++++---
.../spark/sql/execution/datasources/json/JsonSuite.scala | 14 ++++++++++++++
2 files changed, 20 insertions(+), 3 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index bbcff49..8f3f703 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -108,7 +108,7 @@ class JacksonParser(
// List([str_a_2,null], [null,str_b_3])
//
case START_ARRAY if allowArrayAsStructs =>
- val array = convertArray(parser, elementConverter)
+ val array = convertArray(parser, elementConverter, isRoot = true)
// Here, as we support reading top level JSON arrays and take every element
// in such an array as a row, this case is possible.
if (array.numElements() == 0) {
@@ -439,10 +439,13 @@ class JacksonParser(
*/
private def convertArray(
parser: JsonParser,
- fieldConverter: ValueConverter): ArrayData = {
+ fieldConverter: ValueConverter,
+ isRoot: Boolean = false): ArrayData = {
val values = ArrayBuffer.empty[Any]
while (nextUntil(parser, JsonToken.END_ARRAY)) {
- values += fieldConverter.apply(parser)
+ val v = fieldConverter.apply(parser)
+ if (isRoot && v == null) throw new RuntimeException("Root converter returned null")
+ values += v
}
new GenericArrayData(values.toArray)
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 76e05a2..c60cba2 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
@@ -2844,6 +2844,20 @@ abstract class JsonSuite
assert(readback.collect sameElements Array(Row(0), Row(1), Row(2)))
}
}
+
+ test("SPARK-36379: proceed parsing with root nulls in permissive mode") {
+ assert(intercept[SparkException] {
+ spark.read.option("mode", "failfast")
+ .schema("a string").json(Seq("""[{"a": "str"}, null]""").toDS).collect()
+ }.getMessage.contains("Malformed records are detected"))
+
+ // Permissive modes should proceed parsing malformed records (null).
+ // Here, since an array fails to parse in the middle, we will return one row.
+ checkAnswer(
+ spark.read.option("mode", "permissive")
+ .json(Seq("""[{"a": "str"}, null, {"a": "str"}]""").toDS),
+ Row(null) :: Nil)
+ }
}
class JsonV1Suite extends JsonSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org