fix: only emit early after skip probe evaluates and decides not to skip

The early emission was firing before the skip aggregation probe
could evaluate (needs 100K rows), causing regressions for very
high-cardinality GROUP BY queries (e.g. Q32 GROUP BY WatchID,
ClientIP was 1.53x slower).

Fix: only enable early emission AFTER the skip probe has evaluated
and decided NOT to skip. This ensures:
- Before 100K rows: no early emission (let probe evaluate first)
- High cardinality (ratio >= 0.8): skip probe takes over, no emission
- Medium cardinality (ratio < 0.8): early emission keeps hash table
  cache-friendly

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

Author: Dandandan

datafusion/physical-plan/src/aggregates/row_hash.rs

@@ -825,14 +825,11 @@ impl Stream for GroupedHashAggregateStream {
                             // prevents early emission from interfering with
                             // very-high-cardinality queries (ratio >= 0.8)
                             // where the skip probe should take over entirely.
-                            let early_emit_enabled =
-                                self.early_emit_max_table_size > 0
-                                    && match &self.skip_aggregation_probe {
-                                        None => true,
-                                        Some(p) => {
-                                            p.ratio().is_some() && !p.should_skip()
-                                        }
-                                    };
+                            let early_emit_enabled = self.early_emit_max_table_size > 0
+                                && match &self.skip_aggregation_probe {
+                                    None => true,
+                                    Some(p) => p.ratio().is_some() && !p.should_skip(),
+                                };
                             if early_emit_enabled {
                                 let table_size = self.group_values.size()
                                     + self