try to fix

Author: xudong963

datafusion/physical-optimizer/src/limit_pushdown.rs

@@ -29,6 +29,8 @@ use datafusion_common::tree_node::{Transformed, TreeNodeRecursion};
 use datafusion_common::utils::combine_limit;
 use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
+use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
 /// This rule inspects [`ExecutionPlan`]'s and pushes down the fetch limit from
@@ -159,6 +161,18 @@ pub fn pushdown_limit_helper(
         ));
     }
 
+    if let Some(projection) = pushdown_plan.as_any().downcast_ref::<ProjectionExec>() {
+        // If `PlaceholderRowExec`, it's possible that the projection is from aggregate_statistics,
+        // So to be conservative, we should not push down the limit.
+        // Like this:
+        // LimitExec(fetch=1)
+        //    ProjectionExec(expr=[10 as count(*)])
+        //       PlaceholderRowExec
+        if projection.input().as_any().is::<PlaceholderRowExec>() {
+            return Ok((Transformed::no(pushdown_plan), global_state));
+        }
+    }
+
     // If we have a non-limit operator with fetch capability, update global
     // state as necessary:
     if pushdown_plan.fetch().is_some() {