refactor: extract shared runtime filter components

[pingzh]

Which issue does this PR close?

Part 1 of the five-PR plan for #5775. The issue remains open for the later steps.

Rationale for this change

Give join wiring, decoded-batch filtering, and Parquet reader attachment clear ownership so the latter two can be reused by TopK.

What changes are included in this PR?

Four commits:

1. Separate DynamicFilterJoinExec and its existing tests from the shared decoded-batch consumer, with an explicit owner-metrics constructor.

2. Extract the existing Parquet reader traversal/remapping and its null-check recognition test.

3. Use dynamic_filter_join_* for all six join filter metrics across Rust, Spark, tests, and docs. The shared consumer receives its metric prefix from its owner.

4. Group the components under dynamic_filter/ in response to review, keeping each unit-test module beneath its implementation so tests retain access to private state:

   dynamic_filter/
   ├── mod.rs
   ├── tests.rs
   ├── join.rs
   ├── join/
   │   └── tests.rs
   ├── parquet_reader.rs
   └── parquet_reader/
       └── tests.rs
   

Filtering behavior, eligibility, and execution lifetime are preserved. Most of the diff moves existing code and tests. Metric consumers must use the renamed keys.

How are these changes tested?

After the module reorganization:

• All 19 focused native tests passed: the 18 runtime-filter tests and the planner metric-export test.
• Cargo formatting and diff whitespace checks passed.
• An independent sub-agent reviewed the moves, imports, visibility, and preservation of all existing tests, with no remaining concerns.

Initial refactor validation, before the module reorganization:

• 19 native tests passed: all 18 existing runtime-filter tests plus the planner metric-export test. Each extraction commit also passed the 18 operator tests independently.
• 52 Spark 4.1.3 join tests passed on JDK 21. The freshly rebuilt native library and Spark-packaged copy have matching SHA-256 hashes.
• Whole-workspace Clippy with all targets and warnings denied, Cargo formatting, Spotless, Scalastyle, and Prettier passed. Native build and Spark 4.1 test compilation passed.
• Three independent sub-agents reviewed the initial refactor and found no remaining concerns; each commit was reviewed before committing.

[sunchao]

Reviewed 93b3061 against 39da850. No actionable findings after five independent review scopes covering native join wiring, Spark metrics, memory/error cleanup, Parquet reader compatibility, and test/edge-case preservation.

The extraction preserves join eligibility, build/probe selection, independent execution state, reset/child-replacement behavior, and EOF/error/cancellation cleanup. Reader remapping and boundaries around limits, projections, and stateful residual expressions are unchanged. All six renamed metric keys match across native producers, Spark consumers, tests, and documentation. The existing native and Scala test assertions remain intact.

Validation:

• Inspected the native CI job: 1,447 tests passed, 5 skipped, including all 18 runtime-filter operator tests and the planner metric-export test.
• Verified 52 passing CometJoinSuite tests each in the Spark 3.4 and Spark 4.2 execution jobs. Those whole-job runs had zero failures, with 15 canceled/5 ignored and 7 canceled/5 ignored tests respectively.
• CI ran merge commit e2155db. Its source tree is identical to the reviewed head.
• Local formatting and diff-whitespace checks passed. Local native compilation stopped during dependency resolution because the configured package mirror lacks the locked DataFusion 55.1.0 dependency, so native/JVM runtime tests were not reproduced locally.

At submission, CI has 43 successful checks, 8 running, 1 queued, 9 skipped, and no failures. The remaining workflow results are pending.

LGTM. Approving the reviewed revision.

[rich7420]

@pingzh thanks for the patch

[sunchao]

Rechecked unchanged 93b306177abeb5ab72cd10bfb63c0ab35fc4735c against 39da85081b140dae57492f96e3478337ee345a8f. My existing approval remains. The decoded-batch filtering and eligibility bodies are byte-identical to the base. The moved lifecycle and reader logic retain their behavior, and all 18 operator tests retain their assertions. The four formatted metric names are constructed when an execution opens, outside the batch loop. All six renamed keys match the native producers, Spark consumers and documentation.

