andygrove commented on code in PR #5457:
URL: https://github.com/apache/datafusion-comet/pull/5457#discussion_r4283988192
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -250,6 +250,7 @@ pub(crate) fn cast_array(
return Ok(Arc::new(array));
}
+ let array = prepare_nested_cast_input(array, to_type)?;
Review Comment:
This runs on every cast of a struct, list or map. It copies the input with
`take` whenever a struct has a null row, a list or map is sliced, or more than
a quarter of its rows are null. Only signed integer widening and identity casts
on primitives skip it, so most casts that can't fail pay for the copy too.
Against the merge base, at 8,192 rows, `struct<int, string>` to
`struct<bigint, string>` went from 1.6 µs to 70 µs with 10% null rows, because
the string child is now copied instead of reused. At 50% null rows, `list<int>`
to `list<bigint>` went from 19.6 µs to 66 µs and `map<int, int>` to `map<int,
bigint>` from 19.8 µs to 112 µs. A sliced `list<int>` with no nulls went from
19.5 µs to 111 µs. ANSI gave the same numbers, and shapes with no nulls and no
slice didn't change.
The copy only matters when a hidden child value could fail the cast. That's
DATE to TIMESTAMP in every mode, plus the leaves that can fail under ANSI or
TRY. For a cast that can't fail, `main` already reuses the children, so the
copy is all cost. Could the gate be whether some leaf of the cast can fail in
this eval mode, rather than whether the leaves are fixed width?
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -163,6 +175,15 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
withFallbackReason(scanExec, "Full native scan disabled because
ignoreMissingFiles enabled")
}
+ // A requested NTZ field can be backed by a physical DATE column. Filtered
native scans
+ // currently use a safe cast that turns DATE overflow into NULL. Checking
eagerly instead
+ // would fail on rows Spark can skip through pruning or an early LIMIT.
The physical schema
+ // is only available when each file opens, so conservatively retain
Spark's reader for this
+ // projected type, including genuine NTZ files, until conversion follows
Spark's row demand.
+ if (scanExec.dataFilters.nonEmpty &&
containsTimestampNTZ(scanExec.requiredSchema)) {
Review Comment:
This sends every filtered Parquet scan that reads a `TIMESTAMP_NTZ` field to
Spark's reader, and almost all of those are genuine NTZ files with no DATE
column anywhere. Spark adds filters by itself too. An inner join's `isnotnull`
key filter or a `window()` is enough. That's the six red tests on every
profile. `window.sql` keeps 2 of 8 operators native, `unix_timestamp.sql` keeps
0 of 3, and the fuzz `join` test finds 0 native scans instead of 2.
I don't think it buys the row-demand guarantee either. With no filter the
guard doesn't apply, so `read().limit(1)` on the same 6,000-row file runs
natively and throws. Comet's 8,192-row batch converts rows that Spark's
4,096-row batch never reaches. That's the batch-evaluation difference the new
cast docs already accept. `main` and `branch-1.1` already throw for a top-level
DATE to NTZ overflow in a filtered scan, because that column goes through the
native `Cast` in legacy mode, whose NTZ branch calls Arrow's cast with `safe:
false`. Spark can't prune on the converted column either, apart from `IS NULL`
and `IS NOT NULL`. `ParquetFilters.valueCanMakeFilterOn` only builds a filter
on a DATE column for a `Date`, a `LocalDate` or a null, and an NTZ literal is a
`LocalDateTime`.
Could the DATE to NTZ arm in `parquet_convert_array_impl` stay checked
whether or not the scan has filters, and this guard and its compatibility
paragraph go? That keeps genuine NTZ scans native, and a selected overflowing
row still gets the plain `ArithmeticException`. It also fixes filtered nested
fields, which `main` silently nulls. If you want Spark's `IS NULL` pruning
covered exactly, a null check through the checked conversion is the same as a
null check on the DATE column, so it can be unwrapped. For the test, a file
whose overflowing rows sit in a row group that both engines prune on `id`
statistics would cover the pruning case without asserting a fallback.
##########
spark/src/main/scala/org/apache/comet/expressions/CometCast.scala:
##########
@@ -520,6 +568,18 @@ object CometCast
case _ => Unsupported(Some(s"Cast from DateType to $toType is not
supported"))
}
+ /**
+ * Resolve fixed-zone aliases with Spark's rules, then serialize only the
subset understood by
+ * the native fixed-offset parser. Checking `isFixedOffset` alone would
admit UTC aliases that
+ * chrono rejects and offsets with seconds that chrono can silently truncate
to whole minutes.
+ */
+ private[comet] def nativeDateTimestampTimezone(timezone: String):
Option[String] =
Review Comment:
This duplicates `CometTimeZone.nativeId`. Both normalize with
`DateTimeUtils.getZoneId(...).normalized()`, map a zero offset to `UTC`, keep
whole-minute offsets as `+HH:MM` and reject offsets with seconds. `castToProto`
already runs the result through `nativeId`, so the remap in `convert` doesn't
change anything. Could `canCastFromDate` ask `CometTimeZone` instead, for
example through an `isFixedOffset` helper next to `nativeId`, and the remap go?
`timezones.md` asks for timezones to reach native code only through
`CometTimeZone`, and keeping one parser for these IDs was the point of the
alias change in the first round.
##########
native/spark-expr/src/conversion_funcs/temporal.rs:
##########
@@ -42,50 +38,47 @@ pub(crate) fn cast_date_to_timestamp(
cast_options: &SparkCastOptions,
target_tz: &Option<Arc<str>>,
) -> SparkResult<ArrayRef> {
- if target_tz.is_none() {
- // TIMESTAMP_NTZ ignores session TZ. Only TRY_CAST turns overflow into
null.
- return Ok(cast_with_options(
- array_ref.as_ref(),
- &DataType::Timestamp(TimeUnit::Microsecond, None),
- &CastOptions {
- safe: cast_options.eval_mode == EvalMode::Try,
- ..DEFAULT_CAST_OPTIONS
- },
- )?);
- }
-
+ // Retain checked arithmetic instead of Arrow's generic NTZ cast so
failures preserve
+ // Spark's plain ArithmeticException("long overflow") in both ANSI and
legacy modes.
let date_array = array_ref.as_primitive::<Date32Type>();
let mut builder =
TimestampMicrosecondBuilder::with_capacity(date_array.len());
- // TIMESTAMP: midnight in session TZ → UTC epoch μs
- let tz_str = if cast_options.timezone.is_empty() {
- "UTC"
+
+ // Date32 has a wider range than both chrono's calendar and an i64
microsecond timestamp.
+ // Use day arithmetic for NTZ and fixed-offset zones. Region-zone casts
run through Spark's
+ // JVM codegen dispatcher (see CometCast.canCastFromDate), which owns
their full-range rules.
+ let offset_seconds = if target_tz.is_none()
+ || cast_options.timezone.is_empty()
+ || cast_options.timezone == "UTC"
+ {
+ 0
} else {
- cast_options.timezone.as_str()
+ // Reject any unexpected region-zone call safely instead of
constructing a chrono date.
+ // CometCast canonicalizes fixed-zone aliases to UTC or +/-HH:MM
before serialization.
+ cast_options
+ .timezone
+ .parse::<FixedOffset>()
+ .map_err(|_| {
+ ArrowError::ParseError(format!(
+ "Invalid timezone \"{}\": expected a fixed offset",
+ cast_options.timezone
+ ))
+ })?
+ .local_minus_utc()
};
- // safe to unwrap since we are falling back to UTC above
- let tz = Tz::from_str(tz_str)?;
- let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
for date in date_array.iter() {
Review Comment:
This loop is about 2.7x slower than the Arrow cast the merge base uses for
NTZ, about 8 µs there against 21.6 µs here for 8,192 dates with 1% nulls, in
both legacy and TRY mode. The fixed-offset `TIMESTAMP` case went the other way,
from about 165 µs to 22 µs in UTC, because chrono is gone.
You can keep the plain `ArithmeticException` and still get the vectorized
loop for both. I tried `try_unary` with a closure that returns
`to_micros(d).ok_or(())`, then `.map_err(|_| SparkError::LongOverflow)`, plus
`unary_opt` for TRY. That's the same shape as your new Parquet arm. It ran at
about 9 µs for NTZ, UTC and `+05:30`, and the 125 `conversion_funcs` tests
still pass. Returning `SparkError` straight from the closure only got to 15.6
µs, because the wide `Result` slows the hot loop. Could the cast use that form?
--
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]
