fix-code-style

Author: GuoPhilipse

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -61,7 +61,6 @@ object Cast {
     case (DateType, TimestampType) => true
     case (_: NumericType, TimestampType) => if (SQLConf.get.numericConvertToTimestampEnable) true
     else false
-
     case (StringType, DateType) => true
     case (TimestampType, DateType) => true
 
@@ -464,7 +463,6 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
     if (SQLConf.get.numericConvertToTimestampInSeconds) t * MICROS_PER_SECOND
     else t * MILLIS_PER_SECOND
   }
-
   // converting us to seconds
   private[this] def timestampToLong(ts: Long): Long = {
     Math.floorDiv(ts, MICROS_PER_SECOND)
@@ -1287,13 +1285,11 @@ abstract class CastBase extends UnaryExpression with TimeZoneAwareExpression wit
     val block = inline"new java.math.BigDecimal($MICROS_PER_SECOND)"
     code"($d.toBigDecimal().bigDecimal().multiply($block)).longValue()"
   }
-
   private[this] def longToTimeStampCode(l: ExprValue): Block = {
     if (SQLConf.get.numericConvertToTimestampInSeconds) code"" +
       code"$l * $MICROS_PER_SECOND"
     else code"$l * $MILLIS_PER_SECOND"
   }
-
   private[this] def timestampToLongCode(ts: ExprValue): Block =
     code"java.lang.Math.floorDiv($ts, $MICROS_PER_SECOND)"
   private[this] def timestampToDoubleCode(ts: ExprValue): Block =

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -621,7 +621,7 @@ object SQLConf {
     .stringConf
     .transform(_.toUpperCase(Locale.ROOT))
     .checkValues(ParquetOutputTimestampType.values.map(_.toString))
-    .createWithDefault(ParquetOutputTimestampType.INT96.toString)
+    .createWithDefault(ParquetOutputTimestampType.TIMESTAMP_MICROS.toString)
 
   val PARQUET_COMPRESSION = buildConf("spark.sql.parquet.compression.codec")
     .doc("Sets the compression codec used when writing Parquet files. If either `compression` or " +
@@ -845,10 +845,8 @@ object SQLConf {
     .doc("When true, enable the metadata-only query optimization that use the table's metadata " +
       "to produce the partition columns instead of table scans. It applies when all the columns " +
       "scanned are partition columns and the query has an aggregate operator that satisfies " +
-      "distinct semantics. By default the optimization is disabled, and deprecated as of Spark " +
-      "3.0 since it may return incorrect results when the files are empty, see also SPARK-26709." +
-      "It will be removed in the future releases. If you must use, use 'SparkSessionExtensions' " +
-      "instead to inject it as a custom rule.")
+      "distinct semantics. By default the optimization is disabled, since it may return " +
+      "incorrect results when the files are empty.")
     .version("2.1.1")
     .booleanConf
     .createWithDefault(false)
@@ -2065,18 +2063,16 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
-  val NESTED_PREDICATE_PUSHDOWN_FILE_SOURCE_LIST =
-    buildConf("spark.sql.optimizer.nestedPredicatePushdown.supportedFileSources")
+  val NESTED_PREDICATE_PUSHDOWN_ENABLED =
+    buildConf("spark.sql.optimizer.nestedPredicatePushdown.enabled")
       .internal()
-      .doc("A comma-separated list of data source short names or fully qualified data source " +
-        "implementation class names for which Spark tries to push down predicates for nested " +
-        "columns and/or names containing `dots` to data sources. This configuration is only " +
-        "effective with file-based data source in DSv1. Currently, Parquet implements " +
-        "both optimizations while ORC only supports predicates for names containing `dots`. The " +
-        "other data sources don't support this feature yet. So the default value is 'parquet,orc'.")
+      .doc("When true, Spark tries to push down predicates for nested columns and or names " +
+        "containing `dots` to data sources. Currently, Parquet implements both optimizations " +
+        "while ORC only supports predicates for names containing `dots`. The other data sources" +
+        "don't support this feature yet.")
       .version("3.0.0")
-      .stringConf
-      .createWithDefault("parquet,orc")
+      .booleanConf
+      .createWithDefault(true)
 
   val SERIALIZER_NESTED_SCHEMA_PRUNING_ENABLED =
     buildConf("spark.sql.optimizer.serializer.nestedSchemaPruning.enabled")
@@ -2228,6 +2224,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED =
+    buildConf("spark.sql.legacy.createHiveTableByDefault.enabled")
+      .internal()
+      .doc("When set to true, CREATE TABLE syntax without a provider will use hive " +
+        s"instead of the value of ${DEFAULT_DATA_SOURCE_NAME.key}.")
+      .version("3.0.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val LEGACY_BUCKETED_TABLE_SCAN_OUTPUT_ORDERING =
     buildConf("spark.sql.legacy.bucketedTableScan.outputOrdering")
       .internal()
@@ -2519,72 +2524,57 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
-  val LEGACY_PARQUET_REBASE_MODE_IN_WRITE =
-    buildConf("spark.sql.legacy.parquet.datetimeRebaseModeInWrite")
+  val LEGACY_PARQUET_REBASE_DATETIME_IN_WRITE =
+    buildConf("spark.sql.legacy.parquet.rebaseDateTimeInWrite.enabled")
       .internal()
-      .doc("When LEGACY, Spark will rebase dates/timestamps from Proleptic Gregorian calendar " +
-        "to the legacy hybrid (Julian + Gregorian) calendar when writing Parquet files. " +
-        "When CORRECTED, Spark will not do rebase and write the dates/timestamps as it is. " +
-        "When EXCEPTION, which is the default, Spark will fail the writing if it sees " +
-        "ancient dates/timestamps that are ambiguous between the two calendars.")
+      .doc("When true, rebase dates/timestamps from Proleptic Gregorian calendar " +
+        "to the hybrid calendar (Julian + Gregorian) in write. " +
+        "The rebasing is performed by converting micros/millis/days to " +
+        "a local date/timestamp in the source calendar, interpreting the resulted date/" +
+        "timestamp in the target calendar, and getting the number of micros/millis/days " +
+        "since the epoch 1970-01-01 00:00:00Z.")
       .version("3.0.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .checkValues(LegacyBehaviorPolicy.values.map(_.toString))
-      .createWithDefault(LegacyBehaviorPolicy.EXCEPTION.toString)
-
-  val LEGACY_PARQUET_REBASE_MODE_IN_READ =
-    buildConf("spark.sql.legacy.parquet.datetimeRebaseModeInRead")
-      .internal()
-      .doc("When LEGACY, Spark will rebase dates/timestamps from the legacy hybrid (Julian + " +
-        "Gregorian) calendar to Proleptic Gregorian calendar when reading Parquet files. " +
-        "When CORRECTED, Spark will not do rebase and read the dates/timestamps as it is. " +
-        "When EXCEPTION, which is the default, Spark will fail the reading if it sees " +
-        "ancient dates/timestamps that are ambiguous between the two calendars. This config is " +
-        "only effective if the writer info (like Spark, Hive) of the Parquet files is unknown.")
-      .version("3.0.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .checkValues(LegacyBehaviorPolicy.values.map(_.toString))
-      .createWithDefault(LegacyBehaviorPolicy.EXCEPTION.toString)
+      .booleanConf
+      .createWithDefault(false)
 
-  val LEGACY_AVRO_REBASE_MODE_IN_WRITE =
-    buildConf("spark.sql.legacy.avro.datetimeRebaseModeInWrite")
+  val LEGACY_PARQUET_REBASE_DATETIME_IN_READ =
+    buildConf("spark.sql.legacy.parquet.rebaseDateTimeInRead.enabled")
       .internal()
-      .doc("When LEGACY, Spark will rebase dates/timestamps from Proleptic Gregorian calendar " +
-        "to the legacy hybrid (Julian + Gregorian) calendar when writing Avro files. " +
-        "When CORRECTED, Spark will not do rebase and write the dates/timestamps as it is. " +
-        "When EXCEPTION, which is the default, Spark will fail the writing if it sees " +
-        "ancient dates/timestamps that are ambiguous between the two calendars.")
+      .doc("When true, rebase dates/timestamps " +
+        "from the hybrid calendar to Proleptic Gregorian calendar in read. " +
+        "The rebasing is performed by converting micros/millis/days to " +
+        "a local date/timestamp in the source calendar, interpreting the resulted date/" +
+        "timestamp in the target calendar, and getting the number of micros/millis/days " +
+        "since the epoch 1970-01-01 00:00:00Z.")
       .version("3.0.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .checkValues(LegacyBehaviorPolicy.values.map(_.toString))
-      .createWithDefault(LegacyBehaviorPolicy.EXCEPTION.toString)
-
-  val LEGACY_AVRO_REBASE_MODE_IN_READ =
-    buildConf("spark.sql.legacy.avro.datetimeRebaseModeInRead")
-      .internal()
-      .doc("When LEGACY, Spark will rebase dates/timestamps from the legacy hybrid (Julian + " +
-        "Gregorian) calendar to Proleptic Gregorian calendar when reading Avro files. " +
-        "When CORRECTED, Spark will not do rebase and read the dates/timestamps as it is. " +
-        "When EXCEPTION, which is the default, Spark will fail the reading if it sees " +
-        "ancient dates/timestamps that are ambiguous between the two calendars. This config is " +
-        "only effective if the writer info (like Spark, Hive) of the Avro files is unknown.")
+      .booleanConf
+      .createWithDefault(false)
+
+  val LEGACY_AVRO_REBASE_DATETIME_IN_WRITE =
+    buildConf("spark.sql.legacy.avro.rebaseDateTimeInWrite.enabled")
+      .internal()
+      .doc("When true, rebase dates/timestamps from Proleptic Gregorian calendar " +
+        "to the hybrid calendar (Julian + Gregorian) in write. " +
+        "The rebasing is performed by converting micros/millis/days to " +
+        "a local date/timestamp in the source calendar, interpreting the resulted date/" +
+        "timestamp in the target calendar, and getting the number of micros/millis/days " +
+        "since the epoch 1970-01-01 00:00:00Z.")
       .version("3.0.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .checkValues(LegacyBehaviorPolicy.values.map(_.toString))
-      .createWithDefault(LegacyBehaviorPolicy.EXCEPTION.toString)
+      .booleanConf
+      .createWithDefault(false)
 
-  val SCRIPT_TRANSFORMATION_EXIT_TIMEOUT =
-    buildConf("spark.sql.scriptTransformation.exitTimeoutInSeconds")
+  val LEGACY_AVRO_REBASE_DATETIME_IN_READ =
+    buildConf("spark.sql.legacy.avro.rebaseDateTimeInRead.enabled")
       .internal()
-      .doc("Timeout for executor to wait for the termination of transformation script when EOF.")
+      .doc("When true, rebase dates/timestamps " +
+        "from the hybrid calendar to Proleptic Gregorian calendar in read. " +
+        "The rebasing is performed by converting micros/millis/days to " +
+        "a local date/timestamp in the source calendar, interpreting the resulted date/" +
+        "timestamp in the target calendar, and getting the number of micros/millis/days " +
+        "since the epoch 1970-01-01 00:00:00Z.")
       .version("3.0.0")
-      .timeConf(TimeUnit.SECONDS)
-      .checkValue(_ > 0, "The timeout value must be positive")
-      .createWithDefault(10L)
+      .booleanConf
+      .createWithDefault(false)
 
   val LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_ENABLE =
     buildConf("spark.sql.legacy.numericConvertToTimestampEnable")
@@ -2632,10 +2622,7 @@ object SQLConf {
       DeprecatedConfig(ARROW_FALLBACK_ENABLED.key, "3.0",
         s"Use '${ARROW_PYSPARK_FALLBACK_ENABLED.key}' instead of it."),
       DeprecatedConfig(SHUFFLE_TARGET_POSTSHUFFLE_INPUT_SIZE.key, "3.0",
-        s"Use '${ADVISORY_PARTITION_SIZE_IN_BYTES.key}' instead of it."),
-      DeprecatedConfig(OPTIMIZER_METADATA_ONLY.key, "3.0",
-        "Avoid to depend on this optimization to prevent a potential correctness issue. " +
-          "If you must use, use 'SparkSessionExtensions' instead to inject it as a custom rule.")
+        s"Use '${ADVISORY_PARTITION_SIZE_IN_BYTES.key}' instead of it.")
     )
 
     Map(configs.map { cfg => cfg.key -> cfg } : _*)
@@ -3128,6 +3115,8 @@ class SQLConf extends Serializable with Logging {
 
   def nestedSchemaPruningEnabled: Boolean = getConf(NESTED_SCHEMA_PRUNING_ENABLED)
 
+  def nestedPredicatePushdownEnabled: Boolean = getConf(NESTED_PREDICATE_PUSHDOWN_ENABLED)
+
   def serializerNestedSchemaPruningEnabled: Boolean =
     getConf(SERIALIZER_NESTED_SCHEMA_PRUNING_ENABLED)
 
@@ -3161,6 +3150,9 @@ class SQLConf extends Serializable with Logging {
   def allowNegativeScaleOfDecimalEnabled: Boolean =
     getConf(SQLConf.LEGACY_ALLOW_NEGATIVE_SCALE_OF_DECIMAL_ENABLED)
 
+  def createHiveTableByDefaultEnabled: Boolean =
+    getConf(SQLConf.LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT_ENABLED)
+
   def truncateTableIgnorePermissionAcl: Boolean =
     getConf(SQLConf.TRUNCATE_TABLE_IGNORE_PERMISSION_ACL)
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuite.scala

@@ -1300,15 +1300,97 @@ class CastSuite extends CastSuiteBase {
     }
   }
 
-  test("cast a timestamp before the epoch 1970-01-01 00:00:00Z") {
-    withDefaultTimeZone(UTC) {
-      val negativeTs = Timestamp.valueOf("1900-05-05 18:34:56.1")
-      assert(negativeTs.getTime < 0)
-      val expectedSecs = Math.floorDiv(negativeTs.getTime, MILLIS_PER_SECOND)
-      checkEvaluation(cast(negativeTs, ByteType), expectedSecs.toByte)
-      checkEvaluation(cast(negativeTs, ShortType), expectedSecs.toShort)
-      checkEvaluation(cast(negativeTs, IntegerType), expectedSecs.toInt)
-      checkEvaluation(cast(negativeTs, LongType), expectedSecs)
+  test("SPARK-31710:Add legacy when casting long to timestamp") {
+    withSQLConf(
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_ENABLE.key -> "true",
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_IN_SECONDS.key -> "false") {
+        def checkLongToTimestamp(l: Long, expected: Long): Unit = {
+          checkEvaluation(cast(l, TimestampType), expected)
+        }
+        checkLongToTimestamp(253402272000L, 253402272000000L)
+        checkLongToTimestamp(-5L, -5000L)
+        checkLongToTimestamp(1L, 1000L)
+        checkLongToTimestamp(0L, 0L)
+        checkLongToTimestamp(123L, 123000L)
+    }
+    withSQLConf(
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_ENABLE.key -> "true",
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_IN_SECONDS.key -> "true") {
+        def checkLongToTimestamp(l: Long, expected: Long): Unit = {
+          checkEvaluation(cast(l, TimestampType), expected)
+        }
+        checkLongToTimestamp(253402272000L, 253402272000000000L)
+        checkLongToTimestamp(-5L, -5000000L)
+        checkLongToTimestamp(1L, 1000000L)
+        checkLongToTimestamp(0L, 0L)
+        checkLongToTimestamp(123L, 123000000L)
+    }
+
+    withSQLConf(
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_ENABLE.key -> "false",
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_IN_SECONDS.key -> "false") {
+        def checkByteToTimestamp(b: Byte, expected: Long): Unit = {
+          assert(!cast(b, TimestampType).resolved)
+        }
+        def checkShortToTimestamp(s: Short, expected: Long): Unit = {
+          assert(!cast(s, TimestampType).resolved)
+        }
+        def checkIntToTimestamp(str: Int, expected: Long): Unit = {
+          assert(!cast(str, TimestampType).resolved)
+        }
+        def checkLongToTimestamp(l: Long, expected: Long): Unit = {
+          assert(!cast(l, TimestampType).resolved)
+        }
+        def checkDecimalToTimestamp(d: Decimal, expected: Long): Unit = {
+          assert(!cast(d, TimestampType).resolved)
+        }
+        def checkFloatToTimestamp(f: Float, expected: Long): Unit = {
+          assert(!cast(f, TimestampType).resolved)
+        }
+        def checkDoubleToTimestamp(d: Double, expected: Long): Unit = {
+          assert(!cast(d, TimestampType).resolved)
+        }
+        checkByteToTimestamp(1.toByte, 0L)
+        checkShortToTimestamp(1.toShort, 0L)
+        checkIntToTimestamp(1, 0L)
+        checkLongToTimestamp(1L, 0L)
+        checkDecimalToTimestamp(Decimal(1.5), 0L)
+        checkFloatToTimestamp(1.5f, 0L)
+        checkDoubleToTimestamp(2.1D, 0L)
+    }
+
+    withSQLConf(
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_ENABLE.key -> "false",
+      SQLConf.LEGACY_NUMERIC_CONVERT_TO_TIMESTAMP_IN_SECONDS.key -> "true") {
+        def checkByteToTimestamp(b: Byte, expected: Long): Unit = {
+          assert(!cast(b, TimestampType).resolved)
+        }
+        def checkShortToTimestamp(s: Short, expected: Long): Unit = {
+          assert(!cast(s, TimestampType).resolved)
+        }
+        def checkIntToTimestamp(str: Int, expected: Long): Unit = {
+          assert(!cast(str, TimestampType).resolved)
+        }
+        def checkLongToTimestamp(l: Long, expected: Long): Unit = {
+          assert(!cast(l, TimestampType).resolved)
+        }
+        def checkDecimalToTimestamp(d: Decimal, expected: Long): Unit = {
+          assert(!cast(d, TimestampType).resolved)
+        }
+        def checkFloatToTimestamp(f: Float, expected: Long): Unit = {
+          assert(!cast(f, TimestampType).resolved)
+        }
+        def checkDoubleToTimestamp(d: Double, expected: Long): Unit = {
+          assert(!cast(d, TimestampType).resolved)
+        }
+
+        checkByteToTimestamp(1.toByte, 0L)
+        checkShortToTimestamp(1.toShort, 0L)
+        checkIntToTimestamp(1, 0L)
+        checkLongToTimestamp(1L, 0L)
+        checkDecimalToTimestamp(Decimal(1.5), 0L)
+        checkFloatToTimestamp(1.5f, 0L)
+        checkDoubleToTimestamp(2.1D, 0L)
     }
   }