This is an automated email from the ASF dual-hosted git repository.
taiyangli pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 0918db2eff optimize nested and/or function calls (#8558)
0918db2eff is described below
commit 0918db2eff4aaefd148e48791baa10eb0d791d29
Author: kevinyhzou <[email protected]>
AuthorDate: Thu Apr 3 15:38:49 2025 +0800
optimize nested and/or function calls (#8558)
Co-authored-by: BIGO <[email protected]>
---
.../gluten/backendsapi/clickhouse/CHRuleApi.scala | 1 +
.../clickhouse/CHSparkPlanExecApi.scala | 19 ++-
.../gluten/execution/CHFilterExecTransformer.scala | 13 +-
.../gluten/expression/CHFlattenedExpression.scala | 84 ++++++++++
.../extension/FlattenNestedExpressions.scala | 175 +++++++++++++++++++++
.../execution/GlutenFunctionValidateSuite.scala | 40 +++++
.../Functions/SparkFunctionTupleElement.cpp | 6 -
cpp-ch/local-engine/Parser/FunctionParser.cpp | 1 -
.../gluten/backendsapi/SparkPlanExecApi.scala | 9 ++
.../BasicPhysicalOperatorTransformer.scala | 5 +-
.../gluten/expression/ExpressionConverter.scala | 19 +++
.../org/apache/gluten/config/GlutenConfig.scala | 11 +-
12 files changed, 369 insertions(+), 14 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
index edcdee7fd2..695deaddbf 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
@@ -125,6 +125,7 @@ object CHRuleApi {
injector.injectPostTransform(c =>
AddPreProjectionForHashJoin.apply(c.session))
injector.injectPostTransform(c =>
ReplaceSubStringComparison.apply(c.session))
injector.injectPostTransform(c =>
EliminateDeduplicateAggregateWithAnyJoin(c.session))
+ injector.injectPostTransform(c =>
FlattenNestedExpressions.apply(c.session))
// Gluten columnar: Fallback policies.
injector.injectFallbackPolicy(c => p =>
ExpandFallbackPolicy(c.caller.isAqe(), p))
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
index 3a4267c4b1..dcf19204d2 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala
@@ -581,7 +581,9 @@ class CHSparkPlanExecApi extends SparkPlanExecApi with
Logging {
List(
Sig[CollectList](ExpressionNames.COLLECT_LIST),
Sig[CollectSet](ExpressionNames.COLLECT_SET),
- Sig[MonotonicallyIncreasingID](MONOTONICALLY_INCREASING_ID)
+ Sig[MonotonicallyIncreasingID](MONOTONICALLY_INCREASING_ID),
+ CHFlattenedExpression.sigAnd,
+ CHFlattenedExpression.sigOr
) ++
ExpressionExtensionTrait.expressionExtensionSigList ++
SparkShimLoader.getSparkShims.bloomFilterExpressionMappings()
@@ -947,4 +949,19 @@ class CHSparkPlanExecApi extends SparkPlanExecApi with
Logging {
outputAttributes: Seq[Attribute],
child: Seq[SparkPlan]): ColumnarRangeBaseExec =
CHRangeExecTransformer(start, end, step, numSlices, numElements,
outputAttributes, child)
+
+ override def expressionFlattenSupported(expr: Expression): Boolean = expr
match {
+ case ca: FlattenedAnd => CHFlattenedExpression.supported(ca.name)
+ case co: FlattenedOr => CHFlattenedExpression.supported(co.name)
+ case _ => false
+ }
+
+ override def genFlattenedExpressionTransformer(
+ substraitName: String,
+ children: Seq[ExpressionTransformer],
+ expr: Expression): ExpressionTransformer = expr match {
+ case ce: FlattenedAnd => GenericExpressionTransformer(ce.name, children,
ce)
+ case co: FlattenedOr => GenericExpressionTransformer(co.name, children, co)
+ case _ => super.genFlattenedExpressionTransformer(substraitName, children,
expr)
+ }
}
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHFilterExecTransformer.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHFilterExecTransformer.scala
index 79ddf2942c..686e51fcf6 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHFilterExecTransformer.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHFilterExecTransformer.scala
@@ -16,7 +16,9 @@
*/
package org.apache.gluten.execution
-import org.apache.spark.sql.catalyst.expressions.{And, Expression}
+import org.apache.gluten.expression.CHFlattenedExpression
+
+import org.apache.spark.sql.catalyst.expressions.{And, Expression, ExprId,
IsNotNull}
import org.apache.spark.sql.execution.SparkPlan
case class CHFilterExecTransformer(condition: Expression, child: SparkPlan)
@@ -48,4 +50,13 @@ case class FilterExecTransformer(condition: Expression,
child: SparkPlan)
override protected def getRemainingCondition: Expression = condition
override protected def withNewChildInternal(newChild: SparkPlan):
FilterExecTransformer =
copy(child = newChild)
+ override protected val notNullAttributes: Seq[ExprId] = condition match {
+ case s: CHFlattenedExpression =>
+ val (notNullPreds, _) = s.children.partition {
+ case IsNotNull(a) => isNullIntolerant(a) &&
a.references.subsetOf(child.outputSet)
+ case _ => false
+ }
+ notNullPreds.flatMap(_.references).distinct.map(_.exprId)
+ case _ => notNullPreds.flatMap(_.references).distinct.map(_.exprId)
+ }
}
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/expression/CHFlattenedExpression.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/expression/CHFlattenedExpression.scala
new file mode 100644
index 0000000000..e0a4623ce5
--- /dev/null
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/expression/CHFlattenedExpression.scala
@@ -0,0 +1,84 @@
+/*
+ * 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.expression
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.exception.GlutenException
+import
org.apache.gluten.expression.CHFlattenedExpression.genFlattenedExpression
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
+import org.apache.spark.sql.types.DataType
+
+abstract class CHFlattenedExpression(children: Seq[Expression], name: String)
extends Expression {
+
+ def this() = {
+ this(Seq.empty[Expression], "")
+ }
+
+ override def toString: String = s"$name(${children.mkString(", ")})"
+
+ override def eval(input: InternalRow): Any = null
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = null
+
+ override protected def withNewChildrenInternal(newChildren:
IndexedSeq[Expression]): Expression =
+ genFlattenedExpression(dataType, newChildren, name, nullable) match {
+ case Some(f) => f
+ case None =>
+ throw new GlutenException(
+ String.format(
+ "Logical error, the function %s should not be a clickhouse
Flattened function.",
+ name))
+ }
+}
+
+case class FlattenedAnd(
+ dataType: DataType,
+ children: Seq[Expression],
+ name: String,
+ nullable: Boolean)
+ extends CHFlattenedExpression(children, name) {}
+
+case class FlattenedOr(
+ dataType: DataType,
+ children: Seq[Expression],
+ name: String,
+ nullable: Boolean)
+ extends CHFlattenedExpression(children, name) {}
+
+object CHFlattenedExpression {
+
+ def sigAnd: Sig = Sig[FlattenedAnd]("FlattenedAnd")
+ def sigOr: Sig = Sig[FlattenedOr]("FlattenedOr")
+
+ def supported(name: String): Boolean = {
+ GlutenConfig.get.getSupportedFlattenedExpressions.split(",").exists(p =>
p.equals(name))
+ }
+
+ def genFlattenedExpression(
+ dataType: DataType,
+ children: Seq[Expression],
+ name: String,
+ nullable: Boolean): Option[CHFlattenedExpression] = name match {
+ case "and" => Option.apply(FlattenedAnd(dataType, children, name,
nullable))
+ case "or" => Option.apply(FlattenedOr(dataType, children, name, nullable))
+ case _ => Option.empty
+ }
+
+}
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/extension/FlattenNestedExpressions.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/FlattenNestedExpressions.scala
new file mode 100644
index 0000000000..8129f75633
--- /dev/null
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/extension/FlattenNestedExpressions.scala
@@ -0,0 +1,175 @@
+/*
+ * 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.extension
+
+import org.apache.gluten.execution.{FilterExecTransformer,
ProjectExecTransformer}
+import org.apache.gluten.expression.{CHFlattenedExpression, ExpressionMappings}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.types.DataType
+
+/**
+ * Flatten nested expressions for optimization, to reduce expression calls.
Now support `and`, `or`.
+ * e.g. select ... and(and(a=1, b=2), c=3) => select ... and(a=1, b=2, c=3).
+ */
+case class FlattenNestedExpressions(spark: SparkSession) extends
Rule[SparkPlan] {
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ if (canBeOptimized(plan)) {
+ visitPlan(plan)
+ } else {
+ plan
+ }
+ }
+
+ private def canBeOptimized(plan: SparkPlan): Boolean = plan match {
+ case p: ProjectExecTransformer =>
+ var res = p.projectList.exists(c => c.isInstanceOf[And] ||
c.isInstanceOf[Or])
+ if (res) {
+ return false
+ }
+ res = p.projectList.exists(c => canBeOptimized(c))
+ if (!res) {
+ res = p.children.exists(c => canBeOptimized(c))
+ }
+ res
+ case f: FilterExecTransformer =>
+ var res = canBeOptimized(f.condition)
+ if (!res) {
+ res = canBeOptimized(f.child)
+ }
+ res
+ case _ => plan.children.exists(c => canBeOptimized(c))
+ }
+
+ private def canBeOptimized(expr: Expression): Boolean = {
+ var exprCall = expr
+ expr match {
+ case a: Alias => exprCall = a.child
+ case _ =>
+ }
+ val exprName = getExpressionName(exprCall)
+ exprName match {
+ case None =>
+ exprCall match {
+ case _: LeafExpression => false
+ case _ => exprCall.children.exists(c => canBeOptimized(c))
+ }
+ case Some(f) =>
+ CHFlattenedExpression.supported(f)
+ }
+ }
+
+ private def getExpressionName(expr: Expression): Option[String] = expr match
{
+ case _: And => ExpressionMappings.expressionsMap.get(classOf[And])
+ case _: Or => ExpressionMappings.expressionsMap.get(classOf[Or])
+ case _ => Option.empty[String]
+ }
+
+ private def visitPlan(plan: SparkPlan): SparkPlan = plan match {
+ case p: ProjectExecTransformer =>
+ var newProjectList = Seq.empty[NamedExpression]
+ p.projectList.foreach {
+ case a: Alias =>
+ val newAlias = Alias(optimize(a.child), a.name)(a.exprId)
+ newProjectList :+= newAlias
+ case p =>
+ newProjectList :+= p
+ }
+ val newChild = visitPlan(p.child)
+ ProjectExecTransformer(newProjectList, newChild)
+ case f: FilterExecTransformer =>
+ val newCondition = optimize(f.condition)
+ val newChild = visitPlan(f.child)
+ FilterExecTransformer(newCondition, newChild)
+ case _ =>
+ val newChildren = plan.children.map(p => visitPlan(p))
+ plan.withNewChildren(newChildren)
+ }
+
+ private def optimize(expr: Expression): Expression = {
+ var resultExpr = expr
+ var name = getExpressionName(expr)
+ var children = Seq.empty[Expression]
+ var dataType = null.asInstanceOf[DataType]
+ var nestedFunctions = 0
+
+ def f(e: Expression, parent: Option[Expression] =
Option.empty[Expression]): Unit = {
+ parent match {
+ case None =>
+ name = getExpressionName(e)
+ dataType = e.dataType
+ case _ =>
+ }
+ e match {
+ case a: And if canBeOptimized(a) =>
+ parent match {
+ case Some(_: And) | None =>
+ f(a.left, Option.apply(a))
+ f(a.right, Option.apply(a))
+ nestedFunctions += 1
+ case _ =>
+ children +:= optimize(a)
+ }
+ case o: Or if canBeOptimized(o) =>
+ parent match {
+ case Some(_: Or) | None =>
+ f(o.left, parent = Option.apply(o))
+ f(o.right, parent = Option.apply(o))
+ nestedFunctions += 1
+ case _ =>
+ children +:= optimize(o)
+ }
+ case _ =>
+ if (parent.nonEmpty) {
+ children +:= optimize(e)
+ } else {
+ children = Seq.empty[Expression]
+ nestedFunctions = 0
+ val exprNewChildren = e.children.map(p => optimize(p))
+ resultExpr = e.withNewChildren(exprNewChildren)
+ }
+ }
+ }
+ f(expr)
+ if ((nestedFunctions > 1 && name.isDefined) ||
flattenedExpressionExists(children)) {
+ CHFlattenedExpression.genFlattenedExpression(
+ dataType,
+ children,
+ name.getOrElse(""),
+ expr.nullable) match {
+ case Some(f) => f
+ case None => resultExpr
+ }
+ } else {
+ resultExpr
+ }
+ }
+
+ private def flattenedExpressionExists(children: Seq[Expression]): Boolean = {
+ var res = false
+ children.foreach {
+ case _: CHFlattenedExpression if !res => res = true
+ case c if !res => res = flattenedExpressionExists(c.children)
+ case _ =>
+ }
+ res
+ }
+}
diff --git
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
index c50d6cd717..7bbcc3c363 100644
---
a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
+++
b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala
@@ -17,6 +17,7 @@
package org.apache.gluten.execution
import org.apache.gluten.backendsapi.clickhouse.CHConfig
+import org.apache.gluten.expression.{FlattenedAnd, FlattenedOr}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, GlutenTestUtils, Row}
@@ -385,6 +386,45 @@ class GlutenFunctionValidateSuite extends
GlutenClickHouseWholeStageTransformerS
}
}
+ test("GLUTEN-8557: Optimize nested and/or") {
+ def checkFlattenedFunctions(plan: SparkPlan, functionName: String, argNum:
Int): Boolean = {
+
+ def checkExpression(expr: Expression, functionName: String, argNum:
Int): Boolean =
+ expr match {
+ case s: FlattenedAnd if s.name.equals(functionName) &&
s.children.size == argNum =>
+ true
+ case o: FlattenedOr if o.name.equals(functionName) &&
o.children.size == argNum =>
+ true
+ case _ => expr.children.exists(c => checkExpression(c, functionName,
argNum))
+ }
+ plan match {
+ case f: FilterExecTransformer => return checkExpression(f.condition,
functionName, argNum)
+ case _ => return plan.children.exists(c => checkFlattenedFunctions(c,
functionName, argNum))
+ }
+ false
+ }
+ runQueryAndCompare(
+ "SELECT count(1) from json_test where int_field1 = 5 and double_field1 >
1.0" +
+ " and string_field1 is not null") {
+ x => assert(checkFlattenedFunctions(x.queryExecution.executedPlan,
"and", 5))
+ }
+ runQueryAndCompare(
+ "SELECT count(1) from json_test where int_field1 = 5 or double_field1 >
1.0" +
+ " or string_field1 is not null") {
+ x => assert(checkFlattenedFunctions(x.queryExecution.executedPlan, "or",
3))
+ }
+ runQueryAndCompare(
+ "SELECT count(1) from json_test where int_field1 = 5 and double_field1 >
1.0" +
+ " and double_field1 < 10 or int_field1 = 12 or string_field1 is not
null") {
+ x =>
+ assert(
+ checkFlattenedFunctions(
+ x.queryExecution.executedPlan,
+ "and",
+ 3) && checkFlattenedFunctions(x.queryExecution.executedPlan, "or",
3))
+ }
+ }
+
test("Test covar_samp") {
runQueryAndCompare("SELECT covar_samp(double_field1, int_field1) from
json_test") { _ => }
}
diff --git a/cpp-ch/local-engine/Functions/SparkFunctionTupleElement.cpp
b/cpp-ch/local-engine/Functions/SparkFunctionTupleElement.cpp
index d52abde2cc..884adbaf2b 100644
--- a/cpp-ch/local-engine/Functions/SparkFunctionTupleElement.cpp
+++ b/cpp-ch/local-engine/Functions/SparkFunctionTupleElement.cpp
@@ -68,14 +68,12 @@ public:
DataTypePtr getReturnTypeImpl(const ColumnsWithTypeAndName & arguments)
const override
{
const size_t number_of_arguments = arguments.size();
-
if (number_of_arguments < 2 || number_of_arguments > 3)
throw Exception(
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
"Number of arguments for function {} doesn't match: passed {},
should be 2 or 3",
getName(),
number_of_arguments);
-
std::vector<bool> arrays_is_nullable;
DataTypePtr input_type = arguments[0].type;
while (const DataTypeArray * array =
checkAndGetDataType<DataTypeArray>(removeNullable(input_type).get()))
@@ -108,9 +106,6 @@ public:
if (*it)
return_type = makeNullable(return_type);
}
-
- // std::cout << "return_type:" << return_type->getName() <<
std::endl;
-
return return_type;
}
else
@@ -163,7 +158,6 @@ public:
return arguments[2].column;
ColumnPtr res = input_col_as_tuple->getColumns()[index.value()];
-
/// Wrap into Nullable if needed
if (input_col_as_nullable_tuple)
{
diff --git a/cpp-ch/local-engine/Parser/FunctionParser.cpp
b/cpp-ch/local-engine/Parser/FunctionParser.cpp
index 4bd35cb8c9..3d9da46ad1 100644
--- a/cpp-ch/local-engine/Parser/FunctionParser.cpp
+++ b/cpp-ch/local-engine/Parser/FunctionParser.cpp
@@ -100,7 +100,6 @@ std::pair<DataTypePtr, Field>
FunctionParser::parseLiteral(const substrait::Expr
ActionsDAG::NodeRawConstPtrs
FunctionParser::parseFunctionArguments(const
substrait::Expression_ScalarFunction & substrait_func, ActionsDAG &
actions_dag) const
{
- ActionsDAG::NodeRawConstPtrs parsed_args;
return expression_parser->parseFunctionArguments(actions_dag,
substrait_func);
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
index ccd697eb72..ea24616ef3 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
@@ -325,6 +325,7 @@ trait SparkPlanExecApi {
*
* childOutputAttributes may be different from outputAttributes, for
example, the
* childOutputAttributes include additional shuffle key columns
+ *
* @return
*/
// scalastyle:off argcount
@@ -711,4 +712,12 @@ trait SparkPlanExecApi {
numElements: BigInt,
outputAttributes: Seq[Attribute],
child: Seq[SparkPlan]): ColumnarRangeBaseExec
+
+ def expressionFlattenSupported(expr: Expression): Boolean = false
+
+ def genFlattenedExpressionTransformer(
+ substraitName: String,
+ children: Seq[ExpressionTransformer],
+ expr: Expression): ExpressionTransformer =
+ GenericExpressionTransformer(substraitName, children, expr)
}
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
index d7f729f4d2..d1f3462564 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala
@@ -47,13 +47,14 @@ abstract class FilterExecTransformerBase(val cond:
Expression, val input: SparkP
BackendsApiManager.getMetricsApiInstance.genFilterTransformerMetrics(sparkContext)
// Split out all the IsNotNulls from condition.
- private val (notNullPreds, _) = splitConjunctivePredicates(cond).partition {
+ protected val (notNullPreds, _) = splitConjunctivePredicates(cond).partition
{
case IsNotNull(a) => isNullIntolerant(a) &&
a.references.subsetOf(child.outputSet)
case _ => false
}
// The columns that will filtered out by `IsNotNull` could be considered as
not nullable.
- private val notNullAttributes =
notNullPreds.flatMap(_.references).distinct.map(_.exprId)
+ protected val notNullAttributes: Seq[ExprId] =
+ notNullPreds.flatMap(_.references).distinct.map(_.exprId)
override def isNullIntolerant(expr: Expression): Boolean = expr match {
case e: NullIntolerant => e.children.forall(isNullIntolerant)
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
index 48f33c3d50..04a4e47e0b 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
@@ -109,6 +109,19 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
}
}
+ private def replaceFlattenedExpressionWithExpressionTransformer(
+ substraitName: String,
+ expr: Expression,
+ attributeSeq: Seq[Attribute],
+ expressionsMap: Map[Class[_], String]): ExpressionTransformer = {
+ val children =
+ expr.children.map(replaceWithExpressionTransformer0(_, attributeSeq,
expressionsMap))
+
BackendsApiManager.getSparkPlanExecApiInstance.genFlattenedExpressionTransformer(
+ substraitName,
+ children,
+ expr)
+ }
+
private def genRescaleDecimalTransformer(
substraitName: String,
b: BinaryArithmetic,
@@ -742,6 +755,12 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
substraitExprName,
expr.children.map(replaceWithExpressionTransformer0(_, attributeSeq,
expressionsMap)),
j)
+ case ce if
BackendsApiManager.getSparkPlanExecApiInstance.expressionFlattenSupported(ce) =>
+ replaceFlattenedExpressionWithExpressionTransformer(
+ substraitExprName,
+ ce,
+ attributeSeq,
+ expressionsMap)
case expr =>
GenericExpressionTransformer(
substraitExprName,
diff --git
a/shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
b/shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
index 8c0ce01b5c..b6cd5de9ae 100644
--- a/shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
+++ b/shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala
@@ -355,12 +355,10 @@ class GlutenConfig(conf: SQLConf) extends Logging {
def autoAdjustStageFallenNodeThreshold: Double =
getConf(AUTO_ADJUST_STAGE_RESOURCES_FALLEN_NODE_RATIO_THRESHOLD)
-
def parquetEncryptionValidationFileLimit: Int =
getConf(ENCRYPTED_PARQUET_FALLBACK_FILE_LIMIT)
-
def enableColumnarRange: Boolean = getConf(COLUMNAR_RANGE_ENABLED)
-
def enableColumnarCollectLimit: Boolean =
getConf(COLUMNAR_COLLECT_LIMIT_ENABLED)
+ def getSupportedFlattenedExpressions: String =
getConf(GLUTEN_SUPPORTED_FLATTENED_FUNCTIONS)
}
object GlutenConfig {
@@ -695,6 +693,13 @@ object GlutenConfig {
.stringConf
.createWithDefault("")
+ val GLUTEN_SUPPORTED_FLATTENED_FUNCTIONS =
+ buildConf("spark.gluten.sql.supported.flattenNestedFunctions")
+ .internal()
+ .doc("Flatten nested functions as one for optimization.")
+ .stringConf
+ .createWithDefault("and,or");
+
val GLUTEN_SOFT_AFFINITY_ENABLED =
buildConf("spark.gluten.soft-affinity.enabled")
.doc("Whether to enable Soft Affinity scheduling.")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
