[WIP][SPARK-57811][SQL] Support string to nanosecond-precision timestamp coercion in comparisons and predicates

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

@@ -29,6 +29,7 @@ import org.apache.spark.sql.catalyst.expressions.{
 }
 import org.apache.spark.sql.types.{
   AnsiIntervalType,
+  AnyTimestampNanoTypeExpression,
   CalendarIntervalType,
   DataType,
   DoubleType,
@@ -59,6 +60,10 @@ object StringPromotionTypeCoercion {
       p.withNewChildren(Seq(Cast(left, TimestampType), right))
     case p @ Equality(left @ TimestampTypeExpression(), right @ StringTypeExpression()) =>
       p.withNewChildren(Seq(left, Cast(right, TimestampType)))
+    case p @ Equality(left @ StringTypeExpression(), right @ AnyTimestampNanoTypeExpression()) =>
+      p.withNewChildren(Seq(Cast(left, right.dataType), right))
+    case p @ Equality(left @ AnyTimestampNanoTypeExpression(), right @ StringTypeExpression()) =>
+      p.withNewChildren(Seq(left, Cast(right, left.dataType)))
 
     case p @ BinaryComparison(left, right)
         if findCommonTypeForBinaryComparison(left.dataType, right.dataType, conf).isDefined =>

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

@@ -145,6 +145,14 @@ object TypeCoercion extends TypeCoercionBase {
       => if (conf.castDatetimeToString) Some(st) else Some(TimestampType)
     case (TimestampType, st: StringType)
       => if (conf.castDatetimeToString) Some(st) else Some(TimestampType)
+    // These arms match both TIMESTAMP_LTZ(p) and TIMESTAMP_NTZ(p) (AnyTimestampNanoType), so the
+    // nanos types follow the micros TimestampType (LTZ) branch above and honor castDatetimeToString
+    // for both families. This is intentionally more precise than micros TimestampNTZType, which has
+    // no arm here and falls through to the config-blind canPromoteAsInBinaryComparison line below.
+    case (st: StringType, tsNanos: AnyTimestampNanoType)
+      => if (conf.castDatetimeToString) Some(st) else Some(tsNanos)
+    case (tsNanos: AnyTimestampNanoType, st: StringType)
+      => if (conf.castDatetimeToString) Some(st) else Some(tsNanos)
     case (st: StringType, NullType) => Some(st)
     case (NullType, st: StringType) => Some(st)
 

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

@@ -1701,6 +1701,44 @@ class TypeCoercionSuite extends TypeCoercionSuiteBase {
       EqualTo(Cast(date0301, TimestampType), timestamp0301000000))
     ruleTest(rule, LessThan(date0301, timestamp0301000001),
       LessThan(Cast(date0301, TimestampType), timestamp0301000001))
+
+    // SPARK-57811: a string operand is coerced to the nanosecond timestamp type in comparisons
+    // and predicates, mirroring the microsecond timestamp handling above. The concrete operand
+    // type (family + precision) is preserved, so the string is cast to that exact nanos type.
+    Seq(7, 8, 9).foreach { p =>
+      Seq(TimestampLTZNanosType(p), TimestampNTZNanosType(p)).foreach { nt =>
+        val tsn = AttributeReference("tsn", nt)()
+        val strLit = Literal("2020-01-02 03:04:05.123456789")
+        // Equality path (StringPromotionTypeCoercion Equality arms): the string is cast to the
+        // nanos operand's own type so subsecond rounding does not affect the comparison.
+        ruleTest(rule, EqualTo(tsn, strLit), EqualTo(tsn, Cast(strLit, nt)))
+        ruleTest(rule, EqualTo(strLit, tsn), EqualTo(Cast(strLit, nt), tsn))
+        // Range path (findCommonTypeForBinaryComparison).
+        ruleTest(rule, LessThan(tsn, strLit), LessThan(tsn, Cast(strLit, nt)))
+        ruleTest(rule, GreaterThanOrEqual(strLit, tsn),
+          GreaterThanOrEqual(Cast(strLit, nt), tsn))
+      }
+    }
+
+    // SPARK-57811: legacy `castDatetimeToString` parity with microsecond timestamps. Range
+    // comparisons promote both operands to string; equality still casts the string to the nanos
+    // type (the Equality arm fires before the range arm). This block is the assertion that
+    // actually exercises the new non-ANSI production arms -- in the default config the generic
+    // string-promotion fall-through already yields the nanos common type, so only the legacy
+    // branch distinguishes the new code (the range case below fails without the
+    // findCommonTypeForBinaryComparison edit).
+    withSQLConf(SQLConf.LEGACY_CAST_DATETIME_TO_STRING.key -> "true") {
+      val nt = TimestampNTZNanosType(9)
+      val tsn = AttributeReference("tsn", nt)()
+      val strLit = Literal("2020-01-02 03:04:05.123456789")
+      // Range: both operands become strings.
+      ruleTest(rule, LessThan(tsn, strLit), LessThan(Cast(tsn, StringType), strLit))
+      ruleTest(rule, GreaterThanOrEqual(strLit, tsn),
+        GreaterThanOrEqual(strLit, Cast(tsn, StringType)))
+      // Equality: the string is still cast to nanos so subseconds are compared exactly.
+      ruleTest(rule, EqualTo(tsn, strLit), EqualTo(tsn, Cast(strLit, nt)))
+      ruleTest(rule, EqualTo(strLit, tsn), EqualTo(Cast(strLit, nt), tsn))
+    }
   }
 
   test("cast WindowFrame boundaries to the type they operate upon") {

sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ltz-nanos.sql.out

@@ -1034,3 +1034,26 @@ Sort [v#x ASC NULLS FIRST], true
                   :     +- OneRowRelation
                   +- Project [2020-01-01 00:00:00.0000005 AS TIMESTAMP_LTZ '2020-01-01 00:00:00.000000500'#x]
                      +- OneRowRelation
+
+
+-- !query
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789') AS t(c)
+-- !query analysis
+Project [(c#x = cast(2020-01-02 03:04:05.123456789 as timestamp_ltz(9))) AS (c = 2020-01-02 03:04:05.123456789)#x, (c#x = cast(2020-01-02 03:04:05.123456788 as timestamp_ltz(9))) AS (c = 2020-01-02 03:04:05.123456788)#x, (c#x < cast(2020-01-02 03:04:05.123456790 as timestamp_ltz(9))) AS (c < 2020-01-02 03:04:05.123456790)#x]
++- SubqueryAlias t
+   +- LocalRelation [c#x]
+
+
+-- !query
+SELECT c FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005'
+-- !query analysis
+Project [c#x]
++- Filter between(c#x, 2020-01-02 03:04:05.000000001, 2020-01-02 03:04:05.000000005)
+   +- SubqueryAlias t
+      +- LocalRelation [c#x]

sql/core/src/test/resources/sql-tests/analyzer-results/timestamp-ntz-nanos.sql.out

@@ -868,3 +868,26 @@ Sort [v#x ASC NULLS FIRST], true
                   :     +- OneRowRelation
                   +- Project [2020-01-01 00:00:00.0000005 AS TIMESTAMP_NTZ '2020-01-01 00:00:00.000000500'#x]
                      +- OneRowRelation
+
+
+-- !query
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789') AS t(c)
+-- !query analysis
+Project [(c#x = cast(2020-01-02 03:04:05.123456789 as timestamp_ntz(9))) AS (c = 2020-01-02 03:04:05.123456789)#x, (c#x = cast(2020-01-02 03:04:05.123456788 as timestamp_ntz(9))) AS (c = 2020-01-02 03:04:05.123456788)#x, (c#x < cast(2020-01-02 03:04:05.123456790 as timestamp_ntz(9))) AS (c < 2020-01-02 03:04:05.123456790)#x]
++- SubqueryAlias t
+   +- LocalRelation [c#x]
+
+
+-- !query
+SELECT c FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005'
+-- !query analysis
+Project [c#x]
++- Filter between(c#x, 2020-01-02 03:04:05.000000001, 2020-01-02 03:04:05.000000005)
+   +- SubqueryAlias t
+      +- LocalRelation [c#x]

sql/core/src/test/resources/sql-tests/inputs/timestamp-ltz-nanos.sql

@@ -328,3 +328,17 @@ SELECT v, lead(v) OVER (ORDER BY v) AS next_v FROM (
     SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000900' AS v
     UNION ALL SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000100'
     UNION ALL SELECT TIMESTAMP_LTZ '2020-01-01 00:00:00.000000500') ORDER BY v;
+
+-- SPARK-57811: a string operand is coerced to the nanosecond timestamp type in comparisons and
+-- predicates (not truncated to micros, not promoted to string). The 9th fractional digit is
+-- significant, so an off-by-one-nanosecond literal does not compare equal.
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789') AS t(c);
+
+-- BETWEEN over nanosecond timestamps: only the value inside the sub-microsecond range qualifies.
+SELECT c FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005';

sql/core/src/test/resources/sql-tests/inputs/timestamp-ntz-nanos.sql

@@ -268,3 +268,17 @@ SELECT v, lead(v) OVER (ORDER BY v) AS next_v FROM (
     SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000900' AS v
     UNION ALL SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000100'
     UNION ALL SELECT TIMESTAMP_NTZ '2020-01-01 00:00:00.000000500') ORDER BY v;
+
+-- SPARK-57811: a string operand is coerced to the nanosecond timestamp type in comparisons and
+-- predicates (not truncated to micros, not promoted to string). The 9th fractional digit is
+-- significant, so an off-by-one-nanosecond literal does not compare equal.
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789') AS t(c);
+
+-- BETWEEN over nanosecond timestamps: only the value inside the sub-microsecond range qualifies.
+SELECT c FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005';

sql/core/src/test/resources/sql-tests/results/timestamp-ltz-nanos.sql.out

@@ -1125,3 +1125,25 @@ struct<v:timestamp_ltz(9),next_v:timestamp_ltz(9)>
 2020-01-01 00:00:00.0000001	2020-01-01 00:00:00.0000005
 2020-01-01 00:00:00.0000005	2020-01-01 00:00:00.0000009
 2020-01-01 00:00:00.0000009	NULL
+
+
+-- !query
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_LTZ '2020-01-02 03:04:05.123456789') AS t(c)
+-- !query schema
+struct<(c = 2020-01-02 03:04:05.123456789):boolean,(c = 2020-01-02 03:04:05.123456788):boolean,(c < 2020-01-02 03:04:05.123456790):boolean>
+-- !query output
+true	false	true
+
+
+-- !query
+SELECT c FROM VALUES
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_LTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005'
+-- !query schema
+struct<c:timestamp_ltz(9)>
+-- !query output
+2020-01-02 03:04:05.000000001

sql/core/src/test/resources/sql-tests/results/timestamp-ntz-nanos.sql.out

@@ -922,3 +922,25 @@ struct<v:timestamp_ntz(9),next_v:timestamp_ntz(9)>
 2020-01-01 00:00:00.0000001	2020-01-01 00:00:00.0000005
 2020-01-01 00:00:00.0000005	2020-01-01 00:00:00.0000009
 2020-01-01 00:00:00.0000009	NULL
+
+
+-- !query
+SELECT c = '2020-01-02 03:04:05.123456789',
+       c = '2020-01-02 03:04:05.123456788',
+       c < '2020-01-02 03:04:05.123456790'
+  FROM VALUES (TIMESTAMP_NTZ '2020-01-02 03:04:05.123456789') AS t(c)
+-- !query schema
+struct<(c = 2020-01-02 03:04:05.123456789):boolean,(c = 2020-01-02 03:04:05.123456788):boolean,(c < 2020-01-02 03:04:05.123456790):boolean>
+-- !query output
+true	false	true
+
+
+-- !query
+SELECT c FROM VALUES
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000001'),
+  (TIMESTAMP_NTZ '2020-01-02 03:04:05.000000009') AS t(c)
+  WHERE c BETWEEN '2020-01-02 03:04:05.000000001' AND '2020-01-02 03:04:05.000000005'
+-- !query schema
+struct<c:timestamp_ntz(9)>
+-- !query output
+2020-01-02 03:04:05.000000001

sql/core/src/test/scala/org/apache/spark/sql/TimestampNanosWideningSuiteBase.scala

@@ -132,6 +132,29 @@ abstract class TimestampNanosWideningSuiteBase extends SharedSparkSession {
     val ntzEq = twoCols(TimestampNTZType, ldtA, TimestampNTZNanosType(9), ldtA)
     checkAnswer(ntzEq.selectExpr("a = b", "a < b"), Row(true, false))
   }
+
+  test("SPARK-57811: string operand coerces to the nanosecond timestamp type") {
+    // Equality on the exact sub-microsecond value: the string is parsed at nanos precision, so a
+    // string that differs only in the 9th digit does NOT compare equal. The paired
+    // `= '...789' => true` / `= '...788' => false` is the load-bearing pair: it rules out
+    // micro-truncation (truncating the operand to micros would make both strings equal the column).
+    // The nanos-vs-string-promotion distinction is locked separately by the analyzer goldens
+    // (explicit `cast as timestamp_ntz(9)`) and the unit tests in TypeCoercionSuite.
+    val ntz =
+      single(TimestampNTZNanosType(9), LocalDateTime.parse("2020-01-02T03:04:05.123456789"))
+    checkAnswer(
+      ntz.selectExpr(
+        "c = '2020-01-02 03:04:05.123456789'",
+        "c = '2020-01-02 03:04:05.123456788'",
+        "c < '2020-01-02 03:04:05.123456790'",
+        "c > '2020-01-02 03:04:05.123456788'"),
+      Row(true, false, true, true))
+
+    // LTZ operand: the string literal is parsed in the session zone (America/Los_Angeles), so the
+    // 2020-01-01T00:00:00Z instant equals the local wall-clock time eight hours earlier.
+    val ltz = single(TimestampLTZNanosType(9), Instant.parse("2020-01-01T00:00:00Z"))
+    checkAnswer(ltz.selectExpr("c = '2019-12-31 16:00:00'"), Row(true))
+  }
 }
 
 // Runs the nanosecond timestamp widening tests with ANSI mode enabled explicitly.