diff --git a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala index 54cd3ba3464e9..5bb51a92977af 100644 --- a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala +++ b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala @@ -34,10 +34,12 @@ import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters} import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow, UnsafeArrayData} import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, DateTimeUtils, GenericArrayData} import org.apache.spark.sql.catalyst.util.DateTimeConstants.MILLIS_PER_DAY +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.execution.datasources.DataSourceUtils import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String + /** * A deserializer to deserialize data in avro format to data in catalyst format. */ @@ -45,7 +47,7 @@ private[sql] class AvroDeserializer( rootAvroType: Schema, rootCatalystType: DataType, positionalFieldMatch: Boolean, - datetimeRebaseMode: LegacyBehaviorPolicy.Value, + datetimeRebaseSpec: RebaseSpec, filters: StructFilters) { def this( @@ -56,17 +58,17 @@ private[sql] class AvroDeserializer( rootAvroType, rootCatalystType, positionalFieldMatch = false, - LegacyBehaviorPolicy.withName(datetimeRebaseMode), + RebaseSpec(LegacyBehaviorPolicy.withName(datetimeRebaseMode)), new NoopFilters) } private lazy val decimalConversions = new DecimalConversion() private val dateRebaseFunc = DataSourceUtils.createDateRebaseFuncInRead( - datetimeRebaseMode, "Avro") + datetimeRebaseSpec.mode, "Avro") private val timestampRebaseFunc = DataSourceUtils.createTimestampRebaseFuncInRead( - datetimeRebaseMode, "Avro") + datetimeRebaseSpec, "Avro") private val converter: Any => Option[Any] = try { rootCatalystType match { diff --git a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala index 398cb02242e08..a13e0624f351d 100755 --- a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala +++ b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala @@ -123,7 +123,7 @@ private[sql] class AvroFileFormat extends FileFormat reader.sync(file.start) - val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode( + val datetimeRebaseMode = DataSourceUtils.datetimeRebaseSpec( reader.asInstanceOf[DataFileReader[_]].getMetaString, datetimeRebaseModeInRead) diff --git a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala index 2efc4f535e7b7..44520bb92d14e 100644 --- a/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala +++ b/external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala @@ -29,7 +29,7 @@ import org.apache.hadoop.io.NullWritable import org.apache.hadoop.mapreduce.{RecordWriter, TaskAttemptContext} import org.apache.spark.SPARK_VERSION_SHORT -import org.apache.spark.sql.{SPARK_LEGACY_DATETIME, SPARK_VERSION_METADATA_KEY} +import org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.execution.datasources.OutputWriter import org.apache.spark.sql.internal.SQLConf @@ -59,9 +59,11 @@ private[avro] class AvroOutputWriter( private val recordWriter: RecordWriter[AvroKey[GenericRecord], NullWritable] = { val fileMeta = Map(SPARK_VERSION_METADATA_KEY -> SPARK_VERSION_SHORT) ++ { if (datetimeRebaseMode == LegacyBehaviorPolicy.LEGACY) { - Some(SPARK_LEGACY_DATETIME -> "") + Map( + SPARK_LEGACY_DATETIME_METADATA_KEY -> "", + SPARK_TIMEZONE_METADATA_KEY -> SQLConf.get.sessionLocalTimeZone) } else { - None + Map.empty } } diff --git a/external/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala b/external/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala index b43c7faddcd93..a4dfdbfe68f9c 100644 --- a/external/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala +++ b/external/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala @@ -89,7 +89,7 @@ case class AvroPartitionReaderFactory( reader.sync(partitionedFile.start) - val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode( + val datetimeRebaseMode = DataSourceUtils.datetimeRebaseSpec( reader.asInstanceOf[DataFileReader[_]].getMetaString, datetimeRebaseModeInRead) diff --git a/external/avro/src/test/resources/before_1582_date_v3_2_0.avro b/external/avro/src/test/resources/before_1582_date_v3_2_0.avro new file mode 100644 index 0000000000000..da319f6874d91 Binary files /dev/null and b/external/avro/src/test/resources/before_1582_date_v3_2_0.avro differ diff --git a/external/avro/src/test/resources/before_1582_timestamp_micros_v3_2_0.avro b/external/avro/src/test/resources/before_1582_timestamp_micros_v3_2_0.avro new file mode 100644 index 0000000000000..27b822b4239ce Binary files /dev/null and b/external/avro/src/test/resources/before_1582_timestamp_micros_v3_2_0.avro differ diff --git a/external/avro/src/test/resources/before_1582_timestamp_millis_v3_2_0.avro b/external/avro/src/test/resources/before_1582_timestamp_millis_v3_2_0.avro new file mode 100644 index 0000000000000..e6e2b9474b139 Binary files /dev/null and b/external/avro/src/test/resources/before_1582_timestamp_millis_v3_2_0.avro differ diff --git a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala index 98a7d89ae83b2..a43d171fb52d3 100644 --- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala +++ b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala @@ -29,6 +29,7 @@ import org.apache.spark.sql.{RandomDataGenerator, Row} import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, NoopFilters, OrderedFilters, StructFilters} import org.apache.spark.sql.catalyst.expressions.{ExpressionEvalHelper, GenericInternalRow, Literal} import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData, MapData} +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.sources.{EqualTo, Not} import org.apache.spark.sql.test.SharedSparkSession @@ -285,7 +286,7 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite schema, dataType, false, - SQLConf.LegacyBehaviorPolicy.CORRECTED, + RebaseSpec(SQLConf.LegacyBehaviorPolicy.CORRECTED), filters) val deserialized = deserializer.deserialize(data) expected match { diff --git a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala index 9bf2e569f7546..890b615b18c5f 100644 --- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala +++ b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala @@ -20,7 +20,7 @@ package org.apache.spark.sql.avro import java.io._ import java.net.URI -import org.apache.avro.file.{DataFileReader} +import org.apache.avro.file.DataFileReader import org.apache.avro.generic.{GenericDatumReader, GenericRecord} import org.apache.avro.mapred.FsInput import org.apache.hadoop.conf.Configuration @@ -29,6 +29,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.SparkConf import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters} +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._ @@ -71,7 +72,7 @@ class AvroRowReaderSuite reader.getSchema, StructType(new StructField("value", IntegerType, true) :: Nil), false, - CORRECTED, + RebaseSpec(CORRECTED), new NoopFilters) override val stopPosition = fileSize diff --git a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala index 8be3cf0c2ec65..6d0a734f381ee 100644 --- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala +++ b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala @@ -21,6 +21,7 @@ import org.apache.avro.generic.GenericRecordBuilder import org.apache.spark.SparkFunSuite import org.apache.spark.sql.catalyst.NoopFilters +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy.CORRECTED import org.apache.spark.sql.types.{IntegerType, StructType} @@ -209,7 +210,12 @@ object AvroSerdeSuite { } private object Deserializer extends SerdeFactory[AvroDeserializer] { override def create(sql: StructType, avro: Schema, matchType: MatchType): AvroDeserializer = - new AvroDeserializer(avro, sql, isPositional(matchType), CORRECTED, new NoopFilters) + new AvroDeserializer( + avro, + sql, + isPositional(matchType), + RebaseSpec(CORRECTED), + new NoopFilters) } /** diff --git a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index 11588ea50cec4..d85baeb9386f2 100644 --- a/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -39,6 +39,7 @@ import org.apache.spark.sql._ import org.apache.spark.sql.TestingUDT.IntervalData import org.apache.spark.sql.catalyst.expressions.AttributeReference import org.apache.spark.sql.catalyst.plans.logical.Filter +import org.apache.spark.sql.catalyst.util.DateTimeTestUtils import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC} import org.apache.spark.sql.execution.{FormattedMode, SparkPlan} import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition} @@ -1786,10 +1787,10 @@ abstract class AvroSuite } // It generates input files for the test below: - // "SPARK-31183: compatibility with Spark 2.4 in reading dates/timestamps" - ignore("SPARK-31855: generate test files for checking compatibility with Spark 2.4") { + // "SPARK-31183, SPARK-37705: compatibility with Spark 2.4/3.2 in reading dates/timestamps" + ignore("SPARK-31855: generate test files for checking compatibility with Spark 2.4/3.2") { val resourceDir = "external/avro/src/test/resources" - val version = "2_4_6" + val version = SPARK_VERSION_SHORT.replaceAll("\\.", "_") def save( in: Seq[String], t: String, @@ -1856,8 +1857,8 @@ abstract class AvroSuite } } - test("SPARK-31183: compatibility with Spark 2.4 in reading dates/timestamps") { - // test reading the existing 2.4 files and new 3.0 files (with rebase on/off) together. + test("SPARK-31183, SPARK-37705: compatibility with Spark 2.4/3.2 in reading dates/timestamps") { + // test reading the existing 2.4/3.2 files and new 3.x files (with rebase on/off) together. def checkReadMixedFiles( fileName: String, dt: String, @@ -1865,26 +1866,26 @@ abstract class AvroSuite checkDefaultLegacyRead: String => Unit): Unit = { withTempPaths(2) { paths => paths.foreach(_.delete()) - val path2_4 = getResourceAvroFilePath(fileName) - val path3_0 = paths(0).getCanonicalPath - val path3_0_rebase = paths(1).getCanonicalPath + val oldPath = getResourceAvroFilePath(fileName) + val path3_x = paths(0).getCanonicalPath + val path3_x_rebase = paths(1).getCanonicalPath if (dt == "date") { val df = Seq(dataStr).toDF("str").select($"str".cast("date").as("dt")) - checkDefaultLegacyRead(path2_4) + checkDefaultLegacyRead(oldPath) withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> CORRECTED.toString) { - df.write.format("avro").mode("overwrite").save(path3_0) + df.write.format("avro").mode("overwrite").save(path3_x) } withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { - df.write.format("avro").save(path3_0_rebase) + df.write.format("avro").save(path3_x_rebase) } // For Avro files written by Spark 3.0, we know the writer info and don't need the config // to guide the rebase behavior. runInMode(Seq(LEGACY)) { options => checkAnswer( - spark.read.options(options).format("avro").load(path2_4, path3_0, path3_0_rebase), + spark.read.options(options).format("avro").load(oldPath, path3_x, path3_x_rebase), 1.to(3).map(_ => Row(java.sql.Date.valueOf(dataStr)))) } } else { @@ -1901,23 +1902,23 @@ abstract class AvroSuite // By default we should fail to write ancient datetime values. val e = intercept[SparkException] { - df.write.format("avro").option("avroSchema", avroSchema).save(path3_0) + df.write.format("avro").option("avroSchema", avroSchema).save(path3_x) } assert(e.getCause.getCause.getCause.isInstanceOf[SparkUpgradeException]) - checkDefaultLegacyRead(path2_4) + checkDefaultLegacyRead(oldPath) withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> CORRECTED.toString) { - df.write.format("avro").option("avroSchema", avroSchema).mode("overwrite").save(path3_0) + df.write.format("avro").option("avroSchema", avroSchema).mode("overwrite").save(path3_x) } withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { - df.write.format("avro").option("avroSchema", avroSchema).save(path3_0_rebase) + df.write.format("avro").option("avroSchema", avroSchema).save(path3_x_rebase) } // For Avro files written by Spark 3.0, we know the writer info and don't need the config // to guide the rebase behavior. runInMode(Seq(LEGACY)) { options => checkAnswer( - spark.read.options(options).format("avro").load(path2_4, path3_0, path3_0_rebase), + spark.read.options(options).format("avro").load(oldPath, path3_x, path3_x_rebase), 1.to(3).map(_ => Row(java.sql.Timestamp.valueOf(dataStr)))) } } @@ -1933,7 +1934,8 @@ abstract class AvroSuite // By default we should fail to read ancient datetime values when parquet files don't // contain Spark version. "2_4_5" -> failInRead _, - "2_4_6" -> successInRead _ + "2_4_6" -> successInRead _, + "3_2_0" -> successInRead _ ).foreach { case (version, checkDefaultRead) => checkReadMixedFiles( s"before_1582_date_v$version.avro", @@ -1953,75 +1955,94 @@ abstract class AvroSuite } } - test("SPARK-31183: rebasing microseconds timestamps in write") { - val tsStr = "1001-01-01 01:02:03.123456" - val nonRebased = "1001-01-07 01:09:05.123456" - withTempPath { dir => - val path = dir.getAbsolutePath - withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { - Seq(tsStr).toDF("tsS") - .select($"tsS".cast("timestamp").as("ts")) - .write.format("avro") - .save(path) - } + test("SPARK-31183, SPARK-37705: rebasing microseconds timestamps in write") { + // Ignore the default JVM time zone and use the session time zone instead of it in rebasing. + DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.JST) { + withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> DateTimeTestUtils.LA.getId) { + val tsStr = "1001-01-01 01:02:03.123456" + val nonRebased = "1001-01-07 01:09:05.123456" + withTempPath { dir => + val path = dir.getAbsolutePath + withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { + Seq(tsStr).toDF("tsS") + .select($"tsS".cast("timestamp").as("ts")) + .write.format("avro") + .save(path) + } - // The file metadata indicates if it needs rebase or not, so we can always get the correct - // result regardless of the "rebase mode" config. - runInMode(Seq(LEGACY, CORRECTED, EXCEPTION)) { options => - checkAnswer( - spark.read.options(options).format("avro").load(path), - Row(Timestamp.valueOf(tsStr))) - } + // The file metadata indicates if it needs rebase or not, so we can always get the correct + // result regardless of the "rebase mode" config. + runInMode(Seq(LEGACY, CORRECTED, EXCEPTION)) { options => + checkAnswer( + spark.read.options(options).format("avro").load(path).select($"ts".cast("string")), + Row(tsStr)) + } - // Force to not rebase to prove the written datetime values are rebased and we will get - // wrong result if we don't rebase while reading. - withSQLConf("spark.test.forceNoRebase" -> "true") { - checkAnswer(spark.read.format("avro").load(path), Row(Timestamp.valueOf(nonRebased))) + // Force to not rebase to prove the written datetime values are rebased and we will get + // wrong result if we don't rebase while reading. + withSQLConf("spark.test.forceNoRebase" -> "true") { + checkAnswer( + spark.read.format("avro").load(path).select($"ts".cast("string")), + Row(nonRebased)) + } + } } } } - test("SPARK-31183: rebasing milliseconds timestamps in write") { - val tsStr = "1001-01-01 01:02:03.123456" - val rebased = "1001-01-01 01:02:03.123" - val nonRebased = "1001-01-07 01:09:05.123" - Seq( - """{"type": "long","logicalType": "timestamp-millis"}""", - """"long"""").foreach { tsType => - val timestampSchema = s""" - |{ - | "namespace": "logical", - | "type": "record", - | "name": "test", - | "fields": [ - | {"name": "ts", "type": $tsType} - | ] - |}""".stripMargin - withTempPath { dir => - val path = dir.getAbsolutePath - withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { - Seq(tsStr).toDF("tsS") - .select($"tsS".cast("timestamp").as("ts")) - .write - .option("avroSchema", timestampSchema) - .format("avro") - .save(path) - } - - // The file metadata indicates if it needs rebase or not, so we can always get the correct - // result regardless of the "rebase mode" config. - runInMode(Seq(LEGACY, CORRECTED, EXCEPTION)) { options => - checkAnswer( - spark.read.options(options).schema("ts timestamp").format("avro").load(path), - Row(Timestamp.valueOf(rebased))) - } - - // Force to not rebase to prove the written datetime values are rebased and we will get - // wrong result if we don't rebase while reading. - withSQLConf("spark.test.forceNoRebase" -> "true") { - checkAnswer( - spark.read.schema("ts timestamp").format("avro").load(path), - Row(Timestamp.valueOf(nonRebased))) + test("SPARK-31183, SPARK-37705: rebasing milliseconds timestamps in write") { + // Ignore the default JVM time zone and use the session time zone instead of it in rebasing. + DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.JST) { + withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> DateTimeTestUtils.LA.getId) { + val tsStr = "1001-01-01 01:02:03.123456" + val rebased = "1001-01-01 01:02:03.123" + val nonRebased = "1001-01-07 01:09:05.123" + Seq( + """{"type": "long","logicalType": "timestamp-millis"}""", + """"long"""").foreach { tsType => + val timestampSchema = s""" + |{ + | "namespace": "logical", + | "type": "record", + | "name": "test", + | "fields": [ + | {"name": "ts", "type": $tsType} + | ] + |}""".stripMargin + withTempPath { dir => + val path = dir.getAbsolutePath + withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { + Seq(tsStr).toDF("tsS") + .select($"tsS".cast("timestamp").as("ts")) + .write + .option("avroSchema", timestampSchema) + .format("avro") + .save(path) + } + + // The file metadata indicates if it needs rebase or not, so we can always get + // the correct result regardless of the "rebase mode" config. + runInMode(Seq(LEGACY, CORRECTED, EXCEPTION)) { options => + checkAnswer( + spark.read + .options(options) + .schema("ts timestamp") + .format("avro").load(path) + .select($"ts".cast("string")), + Row(rebased)) + } + + // Force to not rebase to prove the written datetime values are rebased and we will get + // wrong result if we don't rebase while reading. + withSQLConf("spark.test.forceNoRebase" -> "true") { + checkAnswer( + spark.read + .schema("ts timestamp") + .format("avro").load(path) + .select($"ts".cast("string")), + Row(nonRebased)) + } + } } } } @@ -2069,7 +2090,8 @@ abstract class AvroSuite } } - test("SPARK-33163: write the metadata key 'org.apache.spark.legacyDateTime'") { + test("SPARK-33163, SPARK-37705: write the metadata key 'org.apache.spark.legacyDateTime' " + + "and 'org.apache.spark.timeZone'") { def saveTs(dir: java.io.File): Unit = { Seq(Timestamp.valueOf("2020-10-15 01:02:03")).toDF() .repartition(1) @@ -2080,14 +2102,16 @@ abstract class AvroSuite withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { withTempPath { dir => saveTs(dir) - checkMetaData(dir, SPARK_LEGACY_DATETIME, "") + checkMetaData(dir, SPARK_LEGACY_DATETIME_METADATA_KEY, "") + checkMetaData(dir, SPARK_TIMEZONE_METADATA_KEY, SQLConf.get.sessionLocalTimeZone) } } Seq(CORRECTED, EXCEPTION).foreach { mode => withSQLConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE.key -> mode.toString) { withTempPath { dir => saveTs(dir) - checkMetaData(dir, SPARK_LEGACY_DATETIME, null) + checkMetaData(dir, SPARK_LEGACY_DATETIME_METADATA_KEY, null) + checkMetaData(dir, SPARK_TIMEZONE_METADATA_KEY, null) } } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala index 2999d475fc8f6..72bb43b1259f9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala @@ -29,6 +29,7 @@ import com.fasterxml.jackson.module.scala.{DefaultScalaModule, ScalaObjectMapper import org.apache.spark.sql.catalyst.util.DateTimeConstants._ import org.apache.spark.sql.catalyst.util.DateTimeUtils._ +import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy import org.apache.spark.util.Utils /** @@ -37,6 +38,13 @@ import org.apache.spark.util.Utils * to/from Proleptic Gregorian calendar which is used by Spark since version 3.0. See SPARK-26651. */ object RebaseDateTime { + + // Specification of rebase operation including `mode` and the time zone in which it is performed + case class RebaseSpec(mode: LegacyBehaviorPolicy.Value, originTimeZone: Option[String] = None) { + // Use the default JVM time zone for backward compatibility + def timeZone: String = originTimeZone.getOrElse(TimeZone.getDefault.getID) + } + /** * Rebases days since the epoch from an original to an target calendar, for instance, * from a hybrid (Julian + Gregorian) to Proleptic Gregorian calendar. @@ -363,33 +371,46 @@ object RebaseDateTime { } /** - * An optimized version of [[rebaseGregorianToJulianMicros(ZoneId, Long)]]. This method leverages - * the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't contain - * information about the current JVM system time zone or `micros` is related to Before Common Era, - * the function falls back to the regular unoptimized version. - * - * Note: The function assumes that the input micros was derived from a local timestamp - * at the default system JVM time zone in Proleptic Gregorian calendar. + * An optimized version of [[rebaseGregorianToJulianMicros(TimeZone, Long)]]. This method + * leverages the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't + * contain information about the given time zone `timeZoneId` or `micros` is related to Before + * Common Era, the function falls back to the regular unoptimized version. * + * @param timeZoneId A string identifier of a time zone. * @param micros The number of microseconds since the epoch '1970-01-01T00:00:00Z' * in Proleptic Gregorian calendar. It can be negative. * @return The rebased microseconds since the epoch in Julian calendar. */ - def rebaseGregorianToJulianMicros(micros: Long): Long = { + def rebaseGregorianToJulianMicros(timeZoneId: String, micros: Long): Long = { if (micros >= lastSwitchGregorianTs) { micros } else { - val timeZone = TimeZone.getDefault - val tzId = timeZone.getID - val rebaseRecord = gregJulianRebaseMap.getOrNull(tzId) + val rebaseRecord = gregJulianRebaseMap.getOrNull(timeZoneId) if (rebaseRecord == null || micros < rebaseRecord.switches(0)) { - rebaseGregorianToJulianMicros(timeZone, micros) + rebaseGregorianToJulianMicros(TimeZone.getTimeZone(timeZoneId), micros) } else { rebaseMicros(rebaseRecord, micros) } } } + /** + * An optimized version of [[rebaseGregorianToJulianMicros(TimeZone, Long)]]. This method + * leverages the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't + * contain information about the current JVM system time zone or `micros` is related to Before + * Common Era, the function falls back to the regular unoptimized version. + * + * Note: The function assumes that the input micros was derived from a local timestamp + * at the default system JVM time zone in Proleptic Gregorian calendar. + * + * @param micros The number of microseconds since the epoch '1970-01-01T00:00:00Z' + * in Proleptic Gregorian calendar. It can be negative. + * @return The rebased microseconds since the epoch in Julian calendar. + */ + def rebaseGregorianToJulianMicros(micros: Long): Long = { + rebaseGregorianToJulianMicros(TimeZone.getDefault.getID, micros) + } + /** * Converts the given number of microseconds since the epoch '1970-01-01T00:00:00Z', to a local * date-time in Julian calendar with timezone identified by `zoneId`, interprets the result as a @@ -464,30 +485,43 @@ object RebaseDateTime { final val lastSwitchJulianTs: Long = getLastSwitchTs(julianGregRebaseMap) /** - * An optimized version of [[rebaseJulianToGregorianMicros(ZoneId, Long)]]. This method leverages - * the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't contain - * information about the current JVM system time zone or `micros` is related to Before Common Era, - * the function falls back to the regular unoptimized version. - * - * Note: The function assumes that the input micros was derived from a local timestamp - * at the default system JVM time zone in the Julian calendar. + * An optimized version of [[rebaseJulianToGregorianMicros(TimeZone, Long)]]. This method + * leverages the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't + * contain information about the given time zone `timeZoneId` or `micros` is related to Before + * Common Era, the function falls back to the regular unoptimized version. * + * @param timeZoneId A string identifier of a time zone. * @param micros The number of microseconds since the epoch '1970-01-01T00:00:00Z' * in the Julian calendar. It can be negative. * @return The rebased microseconds since the epoch in Proleptic Gregorian calendar. */ - def rebaseJulianToGregorianMicros(micros: Long): Long = { + def rebaseJulianToGregorianMicros(timeZoneId: String, micros: Long): Long = { if (micros >= lastSwitchJulianTs) { micros } else { - val timeZone = TimeZone.getDefault - val tzId = timeZone.getID - val rebaseRecord = julianGregRebaseMap.getOrNull(tzId) + val rebaseRecord = julianGregRebaseMap.getOrNull(timeZoneId) if (rebaseRecord == null || micros < rebaseRecord.switches(0)) { - rebaseJulianToGregorianMicros(timeZone, micros) + rebaseJulianToGregorianMicros(TimeZone.getTimeZone(timeZoneId), micros) } else { rebaseMicros(rebaseRecord, micros) } } } + + /** + * An optimized version of [[rebaseJulianToGregorianMicros(TimeZone, Long)]]. This method + * leverages the pre-calculated rebasing maps to save calculation. If the rebasing map doesn't + * contain information about the current JVM system time zone or `micros` is related to Before + * Common Era, the function falls back to the regular unoptimized version. + * + * Note: The function assumes that the input micros was derived from a local timestamp + * at the default system JVM time zone in the Julian calendar. + * + * @param micros The number of microseconds since the epoch '1970-01-01T00:00:00Z' + * in the Julian calendar. It can be negative. + * @return The rebased microseconds since the epoch in Proleptic Gregorian calendar. + */ + def rebaseJulianToGregorianMicros(micros: Long): Long = { + rebaseJulianToGregorianMicros(TimeZone.getDefault.getID, micros) + } } diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java index 5a5a099992179..cce53001ea621 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java @@ -45,17 +45,23 @@ public class ParquetVectorUpdaterFactory { // The timezone conversion to apply to int96 timestamps. Null if no conversion. private final ZoneId convertTz; private final String datetimeRebaseMode; + private final String datetimeRebaseTz; private final String int96RebaseMode; + private final String int96RebaseTz; ParquetVectorUpdaterFactory( LogicalTypeAnnotation logicalTypeAnnotation, ZoneId convertTz, String datetimeRebaseMode, - String int96RebaseMode) { + String datetimeRebaseTz, + String int96RebaseMode, + String int96RebaseTz) { this.logicalTypeAnnotation = logicalTypeAnnotation; this.convertTz = convertTz; this.datetimeRebaseMode = datetimeRebaseMode; + this.datetimeRebaseTz = datetimeRebaseTz; this.int96RebaseMode = int96RebaseMode; + this.int96RebaseTz = int96RebaseTz; } public ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType sparkType) { @@ -109,7 +115,7 @@ public ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType spa return new LongUpdater(); } else { boolean failIfRebase = "EXCEPTION".equals(datetimeRebaseMode); - return new LongWithRebaseUpdater(failIfRebase); + return new LongWithRebaseUpdater(failIfRebase, datetimeRebaseTz); } } else if (isTimestampTypeMatched(LogicalTypeAnnotation.TimeUnit.MILLIS)) { validateTimestampType(sparkType); @@ -117,7 +123,7 @@ public ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType spa return new LongAsMicrosUpdater(); } else { final boolean failIfRebase = "EXCEPTION".equals(datetimeRebaseMode); - return new LongAsMicrosRebaseUpdater(failIfRebase); + return new LongAsMicrosRebaseUpdater(failIfRebase, datetimeRebaseTz); } } else if (sparkType instanceof DayTimeIntervalType) { return new LongUpdater(); @@ -142,13 +148,16 @@ public ParquetVectorUpdater getUpdater(ColumnDescriptor descriptor, DataType spa if ("CORRECTED".equals(int96RebaseMode)) { return new BinaryToSQLTimestampUpdater(); } else { - return new BinaryToSQLTimestampRebaseUpdater(failIfRebase); + return new BinaryToSQLTimestampRebaseUpdater(failIfRebase, int96RebaseTz); } } else { if ("CORRECTED".equals(int96RebaseMode)) { return new BinaryToSQLTimestampConvertTzUpdater(convertTz); } else { - return new BinaryToSQLTimestampConvertTzRebaseUpdater(failIfRebase, convertTz); + return new BinaryToSQLTimestampConvertTzRebaseUpdater( + failIfRebase, + convertTz, + int96RebaseTz); } } } @@ -517,9 +526,11 @@ public void decodeSingleDictionaryId( private static class LongWithRebaseUpdater implements ParquetVectorUpdater { private final boolean failIfRebase; + private final String timeZone; - LongWithRebaseUpdater(boolean failIfRebase) { + LongWithRebaseUpdater(boolean failIfRebase, String timeZone) { this.failIfRebase = failIfRebase; + this.timeZone = timeZone; } @Override @@ -528,7 +539,7 @@ public void readValues( int offset, WritableColumnVector values, VectorizedValuesReader valuesReader) { - valuesReader.readLongsWithRebase(total, values, offset, failIfRebase); + valuesReader.readLongsWithRebase(total, values, offset, failIfRebase, timeZone); } @Override @@ -542,7 +553,7 @@ public void readValue( WritableColumnVector values, VectorizedValuesReader valuesReader) { long julianMicros = valuesReader.readLong(); - values.putLong(offset, rebaseMicros(julianMicros, failIfRebase)); + values.putLong(offset, rebaseMicros(julianMicros, failIfRebase, timeZone)); } @Override @@ -552,7 +563,7 @@ public void decodeSingleDictionaryId( WritableColumnVector dictionaryIds, Dictionary dictionary) { long julianMicros = dictionary.decodeToLong(dictionaryIds.getDictId(offset)); - values.putLong(offset, rebaseMicros(julianMicros, failIfRebase)); + values.putLong(offset, rebaseMicros(julianMicros, failIfRebase, timeZone)); } } @@ -594,9 +605,11 @@ public void decodeSingleDictionaryId( private static class LongAsMicrosRebaseUpdater implements ParquetVectorUpdater { private final boolean failIfRebase; + private final String timeZone; - LongAsMicrosRebaseUpdater(boolean failIfRebase) { + LongAsMicrosRebaseUpdater(boolean failIfRebase, String timeZone) { this.failIfRebase = failIfRebase; + this.timeZone = timeZone; } @Override @@ -621,7 +634,7 @@ public void readValue( WritableColumnVector values, VectorizedValuesReader valuesReader) { long julianMicros = DateTimeUtils.millisToMicros(valuesReader.readLong()); - values.putLong(offset, rebaseMicros(julianMicros, failIfRebase)); + values.putLong(offset, rebaseMicros(julianMicros, failIfRebase, timeZone)); } @Override @@ -632,7 +645,7 @@ public void decodeSingleDictionaryId( Dictionary dictionary) { long julianMillis = dictionary.decodeToLong(dictionaryIds.getDictId(offset)); long julianMicros = DateTimeUtils.millisToMicros(julianMillis); - values.putLong(offset, rebaseMicros(julianMicros, failIfRebase)); + values.putLong(offset, rebaseMicros(julianMicros, failIfRebase, timeZone)); } } @@ -823,9 +836,11 @@ public void decodeSingleDictionaryId( private static class BinaryToSQLTimestampRebaseUpdater implements ParquetVectorUpdater { private final boolean failIfRebase; + private final String timeZone; - BinaryToSQLTimestampRebaseUpdater(boolean failIfRebase) { + BinaryToSQLTimestampRebaseUpdater(boolean failIfRebase, String timeZone) { this.failIfRebase = failIfRebase; + this.timeZone = timeZone; } @Override @@ -851,7 +866,7 @@ public void readValue( VectorizedValuesReader valuesReader) { // Read 12 bytes for INT96 long julianMicros = ParquetRowConverter.binaryToSQLTimestamp(valuesReader.readBinary(12)); - long gregorianMicros = rebaseInt96(julianMicros, failIfRebase); + long gregorianMicros = rebaseInt96(julianMicros, failIfRebase, timeZone); values.putLong(offset, gregorianMicros); } @@ -863,7 +878,7 @@ public void decodeSingleDictionaryId( Dictionary dictionary) { Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(offset)); long julianMicros = ParquetRowConverter.binaryToSQLTimestamp(v); - long gregorianMicros = rebaseInt96(julianMicros, failIfRebase); + long gregorianMicros = rebaseInt96(julianMicros, failIfRebase, timeZone); values.putLong(offset, gregorianMicros); } } @@ -871,10 +886,15 @@ public void decodeSingleDictionaryId( private static class BinaryToSQLTimestampConvertTzRebaseUpdater implements ParquetVectorUpdater { private final boolean failIfRebase; private final ZoneId convertTz; + private final String timeZone; - BinaryToSQLTimestampConvertTzRebaseUpdater(boolean failIfRebase, ZoneId convertTz) { + BinaryToSQLTimestampConvertTzRebaseUpdater( + boolean failIfRebase, + ZoneId convertTz, + String timeZone) { this.failIfRebase = failIfRebase; this.convertTz = convertTz; + this.timeZone = timeZone; } @Override @@ -900,7 +920,7 @@ public void readValue( VectorizedValuesReader valuesReader) { // Read 12 bytes for INT96 long julianMicros = ParquetRowConverter.binaryToSQLTimestamp(valuesReader.readBinary(12)); - long gregorianMicros = rebaseInt96(julianMicros, failIfRebase); + long gregorianMicros = rebaseInt96(julianMicros, failIfRebase, timeZone); long adjTime = DateTimeUtils.convertTz(gregorianMicros, convertTz, UTC); values.putLong(offset, adjTime); } @@ -913,7 +933,7 @@ public void decodeSingleDictionaryId( Dictionary dictionary) { Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(offset)); long julianMicros = ParquetRowConverter.binaryToSQLTimestamp(v); - long gregorianMicros = rebaseInt96(julianMicros, failIfRebase); + long gregorianMicros = rebaseInt96(julianMicros, failIfRebase, timeZone); long adjTime = DateTimeUtils.convertTz(gregorianMicros, convertTz, UTC); values.putLong(offset, adjTime); } @@ -1062,7 +1082,8 @@ private static int rebaseDays(int julianDays, final boolean failIfRebase) { private static long rebaseTimestamp( long julianMicros, final boolean failIfRebase, - final String format) { + final String format, + final String timeZone) { if (failIfRebase) { if (julianMicros < RebaseDateTime.lastSwitchJulianTs()) { throw DataSourceUtils.newRebaseExceptionInRead(format); @@ -1070,16 +1091,22 @@ private static long rebaseTimestamp( return julianMicros; } } else { - return RebaseDateTime.rebaseJulianToGregorianMicros(julianMicros); + return RebaseDateTime.rebaseJulianToGregorianMicros(timeZone, julianMicros); } } - private static long rebaseMicros(long julianMicros, final boolean failIfRebase) { - return rebaseTimestamp(julianMicros, failIfRebase, "Parquet"); + private static long rebaseMicros( + long julianMicros, + final boolean failIfRebase, + final String timeZone) { + return rebaseTimestamp(julianMicros, failIfRebase, "Parquet", timeZone); } - private static long rebaseInt96(long julianMicros, final boolean failIfRebase) { - return rebaseTimestamp(julianMicros, failIfRebase, "Parquet INT96"); + private static long rebaseInt96( + long julianMicros, + final boolean failIfRebase, + final String timeZone) { + return rebaseTimestamp(julianMicros, failIfRebase, "Parquet INT96", timeZone); } private boolean shouldConvertTimestamps() { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java index 92dea08102dfc..cedb239541c71 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java @@ -93,13 +93,20 @@ public VectorizedColumnReader( PrimitiveIterator.OfLong rowIndexes, ZoneId convertTz, String datetimeRebaseMode, - String int96RebaseMode) throws IOException { + String datetimeRebaseTz, + String int96RebaseMode, + String int96RebaseTz) throws IOException { this.descriptor = descriptor; this.pageReader = pageReader; this.readState = new ParquetReadState(descriptor.getMaxDefinitionLevel(), rowIndexes); this.logicalTypeAnnotation = logicalTypeAnnotation; this.updaterFactory = new ParquetVectorUpdaterFactory( - logicalTypeAnnotation, convertTz, datetimeRebaseMode, int96RebaseMode); + logicalTypeAnnotation, + convertTz, + datetimeRebaseMode, + datetimeRebaseTz, + int96RebaseMode, + int96RebaseTz); DictionaryPage dictionaryPage = pageReader.readDictionaryPage(); if (dictionaryPage != null) { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java index 9e0868b7550f5..50056bf4073e9 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java @@ -94,11 +94,15 @@ public class VectorizedParquetRecordReader extends SpecificParquetRecordReaderBa * The mode of rebasing date/timestamp from Julian to Proleptic Gregorian calendar. */ private final String datetimeRebaseMode; + // The time zone Id in which rebasing of date/timestamp is performed + private final String datetimeRebaseTz; /** * The mode of rebasing INT96 timestamp from Julian to Proleptic Gregorian calendar. */ private final String int96RebaseMode; + // The time zone Id in which rebasing of INT96 is performed + private final String int96RebaseTz; /** * columnBatch object that is used for batch decoding. This is created on first use and triggers @@ -131,19 +135,30 @@ public class VectorizedParquetRecordReader extends SpecificParquetRecordReaderBa public VectorizedParquetRecordReader( ZoneId convertTz, String datetimeRebaseMode, + String datetimeRebaseTz, String int96RebaseMode, + String int96RebaseTz, boolean useOffHeap, int capacity) { this.convertTz = convertTz; this.datetimeRebaseMode = datetimeRebaseMode; + this.datetimeRebaseTz = datetimeRebaseTz; this.int96RebaseMode = int96RebaseMode; + this.int96RebaseTz = int96RebaseTz; MEMORY_MODE = useOffHeap ? MemoryMode.OFF_HEAP : MemoryMode.ON_HEAP; this.capacity = capacity; } // For test only. public VectorizedParquetRecordReader(boolean useOffHeap, int capacity) { - this(null, "CORRECTED", "LEGACY", useOffHeap, capacity); + this( + null, + "CORRECTED", + "UTC", + "LEGACY", + ZoneId.systemDefault().getId(), + useOffHeap, + capacity); } /** @@ -350,7 +365,9 @@ private void checkEndOfRowGroup() throws IOException { pages.getRowIndexes().orElse(null), convertTz, datetimeRebaseMode, - int96RebaseMode); + datetimeRebaseTz, + int96RebaseMode, + int96RebaseTz); } totalCountLoadedSoFar += pages.getRowCount(); } diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java index 0eb5d65a4a8f7..fb40a131d2a6b 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java @@ -206,7 +206,11 @@ public final void readUnsignedLongs(int total, WritableColumnVector c, int rowId // if rebase is not needed. @Override public final void readLongsWithRebase( - int total, WritableColumnVector c, int rowId, boolean failIfRebase) { + int total, + WritableColumnVector c, + int rowId, + boolean failIfRebase, + String timeZone) { int requiredBytes = total * 8; ByteBuffer buffer = getBuffer(requiredBytes); boolean rebase = false; @@ -218,7 +222,9 @@ public final void readLongsWithRebase( throw DataSourceUtils.newRebaseExceptionInRead("Parquet"); } else { for (int i = 0; i < total; i += 1) { - c.putLong(rowId + i, RebaseDateTime.rebaseJulianToGregorianMicros(buffer.getLong())); + c.putLong( + rowId + i, + RebaseDateTime.rebaseJulianToGregorianMicros(timeZone, buffer.getLong())); } } } else { diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java index af739a52d8ed0..cd97fb6c3cd55 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java @@ -354,7 +354,11 @@ public void readLongs(int total, WritableColumnVector c, int rowId) { @Override public void readLongsWithRebase( - int total, WritableColumnVector c, int rowId, boolean failIfRebase) { + int total, + WritableColumnVector c, + int rowId, + boolean failIfRebase, + String timeZone) { throw new UnsupportedOperationException("only readInts is valid."); } diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java index fc4eac94d1c46..92161f1f59cc1 100644 --- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java @@ -46,7 +46,12 @@ public interface VectorizedValuesReader { void readUnsignedIntegers(int total, WritableColumnVector c, int rowId); void readUnsignedLongs(int total, WritableColumnVector c, int rowId); void readLongs(int total, WritableColumnVector c, int rowId); - void readLongsWithRebase(int total, WritableColumnVector c, int rowId, boolean failIfRebase); + void readLongsWithRebase( + int total, + WritableColumnVector c, + int rowId, + boolean failIfRebase, + String timeZone); void readFloats(int total, WritableColumnVector c, int rowId); void readDoubles(int total, WritableColumnVector c, int rowId); void readBinary(int total, WritableColumnVector c, int rowId); diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala index 76bc03ad5e946..6ceb44ab15020 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala @@ -26,10 +26,11 @@ import org.json4s.NoTypeHints import org.json4s.jackson.Serialization import org.apache.spark.SparkUpgradeException -import org.apache.spark.sql.{SPARK_LEGACY_DATETIME, SPARK_LEGACY_INT96, SPARK_VERSION_METADATA_KEY} +import org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY} import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils} import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, Expression, ExpressionSet, PredicateHelper} import org.apache.spark.sql.catalyst.util.RebaseDateTime +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors} import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions import org.apache.spark.sql.internal.SQLConf @@ -109,44 +110,53 @@ object DataSourceUtils extends PredicateHelper { case _ => false } - def datetimeRebaseMode( + private def getRebaseSpec( lookupFileMeta: String => String, - modeByConfig: String): LegacyBehaviorPolicy.Value = { - if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") { - return LegacyBehaviorPolicy.CORRECTED + modeByConfig: String, + minVersion: String, + metadataKey: String): RebaseSpec = { + val policy = if (Utils.isTesting && + SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") { + LegacyBehaviorPolicy.CORRECTED + } else { + // If there is no version, we return the mode specified by the config. + Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version => + // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to + // rebase the datetime values. + // Files written by `minVersion` and latter may also need the rebase if they were written + // with the "LEGACY" rebase mode. + if (version < minVersion || lookupFileMeta(metadataKey) != null) { + LegacyBehaviorPolicy.LEGACY + } else { + LegacyBehaviorPolicy.CORRECTED + } + }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig)) + } + policy match { + case LegacyBehaviorPolicy.LEGACY => + RebaseSpec(LegacyBehaviorPolicy.LEGACY, Option(lookupFileMeta(SPARK_TIMEZONE_METADATA_KEY))) + case _ => RebaseSpec(policy) } - // If there is no version, we return the mode specified by the config. - Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version => - // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to - // rebase the datetime values. - // Files written by Spark 3.0 and latter may also need the rebase if they were written with - // the "LEGACY" rebase mode. - if (version < "3.0.0" || lookupFileMeta(SPARK_LEGACY_DATETIME) != null) { - LegacyBehaviorPolicy.LEGACY - } else { - LegacyBehaviorPolicy.CORRECTED - } - }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig)) } - def int96RebaseMode( + def datetimeRebaseSpec( lookupFileMeta: String => String, - modeByConfig: String): LegacyBehaviorPolicy.Value = { - if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") { - return LegacyBehaviorPolicy.CORRECTED - } - // If there is no version, we return the mode specified by the config. - Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version => - // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to - // rebase the INT96 timestamp values. - // Files written by Spark 3.1 and latter may also need the rebase if they were written with - // the "LEGACY" rebase mode. - if (version < "3.1.0" || lookupFileMeta(SPARK_LEGACY_INT96) != null) { - LegacyBehaviorPolicy.LEGACY - } else { - LegacyBehaviorPolicy.CORRECTED - } - }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig)) + modeByConfig: String): RebaseSpec = { + getRebaseSpec( + lookupFileMeta, + modeByConfig, + "3.0.0", + SPARK_LEGACY_DATETIME_METADATA_KEY) + } + + def int96RebaseSpec( + lookupFileMeta: String => String, + modeByConfig: String): RebaseSpec = { + getRebaseSpec( + lookupFileMeta, + modeByConfig, + "3.1.0", + SPARK_LEGACY_INT96_METADATA_KEY) } def newRebaseExceptionInRead(format: String): SparkUpgradeException = { @@ -197,14 +207,15 @@ object DataSourceUtils extends PredicateHelper { } def createTimestampRebaseFuncInRead( - rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Long => Long = rebaseMode match { + rebaseSpec: RebaseSpec, + format: String): Long => Long = rebaseSpec.mode match { case LegacyBehaviorPolicy.EXCEPTION => micros: Long => if (micros < RebaseDateTime.lastSwitchJulianTs) { throw DataSourceUtils.newRebaseExceptionInRead(format) } micros - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseJulianToGregorianMicros + case LegacyBehaviorPolicy.LEGACY => + RebaseDateTime.rebaseJulianToGregorianMicros(rebaseSpec.timeZone, _) case LegacyBehaviorPolicy.CORRECTED => identity[Long] } @@ -216,7 +227,9 @@ object DataSourceUtils extends PredicateHelper { throw DataSourceUtils.newRebaseExceptionInWrite(format) } micros - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseGregorianToJulianMicros + case LegacyBehaviorPolicy.LEGACY => + val timeZone = SQLConf.get.sessionLocalTimeZone + RebaseDateTime.rebaseGregorianToJulianMicros(timeZone, _) case LegacyBehaviorPolicy.CORRECTED => identity[Long] } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala index b400e3688f4b7..4515387bdaa90 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala @@ -266,7 +266,7 @@ class ParquetFileFormat lazy val footerFileMetaData = ParquetFooterReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData - val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode( + val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec( footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead) // Try to push down filters when filter push-down is enabled. @@ -280,7 +280,7 @@ class ParquetFileFormat pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive, - datetimeRebaseMode) + datetimeRebaseSpec) filters // Collects all converted Parquet filter predicates. Notice that not all predicates can be // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap` @@ -306,7 +306,7 @@ class ParquetFileFormat None } - val int96RebaseMode = DataSourceUtils.int96RebaseMode( + val int96RebaseSpec = DataSourceUtils.int96RebaseSpec( footerFileMetaData.getKeyValueMetaData.get, int96RebaseModeInRead) @@ -323,8 +323,10 @@ class ParquetFileFormat if (enableVectorizedReader) { val vectorizedReader = new VectorizedParquetRecordReader( convertTz.orNull, - datetimeRebaseMode.toString, - int96RebaseMode.toString, + datetimeRebaseSpec.mode.toString, + datetimeRebaseSpec.timeZone, + int96RebaseSpec.mode.toString, + int96RebaseSpec.timeZone, enableOffHeapColumnVector && taskContext.isDefined, capacity) // SPARK-37089: We cannot register a task completion listener to close this iterator here @@ -358,8 +360,8 @@ class ParquetFileFormat val readSupport = new ParquetReadSupport( convertTz, enableVectorizedReader = false, - datetimeRebaseMode, - int96RebaseMode) + datetimeRebaseSpec, + int96RebaseSpec) val reader = if (pushed.isDefined && enableRecordFilter) { val parquetFilter = FilterCompat.get(pushed.get, null) new ParquetRecordReader[InternalRow](readSupport, parquetFilter) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index c81d8adc078ec..75060cfca24e6 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -35,7 +35,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName._ import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils, IntervalUtils} -import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros} +import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros, RebaseSpec} import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy import org.apache.spark.sql.sources import org.apache.spark.unsafe.types.UTF8String @@ -51,7 +51,7 @@ class ParquetFilters( pushDownStartWith: Boolean, pushDownInFilterThreshold: Int, caseSensitive: Boolean, - datetimeRebaseMode: LegacyBehaviorPolicy.Value) { + datetimeRebaseSpec: RebaseSpec) { // A map which contains parquet field name and data type, if predicate push down applies. // // Each key in `nameToParquetField` represents a column; `dots` are used as separators for @@ -137,7 +137,7 @@ class ParquetFilters( case d: Date => DateTimeUtils.fromJavaDate(d) case ld: LocalDate => DateTimeUtils.localDateToDays(ld) } - datetimeRebaseMode match { + datetimeRebaseSpec.mode match { case LegacyBehaviorPolicy.LEGACY => rebaseGregorianToJulianDays(gregorianDays) case _ => gregorianDays } @@ -148,8 +148,9 @@ class ParquetFilters( case i: Instant => DateTimeUtils.instantToMicros(i) case t: Timestamp => DateTimeUtils.fromJavaTimestamp(t) } - datetimeRebaseMode match { - case LegacyBehaviorPolicy.LEGACY => rebaseGregorianToJulianMicros(gregorianMicros) + datetimeRebaseSpec.mode match { + case LegacyBehaviorPolicy.LEGACY => + rebaseGregorianToJulianMicros(datetimeRebaseSpec.timeZone, gregorianMicros) case _ => gregorianMicros } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala index 597a1bba8c628..bdab0f7892f00 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala @@ -32,6 +32,7 @@ import org.apache.parquet.schema.Type.Repetition import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy @@ -56,8 +57,8 @@ import org.apache.spark.sql.types._ class ParquetReadSupport( val convertTz: Option[ZoneId], enableVectorizedReader: Boolean, - datetimeRebaseMode: LegacyBehaviorPolicy.Value, - int96RebaseMode: LegacyBehaviorPolicy.Value) + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec) extends ReadSupport[InternalRow] with Logging { private var catalystRequestedSchema: StructType = _ @@ -68,8 +69,8 @@ class ParquetReadSupport( this( None, enableVectorizedReader = true, - datetimeRebaseMode = LegacyBehaviorPolicy.CORRECTED, - int96RebaseMode = LegacyBehaviorPolicy.LEGACY) + datetimeRebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED), + int96RebaseSpec = RebaseSpec(LegacyBehaviorPolicy.LEGACY)) } /** @@ -138,8 +139,8 @@ class ParquetReadSupport( ParquetReadSupport.expandUDT(catalystRequestedSchema), new ParquetToSparkSchemaConverter(conf), convertTz, - datetimeRebaseMode, - int96RebaseMode) + datetimeRebaseSpec, + int96RebaseSpec) } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala index 80763ef019b01..2206b43b3d701 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala @@ -23,7 +23,7 @@ import org.apache.parquet.io.api.{GroupConverter, RecordMaterializer} import org.apache.parquet.schema.MessageType import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.types.StructType /** @@ -33,18 +33,18 @@ import org.apache.spark.sql.types.StructType * @param catalystSchema Catalyst schema of the rows to be constructed * @param schemaConverter A Parquet-Catalyst schema converter that helps initializing row converters * @param convertTz the optional time zone to convert to int96 data - * @param datetimeRebaseMode the mode of rebasing date/timestamp from Julian to Proleptic Gregorian - * calendar - * @param int96RebaseMode the mode of rebasing INT96 timestamp from Julian to Proleptic Gregorian - * calendar + * @param datetimeRebaseSpec the specification of rebasing date/timestamp from Julian to Proleptic + * Gregorian calendar: mode + optional original time zone + * @param int96RebaseSpec the specification of rebasing INT96 timestamp from Julian to Proleptic + * Gregorian calendar */ private[parquet] class ParquetRecordMaterializer( parquetSchema: MessageType, catalystSchema: StructType, schemaConverter: ParquetToSparkSchemaConverter, convertTz: Option[ZoneId], - datetimeRebaseMode: LegacyBehaviorPolicy.Value, - int96RebaseMode: LegacyBehaviorPolicy.Value) + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec) extends RecordMaterializer[InternalRow] { private val rootConverter = new ParquetRowConverter( @@ -52,8 +52,8 @@ private[parquet] class ParquetRecordMaterializer( parquetSchema, catalystSchema, convertTz, - datetimeRebaseMode, - int96RebaseMode, + datetimeRebaseSpec, + int96RebaseSpec, NoopUpdater) override def getCurrentRecord: InternalRow = rootConverter.currentRecord diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala index 243bc164183d4..b12898360dcf4 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala @@ -35,10 +35,10 @@ import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, CaseInsensitiveMap, DateTimeUtils, GenericArrayData} +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.execution.datasources.DataSourceUtils import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String @@ -127,10 +127,10 @@ private[parquet] class ParquetPrimitiveConverter(val updater: ParentContainerUpd * @param catalystType Spark SQL schema that corresponds to the Parquet record type. User-defined * types should have been expanded. * @param convertTz the optional time zone to convert to int96 data - * @param datetimeRebaseMode the mode of rebasing date/timestamp from Julian to Proleptic Gregorian - * calendar - * @param int96RebaseMode the mode of rebasing INT96 timestamp from Julian to Proleptic Gregorian - * calendar + * @param datetimeRebaseSpec the specification of rebasing date/timestamp from Julian to Proleptic + * Gregorian calendar: mode + optional original time zone + * @param int96RebaseSpec the specification of rebasing INT96 timestamp from Julian to Proleptic + * Gregorian calendar * @param updater An updater which propagates converted field values to the parent container */ private[parquet] class ParquetRowConverter( @@ -138,8 +138,8 @@ private[parquet] class ParquetRowConverter( parquetType: GroupType, catalystType: StructType, convertTz: Option[ZoneId], - datetimeRebaseMode: LegacyBehaviorPolicy.Value, - int96RebaseMode: LegacyBehaviorPolicy.Value, + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec, updater: ParentContainerUpdater) extends ParquetGroupConverter(updater) with Logging { @@ -191,13 +191,13 @@ private[parquet] class ParquetRowConverter( def currentRecord: InternalRow = currentRow private val dateRebaseFunc = DataSourceUtils.createDateRebaseFuncInRead( - datetimeRebaseMode, "Parquet") + datetimeRebaseSpec.mode, "Parquet") private val timestampRebaseFunc = DataSourceUtils.createTimestampRebaseFuncInRead( - datetimeRebaseMode, "Parquet") + datetimeRebaseSpec, "Parquet") private val int96RebaseFunc = DataSourceUtils.createTimestampRebaseFuncInRead( - int96RebaseMode, "Parquet INT96") + int96RebaseSpec, "Parquet INT96") // Converters for each field. private[this] val fieldConverters: Array[Converter with HasParentContainerUpdater] = { @@ -443,8 +443,8 @@ private[parquet] class ParquetRowConverter( parquetType.asGroupType(), t, convertTz, - datetimeRebaseMode, - int96RebaseMode, + datetimeRebaseSpec, + int96RebaseSpec, wrappedUpdater) case t => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala index 5bd712e9c583c..87a0d9c860f31 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala @@ -32,6 +32,7 @@ import org.apache.spark.SparkException import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.JoinedRow +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.connector.expressions.aggregate.{Aggregation, Count, CountStar, Max, Min} import org.apache.spark.sql.execution.datasources.AggregatePushDownUtils import org.apache.spark.sql.internal.SQLConf.{LegacyBehaviorPolicy, PARQUET_AGGREGATE_PUSHDOWN_ENABLED} @@ -159,7 +160,7 @@ object ParquetUtils { aggregation: Aggregation, aggSchema: StructType, partitionValues: InternalRow, - datetimeRebaseMode: LegacyBehaviorPolicy.Value): InternalRow = { + datetimeRebaseSpec: RebaseSpec): InternalRow = { val (primitiveTypes, values) = getPushedDownAggResult( footer, filePath, dataSchema, partitionSchema, aggregation) @@ -173,8 +174,14 @@ object ParquetUtils { AggregatePushDownUtils.getSchemaWithoutGroupingExpression(aggSchema, aggregation) val schemaConverter = new ParquetToSparkSchemaConverter - val converter = new ParquetRowConverter(schemaConverter, parquetSchema, schemaWithoutGroupBy, - None, datetimeRebaseMode, LegacyBehaviorPolicy.CORRECTED, NoopUpdater) + val converter = new ParquetRowConverter( + schemaConverter, + parquetSchema, + schemaWithoutGroupBy, + None, + datetimeRebaseSpec, + RebaseSpec(LegacyBehaviorPolicy.CORRECTED), + NoopUpdater) val primitiveTypeNames = primitiveTypes.map(_.getPrimitiveTypeName) primitiveTypeNames.zipWithIndex.foreach { case (PrimitiveType.PrimitiveTypeName.BOOLEAN, i) => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala index 183ea394d8b15..9d38c967cb031 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala @@ -31,7 +31,7 @@ import org.apache.parquet.io.api.{Binary, RecordConsumer} import org.apache.spark.SPARK_VERSION_SHORT import org.apache.spark.internal.Logging -import org.apache.spark.sql.{SPARK_LEGACY_DATETIME, SPARK_LEGACY_INT96, SPARK_VERSION_METADATA_KEY} +import org.apache.spark.sql.{SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.SpecializedGetters import org.apache.spark.sql.catalyst.util.DateTimeUtils @@ -117,15 +117,19 @@ class ParquetWriteSupport extends WriteSupport[InternalRow] with Logging { ParquetReadSupport.SPARK_METADATA_KEY -> schemaString ) ++ { if (datetimeRebaseMode == LegacyBehaviorPolicy.LEGACY) { - Some(SPARK_LEGACY_DATETIME -> "") + Map( + SPARK_LEGACY_DATETIME_METADATA_KEY -> "", + SPARK_TIMEZONE_METADATA_KEY -> SQLConf.get.sessionLocalTimeZone) } else { - None + Map.empty } } ++ { if (int96RebaseMode == LegacyBehaviorPolicy.LEGACY) { - Some(SPARK_LEGACY_INT96 -> "") + Map( + SPARK_LEGACY_INT96_METADATA_KEY -> "", + SPARK_TIMEZONE_METADATA_KEY -> SQLConf.get.sessionLocalTimeZone) } else { - None + Map.empty } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala index ee94e990b2703..41ee98a4f47b8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala @@ -34,13 +34,13 @@ import org.apache.spark.broadcast.Broadcast import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.util.DateTimeUtils +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.connector.expressions.aggregate.Aggregation import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader} import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, DataSourceUtils, PartitionedFile, RecordReaderIterator} import org.apache.spark.sql.execution.datasources.parquet._ import org.apache.spark.sql.execution.datasources.v2._ import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy import org.apache.spark.sql.sources.Filter import org.apache.spark.sql.types.{AtomicType, StructType} import org.apache.spark.sql.vectorized.ColumnarBatch @@ -96,9 +96,9 @@ case class ParquetPartitionReaderFactory( } } - private def getDatetimeRebaseMode( - footerFileMetaData: FileMetaData): LegacyBehaviorPolicy.Value = { - DataSourceUtils.datetimeRebaseMode( + private def getDatetimeRebaseSpec( + footerFileMetaData: FileMetaData): RebaseSpec = { + DataSourceUtils.datetimeRebaseSpec( footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead) } @@ -133,7 +133,7 @@ case class ParquetPartitionReaderFactory( if (footer != null && footer.getBlocks.size > 0) { ParquetUtils.createAggInternalRowFromFooter(footer, file.filePath, dataSchema, partitionSchema, aggregation.get, readDataSchema, file.partitionValues, - getDatetimeRebaseMode(footer.getFileMetaData)) + getDatetimeRebaseSpec(footer.getFileMetaData)) } else { null } @@ -176,7 +176,7 @@ case class ParquetPartitionReaderFactory( if (footer != null && footer.getBlocks.size > 0) { val row = ParquetUtils.createAggInternalRowFromFooter(footer, file.filePath, dataSchema, partitionSchema, aggregation.get, readDataSchema, file.partitionValues, - getDatetimeRebaseMode(footer.getFileMetaData)) + getDatetimeRebaseSpec(footer.getFileMetaData)) AggregatePushDownUtils.convertAggregatesRowToBatch( row, readDataSchema, enableOffHeapColumnVector && Option(TaskContext.get()).isDefined) } else { @@ -204,15 +204,15 @@ case class ParquetPartitionReaderFactory( buildReaderFunc: ( FileSplit, InternalRow, TaskAttemptContextImpl, Option[FilterPredicate], Option[ZoneId], - LegacyBehaviorPolicy.Value, - LegacyBehaviorPolicy.Value) => RecordReader[Void, T]): RecordReader[Void, T] = { + RebaseSpec, + RebaseSpec) => RecordReader[Void, T]): RecordReader[Void, T] = { val conf = broadcastedConf.value.value val filePath = new Path(new URI(file.filePath)) val split = new FileSplit(filePath, file.start, file.length, Array.empty[String]) lazy val footerFileMetaData = getFooter(file).getFileMetaData - val datetimeRebaseMode = getDatetimeRebaseMode(footerFileMetaData) + val datetimeRebaseSpec = getDatetimeRebaseSpec(footerFileMetaData) // Try to push down filters when filter push-down is enabled. val pushed = if (enableParquetFilterPushDown) { val parquetSchema = footerFileMetaData.getSchema @@ -224,7 +224,7 @@ case class ParquetPartitionReaderFactory( pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive, - datetimeRebaseMode) + datetimeRebaseSpec) filters // Collects all converted Parquet filter predicates. Notice that not all predicates can be // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap` @@ -257,7 +257,7 @@ case class ParquetPartitionReaderFactory( if (pushed.isDefined) { ParquetInputFormat.setFilterPredicate(hadoopAttemptContext.getConfiguration, pushed.get) } - val int96RebaseMode = DataSourceUtils.int96RebaseMode( + val int96RebaseSpec = DataSourceUtils.int96RebaseSpec( footerFileMetaData.getKeyValueMetaData.get, int96RebaseModeInRead) val reader = buildReaderFunc( @@ -266,8 +266,8 @@ case class ParquetPartitionReaderFactory( hadoopAttemptContext, pushed, convertTz, - datetimeRebaseMode, - int96RebaseMode) + datetimeRebaseSpec, + int96RebaseSpec) reader.initialize(split, hadoopAttemptContext) reader } @@ -282,16 +282,16 @@ case class ParquetPartitionReaderFactory( hadoopAttemptContext: TaskAttemptContextImpl, pushed: Option[FilterPredicate], convertTz: Option[ZoneId], - datetimeRebaseMode: LegacyBehaviorPolicy.Value, - int96RebaseMode: LegacyBehaviorPolicy.Value): RecordReader[Void, InternalRow] = { + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec): RecordReader[Void, InternalRow] = { logDebug(s"Falling back to parquet-mr") val taskContext = Option(TaskContext.get()) // ParquetRecordReader returns InternalRow val readSupport = new ParquetReadSupport( convertTz, enableVectorizedReader = false, - datetimeRebaseMode, - int96RebaseMode) + datetimeRebaseSpec, + int96RebaseSpec) val reader = if (pushed.isDefined && enableRecordFilter) { val parquetFilter = FilterCompat.get(pushed.get, null) new ParquetRecordReader[InternalRow](readSupport, parquetFilter) @@ -317,13 +317,15 @@ case class ParquetPartitionReaderFactory( hadoopAttemptContext: TaskAttemptContextImpl, pushed: Option[FilterPredicate], convertTz: Option[ZoneId], - datetimeRebaseMode: LegacyBehaviorPolicy.Value, - int96RebaseMode: LegacyBehaviorPolicy.Value): VectorizedParquetRecordReader = { + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec): VectorizedParquetRecordReader = { val taskContext = Option(TaskContext.get()) val vectorizedReader = new VectorizedParquetRecordReader( convertTz.orNull, - datetimeRebaseMode.toString, - int96RebaseMode.toString, + datetimeRebaseSpec.mode.toString, + datetimeRebaseSpec.timeZone, + int96RebaseSpec.mode.toString, + int96RebaseSpec.timeZone, enableOffHeapColumnVector && taskContext.isDefined, capacity) val iter = new RecordReaderIterator(vectorizedReader) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala index 74d11b62b4c26..1f2f75aebd7bf 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.v2.parquet import scala.collection.JavaConverters._ import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.connector.expressions.aggregate.Aggregation import org.apache.spark.sql.connector.read.{Scan, SupportsPushDownAggregates} import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex} @@ -65,7 +66,7 @@ case class ParquetScanBuilder( isCaseSensitive, // The rebase mode doesn't matter here because the filters are used to determine // whether they is convertible. - LegacyBehaviorPolicy.CORRECTED) + RebaseSpec(LegacyBehaviorPolicy.CORRECTED)) parquetFilters.convertibleFilters(pushedDataFilters).toArray } else { Array.empty[Filter] diff --git a/sql/core/src/main/scala/org/apache/spark/sql/package.scala b/sql/core/src/main/scala/org/apache/spark/sql/package.scala index 022fecf1ae412..1794ac513749f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/package.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/package.scala @@ -55,15 +55,22 @@ package object sql { */ private[sql] val SPARK_VERSION_METADATA_KEY = "org.apache.spark.version" + /** + * The metadata key which is used to write the current session time zone into: + * - Parquet file metadata + * - Avro file metadata + */ + private[sql] val SPARK_TIMEZONE_METADATA_KEY = "org.apache.spark.timeZone" + /** * Parquet/Avro file metadata key to indicate that the file was written with legacy datetime * values. */ - private[sql] val SPARK_LEGACY_DATETIME = "org.apache.spark.legacyDateTime" + private[sql] val SPARK_LEGACY_DATETIME_METADATA_KEY = "org.apache.spark.legacyDateTime" /** * Parquet file metadata key to indicate that the file with INT96 column type was written * with rebasing. */ - private[sql] val SPARK_LEGACY_INT96 = "org.apache.spark.legacyINT96" + private[sql] val SPARK_LEGACY_INT96_METADATA_KEY = "org.apache.spark.legacyINT96" } diff --git a/sql/core/src/test/resources/test-data/before_1582_date_v3_2_0.snappy.parquet b/sql/core/src/test/resources/test-data/before_1582_date_v3_2_0.snappy.parquet new file mode 100644 index 0000000000000..25df53eb3d472 Binary files /dev/null and b/sql/core/src/test/resources/test-data/before_1582_date_v3_2_0.snappy.parquet differ diff --git a/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_dict_v3_2_0.snappy.parquet b/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_dict_v3_2_0.snappy.parquet new file mode 100644 index 0000000000000..78120fdb23337 Binary files /dev/null and b/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_dict_v3_2_0.snappy.parquet differ diff --git a/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_plain_v3_2_0.snappy.parquet b/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_plain_v3_2_0.snappy.parquet new file mode 100644 index 0000000000000..3f5f85e244c05 Binary files /dev/null and b/sql/core/src/test/resources/test-data/before_1582_timestamp_int96_plain_v3_2_0.snappy.parquet differ diff --git a/sql/core/src/test/resources/test-data/before_1582_timestamp_micros_v3_2_0.snappy.parquet b/sql/core/src/test/resources/test-data/before_1582_timestamp_micros_v3_2_0.snappy.parquet new file mode 100644 index 0000000000000..809f6f50e0926 Binary files /dev/null and b/sql/core/src/test/resources/test-data/before_1582_timestamp_micros_v3_2_0.snappy.parquet differ diff --git a/sql/core/src/test/resources/test-data/before_1582_timestamp_millis_v3_2_0.snappy.parquet b/sql/core/src/test/resources/test-data/before_1582_timestamp_millis_v3_2_0.snappy.parquet new file mode 100644 index 0000000000000..f9cfa4a1c0aa8 Binary files /dev/null and b/sql/core/src/test/resources/test-data/before_1582_timestamp_millis_v3_2_0.snappy.parquet differ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index 04c0a68bc4435..9b554b626df85 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -39,6 +39,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints import org.apache.spark.sql.catalyst.planning.PhysicalOperation +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath import org.apache.spark.sql.execution.ExplainMode import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, HadoopFsRelation, LogicalRelation, PushableColumnAndNestedColumn} @@ -77,13 +78,13 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared protected def createParquetFilters( schema: MessageType, caseSensitive: Option[Boolean] = None, - datetimeRebaseMode: LegacyBehaviorPolicy.Value = LegacyBehaviorPolicy.CORRECTED + datetimeRebaseSpec: RebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED) ): ParquetFilters = new ParquetFilters(schema, conf.parquetFilterPushDownDate, conf.parquetFilterPushDownTimestamp, conf.parquetFilterPushDownDecimal, conf.parquetFilterPushDownStringStartWith, conf.parquetFilterPushDownInFilterThreshold, caseSensitive.getOrElse(conf.caseSensitiveAnalysis), - datetimeRebaseMode) + datetimeRebaseSpec) override def beforeEach(): Unit = { super.beforeEach() diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRebaseDatetimeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRebaseDatetimeSuite.scala index e0c71cde51b4d..d9683fa47ebe6 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRebaseDatetimeSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRebaseDatetimeSuite.scala @@ -20,8 +20,8 @@ package org.apache.spark.sql.execution.datasources.parquet import java.nio.file.{Files, Paths, StandardCopyOption} import java.sql.{Date, Timestamp} -import org.apache.spark.{SparkConf, SparkException, SparkUpgradeException} -import org.apache.spark.sql.{QueryTest, Row, SPARK_LEGACY_DATETIME, SPARK_LEGACY_INT96} +import org.apache.spark.{SPARK_VERSION_SHORT, SparkConf, SparkException, SparkUpgradeException} +import org.apache.spark.sql.{QueryTest, Row, SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY} import org.apache.spark.sql.catalyst.util.DateTimeTestUtils import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.SQLConf.{LegacyBehaviorPolicy, ParquetOutputTimestampType} @@ -37,10 +37,10 @@ abstract class ParquetRebaseDatetimeSuite import testImplicits._ // It generates input files for the test below: - // "SPARK-31159: compatibility with Spark 2.4 in reading dates/timestamps" - ignore("SPARK-31806: generate test files for checking compatibility with Spark 2.4") { + // "SPARK-31159, SPARK-37705: compatibility with Spark 2.4/3.2 in reading dates/timestamps" + ignore("SPARK-31806: generate test files for checking compatibility with Spark 2.4/3.2") { val resourceDir = "sql/core/src/test/resources/test-data" - val version = "2_4_5" + val version = SPARK_VERSION_SHORT.replaceAll("\\.", "_") val N = 8 def save( in: Seq[(String, String)], @@ -62,7 +62,10 @@ abstract class ParquetRebaseDatetimeSuite } } DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.LA) { - withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> DateTimeTestUtils.LA.getId) { + withSQLConf( + SQLConf.SESSION_LOCAL_TIMEZONE.key -> DateTimeTestUtils.LA.getId, + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> LEGACY.toString, + SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { save( (1 to N).map(i => ("1001-01-01", s"1001-01-0$i")), "date", @@ -119,7 +122,7 @@ abstract class ParquetRebaseDatetimeSuite } } - test("SPARK-31159: compatibility with Spark 2.4 in reading dates/timestamps") { + test("SPARK-31159, SPARK-37705: compatibility with Spark 2.4/3.2 in reading dates/timestamps") { val N = 8 // test reading the existing 2.4 files and new 3.0 files (with rebase on/off) together. def checkReadMixedFiles[T]( @@ -133,25 +136,25 @@ abstract class ParquetRebaseDatetimeSuite inReadConf: String = SQLConf.PARQUET_REBASE_MODE_IN_READ.key): Unit = { withTempPaths(2) { paths => paths.foreach(_.delete()) - val path2_4 = getResourceParquetFilePath("test-data/" + fileName) - val path3_0 = paths(0).getCanonicalPath - val path3_0_rebase = paths(1).getCanonicalPath + val oldPath = getResourceParquetFilePath("test-data/" + fileName) + val path3_x = paths(0).getCanonicalPath + val path3_x_rebase = paths(1).getCanonicalPath val df = Seq.tabulate(N)(rowFunc).toDF("dict", "plain") .select($"dict".cast(catalystType), $"plain".cast(catalystType)) withSQLConf(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> tsOutputType) { - checkDefaultLegacyRead(path2_4) + checkDefaultLegacyRead(oldPath) withSQLConf(inWriteConf -> CORRECTED.toString) { - df.write.mode("overwrite").parquet(path3_0) + df.write.mode("overwrite").parquet(path3_x) } withSQLConf(inWriteConf -> LEGACY.toString) { - df.write.parquet(path3_0_rebase) + df.write.parquet(path3_x_rebase) } } // For Parquet files written by Spark 3.0, we know the writer info and don't need the // config to guide the rebase behavior. runInMode(inReadConf, Seq(LEGACY)) { options => checkAnswer( - spark.read.format("parquet").options(options).load(path2_4, path3_0, path3_0_rebase), + spark.read.format("parquet").options(options).load(oldPath, path3_x, path3_x_rebase), (0 until N).flatMap { i => val (dictS, plainS) = rowFunc(i) Seq.tabulate(3) { _ => @@ -170,7 +173,8 @@ abstract class ParquetRebaseDatetimeSuite // By default we should fail to read ancient datetime values when parquet files don't // contain Spark version. "2_4_5" -> failInRead _, - "2_4_6" -> successInRead _).foreach { case (version, checkDefaultRead) => + "2_4_6" -> successInRead _, + "3_2_0" -> successInRead _).foreach { case (version, checkDefaultRead) => withAllParquetReaders { checkReadMixedFiles( s"before_1582_date_v$version.snappy.parquet", @@ -212,7 +216,7 @@ abstract class ParquetRebaseDatetimeSuite } } - test("SPARK-31159: rebasing timestamps in write") { + test("SPARK-31159, SPARK-37705: rebasing timestamps in write") { val N = 8 Seq(false, true).foreach { dictionaryEncoding => Seq( @@ -236,34 +240,39 @@ abstract class ParquetRebaseDatetimeSuite SQLConf.PARQUET_INT96_REBASE_MODE_IN_READ.key ) ).foreach { case (outType, tsStr, nonRebased, inWriteConf, inReadConf) => - withClue(s"output type $outType") { - withSQLConf(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outType) { - withTempPath { dir => - val path = dir.getAbsolutePath - withSQLConf(inWriteConf -> LEGACY.toString) { - Seq.tabulate(N)(_ => tsStr).toDF("tsS") - .select($"tsS".cast("timestamp").as("ts")) - .repartition(1) - .write - .option("parquet.enable.dictionary", dictionaryEncoding) - .parquet(path) - } + // Ignore the default JVM time zone and use the session time zone instead of it in rebasing. + DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.JST) { + withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> DateTimeTestUtils.LA.getId) { + withClue(s"output type $outType") { + withSQLConf(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outType) { + withTempPath { dir => + val path = dir.getAbsolutePath + withSQLConf(inWriteConf -> LEGACY.toString) { + Seq.tabulate(N)(_ => tsStr).toDF("tsS") + .select($"tsS".cast("timestamp").as("ts")) + .repartition(1) + .write + .option("parquet.enable.dictionary", dictionaryEncoding) + .parquet(path) + } - withAllParquetReaders { - // The file metadata indicates if it needs rebase or not, so we can always get the - // correct result regardless of the "rebase mode" config. - runInMode(inReadConf, Seq(LEGACY, CORRECTED, EXCEPTION)) { options => - checkAnswer( - spark.read.options(options).parquet(path), - Seq.tabulate(N)(_ => Row(Timestamp.valueOf(tsStr)))) - } + withAllParquetReaders { + // The file metadata indicates if it needs rebase or not, so we can always get + // the correct result regardless of the "rebase mode" config. + runInMode(inReadConf, Seq(LEGACY, CORRECTED, EXCEPTION)) { options => + checkAnswer( + spark.read.options(options).parquet(path).select($"ts".cast("string")), + Seq.tabulate(N)(_ => Row(tsStr))) + } - // Force to not rebase to prove the written datetime values are rebased - // and we will get wrong result if we don't rebase while reading. - withSQLConf("spark.test.forceNoRebase" -> "true") { - checkAnswer( - spark.read.parquet(path), - Seq.tabulate(N)(_ => Row(Timestamp.valueOf(nonRebased)))) + // Force to not rebase to prove the written datetime values are rebased + // and we will get wrong result if we don't rebase while reading. + withSQLConf("spark.test.forceNoRebase" -> "true") { + checkAnswer( + spark.read.parquet(path).select($"ts".cast("string")), + Seq.tabulate(N)(_ => Row(nonRebased))) + } + } } } } @@ -310,7 +319,8 @@ abstract class ParquetRebaseDatetimeSuite } } - test("SPARK-33163: write the metadata key 'org.apache.spark.legacyDateTime'") { + test("SPARK-33163, SPARK-37705: write the metadata keys 'org.apache.spark.legacyDateTime' " + + "and 'org.apache.spark.timeZone'") { def checkMetadataKey(dir: java.io.File, exists: Boolean): Unit = { Seq("timestamp '1000-01-01 01:02:03'", "date '1000-01-01'").foreach { dt => withSQLConf(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> @@ -322,7 +332,9 @@ abstract class ParquetRebaseDatetimeSuite .parquet(dir.getAbsolutePath) val metaData = getMetaData(dir) val expected = if (exists) Some("") else None - assert(metaData.get(SPARK_LEGACY_DATETIME) === expected) + assert(metaData.get(SPARK_LEGACY_DATETIME_METADATA_KEY) === expected) + val expectedTz = if (exists) Some(SQLConf.get.sessionLocalTimeZone) else None + assert(metaData.get(SPARK_TIMEZONE_METADATA_KEY) === expectedTz) } } } @@ -341,7 +353,8 @@ abstract class ParquetRebaseDatetimeSuite } } - test("SPARK-33160: write the metadata key 'org.apache.spark.legacyINT96'") { + test("SPARK-33160, SPARK-37705: write the metadata key 'org.apache.spark.legacyINT96' " + + "and 'org.apache.spark.timeZone'") { def saveTs(dir: java.io.File, ts: String = "1000-01-01 01:02:03"): Unit = { Seq(Timestamp.valueOf(ts)).toDF() .repartition(1) @@ -351,13 +364,15 @@ abstract class ParquetRebaseDatetimeSuite withSQLConf(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> LEGACY.toString) { withTempPath { dir => saveTs(dir) - assert(getMetaData(dir)(SPARK_LEGACY_INT96) === "") + assert(getMetaData(dir)(SPARK_LEGACY_INT96_METADATA_KEY) === "") + assert(getMetaData(dir)(SPARK_TIMEZONE_METADATA_KEY) === SQLConf.get.sessionLocalTimeZone) } } withSQLConf(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> CORRECTED.toString) { withTempPath { dir => saveTs(dir) - assert(getMetaData(dir).get(SPARK_LEGACY_INT96).isEmpty) + assert(getMetaData(dir).get(SPARK_LEGACY_INT96_METADATA_KEY).isEmpty) + assert(getMetaData(dir).get(SPARK_TIMEZONE_METADATA_KEY).isEmpty) } } withSQLConf(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> EXCEPTION.toString) { @@ -366,7 +381,8 @@ abstract class ParquetRebaseDatetimeSuite withSQLConf(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> EXCEPTION.toString) { withTempPath { dir => saveTs(dir, "2020-10-22 01:02:03") - assert(getMetaData(dir).get(SPARK_LEGACY_INT96).isEmpty) + assert(getMetaData(dir).get(SPARK_LEGACY_INT96_METADATA_KEY).isEmpty) + assert(getMetaData(dir).get(SPARK_TIMEZONE_METADATA_KEY).isEmpty) } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala index e03eda2d9412b..36a52b60688e9 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorizedSuite.scala @@ -293,7 +293,13 @@ class ParquetVectorizedSuite extends QueryTest with ParquetTest with SharedSpark import collection.JavaConverters._ val recordReader = new VectorizedParquetRecordReader( - DateTimeUtils.getZoneId("EST"), "CORRECTED", "CORRECTED", true, batchSize) + DateTimeUtils.getZoneId("EST"), + "CORRECTED", + "UTC", + "CORRECTED", + "UTC", + true, + batchSize) recordReader.initialize(fileSchema, fileSchema, TestParquetRowGroupReader(Seq(readStore)), totalRowCount)