refactor: move the plan entry point logic into core functions

[viirya]

Which issue does this PR close?

Part of #6434.

Rationale for this change

#6435 and #6534 separated the native logic of the JNI entry points that do not call back into the
JVM from their JNI argument handling. This does the same for the plan entry points: createPlan,
setShufflePartitionPusher, executePlan and releasePlan. Their logic was spread across the
JNI exports, so a plan could not be created, run and released in a Rust unit test.

What changes are included in this PR?

Each export keeps converting its JNI arguments and calls a core function with no Env or JNI
local reference in its signature:

• createPlan: prepare_plan_settings parses the Spark configs and initializes the Tokio runtime.
  create_execution_context(PlanInputs, memory_pool_for) deserializes the plan and sets up its
  memory pool, DataFusion session and tracing registration.
• setShufflePartitionPusher: register_shuffle_partition_pusher(ctx, make_pusher) checks the plan
  state, then builds the pusher, so a rejected call still creates no global reference.
• executePlan: execute_plan(ctx, stage_id, partition, publish_metrics) returns the plan's next
  batch, or None at the end. The export still exports the batch through prepare_output inside
  the same trace span.
• releasePlan: release_plan(ctx, publish_metrics) stops the plan, publishes its final metrics,
  drops it and waits for its reservations.

These plans call back into the JVM, so the JVM objects remain JNI global references held by the
execution context (PlanJvmRefs). The steps that need an Env are passed in as callbacks: the
memory pool factory (the pool acquires memory from Spark's task memory manager), the pusher
factory, and PublishMetrics, which publishes to the CometMetricNode.
ExecutionContext::metrics becomes an Option. createPlan always sets it, and None is only
used without a JVM.

Exception classes and messages are unchanged. In createPlan the JNI conversions now all happen
before the plan is deserialized, so when two steps fail at once, a different one of the two errors
can be reported. Successful plan creation is unchanged.

How are these changes tested?

• New Rust unit tests without a JVM:
  • A RangeScan plan is created, executed to the end and released. The test checks the values,
    the batch size, that the plan runs on a Tokio task, that metrics are published once on
    release, and that no memory stays reserved.
  • release_plan reports a failed metrics publish.
  • A malformed plan is rejected.
  • A shuffle pusher can only be registered once, before execution, and a rejected pusher is not
    built.
• Ran CometExecSuite, CometNativeShuffleSuite, CelebornShufflePartitionPusherSuite,
  CometTaskMetricsSuite and ParquetEncryptionITCase locally.

This pull request and its description were written by Isaac.

[sunchao]

Summary

• Prior state and problem: Plan lifecycle logic lived inside JNI exports, making it difficult to test without a JVM.
• Design approach: Extract creation, execution, shuffle callback registration and release into internal Rust functions, with callbacks for JVM-dependent operations.
• Correctness / compatibility analysis: JNI signatures, execution paths, Arrow export, metrics cadence and cleanup ordering are preserved. Checked Spark task-context and completion-listener semantics across 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. No introduced P1/P2 issues found within this review.
• Key design decisions: Global references retain ownership across calls. Rejected shuffle registrations still avoid constructing a callback. The small input structs and callbacks provide useful test boundaries without adding batch copies or per-row JNI calls.
• Implementation sketch: PlanSettings, PlanInputs and PlanJvmRefs feed create_execution_context. execute_plan returns batches for the existing JNI exporter, and release_plan preserves cleanup even when metrics publication fails.
• Behavioral changes worth calling out: JNI conversions now precede plan deserialization, so simultaneous creation failures can report a different first error. Compared with branch-1.1, cancellation detection, metrics throttling and reservation-release waiting are inherited changes, not introduced here.
• Suggested improvements: None meet the requested P1/P2 reporting threshold.

Reviewed full SHA eddc691be21972957c1d215b370479df7eb37ef0 against base 1aa166fde858b2bd20ac1dfa83eb2c16e2f6208c. The complete PR diff changes jni_api.rs. Verified that the additional cache/Iceberg differences in a direct tree comparison originate exclusively from two newer base commits. Existing reviews, comments and threads were empty.

Routed skills: review-comet-pr, review-comet-ffi-pr, review-comet-memory-pr and review-comet-shuffle-pr. Also consulted review-comet-iceberg-write-pr while checking the base-only differences.

Exact-head CI: 38 successful checks, 31 skipped and one cancelled. Rust CI recorded 2,089 passing tests, including all four new tests. Linux Comet suites, TPC checks and all nine Spark 4.1 SQL shards passed. The label-run aggregate passed. The main workflow reports failure with its Required Checks job cancelled, so its aggregate verdict is not green.

Validation: All 35 focused execution::jni_api::tests passed locally with --no-default-features. The default-feature build failed because this environment lacks jni.h for HDFS. No local JVM suite or performance benchmark was run. macOS, other Spark runtime profiles and Iceberg suites were skipped in exact-head CI.

[andygrove]

I built the merge preview (main plus this PR) and the jni_api unit tests pass, including repeated runs under load. One question on the lifecycle test inline.

[andygrove]

Thanks for the quick fix on the lifecycle test. I ran the release tests against the leaking release_plan from my first comment, the one that skips BatchProducer::stop and forgets the context, and release_plan_returns_the_plans_memory and release_plan_reports_a_failed_metrics_publish both fail now with the sort's reservation still held. The reserved() > 0 check before the release also means the tests will say so if the sort ever stops reserving at this size. This looks good to me.

[viirya]

Thank you @sunchao @andygrove !