You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "minni31 (via GitHub)" <gi...@apache.org> on 2026/07/27 07:56:03 UTC
[PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
minni31 opened a new pull request, #12631:
URL: https://github.com/apache/gluten/pull/12631
## What changes were proposed in this pull request?
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.
## 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.
- Config default is `true`, matching the convention used by other columnar operator toggles in `GlutenConfig`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5089565322
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12631:
URL: https://github.com/apache/gluten/pull/12631#discussion_r3655370619
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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)
+ }
+ }
+
+ logInfo(
+ 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
+
+ if (rows.isEmpty) {
+ sparkContext.emptyRDD[ColumnarBatch]
+ } else {
+ // Materialize rows as UnsafeRow on the driver, then parallelize
Review Comment:
`rows` is marked `@transient`, which means it can become `null` after Java serialization. `doExecuteColumnar()` calls `rows.isEmpty` without a null check, which would throw an NPE if this transformer instance is ever deserialized (similar to the `LocalTableScanExec.rows` issue guarded elsewhere). Add a defensive null check with a clear error message (or other safe handling) before using `rows`.
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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)
+ }
+ }
+
+ logInfo(
Review Comment:
`doValidateInternal()` logs a success message at INFO level for every validated LocalTableScan, which can be very noisy in production (validation runs per plan node per query). Consider downgrading this to DEBUG (or removing it) to avoid log spam.
##########
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
+ }
Review Comment:
The PR description mentions a test covering the streaming-source skip path, but the added `VeloxLocalTableScanSuite` doesn't include a case that exercises `getLocalTableScanStream(plan).isDefined` (Spark 4.0+). Either add a focused test for this behavior (Spark 4.x profiles) or update the PR description to match what's actually covered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5090117604
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5089833178
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5088750090
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5276326013
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "minni31 (via GitHub)" <gi...@apache.org>.
minni31 commented on code in PR #12631:
URL: https://github.com/apache/gluten/pull/12631#discussion_r3656030459
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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)
+ }
+ }
+
+ logInfo(
+ 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
+
+ if (rows.isEmpty) {
+ sparkContext.emptyRDD[ColumnarBatch]
+ } else {
+ // Materialize rows as UnsafeRow on the driver, then parallelize
Review Comment:
Good catch — fixed in b2323f7d0. `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`).
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxLocalTableScanTransformer.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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)
+ }
+ }
+
+ logInfo(
Review Comment:
Done in b2323f7d0 — 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "minni31 (via GitHub)" <gi...@apache.org>.
minni31 commented on code in PR #12631:
URL: https://github.com/apache/gluten/pull/12631#discussion_r3656032943
##########
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
+ }
Review Comment:
Addressed in b2323f7d0 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5276315423
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "minni31 (via GitHub)" <gi...@apache.org>.
minni31 commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5277690403
**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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12631:
URL: https://github.com/apache/gluten/pull/12631#discussion_r3656083605
##########
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)
+
Review Comment:
`doExecuteColumnar()` eagerly converts every row to an `UnsafeRow` and copies it on the driver (`rows.map(proj(_).copy()).toArray`) before parallelizing. This is redundant because `RowToVeloxColumnarExec.toColumnarBatchIterator` already converts non-`UnsafeRow` inputs to `UnsafeRow` internally (via an `UnsafeProjection`), and the driver-side materialization can double memory usage for large local relations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12631:
URL: https://github.com/apache/gluten/pull/12631#issuecomment-5091164543
Run Gluten Clickhouse CI on x86
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [CORE][VL] Add LocalTableScanExec offload support to Velox backend [gluten]
Posted by "weiting-chen (via GitHub)" <gi...@apache.org>.
weiting-chen merged PR #12631:
URL: https://github.com/apache/gluten/pull/12631
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org