[SPARK-29774][SQL] Date and Timestamp type +/- null should be null as Postgres

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -34,7 +34,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules._
+import org.apache.spark.sql.catalyst.rules.{Rule, _}
 import org.apache.spark.sql.catalyst.trees.TreeNodeRef
 import org.apache.spark.sql.catalyst.util.toPrettySQL
 import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogPlugin, CatalogV2Util, Identifier, LookupCatalog, Table, TableCatalog, TableChange, V1Table}
@@ -228,6 +228,7 @@ class Analyzer(
       ResolveLambdaVariables(conf) ::
       ResolveTimeZone(conf) ::
       ResolveRandomSeed ::
+      ResolveBinaryArithmetic(conf) ::
       TypeCoercion.typeCoercionRules(conf) ++
       extendedResolutionRules : _*),
     Batch("PostgreSQL Dialect", Once, PostgreSQLDialect.postgreSQLDialectRules: _*),
@@ -246,6 +247,68 @@ class Analyzer(
       CleanupAliases)
   )
 
+  /**
+   * For [[UnresolvedAdd]]:
+   * 1. If one side is timestamp/date/string and the other side is interval, turns it to
+   * [[TimeAdd]];
+   * 2. else if one side is date, turns it to [[DateAdd]] ;
+   * 3. else turns it to [[Add]].
+   *
+   * For [[UnresolvedSubtract]]:
+   * 1. If the left side is timestamp/date/string and the right side is an interval, turns it to
+   * [[TimeSub]];
+   * 2. else if one side is timestamp and the other side is date/timestamp, turns it to
+   * [[SubtractTimestamps]];
+   * 3. else if both side are dates, turns it to [[DateDiff]]/[[SubtractDates]];
+   * 4. else if the left side is date, turns it to [[DateSub]];
+   * 5. else turns it to [[Subtract]].
+   *
+   * For [[UnresolvedMultiply]]:
+   * 1. If one side is interval and the other side is numeric, turns it to [[MultiplyInterval]];
+   * 2. otherwise, turns it to [[Multiply]].
+   *
+   * For [[UnresolvedDivide]]:
+   * 1. If the left side is interval and the right side is numeric, turns it to
+   * [[DivideInterval]];
+   * 2. otherwise, turns it to [[Divide]].
+   */
+  case class ResolveBinaryArithmetic(conf: SQLConf) extends Rule[LogicalPlan] {
+    override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+      case p: LogicalPlan => p.transformExpressionsUp {
+        case u @ UnresolvedAdd(l, r) if u.childrenResolved => (l.dataType, r.dataType) match {
+          case (TimestampType | DateType | StringType, CalendarIntervalType) =>
+            Cast(TimeAdd(l, r), l.dataType)
+          case (CalendarIntervalType, TimestampType | DateType | StringType) =>
+            Cast(TimeAdd(r, l), r.dataType)
+          case (DateType, _) => DateAdd(l, r)
+          case (_, DateType) => DateAdd(r, l)
+          case (_, _) => Add(l, r)
+        }
+        case u @ UnresolvedSubtract(l, r) if u.childrenResolved => (l.dataType, r.dataType) match {
+          case (TimestampType | DateType | StringType, CalendarIntervalType) =>
+            Cast(TimeSub(l, r), l.dataType)
+          case (TimestampType, TimestampType | DateType | NullType) => SubtractTimestamps(l, r)
+          case (DateType | NullType, TimestampType) => SubtractTimestamps(Cast(l, TimestampType), r)
+          case (DateType | NullType, DateType) => if (conf.usePostgreSQLDialect) {
+            DateDiff(l, r)
+          } else {
+            SubtractDates(l, r)
+          }
+          case (DateType, _) => DateSub(l, r)
+          case (_, _) => Subtract(l, r)
+        }
+        case u @ UnresolvedMultiply(l, r) if u.childrenResolved => (l.dataType, r.dataType) match {
+          case (CalendarIntervalType, _: NumericType | NullType) => MultiplyInterval(l, r)
+          case (_: NumericType | NullType, CalendarIntervalType) => MultiplyInterval(r, l)
+          case (_, _) => Multiply(l, r)
+        }
+        case u @ UnresolvedDivide(l, r) if u.childrenResolved => (l.dataType, r.dataType) match {
+          case (CalendarIntervalType, _: NumericType | NullType) => DivideInterval(l, r)
+          case (_, _) => Divide(l, r)
+        }
+      }
+    }
+  }
   /**
    * Substitute child plan with WindowSpecDefinitions.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

@@ -61,7 +61,6 @@ object TypeCoercion {
       StackCoercion ::
       Division(conf) ::
       ImplicitTypeCasts ::
-      DateTimeOperations ::
       WindowFrameCoercion ::
       Nil
 
@@ -822,55 +821,6 @@ object TypeCoercion {
     }
   }
 
-  /**
-   * 1. Turns Add/Subtract of DateType/TimestampType/StringType and CalendarIntervalType
-   *    to TimeAdd/TimeSub.
-   * 2. Turns Add/Subtract of TimestampType/DateType/IntegerType
-   *    and TimestampType/IntegerType/DateType to DateAdd/DateSub/SubtractDates and
-   *    to SubtractTimestamps.
-   * 3. Turns Multiply/Divide of CalendarIntervalType and NumericType
-   *    to MultiplyInterval/DivideInterval
-   */
-  object DateTimeOperations extends Rule[LogicalPlan] {
-
-    private val acceptedTypes = Seq(DateType, TimestampType, StringType)
-
-    def apply(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
-      // Skip nodes who's children have not been resolved yet.
-      case e if !e.childrenResolved => e
-
-      case Add(l @ CalendarIntervalType(), r) if acceptedTypes.contains(r.dataType) =>
-        Cast(TimeAdd(r, l), r.dataType)
-      case Add(l, r @ CalendarIntervalType()) if acceptedTypes.contains(l.dataType) =>
-        Cast(TimeAdd(l, r), l.dataType)
-      case Subtract(l, r @ CalendarIntervalType()) if acceptedTypes.contains(l.dataType) =>
-        Cast(TimeSub(l, r), l.dataType)
-      case Multiply(l @ CalendarIntervalType(), r @ NumericType()) =>
-        MultiplyInterval(l, r)
-      case Multiply(l @ NumericType(), r @ CalendarIntervalType()) =>
-        MultiplyInterval(r, l)
-      case Divide(l @ CalendarIntervalType(), r @ NumericType()) =>
-        DivideInterval(l, r)
-
-      case b @ BinaryOperator(l @ CalendarIntervalType(), r @ NullType()) =>
-        b.withNewChildren(Seq(l, Cast(r, CalendarIntervalType)))
-      case b @ BinaryOperator(l @ NullType(), r @ CalendarIntervalType()) =>
-        b.withNewChildren(Seq(Cast(l, CalendarIntervalType), r))
-
-      case Add(l @ DateType(), r @ IntegerType()) => DateAdd(l, r)
-      case Add(l @ IntegerType(), r @ DateType()) => DateAdd(r, l)
-      case Subtract(l @ DateType(), r @ IntegerType()) => DateSub(l, r)
-      case Subtract(l @ DateType(), r @ DateType()) =>
-        if (SQLConf.get.usePostgreSQLDialect) DateDiff(l, r) else SubtractDates(l, r)
-      case Subtract(l @ TimestampType(), r @ TimestampType()) =>
-        SubtractTimestamps(l, r)
-      case Subtract(l @ TimestampType(), r @ DateType()) =>
-        SubtractTimestamps(l, Cast(r, TimestampType))
-      case Subtract(l @ DateType(), r @ TimestampType()) =>
-        SubtractTimestamps(Cast(l, TimestampType), r)
-    }
-  }
-
   /**
    * Casts types according to the expected input types for [[Expression]]s.
    */
