Skip to content

[SPARK-37705][SQL] Rebase timestamps in the session time zone saved in Parquet/Avro metadata #34973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MaxGekk wants to merge 26 commits into from
Closed

[SPARK-37705][SQL] Rebase timestamps in the session time zone saved in Parquet/Avro metadata #34973

Show file tree
Hide file tree
Changes from 25 commits
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
158e6cb
Write the metadata key 'org.apache.spark.timeZone' to parquet files
MaxGekk Dec 21, 2021
7e3ee92
Update tests
MaxGekk Dec 21, 2021
c6d3a64
Pass time zone id to the optimized version of rebaseJulianToGregorian…
MaxGekk Dec 21, 2021
8ec7883
Pass time zone id to the optimized version of rebaseGregorianToJulian…
MaxGekk Dec 21, 2021
b510ff1
Use the session time zone in timestamp rebasing
MaxGekk Dec 21, 2021
609717c
Merge remote-tracking branch 'origin/master' into parquet-rebase-save-tz
MaxGekk Dec 22, 2021
c440169
Write the session time zone to Avro metadata
MaxGekk Dec 22, 2021
37dc837
SPARK_LEGACY_DATETIME -> SPARK_LEGACY_DATETIME_METADATA_KEY
MaxGekk Dec 22, 2021
08d4f78
SPARK_LEGACY_INT96 -> SPARK_LEGACY_INT96_METADATA_KEY
MaxGekk Dec 22, 2021
70bcd62
Extract common code to getRebaseMode()
MaxGekk Dec 22, 2021
8d186ad
Add RebaseSpec
MaxGekk Dec 22, 2021
3a28d07
getRebaseMode -> getRebaseSpec
MaxGekk Dec 22, 2021
e01a29f
Use saved time zone in loading
MaxGekk Dec 22, 2021
21e688b
Change the Parquet Vectorized reader
MaxGekk Dec 22, 2021
64c39a7
Make Java style checker happy
MaxGekk Dec 22, 2021
ece9afc
Modified a test to check JVM and session time zones
MaxGekk Dec 23, 2021
2a3fd19
Test loading of mixed files
MaxGekk Dec 23, 2021
6282a9e
Add parquet files
MaxGekk Dec 24, 2021
605f350
Check rebasing microseconds/milliseconds in Avro
MaxGekk Dec 24, 2021
68a3504
Check backward compatibility in Avro
MaxGekk Dec 24, 2021
e06f5f4
Fix comments
MaxGekk Dec 24, 2021
0f77adb
Fix indentations
MaxGekk Dec 27, 2021
12a894a
Merge remote-tracking branch 'origin/master' into parquet-rebase-save-tz
MaxGekk Dec 27, 2021
8eebcf8
Re-gen Avro golden files by Spark 3.2.0
MaxGekk Dec 27, 2021
7b734e4
Re-gen Parquet golden files by Spark 3.2.0
MaxGekk Dec 27, 2021
5b7de02
Use SPARK_VERSION_SHORT in generating golden files
MaxGekk Dec 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,18 +34,20 @@ 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.
*/
private[sql] class AvroDeserializer(
rootAvroType: Schema,
rootCatalystType: DataType,
positionalFieldMatch: Boolean,
datetimeRebaseMode: LegacyBehaviorPolicy.Value,
datetimeRebaseSpec: RebaseSpec,
filters: StructFilters) {

def this(
Expand All @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion external/avro/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
8 changes: 5 additions & 3 deletions external/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}
}

Expand Down
2 changes: 1 addition & 1 deletion external/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ case class AvroPartitionReaderFactory(

reader.sync(partitionedFile.start)

val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode(
val datetimeRebaseMode = DataSourceUtils.datetimeRebaseSpec(
reader.asInstanceOf[DataFileReader[_]].getMetaString,
datetimeRebaseModeInRead)

Expand Down
Binary file added external/avro/src/test/resources/before_1582_date_v3_2_0.avro
View file
Open in desktop
Binary file not shown.
Binary file added external/avro/src/test/resources/before_1582_timestamp_micros_v3_2_0.avro
View file
Open in desktop
Binary file not shown.
Binary file added external/avro/src/test/resources/before_1582_timestamp_millis_v3_2_0.avro
View file
Open in desktop
Binary file not shown.
3 changes: 2 additions & 1 deletion external/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down
5 changes: 3 additions & 2 deletions external/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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._
Expand Down Expand Up @@ -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

Expand Down
8 changes: 7 additions & 1 deletion external/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}

Expand Down Expand Up @@ -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)
}

/**
Expand Down
Loading