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

Support round and bround SQL functions #1244

Merged
merged 16 commits into from
Jan 8, 2021
Merged

Support round and bround SQL functions #1244

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
c9c6a72
Support round and bround
nartal1 Nov 18, 2020
4af74eb
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Dec 2, 2020
fbb4a1e
remove unused import
nartal1 Dec 2, 2020
3c87a5f
Rework to use GpuBinaryExpression
nartal1 Dec 8, 2020
ac2f729
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Dec 8, 2020
c4185fc
update round operators to take in decimal config
nartal1 Dec 8, 2020
2bf3824
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Dec 10, 2020
8dff4b6
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Dec 11, 2020
9318898
addressed review comments
nartal1 Dec 14, 2020
ac14945
updated tests
nartal1 Dec 16, 2020
a9f7614
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Dec 18, 2020
a729637
fix build after upmerge
nartal1 Dec 18, 2020
cd8b85f
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Jan 5, 2021
a4cfb2d
upmerge and update configs.md, supported_ops.md
nartal1 Jan 5, 2021
40fa6be
Merge branch 'branch-0.4' of github.com:NVIDIA/spark-rapids into brou…
nartal1 Jan 6, 2021
80a5973
addressed review comments
nartal1 Jan 6, 2021
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
20 changes: 20 additions & 0 deletions integration_tests/src/main/python/arithmetic_ops_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,26 @@ def test_shift_right_unsigned(data_gen):
'shiftrightunsigned(a, cast(null as INT))',
'shiftrightunsigned(a, b)'))

@approximate_float
@pytest.mark.parametrize('data_gen', round_gens, ids=idfn)
def test_decimal_bround(data_gen):
assert_gpu_and_cpu_are_equal_collect(
lambda spark: debug_df(unary_op_df(spark, data_gen).selectExpr(
'bround(a)',
#'bround(a, -2)',
'bround(a, 10)')),
conf=allow_negative_scale_of_decimal_conf)

@approximate_float
@pytest.mark.parametrize('data_gen', round_gens, ids=idfn)
def test_decimal_round(data_gen):
assert_gpu_and_cpu_are_equal_collect(
lambda spark: debug_df(unary_op_df(spark, data_gen).selectExpr(
'round(a)',
#'round(a, -2)',
'round(a, 10)')),
conf=allow_negative_scale_of_decimal_conf)

@approximate_float
@pytest.mark.parametrize('data_gen', double_gens, ids=idfn)
def test_cbrt(data_gen):
Expand Down
3 changes: 3 additions & 0 deletions integration_tests/src/main/python/data_gen.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -753,6 +753,9 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False):
# Include decimal type while testing equalTo and notEqualTo
eq_gens_with_decimal_gen = eq_gens + decimal_gens

#gen for testing round operator
round_gens = numeric_gens + decimal_gens

date_gens = [date_gen]
date_n_time_gens = [date_gen, timestamp_gen]

Expand Down
6 changes: 5 additions & 1 deletion sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

package com.nvidia.spark.rapids

import ai.rapids.cudf.{BinaryOp, BinaryOperable, ColumnVector, DType, Scalar, UnaryOp}
import ai.rapids.cudf.{BinaryOp, BinaryOperable, ColumnVector, DType, RoundMode, Scalar, UnaryOp}
import com.nvidia.spark.rapids.RapidsPluginImplicits._

import org.apache.spark.sql.catalyst.expressions._
Expand Down Expand Up @@ -148,6 +148,10 @@ trait CudfUnaryExpression extends GpuUnaryExpression {
override def doColumnar(input: GpuColumnVector): ColumnVector = input.getBase.unaryOp(unaryOp)
}

trait GpuRoundBase extends GpuBinaryExpression {
revans2 marked this conversation as resolved.
Show resolved Hide resolved
def roundMode: RoundMode
}

trait GpuBinaryExpression extends BinaryExpression with GpuExpression {

def doColumnar(lhs: GpuColumnVector, rhs: GpuColumnVector): ColumnVector
Expand Down
16 changes: 16 additions & 0 deletions sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1586,6 +1586,22 @@ object GpuOverrides {

override def convertToGpu(child: Expression): GpuExpression = GpuAverage(child)
}),
expr[BRound](
"Round an expression to d decimal places using HALF_EVEN rounding mode",
(a, conf, p, r) => new BinaryExprMeta[BRound](a, conf, p, r) {
override def isSupportedType(t: DataType): Boolean =
GpuOverrides.isSupportedType(t, allowDecimal = conf.decimalTypeEnabled)
override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression =
GpuBRound(lhs, rhs)
}),
expr[Round](
"Round an expression to d decimal places using HALF_UP rounding mode",
(a, conf, p, r) => new BinaryExprMeta[Round](a, conf, p, r) {
override def isSupportedType(t: DataType): Boolean =
GpuOverrides.isSupportedType(t, allowDecimal = conf.decimalTypeEnabled)
override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression =
GpuRound(lhs, rhs)
}),
expr[PythonUDF](
"UDF run in an external python process. Does not actually run on the GPU, but " +
"the transfer of data to/from it can be accelerated.",
Expand Down
71 changes: 68 additions & 3 deletions sql-plugin/src/main/scala/org/apache/spark/sql/rapids/mathExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,11 @@ package org.apache.spark.sql.rapids

import java.io.Serializable

import ai.rapids.cudf.{BinaryOp, ColumnVector, DType, Scalar, UnaryOp}
import com.nvidia.spark.rapids.{Arm, CudfBinaryExpression, CudfUnaryExpression, FloatUtils, GpuColumnVector, GpuUnaryExpression}
import ai.rapids.cudf.{BinaryOp, ColumnVector, DType, RoundMode, Scalar, UnaryOp}
import com.nvidia.spark.rapids.{Arm, CudfBinaryExpression, CudfUnaryExpression, FloatUtils, GpuBinaryExpression, GpuColumnVector, GpuExpression, GpuUnaryExpression}
import com.nvidia.spark.rapids.RapidsPluginImplicits.ReallyAGpuExpression

import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes}
import org.apache.spark.sql.catalyst.expressions.{EmptyRow, Expression, ImplicitCastInputTypes}
import org.apache.spark.sql.types._

