fix: limit row cnt (#138)

Change the `LIMIT` output cardinality to `min(child, fetch)`. In the
below, I hard-coded `PhysicalScan`'s cardinality to 10, because cli does
not support stats yet.

```
❯     create table t1(v1 int);

0 rows in set. Query took 0.011 seconds.

Execution took 0.000 secs, Planning took 0.000 secs
❯     insert into t1 values (0), (1), (2), (3);

❯ explain verbose select * from t1 limit 2;
+--------------------------------------------------+-------------------------------------------------------------------------------------------------+
| plan_type                                        | plan                                                                                            |
+--------------------------------------------------+-------------------------------------------------------------------------------------------------+
| logical_plan after datafusion                    | Limit: skip=0, fetch=2                                                                          |
|                                                  |   Projection: t1.v1                                                                             |
|                                                  |     TableScan: t1                                                                               |
| logical_plan after optd                          | LogicalLimit { skip: 0, fetch: 2 }                                                              |
|                                                  | └── LogicalProjection { exprs: [ #0 ] }                                                         |
|                                                  |     └── LogicalScan { table: t1 }                                                               |
| physical_plan after optd                         | PhysicalLimit { skip: 0, fetch: 2, cost: weighted=3.12,row_cnt=1.00,compute=2.12,io=1.00 }      |
|                                                  | └── PhysicalProjection { exprs: [ #0 ], cost: weighted=1.06,row_cnt=1.00,compute=0.06,io=1.00 } |
|                                                  |     └── PhysicalScan { table: t1, cost: weighted=1.00,row_cnt=1.00,compute=0.00,io=1.00 }       |
| physical_plan after optd-join-order              | t1                                                                                              |
| physical_plan after optd-all-join-orders         | SAME TEXT AS ABOVE                                                                              |
| physical_plan after optd-all-logical-join-orders | SAME TEXT AS ABOVE                                                                              |
| physical_plan                                    | GlobalLimitExec: skip=0, fetch=2                                                                |
|                                                  |   ProjectionExec: expr=[<expr>@0 as col0]                                                       |
|                                                  |     MemoryExec: partitions=1, partition_sizes=[1]                                               |
|                                                  |                                                                                                 |
+--------------------------------------------------+-------------------------------------------------------------------------------------------------+
```

Author: Gun9niR

optd-datafusion-repr/src/cost/base_cost.rs

@@ -1,6 +1,8 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::plan_nodes::{BinOpType, ColumnRefExpr, LogOpType, OptRelNode, UnOpType};
+use crate::plan_nodes::{
+    BinOpType, ColumnRefExpr, ConstantExpr, ConstantType, LogOpType, OptRelNode, UnOpType,
+};
 use crate::properties::column_ref::{ColumnRefPropertyBuilder, GroupColumnRefs};
 use crate::{
     plan_nodes::{OptRelNodeRef, OptRelNodeTyp},
@@ -392,8 +394,39 @@ impl<M: MostCommonValues, D: Distribution> CostModel<OptRelNodeTyp> for OptCostM
             OptRelNodeTyp::PhysicalEmptyRelation => Self::cost(0.5, 0.01, 0.0),
             OptRelNodeTyp::PhysicalLimit => {
                 let (row_cnt, compute_cost, _) = Self::cost_tuple(&children[0]);
-                let selectivity = 0.001;
-                Self::cost((row_cnt * selectivity).max(1.0), compute_cost, 0.0)
+                let row_cnt = if let Some(context) = context {
+                    if let Some(optimizer) = optimizer {
+                        let mut fetch_expr =
+                            optimizer.get_all_group_bindings(context.children_group_ids[2], false);
+                        assert!(
+                            fetch_expr.len() == 1,
+                            "fetch expression should be the only expr in the group"
+                        );
+                        let fetch_expr = fetch_expr.pop().unwrap();
+                        assert!(
+                            matches!(
+                                fetch_expr.typ,
+                                OptRelNodeTyp::Constant(ConstantType::UInt64)
+                            ),
+                            "fetch type can only be UInt64"
+                        );
+                        let fetch = ConstantExpr::from_rel_node(fetch_expr)
+                            .unwrap()
+                            .value()
+                            .as_u64();
+                        // u64::MAX represents None
+                        if fetch == u64::MAX {
+                            row_cnt
+                        } else {
+                            row_cnt.min(fetch as f64)
+                        }
+                    } else {
+                        (row_cnt * INVALID_SELECTIVITY).max(1.0)
+                    }
+                } else {
+                    (row_cnt * INVALID_SELECTIVITY).max(1.0)
+                };
+                Self::cost(row_cnt, compute_cost, 0.0)
             }
             OptRelNodeTyp::PhysicalFilter => {
                 let (row_cnt, _, _) = Self::cost_tuple(&children[0]);