This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 8bb224895e95 [SPARK-57455][SQL] Support nanosecond timestamp types in
ORC
8bb224895e95 is described below
commit 8bb224895e957f1cb4d279a213003c1d5ddd1aaa
Author: Maxim Gekk <[email protected]>
AuthorDate: Thu Jun 18 13:17:03 2026 +0200
[SPARK-57455][SQL] Support nanosecond timestamp types in ORC
### What changes were proposed in this pull request?
Add ORC read/write support for nanosecond timestamp types
(`TimestampNTZNanosType(p)` and `TimestampLTZNanosType(p)`, `p` in 7..9) across
native ORC (v1/v2, vectorized and non-vectorized) and the Hive ORC serde paths.
The Spark SQL types map onto ORC column types as follows:
| Spark SQL type | ORC column type | `spark.sql.catalyst.type` attribute |
|---|---|---|
| `TimestampType` (micros) | `TIMESTAMP` | `timestamp` |
| `TimestampNTZType` (micros) | `TIMESTAMP` | `timestamp_ntz` |
| `TimestampNTZNanosType(p)` | `TIMESTAMP` | `timestamp_ntz(p)` |
| `TimestampLTZNanosType(p)` | `TIMESTAMP_INSTANT` | `timestamp_ltz(p)` |
The ORC physical category is shared with the existing micros types
(wall-clock `TIMESTAMP` for NTZ, instant-based `TIMESTAMP_INSTANT` for LTZ),
and the exact Spark type/precision is carried in the `spark.sql.catalyst.type`
ORC attribute. On read, that attribute restores the nanos Catalyst type; a
plain ORC `TIMESTAMP` without it stays `TimestampType` for backward
compatibility.
### Why are the changes needed?
ORC rejected nanos timestamp types in its datasource capability checks and
lacked the conversions to round-trip them, so these columns could not be
written or read through ORC.
### Does this PR introduce _any_ user-facing change?
Yes. Users can write and read `TimestampNTZNanosType(p)` /
`TimestampLTZNanosType(p)` (`p` in 7..9) with ORC in datasource v1/v2 and Hive
ORC, including inside nested struct/array/map columns.
### How was this patch tested?
New tests in `OrcQuerySuite` (v1/v2) and `HiveOrcSourceSuite` covering
precisions 7/8/9:
- vectorized vs non-vectorized read parity;
- time-zone independence (NTZ wall clock and LTZ instant preserved across a
JVM default-zone change);
- schema inference from the ORC attribute;
- documented min/max range round-trip;
- nested/complex types (struct/array/map);
- backward compatibility (plain ORC `timestamp` stays `TimestampType`).
```
build/sbt 'sql/testOnly *.orc.OrcV1QuerySuite *.orc.OrcV2QuerySuite -- -z
"SPARK-57455"'
build/sbt 'hive/testOnly *.orc.HiveOrcSourceSuite -- -z "SPARK-57455"'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Codex 5.3
Closes #56557 from MaxGekk/spark-57455-orc-nanos.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit bdc42527b2045e87452cf307ab527fdbdecf152e)
Signed-off-by: Max Gekk <[email protected]>
---
.../spark/sql/errors/QueryExecutionErrors.scala | 8 ++
.../datasources/orc/OrcAtomicColumnVector.java | 29 +++++
.../datasources/orc/OrcDeserializer.scala | 10 ++
.../execution/datasources/orc/OrcFileFormat.scala | 3 -
.../execution/datasources/orc/OrcSerializer.scala | 14 ++
.../sql/execution/datasources/orc/OrcUtils.scala | 46 ++++++-
.../execution/datasources/v2/orc/OrcTable.scala | 3 -
.../spark/sql/FileBasedDataSourceSuite.scala | 42 +++++-
.../sql/errors/QueryExecutionErrorsSuite.scala | 50 ++++++-
.../execution/datasources/orc/OrcQuerySuite.scala | 143 +++++++++++++++++++++
.../execution/datasources/orc/OrcSourceSuite.scala | 26 ++++
.../sql/execution/datasources/orc/OrcTest.scala | 17 ++-
.../org/apache/spark/sql/hive/HiveInspectors.scala | 120 ++++++++++++++++-
.../org/apache/spark/sql/hive/TableReader.scala | 16 ++-
.../apache/spark/sql/hive/orc/OrcFileFormat.scala | 42 +++++-
.../spark/sql/hive/orc/HiveOrcSourceSuite.scala | 135 ++++++++++++++-----
16 files changed, 642 insertions(+), 62 deletions(-)
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 4b50a90a30f7..259dce745662 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
@@ -2543,6 +2543,14 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
"toType" -> toSQLType(TimestampType)))
}
+ def cannotCastOrcTimestampError(orcType: DataType, toType: DataType):
Throwable = {
+ new SparkUnsupportedOperationException(
+ errorClass = "UNSUPPORTED_FEATURE.ORC_TYPE_CAST",
+ messageParameters = Map(
+ "orcType" -> toSQLType(orcType),
+ "toType" -> toSQLType(toType)))
+ }
+
def writePartitionExceedConfigSizeWhenDynamicPartitionError(
numWrittenParts: Int,
maxDynamicPartitions: Int,
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcAtomicColumnVector.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcAtomicColumnVector.java
index 36e5da64bb75..3ea79f83ac53 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcAtomicColumnVector.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcAtomicColumnVector.java
@@ -27,9 +27,12 @@ import org.apache.spark.sql.catalyst.util.RebaseDateTime;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DateType;
import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.TimestampLTZNanosType;
+import org.apache.spark.sql.types.TimestampNTZNanosType;
import org.apache.spark.sql.types.TimestampType;
import org.apache.spark.sql.vectorized.ColumnarArray;
import org.apache.spark.sql.vectorized.ColumnarMap;
+import org.apache.spark.unsafe.types.TimestampNanosVal;
import org.apache.spark.unsafe.types.UTF8String;
/**
@@ -37,6 +40,8 @@ import org.apache.spark.unsafe.types.UTF8String;
*/
public class OrcAtomicColumnVector extends OrcColumnVector {
private final boolean isTimestamp;
+ private final boolean isTimestampNTZNanos;
+ private final boolean isTimestampLTZNanos;
private final boolean isDate;
// Column vector for each type. Only 1 is populated for any type.
@@ -54,6 +59,8 @@ public class OrcAtomicColumnVector extends OrcColumnVector {
} else {
isTimestamp = false;
}
+ isTimestampNTZNanos = type instanceof TimestampNTZNanosType;
+ isTimestampLTZNanos = type instanceof TimestampLTZNanosType;
if (type instanceof DateType) {
isDate = true;
@@ -111,6 +118,28 @@ public class OrcAtomicColumnVector extends OrcColumnVector
{
}
}
+ @Override
+ public TimestampNanosVal getTimestampNTZNanos(int rowId) {
+ if (!isTimestampNTZNanos || isNullAt(rowId)) {
+ return null;
+ }
+ int index = getRowIndex(rowId);
+ return DateTimeUtils.localDateTimeToTimestampNanos(
+ timestampData.asScratchTimestamp(index).toLocalDateTime(),
+ ((TimestampNTZNanosType) type).precision());
+ }
+
+ @Override
+ public TimestampNanosVal getTimestampLTZNanos(int rowId) {
+ if (!isTimestampLTZNanos || isNullAt(rowId)) {
+ return null;
+ }
+ int index = getRowIndex(rowId);
+ return DateTimeUtils.instantToTimestampNanos(
+ timestampData.asScratchTimestamp(index).toInstant(),
+ ((TimestampLTZNanosType) type).precision());
+ }
+
@Override
public float getFloat(int rowId) {
return (float) doubleData.vector[getRowIndex(rowId)];
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
index 861e271acd3c..04dd37dec50d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
@@ -149,6 +149,16 @@ class OrcDeserializer(
case TimestampType => (ordinal, value) =>
updater.setLong(ordinal,
DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
+ case t: TimestampLTZNanosType => (ordinal, value) =>
+ val ts = value.asInstanceOf[OrcTimestamp]
+ val instant = ts.toInstant
+ updater.set(ordinal, DateTimeUtils.instantToTimestampNanos(instant,
t.precision))
+ case t: TimestampNTZNanosType => (ordinal, value) =>
+ val ts = value.asInstanceOf[OrcTimestamp]
+ val localDateTime = ts.toLocalDateTime
+ updater.set(
+ ordinal,
+ DateTimeUtils.localDateTimeToTimestampNanos(localDateTime,
t.precision))
case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
val v = OrcShimUtils.getDecimal(value)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
index 7065693e59ca..633ac8107f32 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
@@ -251,9 +251,6 @@ class OrcFileFormat
case _: GeometryType | _: GeographyType => false
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
-
case _: AtomicType => true
case st: StructType => st.forall { f => supportDataType(f.dataType) }
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
index dc124dc2f7c9..64bc3e429212 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
@@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.TimestampNanosVal
/**
* A serializer to serialize Spark rows to ORC structs.
@@ -154,6 +155,19 @@ class OrcSerializer(dataSchema: StructType) {
val result = new OrcTimestamp(ts.getTime)
result.setNanos(ts.getNanos)
result
+ case t: TimestampLTZNanosType => (getter, ordinal) =>
+ val v = getter.get(ordinal, t).asInstanceOf[TimestampNanosVal]
+ val instant = DateTimeUtils.timestampNanosToInstant(v)
+ val result = new OrcTimestamp(instant.toEpochMilli)
+ result.setNanos(instant.getNano)
+ result
+ case t: TimestampNTZNanosType => (getter, ordinal) =>
+ val v = getter.get(ordinal, t).asInstanceOf[TimestampNanosVal]
+ val localDateTime = DateTimeUtils.timestampNanosToLocalDateTime(v)
+ val ts = java.sql.Timestamp.valueOf(localDateTime)
+ val result = new OrcTimestamp(ts.getTime)
+ result.setNanos(ts.getNanos)
+ result
case DecimalType.Fixed(precision, scale) =>
OrcShimUtils.getHiveDecimalWritable(precision, scale)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index 72c7ce5232e8..2c1e8beeb2b1 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -196,15 +196,45 @@ object OrcUtils extends Logging {
requiredSchema: StructType,
orcSchema: TypeDescription,
conf: Configuration): Option[(Array[Int], Boolean)] = {
- def checkTimestampCompatibility(orcCatalystSchema: StructType, dataSchema:
StructType): Unit = {
-
orcCatalystSchema.fields.map(_.dataType).zip(dataSchema.fields.map(_.dataType)).foreach
{
+ def isOrcTimestamp(dt: DataType): Boolean = dt match {
+ case TimestampType | TimestampNTZType | _: AnyTimestampNanoType => true
+ case _ => false
+ }
+
+ // The ORC reader does not coerce between timestamp families/precisions,
except between
+ // nanos timestamps of the same kind (NTZ or LTZ), which share an ORC
physical category and
+ // only differ by the precision applied on read. Any other mismatch (zone
or micros<->nanos)
+ // would otherwise fail obscurely, so reject it with a clear error.
+ def timestampReadCompatible(orcType: DataType, dataType: DataType):
Boolean =
+ (orcType, dataType) match {
+ case _ if orcType == dataType => true
+ case (_: TimestampNTZNanosType, _: TimestampNTZNanosType) => true
+ case (_: TimestampLTZNanosType, _: TimestampLTZNanosType) => true
+ case _ => false
+ }
+
+ // Recurse into struct/array/map so timestamp mismatches nested inside
containers are caught
+ // too, not just top-level and struct fields.
+ def checkTypeCompatibility(orcType: DataType, dataType: DataType): Unit =
+ (orcType, dataType) match {
case (TimestampType, TimestampNTZType) =>
throw
QueryExecutionErrors.cannotConvertOrcTimestampToTimestampNTZError()
case (TimestampNTZType, TimestampType) =>
throw
QueryExecutionErrors.cannotConvertOrcTimestampNTZToTimestampLTZError()
- case (t1: StructType, t2: StructType) =>
checkTimestampCompatibility(t1, t2)
+ case (o, d) if isOrcTimestamp(o) && isOrcTimestamp(d) &&
!timestampReadCompatible(o, d) =>
+ throw QueryExecutionErrors.cannotCastOrcTimestampError(o, d)
+ case (o: StructType, d: StructType) => checkTimestampCompatibility(o,
d)
+ case (ArrayType(o, _), ArrayType(d, _)) => checkTypeCompatibility(o, d)
+ case (MapType(ok, ov, _), MapType(dk, dv, _)) =>
+ checkTypeCompatibility(ok, dk)
+ checkTypeCompatibility(ov, dv)
case _ =>
}
+
+ def checkTimestampCompatibility(orcCatalystSchema: StructType, dataSchema:
StructType): Unit = {
+
orcCatalystSchema.fields.map(_.dataType).zip(dataSchema.fields.map(_.dataType)).foreach
{
+ case (orcType, dataType) => checkTypeCompatibility(orcType, dataType)
+ }
}
checkTimestampCompatibility(toCatalystSchema(orcSchema), dataSchema)
@@ -291,6 +321,8 @@ object OrcUtils extends Logging {
case m: MapType =>
s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
case _: DayTimeIntervalType | _: TimestampNTZType | _: TimeType =>
LongType.catalogString
+ case _: TimestampLTZNanosType => "timestamp with local time zone"
+ case _: TimestampNTZNanosType => "timestamp"
case _: YearMonthIntervalType => IntegerType.catalogString
case _ => dt.catalogString
}
@@ -318,6 +350,14 @@ object OrcUtils extends Logging {
val typeDesc = new
TypeDescription(TypeDescription.Category.TIMESTAMP)
typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, t.typeName)
Some(typeDesc)
+ case t: TimestampLTZNanosType =>
+ val typeDesc = new
TypeDescription(TypeDescription.Category.TIMESTAMP_INSTANT)
+ typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, t.typeName)
+ Some(typeDesc)
+ case t: TimestampNTZNanosType =>
+ val typeDesc = new
TypeDescription(TypeDescription.Category.TIMESTAMP)
+ typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, t.typeName)
+ Some(typeDesc)
case _: StringType =>
val typeDesc = new TypeDescription(TypeDescription.Category.STRING)
typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME,
StringType.typeName)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
index 24cd5e60321e..08cd89fdacc6 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
@@ -53,9 +53,6 @@ case class OrcTable(
override def supportsDataType(dataType: DataType): Boolean = dataType match {
case _: GeometryType | _: GeographyType => false
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
-
case _: AtomicType => true
case st: StructType => st.forall { f => supportsDataType(f.dataType) }
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
index dbf5dee4146c..22f3ae6126ee 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql
import java.io.{File, FileNotFoundException}
import java.net.URI
import java.nio.file.{Files, StandardOpenOption}
+import java.time.{LocalDateTime, ZoneOffset}
import scala.collection.mutable
@@ -33,6 +34,7 @@ import
org.apache.spark.sql.catalyst.expressions.{AttributeReference, GreaterTha
import
org.apache.spark.sql.catalyst.expressions.IntegralLiteralTestUtils.{negativeInt,
positiveInt}
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.execution.{FileSourceScanLike, SimpleMode}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.FilePartition
@@ -1336,10 +1338,9 @@ class FileBasedDataSourceSuite extends SharedSparkSession
}
}
- test("SPARK-57166: nanosecond timestamp types are not supported in
non-Parquet file sources") {
- // These built-in file formats do not support nanosecond-capable
timestamps. Parquet support is
- // covered separately in ParquetTimestampNanosSuite.
- val unsupportedDataSources = Seq("orc", "json", "csv", "xml")
+ test("SPARK-57166: nanosecond timestamp types are not supported in selected
file data sources") {
+ // Parquet and ORC support nanosecond-capable timestamps, while these
formats still reject them.
+ val unsupportedDataSources = Seq("json", "csv", "xml")
val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
// Test both v1 and v2 data sources.
@@ -1395,6 +1396,39 @@ class FileBasedDataSourceSuite extends SharedSparkSession
}
}
}
+
+ test("SPARK-57166: ORC supports nanosecond timestamp types in v1 and v2") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ // Validate both v1 and v2 ORC paths.
+ Seq(true, false).foreach { useV1 =>
+ val useV1List = if (useV1) "orc" else ""
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempDir { dir =>
+ // Build the row from an external java.time value; the
column schema carries the
+ // precision and truncates the sub-microsecond digits,
matching the ORC suites.
+ val wallClock = LocalDateTime.of(1970, 1, 1, 0, 20, 34,
567890123)
+ val value: Any = nanosType match {
+ case _: TimestampNTZNanosType => wallClock
+ case _: TimestampLTZNanosType =>
wallClock.toInstant(ZoneOffset.UTC)
+ }
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(value))),
+ new StructType().add("ts", nanosType))
+ val path = new File(dir,
s"orc_nanos_${nanosType.typeName}").getCanonicalPath
+ df.write.format("orc").mode("overwrite").save(path)
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .format("orc").load(path)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
object TestingUDT {
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 ce549da03b46..058b47ee4f67 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
@@ -20,6 +20,7 @@ package org.apache.spark.sql.errors
import java.io.{File, IOException}
import java.net.{URI, URL}
import java.sql.{Connection, DatabaseMetaData, Driver, DriverManager,
PreparedStatement, ResultSet, ResultSetMetaData}
+import java.time.LocalDateTime
import java.util.{Locale, Properties, ServiceConfigurationError}
import scala.jdk.CollectionConverters._
@@ -54,7 +55,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects}
import org.apache.spark.sql.streaming.StreamingQueryException
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{ArrayType, BooleanType, DataType,
DecimalType, IntegerType, LongType, MetadataBuilder, StructField, StructType}
+import org.apache.spark.sql.types.{ArrayType, BooleanType, DataType,
DecimalType, IntegerType, LongType, MetadataBuilder, StructField, StructType,
TimestampNTZNanosType}
import org.apache.spark.sql.vectorized.ColumnarArray
import org.apache.spark.unsafe.array.ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
import org.apache.spark.util.ThreadUtils
@@ -322,6 +323,53 @@ class QueryExecutionErrorsSuite
}
}
+ test("UNSUPPORTED_FEATURE - SPARK-57455: can't read nanos timestamp as
micros timestamp") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ withTempPath { file =>
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(LocalDateTime.of(1970, 1, 1,
0, 0, 0, 1)))),
+ new StructType().add("ts", TimestampNTZNanosType(9)))
+ df.write.orc(file.getCanonicalPath)
+ withAllNativeOrcReaders {
+ val ex = intercept[SparkException] {
+ spark.read.schema("ts
timestamp_ntz").orc(file.getCanonicalPath).collect()
+ }
+ assert(ex.getCondition.startsWith("FAILED_READ_FILE"))
+ checkError(
+ exception =
ex.getCause.asInstanceOf[SparkUnsupportedOperationException],
+ condition = "UNSUPPORTED_FEATURE.ORC_TYPE_CAST",
+ parameters = Map("orcType" -> "\"TIMESTAMP_NTZ(9)\"",
+ "toType" -> "\"TIMESTAMP_NTZ\""),
+ sqlState = "0A000")
+ }
+ }
+ }
+ }
+
+ test("UNSUPPORTED_FEATURE - SPARK-57455: nanos timestamp mismatch nested in
array is caught") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ withTempPath { file =>
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(
+ Seq(Row(Seq(LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1))))),
+ new StructType().add("ts", ArrayType(TimestampNTZNanosType(9))))
+ df.write.orc(file.getCanonicalPath)
+ withAllNativeOrcReaders {
+ val ex = intercept[SparkException] {
+ spark.read.schema("ts
array<timestamp_ntz>").orc(file.getCanonicalPath).collect()
+ }
+ assert(ex.getCondition.startsWith("FAILED_READ_FILE"))
+ checkError(
+ exception =
ex.getCause.asInstanceOf[SparkUnsupportedOperationException],
+ condition = "UNSUPPORTED_FEATURE.ORC_TYPE_CAST",
+ parameters = Map("orcType" -> "\"TIMESTAMP_NTZ(9)\"",
+ "toType" -> "\"TIMESTAMP_NTZ\""),
+ sqlState = "0A000")
+ }
+ }
+ }
+ }
+
test("SPARK-42290: NotEnoughMemory error can't be create") {
QueryExecutionErrors.notEnoughMemoryToBuildAndBroadcastTableError(new
OutOfMemoryError(), Seq())
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
index a42c004e3aaf..c4e041e13ebd 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
@@ -36,6 +36,7 @@ import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
+import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.{HadoopFsRelation,
LogicalRelation, RecordReaderIterator}
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
@@ -917,6 +918,148 @@ abstract class OrcQuerySuite extends OrcQueryTest with
SharedSparkSession {
}
}
+ test("SPARK-57455: ORC reads nanos timestamps with vectorized and
non-vectorized readers") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val wallClocks = Seq(
+ LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0),
+ LocalDateTime.of(1970, 1, 1, 0, 0, 1, 900),
+ LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, wallClocks)
+ val expected = inputDf.collect()
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ inputDf.write.mode("overwrite").orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ checkAnswer(
+ spark.read.schema(new StructType().add("ts",
nanosType)).orc(path),
+ expected)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: nanos timestamps survive a time-zone change across ORC
write/read") {
+ // The NTZ wall clock stays zone-independent and the LTZ instant is
preserved even when the
+ // writer's and reader's JVM default time zones differ: NTZ goes through
an ORC TIMESTAMP
+ // (which preserves the local wall-clock fields) via the default-zone
valueOf/toLocalDateTime
+ // on both ends, and LTZ goes through an ORC TIMESTAMP_INSTANT
(instant-preserving) via
+ // epoch-based conversions, so neither depends on the default zone
matching across the round
+ // trip.
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val wallClock = Seq(LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, wallClock)
+ val expected = inputDf.collect()
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.LA) {
+ inputDf.write.mode("overwrite").orc(path)
+ }
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.UTC)
{
+ checkAnswer(
+ spark.read.schema(new StructType().add("ts",
nanosType)).orc(path),
+ expected)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: nanos timestamp type is inferred from ORC without an
explicit schema") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val wallClock = Seq(LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, wallClock)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ inputDf.write.mode("overwrite").orc(path)
+ // Read without supplying a schema: the nanos type is recovered
from the
+ // spark.sql.catalyst.type attribute stored in the ORC type
description.
+ val readBack = spark.read.orc(path)
+ assert(readBack.schema("ts").dataType === nanosType)
+ checkAnswer(readBack, inputDf)
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: ORC round-trips the min/max values of nanos timestamp
types") {
+ // The documented range is [0001-01-01T00:00:00.000000000,
9999-12-31T23:59:59.999999999]
+ // (at UTC for LTZ); check both ends round-trip through ORC with both
readers, at every
+ // supported precision (the max's sub-micro digits are truncated to the
precision).
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val bounds = Seq(
+ LocalDateTime.of(1, 1, 1, 0, 0, 0, 0),
+ LocalDateTime.of(9999, 12, 31, 23, 59, 59, 999999999))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val inputDf = nanosTimestampDf(nanosType, bounds)
+ val expected = inputDf.collect()
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ inputDf.write.mode("overwrite").orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ checkAnswer(
+ spark.read.schema(new StructType().add("ts",
nanosType)).orc(path),
+ expected)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: ORC round-trips nanos timestamps in nested/complex
types") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ val wallClocks = Seq(LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempView("nanos_input") {
+ nanosTimestampDf(nanosType,
wallClocks).createOrReplaceTempView("nanos_input")
+ val nested = sql(
+ """SELECT
+ | named_struct('ts', ts) AS struct_ts,
+ | array(ts) AS array_ts,
+ | map('k', ts) AS map_ts
+ |FROM nanos_input
+ |""".stripMargin)
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ nested.write.mode("overwrite").orc(path)
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
+ val readBack = spark.read.schema(nested.schema).orc(path)
+ checkAnswer(readBack, nested)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
// SPARK-39519: Ignore this case because it requires more than 4g heap
memory to ensure test
// stability when use Java 11. Should test it manually when upgrading
`hive-storage-api`
ignore("SPARK-39387: BytesColumnVector should not throw RuntimeException due
to overflow") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
index a65eb5fa43fc..3e7d28fb8843 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala
@@ -25,11 +25,13 @@ import java.util.Locale
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
+import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector
import org.apache.logging.log4j.Level
import org.apache.orc.OrcConf.COMPRESS
import org.apache.orc.OrcFile
import org.apache.orc.OrcProto.ColumnEncoding.Kind.{DICTIONARY_V2, DIRECT,
DIRECT_V2}
import org.apache.orc.OrcProto.Stream.Kind
+import org.apache.orc.TypeDescription
import org.apache.orc.impl.RecordReaderImpl
import org.apache.spark.{SPARK_VERSION_SHORT, SparkConf, SparkException}
@@ -875,6 +877,30 @@ abstract class OrcSourceSuite extends OrcSuite with
SharedSparkSession {
}
}
+ test("SPARK-57455: plain ORC timestamp stays TimestampType without Spark
nanos metadata") {
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ val outputFile = new Path(new File(dir,
"part-00000.orc").getCanonicalPath)
+ val schema = TypeDescription.fromString("struct<ts:timestamp>")
+ val writerOptions = OrcFile
+ .writerOptions(spark.sessionState.newHadoopConf())
+ .setSchema(schema)
+ Utils.tryWithResource(OrcFile.createWriter(outputFile, writerOptions))
{ writer =>
+ val batch = schema.createRowBatch()
+ val tsCol = batch.cols(0).asInstanceOf[TimestampColumnVector]
+ batch.size = 1
+ tsCol.time(0) = 0L
+ tsCol.nanos(0) = 123000000
+ writer.addRowBatch(batch)
+ }
+
+ val readBack = spark.read.orc(dir.getCanonicalPath)
+ assert(readBack.schema("ts").dataType === TimestampType)
+ assert(readBack.count() === 1)
+ }
+ }
+ }
+
withAllNativeOrcReaders {
Seq(true, false).foreach { vecReaderNestedColEnabled =>
val vecReaderEnabled = SQLConf.get.orcVectorizedReaderEnabled
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcTest.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcTest.scala
index 7c340746413c..da6e6b11969a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcTest.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcTest.scala
@@ -18,11 +18,12 @@
package org.apache.spark.sql.execution.datasources.orc
import java.io.File
+import java.time.{LocalDateTime, ZoneOffset}
import scala.reflect.ClassTag
import scala.reflect.runtime.universe.TypeTag
-import org.apache.spark.sql.{Column, DataFrame, QueryTest}
+import org.apache.spark.sql.{Column, DataFrame, QueryTest, Row}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Predicate}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.classic.ClassicConversions._
@@ -31,6 +32,7 @@ import
org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan
import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.ORC_IMPLEMENTATION
+import org.apache.spark.sql.types.{DataType, StructType,
TimestampLTZNanosType, TimestampNTZNanosType}
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
@@ -152,6 +154,19 @@ trait OrcTest extends QueryTest with
FileBasedDataSourceTest {
withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> "true")(code)
}
+ // Builds a single-column ("ts") DataFrame from external java.time values,
letting the schema
+ // precision truncate the sub-microsecond digits: an NTZ column takes
java.time.LocalDateTime
+ // values, an LTZ column takes the same wall clocks as java.time.Instant at
UTC.
+ protected def nanosTimestampDf(nanosType: DataType, wallClocks:
Seq[LocalDateTime]): DataFrame = {
+ val values: Seq[Any] = nanosType match {
+ case _: TimestampNTZNanosType => wallClocks
+ case _: TimestampLTZNanosType =>
wallClocks.map(_.toInstant(ZoneOffset.UTC))
+ }
+ spark.createDataFrame(
+ spark.sparkContext.parallelize(values.map(Row(_))),
+ new StructType().add("ts", nanosType))
+ }
+
/**
* Takes a sequence of products `data` to generate multi-level nested
* dataframes as new test data. It tests both non-nested and nested
dataframes
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
index bf40327bd991..49b9cc798a1b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
@@ -38,6 +38,7 @@ import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.execution.datasources.DaysWritable
import org.apache.spark.sql.types
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.TimestampNanosVal
import org.apache.spark.unsafe.types.UTF8String
/**
@@ -330,8 +331,22 @@ private[hive] trait HiveInspectors {
withNullSafe(o =>
DateTimeUtils.toJavaDate(o.asInstanceOf[Int]))
case _: JavaTimestampObjectInspector =>
- withNullSafe(o =>
- DateTimeUtils.toJavaTimestamp(o.asInstanceOf[Long]))
+ withNullSafe(o => dataType match {
+ case _: TimestampLTZNanosType =>
+ o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.from(
+ DateTimeUtils.timestampNanosToInstant(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ }
+ case _: TimestampNTZNanosType =>
+ o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.valueOf(
+ DateTimeUtils.timestampNanosToLocalDateTime(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ }
+ case _ =>
+ DateTimeUtils.toJavaTimestamp(o.asInstanceOf[Long])
+ })
case _: HiveDecimalObjectInspector if x.preferWritable() =>
withNullSafe(o => getDecimalWritable(o.asInstanceOf[Decimal]))
case _: HiveDecimalObjectInspector =>
@@ -346,9 +361,39 @@ private[hive] trait HiveInspectors {
case _: DateObjectInspector =>
withNullSafe(o => DateTimeUtils.toJavaDate(o.asInstanceOf[Int]))
case _: TimestampObjectInspector if x.preferWritable() =>
- withNullSafe(o => getTimestampWritable(o))
+ withNullSafe(o => dataType match {
+ case _: TimestampLTZNanosType =>
+ new hiveIo.TimestampWritable(o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.from(
+ DateTimeUtils.timestampNanosToInstant(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ })
+ case _: TimestampNTZNanosType =>
+ new hiveIo.TimestampWritable(o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.valueOf(
+ DateTimeUtils.timestampNanosToLocalDateTime(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ })
+ case _ =>
+ getTimestampWritable(o)
+ })
case _: TimestampObjectInspector =>
- withNullSafe(o => DateTimeUtils.toJavaTimestamp(o.asInstanceOf[Long]))
+ withNullSafe(o => dataType match {
+ case _: TimestampLTZNanosType =>
+ o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.from(
+ DateTimeUtils.timestampNanosToInstant(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ }
+ case _: TimestampNTZNanosType =>
+ o match {
+ case v: TimestampNanosVal => java.sql.Timestamp.valueOf(
+ DateTimeUtils.timestampNanosToLocalDateTime(v))
+ case micros: Long => DateTimeUtils.toJavaTimestamp(micros)
+ }
+ case _ =>
+ DateTimeUtils.toJavaTimestamp(o.asInstanceOf[Long])
+ })
case _: HiveIntervalDayTimeObjectInspector if x.preferWritable() =>
withNullSafe(o => getHiveIntervalDayTimeWritable(o))
case _: HiveIntervalDayTimeObjectInspector =>
@@ -752,6 +797,73 @@ private[hive] trait HiveInspectors {
}
}
+ /**
+ * Returns an unwrapper that converts a Hive value into a Catalyst value,
using the target
+ * Catalyst `dataType` to preserve nanosecond timestamp precision. The plain
+ * `unwrapperFor(ObjectInspector)` cannot do this because a Hive
`TimestampObjectInspector`
+ * maps to micros by default; here the nanos timestamp types are produced as
`TimestampNanosVal`,
+ * recursing through array/map/struct so nested nanos timestamps round-trip
correctly. Any other
+ * type is delegated to the `ObjectInspector`-only overload.
+ */
+ def unwrapperFor(objectInspector: ObjectInspector, dataType: DataType): Any
=> Any =
+ (objectInspector, dataType) match {
+ case (ti: TimestampObjectInspector, t: TimestampNTZNanosType) =>
+ data: Any => {
+ if (data != null) {
+ DateTimeUtils.localDateTimeToTimestampNanos(
+ ti.getPrimitiveJavaObject(data).toLocalDateTime, t.precision)
+ } else {
+ null
+ }
+ }
+ case (ti: TimestampObjectInspector, t: TimestampLTZNanosType) =>
+ data: Any => {
+ if (data != null) {
+ DateTimeUtils.instantToTimestampNanos(
+ ti.getPrimitiveJavaObject(data).toInstant, t.precision)
+ } else {
+ null
+ }
+ }
+ case (li: ListObjectInspector, ArrayType(elementType, _)) =>
+ val unwrapper = unwrapperFor(li.getListElementObjectInspector,
elementType)
+ data: Any => {
+ if (data != null) {
+ Option(li.getList(data))
+ .map(l => new GenericArrayData(l.asScala.map(unwrapper).toArray))
+ .orNull
+ } else {
+ null
+ }
+ }
+ case (mi: MapObjectInspector, MapType(keyType, valueType, _)) =>
+ val keyUnwrapper = unwrapperFor(mi.getMapKeyObjectInspector, keyType)
+ val valueUnwrapper = unwrapperFor(mi.getMapValueObjectInspector,
valueType)
+ data: Any => {
+ if (data != null) {
+ val map = mi.getMap(data)
+ if (map == null) null else ArrayBasedMapData(map, keyUnwrapper,
valueUnwrapper)
+ } else {
+ null
+ }
+ }
+ case (si: StructObjectInspector, st: StructType) =>
+ val fields = si.getAllStructFieldRefs.asScala
+ val unwrappers = fields.zip(st.fields).map { case (field, structField)
=>
+ val unwrapper = unwrapperFor(field.getFieldObjectInspector,
structField.dataType)
+ data: Any => unwrapper(si.getStructFieldData(data, field))
+ }
+ data: Any => {
+ if (data != null) {
+ new GenericInternalRow(unwrappers.map(_(data)).toArray)
+ } else {
+ null
+ }
+ }
+ case _ =>
+ unwrapperFor(objectInspector)
+ }
+
/**
* Builds unwrappers ahead of time according to object inspector
* types to avoid pattern matching and branching costs per row.
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
index d6f62284d2e7..fedf2e51ffc3 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
@@ -45,6 +45,7 @@ import org.apache.spark.sql.catalyst.analysis.CastSupport
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{TimestampLTZNanosType,
TimestampNTZNanosType}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.{SerializableConfiguration, Utils}
import org.apache.spark.util.ArrayImplicits._
@@ -461,8 +462,8 @@ private[hive] object HadoopTableReader extends
HiveInspectors with Logging {
* Builds specific unwrappers ahead of time according to object inspector
* types to avoid pattern matching and branching costs per row.
*/
- val unwrappers: Seq[(Any, InternalRow, Int) => Unit] = fieldRefs.map {
- _.getFieldObjectInspector match {
+ val unwrappers: Seq[(Any, InternalRow, Int) => Unit] =
fieldRefs.zip(nonPartitionKeyAttrs).map {
+ case (fieldRef, (attr, _)) => fieldRef.getFieldObjectInspector match {
case oi: BooleanObjectInspector =>
(value: Any, row: InternalRow, ordinal: Int) =>
row.setBoolean(ordinal, oi.get(value))
case oi: ByteObjectInspector =>
@@ -486,9 +487,18 @@ private[hive] object HadoopTableReader extends
HiveInspectors with Logging {
case oi: HiveDecimalObjectInspector =>
(value: Any, row: InternalRow, ordinal: Int) =>
row.update(ordinal, HiveShim.toCatalystDecimal(oi, value))
+ case oi: TimestampObjectInspector
+ if attr.dataType.isInstanceOf[TimestampLTZNanosType] ||
+ attr.dataType.isInstanceOf[TimestampNTZNanosType] =>
+ // Nanos timestamps need the target Catalyst type to produce
TimestampNanosVal; reuse
+ // the data-type-aware unwrapper instead of duplicating the
conversion here.
+ val unwrapper = unwrapperFor(oi, attr.dataType)
+ (value: Any, row: InternalRow, ordinal: Int) => row.update(ordinal,
unwrapper(value))
case oi: TimestampObjectInspector =>
(value: Any, row: InternalRow, ordinal: Int) =>
- row.setLong(ordinal,
DateTimeUtils.fromJavaTimestamp(oi.getPrimitiveJavaObject(value)))
+ row.setLong(
+ ordinal,
+
DateTimeUtils.fromJavaTimestamp(oi.getPrimitiveJavaObject(value)))
case oi: DateObjectInspector =>
(value: Any, row: InternalRow, ordinal: Int) =>
row.setInt(ordinal,
DateTimeUtils.fromJavaDate(oi.getPrimitiveJavaObject(value)))
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
index 6c3fbb5a8314..04857cff9d9b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
@@ -197,8 +197,6 @@ case class OrcFileFormat() extends FileFormat
case _: AnsiIntervalType => false
case _: TimeType => false
- // Nanosecond-capable timestamps are not yet supported by this datasource.
- case _: AnyTimestampNanoType => false
case _: AtomicType => true
case st: StructType => st.forall { f => supportDataType(f.dataType) }
@@ -226,6 +224,26 @@ case class OrcFileFormat() extends FileFormat
private[orc] class OrcSerializer(dataSchema: StructType, conf: Configuration)
extends HiveInspectors {
+ private def toHiveCompatibleDataType(dataType: DataType): DataType = {
+ dataType match {
+ case _: TimestampNTZNanosType | _: TimestampLTZNanosType =>
+ TimestampType
+ case StructType(fields) =>
+ StructType(fields.map(f => f.copy(dataType =
toHiveCompatibleDataType(f.dataType))))
+ case ArrayType(elementType, containsNull) =>
+ ArrayType(toHiveCompatibleDataType(elementType), containsNull)
+ case MapType(keyType, valueType, valueContainsNull) =>
+ MapType(
+ toHiveCompatibleDataType(keyType),
+ toHiveCompatibleDataType(valueType),
+ valueContainsNull)
+ case other =>
+ other
+ }
+ }
+
+ private[this] val hiveCompatibleSchema = StructType(
+ dataSchema.map(f => f.copy(dataType =
toHiveCompatibleDataType(f.dataType))))
def serialize(row: InternalRow): Writable = {
wrapOrcStruct(cachedOrcStruct, structOI, row)
@@ -235,7 +253,9 @@ private[orc] class OrcSerializer(dataSchema: StructType,
conf: Configuration)
private[this] val serializer = {
val table = new Properties()
table.setProperty("columns", dataSchema.fieldNames.mkString(","))
- table.setProperty("columns.types",
dataSchema.map(_.dataType.catalogString).mkString(":"))
+ table.setProperty(
+ "columns.types",
+ hiveCompatibleSchema.map(_.dataType.catalogString).mkString(":"))
val serde = new OrcSerde
serde.initialize(conf, table)
@@ -244,7 +264,7 @@ private[orc] class OrcSerializer(dataSchema: StructType,
conf: Configuration)
// Object inspector converted from the schema of the relation to be
serialized.
val structOI = {
- val typeInfo =
TypeInfoUtils.getTypeInfoFromTypeString(dataSchema.catalogString)
+ val typeInfo =
TypeInfoUtils.getTypeInfoFromTypeString(hiveCompatibleSchema.catalogString)
OrcStruct.createObjectInspector(typeInfo.asInstanceOf[StructTypeInfo])
.asInstanceOf[SettableStructObjectInspector]
}
@@ -337,6 +357,8 @@ private[orc] object OrcFileFormat extends HiveInspectors
with Logging {
val unsafeProjection = UnsafeProjection.create(requiredSchema)
val forcePositionalEvolution =
OrcConf.FORCE_POSITIONAL_EVOLUTION.getBoolean(conf)
+ def isNanosTimestamp(dt: DataType): Boolean =
dt.isInstanceOf[AnyTimestampNanoType]
+
def unwrap(oi: StructObjectInspector): Iterator[InternalRow] = {
val (fieldRefs, fieldOrdinals) = requiredSchema.zipWithIndex.map {
case (field, ordinal) =>
@@ -352,7 +374,17 @@ private[orc] object OrcFileFormat extends HiveInspectors
with Logging {
ref -> ordinal
}.unzip
- val unwrappers = fieldRefs.map(r => if (r == null) null else
unwrapperFor(r))
+ val unwrappers = fieldRefs.zip(requiredSchema).map {
+ case (null, _) => null
+ // Nanos timestamps (including those nested in struct/array/map) need
the target Catalyst
+ // type to produce TimestampNanosVal; the data-type-aware unwrapper
handles both the
+ // top-level and nested cases, while everything else keeps the
primitive fast paths.
+ case (fieldRef, field) if
field.dataType.existsRecursively(isNanosTimestamp) =>
+ val unwrapper = unwrapperFor(fieldRef.getFieldObjectInspector,
field.dataType)
+ (value: Any, row: InternalRow, ordinal: Int) => row.update(ordinal,
unwrapper(value))
+ case (fieldRef, _) =>
+ unwrapperFor(fieldRef)
+ }
iterator.map { value =>
val raw = deserializer.deserialize(value)
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
index 504d1a5881a0..96ac15893d07 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
@@ -18,17 +18,17 @@
package org.apache.spark.sql.hive.orc
import java.io.File
+import java.time.LocalDateTime
-import org.apache.spark.sql.{AnalysisException, Column, Row}
+import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.TestingUDT.{IntervalData, IntervalUDT}
-import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.classic.ClassicConversions._
+import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
+import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
import org.apache.spark.sql.execution.datasources.orc.OrcSuite
import org.apache.spark.sql.hive.HiveUtils
import org.apache.spark.sql.hive.test.TestHiveSingleton
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.TimestampNanosVal
import org.apache.spark.util.Utils
class HiveOrcSourceSuite extends OrcSuite with TestHiveSingleton {
@@ -350,39 +350,104 @@ class HiveOrcSourceSuite extends OrcSuite with
TestHiveSingleton {
}
}
- test("SPARK-57166: nanosecond timestamp types are not supported in Hive
ORC") {
- val nanosTypes = Seq(TimestampNTZNanosType(9), TimestampLTZNanosType(9))
+ test("SPARK-57166: nanosecond timestamp types are supported in Hive ORC") {
withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true") {
- nanosTypes.foreach { nanosType =>
- val expectedType = s""""${nanosType.sql}""""
+ val wallClock = Seq(LocalDateTime.of(1970, 1, 1, 0, 0, 0, 0))
+ Seq(true, false).foreach { convertMetastore =>
+ withSQLConf(HiveUtils.CONVERT_METASTORE_ORC.key ->
s"$convertMetastore") {
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempDir { dir =>
+ val df = nanosTimestampDf(nanosType, wallClock)
+ val path = new File(dir, "nanos").getCanonicalPath
+ df.write.format("orc").mode("overwrite").save(path)
+
+ val readBack = spark.read.schema(new StructType().add("ts",
nanosType))
+ .format("orc").load(path)
+ checkAnswer(readBack, df)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: Hive ORC serde nanos round-trip; NTZ time-zone
independent") {
+ // Force spark.sql.orc.impl=hive so the Hive serde write/read conversion
is exercised
+ // (HiveInspectors.wrapperFor on write, the hive OrcFileFormat unwrappers
on read) rather
+ // than the native datasource that the other nanos tests use.
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.ORC_IMPLEMENTATION.key -> "hive") {
+ // A mid-range value plus the documented min/max ends
+ // [0001-01-01T00:00:00, 9999-12-31T23:59:59.999999999] (at UTC for LTZ).
+ val wallClocks = Seq(
+ LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123),
+ LocalDateTime.of(1, 1, 1, 0, 0, 0, 0),
+ LocalDateTime.of(9999, 12, 31, 23, 59, 59, 999999999))
+ foreachNanosPrecision { precision =>
+ // Same-zone round trip through the Hive serde path for both nanos
types, including the
+ // min and max of the documented range.
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ val input = nanosTimestampDf(nanosType, wallClocks)
+ withTempDir { dir =>
+ val path = new File(dir, "nanos").getCanonicalPath
+ input.write.format("orc").mode("overwrite").save(path)
+ checkAnswer(
+ spark.read.schema(new StructType().add("ts",
nanosType)).format("orc").load(path),
+ input)
+ }
+ }
+
+ // The NTZ wall clock stays zone-independent across a JVM default
time-zone change. Hive
+ // ORC stores zone-naive wall-clock fields, so the instant-based LTZ
type is not
+ // zone-stable through the Hive serde path -- the same caveat as the
legacy TimestampType
+ // -- so LTZ is only round-tripped within a single zone above.
+ val ntzType = TimestampNTZNanosType(precision)
+ val ntzInput = nanosTimestampDf(ntzType, wallClocks.take(1))
+ val ntzExpected = ntzInput.collect()
withTempDir { dir =>
- // Write path
- val nanosLiteral = Literal.create(new TimestampNanosVal(0L,
0.toShort), nanosType)
- val df = spark.range(1).select(Column(nanosLiteral).as("ts"))
- val writeDir = new File(dir, "write").getCanonicalPath
- checkError(
- exception = intercept[AnalysisException] {
- df.write.format("orc").mode("overwrite").save(writeDir)
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> "ORC"))
-
- // Read path
- val readDir = new File(dir, "read").getCanonicalPath
- spark.range(1).write.format("orc").mode("overwrite").save(readDir)
- checkError(
- exception = intercept[AnalysisException] {
- spark.read.schema(new StructType().add("ts", nanosType))
- .format("orc").load(readDir).collect()
- },
- condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
- parameters = Map(
- "columnName" -> "`ts`",
- "columnType" -> expectedType,
- "format" -> "ORC"))
+ val path = new File(dir, "ntz-tz").getCanonicalPath
+ DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.LA) {
+ ntzInput.write.format("orc").mode("overwrite").save(path)
+ }
+ DateTimeTestUtils.withDefaultTimeZone(DateTimeTestUtils.UTC) {
+ checkAnswer(
+ spark.read.schema(new StructType().add("ts",
ntzType)).format("orc").load(path),
+ ntzExpected)
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57455: Hive ORC serde round-trips nanos timestamps in
nested/complex types") {
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.ORC_IMPLEMENTATION.key -> "hive") {
+ val wallClocks = Seq(LocalDateTime.of(1970, 1, 1, 0, 20, 34, 567890123))
+ foreachNanosPrecision { precision =>
+ Seq(TimestampNTZNanosType(precision),
TimestampLTZNanosType(precision)).foreach {
+ nanosType =>
+ withTempView("nanos_input") {
+ nanosTimestampDf(nanosType,
wallClocks).createOrReplaceTempView("nanos_input")
+ val nested = sql(
+ """SELECT
+ | named_struct('ts', ts) AS struct_ts,
+ | array(ts) AS array_ts,
+ | map('k', ts) AS map_ts
+ |FROM nanos_input
+ |""".stripMargin)
+ withTempDir { dir =>
+ val path = new File(dir, "nanos-nested").getCanonicalPath
+ nested.write.format("orc").mode("overwrite").save(path)
+ val readBack =
spark.read.schema(nested.schema).format("orc").load(path)
+ checkAnswer(readBack, nested)
+ }
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