@@ -881,11 +831,8 @@ object TypeCoercion {
       case e if !e.childrenResolved => e
 
       // If DecimalType operands are involved, DecimalPrecision will handle it
-      // If CalendarIntervalType operands are involved, DateTimeOperations will handle it
       case b @ BinaryOperator(left, right) if !left.dataType.isInstanceOf[DecimalType] &&
           !right.dataType.isInstanceOf[DecimalType] &&
-          !left.dataType.isInstanceOf[CalendarIntervalType] &&
-          !right.dataType.isInstanceOf[CalendarIntervalType] &&
           left.dataType != right.dataType =>
         findTightestCommonType(left.dataType, right.dataType).map { commonType =>
           if (b.inputType.acceptsType(commonType)) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala

@@ -544,3 +544,19 @@ case class UnresolvedOrdinal(ordinal: Int)
   override def nullable: Boolean = throw new UnresolvedException(this, "nullable")
   override lazy val resolved = false
 }
+
+trait UnresolvedBinaryExpression extends BinaryExpression with Unevaluable {
+  override lazy val resolved: Boolean = false
+  override def dataType: DataType = throw new UnresolvedException(this, "dataType")
+}
+
+case class UnresolvedAdd(left: Expression, right: Expression) extends UnresolvedBinaryExpression
+
+case class UnresolvedSubtract(left: Expression, right: Expression)
+  extends UnresolvedBinaryExpression
+
+case class UnresolvedMultiply(left: Expression, right: Expression)
+  extends UnresolvedBinaryExpression
+
+case class UnresolvedDivide(left: Expression, right: Expression)
+  extends UnresolvedBinaryExpression

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1429,17 +1429,17 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     val right = expression(ctx.right)
     ctx.operator.getType match {
       case SqlBaseParser.ASTERISK =>
-        Multiply(left, right)
+        UnresolvedMultiply(left, right)
       case SqlBaseParser.SLASH =>
-        Divide(left, right)
+        UnresolvedDivide(left, right)
       case SqlBaseParser.PERCENT =>
         Remainder(left, right)
       case SqlBaseParser.DIV =>
         IntegralDivide(left, right)
       case SqlBaseParser.PLUS =>
-        Add(left, right)
+        UnresolvedAdd(left, right)
       case SqlBaseParser.MINUS =>
-        Subtract(left, right)
+        UnresolvedSubtract(left, right)
       case SqlBaseParser.CONCAT_PIPE =>
         Concat(left :: right :: Nil)
       case SqlBaseParser.AMPERSAND =>

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala

@@ -1401,44 +1401,6 @@ class TypeCoercionSuite extends AnalysisTest {
     }
   }
 
-  test("rule for date/timestamp operations") {
-    val dateTimeOperations = TypeCoercion.DateTimeOperations
-    val date = Literal(new java.sql.Date(0L))
-    val timestamp = Literal(new Timestamp(0L))
-    val interval = Literal(new CalendarInterval(0, 0, 0))
-    val str = Literal("2015-01-01")
-    val intValue = Literal(0, IntegerType)
-
-    ruleTest(dateTimeOperations, Add(date, interval), Cast(TimeAdd(date, interval), DateType))
-    ruleTest(dateTimeOperations, Add(interval, date), Cast(TimeAdd(date, interval), DateType))
-    ruleTest(dateTimeOperations, Add(timestamp, interval),
-      Cast(TimeAdd(timestamp, interval), TimestampType))
-    ruleTest(dateTimeOperations, Add(interval, timestamp),
-      Cast(TimeAdd(timestamp, interval), TimestampType))
-    ruleTest(dateTimeOperations, Add(str, interval), Cast(TimeAdd(str, interval), StringType))
-    ruleTest(dateTimeOperations, Add(interval, str), Cast(TimeAdd(str, interval), StringType))
-
-    ruleTest(dateTimeOperations, Subtract(date, interval), Cast(TimeSub(date, interval), DateType))
-    ruleTest(dateTimeOperations, Subtract(timestamp, interval),
-      Cast(TimeSub(timestamp, interval), TimestampType))
-    ruleTest(dateTimeOperations, Subtract(str, interval), Cast(TimeSub(str, interval), StringType))
-
-    // interval operations should not be effected
-    ruleTest(dateTimeOperations, Add(interval, interval), Add(interval, interval))
-    ruleTest(dateTimeOperations, Subtract(interval, interval), Subtract(interval, interval))
-
-    ruleTest(dateTimeOperations, Add(date, intValue), DateAdd(date, intValue))
-    ruleTest(dateTimeOperations, Add(intValue, date), DateAdd(date, intValue))
-    ruleTest(dateTimeOperations, Subtract(date, intValue), DateSub(date, intValue))
-    ruleTest(dateTimeOperations, Subtract(date, date), SubtractDates(date, date))
-    ruleTest(dateTimeOperations, Subtract(timestamp, timestamp),
-      SubtractTimestamps(timestamp, timestamp))
-    ruleTest(dateTimeOperations, Subtract(timestamp, date),
-      SubtractTimestamps(timestamp, Cast(date, TimestampType)))
-    ruleTest(dateTimeOperations, Subtract(date, timestamp),
-      SubtractTimestamps(Cast(date, TimestampType), timestamp))
-  }
-
   /**
    * There are rules that need to not fire before child expressions get resolved.
    * We use this test to make sure those rules do not fire early.
@@ -1586,27 +1548,6 @@ class TypeCoercionSuite extends AnalysisTest {
       Multiply(CaseWhen(Seq((EqualTo(1, 2), Cast(1, DecimalType(34, 24)))),
         Cast(100, DecimalType(34, 24))), Cast(1, IntegerType)))
   }
-
-  test("rule for interval operations") {
-    val dateTimeOperations = TypeCoercion.DateTimeOperations
-    val interval = Literal(new CalendarInterval(0, 0, 0))
-
-    Seq(
-      Literal(10.toByte, ByteType),
-      Literal(10.toShort, ShortType),
-      Literal(10, IntegerType),
-      Literal(10L, LongType),
-      Literal(Decimal(10), DecimalType.SYSTEM_DEFAULT),
-      Literal(10.5.toFloat, FloatType),
-      Literal(10.5, DoubleType)).foreach { num =>
-      ruleTest(dateTimeOperations, Multiply(interval, num),
-        MultiplyInterval(interval, num))
-      ruleTest(dateTimeOperations, Multiply(num, interval),
-        MultiplyInterval(interval, num))
-      ruleTest(dateTimeOperations, Divide(interval, num),
-        DivideInterval(interval, num))
-    }
-  }
 }
 
 

sql/core/src/test/resources/sql-tests/inputs/datetime.sql

@@ -39,3 +39,10 @@ select timestamp'2019-10-06 10:11:12.345678' - date'2020-01-01';
 
 select date '2019-01-01\t';
 select timestamp '2019-01-01\t';
+
+select date '2001-09-28' + null;
+select date '2001-09-28' - null;
+select null + date '2001-09-28';
+select null - date '2019-10-06';
+select null - timestamp '2019-10-06 10:11:12.345678';
+select timestamp '2019-10-06 10:11:12.345678' - null;

sql/core/src/test/resources/sql-tests/results/datetime.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 19
+-- Number of queries: 25
 
 
 -- !query 0
@@ -161,3 +161,51 @@ select timestamp '2019-01-01\t'
 struct<TIMESTAMP '2019-01-01 00:00:00':timestamp>
 -- !query 18 output
 2019-01-01 00:00:00
+
+
+-- !query 19
+select date '2001-09-28' + null
+-- !query 19 schema
+struct<date_add(DATE '2001-09-28', CAST(NULL AS INT)):date>
+-- !query 19 output
+NULL
+
+
+-- !query 20
+select date '2001-09-28' - null
+-- !query 20 schema
+struct<date_sub(DATE '2001-09-28', CAST(NULL AS INT)):date>
+-- !query 20 output
+NULL
+
+
+-- !query 21
+select null + date '2001-09-28'
+-- !query 21 schema
+struct<date_add(DATE '2001-09-28', CAST(NULL AS INT)):date>
+-- !query 21 output
+NULL
+
+
+-- !query 22
+select null - date '2019-10-06'
+-- !query 22 schema
+struct<subtractdates(CAST(NULL AS DATE), DATE '2019-10-06'):interval>
+-- !query 22 output
+NULL
+
+
+-- !query 23
+select null - timestamp '2019-10-06 10:11:12.345678'
+-- !query 23 schema
+struct<subtracttimestamps(CAST(NULL AS TIMESTAMP), TIMESTAMP '2019-10-06 10:11:12.345678'):interval>
+-- !query 23 output
+NULL
+
+
+-- !query 24
+select timestamp '2019-10-06 10:11:12.345678' - null
+-- !query 24 schema
+struct<subtracttimestamps(TIMESTAMP '2019-10-06 10:11:12.345678', CAST(NULL AS TIMESTAMP)):interval>
+-- !query 24 output
+NULL

sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out

@@ -54,7 +54,7 @@ struct<>
 -- !query 6
 select id, a+b, a-b, a*b, a/b from decimals_test order by id
 -- !query 6 schema
-struct<id:int,(a + b):decimal(38,17),(a - b):decimal(38,17),(a * b):decimal(38,6),(a / b):decimal(38,6)>
+struct<id:int,(CAST(a AS DECIMAL(38,17)) + CAST(b AS DECIMAL(38,17))):decimal(38,17),(CAST(a AS DECIMAL(38,17)) - CAST(b AS DECIMAL(38,17))):decimal(38,17),(CAST(a AS DECIMAL(38,18)) * CAST(b AS DECIMAL(38,18))):decimal(38,6),(CAST(a AS DECIMAL(38,18)) / CAST(b AS DECIMAL(38,18))):decimal(38,6)>
 -- !query 6 output
 1	1099.00000000000000000	-899.00000000000000000	99900.000000	0.100100
 2	24690.24600000000000000	0.00000000000000000	152402061.885129	1.000000
@@ -140,7 +140,7 @@ NULL
 -- !query 16
 select 12345678901234567890.0 * 12345678901234567890.0
 -- !query 16 schema
-struct<(12345678901234567890.0 * 12345678901234567890.0):decimal(38,2)>
+struct<(CAST(12345678901234567890.0 AS DECIMAL(21,1)) * CAST(12345678901234567890.0 AS DECIMAL(21,1))):decimal(38,2)>
 -- !query 16 output
 NULL
 
@@ -196,7 +196,7 @@ spark.sql.decimalOperations.allowPrecisionLoss	false
 -- !query 23
 select id, a+b, a-b, a*b, a/b from decimals_test order by id
 -- !query 23 schema
-struct<id:int,(a + b):decimal(38,18),(a - b):decimal(38,18),(a * b):decimal(38,36),(a / b):decimal(38,18)>
+struct<id:int,(CAST(a AS DECIMAL(38,18)) + CAST(b AS DECIMAL(38,18))):decimal(38,18),(CAST(a AS DECIMAL(38,18)) - CAST(b AS DECIMAL(38,18))):decimal(38,18),(CAST(a AS DECIMAL(38,18)) * CAST(b AS DECIMAL(38,18))):decimal(38,36),(CAST(a AS DECIMAL(38,18)) / CAST(b AS DECIMAL(38,18))):decimal(38,18)>
 -- !query 23 output
 1	1099.000000000000000000	-899.000000000000000000	NULL	0.100100100100100100
 2	24690.246000000000000000	0.000000000000000000	NULL	1.000000000000000000
@@ -282,7 +282,7 @@ NULL
 -- !query 33
 select 12345678901234567890.0 * 12345678901234567890.0
 -- !query 33 schema
-struct<(12345678901234567890.0 * 12345678901234567890.0):decimal(38,2)>
+struct<(CAST(12345678901234567890.0 AS DECIMAL(21,1)) * CAST(12345678901234567890.0 AS DECIMAL(21,1))):decimal(38,2)>
 -- !query 33 output
 NULL