You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by hv...@apache.org on 2023/07/28 11:13:48 UTC
[spark] branch branch-3.5 updated: [SPARK-44538][CONNECT][SQL] Reinstate Row.jsonValue and friends
This is an automated email from the ASF dual-hosted git repository.
hvanhovell pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 02922313076 [SPARK-44538][CONNECT][SQL] Reinstate Row.jsonValue and friends
02922313076 is described below
commit 0292231307681e70674d743ba382fed13b30d47a
Author: Herman van Hovell <he...@databricks.com>
AuthorDate: Fri Jul 28 07:03:28 2023 -0400
[SPARK-44538][CONNECT][SQL] Reinstate Row.jsonValue and friends
This PR reinstates `Row.jsonValue`, `Row.prettyJson` & `Row.json`.
`Row.jsonValue` and other json rendering is used in a couple of hard to remove places and there are also a couple of users using it. In order to support this we needed to move quite a few date time related classes. We were going to move them anyway because they are also needed to decouple direct arrow encoding from catalyst.
No.
Existing tests.
Closes #42164 from hvanhovell/SPARK-44538.
Authored-by: Herman van Hovell <he...@databricks.com>
Signed-off-by: Herman van Hovell <he...@databricks.com>
(cherry picked from commit 3dcee0ccf7e02acd162da652450b85d34dc4e297)
Signed-off-by: Herman van Hovell <he...@databricks.com>
---
.../apache/spark/sql/avro/AvroDeserializer.scala | 3 +-
.../apache/spark/sql/avro/AvroOutputWriter.scala | 3 +-
.../org/apache/spark/sql/avro/AvroSerializer.scala | 3 +-
.../sql/avro/AvroCatalystDataConversionSuite.scala | 4 +-
.../apache/spark/sql/avro/AvroRowReaderSuite.scala | 2 +-
.../org/apache/spark/sql/avro/AvroSerdeSuite.scala | 2 +-
.../org/apache/spark/sql/avro/AvroSuite.scala | 4 +-
.../connect/client/arrow/ArrowDeserializer.scala | 4 +-
.../sql/connect/client/arrow/ArrowSerializer.scala | 20 +-
.../connect/client/arrow/ArrowVectorReader.scala | 2 +-
.../org/apache/spark/sql/streaming/progress.scala | 5 +-
.../connect/client/arrow/ArrowEncoderSuite.scala | 2 +-
.../connect/common/DataTypeProtoConverter.scala | 4 +-
.../common/LiteralValueProtoConverter.scala | 17 +-
.../resources/gregorian-julian-rebase-micros.json | 0
.../resources/julian-gregorian-rebase-micros.json | 0
.../src/main/scala/org/apache/spark/sql/Row.scala | 103 +++-
.../spark/sql/catalyst/JavaTypeInference.scala | 8 +-
.../spark/sql/catalyst/ScalaReflection.scala | 12 +-
.../spark/sql/catalyst/encoders/RowEncoder.scala | 7 +-
.../sql/catalyst/parser/DataTypeAstBuilder.scala | 2 +-
.../apache/spark/sql/catalyst/parser/parsers.scala | 3 +-
.../spark/sql/catalyst/util/DateFormatter.scala | 16 +-
.../catalyst/util/DateTimeFormatterHelper.scala | 43 +-
.../apache/spark/sql/catalyst/util/MathUtils.scala | 4 +-
.../spark/sql/catalyst/util/RebaseDateTime.scala | 8 +-
.../sql/catalyst/util/SparkCharVarcharUtils.scala | 2 +-
.../sql/catalyst/util/SparkDateTimeUtils.scala | 572 +++++++++++++++++++-
.../sql/catalyst/util/TimestampFormatter.scala | 26 +-
.../apache/spark/sql/catalyst/util/UDTUtils.scala | 58 ++
.../apache/spark/sql/errors/DataTypeErrors.scala | 6 -
.../spark/sql/errors/DataTypeErrorsBase.scala | 33 +-
.../apache/spark/sql/errors/EncoderErrors.scala | 74 ---
.../apache/spark/sql/errors/ExecutionErrors.scala | 211 ++++++++
.../spark/sql/errors/QueryParsingErrors.scala | 4 +-
.../LegacyBehaviorPolicy.scala} | 24 +-
.../spark/sql/{ => internal}/SqlApiConf.scala | 12 +-
.../org/apache/spark/sql/types/DataType.scala | 2 +-
.../scala/org/apache/spark/sql/types/Decimal.scala | 2 +-
.../org/apache/spark/sql/types/DecimalType.scala | 2 +-
.../org/apache/spark/sql/types/StructType.scala | 2 +-
.../org/apache/spark/sql/util/ArrowUtils.scala | 8 +-
.../sql/catalyst/analysis/CTESubstitution.scala | 3 +-
.../apache/spark/sql/catalyst/csv/CSVOptions.scala | 3 +-
.../spark/sql/catalyst/csv/UnivocityParser.scala | 6 +-
.../spark/sql/catalyst/json/JSONOptions.scala | 3 +-
.../spark/sql/catalyst/json/JacksonParser.scala | 6 +-
.../spark/sql/catalyst/util/DateTimeUtils.scala | 586 +--------------------
.../spark/sql/catalyst/util/UDTUtilsImpl.scala} | 25 +-
.../spark/sql/errors/QueryExecutionErrors.scala | 94 +---
.../spark/sql/execution/arrow/ArrowWriter.scala | 4 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 13 +-
.../org/apache/spark/sql/util/ToJsonUtil.scala | 129 -----
.../scala/org/apache/spark/sql/RowJsonSuite.scala | 7 +-
.../test/scala/org/apache/spark/sql/RowTest.scala | 3 +-
.../sql/catalyst/util/DateFormatterSuite.scala | 3 +-
.../catalyst/util/TimestampFormatterSuite.scala | 3 +-
.../org/apache/spark/sql/execution/Columnar.scala | 4 +-
.../execution/datasources/DataSourceUtils.scala | 3 +-
.../datasources/parquet/ParquetFilters.scala | 2 +-
.../datasources/parquet/ParquetReadSupport.scala | 3 +-
.../datasources/parquet/ParquetUtils.scala | 4 +-
.../datasources/parquet/ParquetWriteSupport.scala | 3 +-
.../v2/parquet/ParquetScanBuilder.scala | 2 +-
.../org/apache/spark/sql/streaming/progress.scala | 3 +-
.../sql/errors/QueryExecutionErrorsSuite.scala | 2 +-
.../benchmark/DateTimeRebaseBenchmark.scala | 4 +-
.../sql/execution/datasources/csv/CSVSuite.scala | 3 +-
.../datasources/parquet/ParquetFilterSuite.scala | 5 +-
.../parquet/ParquetRebaseDatetimeSuite.scala | 6 +-
.../spark/sql/sources/HadoopFsRelationTest.scala | 2 +-
71 files changed, 1144 insertions(+), 1109 deletions(-)
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
index 092a8f80b77..d4d34a891e9 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
@@ -37,8 +37,7 @@ 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.errors.QueryCompilationErrors
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.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala
index 44520bb92d1..23dfe86b6bc 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOutputWriter.scala
@@ -32,8 +32,7 @@ import org.apache.spark.SPARK_VERSION_SHORT
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
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
// NOTE: This class is instantiated and used on executor side only, no need to be serializable.
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
index c95d731f0de..34bf47613e7 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala
@@ -37,8 +37,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
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.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
/**
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
index 5c0d64b4d55..1cb34a0bc4d 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, NoopF
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.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.sources.{EqualTo, Not}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
@@ -286,7 +286,7 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite
schema,
dataType,
false,
- RebaseSpec(SQLConf.LegacyBehaviorPolicy.CORRECTED),
+ RebaseSpec(LegacyBehaviorPolicy.CORRECTED),
filters)
val deserialized = deserializer.deserialize(data)
expected match {
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
index cc0e178c617..70d0bc6c0ad 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
@@ -31,8 +31,8 @@ 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.LegacyBehaviorPolicy._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.v2.avro.AvroScan
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
index bfd56613fd6..7f99f3c737c 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
@@ -22,7 +22,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.internal.LegacyBehaviorPolicy.CORRECTED
import org.apache.spark.sql.types.{IntegerType, StructType}
/**
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 35e9f43289c..516c173c118 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -46,9 +46,9 @@ import org.apache.spark.sql.execution.{FormattedMode, SparkPlan}
import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition}
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.v2.avro.AvroScan
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
index 4177a88ba52..cd3afeb7e90 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowDeserializer.scala
@@ -37,7 +37,7 @@ import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
-import org.apache.spark.sql.errors.{DataTypeErrors, QueryCompilationErrors}
+import org.apache.spark.sql.errors.{ExecutionErrors, QueryCompilationErrors}
import org.apache.spark.sql.types.Decimal
/**
@@ -341,7 +341,7 @@ object ArrowDeserializers {
}
case (CalendarIntervalEncoder | _: UDTEncoder[_], _) =>
- throw DataTypeErrors.unsupportedDataTypeError(encoder.dataType)
+ throw ExecutionErrors.unsupportedDataTypeError(encoder.dataType)
case _ =>
throw new RuntimeException(
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
index 9b39a75ceed..94e01bbef31 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowSerializer.scala
@@ -37,8 +37,8 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.DefinedByConstructorParams
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
-import org.apache.spark.sql.catalyst.util.{DateTimeUtils, IntervalUtils}
-import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.catalyst.util.{IntervalUtils, SparkDateTimeUtils}
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.types.Decimal
import org.apache.spark.sql.util.ArrowUtils
@@ -323,37 +323,37 @@ object ArrowSerializer {
case (DateEncoder(true) | LocalDateEncoder(true), v: DateDayVector) =>
new FieldSerializer[Any, DateDayVector](v) {
override def set(index: Int, value: Any): Unit =
- vector.setSafe(index, DateTimeUtils.anyToDays(value))
+ vector.setSafe(index, SparkDateTimeUtils.anyToDays(value))
}
case (DateEncoder(false), v: DateDayVector) =>
new FieldSerializer[java.sql.Date, DateDayVector](v) {
override def set(index: Int, value: java.sql.Date): Unit =
- vector.setSafe(index, DateTimeUtils.fromJavaDate(value))
+ vector.setSafe(index, SparkDateTimeUtils.fromJavaDate(value))
}
case (LocalDateEncoder(false), v: DateDayVector) =>
new FieldSerializer[LocalDate, DateDayVector](v) {
override def set(index: Int, value: LocalDate): Unit =
- vector.setSafe(index, DateTimeUtils.localDateToDays(value))
+ vector.setSafe(index, SparkDateTimeUtils.localDateToDays(value))
}
case (TimestampEncoder(true) | InstantEncoder(true), v: TimeStampMicroTZVector) =>
new FieldSerializer[Any, TimeStampMicroTZVector](v) {
override def set(index: Int, value: Any): Unit =
- vector.setSafe(index, DateTimeUtils.anyToMicros(value))
+ vector.setSafe(index, SparkDateTimeUtils.anyToMicros(value))
}
case (TimestampEncoder(false), v: TimeStampMicroTZVector) =>
new FieldSerializer[java.sql.Timestamp, TimeStampMicroTZVector](v) {
override def set(index: Int, value: java.sql.Timestamp): Unit =
- vector.setSafe(index, DateTimeUtils.fromJavaTimestamp(value))
+ vector.setSafe(index, SparkDateTimeUtils.fromJavaTimestamp(value))
}
case (InstantEncoder(false), v: TimeStampMicroTZVector) =>
new FieldSerializer[Instant, TimeStampMicroTZVector](v) {
override def set(index: Int, value: Instant): Unit =
- vector.setSafe(index, DateTimeUtils.instantToMicros(value))
+ vector.setSafe(index, SparkDateTimeUtils.instantToMicros(value))
}
case (LocalDateTimeEncoder, v: TimeStampMicroVector) =>
new FieldSerializer[LocalDateTime, TimeStampMicroVector](v) {
override def set(index: Int, value: LocalDateTime): Unit =
- vector.setSafe(index, DateTimeUtils.localDateTimeToMicros(value))
+ vector.setSafe(index, SparkDateTimeUtils.localDateTimeToMicros(value))
}
case (OptionEncoder(value), v) =>
@@ -439,7 +439,7 @@ object ArrowSerializer {
}
case (CalendarIntervalEncoder | _: UDTEncoder[_], _) =>
- throw DataTypeErrors.unsupportedDataTypeError(encoder.dataType)
+ throw ExecutionErrors.unsupportedDataTypeError(encoder.dataType)
case _ =>
throw new RuntimeException(s"Unsupported Encoder($encoder)/Vector($v) combination.")
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
index 9111b3b9ccf..63c6a0df5c8 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/arrow/ArrowVectorReader.scala
@@ -26,8 +26,8 @@ import org.apache.arrow.vector.util.Text
import org.apache.spark.sql.catalyst.expressions.Cast
import org.apache.spark.sql.catalyst.util.{DateFormatter, IntervalUtils, StringUtils, TimestampFormatter}
import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_SECOND
-import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.catalyst.util.IntervalStringStyles.ANSI_STYLE
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
import org.apache.spark.sql.types.{DataType, DayTimeIntervalType, Decimal, YearMonthIntervalType}
import org.apache.spark.sql.util.ArrowUtils
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala
index 123b3306f2a..593311efb9c 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala
@@ -37,7 +37,6 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
-import org.apache.spark.sql.util.ToJsonUtil
/**
* Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
@@ -190,9 +189,7 @@ class StreamingQueryProgress private[spark] (
("stateOperators" -> JArray(stateOperators.map(_.jsonValue).toList)) ~
("sources" -> JArray(sources.map(_.jsonValue).toList)) ~
("sink" -> sink.jsonValue) ~
- ("observedMetrics" -> safeMapToJValue[Row](
- observedMetrics,
- row => ToJsonUtil.jsonValue(row)))
+ ("observedMetrics" -> safeMapToJValue[Row](observedMetrics, row => row.jsonValue))
}
}
diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
index 5c035a613fe..7d5311e7312 100644
--- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
+++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
@@ -38,9 +38,9 @@ import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{BinaryEncoder, B
import org.apache.spark.sql.catalyst.encoders.RowEncoder.{encoderFor => toRowEncoder}
import org.apache.spark.sql.catalyst.util.{DateFormatter, StringUtils, TimestampFormatter}
import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_SECOND
-import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.catalyst.util.IntervalStringStyles.ANSI_STYLE
import org.apache.spark.sql.catalyst.util.IntervalUtils._
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
import org.apache.spark.sql.connect.client.arrow.FooEnum.FooEnum
import org.apache.spark.sql.connect.client.util.ConnectFunSuite
import org.apache.spark.sql.types.{ArrayType, DataType, DayTimeIntervalType, Decimal, DecimalType, IntegerType, Metadata, SQLUserDefinedType, StructType, UserDefinedType, YearMonthIntervalType}
diff --git a/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala b/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
index c2ef171498d..bac3da25461 100644
--- a/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
+++ b/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
@@ -21,7 +21,7 @@ import scala.collection.convert.ImplicitConversions._
import org.apache.spark.connect.proto
import org.apache.spark.sql.types._
-import org.apache.spark.util.Utils
+import org.apache.spark.util.SparkClassUtils
/**
* Helper class for conversions between [[DataType]] and [[proto.DataType]].
@@ -124,7 +124,7 @@ object DataTypeProtoConverter {
}
if (t.hasJvmClass) {
- Utils
+ SparkClassUtils
.classForName[UserDefinedType[_]](t.getJvmClass)
.getConstructor()
.newInstance()
diff --git a/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala b/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
index fdd8559be48..f2abbf9c723 100644
--- a/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
+++ b/connector/connect/common/src/main/scala/org/apache/spark/sql/connect/common/LiteralValueProtoConverter.scala
@@ -32,7 +32,7 @@ import com.google.protobuf.ByteString
import org.apache.spark.connect.proto
import org.apache.spark.sql.catalyst.ScalaReflection
-import org.apache.spark.sql.catalyst.util.{DateTimeUtils, IntervalUtils}
+import org.apache.spark.sql.catalyst.util.{IntervalUtils, SparkDateTimeUtils}
import org.apache.spark.sql.connect.common.DataTypeProtoConverter._
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.CalendarInterval
@@ -88,10 +88,11 @@ object LiteralValueProtoConverter {
case v: LocalDate => builder.setDate(v.toEpochDay.toInt)
case v: Decimal =>
builder.setDecimal(decimalBuilder(Math.max(v.precision, v.scale), v.scale, v.toString))
- case v: Instant => builder.setTimestamp(DateTimeUtils.instantToMicros(v))
- case v: Timestamp => builder.setTimestamp(DateTimeUtils.fromJavaTimestamp(v))
- case v: LocalDateTime => builder.setTimestampNtz(DateTimeUtils.localDateTimeToMicros(v))
- case v: Date => builder.setDate(DateTimeUtils.fromJavaDate(v))
+ case v: Instant => builder.setTimestamp(SparkDateTimeUtils.instantToMicros(v))
+ case v: Timestamp => builder.setTimestamp(SparkDateTimeUtils.fromJavaTimestamp(v))
+ case v: LocalDateTime =>
+ builder.setTimestampNtz(SparkDateTimeUtils.localDateTimeToMicros(v))
+ case v: Date => builder.setDate(SparkDateTimeUtils.fromJavaDate(v))
case v: Duration => builder.setDayTimeInterval(IntervalUtils.durationToMicros(v))
case v: Period => builder.setYearMonthInterval(IntervalUtils.periodToMonths(v))
case v: Array[_] => builder.setArray(arrayBuilder(v))
@@ -263,13 +264,13 @@ object LiteralValueProtoConverter {
case proto.Expression.Literal.LiteralTypeCase.STRING => literal.getString
case proto.Expression.Literal.LiteralTypeCase.DATE =>
- DateTimeUtils.toJavaDate(literal.getDate)
+ SparkDateTimeUtils.toJavaDate(literal.getDate)
case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP =>
- DateTimeUtils.toJavaTimestamp(literal.getTimestamp)
+ SparkDateTimeUtils.toJavaTimestamp(literal.getTimestamp)
case proto.Expression.Literal.LiteralTypeCase.TIMESTAMP_NTZ =>
- DateTimeUtils.microsToLocalDateTime(literal.getTimestampNtz)
+ SparkDateTimeUtils.microsToLocalDateTime(literal.getTimestampNtz)
case proto.Expression.Literal.LiteralTypeCase.CALENDAR_INTERVAL =>
new CalendarInterval(
diff --git a/sql/catalyst/src/main/resources/gregorian-julian-rebase-micros.json b/sql/api/src/main/resources/gregorian-julian-rebase-micros.json
similarity index 100%
rename from sql/catalyst/src/main/resources/gregorian-julian-rebase-micros.json
rename to sql/api/src/main/resources/gregorian-julian-rebase-micros.json
diff --git a/sql/catalyst/src/main/resources/julian-gregorian-rebase-micros.json b/sql/api/src/main/resources/julian-gregorian-rebase-micros.json
similarity index 100%
rename from sql/catalyst/src/main/resources/julian-gregorian-rebase-micros.json
rename to sql/api/src/main/resources/julian-gregorian-rebase-micros.json
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/Row.scala b/sql/api/src/main/scala/org/apache/spark/sql/Row.scala
index 1765a5444df..94b446bf326 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/Row.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/Row.scala
@@ -17,13 +17,25 @@
package org.apache.spark.sql
+import java.sql.{Date, Timestamp}
+import java.time.{Instant, LocalDate}
+import java.util.Base64
+
import scala.collection.JavaConverters._
+import scala.collection.mutable
import scala.util.hashing.MurmurHash3
-import org.apache.spark.annotation.Stable
+import org.json4s.{JArray, JBool, JDecimal, JDouble, JField, JLong, JNull, JObject, JString}
+import org.json4s.JsonAST.JValue
+import org.json4s.jackson.JsonMethods.{compact, pretty, render}
+
+import org.apache.spark.annotation.{Stable, Unstable}
import org.apache.spark.sql.catalyst.expressions.GenericRow
+import org.apache.spark.sql.catalyst.util.{DateFormatter, SparkDateTimeUtils, TimestampFormatter, UDTUtils}
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.CalendarInterval
/**
* @since 1.3.0
@@ -513,4 +525,93 @@ trait Row extends Serializable {
private def getAnyValAs[T <: AnyVal](i: Int): T =
if (isNullAt(i)) throw DataTypeErrors.valueIsNullError(i)
else getAs[T](i)
+
+ /**
+ * The compact JSON representation of this row.
+ * @since 3.0
+ */
+ @Unstable
+ def json: String = compact(jsonValue)
+
+ /**
+ * The pretty (i.e. indented) JSON representation of this row.
+ * @since 3.0
+ */
+ @Unstable
+ def prettyJson: String = pretty(render(jsonValue))
+
+ /**
+ * JSON representation of the row.
+ *
+ * Note that this only supports the data types that are also supported by
+ * [[org.apache.spark.sql.catalyst.encoders.RowEncoder]].
+ *
+ * @return the JSON representation of the row.
+ */
+ private[sql] def jsonValue: JValue = {
+ require(schema != null, "JSON serialization requires a non-null schema.")
+
+ lazy val zoneId = SparkDateTimeUtils.getZoneId(SqlApiConf.get.sessionLocalTimeZone)
+ lazy val dateFormatter = DateFormatter()
+ lazy val timestampFormatter = TimestampFormatter(zoneId)
+
+ // Convert an iterator of values to a json array
+ def iteratorToJsonArray(iterator: Iterator[_], elementType: DataType): JArray = {
+ JArray(iterator.map(toJson(_, elementType)).toList)
+ }
+
+ // Convert a value to json.
+ def toJson(value: Any, dataType: DataType): JValue = (value, dataType) match {
+ case (null, _) => JNull
+ case (b: Boolean, _) => JBool(b)
+ case (b: Byte, _) => JLong(b)
+ case (s: Short, _) => JLong(s)
+ case (i: Int, _) => JLong(i)
+ case (l: Long, _) => JLong(l)
+ case (f: Float, _) => JDouble(f)
+ case (d: Double, _) => JDouble(d)
+ case (d: BigDecimal, _) => JDecimal(d)
+ case (d: java.math.BigDecimal, _) => JDecimal(d)
+ case (d: Decimal, _) => JDecimal(d.toBigDecimal)
+ case (s: String, _) => JString(s)
+ case (b: Array[Byte], BinaryType) =>
+ JString(Base64.getEncoder.encodeToString(b))
+ case (d: LocalDate, _) => JString(dateFormatter.format(d))
+ case (d: Date, _) => JString(dateFormatter.format(d))
+ case (i: Instant, _) => JString(timestampFormatter.format(i))
+ case (t: Timestamp, _) => JString(timestampFormatter.format(t))
+ case (i: CalendarInterval, _) => JString(i.toString)
+ case (a: Array[_], ArrayType(elementType, _)) =>
+ iteratorToJsonArray(a.iterator, elementType)
+ case (a: mutable.ArraySeq[_], ArrayType(elementType, _)) =>
+ iteratorToJsonArray(a.iterator, elementType)
+ case (s: Seq[_], ArrayType(elementType, _)) =>
+ iteratorToJsonArray(s.iterator, elementType)
+ case (m: Map[String @unchecked, _], MapType(StringType, valueType, _)) =>
+ new JObject(m.toList.sortBy(_._1).map {
+ case (k, v) => k -> toJson(v, valueType)
+ })
+ case (m: Map[_, _], MapType(keyType, valueType, _)) =>
+ new JArray(m.iterator.map {
+ case (k, v) =>
+ new JObject("key" -> toJson(k, keyType) :: "value" -> toJson(v, valueType) :: Nil)
+ }.toList)
+ case (row: Row, schema: StructType) =>
+ var n = 0
+ val elements = new mutable.ListBuffer[JField]
+ val len = row.length
+ while (n < len) {
+ val field = schema(n)
+ elements += (field.name -> toJson(row(n), field.dataType))
+ n += 1
+ }
+ new JObject(elements.toList)
+ case (v: Any, udt: UserDefinedType[Any @unchecked]) =>
+ toJson(UDTUtils.toRow(v, udt), udt.sqlType)
+ case _ =>
+ throw new IllegalArgumentException(s"Failed to convert value $value " +
+ s"(class of ${value.getClass}}) with the type of $dataType to JSON.")
+ }
+ toJson(this, schema)
+ }
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
index ec9e704c66d..f352d28a7b5 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala
@@ -26,7 +26,7 @@ import scala.reflect.ClassTag
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ArrayEncoder, BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, DayTimeIntervalEncoder, DEFAULT_JAVA_DECIMAL_ENCODER, EncoderField, IterableEncoder, JavaBeanEncoder, JavaBigIntEncoder, JavaEnumEncoder, LocalDateTimeEncoder, MapEncoder, PrimitiveBooleanEncoder, PrimitiveByteEncoder, PrimitiveDoubleEncoder, PrimitiveFloatEncoder, P [...]
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.types._
/**
@@ -115,7 +115,7 @@ object JavaTypeInference {
case c: Class[_] =>
if (seenTypeSet.contains(c)) {
- throw EncoderErrors.cannotHaveCircularReferencesInBeanClassError(c)
+ throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
// TODO: we should only collect properties that have getter and setter. However, some tests
@@ -138,7 +138,7 @@ object JavaTypeInference {
JavaBeanEncoder(ClassTag(c), fields)
case _ =>
- throw EncoderErrors.cannotFindEncoderForTypeError(t.toString)
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
}
def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
@@ -196,7 +196,7 @@ object JavaTypeInference {
}
}
}
- throw EncoderErrors.unreachableError()
+ throw ExecutionErrors.unreachableError()
}
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
index 1f366393d25..5f063b4a9a6 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
@@ -30,7 +30,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders._
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.CalendarInterval
@@ -377,13 +377,13 @@ object ScalaReflection extends ScalaReflection {
case t if definedByConstructorParams(t) =>
if (seenTypeSet.contains(t)) {
- throw EncoderErrors.cannotHaveCircularReferencesInClassError(t.toString)
+ throw ExecutionErrors.cannotHaveCircularReferencesInClassError(t.toString)
}
val params = getConstructorParameters(t).map {
case (fieldName, fieldType) =>
if (SourceVersion.isKeyword(fieldName) ||
!SourceVersion.isIdentifier(encodeFieldNameToIdentifier(fieldName))) {
- throw EncoderErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
+ throw ExecutionErrors.cannotUseInvalidJavaIdentifierAsFieldNameError(
fieldName,
path)
}
@@ -396,7 +396,7 @@ object ScalaReflection extends ScalaReflection {
}
ProductEncoder(ClassTag(getClassFromType(t)), params)
case _ =>
- throw EncoderErrors.cannotFindEncoderForTypeError(tpe.toString)
+ throw ExecutionErrors.cannotFindEncoderForTypeError(tpe.toString)
}
}
}
@@ -477,7 +477,7 @@ trait ScalaReflection extends Logging {
*/
private def getCompanionConstructor(tpe: Type): Symbol = {
def throwUnsupportedOperation = {
- throw EncoderErrors.cannotFindConstructorForTypeError(tpe.toString)
+ throw ExecutionErrors.cannotFindConstructorForTypeError(tpe.toString)
}
tpe.typeSymbol.asClass.companion match {
case NoSymbol => throwUnsupportedOperation
@@ -500,7 +500,7 @@ trait ScalaReflection extends Logging {
val primaryConstructorSymbol: Option[Symbol] = constructorSymbol.asTerm.alternatives.find(
s => s.isMethod && s.asMethod.isPrimaryConstructor)
if (primaryConstructorSymbol.isEmpty) {
- throw EncoderErrors.primaryConstructorNotFoundError(tpe.getClass)
+ throw ExecutionErrors.primaryConstructorNotFoundError(tpe.getClass)
} else {
primaryConstructorSymbol.get.asMethod.paramLists
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
index 5ab9c3bfc48..7d214768fbc 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
@@ -20,9 +20,10 @@ package org.apache.spark.sql.catalyst.encoders
import scala.collection.mutable
import scala.reflect.classTag
-import org.apache.spark.sql.{Row, SqlApiConf}
+import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{BinaryEncoder, BoxedBooleanEncoder, BoxedByteEncoder, BoxedDoubleEncoder, BoxedFloatEncoder, BoxedIntEncoder, BoxedLongEncoder, BoxedShortEncoder, CalendarIntervalEncoder, DateEncoder, DayTimeIntervalEncoder, EncoderField, InstantEncoder, IterableEncoder, JavaDecimalEncoder, LocalDateEncoder, LocalDateTimeEncoder, MapEncoder, NullEncoder, RowEncoder => AgnosticRowEncoder, StringEncoder, TimestampEncoder, UDTEncoder, YearMont [...]
-import org.apache.spark.sql.errors.EncoderErrors
+import org.apache.spark.sql.errors.ExecutionErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types._
/**
@@ -97,7 +98,7 @@ object RowEncoder {
annotation.udt()
} else {
UDTRegistration.getUDTFor(udt.userClass.getName).getOrElse {
- throw EncoderErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
+ throw ExecutionErrors.userDefinedTypeNotAnnotatedAndRegisteredError(udt)
}
}
UDTEncoder(udt, udtClass.asInstanceOf[Class[_ <: UserDefinedType[_]]])
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/DataTypeAstBuilder.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/DataTypeAstBuilder.scala
index f42137e2d3f..97027604fd8 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/DataTypeAstBuilder.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/DataTypeAstBuilder.scala
@@ -23,10 +23,10 @@ import scala.collection.JavaConverters._
import org.antlr.v4.runtime.Token
import org.antlr.v4.runtime.tree.ParseTree
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
import org.apache.spark.sql.catalyst.util.SparkParserUtils.{string, withOrigin}
import org.apache.spark.sql.errors.QueryParsingErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType, CalendarIntervalType, CharType, DataType, DateType, DayTimeIntervalType, DecimalType, DoubleType, FloatType, IntegerType, LongType, MapType, MetadataBuilder, NullType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, VarcharType, YearMonthIntervalType}
class DataTypeAstBuilder extends SqlBaseParserBaseVisitor[AnyRef] {
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala
index 39d369fed94..c3a051be89b 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/parser/parsers.scala
@@ -25,10 +25,11 @@ import org.antlr.v4.runtime.tree.TerminalNodeImpl
import org.apache.spark.{QueryContext, SparkThrowable, SparkThrowableHelper}
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.{AnalysisException, SqlApiConf}
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, WithOrigin}
import org.apache.spark.sql.catalyst.util.SparkParserUtils
import org.apache.spark.sql.errors.QueryParsingErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types.{DataType, StructType}
/**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala
similarity index 92%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala
index 0aa64bcf5b7..a41d2a0df48 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala
@@ -23,9 +23,9 @@ import java.util.{Date, Locale}
import org.apache.commons.lang3.time.FastDateFormat
-import org.apache.spark.sql.catalyst.util.DateTimeUtils._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.unsafe.types.UTF8String
sealed trait DateFormatter extends Serializable {
@@ -99,7 +99,7 @@ class DefaultDateFormatter(
override def parse(s: String): Int = {
try {
- DateTimeUtils.stringToDateAnsi(UTF8String.fromString(s))
+ SparkDateTimeUtils.stringToDateAnsi(UTF8String.fromString(s))
} catch checkParsedDiff(s, legacyFormatter.parse)
}
}
@@ -112,7 +112,7 @@ trait LegacyDateFormatter extends DateFormatter {
}
override def format(days: Int): String = {
- format(DateTimeUtils.toJavaDate(days))
+ format(SparkDateTimeUtils.toJavaDate(days))
}
override def format(localDate: LocalDate): String = {
@@ -125,7 +125,7 @@ trait LegacyDateFormatter extends DateFormatter {
* JVM time zone intentionally for compatibility with Spark 2.4 and earlier versions.
*
* Note: Using of the default JVM time zone makes the formatter compatible with the legacy
- * `DateTimeUtils` methods `toJavaDate` and `fromJavaDate` that are based on the default
+ * `SparkDateTimeUtils` methods `toJavaDate` and `fromJavaDate` that are based on the default
* JVM time zone too.
*
* @param pattern `java.text.SimpleDateFormat` compatible pattern.
@@ -145,7 +145,7 @@ class LegacyFastDateFormatter(pattern: String, locale: Locale) extends LegacyDat
* JVM time zone intentionally for compatibility with Spark 2.4 and earlier versions.
*
* Note: Using of the default JVM time zone makes the formatter compatible with the legacy
- * `DateTimeUtils` methods `toJavaDate` and `fromJavaDate` that are based on the default
+ * `SparkDateTimeUtils` methods `toJavaDate` and `fromJavaDate` that are based on the default
* JVM time zone too.
*
* @param pattern The pattern describing the date and time format.
@@ -176,7 +176,7 @@ object DateFormatter {
locale: Locale = defaultLocale,
legacyFormat: LegacyDateFormat = LENIENT_SIMPLE_DATE_FORMAT,
isParsing: Boolean): DateFormatter = {
- if (SQLConf.get.legacyTimeParserPolicy == LEGACY) {
+ if (SqlApiConf.get.legacyTimeParserPolicy == LEGACY) {
getLegacyFormatter(format.getOrElse(defaultPattern), locale, legacyFormat)
} else {
val df = format
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala
similarity index 93%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala
index fe14d74488c..43701d1d8ff 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeFormatterHelper.scala
@@ -21,15 +21,13 @@ import java.time._
import java.time.chrono.IsoChronology
import java.time.format.{DateTimeFormatter, DateTimeFormatterBuilder, ResolverStyle}
import java.time.temporal.{ChronoField, TemporalAccessor, TemporalQueries}
-import java.util.{Date, Locale}
+import java.util
+import java.util.{Collections, Date, Locale}
-import com.google.common.cache.CacheBuilder
-
-import org.apache.spark.SPARK_DOC_ROOT
import org.apache.spark.sql.catalyst.util.DateTimeFormatterHelper._
-import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
+import org.apache.spark.sql.errors.ExecutionErrors
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
+import org.apache.spark.sql.internal.SqlApiConf
trait DateTimeFormatterHelper {
private def getOrDefault(accessor: TemporalAccessor, field: ChronoField, default: Int): Int = {
@@ -46,7 +44,7 @@ trait DateTimeFormatterHelper {
val actual = accessor.get(field)
val expected = candidate.get(field)
if (actual != expected) {
- throw QueryExecutionErrors.fieldDiffersFromDerivedLocalDateError(
+ throw ExecutionErrors.fieldDiffersFromDerivedLocalDateError(
field, actual, expected, candidate)
}
}
@@ -122,16 +120,11 @@ trait DateTimeFormatterHelper {
val newPattern = convertIncompatiblePattern(pattern, isParsing)
val useVarLen = isParsing && newPattern.contains('S')
val key = (newPattern, locale, useVarLen)
- var formatter = cache.getIfPresent(key)
- if (formatter == null) {
- formatter = buildFormatter(newPattern, locale, useVarLen)
- cache.put(key, formatter)
- }
- formatter
+ cache.computeIfAbsent(key, _ => buildFormatter(newPattern, locale, useVarLen))
}
private def needConvertToSparkUpgradeException(e: Throwable): Boolean = e match {
- case _: DateTimeException if SQLConf.get.legacyTimeParserPolicy == EXCEPTION => true
+ case _: DateTimeException if SqlApiConf.get.legacyTimeParserPolicy == EXCEPTION => true
case _ => false
}
// When legacy time parser policy set to EXCEPTION, check whether we will get different results
@@ -146,7 +139,7 @@ trait DateTimeFormatterHelper {
} catch {
case _: Throwable => throw e
}
- throw QueryExecutionErrors.failToParseDateTimeInNewParserError(s, e)
+ throw ExecutionErrors.failToParseDateTimeInNewParserError(s, e)
}
// When legacy time parser policy set to EXCEPTION, check whether we will get different results
@@ -162,7 +155,7 @@ trait DateTimeFormatterHelper {
} catch {
case _: Throwable => throw e
}
- throw QueryExecutionErrors.failToParseDateTimeInNewParserError(resultCandidate, e)
+ throw ExecutionErrors.failToParseDateTimeInNewParserError(resultCandidate, e)
}
/**
@@ -185,20 +178,24 @@ trait DateTimeFormatterHelper {
} catch {
case _: Throwable => throw e
}
- throw QueryExecutionErrors.failToRecognizePatternAfterUpgradeError(
- pattern, e, SPARK_DOC_ROOT)
+ throw ExecutionErrors.failToRecognizePatternAfterUpgradeError(pattern, e)
}
protected def checkInvalidPattern(pattern: String): PartialFunction[Throwable, Nothing] = {
case e: IllegalArgumentException =>
- throw QueryExecutionErrors.failToRecognizePatternError(pattern, e, SPARK_DOC_ROOT)
+ throw ExecutionErrors.failToRecognizePatternError(pattern, e)
}
}
private object DateTimeFormatterHelper {
- val cache = CacheBuilder.newBuilder()
- .maximumSize(128)
- .build[(String, Locale, Boolean), DateTimeFormatter]()
+ private type CacheKey = (String, Locale, Boolean)
+ private val cache: util.Map[CacheKey, DateTimeFormatter] =
+ Collections.synchronizedMap(new util.LinkedHashMap[CacheKey, DateTimeFormatter] {
+ override def removeEldestEntry(
+ eldest: util.Map.Entry[CacheKey, DateTimeFormatter]): Boolean = {
+ size() > 128
+ }
+ })
final val extractor = "^([^S]*)(S*)(.*)$".r
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala
similarity index 95%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala
index b285b1df572..7c1b37e9e58 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/MathUtils.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.catalyst.util
import org.apache.spark.sql.catalyst.trees.SQLQueryContext
-import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.errors.ExecutionErrors
/**
* Helper functions for arithmetic operations with overflow.
@@ -83,7 +83,7 @@ object MathUtils {
f
} catch {
case e: ArithmeticException =>
- throw QueryExecutionErrors.arithmeticOverflowError(e.getMessage, hint, context)
+ throw ExecutionErrors.arithmeticOverflowError(e.getMessage, hint, context)
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
similarity index 99%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
index a2a63e2af42..f9566b0e1fb 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/RebaseDateTime.scala
@@ -28,9 +28,9 @@ import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
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
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
+import org.apache.spark.util.SparkClassUtils
/**
* The collection of functions for rebasing days and microseconds from/to the hybrid calendar
@@ -278,7 +278,7 @@ object RebaseDateTime {
// `JsonRebaseRecord`. AnyRefMap is used here instead of Scala's immutable map because
// it is 2 times faster in DateTimeRebaseBenchmark.
private[sql] def loadRebaseRecords(fileName: String): AnyRefMap[String, RebaseInfo] = {
- val file = Utils.getSparkClassLoader.getResource(fileName)
+ val file = SparkClassUtils.getSparkClassLoader.getResource(fileName)
val jsonRebaseRecords = mapper.readValue[Seq[JsonRebaseRecord]](file)
val anyRefMap = new AnyRefMap[String, RebaseInfo]((3 * jsonRebaseRecords.size) / 2)
jsonRebaseRecords.foreach { jsonRecord =>
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
index 47b3ef9b05a..498eb83566e 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala
@@ -16,8 +16,8 @@
*/
package org.apache.spark.sql.catalyst.util
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types.{ArrayType, CharType, DataType, MapType, StringType, StructType, VarcharType}
trait SparkCharVarcharUtils {
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
index e96da99c6b0..698e7b37a9e 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala
@@ -16,9 +16,61 @@
*/
package org.apache.spark.sql.catalyst.util
-import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_MILLIS
+import java.sql.{Date, Timestamp}
+import java.time.{Instant, LocalDate, LocalDateTime, LocalTime, ZonedDateTime, ZoneId, ZoneOffset}
+import java.util.TimeZone
+import java.util.concurrent.TimeUnit.{MICROSECONDS, NANOSECONDS}
+
+import scala.util.control.NonFatal
+
+import sun.util.calendar.ZoneInfo
+
+import org.apache.spark.sql.catalyst.trees.SQLQueryContext
+import org.apache.spark.sql.catalyst.util.DateTimeConstants._
+import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros, rebaseJulianToGregorianDays, rebaseJulianToGregorianMicros}
+import org.apache.spark.sql.errors.ExecutionErrors
+import org.apache.spark.sql.types.{DateType, TimestampType}
+import org.apache.spark.unsafe.types.UTF8String
+
+trait SparkDateTimeUtils {
+
+ final val TimeZoneUTC = TimeZone.getTimeZone("UTC")
+
+ def getZoneId(timeZoneId: String): ZoneId = {
+ val formattedZoneId = timeZoneId
+ // To support the (+|-)h:mm format because it was supported before Spark 3.0.
+ .replaceFirst("(\\+|\\-)(\\d):", "$10$2:")
+ // To support the (+|-)hh:m format because it was supported before Spark 3.0.
+ .replaceFirst("(\\+|\\-)(\\d\\d):(\\d)$", "$1$2:0$3")
+
+ ZoneId.of(formattedZoneId, ZoneId.SHORT_IDS)
+ }
+
+ def getTimeZone(timeZoneId: String): TimeZone = TimeZone.getTimeZone(getZoneId(timeZoneId))
+
+ /**
+ * Converts an Java object to days.
+ *
+ * @param obj Either an object of `java.sql.Date` or `java.time.LocalDate`.
+ * @return The number of days since 1970-01-01.
+ */
+ def anyToDays(obj: Any): Int = obj match {
+ case d: Date => fromJavaDate(d)
+ case ld: LocalDate => localDateToDays(ld)
+ }
+
+ /**
+ * Converts an Java object to microseconds.
+ *
+ * @param obj Either an object of `java.sql.Timestamp` or `java.time.{Instant,LocalDateTime}`.
+ * @return The number of micros since the epoch.
+ */
+ def anyToMicros(obj: Any): Long = obj match {
+ case t: Timestamp => fromJavaTimestamp(t)
+ case i: Instant => instantToMicros(i)
+ case ldt: LocalDateTime => localDateTimeToMicros(ldt)
+ }
-object SparkDateTimeUtils {
/**
* Converts the timestamp to milliseconds since epoch. In Spark timestamp values have microseconds
* precision, so this conversion is lossy.
@@ -36,4 +88,520 @@ object SparkDateTimeUtils {
def millisToMicros(millis: Long): Long = {
Math.multiplyExact(millis, MICROS_PER_MILLIS)
}
+
+ // See issue SPARK-35679
+ // min second cause overflow in instant to micro
+ private val MIN_SECONDS = Math.floorDiv(Long.MinValue, MICROS_PER_SECOND)
+
+ /**
+ * Obtains an instance of `java.time.Instant` using microseconds from
+ * the epoch of 1970-01-01 00:00:00Z.
+ */
+ def microsToInstant(micros: Long): Instant = {
+ val secs = Math.floorDiv(micros, MICROS_PER_SECOND)
+ // Unfolded Math.floorMod(us, MICROS_PER_SECOND) to reuse the result of
+ // the above calculation of `secs` via `floorDiv`.
+ val mos = micros - secs * MICROS_PER_SECOND
+ Instant.ofEpochSecond(secs, mos * NANOS_PER_MICROS)
+ }
+
+ /**
+ * Gets the number of microseconds since the epoch of 1970-01-01 00:00:00Z from the given
+ * instance of `java.time.Instant`. The epoch microsecond count is a simple incrementing count of
+ * microseconds where microsecond 0 is 1970-01-01 00:00:00Z.
+ */
+ def instantToMicros(instant: Instant): Long = {
+ val secs = instant.getEpochSecond
+ if (secs == MIN_SECONDS) {
+ val us = Math.multiplyExact(secs + 1, MICROS_PER_SECOND)
+ Math.addExact(us, NANOSECONDS.toMicros(instant.getNano) - MICROS_PER_SECOND)
+ } else {
+ val us = Math.multiplyExact(secs, MICROS_PER_SECOND)
+ Math.addExact(us, NANOSECONDS.toMicros(instant.getNano))
+ }
+ }
+
+ /**
+ * Converts the timestamp `micros` from one timezone to another.
+ *
+ * Time-zone rules, such as daylight savings, mean that not every local date-time
+ * is valid for the `toZone` time zone, thus the local date-time may be adjusted.
+ */
+ def convertTz(micros: Long, fromZone: ZoneId, toZone: ZoneId): Long = {
+ val rebasedDateTime = getLocalDateTime(micros, toZone).atZone(fromZone)
+ instantToMicros(rebasedDateTime.toInstant)
+ }
+
+ // Gets the local date-time parts (year, month, day and time) of the instant expressed as the
+ // number of microseconds since the epoch at the given time zone ID.
+ protected def getLocalDateTime(micros: Long, zoneId: ZoneId): LocalDateTime = {
+ microsToInstant(micros).atZone(zoneId).toLocalDateTime
+ }
+
+ def microsToLocalDateTime(micros: Long): LocalDateTime = {
+ getLocalDateTime(micros, ZoneOffset.UTC)
+ }
+
+ def localDateTimeToMicros(localDateTime: LocalDateTime): Long = {
+ instantToMicros(localDateTime.toInstant(ZoneOffset.UTC))
+ }
+
+ /**
+ * Converts the local date to the number of days since 1970-01-01.
+ */
+ def localDateToDays(localDate: LocalDate): Int = MathUtils.toIntExact(localDate.toEpochDay)
+
+ /**
+ * Obtains an instance of `java.time.LocalDate` from the epoch day count.
+ */
+ def daysToLocalDate(days: Int): LocalDate = LocalDate.ofEpochDay(days)
+
+ /**
+ * Converts microseconds since 1970-01-01 00:00:00Z to days since 1970-01-01 at the given zone ID.
+ */
+ def microsToDays(micros: Long, zoneId: ZoneId): Int = {
+ localDateToDays(getLocalDateTime(micros, zoneId).toLocalDate)
+ }
+
+ /**
+ * Converts days since 1970-01-01 at the given zone ID to microseconds since 1970-01-01 00:00:00Z.
+ */
+ def daysToMicros(days: Int, zoneId: ZoneId): Long = {
+ val instant = daysToLocalDate(days).atStartOfDay(zoneId).toInstant
+ instantToMicros(instant)
+ }
+
+ /**
+ * Converts a local date at the default JVM time zone to the number of days since 1970-01-01
+ * in the hybrid calendar (Julian + Gregorian) by discarding the time part. The resulted days are
+ * rebased from the hybrid to Proleptic Gregorian calendar. The days rebasing is performed via
+ * UTC time zone for simplicity because the difference between two calendars is the same in
+ * any given time zone and UTC time zone.
+ *
+ * Note: The date is shifted by the offset of the default JVM time zone for backward compatibility
+ * with Spark 2.4 and earlier versions. The goal of the shift is to get a local date derived
+ * from the number of days that has the same date fields (year, month, day) as the original
+ * `date` at the default JVM time zone.
+ *
+ * @param date It represents a specific instant in time based on the hybrid calendar which
+ * combines Julian and Gregorian calendars.
+ * @return The number of days since the epoch in Proleptic Gregorian calendar.
+ */
+ def fromJavaDate(date: Date): Int = {
+ val millisUtc = date.getTime
+ val millisLocal = millisUtc + TimeZone.getDefault.getOffset(millisUtc)
+ val julianDays = Math.toIntExact(Math.floorDiv(millisLocal, MILLIS_PER_DAY))
+ rebaseJulianToGregorianDays(julianDays)
+ }
+
+ /**
+ * Converts days since the epoch 1970-01-01 in Proleptic Gregorian calendar to a local date
+ * at the default JVM time zone in the hybrid calendar (Julian + Gregorian). It rebases the given
+ * days from Proleptic Gregorian to the hybrid calendar at UTC time zone for simplicity because
+ * the difference between two calendars doesn't depend on any time zone. The result is shifted
+ * by the time zone offset in wall clock to have the same date fields (year, month, day)
+ * at the default JVM time zone as the input `daysSinceEpoch` in Proleptic Gregorian calendar.
+ *
+ * Note: The date is shifted by the offset of the default JVM time zone for backward compatibility
+ * with Spark 2.4 and earlier versions.
+ *
+ * @param days The number of days since 1970-01-01 in Proleptic Gregorian calendar.
+ * @return A local date in the hybrid calendar as `java.sql.Date` from number of days since epoch.
+ */
+ def toJavaDate(days: Int): Date = {
+ val rebasedDays = rebaseGregorianToJulianDays(days)
+ val localMillis = Math.multiplyExact(rebasedDays, MILLIS_PER_DAY)
+ val timeZoneOffset = TimeZone.getDefault match {
+ case zoneInfo: ZoneInfo => zoneInfo.getOffsetsByWall(localMillis, null)
+ case timeZone: TimeZone => timeZone.getOffset(localMillis - timeZone.getRawOffset)
+ }
+ new Date(localMillis - timeZoneOffset)
+ }
+
+ /**
+ * Converts microseconds since the epoch to an instance of `java.sql.Timestamp`
+ * via creating a local timestamp at the system time zone in Proleptic Gregorian
+ * calendar, extracting date and time fields like `year` and `hours`, and forming
+ * new timestamp in the hybrid calendar from the extracted fields.
+ *
+ * The conversion is based on the JVM system time zone because the `java.sql.Timestamp`
+ * uses the time zone internally.
+ *
+ * The method performs the conversion via local timestamp fields to have the same date-time
+ * representation as `year`, `month`, `day`, ..., `seconds` in the original calendar
+ * and in the target calendar.
+ *
+ * @param micros The number of microseconds since 1970-01-01T00:00:00.000000Z.
+ * @return A `java.sql.Timestamp` from number of micros since epoch.
+ */
+ def toJavaTimestamp(micros: Long): Timestamp =
+ toJavaTimestampNoRebase(rebaseGregorianToJulianMicros(micros))
+
+ /**
+ * Converts microseconds since the epoch to an instance of `java.sql.Timestamp`.
+ *
+ * @param micros The number of microseconds since 1970-01-01T00:00:00.000000Z.
+ * @return A `java.sql.Timestamp` from number of micros since epoch.
+ */
+ def toJavaTimestampNoRebase(micros: Long): Timestamp = {
+ val seconds = Math.floorDiv(micros, MICROS_PER_SECOND)
+ val ts = new Timestamp(seconds * MILLIS_PER_SECOND)
+ val nanos = (micros - seconds * MICROS_PER_SECOND) * NANOS_PER_MICROS
+ ts.setNanos(nanos.toInt)
+ ts
+ }
+
+ /**
+ * Converts an instance of `java.sql.Timestamp` to the number of microseconds since
+ * 1970-01-01T00:00:00.000000Z. It extracts date-time fields from the input, builds
+ * a local timestamp in Proleptic Gregorian calendar from the fields, and binds
+ * the timestamp to the system time zone. The resulted instant is converted to
+ * microseconds since the epoch.
+ *
+ * The conversion is performed via the system time zone because it is used internally
+ * in `java.sql.Timestamp` while extracting date-time fields.
+ *
+ * The goal of the function is to have the same local date-time in the original calendar
+ * - the hybrid calendar (Julian + Gregorian) and in the target calendar which is
+ * Proleptic Gregorian calendar, see SPARK-26651.
+ *
+ * @param t It represents a specific instant in time based on
+ * the hybrid calendar which combines Julian and
+ * Gregorian calendars.
+ * @return The number of micros since epoch from `java.sql.Timestamp`.
+ */
+ def fromJavaTimestamp(t: Timestamp): Long =
+ rebaseJulianToGregorianMicros(fromJavaTimestampNoRebase(t))
+
+ /**
+ * Converts an instance of `java.sql.Timestamp` to the number of microseconds since
+ * 1970-01-01T00:00:00.000000Z.
+ *
+ * @param t an instance of `java.sql.Timestamp`.
+ * @return The number of micros since epoch from `java.sql.Timestamp`.
+ */
+ def fromJavaTimestampNoRebase(t: Timestamp): Long =
+ millisToMicros(t.getTime) + (t.getNanos / NANOS_PER_MICROS) % MICROS_PER_MILLIS
+
+ /**
+ * Trims and parses a given UTF8 date string to a corresponding [[Int]] value.
+ * The return type is [[Option]] in order to distinguish between 0 and null. The following
+ * formats are allowed:
+ *
+ * `[+-]yyyy*`
+ * `[+-]yyyy*-[m]m`
+ * `[+-]yyyy*-[m]m-[d]d`
+ * `[+-]yyyy*-[m]m-[d]d `
+ * `[+-]yyyy*-[m]m-[d]d *`
+ * `[+-]yyyy*-[m]m-[d]dT*`
+ */
+ def stringToDate(s: UTF8String): Option[Int] = {
+ def isValidDigits(segment: Int, digits: Int): Boolean = {
+ // An integer is able to represent a date within [+-]5 million years.
+ val maxDigitsYear = 7
+ (segment == 0 && digits >= 4 && digits <= maxDigitsYear) ||
+ (segment != 0 && digits > 0 && digits <= 2)
+ }
+ if (s == null || s.trimAll().numBytes() == 0) {
+ return None
+ }
+ val segments: Array[Int] = Array[Int](1, 1, 1)
+ var sign = 1
+ var i = 0
+ var currentSegmentValue = 0
+ var currentSegmentDigits = 0
+ val bytes = s.trimAll().getBytes
+ var j = 0
+ if (bytes(j) == '-' || bytes(j) == '+') {
+ sign = if (bytes(j) == '-') -1 else 1
+ j += 1
+ }
+ while (j < bytes.length && (i < 3 && !(bytes(j) == ' ' || bytes(j) == 'T'))) {
+ val b = bytes(j)
+ if (i < 2 && b == '-') {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return None
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else {
+ val parsedValue = b - '0'.toByte
+ if (parsedValue < 0 || parsedValue > 9) {
+ return None
+ } else {
+ currentSegmentValue = currentSegmentValue * 10 + parsedValue
+ currentSegmentDigits += 1
+ }
+ }
+ j += 1
+ }
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return None
+ }
+ if (i < 2 && j < bytes.length) {
+ // For the `yyyy` and `yyyy-[m]m` formats, entire input must be consumed.
+ return None
+ }
+ segments(i) = currentSegmentValue
+ try {
+ val localDate = LocalDate.of(sign * segments(0), segments(1), segments(2))
+ Some(localDateToDays(localDate))
+ } catch {
+ case NonFatal(_) => None
+ }
+ }
+
+ def stringToDateAnsi(
+ s: UTF8String,
+ context: SQLQueryContext = null): Int = {
+ stringToDate(s).getOrElse {
+ throw ExecutionErrors.invalidInputInCastToDatetimeError(s, DateType, context)
+ }
+ }
+
+ /**
+ * Trims and parses a given UTF8 timestamp string to the corresponding timestamp segments,
+ * time zone id and whether it is just time without a date.
+ * value. The return type is [[Option]] in order to distinguish between 0L and null. The following
+ * formats are allowed:
+ *
+ * `[+-]yyyy*`
+ * `[+-]yyyy*-[m]m`
+ * `[+-]yyyy*-[m]m-[d]d`
+ * `[+-]yyyy*-[m]m-[d]d `
+ * `[+-]yyyy*-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
+ * `[+-]yyyy*-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
+ * `[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
+ * `T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
+ *
+ * where `zone_id` should have one of the forms:
+ * - Z - Zulu time zone UTC+0
+ * - +|-[h]h:[m]m
+ * - A short id, see https://docs.oracle.com/javase/8/docs/api/java/time/ZoneId.html#SHORT_IDS
+ * - An id with one of the prefixes UTC+, UTC-, GMT+, GMT-, UT+ or UT-,
+ * and a suffix in the formats:
+ * - +|-h[h]
+ * - +|-hh[:]mm
+ * - +|-hh:mm:ss
+ * - +|-hhmmss
+ * - Region-based zone IDs in the form `area/city`, such as `Europe/Paris`
+ *
+ * @return timestamp segments, time zone id and whether the input is just time without a date. If
+ * the input string can't be parsed as timestamp, the result timestamp segments are empty.
+ */
+ def parseTimestampString(s: UTF8String): (Array[Int], Option[ZoneId], Boolean) = {
+ def isValidDigits(segment: Int, digits: Int): Boolean = {
+ // A Long is able to represent a timestamp within [+-]200 thousand years
+ val maxDigitsYear = 6
+ // For the nanosecond part, more than 6 digits is allowed, but will be truncated.
+ segment == 6 || (segment == 0 && digits >= 4 && digits <= maxDigitsYear) ||
+ // For the zoneId segment(7), it's could be zero digits when it's a region-based zone ID
+ (segment == 7 && digits <= 2) ||
+ (segment != 0 && segment != 6 && segment != 7 && digits > 0 && digits <= 2)
+ }
+ if (s == null || s.trimAll().numBytes() == 0) {
+ return (Array.empty, None, false)
+ }
+ var tz: Option[String] = None
+ val segments: Array[Int] = Array[Int](1, 1, 1, 0, 0, 0, 0, 0, 0)
+ var i = 0
+ var currentSegmentValue = 0
+ var currentSegmentDigits = 0
+ val bytes = s.trimAll().getBytes
+ var j = 0
+ var digitsMilli = 0
+ var justTime = false
+ var yearSign: Option[Int] = None
+ if (bytes(j) == '-' || bytes(j) == '+') {
+ yearSign = if (bytes(j) == '-') Some(-1) else Some(1)
+ j += 1
+ }
+ while (j < bytes.length) {
+ val b = bytes(j)
+ val parsedValue = b - '0'.toByte
+ if (parsedValue < 0 || parsedValue > 9) {
+ if (j == 0 && b == 'T') {
+ justTime = true
+ i += 3
+ } else if (i < 2) {
+ if (b == '-') {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else if (i == 0 && b == ':' && yearSign.isEmpty) {
+ justTime = true
+ if (!isValidDigits(3, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(3) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i = 4
+ } else {
+ return (Array.empty, None, false)
+ }
+ } else if (i == 2) {
+ if (b == ' ' || b == 'T') {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else {
+ return (Array.empty, None, false)
+ }
+ } else if (i == 3 || i == 4) {
+ if (b == ':') {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else {
+ return (Array.empty, None, false)
+ }
+ } else if (i == 5 || i == 6) {
+ if (b == '.' && i == 5) {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ tz = Some(new String(bytes, j, bytes.length - j))
+ j = bytes.length - 1
+ }
+ if (i == 6 && b != '.') {
+ i += 1
+ }
+ } else {
+ if (i < segments.length && (b == ':' || b == ' ')) {
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+ currentSegmentValue = 0
+ currentSegmentDigits = 0
+ i += 1
+ } else {
+ return (Array.empty, None, false)
+ }
+ }
+ } else {
+ if (i == 6) {
+ digitsMilli += 1
+ }
+ // We will truncate the nanosecond part if there are more than 6 digits, which results
+ // in loss of precision
+ if (i != 6 || currentSegmentDigits < 6) {
+ currentSegmentValue = currentSegmentValue * 10 + parsedValue
+ }
+ currentSegmentDigits += 1
+ }
+ j += 1
+ }
+
+ if (!isValidDigits(i, currentSegmentDigits)) {
+ return (Array.empty, None, false)
+ }
+ segments(i) = currentSegmentValue
+
+ while (digitsMilli < 6) {
+ segments(6) *= 10
+ digitsMilli += 1
+ }
+
+ // This step also validates time zone part
+ val zoneId = tz.map(zoneName => getZoneId(zoneName.trim))
+ segments(0) *= yearSign.getOrElse(1)
+ (segments, zoneId, justTime)
+ }
+
+ /**
+ * Trims and parses a given UTF8 timestamp string to the corresponding a corresponding [[Long]]
+ * value. The return type is [[Option]] in order to distinguish between 0L and null. Please
+ * refer to `parseTimestampString` for the allowed formats
+ */
+ def stringToTimestamp(s: UTF8String, timeZoneId: ZoneId): Option[Long] = {
+ try {
+ val (segments, parsedZoneId, justTime) = parseTimestampString(s)
+ if (segments.isEmpty) {
+ return None
+ }
+ val zoneId = parsedZoneId.getOrElse(timeZoneId)
+ val nanoseconds = MICROSECONDS.toNanos(segments(6))
+ val localTime = LocalTime.of(segments(3), segments(4), segments(5), nanoseconds.toInt)
+ val localDate = if (justTime) {
+ LocalDate.now(zoneId)
+ } else {
+ LocalDate.of(segments(0), segments(1), segments(2))
+ }
+ val localDateTime = LocalDateTime.of(localDate, localTime)
+ val zonedDateTime = ZonedDateTime.of(localDateTime, zoneId)
+ val instant = Instant.from(zonedDateTime)
+ Some(instantToMicros(instant))
+ } catch {
+ case NonFatal(_) => None
+ }
+ }
+
+ def stringToTimestampAnsi(
+ s: UTF8String,
+ timeZoneId: ZoneId,
+ context: SQLQueryContext = null): Long = {
+ stringToTimestamp(s, timeZoneId).getOrElse {
+ throw ExecutionErrors.invalidInputInCastToDatetimeError(s, TimestampType, context)
+ }
+ }
+
+ /**
+ * Trims and parses a given UTF8 string to a corresponding [[Long]] value which representing the
+ * number of microseconds since the epoch. The result will be independent of time zones.
+ *
+ * If the input string contains a component associated with time zone, the method will return
+ * `None` if `allowTimeZone` is set to `false`. If `allowTimeZone` is set to `true`, the method
+ * will simply discard the time zone component. Enable the check to detect situations like parsing
+ * a timestamp with time zone as TimestampNTZType.
+ *
+ * The return type is [[Option]] in order to distinguish between 0L and null. Please
+ * refer to `parseTimestampString` for the allowed formats.
+ */
+ def stringToTimestampWithoutTimeZone(s: UTF8String, allowTimeZone: Boolean): Option[Long] = {
+ try {
+ val (segments, zoneIdOpt, justTime) = parseTimestampString(s)
+ // If the input string can't be parsed as a timestamp without time zone, or it contains only
+ // the time part of a timestamp and we can't determine its date, return None.
+ if (segments.isEmpty || justTime || !allowTimeZone && zoneIdOpt.isDefined) {
+ return None
+ }
+ val nanoseconds = MICROSECONDS.toNanos(segments(6))
+ val localTime = LocalTime.of(segments(3), segments(4), segments(5), nanoseconds.toInt)
+ val localDate = LocalDate.of(segments(0), segments(1), segments(2))
+ val localDateTime = LocalDateTime.of(localDate, localTime)
+ Some(localDateTimeToMicros(localDateTime))
+ } catch {
+ case NonFatal(_) => None
+ }
+ }
}
+
+object SparkDateTimeUtils extends SparkDateTimeUtils
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala
similarity index 95%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala
index aab90ec3844..8a288d0e9f3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/TimestampFormatter.scala
@@ -30,12 +30,12 @@ import scala.util.control.NonFatal
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
-import org.apache.spark.sql.catalyst.util.DateTimeUtils._
import org.apache.spark.sql.catalyst.util.LegacyDateFormats.{LegacyDateFormat, LENIENT_SIMPLE_DATE_FORMAT}
import org.apache.spark.sql.catalyst.util.RebaseDateTime._
-import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
+import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils._
+import org.apache.spark.sql.errors.ExecutionErrors
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types.{Decimal, TimestampNTZType}
import org.apache.spark.unsafe.types.UTF8String
@@ -212,11 +212,11 @@ class Iso8601TimestampFormatter(
parsed: TemporalAccessor,
allowTimeZone: Boolean): Long = {
if (!allowTimeZone && parsed.query(TemporalQueries.zone()) != null) {
- throw QueryExecutionErrors.cannotParseStringAsDataTypeError(pattern, s, TimestampNTZType)
+ throw ExecutionErrors.cannotParseStringAsDataTypeError(pattern, s, TimestampNTZType)
}
val localDate = toLocalDate(parsed)
val localTime = toLocalTime(parsed)
- DateTimeUtils.localDateTimeToMicros(LocalDateTime.of(localDate, localTime))
+ SparkDateTimeUtils.localDateTimeToMicros(LocalDateTime.of(localDate, localTime))
}
override def parseWithoutTimeZone(s: String, allowTimeZone: Boolean): Long = {
@@ -234,7 +234,7 @@ class Iso8601TimestampFormatter(
}
override def format(us: Long): String = {
- val instant = DateTimeUtils.microsToInstant(us)
+ val instant = SparkDateTimeUtils.microsToInstant(us)
format(instant)
}
@@ -281,18 +281,18 @@ class DefaultTimestampFormatter(
override def parse(s: String): Long = {
try {
- DateTimeUtils.stringToTimestampAnsi(UTF8String.fromString(s), zoneId)
+ SparkDateTimeUtils.stringToTimestampAnsi(UTF8String.fromString(s), zoneId)
} catch checkParsedDiff(s, legacyFormatter.parse)
}
override def parseOptional(s: String): Option[Long] =
- DateTimeUtils.stringToTimestamp(UTF8String.fromString(s), zoneId)
+ SparkDateTimeUtils.stringToTimestamp(UTF8String.fromString(s), zoneId)
override def parseWithoutTimeZone(s: String, allowTimeZone: Boolean): Long = {
try {
val utf8Value = UTF8String.fromString(s)
- DateTimeUtils.stringToTimestampWithoutTimeZone(utf8Value, allowTimeZone).getOrElse {
- throw QueryExecutionErrors.cannotParseStringAsDataTypeError(
+ SparkDateTimeUtils.stringToTimestampWithoutTimeZone(utf8Value, allowTimeZone).getOrElse {
+ throw ExecutionErrors.cannotParseStringAsDataTypeError(
TimestampFormatter.defaultPattern(), s, TimestampNTZType)
}
} catch checkParsedDiff(s, legacyFormatter.parse)
@@ -300,7 +300,7 @@ class DefaultTimestampFormatter(
override def parseWithoutTimeZoneOptional(s: String, allowTimeZone: Boolean): Option[Long] = {
val utf8Value = UTF8String.fromString(s)
- DateTimeUtils.stringToTimestampWithoutTimeZone(utf8Value, allowTimeZone)
+ SparkDateTimeUtils.stringToTimestampWithoutTimeZone(utf8Value, allowTimeZone)
}
}
@@ -507,7 +507,7 @@ object TimestampFormatter {
legacyFormat: LegacyDateFormat = LENIENT_SIMPLE_DATE_FORMAT,
isParsing: Boolean,
forTimestampNTZ: Boolean = false): TimestampFormatter = {
- val formatter = if (SQLConf.get.legacyTimeParserPolicy == LEGACY && !forTimestampNTZ) {
+ val formatter = if (SqlApiConf.get.legacyTimeParserPolicy == LEGACY && !forTimestampNTZ) {
getLegacyFormatter(format.getOrElse(defaultPattern), zoneId, locale, legacyFormat)
} else {
format
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtils.scala
new file mode 100644
index 00000000000..98768a35e8a
--- /dev/null
+++ b/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtils.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.util
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.sql.types.UserDefinedType
+import org.apache.spark.util.SparkClassUtils
+
+/**
+ * Utilities for working with [[UserDefinedType]]s from within the `sql/api` project. UDTs are not
+ * not fully supported in `sql/api` (and connect), they can currently only be used in conjunction
+ * with catalyst because they (amongst others) require access to Spark SQLs internal data
+ * representation.
+ *
+ * This interface and its companion object provide an escape hatch for working with UDTs from within
+ * the api project (e.g. Row.toJSON). The companion will try to bind to an implementation of the
+ * interface in catalyst, if none is found it will bind to [[DefaultUDTUtils]].
+ */
+private[sql] trait UDTUtils {
+ /**
+ * Convert the UDT instance to something that is compatible with [[org.apache.spark.sql.Row]].
+ * The returned value must conform to the schema of the UDT.
+ */
+ def toRow(value: Any, udt: UserDefinedType[Any]): Any
+}
+
+private[sql] object UDTUtils extends UDTUtils {
+ private val delegate = try {
+ val cls = SparkClassUtils.classForName("org.apache.spark.sql.catalyst.util.UDTUtilsImpl")
+ cls.getConstructor().newInstance().asInstanceOf[UDTUtils]
+ } catch {
+ case NonFatal(_) =>
+ DefaultUDTUtils
+ }
+
+ override def toRow(value: Any, udt: UserDefinedType[Any]): Any = delegate.toRow(value, udt)
+}
+
+private[sql] object DefaultUDTUtils extends UDTUtils {
+ override def toRow(value: Any, udt: UserDefinedType[Any]): Any = {
+ throw new UnsupportedOperationException()
+ }
+}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
index 1dd3cf3dd5d..fcc3086b573 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala
@@ -295,10 +295,4 @@ private[sql] object DataTypeErrors extends DataTypeErrorsBase {
messageParameters = Map("operation" -> operation),
cause = null)
}
-
- def unsupportedDataTypeError(typeName: DataType): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "UNSUPPORTED_DATATYPE",
- messageParameters = Map("typeName" -> toSQLType(typeName)))
- }
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrorsBase.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrorsBase.scala
index 0f4cb88dcf1..95962f192f2 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrorsBase.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrorsBase.scala
@@ -18,10 +18,11 @@ package org.apache.spark.sql.errors
import java.util.Locale
-import org.apache.spark.{QueryContext, SparkRuntimeException}
+import org.apache.spark.QueryContext
import org.apache.spark.sql.catalyst.trees.SQLQueryContext
import org.apache.spark.sql.catalyst.util.{AttributeNameParser, QuotingUtils, SparkStringUtils}
import org.apache.spark.sql.types.{AbstractDataType, DataType, TypeCollection}
+import org.apache.spark.unsafe.types.UTF8String
private[sql] trait DataTypeErrorsBase {
def toSQLId(parts: String): String = {
@@ -54,7 +55,7 @@ private[sql] trait DataTypeErrorsBase {
case at => quoteByDefault(at.simpleString.toUpperCase(Locale.ROOT))
}
- def dataTypeToSQLValue(value: String): String = {
+ def toSQLValue(value: String): String = {
if (value == null) {
"NULL"
} else {
@@ -62,6 +63,28 @@ private[sql] trait DataTypeErrorsBase {
}
}
+ def toSQLValue(value: UTF8String): String = toSQLValue(value.toString)
+
+ def toSQLValue(value: Short): String = String.valueOf(value) + "S"
+
+ def toSQLValue(value: Int): String = String.valueOf(value)
+
+ def toSQLValue(value: Long): String = String.valueOf(value) + "L"
+
+ def toSQLValue(value: Float): String = {
+ if (value.isNaN) "NaN"
+ else if (value.isPosInfinity) "Infinity"
+ else if (value.isNegInfinity) "-Infinity"
+ else value.toString
+ }
+
+ def toSQLValue(value: Double): String = {
+ if (value.isNaN) "NaN"
+ else if (value.isPosInfinity) "Infinity"
+ else if (value.isNegInfinity) "-Infinity"
+ else value.toString
+ }
+
protected def quoteByDefault(elem: String): String = {
"\"" + elem + "\""
}
@@ -73,10 +96,4 @@ private[sql] trait DataTypeErrorsBase {
def getQueryContext(sqlContext: SQLQueryContext): Array[QueryContext] = {
if (sqlContext == null) Array.empty else Array(sqlContext.asInstanceOf[QueryContext])
}
-
- def unreachableError(err: String = ""): SparkRuntimeException = {
- new SparkRuntimeException(
- errorClass = "_LEGACY_ERROR_TEMP_2028",
- messageParameters = Map("err" -> err))
- }
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/EncoderErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/EncoderErrors.scala
deleted file mode 100644
index e70a5124ce7..00000000000
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/EncoderErrors.scala
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.errors
-
-import org.apache.spark.{SparkBuildInfo, SparkException, SparkRuntimeException, SparkUnsupportedOperationException}
-import org.apache.spark.sql.catalyst.WalkedTypePath
-import org.apache.spark.sql.types.UserDefinedType
-
-object EncoderErrors extends DataTypeErrorsBase {
- def userDefinedTypeNotAnnotatedAndRegisteredError(udt: UserDefinedType[_]): Throwable = {
- new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_2155",
- messageParameters = Map(
- "userClass" -> udt.userClass.getName),
- cause = null)
- }
-
- def cannotFindEncoderForTypeError(typeName: String): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "ENCODER_NOT_FOUND",
- messageParameters = Map(
- "typeName" -> typeName,
- "docroot" -> SparkBuildInfo.spark_doc_root))
- }
-
- def cannotHaveCircularReferencesInBeanClassError(
- clazz: Class[_]): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "_LEGACY_ERROR_TEMP_2138",
- messageParameters = Map("clazz" -> clazz.toString()))
- }
-
- def cannotFindConstructorForTypeError(tpe: String): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "_LEGACY_ERROR_TEMP_2144",
- messageParameters = Map(
- "tpe" -> tpe))
- }
-
- def cannotHaveCircularReferencesInClassError(t: String): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "_LEGACY_ERROR_TEMP_2139",
- messageParameters = Map("t" -> t))
- }
-
- def cannotUseInvalidJavaIdentifierAsFieldNameError(
- fieldName: String, walkedTypePath: WalkedTypePath): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "_LEGACY_ERROR_TEMP_2140",
- messageParameters = Map(
- "fieldName" -> fieldName,
- "walkedTypePath" -> walkedTypePath.toString()))
- }
-
- def primaryConstructorNotFoundError(cls: Class[_]): SparkRuntimeException = {
- new SparkRuntimeException(
- errorClass = "_LEGACY_ERROR_TEMP_2021",
- messageParameters = Map("cls" -> cls.toString()))
- }
-}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/ExecutionErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/ExecutionErrors.scala
new file mode 100644
index 00000000000..1e8e0ef5f6a
--- /dev/null
+++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/ExecutionErrors.scala
@@ -0,0 +1,211 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.errors
+
+import java.time.LocalDate
+import java.time.temporal.ChronoField
+
+import org.apache.arrow.vector.types.pojo.ArrowType
+
+import org.apache.spark.{SparkArithmeticException, SparkBuildInfo, SparkDateTimeException, SparkException, SparkRuntimeException, SparkUnsupportedOperationException, SparkUpgradeException}
+import org.apache.spark.sql.catalyst.WalkedTypePath
+import org.apache.spark.sql.catalyst.trees.SQLQueryContext
+import org.apache.spark.sql.internal.SqlApiConf
+import org.apache.spark.sql.types.{DataType, DoubleType, StringType, UserDefinedType}
+import org.apache.spark.unsafe.types.UTF8String
+
+private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
+ def fieldDiffersFromDerivedLocalDateError(
+ field: ChronoField,
+ actual: Int,
+ expected: Int,
+ candidate: LocalDate): SparkDateTimeException = {
+ new SparkDateTimeException(
+ errorClass = "_LEGACY_ERROR_TEMP_2129",
+ messageParameters = Map(
+ "field" -> field.toString,
+ "actual" -> actual.toString,
+ "expected" -> expected.toString,
+ "candidate" -> candidate.toString),
+ context = Array.empty,
+ summary = "")
+ }
+
+ def failToParseDateTimeInNewParserError(s: String, e: Throwable): SparkUpgradeException = {
+ new SparkUpgradeException(
+ errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER",
+ messageParameters = Map(
+ "datetime" -> toSQLValue(s),
+ "config" -> toSQLConf(SqlApiConf.LEGACY_TIME_PARSER_POLICY_KEY)),
+ e)
+ }
+
+ def failToRecognizePatternAfterUpgradeError(
+ pattern: String, e: Throwable): SparkUpgradeException = {
+ new SparkUpgradeException(
+ errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.DATETIME_PATTERN_RECOGNITION",
+ messageParameters = Map(
+ "pattern" -> toSQLValue(pattern),
+ "config" -> toSQLConf(SqlApiConf.LEGACY_TIME_PARSER_POLICY_KEY),
+ "docroot" -> SparkBuildInfo.spark_doc_root),
+ e)
+ }
+
+ def failToRecognizePatternError(pattern: String, e: Throwable): SparkRuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "_LEGACY_ERROR_TEMP_2130",
+ messageParameters = Map(
+ "pattern" -> toSQLValue(pattern),
+ "docroot" -> SparkBuildInfo.spark_doc_root),
+ cause = e)
+ }
+
+ def unreachableError(err: String = ""): SparkRuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "_LEGACY_ERROR_TEMP_2028",
+ messageParameters = Map("err" -> err))
+ }
+
+ def invalidInputInCastToDatetimeError(
+ value: UTF8String,
+ to: DataType,
+ context: SQLQueryContext): SparkDateTimeException = {
+ invalidInputInCastToDatetimeErrorInternal(toSQLValue(value), StringType, to, context)
+ }
+
+ def invalidInputInCastToDatetimeError(
+ value: Double,
+ to: DataType,
+ context: SQLQueryContext): SparkDateTimeException = {
+ invalidInputInCastToDatetimeErrorInternal(toSQLValue(value), DoubleType, to, context)
+ }
+
+ protected def invalidInputInCastToDatetimeErrorInternal(
+ sqlValue: String,
+ from: DataType,
+ to: DataType,
+ context: SQLQueryContext): SparkDateTimeException = {
+ new SparkDateTimeException(
+ errorClass = "CAST_INVALID_INPUT",
+ messageParameters = Map(
+ "expression" -> sqlValue,
+ "sourceType" -> toSQLType(from),
+ "targetType" -> toSQLType(to),
+ "ansiConfig" -> toSQLConf(SqlApiConf.ANSI_ENABLED_KEY)),
+ context = getQueryContext(context),
+ summary = getSummary(context))
+ }
+
+ def arithmeticOverflowError(
+ message: String,
+ hint: String = "",
+ context: SQLQueryContext = null): ArithmeticException = {
+ val alternative = if (hint.nonEmpty) {
+ s" Use '$hint' to tolerate overflow and return NULL instead."
+ } else ""
+ new SparkArithmeticException(
+ errorClass = "ARITHMETIC_OVERFLOW",
+ messageParameters = Map(
+ "message" -> message,
+ "alternative" -> alternative,
+ "config" -> toSQLConf(SqlApiConf.ANSI_ENABLED_KEY)),
+ context = getQueryContext(context),
+ summary = getSummary(context))
+ }
+
+ def cannotParseStringAsDataTypeError(pattern: String, value: String, dataType: DataType)
+ : SparkRuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "_LEGACY_ERROR_TEMP_2134",
+ messageParameters = Map(
+ "value" -> toSQLValue(value),
+ "pattern" -> toSQLValue(pattern),
+ "dataType" -> dataType.toString))
+ }
+
+ def unsupportedArrowTypeError(typeName: ArrowType): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "UNSUPPORTED_ARROWTYPE",
+ messageParameters = Map("typeName" -> typeName.toString))
+ }
+
+ def duplicatedFieldNameInArrowStructError(
+ fieldNames: Seq[String]): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT",
+ messageParameters = Map("fieldNames" -> fieldNames.mkString("[", ", ", "]")))
+ }
+
+ def unsupportedDataTypeError(typeName: DataType): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "UNSUPPORTED_DATATYPE",
+ messageParameters = Map("typeName" -> toSQLType(typeName)))
+ }
+
+ def userDefinedTypeNotAnnotatedAndRegisteredError(udt: UserDefinedType[_]): Throwable = {
+ new SparkException(
+ errorClass = "_LEGACY_ERROR_TEMP_2155",
+ messageParameters = Map(
+ "userClass" -> udt.userClass.getName),
+ cause = null)
+ }
+
+ def cannotFindEncoderForTypeError(typeName: String): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "ENCODER_NOT_FOUND",
+ messageParameters = Map(
+ "typeName" -> typeName,
+ "docroot" -> SparkBuildInfo.spark_doc_root))
+ }
+
+ def cannotHaveCircularReferencesInBeanClassError(
+ clazz: Class[_]): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "_LEGACY_ERROR_TEMP_2138",
+ messageParameters = Map("clazz" -> clazz.toString))
+ }
+
+ def cannotFindConstructorForTypeError(tpe: String): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "_LEGACY_ERROR_TEMP_2144",
+ messageParameters = Map(
+ "tpe" -> tpe))
+ }
+
+ def cannotHaveCircularReferencesInClassError(t: String): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "_LEGACY_ERROR_TEMP_2139",
+ messageParameters = Map("t" -> t))
+ }
+
+ def cannotUseInvalidJavaIdentifierAsFieldNameError(
+ fieldName: String, walkedTypePath: WalkedTypePath): SparkUnsupportedOperationException = {
+ new SparkUnsupportedOperationException(
+ errorClass = "_LEGACY_ERROR_TEMP_2140",
+ messageParameters = Map(
+ "fieldName" -> fieldName,
+ "walkedTypePath" -> walkedTypePath.toString))
+ }
+
+ def primaryConstructorNotFoundError(cls: Class[_]): SparkRuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "_LEGACY_ERROR_TEMP_2021",
+ messageParameters = Map("cls" -> cls.toString))
+ }
+}
+
+private[sql] object ExecutionErrors extends ExecutionErrors
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
index 2bd385787dd..f63fc8c4785 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
@@ -184,7 +184,7 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
def invalidEscapeStringError(invalidEscape: String, ctx: PredicateContext): Throwable = {
new ParseException(
errorClass = "INVALID_ESC",
- messageParameters = Map("invalidEscape" -> dataTypeToSQLValue(invalidEscape)),
+ messageParameters = Map("invalidEscape" -> toSQLValue(invalidEscape)),
ctx)
}
@@ -208,7 +208,7 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
errorClass = "INVALID_TYPED_LITERAL",
messageParameters = Map(
"valueType" -> toSQLType(valueType),
- "value" -> dataTypeToSQLValue(value)
+ "value" -> toSQLValue(value)
),
ctx)
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/internal/LegacyBehaviorPolicy.scala
similarity index 51%
copy from sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala
copy to sql/api/src/main/scala/org/apache/spark/sql/internal/LegacyBehaviorPolicy.scala
index 59e35c9c9ba..03ff85f2b97 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/LegacyBehaviorPolicy.scala
@@ -14,26 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.errors
+package org.apache.spark.sql.internal
-import org.apache.arrow.vector.types.pojo.ArrowType
-
-import org.apache.spark.SparkUnsupportedOperationException
-
-trait ArrowErrors {
-
- def unsupportedArrowTypeError(typeName: ArrowType): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "UNSUPPORTED_ARROWTYPE",
- messageParameters = Map("typeName" -> typeName.toString))
- }
-
- def duplicatedFieldNameInArrowStructError(
- fieldNames: Seq[String]): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT",
- messageParameters = Map("fieldNames" -> fieldNames.mkString("[", ", ", "]")))
- }
+object LegacyBehaviorPolicy extends Enumeration {
+ val EXCEPTION, LEGACY, CORRECTED = Value
}
-
-object ArrowErrors extends ArrowErrors
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
similarity index 86%
rename from sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala
rename to sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
index 66398495d90..bdc4a5759a1 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala
@@ -14,9 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql
-
+package org.apache.spark.sql.internal
+import java.util.TimeZone
import java.util.concurrent.atomic.AtomicReference
import scala.util.Try
@@ -42,9 +42,15 @@ private[sql] trait SqlApiConf {
def allowNegativeScaleOfDecimalEnabled: Boolean
def charVarcharAsString: Boolean
def datetimeJava8ApiEnabled: Boolean
+ def sessionLocalTimeZone: String
+ def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value
}
private[sql] object SqlApiConf {
+ // Shared keys.
+ val ANSI_ENABLED_KEY: String = "spark.sql.ansi.enabled"
+ val LEGACY_TIME_PARSER_POLICY_KEY: String = "spark.sql.legacy.timeParserPolicy"
+
/**
* Defines a getter that returns the [[SqlApiConf]] within scope.
*/
@@ -78,4 +84,6 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
override def allowNegativeScaleOfDecimalEnabled: Boolean = false
override def charVarcharAsString: Boolean = false
override def datetimeJava8ApiEnabled: Boolean = false
+ override def sessionLocalTimeZone: String = TimeZone.getDefault.getID
+ override def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value = LegacyBehaviorPolicy.EXCEPTION
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
index 632091a9a8c..2bd88d59756 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
@@ -29,12 +29,12 @@ import org.json4s.jackson.JsonMethods._
import org.apache.spark.SparkThrowable
import org.apache.spark.annotation.Stable
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.analysis.SqlApiAnalysis
import org.apache.spark.sql.catalyst.parser.DataTypeParser
import org.apache.spark.sql.catalyst.util.DataTypeJsonUtils.{DataTypeJsonDeserializer, DataTypeJsonSerializer}
import org.apache.spark.sql.catalyst.util.StringConcat
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.sql.types.DayTimeIntervalType._
import org.apache.spark.sql.types.YearMonthIntervalType._
import org.apache.spark.util.SparkClassUtils
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/Decimal.scala
index 40186dab822..afe73635a68 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/Decimal.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/Decimal.scala
@@ -22,9 +22,9 @@ import java.math.{BigDecimal => JavaBigDecimal, BigInteger, MathContext, Roundin
import scala.util.Try
import org.apache.spark.annotation.Unstable
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.trees.SQLQueryContext
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
import org.apache.spark.unsafe.types.UTF8String
/**
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/DecimalType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/DecimalType.scala
index bc06888e4cf..af545341dee 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/DecimalType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/DecimalType.scala
@@ -22,8 +22,8 @@ import java.util.Locale
import scala.annotation.tailrec
import org.apache.spark.annotation.Stable
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
/**
* The data type representing `java.math.BigDecimal` values.
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala
index 9eabc6a70bc..8edc7cf370b 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala
@@ -24,12 +24,12 @@ import scala.util.control.NonFatal
import org.json4s.JsonDSL._
import org.apache.spark.annotation.Stable
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.analysis.SqlApiAnalysis
import org.apache.spark.sql.catalyst.parser.{DataTypeParser, LegacyTypeStringParser}
import org.apache.spark.sql.catalyst.trees.Origin
import org.apache.spark.sql.catalyst.util.{SparkCollectionUtils, SparkStringUtils, StringConcat}
import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.internal.SqlApiConf
/**
* A [[StructType]] object can be constructed by
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
index 49990f7f033..506cb8c564c 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala
@@ -26,7 +26,7 @@ import org.apache.arrow.vector.complex.MapVector
import org.apache.arrow.vector.types.{DateUnit, FloatingPointPrecision, IntervalUnit, TimeUnit}
import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType, Schema}
-import org.apache.spark.sql.errors.{ArrowErrors, DataTypeErrors}
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.types._
private[sql] object ArrowUtils {
@@ -60,7 +60,7 @@ private[sql] object ArrowUtils {
case _: YearMonthIntervalType => new ArrowType.Interval(IntervalUnit.YEAR_MONTH)
case _: DayTimeIntervalType => new ArrowType.Duration(TimeUnit.MICROSECOND)
case _ =>
- throw DataTypeErrors.unsupportedDataTypeError(dt)
+ throw ExecutionErrors.unsupportedDataTypeError(dt)
}
def fromArrowType(dt: ArrowType): DataType = dt match {
@@ -85,7 +85,7 @@ private[sql] object ArrowUtils {
case ArrowType.Null.INSTANCE => NullType
case yi: ArrowType.Interval if yi.getUnit == IntervalUnit.YEAR_MONTH => YearMonthIntervalType()
case di: ArrowType.Duration if di.getUnit == TimeUnit.MICROSECOND => DayTimeIntervalType()
- case _ => throw ArrowErrors.unsupportedArrowTypeError(dt)
+ case _ => throw ExecutionErrors.unsupportedArrowTypeError(dt)
}
/** Maps field from Spark to Arrow. NOTE: timeZoneId required for TimestampType */
@@ -179,7 +179,7 @@ private[sql] object ArrowUtils {
st.names
} else {
if (errorOnDuplicatedFieldNames) {
- throw ArrowErrors.duplicatedFieldNameInArrowStructError(st.names)
+ throw ExecutionErrors.duplicatedFieldNameInArrowStructError(st.names)
}
val genNawName = st.names.groupBy(identity).map {
case (name, names) if names.length > 1 =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala
index 4e3234f9c0d..954f5f19cd3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala
@@ -25,7 +25,8 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.util.TypeUtils._
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.internal.SQLConf.{LEGACY_CTE_PRECEDENCE_POLICY, LegacyBehaviorPolicy}
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.SQLConf.LEGACY_CTE_PRECEDENCE_POLICY
/**
* Analyze WITH nodes and substitute child plan with CTE references or CTE definitions depending
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
index f4ad1f2f2e5..845c815c564 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
@@ -27,8 +27,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
class CSVOptions(
@transient val parameters: CaseInsensitiveMap[String],
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
index a02d57c0bc7..b99ee630d4b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -126,14 +126,14 @@ class UnivocityParser(
options.enableDateTimeParsingFallback
.orElse(SQLConf.get.csvEnableDateTimeParsingFallback)
.getOrElse {
- SQLConf.get.legacyTimeParserPolicy == SQLConf.LegacyBehaviorPolicy.LEGACY ||
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
options.timestampFormatInRead.isEmpty
}
private val enableParsingFallbackForDateType =
options.enableDateTimeParsingFallback
.orElse(SQLConf.get.csvEnableDateTimeParsingFallback)
.getOrElse {
- SQLConf.get.legacyTimeParserPolicy == SQLConf.LegacyBehaviorPolicy.LEGACY ||
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
options.dateFormatOption.isEmpty
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
index c06f411c505..596d9e39b94 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JSONOptions.scala
@@ -27,8 +27,7 @@ import com.fasterxml.jackson.core.json.JsonReadFeature
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
import org.apache.spark.sql.catalyst.util._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
/**
* Options for parsing JSON data into Spark SQL rows.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index 03dce431837..388edb9024c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
@@ -85,14 +85,14 @@ class JacksonParser(
options.enableDateTimeParsingFallback
.orElse(SQLConf.get.jsonEnableDateTimeParsingFallback)
.getOrElse {
- SQLConf.get.legacyTimeParserPolicy == SQLConf.LegacyBehaviorPolicy.LEGACY ||
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
options.timestampFormatInRead.isEmpty
}
private val enableParsingFallbackForDateType =
options.enableDateTimeParsingFallback
.orElse(SQLConf.get.jsonEnableDateTimeParsingFallback)
.getOrElse {
- SQLConf.get.legacyTimeParserPolicy == SQLConf.LegacyBehaviorPolicy.LEGACY ||
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
options.dateFormatInRead.isEmpty
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
index 1142b6184dd..23bbc91c16d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
@@ -17,21 +17,17 @@
package org.apache.spark.sql.catalyst.util
-import java.sql.{Date, Timestamp}
import java.time._
import java.time.temporal.{ChronoField, ChronoUnit, IsoFields, Temporal}
-import java.util.{Locale, TimeZone}
+import java.util.Locale
import java.util.concurrent.TimeUnit._
import scala.util.control.NonFatal
-import sun.util.calendar.ZoneInfo
-
import org.apache.spark.sql.catalyst.trees.SQLQueryContext
import org.apache.spark.sql.catalyst.util.DateTimeConstants._
-import org.apache.spark.sql.catalyst.util.RebaseDateTime._
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.types.{DateType, Decimal, DoubleExactNumeric, DoubleType, StringType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{Decimal, DoubleExactNumeric, TimestampNTZType, TimestampType}
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
/**
@@ -41,185 +37,14 @@ import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
* and are stored internally as longs, which are capable of storing timestamps with microsecond
* precision.
*/
-object DateTimeUtils {
+object DateTimeUtils extends SparkDateTimeUtils {
// See http://stackoverflow.com/questions/466321/convert-unix-timestamp-to-julian
// It's 2440587.5, rounding up to be compatible with Hive.
final val JULIAN_DAY_OF_EPOCH = 2440588
- final val TimeZoneUTC = TimeZone.getTimeZone("UTC")
-
val TIMEZONE_OPTION = "timeZone"
- def getZoneId(timeZoneId: String): ZoneId = {
- val formattedZoneId = timeZoneId
- // To support the (+|-)h:mm format because it was supported before Spark 3.0.
- .replaceFirst("(\\+|\\-)(\\d):", "$10$2:")
- // To support the (+|-)hh:m format because it was supported before Spark 3.0.
- .replaceFirst("(\\+|\\-)(\\d\\d):(\\d)$", "$1$2:0$3")
-
- ZoneId.of(formattedZoneId, ZoneId.SHORT_IDS)
- }
- def getTimeZone(timeZoneId: String): TimeZone = TimeZone.getTimeZone(getZoneId(timeZoneId))
-
- /**
- * Converts microseconds since 1970-01-01 00:00:00Z to days since 1970-01-01 at the given zone ID.
- */
- def microsToDays(micros: Long, zoneId: ZoneId): Int = {
- localDateToDays(getLocalDateTime(micros, zoneId).toLocalDate)
- }
-
- /**
- * Converts days since 1970-01-01 at the given zone ID to microseconds since 1970-01-01 00:00:00Z.
- */
- def daysToMicros(days: Int, zoneId: ZoneId): Long = {
- val instant = daysToLocalDate(days).atStartOfDay(zoneId).toInstant
- instantToMicros(instant)
- }
-
- def microsToLocalDateTime(micros: Long): LocalDateTime = {
- getLocalDateTime(micros, ZoneOffset.UTC)
- }
-
- def localDateTimeToMicros(localDateTime: LocalDateTime): Long = {
- instantToMicros(localDateTime.toInstant(ZoneOffset.UTC))
- }
-
- /**
- * Converts a local date at the default JVM time zone to the number of days since 1970-01-01
- * in the hybrid calendar (Julian + Gregorian) by discarding the time part. The resulted days are
- * rebased from the hybrid to Proleptic Gregorian calendar. The days rebasing is performed via
- * UTC time zone for simplicity because the difference between two calendars is the same in
- * any given time zone and UTC time zone.
- *
- * Note: The date is shifted by the offset of the default JVM time zone for backward compatibility
- * with Spark 2.4 and earlier versions. The goal of the shift is to get a local date derived
- * from the number of days that has the same date fields (year, month, day) as the original
- * `date` at the default JVM time zone.
- *
- * @param date It represents a specific instant in time based on the hybrid calendar which
- * combines Julian and Gregorian calendars.
- * @return The number of days since the epoch in Proleptic Gregorian calendar.
- */
- def fromJavaDate(date: Date): Int = {
- val millisUtc = date.getTime
- val millisLocal = millisUtc + TimeZone.getDefault.getOffset(millisUtc)
- val julianDays = Math.toIntExact(Math.floorDiv(millisLocal, MILLIS_PER_DAY))
- rebaseJulianToGregorianDays(julianDays)
- }
-
- /**
- * Converts an Java object to days.
- *
- * @param obj Either an object of `java.sql.Date` or `java.time.LocalDate`.
- * @return The number of days since 1970-01-01.
- */
- def anyToDays(obj: Any): Int = obj match {
- case d: Date => fromJavaDate(d)
- case ld: LocalDate => localDateToDays(ld)
- }
-
- /**
- * Converts days since the epoch 1970-01-01 in Proleptic Gregorian calendar to a local date
- * at the default JVM time zone in the hybrid calendar (Julian + Gregorian). It rebases the given
- * days from Proleptic Gregorian to the hybrid calendar at UTC time zone for simplicity because
- * the difference between two calendars doesn't depend on any time zone. The result is shifted
- * by the time zone offset in wall clock to have the same date fields (year, month, day)
- * at the default JVM time zone as the input `daysSinceEpoch` in Proleptic Gregorian calendar.
- *
- * Note: The date is shifted by the offset of the default JVM time zone for backward compatibility
- * with Spark 2.4 and earlier versions.
- *
- * @param days The number of days since 1970-01-01 in Proleptic Gregorian calendar.
- * @return A local date in the hybrid calendar as `java.sql.Date` from number of days since epoch.
- */
- def toJavaDate(days: Int): Date = {
- val rebasedDays = rebaseGregorianToJulianDays(days)
- val localMillis = Math.multiplyExact(rebasedDays, MILLIS_PER_DAY)
- val timeZoneOffset = TimeZone.getDefault match {
- case zoneInfo: ZoneInfo => zoneInfo.getOffsetsByWall(localMillis, null)
- case timeZone: TimeZone => timeZone.getOffset(localMillis - timeZone.getRawOffset)
- }
- new Date(localMillis - timeZoneOffset)
- }
-
- /**
- * Converts microseconds since the epoch to an instance of `java.sql.Timestamp`
- * via creating a local timestamp at the system time zone in Proleptic Gregorian
- * calendar, extracting date and time fields like `year` and `hours`, and forming
- * new timestamp in the hybrid calendar from the extracted fields.
- *
- * The conversion is based on the JVM system time zone because the `java.sql.Timestamp`
- * uses the time zone internally.
- *
- * The method performs the conversion via local timestamp fields to have the same date-time
- * representation as `year`, `month`, `day`, ..., `seconds` in the original calendar
- * and in the target calendar.
- *
- * @param micros The number of microseconds since 1970-01-01T00:00:00.000000Z.
- * @return A `java.sql.Timestamp` from number of micros since epoch.
- */
- def toJavaTimestamp(micros: Long): Timestamp =
- toJavaTimestampNoRebase(rebaseGregorianToJulianMicros(micros))
-
- /**
- * Converts microseconds since the epoch to an instance of `java.sql.Timestamp`.
- *
- * @param micros The number of microseconds since 1970-01-01T00:00:00.000000Z.
- * @return A `java.sql.Timestamp` from number of micros since epoch.
- */
- def toJavaTimestampNoRebase(micros: Long): Timestamp = {
- val seconds = Math.floorDiv(micros, MICROS_PER_SECOND)
- val ts = new Timestamp(seconds * MILLIS_PER_SECOND)
- val nanos = (micros - seconds * MICROS_PER_SECOND) * NANOS_PER_MICROS
- ts.setNanos(nanos.toInt)
- ts
- }
-
- /**
- * Converts an instance of `java.sql.Timestamp` to the number of microseconds since
- * 1970-01-01T00:00:00.000000Z. It extracts date-time fields from the input, builds
- * a local timestamp in Proleptic Gregorian calendar from the fields, and binds
- * the timestamp to the system time zone. The resulted instant is converted to
- * microseconds since the epoch.
- *
- * The conversion is performed via the system time zone because it is used internally
- * in `java.sql.Timestamp` while extracting date-time fields.
- *
- * The goal of the function is to have the same local date-time in the original calendar
- * - the hybrid calendar (Julian + Gregorian) and in the target calendar which is
- * Proleptic Gregorian calendar, see SPARK-26651.
- *
- * @param t It represents a specific instant in time based on
- * the hybrid calendar which combines Julian and
- * Gregorian calendars.
- * @return The number of micros since epoch from `java.sql.Timestamp`.
- */
- def fromJavaTimestamp(t: Timestamp): Long =
- rebaseJulianToGregorianMicros(fromJavaTimestampNoRebase(t))
-
- /**
- * Converts an instance of `java.sql.Timestamp` to the number of microseconds since
- * 1970-01-01T00:00:00.000000Z.
- *
- * @param t an instance of `java.sql.Timestamp`.
- * @return The number of micros since epoch from `java.sql.Timestamp`.
- */
- def fromJavaTimestampNoRebase(t: Timestamp): Long =
- millisToMicros(t.getTime) + (t.getNanos / NANOS_PER_MICROS) % MICROS_PER_MILLIS
-
- /**
- * Converts an Java object to microseconds.
- *
- * @param obj Either an object of `java.sql.Timestamp` or `java.time.{Instant,LocalDateTime}`.
- * @return The number of micros since the epoch.
- */
- def anyToMicros(obj: Any): Long = obj match {
- case t: Timestamp => fromJavaTimestamp(t)
- case i: Instant => instantToMicros(i)
- case ldt: LocalDateTime => localDateTimeToMicros(ldt)
- }
-
/**
* Returns the number of microseconds since epoch from Julian day and nanoseconds in a day.
*/
@@ -240,281 +65,19 @@ object DateTimeUtils {
(days.toInt, MICROSECONDS.toNanos(us))
}
- /**
- * Converts the timestamp to milliseconds since epoch. In Spark timestamp values have microseconds
- * precision, so this conversion is lossy.
- */
- def microsToMillis(micros: Long): Long = {
- // When the timestamp is negative i.e before 1970, we need to adjust the milliseconds portion.
- // Example - 1965-01-01 10:11:12.123456 is represented as (-157700927876544) in micro precision.
- // In millis precision the above needs to be represented as (-157700927877).
- Math.floorDiv(micros, MICROS_PER_MILLIS)
- }
-
- /**
- * Converts milliseconds since the epoch to microseconds.
- */
- def millisToMicros(millis: Long): Long = {
- SparkDateTimeUtils.millisToMicros(millis)
- }
-
private final val gmtUtf8 = UTF8String.fromString("GMT")
// The method is called by JSON/CSV parser to clean up the legacy timestamp string by removing
// the "GMT" string. For example, it returns 2000-01-01T00:00+01:00 for 2000-01-01T00:00GMT+01:00.
def cleanLegacyTimestampStr(s: UTF8String): UTF8String = s.replace(gmtUtf8, UTF8String.EMPTY_UTF8)
- /**
- * Trims and parses a given UTF8 timestamp string to the corresponding timestamp segments,
- * time zone id and whether it is just time without a date.
- * value. The return type is [[Option]] in order to distinguish between 0L and null. The following
- * formats are allowed:
- *
- * `[+-]yyyy*`
- * `[+-]yyyy*-[m]m`
- * `[+-]yyyy*-[m]m-[d]d`
- * `[+-]yyyy*-[m]m-[d]d `
- * `[+-]yyyy*-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
- * `[+-]yyyy*-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
- * `[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
- * `T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us][zone_id]`
- *
- * where `zone_id` should have one of the forms:
- * - Z - Zulu time zone UTC+0
- * - +|-[h]h:[m]m
- * - A short id, see https://docs.oracle.com/javase/8/docs/api/java/time/ZoneId.html#SHORT_IDS
- * - An id with one of the prefixes UTC+, UTC-, GMT+, GMT-, UT+ or UT-,
- * and a suffix in the formats:
- * - +|-h[h]
- * - +|-hh[:]mm
- * - +|-hh:mm:ss
- * - +|-hhmmss
- * - Region-based zone IDs in the form `area/city`, such as `Europe/Paris`
- *
- * @return timestamp segments, time zone id and whether the input is just time without a date. If
- * the input string can't be parsed as timestamp, the result timestamp segments are empty.
- */
- def parseTimestampString(s: UTF8String): (Array[Int], Option[ZoneId], Boolean) = {
- def isValidDigits(segment: Int, digits: Int): Boolean = {
- // A Long is able to represent a timestamp within [+-]200 thousand years
- val maxDigitsYear = 6
- // For the nanosecond part, more than 6 digits is allowed, but will be truncated.
- segment == 6 || (segment == 0 && digits >= 4 && digits <= maxDigitsYear) ||
- // For the zoneId segment(7), it's could be zero digits when it's a region-based zone ID
- (segment == 7 && digits <= 2) ||
- (segment != 0 && segment != 6 && segment != 7 && digits > 0 && digits <= 2)
- }
- if (s == null || s.trimAll().numBytes() == 0) {
- return (Array.empty, None, false)
- }
- var tz: Option[String] = None
- val segments: Array[Int] = Array[Int](1, 1, 1, 0, 0, 0, 0, 0, 0)
- var i = 0
- var currentSegmentValue = 0
- var currentSegmentDigits = 0
- val bytes = s.trimAll().getBytes
- var j = 0
- var digitsMilli = 0
- var justTime = false
- var yearSign: Option[Int] = None
- if (bytes(j) == '-' || bytes(j) == '+') {
- yearSign = if (bytes(j) == '-') Some(-1) else Some(1)
- j += 1
- }
- while (j < bytes.length) {
- val b = bytes(j)
- val parsedValue = b - '0'.toByte
- if (parsedValue < 0 || parsedValue > 9) {
- if (j == 0 && b == 'T') {
- justTime = true
- i += 3
- } else if (i < 2) {
- if (b == '-') {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else if (i == 0 && b == ':' && yearSign.isEmpty) {
- justTime = true
- if (!isValidDigits(3, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(3) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i = 4
- } else {
- return (Array.empty, None, false)
- }
- } else if (i == 2) {
- if (b == ' ' || b == 'T') {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else {
- return (Array.empty, None, false)
- }
- } else if (i == 3 || i == 4) {
- if (b == ':') {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else {
- return (Array.empty, None, false)
- }
- } else if (i == 5 || i == 6) {
- if (b == '.' && i == 5) {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- tz = Some(new String(bytes, j, bytes.length - j))
- j = bytes.length - 1
- }
- if (i == 6 && b != '.') {
- i += 1
- }
- } else {
- if (i < segments.length && (b == ':' || b == ' ')) {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else {
- return (Array.empty, None, false)
- }
- }
- } else {
- if (i == 6) {
- digitsMilli += 1
- }
- // We will truncate the nanosecond part if there are more than 6 digits, which results
- // in loss of precision
- if (i != 6 || currentSegmentDigits < 6) {
- currentSegmentValue = currentSegmentValue * 10 + parsedValue
- }
- currentSegmentDigits += 1
- }
- j += 1
- }
-
- if (!isValidDigits(i, currentSegmentDigits)) {
- return (Array.empty, None, false)
- }
- segments(i) = currentSegmentValue
-
- while (digitsMilli < 6) {
- segments(6) *= 10
- digitsMilli += 1
- }
-
- // This step also validates time zone part
- val zoneId = tz.map(zoneName => getZoneId(zoneName.trim))
- segments(0) *= yearSign.getOrElse(1)
- (segments, zoneId, justTime)
- }
-
- /**
- * Trims and parses a given UTF8 timestamp string to the corresponding a corresponding [[Long]]
- * value. The return type is [[Option]] in order to distinguish between 0L and null. Please
- * refer to `parseTimestampString` for the allowed formats
- */
- def stringToTimestamp(s: UTF8String, timeZoneId: ZoneId): Option[Long] = {
- try {
- val (segments, parsedZoneId, justTime) = parseTimestampString(s)
- if (segments.isEmpty) {
- return None
- }
- val zoneId = parsedZoneId.getOrElse(timeZoneId)
- val nanoseconds = MICROSECONDS.toNanos(segments(6))
- val localTime = LocalTime.of(segments(3), segments(4), segments(5), nanoseconds.toInt)
- val localDate = if (justTime) {
- LocalDate.now(zoneId)
- } else {
- LocalDate.of(segments(0), segments(1), segments(2))
- }
- val localDateTime = LocalDateTime.of(localDate, localTime)
- val zonedDateTime = ZonedDateTime.of(localDateTime, zoneId)
- val instant = Instant.from(zonedDateTime)
- Some(instantToMicros(instant))
- } catch {
- case NonFatal(_) => None
- }
- }
-
- def stringToTimestampAnsi(
- s: UTF8String,
- timeZoneId: ZoneId,
- context: SQLQueryContext = null): Long = {
- stringToTimestamp(s, timeZoneId).getOrElse {
- throw QueryExecutionErrors.invalidInputInCastToDatetimeError(
- s, StringType, TimestampType, context)
- }
- }
-
def doubleToTimestampAnsi(d: Double, context: SQLQueryContext): Long = {
if (d.isNaN || d.isInfinite) {
- throw QueryExecutionErrors.invalidInputInCastToDatetimeError(
- d, DoubleType, TimestampType, context)
+ throw QueryExecutionErrors.invalidInputInCastToDatetimeError(d, TimestampType, context)
} else {
DoubleExactNumeric.toLong(d * MICROS_PER_SECOND)
}
}
- /**
- * Trims and parses a given UTF8 string to a corresponding [[Long]] value which representing the
- * number of microseconds since the epoch. The result will be independent of time zones.
- *
- * If the input string contains a component associated with time zone, the method will return
- * `None` if `allowTimeZone` is set to `false`. If `allowTimeZone` is set to `true`, the method
- * will simply discard the time zone component. Enable the check to detect situations like parsing
- * a timestamp with time zone as TimestampNTZType.
- *
- * The return type is [[Option]] in order to distinguish between 0L and null. Please
- * refer to `parseTimestampString` for the allowed formats.
- */
- def stringToTimestampWithoutTimeZone(s: UTF8String, allowTimeZone: Boolean): Option[Long] = {
- try {
- val (segments, zoneIdOpt, justTime) = parseTimestampString(s)
- // If the input string can't be parsed as a timestamp without time zone, or it contains only
- // the time part of a timestamp and we can't determine its date, return None.
- if (segments.isEmpty || justTime || !allowTimeZone && zoneIdOpt.isDefined) {
- return None
- }
- val nanoseconds = MICROSECONDS.toNanos(segments(6))
- val localTime = LocalTime.of(segments(3), segments(4), segments(5), nanoseconds.toInt)
- val localDate = LocalDate.of(segments(0), segments(1), segments(2))
- val localDateTime = LocalDateTime.of(localDate, localTime)
- Some(localDateTimeToMicros(localDateTime))
- } catch {
- case NonFatal(_) => None
- }
- }
-
/**
* Trims and parses a given UTF8 string to a corresponding [[Long]] value which representing the
* number of microseconds since the epoch. The result is independent of time zones. Zone id
@@ -530,138 +93,10 @@ object DateTimeUtils {
s: UTF8String,
context: SQLQueryContext): Long = {
stringToTimestampWithoutTimeZone(s, true).getOrElse {
- throw QueryExecutionErrors.invalidInputInCastToDatetimeError(
- s, StringType, TimestampNTZType, context)
- }
- }
-
- // See issue SPARK-35679
- // min second cause overflow in instant to micro
- private val MIN_SECONDS = Math.floorDiv(Long.MinValue, MICROS_PER_SECOND)
-
- /**
- * Gets the number of microseconds since the epoch of 1970-01-01 00:00:00Z from the given
- * instance of `java.time.Instant`. The epoch microsecond count is a simple incrementing count of
- * microseconds where microsecond 0 is 1970-01-01 00:00:00Z.
- */
- def instantToMicros(instant: Instant): Long = {
- val secs = instant.getEpochSecond
- if (secs == MIN_SECONDS) {
- val us = Math.multiplyExact(secs + 1, MICROS_PER_SECOND)
- Math.addExact(us, NANOSECONDS.toMicros(instant.getNano) - MICROS_PER_SECOND)
- } else {
- val us = Math.multiplyExact(secs, MICROS_PER_SECOND)
- Math.addExact(us, NANOSECONDS.toMicros(instant.getNano))
+ throw QueryExecutionErrors.invalidInputInCastToDatetimeError(s, TimestampNTZType, context)
}
}
- /**
- * Obtains an instance of `java.time.Instant` using microseconds from
- * the epoch of 1970-01-01 00:00:00Z.
- */
- def microsToInstant(micros: Long): Instant = {
- val secs = Math.floorDiv(micros, MICROS_PER_SECOND)
- // Unfolded Math.floorMod(us, MICROS_PER_SECOND) to reuse the result of
- // the above calculation of `secs` via `floorDiv`.
- val mos = micros - secs * MICROS_PER_SECOND
- Instant.ofEpochSecond(secs, mos * NANOS_PER_MICROS)
- }
-
- /**
- * Converts the local date to the number of days since 1970-01-01.
- */
- def localDateToDays(localDate: LocalDate): Int = MathUtils.toIntExact(localDate.toEpochDay)
-
- /**
- * Obtains an instance of `java.time.LocalDate` from the epoch day count.
- */
- def daysToLocalDate(days: Int): LocalDate = LocalDate.ofEpochDay(days)
-
- /**
- * Trims and parses a given UTF8 date string to a corresponding [[Int]] value.
- * The return type is [[Option]] in order to distinguish between 0 and null. The following
- * formats are allowed:
- *
- * `[+-]yyyy*`
- * `[+-]yyyy*-[m]m`
- * `[+-]yyyy*-[m]m-[d]d`
- * `[+-]yyyy*-[m]m-[d]d `
- * `[+-]yyyy*-[m]m-[d]d *`
- * `[+-]yyyy*-[m]m-[d]dT*`
- */
- def stringToDate(s: UTF8String): Option[Int] = {
- def isValidDigits(segment: Int, digits: Int): Boolean = {
- // An integer is able to represent a date within [+-]5 million years.
- val maxDigitsYear = 7
- (segment == 0 && digits >= 4 && digits <= maxDigitsYear) ||
- (segment != 0 && digits > 0 && digits <= 2)
- }
- if (s == null || s.trimAll().numBytes() == 0) {
- return None
- }
- val segments: Array[Int] = Array[Int](1, 1, 1)
- var sign = 1
- var i = 0
- var currentSegmentValue = 0
- var currentSegmentDigits = 0
- val bytes = s.trimAll().getBytes
- var j = 0
- if (bytes(j) == '-' || bytes(j) == '+') {
- sign = if (bytes(j) == '-') -1 else 1
- j += 1
- }
- while (j < bytes.length && (i < 3 && !(bytes(j) == ' ' || bytes(j) == 'T'))) {
- val b = bytes(j)
- if (i < 2 && b == '-') {
- if (!isValidDigits(i, currentSegmentDigits)) {
- return None
- }
- segments(i) = currentSegmentValue
- currentSegmentValue = 0
- currentSegmentDigits = 0
- i += 1
- } else {
- val parsedValue = b - '0'.toByte
- if (parsedValue < 0 || parsedValue > 9) {
- return None
- } else {
- currentSegmentValue = currentSegmentValue * 10 + parsedValue
- currentSegmentDigits += 1
- }
- }
- j += 1
- }
- if (!isValidDigits(i, currentSegmentDigits)) {
- return None
- }
- if (i < 2 && j < bytes.length) {
- // For the `yyyy` and `yyyy-[m]m` formats, entire input must be consumed.
- return None
- }
- segments(i) = currentSegmentValue
- try {
- val localDate = LocalDate.of(sign * segments(0), segments(1), segments(2))
- Some(localDateToDays(localDate))
- } catch {
- case NonFatal(_) => None
- }
- }
-
- def stringToDateAnsi(
- s: UTF8String,
- context: SQLQueryContext = null): Int = {
- stringToDate(s).getOrElse {
- throw QueryExecutionErrors.invalidInputInCastToDatetimeError(
- s, StringType, DateType, context)
- }
- }
-
- // Gets the local date-time parts (year, month, day and time) of the instant expressed as the
- // number of microseconds since the epoch at the given time zone ID.
- private def getLocalDateTime(micros: Long, zoneId: ZoneId): LocalDateTime = {
- microsToInstant(micros).atZone(zoneId).toLocalDateTime
- }
-
/**
* Returns the hour value of a given timestamp value. The timestamp is expressed in microseconds.
*/
@@ -1036,17 +471,6 @@ object DateTimeUtils {
}
}
- /**
- * Converts the timestamp `micros` from one timezone to another.
- *
- * Time-zone rules, such as daylight savings, mean that not every local date-time
- * is valid for the `toZone` time zone, thus the local date-time may be adjusted.
- */
- def convertTz(micros: Long, fromZone: ZoneId, toZone: ZoneId): Long = {
- val rebasedDateTime = getLocalDateTime(micros, toZone).atZone(fromZone)
- instantToMicros(rebasedDateTime.toInstant)
- }
-
/**
* Converts a timestamp without time zone from a source to target time zone.
*
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtilsImpl.scala
similarity index 51%
rename from sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtilsImpl.scala
index 59e35c9c9ba..fb480830d48 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/errors/ArrowErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/UDTUtilsImpl.scala
@@ -14,26 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.errors
+package org.apache.spark.sql.catalyst.util
-import org.apache.arrow.vector.types.pojo.ArrowType
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.types.UserDefinedType
-import org.apache.spark.SparkUnsupportedOperationException
-
-trait ArrowErrors {
-
- def unsupportedArrowTypeError(typeName: ArrowType): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "UNSUPPORTED_ARROWTYPE",
- messageParameters = Map("typeName" -> typeName.toString))
- }
-
- def duplicatedFieldNameInArrowStructError(
- fieldNames: Seq[String]): SparkUnsupportedOperationException = {
- new SparkUnsupportedOperationException(
- errorClass = "DUPLICATED_FIELD_NAME_IN_ARROW_STRUCT",
- messageParameters = Map("fieldNames" -> fieldNames.mkString("[", ", ", "]")))
+private[sql] class UDTUtilsImpl extends UDTUtils {
+ override def toRow(value: Any, udt: UserDefinedType[Any]): Any = {
+ CatalystTypeConverters.convertToScala(udt.serialize(value), udt.sqlType)
}
}
-
-object ArrowErrors extends ArrowErrors
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 5b1eaa20d22..c7245eb0a6d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -20,8 +20,7 @@ package org.apache.spark.sql.errors
import java.io.{FileNotFoundException, IOException}
import java.lang.reflect.InvocationTargetException
import java.net.{URISyntaxException, URL}
-import java.time.{DateTimeException, LocalDate}
-import java.time.temporal.ChronoField
+import java.time.DateTimeException
import java.util.concurrent.TimeoutException
import com.fasterxml.jackson.core.{JsonParser, JsonToken}
@@ -60,7 +59,7 @@ import org.apache.spark.util.{CircularBuffer, Utils}
* This does not include exceptions thrown during the eager execution of commands, which are
* grouped into [[QueryCompilationErrors]].
*/
-private[sql] object QueryExecutionErrors extends QueryErrorsBase {
+private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionErrors {
def cannotEvaluateExpressionError(expression: Expression): Throwable = {
SparkException.internalError(s"Cannot evaluate expression: $expression")
@@ -117,22 +116,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
summary = getSummary(context))
}
- def invalidInputInCastToDatetimeError(
- value: Any,
- from: DataType,
- to: DataType,
- context: SQLQueryContext): Throwable = {
- new SparkDateTimeException(
- errorClass = "CAST_INVALID_INPUT",
- messageParameters = Map(
- "expression" -> toSQLValue(value, from),
- "sourceType" -> toSQLType(from),
- "targetType" -> toSQLType(to),
- "ansiConfig" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
- context = getQueryContext(context),
- summary = getSummary(context))
- }
-
def invalidInputSyntaxForBooleanError(
s: UTF8String,
context: SQLQueryContext): SparkRuntimeException = {
@@ -632,23 +615,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
summary = "")
}
- def arithmeticOverflowError(
- message: String,
- hint: String = "",
- context: SQLQueryContext = null): ArithmeticException = {
- val alternative = if (hint.nonEmpty) {
- s" Use '$hint' to tolerate overflow and return NULL instead."
- } else ""
- new SparkArithmeticException(
- errorClass = "ARITHMETIC_OVERFLOW",
- messageParameters = Map(
- "message" -> message,
- "alternative" -> alternative,
- "config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
- context = getQueryContext(context),
- summary = getSummary(context))
- }
-
def unaryMinusCauseOverflowError(originValue: Int): SparkArithmeticException = {
new SparkArithmeticException(
errorClass = "_LEGACY_ERROR_TEMP_2043",
@@ -1316,52 +1282,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
messageParameters = Map.empty)
}
- def fieldDiffersFromDerivedLocalDateError(
- field: ChronoField,
- actual: Int,
- expected: Int,
- candidate: LocalDate): SparkDateTimeException = {
- new SparkDateTimeException(
- errorClass = "_LEGACY_ERROR_TEMP_2129",
- messageParameters = Map(
- "field" -> field.toString(),
- "actual" -> actual.toString(),
- "expected" -> expected.toString(),
- "candidate" -> candidate.toString()),
- context = Array.empty,
- summary = "")
- }
-
- def failToParseDateTimeInNewParserError(s: String, e: Throwable): Throwable = {
- new SparkUpgradeException(
- errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER",
- messageParameters = Map(
- "datetime" -> toSQLValue(s, StringType),
- "config" -> toSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key)),
- e)
- }
-
- def failToRecognizePatternAfterUpgradeError(
- pattern: String, e: Throwable, docroot: String): Throwable = {
- new SparkUpgradeException(
- errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION.DATETIME_PATTERN_RECOGNITION",
- messageParameters = Map(
- "pattern" -> toSQLValue(pattern, StringType),
- "config" -> toSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key),
- "docroot" -> docroot),
- e)
- }
-
- def failToRecognizePatternError(
- pattern: String, e: Throwable, docroot: String): SparkRuntimeException = {
- new SparkRuntimeException(
- errorClass = "_LEGACY_ERROR_TEMP_2130",
- messageParameters = Map(
- "pattern" -> toSQLValue(pattern, StringType),
- "docroot" -> docroot),
- cause = e)
- }
-
def registeringStreamingQueryListenerError(e: Exception): Throwable = {
new SparkException(
errorClass = "_LEGACY_ERROR_TEMP_2131",
@@ -1401,16 +1321,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
"dataType" -> dataType.toString()))
}
- def cannotParseStringAsDataTypeError(pattern: String, value: String, dataType: DataType)
- : SparkRuntimeException = {
- new SparkRuntimeException(
- errorClass = "_LEGACY_ERROR_TEMP_2134",
- messageParameters = Map(
- "value" -> toSQLValue(value, StringType),
- "pattern" -> toSQLValue(pattern, StringType),
- "dataType" -> dataType.toString()))
- }
-
def emptyJsonFieldValueError(dataType: DataType): SparkRuntimeException = {
new SparkRuntimeException(
errorClass = "EMPTY_JSON_FIELD_VALUE",
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
index 534cd8f9ab0..0dc8f75705a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowWriter.scala
@@ -24,7 +24,7 @@ import org.apache.arrow.vector.complex._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
-import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.ArrowUtils
@@ -83,7 +83,7 @@ object ArrowWriter {
case (_: YearMonthIntervalType, vector: IntervalYearVector) => new IntervalYearWriter(vector)
case (_: DayTimeIntervalType, vector: DurationVector) => new DurationWriter(vector)
case (dt, _) =>
- throw DataTypeErrors.unsupportedDataTypeError(dt)
+ throw ExecutionErrors.unsupportedDataTypeError(dt)
}
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 7fd960de37f..e74c3ada491 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -36,7 +36,6 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.{IGNORE_MISSING_FILES => SPARK_IGNORE_MISSING_FILES}
import org.apache.spark.network.util.ByteUnit
-import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.catalyst.analysis.{HintErrorLogger, Resolver}
import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
@@ -3162,7 +3161,7 @@ object SQLConf {
.checkValues(StoreAssignmentPolicy.values.map(_.toString))
.createWithDefault(StoreAssignmentPolicy.ANSI.toString)
- val ANSI_ENABLED = buildConf("spark.sql.ansi.enabled")
+ val ANSI_ENABLED = buildConf(SqlApiConf.ANSI_ENABLED_KEY)
.doc("When true, Spark SQL uses an ANSI compliant dialect instead of being Hive compliant. " +
"For example, Spark will throw an exception at runtime instead of returning null results " +
"when the inputs to a SQL operator/function are invalid." +
@@ -3749,10 +3748,6 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
- object LegacyBehaviorPolicy extends Enumeration {
- val EXCEPTION, LEGACY, CORRECTED = Value
- }
-
val LEGACY_CTE_PRECEDENCE_POLICY = buildConf("spark.sql.legacy.ctePrecedencePolicy")
.internal()
.doc("When LEGACY, outer CTE definitions takes precedence over inner definitions. If set to " +
@@ -3765,7 +3760,7 @@ object SQLConf {
.checkValues(LegacyBehaviorPolicy.values.map(_.toString))
.createWithDefault(LegacyBehaviorPolicy.EXCEPTION.toString)
- val LEGACY_TIME_PARSER_POLICY = buildConf("spark.sql.legacy.timeParserPolicy")
+ val LEGACY_TIME_PARSER_POLICY = buildConf(SqlApiConf.LEGACY_TIME_PARSER_POLICY_KEY)
.internal()
.doc("When LEGACY, java.text.SimpleDateFormat is used for formatting and parsing " +
"dates/timestamps in a locale-sensitive manner, which is the approach before Spark 3.0. " +
@@ -4735,7 +4730,7 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def legacyMsSqlServerNumericMappingEnabled: Boolean =
getConf(LEGACY_MSSQLSERVER_NUMERIC_MAPPING_ENABLED)
- def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value = {
+ override def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value = {
LegacyBehaviorPolicy.withName(getConf(SQLConf.LEGACY_TIME_PARSER_POLICY))
}
@@ -4875,7 +4870,7 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def crossJoinEnabled: Boolean = getConf(SQLConf.CROSS_JOINS_ENABLED)
- def sessionLocalTimeZone: String = getConf(SQLConf.SESSION_LOCAL_TIMEZONE)
+ override def sessionLocalTimeZone: String = getConf(SQLConf.SESSION_LOCAL_TIMEZONE)
def jsonGeneratorIgnoreNullFields: Boolean = getConf(SQLConf.JSON_GENERATOR_IGNORE_NULL_FIELDS)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/ToJsonUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/util/ToJsonUtil.scala
deleted file mode 100644
index cebd0bacc74..00000000000
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/ToJsonUtil.scala
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.util
-
-import java.sql.{Date, Timestamp}
-import java.time.{Instant, LocalDate}
-import java.util.Base64
-
-import scala.collection.mutable
-
-import org.json4s._
-import org.json4s.JsonAST.JValue
-import org.json4s.jackson.JsonMethods.{compact, pretty, render}
-
-import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.CatalystTypeConverters
-import org.apache.spark.sql.catalyst.util.{DateFormatter, DateTimeUtils, TimestampFormatter}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.CalendarInterval
-
-
-object ToJsonUtil {
- /**
- * The compact JSON representation of this row.
- * @since 3.5
- */
- @deprecated
- def json(row: Row): String = compact(jsonValue(row))
-
- /**
- * The pretty (i.e. indented) JSON representation of this row.
- * @since 3.5
- */
- @deprecated
- def prettyJson(row: Row): String = pretty(render(jsonValue(row)))
-
- /**
- * JSON representation of the row.
- *
- * Note that this only supports the data types that are also supported by
- * [[org.apache.spark.sql.catalyst.encoders.RowEncoder]].
- *
- * @return the JSON representation of the row.
- */
- private[sql] def jsonValue(row: Row): JValue = {
- require(row.schema != null, "JSON serialization requires a non-null schema.")
-
- lazy val zoneId = DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone)
- lazy val dateFormatter = DateFormatter()
- lazy val timestampFormatter = TimestampFormatter(zoneId)
-
- // Convert an iterator of values to a json array
- def iteratorToJsonArray(iterator: Iterator[_], elementType: DataType): JArray = {
- JArray(iterator.map(toJson(_, elementType)).toList)
- }
-
- // Convert a value to json.
- def toJson(value: Any, dataType: DataType): JValue = (value, dataType) match {
- case (null, _) => JNull
- case (b: Boolean, _) => JBool(b)
- case (b: Byte, _) => JLong(b)
- case (s: Short, _) => JLong(s)
- case (i: Int, _) => JLong(i)
- case (l: Long, _) => JLong(l)
- case (f: Float, _) => JDouble(f)
- case (d: Double, _) => JDouble(d)
- case (d: BigDecimal, _) => JDecimal(d)
- case (d: java.math.BigDecimal, _) => JDecimal(d)
- case (d: Decimal, _) => JDecimal(d.toBigDecimal)
- case (s: String, _) => JString(s)
- case (b: Array[Byte], BinaryType) =>
- JString(Base64.getEncoder.encodeToString(b))
- case (d: LocalDate, _) => JString(dateFormatter.format(d))
- case (d: Date, _) => JString(dateFormatter.format(d))
- case (i: Instant, _) => JString(timestampFormatter.format(i))
- case (t: Timestamp, _) => JString(timestampFormatter.format(t))
- case (i: CalendarInterval, _) => JString(i.toString)
- case (a: Array[_], ArrayType(elementType, _)) =>
- iteratorToJsonArray(a.iterator, elementType)
- case (a: mutable.ArraySeq[_], ArrayType(elementType, _)) =>
- iteratorToJsonArray(a.iterator, elementType)
- case (s: Seq[_], ArrayType(elementType, _)) =>
- iteratorToJsonArray(s.iterator, elementType)
- case (m: Map[String @unchecked, _], MapType(StringType, valueType, _)) =>
- new JObject(m.toList.sortBy(_._1).map {
- case (k, v) => k -> toJson(v, valueType)
- })
- case (m: Map[_, _], MapType(keyType, valueType, _)) =>
- new JArray(m.iterator.map {
- case (k, v) =>
- new JObject("key" -> toJson(k, keyType) :: "value" -> toJson(v, valueType) :: Nil)
- }.toList)
- case (r: Row, _) => jsonValue(r)
- case (v: Any, udt: UserDefinedType[Any @unchecked]) =>
- val dataType = udt.sqlType
- toJson(CatalystTypeConverters.convertToScala(udt.serialize(v), dataType), dataType)
- case _ =>
- throw new IllegalArgumentException(s"Failed to convert value $value " +
- s"(class of ${value.getClass}}) with the type of $dataType to JSON.")
- }
-
- // Convert the row fields to json
- var n = 0
- val elements = new mutable.ListBuffer[JField]
- val len = row.length
- while (n < len) {
- val field = row.schema(n)
- elements += (field.name -> toJson(row.apply(n), field.dataType))
- n += 1
- }
- new JObject(elements.toList)
- }
-}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/RowJsonSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/RowJsonSuite.scala
index 916edd9f2ed..1962fca66c0 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/RowJsonSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/RowJsonSuite.scala
@@ -25,7 +25,6 @@ import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.encoders.{ExamplePoint, ExamplePointUDT}
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types._
-import org.apache.spark.sql.util.ToJsonUtil
/**
* Test suite for [[Row]] JSON serialization.
@@ -38,7 +37,7 @@ class RowJsonSuite extends SparkFunSuite {
private def testJson(name: String, value: Any, dt: DataType, expected: JValue): Unit = {
test(name) {
val row = new GenericRowWithSchema(Array(value), new StructType().add("a", dt))
- assert(ToJsonUtil.jsonValue(row) === JObject("a" -> expected))
+ assert(row.jsonValue === JObject("a" -> expected))
}
}
@@ -123,7 +122,7 @@ class RowJsonSuite extends SparkFunSuite {
test("no schema") {
val e = intercept[IllegalArgumentException] {
- ToJsonUtil.jsonValue(Row("a"))
+ Row("a").jsonValue
}
assert(e.getMessage.contains("requires a non-null schema"))
}
@@ -133,7 +132,7 @@ class RowJsonSuite extends SparkFunSuite {
val row = new GenericRowWithSchema(
Array((1, 2)),
new StructType().add("a", ObjectType(classOf[(Int, Int)])))
- ToJsonUtil.jsonValue(row)
+ row.jsonValue
}
assert(e.getMessage.contains("Failed to convert value"))
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/RowTest.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/RowTest.scala
index ce0dd767c64..ec40989e6b7 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/RowTest.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/RowTest.scala
@@ -28,7 +28,6 @@ import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{GenericRow, GenericRowWithSchema}
import org.apache.spark.sql.types._
-import org.apache.spark.sql.util.ToJsonUtil
class RowTest extends AnyFunSpec with Matchers {
@@ -102,7 +101,7 @@ class RowTest extends AnyFunSpec with Matchers {
val values = ArraySeq("1", "2", "3")
val row = new GenericRowWithSchema(Array(values), schema)
val expectedList = JArray(JString("1") :: JString("2") :: JString("3") :: Nil)
- ToJsonUtil.jsonValue(row) shouldBe new JObject(("list", expectedList) :: Nil)
+ row.jsonValue shouldBe new JObject(("list", expectedList) :: Nil)
}
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateFormatterSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateFormatterSuite.scala
index 4b703285434..20b1ade81f5 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateFormatterSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/DateFormatterSuite.scala
@@ -22,8 +22,7 @@ import java.time.{DateTimeException, LocalDate}
import org.apache.spark.SparkUpgradeException
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
class DateFormatterSuite extends DatetimeFormatterSuite {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/TimestampFormatterSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/TimestampFormatterSuite.scala
index e050a6d1789..eb173bc7f8c 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/TimestampFormatterSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/TimestampFormatterSuite.scala
@@ -24,8 +24,7 @@ import org.apache.commons.lang3.{JavaVersion, SystemUtils}
import org.apache.spark.SparkUpgradeException
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
import org.apache.spark.sql.catalyst.util.DateTimeUtils._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.unsafe.types.UTF8String
class TimestampFormatterSuite extends DatetimeFormatterSuite {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
index fc879f7e98f..3fec13a7f9b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.errors.DataTypeErrors
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.execution.command.DataWritingCommandExec
import org.apache.spark.sql.execution.datasources.V1WriteCommand
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -273,7 +273,7 @@ private object RowToColumnConverter {
case dt: DecimalType => new DecimalConverter(dt)
case mt: MapType => MapConverter(getConverterForType(mt.keyType, nullable = false),
getConverterForType(mt.valueType, mt.valueContainsNull))
- case unknown => throw DataTypeErrors.unsupportedDataTypeError(unknown)
+ case unknown => throw ExecutionErrors.unsupportedDataTypeError(unknown)
}
if (nullable) {
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 cedca062074..a72e4c339af 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
@@ -33,8 +33,7 @@ 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
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.sources.BaseRelation
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CaseInsensitiveStringMap
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 6994e1ba39d..5899b6621ad 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
@@ -38,7 +38,7 @@ import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils, IntervalUtils}
import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros, RebaseSpec}
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.sources
import org.apache.spark.unsafe.types.UTF8String
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 6e29afce491..025ee3dd034 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
@@ -35,8 +35,7 @@ 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
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
/**
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 a6d13d072f4..8a6221c4b87 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
@@ -42,8 +42,8 @@ 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, OutputWriter, OutputWriterFactory}
import org.apache.spark.sql.execution.datasources.v2.V2ColumnUtils
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.{LegacyBehaviorPolicy, PARQUET_AGGREGATE_PUSHDOWN_ENABLED}
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+import org.apache.spark.sql.internal.SQLConf.PARQUET_AGGREGATE_PUSHDOWN_ENABLED
import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType, StructField, StructType, UserDefinedType}
object ParquetUtils extends Logging {
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 9f3ee0af4e5..e546816dd1d 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
@@ -36,8 +36,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util.DateTimeUtils
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.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
/**
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 0d63d295ccd..7987372abde 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
@@ -26,7 +26,7 @@ import org.apache.spark.sql.connector.read.SupportsPushDownAggregates
import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFilters, SparkToParquetSchemaConverter}
import org.apache.spark.sql.execution.datasources.v2.FileScanBuilder
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala
index 5cce52b9c0f..b3927699756 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala
@@ -37,7 +37,6 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
-import org.apache.spark.sql.util.ToJsonUtil
/**
* Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
@@ -174,7 +173,7 @@ class StreamingQueryProgress private[spark](
("stateOperators" -> JArray(stateOperators.map(_.jsonValue).toList)) ~
("sources" -> JArray(sources.map(_.jsonValue).toList)) ~
("sink" -> sink.jsonValue) ~
- ("observedMetrics" -> safeMapToJValue[Row](observedMetrics, row => ToJsonUtil.jsonValue(row)))
+ ("observedMetrics" -> safeMapToJValue[Row](observedMetrics, row => row.jsonValue))
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
index df82e3c268f..e70d04b7b5a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
@@ -42,8 +42,8 @@ import org.apache.spark.sql.execution.datasources.parquet.ParquetTest
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
import org.apache.spark.sql.execution.streaming.FileSystemBasedCheckpointFileManager
import org.apache.spark.sql.functions.{lit, lower, struct, sum, udf}
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy.EXCEPTION
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy.EXCEPTION
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects}
import org.apache.spark.sql.streaming.StreamingQueryException
import org.apache.spark.sql.test.SharedSparkSession
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/DateTimeRebaseBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/DateTimeRebaseBenchmark.scala
index e7f1f15ae32..800c22a78d0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/DateTimeRebaseBenchmark.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/DateTimeRebaseBenchmark.scala
@@ -24,8 +24,8 @@ import org.apache.spark.benchmark.Benchmark
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.catalyst.util.DateTimeConstants.SECONDS_PER_DAY
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.{LegacyBehaviorPolicy, ParquetOutputTimestampType}
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
object DateTime extends Enumeration {
type DateTime = Value
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
index ceb79eaa866..3bd45ca0dcd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
@@ -41,8 +41,7 @@ import org.apache.spark.sql.{AnalysisException, Column, DataFrame, Encoders, Que
import org.apache.spark.sql.catalyst.csv.CSVOptions
import org.apache.spark.sql.catalyst.util.{DateTimeTestUtils, DateTimeUtils}
import org.apache.spark.sql.execution.datasources.CommonFileDataSourceSuite
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
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 9adcb43c838..269a3efb736 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
@@ -48,9 +48,8 @@ import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, HadoopFsR
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.functions._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType.{INT96, TIMESTAMP_MICROS, TIMESTAMP_MILLIS}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
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 240bb4e6dcb..4f906411345 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
@@ -23,9 +23,9 @@ import java.sql.{Date, Timestamp}
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}
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy.{CORRECTED, EXCEPTION, LEGACY}
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy.{CORRECTED, EXCEPTION, LEGACY}
+import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType.{INT96, TIMESTAMP_MICROS, TIMESTAMP_MILLIS}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.tags.SlowSQLTest
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala
index c6e5585f619..b4140f17775 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala
@@ -30,8 +30,8 @@ import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAM
import org.apache.spark.sql.execution.DataSourceScanExec
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
import org.apache.spark.sql.test.SQLTestUtils
import org.apache.spark.sql.types._
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org