sunchao commented on code in PR #5177:
URL: https://github.com/apache/datafusion-comet/pull/5177#discussion_r3918262116
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -195,6 +217,28 @@ fn parquet_convert_array(
list_arr.nulls().cloned(),
)))
}
+ (
+ Timestamp(TimeUnit::Millisecond, _),
+ Timestamp(TimeUnit::Microsecond, target_tz),
+ ) if top_level && parquet_options.checked_timestamp_overflow => {
+ // Spark's Parquet reader calls the checked `millisToMicros`
conversion for both
+ // direct and dictionary values, independent of CAST evaluation
mode:
+ //
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
+ //
+ // The checked conversion is limited to TOP-LEVEL columns. Spark
only avoids the
+ // error for filtered-out values through row-group statistics
pruning, and
+ // DataFusion's PruningPredicate does not support nested fields
yet, so a checked
+ // conversion on a nested field would fail queries whose
predicates Spark prunes
+ // (e.g. `WHERE s.ts < X` over an all-overflowing file). Nested
fields keep the
+ // pre-existing safe-cast behavior below (overflow -> NULL).
+ let micros = array
+ .as_primitive::<TimestampMillisecondType>()
+ .try_unary::<_, TimestampMicrosecondType, _>(|value|
value.mul_checked(1_000))?
Review Comment:
**[P2] Preserve dictionary pruning before checked timestamp conversion**
A row group can be excluded by Spark's dictionary filter even when its
min/max statistics overlap the predicate. I reproduced this with one 32-row
group containing `id: INT32` alternating between 1 and 3, and `ts:
TIMESTAMP_MILLIS = 9223372036854776`. For `SELECT ts WHERE id = 2`, Spark 4.1.3
and base `215ab706` return zero rows, but this head throws `Overflow happened
on: 9223372036854776 * 1000` with
`spark.comet.parquet.rowFilterPushdown.enabled=false` (the default).
The footer verifies dictionary-only data pages and `id` min/max of 1/3;
there is no schema widening. With plain encoding, both Spark and the head
overflow, while enabling Comet row-filter pushdown makes the dictionary case
succeed. This isolates a pruning gap beyond the timestamp-expression rewrite:
checked conversion now runs on data Spark never decodes. Please preserve
dictionary-based elimination, or use a fallback that preserves the successful
query, before applying checked conversion to these scans; add the
dictionary-only exclusion as a regression test.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -522,7 +534,220 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
}
}
+/// The wrapped expression and the file column's timezone of a millis->micros
cast.
+type MillisCastParts = (Arc<dyn PhysicalExpr>, Option<Arc<str>>);
+
+/// If `expr` is a checked millis->micros [`CometCastColumnExpr`], return its
wrapped
+/// expression and the file column's timezone.
+fn as_millis_to_micros_cast(expr: &Arc<dyn PhysicalExpr>) ->
Option<MillisCastParts> {
+ let cast = expr.downcast_ref::<CometCastColumnExpr>()?;
+ match (
+ cast.input_physical_field().data_type(),
+ cast.target_field().data_type(),
+ ) {
+ (
+ DataType::Timestamp(TimeUnit::Millisecond, file_tz),
+ DataType::Timestamp(TimeUnit::Microsecond, _),
+ ) => Some((Arc::clone(cast.expr()), file_tz.clone())),
+ _ => None,
+ }
+}
+
+/// If `expr` is a microsecond timestamp literal (null or not), return its
value.
+fn as_micros_literal(expr: &Arc<dyn PhysicalExpr>) -> Option<Option<i64>> {
+ match expr.downcast_ref::<Literal>()?.value() {
+ ScalarValue::TimestampMicrosecond(micros, _) => Some(*micros),
+ _ => None,
+ }
+}
+
impl SparkPhysicalExprAdapter {
+ /// Rewrite predicate expressions over a `TIMESTAMP_MILLIS` file column
read as
+ /// microseconds into the millisecond domain — comparisons (including
null-safe
+ /// `<=>`), `IN` lists, and null checks — mirroring Spark's
`ParquetFilters`,
+ /// which pushes timestamp predicates down in the file's physical unit:
+ ///
https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala#L192-L196
+ ///
+ /// The wrapped `CometCastColumnExpr` is opaque to DataFusion's
pruning-predicate
+ /// analyzer, so leaving the conversion inside the predicate defeats
row-group
+ /// statistics pruning — and the checked conversion then raises overflow
errors for
+ /// values Spark never reads, because Spark prunes them from the
millisecond
+ /// statistics. Comparing raw millisecond values against a rescaled
literal is exact
+ /// (`m * 1000 OP lit` over the integers), never overflows, and DataFusion
can prune
+ /// with it. Values that survive the filter still go through the checked
conversion
+ /// when the scan output is adapted, so genuine reads of overflowing
values keep
+ /// failing like Spark's `millisToMicros`.
+ fn rewrite_millis_timestamp_comparison(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ // `ts IS NULL` / `ts IS NOT NULL`: the checked conversion preserves
null-ness
+ // (it errors rather than producing nulls), so test the raw column
directly.
+ if let Some(is_null) = expr.downcast_ref::<IsNullExpr>() {
+ if let Some((inner, _)) = as_millis_to_micros_cast(is_null.arg()) {
+ return Ok(Transformed::yes(Arc::new(IsNullExpr::new(inner))));
+ }
+ }
+ if let Some(is_not_null) = expr.downcast_ref::<IsNotNullExpr>() {
+ if let Some((inner, _)) =
as_millis_to_micros_cast(is_not_null.arg()) {
+ return
Ok(Transformed::yes(Arc::new(IsNotNullExpr::new(inner))));
+ }
+ }
+
+ // `ts IN (...)` / `ts NOT IN (...)`: rescale every microsecond
literal in the
+ // list. DataFusion's pruning predicate understands `InListExpr` (up
to 20
+ // elements), so keeping the raw column visible restores row-group
pruning,
+ // mirroring Spark's `ParquetFilters` handling of `In`.
+ if let Some(in_list_expr) = expr.downcast_ref::<InListExpr>() {
+ if let Some((inner, file_tz)) =
as_millis_to_micros_cast(in_list_expr.expr()) {
+ return self.rewrite_millis_in_list(&expr, in_list_expr, inner,
file_tz);
+ }
+ }
+
+ let Some(binary) = expr.downcast_ref::<BinaryExpr>() else {
+ return Ok(Transformed::no(expr));
+ };
+ let matched = if let (Some(cast), Some(micros)) = (
+ as_millis_to_micros_cast(binary.left()),
+ as_micros_literal(binary.right()),
+ ) {
+ Some((cast, *binary.op(), micros))
+ } else if let (Some(micros), Some(cast)) = (
+ as_micros_literal(binary.left()),
+ as_millis_to_micros_cast(binary.right()),
+ ) {
+ binary.op().swap().map(|op| (cast, op, micros))
+ } else {
+ None
+ };
+ let Some(((inner, file_tz), op, micros)) = matched else {
+ return Ok(Transformed::no(expr));
+ };
+
+ let Some(micros) = micros else {
+ // NULL literal. The ordinary comparisons return NULL for every
row on both
+ // sides of the rewrite, so rescaling the literal to a NULL
millisecond
+ // literal is exact. The null-safe comparisons test null-ness of
the value,
+ // which the checked conversion preserves, so probe the raw column.
+ let rewritten: Arc<dyn PhysicalExpr> = match op {
+ Operator::IsNotDistinctFrom =>
Arc::new(IsNullExpr::new(inner)),
+ Operator::IsDistinctFrom =>
Arc::new(IsNotNullExpr::new(inner)),
+ Operator::Lt
+ | Operator::LtEq
+ | Operator::Gt
+ | Operator::GtEq
+ | Operator::Eq
+ | Operator::NotEq => Arc::new(BinaryExpr::new(
+ inner,
+ op,
+ Arc::new(Literal::new(ScalarValue::TimestampMillisecond(
+ None, file_tz,
+ ))),
+ )),
+ _ => return Ok(Transformed::no(expr)),
+ };
+ return Ok(Transformed::yes(rewritten));
+ };
+
+ // For a file value `m` (milliseconds) the logical value is exactly `m
* 1000`
+ // microseconds, so `m * 1000 OP L` rewrites to an exact comparison on
`m`.
+ let floor = micros.div_euclid(1_000);
+ let ceil = floor + i64::from(micros.rem_euclid(1_000) != 0);
+ let exact = micros % 1_000 == 0;
+ let (op, millis) = match op {
+ Operator::Lt => (Operator::Lt, ceil),
+ Operator::LtEq => (Operator::LtEq, floor),
+ Operator::Gt => (Operator::Gt, floor),
+ Operator::GtEq => (Operator::GtEq, ceil),
+ Operator::Eq if exact => (Operator::Eq, floor),
+ Operator::NotEq if exact => (Operator::NotEq, floor),
+ Operator::IsNotDistinctFrom if exact =>
(Operator::IsNotDistinctFrom, floor),
+ Operator::IsDistinctFrom if exact => (Operator::IsDistinctFrom,
floor),
+ // A sub-millisecond literal can never equal `m * 1000`. `col <
i64::MIN`
+ // (resp. `col >= i64::MIN`) is false (resp. true) for every
non-null value
+ // and NULL for nulls, matching `=` / `!=` null semantics while
remaining a
+ // plain, prunable comparison.
+ Operator::Eq => (Operator::Lt, i64::MIN),
+ Operator::NotEq => (Operator::GtEq, i64::MIN),
+ // The null-safe comparisons never return NULL, so an impossible
literal
+ // folds to a constant. DataFusion's pruning predicate evaluates
boolean
+ // literals, so `false` still prunes every row group.
+ Operator::IsNotDistinctFrom => {
+ return Ok(Transformed::yes(Arc::new(Literal::new(
+ ScalarValue::Boolean(Some(false)),
+ ))));
+ }
+ Operator::IsDistinctFrom => {
+ return Ok(Transformed::yes(Arc::new(Literal::new(
+ ScalarValue::Boolean(Some(true)),
+ ))));
+ }
+ _ => return Ok(Transformed::no(expr)),
+ };
+ let literal = Arc::new(Literal::new(ScalarValue::TimestampMillisecond(
+ Some(millis),
+ file_tz,
+ )));
+ Ok(Transformed::yes(Arc::new(BinaryExpr::new(
+ inner, op, literal,
+ ))))
+ }
+
+ /// Rewrite `ts IN (...)` / `ts NOT IN (...)` over a millis->micros
conversion into
+ /// the millisecond domain. Divisible literals rescale exactly;
sub-millisecond
+ /// literals can never equal `m * 1000` and drop out of the list, which
preserves
+ /// IN's three-valued logic (a dropped element contributes `false` to the
OR / `true`
+ /// to the AND for every non-null probe, and the null-probe result stays
NULL either
+ /// way). A list emptied this way degenerates to the same always-false /
+ /// always-true-with-null-semantics sentinels the `=` / `!=` rewrite uses.
+ fn rewrite_millis_in_list(
+ &self,
+ original: &Arc<dyn PhysicalExpr>,
+ in_list_expr: &InListExpr,
+ inner: Arc<dyn PhysicalExpr>,
+ file_tz: Option<Arc<str>>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ let mut millis_list: Vec<Arc<dyn PhysicalExpr>> = Vec::new();
+ for item in in_list_expr.list() {
+ // Any non-literal or non-timestamp element: leave the expression
alone.
+ let Some(micros) = as_micros_literal(item) else {
+ return Ok(Transformed::no(Arc::clone(original)));
+ };
+ match micros {
+ None => millis_list.push(Arc::new(Literal::new(
+ ScalarValue::TimestampMillisecond(None, file_tz.clone()),
+ ))),
+ Some(v) if v % 1_000 == 0 =>
millis_list.push(Arc::new(Literal::new(
+ ScalarValue::TimestampMillisecond(Some(v / 1_000),
file_tz.clone()),
+ ))),
+ Some(_) => {}
+ }
+ }
+ let negated = in_list_expr.negated();
+ if millis_list.is_empty() {
+ let (op, sentinel) = if negated {
+ (Operator::GtEq, i64::MIN)
+ } else {
+ (Operator::Lt, i64::MIN)
+ };
+ return Ok(Transformed::yes(Arc::new(BinaryExpr::new(
+ inner,
+ op,
+ Arc::new(Literal::new(ScalarValue::TimestampMillisecond(
+ Some(sentinel),
+ file_tz,
+ ))),
+ ))));
+ }
+ let rewritten = in_list(
Review Comment:
**[P2] Preserve pruning for timestamp IN lists longer than 20 values**
Returning a raw `InListExpr` only restores statistics pruning up to
DataFusion 54.1's `MAX_LIST_VALUE_SIZE_REWRITE = 20`. I reproduced the boundary
with 32 repeated `9223372036854776` millisecond values and `ts IN (...)`
containing 21 timestamps from `1970-01-01 00:00:00` through `00:00:20`: Spark
4.1.3 and base `215ab706` return zero rows, but this head throws checked
timestamp overflow when row-filter pushdown is disabled (the default).
All four plain/dictionary × ANSI combinations fail on the head; the 20-value
and 2-value controls pass, and enabling row-filter pushdown avoids the error.
The rewritten list is still not statistics-prunable above the dependency's
cutoff. Please provide a prunable representation or a compatible fallback for
longer lists, and cover the 20/21 boundary with the projected overflowing
timestamp.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -195,6 +217,28 @@ fn parquet_convert_array(
list_arr.nulls().cloned(),
)))
}
+ (
+ Timestamp(TimeUnit::Millisecond, _),
+ Timestamp(TimeUnit::Microsecond, target_tz),
+ ) if top_level && parquet_options.checked_timestamp_overflow => {
Review Comment:
**[P2] Preserve pruning when an unrelated filter column is widened**
The checked conversion also turns an existing pruning limitation on a
different column into a query failure. With a file containing `id: INT32 = 1`
and `ts: TIMESTAMP_MILLIS = 9223372036854776`, read `id` using `LongType` and
run `SELECT ts WHERE id < 0`. Spark 4.1.3 and base `215ab706` return zero rows,
but this head throws timestamp overflow with row-filter pushdown disabled (the
default).
The widening filter remains wrapped in an opaque `CometCastColumnExpr`, so
DataFusion cannot use its statistics to discard the group before this new
checked timestamp conversion. All four plain/dictionary × ANSI cases reproduce;
reading `id` as its original IntegerType passes, as does enabling row-filter
pushdown. Please preserve pruning for these supported widening casts, or fall
back safely for affected scans, and add this unrelated-filter/schema-evolution
case alongside the timestamp-predicate tests.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -522,7 +534,220 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
}
}
+/// The wrapped expression and the file column's timezone of a millis->micros
cast.
+type MillisCastParts = (Arc<dyn PhysicalExpr>, Option<Arc<str>>);
+
+/// If `expr` is a checked millis->micros [`CometCastColumnExpr`], return its
wrapped
+/// expression and the file column's timezone.
+fn as_millis_to_micros_cast(expr: &Arc<dyn PhysicalExpr>) ->
Option<MillisCastParts> {
+ let cast = expr.downcast_ref::<CometCastColumnExpr>()?;
+ match (
+ cast.input_physical_field().data_type(),
+ cast.target_field().data_type(),
+ ) {
+ (
+ DataType::Timestamp(TimeUnit::Millisecond, file_tz),
+ DataType::Timestamp(TimeUnit::Microsecond, _),
+ ) => Some((Arc::clone(cast.expr()), file_tz.clone())),
+ _ => None,
+ }
+}
+
+/// If `expr` is a microsecond timestamp literal (null or not), return its
value.
+fn as_micros_literal(expr: &Arc<dyn PhysicalExpr>) -> Option<Option<i64>> {
+ match expr.downcast_ref::<Literal>()?.value() {
+ ScalarValue::TimestampMicrosecond(micros, _) => Some(*micros),
+ _ => None,
+ }
+}
+
impl SparkPhysicalExprAdapter {
+ /// Rewrite predicate expressions over a `TIMESTAMP_MILLIS` file column
read as
+ /// microseconds into the millisecond domain — comparisons (including
null-safe
+ /// `<=>`), `IN` lists, and null checks — mirroring Spark's
`ParquetFilters`,
+ /// which pushes timestamp predicates down in the file's physical unit:
+ ///
https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala#L192-L196
+ ///
+ /// The wrapped `CometCastColumnExpr` is opaque to DataFusion's
pruning-predicate
+ /// analyzer, so leaving the conversion inside the predicate defeats
row-group
+ /// statistics pruning — and the checked conversion then raises overflow
errors for
+ /// values Spark never reads, because Spark prunes them from the
millisecond
+ /// statistics. Comparing raw millisecond values against a rescaled
literal is exact
+ /// (`m * 1000 OP lit` over the integers), never overflows, and DataFusion
can prune
+ /// with it. Values that survive the filter still go through the checked
conversion
+ /// when the scan output is adapted, so genuine reads of overflowing
values keep
+ /// failing like Spark's `millisToMicros`.
+ fn rewrite_millis_timestamp_comparison(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ // `ts IS NULL` / `ts IS NOT NULL`: the checked conversion preserves
null-ness
+ // (it errors rather than producing nulls), so test the raw column
directly.
+ if let Some(is_null) = expr.downcast_ref::<IsNullExpr>() {
+ if let Some((inner, _)) = as_millis_to_micros_cast(is_null.arg()) {
+ return Ok(Transformed::yes(Arc::new(IsNullExpr::new(inner))));
+ }
+ }
+ if let Some(is_not_null) = expr.downcast_ref::<IsNotNullExpr>() {
+ if let Some((inner, _)) =
as_millis_to_micros_cast(is_not_null.arg()) {
+ return
Ok(Transformed::yes(Arc::new(IsNotNullExpr::new(inner))));
+ }
+ }
+
+ // `ts IN (...)` / `ts NOT IN (...)`: rescale every microsecond
literal in the
+ // list. DataFusion's pruning predicate understands `InListExpr` (up
to 20
+ // elements), so keeping the raw column visible restores row-group
pruning,
+ // mirroring Spark's `ParquetFilters` handling of `In`.
+ if let Some(in_list_expr) = expr.downcast_ref::<InListExpr>() {
+ if let Some((inner, file_tz)) =
as_millis_to_micros_cast(in_list_expr.expr()) {
+ return self.rewrite_millis_in_list(&expr, in_list_expr, inner,
file_tz);
+ }
+ }
+
+ let Some(binary) = expr.downcast_ref::<BinaryExpr>() else {
+ return Ok(Transformed::no(expr));
+ };
+ let matched = if let (Some(cast), Some(micros)) = (
+ as_millis_to_micros_cast(binary.left()),
+ as_micros_literal(binary.right()),
+ ) {
+ Some((cast, *binary.op(), micros))
+ } else if let (Some(micros), Some(cast)) = (
+ as_micros_literal(binary.left()),
+ as_millis_to_micros_cast(binary.right()),
+ ) {
+ binary.op().swap().map(|op| (cast, op, micros))
+ } else {
+ None
+ };
+ let Some(((inner, file_tz), op, micros)) = matched else {
+ return Ok(Transformed::no(expr));
+ };
+
+ let Some(micros) = micros else {
+ // NULL literal. The ordinary comparisons return NULL for every
row on both
+ // sides of the rewrite, so rescaling the literal to a NULL
millisecond
+ // literal is exact. The null-safe comparisons test null-ness of
the value,
+ // which the checked conversion preserves, so probe the raw column.
+ let rewritten: Arc<dyn PhysicalExpr> = match op {
+ Operator::IsNotDistinctFrom =>
Arc::new(IsNullExpr::new(inner)),
+ Operator::IsDistinctFrom =>
Arc::new(IsNotNullExpr::new(inner)),
+ Operator::Lt
+ | Operator::LtEq
+ | Operator::Gt
+ | Operator::GtEq
+ | Operator::Eq
+ | Operator::NotEq => Arc::new(BinaryExpr::new(
+ inner,
+ op,
+ Arc::new(Literal::new(ScalarValue::TimestampMillisecond(
+ None, file_tz,
+ ))),
+ )),
+ _ => return Ok(Transformed::no(expr)),
+ };
+ return Ok(Transformed::yes(rewritten));
+ };
+
+ // For a file value `m` (milliseconds) the logical value is exactly `m
* 1000`
+ // microseconds, so `m * 1000 OP L` rewrites to an exact comparison on
`m`.
+ let floor = micros.div_euclid(1_000);
+ let ceil = floor + i64::from(micros.rem_euclid(1_000) != 0);
+ let exact = micros % 1_000 == 0;
+ let (op, millis) = match op {
+ Operator::Lt => (Operator::Lt, ceil),
+ Operator::LtEq => (Operator::LtEq, floor),
+ Operator::Gt => (Operator::Gt, floor),
+ Operator::GtEq => (Operator::GtEq, ceil),
+ Operator::Eq if exact => (Operator::Eq, floor),
+ Operator::NotEq if exact => (Operator::NotEq, floor),
+ Operator::IsNotDistinctFrom if exact =>
(Operator::IsNotDistinctFrom, floor),
+ Operator::IsDistinctFrom if exact => (Operator::IsDistinctFrom,
floor),
+ // A sub-millisecond literal can never equal `m * 1000`. `col <
i64::MIN`
+ // (resp. `col >= i64::MIN`) is false (resp. true) for every
non-null value
+ // and NULL for nulls, matching `=` / `!=` null semantics while
remaining a
+ // plain, prunable comparison.
+ Operator::Eq => (Operator::Lt, i64::MIN),
+ Operator::NotEq => (Operator::GtEq, i64::MIN),
+ // The null-safe comparisons never return NULL, so an impossible
literal
+ // folds to a constant. DataFusion's pruning predicate evaluates
boolean
+ // literals, so `false` still prunes every row group.
+ Operator::IsNotDistinctFrom => {
+ return Ok(Transformed::yes(Arc::new(Literal::new(
+ ScalarValue::Boolean(Some(false)),
+ ))));
+ }
+ Operator::IsDistinctFrom => {
+ return Ok(Transformed::yes(Arc::new(Literal::new(
+ ScalarValue::Boolean(Some(true)),
+ ))));
+ }
+ _ => return Ok(Transformed::no(expr)),
+ };
+ let literal = Arc::new(Literal::new(ScalarValue::TimestampMillisecond(
+ Some(millis),
+ file_tz,
+ )));
+ Ok(Transformed::yes(Arc::new(BinaryExpr::new(
+ inner, op, literal,
+ ))))
+ }
+
+ /// Rewrite `ts IN (...)` / `ts NOT IN (...)` over a millis->micros
conversion into
+ /// the millisecond domain. Divisible literals rescale exactly;
sub-millisecond
+ /// literals can never equal `m * 1000` and drop out of the list, which
preserves
+ /// IN's three-valued logic (a dropped element contributes `false` to the
OR / `true`
+ /// to the AND for every non-null probe, and the null-probe result stays
NULL either
+ /// way). A list emptied this way degenerates to the same always-false /
+ /// always-true-with-null-semantics sentinels the `=` / `!=` rewrite uses.
+ fn rewrite_millis_in_list(
+ &self,
+ original: &Arc<dyn PhysicalExpr>,
+ in_list_expr: &InListExpr,
+ inner: Arc<dyn PhysicalExpr>,
+ file_tz: Option<Arc<str>>,
+ ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
+ let mut millis_list: Vec<Arc<dyn PhysicalExpr>> = Vec::new();
+ for item in in_list_expr.list() {
+ // Any non-literal or non-timestamp element: leave the expression
alone.
+ let Some(micros) = as_micros_literal(item) else {
+ return Ok(Transformed::no(Arc::clone(original)));
+ };
+ match micros {
+ None => millis_list.push(Arc::new(Literal::new(
+ ScalarValue::TimestampMillisecond(None, file_tz.clone()),
+ ))),
+ Some(v) if v % 1_000 == 0 =>
millis_list.push(Arc::new(Literal::new(
+ ScalarValue::TimestampMillisecond(Some(v / 1_000),
file_tz.clone()),
+ ))),
+ Some(_) => {}
+ }
+ }
+ let negated = in_list_expr.negated();
+ if millis_list.is_empty() {
Review Comment:
**[P2] Preserve NULL rows for an originally empty NOT IN list**
This branch conflates an originally empty list with a nonempty list emptied
by dropping sub-millisecond literals. For original empty `NOT IN`, Spark's
nonlegacy semantics and DataFusion return true even for a NULL probe; the
generated `ts >= i64::MIN` instead returns NULL and can remove matching rows.
I reproduced this on Spark 4.1.3 with `df.filter(!col("ts").isin())`,
`spark.sql.legacy.nullInEmptyListBehavior=false`, and `OptimizeIn` plus
`ConvertToLocalRelation` excluded so the empty expression survives
optimization. For `[NULL, epoch]`, row-filter pushdown drops the NULL row; for
an all-NULL group, the head returns zero rows with pushdown either on or off.
Spark and base `215ab706` retain both rows in all cases. The direct expression
check likewise changes `[true, true]` to `[NULL, true]`.
Please distinguish an originally empty list before using the sentinel
comparison, and add NULL-input coverage. The normal optimizer folds this case
away, so the Spark exposure is conditional on the stated optimizer
configuration.
--
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]
