andygrove commented on PR #5192: URL: https://github.com/apache/datafusion-comet/pull/5192#issuecomment-5160165555
I dug into the `ListPositionsExpr` concern from my earlier review and filed https://github.com/apache/datafusion-comet/issues/5224 for it. Short version: `ListPositionsExpr` panics when its input `ListArray` has a non-zero offset base. It builds a fresh values array numbered from zero but reuses the input's original offset buffer, so `ListArray::new` unwraps an `InvalidArgumentError` from Arrow. `GlobalLimitExec` with a non-zero skip produces exactly that shape, since `LimitStream` does `batch.slice(self.skip, ...)`. This is pre-existing and not something you introduced. It reproduces on `main` today with plain `posexplode`. I had guessed the `CometFilter` that Spark inserts above the limit via `InferFiltersFromGenerate` would reset the offsets and keep the non-outer path safe, but it does not. All rows pass the predicate, so Arrow's `filter` returns the input arrays untouched and the slice survives. What this PR changes is the blast radius. `posexplode_outer` falls back today, so it is safe by default. Once it runs natively it hits the same panic. Here is a test that shows the difference. It passes on `main` and fails with this PR: ```scala test("posexplode_outer 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_outer(arr) FROM (SELECT id, arr FROM t LIMIT 4 OFFSET 1)") checkSparkAnswerAndOperator(df) } } ``` With this PR the plan becomes `CometExplode` directly over `CometGlobalLimit -1, 1` and the query fails with: ``` org.apache.comet.CometNativeException: called `Result::unwrap()` on an `Err` value: InvalidArgumentError("Max offset of 9 exceeds length of values 6") ``` I verified the same failure on `main` by setting `spark.comet.operator.GenerateExec.allowIncompatible=true`, which is the behavior this PR makes the default. One note on the test. `spark.sql.leafNodeDefaultParallelism = 1` is required rather than cosmetic. With the default parallelism each partition produces a one-row batch, `LimitStream` discards whole batches instead of slicing, and the bug is masked. AQE off just keeps the plan readable. I would suggest the Scala form over a `posexplode.sql` entry here. The SQL harness does support `-- Config:`, but the reproduction depends on how many files the `INSERT` writes and how the shuffle reader batches them, and that is harder to pin down from a `.sql` file. Could you either pick up the fix from #5224 as a prerequisite and include this test passing, or add it as `ignore` referencing #5224? The fix itself is small, rebasing the offsets to zero so they line up with the newly built values array: ```rust let base = offsets[0]; let rebased = OffsetBuffer::new(offsets.iter().map(|o| o - base).collect::<Vec<_>>().into()); ``` I lean toward fixing it in this PR since this is the change that makes it reachable by default, but I am fine either way as long as it is tracked rather than silent. -- 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]