abstract class CudfUnaryMathExpression(name: String) extends GpuUnaryMathExpression(name)
Expand Down Expand Up @@ -349,6 +350,70 @@ abstract class CudfBinaryMathExpression(name: String) extends CudfBinaryExpressi
override def dataType: DataType = DoubleType
}

abstract class GpuRoundBase(child: Expression, scale: Expression) extends GpuBinaryExpression
with Serializable with ImplicitCastInputTypes {

override def left: Expression = child
override def right: Expression = scale

def roundMode: RoundMode

override lazy val dataType: DataType = child.dataType match {
// if the new scale is bigger which means we are scaling up,
// keep the original scale as `Decimal` does
case DecimalType.Fixed(p, s) => DecimalType(p, if (_scale > s) s else _scale)
case t => t
}

// Avoid repeated evaluation since `scale` is a constant int,
// avoid unnecessary `child` evaluation in both codegen and non-codegen eval
// by checking if scaleV == null as well.
private lazy val scaleV: Any = scale match {
case _: GpuExpression => scale.columnarEval(null)
case _ => scale.eval(EmptyRow)
}
private lazy val _scale: Int = scaleV.asInstanceOf[Int]

override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, IntegerType)

override def doColumnar(val0: GpuColumnVector, val1: Scalar): ColumnVector = {
revans2 marked this conversation as resolved.
Show resolved Hide resolved
val scaleVal=val1.getInt
revans2 marked this conversation as resolved.
Show resolved Hide resolved
val scale = dataType match {
revans2 marked this conversation as resolved.
Show resolved Hide resolved
case DecimalType.Fixed(p, s) => s
case ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType => val1.getInt
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@revans2 Could you please suggest how we handle overflow when for each types.
For example(considering short type), pyspark results in 0 for the below round operation:

>>> df2= spark.createDataFrame(data=[32562], schema=ShortType())
>>> ret = df2.selectExpr('round(value, -5)')
>>> ret.show()
+----------------+
|round(value, -5)|
+----------------+
|               0|
+----------------+

But we see different GPU result(-31072) as overflow results in undefined behavior in libcudf. 
Should we throw an exception whenever we sense an overflow for each type at this point ?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem is completely in how we implement round/bround vs how spark does it, and I am not 100% sure how to make them sync up without a lot of work on the cudf side for these corner cases.

cudf tries to do the round on the native type, which can result in an overflow. Spark will convert the native value to a decimal value (128-bits if needed), set the scale to do the rounding, and then convert the value back (with some special cases for NaN and Infinite in floating point).

https://github.com/apache/spark/blob/0626901bcbeebceb6937001e1f32934c71876210/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala#L1220-L1249

There can be no overflow in those cases because all of the processing is happening on 128-bits. For integer smaller than a long we could cast it to a long first, do the round/bround, and then cast it back. But we would still end up with issues in long because of overflow.

Similar with float/double.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have tests for overflow to check if it is working correctly or are we going to mark the operators and incompatible until we can figure out a way to make it work properly?

case _ => throw new IllegalArgumentException(s"Round operator doesn't support $dataType")
}
val lhs = val0.getBase
lhs.round(scale, roundMode)
}

override def doColumnar(lhs: GpuColumnVector, rhs: GpuColumnVector): ColumnVector = {
throw new IllegalArgumentException("lhs has to be a vector and rhs has to be a scalar for " +
"the round operator to work")
}

override def doColumnar(lhs: Scalar, rhs: GpuColumnVector): ColumnVector = {
throw new IllegalArgumentException("lhs has to be a vector and rhs has to be a scalar for " +
"the round operator to work")
}

override def doColumnar(numRows: Int, lhs: Scalar, rhs: Scalar): ColumnVector = {
withResource(GpuColumnVector.from(lhs, numRows, left.dataType)) { expandedLhs =>
doColumnar(expandedLhs, rhs)
}
}
}

case class GpuBRound(child: Expression, scale: Expression) extends
GpuRoundBase(child, scale) {
override def roundMode: RoundMode = RoundMode.HALF_EVEN
}

case class GpuRound(child: Expression, scale: Expression) extends
GpuRoundBase(child, scale) {
override def roundMode: RoundMode = RoundMode.HALF_UP
}

case class GpuPow(left: Expression, right: Expression)
extends CudfBinaryMathExpression("POWER") {
override def binaryOp: BinaryOp = BinaryOp.POW
Expand Down