[Kernel] Fix long overflow when parsing far-future timestamp stats

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/data/DefaultJsonRow.java

@@ -30,7 +30,6 @@
 import java.sql.Date;
 import java.time.Instant;
 import java.time.OffsetDateTime;
-import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -255,7 +254,11 @@ private static Object decodeElement(JsonNode jsonValue, DataType dataType) {
     if (dataType instanceof TimestampType) {
       throwIfTypeMismatch("timestamp", jsonValue.isTextual(), jsonValue);
       Instant time = OffsetDateTime.parse(jsonValue.textValue()).toInstant();
-      return ChronoUnit.MICROS.between(Instant.EPOCH, time);
+      // Don't use ChronoUnit.MICROS.between() - it computes (seconds * 1_000_000_000) / 1000,
+      // where the intermediate nanoseconds value overflows. We compute seconds * 1_000_000
+      // directly, which gives the same result without the overflow.
+      return Math.addExact(
+          Math.multiplyExact(time.getEpochSecond(), 1_000_000L), time.getNano() / 1000);
     }
 
     if (dataType instanceof TimestampNTZType) {

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableReadsSuite.scala

@@ -272,6 +272,21 @@ trait AbstractDeltaTableReadsSuite extends AnyFunSuite { self: AbstractTestUtils
     }
   }
 
+  test("read table with far-future timestamp in stats") {
+    withTempDir { tempDir =>
+      val path = tempDir.getCanonicalPath
+      spark.sql(s"CREATE TABLE delta.`$path` (ts TIMESTAMP) USING DELTA")
+      spark.sql(s"INSERT INTO delta.`$path` VALUES (TIMESTAMP'2020-01-01 00:00:00')")
+      spark.sql(s"INSERT INTO delta.`$path` VALUES (TIMESTAMP'9999-12-31 23:59:59')")
+      val filter = new Predicate("<", new Column("ts"), Literal.ofTimestamp(253402300799000000L))
+      checkTable(
+        path,
+        Seq(TestRow(1577836800000000L)),
+        filter = filter,
+        expectedRemainingFilter = filter)
+    }
+  }
+
   //////////////////////////////////////////////////////////////////////////////////
   // Timestamp_NTZ tests
   //////////////////////////////////////////////////////////////////////////////////

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/engine/DefaultJsonHandlerSuite.scala

@@ -212,6 +212,17 @@ class DefaultJsonHandlerSuite extends AnyFunSuite with TestUtils with DefaultVec
       TestRow(2524636800000000L, 23423523000L, -315583200000000L, null))
   }
 
+  test("parse timestamp type with large values") {
+    // Timestamps far in the future should not cause overflow.
+    // ChronoUnit.MICROS.between() internally computes nanoseconds first, which overflows
+    // for timestamps more than ~292 years from epoch.
+    testJsonParserForSingleType(
+      jsonString = """{"col1":"9999-12-31T23:59:59.000+00:00"}""",
+      dataType = TimestampType.TIMESTAMP,
+      numColumns = 1,
+      TestRow(253402300799000000L))
+  }
+
   test("parse null input") {
     val schema = new StructType()
       .add("nested_struct", new StructType().add("foo", IntegerType.INTEGER))