perf: replace SMJ's join_filter_not_matched_map HashMap with Vec<FilterState> (#21517)

## Which issue does this PR close?

Partially addresses #20910, might be the last one for now.

## Rationale for this change
In full outer joins with filters, `BufferedBatch` tracks which buffered
rows had all filter evaluations fail using a `HashMap<u64, bool>`. This
map is read and written per-row in a hot loop during
`freeze_streamed_matched`. The HashMap pays ~40-60 bytes per entry
(8-byte u64 key + 1-byte bool value + hash table overhead), hashes every
key twice per iteration (once for `get`, once for `insert`), and
scatters entries across heap allocations with poor cache locality.

## What changes are included in this PR?

Replaces `HashMap<u64, bool>` with `Vec<FilterState>` indexed by
absolute row position within the batch. `FilterState` is a `#[repr(u8)]`
enum with three variants (`Unvisited`, `AllFailed`, `SomePassed`), so
the Vec is 1 byte per row — allocated once, direct-indexed, no hashing.
At the default batch size of 8192 rows the Vec is 8 KB (fits in L1
cache). Even at large batch sizes (32K+), 32 KB is still within L1 on
most machines, while the HashMap at 32K entries would consume ~1-2 MB of
scattered heap memory.

Three states are needed because a simple `Vec<bool>` cannot distinguish
"never matched" (handled separately by `null_joined`) from "matched but
all filters failed" (must be emitted as null-joined). The enum variant
names are self-documenting, unlike `Option<bool>` where
`None`/`Some(true)`/`Some(false)` would be opaque.

## Are these changes tested?
Existing tests.

## Are there any user-facing changes?
No.

Author: mbutrovich

datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs

@@ -202,6 +202,24 @@ impl StreamedBatch {
     }
 }
 
+/// Per-row filter outcome tracking for full outer joins.
+///
+/// In a full outer join with a filter, buffered rows that match on join
+/// keys but fail every filter evaluation must be emitted with NULLs on
+/// the streamed side. Three states are needed because a simple boolean
+/// cannot distinguish "never matched" (handled by [`BufferedBatch::null_joined`])
+/// from "matched but all filters failed" (must be emitted as null-joined).
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum FilterState {
+    /// Row never appeared in a matched pair.
+    Unvisited = 0,
+    /// Row matched streamed rows, but all filter evaluations failed.
+    AllFailed = 1,
+    /// Row matched and at least one filter evaluation passed.
+    SomePassed = 2,
+}
+
 /// A buffered batch that contains contiguous rows with same join key
 ///
 /// `BufferedBatch` can exist as either an in-memory `RecordBatch` or a `RefCountedTempFile` on disk.
@@ -217,11 +235,9 @@ pub(super) struct BufferedBatch {
     pub null_joined: Vec<usize>,
     /// Size estimation used for reserving / releasing memory
     pub size_estimation: usize,
-    /// The indices of buffered batch that the join filter doesn't satisfy.
-    /// This is a map between right row index and a boolean value indicating whether all joined row
-    /// of the right row does not satisfy the filter .
-    /// When dequeuing the buffered batch, we need to produce null joined rows for these indices.
-    pub join_filter_not_matched_map: HashMap<u64, bool>,
+    /// Tracks filter outcomes for buffered rows in full outer joins.
+    /// Indexed by absolute row position within the batch. See [`FilterState`].
+    pub join_filter_status: Vec<FilterState>,
     /// Current buffered batch number of rows. Equal to batch.num_rows()
     /// but if batch is spilled to disk this property is preferable
     /// and less expensive
@@ -258,7 +274,7 @@ impl BufferedBatch {
             join_arrays,
             null_joined: vec![],
             size_estimation,
-            join_filter_not_matched_map: HashMap::new(),
+            join_filter_status: vec![FilterState::Unvisited; num_rows],
             num_rows,
         }
     }
@@ -1250,12 +1266,16 @@ impl MaterializingSortMergeJoinStream {
             return Ok(());
         }
 
-        // For buffered row which is joined with streamed side rows but all joined rows
-        // don't satisfy the join filter
+        // Collect buffered rows that matched on join keys but had every
+        // filter evaluation fail — these must be emitted with NULLs on
+        // the streamed side to satisfy full outer join semantics.
         let not_matched_buffered_indices = buffered_batch
-            .join_filter_not_matched_map
+            .join_filter_status
             .iter()
-            .filter_map(|(idx, failed)| if *failed { Some(*idx) } else { None })
+            .enumerate()
+            .filter_map(|(i, state)| {
+                matches!(state, FilterState::AllFailed).then_some(i as u64)
+            })
             .collect::<Vec<_>>();
 
         let buffered_indices =
@@ -1270,7 +1290,9 @@ impl MaterializingSortMergeJoinStream {
             self.joined_record_batches
                 .push_batch_with_null_metadata(record_batch, self.join_type);
         }
-        buffered_batch.join_filter_not_matched_map.clear();
+        buffered_batch
+            .join_filter_status
+            .fill(FilterState::Unvisited);
 
         Ok(())
     }
@@ -1443,15 +1465,18 @@ impl MaterializingSortMergeJoinStream {
                             if right.is_null(i) {
                                 continue;
                             }
-                            let buffered_index = right.value(i);
-                            buffered_batch.join_filter_not_matched_map.insert(
-                                buffered_index,
-                                *buffered_batch
-                                    .join_filter_not_matched_map
-                                    .get(&buffered_index)
-                                    .unwrap_or(&true)
-                                    && !pre_mask.value(offset + i),
-                            );
+                            let idx = right.value(i) as usize;
+                            match buffered_batch.join_filter_status[idx] {
+                                FilterState::SomePassed => {}
+                                _ if pre_mask.value(offset + i) => {
+                                    buffered_batch.join_filter_status[idx] =
+                                        FilterState::SomePassed;
+                                }
+                                _ => {
+                                    buffered_batch.join_filter_status[idx] =
+                                        FilterState::AllFailed;
+                                }
+                            }
                         }
                         offset += chunk_len;
                     }