address comment

Author: cloud-fan

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -55,11 +55,11 @@ class AvroDeserializer(
 
   private lazy val decimalConversions = new DecimalConversion()
 
-  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFunc(
-    datetimeRebaseMode, "Avro", isRead = true)
+  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFuncInRead(
+    datetimeRebaseMode, "Avro")
 
-  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFunc(
-    datetimeRebaseMode, "Avro", isRead = true)
+  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFuncInRead(
+    datetimeRebaseMode, "Avro")
 
   private val converter: Any => Any = rootCatalystType match {
     // A shortcut for empty schema.

external/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala

@@ -59,11 +59,11 @@ class AvroSerializer(
     converter.apply(catalystData)
   }
 
-  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFunc(
-    datetimeRebaseMode, "Avro", isRead = false)
+  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFuncInWrite(
+    datetimeRebaseMode, "Avro")
 
-  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFunc(
-    datetimeRebaseMode, "Avro", isRead = false)
+  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFuncInWrite(
+    datetimeRebaseMode, "Avro")
 
   private val converter: Any => Any = {
     val actualAvroType = resolveNullableType(rootAvroType, nullable)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala

@@ -143,47 +143,51 @@ object DataSourceUtils {
       "Gregorian calendar.", null)
   }
 
-  def creteDateRebaseFunc(
+  def creteDateRebaseFuncInRead(
       rebaseMode: LegacyBehaviorPolicy.Value,
-      format: String,
-      isRead: Boolean): Int => Int = rebaseMode match {
-    case LegacyBehaviorPolicy.EXCEPTION if isRead => days: Int =>
+      format: String): Int => Int = rebaseMode match {
+    case LegacyBehaviorPolicy.EXCEPTION => days: Int =>
       if (days < RebaseDateTime.lastSwitchJulianDay) {
         throw DataSourceUtils.newRebaseExceptionInRead(format)
       }
       days
+    case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseJulianToGregorianDays
+    case LegacyBehaviorPolicy.CORRECTED => identity[Int]
+  }
+
+  def creteDateRebaseFuncInWrite(
+      rebaseMode: LegacyBehaviorPolicy.Value,
+      format: String): Int => Int = rebaseMode match {
     case LegacyBehaviorPolicy.EXCEPTION => days: Int =>
       if (days < RebaseDateTime.lastSwitchGregorianDay) {
         throw DataSourceUtils.newRebaseExceptionInWrite(format)
       }
       days
-    case LegacyBehaviorPolicy.LEGACY => if (isRead) {
-      RebaseDateTime.rebaseJulianToGregorianDays
-    } else {
-      RebaseDateTime.rebaseGregorianToJulianDays
-    }
+    case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseGregorianToJulianDays
     case LegacyBehaviorPolicy.CORRECTED => identity[Int]
   }
 
-  def creteTimestampRebaseFunc(
+  def creteTimestampRebaseFuncInRead(
       rebaseMode: LegacyBehaviorPolicy.Value,
-      format: String,
-      isRead: Boolean): Long => Long = rebaseMode match {
-    case LegacyBehaviorPolicy.EXCEPTION if isRead => micros: Long =>
+      format: String): Long => Long = rebaseMode match {
+    case LegacyBehaviorPolicy.EXCEPTION => micros: Long =>
       if (micros < RebaseDateTime.lastSwitchJulianTs) {
         throw DataSourceUtils.newRebaseExceptionInRead(format)
       }
       micros
+    case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseJulianToGregorianMicros
+    case LegacyBehaviorPolicy.CORRECTED => identity[Long]
+  }
+
+  def creteTimestampRebaseFuncInWrite(
+      rebaseMode: LegacyBehaviorPolicy.Value,
+      format: String): Long => Long = rebaseMode match {
     case LegacyBehaviorPolicy.EXCEPTION => micros: Long =>
       if (micros < RebaseDateTime.lastSwitchGregorianTs) {
         throw DataSourceUtils.newRebaseExceptionInWrite(format)
       }
       micros
-    case LegacyBehaviorPolicy.LEGACY => if (isRead) {
-      RebaseDateTime.rebaseJulianToGregorianMicros
-    } else {
-      RebaseDateTime.rebaseGregorianToJulianMicros
-    }
+    case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseGregorianToJulianMicros
     case LegacyBehaviorPolicy.CORRECTED => identity[Long]
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala

@@ -182,11 +182,11 @@ private[parquet] class ParquetRowConverter(
    */
   def currentRecord: InternalRow = currentRow
 
-  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFunc(
-    datetimeRebaseMode, "Parquet", isRead = true)
+  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFuncInRead(
+    datetimeRebaseMode, "Parquet")
 
-  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFunc(
-    datetimeRebaseMode, "Parquet", isRead = true)
+  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFuncInRead(
+    datetimeRebaseMode, "Parquet")
 
   // Converters for each field.
   private[this] val fieldConverters: Array[Converter with HasParentContainerUpdater] = {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala

@@ -82,11 +82,11 @@ class ParquetWriteSupport extends WriteSupport[InternalRow] with Logging {
   private val datetimeRebaseMode = LegacyBehaviorPolicy.withName(
     SQLConf.get.getConf(SQLConf.LEGACY_PARQUET_REBASE_MODE_IN_WRITE))
 
-  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFunc(
-    datetimeRebaseMode, "Parquet", isRead = false)
+  private val dateRebaseFunc = DataSourceUtils.creteDateRebaseFuncInWrite(
+    datetimeRebaseMode, "Parquet")
 
-  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFunc(
-    datetimeRebaseMode, "Parquet", isRead = false)
+  private val timestampRebaseFunc = DataSourceUtils.creteTimestampRebaseFuncInWrite(
+    datetimeRebaseMode, "Parquet")
 
   override def init(configuration: Configuration): WriteContext = {
     val schemaString = configuration.get(ParquetWriteSupport.SPARK_ROW_SCHEMA)