You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by we...@apache.org on 2026/09/11 05:52:02 UTC
(gluten) branch main updated: [VL] Add RDDScanExec support to Velox backend (#12982)
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 <mi...@gmail.com>
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 <22...@users.noreply.github.com>
* [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 <22...@users.noreply.github.com>
---------
Co-authored-by: Copilot <22...@users.noreply.github.com>
---
.../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: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org