You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/30 00:25:29 UTC
(datafusion-comet) 01/01: fix: fall back from the native CSV scan for timestamps outside UTC (#6349)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6349-c7dd26a9c1c05041c4d60465ce31ef147d25b48c
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 0ada70847133eba56112aac16d2f92528b00ce22
Author: Andy Grove <ag...@apache.org>
AuthorDate: Tue Sep 29 23:23:44 2026 +0000
fix: fall back from the native CSV scan for timestamps outside UTC (#6349)
The native CSV V2 reader hands arrow-csv the Spark schema, whose TimestampType is labelled UTC, so a timestamp without an offset is parsed as UTC. Spark parses it in the CSV timeZone option, which defaults to the session timezone. Fall back when the read schema has a timestamp column and that timezone is not UTC.
Closes #6332.
---
.../org/apache/comet/rules/CometScanRule.scala | 16 ++++++++--
.../apache/comet/csv/CometCsvNativeReadSuite.scala | 35 +++++++++++++++++++++-
2 files changed, 48 insertions(+), 3 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index f015ed70f8..373541d63a 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -21,6 +21,7 @@ package org.apache.comet.rules
import java.lang.{Boolean => JBoolean}
import java.net.URI
+import java.time.ZoneOffset
import java.util.Locale
import java.util.concurrent.ConcurrentHashMap
@@ -33,7 +34,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, GenericArrayData, MetadataColumnHelper}
+import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, DateTimeUtils, GenericArrayData, MetadataColumnHelper}
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec}
import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec}
@@ -435,8 +436,19 @@ case class CometScanRule(session: SparkSession)
fallbackReasons +=
s"Comet supports only single-character delimiters, but got: '$delimiter'"
}
+ // The native reader parses a timestamp without an offset as UTC. Spark parses it in the
+ // CSV `timeZone` option, which defaults to the session timezone.
+ val timeZone = Option(scan.options.get(DateTimeUtils.TIMEZONE_OPTION))
+ .getOrElse(SQLConf.get.sessionLocalTimeZone)
+ val parsesTimestampsLikeSpark =
+ !scan.readDataSchema.exists(_.dataType == TimestampType) ||
+ DateTimeUtils.getZoneId(timeZone).normalized() == ZoneOffset.UTC
+ if (!parsesTimestampsLikeSpark) {
+ fallbackReasons += "Comet's native CSV reader parses timestamps in UTC, but the CSV " +
+ s"timezone is $timeZone"
+ }
if (schemaSupported && partitionSchemaSupported && containsCorruptedRecordsColumn
- && !isInferSchemaEnabled && isSingleCharacterDelimiter) {
+ && !isInferSchemaEnabled && isSingleCharacterDelimiter && parsesTimestampsLikeSpark) {
CometBatchScanExec(
scanExec.clone().asInstanceOf[BatchScanExec],
runtimeFilters = scanExec.runtimeFilters)
diff --git a/spark/src/test/scala/org/apache/comet/csv/CometCsvNativeReadSuite.scala b/spark/src/test/scala/org/apache/comet/csv/CometCsvNativeReadSuite.scala
index e9a18a18dc..7ac8faddaf 100644
--- a/spark/src/test/scala/org/apache/comet/csv/CometCsvNativeReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/csv/CometCsvNativeReadSuite.scala
@@ -19,9 +19,12 @@
package org.apache.comet.csv
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
import org.apache.spark.sql.CometTestBase
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType, TimestampType}
import org.apache.comet.CometConf
@@ -86,4 +89,34 @@ class CometCsvNativeReadSuite extends CometTestBase {
"Comet supports only single-character delimiters, but got: ',,'")
}
}
+
+ test("Native csv read - timestamps fall back unless the CSV timezone is UTC") {
+ withTempDir { dir =>
+ Files.write(
+ dir.toPath.resolve("part-0.csv"),
+ "id,ts\n0,2024-01-15 18:30:45\n1,2024-06-30T23:30:00\n".getBytes(StandardCharsets.UTF_8))
+ val schema = new StructType().add("id", IntegerType).add("ts", TimestampType)
+ def read(options: Map[String, String] = Map.empty) =
+ spark.read.option("header", "true").options(options).schema(schema).csv(dir.toString)
+
+ withSQLConf(
+ CometConf.COMET_CSV_V2_NATIVE_ENABLED.key -> "true",
+ SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ checkSparkAnswerAndFallbackReason(
+ read(),
+ "Comet's native CSV reader parses timestamps in UTC, but the CSV timezone is " +
+ "America/Los_Angeles")
+ checkSparkAnswerAndOperator(read(Map("timeZone" -> "UTC")))
+ }
+ withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ checkSparkAnswerAndOperator(read())
+ checkSparkAnswerAndFallbackReason(
+ read(Map("timeZone" -> "Asia/Tokyo")),
+ "Comet's native CSV reader parses timestamps in UTC, but the CSV timezone is " +
+ "Asia/Tokyo")
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org