Implement native function: round()

[richox]

Is your feature request related to a problem? Please describe.

currently, round() function has no native implementation because we found there are different behaviors from spark:

auron/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala

Line 818 in 9f03d94

// TODO: datafusion's round() has different behavior from spark

Describe the solution you'd like

implement native round() with the same behavior as spark (to verify this, we can bring some codes from spark's MathExpressionSuite.scala). we can use AI to directly translate spark's implementation to rust.

Describe alternatives you've considered

Additional context

[slfan1989]

@richox May I take this feature and work on it? Thanks a lot!

[slfan1989]

Background

I’m continuing to follow up on this issue and have completed part of the research work. I’m sharing the relevant information below to help clarify what improvements we need to make.

Spark

We can refer to the mathExpressions.scala class, where the Round function uses the ROUND_HALF_UP algorithm.

https://github.com/apache/spark/blob/18f0463a97bde206595a0a0d3a8c2b6d37d38975/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala#L1704-L1727

@ExpressionDescription(
  usage = "_FUNC_(expr, d) - Returns `expr` rounded to `d` decimal places using HALF_UP rounding mode.",
  examples = """
    Examples:
      > SELECT _FUNC_(2.5, 0);
       3
  """,
  since = "1.5.0",
  group = "math_funcs")
// scalastyle:on line.size.limit
case class Round(
    child: Expression,
    scale: Expression,
    override val ansiEnabled: Boolean = SQLConf.get.ansiEnabled)
  extends RoundBase(child, scale, BigDecimal.RoundingMode.HALF_UP, "ROUND_HALF_UP") {
  def this(child: Expression) = this(child, Literal(0), SQLConf.get.ansiEnabled)


  def this(child: Expression, scale: Expression) = this(child, scale, SQLConf.get.ansiEnabled)


  override def flatArguments: Iterator[Any] = Iterator(child, scale)


  override protected def withNewChildrenInternal(newLeft: Expression, newRight: Expression): Round =
    copy(child = newLeft, scale = newRight)
}

DataFushion

round.rs

For example, the implementation of Float64:

(value * 10.0_f64.powi(decimal_places)).round() / 10.0_f64.powi(decimal_places)

The core here is:

f64::round()

The definition of f64::round() in the Rust standard library is:

Doc

Returns the nearest integer to a number. Rounds halfway cases away from zero.

So from an algorithmic perspective, the two should be consistent, and we need to look at some differences from specific cases.

Test scenario

Double

val scales = -6 to 6
val doublePi: Double = math.Pi
Round(doublePi, scale)

• Datafusion Test Code

#[tokio::test]
async fn test_datafusion_round_pi_scales() -> Result<()> {
    let ctx = SessionContext::new();

    let pi = std::f64::consts::PI;
    let scales = -6..=6;

    println!("DataFusion round(pi, scale) test:\n");

    for scale in scales {
        let sql = format!("SELECT round({pi}, {scale}) AS r");

        let df = ctx.sql(&sql).await?;
        let results = df.collect().await?;

        let arr = results[0].column(0);
        let out = as_float64_array(arr)?;
        let val = out.value(0);

        println!("scale = {scale:>2}, round(pi, {scale}) = {val}");
    }

    Ok(())
}

Engines	Scales : -6	Scales : -5	Scales : -4	Scales : -3	Scales : -2	Scales : -1
DataFushion	0	0	0	0	0	0
Spark	0.0	0.0	0.0	0.0	0.0	0.0

Engines	Scales : 0
DataFushion	0
Spark	0.0

Engines	Scales : 1	Scales : 2	Scales : 3	Scales : 4	Scales : 5	Scales : 6
DataFushion	3	3.1	3.14	3.142	3.1416	3.14159
Spark	3.0	3.1	3.14	3.142	3.1416	3.14159

Float

val scales = -6 to 6
val floatPi: Float = 3.1415f
Round(floatPi, scale)

• Datafusion Test Code

#[tokio::test]
async fn test_datafusion_round_float_pi_scales() -> Result<()> {
    let ctx = SessionContext::new();

    let float_pi: f32 = 3.1415_f32;
    let scales = -6..=6;

    println!("DataFusion round(float_pi, scale) test:\n");

    for scale in scales {
        let sql = format!("SELECT round(CAST({float_pi} AS FLOAT), {scale}) AS r");

        let df = ctx.sql(&sql).await?;
        let results = df.collect().await?;

        let arr = results[0].column(0);
        let out = as_float32_array(arr)?;
        let val = out.value(0);

        println!("scale = {scale:>2}, round(float_pi, {scale}) = {val}");
    }

    Ok(())
}

Engines	Scales : -6	Scales : -5	Scales : -4	Scales : -3	Scales : -2	Scales : -1
DataFushion	0	0	0	0	0	0
Spark	0.0	0.0	0.0	0.0	0.0	0.0

Engines	Scales : 0
DataFushion	0
Spark	0.0

Engines	Scales : 1	Scales : 2	Scales : 3	Scales : 4	Scales : 5	Scales : 6
DataFushion	3	3.1	3.14	3.142	3.1415	3.1415
Spark	3.0	3.1	3.14	3.141	3.1415	3.1415

Int

val scales = -6 to 6
val intPi: Int = 314159265
Round(Int, scale)

• Datafusion Test Code

#[tokio::test]
async fn test_datafusion_round_int_pi_scales() -> Result<()> {
    let ctx = SessionContext::new();

    let int_pi: i64 = 314159265;
    let scales = -6..=6;

    println!("DataFusion round(int_pi, scale) test:\n");

    for scale in scales {
        let sql = format!("SELECT round({int_pi}, {scale}) AS r");
        let df = ctx.sql(&sql).await?;
        let results = df.collect().await?;

        let arr = results[0].column(0);
        let out = as_float64_array(arr)?;
        let val = out.value(0);

        println!("scale = {scale:>2}, round(int_pi, {scale}) = {val}");
    }

    Ok(())
}

Engines	Scales : -6	Scales : -5	Scales : -4	Scales : -3	Scales : -2	Scales : -1
DataFushion	314000000	314200000	314160000	314159000	314159300	314159270
Spark	314000000	314200000	314160000	314159000	314159300	314159270

Engines	Scales : 0
DataFushion	314159265
Spark	314159265

Engines	Scales : 0	Scales : 1	Scales : 2	Scales : 3	Scales : 4	Scales : 5
DataFushion	314159265	314159265	314159265	314159265	314159265	314159265
Spark	314159265	314159265	314159265	314159265	314159265	314159265

Long

val scales = -6 to 6
val longPi: Long = 31415926535897932L
Round(long, scale)

• Datafusion Test Code

#[tokio::test]
async fn test_datafusion_round_long_pi_scales() -> Result<()> {
    let ctx = SessionContext::new();

    let long_pi: i64 = 31415926535897932;
    let scales = -6..=6;

    println!("DataFusion round(long_pi, scale) test:\n");

    for scale in scales {
        let sql = format!("SELECT round({long_pi}, {scale}) AS r");
        let df = ctx.sql(&sql).await?;
        let results = df.collect().await?;

        let arr = results[0].column(0);
        let out = as_float64_array(arr)?;
        let val = out.value(0);

        println!("scale = {scale:>2}, round(long_pi, {scale}) = {val}");
    }

    Ok(())
}

Engines	Scales : -6	Scales : -5	Scales : -4	Scales : -3	Scales : -2	Scales : -1
DataFushion	31415926536000000	31415926535899996	31415926535900000	31415926535898000	31415926535897900	31415926535897940
Spark	31415926536000000L	31415926535900000L	31415926535900000L	31415926535898000L	31415926535897900L	31415926535897930L

Engines	Scales : 0
DataFushion	31415926535897932
Spark	31415926535897932L

Engines	Scales : 0	Scales : 1	Scales : 2	Scales : 3	Scales : 4	Scales : 5
DataFushion	31415926535897932	31415926535897932	31415926535897932	31415926535897932	31415926535897932	31415926535897932
Spark	31415926535897932L	31415926535897932L	31415926535897932L	31415926535897932L	31415926535897932L	31415926535897932L

Short

val scales = -6 to 6
val shortPi: Short = 31415
Round(short, scale)

• Datafusion Test Code

#[tokio::test]
async fn test_datafusion_round_short_pi_scales() -> Result<()> {

    let ctx = SessionContext::new();

    let short_pi: i16 = 31415;
    let scales = -6..=6;

    println!("DataFusion ROUND(short_pi, scale) test:\n");

    for scale in scales {
        let sql = format!("SELECT round({short_pi}, {scale}) AS r");
        let df = ctx.sql(&sql).await?;
        let results = df.collect().await?;

        let arr = results[0].column(0);
        let out = as_float64_array(arr)?;
        let val = out.value(0);

        println!("scale = {scale:>2}, round(short_pi, {scale}) = {val}");
    }

    Ok(())
}

Engines	Scales : -6	Scales : -5	Scales : -4	Scales : -3	Scales : -2	Scales : -1
DataFushion	0	0	30000	31000	31400	31420
Spark	0	0	30000	31000	31400	31420

Engines	Scales : 0
DataFushion	31415
Spark	31415

Engines	Scales : 0	Scales : 1	Scales : 2	Scales : 3	Scales : 4	Scales : 5
DataFushion	31415	31415	31415	31415	31415	31415
Spark	31415	31415	31415	31415	31415	31415