fix ut: test_no_pushdown_through_aggregates & test_plan_with_order_preserving_variants_preserves_fetch

Author: xudong963

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

@@ -276,7 +276,7 @@ fn test_no_pushdown_through_aggregates() {
         Ok:
           - FilterExec: b@1 = bar
           -   CoalesceBatchesExec: target_batch_size=100
-          -     AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt]
+          -     AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt], ordering_mode=PartiallySorted([0])
           -       CoalesceBatchesExec: target_batch_size=10
           -         DataSourceExec: file_groups={1 group: [[test.paqruet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
     "

datafusion/physical-optimizer/src/enforce_sorting/replace_with_order_preserving_variants.rs

@@ -139,11 +139,6 @@ pub fn plan_with_order_preserving_variants(
         return Ok(sort_input);
     } else if is_coalesce_partitions(&sort_input.plan) && is_spm_better {
         let child = &sort_input.children[0].plan;
-        let mut fetch = fetch;
-        if let Some(coalesce_fetch) = sort_input.plan.fetch() {
-            // Get the min fetch between the `fetch` and the coalesce's fetch:
-            fetch = Some(coalesce_fetch.min(fetch.unwrap_or(usize::MAX)))
-        };
         if let Some(ordering) = child.output_ordering() {
             let mut fetch = fetch;
             if let Some(coalesce_fetch) = sort_input.plan.fetch() {