[CORE][VL] Add LocalTableScanExec offload support to Velox backend

[minni31]

What changes were proposed in this pull request?

Offload LocalTableScanExec (a driver-side local collection, e.g. from Seq(...).toDF, VALUES, or a small LocalRelation) to native columnar execution on the Velox backend by converting its rows into columnar batches via the existing RowToVeloxColumnar native row-to-columnar 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.
• 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.

Default is false for now (known write-path limitation)

The config currently defaults to false. 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 does not implement CodegenSupport, so FileFormatWriter's whole-stage codegen can cast it to CodegenSupport and fail with a ClassCastException (reproducible on Spark 3.3 via INSERT ... VALUES whose casts become an AnsiCast that falls back).

The feature itself is complete and fully covered by tests (which enable the config explicitly). A follow-up will make the transition codegen-safe — reverting a no-op offload that is immediately topped by a columnar-to-row transition back to a vanilla LocalTableScanExec — and flip the default to true.

How was this patch tested?

• New unit/integration suite VeloxLocalTableScanSuite covering: successful offload, fallback when disabled, fallback for unsupported (Map/Interval) types, the deserialized-null-rows NPE guard, and streaming-source skip. The suite sets spark.gluten.sql.columnar.localTableScan=true explicitly, so it is independent of the default.
• Streaming-source skip covered by shim-level suites on Spark 4.0/4.1.
• MiscOperatorSuite's RowToVeloxColumnar preferredBatchBytes test pins the config off so its plan expectations are unaffected by this operator.

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

This PR adds Velox-backend offload support for Spark’s LocalTableScanExec by introducing a transformer that converts local InternalRow collections into native columnar batches (via the existing Velox row-to-columnar JNI path), wiring it into the single-node offload rule, and gating it behind a new SQL config.

Changes:

• Introduce a new LocalTableScanTransformer base + backend API hooks to enable/construct backend-specific replacements for LocalTableScanExec.
• Add Velox implementation (VeloxLocalTableScanTransformer) with schema/Arrow-compatibility validation and config gating, including shims to detect Spark 4.x streaming sources.
• Add Velox test suite coverage and document the new spark.gluten.sql.columnar.localTableScan config.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala	Exposes LocalTableScanExec.stream (Spark 4.1) via a shim accessor.
shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala	Exposes LocalTableScanExec.stream (Spark 4.0) via a shim accessor.
shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala	Adds cross-version shim hook for LocalTableScan streaming-source detection (defaults to None).
gluten-substrait/src/main/scala/org/apache/spark/sql/execution/LocalTableScanTransformer.scala	Adds a transformer base + companion helpers to route backend support/creation.
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/offload/OffloadSingleNodeRules.scala	Wires LocalTableScanExec into the “others” offload rule when backend reports support.
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala	Adds spark.gluten.sql.columnar.localTableScan config and accessor.
gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala	Adds backend API hooks (isSupportLocalTableScanExec, getLocalTableScanTransform).
docs/Configuration.md	Documents the new spark.gluten.sql.columnar.localTableScan toggle.
backends-velox/src/test/scala/org/apache/gluten/execution/VeloxLocalTableScanSuite.scala	Adds Velox-side test coverage for offload and fallback scenarios.
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala	Implements Velox local-table-scan offload via driver materialization + JNI row-to-columnar conversion.
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala	Enables Velox support checks + transformer creation, including streaming-source and null-transient-row guards.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[minni31]

Good catch — fixed in b2323f7. doExecuteColumnar() now guards rows before use: if it's null (the deserialized @transient case) it throws an IllegalStateException with a clear message explaining the plan should never have been offloaded, instead of a bare NPE. Since offload is already blocked at planning time in isSupportLocalTableScanExec (which returns false for null rows), this is a defensive execution-time backstop; IllegalStateException matches the invariant-violation convention used elsewhere in this package (e.g. GenerateExecTransformer, VeloxSerializedBroadcastRDD).

[minni31]

Done in b2323f7 — downgraded the per-node validation-success log from logInfo to logDebug. doValidateInternal() runs per plan node per query, so INFO was too noisy for production; DEBUG keeps it available for diagnostics without the spam.