CI has now completed with 53 successful checks and 10 skipped. I verified 1,447 native tests passing, including all 18 runtime-filter operator tests and the metric-export test, with 5 tests skipped. The logs also show 52 passing CometJoinSuite tests on each of Spark 3.4, 3.5, 4.0 and 4.1. The executed merge e2155db8 has the same source tree as this head, and the native artifact's upload/download hashes agree.

Source-equivalence and diff-whitespace checks pass. I found no new P1/P2 finding. I did not run native/JVM tests or benchmarks locally. Maintained Spark 3.5/4.0 source was checked for null-key and duplicate-match semantics. Maintained 3.4/4.1 sources remain unavailable, independently of the CI results.

[peterxcli]

let's make the structure for dyn filter like:

dynamic_filter/
├─ mod.rs              (shared runtime-filter components; from dynamic_filter.rs)
├─ join.rs             (DynamicFilterJoinExec)
├─ parquet_reader.rs   (try_attach_parquet_reader_filter, null-check logic)
└─ tests/
   ├─ mod.rs           (#[cfg(test)] mod join; mod parquet_reader;)
   ├─ join.rs
   └─ parquet_reader.rs

as we will have topk.rs, agg.rs and their tests

[ajsquared]

Reviewed all 11 changed files and relevant callers at 93b3061. Runtime-filter behavior and lifetime handling are preserved, and metric names align across Rust and Spark. No independently confirmed P1 findings. Static review only; tests and CI were not run or inspected.

[pingzh]

let's make the structure for dyn filter like:

dynamic_filter/
├─ mod.rs              (shared runtime-filter components; from dynamic_filter.rs)
├─ join.rs             (DynamicFilterJoinExec)
├─ parquet_reader.rs   (try_attach_parquet_reader_filter, null-check logic)
└─ tests/
   ├─ mod.rs           (#[cfg(test)] mod join; mod parquet_reader;)
   ├─ join.rs
   └─ parquet_reader.rs

as we will have topk.rs, agg.rs and their tests

@peterxcli that is a good idea. let me update it

[sunchao]

Rechecked 7b2f35a7187acab9f7a34ee8bc65b78587d93f55 against 39da85081b140dae57492f96e3478337ee345a8f, including the changes since 93b306177abeb5ab72cd10bfb63c0ab35fc4735c. No new or remaining P1/P2 findings.

The new dynamic_filter/ layout groups the shared consumer, join wrapper and Parquet reader helper. Each test module remains beneath its implementation, preserving private access and the shared consumer's independent predicate-update/reset/error coverage. The incremental source changes are module declarations, imports, re-exports, a module comment and the label path. The filtering bodies and all three test files are unchanged by this reorganization. The planner-facing export is preserved.

Source-equivalence and diff-whitespace checks pass. Current CI shows 16 successful checks, three running and 11 skipped at September 15, 17:23 UTC. I inspected the successful Rust-format and Scala-lint logs at merge f0bf7abf. Its reviewed files and Cargo.lock match this head, but its complete tree and first parent differ from the reviewed base. The current Rust test log confirms 1,451 passed and five skipped, including all 18 reorganized operator tests and the planner metrics test. Native-library and JVM builds are still running. Earlier JVM runtime results apply to the previous head.

No local native/JVM tests or benchmarks were run. Maintained Spark 3.5/4.0 source was checked. Maintained 3.4/4.1 sources remain unavailable.

[sunchao]

Rechecked 7b2f35a7187acab9f7a34ee8bc65b78587d93f55 against 39da85081b140dae57492f96e3478337ee345a8f. Source and discussion are unchanged since my approval. No new or remaining P1/P2 findings.

Current CI has finished with 19 successful checks, 11 skipped, 34 cancelled and one failed. Required Checks failed because the Linux, Spark 4.1 and Iceberg 1.11 upstream groups were cancelled. The logs do not establish why the run was cancelled.

The Rust log confirms 1,451 passed and five skipped, including all 18 runtime-filter operator tests and the planner metrics test. Native-library and Spark 4.1 test-class builds completed. The native artifact upload and Spark 4.1 execution-job download digests match, but that job's Java test step was cancelled. This adds no passing JVM-suite result.

These results used merge f0bf7abf, whose reviewed files and native lockfile match this head, while its full tree and first parent differ from the reviewed head/base. No local native/JVM tests or benchmarks were run.