proto: preserve DynamicFilterPhysicalExpr identity across round-trip

[adriangb]

Which issue does this PR close?

• Alternative to proto: serialize and dedupe dynamic filters #20416. Informs Push dynamic filters across network boundaries datafusion-contrib/datafusion-distributed#180 and closes Serialize dynamic filters across network boundaries #20418.

Rationale for this change

Given a plan like

HashJoinExec(dynamic_filter_1 on a@0)
  (...left side of join)
  ProjectionExec(a := Column("a", source_index))
    DataSourceExec
      ParquetSource(predicate = dynamic_filter_2)

after serialize/deserialize the two DynamicFilterPhysicalExpr wrappers should still share the same mutable inner, so that a HashJoinExec update during execution is visible at the pushed-down ParquetSource and the scan prunes rows. Today this breaks for several reasons:

1. serialize_physical_expr_with_converter calls snapshot_physical_expr, which replaces DynamicFilterPhysicalExpr with its current inner expression (often lit(true)) — identity is lost.
2. The existing dedup key hashes the outer Arc::as_ptr, but the HashJoinExec side and the ParquetSource side hold different outer Arcs (one comes from with_new_children), so they never share expr_id.
3. Even if the filter survived the top-level trip, HashJoinExec and SortExec (TopK) don't serialize their own dynamic_filter field — on deserialize they mint a fresh filter that's disconnected from anything pushed into the scan.

What changes are included in this PR?

Seven commits, grouped by concern. Refactor commits change API shape only and should be reviewable independently. Change commits add behavior.

Base

1. proto: preserve DynamicFilterPhysicalExpr identity across round-trip — Adds PhysicalExpr::expression_id(&self) -> Option<u64> (default None). DynamicFilterPhysicalExpr gains a stable u64 id that follows the shared Arc<RwLock<Inner>> across with_new_children. New proto variant PhysicalDynamicFilterExprNode { current_expr, children, generation, is_complete } — the inner expr tree is serialized natively instead of snapshotted away. DeduplicatingSerializer stamps expr_id = expr.expression_id() (no more Arc::as_ptr/session_id/pid hashing). DeduplicatingDeserializer uses a single unified cache path: on miss, parse + cache; on hit, parse to recover this site's children and overlay via with_new_children on the cached canonical — gives DynamicFilter its shared-inner semantics without any type-specific code in the deserializer. Three proto-level round-trip tests cover shared-inner preservation, per-site remapped children with update propagation, and the FilterExec → ProjectionExec → DataSourceExec(ParquetSource) shape.

Refactor

2. physical-expr: builder setters on DynamicFilterPhysicalExpr — Replaces a growing 5-arg constructor with fluent with_id(u64) / with_generation(u64) / with_is_complete(bool) setters on Self, following the HashJoinExecBuilder style. new(children, inner) keeps the default case.
3. hash_join: public builder setter + accessor for the dynamic filter — Makes HashJoinExecBuilder::with_dynamic_filter public and changes its signature from the crate-private Option<HashJoinExecDynamicFilter> to plain Arc<DynamicFilterPhysicalExpr>, wrapping it internally. The inner SharedBuildAccumulator stays private; only the filter crosses the API boundary. Promotes dynamic_filter_for_test to dynamic_filter().
4. sort: add SortExec::with_dynamic_filter builder setter + accessor — Mirror for SortExec. Fluent setter on Self + a dynamic_filter() accessor that reads through the TopKDynamicFilters wrapper.

Change

5. proto: round-trip HashJoinExec dynamic_filter — Adds optional PhysicalExprNode dynamic_filter to HashJoinExecNode. to_proto emits the filter via the new accessor; from_proto parses it (hitting the id cache populated by the scan-side pushed-down predicate) and installs it via HashJoinExecBuilder::with_dynamic_filter. Because the cache returns the same canonical Arc, the join's filter and the scan's filter share inner automatically.
6. proto: round-trip SortExec's TopK dynamic filter — Same pattern for SortExecNode. with_fetch may auto-create a TopK filter; with_dynamic_filter then replaces it with the sender's so the id matches the pushed-down scan's copy.

