zhztheplayer commented on code in PR #7360: URL: https://github.com/apache/incubator-gluten/pull/7360#discussion_r1778166370
########## backends-velox/src/main/scala/org/apache/gluten/execution/SparkPartialProjectColumnarExec.scala: ########## @@ -0,0 +1,408 @@ +/* + * 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.GlutenConfig +import org.apache.gluten.columnarbatch.ColumnarBatches +import org.apache.gluten.extension.{GlutenPlan, ValidationResult} +import org.apache.gluten.extension.columnar.validator.Validator.Passed +import org.apache.gluten.extension.columnar.validator.Validators.FallbackComplexExpressions +import org.apache.gluten.iterator.Iterators +import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators +import org.apache.gluten.sql.shims.SparkShimLoader +import org.apache.gluten.vectorized.ArrowWritableColumnVector + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Coalesce, Expression, If, LambdaFunction, MutableProjection, NamedExpression, NaNvl, ScalaUDF, UnsafeProjection} +import org.apache.spark.sql.execution.{ExplainUtils, ProjectExec, SparkPlan, UnaryExecNode} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.execution.vectorized.{MutableColumnarRow, WritableColumnVector} +import org.apache.spark.sql.hive.HiveUdfUtil +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampType, YearMonthIntervalType} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +import scala.collection.mutable.ListBuffer + +/** + * Change the Project to ProjectExecTransformer + SparkPartialProjectColumnarExec e.g. sum(myudf(a) Review Comment: ``` By rule <rule name>, the project that is not offload-able is changed to .... ``` ########## backends-velox/src/main/scala/org/apache/gluten/execution/SparkPartialProjectColumnarExec.scala: ########## @@ -0,0 +1,408 @@ +/* + * 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.GlutenConfig +import org.apache.gluten.columnarbatch.ColumnarBatches +import org.apache.gluten.extension.{GlutenPlan, ValidationResult} +import org.apache.gluten.extension.columnar.validator.Validator.Passed +import org.apache.gluten.extension.columnar.validator.Validators.FallbackComplexExpressions +import org.apache.gluten.iterator.Iterators +import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators +import org.apache.gluten.sql.shims.SparkShimLoader +import org.apache.gluten.vectorized.ArrowWritableColumnVector + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Coalesce, Expression, If, LambdaFunction, MutableProjection, NamedExpression, NaNvl, ScalaUDF, UnsafeProjection} +import org.apache.spark.sql.execution.{ExplainUtils, ProjectExec, SparkPlan, UnaryExecNode} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.execution.vectorized.{MutableColumnarRow, WritableColumnVector} +import org.apache.spark.sql.hive.HiveUdfUtil +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampType, YearMonthIntervalType} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +import scala.collection.mutable.ListBuffer + +/** + * Change the Project to ProjectExecTransformer + SparkPartialProjectColumnarExec e.g. sum(myudf(a) + * + b + hash(c)), child is (a, b,c ) SparkPartialProjectColumnarExec (a, b, c, myudf(a)), + * ProjectExecTransformer(myudf(a) + b + hash(c)) + * + * @param original + * extract the ScalaUDF from original project list as Alias in UnsafeProjection and + * AttributeReference in SparkPartialProjectColumnarExec output + * @param child + * child plan + */ +case class SparkPartialProjectColumnarExec(original: ProjectExec, child: SparkPlan)( Review Comment: Could put `Columnar` on LHS for plan naming. Like `ColumnarSparkPartialProjectExec`. Though it's still weird to insert `Spark` into a plan name. Do we have better choice here? ########## gluten-substrait/src/test/scala/org/apache/gluten/execution/WholeStageTransformerSuite.scala: ########## @@ -316,7 +316,6 @@ abstract class WholeStageTransformerSuite } } checkDataFrame(noFallBack, customCheck, df) - df.explain(true) Review Comment: My mistake. Thanks for fixing. ########## backends-velox/src/main/scala/org/apache/gluten/execution/SparkPartialProjectColumnarExec.scala: ########## @@ -0,0 +1,408 @@ +/* + * 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.GlutenConfig +import org.apache.gluten.columnarbatch.ColumnarBatches +import org.apache.gluten.extension.{GlutenPlan, ValidationResult} +import org.apache.gluten.extension.columnar.validator.Validator.Passed +import org.apache.gluten.extension.columnar.validator.Validators.FallbackComplexExpressions +import org.apache.gluten.iterator.Iterators +import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators +import org.apache.gluten.sql.shims.SparkShimLoader +import org.apache.gluten.vectorized.ArrowWritableColumnVector + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Coalesce, Expression, If, LambdaFunction, MutableProjection, NamedExpression, NaNvl, ScalaUDF, UnsafeProjection} +import org.apache.spark.sql.execution.{ExplainUtils, ProjectExec, SparkPlan, UnaryExecNode} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.execution.vectorized.{MutableColumnarRow, WritableColumnVector} +import org.apache.spark.sql.hive.HiveUdfUtil +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampType, YearMonthIntervalType} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +import scala.collection.mutable.ListBuffer + +/** + * Change the Project to ProjectExecTransformer + SparkPartialProjectColumnarExec e.g. sum(myudf(a) + * + b + hash(c)), child is (a, b,c ) SparkPartialProjectColumnarExec (a, b, c, myudf(a)), + * ProjectExecTransformer(myudf(a) + b + hash(c)) + * + * @param original + * extract the ScalaUDF from original project list as Alias in UnsafeProjection and + * AttributeReference in SparkPartialProjectColumnarExec output + * @param child + * child plan + */ +case class SparkPartialProjectColumnarExec(original: ProjectExec, child: SparkPlan)( + replacedAliasUdf: ListBuffer[Alias]) + extends UnaryExecNode + with GlutenPlan { + + private val debug = GlutenConfig.getConf.debug + + private val projectAttributes: ListBuffer[Attribute] = ListBuffer() + private val projectIndexInChild: ListBuffer[Int] = ListBuffer() + private var UDFAttrNotExists = false + private var hasComplexDataType = replacedAliasUdf.exists(a => !validateDataType(a.dataType)) + if (!hasComplexDataType) { + getProjectIndexInChildOutput(replacedAliasUdf) + } + + @transient override lazy val metrics = Map( + "time" -> SQLMetrics.createTimingMetric(sparkContext, "time of project"), + "column_to_row_time" -> SQLMetrics.createTimingMetric( + sparkContext, + "time of velox to Arrow ColumnarBatch"), + "row_to_column_time" -> SQLMetrics.createTimingMetric( + sparkContext, + "time of Arrow ColumnarBatch to velox") + ) + + override def output: Seq[Attribute] = child.output ++ replacedAliasUdf.map(_.toAttribute) + + final override def doExecute(): RDD[InternalRow] = { + throw new UnsupportedOperationException( + s"${this.getClass.getSimpleName} doesn't support doExecute") + } + + final override protected def otherCopyArgs: Seq[AnyRef] = { + replacedAliasUdf :: Nil + } + + final override lazy val supportsColumnar: Boolean = true + + private def validateExpression(expr: Expression): Boolean = { + expr.deterministic && !expr.isInstanceOf[LambdaFunction] && expr.children + .forall(validateExpression) + } + + private def validateDataType(dataType: DataType): Boolean = { + dataType match { + case _: BooleanType => true + case _: ByteType => true + case _: ShortType => true + case _: IntegerType => true + case _: LongType => true + case _: FloatType => true + case _: DoubleType => true + case _: StringType => true + case _: TimestampType => true + case _: DateType => true + case _: BinaryType => true + case _: DecimalType => true + case YearMonthIntervalType.DEFAULT => true + case _: NullType => true + case _ => false + } + } + + private def getProjectIndexInChildOutput(exprs: Seq[Expression]): Unit = { + exprs.foreach { + case a: AttributeReference => + val index = child.output.indexWhere(s => s.exprId.equals(a.exprId)) + // Some child operator as HashAggregateTransformer will not have udf child column + if (index < 0) { + UDFAttrNotExists = true + log.debug(s"Expression $a should exist in child output ${child.output}") + return + } else if (!validateDataType(a.dataType)) { + hasComplexDataType = true + log.debug(s"Expression $a contains unsupported data type ${a.dataType}") + } else if (!projectIndexInChild.contains(index)) { + projectAttributes.append(a.toAttribute) + projectIndexInChild.append(index) + } + case p => getProjectIndexInChildOutput(p.children) + } + } + + override protected def doValidateInternal(): ValidationResult = { + if (!GlutenConfig.getConf.enableColumnarPartialProject) { + return ValidationResult.failed("Config disable this feature") + } + if (UDFAttrNotExists) { + ValidationResult.failed("Attribute in the UDF does not exists in its child") + } else if (hasComplexDataType) { + ValidationResult.failed("Attribute in the UDF contains unsupported type") + } else if (projectAttributes.size == child.output.size) { + ValidationResult.failed("UDF need all the columns in child output") + } else if (original.output.isEmpty) { + ValidationResult.failed("Project fallback because output is empty") + } else if (replacedAliasUdf.isEmpty) { + ValidationResult.failed("No UDF") + } else if (replacedAliasUdf.size > original.output.size) { + // e.g. udf1(col) + udf2(col), it will introduce 2 cols for r2c + ValidationResult.failed("Number of RowToColumn columns is more than ProjectExec") + } else if (!original.projectList.forall(validateExpression(_))) { + ValidationResult.failed("Contains expression not supported") + } else if (isComplexExpression()) { + ValidationResult.failed("Fallback by complex expression") + } else { + ValidationResult.succeeded + } Review Comment: Can we break this long `if-else` sequence into individual `if-return` blocks? ########## gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala: ########## @@ -72,6 +73,8 @@ trait SparkPlanExecApi { child: SparkPlan): ProjectExecTransformer = ProjectExecTransformer.createUnsafe(projectList, child) + def genSparkPartialProjectColumnarExec(original: ProjectExec): GlutenPlan = null Review Comment: Instead of adding code to `OffloadSingleNode`, can we just add a new rule to [Velox's rule list](https://github.com/apache/incubator-gluten/blob/5790c30959dad59284a5e50015d94912b79f1b23/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala#L47), then remove this API from backend API? ########## backends-velox/src/main/scala/org/apache/gluten/execution/SparkPartialProjectColumnarExec.scala: ########## @@ -0,0 +1,408 @@ +/* + * 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.GlutenConfig +import org.apache.gluten.columnarbatch.ColumnarBatches +import org.apache.gluten.extension.{GlutenPlan, ValidationResult} +import org.apache.gluten.extension.columnar.validator.Validator.Passed +import org.apache.gluten.extension.columnar.validator.Validators.FallbackComplexExpressions +import org.apache.gluten.iterator.Iterators +import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators +import org.apache.gluten.sql.shims.SparkShimLoader +import org.apache.gluten.vectorized.ArrowWritableColumnVector + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Coalesce, Expression, If, LambdaFunction, MutableProjection, NamedExpression, NaNvl, ScalaUDF, UnsafeProjection} +import org.apache.spark.sql.execution.{ExplainUtils, ProjectExec, SparkPlan, UnaryExecNode} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.execution.vectorized.{MutableColumnarRow, WritableColumnVector} +import org.apache.spark.sql.hive.HiveUdfUtil +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampType, YearMonthIntervalType} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +import scala.collection.mutable.ListBuffer + +/** + * Change the Project to ProjectExecTransformer + SparkPartialProjectColumnarExec e.g. sum(myudf(a) + * + b + hash(c)), child is (a, b,c ) SparkPartialProjectColumnarExec (a, b, c, myudf(a)), + * ProjectExecTransformer(myudf(a) + b + hash(c)) + * + * @param original + * extract the ScalaUDF from original project list as Alias in UnsafeProjection and + * AttributeReference in SparkPartialProjectColumnarExec output + * @param child + * child plan + */ +case class SparkPartialProjectColumnarExec(original: ProjectExec, child: SparkPlan)( + replacedAliasUdf: ListBuffer[Alias]) + extends UnaryExecNode + with GlutenPlan { + + private val debug = GlutenConfig.getConf.debug + + private val projectAttributes: ListBuffer[Attribute] = ListBuffer() + private val projectIndexInChild: ListBuffer[Int] = ListBuffer() + private var UDFAttrNotExists = false + private var hasComplexDataType = replacedAliasUdf.exists(a => !validateDataType(a.dataType)) + if (!hasComplexDataType) { + getProjectIndexInChildOutput(replacedAliasUdf) + } + + @transient override lazy val metrics = Map( + "time" -> SQLMetrics.createTimingMetric(sparkContext, "time of project"), + "column_to_row_time" -> SQLMetrics.createTimingMetric( + sparkContext, + "time of velox to Arrow ColumnarBatch"), + "row_to_column_time" -> SQLMetrics.createTimingMetric( + sparkContext, + "time of Arrow ColumnarBatch to velox") + ) + + override def output: Seq[Attribute] = child.output ++ replacedAliasUdf.map(_.toAttribute) + + final override def doExecute(): RDD[InternalRow] = { + throw new UnsupportedOperationException( + s"${this.getClass.getSimpleName} doesn't support doExecute") + } + + final override protected def otherCopyArgs: Seq[AnyRef] = { + replacedAliasUdf :: Nil + } + + final override lazy val supportsColumnar: Boolean = true Review Comment: Could remove `lazy` ########## backends-velox/src/main/scala/org/apache/gluten/execution/SparkPartialProjectColumnarExec.scala: ########## @@ -0,0 +1,408 @@ +/* + * 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.GlutenConfig +import org.apache.gluten.columnarbatch.ColumnarBatches +import org.apache.gluten.extension.{GlutenPlan, ValidationResult} +import org.apache.gluten.extension.columnar.validator.Validator.Passed +import org.apache.gluten.extension.columnar.validator.Validators.FallbackComplexExpressions +import org.apache.gluten.iterator.Iterators +import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators +import org.apache.gluten.sql.shims.SparkShimLoader +import org.apache.gluten.vectorized.ArrowWritableColumnVector + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, CaseWhen, Coalesce, Expression, If, LambdaFunction, MutableProjection, NamedExpression, NaNvl, ScalaUDF, UnsafeProjection} +import org.apache.spark.sql.execution.{ExplainUtils, ProjectExec, SparkPlan, UnaryExecNode} +import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.execution.vectorized.{MutableColumnarRow, WritableColumnVector} +import org.apache.spark.sql.hive.HiveUdfUtil +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, NullType, ShortType, StringType, TimestampType, YearMonthIntervalType} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +import scala.collection.mutable.ListBuffer + +/** + * Change the Project to ProjectExecTransformer + SparkPartialProjectColumnarExec e.g. sum(myudf(a) + * + b + hash(c)), child is (a, b,c ) SparkPartialProjectColumnarExec (a, b, c, myudf(a)), Review Comment: s/`(a, b,c )`/`(a, b, c)` ########## gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala: ########## @@ -116,7 +116,7 @@ object Validators { } } - private class FallbackComplexExpressions(threshold: Int) extends Validator { + class FallbackComplexExpressions(threshold: Int) extends Validator { Review Comment: Instead of making this public, could move the shared code into `ExpressionUtils` ########## shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala: ########## @@ -1864,6 +1866,12 @@ object GlutenConfig { .booleanConf .createWithDefault(true) + val ENABLE_COLUMNAR_PARTIAL_PROJECT = + buildConf("spark.gluten.sql.columnar.partial.project") + .doc("Execute partial project which is not supported in backend in Spark") Review Comment: ``` Break up one project node into 2 phases when some of the expressions are non offload-able. Stage one is a regular offloaded project transformer that evaluates the offload-able expressions in native, stage two preserves the output from stage one and evaluates the remaining non-offload-able expressions using vanilla Spark projections. ``` -- 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]
