andygrove commented on code in PR #5177:
URL: https://github.com/apache/datafusion-comet/pull/5177#discussion_r3728616934
##########
native/spark-expr/src/utils.rs:
##########
@@ -81,12 +82,14 @@ pub fn array_with_timezone(
// so the result has the exact annotation the caller
expects.
timestamp_ntz_to_timestamp(array, timezone.as_str(),
Some(target_tz.as_ref()))
}
- Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
- // Convert from Timestamp(Millisecond, None) to
Timestamp(Microsecond, None)
- let millis_array =
as_primitive_array::<TimestampMillisecondType>(&array);
- let micros_array: TimestampMicrosecondArray =
- arrow::compute::kernels::arity::unary(millis_array,
|v| v * 1000);
- Ok(Arc::new(micros_array))
+ Some(to_type @ DataType::Timestamp(TimeUnit::Microsecond,
None)) => {
+ // This defensive conversion intentionally errors in every
CAST eval mode:
Review Comment:
Related to my other comment, do you know whether this arm is reachable today?
Spark logical timestamps are always microseconds, per `to_arrow_datatype` at
`serde.rs:93-96` and `Utils.scala:157-160`. The only millisecond arrays I can
find come from the Parquet physical schema, and those are converted inside
`spark_parquet_convert` rather than `cast_array`. So I could not find a path
that gets here with a millisecond input and a `Timestamp(Microsecond, None)`
target.
If it is dead, then the new test is pinning behavior nothing exercises, and
a note saying so would help the next reader. If you can point me at a query
that reaches it, I am happy to be wrong.
##########
native/spark-expr/src/utils.rs:
##########
@@ -81,12 +82,14 @@ pub fn array_with_timezone(
// so the result has the exact annotation the caller
expects.
timestamp_ntz_to_timestamp(array, timezone.as_str(),
Some(target_tz.as_ref()))
}
- Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
- // Convert from Timestamp(Millisecond, None) to
Timestamp(Microsecond, None)
- let millis_array =
as_primitive_array::<TimestampMillisecondType>(&array);
- let micros_array: TimestampMicrosecondArray =
- arrow::compute::kernels::arity::unary(millis_array,
|v| v * 1000);
- Ok(Arc::new(micros_array))
+ Some(to_type @ DataType::Timestamp(TimeUnit::Microsecond,
None)) => {
+ // This defensive conversion intentionally errors in every
CAST eval mode:
+ // Spark's vectorized Parquet reader calls
`millisToMicros` for both direct
+ // and dictionary values, independent of CAST evaluation.
+ //
https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java#L817-L833
+ // `millisToMicros` uses `Math.multiplyExact`:
+ //
https://github.com/apache/spark/blob/v4.2.0/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L103-L108
+ cast_with_options(array.as_ref(), to_type,
&DEFAULT_CAST_OPTIONS)
Review Comment:
I went to confirm the overflow fix end to end and I do not think it reaches
the Parquet reader. I wrote a scratch test driving
`CometCastColumnExpr::evaluate` with an `i64::MAX` millisecond value on this
branch, and it returns `null` rather than erroring, for both `Timestamp(ms,
None)` and `Timestamp(ms, Some("UTC"))`.
The reason is that the scan never goes through `array_with_timezone`.
`evaluate` falls through the relabel arm, because
`types_differ_only_in_field_names` is `false` for a flat timestamp pair. It
lands in `spark_parquet_convert`. `parquet_convert_array` has no timestamp-unit
arm, so it reaches the `can_cast_types` fallthrough at `parquet_support.rs:257`
and calls `cast_with_options` with `PARQUET_OPTIONS`. That static is `safe:
true` at `parquet_support.rs:57`. I also grepped for callers, and
`array_with_timezone` is only reached from `cast.rs:262` and the three datetime
functions. Nothing under `native/core/src/parquet/` calls it.
On the Spark side this is a real divergence.
`ParquetVectorUpdaterFactory:155-170` sends both `TimestampType` and
`TimestampNTZType` with MILLIS to `LongAsMicrosUpdater`, which calls
`millisToMicros`, which is `Math.multiplyExact`. So Spark throws regardless of
ANSI mode, and we return null.
Would you be willing to add an explicit `(Timestamp(ms, _), Timestamp(us,
_))` arm to `parquet_convert_array` that does the checked multiply? I would
avoid flipping `PARQUET_OPTIONS` to `safe: false` wholesale, since that static
also covers narrowing casts where returning null is what Spark does, so a
targeted arm seems safer than changing the global.
If you would rather keep the scope as it is, could we retitle away from
`fix:` and open an issue for the Parquet path? Our changelog is generated from
PR titles, and as written this would tell users the Parquet overflow is fixed
when it is not.
##########
native/core/src/parquet/cast_column.rs:
##########
@@ -214,20 +176,41 @@ impl Hash for CometCastColumnExpr {
}
impl CometCastColumnExpr {
- /// Create a new [`CometCastColumnExpr`].
- pub fn new(
+ /// Try to create a new [`CometCastColumnExpr`].
+ pub fn try_new(
expr: Arc<dyn PhysicalExpr>,
physical_field: FieldRef,
target_field: FieldRef,
cast_options: Option<CastOptions<'static>>,
- ) -> Self {
- Self {
+ ) -> DataFusionResult<Self> {
+ let physical_type = physical_field.data_type();
+ let target_type = target_field.data_type();
+ // `target_field` is the Spark logical field, while `physical_field`
comes from the
+ // Parquet or Iceberg file. Comet represents Spark's TimestampType and
TimestampNTZType
+ // as Arrow microseconds, and Spark maps both TIMESTAMP_MICROS and
TIMESTAMP_MILLIS files
+ // to those logical types. A millisecond target is therefore invalid
at this read-adapter
+ // boundary:
+ //
https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala#L318-L324
+ if matches!(
Review Comment:
The rejection reads well, and I confirmed the claim holds. Both the Parquet
scan and `IcebergScanExec` build their logical schema through
`convert_spark_types_to_arrow_schema`, so a millisecond target cannot occur.
One scoping question. The check looks at the top-level pair only, so a
target like `Struct<ts: Timestamp(ms)>` passes `try_new` and goes on to
`spark_parquet_convert`. Since the invariant you are asserting is about the
whole read schema rather than just flat columns, the guard currently reads more
complete than it is. Could we either walk nested types or narrow the comment to
say it covers the top level? Either is fine. I mostly want the next person not
to trust it further than it goes.
--
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]
