Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Example of HOST_UDF aggregation #11787

Draft
wants to merge 3 commits into
base: branch-25.02
Choose a base branch
from
Draft

Example of HOST_UDF aggregation #11787

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d2ae155
Example of HostUDF usage
ttnghia Nov 27, 2024
c17f007
Fix type
ttnghia Nov 27, 2024
4c5d91c
Fix compile error
ttnghia Dec 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions sql-plugin/src/main/scala/org/apache/spark/sql/rapids/aggregate/aggregateFunctions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import ai.rapids.cudf.{Aggregation128Utils, BinaryOp, ColumnVector, DType, Group
import com.nvidia.spark.rapids._
import com.nvidia.spark.rapids.Arm.withResource
import com.nvidia.spark.rapids.RapidsPluginImplicits.ReallyAGpuExpression
import com.nvidia.spark.rapids.jni.TestHostUDF
import com.nvidia.spark.rapids.shims.{GpuDeterministicFirstLastCollectShim, ShimExpression, TypeUtilsShims}
import com.nvidia.spark.rapids.window._

Expand Down Expand Up @@ -59,13 +60,15 @@ class CudfSum(override val dataType: DataType) extends CudfAggregate {
// sum(shorts): bigint
// Aggregate [sum(shorts#33) AS sum(shorts)#50L]
//
@transient lazy val rapidsSumType: DType = GpuColumnVector.getNonNestedRapidsType(dataType)
//@transient lazy val rapidsSumType: DType = GpuColumnVector.getNonNestedRapidsType(dataType)

override val reductionAggregate: cudf.ColumnVector => cudf.Scalar =
(col: cudf.ColumnVector) => col.sum(rapidsSumType)

(col: cudf.ColumnVector) => col.reduce(ReductionAggregation.hostUDF(
TestHostUDF.createTestHostUDF(TestHostUDF.AggregationType.Reduction)),
DType.FLOAT64)
override lazy val groupByAggregate: GroupByAggregation =
GroupByAggregation.sum()
GroupByAggregation.hostUDF(
TestHostUDF.createTestHostUDF(TestHostUDF.AggregationType.GroupByAggregation))

override val name: String = "CudfSum"
}
Expand Down Expand Up @@ -2025,7 +2028,7 @@ case class GpuVarianceSamp(child: Expression, nullOnDivideByZero: Boolean)
}

case class GpuReplaceNullmask(
input: Expression,
input: Expression,
mask: Expression) extends GpuExpression with ShimExpression {

override def dataType: DataType = input.dataType
Expand Down Expand Up @@ -2092,15 +2095,15 @@ abstract class GpuMaxMinByBase(valueExpr: Expression, orderingExpr: Expression)

protected val cudfMaxMinByAggregate: CudfAggregate

private lazy val bufferOrdering: AttributeReference =
private lazy val bufferOrdering: AttributeReference =
AttributeReference("ordering", orderingExpr.dataType)()

private lazy val bufferValue: AttributeReference =
AttributeReference("value", valueExpr.dataType)()

// Cudf allows only one column as input, so wrap value and ordering columns by
// a struct before just going into cuDF.
private def createStructExpression(order: Expression, value: Expression): Expression =
private def createStructExpression(order: Expression, value: Expression): Expression =
GpuReplaceNullmask(
GpuCreateNamedStruct(Seq(
GpuLiteral(CudfMaxMinBy.KEY_ORDERING, StringType), order,
Expand Down
Loading