fix: remove nested planner placeholders after recursive conversion

[sunchao]

Which issue does this PR close?

Closes #6438.

Rationale for this change

Comet temporarily wraps Spark query stages while converting a physical plan. When aggregate fallback revisits an already wrapped stage, it can create CometSinkPlaceHolder(CometSinkPlaceHolder(stage)). The top-down cleanup removes only the outer wrapper, leaving a planning-only operator that fails during execution.

For example, with native shuffle enabled, native hash partitioning disabled, and native round-robin partitioning enabled, SELECT k, AVG(v) FROM (SELECT /*+ REPARTITION(4) */ * FROM t) GROUP BY k fails after AQE materializes the repartition stage with CometNativeExec should not be executed directly without a serialized plan.

What changes are included in this PR?

Remove sink placeholders from the bottom up. Explicitly recurse into the plan hidden by a leaf CometScanWrapper, preserving cleanup of its descendants as well as removing wrappers at its root. Preserve existing query-stage objects and AQE partition specifications.

Regressions cover mixed sink/scan nesting, aggregate-buffer repair around an existing native round-robin stage under one configuration, and the end-to-end SQL query above. The SQL test checks nullable AVG results and confirms that a native round-robin shuffle executes beneath restored Spark aggregates without surviving placeholders.

How are these changes tested?

• Built the candidate native library with cargo build --locked, without dependency patches; verified the JVM resource matches the built library's SHA256.
• Spark 4.1.3 / JDK 21: all 86 tests in CometExecRuleSuite pass.
• Before/after control: restoring only the old cleanup makes all three placeholder regressions fail, including the SQL test with the runtime exception above; the other 83 tests pass. Restoring the fix makes the full suite pass.
• Maven formatting and Scala style checks pass.

Broader Spark SQL and supported-version CI is requested with run-spark-4.1-tests and run-all-spark-profiles.

[andygrove]

The old top-down transform recursed into a replacement's children rather than the replacement itself, so the inner CometSinkPlaceHolder that revertUnsafePartialAggregates wraps around an existing stage was never visited. The bottom-up cleanup fixes that and keeps the stage object and its partitionSpecs intact. For plans without nesting both cleanups give the same tree and tags, and the TPC-DS plan stability suites in this head's [exec] run agree. By my reading this is reachable from SQL too: with spark.comet.shuffle.mode=native, native hash partitioning off and native round-robin on, SELECT k, avg(v) FROM (SELECT /*+ REPARTITION(4) */ * FROM t) GROUP BY k leaves a placeholder directly under the Spark partial once the repartition stage materializes, and the query fails with CometNativeExec should not be executed directly without a serialized plan. Could we add an end-to-end test along those lines?

[viirya]

+1 to Andy's request for the end-to-end REPARTITION test. Since the round-robin stage stays native under a single conf, it would also let the planner-level regression build its input stage without toggling COMET_SHUFFLE_NATIVE_HASH_PARTITIONING_ENABLED between passes.

[sunchao]

@andygrove @viirya Addressed in 1627bd7. Added an end-to-end Parquet-backed REPARTITION(4) / grouped AVG regression with AQE enabled, native hash partitioning disabled, and native round-robin partitioning enabled. It checks nullable AVG results, the executed native round-robin shuffle, restored Spark aggregates, and absence of placeholders. The planner-level test now builds its round-robin input stage under the same configuration, with no hash-partitioning toggle.

Restoring only the old cleanup reproduces CometNativeExec should not be executed directly without a serialized plan on a leftover sink around the materialized native repartition stage. All three placeholder regressions fail in that control; the other 83 tests pass. With the fix restored, all 86 CometExecRuleSuite tests pass on Spark 4.1.3 / JDK 21. Native build, formatting, and Scala style checks also pass. The existing Spark SQL and all-profile CI labels remain applied.

[viirya]

Thanks for the updates. The comments now explain where the nesting comes from and why the scan wrapper recursion matters, and the end-to-end REPARTITION test covers the runtime failure under a single configuration. LGTM.