Support exec summary for limit executor

src/coprocessor/dag/builder.rs

@@ -160,8 +160,13 @@ impl DAGBuilder {
         let first = exec_descriptors
             .next()
             .ok_or_else(|| Error::Other(box_err!("has no executor")))?;
+
         let mut src = Self::build_normal_first_executor::<_, C>(first, store, ranges, collect)?;
+        let mut summary_slot_index = 0;
+
         for mut exec in exec_descriptors {
+            summary_slot_index += 1;
+
             let curr: Box<dyn Executor + Send> = match exec.get_tp() {
                 ExecType::TypeTableScan | ExecType::TypeIndexScan => {
                     return Err(box_err!("got too much *scan exec, should be only one"));
@@ -184,7 +189,11 @@ impl DAGBuilder {
                 ExecType::TypeTopN => {
                     Box::new(TopNExecutor::new(exec.take_topN(), Arc::clone(&ctx), src)?)
                 }
-                ExecType::TypeLimit => Box::new(LimitExecutor::new(exec.take_limit(), src)),
+                ExecType::TypeLimit => Box::new(LimitExecutor::new(
+                    C::new(summary_slot_index),
+                    exec.take_limit(),
+                    src,
+                )),
             };
             src = curr;
         }

src/coprocessor/dag/executor/limit.rs

@@ -3,22 +3,24 @@
 use tipb::executor::Limit;
 
 use super::ExecutorMetrics;
-use crate::coprocessor::dag::exec_summary::ExecSummary;
+use crate::coprocessor::dag::exec_summary::{ExecSummary, ExecSummaryCollector};
 use crate::coprocessor::dag::executor::{Executor, Row};
 use crate::coprocessor::dag::expr::EvalWarnings;
 use crate::coprocessor::Result;
 
 /// Retrieves rows from the source executor and only produces part of the rows.
-pub struct LimitExecutor<'a> {
+pub struct LimitExecutor<C: ExecSummaryCollector> {
+    summary_collector: C,
     limit: u64,
     cursor: u64,
-    src: Box<dyn Executor + Send + 'a>,
+    src: Box<dyn Executor + Send>,
     first_collect: bool,
 }
 
-impl<'a> LimitExecutor<'a> {
-    pub fn new(limit: Limit, src: Box<dyn Executor + Send + 'a>) -> LimitExecutor<'_> {
+impl<C: ExecSummaryCollector> LimitExecutor<C> {
+    pub fn new(summary_collector: C, limit: Limit, src: Box<dyn Executor + Send>) -> Self {
         LimitExecutor {
+            summary_collector,
             limit: limit.get_limit(),
             cursor: 0,
             src,
@@ -27,15 +29,19 @@ impl<'a> LimitExecutor<'a> {
     }
 }
 
-impl<'a> Executor for LimitExecutor<'a> {
+impl<C: ExecSummaryCollector> Executor for LimitExecutor<C> {
     fn next(&mut self) -> Result<Option<Row>> {
+        let timer = self.summary_collector.on_start_iterate();
         if self.cursor >= self.limit {
+            self.summary_collector.on_finish_iterate(timer, 0);
             return Ok(None);
         }
         if let Some(row) = self.src.next()? {
             self.cursor += 1;
+            self.summary_collector.on_finish_iterate(timer, 1);
             Ok(Some(row))
         } else {
+            self.summary_collector.on_finish_iterate(timer, 0);
             Ok(None)
         }
     }
@@ -62,6 +68,7 @@ impl<'a> Executor for LimitExecutor<'a> {
 
     fn collect_execution_summaries(&mut self, target: &mut [ExecSummary]) {
         self.src.collect_execution_summaries(target);
+        self.summary_collector.collect_into(target);
     }
 }
 
@@ -72,6 +79,7 @@ mod tests {
 
     use super::super::tests::*;
     use super::*;
+    use crate::coprocessor::dag::exec_summary::ExecSummaryCollectorDisabled;
 
     #[test]
     fn test_limit_executor() {
@@ -101,7 +109,7 @@ mod tests {
         let limit = 5;
         limit_meta.set_limit(limit);
         // init topn executor
-        let mut limit_ect = LimitExecutor::new(limit_meta, ts_ect);
+        let mut limit_ect = LimitExecutor::new(ExecSummaryCollectorDisabled, limit_meta, ts_ect);
         let mut limit_rows = Vec::with_capacity(limit as usize);
         while let Some(row) = limit_ect.next().unwrap() {
             limit_rows.push(row.take_origin());