cherry pick stats related fix to 1.2-dev

pkg/pb/statsinfo/statsinfo.go

@@ -20,7 +20,7 @@ func (sc *StatsInfo) NeedUpdate(currentApproxObjNum int64) bool {
 	if sc.ApproxObjectNumber == 0 || sc.AccurateObjectNumber == 0 {
 		return true
 	}
-	if math.Abs(float64(sc.ApproxObjectNumber-currentApproxObjNum)) >= 10 {
+	if math.Abs(float64(sc.ApproxObjectNumber-currentApproxObjNum)) >= float64(sc.AccurateObjectNumber) {
 		return true
 	}
 	if float64(currentApproxObjNum)/float64(sc.ApproxObjectNumber) > 1.05 || float64(currentApproxObjNum)/float64(sc.ApproxObjectNumber) < 0.95 {

pkg/sql/plan/apply_indices.go

@@ -514,9 +514,13 @@ func (builder *QueryBuilder) applyIndicesForFiltersRegularIndex(nodeID int32, no
 						idxColMap[[2]int32{node.BindingTags[0], colIdx}] = mappedExpr
 					}
 
+					compositeFilterSel := 1.0
 					serialArgs := make([]*plan.Expr, len(hitFilterIdx))
 					for i := range hitFilterIdx {
-						serialArgs[i] = DeepCopyExpr(node.FilterList[hitFilterIdx[i]].GetF().Args[1])
+						filter := node.FilterList[hitFilterIdx[i]]
+						serialArgs[i] = DeepCopyExpr(filter.GetF().Args[1])
+						estimateExprSelectivity(filter, builder)
+						compositeFilterSel = compositeFilterSel * filter.Selectivity
 					}
 					rightArg, _ := BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", serialArgs)
 
@@ -536,6 +540,7 @@ func (builder *QueryBuilder) applyIndicesForFiltersRegularIndex(nodeID int32, no
 						},
 						rightArg,
 					})
+					idxFilter.Selectivity = compositeFilterSel
 
 					newFilterList := make([]*plan.Expr, 0, len(missFilterIdx)+1)
 					for _, idx := range missFilterIdx {
@@ -658,9 +663,15 @@ END0:
 			col.RelPos = idxTag
 			col.ColPos = 0
 		} else {
+
+			compositeFilterSel := 1.0
+
 			serialArgs := make([]*plan.Expr, len(filterIdx))
 			for i := range filterIdx {
-				serialArgs[i] = DeepCopyExpr(node.FilterList[filterIdx[i]].GetF().Args[1])
+				filter := node.FilterList[filterIdx[i]]
+				serialArgs[i] = DeepCopyExpr(filter.GetF().Args[1])
+				estimateExprSelectivity(filter, builder)
+				compositeFilterSel = compositeFilterSel * filter.Selectivity
 			}
 			rightArg, _ := BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", serialArgs)
 
@@ -680,6 +691,7 @@ END0:
 				},
 				rightArg,
 			})
+			idxFilter.Selectivity = compositeFilterSel
 		}
 
 		idxTableNodeID := builder.appendNode(&plan.Node{
@@ -1001,6 +1013,7 @@ func (builder *QueryBuilder) applyIndicesForJoins(nodeID int32, node *plan.Node,
 		}, builder.ctxByNode[nodeID])
 
 		node.RuntimeFilterBuildList = append(node.RuntimeFilterBuildList, MakeRuntimeFilter(rfTag, len(condIdx) < numParts, GetInFilterCardLimitOnPK(leftChild.Stats.TableCnt), rfBuildExpr))
+		recalcStatsByRuntimeFilter(builder.qry.Nodes[idxTableNodeID], node, builder)
 
 		pkIdx := leftChild.TableDef.Name2ColIndex[leftChild.TableDef.Pkey.PkeyColName]
 		pkExpr := &plan.Expr{

pkg/sql/plan/build_constraint_util.go

@@ -1370,10 +1370,7 @@ func appendPrimaryConstraintPlan(
 				scanNode.RuntimeFilterProbeList = nil // can not use both
 			} else {
 				tableScanId = builder.appendNode(scanNode, bindCtx)
-				// temporary solution for the plan of dml go without optimizer
-				// prevent table scan from running on multiple CNs.
-				// because the runtime filter can only run on one now.
-				scanNode.Stats = DefaultMinimalStats()
+				scanNode.Stats.ForceOneCN = true
 			}
 
 			// Perform partition pruning on the full table scan of the partitioned table in the insert statement
@@ -1407,6 +1404,7 @@ func appendPrimaryConstraintPlan(
 					},
 				}
 				fuzzyFilterNode.RuntimeFilterBuildList = []*plan.RuntimeFilterSpec{MakeRuntimeFilter(rfTag, false, GetInFilterCardLimitOnPK(scanNode.Stats.TableCnt), buildExpr)}
+				recalcStatsByRuntimeFilter(scanNode, fuzzyFilterNode, builder)
 			}
 
 			lastNodeId = builder.appendNode(fuzzyFilterNode, bindCtx)
