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 ae61225585 [CORE][VL] Add LocalTableScanExec offload support to Velox
backend (#12631)
ae61225585 is described below
commit ae612255859e152ad23db29085097e57411d37b5
Author: Minni Mittal <[email protected]>
AuthorDate: Wed Aug 19 20:20:54 2026 +0530
[CORE][VL] Add LocalTableScanExec offload support to Velox backend (#12631)
* [CORE][VL] Add LocalTableScanExec offload support to Velox backend
Offload LocalTableScanExec (a driver-side local collection) to native
columnar execution on the Velox backend by converting its rows into
columnar batches via the existing RowToVeloxColumnar JNI path.
- Add gluten-substrait LocalTableScanTransformer base and SparkPlanExecApi
hooks (isSupportLocalTableScanExec / getLocalTableScanTransform), wired
into OffloadOthers.
- Add VeloxLocalTableScanTransformer with schema/Arrow-compatibility
validation (falls back for Map/Interval and other unsupported types).
- Gate offload behind spark.gluten.sql.columnar.localTableScan (default
true,
consistent with other columnar operator toggles).
- Skip offload for deserialized plans whose @transient rows became null
(avoids an NPE when an AQE sub-plan is shipped across an RPC boundary),
and for streaming sources via a new SparkShims.getLocalTableScanStream
accessor (None on Spark 3.x, plan.stream on Spark 4.0+).
- Add unit/integration tests and document the new config.
Generated-by: Copilot claude-opus-4.8
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 677ac585-63d4-4657-9fef-89195c8751e8
* [VL] Address review comments on LocalTableScanExec offload
- Guard against null @transient rows in VeloxLocalTableScanTransformer
.doExecuteColumnar with a clear IllegalStateException instead of a bare
NPE.
- Lower per-node validation-success log from INFO to DEBUG to avoid log
spam.
- Add a version-agnostic test asserting a batch LocalTableScanExec is not
classified as a streaming source (not skipped by the streaming guard).
- Add Spark 4.0/4.1 shim suites covering the streaming-source detection path
(getLocalTableScanStream returns the stream), which cannot live in the
shared
cross-version suite because the `stream` ctor param is Spark 4.x-only.
Generated-by: GitHub Copilot CLI claude-opus-4.8
Co-authored-by: Copilot <[email protected]>
Copilot-Session: a4ac0059-8498-4138-bdeb-0c66249bde12
* [VL] Fix Spark 4.0 test compile in VeloxLocalTableScanSuite
LocalTableScanExec gained a required `stream` parameter on Spark 4.0, so
the direct 2-arg constructor calls in the suite failed to compile under
the Spark 4.0 profile ("Unspecified value parameter stream").
Build the plan through the physical planner via a version-agnostic
`newBatchLocalTableScan()` helper instead of calling the constructor,
keeping the suite compilable on all supported Spark versions.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: a4ac0059-8498-4138-bdeb-0c66249bde12
* [VL] Fix config-doc padding for localTableScan row
The generated docs/Configuration.md row for
spark.gluten.sql.columnar.localTableScan carried two extra trailing
padding spaces, causing the "Check gluten configs" test to fail
(generated output vs committed file mismatch on this line). Trim the
row to the correct column width so it matches gen-all-config-docs.sh
output.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: a4ac0059-8498-4138-bdeb-0c66249bde12
* [VL] Default localTableScan offload to opt-in to fix CI failures
The LocalTableScanExec offload defaulted to enabled, which changed plan
shapes across the whole test suite and caused deterministic CI failures:
- MiscOperatorSuite "RowToVeloxColumnar preferredBatchBytes": with offload
on, the local scan produces columnar batches itself, so no
RowToVeloxColumnarExec node is inserted and the assertion on its count
fails. Pin the offload off in that test so it keeps exercising the
RowToVeloxColumnarExec batching path it targets.
- VeloxParquetWriteForHiveSuite hive VALUES write (Spark 3.3): when the
scan is offloaded but the parent projection falls back to vanilla row
execution (AnsiCast is not mappable to Substrait on 3.3), the inserted
VeloxColumnarToRowExec (not CodegenSupport) sits under FileFormatWriter
whole-stage codegen with no InputAdapter, throwing a ClassCastException.
Make the feature opt-in (default false) so existing behavior is preserved,
and document the write-path codegen limitation. The feature stays fully
covered by VeloxLocalTableScanSuite, which enables the config explicitly.
The codegen-safe write path is left as a follow-up before flipping the
default back on.
Generated-by: Copilot claude-opus-4.8
---------
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Minni Mittal <[email protected]>
Copilot-Session: 677ac585-63d4-4657-9fef-89195c8751e8
Copilot-Session: a4ac0059-8498-4138-bdeb-0c66249bde12
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 25 ++
.../execution/VeloxLocalTableScanTransformer.scala | 154 ++++++++++
.../gluten/execution/MiscOperatorSuite.scala | 6 +-
.../execution/VeloxLocalTableScanSuite.scala | 325 +++++++++++++++++++++
docs/Configuration.md | 1 +
.../gluten/backendsapi/SparkPlanExecApi.scala | 7 +
.../org/apache/gluten/config/GlutenConfig.scala | 15 +
.../columnar/offload/OffloadSingleNodeRules.scala | 3 +
.../sql/execution/LocalTableScanTransformer.scala | 50 ++++
.../org/apache/gluten/sql/shims/SparkShims.scala | 8 +
.../gluten/sql/shims/spark40/Spark40Shims.scala | 3 +
.../spark40/Spark40LocalTableScanStreamSuite.scala | 59 ++++
.../gluten/sql/shims/spark41/Spark41Shims.scala | 3 +
.../spark41/Spark41LocalTableScanStreamSuite.scala | 59 ++++
14 files changed, 717 insertions(+), 1 deletion(-)
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 d588695cd4..d45b7a3f20 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
@@ -1423,6 +1423,31 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi
with Logging {
VeloxColumnarToCarrierRowExec.enforce(plan)
}
+ override def isSupportLocalTableScanExec(plan: LocalTableScanExec): Boolean
= {
+ // `rows` is @transient, so it becomes null after Java serialization (e.g.
an AQE sub-plan
+ // shipped across an RPC boundary). A null rows payload signals a
deserialized plan that can
+ // no longer be executed natively, so offload must be skipped to avoid a
later NPE.
+ if (plan.rows == null) {
+ logDebug("LocalTableScan offload skipped: deserialized plan with null
transient rows")
+ return false
+ }
+ // A streaming source (Spark 4.0+ only) must keep vanilla execution.
+ if (SparkShimLoader.getSparkShims.getLocalTableScanStream(plan).isDefined)
{
+ logDebug("LocalTableScan offload skipped: streaming source detected")
+ return false
+ }
+ if (!GlutenConfig.get.enableColumnarLocalTableScan) {
+ logDebug(
+ "LocalTableScan offload skipped: " +
+ s"${GlutenConfig.COLUMNAR_LOCAL_TABLE_SCAN_ENABLED.key}=false")
+ return false
+ }
+ true
+ }
+
+ override def getLocalTableScanTransform(plan: LocalTableScanExec):
LocalTableScanTransformer =
+ VeloxLocalTableScanTransformer.replace(plan)
+
override def genTimestampAddTransformer(
substraitExprName: String,
left: ExpressionTransformer,
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala
new file mode 100644
index 0000000000..9110cf2188
--- /dev/null
+++
b/backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala
@@ -0,0 +1,154 @@
+/*
+ * 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, SortOrder,
UnsafeProjection}
+import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.execution.{LocalTableScanTransformer, SparkPlan}
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * Velox-backend implementation of LocalTableScanTransformer.
+ *
+ * Converts a driver-side local collection (Seq[InternalRow]) into columnar
batches using Velox's
+ * native row-to-columnar conversion (same JNI path as RowToVeloxColumnarExec).
+ */
+case class VeloxLocalTableScanTransformer(
+ outputAttributes: Seq[Attribute],
+ @transient rows: Seq[InternalRow],
+ // Row-to-columnar conversion preserves data distribution, so we carry
through
+ // the original partitioning, consistent with RowToVeloxColumnarExec's
behavior.
+ override val outputPartitioning: Partitioning,
+ override val outputOrdering: Seq[SortOrder]
+) extends LocalTableScanTransformer(outputAttributes, outputPartitioning,
outputOrdering)
+ with Logging {
+
+ @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 = {
+ 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)
+ }
+ }
+
+ logDebug(
+ s"local_table_scan native validation succeeded: " +
+ s"schema=${schema.fields.map(_.dataType.simpleString).mkString(",")},
" +
+ s"appId=${sparkContext.applicationId}")
+
+ ValidationResult.succeeded
+ }
+
+ /**
+ * Validates that data types are compatible with the Arrow ABI export path
used by
+ * RowToVeloxColumnarExec.toColumnarBatchIterator:
+ * - 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 LocalTableScan 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 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
+
+ // `rows` is @transient and becomes null if this transformer is
deserialized (e.g. an AQE
+ // sub-plan shipped across an RPC boundary). Offload is guarded against
null rows in
+ // VeloxSparkPlanExecApi.isSupportLocalTableScanExec, so reaching
execution with null rows
+ // indicates an inconsistent plan; fail fast with a clear message rather
than a bare NPE.
+ if (rows == null) {
+ throw new IllegalStateException(
+ "VeloxLocalTableScanTransformer.rows is null (deserialized plan cannot
be executed " +
+ "natively); this plan should not have been offloaded")
+ }
+
+ if (rows.isEmpty) {
+ sparkContext.emptyRDD[ColumnarBatch]
+ } else {
+ // Materialize rows as UnsafeRow on the driver, then parallelize
+ val proj = UnsafeProjection.create(outputAttributes, outputAttributes)
+ val unsafeRows = rows.map(r => proj(r).copy()).toArray
+ val numSlices = math.min(
+ unsafeRows.length,
+ SQLConf.get
+ .getConf(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM)
+ .getOrElse(sparkContext.defaultParallelism))
+ val rowRdd = sparkContext.parallelize(unsafeRows.toSeq, numSlices)
+
+ rowRdd.mapPartitions {
+ iter =>
+ RowToVeloxColumnarExec.toColumnarBatchIterator(
+ iter,
+ localSchema,
+ numInputRows,
+ numOutputBatches,
+ convertTime,
+ batchSize,
+ batchBytes)
+ }
+ }
+ }
+
+ override protected def withNewChildrenInternal(newChildren:
IndexedSeq[SparkPlan]): SparkPlan = {
+ assert(newChildren.isEmpty, "VeloxLocalTableScanTransformer is a leaf
node")
+ copy(outputAttributes, rows, outputPartitioning, outputOrdering)
+ }
+}
+
+object VeloxLocalTableScanTransformer {
+
+ def replace(plan: org.apache.spark.sql.execution.LocalTableScanExec):
LocalTableScanTransformer =
+ VeloxLocalTableScanTransformer(
+ plan.output,
+ plan.rows,
+ plan.outputPartitioning,
+ plan.outputOrdering)
+}
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
index 98c82217ad..6ddeaad2ef 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
@@ -2287,7 +2287,11 @@ class MiscOperatorSuite extends
VeloxWholeStageTransformerSuite with AdaptiveSpa
Seq("1", "80", "100000000").foreach(
preferredBatchBytes => {
withSQLConf(
- VeloxConfig.COLUMNAR_VELOX_PREFERRED_BATCH_BYTES.key ->
preferredBatchBytes
+ VeloxConfig.COLUMNAR_VELOX_PREFERRED_BATCH_BYTES.key ->
preferredBatchBytes,
+ // This test targets the RowToVeloxColumnarExec batching path, so
the LocalTableScan
+ // offload must stay disabled here; otherwise the local scan
produces columnar batches
+ // itself and no RowToVeloxColumnarExec node is inserted.
+ GlutenConfig.COLUMNAR_LOCAL_TABLE_SCAN_ENABLED.key -> "false"
) {
val df = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9,
10).toDF("Col").select($"Col".plus(1))
assert(df.collect().length == 10)
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxLocalTableScanSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxLocalTableScanSuite.scala
new file mode 100644
index 0000000000..cdb63cc782
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxLocalTableScanSuite.scala
@@ -0,0 +1,325 @@
+/*
+ * 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.BackendsApiManager
+import org.apache.gluten.sql.shims.SparkShimLoader
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.LocalTableScanExec
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.types._
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream,
ObjectInputStream, ObjectOutputStream}
+import java.util.{Arrays => JArrays}
+
+class VeloxLocalTableScanSuite
+ 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.localTableScan", "true")
+ }
+
+ private def assertHasVeloxLocalTableScan(df: DataFrame): Unit = {
+ val found = collect(df.queryExecution.executedPlan) {
+ case _: VeloxLocalTableScanTransformer => true
+ }
+ assert(found.nonEmpty, "Expected VeloxLocalTableScanTransformer in plan")
+ }
+
+ private def createDF(rows: Seq[Row], schema: StructType): DataFrame = {
+ spark.createDataFrame(JArrays.asList(rows: _*), schema)
+ }
+
+ // Builds a genuine batch [[LocalTableScanExec]] via the physical planner.
This avoids calling the
+ // constructor directly, whose arity differs across Spark versions (Spark
4.0+ adds a required
+ // `stream` parameter), keeping this suite compilable on every supported
Spark version.
+ private def newBatchLocalTableScan(): LocalTableScanExec = {
+ val schema = StructType(Seq(StructField("id", IntegerType)))
+ val df = createDF(Seq(Row(1)), schema)
+ df.queryExecution.sparkPlan
+ .collectFirst { case l: LocalTableScanExec => l }
+ .getOrElse(fail("Expected a LocalTableScanExec in the spark plan"))
+ }
+
+ test("basic LocalTableScanExec with int and string columns") {
+ val schema = StructType(Seq(StructField("id", IntegerType),
StructField("name", StringType)))
+ val rows = Seq(Row(1, "a"), Row(2, "b"), Row(3, "c"))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with numeric types") {
+ val schema = StructType(
+ Seq(
+ StructField("lng", LongType),
+ StructField("dbl", DoubleType),
+ StructField("flt", FloatType),
+ StructField("shrt", ShortType),
+ StructField("byt", ByteType)))
+ val rows = Seq(Row(1L, 1.5, 2.5f, 100.toShort, 42.toByte))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with boolean and null types") {
+ val schema = StructType(
+ Seq(StructField("flag", BooleanType), StructField("value", IntegerType,
nullable = true)))
+ val rows = Seq(Row(true, 1), Row(false, null))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with empty collection") {
+ val schema = StructType(Seq(StructField("id", IntegerType),
StructField("name", StringType)))
+ val df = createDF(Seq.empty, schema)
+ checkAnswer(df, Seq.empty[Row])
+ }
+
+ test("LocalTableScan with aggregation downstream") {
+ val schema = StructType(Seq(StructField("key", StringType),
StructField("value", IntegerType)))
+ val rows = Seq(Row("a", 10), Row("b", 20), Row("a", 30))
+ val df = createDF(rows, schema)
+ val result = df.groupBy("key").sum("value")
+ checkAnswer(result, Seq(Row("a", 40), Row("b", 20)))
+ assertHasVeloxLocalTableScan(result)
+ }
+
+ test("LocalTableScan with filter downstream") {
+ val schema = StructType(Seq(StructField("x", IntegerType)))
+ val rows = Seq(Row(1), Row(2), Row(3), Row(4), Row(5))
+ val df = createDF(rows, schema).filter("x > 3")
+ checkAnswer(df, Seq(Row(4), Row(5)))
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with join") {
+ val leftSchema =
+ StructType(Seq(StructField("id", IntegerType), StructField("name",
StringType)))
+ val rightSchema =
+ StructType(Seq(StructField("id", IntegerType), StructField("score",
IntegerType)))
+ val left = createDF(Seq(Row(1, "a"), Row(2, "b")), leftSchema)
+ val right = createDF(Seq(Row(1, 100), Row(2, 200)), rightSchema)
+ val result = left.join(right, "id")
+ checkAnswer(result, Seq(Row(1, "a", 100), Row(2, "b", 200)))
+ assertHasVeloxLocalTableScan(result)
+ }
+
+ test("LocalTableScan with all supported primitive types in one schema") {
+ 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 rows = 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),
+ new java.math.BigDecimal("123.45")
+ ),
+ Row(
+ false,
+ (-1).toByte,
+ (-2).toShort,
+ -3,
+ -4L,
+ -5.0f,
+ -6.0,
+ "world",
+ java.sql.Date.valueOf("1970-01-01"),
+ java.sql.Timestamp.valueOf("1970-01-01 00:00:00"),
+ Array[Byte](),
+ new java.math.BigDecimal("-123.45")
+ )
+ )
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with array type") {
+ val schema = StructType(
+ Seq(
+ StructField("id", IntegerType),
+ StructField("arr", ArrayType(IntegerType, containsNull = true),
nullable = true)))
+ val rows =
+ Seq(Row(1, Seq(10, 20, 30)), Row(2, Seq.empty[Int]), Row(3, null),
Row(4, Seq(-1, 0, 1)))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan with map type falls back") {
+ val schema = StructType(
+ Seq(
+ StructField("id", IntegerType),
+ StructField(
+ "m",
+ MapType(StringType, IntegerType, valueContainsNull = true),
+ nullable = true)))
+ val rows = Seq(Row(1, Map("a" -> 1, "b" -> 2)), Row(2, Map.empty[String,
Int]), Row(3, null))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ // MapType is not supported in Arrow export path - should fall back
+ val cnt = collect(df.queryExecution.executedPlan) {
+ case _: VeloxLocalTableScanTransformer => true
+ }
+ assert(cnt.isEmpty, "Expected fallback - MapType not supported in Arrow
export")
+ }
+
+ test("LocalTableScan with nested struct type") {
+ val innerSchema = StructType(
+ Seq(
+ StructField("name", StringType, nullable = true),
+ StructField("age", IntegerType, nullable = true)))
+ val schema = StructType(
+ Seq(StructField("id", IntegerType), StructField("person", innerSchema,
nullable = true)))
+ val rows = Seq(Row(1, Row("alice", 30)), Row(2, Row("bob", null)), Row(3,
null))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan falls back for unsupported types") {
+ val schema = StructType(
+ Seq(StructField("id", IntegerType), StructField("duration",
DayTimeIntervalType())))
+ val rows = Seq(Row(1, java.time.Duration.ofHours(3)), Row(2,
java.time.Duration.ofDays(1)))
+ val df = createDF(rows, schema)
+ // Should still produce correct results via fallback to vanilla Spark
+ checkAnswer(df, rows)
+ val cnt = collect(df.queryExecution.executedPlan) {
+ case _: VeloxLocalTableScanTransformer => true
+ }
+ assert(cnt.isEmpty, "Expected fallback - VeloxLocalTableScanTransformer
should NOT be in plan")
+ }
+
+ test("LocalTableScan idempotent re-reads") {
+ val schema = StructType(Seq(StructField("id", IntegerType),
StructField("name", StringType)))
+ val rows = Seq(Row(1, "a"), Row(2, "b"), Row(3, "c"), Row(4, "d"), Row(5,
"e"))
+ val df = createDF(rows, schema)
+ // Read twice to verify idempotency
+ checkAnswer(df, rows)
+ checkAnswer(df, rows)
+ assertHasVeloxLocalTableScan(df)
+ }
+
+ test("LocalTableScan falls back when localTableScan offload is disabled") {
+ withSQLConf("spark.gluten.sql.columnar.localTableScan" -> "false") {
+ val schema =
+ StructType(Seq(StructField("id", IntegerType), StructField("name",
StringType)))
+ val rows = Seq(Row(1, "a"), Row(2, "b"))
+ val df = createDF(rows, schema)
+ checkAnswer(df, rows)
+ val cnt = collect(df.queryExecution.executedPlan) {
+ case _: VeloxLocalTableScanTransformer => true
+ }
+ assert(cnt.isEmpty, "Expected fallback when localTableScan offload is
disabled")
+ }
+ }
+
+ test("isSupportLocalTableScanExec skips deserialized plan with null
transient rows") {
+ // Simulates a plan that was serialized and shipped across an RPC boundary
(e.g. an AQE
+ // sub-plan), where the @transient rows field becomes null after Java
serialization.
+ // Offload must be skipped to avoid a downstream NPE.
+ val plan = newBatchLocalTableScan()
+
+ // Serialize and deserialize to null out the @transient rows field.
+ val baos = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(baos)
+ oos.writeObject(plan)
+ oos.close()
+ val bais = new ByteArrayInputStream(baos.toByteArray)
+ val ois = new ObjectInputStream(bais)
+ val deserialized = ois.readObject().asInstanceOf[LocalTableScanExec]
+
+ assert(deserialized.rows == null, "Expected null rows after
deserialization")
+
+ // Should NOT throw NPE - this is the bug the v2 change fixes.
+ val api = BackendsApiManager.getSparkPlanExecApiInstance
+ val result = api.isSupportLocalTableScanExec(deserialized)
+ assert(!result, "Deserialized plan with null transient rows should not be
offloaded")
+ }
+
+ test("getLocalTableScanStream returns None for a batch LocalTableScanExec
(no streaming skip)") {
+ // The streaming-source skip in isSupportLocalTableScanExec only triggers
on Spark 4.0+, where
+ // LocalTableScanExec may carry a streaming SparkDataStream. This
version-agnostic test asserts
+ // the shim classifies an ordinary batch plan as non-streaming (None) on
every supported Spark
+ // version, so the streaming guard does not falsely skip batch offload.
The true streaming path
+ // (getLocalTableScanStream(plan).isDefined) is exercised by Spark 4.x
profiles only.
+ val plan = newBatchLocalTableScan()
+
+ val stream = SparkShimLoader.getSparkShims.getLocalTableScanStream(plan)
+ assert(
+ stream.isEmpty,
+ "A batch LocalTableScanExec must not be classified as a streaming
source")
+
+ withSQLConf("spark.gluten.sql.columnar.localTableScan" -> "true") {
+ val api = BackendsApiManager.getSparkPlanExecApiInstance
+ assert(
+ api.isSupportLocalTableScanExec(plan),
+ "Batch plan must not be skipped by the streaming-source guard")
+ }
+ }
+
+ test("isSupportLocalTableScanExec returns true for normal plan") {
+ withSQLConf("spark.gluten.sql.columnar.localTableScan" -> "true") {
+ val plan = newBatchLocalTableScan()
+
+ val api = BackendsApiManager.getSparkPlanExecApiInstance
+ val result = api.isSupportLocalTableScanExec(plan)
+ assert(result, "Normal plan should be supported for offload")
+ }
+ }
+
+ test("isSupportLocalTableScanExec returns false when localTableScan offload
is disabled") {
+ withSQLConf("spark.gluten.sql.columnar.localTableScan" -> "false") {
+ val plan = newBatchLocalTableScan()
+
+ val api = BackendsApiManager.getSparkPlanExecApiInstance
+ assert(!api.isSupportLocalTableScanExec(plan))
+ }
+ }
+}
diff --git a/docs/Configuration.md b/docs/Configuration.md
index 0299edce1e..3926053a17 100644
--- a/docs/Configuration.md
+++ b/docs/Configuration.md
@@ -72,6 +72,7 @@ nav_order: 15
| spark.gluten.sql.columnar.libname | 🔄
Dynamic | gluten | The gluten library name.
[...]
| spark.gluten.sql.columnar.libpath | 🔄
Dynamic || The gluten library path.
[...]
| spark.gluten.sql.columnar.limit | 🔄
Dynamic | true |
+| spark.gluten.sql.columnar.localTableScan | 🔄
Dynamic | false | Enable or disable native columnar execution of
LocalTableScanExec. When true, Gluten attempts to replace LocalTableScanExec (a
driver-side local collection) with a backend transformer that converts the rows
into columnar batches natively.
[...]
| spark.gluten.sql.columnar.maxBatchSize | 🔄
Dynamic | 4096 |
| spark.gluten.sql.columnar.overwriteByExpression | 🔄
Dynamic | true | Enable or disable columnar v2 command
overwrite by expression.
[...]
| spark.gluten.sql.columnar.overwritePartitionsDynamic | 🔄
Dynamic | true | Enable or disable columnar v2 command
overwrite partitions dynamic.
[...]
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
index f5dc9f3f21..4ca08a5ad6 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
@@ -364,6 +364,13 @@ trait SparkPlanExecApi {
throw new GlutenNotSupportException("ArrayInsert is not supported")
}
+ /** Whether the backend supports offloading the given [[LocalTableScanExec]]
to native. */
+ def isSupportLocalTableScanExec(plan: LocalTableScanExec): Boolean = false
+
+ /** Returns the backend transformer that replaces the given
[[LocalTableScanExec]]. */
+ def getLocalTableScanTransform(plan: LocalTableScanExec):
LocalTableScanTransformer =
+ throw new GlutenNotSupportException("LocalTableScanExec is not supported")
+
// For date_add(cast('2001-01-01' as Date), interval 1 day), backends may
handle it in different
// ways
def genDateAddTransformer(
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index 5a04647d25..21a60b57bf 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
@@ -95,6 +95,8 @@ class GlutenConfig(conf: SQLConf) extends
GlutenCoreConfig(conf) {
def enableColumnarWindowGroupLimit: Boolean =
getConf(COLUMNAR_WINDOW_GROUP_LIMIT_ENABLED)
+ def enableColumnarLocalTableScan: Boolean =
getConf(COLUMNAR_LOCAL_TABLE_SCAN_ENABLED)
+
def enableAppendData: Boolean = getConf(COLUMNAR_APPEND_DATA_ENABLED)
def enableReplaceData: Boolean = getConf(COLUMNAR_REPLACE_DATA_ENABLED)
@@ -926,6 +928,19 @@ object GlutenConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)
+ val COLUMNAR_LOCAL_TABLE_SCAN_ENABLED =
+ // NOTE: Disabled by default. When an offloaded local scan feeds an
operator that falls back
+ // to vanilla row execution under the write path, the inserted
columnar-to-row transition is
+ // not yet codegen-safe (VeloxColumnarToRowExec is not CodegenSupport),
which can fail
+ // FileFormatWriter codegen. Flip the default to true once that path is
handled.
+ buildConf("spark.gluten.sql.columnar.localTableScan")
+ .doc(
+ "Enable or disable native columnar execution of LocalTableScanExec.
When true, Gluten " +
+ "attempts to replace LocalTableScanExec (a driver-side local
collection) with a " +
+ "backend transformer that converts the rows into columnar batches
natively.")
+ .booleanConf
+ .createWithDefault(false)
+
val COLUMNAR_SORT_ENABLED =
buildConf("spark.gluten.sql.columnar.sort")
.doc("Enable or disable columnar sort.")
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
index 2cafcae1b1..823f72642d 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala
@@ -314,6 +314,9 @@ object OffloadOthers {
child)
case plan: RDDScanExec if
RDDScanTransformer.isSupportRDDScanExec(plan) =>
RDDScanTransformer.getRDDScanTransform(plan)
+ case plan: LocalTableScanExec
+ if LocalTableScanTransformer.isSupportLocalTableScanExec(plan) =>
+ LocalTableScanTransformer.getLocalTableScanTransform(plan)
case p if !p.isInstanceOf[GlutenPlan] =>
logDebug(s"Transformation for ${p.getClass} is currently not
supported.")
p
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/LocalTableScanTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/LocalTableScanTransformer.scala
new file mode 100644
index 0000000000..d76a6092ac
--- /dev/null
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/execution/LocalTableScanTransformer.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.execution.ValidatablePlan
+import org.apache.gluten.extension.columnar.transition.Convention
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.catalyst.plans.physical.{Partitioning,
UnknownPartitioning}
+
+abstract class LocalTableScanTransformer(
+ outputAttributes: Seq[Attribute],
+ override val outputPartitioning: Partitioning = UnknownPartitioning(0),
+ override val outputOrdering: Seq[SortOrder] = Nil
+) extends ValidatablePlan {
+
+ override def rowType(): Convention.RowType = Convention.RowType.None
+ override def batchType(): Convention.BatchType =
BackendsApiManager.getSettings.primaryBatchType
+ override def output: Seq[Attribute] = outputAttributes
+
+ override protected def doExecute()
+ : org.apache.spark.rdd.RDD[org.apache.spark.sql.catalyst.InternalRow] = {
+ throw new UnsupportedOperationException(s"This operator doesn't support
doExecute().")
+ }
+
+ override def children: Seq[SparkPlan] = Seq.empty
+}
+
+object LocalTableScanTransformer {
+ def isSupportLocalTableScanExec(plan: LocalTableScanExec): Boolean =
+
BackendsApiManager.getSparkPlanExecApiInstance.isSupportLocalTableScanExec(plan)
+
+ def getLocalTableScanTransform(plan: LocalTableScanExec):
LocalTableScanTransformer =
+
BackendsApiManager.getSparkPlanExecApiInstance.getLocalTableScanTransform(plan)
+}
diff --git
a/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
b/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
index 068ccdda6f..4246feb0c9 100644
--- a/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
+++ b/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
@@ -250,6 +250,14 @@ trait SparkShims {
/** Shim method for usages from GlutenExplainUtils.scala. */
def unsetOperatorId(plan: QueryPlan[_]): Unit
+ /**
+ * Returns the streaming source associated with a [[LocalTableScanExec]], if
any. The `stream`
+ * field only exists on Spark 4.0+ (where `LocalTableScanExec` mixes in
+ * `StreamSourceAwareSparkPlan`); on Spark 3.x local relations have no
streaming concept, so the
+ * default implementation returns None.
+ */
+ def getLocalTableScanStream(plan: LocalTableScanExec):
Option[SparkDataStream] = None
+
def isParquetFileEncrypted(footer: ParquetMetadata): Boolean
def shouldFallbackForParquetVariantAnnotation(footer: ParquetMetadata):
Boolean = false
diff --git
a/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
b/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
index 2f4dafd188..c3bb77fab2 100644
---
a/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
+++
b/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
@@ -67,6 +67,9 @@ import scala.reflect.ClassTag
class Spark40Shims extends SparkShims {
+ override def getLocalTableScanStream(plan: LocalTableScanExec):
Option[SparkDataStream] =
+ plan.stream
+
override def scalarExpressionMappings: Seq[Sig] = {
Seq(
Sig[Empty2Null](ExpressionNames.EMPTY2NULL),
diff --git
a/shims/spark40/src/test/scala/org/apache/gluten/sql/shims/spark40/Spark40LocalTableScanStreamSuite.scala
b/shims/spark40/src/test/scala/org/apache/gluten/sql/shims/spark40/Spark40LocalTableScanStreamSuite.scala
new file mode 100644
index 0000000000..937f8c6eb8
--- /dev/null
+++
b/shims/spark40/src/test/scala/org/apache/gluten/sql/shims/spark40/Spark40LocalTableScanStreamSuite.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.sql.shims.spark40
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream}
+import org.apache.spark.sql.execution.LocalTableScanExec
+import org.apache.spark.sql.types.IntegerType
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Spark 4.0+ only: LocalTableScanExec carries an optional streaming source
(`stream`). This suite
+ * verifies the Spark40Shims accessor that
VeloxSparkPlanExecApi.isSupportLocalTableScanExec uses to
+ * skip offload for streaming sources. The Velox backend test lives here (not
in the shared
+ * cross-version suite) because the `stream` constructor parameter does not
exist on Spark 3.x.
+ */
+class Spark40LocalTableScanStreamSuite extends AnyFunSuite {
+
+ private val shims = new Spark40Shims
+
+ private def output = Seq(AttributeReference("id", IntegerType)())
+
+ private val stubStream: SparkDataStream = new SparkDataStream {
+ override def initialOffset(): Offset = null
+ override def deserializeOffset(json: String): Offset = null
+ override def commit(end: Offset): Unit = {}
+ override def stop(): Unit = {}
+ }
+
+ test("getLocalTableScanStream returns the stream for a streaming
LocalTableScanExec") {
+ val plan = LocalTableScanExec(output, Seq.empty[InternalRow],
Some(stubStream))
+ val stream = shims.getLocalTableScanStream(plan)
+ assert(stream.isDefined, "Streaming LocalTableScanExec must be detected as
a streaming source")
+ assert(stream.get eq stubStream)
+ }
+
+ test("getLocalTableScanStream returns None for a batch LocalTableScanExec") {
+ val plan = LocalTableScanExec(output, Seq.empty[InternalRow], None)
+ assert(
+ shims.getLocalTableScanStream(plan).isEmpty,
+ "Batch LocalTableScanExec must not be detected as a streaming source")
+ }
+}
diff --git
a/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
b/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
index 3934f6fe84..8d2a1d6e5c 100644
---
a/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
+++
b/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
@@ -66,6 +66,9 @@ import scala.reflect.ClassTag
class Spark41Shims extends SparkShims {
+ override def getLocalTableScanStream(plan: LocalTableScanExec):
Option[SparkDataStream] =
+ plan.stream
+
override def scalarExpressionMappings: Seq[Sig] = {
Seq(
Sig[Empty2Null](ExpressionNames.EMPTY2NULL),
diff --git
a/shims/spark41/src/test/scala/org/apache/gluten/sql/shims/spark41/Spark41LocalTableScanStreamSuite.scala
b/shims/spark41/src/test/scala/org/apache/gluten/sql/shims/spark41/Spark41LocalTableScanStreamSuite.scala
new file mode 100644
index 0000000000..5b7dd440ef
--- /dev/null
+++
b/shims/spark41/src/test/scala/org/apache/gluten/sql/shims/spark41/Spark41LocalTableScanStreamSuite.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.sql.shims.spark41
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.connector.read.streaming.{Offset, SparkDataStream}
+import org.apache.spark.sql.execution.LocalTableScanExec
+import org.apache.spark.sql.types.IntegerType
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Spark 4.1+ only: LocalTableScanExec carries an optional streaming source
(`stream`). This suite
+ * verifies the Spark41Shims accessor that
VeloxSparkPlanExecApi.isSupportLocalTableScanExec uses to
+ * skip offload for streaming sources. The Velox backend test lives here (not
in the shared
+ * cross-version suite) because the `stream` constructor parameter does not
exist on Spark 3.x.
+ */
+class Spark41LocalTableScanStreamSuite extends AnyFunSuite {
+
+ private val shims = new Spark41Shims
+
+ private def output = Seq(AttributeReference("id", IntegerType)())
+
+ private val stubStream: SparkDataStream = new SparkDataStream {
+ override def initialOffset(): Offset = null
+ override def deserializeOffset(json: String): Offset = null
+ override def commit(end: Offset): Unit = {}
+ override def stop(): Unit = {}
+ }
+
+ test("getLocalTableScanStream returns the stream for a streaming
LocalTableScanExec") {
+ val plan = LocalTableScanExec(output, Seq.empty[InternalRow],
Some(stubStream))
+ val stream = shims.getLocalTableScanStream(plan)
+ assert(stream.isDefined, "Streaming LocalTableScanExec must be detected as
a streaming source")
+ assert(stream.get eq stubStream)
+ }
+
+ test("getLocalTableScanStream returns None for a batch LocalTableScanExec") {
+ val plan = LocalTableScanExec(output, Seq.empty[InternalRow], None)
+ assert(
+ shims.getLocalTableScanStream(plan).isEmpty,
+ "Batch LocalTableScanExec must not be detected as a streaming source")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
