You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/12/16 01:22:26 UTC
(spark) branch master updated: [SPARK-49954][SQL][FOLLOWUP] Move states to SchemaOfJsonEvaluator
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 769e569b7916 [SPARK-49954][SQL][FOLLOWUP] Move states to SchemaOfJsonEvaluator
769e569b7916 is described below
commit 769e569b7916467a9310a784048782c9df13f336
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Mon Dec 16 09:22:09 2024 +0800
[SPARK-49954][SQL][FOLLOWUP] Move states to SchemaOfJsonEvaluator
### What changes were proposed in this pull request?
This is a followup of https://github.com/apache/spark/pull/48473.
We should follow the same approach of `schema_of_csv` and create `SchemaOfJsonEvaluator` to keep the states and the `schema_of_json` function can leverage it using `Invoke`.
### Why are the changes needed?
code cleanup
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
existing tests
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #49184 from cloud-fan/json.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../expressions/json/JsonExpressionEvalUtils.scala | 62 +++++++++++-----------
.../sql/catalyst/expressions/jsonExpressions.scala | 29 +++-------
.../function_schema_of_json.explain | 2 +-
.../function_schema_of_json_with_options.explain | 2 +-
4 files changed, 42 insertions(+), 53 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
index 7a399aba4382..cc2bc19c4f77 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
@@ -18,8 +18,6 @@ package org.apache.spark.sql.catalyst.expressions.json
import java.io.CharArrayWriter
-import com.fasterxml.jackson.core.JsonFactory
-
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.ExprUtils
import org.apache.spark.sql.catalyst.expressions.variant.VariantExpressionEvalUtils
@@ -31,34 +29,6 @@ import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, St
import org.apache.spark.unsafe.types.{UTF8String, VariantVal}
import org.apache.spark.util.Utils
-object JsonExpressionEvalUtils {
-
- def schemaOfJson(
- jsonFactory: JsonFactory,
- jsonOptions: JSONOptions,
- jsonInferSchema: JsonInferSchema,
- json: UTF8String): UTF8String = {
- val dt = Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, json)) { parser =>
- parser.nextToken()
- // To match with schema inference from JSON datasource.
- jsonInferSchema.inferField(parser) match {
- case st: StructType =>
- jsonInferSchema.canonicalizeType(st, jsonOptions).getOrElse(StructType(Nil))
- case at: ArrayType if at.elementType.isInstanceOf[StructType] =>
- jsonInferSchema
- .canonicalizeType(at.elementType, jsonOptions)
- .map(ArrayType(_, containsNull = at.containsNull))
- .getOrElse(ArrayType(StructType(Nil), containsNull = at.containsNull))
- case other: DataType =>
- jsonInferSchema.canonicalizeType(other, jsonOptions).getOrElse(
- SQLConf.get.defaultStringType)
- }
- }
-
- UTF8String.fromString(dt.sql)
- }
-}
-
case class JsonToStructsEvaluator(
options: Map[String, String],
nullableSchema: DataType,
@@ -160,3 +130,35 @@ case class StructsToJsonEvaluator(
converter(value)
}
}
+
+case class SchemaOfJsonEvaluator(options: Map[String, String]) {
+ @transient
+ private lazy val jsonOptions = new JSONOptions(options, "UTC")
+
+ @transient
+ private lazy val jsonFactory = jsonOptions.buildJsonFactory()
+
+ @transient
+ private lazy val jsonInferSchema = new JsonInferSchema(jsonOptions)
+
+ final def evaluate(json: UTF8String): Any = {
+ val dt = Utils.tryWithResource(CreateJacksonParser.utf8String(jsonFactory, json)) { parser =>
+ parser.nextToken()
+ // To match with schema inference from JSON datasource.
+ jsonInferSchema.inferField(parser) match {
+ case st: StructType =>
+ jsonInferSchema.canonicalizeType(st, jsonOptions).getOrElse(StructType(Nil))
+ case at: ArrayType if at.elementType.isInstanceOf[StructType] =>
+ jsonInferSchema
+ .canonicalizeType(at.elementType, jsonOptions)
+ .map(ArrayType(_, containsNull = at.containsNull))
+ .getOrElse(ArrayType(StructType(Nil), containsNull = at.containsNull))
+ case other: DataType =>
+ jsonInferSchema.canonicalizeType(other, jsonOptions).getOrElse(
+ SQLConf.get.defaultStringType)
+ }
+ }
+
+ UTF8String.fromString(dt.sql)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index affc8261dc88..bb8cbdd3e6f1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, CodegenFallback, ExprCode}
import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
-import org.apache.spark.sql.catalyst.expressions.json.{JsonExpressionEvalUtils, JsonExpressionUtils, JsonToStructsEvaluator, StructsToJsonEvaluator}
+import org.apache.spark.sql.catalyst.expressions.json.{JsonExpressionUtils, JsonToStructsEvaluator, SchemaOfJsonEvaluator, StructsToJsonEvaluator}
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
import org.apache.spark.sql.catalyst.json._
import org.apache.spark.sql.catalyst.trees.TreePattern.{JSON_TO_STRUCT, RUNTIME_REPLACEABLE, TreePattern}
@@ -849,15 +849,6 @@ case class SchemaOfJson(
override def nullable: Boolean = false
- @transient
- private lazy val jsonOptions = new JSONOptions(options, "UTC")
-
- @transient
- private lazy val jsonFactory = jsonOptions.buildJsonFactory()
-
- @transient
- private lazy val jsonInferSchema = new JsonInferSchema(jsonOptions)
-
@transient
private lazy val json = child.eval().asInstanceOf[UTF8String]
@@ -878,19 +869,15 @@ case class SchemaOfJson(
}
}
- @transient private lazy val jsonFactoryObjectType = ObjectType(classOf[JsonFactory])
- @transient private lazy val jsonOptionsObjectType = ObjectType(classOf[JSONOptions])
- @transient private lazy val jsonInferSchemaObjectType = ObjectType(classOf[JsonInferSchema])
+ @transient
+ private lazy val evaluator: SchemaOfJsonEvaluator = SchemaOfJsonEvaluator(options)
- override def replacement: Expression = StaticInvoke(
- JsonExpressionEvalUtils.getClass,
+ override def replacement: Expression = Invoke(
+ Literal.create(evaluator, ObjectType(classOf[SchemaOfJsonEvaluator])),
+ "evaluate",
dataType,
- "schemaOfJson",
- Seq(Literal(jsonFactory, jsonFactoryObjectType),
- Literal(jsonOptions, jsonOptionsObjectType),
- Literal(jsonInferSchema, jsonInferSchemaObjectType),
- child),
- Seq(jsonFactoryObjectType, jsonOptionsObjectType, jsonInferSchemaObjectType, child.dataType),
+ Seq(child),
+ Seq(child.dataType),
returnNullable = false)
override def prettyName: String = "schema_of_json"
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json.explain
index b400aeeca5af..d75545d8766d 100644
--- a/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json.explain
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json.explain
@@ -1,2 +1,2 @@
-Project [static_invoke(JsonExpressionEvalUtils.schemaOfJson(com.fasterxml.jackson.core.JsonFactory, org.apache.spark.sql.catalyst.json.JSONOptions, org.apache.spark.sql.catalyst.json.JsonInferSchema, [{"col":01}])) AS schema_of_json([{"col":01}])#0]
+Project [invoke(SchemaOfJsonEvaluator(Map()).evaluate([{"col":01}])) AS schema_of_json([{"col":01}])#0]
+- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json_with_options.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json_with_options.explain
index b400aeeca5af..37321af1deed 100644
--- a/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json_with_options.explain
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_schema_of_json_with_options.explain
@@ -1,2 +1,2 @@
-Project [static_invoke(JsonExpressionEvalUtils.schemaOfJson(com.fasterxml.jackson.core.JsonFactory, org.apache.spark.sql.catalyst.json.JSONOptions, org.apache.spark.sql.catalyst.json.JsonInferSchema, [{"col":01}])) AS schema_of_json([{"col":01}])#0]
+Project [invoke(SchemaOfJsonEvaluator(Map(allowNumericLeadingZeros -> true)).evaluate([{"col":01}])) AS schema_of_json([{"col":01}])#0]
+- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org