docs: rewrite comments to describe the code, not the change

Drop comments that read like PR-review notes ("X no longer Y", "the
legacy CASE", "this drops the routing") in favour of comments that
describe the current behaviour for someone reading the file cold.
Trim some now-redundant field-level docs and tighten doc strings on
`MultiMapLookupExpr`, `PushdownStrategy`, `build_partitioned_filter`,
and `try_build_merged_inlist`.

No functional change.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1072,14 +1072,11 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
         .await
         .unwrap();
 
-    // Now check what our filter looks like. When the cross-partition InList
-    // is small enough (≤ MERGED_INLIST_MAX_TOTAL_LEN) we collapse it into a
-    // single global `IN (SET)` regardless of how the data was repartitioned —
-    // this lets the merged set participate in parquet stats / bloom-filter
-    // pruning at the scan, which a per-partition `CASE` could not. Both the
-    // normal repartition and the `force_hash_collisions` path produce the
-    // same logical shape; they only differ in the partition-iteration order
-    // that controls the InList element order.
+    // The dynamic filter for a `Partitioned` hash join with a small enough
+    // cross-partition InList collapses to a single global `struct(...) IN
+    // (SET) ([...])`. The two `#[cfg]` arms differ only in the order of the
+    // InList elements, which is determined by partition-iteration order
+    // (normal repartition vs. the `force_hash_collisions` collapse).
     #[cfg(not(feature = "force_hash_collisions"))]
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
@@ -2575,10 +2572,9 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() {
         .await
         .unwrap();
 
-    // Verify the all-Map fast path collapses per-partition routing into a
-    // single shared `multi_hash_lookup` rather than a
-    // `CASE hash_repartition % N WHEN p THEN hash_lookup ELSE false END`
-    // expression.
+    // The dynamic filter for an all-Map Partitioned hash join uses a single
+    // shared `multi_hash_lookup` over every partition's hash table. There is
+    // no per-row `hash_repartition` routing.
     let plan_str = format_plan_for_test(&plan).to_string();
     assert!(
         plan_str.contains("multi_hash_lookup"),

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -1316,8 +1316,7 @@ impl ExecutionPlan for HashJoinExec {
             .counter(ARRAY_MAP_CREATED_COUNT_METRIC_NAME, partition);
 
         // Initialize build_accumulator lazily with runtime partition counts
-        // (only if enabled). The dynamic filter no longer routes by repartition
-        // hash, so REPARTITION_RANDOM_STATE is not needed here.
+        // (only when dynamic filter pushdown is enabled).
         let build_accumulator = enable_dynamic_filter_pushdown
             .then(|| {
                 self.dynamic_filter.as_ref().map(|df| {
@@ -2042,11 +2041,11 @@ async fn collect_left_input(
 
     let map = Arc::new(join_hash_map);
 
-    // The `Map` is always built (the join itself uses it). The optional
-    // `inlist` array is set when the build side fit under the per-partition
-    // InList caps — that's our signal that this partition's keys are small
-    // enough to participate in parquet stats / bloom-filter pruning when
-    // collapsed across partitions on the probe-side scan.
+    // The hash map is needed by the join itself, so it always travels in
+    // `PushdownStrategy::map`. The optional `inlist` array is attached when
+    // the build side fits under the per-partition InList caps; the
+    // `SharedBuildAccumulator` may then merge those arrays across partitions
+    // into a single `IN (SET)` for scan-side pruning.
     let membership = if num_rows == 0 {
         PushdownStrategy::empty()
     } else {

datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs

@@ -346,29 +346,24 @@ impl PhysicalExpr for HashTableLookupExpr {
 /// Physical expression that probes the same join keys against multiple [`Map`]s
 /// and returns `true` for any row whose join keys match at least one map.
 ///
-/// Equivalent to `OR`-ing several [`HashTableLookupExpr`]s but evaluates
-/// `create_hashes` exactly once for the whole batch — every [`Map::HashMap`]
-/// probe shares that hash buffer. Used for the global-first dynamic filter
-/// when every reported partition uses a hash-table pushdown strategy.
+/// Equivalent to `OR`-ing several [`HashTableLookupExpr`]s, but
+/// `create_hashes` runs exactly once for the whole batch and every
+/// [`Map::HashMap`] probe shares the same hash buffer. All `HashMap`
+/// entries must therefore have been built with the same `RandomState`;
+/// [`Map::ArrayMap`] entries are queried via `contain_keys` and do not
+/// consume hashes.
 pub struct MultiMapLookupExpr {
-    /// Columns in the ON clause used to compute the join key for lookups
+    /// Join-key expressions evaluated against each input batch.
     on_columns: Vec<PhysicalExprRef>,
-    /// Random state for hashing — every map must have been built with the
-    /// same `RandomState`, otherwise the shared hash buffer is meaningless.
+    /// Hashing seed shared by every entry in `maps`.
     random_state: SeededRandomState,
-    /// Maps to OR over (each is one partition's build-side data)
+    /// Build-side maps to OR over, one per partition.
     maps: Vec<Arc<Map>>,
-    /// Description for display
+    /// Display name used in `EXPLAIN` output (e.g. `"multi_hash_lookup"`).
     description: String,
 }
 
 impl MultiMapLookupExpr {
-    /// Create a new MultiMapLookupExpr.
-    ///
-    /// `maps` is the (per-partition) sequence of build-side maps to probe.
-    /// All `Map::HashMap` entries are expected to use the same `random_state`;
-    /// `Map::ArrayMap` entries do not consume hashes and are queried via
-    /// `contain_keys`.
     pub fn new(
         on_columns: Vec<PhysicalExprRef>,
         random_state: SeededRandomState,
@@ -466,8 +461,8 @@ impl PhysicalExpr for MultiMapLookupExpr {
         let join_keys = evaluate_columns(&self.on_columns, batch)?;
 
         if self.maps.is_empty() || num_rows == 0 {
-            // No maps to probe — this should not happen in practice but
-            // returning all-false matches the semantics of an empty `OR`.
+            // Empty `maps` would not be constructed by the dynamic-filter
+            // builder — guard anyway: an empty OR is `false` for every row.
             let buffer = BooleanBufferBuilder::new(num_rows);
             let mut buffer = buffer;
             buffer.append_n(num_rows, false);
@@ -477,14 +472,13 @@ impl PhysicalExpr for MultiMapLookupExpr {
             ))));
         }
 
-        // Whether any map needs hashes. We only compute hashes if at least
-        // one map is a HashMap.
+        // Hashes are only needed for `HashMap` probes; `ArrayMap` queries
+        // its keys directly via `contain_keys`.
         let needs_hashes = self
             .maps
             .iter()
             .any(|m| matches!(m.as_ref(), Map::HashMap(_)));
 
-        // Result buffer accumulates the OR of every map's `contain_*` result.
         let mut result = vec![false; num_rows];
 
         let process_one_map =
@@ -494,8 +488,8 @@ impl PhysicalExpr for MultiMapLookupExpr {
                         let hashes = hashes
                             .expect("hashes computed when at least one map is a HashMap");
                         let arr = hm.contain_hashes(hashes);
-                        // OR into the running result. `arr` has no nulls
-                        // (`contain_hashes` returns a non-nullable BooleanArray).
+                        // `contain_hashes` always returns a non-null
+                        // `BooleanArray`; OR its bits into `result`.
                         for (slot, hit) in result.iter_mut().zip(arr.values().iter()) {
                             *slot |= hit;
                         }
@@ -511,8 +505,8 @@ impl PhysicalExpr for MultiMapLookupExpr {
             };
 
         if needs_hashes {
-            // Compute the join-key hashes ONCE, then probe every HashMap
-            // against the same buffer. ArrayMap probes ignore the hashes.
+            // Hash the join keys once and reuse the buffer for every
+            // `HashMap` probe; `ArrayMap` probes pass `None` for hashes.
             with_hashes(&join_keys, self.random_state.random_state(), |hashes| {
                 for map in &self.maps {
                     process_one_map(&mut result, map, Some(hashes))?;

datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs

@@ -303,14 +303,11 @@ pub(crate) struct SharedBuildAccumulator {
     on_right: Vec<PhysicalExprRef>,
     /// Schema of the probe (right) side for evaluating filter expressions
     probe_schema: Arc<Schema>,
-    /// Cap on the cross-partition merged-InList distinct count. Reuses
-    /// `optimizer.hash_join_inlist_pushdown_max_distinct_values` (the same
-    /// option that gates per-partition InList pushdown). When the union of
-    /// every reported partition's deduplicated InList values stays at or
-    /// below this many distinct entries we collapse them into a single
-    /// `IN (SET)` predicate that can participate in parquet stats /
-    /// bloom-filter pruning at the scan; otherwise we use
-    /// `multi_hash_lookup` over every partition's hash table instead.
+    /// Maximum distinct entries the cross-partition merged InList may
+    /// contain before we fall back to `multi_hash_lookup`. Sourced from
+    /// `optimizer.hash_join_inlist_pushdown_max_distinct_values` so the
+    /// same threshold caps both per-partition and cross-partition InList
+    /// pushdown.
     inlist_max_distinct_values: usize,
 }
 
@@ -734,31 +731,26 @@ impl SharedBuildAccumulator {
         }))
     }
 
-    /// Build the dynamic filter for `PartitionMode::Partitioned`. The filter
-    /// is decoupled from the repartition strategy: regardless of whether
-    /// individual partitions chose Map or InList for their pushdown, we
-    /// always emit `(global_minmax AND ([merged_in_list AND] multi_hash_lookup))`.
-    /// This drops the legacy `CASE hash_repartition % N WHEN p THEN … END`
-    /// routing expression entirely.
+    /// Build the dynamic filter for `PartitionMode::Partitioned`. Emits
+    /// `global_minmax AND ([merged_in_list AND] multi_hash_lookup)` —
+    /// independent of how the build side was repartitioned.
     ///
     /// * `global_minmax` — envelope of every partition's per-column min/max.
     ///   Cheap short-circuit and the only piece visible to scan-level
     ///   `pruning_predicate` extraction.
-    /// * `merged_in_list` — concatenated build keys when every reported
-    ///   partition produced an `InList` array and the cross-partition
-    ///   *deduplicated* set has at most `inlist_max_distinct_values` distinct
-    ///   entries (the same option that gates the per-partition InList path,
-    ///   `optimizer.hash_join_inlist_pushdown_max_distinct_values`). Worth
-    ///   carrying because a small `IN (SET)` participates in parquet stats /
-    ///   bloom-filter pruning, which `multi_hash_lookup` cannot.
-    /// * `multi_hash_lookup` — runtime hash-table probe across every
-    ///   partition's `Map`, hashing the join keys once.
+    /// * `merged_in_list` — concatenated, deduplicated build keys when every
+    ///   reported partition contributed an `InList` array and the
+    ///   cross-partition union fits under
+    ///   `optimizer.hash_join_inlist_pushdown_max_distinct_values`. A small
+    ///   `IN (SET)` participates in parquet stats / bloom-filter pruning,
+    ///   which `multi_hash_lookup` does not. When present it fully replaces
+    ///   the lookup.
+    /// * `multi_hash_lookup` — hashes the join keys once and ORs
+    ///   `contain_hashes()` across every partition's hash table.
     ///
-    /// The `has_canceled_unknown` case is the only one that can't safely use
-    /// this shape (we'd be missing maps for the canceled partitions). We
-    /// could keep a CASE just for that case, but the query is in the middle
-    /// of being torn down — emit `lit(true)` and let the join do whatever
-    /// filtering it can on its own.
+    /// `has_canceled_unknown` partitions short-circuit to `lit(true)`: we
+    /// don't have their maps, so we cannot include them in the lookup, and
+    /// the query is being torn down anyway.
     fn build_partitioned_filter(
         &self,
         real_partitions: &[&PartitionData],
@@ -777,12 +769,10 @@ impl SharedBuildAccumulator {
             .as_ref()
             .and_then(|b| create_bounds_predicate(&self.on_right, b));
 
-        // Try to build a merged InList. Only fires when *every* reported
-        // partition contributed an InList array AND the cross-partition
-        // deduplicated union has at most `inlist_max_distinct_values`
-        // entries. The merged InList already covers the union of every
-        // partition's build-side keys, so when present it fully replaces
-        // `multi_hash_lookup` — no need to AND a redundant probe on top.
+        // The merged InList covers the union of every partition's
+        // build-side keys, so when it fires it stands alone — there is no
+        // need to also AND a `multi_hash_lookup` (which would just probe
+        // the same data via a different structure).
         let membership_expr =
             if let Some(merged) = self.try_build_merged_inlist(real_partitions)? {
                 Some(merged)
@@ -815,14 +805,13 @@ impl SharedBuildAccumulator {
     /// If every reported partition contributed an InList array, concatenate
     /// them, deduplicate by scalar value, and gate on the
     /// `inlist_max_distinct_values` cap. Returns the merged
-    /// `(struct(...))? IN (SET) ([…])` predicate built over the *deduplicated*
-    /// keys when the cap is satisfied; `None` otherwise.
+    /// `(struct(...))? IN (SET) ([…])` predicate built over the
+    /// deduplicated keys when the cap is satisfied; `None` otherwise.
     ///
-    /// Per-partition arrays carry duplicates (the build side never dedups
-    /// before shipping), so each partition's `arr.len()` is an upper bound on
-    /// its distinct count. We start with a cheap pre-check (sum of lengths ≤
-    /// some fast-reject limit) before the dedup walk to keep the cost
-    /// proportional to actual partition sizes.
+    /// Per-partition arrays carry duplicates — each partition ships its raw
+    /// build-side join keys, dedup happens here. The dedup walk early-aborts
+    /// the moment we cross the cap, so the cost stays bounded by
+    /// `O(rows-until-cap+1-distinct-found)` rather than total input size.
     fn try_build_merged_inlist(
         &self,
         real_partitions: &[&PartitionData],