This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ef3a470 [SPARK-37705][SQL] Rebase timestamps in the session time zone
saved in Parquet/Avro metadata
ef3a470 is described below
commit ef3a47038606ea426c15844b0400f5141acd5108
Author: Max Gekk <[email protected]>
AuthorDate: Tue Dec 28 21:39:26 2021 +0800
[SPARK-37705][SQL] Rebase timestamps in the session time zone saved in
Parquet/Avro metadata
### What changes were proposed in this pull request?
In the PR, I propose to add new metadata key `org.apache.spark.timeZone`
which Spark writes to Parquet/Avro matadata while performing of datetimes
rebase in the `LEGACY` mode (see the SQL configs:
- `spark.sql.parquet.datetimeRebaseModeInWrite`,
- `spark.sql.parquet.int96RebaseModeInWrite` and
- `spark.sql.avro.datetimeRebaseModeInWrite`).
The writers uses the current session time zone (see the SQL config
`spark.sql.session.timeZone`) in rebasing of Parquet/Avro timestamp columns.
At the reader side, Spark tries to get info about the writer's time zone
from the new metadata property:
```
$ java -jar ~parquet-tools-1.12.0.jar meta
./part-00000-b0d90bf0-ce60-4b4f-b453-b33f61ab2b2a-c000.snappy.parquet
...
extra: org.apache.spark.timeZone = America/Los_Angeles
extra: org.apache.spark.legacyDateTime =
```
and use it in rebasing timestamps to the Proleptic Gregorian calendar. In
the case when the reader cannot retrieve the original time zone from
Parquet/Avro metadata, it uses the default JVM time zone for backward
compatibility.
### Why are the changes needed?
Before the changes, Spark assumes that a writer uses the default JVM time
zone while rebasing of dates/timestamps. And if a reader and the writer have
different JVM time zone settings, the reader cannot load such columns in the
`LEGACY` mode correctly. So, the reader will have full info about writer
settings after the changes.
### Does this PR introduce _any_ user-facing change?
Yes. After the changes, Parquet/Avro writers use the session time zone
while timestamp rebasing in the `LEGACY` mode instead of the default JVM time
zone. Need to highlight that the session time zone is set to the JVM time zone
by default.
### How was this patch tested?
1. By running new tests:
```
$ build/sbt "test:testOnly *ParquetRebaseDatetimeV1Suite"
$ build/sbt "test:testOnly *ParquetRebaseDatetimeV2Suite"
$ build/sbt "test:testOnly *AvroV1Suite"
$ build/sbt "test:testOnly *AvroV2Suite"
```
2. And related existing test suites:
```
$ build/sbt "test:testOnly *DateTimeUtilsSuite"
$ build/sbt "test:testOnly *RebaseDateTimeSuite"
$ build/sbt "test:testOnly *TimestampFormatterSuite"
$ build/sbt "avro/test:testOnly
org.apache.spark.sql.avro.AvroCatalystDataConversionSuite"
$ build/sbt "test:testOnly *AvroRowReaderSuite"
$ build/sbt "test:testOnly *AvroSerdeSuite"
$ build/sbt "test:testOnly *ParquetVectorizedSuite"
```
3. Also modified the test `SPARK-31159: rebasing timestamps in write` to
check loading timestamps in the LEGACY mode when the session time zone and JVM
time zone are different.
4. Generated parquet files by Spark 3.2.0 (the commit
https://github.com/apache/spark/commit/5d45a415f3a29898d92380380cfd82bfc7f579ea)
using the test `"SPARK-31806: generate test files for checking compatibility
with Spark 2.4"`. The parquet files don't contain info about the original time
zone:
```
$ java -jar ~/Downloads/parquet-tools-1.12.0.jar meta
sql/core/src/test/resources/test-data/before_1582_timestamp_micros_v3_2_0.snappy.parquet
file:
file:/Users/maximgekk/proj/parquet-rebase-save-tz/sql/core/src/test/resources/test-data/before_1582_timestamp_micros_v3_2_0.snappy.parquet
creator: parquet-mr version 1.12.1 (build
2a5c06c58fa987f85aa22170be14d927d5ff6e7d)
extra: org.apache.spark.version = 3.2.0
extra: org.apache.spark.legacyINT96 =
extra: org.apache.spark.sql.parquet.row.metadata =
{"type":"struct","fields":[{"name":"dict","type":"timestamp","nullable":true,"metadata":{}},{"name":"plain","type":"timestamp","nullable":true,"metadata":{}}]}
extra: org.apache.spark.legacyDateTime =
file schema: spark_schema
--------------------------------------------------------------------------------
dict: OPTIONAL INT64 L:TIMESTAMP(MICROS,true) R:0 D:1
plain: OPTIONAL INT64 L:TIMESTAMP(MICROS,true) R:0 D:1
```
By running the test `"SPARK-31159, SPARK-37705: compatibility with Spark
2.4/3.2 in reading dates/timestamps"`, check loading of mixed parquet files
generated by Spark 2.4.5/2.4.6 and 3.2.0/master.
5. Generated avro files by Spark 3.2.0 (the commit
https://github.com/apache/spark/commit/5d45a415f3a29898d92380380cfd82bfc7f579ea)
using the test `"SPARK-31855: generate test files for checking compatibility
with Spark 2.4"`. The avro files don't contain info about the original time
zone:
```
$ java -jar ~/Downloads/avro-tools-1.9.2.jar getmeta
external/avro/src/test/resources/before_1582_timestamp_micros_v3_2_0.avro
avro.schema
{"type":"record","name":"topLevelRecord","fields":[{"name":"dt","type":[{"type":"long","logicalType":"timestamp-micros"},"null"]}]}
org.apache.spark.version 3.2.0
avro.codec snappy
org.apache.spark.legacyDateTime
```
By running the test `"SPARK-31159, SPARK-37705: compatibility with Spark
2.4/3.2 in reading dates/timestamps"`, check loading of mixed avro files
generated by Spark 2.4.5/2.4.6 and 3.2.0/master.
Closes #34973 from MaxGekk/parquet-rebase-save-tz.
Authored-by: Max Gekk <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../apache/spark/sql/avro/AvroDeserializer.scala | 10 +-
.../org/apache/spark/sql/avro/AvroFileFormat.scala | 2 +-
.../apache/spark/sql/avro/AvroOutputWriter.scala | 8 +-
.../sql/v2/avro/AvroPartitionReaderFactory.scala | 2 +-
.../test/resources/before_1582_date_v3_2_0.avro | Bin 0 -> 264 bytes
.../before_1582_timestamp_micros_v3_2_0.avro | Bin 0 -> 282 bytes
.../before_1582_timestamp_millis_v3_2_0.avro | Bin 0 -> 308 bytes
.../sql/avro/AvroCatalystDataConversionSuite.scala | 3 +-
.../apache/spark/sql/avro/AvroRowReaderSuite.scala | 5 +-
.../org/apache/spark/sql/avro/AvroSerdeSuite.scala | 8 +-
.../org/apache/spark/sql/avro/AvroSuite.scala | 192 ++++++++++++---------
.../spark/sql/catalyst/util/RebaseDateTime.scala | 82 ++++++---
.../parquet/ParquetVectorUpdaterFactory.java | 75 +++++---
.../parquet/VectorizedColumnReader.java | 11 +-
.../parquet/VectorizedParquetRecordReader.java | 21 ++-
.../parquet/VectorizedPlainValuesReader.java | 10 +-
.../parquet/VectorizedRleValuesReader.java | 6 +-
.../parquet/VectorizedValuesReader.java | 7 +-
.../execution/datasources/DataSourceUtils.scala | 89 ++++++----
.../datasources/parquet/ParquetFileFormat.scala | 16 +-
.../datasources/parquet/ParquetFilters.scala | 11 +-
.../datasources/parquet/ParquetReadSupport.scala | 13 +-
.../parquet/ParquetRecordMaterializer.scala | 18 +-
.../datasources/parquet/ParquetRowConverter.scala | 24 +--
.../datasources/parquet/ParquetUtils.scala | 13 +-
.../datasources/parquet/ParquetWriteSupport.scala | 14 +-
.../v2/parquet/ParquetPartitionReaderFactory.scala | 44 ++---
.../v2/parquet/ParquetScanBuilder.scala | 3 +-
.../main/scala/org/apache/spark/sql/package.scala | 11 +-
.../before_1582_date_v3_2_0.snappy.parquet | Bin 0 -> 822 bytes
...1582_timestamp_int96_dict_v3_2_0.snappy.parquet | Bin 0 -> 842 bytes
...582_timestamp_int96_plain_v3_2_0.snappy.parquet | Bin 0 -> 779 bytes
...ore_1582_timestamp_micros_v3_2_0.snappy.parquet | Bin 0 -> 929 bytes
...ore_1582_timestamp_millis_v3_2_0.snappy.parquet | Bin 0 -> 922 bytes
.../datasources/parquet/ParquetFilterSuite.scala | 5 +-
.../parquet/ParquetRebaseDatetimeSuite.scala | 112 ++++++------
.../parquet/ParquetVectorizedSuite.scala | 8 +-
37 files changed, 510 insertions(+), 313 deletions(-)
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 54cd3ba..5bb51a9 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 398cb02..a13e062 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 2efc4f5..44520bb 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 b43c7fa..a4dfdbf 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 0000000..da319f6
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 0000000..27b822b
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 0000000..e6e2b94
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 98a7d89..a43d171 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 9bf2e56..890b615 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 8be3cf0..6d0a734 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 11588ea..d85baeb 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 2999d47..72bb43b 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,27 +371,23 @@ 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)
}
@@ -391,6 +395,23 @@ object RebaseDateTime {
}
/**
+ * 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
* local date-time in Proleptic Gregorian calendar with the same timezone,
and takes microseconds
@@ -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 5a5a099..cce5300 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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 @@ public class ParquetVectorUpdaterFactory {
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 @@ public class ParquetVectorUpdaterFactory {
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 92dea08..cedb2395 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 class 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 9e0868b..50056bf 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 @@ public class VectorizedParquetRecordReader extends
SpecificParquetRecordReaderBa
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 0eb5d65..fb40a13 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 class VectorizedPlainValuesReader extends
ValuesReader implements Vectori
// 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 class VectorizedPlainValuesReader extends
ValuesReader implements Vectori
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 af739a5..cd97fb6 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 final class VectorizedRleValuesReader extends
ValuesReader
@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 fc4eac9..92161f1 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 76bc03a..6ceb44a 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 b400e36..4515387 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 c81d8ad..75060cf 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 597a1bb..bdab0f7 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 80763ef..2206b43b 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 243bc16..b128983 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 5bd712e..87a0d9c 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 183ea39..9d38c96 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 ee94e99..41ee98a 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 74d11b6..1f2f75aeb 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 022fecf..1794ac5 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
@@ -56,14 +56,21 @@ 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 0000000..25df53e
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 0000000..78120fd
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 0000000..3f5f85e
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 0000000..809f6f5
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 0000000..f9cfa4a
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 04c0a68..9b554b6 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 e0c71cd..d9683fa 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 e03eda2..36a52b6 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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
