You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "chenhao-db (via GitHub)" <gi...@apache.org> on 2026/06/14 18:57:56 UTC
[PR] [SPARK-57445] Fix PushVariantIntoScan exception semantics. [spark]
chenhao-db opened a new pull request, #56505:
URL: https://github.com/apache/spark/pull/56505
### What changes were proposed in this pull request?
Today, when `PushVariantIntoScan` rewrites a strict variant cast/`variant_get` into a typed scan field, the cast is evaluated eagerly inside the scan. An `INVALID_VARIANT_CAST` from any row aborts the query, even when the user expression that requested the cast (e.g., a predicate that prunes the bad row) would never actually consume it.
This PR adds an opt-in (`spark.sql.variant.pushVariantIntoScan.deferCastError`, default off) that defers the cast error to the row's consumer. The mechanism:
- **Wrapper schema** — For each pushed strict-cast field `<n>`, add a new field with a special metadata entry `castErrorFor: <n>` to the variant struct schema. This field name will be use for paring the target field and its cast-error companion.
- **Reader** — `SparkShreddingUtils.assembleVariantStruct` catches `INVALID_VARIANT_CAST`, writes the offending value into `cast_error`, and leaves `field_value` null on failure (and the reverse on success).
- **Consumer** — New Catalyst expression `UnwrapVariantCastError(cast_error, field_value)` is equivalent to `if(cast_error IS NOT NULL, raise_error('INVALID_VARIANT_CAST', ...), field_value)` but kept as a single named expression so downstream operators (physical sacn) can easily recognize it)
### Why are the changes needed?
To ensure that user doesn't get surprising result when `PushVariantIntoScan` is enabled.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Co-authored with Claude Opus 4.8.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "chenhao-db (via GitHub)" <gi...@apache.org>.
chenhao-db commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3431310671
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala:
##########
@@ -836,8 +865,26 @@ case object SparkShreddingUtils {
val resultRow = new GenericInternalRow(numFields)
var fieldIdx = 0
while (fieldIdx < numFields) {
- resultRow.update(fieldIdx, extractField(inputRow, topLevelMetadata, schema,
- fields(fieldIdx).path, fields(fieldIdx).reader))
+ val field = fields(fieldIdx)
+ if (field.isCastError) {
+ // Filled by the paired data field on failure; left null otherwise.
+ } else if (field.castErrorOrdinal >= 0) {
+ try {
+ val value = extractField(inputRow, topLevelMetadata, schema, field.path, field.reader)
+ resultRow.update(fieldIdx, value)
+ } catch {
+ case e: SparkRuntimeException if e.getCondition == "INVALID_VARIANT_CAST" =>
+ // Recover the offending value from the error's `value` message parameter so the
Review Comment:
The observation is correct, but it may not worth the complexity to retain the data type in error message (need to add more information to the companion field). The whole type is more informational in some sense.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #56505:
URL: https://github.com/apache/spark/pull/56505#issuecomment-4737995685
Hi, @chenhao-db and @cloud-fan .
This broke `branch-4.2`.
- https://github.com/apache/spark/actions/runs/27725205574/job/82020200235
```
[error] /home/runner/work/spark/spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:451:19: not found: value VariantMetadata
[error] VariantMetadata.castErrorCompanionMetadata(ordinal.toString),
[error] ^
[error] /home/runner/work/spark/spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:39:38: Unused import
[error] Applicable -Wconf / @nowarn filters for this fatal warning: msg=<part of the message>, cat=unused-imports, site=org.apache.spark.sql.execution.datasources.v2
[error] import org.apache.spark.sql.internal.SQLConf
[error]
```
I also verified locally.
```
$ build/sbt "sql/compile"
...
[error] /Users/dongjoon/APACHE/spark-merge/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:451:19: not found: value VariantMetadata
[error] VariantMetadata.castErrorCompanionMetadata(ordinal.toString),
[error] ^
[error] /Users/dongjoon/APACHE/spark-merge/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:39:38: Unused import
[error] Applicable -Wconf / @nowarn filters for this fatal warning: msg=<part of the message>, cat=unused-imports, site=org.apache.spark.sql.execution.datasources.v2
[error] import org.apache.spark.sql.internal.SQLConf
[error] ^
[warn] 5 warnings found
[error] two errors found
[error] (sql / Compile / compileIncremental) Compilation failed
[error] Total time: 14 s, completed Jun 17, 2026, 9:22:03 PM
```
Let me revert this from `branch-4.2` in oder to recover `branch-4.2` and unblock next RC.
Please make an official backporting PR for `branch-4.2`.
cc @huaxingao
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "chenhao-db (via GitHub)" <gi...@apache.org>.
chenhao-db commented on PR #56505:
URL: https://github.com/apache/spark/pull/56505#issuecomment-4710051819
@cloud-fan could you take a look? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #56505:
URL: https://github.com/apache/spark/pull/56505#issuecomment-4737999267
This is reverted from `branch-4.2` via the following.
- https://github.com/apache/spark/commit/16086bc2782488980629f797a343a0fd34e67c6f
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "shrirangmhalgi (via GitHub)" <gi...@apache.org>.
shrirangmhalgi commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3425894251
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -427,6 +427,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
// If no variant columns remain after collection, return original plan
if (variants.mapping.forall(_._2.isEmpty)) return originalPlan
+ // Cast-error deferral needs reader support for synthetic companion fields. If a scan only
+ // supports plain variant extraction pushdown, leave the plan untouched while deferral is on.
+ if (variants.deferCastErrorEnabled && !builder.supportsDeferCastError()) return originalPlan
Review Comment:
The early return here disables ALL variant pushdown when the scan doesn't support deferral - including non-strict accesses (`try_variant_get`, plain variant reads) that don't need companion support. Could we limit this to plans that actually contain strict-cast extractions? Non-strict pushdowns are independently safe.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #56505: [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics.
URL: https://github.com/apache/spark/pull/56505
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3433025994
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6329,6 +6329,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR =
+ buildConf("spark.sql.variant.pushVariantIntoScan.deferCastError")
+ .internal()
+ .doc("When true, strict variant casts that get pushed into the scan are wrapped with a " +
+ "per-row cast-error companion column (nullable string) so that the cast error is only " +
+ "raised when the row is consumed by the user expression. Without this flag, the cast is " +
+ "always evaluated and any failure raises immediately, even when the surrounding " +
+ "expression would not have consumed the failing row.")
+ .version("4.3.0")
Review Comment:
BTW, this PR code itself seems to have a wrong version, @cloud-fan .
I'd like to recommend to fix `master` and `branch-4.x` first before making a backport to `branch-4.2`, @chenhao-db .
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "chenhao-db (via GitHub)" <gi...@apache.org>.
chenhao-db commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3431287091
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -427,6 +427,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
// If no variant columns remain after collection, return original plan
if (variants.mapping.forall(_._2.isEmpty)) return originalPlan
+ // Cast-error deferral needs reader support for synthetic companion fields. If a scan only
+ // supports plain variant extraction pushdown, leave the plan untouched while deferral is on.
+ if (variants.deferCastErrorEnabled && !builder.supportsDeferCastError()) return originalPlan
Review Comment:
Done. Now I check `builder.supportsDeferCastError()` after we know there is at least a companion field.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3433061382
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6329,6 +6329,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR =
+ buildConf("spark.sql.variant.pushVariantIntoScan.deferCastError")
+ .internal()
+ .doc("When true, strict variant casts that get pushed into the scan are wrapped with a " +
+ "per-row cast-error companion column (nullable string) so that the cast error is only " +
+ "raised when the row is consumed by the user expression. Without this flag, the cast is " +
+ "always evaluated and any failure raises immediately, even when the surrounding " +
+ "expression would not have consumed the failing row.")
+ .version("4.3.0")
Review Comment:
Got it. Thank you for confirming that!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3430465052
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/SparkShreddingUtils.scala:
##########
@@ -836,8 +865,26 @@ case object SparkShreddingUtils {
val resultRow = new GenericInternalRow(numFields)
var fieldIdx = 0
while (fieldIdx < numFields) {
- resultRow.update(fieldIdx, extractField(inputRow, topLevelMetadata, schema,
- fields(fieldIdx).path, fields(fieldIdx).reader))
+ val field = fields(fieldIdx)
+ if (field.isCastError) {
+ // Filled by the paired data field on failure; left null otherwise.
+ } else if (field.castErrorOrdinal >= 0) {
+ try {
+ val value = extractField(inputRow, topLevelMetadata, schema, field.path, field.reader)
+ resultRow.update(fieldIdx, value)
+ } catch {
+ case e: SparkRuntimeException if e.getCondition == "INVALID_VARIANT_CAST" =>
+ // Recover the offending value from the error's `value` message parameter so the
Review Comment:
The recovered `value` matches the eager path, but the re-raised `dataType` does not. For a composite target (`array`/`struct`/`map`) where an *inner* element fails, the eager cast raises `INVALID_VARIANT_CAST` with the **element** type (e.g. `INT`), since `VariantGet.cast` recurses with the element type. The deferred path stores only the `value` param here, and `UnwrapVariantCastError` re-raises with `dataType = value.dataType` — the **field's full target type** (e.g. `ARRAY<INT>`). So the condition and offending value match, but the displayed type differs for composite targets (top-level scalar casts are unaffected). The new tests assert only the condition via `hasCastCondition`, so this isn't covered. Since the PR targets exception-semantics fidelity ("surface the same value that an eager raise would have"), is exact message parity with the eager path intended?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala:
##########
@@ -49,13 +73,235 @@ trait PushVariantIntoScanSuiteBase extends SharedSparkSession {
}
}
+ test(s"Strict cast wraps with cast-error-deferred error") {
+ withTable("T") {
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ sql("create table T (v variant) using parquet")
+ sql("select cast(v as int) as a, try_variant_get(v, '$.b', 'string') as b from T")
+ .queryExecution.optimizedPlan match {
+ case Project(projectList, l: LogicalRelation) =>
+ val output = l.output
+ val v = output(0)
+ // Strict cast should be wrapped with `UnwrapVariantCastError` over the sibling
+ // companion field whose `castErrorFor` metadata names the data field.
+ projectList(0) match {
+ case Alias(UnwrapVariantCastError(
+ GetStructField(_, errOrd, _), GetStructField(_, 0, _)), "a") =>
+ assert(errOrd == 2, s"Expected companion ordinal 2, got $errOrd")
+ case other => fail(s"Unexpected projection 0: $other")
+ }
+ // try_variant_get is non-strict and should NOT be wrapped.
+ projectList(1) match {
+ case Alias(GetStructField(_, 1, _), "b") =>
+ case other => fail(s"Unexpected projection 1: $other")
+ }
+ val expected = StructType(Array(
+ field(0, IntegerType, "$", failOnError = true),
+ field(1, StringType, "$.b", failOnError = false),
+ StructField("2", StringType,
+ metadata = VariantMetadata.castErrorCompanionMetadata("0"))
+ ))
+ assert(v.dataType == expected, s"Got ${v.dataType}")
+ case other => fail(s"Unexpected plan: $other")
+ }
+ }
+ }
+ }
+
+ test(s"Cast-error companion is skipped for full-variant access") {
+ withTable("T") {
+ withSQLConf(
+ SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ sql("create table T (v variant) using parquet")
+ // Selecting `v` alone produces only the full-variant request. cast-to-variant never
+ // fails, so no cast-error companion should be emitted.
+ sql("select v from T").queryExecution.optimizedPlan match {
+ case Project(_, l: LogicalRelation) =>
+ val v = l.output(0)
+ val expected = StructType(Array(
+ field(0, VariantType, "$", timeZone = "UTC")
+ ))
+ assert(v.dataType == expected, s"Got ${v.dataType}")
+ case other => fail(s"Unexpected plan: $other")
+ }
+ }
+ }
+ }
+
+ test(s"Reader defers strict-cast errors when cast-error companion is present") {
+ // Row 0: number 1 (LONG in variant) -> cast(v as int) succeeds.
+ // Row 1: string -> cast(v as int) would raise INVALID_VARIANT_CAST. With the deferral, the
+ // surrounding `if(schema_of_variant(v) = 'BIGINT', cast(v as int), null)`
+ // short-circuits to null before the error is observed.
+ withVariantParquetData("v variant",
+ "(parse_json('1'))",
+ "(parse_json('\"hello\"'))") {
+ val query =
+ "select if(schema_of_variant(v) = 'BIGINT', cast(v as int), null) as a from T"
+
+ // Without the deferral, the strict cast pushed into the scan raises at the failing row
+ // even though the `if` would have filtered it out.
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "false") {
+ val ex = intercept[Exception](sql(query).collect())
+ // The cast failure may surface directly or be wrapped in a task failure.
+ def hasCastCondition(t: Throwable): Boolean = t match {
+ case null => false
+ case s: org.apache.spark.SparkThrowable if s.getCondition == "INVALID_VARIANT_CAST" =>
+ true
+ case _ => hasCastCondition(t.getCause)
+ }
+ assert(hasCastCondition(ex), s"Expected INVALID_VARIANT_CAST, got $ex")
+ }
+
+ // With the deferral, the strict cast emits a cast-error companion and the `if`
+ // short-circuits before the failing row is consumed.
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ val rows = sql(query).collect()
+ val values = rows.map(r => if (r.isNullAt(0)) null else r.getInt(0).asInstanceOf[Any])
+ .toSet
+ assert(values == Set(1, null), s"Got ${values.mkString(",")}")
+ }
+ }
+ }
+
+ test(s"Reader defers strict-cast errors for struct target") {
+ // Row 0: object with int field -> cast(v as struct<x int>) succeeds.
+ // Row 1: scalar -> cast(v as struct<x int>) would raise (wrong kind).
+ withVariantParquetData("v variant",
+ "(parse_json('{\"x\": 1}'))",
+ "(parse_json('\"hello\"'))") {
+ val query =
+ "select if(schema_of_variant(v) like 'OBJECT<%>', cast(v as struct<x: int>), null) as a " +
+ "from T"
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ val rows = sql(query).collect()
+ val xs = rows.map { r =>
+ if (r.isNullAt(0)) null else r.getStruct(0).getInt(0).asInstanceOf[Any]
+ }.toSet
+ assert(xs == Set(1, null), s"Got ${xs.mkString(",")}")
+ }
+ }
+ }
+
+ test(s"Reader defers strict-cast errors for array target") {
+ // Row 0: array of ints -> cast(v as array<int>) succeeds.
+ // Row 1: scalar -> cast(v as array<int>) wrong-kind failure.
+ withVariantParquetData("v variant",
+ "(parse_json('[1, 2, 3]'))",
+ "(parse_json('\"hello\"'))") {
+ val query =
+ "select if(schema_of_variant(v) like 'ARRAY<%>', cast(v as array<int>), null) as a " +
+ "from T"
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ val rows = sql(query).collect()
+ val arrs = rows.map { r =>
+ if (r.isNullAt(0)) null else r.getList[Int](0).toArray.toSeq
+ }.toSet
+ assert(arrs == Set(Seq(1, 2, 3), null), s"Got ${arrs.mkString(",")}")
+ }
+ }
+ }
+
+ test(s"Reader surfaces deferred error for array target with inner-element failure") {
+ // Row 0: heterogeneous array; cast(v as array<int>) fails on the inner string element.
+ // With deferred errors enabled, the failure must surface when the row is consumed by the
+ // outer expression -- i.e., the element-level companion buffer was correctly aggregated to
+ // the outer row.
+ withVariantParquetData("v variant",
+ "(parse_json('[1, \"abc\"]'))") {
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ val ex = intercept[Exception](sql("select cast(v as array<int>) from T").collect())
+ def hasCastCondition(t: Throwable): Boolean = t match {
+ case null => false
+ case s: org.apache.spark.SparkThrowable if s.getCondition == "INVALID_VARIANT_CAST" =>
+ true
+ case _ => hasCastCondition(t.getCause)
+ }
+ assert(hasCastCondition(ex), s"Expected INVALID_VARIANT_CAST, got $ex")
+ }
+ }
+ }
+
+ test(s"Reader surfaces deferred error for struct target with field cast failure") {
+ // Force the writer to shred `x` as int. The inner string `"abc"` lands in the unshredded
+ // `value` part, and `cast(v as struct<x: int>)` reads the int via the shredded path, which
+ // exercises `SparkShreddingUtils.getFieldsToExtract` / `assembleVariantStruct` with the new
+ // companion-field pairing.
+ withSQLConf(
+ SQLConf.VARIANT_WRITE_SHREDDING_ENABLED.key -> "true",
+ SQLConf.VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST.key -> "x int") {
+ withVariantParquetData("v variant",
+ "(parse_json('{\"x\": \"abc\"}'))") {
+ withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key -> "true") {
+ val ex =
+ intercept[Exception](sql("select cast(v as struct<x: int>) from T").collect())
+ def hasCastCondition(t: Throwable): Boolean = t match {
+ case null => false
+ case s: org.apache.spark.SparkThrowable
+ if s.getCondition == "INVALID_VARIANT_CAST" => true
+ case _ => hasCastCondition(t.getCause)
+ }
+ assert(hasCastCondition(ex), s"Expected INVALID_VARIANT_CAST, got $ex")
+ }
+ }
+ }
+ }
+
+ // Returns true iff `t` or any of its causes is an INVALID_VARIANT_CAST error.
+ private def hasCastCondition(t: Throwable): Boolean = t match {
Review Comment:
`hasCastCondition` is also defined inline in three earlier tests (the `if(...)` scalar, array, and struct cases). Could those reuse this private helper to drop the duplicated copies?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3433057935
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6329,6 +6329,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR =
+ buildConf("spark.sql.variant.pushVariantIntoScan.deferCastError")
+ .internal()
+ .doc("When true, strict variant casts that get pushed into the scan are wrapped with a " +
+ "per-row cast-error companion column (nullable string) so that the cast error is only " +
+ "raised when the row is consumed by the user expression. Without this flag, the cast is " +
+ "always evaluated and any failure raises immediately, even when the surrounding " +
+ "expression would not have consumed the failing row.")
+ .version("4.3.0")
Review Comment:
let's leave the fix to 4.3 then, let me update the JIRA fixed version
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56505:
URL: https://github.com/apache/spark/pull/56505#discussion_r3433059697
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6329,6 +6329,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR =
+ buildConf("spark.sql.variant.pushVariantIntoScan.deferCastError")
+ .internal()
+ .doc("When true, strict variant casts that get pushed into the scan are wrapped with a " +
+ "per-row cast-error companion column (nullable string) so that the cast error is only " +
+ "raised when the row is consumed by the user expression. Without this flag, the cast is " +
+ "always evaluated and any failure raises immediately, even when the surrounding " +
+ "expression would not have consumed the failing row.")
+ .version("4.3.0")
Review Comment:
oh you already did so, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics. [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #56505:
URL: https://github.com/apache/spark/pull/56505#issuecomment-4736251450
thanks, merging to master/4.x/4.2 (bug fix)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org