[minni31]

Addressed in b2323f7 by adding explicit test coverage for the streaming path:

• Spark 4.x streaming-detected path — new Spark40LocalTableScanStreamSuite and Spark41LocalTableScanStreamSuite construct a LocalTableScanExec carrying a stream and assert getLocalTableScanStream(plan).isDefined (plus None for a batch plan). These live in the shims/spark40 and shims/spark41 modules because the stream constructor parameter is Spark 4.x-only and cannot compile in the shared cross-version VeloxLocalTableScanSuite.
• Version-agnostic guard — added a case to VeloxLocalTableScanSuite asserting a batch LocalTableScanExec is classified as non-streaming (None) and is therefore not falsely skipped by the streaming guard, on every supported Spark version.

Both shim suites pass locally (2/2 each) under -Pspark-4.0 and -Pspark-4.1.

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:70

• This logs a validation-success message at INFO for every LocalTableScan offload, which can be very noisy in production (local relations can appear frequently, e.g. in AQE/broadcast-related plans). Consider lowering this to DEBUG to avoid inflating driver logs without losing troubleshooting capability.

    logInfo(
      s"local_table_scan native validation succeeded: " +
        s"schema=${schema.fields.map(_.dataType.simpleString).mkString(",")}, " +
        s"appId=${sparkContext.applicationId}")

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:108

• rows.map(...).toArray materializes and copies the full local relation as UnsafeRows on the driver before parallelizing. This duplicates memory (rows already exist on the driver) and can significantly increase driver heap pressure for large local relations. It’s safer to parallelize the original rows first and project/copy per-partition on executors (and also guard against rows == null in case the transformer is ever serialized).

    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

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.

Suppressed comments (3)

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:118

• This always runs an UnsafeProjection for every row, even if a row is already an UnsafeRow. Since LocalTableScanExec rows can already be unsafe, you can avoid unnecessary projection work by copying UnsafeRow inputs directly (similar to RowToVeloxColumnarExec).

      // Materialize rows as UnsafeRow on the driver, then parallelize
      val proj = UnsafeProjection.create(outputAttributes, outputAttributes)
      val unsafeRows = rows.map(r => proj(r).copy()).toArray

gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:930

• The PR description says spark.gluten.sql.columnar.localTableScan defaults to true, but this config is created with a default of false (and the docs entry also shows false). Please reconcile the PR description vs. the shipped default (either update the description or flip the default once the codegen-safe transition issue is resolved).

          "attempts to replace LocalTableScanExec (a driver-side local collection) with a " +

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxLocalTableScanSuite.scala:101

• The empty-collection case only asserts correctness via Spark fallback (checkAnswer), but doesn’t assert that the new LocalTableScan offload path is actually exercised. Adding an assertHasVeloxLocalTableScan(df) here would cover the rows.isEmpty execution branch in VeloxLocalTableScanTransformer.doExecuteColumnar() and guard against regressions where empty local scans silently stop offloading.

  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])
  }

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.

Suppressed comments (1)

gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala:934

• The PR description says spark.gluten.sql.columnar.localTableScan defaults to true, but this config entry is created with createWithDefault(false) (and docs also show false). Please align the PR description with the actual default (or change the default here if true is intended).

  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)

[minni31]

Note on the default value (spark.gluten.sql.columnar.localTableScan)

For this PR the config defaults to false. There is one known limitation: 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 does not implement CodegenSupport, so FileFormatWriter's whole-stage codegen can cast it to CodegenSupport and fail with a ClassCastException (reproducible on Spark 3.3 via INSERT ... VALUES whose casts fall back through an AnsiCast).

The feature itself is complete and fully covered by tests (which enable the config explicitly). This limitation will be addressed in a follow-up PR that makes the transition codegen-safe — reverting a no-op offload that is immediately topped by a columnar-to-row transition back to a vanilla LocalTableScanExec — after which the default will be flipped to true.

Keeping the default off here lets this PR land safely while the codegen-safe handling is reviewed separately.

[weiting-chen]

Looks good to me. BTW, this feature set to false by default so no risk for me as well.