@@ -1527,6 +1525,7 @@ func appendPrimaryConstraintPlan(
 					RuntimeFilterBuildList: []*plan.RuntimeFilterSpec{MakeRuntimeFilter(rfTag, false, GetInFilterCardLimitOnPK(scanNode.Stats.TableCnt), buildExpr)},
 				}
 				lastNodeId = builder.appendNode(joinNode, bindCtx)
+				recalcStatsByRuntimeFilter(scanNode, joinNode, builder)
 
 				// append agg node.
 				aggGroupBy := []*Expr{

pkg/sql/plan/build_dml_util.go

@@ -3092,6 +3092,7 @@ func appendDeleteIndexTablePlan(
 		ProjectList:            projectList,
 		RuntimeFilterBuildList: []*plan.RuntimeFilterSpec{MakeRuntimeFilter(rfTag, false, GetInFilterCardLimitOnPK(builder.qry.Nodes[leftId].Stats.TableCnt), buildExpr)},
 	}, bindCtx)
+	recalcStatsByRuntimeFilter(builder.qry.Nodes[leftId], builder.qry.Nodes[lastNodeId], builder)
 	return lastNodeId, nil
 }
 

pkg/sql/plan/explain/explain_node.go

@@ -989,7 +989,7 @@ func (c *CostDescribeImpl) GetDescription(ctx context.Context, options *ExplainO
 		if c.Stats.BlockNum > 0 {
 			blockNumStr = " blockNum=" + strconv.FormatInt(int64(c.Stats.BlockNum), 10)
 		}
-		if c.Stats.HashmapStats != nil && c.Stats.HashmapStats.HashmapSize > 0 {
+		if c.Stats.HashmapStats != nil && c.Stats.HashmapStats.HashmapSize > 1 {
 			hashmapSizeStr = " hashmapSize=" + strconv.FormatFloat(c.Stats.HashmapStats.HashmapSize, 'f', 2, 64)
 		}
 		buf.WriteString(" (cost=" + strconv.FormatFloat(c.Stats.Cost, 'f', 2, 64) +

pkg/sql/plan/runtime_filter.go

@@ -177,8 +177,7 @@ func (builder *QueryBuilder) generateRuntimeFilters(nodeID int32) {
 			},
 		}
 		node.RuntimeFilterBuildList = append(node.RuntimeFilterBuildList, MakeRuntimeFilter(rfTag, false, inLimit, buildExpr))
-
-		recalcStatsByRuntimeFilter(leftChild, node, rightChild.Stats.Selectivity)
+		recalcStatsByRuntimeFilter(leftChild, node, builder)
 		return
 	}
 
@@ -255,5 +254,5 @@ func (builder *QueryBuilder) generateRuntimeFilters(nodeID int32) {
 	buildExpr, _ := BindFuncExprImplByPlanExpr(builder.GetContext(), "serial", buildArgs)
 
 	node.RuntimeFilterBuildList = append(node.RuntimeFilterBuildList, MakeRuntimeFilter(rfTag, cnt < len(tableDef.Pkey.Names), GetInFilterCardLimitOnPK(leftChild.Stats.TableCnt), buildExpr))
-	recalcStatsByRuntimeFilter(leftChild, node, rightChild.Stats.Selectivity)
+	recalcStatsByRuntimeFilter(leftChild, node, builder)
 }

pkg/sql/plan/stats.go

@@ -470,47 +470,46 @@ func estimateNonEqualitySelectivity(expr *plan.Expr, funcName string, builder *Q
 
 func estimateExprSelectivity(expr *plan.Expr, builder *QueryBuilder) float64 {
 	if expr == nil {
-		return 1
+		return 1.0
+	}
+	if expr.Selectivity != 0 {
+		return expr.Selectivity // already calculated
 	}
 
+	ret := 1.0
 	switch exprImpl := expr.Expr.(type) {
 	case *plan.Expr_F:
 		funcName := exprImpl.F.Func.ObjName
 		switch funcName {
 		case "=":
-			return estimateEqualitySelectivity(expr, builder)
+			ret = estimateEqualitySelectivity(expr, builder)
 		case "!=", "<>":
-			return 0.9
+			ret = 0.9
 		case ">", "<", ">=", "<=", "between":
-			return estimateNonEqualitySelectivity(expr, funcName, builder)
+			ret = estimateNonEqualitySelectivity(expr, funcName, builder)
 		case "and":
 			sel1 := estimateExprSelectivity(exprImpl.F.Args[0], builder)
 			sel2 := estimateExprSelectivity(exprImpl.F.Args[1], builder)
 			if canMergeToBetweenAnd(exprImpl.F.Args[0], exprImpl.F.Args[1]) && (sel1+sel2) > 1 {
-				return sel1 + sel2 - 1
+				ret = sel1 + sel2 - 1
 			} else {
-				return andSelectivity(sel1, sel2)
+				ret = andSelectivity(sel1, sel2)
 			}
 		case "or":
 			sel1 := estimateExprSelectivity(exprImpl.F.Args[0], builder)
 			sel2 := estimateExprSelectivity(exprImpl.F.Args[1], builder)
-			return orSelectivity(sel1, sel2)
+			ret = orSelectivity(sel1, sel2)
 		case "not":
-			return 1 - estimateExprSelectivity(exprImpl.F.Args[0], builder)
+			ret = 1 - estimateExprSelectivity(exprImpl.F.Args[0], builder)
 		case "like":
-			return 0.2
+			ret = 0.2
 		case "prefix_eq":
-			ndv := getExprNdv(expr, builder)
-			if ndv > 10 {
-				return 10 / ndv
-			}
-			return 0.5
+			ret = 0.0001 // should never go here
 		case "in", "prefix_in":
-			card := float64(1)
+			card := 1.0
 			switch arg1Impl := exprImpl.F.Args[1].Expr.(type) {
 			case *plan.Expr_Vec:
 				card = float64(arg1Impl.Vec.Len)
-
 			case *plan.Expr_List:
 				card = float64(len(arg1Impl.List.List))
 			}
@@ -519,22 +518,24 @@ func estimateExprSelectivity(expr *plan.Expr, builder *QueryBuilder) float64 {
 			}
 			ndv := getExprNdv(expr, builder)
 			if ndv > card {
-				return card / ndv
+				ret = card / ndv
+			} else {
+				ret = 0.5
 			}
-			return 0.5
 		case "prefix_between":
-			return 0.1
+			ret = 0.1
 		case "isnull", "is_null":
-			return getNullSelectivity(exprImpl.F.Args[0], builder, true)
+			ret = getNullSelectivity(exprImpl.F.Args[0], builder, true)
 		case "isnotnull", "is_not_null":
-			return getNullSelectivity(exprImpl.F.Args[0], builder, false)
+			ret = getNullSelectivity(exprImpl.F.Args[0], builder, false)
 		default:
-			return 0.15
+			ret = 0.15
 		}
 	case *plan.Expr_Lit:
-		return 1
+		ret = 1.0
 	}
-	return 1
+	expr.Selectivity = ret
+	return ret
 }
 
 func estimateFilterWeight(expr *plan.Expr, w float64) float64 {
@@ -867,6 +868,7 @@ func ReCalcNodeStats(nodeID int32, builder *QueryBuilder, recursive bool, leafNo
 			node.Stats.Outcnt = childStats.Outcnt
 			node.Stats.Cost = childStats.Outcnt
 			node.Stats.Selectivity = childStats.Selectivity
+			node.Stats.BlockNum = childStats.BlockNum
 		}
 	}
 
@@ -992,16 +994,33 @@ func foldTableScanFilters(proc *process.Process, qry *Query, nodeId int32) error
 	return nil
 }
 
-func recalcStatsByRuntimeFilter(node *plan.Node, joinNode *plan.Node, runtimeFilterSel float64) {
-	if node.NodeType != plan.Node_TABLE_SCAN {
+func recalcStatsByRuntimeFilter(scanNode *plan.Node, joinNode *plan.Node, builder *QueryBuilder) {
+	if scanNode.NodeType != plan.Node_TABLE_SCAN {
+		return
+	}
+	if joinNode.NodeType != plan.Node_JOIN && joinNode.NodeType != plan.Node_FUZZY_FILTER {
+		return
+	}
+
+	if joinNode.JoinType == plan.Node_INDEX || joinNode.NodeType == plan.Node_FUZZY_FILTER {
+		scanNode.Stats.Outcnt = builder.qry.Nodes[joinNode.Children[1]].Stats.Outcnt
+		scanNode.Stats.BlockNum = int32(scanNode.Stats.Outcnt/3) + 1
+		scanNode.Stats.Cost = float64(scanNode.Stats.BlockNum * DefaultBlockMaxRows)
+		if scanNode.Stats.Cost > scanNode.Stats.TableCnt {
+			scanNode.Stats.Cost = scanNode.Stats.TableCnt
+			scanNode.Stats.BlockNum = int32(scanNode.Stats.TableCnt / DefaultBlockMaxRows)
+		}
+		scanNode.Stats.Selectivity = scanNode.Stats.Outcnt / scanNode.Stats.TableCnt
 		return
 	}
-	node.Stats.Cost *= runtimeFilterSel
-	node.Stats.Outcnt *= runtimeFilterSel
-	if node.Stats.Cost < 1 {
-		node.Stats.Cost = 1
+	runtimeFilterSel := builder.qry.Nodes[joinNode.Children[1]].Stats.Selectivity
+	scanNode.Stats.Cost *= runtimeFilterSel
+	scanNode.Stats.Outcnt *= runtimeFilterSel
+	if scanNode.Stats.Cost < 1 {
+		scanNode.Stats.Cost = 1
 	}
-	node.Stats.BlockNum = int32(node.Stats.Outcnt/2) + 1
+	scanNode.Stats.BlockNum = int32(scanNode.Stats.Outcnt/3) + 1
+	scanNode.Stats.Selectivity = andSelectivity(scanNode.Stats.Selectivity, runtimeFilterSel)
 }
 
 func calcScanStats(node *plan.Node, builder *QueryBuilder) *plan.Stats {

pkg/sql/plan/utils.go

@@ -1798,6 +1798,7 @@ func doFormatExpr(expr *plan.Expr, out *bytes.Buffer, depth int) {
 	default:
 		out.WriteString(fmt.Sprintf("%sExpr_Unknown(%s)", prefix, expr.String()))
 	}
+	out.WriteString(fmt.Sprintf("%sExpr_Selectivity(%v)", prefix, expr.Selectivity))
 }
 
 // databaseIsValid checks whether the database exists or not.

pkg/vm/engine/disttae/stats.go

@@ -168,7 +168,7 @@ func (gs *GlobalStats) ShouldUpdate(key pb.StatsInfoKey, entryNum int64) bool {
 	gs.mu.Lock()
 	defer gs.mu.Unlock()
 	info, ok := gs.mu.statsInfoMap[key]
-	if ok && info != nil && info.BlockNumber > entryNum {
+	if ok && info != nil && info.BlockNumber-entryNum > 64 {
 		return false
 	}
 	return true