This is an automated email from the ASF dual-hosted git repository.
weiting-chen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new ea80387689 [VL] Add RDDScanExec support to Velox backend (#12982)
ea80387689 is described below
commit ea80387689cb35746bcd7dc94ee68e034f6e04e4
Author: Minni Mittal <[email protected]>
AuthorDate: Fri Sep 11 11:21:54 2026 +0530
[VL] Add RDDScanExec support to Velox backend (#12982)
* [VL] Add RDDScanExec support to Velox backend
Offload RDDScanExec to the Velox backend by converting the underlying
RDD[InternalRow] into columnar batches through the native
row-to-columnar path (the same JNI path used by RowToVeloxColumnarExec).
Highlights:
- New VeloxRDDScanTransformer implementing the RDDScanTransformer
contract. Schemas not supported by the Arrow export path (e.g. map or
interval types) are rejected in validation and fall back to vanilla
Spark.
- Wire isSupportRDDScanExec / getRDDScanTransform in VeloxSparkPlanExecApi.
Offload is gated by a new config and skipped inside a Structured
Streaming query (micro-batch / foreachBatch), where offloading a
materialized per-batch source into a state-store pipeline can deadlock.
- New config spark.gluten.sql.columnar.backend.velox.rddScan.enabled
(default true), which also acts as a runtime kill-switch.
- Nullability-aware row conversion: for schemas with non-nullable fields,
rows are pre-projected with a nullability-honoring UnsafeProjection so a
null in a non-nullable field yields the type default, matching Spark's
WholeStageCodegen behavior (SPARK-35912). The guard recurses into nested
struct/array/map types.
- Guard ColumnarPartialProjectExec against an empty projectAttributes set
(a projection that references no child column), which can arise once
RDDScan feeds downstream partial projection.
- Add VeloxRDDScanSuite covering type coverage, aggregation downstream,
empty/duplicate reads, null handling, checkpoint BatchCarrierRow reuse,
map/interval fallback, nullability coercion, kill-switch fallback, and
the streaming-query skip.
Co-authored-by: Copilot <[email protected]>
* [VL] Fix RDDScan CI regressions
Reject zero-column RDDScan schemas so Spark's OneRowRelation remains on the
row path, preserving existing plan assertions and parameterized EXPLAIN
output.
Add a regression test for the fallback and regenerate the Velox
configuration
documentation for the RDDScan setting.
Co-authored-by: Copilot <[email protected]>
---------
Co-authored-by: Copilot <[email protected]>
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 36 ++
.../org/apache/gluten/config/VeloxConfig.scala | 12 +
.../execution/ColumnarPartialProjectExec.scala | 4 +
.../gluten/execution/VeloxRDDScanTransformer.scala | 225 ++++++++++++
.../spark/sql/execution/VeloxRDDScanSuite.scala | 378 +++++++++++++++++++++
docs/velox-configuration.md | 1 +
6 files changed, 656 insertions(+)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
index c00803c521..5e3a88941a 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -37,6 +37,7 @@ import org.apache.spark.memory.SparkMemoryUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.serializer.Serializer
import org.apache.spark.shuffle.{GenShuffleReaderParameters,
GenShuffleWriterParameters, GlutenShuffleReaderWrapper,
GlutenShuffleWriterWrapper, VeloxShuffleUtils}
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions._
@@ -1416,6 +1417,41 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi
with Logging {
override def genColumnarRangeExec(rangeExec: RangeExec):
ColumnarRangeBaseExec =
ColumnarRangeExec(rangeExec.range)
+ override def isSupportRDDScanExec(plan: RDDScanExec): Boolean = {
+ if (!VeloxConfig.get.enableRddScan) {
+ logDebug(
+ "RDDScan offload skipped: " +
+ s"${VeloxConfig.COLUMNAR_VELOX_RDD_SCAN_ENABLED.key}=false")
+ return false
+ }
+ // Exclude any scan planned within a Structured Streaming query
(micro-batch or its
+ // foreachBatch callback). The per-batch source RDD is a materialized
snapshot that
+ // otherwise slips past the plan-level logicalLink.isStreaming fallback,
yet offloading
+ // it into a streaming/state-store pipeline can deadlock the micro-batch.
+ if (isWithinStreamingQuery) {
+ logDebug("RDDScan offload skipped: within a streaming query
(micro-batch/foreachBatch)")
+ return false
+ }
+ true
+ }
+
+ /**
+ * Whether the current thread is planning/executing a Structured Streaming
query -- including the
+ * `DataFrameWriter.foreachBatch` user callback, which Spark runs on the
StreamExecution driver
+ * thread. That thread sets the `sql.streaming.queryId` local property
+ * (`StreamExecution.QUERY_ID_KEY`) for the whole lifetime of the query. We
match on the literal
+ * key rather than referencing the class so this stays agnostic to the
per-Spark-version package
+ * of `StreamExecution` across shims.
+ */
+ private def isWithinStreamingQuery: Boolean =
+ SparkSession.getActiveSession
+ .map(_.sparkContext)
+ .flatMap(sc => Option(sc.getLocalProperty("sql.streaming.queryId")))
+ .isDefined
+
+ override def getRDDScanTransform(plan: RDDScanExec): RDDScanTransformer =
+ VeloxRDDScanTransformer.replace(plan)
+
override def genColumnarTailExec(limit: Int, child: SparkPlan):
ColumnarCollectTailBaseExec =
ColumnarCollectTailExec(limit, child)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index d735184916..7fb3f69dd3 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -89,6 +89,8 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
def veloxPreferredBatchBytes: Long =
getConf(COLUMNAR_VELOX_PREFERRED_BATCH_BYTES)
+ def enableRddScan: Boolean = getConf(COLUMNAR_VELOX_RDD_SCAN_ENABLED)
+
def cudfEnableTableScan: Boolean = getConf(CUDF_ENABLE_TABLE_SCAN)
def cudfEnableValidation: Boolean = getConf(CUDF_ENABLE_VALIDATION)
@@ -940,6 +942,16 @@ object VeloxConfig extends ConfigRegistry {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("10MB")
+ val COLUMNAR_VELOX_RDD_SCAN_ENABLED =
+ buildConf("spark.gluten.sql.columnar.backend.velox.rddScan.enabled")
+ .doc(
+ "When true, offload RDDScanExec to Velox by converting the
RDD[InternalRow] into" +
+ " columnar batches through the native row-to-columnar path. Schemas
that are not" +
+ " supported by the Arrow export path (e.g. map or interval types)
fall back to" +
+ " vanilla Spark.")
+ .booleanConf
+ .createWithDefault(true)
+
val VELOX_MAX_COMPILED_REGEXES =
buildConf("spark.gluten.sql.columnar.backend.velox.maxCompiledRegexes")
.doc(
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
index d657c3e303..f2884af5ce 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala
@@ -133,6 +133,10 @@ case class ColumnarPartialProjectExec(projectList:
Seq[Expression], child: Spark
return ValidationResult.failed(
"Attribute in the partial projected expressions contains unsupported
type")
}
+ if (projectAttributes.isEmpty) {
+ return ValidationResult.failed(
+ "The partial projected expressions do not reference any child column")
+ }
if (projectAttributes.size == child.output.size) {
return ValidationResult.failed(
"The partial projected expressions need all the columns in child
output")
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxRDDScanTransformer.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxRDDScanTransformer.scala
new file mode 100644
index 0000000000..d875122f56
--- /dev/null
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxRDDScanTransformer.scala
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.backendsapi.velox.VeloxValidatorApi
+import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
SortOrder, UnsafeProjection}
+import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.execution.{RDDScanTransformer, SparkPlan}
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * Velox-backend implementation of RDDScanTransformer.
+ *
+ * Converts an RDD[InternalRow] into columnar batches using Velox's native
row-to-columnar
+ * conversion (same JNI path as RowToVeloxColumnarExec).
+ */
+case class VeloxRDDScanTransformer(
+ outputAttributes: Seq[Attribute],
+ rdd: RDD[InternalRow],
+ name: String,
+ // Row-to-columnar conversion preserves data distribution, so we carry
through
+ // the original partitioning. This differs from CH which uses
UnknownPartitioning(0)
+ // but is consistent with RowToVeloxColumnarExec's behavior.
+ override val outputPartitioning: Partitioning,
+ override val outputOrdering: Seq[SortOrder]
+) extends RDDScanTransformer(outputAttributes, outputPartitioning,
outputOrdering)
+ with Logging {
+
+ override def nodeName: String = name
+
+ @transient override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numInputRows" -> SQLMetrics.createMetric(sparkContext, "number of input
rows"),
+ "numOutputBatches" -> SQLMetrics.createMetric(sparkContext, "number of
output batches"),
+ "convertTime" -> SQLMetrics.createTimingMetric(sparkContext, "time to
convert")
+ )
+
+ override protected def doValidateInternal(): ValidationResult = {
+ if (schema.isEmpty) {
+ return ValidationResult.failed("RDDScan with an empty schema is not
supported")
+ }
+ for (field <- schema.fields) {
+ val reason = VeloxValidatorApi.validateSchema(field.dataType)
+ if (reason.isDefined) {
+ return ValidationResult.failed(reason.get)
+ }
+ val arrowReason = validateArrowCompatibility(field.dataType)
+ if (arrowReason.isDefined) {
+ return ValidationResult.failed(arrowReason.get)
+ }
+ }
+ ValidationResult.succeeded
+ }
+
+ override def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ val numInputRows = longMetric("numInputRows")
+ val numOutputBatches = longMetric("numOutputBatches")
+ val convertTime = longMetric("convertTime")
+ val localSchema = this.schema
+ val batchSize = GlutenConfig.get.maxBatchSize
+ val batchBytes = VeloxConfig.get.veloxPreferredBatchBytes
+ rdd.mapPartitions {
+ iter =>
+ if (iter.hasNext) {
+ val first = iter.next()
+ // A partition is homogeneous by construction: a checkpoint/cache of
a Gluten
+ // columnar plan yields all BatchCarrierRows, while any other RDD
yields all
+ // InternalRows. We therefore select the path once, based on the
first row, and
+ // apply it to the whole partition.
+ first match {
+ case _: BatchCarrierRow =>
+ // RDD already contains columnar batches wrapped as carrier rows
+ // (e.g., from df.checkpoint() on a Gluten plan). Unwrap
directly.
+ // No row conversion happens here, so convertTime is
intentionally left at 0;
+ // numInputRows is credited with the already-batched row count
for observability.
+ (Iterator.single(first) ++ iter).flatMap {
+ row =>
+ BatchCarrierRow.unwrap(row).map {
+ batch =>
+ numOutputBatches += 1
+ numInputRows += batch.numRows()
+ batch
+ }
+ }
+ case _ =>
+ // Standard InternalRow path - convert via native
row-to-columnar.
+ val rowIter = Iterator.single(first) ++ iter
+ val processedIter = if (schemaHasNonNullableField(localSchema)) {
+ // Pre-convert rows to UnsafeRow respecting schema nullability.
+ // This matches Spark's codegen behavior: for non-nullable
fields, getLong/getInt
+ // is called directly without isNullAt check, so null values
in non-nullable
+ // columns produce the type's default (0 for Long) rather than
setting null bits.
+ // Without this, UnsafeProjection.create(schema) always uses
nullable=true,
+ // which would incorrectly propagate nulls for non-nullable
fields (SPARK-35912).
+ // The guard recurses into nested struct/array/map types so a
non-nullable field
+ // nested under a nullable top-level column still routes
through the projection
+ // (which itself handles nested nullability), rather than
silently skipping the fix.
+ createNullabilityAwareIterator(rowIter, localSchema)
+ } else {
+ rowIter
+ }
+ RowToVeloxColumnarExec.toColumnarBatchIterator(
+ processedIter,
+ localSchema,
+ numInputRows,
+ numOutputBatches,
+ convertTime,
+ batchSize,
+ batchBytes)
+ }
+ } else {
+ Iterator.empty
+ }
+ }
+ }
+
+ /**
+ * Returns true if `dataType` declares any non-nullable field at any nesting
level. The top-level
+ * routing guard uses this so that a non-nullable field nested inside an
otherwise-nullable
+ * struct/array/map column still triggers the nullability-aware projection,
matching Spark's
+ * codegen null->default behavior for non-nullable fields (SPARK-35912).
+ *
+ * Coercion is exact for (nested) struct fields, which UnsafeProjection
writes via typed getters
+ * that turn a null into the primitive default. It is best-effort for
elements nested inside a
+ * `containsNull = false` array or a `valueContainsNull = false` map: the
projection copies the
+ * array/map payload wholesale and does not rewrite individual element
nulls. Such payloads (a
+ * null element in a declared non-null collection produced by a raw RDD) are
pathological and not
+ * expected in practice; the guard still routes them through the projection
for consistency.
+ */
+ private def schemaHasNonNullableField(dataType: DataType): Boolean =
dataType match {
+ case s: StructType =>
+ s.fields.exists(f => !f.nullable ||
schemaHasNonNullableField(f.dataType))
+ case a: ArrayType =>
+ !a.containsNull || schemaHasNonNullableField(a.elementType)
+ case m: MapType =>
+ // Map keys are non-null by Spark semantics (not a data-null risk), so
they do not by
+ // themselves trigger the guard; only declared non-null values or nested
non-nullable
+ // fields in the key/value types do.
+ !m.valueContainsNull || schemaHasNonNullableField(m.keyType) ||
+ schemaHasNonNullableField(m.valueType)
+ case _ => false
+ }
+
+ /**
+ * Creates an iterator that converts InternalRows to UnsafeRows while
respecting schema
+ * nullability. For non-nullable fields, values are read via typed getters
(getLong, getInt, etc.)
+ * which return default values (0) for null inputs, matching Spark's
WholeStageCodegen behavior.
+ */
+ private def createNullabilityAwareIterator(
+ iter: Iterator[InternalRow],
+ schema: StructType): Iterator[InternalRow] = {
+ // Create BoundReferences that respect the schema's declared nullability.
+ // When nullable=false, the generated code calls getLong/getInt directly
without
+ // checking isNullAt, so null.asInstanceOf[Long] unboxes to 0.
+ val boundRefs = schema.fields.zipWithIndex.map {
+ case (field, i) => BoundReference(i, field.dataType, field.nullable)
+ }.toSeq
+ val projection = UnsafeProjection.create(boundRefs)
+ iter.map {
+ row =>
+ // The projection returns a mutable UnsafeRow that is reused across
calls. This is safe
+ // for two reasons: (1) toColumnarBatchIterator copies each row's
bytes into an ArrowBuf
+ // via Platform.copyMemory before advancing the iterator, and (2) its
convertToUnsafeRow
+ // passes an already-UnsafeRow straight through without re-projecting,
so this
+ // nullability-aware projection replaces (not adds to) the converter's
internal one.
+ projection.apply(row)
+ }
+ }
+
+ /**
+ * Additional validation for Arrow export compatibility. The RDDScan path
transfers data via Arrow
+ * ABI, which has stricter constraints than Velox's type system:
+ * - Map types can trigger "Map data key type should be a non-nullable" in
Arrow export
+ * - Interval types are not supported by ArrowWritableColumnVector
+ */
+ private def validateArrowCompatibility(dataType: DataType): Option[String] =
{
+ dataType match {
+ case _: MapType =>
+ Some(s"Map type is not supported in RDDScan Arrow export path:
$dataType")
+ case _: YearMonthIntervalType | _: DayTimeIntervalType |
CalendarIntervalType =>
+ Some(s"Interval type is not supported in Arrow export: $dataType")
+ case struct: StructType =>
+ struct.fields.flatMap(f =>
validateArrowCompatibility(f.dataType)).headOption
+ case array: ArrayType =>
+ validateArrowCompatibility(array.elementType)
+ case _ => None
+ }
+ }
+
+ override protected def withNewChildrenInternal(newChildren:
IndexedSeq[SparkPlan]): SparkPlan = {
+ assert(newChildren.isEmpty, "VeloxRDDScanTransformer is a leaf node")
+ copy(outputAttributes, rdd, name, outputPartitioning, outputOrdering)
+ }
+}
+
+object VeloxRDDScanTransformer {
+
+ def replace(plan: org.apache.spark.sql.execution.RDDScanExec):
RDDScanTransformer =
+ VeloxRDDScanTransformer(
+ plan.output,
+ plan.inputRDD,
+ plan.nodeName,
+ plan.outputPartitioning,
+ plan.outputOrdering)
+}
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxRDDScanSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxRDDScanSuite.scala
new file mode 100644
index 0000000000..df55258f0e
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxRDDScanSuite.scala
@@ -0,0 +1,378 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution
+
+import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.config.VeloxConfig
+import org.apache.gluten.execution._
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference}
+import org.apache.spark.sql.classic.{ClassicDataset, ClassicTypes}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.types._
+import org.apache.spark.util.Utils
+
+class VeloxRDDScanSuite extends VeloxWholeStageTransformerSuite with
AdaptiveSparkPlanHelper {
+
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.sql.ansi.enabled", "false")
+ .set("spark.gluten.sql.columnar.backend.velox.rddScan.enabled", "true")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
+ }
+
+ /** Creates a DataFrame backed by LogicalRDD/RDDScanExec from an existing
DataFrame. */
+ private def asRDDScanDF(data: DataFrame): DataFrame = {
+ val node = LogicalRDD(data.queryExecution.analyzed.output,
data.queryExecution.toRdd)(
+ data.sparkSession.asInstanceOf[ClassicTypes.ClassicSparkSession])
+ ClassicDataset.ofRows(spark, node).toDF()
+ }
+
+ /**
+ * Builds a DataFrame backed by LogicalRDD/RDDScanExec directly from raw
InternalRows and an
+ * explicit output schema. Unlike SparkSession.createDataFrame(RDD[Row],
schema), this bypasses
+ * the RowEncoder (which inserts assertnotnull for non-nullable fields and
would reject nulls), so
+ * we can inject nulls into non-nullable columns - exactly the case
createNullabilityAwareIterator
+ * handles.
+ */
+ private def rddScanDF(output: Seq[Attribute], rows: Seq[InternalRow]):
DataFrame = {
+ val rdd = spark.sparkContext.parallelize(rows)
+ val node = LogicalRDD(output,
rdd)(spark.asInstanceOf[ClassicTypes.ClassicSparkSession])
+ ClassicDataset.ofRows(spark, node).toDF()
+ }
+
+ test("basic RDDScanExec is replaced by VeloxRDDScanTransformer") {
+ val data = spark.sql("SELECT l_orderkey, l_partkey FROM lineitem LIMIT 10")
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with string and numeric types") {
+ val data = spark.sql("""SELECT l_returnflag, l_linestatus, l_quantity,
l_extendedprice
+ |FROM lineitem LIMIT 20""".stripMargin)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with aggregation downstream") {
+ val query =
+ """SELECT l_returnflag, sum(l_quantity) AS sum_qty
+ |FROM lineitem
+ |WHERE l_shipdate <= date'1998-09-02'
+ |GROUP BY l_returnflag""".stripMargin
+ val data = spark.sql(query)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with empty RDD") {
+ val data = spark.sql("SELECT l_orderkey FROM lineitem WHERE 1 = 0")
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ assert(df.count() == 0)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan preserves data correctness with multiple re-reads") {
+ val data = spark.sql("SELECT l_orderkey, l_partkey FROM lineitem LIMIT 50")
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ // Read twice to verify idempotency
+ checkAnswer(df, expectedAnswer)
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with null values") {
+ val rdd = spark.sparkContext.parallelize(
+ Seq(
+ Row(1, "a", null),
+ Row(null, "b", 2.0),
+ Row(3, null, 3.0)
+ ))
+ val schema = StructType(
+ Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("name", StringType, nullable = true),
+ StructField("value", DoubleType, nullable = true)
+ ))
+ val data = spark.createDataFrame(rdd, schema)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with all supported primitive types") {
+ val rdd = spark.sparkContext.parallelize(
+ Seq(
+ Row(
+ true,
+ 1.toByte,
+ 2.toShort,
+ 3,
+ 4L,
+ 5.0f,
+ 6.0,
+ "hello",
+ java.sql.Date.valueOf("2024-01-01"),
+ java.sql.Timestamp.valueOf("2024-01-01 12:00:00"),
+ Array[Byte](1, 2, 3),
+ BigDecimal("123.45").underlying()
+ )
+ ))
+ val schema = StructType(
+ Seq(
+ StructField("bool", BooleanType),
+ StructField("byte", ByteType),
+ StructField("short", ShortType),
+ StructField("int", IntegerType),
+ StructField("long", LongType),
+ StructField("float", FloatType),
+ StructField("double", DoubleType),
+ StructField("string", StringType),
+ StructField("date", DateType),
+ StructField("timestamp", TimestampType),
+ StructField("binary", BinaryType),
+ StructField("decimal", DecimalType(10, 2))
+ ))
+ val data = spark.createDataFrame(rdd, schema)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with array type") {
+ val rdd = spark.sparkContext.parallelize(
+ Seq(
+ Row(Seq(1, 2, 3)),
+ Row(Seq(4, 5))
+ ))
+ val schema = StructType(Seq(StructField("arr", ArrayType(IntegerType))))
+ val data = spark.createDataFrame(rdd, schema)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan with map type falls back to row-based") {
+ val rdd = spark.sparkContext.parallelize(
+ Seq(
+ Row(Map("a" -> 1, "b" -> 2)),
+ Row(Map("c" -> 3))
+ ))
+ val schema = StructType(Seq(StructField("m", MapType(StringType,
IntegerType))))
+ val data = spark.createDataFrame(rdd, schema)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ // MapType is not supported in Arrow export, so falls back to row-based
processing
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.isEmpty, "MapType schema should fall back from
VeloxRDDScanTransformer")
+ }
+
+ test("RDDScan with an empty schema falls back to row-based") {
+ val plan = RDDScanExec(
+ Seq.empty,
+ spark.sparkContext.parallelize(Seq(InternalRow.empty)),
+ "OneRowRelation")
+ val transformer = RDDScanTransformer.getRDDScanTransform(plan)
+
+ assert(!transformer.doValidate().ok(), "Empty-schema RDDScan should not be
offloaded")
+ }
+
+ test("RDDScan with struct type") {
+ val rdd = spark.sparkContext.parallelize(
+ Seq(
+ Row(Row("hello", 1)),
+ Row(Row("world", 2))
+ ))
+ val innerSchema =
+ StructType(Seq(StructField("name", StringType), StructField("value",
IntegerType)))
+ val schema = StructType(Seq(StructField("s", innerSchema)))
+ val data = spark.createDataFrame(rdd, schema)
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan falls back for unsupported types") {
+ val data = spark.sql("SELECT INTERVAL '1' DAY AS di")
+ val expectedAnswer = data.collect()
+ val result = asRDDScanDF(data)
+
+ // Should still produce correct results via fallback to vanilla Spark
+ checkAnswer(result, expectedAnswer)
+ val cnt = collect(result.queryExecution.executedPlan) {
+ case _: VeloxRDDScanTransformer => true
+ }
+ assert(cnt.isEmpty, "Expected fallback - VeloxRDDScanTransformer should
NOT be in plan")
+ }
+
+ test("RDDScan handles BatchCarrierRow from checkpoint") {
+ val tempDir = Utils.createTempDir()
+ try {
+ spark.sparkContext.setCheckpointDir(tempDir.getAbsolutePath)
+ val df = spark.range(100).selectExpr("id", "id * 2 as value")
+ val checkpointed = df.localCheckpoint()
+ val result = asRDDScanDF(checkpointed)
+
+ checkAnswer(result, df.collect())
+ val cnt = collect(result.queryExecution.executedPlan) {
+ case _: VeloxRDDScanTransformer => true
+ }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ } finally {
+ Utils.deleteRecursively(tempDir)
+ }
+ }
+
+ test("falls back for schemas with interval types") {
+ val df = spark.sql("SELECT INTERVAL '1' YEAR as y")
+ val rddDf = asRDDScanDF(df)
+ checkAnswer(rddDf, df.collect())
+ // Should NOT use VeloxRDDScanTransformer (falls back due to interval type)
+ val veloxScans = collect(rddDf.queryExecution.executedPlan) {
+ case _: VeloxRDDScanTransformer => true
+ }
+ assert(veloxScans.isEmpty, "Interval type schema should fall back from
VeloxRDDScanTransformer")
+ }
+
+ test("RDDScan falls back when native kill-switch is disabled") {
+ withSQLConf(VeloxConfig.COLUMNAR_VELOX_RDD_SCAN_ENABLED.key -> "false") {
+ val data = spark.sql("SELECT l_orderkey, l_partkey FROM lineitem LIMIT
10")
+ val expectedAnswer = data.collect()
+ val df = asRDDScanDF(data)
+
+ checkAnswer(df, expectedAnswer)
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.isEmpty, "Expected fallback when native rddScan kill-switch
is disabled")
+ }
+ }
+
+ test("RDDScan nullability: null in non-nullable field yields Spark default
(SPARK-35912)") {
+ // Output attributes declare numeric fields non-nullable, but the backing
InternalRows
+ // carry nulls. This drives the createNullabilityAwareIterator branch
(entered only when
+ // the schema has a non-nullable field). Per SPARK-35912, a null read from
a non-nullable
+ // field via a typed getter unboxes to the primitive default (0), matching
WholeStageCodegen.
+ val output = Seq(
+ AttributeReference("i", IntegerType, nullable = false)(),
+ AttributeReference("l", LongType, nullable = false)(),
+ AttributeReference("d", DoubleType, nullable = false)()
+ )
+ val rows = Seq(InternalRow(1, 10L, 1.5d), InternalRow(null, null, null))
+ val df = rddScanDF(output, rows)
+
+ // The null row must surface Spark's type defaults (0 / 0L / 0.0), not
propagated nulls.
+ checkAnswer(df, Seq(Row(1, 10L, 1.5d), Row(0, 0L, 0.0d)))
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan nullability: mixed nullable and non-nullable fields") {
+ // Only the non-nullable field coerces null to a default; the nullable
field keeps null.
+ val output = Seq(
+ AttributeReference("nn", LongType, nullable = false)(),
+ AttributeReference("nl", LongType, nullable = true)())
+ val rows = Seq(InternalRow(5L, 7L), InternalRow(null, null))
+ val df = rddScanDF(output, rows)
+
+ checkAnswer(df, Seq(Row(5L, 7L), Row(0L, null)))
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("RDDScan nullability: non-nullable field nested under a nullable struct
coerces null") {
+ // Top-level field `a` is nullable, but its nested field `x` is
non-nullable. Previously the
+ // routing guard only inspected top-level nullability, so this schema
skipped the
+ // nullability-aware projection and a null nested `x` could
propagate/diverge from Spark.
+ // The recursive guard now routes it through the projection, coercing the
nested null to the
+ // type default (0L) per SPARK-35912, while a fully-null top-level struct
stays null.
+ val inner = StructType(Seq(StructField("x", LongType, nullable = false)))
+ val output = Seq(AttributeReference("a", inner, nullable = true)())
+ val rows = Seq(InternalRow(InternalRow(3L)),
InternalRow(InternalRow(null)), InternalRow(null))
+ val df = rddScanDF(output, rows)
+
+ checkAnswer(df, Seq(Row(Row(3L)), Row(Row(0L)), Row(null)))
+ val cnt = collect(df.queryExecution.executedPlan) { case _:
VeloxRDDScanTransformer => true }
+ assert(cnt.nonEmpty, "Expected VeloxRDDScanTransformer in plan")
+ }
+
+ test("isSupportRDDScanExec skips offload inside a streaming query
(foreachBatch)") {
+ // A Structured Streaming micro-batch (and its foreachBatch callback) runs
on the
+ // StreamExecution driver thread, which sets the `sql.streaming.queryId`
local property.
+ // The per-batch source is a materialized snapshot, so it slips past the
plan-level
+ // streaming fallback; offloading it into a streaming/state-store pipeline
can deadlock.
+ val output = Seq(AttributeReference("id", IntegerType)())
+ val rdd = spark.sparkContext.parallelize(Seq.empty[InternalRow])
+ val plan = RDDScanExec(output, rdd, "ExistingRDD")
+ val api = BackendsApiManager.getSparkPlanExecApiInstance
+
+ // Sanity: outside a streaming query this plan is offloadable.
+ assert(api.isSupportRDDScanExec(plan), "Non-streaming RDDScan should be
offloadable")
+
+ val sc = spark.sparkContext
+ val prev = sc.getLocalProperty("sql.streaming.queryId")
+ try {
+ sc.setLocalProperty("sql.streaming.queryId", "test-streaming-query-id")
+ assert(
+ !api.isSupportRDDScanExec(plan),
+ "RDDScan inside a streaming query must not be offloaded")
+ } finally {
+ sc.setLocalProperty("sql.streaming.queryId", prev)
+ }
+ }
+}
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 3eca94a7f9..2c14b37c45 100644
--- a/docs/velox-configuration.md
+++ b/docs/velox-configuration.md
@@ -69,6 +69,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames
| 🔄 Dynamic | true | Maps table field names to file field
names using names, not indices for Parquet files.
[...]
| spark.gluten.sql.columnar.backend.velox.prefetchRowGroups
| âš“ Static | 1 | Set the prefetch row groups for velox
file scan
[...]
| spark.gluten.sql.columnar.backend.velox.queryTraceEnabled
| 🔄 Dynamic | false | Enable query tracing flag.
[...]
+| spark.gluten.sql.columnar.backend.velox.rddScan.enabled
| 🔄 Dynamic | true | When true, offload RDDScanExec to
Velox by converting the RDD[InternalRow] into columnar batches through the
native row-to-columnar path. Schemas that are not supported by the Arrow export
path (e.g. map or interval types) fall back to vanilla Spark.
[...]
| spark.gluten.sql.columnar.backend.velox.reclaimMaxWaitMs
| 🔄 Dynamic | 3600000ms | The max time in ms to wait for memory
reclaim.
[...]
| spark.gluten.sql.columnar.backend.velox.resizeBatches.copyRanges.enabled
| 🔄 Dynamic | true | Enables a VeloxResizeBatchesExec fast
path that combines eligible batches using Velox vector copyRanges instead of
generic RowVector append. When possible, it collects the small input batches
for one VeloxResizeBatchesExec output, allocates the output RowVector once, and
bulk-copies child vector ranges. This is most useful for shuffle-read outputs
where plain hash shuff [...]
| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput
| 🔄 Dynamic | true | If true, combine small columnar
batches together before sending to shuffle. The default minimum output batch
size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize
[...]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
