[SPARK-31761][SQL] cast integer to Long to avoid IntegerOverflow for IntegralDivide operator

[sandeep-katta]

What changes were proposed in this pull request?

IntegralDivide operator returns Long DataType, so integer overflow case should be handled.
If the operands are of type Int it will be casted to Long

Why are the changes needed?

As IntegralDivide returns Long datatype, integer overflow should not happen

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added UT and also tested in the local cluster

After fix

SQL Test

After fix

Before Fix

[SparkQA]

Test build #122943 has finished for PR 28600 at commit cac15d9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #122969 has finished for PR 28600 at commit b2eea1f.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #122979 has finished for PR 28600 at commit 7cdc7b5.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #122982 has finished for PR 28600 at commit e2d8e1f.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

there is a failure in org.apache.spark.sql.catalyst.expressions.ArithmeticExpressionSuite./ (Divide) for integral type

[sandeep-katta]

this is because I changed the acceptedType to Long and Decimal

checkEvaluation(IntegralDivide(Literal(1.toByte), Literal(2.toByte)), 0L) checkEvaluation(IntegralDivide(Literal(1.toShort), Literal(2.toShort)), 0L) checkEvaluation(IntegralDivide(Literal(1), Literal(2)), 0L) checkEvaluation(IntegralDivide(Literal(1.toLong), Literal(2.toLong)), 0L) checkEvaluation(IntegralDivide(positiveShortLit, negativeShortLit), 0L) checkEvaluation(IntegralDivide(positiveIntLit, negativeIntLit), 0L) checkEvaluation(IntegralDivide(positiveLongLit, negativeLongLit), 0L)

from the test case shall I remove checkEvaluation for byte,short and int ?

[SparkQA]

Test build #122988 has finished for PR 28600 at commit d9f4616.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #122992 has finished for PR 28600 at commit 351e49a.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

from the test case shall I remove checkEvaluation for byte,short and int ?

Yes please

[cloud-fan]

org.scalatest.exceptions.TestFailedException: "struct<([3 div 2]):bigint>" did not equal "struct<([CAST(3 AS BIGINT) div CAST(2 AS BIGINT)]):bigint>" Schema did not match for query SELECT 3 div 2

OK so the problem here is we change the auto-generated column name of 3 div 2, because we add cast.

It's not a big deal, we can just update all the tests. But I'm thinking if it's better to embed the cast in the IntegralDivide itself. e.g.

case class IntegralDivide(...) {
  private lazy val div: (Any, Any) => Any = {
    val integral = left.dataType match {
      case _: IntegralType =>
        LongType.integral.asInstanceOf[Integral[Any]] // see the change in this line
      case d: DecimalType =>
        d.asIntegral.asInstanceOf[Integral[Any]]
    }
    (x, y) => {
      val res = integral.quot(x, y)
      if (res == null) {
        null
      } else {
        integral.asInstanceOf[Integral[Any]].toLong(res)
      }
    }
  }

  override def operationCode(v1: String, v2: String): String = {
    if (decimalType) ... else {
      s"((long) $v1) / $v2"
    }
  }
}

This is probably more efficient as there are less expressions

[sandeep-katta]

org.scalatest.exceptions.TestFailedException: "struct<([3 div 2]):bigint>" did not equal "struct<([CAST(3 AS BIGINT) div CAST(2 AS BIGINT)]):bigint>" Schema did not match for query SELECT 3 div 2

OK so the problem here is we change the auto-generated column name of 3 div 2, because we add cast.

It's not a big deal, we can just update all the tests. But I'm thinking if it's better to embed the cast in the IntegralDivide itself. e.g.

case class IntegralDivide(...) {
  private lazy val div: (Any, Any) => Any = {
    val integral = left.dataType match {
      case _: IntegralType =>
        LongType.integral.asInstanceOf[Integral[Any]] // see the change in this line
      case d: DecimalType =>
        d.asIntegral.asInstanceOf[Integral[Any]]
    }
    (x, y) => {
      val res = integral.quot(x, y)
      if (res == null) {
        null
      } else {
        integral.asInstanceOf[Integral[Any]].toLong(res)
      }
    }
  }

  override def operationCode(v1: String, v2: String): String = {
    if (decimalType) ... else {
      s"((long) $v1) / $v2"
    }
  }
}

This is probably more efficient as there are less expressions

this was my first approach, now I am updating the golden files.

Let me know which way we need to do, accordingly I will update the code

[SparkQA]

Test build #123005 has finished for PR 28600 at commit db605e6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

LGTM if the tests passed.

[SparkQA]

Test build #123026 has finished for PR 28600 at commit 3a1a061.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[sandeep-katta]

retest this please

[SparkQA]

Test build #123029 has finished for PR 28600 at commit 3a1a061.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HyukjinKwon]

LGTM too

[HyukjinKwon]

Merged to master.

@sandeep-katta, it has a conflict against branch-3.0. Although it's minor, can you create a PR to backport it? Looks like these SQL files have some diff in the tests results lately. Let's just create a PR to backport just to doubly make sure.

[sandeep-katta]

okay will raise BackPort PR