You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "zhztheplayer (via GitHub)" <gi...@apache.org> on 2024/05/08 06:42:13 UTC
[PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
zhztheplayer opened a new pull request, #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655
Velox backend's collect_list / collect_set implementations require for ARRAY intermediate data however Spark uses BINARY. To address this we did some tricks to forcibly modify the physical plan to modify the output schema of partial aggregate operator to align with Velox, but that way the actual execution procedure for the two functions in Velox backend is still hidden from query plan so advanced optimizations or compatibility checks are made difficult during planning phase.
This patch adds new functions velox_collect_list / velox_collect_set to correctly map to Velox backend's implementation for the two functions and does essential code cleanup and refactors.
1. Add functions velox_collect_list / velox_collect_set.
2. Remove physical rule `RewriteCollect` / `RewriteTypedImperativeAggregate`, add logical rule `CollectRewriteRule` to incorporate functionalities of the former.
3. `CollectRewriteRule` will replace collect_list / collect_set to velox_collect_list / velox_collect_set.
4. Since velox_collect_list / velox_collect_set becomes `DeclarativeAggregate`, some UTs should be disabled as they do some plan checks for existence of `ObjectHashAggregateExec`.
5. Some optimizations for UT facility code.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597370680
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
+ out
+ }
+ case other => other
+ }
+ }
+}
+
+object CollectRewriteRule {
+ private def ensureNonNull(expr: Expression): Expression = {
+ val out =
+ Coalesce(List(expr, Literal.create(Seq.empty, expr.dataType)))
+ assert(!out.nullable)
+ assert(!out.dataType.asInstanceOf[ArrayType].containsNull)
+ out
+ }
+
+ private object ToVeloxCollectSet {
+ def unapply(expr: Expression): Option[Expression] = expr match {
+ case aggFunc @ AggregateExpression(s: CollectSet, _, _, filter, _) if has[VeloxCollectSet] =>
+ val newFilter = (filter ++ Some(IsNotNull(s.child))).reduceOption(And)
+ val newAggFunc =
+ aggFunc.copy(aggregateFunction = VeloxCollectSet(s.child), filter = newFilter)
+ Some(newAggFunc)
+ case _ => None
+ }
+ }
+
+ private def has[T <: Expression: ClassTag]: Boolean = {
Review Comment:
I agree with removing the check. Although we can do that later in some version...
For example `VeloxCollectSet` is still expensive than vanilla Spark so removing this check may cause a user who disabled `collect_set` experience performance regression.
BTW I'll raise some optimizations on `VeloxCollectSet` after this patch is merged. When I am confident with vanilla Spark + velox_collect_list / velox_collect_set being used in production enough, I'll remove this check.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596439098
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
Use ArrayUnion for updateExpressions and mergeExpressions ?
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
+ override def prettyName: String = "velox_collect_set"
+
+ override def nullable: Boolean = true
Review Comment:
can we preserve the comment ?
https://github.com/apache/incubator-gluten/blob/5d8ac72966b9c23870654fe56ffe276579c93c15/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/rewrite/RewriteCollect.scala#L101-L103
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
Review Comment:
unnecessary variable
##########
backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala:
##########
@@ -106,15 +112,44 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
}
}
- // java.lang.NullPointerException
- ignore("fallback final aggregate of collect_list") {
+ test("fallback collect_list") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_list"
+ ) {
+ CollectRewriteRule.forceEnableAndRun {
+ val df = spark.sql("SELECT c1, collect_list(c2) FROM tmp3 GROUP BY c1")
+ val result = df.collect()
+ assert(result.length == 3)
+ assert(result.map(_.getList[Int](1).size()).sum == 100)
+ val plan = df.queryExecution.executedPlan
+ assert(collect(plan) { case v: HashAggregateExecBaseTransformer => v }.isEmpty)
+ }
+ }
+ }
+
+ test("fallback collect_set") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_set"
+ ) {
+ CollectRewriteRule.forceEnableAndRun {
Review Comment:
why do we need to force enable the rule ? Should it it already enabled if have collect_set ?
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
Review Comment:
what does this check for ? I did not see Spark call it at optimizer phase...
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
Review Comment:
shall we `ensureNonNull` for newAggFunc rather than window func ?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596496889
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
It internally uses Array. A single de-dup operation should cost ~O(n). I would do a distinct at end of aggregation.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596523132
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
Review Comment:
I usually dig the internal method returned value.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596323725
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
Sure, let me add more test against this case.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597342278
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
Review Comment:
It's evaluated with the project created in `WindoeExecBase#createResultProjection`. Velox's window implementation doesn't support `collect_set` yet. I'd add a case for vanilla Spark + velox_collect_set if you are concerned about this part of code.
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
Review Comment:
It's evaluated with the project created in `WindowExecBase#createResultProjection`. Velox's window implementation doesn't support `collect_set` yet. I'd add a case for vanilla Spark + velox_collect_set if you are concerned about this part of code.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596431877
##########
backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala:
##########
@@ -106,15 +112,38 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
}
}
- // java.lang.NullPointerException
- ignore("fallback final aggregate of collect_list") {
+ test("fallback collect_list") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_list"
Review Comment:
Added a set of cases here https://github.com/apache/incubator-gluten/pull/5655/files#diff-742ce9027f0d774d324125a130d9993bad09d4ea5eb6d35d791b39bb57138c77R983-R1057 for better coverage.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2104125254
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2100052781
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2100058191
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099853312
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2101713967
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2105505009
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597372264
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
Isn't binary data only able to obtain when it's a `ImperativeAggregate`? Would you elaborate on the suggestion?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2101733423
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "PHILO-HE (via GitHub)" <gi...@apache.org>.
PHILO-HE commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596181272
##########
backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala:
##########
@@ -106,15 +112,38 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
}
}
- // java.lang.NullPointerException
- ignore("fallback final aggregate of collect_list") {
+ test("fallback collect_list") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_list"
Review Comment:
Is it possible to only make partial fallback happen in the test? Maybe, needs a config to control.
##########
gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala:
##########
@@ -279,4 +283,55 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS
randn(Random.nextLong())
).foreach(assertNoExceptions)
}
+
+ Seq(true, false).foreach {
+ value =>
+ testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") {
Review Comment:
Do we need to also add this rewritten test for other spark versions? Generally, we do that even though duplicate code is introduced.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596208826
##########
gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala:
##########
@@ -279,4 +283,55 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS
randn(Random.nextLong())
).foreach(assertNoExceptions)
}
+
+ Seq(true, false).foreach {
+ value =>
+ testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") {
Review Comment:
> Do we need to also add this rewritten test for other spark versions? Generally, we do that even though duplicate code is introduced.
For this case I felt it's OK to just test with one version (as long as it's newer version, e.g., 3.4, 3.5). `collect_list` / `collect_set` in Spark shouldn't have actual differences among these 4 Spark versions. Having it tested with all Spark version definitely provides better coverage but we can also make trade-off taking maintenance effort into account.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103637430
@liujiayi771
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2100058237
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596211736
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
> For each row, we wrap a array. Will it cause performance regression with vanilla Spark ?
Yes that's possible. But I assume it should be a comparatively rare case when user enables Gluten however these two functions are fallen back. So I think it's fine until getting real complaints from users.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2105506109
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2102039798
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597365504
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
+ out
+ }
+ case other => other
+ }
+ }
+}
+
+object CollectRewriteRule {
+ private def ensureNonNull(expr: Expression): Expression = {
+ val out =
+ Coalesce(List(expr, Literal.create(Seq.empty, expr.dataType)))
+ assert(!out.nullable)
+ assert(!out.dataType.asInstanceOf[ArrayType].containsNull)
+ out
+ }
+
+ private object ToVeloxCollectSet {
+ def unapply(expr: Expression): Option[Expression] = expr match {
+ case aggFunc @ AggregateExpression(s: CollectSet, _, _, filter, _) if has[VeloxCollectSet] =>
+ val newFilter = (filter ++ Some(IsNotNull(s.child))).reduceOption(And)
+ val newAggFunc =
+ aggFunc.copy(aggregateFunction = VeloxCollectSet(s.child), filter = newFilter)
+ Some(newAggFunc)
+ case _ => None
+ }
+ }
+
+ private def has[T <: Expression: ClassTag]: Boolean = {
Review Comment:
After droping test, do we still need this check ? It seems this rule is only added by velox backend and the same is VeloxCollectSet. So when we going to this rule, there must have VeloxCollectSet ?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099887860
/Benchmark Velox
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "GlutenPerfBot (via GitHub)" <gi...@apache.org>.
GlutenPerfBot commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099979502
===== Performance report for TPCH SF2000 with Velox backend, for reference only ====
<table border="1" cellpadding="0" cellspacing="0">
<tbody>
<tr>
<td><font color=#000000>query</font></td>
<td><font color=#000000>log/native_5655_time.csv</font></td>
<td><font color=#000000>log/native_master_05_07_2024_254d62e72_time.csv</font></td>
<td><font color=#000000>difference</font></td>
<td><font color=#000000>percentage</font></td>
</tr>
<tr>
<td><font color=#F92663>q1</font></td>
<td><font color=#F92663>34.16</font></td>
<td><font color=#F92663>34.69</font></td>
<td><font color=#F92663>0.528</font></td>
<td><font color=#F92663>101.55%</font></td>
</tr>
<tr>
<td><font color=#F92663>q2</font></td>
<td><font color=#F92663>22.17</font></td>
<td><font color=#F92663>23.52</font></td>
<td><font color=#F92663>1.349</font></td>
<td><font color=#F92663>106.08%</font></td>
</tr>
<tr>
<td><font color=#F92663>q3</font></td>
<td><font color=#F92663>36.72</font></td>
<td><font color=#F92663>39.12</font></td>
<td><font color=#F92663>2.402</font></td>
<td><font color=#F92663>106.54%</font></td>
</tr>
<tr>
<td><font color=#F92663>q4</font></td>
<td><font color=#F92663>38.01</font></td>
<td><font color=#F92663>38.99</font></td>
<td><font color=#F92663>0.982</font></td>
<td><font color=#F92663>102.58%</font></td>
</tr>
<tr>
<td><font color=#F92663>q5</font></td>
<td><font color=#F92663>69.14</font></td>
<td><font color=#F92663>70.47</font></td>
<td><font color=#F92663>1.331</font></td>
<td><font color=#F92663>101.93%</font></td>
</tr>
<tr>
<td><font color=#F92663>q6</font></td>
<td><font color=#F92663>5.85</font></td>
<td><font color=#F92663>7.99</font></td>
<td><font color=#F92663>2.145</font></td>
<td><font color=#F92663>136.68%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q7</font></td>
<td><font color=#6F9915>82.92</font></td>
<td><font color=#6F9915>82.58</font></td>
<td><font color=#6F9915>-0.335</font></td>
<td><font color=#6F9915>99.60%</font></td>
</tr>
<tr>
<td><font color=#F92663>q8</font></td>
<td><font color=#F92663>84.23</font></td>
<td><font color=#F92663>85.88</font></td>
<td><font color=#F92663>1.654</font></td>
<td><font color=#F92663>101.96%</font></td>
</tr>
<tr>
<td><font color=#F92663>q9</font></td>
<td><font color=#F92663>124.80</font></td>
<td><font color=#F92663>125.69</font></td>
<td><font color=#F92663>0.890</font></td>
<td><font color=#F92663>100.71%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q10</font></td>
<td><font color=#6F9915>46.97</font></td>
<td><font color=#6F9915>44.17</font></td>
<td><font color=#6F9915>-2.799</font></td>
<td><font color=#6F9915>94.04%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q11</font></td>
<td><font color=#6F9915>20.58</font></td>
<td><font color=#6F9915>19.52</font></td>
<td><font color=#6F9915>-1.062</font></td>
<td><font color=#6F9915>94.84%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q12</font></td>
<td><font color=#6F9915>27.60</font></td>
<td><font color=#6F9915>25.07</font></td>
<td><font color=#6F9915>-2.524</font></td>
<td><font color=#6F9915>90.85%</font></td>
</tr>
<tr>
<td><font color=#F92663>q13</font></td>
<td><font color=#F92663>53.40</font></td>
<td><font color=#F92663>54.78</font></td>
<td><font color=#F92663>1.379</font></td>
<td><font color=#F92663>102.58%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q14</font></td>
<td><font color=#6F9915>18.46</font></td>
<td><font color=#6F9915>16.46</font></td>
<td><font color=#6F9915>-1.993</font></td>
<td><font color=#6F9915>89.20%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q15</font></td>
<td><font color=#6F9915>31.18</font></td>
<td><font color=#6F9915>29.15</font></td>
<td><font color=#6F9915>-2.027</font></td>
<td><font color=#6F9915>93.50%</font></td>
</tr>
<tr>
<td><font color=#F92663>q16</font></td>
<td><font color=#F92663>13.60</font></td>
<td><font color=#F92663>14.27</font></td>
<td><font color=#F92663>0.674</font></td>
<td><font color=#F92663>104.96%</font></td>
</tr>
<tr>
<td><font color=#F92663>q17</font></td>
<td><font color=#F92663>102.39</font></td>
<td><font color=#F92663>104.34</font></td>
<td><font color=#F92663>1.951</font></td>
<td><font color=#F92663>101.91%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q18</font></td>
<td><font color=#6F9915>148.53</font></td>
<td><font color=#6F9915>144.89</font></td>
<td><font color=#6F9915>-3.634</font></td>
<td><font color=#6F9915>97.55%</font></td>
</tr>
<tr>
<td><font color=#F92663>q19</font></td>
<td><font color=#F92663>13.42</font></td>
<td><font color=#F92663>15.16</font></td>
<td><font color=#F92663>1.737</font></td>
<td><font color=#F92663>112.95%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q20</font></td>
<td><font color=#6F9915>27.16</font></td>
<td><font color=#6F9915>26.39</font></td>
<td><font color=#6F9915>-0.771</font></td>
<td><font color=#6F9915>97.16%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q21</font></td>
<td><font color=#6F9915>289.52</font></td>
<td><font color=#6F9915>283.25</font></td>
<td><font color=#6F9915>-6.273</font></td>
<td><font color=#6F9915>97.83%</font></td>
</tr>
<tr>
<td><font color=#6F9915>q22</font></td>
<td><font color=#6F9915>16.11</font></td>
<td><font color=#6F9915>14.47</font></td>
<td><font color=#6F9915>-1.643</font></td>
<td><font color=#6F9915>89.80%</font></td>
</tr>
<tr>
<td><font color=#6F9915>total</font></td>
<td><font color=#6F9915>1306.89</font></td>
<td><font color=#6F9915>1300.85</font></td>
<td><font color=#6F9915>-6.038</font></td>
<td><font color=#6F9915>99.54%</font></td>
</tr>
</tbody>
</table>
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596502498
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
Review Comment:
I didn't find a proper way for placing breakpoint for debugging code without doing this. Do you know some?
If there is not a good way, Personally I would keep something like this and it doesn't increase code complexity.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596188187
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
For each row, we wrap a array. Will it cause performance regression with vanilla Spark ?
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
Does the semaintics of `ArrayUnion` equal to `Collect` ?
- `ArrayUnion` always returns distinct value in an array
- `ArrayUnion` allows null but CollectList/CollectSet do not allow
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103979937
> @zhztheplayer Do you think this approach is better, or is it better for us to convert it to binary in the project during row extraction after partial agg?
I am not 100% sure but seems to be that the PR's approach looks more portable? Row extraction would work but if it's done at operator level, the function still outputs ARRAY intermediate data which doesn't match Spark's CollectList / CollectSet definition.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597340489
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
+ override def prettyName: String = "velox_collect_set"
+
+ override def nullable: Boolean = true
Review Comment:
Added some comment to VeloxCollectSet.
For now we don't need the `withNullability` part of the comment.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103833029
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596310713
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
Instead of adding more cutsom Gluten expression, cann't we just add project to adapt data type (convert array to binary, etc) in columnar rule ?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596500210
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
Review Comment:
That doesn't work. Spark has some checks in `checkAnalysis` to enforce "WindowExpression(WindowFunction, ...)" pattern.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596208826
##########
gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala:
##########
@@ -279,4 +283,55 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS
randn(Random.nextLong())
).foreach(assertNoExceptions)
}
+
+ Seq(true, false).foreach {
+ value =>
+ testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") {
Review Comment:
> Do we need to also add this rewritten test for other spark versions? Generally, we do that even though duplicate code is introduced.
For this case I felt it's OK to just test with one version (as long as it's newer version, e.g., 3.4, 3.5). `collect_list` / `collect_set` in Spark shouldn't have actual differences among different Spark versions. Having it tested with all Spark version definitely provides better coverage but we can also make trade-off taking maintenance effort into account.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596516354
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
It's mainly for reduce shuffle data if the partial mode aggreagte fallback.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2105630103
Thank you all for reviewing!
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2102035458
/Benchmark Velox
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2102033961
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099857922
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099860435
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099881029
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596222302
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
So [concat](https://spark.apache.org/docs/latest/api/sql/index.html#concat) could be the right choice here?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596504202
##########
backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala:
##########
@@ -106,15 +112,44 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
}
}
- // java.lang.NullPointerException
- ignore("fallback final aggregate of collect_list") {
+ test("fallback collect_list") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_list"
+ ) {
+ CollectRewriteRule.forceEnableAndRun {
+ val df = spark.sql("SELECT c1, collect_list(c2) FROM tmp3 GROUP BY c1")
+ val result = df.collect()
+ assert(result.length == 3)
+ assert(result.map(_.getList[Int](1).size()).sum == 100)
+ val plan = df.queryExecution.executedPlan
+ assert(collect(plan) { case v: HashAggregateExecBaseTransformer => v }.isEmpty)
+ }
+ }
+ }
+
+ test("fallback collect_set") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_set"
+ ) {
+ CollectRewriteRule.forceEnableAndRun {
Review Comment:
I'll remove this API and the two tests. They are not useful for now.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2104259779
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596538120
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
Yes, It costs more space. May be we can use Spark Map type for collect_set?
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
Yes, it costs more space. May be we can use Spark Map type for collect_set?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596532823
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
+ }
+ case w: Window =>
+ w.transformExpressions {
+ case func @ WindowExpression(ToVeloxCollectSet(newAggFunc), _) =>
+ val out = ensureNonNull(func.copy(newAggFunc))
Review Comment:
I'm not sure adding ensureNonNull for window expression is been actually evaluated. It seems window operator only collects the window expression to eval. Is there a test for window + collect_set with null input ?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099862054
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2099853027
<!--
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/LICEGJ-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.
-->
Thanks for opening a pull request!
Could you open an issue for this pull request on Github Issues?
https://github.com/apache/incubator-gluten/issues
Then could you also rename ***commit message*** and ***pull request title*** in the following format?
[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}
See also:
* [Other pull requests](https://github.com/apache/incubator-gluten/pulls/)
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2102046841
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103824355
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596405895
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
@ulysses-you Please take a look again.
Change:
https://github.com/apache/incubator-gluten/pull/5655/files#diff-1ebbabef839649a226bf6bed298759976ab8adafe31d3dbfe07c559a1a95842dR33-R38
Test case:
https://github.com/apache/incubator-gluten/pull/5655/files#diff-742ce9027f0d774d324125a130d9993bad09d4ea5eb6d35d791b39bb57138c77R983-R1057
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2104128442
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103681867
@PHILO-HE
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596209504
##########
backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala:
##########
@@ -106,15 +112,38 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
}
}
- // java.lang.NullPointerException
- ignore("fallback final aggregate of collect_list") {
+ test("fallback collect_list") {
+ withSQLConf(
+ GlutenConfig.EXPRESSION_BLACK_LIST.key -> "collect_list"
Review Comment:
One of the following tests is for partial fallback:
https://github.com/apache/incubator-gluten/pull/5655/files/c40a5ad359fe14b030f85627e0d983a2848c8bd6#diff-493b5fa71a5befc5adcd67b51e47e802b41870287619a7b752bbf85f53b1880cR139
Would you mean that?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596553628
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, AttributeReference, Concat, CreateArray, Expression, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ If(
+ IsNull(child),
+ buffer,
+ Concat(List(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))))
+ )
+
+ override lazy val mergeExpressions: Seq[Expression] = List(
+ Concat(List(buffer.left, buffer.right))
+ )
+
+ override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))
+}
+
+case class VeloxCollectSet(override val child: Expression) extends VeloxCollect {
Review Comment:
Is it possible to fully copy `Collect` code and change `serialize`, `deserialize` to adapt datatype ? It seems we can re-point binary to a UnsafeArrayData and then getArray.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2100054199
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Add velox_collect_list / velox_collect_set to distinguish with vanilla Spark's collect_list / collect_set [incubator-gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2100387450
CC @zhli1142015
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596304860
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
concat allows null but collect_list does not allow:
- the result is null
- the null element value in array
I think we should at least pass test `collect_set/collect_list with null` in `VeloxAggregateFunctionsSuite` with partially fallback.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596502498
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
+ out
+ }
+
+ private def replaceCollectList(node: LogicalPlan): LogicalPlan = {
+ node.transformExpressions {
+ case func @ AggregateExpression(l: CollectList, _, _, _, _) if has[VeloxCollectList] =>
+ func.copy(VeloxCollectList(l.child))
+ }
+ }
+
+ private def replaceCollectSet(node: LogicalPlan): LogicalPlan = {
+ // 1. Replace null result from VeloxCollectSet with empty array to align with
+ // vanilla Spark.
+ // 2. Filter out null inputs from VeloxCollectSet to align with vanilla Spark.
+ //
+ // Since https://github.com/apache/incubator-gluten/pull/4805
+ node match {
+ case agg: Aggregate =>
+ agg.transformExpressions {
+ case ToVeloxCollectSet(newAggFunc) =>
+ val out = ensureNonNull(newAggFunc)
+ out
Review Comment:
I didn't find a proper way for placing breakpoint on return value for debugging code without doing this. Do you know some?
If there is not a good way, Personally I would keep something like this and it doesn't increase code complexity.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2104262701
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103794412
> thank you @zhztheplayer , can you also provide the npe case to help reproduce ? It seems this pr did not contain that case ?
NPE can be repeated from current main code test by enabling some ignored tests.
See https://github.com/apache/incubator-gluten/pull/5649.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1596212647
##########
backends-velox/src/main/scala/org/apache/gluten/expression/aggregate/VeloxCollect.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.aggregate
+
+import org.apache.spark.sql.catalyst.expressions.{ArrayDistinct, ArrayUnion, AttributeReference, CreateArray, Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.types.{ArrayType, DataType}
+
+abstract class VeloxCollect extends DeclarativeAggregate with UnaryLike[Expression] {
+ protected lazy val buffer: AttributeReference = AttributeReference("buffer", dataType)()
+
+ override def dataType: DataType = ArrayType(child.dataType, false)
+
+ override def aggBufferAttributes: Seq[AttributeReference] = List(buffer)
+
+ override lazy val initialValues: Seq[Expression] = List(Literal.create(Seq.empty, dataType))
+
+ override lazy val updateExpressions: Seq[Expression] = List(
+ ArrayUnion(buffer, CreateArray(List(child), useStringTypeWhenEmpty = false))
Review Comment:
> ArrayUnion always returns distinct value in an array
Oops, if it does de-dup then `velox_collect_list` may return wrong result. I'll take a look on this.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "liujiayi771 (via GitHub)" <gi...@apache.org>.
liujiayi771 commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2103729011
@zhztheplayer Do you think this approach is better, or is it better for us to convert it to binary in the project during row extraction?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#discussion_r1597340017
##########
backends-velox/src/main/scala/org/apache/gluten/extension/CollectRewriteRule.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.expression.ExpressionMappings
+import org.apache.gluten.expression.aggregate.{VeloxCollectList, VeloxCollectSet}
+import org.apache.gluten.utils.LogicalPlanSelector
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{And, Coalesce, Expression, IsNotNull, Literal, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Window}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.ArrayType
+
+import scala.reflect.{classTag, ClassTag}
+
+/**
+ * Velox's collect_list / collect_set use array as intermediate data type so aren't compatible with
+ * vanilla Spark. We here replace the two functions with velox_collect_list / velox_collect_set to
+ * distinguish.
+ */
+case class CollectRewriteRule(spark: SparkSession) extends Rule[LogicalPlan] {
+ import CollectRewriteRule._
+ override def apply(plan: LogicalPlan): LogicalPlan = LogicalPlanSelector.maybe(spark, plan) {
+ val out = plan.transformUp {
+ case node =>
+ val out = replaceCollectSet(replaceCollectList(node))
+ out
+ }
+ if (out.fastEquals(plan)) {
+ return plan
+ }
+ spark.sessionState.analyzer.checkAnalysis(out)
Review Comment:
I'll try removing this. It should not be required with current approach either.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2105500318
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Fix NullPointerException when collect_list / collect_set are partially fallen back [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5655:
URL: https://github.com/apache/incubator-gluten/pull/5655#issuecomment-2105504803
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org