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: [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]
