andygrove commented on code in PR #5161:
URL: https://github.com/apache/datafusion-comet/pull/5161#discussion_r3690464235
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -855,6 +855,8 @@ case class CometScanTypeChecker() extends DataTypeSupport
with CometTypeShim {
name: String,
fallbackReasons: ListBuffer[String]): Boolean = {
dt match {
+ case _: YearMonthIntervalType | _: DayTimeIntervalType =>
Review Comment:
`isSchemaSupported` is called for `r.partitionSchema` as well as
`requiredSchema` (lines 831 and 839), and Spark does allow ANSI intervals as
partition columns. `PartitioningUtils.canPartitionOn` takes any `AtomicType`,
and `castPartValueToDesiredType` has an explicit `case it: AnsiIntervalType`
branch. Once the checker admits the type, `partition2Proto` tries
`exprToProto(Literal(value, YearMonthIntervalType))`, and
`CometLiteral.getSupportLevel` only whitelists `DayTimeIntervalType` today
(`literals.scala:60`), because `QueryPlanSerde.supportedDataType` lists neither
interval type. So the literal comes back as `None` and the `assert` at
`serde/operator/package.scala:64` fires.
I reproduced this on your branch with Spark 4.1:
```
java.lang.AssertionError: assertion failed: Unsupported partition value: 38,
type: YearMonthIntervalType(0,1)
at
org.apache.comet.serde.operator.package$.partition2Proto(package.scala:52)
at
org.apache.spark.sql.comet.CometNativeScanExec.$anonfun$serializedPartitionData$10(...)
```
Reverting just this hunk goes back to a clean fallback with `Unsupported p
of type YearMonthIntervalType(0,1)`. Day-time interval partition columns are
fine, because the literal path already handles them.
I think this is also what the Spark 4.1 `sql_core-1` job is hitting.
`PartitionedWriteSuite` has `"SPARK-37231, SPARK-37240: Dynamic writes/reads of
ANSI interval partitions"`, which writes a Parquet dir partitioned by
`YearMonthIntervalType(YEAR)` and `YearMonthIntervalType(YEAR, MONTH)` and
reads it back. The suite is untagged so it runs in that shard, and it is not
excluded in `dev/diffs/4.1.2.diff`.
Could you either teach `CometLiteral` and the native `IntVal` branch about
`YearMonthIntervalType`, or keep the partition-schema gate narrow so interval
partition columns still fall back? Either way a test for both interval types as
partition columns would be good, since nothing covers that today.
Separately, the `assert` in `partition2Proto` turning a planner gap into an
`AssertionError` rather than a fallback looks fragile. That seems worth a
follow-up issue of its own.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -125,6 +127,64 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("ANSI interval types") {
Review Comment:
This is a thorough test. I like that you assert the row group count and
dictionary encoding rather than hoping the write options took effect.
A few cases that are newly reachable through this change and might be worth
covering, since `DataTypeSupport.isTypeSupported` recurses into complex types
via your override:
- interval leaves nested in a struct, an array and a map
- a filter on an interval column, with and without
`spark.sql.parquet.filterPushdown`
- the interval min and max (`INTERVAL '178956970-7' YEAR TO MONTH` and the
day-time equivalents)
- both interval types as partition columns
I checked all of these on your branch and everything except the year-month
partition column produces Spark-matching results, so mostly this is about
locking the behavior in rather than chasing a bug.
One thing worth knowing: a `ym` predicate is currently dropped from the
native data filters because the year-month literal is unsupported, so that
column gets no row group pruning. Results stay correct but the filter is not
accelerated. Same root cause as the partition value problem.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -855,6 +855,8 @@ case class CometScanTypeChecker() extends DataTypeSupport
with CometTypeShim {
name: String,
fallbackReasons: ListBuffer[String]): Boolean = {
dt match {
+ case _: YearMonthIntervalType | _: DayTimeIntervalType =>
+ true
Review Comment:
Did you look at what happens when the Parquet physical type does not match
the requested interval type? Spark's `ParquetVectorUpdaterFactory` accepts any
INT32 as `YearMonthIntervalType` via `IntegerUpdater` and any INT64 as
`DayTimeIntervalType` via `LongUpdater`, and raises
`PARQUET_COLUMN_DATA_TYPE_MISMATCH` for anything else. Comet's
`parquet_convert_array` goes through arrow `can_cast_types` and then the `_ =>
Ok(array)` fallthrough, which gives different answers in two cases. I probed
these on your branch:
| file column | requested schema | Spark | Comet |
| --- | --- | --- | --- |
| `date` | `interval year to month` | succeeds (`P1522Y1M`, ...) | `Invalid
argument error: column types must match schema types, expected
Interval(YearMonth) but found Date32` |
| `int` | `interval day to second` | `PARQUET_COLUMN_DATA_TYPE_MISMATCH` |
succeeds, values reinterpreted as micros |
| `timestamp` | `interval day to second` | error | error |
| `long` | `interval year to month` | error | error |
The `int` row is the one I would want resolved, since Comet returns rows
where Spark raises. arrow's `(_, Duration(_)) if from_type.is_numeric()`
happily accepts that cast. The `date` row is the mirror image, a hard native
failure where Spark succeeds.
Both need schema-on-read to trigger, so I do not think they have to be fixed
in this PR. Could you file a tracking issue and link it here? Left unfiled I
suspect it will not get picked up.
##########
docs/source/user-guide/latest/datatypes.md:
##########
@@ -82,15 +82,15 @@ the tables below and may be reconsidered based on demand:
All three interval types are mapped to Arrow and flow through serde, native
shuffle, and the
codegen dispatcher, so interval columns and the interval-producing expressions
run natively.
-Several operators still gate on the type and fall back: Parquet scans of ANSI
interval columns,
-single-column sorts, hash aggregates (`min` / `max` / `sum` / `avg`), `GROUP
BY`, window
-functions, and hashing a `CalendarInterval`. Remaining work is tracked by
+Comet's V1 native Parquet scan also supports both ANSI interval types. Several
operators still
Review Comment:
Comet has a single native Parquet scan path now, so "V1 native Parquet scan"
may read as if there is a V2 variant with different support. Could this just
say "Comet's native Parquet scan"?
Two more things for this section. The notes should call out whatever the
partition column outcome ends up being. And #5060 asked whether the
`CalendarIntervalType` entry in the shared `DataTypeSupport.isTypeSupported`
should move to `CometLocalTableScanExec` or go away, given it can never appear
in a Parquet file. Did you decide to leave that for a follow-up? A linked issue
would help if so.
Unrelated to the docs, but worth adding to the PR description:
`CometScanTypeChecker` also gates the native Iceberg scan
(`CometScanRule.scala:366` and `:525`), not just Parquet V1. That is moot in
practice since Iceberg has no interval type, but the current wording reads as
if the change is Parquet-only and a future reader may not realize the checker
is shared.
--
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]
