This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 5e9c4b8a310b [SPARK-57445][SQL] Fix PushVariantIntoScan exception
semantics.
5e9c4b8a310b is described below
commit 5e9c4b8a310b8324aaa9425fdec7aba33e2a2488
Author: chenhao-db <[email protected]>
AuthorDate: Wed Jun 17 15:56:50 2026 -0700
[SPARK-57445][SQL] Fix PushVariantIntoScan exception semantics.
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)
To ensure that user doesn't get surprising result when
`PushVariantIntoScan` is enabled.
No.
New unit tests.
Yes. Co-authored with Claude Opus 4.8.
Closes #56505 from chenhao-db/fix_variant_push.
Authored-by: chenhao-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 1284cfc03599ab3db564cdcaef366496e10dbf3b)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../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, 556 insertions(+), 62 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 750e0479e542..2eb0038e6e92 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,6 +38,31 @@ 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 5d78f11bf86f..88ab4676c36d 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,3 +985,63 @@ 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 b69e0428630c..0a208afc5023 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,6 +6202,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")
+ .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 b0b20d08dccb..69aa576f0c94 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,18 +41,26 @@ case class VariantMetadata(
// `[*]` is not supported.
path: String,
failOnError: Boolean,
- timeZoneId: String) {
+ 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) {
// Produce a metadata contain one key-value pair. The key is the special
`METADATA_KEY`.
- // 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()
+ // 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()
+ }
def parsedPath(): Array[VariantPathSegment] = {
VariantPathParser.parse(path).getOrElse {
@@ -67,6 +75,31 @@ 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))
@@ -79,10 +112,17 @@ 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)
+ value.getString(TIME_ZONE_ID_KEY),
+ castErrorFor
)
}
}
@@ -133,6 +173,15 @@ 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 {
@@ -173,7 +222,8 @@ class VariantInRelation {
case _: VariantType =>
mapping.get(attrId).flatMap(_.get(path)) match {
case Some(fields) =>
- var requestedFields = fields.toArray.sortBy(_._2).map { case
(field, ordinal) =>
+ val sorted = fields.toArray.sortBy(_._2)
+ var dataFields = sorted.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
@@ -181,13 +231,30 @@ 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 (requestedFields.isEmpty) {
+ if (dataFields.isEmpty) {
val placeholder = VariantMetadata("$.__placeholder_field__",
failOnError = false, timeZoneId = "UTC")
- requestedFields = Array(StructField("0", BooleanType,
+ dataFields = Array(StructField("0", BooleanType,
metadata = placeholder.toMetadata))
}
- StructType(requestedFields)
+ 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)
+ }
case _ => dataType
}
case s: StructType if !VariantMetadata.isVariantStruct(s) =>
@@ -236,6 +303,30 @@ 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 = {
@@ -248,13 +339,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.
- GetStructField(rewriteAttribute(v), fields(RequestedVariantField(g)))
+ accessRequestedField(fields, RequestedVariantField(g),
rewriteAttribute(v))
case c@Cast(v@StructPathToVariant(fields), _, _, _) =>
- GetStructField(rewriteAttribute(v), fields(RequestedVariantField(c)))
+ accessRequestedField(fields, RequestedVariantField(c),
rewriteAttribute(v))
case i@IsNotNull(StructPath(_, _)) => rewriteAttribute(i)
case i@IsNull(StructPath(_, _)) => rewriteAttribute(i)
case v@StructPathToVariant(fields) =>
- GetStructField(rewriteAttribute(v),
fields(RequestedVariantField.fullVariant))
+ accessRequestedField(fields, RequestedVariantField.fullVariant,
rewriteAttribute(v))
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 0426b41c6b7a..834b8a56b038 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,6 +20,7 @@ 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._
@@ -73,9 +74,17 @@ case class SchemaPathSegment(
// but accessing a raw int should be more efficient than `rawPath`, which
is an `Either`.
extractionIdx: Int)
-// 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 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)
// 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`.
@@ -709,50 +718,66 @@ case object SparkShreddingUtils {
.row
}
- // Return a list of fields to extract. `targetType` must be either variant
or variant struct.
+ // Return a list of output slots. `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 a list of fields matching the
variant struct fields.
+ // extract. If it is variant struct, return one `FieldToExtract` per struct
field.
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)
- 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) {
+ 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 =>
typedIdx = schema.typedIdx
- extractionIdx = fieldIdx
- schema = schema.objectSchema(fieldIdx).schema
- } else {
+ extractionIdx = index
+ schema = schema.arraySchema
+ case _ =>
schema = null
- }
- case ArrayExtraction(index) if schema != null &&
schema.arraySchema != null =>
- typedIdx = schema.typedIdx
- extractionIdx = index
- schema = schema.arraySchema
- case _ =>
- schema = null
+ }
+ schemaPath(i) = SchemaPathSegment(rawPath(i), isObject,
typedIdx, extractionIdx)
}
- 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")
+ }
+ FieldToExtract(schemaPath, reader, castErrorOrdinal =
castErrorOrdinal)
}
- 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}`")
@@ -824,6 +849,10 @@ 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,
@@ -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
+ // 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))
+ }
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 a1c69847c509..a18e229a24e0 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,9 +36,10 @@ 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,
StructField, StructType}
+import org.apache.spark.sql.types.{DataType, DecimalType, IntegerType,
StringType, StructField, StructType}
import org.apache.spark.sql.util.SchemaUtils._
import org.apache.spark.util.ArrayImplicits._
@@ -415,6 +416,9 @@ 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) {
@@ -428,7 +432,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
Seq(topAttr.name) ++
getColumnName(topAttr.dataType.asInstanceOf[StructType],
pathToVariant)
}
- fields.toArray.sortBy(_._2).map { case (field, ordinal) =>
+ // 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) =>
val extraction = new VariantExtractionImpl(
columnName.toArray,
field.path.toMetadata,
@@ -436,6 +443,21 @@ 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
+ }
}
}
}
@@ -443,6 +465,11 @@ 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 d0c7859964e0..9a7edde3f749 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,13 +65,12 @@ case class ParquetScan(
}
private def rewriteVariantPushdownSchema(schema: StructType): StructType = {
- // Group extractions by column name and build extracted schemas
+ // Field names follow the defer-cast-error contract: companion metadata
refers to
+ // the paired data field by its group-local name.
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 94da53f22934..149d7e6f0b72 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,6 +103,8 @@ 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 41b78881b788..d6a9cfc94e8c 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,6 +31,30 @@ 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`.
@@ -49,6 +73,208 @@ 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
@@ -56,6 +282,8 @@ 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)
@@ -238,6 +466,8 @@ 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]
