andygrove opened a new issue, #5224:
URL: https://github.com/apache/datafusion-comet/issues/5224
## Describe the bug
`ListPositionsExpr`
(`native/core/src/execution/expressions/list_positions.rs`), which backs the
`pos` column for native `posexplode` / `posexplode_outer`, panics when its
input `ListArray` has a non-zero offset base (i.e. a sliced list). The panic
surfaces as:
```
org.apache.comet.CometNativeException: called `Result::unwrap()` on an `Err`
value:
InvalidArgumentError("Max offset of 9 exceeds length of values 6")
```
## To Reproduce
Any query where a native limit with a non-zero offset feeds a native
`posexplode` in the same native plan. Reproduced against `main` (3df58d528):
```scala
test("posexplode over limit with offset") {
withSQLConf(
"spark.sql.adaptive.enabled" -> "false",
"spark.sql.leafNodeDefaultParallelism" -> "1",
CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true",
CometConf.COMET_EXEC_EXPLODE_ENABLED.key -> "true") {
Seq((1, Array(1, 2, 3)), (2, Array(4, 5)), (3, Array(6)), (4, Array(7,
8)), (5, Array(9)))
.toDF("id", "arr")
.createOrReplaceTempView("t")
val df = spark.sql(
"SELECT id, posexplode(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET
1)")
checkSparkAnswerAndOperator(df)
}
}
```
`spark.sql.leafNodeDefaultParallelism = 1` matters only so that all rows
arrive in a single batch. With the default parallelism each partition produces
a one-row batch, `LimitStream` discards whole batches instead of slicing, and
the bug is masked.
The resulting native plan is:
```
UnnestExec
ProjectionExec: expr=[col_0@0 as col_0, list_positions(col_1@1) as pos,
col_1@1 as col_1]
GlobalLimitExec: skip=1, fetch=None
ScanExec: source=[Exchange (unknown)], schema=[col_0: Int32, col_1:
List(non-null Int32)]
```
Note that the `CometFilter` Spark inserts above the limit via
`InferFiltersFromGenerate` does not mask the bug. All rows pass the predicate,
so Arrow's `filter` returns the input arrays untouched and the slice survives.
## Root cause
`ListPositionsExpr::evaluate` builds a fresh values array numbered from
zero, but reuses the input's original offset buffer:
https://github.com/apache/datafusion-comet/blob/main/native/core/src/execution/expressions/list_positions.rs#L96-L114
```rust
let offsets = list.offsets();
let total_len = *offsets.last().unwrap() as usize;
let mut values: Vec<i32> = Vec::with_capacity(total_len);
for window in offsets.windows(2) {
let start = window[0];
let end = window[1];
for i in 0..(end - start) {
values.push(i);
}
}
let result = ListArray::new(
element_field,
offsets.clone(), // <-- still based at the original offset
Arc::new(Int32Array::from(values)), // <-- length is only the slice
span
list.nulls().cloned(),
);
```
`GenericListArray::slice` slices `value_offsets` and `nulls` but leaves
`values` unsliced, so a sliced list has `offsets[0] > 0` while `values()` stays
full length. `ListPositionsExpr` pushes only `offsets.last() - offsets[0]`
values but keeps offsets that run up to `offsets.last()`, so
`ListArray::try_new` rejects it on `offsets.last() > values.len()` and
`ListArray::new` unwraps the error into a panic.
`DataFusion`'s `LimitStream` produces exactly this shape: `poll_and_skip`
does `batch.slice(self.skip, batch.num_rows() - self.skip)`. `planner.rs` maps
a Comet limit with a non-zero offset onto `GlobalLimitExec` with that skip. The
plain `LIMIT` path (`skip == 0`) slices from zero, keeps the base at zero, and
is unaffected.
## Suggested fix
Rebase the offsets to zero so they match the newly built values array:
```rust
let base = offsets[0];
let rebased = OffsetBuffer::new(offsets.iter().map(|o| o -
base).collect::<Vec<_>>().into());
```
A unit test that fails before the fix and passes after:
```rust
#[test]
fn sliced_input_with_non_zero_offset_base() {
// 5-row list, then slice out rows 1..4. `offsets()` becomes [2, 3, 3, 5]
// (base 2) while `values()` stays unsliced at length 6.
let values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
let offsets = OffsetBuffer::new(vec![0, 2, 3, 3, 5, 6].into());
let element_field = Arc::new(Field::new("item", DataType::Int32, true));
let input = ListArray::new(element_field, offsets, Arc::new(values),
None);
let sliced: ArrayRef = Arc::new(input.slice(1, 3));
let schema = Schema::new(vec![Field::new("arr",
sliced.data_type().clone(), true)]);
let batch = RecordBatch::try_new(Arc::new(schema),
vec![sliced]).unwrap();
let expr = ListPositionsExpr::new(Arc::new(Column::new("arr", 0)));
let result = expr.evaluate(&batch).unwrap();
let out = result.into_array(batch.num_rows()).unwrap();
let list = out.as_any().downcast_ref::<ListArray>().unwrap();
assert_eq!(list.len(), 3);
assert_eq!(
list.value(0).as_any().downcast_ref::<Int32Array>().unwrap(),
&Int32Array::from(vec![0])
);
assert_eq!(list.value(1).len(), 0);
assert_eq!(
list.value(2).as_any().downcast_ref::<Int32Array>().unwrap(),
&Int32Array::from(vec![0, 1])
);
}
```
Without the fix this test panics with `InvalidArgumentError("Max offset of 5
exceeds length of values 3")`.
It is worth auditing the other native expressions that rebuild a `ListArray`
from `offsets()` plus a freshly built values array for the same assumption.
## Additional context
Affects `posexplode` today. It will also affect `posexplode_outer` once
#5192 lands, since that PR makes the outer variant native by default. On `main`
today `posexplode_outer` is `Incompatible` and falls back to Spark, so the
query above succeeds for the outer variant unless
`spark.comet.operator.GenerateExec.allowIncompatible=true` is set. With that
config set it panics identically, which is the behavior #5192 makes the default.
`ListEmptyToNullExpr`, added in #5192, handles the sliced case correctly (it
preserves the original offsets alongside the unsliced `values()`), but it also
preserves the non-zero offset base and passes it straight into
`ListPositionsExpr` in the `posexplode_outer` plan.
--
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]
