diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/types/VariantVal.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/types/VariantVal.java index a441bab4ac41e..081784e8e83be 100644 --- a/common/unsafe/src/main/java/org/apache/spark/unsafe/types/VariantVal.java +++ b/common/unsafe/src/main/java/org/apache/spark/unsafe/types/VariantVal.java @@ -105,6 +105,12 @@ public String toJson(ZoneId zoneId) { return new Variant(value, metadata).toJson(zoneId); } + // Rejects a value nested more deeply than `maxNestingDepth` when that argument is positive. + // A non-positive value imposes no limit and preserves the previous behavior. + public String toJson(ZoneId zoneId, int maxNestingDepth) { + return new Variant(value, metadata).toJson(zoneId, maxNestingDepth); + } + /** * @return A human-readable representation of the Variant value. It is always a JSON string at * this moment. diff --git a/common/variant/src/main/java/org/apache/spark/types/variant/Variant.java b/common/variant/src/main/java/org/apache/spark/types/variant/Variant.java index cefb957f9b976..4c05cec7885ee 100644 --- a/common/variant/src/main/java/org/apache/spark/types/variant/Variant.java +++ b/common/variant/src/main/java/org/apache/spark/types/variant/Variant.java @@ -237,8 +237,15 @@ public Variant getElementAtIndex(int index) { // Stringify the variant in JSON format. // Throw `MALFORMED_VARIANT` if the variant is malformed. public String toJson(ZoneId zoneId) { + return toJson(zoneId, -1); + } + + // Stringify the variant in JSON format, rejecting a value nested more deeply than + // `maxNestingDepth` when that argument is positive. A non-positive value imposes no limit and + // preserves the previous behavior. + public String toJson(ZoneId zoneId, int maxNestingDepth) { StringBuilder sb = new StringBuilder(); - toJsonImpl(value, metadata, pos, sb, zoneId); + toJsonImpl(value, metadata, pos, sb, zoneId, 1, maxNestingDepth); return sb.toString(); } @@ -280,6 +287,15 @@ private static Instant microsToInstant(long timestamp) { } static void toJsonImpl(byte[] value, byte[] metadata, int pos, StringBuilder sb, ZoneId zoneId) { + toJsonImpl(value, metadata, pos, sb, zoneId, 1, -1); + } + + static void toJsonImpl(byte[] value, byte[] metadata, int pos, StringBuilder sb, ZoneId zoneId, + int depth, int maxNestingDepth) { + if (maxNestingDepth > 0 && depth > maxNestingDepth) { + throw new IllegalStateException( + "Variant value nesting depth exceeds the configured maximum of " + maxNestingDepth + "."); + } switch (VariantUtil.getType(value, pos)) { case OBJECT: handleObject(value, pos, (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> { @@ -291,7 +307,7 @@ static void toJsonImpl(byte[] value, byte[] metadata, int pos, StringBuilder sb, if (i != 0) sb.append(','); sb.append(escapeJson(getMetadataKey(metadata, id))); sb.append(':'); - toJsonImpl(value, metadata, elementPos, sb, zoneId); + toJsonImpl(value, metadata, elementPos, sb, zoneId, depth + 1, maxNestingDepth); } sb.append('}'); return null; @@ -304,7 +320,7 @@ static void toJsonImpl(byte[] value, byte[] metadata, int pos, StringBuilder sb, int offset = readUnsigned(value, offsetStart + offsetSize * i, offsetSize); int elementPos = dataStart + offset; if (i != 0) sb.append(','); - toJsonImpl(value, metadata, elementPos, sb, zoneId); + toJsonImpl(value, metadata, elementPos, sb, zoneId, depth + 1, maxNestingDepth); } sb.append(']'); return null; diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala index 5e5bb00df6c26..b5bfbe98b4e56 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala @@ -810,7 +810,8 @@ case class Cast( private lazy val castArgs = variant.VariantCastArgs( evalMode != EvalMode.TRY, timeZoneId, - zoneId) + zoneId, + SQLConf.get.getConf(SQLConf.VARIANT_MAX_NESTING_DEPTH)) def needsTimeZone: Boolean = Cast.needsTimeZone(child.dataType, dataType) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala index e77ad10ba4876..a1d8daed649ac 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala @@ -449,7 +449,8 @@ case class VariantGet( private lazy val castArgs = VariantCastArgs( failOnError, timeZoneId, - zoneId) + zoneId, + SQLConf.get.getConf(SQLConf.VARIANT_MAX_NESTING_DEPTH)) override def eval(input: InternalRow): Any = { val _ = parsedPath @@ -512,7 +513,8 @@ case class VariantGet( case class VariantCastArgs( failOnError: Boolean, zoneStr: Option[String], - zoneId: ZoneId) + zoneId: ZoneId, + maxNestingDepth: Int = -1) case object VariantGet { /** @@ -594,7 +596,8 @@ case object VariantGet { def cast(v: Variant, dataType: DataType, castArgs: VariantCastArgs): Any = { def invalidCast(): Any = { if (castArgs.failOnError) { - throw QueryExecutionErrors.invalidVariantCast(v.toJson(castArgs.zoneId), dataType) + throw QueryExecutionErrors.invalidVariantCast( + v.toJson(castArgs.zoneId, castArgs.maxNestingDepth), dataType) } else { null } @@ -622,7 +625,7 @@ case object VariantGet { val input = variantType match { case Type.OBJECT | Type.ARRAY => return if (dataType.isInstanceOf[StringType]) { - UTF8String.fromString(v.toJson(castArgs.zoneId)) + UTF8String.fromString(v.toJson(castArgs.zoneId, castArgs.maxNestingDepth)) } else { invalidCast() } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala index 7abf86850fc4d..c9999f4e5d416 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonGenerator.scala @@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.SpecializedGetters import org.apache.spark.sql.catalyst.util._ import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT import org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.VariantVal import org.apache.spark.util.ArrayImplicits._ @@ -111,6 +112,9 @@ class JacksonGenerator( case customPattern => TimeFormatter(customPattern, isParsing = false) } + // Read once at construction (per task) to avoid any per-row lookup overhead. + private val variantMaxNestingDepth = SQLConf.get.getConf(SQLConf.VARIANT_MAX_NESTING_DEPTH) + private def makeWriter(dataType: DataType): ValueWriter = dataType match { case NullType => (row: SpecializedGetters, ordinal: Int) => @@ -351,7 +355,7 @@ class JacksonGenerator( } def write(v: VariantVal): Unit = { - gen.writeRawValue(v.toJson(options.zoneId)) + gen.writeRawValue(v.toJson(options.zoneId, variantMaxNestingDepth)) } def writeLineEnding(): Unit = { diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 0b52c0e34d7a5..193f01282329d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -7074,6 +7074,18 @@ object SQLConf { .booleanConf .createWithDefault(true) + val VARIANT_MAX_NESTING_DEPTH = + buildConf("spark.sql.variant.maxNestingDepth") + .internal() + .doc("The maximum nesting depth allowed when converting a variant value to its JSON " + + "string form. When set to a positive value, converting a variant nested more deeply " + + "than this limit fails instead of recursing. A non-positive value (the default) " + + "imposes no limit and preserves the previous behavior.") + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .intConf + .createWithDefault(-1) + val PUSH_VARIANT_INTO_SCAN = buildConf("spark.sql.variant.pushVariantIntoScan") .internal() diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala index ed790e643adc0..895fe055bbfb8 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/variant/VariantExpressionSuite.scala @@ -2058,6 +2058,25 @@ class VariantExpressionSuite extends SparkFunSuite with ExpressionEvalHelper { name => Map("sizeLimit" -> "16.0 MiB", "functionName" -> s"`$name`")) } + test("Variant.toJson enforces the configured maximum nesting depth") { + // A value nested `depth` arrays deep: [[[ ... 1 ... ]]]. + val depth = 50 + val json = ("[" * depth) + "1" + ("]" * depth) + val variant = VariantBuilder.parseJson(json, false) + + // A non-positive limit imposes no bound and reproduces the previous behavior. + assert(variant.toJson(ZoneOffset.UTC) == json) + assert(variant.toJson(ZoneOffset.UTC, -1) == json) + // A limit at least as large as the actual nesting still renders the whole value. + assert(variant.toJson(ZoneOffset.UTC, depth * 2) == json) + + // A limit smaller than the actual nesting is rejected instead of recursing all the way down. + val e = intercept[IllegalStateException] { + variant.toJson(ZoneOffset.UTC, 10) + } + assert(e.getMessage.contains("nesting depth")) + } + test("variant_strip_nulls") { // Strip `input`, render the result back to JSON, and compare. `includeArrays` defaults to true. def check(input: String, expected: String, includeArrays: Boolean = true): Unit = { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala index 834b8a56b0380..aaa253d7fe0d3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala @@ -31,6 +31,7 @@ import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors import org.apache.spark.sql.execution.RowToColumnConverter import org.apache.spark.sql.execution.datasources.VariantMetadata import org.apache.spark.sql.execution.vectorized.WritableColumnVector +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.types.variant._ import org.apache.spark.types.variant.VariantUtil.Type @@ -211,7 +212,8 @@ class ParquetVariantReader( protected final def invalidCast(row: InternalRow, topLevelMetadata: Array[Byte]): Any = { if (castArgs.failOnError) { throw QueryExecutionErrors.invalidVariantCast( - rebuildVariant(row, topLevelMetadata).toJson(castArgs.zoneId), targetType) + rebuildVariant(row, topLevelMetadata).toJson( + castArgs.zoneId, castArgs.maxNestingDepth), targetType) } else { null } @@ -444,7 +446,8 @@ private[this] final class ScalarReader( override def readFromTyped(row: InternalRow, topLevelMetadata: Array[Byte]): Any = { if (castProject == null) { return if (targetType.isInstanceOf[StringType]) { - UTF8String.fromString(rebuildVariant(row, topLevelMetadata).toJson(castArgs.zoneId)) + UTF8String.fromString( + rebuildVariant(row, topLevelMetadata).toJson(castArgs.zoneId, castArgs.maxNestingDepth)) } else { invalidCast(row, topLevelMetadata) } @@ -769,7 +772,8 @@ case object SparkShreddingUtils { val reader = ParquetVariantReader(schema, f.dataType, VariantCastArgs( metadata.failOnError, Some(metadata.timeZoneId), - DateTimeUtils.getZoneId(metadata.timeZoneId)), + DateTimeUtils.getZoneId(metadata.timeZoneId), + SQLConf.get.getConf(SQLConf.VARIANT_MAX_NESTING_DEPTH)), isTopLevelUnshredded = schemaPath.isEmpty && inputSchema.isUnshredded) val castErrorOrdinal = companionIdxByDataName.getOrElse(f.name, -1) if (castErrorOrdinal >= 0) { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala index 30675e759428d..d1bc39ba1e20d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala @@ -523,4 +523,36 @@ class VariantEndToEndSuite extends SharedSparkSession { } } } + + test("SPARK-59325: spark.sql.variant.maxNestingDepth bounds rendering to JSON") { + def nestedObject(depth: Int): String = + if (depth == 0) "1" else s"""{"a":${nestedObject(depth - 1)}}""" + def nestedArray(depth: Int): String = + if (depth == 0) "1" else s"[${nestedArray(depth - 1)}]" + def causeMessages(t: Throwable): String = + if (t == null) "" else t.getMessage + " " + causeMessages(t.getCause) + + Seq(nestedObject(20), nestedArray(20)).foreach { json => + val df = Seq(json).toDF("v") + + // Default (-1) and a generous limit render fully (to_json and CAST AS STRING). + checkAnswer(df.select(to_json(parse_json(col("v")))), Seq(Row(json))) + checkAnswer(df.selectExpr("CAST(parse_json(v) AS STRING)"), Seq(Row(json))) + withSQLConf(SQLConf.VARIANT_MAX_NESTING_DEPTH.key -> "100") { + checkAnswer(df.select(to_json(parse_json(col("v")))), Seq(Row(json))) + checkAnswer(df.selectExpr("CAST(parse_json(v) AS STRING)"), Seq(Row(json))) + } + + // A limit below the nesting depth rejects on both render paths. + withSQLConf(SQLConf.VARIANT_MAX_NESTING_DEPTH.key -> "5") { + val renders = Seq( + () => df.select(to_json(parse_json(col("v")))).collect(), + () => df.selectExpr("CAST(parse_json(v) AS STRING)").collect()) + renders.foreach { render => + val e = intercept[Exception](render()) + assert(causeMessages(e).contains("nesting depth exceeds the configured maximum")) + } + } + } + } }