This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 16086bc27824 Revert "[SPARK-57445][SQL] Fix PushVariantIntoScan
exception semantics."
16086bc27824 is described below
commit 16086bc2782488980629f797a343a0fd34e67c6f
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Jun 17 21:22:19 2026 -0700
Revert "[SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics."
This reverts commit 5e9c4b8a310b8324aaa9425fdec7aba33e2a2488.
---
.../read/SupportsPushDownVariantExtractions.java | 25 ---
.../expressions/variant/variantExpressions.scala | 60 ------
.../org/apache/spark/sql/internal/SQLConf.scala | 13 --
.../datasources/PushVariantIntoScan.scala | 129 ++----------
.../datasources/parquet/SparkShreddingUtils.scala | 123 ++++-------
.../datasources/v2/V2ScanRelationPushDown.scala | 31 +--
.../datasources/v2/parquet/ParquetScan.scala | 5 +-
.../v2/parquet/ParquetScanBuilder.scala | 2 -
.../datasources/PushVariantIntoScanSuite.scala | 230 ---------------------
9 files changed, 62 insertions(+), 556 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownVariantExtractions.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownVariantExtractions.java
index 2eb0038e6e92..750e0479e542 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownVariantExtractions.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsPushDownVariantExtractions.java
@@ -38,31 +38,6 @@ import org.apache.spark.annotation.Experimental;
@Experimental
public interface SupportsPushDownVariantExtractions extends ScanBuilder {
- /**
- * Returns whether this scan supports deferring strict variant cast errors.
- * <p>
- * When this returns false, Spark will not push down variant extractions if
cast-error deferral
- * is enabled.
- * <p>
- * Returning true opts the scan into receiving synthetic cast-error
companion extractions.
- * Companion extractions are marked by a {@code castErrorFor} metadata key.
Within each
- * {@link VariantExtraction#columnName()} group, the scan output field for
the i-th pushed
- * extraction MUST be named {@code Integer.toString(i)}. A companion
extraction's
- * {@code castErrorFor} value names its paired data field in that same
output struct.
- * <p>
- * Implementations may still reject individual extractions via
- * {@link #pushVariantExtractions(VariantExtraction[])}. However, for any
data extraction that has
- * a cast-error companion, accepting the data extraction requires accepting
its companion
- * extraction as well. Accepting only one side of the pair is invalid
because Spark rewrites the
- * consumed expression as a combined value/companion access.
- * <p>
- * A scan that supports this must preserve the companion metadata and
populate the companion
- * field with the offending value when the paired strict cast fails, or null
otherwise.
- *
- * @return true if this scan supports deferring strict variant cast errors
- */
- default boolean supportsDeferCastError() { return false; }
-
/**
* Pushes down variant field extractions to the data source.
* <p>
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 88ab4676c36d..5d78f11bf86f 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
@@ -985,63 +985,3 @@ case class IsValidVariant(child: Expression) extends
UnaryExpression
override protected def withNewChildInternal(newChild: Expression):
IsValidVariant =
copy(child = newChild)
}
-
-/**
- * Internal expression. It surfaces a deferred cast error produced by
`PushVariantIntoScan` for a
- * strict variant cast. Semantically equivalent to
- *
- * if(castError IS NOT NULL, raise_error('INVALID_VARIANT_CAST', ...), value)
- *
- * but kept as a single named expression so downstream consumers can easily
recognize it.
- */
-case class UnwrapVariantCastError(castError: Expression, value: Expression)
- extends BinaryExpression
- with ExpectsInputTypes
- with QueryErrorsBase {
- override def left: Expression = castError
- override def right: Expression = value
-
- override def inputTypes: Seq[AbstractDataType] = Seq(StringType, AnyDataType)
-
- override def dataType: DataType = value.dataType
-
- override def nullable: Boolean = true
-
- override def eval(input: InternalRow): Any = {
- val err = castError.eval(input)
- if (err != null) {
- throw
QueryExecutionErrors.invalidVariantCast(err.asInstanceOf[UTF8String].toString,
dataType)
- }
- value.eval(input)
- }
-
- override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = {
- val dataTypeRef = ctx.addReferenceObj("dataType", dataType,
classOf[DataType].getName)
- val cls = UnwrapVariantCastError.getClass.getName.stripSuffix("$")
- val errEval = castError.genCode(ctx)
- val valEval = value.genCode(ctx)
- val javaType = CodeGenerator.javaType(dataType)
- val code = code"""
- ${errEval.code}
- if (!${errEval.isNull}) {
- $cls.throwInvalidVariantCast(${errEval.value}, $dataTypeRef);
- }
- ${valEval.code}
- boolean ${ev.isNull} = ${valEval.isNull};
- $javaType ${ev.value} = ${valEval.value};
- """
- ev.copy(code = code)
- }
-
- override protected def withNewChildrenInternal(
- newLeft: Expression, newRight: Expression): UnwrapVariantCastError =
- copy(castError = newLeft, value = newRight)
-}
-
-object UnwrapVariantCastError {
- // Indirection so codegen can throw via a method call; a literal `throw` of
a `Throwable`-typed
- // expression trips Java's checked-exception check.
- def throwInvalidVariantCast(error: UTF8String, dataType: DataType): Unit = {
- throw QueryExecutionErrors.invalidVariantCast(error.toString, dataType)
- }
-}
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 0a208afc5023..b69e0428630c 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
@@ -6202,19 +6202,6 @@ 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")
- .withBindingPolicy(ConfigBindingPolicy.SESSION)
- .booleanConf
- .createWithDefault(false)
-
val VARIANT_WRITE_SHREDDING_ENABLED =
buildConf("spark.sql.variant.writeShredding.enabled")
.internal()
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala
index 69aa576f0c94..b0b20d08dccb 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScan.scala
@@ -41,26 +41,18 @@ case class VariantMetadata(
// `[*]` is not supported.
path: String,
failOnError: Boolean,
- timeZoneId: String,
- // When set, this struct field is a synthetic cast-error companion paired
with the data field
- // of the given NAME in the same variant struct. The companion is
populated by the reader
- // with the offending value when the paired data field's strict cast raises
- // INVALID_VARIANT_CAST. We pair by NAME (not struct ordinal) because
later pruning or
- // reordering of struct fields preserves names but may shift positions.
- castErrorFor: Option[String] = None) {
+ timeZoneId: String) {
// Produce a metadata contain one key-value pair. The key is the special
`METADATA_KEY`.
- // The value contains key-value pairs for `path`, `failOnError`,
`timeZoneId`, and -- for
- // companion fields only -- `castErrorFor`.
- def toMetadata: Metadata = {
- val inner = new MetadataBuilder()
- .putString(VariantMetadata.PATH_KEY, path)
- .putBoolean(VariantMetadata.FAIL_ON_ERROR_KEY, failOnError)
- .putString(VariantMetadata.TIME_ZONE_ID_KEY, timeZoneId)
- castErrorFor.foreach { name =>
- inner.putString(VariantMetadata.CAST_ERROR_FOR_KEY, name)
- }
- new MetadataBuilder().putMetadata(VariantMetadata.METADATA_KEY,
inner.build()).build()
- }
+ // The value contains three key-value pairs for `path`, `failOnError`, and
`timeZoneId`.
+ def toMetadata: Metadata =
+ new MetadataBuilder().putMetadata(
+ VariantMetadata.METADATA_KEY,
+ new MetadataBuilder()
+ .putString(VariantMetadata.PATH_KEY, path)
+ .putBoolean(VariantMetadata.FAIL_ON_ERROR_KEY, failOnError)
+ .putString(VariantMetadata.TIME_ZONE_ID_KEY, timeZoneId)
+ .build()
+ ).build()
def parsedPath(): Array[VariantPathSegment] = {
VariantPathParser.parse(path).getOrElse {
@@ -75,31 +67,6 @@ object VariantMetadata {
val PATH_KEY = "path"
val FAIL_ON_ERROR_KEY = "failOnError"
val TIME_ZONE_ID_KEY = "timeZoneId"
- // Optional metadata key marking a struct field as a synthetic cast-error
companion. When
- // present, the value is the NAME of the paired data field in the same
variant struct. We tag
- // in metadata (rather than by a field-name convention or sentinel path) so
the marker can't
- // collide with a user-supplied variant path, and so scan-layer schema
rewrites that rename
- // fields by ordinal preserve the marker.
- //
- // Example: with two strict-cast requested fields (b::int and obj.b::double)
and one
- // non-strict-cast requested field (try_cast(c as long)), the rewritten
variant struct looks
- // like:
- // scalastyle:off line.size.limit
- // struct<
- // "0": int metadata = { path: "$.b", failOnError: true, ... },
// data slot for b::int
- // "1": double metadata = { path: "$.obj.b", failOnError: true, ... },
// data slot for obj.b::double
- // "2": long metadata = { path: "$.c", failOnError: false, ... },
// data slot for try_cast(c as long), no companion
- // "3": string metadata = { path: "$", castErrorFor: "0", ... },
// companion paired with data field named "0"
- // "4": string metadata = { path: "$", castErrorFor: "1", ... }
// companion paired with data field named "1"
- // >
- // scalastyle:on line.size.limit
- val CAST_ERROR_FOR_KEY = "castErrorFor"
-
- // Build the metadata for a synthetic cast-error companion. `dataFieldName`
is the NAME of the
- // paired data field in the same variant struct.
- def castErrorCompanionMetadata(dataFieldName: String): Metadata =
- VariantMetadata("$", failOnError = false, timeZoneId = "UTC",
- castErrorFor = Some(dataFieldName)).toMetadata
def isVariantStruct(s: StructType): Boolean =
s.fields.length > 0 && s.fields.forall(_.metadata.contains(METADATA_KEY))
@@ -112,17 +79,10 @@ object VariantMetadata {
// Parse the `VariantMetadata` from a metadata produced by `toMetadata`.
def fromMetadata(metadata: Metadata): VariantMetadata = {
val value = metadata.getMetadata(METADATA_KEY)
- val castErrorFor =
- if (value.contains(CAST_ERROR_FOR_KEY)) {
- Some(value.getString(CAST_ERROR_FOR_KEY))
- } else {
- None
- }
VariantMetadata(
value.getString(PATH_KEY),
value.getBoolean(FAIL_ON_ERROR_KEY),
- value.getString(TIME_ZONE_ID_KEY),
- castErrorFor
+ value.getString(TIME_ZONE_ID_KEY)
)
}
}
@@ -173,15 +133,6 @@ class VariantInRelation {
// Final value: the ordinal of a requested field in the final struct of
requested fields.
val mapping = new HashMap[ExprId, HashMap[Seq[Int],
HashMap[RequestedVariantField, Int]]]
- lazy val deferCastErrorEnabled: Boolean =
- SQLConf.get.getConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR)
-
- // Cast to variant/string never triggers an invalid cast error, so there is
no need to wrap.
- def shouldWrapCastError(field: RequestedVariantField): Boolean =
field.targetType match {
- case _: VariantType | _: StringType => false
- case _ => field.path.failOnError && deferCastErrorEnabled
- }
-
// Extract the SQL-struct path where the leaf is a variant.
object StructPathToVariant {
def unapply(expr: Expression): Option[HashMap[RequestedVariantField, Int]]
= expr match {
@@ -222,8 +173,7 @@ class VariantInRelation {
case _: VariantType =>
mapping.get(attrId).flatMap(_.get(path)) match {
case Some(fields) =>
- val sorted = fields.toArray.sortBy(_._2)
- var dataFields = sorted.map { case (field, ordinal) =>
+ var requestedFields = fields.toArray.sortBy(_._2).map { case
(field, ordinal) =>
StructField(ordinal.toString, field.targetType, metadata =
field.path.toMetadata)
}
// Avoid producing an empty struct of requested fields. This is
intended to simplify the
@@ -231,30 +181,13 @@ class VariantInRelation {
// if the variant is not used, or only used in `IsNotNull/IsNull`
expressions. The value
// of the placeholder field doesn't matter, even if the scan
source accidentally
// contains such a field.
- if (dataFields.isEmpty) {
+ if (requestedFields.isEmpty) {
val placeholder = VariantMetadata("$.__placeholder_field__",
failOnError = false, timeZoneId = "UTC")
- dataFields = Array(StructField("0", BooleanType,
+ requestedFields = Array(StructField("0", BooleanType,
metadata = placeholder.toMetadata))
}
- if (deferCastErrorEnabled) {
- // Append a companion field for each strict cast. The reader
populates it with the
- // offending value on failure; the rewrite consumes both slots
through
- // `UnwrapVariantCastError(error, value)`. The companion's
`castErrorFor` metadata
- // stores the data field's NAME so the pairing survives later
field renaming.
- val companionDataNames = sorted.collect {
- case (field, ordinal) if shouldWrapCastError(field) =>
ordinal.toString
- }
- val numData = dataFields.length
- val companionFields = companionDataNames.zipWithIndex.map {
- case (dataFieldName, idx) =>
- StructField((numData + idx).toString, StringType,
- metadata =
VariantMetadata.castErrorCompanionMetadata(dataFieldName))
- }
- StructType(dataFields ++ companionFields)
- } else {
- StructType(dataFields)
- }
+ StructType(requestedFields)
case _ => dataType
}
case s: StructType if !VariantMetadata.isVariantStruct(s) =>
@@ -303,30 +236,6 @@ class VariantInRelation {
case _ => expr.children.foreach(collectRequestedFields)
}
- // Build the access expression for a requested field. For fields that need
cast-error deferral,
- // wrap with `UnwrapVariantCastError` over the paired companion slot;
otherwise return the bare
- // `GetStructField`.
- private def accessRequestedField(
- fields: HashMap[RequestedVariantField, Int],
- field: RequestedVariantField,
- v: Expression): Expression = {
- val ordinal = fields(field)
- val value = GetStructField(v, ordinal)
- if (shouldWrapCastError(field)) {
- // Locate the companion: the companion's `castErrorFor` equals the data
field's name.
- val variantStruct = v.dataType.asInstanceOf[StructType]
- val dataFieldName = variantStruct.fields(ordinal).name
- val companionOrdinal = variantStruct.fields.indexWhere { f =>
-
VariantMetadata.fromMetadata(f.metadata).castErrorFor.contains(dataFieldName)
- }
- assert(companionOrdinal >= 0,
- s"missing cast-error companion for data field $dataFieldName in
${variantStruct.sql}")
- UnwrapVariantCastError(GetStructField(v, companionOrdinal), value)
- } else {
- value
- }
- }
-
def rewriteExpr(
expr: Expression,
attributeMap: Map[ExprId, AttributeReference]): Expression = {
@@ -339,13 +248,13 @@ class VariantInRelation {
case g@VariantGet(v@StructPathToVariant(fields), path, _, _, _) if
path.foldable =>
// Rewrite the attribute in advance, rather than depending on the last
branch to rewrite it.
// Ww need to avoid the `v@StructPathToVariant(fields)` branch to
rewrite the child again.
- accessRequestedField(fields, RequestedVariantField(g),
rewriteAttribute(v))
+ GetStructField(rewriteAttribute(v), fields(RequestedVariantField(g)))
case c@Cast(v@StructPathToVariant(fields), _, _, _) =>
- accessRequestedField(fields, RequestedVariantField(c),
rewriteAttribute(v))
+ GetStructField(rewriteAttribute(v), fields(RequestedVariantField(c)))
case i@IsNotNull(StructPath(_, _)) => rewriteAttribute(i)
case i@IsNull(StructPath(_, _)) => rewriteAttribute(i)
case v@StructPathToVariant(fields) =>
- accessRequestedField(fields, RequestedVariantField.fullVariant,
rewriteAttribute(v))
+ GetStructField(rewriteAttribute(v),
fields(RequestedVariantField.fullVariant))
case a: Attribute => attributeMap.getOrElse(a.exprId, a)
}
}
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 834b8a56b038..0426b41c6b7a 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
@@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.parquet.io.ColumnIOFactory
import org.apache.parquet.schema.{Type => ParquetType, Types => ParquetTypes}
-import org.apache.spark.SparkRuntimeException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen._
@@ -74,17 +73,9 @@ case class SchemaPathSegment(
// but accessing a raw int should be more efficient than `rawPath`, which
is an `Either`.
extractionIdx: Int)
-// A single output slot of a variant struct (see `VariantMetadata`):
-// - Data field: `path` and `reader` are set. `castErrorOrdinal >= 0` means
an
-// INVALID_VARIANT_CAST during extraction is written into the companion
slot at that ordinal
-// instead of propagating; -1 means no companion.
-// - Companion placeholder (`isCastError`): no extraction; written by the
paired
-// data field on failure, left null on success.
-case class FieldToExtract(
- path: Array[SchemaPathSegment],
- reader: ParquetVariantReader,
- castErrorOrdinal: Int = -1,
- isCastError: Boolean = false)
+// Represent a single field in a variant struct (see `VariantMetadata` for
definition), that is, a
+// single requested field that the scan should produce by extracting from the
variant column.
+case class FieldToExtract(path: Array[SchemaPathSegment], reader:
ParquetVariantReader)
// A helper class to cast from scalar `typed_value` into a scalar `dataType`.
Need a custom
// expression because it has different error reporting code than `Cast`.
@@ -718,66 +709,50 @@ case object SparkShreddingUtils {
.row
}
- // Return a list of output slots. `targetType` must be either variant or
variant struct.
+ // Return a list of fields to extract. `targetType` must be either variant
or variant struct.
// If it is variant, return null because the target is the full variant and
there is no field to
- // extract. If it is variant struct, return one `FieldToExtract` per struct
field.
+ // extract. If it is variant struct, return a list of fields matching the
variant struct fields.
def getFieldsToExtract(targetType: DataType, inputSchema: VariantSchema):
Array[FieldToExtract] =
targetType match {
case _: VariantType => null
case s: StructType if VariantMetadata.isVariantStruct(s) =>
- // Companions are identified by their `castErrorFor` metadata key. The
key's value is the
- // NAME of the paired data field, so the pairing is stable across any
field renaming or
- // reordering that the scan layer might apply.
- val companionIdxByDataName: Map[String, Int] =
s.fields.iterator.zipWithIndex.flatMap {
- case (f, idx) =>
- VariantMetadata.fromMetadata(f.metadata).castErrorFor.map(_ -> idx)
- }.toMap
s.fields.map { f =>
val metadata = VariantMetadata.fromMetadata(f.metadata)
- if (metadata.castErrorFor.isDefined) {
- FieldToExtract(path = null, reader = null, isCastError = true)
- } else {
- val rawPath = metadata.parsedPath()
- val schemaPath = new Array[SchemaPathSegment](rawPath.length)
- var schema = inputSchema
- // Search `rawPath` in `schema` to produce `schemaPath`. If a raw
path segment cannot
- // be found at a certain level of the file type, then `typedIdx`
will be -1 starting
- // from this position, and the final `schema` will be null.
- for (i <- rawPath.indices) {
- val isObject = rawPath(i).isInstanceOf[ObjectExtraction]
- var typedIdx = -1
- var extractionIdx = -1
- rawPath(i) match {
- case ObjectExtraction(key) if schema != null &&
schema.objectSchema != null =>
- val fieldIdx = schema.objectSchemaMap.get(key)
- if (fieldIdx != null) {
- typedIdx = schema.typedIdx
- extractionIdx = fieldIdx
- schema = schema.objectSchema(fieldIdx).schema
- } else {
- schema = null
- }
- case ArrayExtraction(index) if schema != null &&
schema.arraySchema != null =>
+ val rawPath = metadata.parsedPath()
+ val schemaPath = new Array[SchemaPathSegment](rawPath.length)
+ var schema = inputSchema
+ // Search `rawPath` in `schema` to produce `schemaPath`. If a raw
path segment cannot be
+ // found at a certain level of the file type, then `typedIdx` will
be -1 starting from
+ // this position, and the final `schema` will be null.
+ for (i <- rawPath.indices) {
+ val isObject = rawPath(i).isInstanceOf[ObjectExtraction]
+ var typedIdx = -1
+ var extractionIdx = -1
+ rawPath(i) match {
+ case ObjectExtraction(key) if schema != null &&
schema.objectSchema != null =>
+ val fieldIdx = schema.objectSchemaMap.get(key)
+ if (fieldIdx != null) {
typedIdx = schema.typedIdx
- extractionIdx = index
- schema = schema.arraySchema
- case _ =>
+ extractionIdx = fieldIdx
+ schema = schema.objectSchema(fieldIdx).schema
+ } else {
schema = null
- }
- schemaPath(i) = SchemaPathSegment(rawPath(i), isObject,
typedIdx, extractionIdx)
- }
- val reader = ParquetVariantReader(schema, f.dataType,
VariantCastArgs(
- metadata.failOnError,
- Some(metadata.timeZoneId),
- DateTimeUtils.getZoneId(metadata.timeZoneId)),
- isTopLevelUnshredded = schemaPath.isEmpty &&
inputSchema.isUnshredded)
- val castErrorOrdinal = companionIdxByDataName.getOrElse(f.name, -1)
- if (castErrorOrdinal >= 0) {
- assert(metadata.failOnError,
- "cast-error-deferred variant field must have failOnError=true")
+ }
+ case ArrayExtraction(index) if schema != null &&
schema.arraySchema != null =>
+ typedIdx = schema.typedIdx
+ extractionIdx = index
+ schema = schema.arraySchema
+ case _ =>
+ schema = null
}
- FieldToExtract(schemaPath, reader, castErrorOrdinal =
castErrorOrdinal)
+ schemaPath(i) = SchemaPathSegment(rawPath(i), isObject, typedIdx,
extractionIdx)
}
+ val reader = ParquetVariantReader(schema, f.dataType,
VariantCastArgs(
+ metadata.failOnError,
+ Some(metadata.timeZoneId),
+ DateTimeUtils.getZoneId(metadata.timeZoneId)),
+ isTopLevelUnshredded = schemaPath.isEmpty &&
inputSchema.isUnshredded)
+ FieldToExtract(schemaPath, reader)
}
case _ =>
throw QueryExecutionErrors.unreachableError(s"Invalid target type:
`${targetType.sql}`")
@@ -849,10 +824,6 @@ case object SparkShreddingUtils {
}
// Assemble a variant struct, in which each field is extracted from the
Parquet variant value.
- // For data fields paired with a cast-error companion (`castErrorFor`
metadata key on the
- // companion field naming the partner data field), an INVALID_VARIANT_CAST
raised by the strict
- // cast is routed into the companion slot so the error is deferred until the
row is consumed by
- // the user expression.
def assembleVariantStruct(
inputRow: InternalRow,
schema: VariantSchema,
@@ -865,26 +836,8 @@ case object SparkShreddingUtils {
val resultRow = new GenericInternalRow(numFields)
var fieldIdx = 0
while (fieldIdx < numFields) {
- 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
- // deferred RaiseError can surface the same value that an eager
raise would have.
- val offendingValue =
- Option(e.getMessageParameters.get("value")).getOrElse("")
- resultRow.update(field.castErrorOrdinal,
- UTF8String.fromString(offendingValue))
- }
- } else {
- resultRow.update(fieldIdx, extractField(inputRow, topLevelMetadata,
schema, field.path,
- field.reader))
- }
+ resultRow.update(fieldIdx, extractField(inputRow, topLevelMetadata,
schema,
+ fields(fieldIdx).path, fields(fieldIdx).reader))
fieldIdx += 1
}
resultRow
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
index a18e229a24e0..a1c69847c509 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
@@ -36,10 +36,9 @@ import
org.apache.spark.sql.connector.expressions.aggregate.{Aggregation, Avg, C
import org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read.{Scan, ScanBuilder,
SupportsPushDownAggregates, SupportsPushDownFilters, SupportsPushDownJoin,
SupportsPushDownVariantExtractions, V1Scan, VariantExtraction}
import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
VariantInRelation}
-import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.connector.VariantExtractionImpl
import org.apache.spark.sql.sources
-import org.apache.spark.sql.types.{DataType, DecimalType, IntegerType,
StringType, StructField, StructType}
+import org.apache.spark.sql.types.{DataType, DecimalType, IntegerType,
StructField, StructType}
import org.apache.spark.sql.util.SchemaUtils._
import org.apache.spark.util.ArrayImplicits._
@@ -416,9 +415,6 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
// Build individual VariantExtraction for each field access
// Track which extraction corresponds to which (attr, field, ordinal)
- // Cast-error deferral attaches a synthetic companion field to every
strict-cast extraction;
- // record whether any are generated so we can require reader support below.
- var hasCompanionExtraction = false
val extractionInfo = schemaAttributes.flatMap { topAttr =>
val variantFields = variants.mapping.get(topAttr.exprId)
if (variantFields.isEmpty || variantFields.get.isEmpty) {
@@ -432,10 +428,7 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
Seq(topAttr.name) ++
getColumnName(topAttr.dataType.asInstanceOf[StructType],
pathToVariant)
}
- // Keep data extractions in the same order as
`VariantInRelation.rewriteType`, so
- // companion fields can refer to their paired data field by name.
- val sorted = fields.toArray.sortBy(_._2)
- val dataExtractions = sorted.map { case (field, ordinal) =>
+ fields.toArray.sortBy(_._2).map { case (field, ordinal) =>
val extraction = new VariantExtractionImpl(
columnName.toArray,
field.path.toMetadata,
@@ -443,21 +436,6 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
)
(extraction, topAttr, field, ordinal)
}
- if (variants.deferCastErrorEnabled) {
- val companionExtractions = sorted.collect {
- case (field, ordinal) if variants.shouldWrapCastError(field) =>
- val extraction = new VariantExtractionImpl(
- columnName.toArray,
- VariantMetadata.castErrorCompanionMetadata(ordinal.toString),
- StringType
- )
- (extraction, topAttr, field, ordinal)
- }
- if (companionExtractions.nonEmpty) hasCompanionExtraction = true
- dataExtractions ++ companionExtractions
- } else {
- dataExtractions
- }
}
}
}
@@ -465,11 +443,6 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
// Call the API to push down variant extractions
if (extractionInfo.isEmpty) return originalPlan
- // Companion extractions can only be honored by readers that support
cast-error deferral. If
- // none were generated, the pushdown carries only non-strict accesses
(`try_variant_get`, plain
- // variant reads, casts to variant/string) that are safe regardless of
deferral support.
- if (hasCompanionExtraction && !builder.supportsDeferCastError()) return
originalPlan
-
val extractions: Array[VariantExtraction] =
extractionInfo.map(_._1).toArray
val pushedResults = builder.pushVariantExtractions(extractions)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
index 9a7edde3f749..d0c7859964e0 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
@@ -65,12 +65,13 @@ case class ParquetScan(
}
private def rewriteVariantPushdownSchema(schema: StructType): StructType = {
- // Field names follow the defer-cast-error contract: companion metadata
refers to
- // the paired data field by its group-local name.
+ // Group extractions by column name and build extracted schemas
val variantSchemaMap: Map[Seq[String], StructType] =
pushedVariantExtractions
.groupBy(e => e.columnName().toSeq)
.map { case (colName, extractions) =>
+ // Build struct schema with ordinal-named fields for each extraction
var fields = extractions.zipWithIndex.map { case (extraction, idx) =>
+ // Attach VariantMetadata so Parquet reader knows this is a variant
extraction
StructField(idx.toString, extraction.expectedDataType(), nullable =
true,
extraction.metadata())
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
index 149d7e6f0b72..94da53f22934 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
@@ -103,8 +103,6 @@ case class ParquetScanBuilder(
}
// SupportsPushDownVariantExtractions API implementation
- override def supportsDeferCastError(): Boolean = true
-
override def pushVariantExtractions(extractions: Array[VariantExtraction]):
Array[Boolean] = {
// Parquet supports variant pushdown for all variant extractions
pushedVariantExtractions = extractions
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
index d6a9cfc94e8c..41b78881b788 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
@@ -31,30 +31,6 @@ trait PushVariantIntoScanSuiteBase extends
SharedSparkSession {
override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.PUSH_VARIANT_INTO_SCAN.key, "true")
- // Whether the reader-deferral tests should exercise the V2 read path.
Subclasses override.
- protected def useV2: Boolean
-
- // Write a parquet dataset via V1, then expose it as the temp view `T`. The
view's read path is
- // V2 when `useV2`, V1 otherwise. Use this for tests that need to actually
execute a scan and
- // compare V1 vs V2 behavior.
- protected def withVariantParquetData(schema: String, inserts: String*)(body:
=> Unit): Unit = {
- withTempPath { dir =>
- val path = dir.getCanonicalPath
- // External (LOCATION) table, so `withTable` only drops the catalog
entry - the parquet
- // files at `path` survive for the subsequent V2 read.
- withTable("temp_variant_setup") {
- sql(s"create table temp_variant_setup ($schema) using PARQUET location
'$path'")
- inserts.foreach(values => sql(s"insert into temp_variant_setup values
$values"))
- }
- val sourceListConf: Seq[(String, String)] =
- if (useV2) Seq(SQLConf.USE_V1_SOURCE_LIST.key -> "") else Nil
- withSQLConf(sourceListConf: _*) {
- spark.read.parquet(path).createOrReplaceTempView("T")
- try body finally spark.catalog.dropTempView("T")
- }
- }
- }
-
protected def localTimeZone = spark.sessionState.conf.sessionLocalTimeZone
// Return a `StructField` with the expected `VariantMetadata`.
@@ -73,208 +49,6 @@ trait PushVariantIntoScanSuiteBase extends
SharedSparkSession {
}
}
- // Returns true iff `t` or any of its causes is an INVALID_VARIANT_CAST
error. The failure may
- // surface directly or be wrapped in a task failure.
- protected 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)
- }
-
- 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())
- 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())
- 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())
- assert(hasCastCondition(ex), s"Expected INVALID_VARIANT_CAST, got
$ex")
- }
- }
- }
- }
-
- test(s"Reader defers strict-cast errors through AND/OR short-circuit") {
- // Row 0: number 1 (LONG in variant) -> cast(v as int) succeeds.
- // Row 1: string -> cast(v as int) would raise INVALID_VARIANT_CAST.
- //
- // The strict cast is a child of an `AND`/`OR` that is projected as a
boolean value. The
- // `AND`/`OR` must be evaluated lazily/left-to-right with short-circuit:
when the left operand
- // already decides the result (false for `AND`, true for `OR`) the right
operand (the wrapped
- // cast) is not consumed, so the deferred cast error on the string row
must not surface.
- withVariantParquetData("v variant",
- "(parse_json('1'))",
- "(parse_json('\"hello\"'))") {
- // For each case: the projected expression, and the expected (sorted)
values with deferral on.
- // - AND: row 0 = 'BIGINT'='BIGINT' (true) AND 1 > 5 (false) -> false;
- // row 1 = 'STRING'='BIGINT' (false) -> false (cast deferred,
never consumed).
- // - OR: row 0 = 'BIGINT'='STRING' (false) OR 1 > 5 (false) -> false;
- // row 1 = 'STRING'='STRING' (true) -> true (cast deferred, never
consumed).
- val cases = Seq(
- "schema_of_variant(v) = 'BIGINT' and cast(v as int) > 5" -> Seq(false,
false),
- "schema_of_variant(v) = 'STRING' or cast(v as int) > 5" -> Seq(false,
true))
-
- for ((expr, expected) <- cases) {
- val query = s"select $expr as a from T"
-
- // Without the deferral, the strict cast pushed into the scan raises
at the failing row even
- // though the `AND`/`OR` would have short-circuited past it.
- withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key ->
"false") {
- val ex = intercept[Exception](sql(query).collect())
- assert(hasCastCondition(ex), s"[$expr] Expected
INVALID_VARIANT_CAST, got $ex")
- }
-
- // With the deferral, the short-circuit happens before the failing row
is consumed.
- // Read order is not guaranteed, so compare the sorted values.
- withSQLConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR.key ->
"true") {
- val values = sql(query).collect().map(_.getBoolean(0)).sorted.toSeq
- assert(values == expected, s"[$expr] Got ${values.mkString(",")}")
- }
- }
- }
- }
}
// V1 DataSource tests with parameterized reader type
@@ -282,8 +56,6 @@ abstract class PushVariantIntoScanV1SuiteBase extends
PushVariantIntoScanSuiteBa
protected def vectorizedReaderEnabled: Boolean
protected def readerName: String
- override protected def useV2: Boolean = false
-
override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key,
vectorizedReaderEnabled.toString)
@@ -466,8 +238,6 @@ abstract class PushVariantIntoScanV2SuiteBase extends
QueryTest with PushVariant
protected def vectorizedReaderEnabled: Boolean
protected def readerName: String
- override protected def useV2: Boolean = true
-
override def sparkConf: SparkConf =
super.sparkConf.set(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key,
vectorizedReaderEnabled.toString)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
