This is an automated email from the ASF dual-hosted git repository.
philo 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 2721484ad1 [GLUTEN-7541][VL] Improve HLLRewriteRule for Velox (#7543)
2721484ad1 is described below
commit 2721484ad14245cd10e2d20d12a23bcac89a4afc
Author: Jiaan Geng <[email protected]>
AuthorDate: Fri Oct 18 16:09:33 2024 +0800
[GLUTEN-7541][VL] Improve HLLRewriteRule for Velox (#7543)
---
.../apache/gluten/extension/HLLRewriteRule.scala | 35 ++++++++++------------
1 file changed, 16 insertions(+), 19 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/extension/HLLRewriteRule.scala
b/backends-velox/src/main/scala/org/apache/gluten/extension/HLLRewriteRule.scala
index 2b17cbfd65..8b44005646 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/extension/HLLRewriteRule.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/extension/HLLRewriteRule.scala
@@ -25,37 +25,34 @@ import org.apache.spark.sql.catalyst.expressions.Literal
import
org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
HyperLogLogPlusPlus}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE,
AGGREGATE_EXPRESSION}
import org.apache.spark.sql.types._
case class HLLRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan =
LogicalPlanSelector.maybe(spark, plan) {
- plan.resolveOperatorsUp {
+ plan.transformUpWithPruning(_.containsPattern(AGGREGATE)) {
case a: Aggregate =>
- a.transformExpressions {
- case hllExpr @ AggregateExpression(hll: HyperLogLogPlusPlus, _, _,
_, _)
+
a.transformExpressionsWithPruning(_.containsPattern(AGGREGATE_EXPRESSION)) {
+ case aggExpr @ AggregateExpression(hll: HyperLogLogPlusPlus, _, _,
_, _)
if GlutenConfig.getConf.enableNativeHyperLogLogAggregateFunction
&&
GlutenConfig.getConf.enableColumnarHashAgg &&
- isDataTypeSupported(hll.child.dataType) =>
- AggregateExpression(
- HLLAdapter(
- hll.child,
- Literal(hll.relativeSD),
- hll.mutableAggBufferOffset,
- hll.inputAggBufferOffset),
- hllExpr.mode,
- hllExpr.isDistinct,
- hllExpr.filter,
- hllExpr.resultId
- )
+ isSupportedDataType(hll.child.dataType) =>
+ val hllAdapter = HLLAdapter(
+ hll.child,
+ Literal(hll.relativeSD),
+ hll.mutableAggBufferOffset,
+ hll.inputAggBufferOffset)
+
+ aggExpr.copy(aggregateFunction = hllAdapter)
}
}
}
- private def isDataTypeSupported(dataType: DataType): Boolean = {
- // HLL in velox only supports below data types. we should not offload HLL
to velox, if
+ private def isSupportedDataType(dataType: DataType): Boolean = {
+ // HLL in Velox only supports below data types. We should not offload HLL
to velox, if
// child's data type is not supported. This prevents the case only partail
agg is fallbacked.
- // As spark and velox have different HLL binary formats, HLL binary
generated by spark can't
- // be parsed by velox, it would cause the error: 'Unexpected type of HLL'.
+ // As Spark and Velox have different HLL binary formats, HLL binary
generated by Spark can't
+ // be parsed by Velox, it would cause the error: 'Unexpected type of HLL'.
dataType match {
case BooleanType | ByteType | ShortType | IntegerType | LongType |
FloatType | DoubleType |
StringType | _: CharType | _: DecimalType | DateType | TimestampType
| BinaryType =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
