refactor: move MergeRows, Explode, aggregate, join and window planning into dedicated modules

[andygrove]

Which issue does this PR close?

Part of #6979.

Rationale for this change

PhysicalPlanner::create_plan builds most operators inline, which makes planner.rs the most frequently edited file in the native code and a regular source of merge conflicts. #6979 proposes finishing the move onto OperatorRegistry. This PR does its first sub-task, moving the largest create_plan arms, which takes create_plan from 1,320 lines to 373 and planner.rs from 8,555 lines to 6,189.

What changes are included in this PR?

Each commit moves one operator family into its own module under native/core/src/execution/planner/, registers its builders in OperatorRegistry, and removes its arms from create_plan. The helpers each family uses move with it.

• merge_rows.rs: MergeRows, with merge_output_type_compatible.
• explode.rs: Explode.
• aggregate.rs: HashAgg, with create_agg_expr and the helpers it uses (create_aggr_func_expr, coerce_collect_child_nullability, make_all_fields_nullable, min_max_udaf and percentile_value).
• join.rs: HashJoin, SortMergeJoin and BroadcastNestedLoopJoin, with parse_join_parameters, which builds the join keys and filter for all three, the join-filter column rewriting, strip_timestamp_tz for sort-merge join keys, and the runtime-filter helpers.
• window.rs: Window and WindowGroupLimit, with create_window_expr, process_agg_func, find_df_window_function and numeric_literal_to_scalar.

OperatorType gains MergeRows, Explode, BroadcastNestedLoopJoin and WindowGroupLimit. #6979 lists BroadcastNestedLoopJoin and WindowGroupLimit under its second sub-task. They move here to keep each family in one module. The window example in the debugging guide now points at WindowBuilder::build.

The moved code is unchanged apart from self becoming planner in the builders and rustfmt reflowing the dedented lines. create_aggr_func_expr and coerce_collect_child_nullability become pub(super) because the planner tests call them, and min_max_udaf because window.rs shares it with aggregate.rs. Reviewing one commit at a time with git show --color-moved=zebra --color-moved-ws=allow-indentation-change marks almost all of the removed lines as moved.

How are these changes tested?

This moves code without changing behavior, so it relies on the existing tests. Each commit passes on its own:

• cargo fmt --all -- --check
• cargo clippy --color=never --all-targets --workspace -- -D warnings
• cargo test -p datafusion-comet --lib (874 passed, 5 ignored)
• the JVM suite for the family it moves

On the last commit:

• ./mvnw test -Dtest=none -Dsuites=... with CometMergeRowsSuite, CometGenerateExecSuite, CometJoinSuite, CometWindowExecSuite and CometAggregateSuite on the default Spark 4.1 profile (362 passed, 1 canceled because it needs Spark 4.2, 2 ignored)
• CometMergeRowsSuite with -Pspark-4.0, whose version of the suite runs the native MERGE tests (10 passed)

[kazuyukitanimura]

pending ci

1. On docs/source/contributor-guide/adding_a_new_operator.md ("Update the Planner" section): "Thanks for splitting this up. The refactor makes OperatorRegistry the
   home for operator planning. Could you update this section to describe adding an OperatorBuilder in a module under planner/ and registering it, instead of adding a
   match case in create_plan? Otherwise new contributors will keep adding arms to the function you just shrank."
2. On native/common/src/schema.rs lines 103 and 211, and native/core/src/execution/planner.rs:134: "These comments say make_all_fields_nullable is in the planner. It
   now lives in planner/aggregate.rs. Could you update the pointers so they stay findable?"

[viirya]

Thanks for picking this up. Splitting the families one commit at a time made the move easy to follow, and --color-moved marks nearly every removed line as moved. I went through each commit, and outside of self becoming planner, the extract_op! and children boilerplate, imports and rustfmt reflow, the builders match the old arms line for line. That includes the MergeRows single-child check and the hash join runtime-filter and swap path. The four new get_operator_type mappings only feed can_handle and OperatorRegistry::create_plan, so operators that have an OperatorType but no registered builder still fall through to the match as before.

One thing outside the diff. Step 5 of docs/source/contributor-guide/adding_a_new_operator.md still tells contributors to add a match arm to create_plan in planner.rs. Now that the registry is the established home for the large operators, and #6979 aims to leave create_plan as a thin dispatcher, anyone following that guide would add code back into the function this work is shrinking. Could that section describe the registry path instead? That means implementing an OperatorBuilder in a module under native/core/src/execution/planner/, mapping the OpStruct variant in get_operator_type, and registering the builder in OperatorRegistry::register_all_operators.

[viirya]

Since this paragraph now points at WindowBuilder::build, could the snippet below it be refreshed as well? The builder picks window_agg between BoundedWindowAggExec and WindowAggExec based on all_bounded (window.rs:134-135). It then may wrap that in a casting ProjectionExec as final_plan (window.rs:155), and final_plan is what goes to SparkPlan::new. The example still shows a single BoundedWindowAggExec::try_new feeding SparkPlan::new directly. That was already stale before this PR, but a reader following the new pointer won't find the lines it describes.

[andygrove]

Good catch. The snippet now shows the all_bounded choice between BoundedWindowAggExec and WindowAggExec, wraps window_agg right after it is built (before the casting projection), and says to wrap final_plan the same way to see the output after the cast. Pushed in 2c28107.

[andygrove]

Thanks @kazuyukitanimura and @viirya. Pushed 2c28107, which covers the three asks:

• adding_a_new_operator.md: the "Update the Planner" step now describes writing an OperatorBuilder in a module under planner/, adding the OperatorType variant, mapping the OpStruct in get_operator_type, and registering it from register_all_operators. It no longer tells contributors to add an arm to create_plan.
• The three make_all_fields_nullable comments in native/common/src/schema.rs (two) and planner.rs now point at planner/aggregate.rs.
• The window example in debugging.md matches WindowBuilder::build (see the inline reply).

The commit only touches comments and docs, so I did not rerun the test suites.

[viirya]

Thanks for the quick turnaround. The new "Update the Planner" section matches how the moved builders are written, so the next operator should land as a builder rather than another arm in create_plan. The window example in debugging.md now lines up with WindowBuilder::build, including the all_bounded choice and the note on wrapping final_plan to see the post-cast output. The make_all_fields_nullable pointers read correctly too. LGTM.