Integration tests

7. test: add datafusion-tests workspace crate for cross-crate integration tests — Introduces a new datafusion-tests workspace member at datafusion/tests/. Tests living here can dev-depend on both datafusion and datafusion-proto simultaneously; putting them in either of those crates' own tests/ directory closes a dev-dep cycle caught by dev/depcheck. Hosts two end-to-end SQL tests that prove the round-trip actually drives scan-level pruning (see next section).

Comparison with #20416

PR #20416 introduces a generic PhysicalExprId { exact, shallow } and keeps an Arc-ptr default so every expression is stamped. This PR instead makes the identity hook opt-in per type (expression_id()), restricting the blast radius to the one type that needs it today. A follow-up can implement expression_id() on other types (e.g. InList, literals) to re-introduce generic within-process dedup on a case-by-case basis.

Are these changes tested?

Yes.

Unit tests (datafusion/physical-expr/src/expressions/dynamic_filters.rs):

• test_expression_id_stable_across_with_new_children — the id survives with_new_children.

Proto-level round-trip tests (datafusion/proto/tests/cases/roundtrip_physical_plan.rs):

• roundtrip_dynamic_filter_preserves_shared_inner — two wrappers in a BinaryExpr(And) predicate share inner after round-trip; an update() on one is observable via the other.
• roundtrip_dynamic_filter_preserves_remapped_children — two wrappers with different effective children; each preserves its site-specific projection, both share identity, update() propagates, and current() on the remapped side applies the column substitution.
• roundtrip_dynamic_filter_in_parquet_pushdown — the plan shape from the PR description (FilterExec → ProjectionExec → DataSourceExec(ParquetSource)).

End-to-end SQL tests (datafusion/tests/tests/dynamic_filter_proto_roundtrip.rs):

• hash_join_dynamic_filter_prunes_via_sql — registers a parquet file, runs INNER JOIN ... WHERE b.n_nationkey = 5, round-trips via DeduplicatingProtoConverter, executes, and asserts the probe-side scan emitted strictly fewer rows than the full table.
• topk_dynamic_filter_prunes_files_via_sql — writes two single-row parquet files (a.parquet key=1, b.parquet key=2), runs ORDER BY ... LIMIT 1 with target_partitions=1, round-trips, executes, and asserts the scan emitted exactly 1 row. b.parquet is pruned by row-group statistics once TopK's dynamic filter tightens after reading a.parquet.

Removed: the pre-existing test_expression_deduplication_arc_sharing, test_deduplication_within_plan_deserialization, test_deduplication_within_expr_deserialization, and two test_session_id_rotation_* tests — they asserted the generic Arc-ptr dedup contract that this PR deliberately drops.

CI is green: cargo fmt, clippy, circular dependency check, detect-unused-dependencies, all cargo test matrixes pass.

Are there any user-facing changes?

• New trait method PhysicalExpr::expression_id(&self) -> Option<u64> with a safe default (None). Existing implementations keep compiling.
• New public DynamicFilterPhysicalExpr surface: with_id, with_generation, with_is_complete, is_complete(). The Debug impl now hides the random id field so plan snapshots stay deterministic.
• HashJoinExecBuilder::with_dynamic_filter promoted to pub with a Arc<DynamicFilterPhysicalExpr> signature; HashJoinExec::dynamic_filter() accessor promoted from dynamic_filter_for_test. Callers of the renamed method need to update.
• New SortExec::with_dynamic_filter setter + SortExec::dynamic_filter() accessor.
• DeduplicatingProtoConverter no longer deduplicates arbitrary expressions by Arc pointer. Plans that relied on that (e.g. for a large shared InList) will serialize independently until a per-type expression_id() is added. Dynamic filters now round-trip with shared state, which is the primary motivation.
• New workspace member datafusion-tests (internal; publish = false) hosting cross-crate integration tests.

🤖 Generated with Claude Code