ump45nose commented on code in PR #7089:
URL: https://github.com/apache/datafusion-comet/pull/7089#discussion_r4332446299
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -504,15 +516,22 @@ fn mark_repeated_primitives(schema: &Schema, parquet:
&SchemaDescriptor) -> Sche
}
DataType::Map(entries, sorted) => DataType::Map(mark(entries,
columns, index), *sorted),
_ => {
- let repeated = columns.get(*index).is_some_and(|column| {
+ let column = columns.get(*index);
+ *index += 1;
+ let repeated = column.is_some_and(|column| {
column.self_type().get_basic_info().repetition()
== parquet::basic::Repetition::REPEATED
});
- *index += 1;
+ let nanos = column.is_some_and(is_nanos_timestamp);
Review Comment:
Thanks, confirmed. Fixed in a7eb8e9: `mark_parquet_leaves` now recurses into
`ListView` and `LargeListView`, matching `spark_enum_field`.
`int96_after_list_view_with_nanos_leaf_reads_as_timestamp` uses your shape (`a:
ListView/LargeListView<struct<x int, n TIMESTAMP(NANOS)>>`, then `t INT96`),
reads only `t` as timestamp with row filters off, and fails without the fix.
The JVM fixture now also has a `list_view` column followed by other leaves.
##########
native/core/src/parquet/eager_page_index_reader_factory.rs:
##########
@@ -734,7 +758,7 @@ impl AsyncFileReader for EagerPageIndexReader {
DataFusionError::External(source) =>
ParquetError::External(source),
other => ParquetError::External(Box::new(other)),
})?;
- if *row_filters {
+ if *row_filters || has_nanos {
Review Comment:
Updated both in a7eb8e9. The module doc and `with_conversion_check` now say
the deferred failure is also raised for a file with a `TIMESTAMP(NANOS)` leaf
when row filters are off.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -1824,6 +1824,70 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("native scan rejects TIMESTAMP(NANOS) read as timestamp like Spark") {
+ // Regression guard for
https://github.com/apache/datafusion-comet/issues/7034. Spark's
+ // writer cannot produce TIMESTAMP(NANOS), so timestamp_nanos.parquet was
written with
+ // PyArrow. Next to `id int`, it holds TIMESTAMP(NANOS) leaves that are
UTC-adjusted (`t`,
+ // `s.t` and the elements of `a`) or not (`u` and the elements of `b`),
all set to
+ // 2020-09-13 12:26:40.123456789 UTC in the row with id 1 and null or
empty in the row with
+ // id 2. Spark has no updater that reads TIMESTAMP(NANOS) as timestamp or
timestamp_ntz.
+ val path = getResourceParquetFilePath("test-data/timestamp_nanos.parquet")
+ val readTypes = Seq(
+ "t timestamp",
+ "u timestamp",
+ "s struct<t: timestamp>",
+ "a array<timestamp>",
+ "b array<timestamp>",
+ "t timestamp_ntz",
+ "u timestamp_ntz")
+ for (pushdown <- Seq(false, true); readType <- readTypes) {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
pushdown.toString) {
+ val df = spark.read.schema(s"id int,
$readType").parquet(path).where("id > 0")
+ checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
+ Seq("Spark" -> sparkError, "Comet" -> cometError).foreach { case
(engine, error) =>
+ val chain = error.toSeq.flatMap(causeChain)
+ assert(
+ chain.exists(_.isInstanceOf[
+
org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException]),
+ s"$engine, $readType, row filter pushdown $pushdown:
${chain.mkString("\n")}")
+ }
+ // The TIMESTAMP(NANOS) leaves only fail the read when they are
requested.
+ checkSparkAnswerAndOperator(spark.read.schema("id
int").parquet(path).where("id > 0"))
Review Comment:
Moved in a7eb8e9. The reads that must succeed now run once per pushdown
setting, after the read-type loop.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -1824,6 +1824,70 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("native scan rejects TIMESTAMP(NANOS) read as timestamp like Spark") {
+ // Regression guard for
https://github.com/apache/datafusion-comet/issues/7034. Spark's
+ // writer cannot produce TIMESTAMP(NANOS), so timestamp_nanos.parquet was
written with
+ // PyArrow. Next to `id int`, it holds TIMESTAMP(NANOS) leaves that are
UTC-adjusted (`t`,
+ // `s.t` and the elements of `a`) or not (`u` and the elements of `b`),
all set to
+ // 2020-09-13 12:26:40.123456789 UTC in the row with id 1 and null or
empty in the row with
+ // id 2. Spark has no updater that reads TIMESTAMP(NANOS) as timestamp or
timestamp_ntz.
+ val path = getResourceParquetFilePath("test-data/timestamp_nanos.parquet")
Review Comment:
Regenerated in a7eb8e9 with PyArrow 26. The leaves are now `id, t, u, s.t,
s.i, a, b, m.key, m.value, v.element.x, v.element.n, w`. `m` is `map<string,
timestamp(ns)>`, `v` is a `list_view<struct<x int, n timestamp(ns)>>` in
ARROW:schema, and `w` is a trailing `TIMESTAMP_MICROS`. The test expects `m`
and `v.n` read as timestamp to fail in both engines, and reads `w`, `s.i` and
`v.x` on their own against Spark. Without the ListView fix it fails.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -1824,6 +1824,70 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("native scan rejects TIMESTAMP(NANOS) read as timestamp like Spark") {
+ // Regression guard for
https://github.com/apache/datafusion-comet/issues/7034. Spark's
+ // writer cannot produce TIMESTAMP(NANOS), so timestamp_nanos.parquet was
written with
+ // PyArrow. Next to `id int`, it holds TIMESTAMP(NANOS) leaves that are
UTC-adjusted (`t`,
+ // `s.t` and the elements of `a`) or not (`u` and the elements of `b`),
all set to
+ // 2020-09-13 12:26:40.123456789 UTC in the row with id 1 and null or
empty in the row with
+ // id 2. Spark has no updater that reads TIMESTAMP(NANOS) as timestamp or
timestamp_ntz.
+ val path = getResourceParquetFilePath("test-data/timestamp_nanos.parquet")
+ val readTypes = Seq(
Review Comment:
Added in a7eb8e9. Comet matches Spark reading `t`, `s.t`, `b` and `m` as
bigint, with pushdown on and off. I also checked
`spark.sql.legacy.parquet.nanosAsLong` on and off; with an explicit read schema
it makes no difference.
--
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]
