add more comments

Author: adriangb

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -1214,9 +1214,67 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     );
 }
 
-/// Test that simulates behavior when some build side partitions are much slower to finish than others.
-/// In these cases we push down _partial_ filters that only apply scan filters to rows the belong to build side partitions that have completed.
-/// Once all build side partitions complete the full filter is pushed down without any hash key partition calculations.
+/// Test demonstrating progressive dynamic filter evolution in partitioned hash joins
+///
+/// This test validates that instead of waiting for all
+/// build-side partitions to complete before applying any filters, we apply partial filters
+/// immediately as each partition completes.
+/// To be able to evaluate partial filters we need to know which partition each probe row belongs to,
+/// so we push down a hash function to the probe side that computes the same hash as repartitioning will later on.
+///
+/// ## Test Scenario Setup
+///
+/// - **Build side**: Values [1, 2, 3, 4] distributed across 3 hash partitions
+/// - **Probe side**: Values [2, 3] that need to be filtered
+/// - **Partition 1 is artificially slowed**: Simulates real-world partition skew
+///
+/// ## Progressive Filter Evolution Demonstration
+///
+/// The test shows how the dynamic filter evolves through three distinct phases:
+///
+/// ### Phase 1: Initial State (All Partitions Building)
+/// ```sql
+/// -- No filter applied yet
+/// predicate=DynamicFilterPhysicalExpr [ true ]
+/// ```
+/// → All probe-side data passes through unfiltered
+///
+/// ### Phase 2: Progressive Filtering (Some Partitions Complete)
+/// ```sql
+/// -- Hash-based progressive filter after partition 0 completes
+/// predicate=DynamicFilterPhysicalExpr [
+///   CASE repartition_hash(id@0) % 3
+///     WHEN 0 THEN id@0 >= 3 AND id@0 <= 3  -- Only partition 0 bounds known
+///     ELSE true                              -- Pass through partitions 1,2 data
+///   END
+/// ]
+/// ```
+/// → Filters probe data for partition 0, passes through everything else safely
+///
+/// ### Phase 3: Final Optimization (All Partitions Complete)
+/// ```sql
+/// -- Optimized bounds-only filter
+/// predicate=DynamicFilterPhysicalExpr [
+///   id@0 >= 3 AND id@0 <= 3 OR    -- Partition 0 bounds
+///   id@0 >= 2 AND id@0 <= 2 OR    -- Partition 1 bounds
+///   id@0 >= 1 AND id@0 <= 4       -- Partition 2 bounds
+/// ]
+/// ```
+/// → Bounds filter with no hash computation overhead
+///
+/// ## Correctness Validation
+///
+/// The test verifies:
+/// 1. **No False Negatives**: All valid join results [2,3] are preserved throughout
+/// 2. **Progressive Improvement**: Filter selectivity increases as partitions complete
+/// 3. **Final Optimization**: Hash-based expressions are removed when all partitions finish
+/// 4. **Partition Isolation**: Each partition's filter only affects its own hash bucket
+///
+/// ## Real-World Impact
+///
+/// This optimization addresses common production scenarios where:
+/// - Some partitions finish much faster than others (data skew)
+/// - Waiting for large build sides before starting the probe sides increases latency
 #[tokio::test]
 #[cfg(not(feature = "force_hash_collisions"))] // this test relies on hash partitioning to separate rows
 async fn test_hashjoin_progressive_filter_reporting() {
@@ -1295,7 +1353,8 @@ async fn test_hashjoin_progressive_filter_reporting() {
         .unwrap(),
     ) as Arc<dyn ExecutionPlan>;
 
-    // expect the predicate to be pushed down into the probe side DataSource
+    // Verify the initial optimization - should show DynamicFilterPhysicalExpr is set up
+    // but not yet populated with any bounds (shows as "true" initially)
     insta::assert_snapshot!(
         OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new_post_optimization(), true),
         @r"
@@ -1347,7 +1406,13 @@ async fn test_hashjoin_progressive_filter_reporting() {
         }
     }
 
-    // Now check what our filter looks like
+    // CRITICAL VALIDATION: This snapshot shows the progressive filter in action!
+    // After partition 0 completes (but partition 1 is still blocked), we see:
+    // - CASE repartition_hash(id@0) % 3 WHEN 0 THEN id@0 >= 3 AND id@0 <= 3 ELSE true END
+    // This means:
+    //   - For rows that hash to partition 0: Apply bounds check (id >= 3 AND id <= 3)
+    //   - For rows that hash to partitions 1,2: Pass everything through (ELSE true)
+    // This is the core of progressive filtering - partial filtering without false negatives!
     #[cfg(not(feature = "force_hash_collisions"))]
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
@@ -1385,7 +1450,11 @@ async fn test_hashjoin_progressive_filter_reporting() {
         batches.push(batch.unwrap());
     }
 
-    // Look at the final plan
+    // FINAL OPTIMIZATION VALIDATION: All partitions complete - filter is now optimized!
+    // The hash-based CASE expression has been replaced with a simple OR of bounds:
+    // - id@0 >= 3 AND id@0 <= 3 OR id@0 >= 2 AND id@0 <= 2 OR id@0 >= 1 AND id@0 <= 4
+    // This is much more efficient - no hash computation needed, just bounds checks.
+    // Each OR clause represents one partition's bounds: [3,3], [2,2], [1,4]
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
         @r"