ListPositionsExpr panics on sliced list input, breaking native posexplode over LIMIT with OFFSET

[andygrove]

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 (3df58d5):

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

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:

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:

#[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.