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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
