harshmotw-db commented on code in PR #57120:
URL: https://github.com/apache/spark/pull/57120#discussion_r3547153885
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala:
##########
@@ -1329,11 +1329,48 @@ class VariantExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
test("variant_insert") {
def checkInsert(input: String, path: String, value: Expression, expected:
String): Unit = {
- val expr = VariantInsert(
- Literal(parseJson(input)), Literal.create(path, StringType), value)
+ Seq(true, false).foreach { failOnError =>
+ val expr = VariantInsert(
+ Literal(parseJson(input)), Literal.create(path, StringType), value,
failOnError)
+ checkEvaluation(
+ ResolveTimeZone.resolveTimeZones(Cast(expr, StringType)),
+ expected)
+ }
+ }
+
+ // A catchable error: `variant_insert` throws, but `try_variant_insert`
returns NULL.
Review Comment:
They are "catchable" errors but not "recoverable" errors. I would prefer the
recoverable terminology.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionEvalUtils.scala:
##########
@@ -131,27 +132,34 @@ object VariantExpressionEvalUtils {
/**
* Insert `value` into `input` at `javaSegments`. `path` is the source
string used in error
* messages. The cast and insert share one try, so any size overflow maps to
`VARIANT_SIZE_LIMIT`
- * and a type mismatch maps to `VARIANT_PATH_TYPE_MISMATCH`.
+ * and a type mismatch maps to `VARIANT_PATH_TYPE_MISMATCH`. When
`failOnError` is false (the
+ * `try_variant_insert` mode), a duplicate key or path type mismatch returns
null instead of
+ * throwing; a size overflow (and a malformed path, rejected earlier during
parsing) is still
+ * raised.
*/
def insertAtPath(
input: VariantVal,
javaSegments: Array[VariantBuilder.PathSegment],
path: String,
value: Any,
valueDataType: DataType,
- functionName: String): VariantVal = {
+ functionName: String,
+ failOnError: Boolean): VariantVal = {
val v = new Variant(input.getValue, input.getMetadata)
try {
val valVal = castToVariant(value, valueDataType)
val valVariant = new Variant(valVal.getValue, valVal.getMetadata)
val out = VariantBuilder.insertAtPath(v, javaSegments, valVariant)
new VariantVal(out.getValue, out.getMetadata)
} catch {
+ case _: VariantPathTypeMismatchException if !failOnError => null
case e: VariantPathTypeMismatchException =>
throw QueryExecutionErrors.variantPathTypeMismatch(
path, renderVariantPath(javaSegments.take(e.depth)), functionName)
case _: VariantSizeLimitException =>
throw
QueryExecutionErrors.variantSizeLimitError(VariantUtil.SIZE_LIMIT, functionName)
+ case e: SparkRuntimeException if !failOnError && e.getCondition ==
"VARIANT_DUPLICATE_KEY" =>
+ null
Review Comment:
I would put all the non-throwing cases first. Just by reading this, we don't
have evidence that there is no overlap between the throwing/non-throwing cases
##########
sql/core/src/test/scala/org/apache/spark/sql/VariantSuite.scala:
##########
@@ -384,6 +384,61 @@ class VariantSuite extends SharedSparkSession with
ExpressionEvalHelper {
}
}
+ test("try_variant_insert with literal arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b',
2))"),
+ rows("""{"a":1,"b":2}"""))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.a',
2))"),
+ rows(null))
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'),
'$.a.b', 2))"),
+ rows(null))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$[0]',
2))"),
+ rows(null))
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 5}'),
'$.a[0]', 2))"),
+ rows(null))
+
+ checkAnswer(
+ sql("SELECT to_json(try_variant_insert(parse_json('{\"a\": 1}'), '$.b',
NULL))"),
+ rows(null))
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT try_variant_insert(parse_json('{}'), '$', 1)").collect()
+ },
+ condition = "INVALID_VARIANT_PATH",
+ parameters = Map("path" -> "$", "functionName" ->
toSQLId("try_variant_insert")))
+ }
+
+ test("try_variant_insert with dynamic arguments") {
+ def rows(results: Any*): Seq[Row] = results.map(Row(_))
+ Seq("CODEGEN_ONLY", "NO_CODEGEN").foreach { codegenMode =>
+ withSQLConf(SQLConf.CODEGEN_FACTORY_MODE.key -> codegenMode) {
+ val df = Seq(
+ ("""{"a": 1}""", "$.b"),
+ ("""{"a": 1}""", "$.a"),
+ ("""{"a": 1}""", "$.a.b"),
+ (null, "$.b")
+ ).toDF("json", "path")
+ val v = parse_json(col("json"))
+ val out = df.select(to_json(try_variant_insert(v, col("path"),
lit(2))).alias("r"))
+ checkAnswer(out, rows("""{"a":1,"b":2}""", null, null, null))
Review Comment:
Do you have a planned expression that also upserts data into variants?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
