MaxGekk commented on code in PR #56407:
URL: https://github.com/apache/spark/pull/56407#discussion_r3384058025
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala:
##########
@@ -268,6 +275,14 @@ class ParquetWriteSupport extends
WriteSupport[InternalRow] with Logging {
// MICROS time unit.
(row: SpecializedGetters, ordinal: Int) =>
recordConsumer.addLong(row.getLong(ordinal))
+ case _: TimestampLTZNanosType =>
Review Comment:
The LTZ MICROS write path (line 262-263) applies
`timestampRebaseFunc(micros)` before writing. This path does not apply any
rebase to `epochMicros`, while the LTZ nanos **read** path does apply
`timestampRebaseFunc(epochMicros)` (`ParquetRowConverter.scala:496-497`). In
the default `CORRECTED` mode `timestampRebaseFunc` is `identity`, so there is
no observable difference in practice. But in `LEGACY` mode (user-configured),
write stores raw Gregorian `epochMicros * 1000 + nanosWithinMicro` while read
applies Julian→Gregorian conversion, silently corrupting round-trips for
pre-Gregorian dates.
The `PARQUET_REBASE_MODE_IN_WRITE` config doc lists DATE / TIMESTAMP_MILLIS
/ TIMESTAMP_MICROS as influenced types and omits NANOS, so the omission here
may be intentional. But if NANOS is always Gregorian and exempt from rebasing,
the read path (`ParquetRowConverter.scala:497`) should also not call
`timestampRebaseFunc`. The current state is inconsistent: write skips the
rebase, read applies it. Could you confirm the intent? Either apply
`timestampRebaseFunc` on write too (full LEGACY-mode symmetry), or explicitly
skip it on read and document why NANOS is exempt.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.example.data.simple.SimpleGroupFactory
+import org.apache.parquet.hadoop.ParquetFileWriter.Mode
+import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.schema.{LogicalTypeAnnotation, MessageType, Types}
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with
SharedSparkSession {
+
+ private def withNanosEnabled(f: => Unit): Unit = {
+ withSQLConf(
+ SQLConf.TYPES_FRAMEWORK_ENABLED.key -> "true",
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true")(f)
+ }
+
+ private def hasCause(t: Throwable, clazz: Class[_]): Boolean = {
+ var cur = t
+ while (cur != null) {
+ if (clazz.isInstance(cur)) return true
+ cur = cur.getCause
+ }
+ false
+ }
+
+ private def writeForeignNanosParquet(
+ file: File,
+ isAdjustedToUTC: Boolean,
+ values: Seq[Option[Long]]): Unit = {
+ val schema: MessageType = Types.buildMessage()
+ .optional(INT64)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named("ts")
+ .named("spark_schema")
+ val conf = spark.sessionState.newHadoopConf()
+ val writer = ExampleParquetWriter.builder(new Path(file.toURI))
+ .withType(schema)
+ .withConf(conf)
+ .withWriteMode(Mode.OVERWRITE)
+ .build()
+ try {
+ val factory = new SimpleGroupFactory(schema)
+ values.foreach { v =>
+ val group = factory.newGroup()
+ v.foreach(x => group.add("ts", x))
+ writer.write(group)
+ }
+ } finally {
+ writer.close()
+ }
+ }
+
+ test("SPARK-57102: Spark write/read round-trips nanos value and precision") {
+ withNanosEnabled {
+ Seq("true", "false").foreach { vectorized =>
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized) {
+ Seq(7, 8, 9).foreach { p =>
+ withTempPath { dir =>
+ val frac = "123456789".take(p)
+ val df = spark.sql(
+ s"""SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '2020-01-01 12:34:56.$frac',
+ | TIMESTAMP_LTZ '2020-01-01 12:34:56.$frac'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.$frac',
+ | TIMESTAMP_LTZ '1969-12-31 23:59:59.$frac'),
+ | (CAST(NULL AS TIMESTAMP_NTZ($p)), CAST(NULL AS
TIMESTAMP_LTZ($p)))
+ | AS t(ntz, ltz)""".stripMargin)
+ assert(df.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(df.schema("ltz").dataType === TimestampLTZNanosType(p))
+
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+
+ assert(read.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(read.schema("ltz").dataType === TimestampLTZNanosType(p))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: read a foreign TIMESTAMP(NANOS) file as nanosecond
timestamp types") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val values = Seq(Some(123L), Some(1000000123L), Some(-1L), None)
+
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_NTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_LTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: legacy nanosAsLong reads a foreign TIMESTAMP(NANOS) file
as LongType") {
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, Seq(Some(123L),
Some(-1L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === LongType)
+ checkAnswer(read, Seq(Row(123L), Row(-1L)))
+ }
+ }
+ }
+
+ test("SPARK-57102: reading a foreign TIMESTAMP(NANOS) file fails when the
feature is disabled") {
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, Seq(Some(123L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "false") {
+ val e = intercept[AnalysisException] {
+ spark.read.parquet(file.getCanonicalPath).schema
+ }
+ assert(e.getMessage.contains("NANOS") ||
e.getMessage.contains("Parquet"))
+ }
+ }
+ }
+
+ test("SPARK-57102: writing a timestamp outside the INT64 epoch-nanos range
fails loudly") {
+ withNanosEnabled {
+ withTempPath { dir =>
+ val df = spark.sql("SELECT TIMESTAMP_NTZ '9999-12-31
23:59:59.999999999' AS ntz")
Review Comment:
`timestampNanosToEpochNanos` is called for both LTZ and NTZ writes via the
same shared helper. This test exercises only `TIMESTAMP_NTZ`. Add a symmetric
`TIMESTAMP_LTZ '9999-12-31 23:59:59.999999999'` case to the same test.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala:
##########
@@ -1111,13 +1111,15 @@ class ParquetSchemaSuite extends ParquetSchemaTest {
test("SPARK-40819: parquet file with TIMESTAMP(NANOS, true) (with default
nanosAsLong=false)") {
val testDataPath = testFile("test-data/timestamp-nanos.parquet")
- checkError(
- exception = intercept[AnalysisException] {
- spark.read.parquet(testDataPath).collect()
- },
- condition = "PARQUET_TYPE_ILLEGAL",
- parameters = Map("parquetType" -> "INT64 (TIMESTAMP(NANOS,true))")
- )
+ withSQLConf(SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false") {
Review Comment:
The `TimestampNTZType` analogue has `testSchema`, `testCatalystToParquet`,
and `testParquetToCatalyst` cases in `ParquetSchemaSuite` (lines 2604-2633)
that document the exact type-mapping contract (Spark type ↔ Parquet schema ↔
logical annotation). The new nanos types have none here;
`ParquetTimestampNanosSuite` covers full round-trips but not schema-level
conversion in isolation. Consider adding analogous cases — e.g.:
```
testSchema("TimestampNTZNanos written as INT64 TIMESTAMP(NANOS,false)",
StructType(Seq(StructField("f1", TimestampNTZNanosType(9)))),
"""message root { optional INT64 f1 (TIMESTAMP(NANOS,false)); }""",
binaryAsString = true, int96AsTimestamp = true, writeLegacyParquetFormat =
false,
timestampNanosTypesEnabled = true)
```
(The helper would need `timestampNanosTypesEnabled` wired through, but the
pattern mirrors the existing NTZ tests.)
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.execution.datasources.parquet
+
+import java.io.File
+
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.example.data.simple.SimpleGroupFactory
+import org.apache.parquet.hadoop.ParquetFileWriter.Mode
+import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.schema.{LogicalTypeAnnotation, MessageType, Types}
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with
SharedSparkSession {
+
+ private def withNanosEnabled(f: => Unit): Unit = {
+ withSQLConf(
+ SQLConf.TYPES_FRAMEWORK_ENABLED.key -> "true",
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true")(f)
+ }
+
+ private def hasCause(t: Throwable, clazz: Class[_]): Boolean = {
+ var cur = t
+ while (cur != null) {
+ if (clazz.isInstance(cur)) return true
+ cur = cur.getCause
+ }
+ false
+ }
+
+ private def writeForeignNanosParquet(
+ file: File,
+ isAdjustedToUTC: Boolean,
+ values: Seq[Option[Long]]): Unit = {
+ val schema: MessageType = Types.buildMessage()
+ .optional(INT64)
+ .as(LogicalTypeAnnotation.timestampType(isAdjustedToUTC, TimeUnit.NANOS))
+ .named("ts")
+ .named("spark_schema")
+ val conf = spark.sessionState.newHadoopConf()
+ val writer = ExampleParquetWriter.builder(new Path(file.toURI))
+ .withType(schema)
+ .withConf(conf)
+ .withWriteMode(Mode.OVERWRITE)
+ .build()
+ try {
+ val factory = new SimpleGroupFactory(schema)
+ values.foreach { v =>
+ val group = factory.newGroup()
+ v.foreach(x => group.add("ts", x))
+ writer.write(group)
+ }
+ } finally {
+ writer.close()
+ }
+ }
+
+ test("SPARK-57102: Spark write/read round-trips nanos value and precision") {
+ withNanosEnabled {
+ Seq("true", "false").foreach { vectorized =>
+ withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized) {
+ Seq(7, 8, 9).foreach { p =>
+ withTempPath { dir =>
+ val frac = "123456789".take(p)
+ val df = spark.sql(
+ s"""SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '2020-01-01 12:34:56.$frac',
+ | TIMESTAMP_LTZ '2020-01-01 12:34:56.$frac'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.$frac',
+ | TIMESTAMP_LTZ '1969-12-31 23:59:59.$frac'),
+ | (CAST(NULL AS TIMESTAMP_NTZ($p)), CAST(NULL AS
TIMESTAMP_LTZ($p)))
+ | AS t(ntz, ltz)""".stripMargin)
+ assert(df.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(df.schema("ltz").dataType === TimestampLTZNanosType(p))
+
+ df.write.parquet(dir.getCanonicalPath)
+ val read = spark.read.parquet(dir.getCanonicalPath)
+
+ assert(read.schema("ntz").dataType === TimestampNTZNanosType(p))
+ assert(read.schema("ltz").dataType === TimestampLTZNanosType(p))
+ checkAnswer(read, df.collect().toSeq)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: read a foreign TIMESTAMP(NANOS) file as nanosecond
timestamp types") {
+ withNanosEnabled {
+ withSQLConf(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val values = Seq(Some(123L), Some(1000000123L), Some(-1L), None)
+
+ withTempPath { dir =>
+ val file = new File(dir, "ntz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampNTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_NTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_NTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_NTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+
+ withTempPath { dir =>
+ val file = new File(dir, "ltz.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = true, values)
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === TimestampLTZNanosType(9))
+ checkAnswer(read, spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:00.000000123'),
+ | (TIMESTAMP_LTZ '1970-01-01 00:00:01.000000123'),
+ | (TIMESTAMP_LTZ '1969-12-31 23:59:59.999999999'),
+ | (CAST(NULL AS TIMESTAMP_LTZ(9)))
+ | AS t(ts)""".stripMargin).collect().toSeq)
+ }
+ }
+ }
+ }
+
+ test("SPARK-57102: legacy nanosAsLong reads a foreign TIMESTAMP(NANOS) file
as LongType") {
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, Seq(Some(123L),
Some(-1L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
+ val read = spark.read.parquet(file.getCanonicalPath)
+ assert(read.schema("ts").dataType === LongType)
+ checkAnswer(read, Seq(Row(123L), Row(-1L)))
+ }
+ }
+ }
+
+ test("SPARK-57102: reading a foreign TIMESTAMP(NANOS) file fails when the
feature is disabled") {
+ withTempPath { dir =>
+ val file = new File(dir, "foreign.parquet")
+ writeForeignNanosParquet(file, isAdjustedToUTC = false, Seq(Some(123L)))
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false",
+ SQLConf.LEGACY_PARQUET_NANOS_AS_LONG.key -> "false") {
+ val e = intercept[AnalysisException] {
+ spark.read.parquet(file.getCanonicalPath).schema
+ }
+ assert(e.getMessage.contains("NANOS") ||
e.getMessage.contains("Parquet"))
Review Comment:
This assertion is weaker than the analogous test in `ParquetSchemaSuite`
(`withSQLConf(TIMESTAMP_NANOS_TYPES_ENABLED.key -> "false")` wrapping
`checkError(..., condition = "PARQUET_TYPE_ILLEGAL", parameters =
Map("parquetType" -> "INT64 (TIMESTAMP(NANOS,true))"))`). Use `checkError` with
the specific condition code so the test pins the error condition, not just a
keyword in the message.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala:
##########
@@ -190,6 +191,12 @@ class ParquetWriteSupport extends
WriteSupport[InternalRow] with Logging {
// `inShredded` indicates whether the current traversal is nested within a
shredded Variant
// schema. This affects how timestamp values are written.
+ private def timestampNanosToEpochNanos(value: TimestampNanosVal): Long = {
Review Comment:
The comment on line 192 (`// \`inShredded\` indicates whether the current
traversal is nested within a shredded Variant schema`) was the doc-comment for
`makeWriter(dataType, inShredded)` in the base branch. This PR inserted
`timestampNanosToEpochNanos` between the comment and `makeWriter`, so the
comment now reads as describing the wrong method (`timestampNanosToEpochNanos`
has no `inShredded` parameter).
Move the comment to sit immediately before `makeWriter`:
```suggestion
private def timestampNanosToEpochNanos(value: TimestampNanosVal): Long = {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
