DataFrameValueWriter serialized a Spark DateType via java.sql.Date.getTime(),

which is midnight in the JVM default timezone. On a non UTC JVM this shifted the
stored value by the timezone offset and could land on a different calendar day.
Convert through the calendar date at UTC so the result is timezone independent.

Applies to the sql-30, sql-35, and sql-40 modules.

Resolves opensearch-project#797

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

CHANGELOG.md

@@ -5,6 +5,9 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Added
 - Add system property `opensearch.hadoop.version.check.skip` to bypass multiple JAR version detection ([#753](https://github.com/opensearch-project/opensearch-hadoop/pull/753))
 
+### Fixed
+- Write Spark `DateType` values as UTC start of day so the stored date does not shift by the JVM timezone offset on non UTC JVMs ([#797](https://github.com/opensearch-project/opensearch-hadoop/issues/797))
+
 ### Dependencies
 - Bumps `commons-logging:commons-logging` from 1.3.5 to 1.3.6
 - Bumps `com.fasterxml.jackson.core:jackson-databind` from 2.21.1 to 2.21.3

spark/sql-30/src/main/scala/org/opensearch/spark/sql/DataFrameValueWriter.scala

@@ -269,7 +269,7 @@ class DataFrameValueWriter(writeUnknownTypes: Boolean = false) extends Filtering
       case DoubleType    => generator.writeNumber(value.asInstanceOf[Double])
       case FloatType     => generator.writeNumber(value.asInstanceOf[Float])
       case TimestampType => generator.writeNumber(value.asInstanceOf[Timestamp].getTime())
-      case DateType      => generator.writeNumber(value.asInstanceOf[Date].getTime())
+      case DateType      => generator.writeNumber(value.asInstanceOf[Date].toLocalDate.atStartOfDay(java.time.ZoneOffset.UTC).toInstant.toEpochMilli)
       case StringType    => generator.writeString(value.toString)
       case _             => {
         val className = schema.getClass().getName()

spark/sql-30/src/test/scala/org/opensearch/spark/sql/DataFrameValueWriterTest.scala

@@ -30,10 +30,13 @@
 package org.opensearch.spark.sql
 
 import java.io.ByteArrayOutputStream
+import java.sql.Date
+import java.util.TimeZone
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.types.ArrayType
+import org.apache.spark.sql.types.DateType
 import org.apache.spark.sql.types.IntegerType
 import org.apache.spark.sql.types.MapType
 import org.apache.spark.sql.types.StringType
@@ -165,4 +168,18 @@ class DataFrameValueWriterTest {
     }
   }
 
+  @Test
+  def testWriteDateUsesUtcStartOfDay(): Unit = {
+    val defaultTimeZone = TimeZone.getDefault
+    try {
+      TimeZone.setDefault(TimeZone.getTimeZone("Asia/Tokyo"))
+      val schema = StructType(Seq(StructField("d", DateType)))
+      val row = Row(Date.valueOf("2023-07-22"))
+      val serialized = serialize(row, schema)
+      assertTrue(serialized.contains(""""d":1689984000000"""))
+    } finally {
+      TimeZone.setDefault(defaultTimeZone)
+    }
+  }
+
 }

spark/sql-35/src/main/scala/org/opensearch/spark/sql/DataFrameValueWriter.scala

@@ -269,7 +269,7 @@ class DataFrameValueWriter(writeUnknownTypes: Boolean = false) extends Filtering
       case DoubleType    => generator.writeNumber(value.asInstanceOf[Double])
       case FloatType     => generator.writeNumber(value.asInstanceOf[Float])
       case TimestampType => generator.writeNumber(value.asInstanceOf[Timestamp].getTime())
-      case DateType      => generator.writeNumber(value.asInstanceOf[Date].getTime())
+      case DateType      => generator.writeNumber(value.asInstanceOf[Date].toLocalDate.atStartOfDay(java.time.ZoneOffset.UTC).toInstant.toEpochMilli)
       case StringType    => generator.writeString(value.toString)
       case _             => {
         val className = schema.getClass().getName()

spark/sql-35/src/test/scala/org/opensearch/spark/sql/DataFrameValueWriterTest.scala

@@ -30,10 +30,13 @@
 package org.opensearch.spark.sql
 
 import java.io.ByteArrayOutputStream
+import java.sql.Date
+import java.util.TimeZone
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.types.ArrayType
+import org.apache.spark.sql.types.DateType
 import org.apache.spark.sql.types.IntegerType
 import org.apache.spark.sql.types.MapType
 import org.apache.spark.sql.types.StringType
@@ -165,4 +168,18 @@ class DataFrameValueWriterTest {
     }
   }
 
+  @Test
+  def testWriteDateUsesUtcStartOfDay(): Unit = {
+    val defaultTimeZone = TimeZone.getDefault
+    try {
+      TimeZone.setDefault(TimeZone.getTimeZone("Asia/Tokyo"))
+      val schema = StructType(Seq(StructField("d", DateType)))
+      val row = Row(Date.valueOf("2023-07-22"))
+      val serialized = serialize(row, schema)
+      assertTrue(serialized.contains(""""d":1689984000000"""))
+    } finally {
+      TimeZone.setDefault(defaultTimeZone)
+    }
+  }
+
 }

spark/sql-40/src/main/scala/org/opensearch/spark/sql/DataFrameValueWriter.scala

@@ -269,7 +269,7 @@ class DataFrameValueWriter(writeUnknownTypes: Boolean = false) extends Filtering
       case DoubleType    => generator.writeNumber(value.asInstanceOf[Double])
       case FloatType     => generator.writeNumber(value.asInstanceOf[Float])
       case TimestampType => generator.writeNumber(value.asInstanceOf[Timestamp].getTime())
-      case DateType      => generator.writeNumber(value.asInstanceOf[Date].getTime())
+      case DateType      => generator.writeNumber(value.asInstanceOf[Date].toLocalDate.atStartOfDay(java.time.ZoneOffset.UTC).toInstant.toEpochMilli)
       case StringType    => generator.writeString(value.toString)
       case _             => {
         val className = schema.getClass().getName()

spark/sql-40/src/test/scala/org/opensearch/spark/sql/DataFrameValueWriterTest.scala

@@ -30,10 +30,13 @@
 package org.opensearch.spark.sql
 
 import java.io.ByteArrayOutputStream
+import java.sql.Date
+import java.util.TimeZone
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.types.ArrayType
+import org.apache.spark.sql.types.DateType
 import org.apache.spark.sql.types.IntegerType
 import org.apache.spark.sql.types.MapType
 import org.apache.spark.sql.types.StringType
@@ -165,4 +168,18 @@ class DataFrameValueWriterTest {
     }
   }
 
+  @Test
+  def testWriteDateUsesUtcStartOfDay(): Unit = {
+    val defaultTimeZone = TimeZone.getDefault
+    try {
+      TimeZone.setDefault(TimeZone.getTimeZone("Asia/Tokyo"))
+      val schema = StructType(Seq(StructField("d", DateType)))
+      val row = Row(Date.valueOf("2023-07-22"))
+      val serialized = serialize(row, schema)
+      assertTrue(serialized.contains(""""d":1689984000000"""))
+    } finally {
+      TimeZone.setDefault(defaultTimeZone)
+    }
+  }
+
 }