// detachConditions distinguishs between access conditions and filter conditions from conditions.
func detachConditions(conditions []expression.Expression, table *model.TableInfo,
	idx *model.IndexInfo, colOffset int) ([]expression.Expression, []expression.Expression) {
	var pkName model.CIStr
	var accessConditions, filterConditions []expression.Expression
	if table.PKIsHandle {
		for _, colInfo := range table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				pkName = colInfo.Name
				break
			}
		}
	}
	for _, con := range conditions {
		if pkName.L != "" {
			checker := conditionChecker{
				tableName:    table.Name,
				pkName:       pkName,
				idx:          idx,
				columnOffset: colOffset}
			if checker.newCheck(con) {
				accessConditions = append(accessConditions, con)
				continue
			}
		}
		filterConditions = append(filterConditions, con)
	}

	return accessConditions, filterConditions
}

func (b *planBuilder) buildTableScanPlan(path *joinPath) Plan {
	tn := path.table
	p := &TableScan{
		Table:     tn.TableInfo,
		TableName: tn,
	}
	// Equal condition contains a column from previous joined table.
	p.RefAccess = len(path.eqConds) > 0
	p.SetFields(tn.GetResultFields())
	p.TableAsName = getTableAsName(p.Fields())
	var pkName model.CIStr
	if p.Table.PKIsHandle {
		for _, colInfo := range p.Table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				pkName = colInfo.Name
			}
		}
	}
	for _, con := range path.conditions {
		if pkName.L != "" {
			checker := conditionChecker{tableName: tn.TableInfo.Name, pkName: pkName}
			if checker.check(con) {
				p.AccessConditions = append(p.AccessConditions, con)
			} else {
				p.FilterConditions = append(p.FilterConditions, con)
			}
		} else {
			p.FilterConditions = append(p.FilterConditions, con)
		}
	}
	return p
}

// detachTableScanConditions distinguishes between access conditions and filter conditions from conditions.
func detachTableScanConditions(conditions []expression.Expression, table *model.TableInfo) ([]expression.Expression, []expression.Expression) {
	var pkName model.CIStr
	if table.PKIsHandle {
		for _, colInfo := range table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				pkName = colInfo.Name
				break
			}
		}
	}
	if pkName.L == "" {
		return nil, conditions
	}

	var accessConditions, filterConditions []expression.Expression
	checker := conditionChecker{
		tableName: table.Name,
		pkName:    pkName}
	for _, cond := range conditions {
		cond = pushDownNot(cond, false)
		if !checker.newCheck(cond) {
			filterConditions = append(filterConditions, cond)
			continue
		}
		accessConditions = append(accessConditions, cond)
		// TODO: it will lead to repeated compution cost.
		if checker.shouldReserve {
			filterConditions = append(filterConditions, cond)
			checker.shouldReserve = false
		}
	}

	return accessConditions, filterConditions
}

func (e *ShowExec) fetchShowIndex() error {
	tb, err := e.getTable()
	if err != nil {
		return errors.Trace(err)
	}
	if tb.Meta().PKIsHandle {
		var pkCol *table.Column
		for _, col := range tb.Cols() {
			if mysql.HasPriKeyFlag(col.Flag) {
				pkCol = col
				break
			}
		}
		data := types.MakeDatums(
			tb.Meta().Name.O, // Table
			0,                // Non_unique
			"PRIMARY",        // Key_name
			1,                // Seq_in_index
			pkCol.Name.O,     // Column_name
			"utf8_bin",       // Colation
			0,                // Cardinality
			nil,              // Sub_part
			nil,              // Packed
			"",               // Null
			"BTREE",          // Index_type
			"",               // Comment
			"",               // Index_comment
		)
		e.rows = append(e.rows, &Row{Data: data})
	}
	for _, idx := range tb.Indices() {
		for i, col := range idx.Meta().Columns {
			nonUniq := 1
			if idx.Meta().Unique {
				nonUniq = 0
			}
			var subPart interface{}
			if col.Length != types.UnspecifiedLength {
				subPart = col.Length
			}
			data := types.MakeDatums(
				tb.Meta().Name.O,  // Table
				nonUniq,           // Non_unique
				idx.Meta().Name.O, // Key_name
				i+1,               // Seq_in_index
				col.Name.O,        // Column_name
				"utf8_bin",        // Colation
				0,                 // Cardinality
				subPart,           // Sub_part
				nil,               // Packed
				"YES",             // Null
				idx.Meta().Tp.String(), // Index_type
				"",                 // Comment
				idx.Meta().Comment, // Index_comment
			)
			e.rows = append(e.rows, &Row{Data: data})
		}
	}
	return nil
}

func (b *planBuilder) buildTableScanPlan(tn *ast.TableName, conditions []ast.ExprNode) Plan {
	p := &TableScan{
		Table: tn.TableInfo,
	}
	p.SetFields(tn.GetResultFields())
	var pkName model.CIStr
	if p.Table.PKIsHandle {
		for _, colInfo := range p.Table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				pkName = colInfo.Name
			}
		}
	}
	for _, con := range conditions {
		if pkName.L != "" {
			checker := conditionChecker{tableName: tn.TableInfo.Name, pkName: pkName}
			if checker.check(con) {
				p.AccessConditions = append(p.AccessConditions, con)
			} else {
				p.FilterConditions = append(p.FilterConditions, con)
			}
		} else {
			p.FilterConditions = append(p.FilterConditions, con)
		}
	}
	return p
}

func (isp *InfoSchemaPlan) fetchStatisticsInTable(is infoschema.InfoSchema, schema *model.DBInfo, table *model.TableInfo) {
	if table.PKIsHandle {
		for _, col := range table.Columns {
			if mysql.HasPriKeyFlag(col.Flag) {
				record := []interface{}{
					catalogVal,    // TABLE_CATALOG
					schema.Name.O, // TABLE_SCHEMA
					table.Name.O,  // TABLE_NAME
					"0",           // NON_UNIQUE
					schema.Name.O, // INDEX_SCHEMA
					"PRIMARY",     // INDEX_NAME
					1,             // SEQ_IN_INDEX
					col.Name.O,    // COLUMN_NAME
					"A",           // COLLATION
					0,             // CARDINALITY
					nil,           // SUB_PART
					nil,           // PACKED
					"",            // NULLABLE
					"BTREE",       // INDEX_TYPE
					"",            // COMMENT
					"",            // INDEX_COMMENT
				}
				isp.rows = append(isp.rows, &plan.Row{Data: record})
			}
		}
	}
	for _, index := range table.Indices {
		nonUnique := "1"
		if index.Unique {
			nonUnique = "0"
		}
		for i, key := range index.Columns {
			col, _ := is.ColumnByName(schema.Name, table.Name, key.Name)
			nullable := "YES"
			if mysql.HasNotNullFlag(col.Flag) {
				nullable = ""
			}
			record := []interface{}{
				catalogVal,    // TABLE_CATALOG
				schema.Name.O, // TABLE_SCHEMA
				table.Name.O,  // TABLE_NAME
				nonUnique,     // NON_UNIQUE
				schema.Name.O, // INDEX_SCHEMA
				index.Name.O,  // INDEX_NAME
				i + 1,         // SEQ_IN_INDEX
				key.Name.O,    // COLUMN_NAME
				"A",           // COLLATION
				0,             // CARDINALITY
				nil,           // SUB_PART
				nil,           // PACKED
				nullable,      // NULLABLE
				"BTREE",       // INDEX_TYPE
				"",            // COMMENT
				"",            // INDEX_COMMENT
			}
			isp.rows = append(isp.rows, &plan.Row{Data: record})
		}
	}
}

// matchOrder checks if the plan has the same ordering as items.
func matchOrder(p Plan, items []*ast.ByItem) bool {
	switch x := p.(type) {
	case *Aggregate:
		return false
	case *IndexScan:
		if len(items) > len(x.Index.Columns) {
			return false
		}
		for i, item := range items {
			if item.Desc {
				return false
			}
			var rf *ast.ResultField
			switch y := item.Expr.(type) {
			case *ast.ColumnNameExpr:
				rf = y.Refer
			case *ast.PositionExpr:
				rf = y.Refer
			default:
				return false
			}
			if rf.Table.Name.L != x.Table.Name.L || rf.Column.Name.L != x.Index.Columns[i].Name.L {
				return false
			}
		}
		return true
	case *TableScan:
		if len(items) != 1 || !x.Table.PKIsHandle {
			return false
		}
		if items[0].Desc {
			return false
		}
		var refer *ast.ResultField
		switch x := items[0].Expr.(type) {
		case *ast.ColumnNameExpr:
			refer = x.Refer
		case *ast.PositionExpr:
			refer = x.Refer
		default:
			return false
		}
		if mysql.HasPriKeyFlag(refer.Column.Flag) {
			return true
		}
		return false
	case *JoinOuter:
		return false
	case *JoinInner:
		return false
	case *Sort:
		// Sort plan should not be checked here as there should only be one sort plan in a plan tree.
		return false
	case WithSrcPlan:
		return matchOrder(x.Src(), items)
	}
	return true
}

func equivHasIndex(rf *ast.ResultField) bool {
	if rf.Table.PKIsHandle && mysql.HasPriKeyFlag(rf.Column.Flag) {
		return true
	}
	for _, idx := range rf.Table.Indices {
		if len(idx.Columns) == 1 && idx.Columns[0].Name.L == rf.Column.Name.L {
			return true
		}
	}
	return false
}

// ColumnsToProto converts a slice of model.ColumnInfo to a slice of tipb.ColumnInfo.
func ColumnsToProto(columns []*model.ColumnInfo, pkIsHandle bool) []*tipb.ColumnInfo {
	cols := make([]*tipb.ColumnInfo, 0, len(columns))
	for _, c := range columns {
		col := columnToProto(c)
		if pkIsHandle && mysql.HasPriKeyFlag(c.Flag) {
			col.PkHandle = true
		} else {
			col.PkHandle = false
		}
		cols = append(cols, col)
	}
	return cols
}

func (r *refiner) sortBypass(p *Sort) {
	if r.indexScan != nil {
		idx := r.indexScan.Index
		if len(p.ByItems) > len(idx.Columns) {
			return
		}
		var desc bool
		for i, val := range p.ByItems {
			if val.Desc {
				desc = true
			}
			cn, ok := val.Expr.(*ast.ColumnNameExpr)
			if !ok {
				return
			}
			if r.indexScan.Table.Name.L != cn.Refer.Table.Name.L {
				return
			}
			indexColumn := idx.Columns[i]
			if indexColumn.Name.L != cn.Refer.Column.Name.L {
				return
			}
		}
		if desc {
			// TODO: support desc when index reverse iterator is supported.
			r.indexScan.Desc = true
			return
		}
		p.Bypass = true
	} else if r.tableScan != nil {
		if len(p.ByItems) != 1 {
			return
		}
		byItem := p.ByItems[0]
		if byItem.Desc {
			// TODO: support desc when table reverse iterator is supported.
			return
		}
		cn, ok := byItem.Expr.(*ast.ColumnNameExpr)
		if !ok {
			return
		}
		if !mysql.HasPriKeyFlag(cn.Refer.Column.Flag) {
			return
		}
		if !cn.Refer.Table.PKIsHandle {
			return
		}
		p.Bypass = true
	}
}

func (e *XSelectIndexExec) indexRowToTableRow(handle int64, indexRow []types.Datum) []types.Datum {
	tableRow := make([]types.Datum, len(e.indexPlan.Columns))
	for i, tblCol := range e.indexPlan.Columns {
		if mysql.HasPriKeyFlag(tblCol.Flag) && e.indexPlan.Table.PKIsHandle {
			tableRow[i] = types.NewIntDatum(handle)
			continue
		}
		for j, idxCol := range e.indexPlan.Index.Columns {
			if tblCol.Name.L == idxCol.Name.L {
				tableRow[i] = indexRow[j]
				break
			}
		}
	}
	return tableRow
}

// TableToProto converts a model.TableInfo to a tipb.TableInfo.
func TableToProto(t *model.TableInfo) *tipb.TableInfo {
	pt := &tipb.TableInfo{
		TableId: proto.Int64(t.ID),
	}
	cols := make([]*tipb.ColumnInfo, 0, len(t.Columns))
	for _, c := range t.Columns {
		col := columnToProto(c)
		if t.PKIsHandle && mysql.HasPriKeyFlag(c.Flag) {
			col.PkHandle = proto.Bool(true)
		} else {
			col.PkHandle = proto.Bool(false)
		}
		cols = append(cols, col)
	}
	pt.Columns = cols
	return pt
}

func isCoveringIndex(columns []*model.ColumnInfo, indexColumns []*model.IndexColumn, pkIsHandle bool) bool {
	for _, colInfo := range columns {
		if pkIsHandle && mysql.HasPriKeyFlag(colInfo.Flag) {
			continue
		}
		isIndexColumn := false
		for _, indexCol := range indexColumns {
			if colInfo.Name.L == indexCol.Name.L && indexCol.Length == types.UnspecifiedLength {
				isIndexColumn = true
				break
			}
		}
		if !isIndexColumn {
			return false
		}
	}
	return true
}

// NewColDesc returns a new ColDesc for a column.
func NewColDesc(col *Column) *ColDesc {
	// TODO: if we have no primary key and a unique index which's columns are all not null
	// we will set these columns' flag as PriKeyFlag
	// see https://dev.mysql.com/doc/refman/5.7/en/show-columns.html
	// create table
	name := col.Name
	nullFlag := "YES"
	if mysql.HasNotNullFlag(col.Flag) {
		nullFlag = "NO"
	}
	keyFlag := ""
	if mysql.HasPriKeyFlag(col.Flag) {
		keyFlag = "PRI"
	} else if mysql.HasUniKeyFlag(col.Flag) {
		keyFlag = "UNI"
	} else if mysql.HasMultipleKeyFlag(col.Flag) {
		keyFlag = "MUL"
	}
	var defaultValue interface{}
	if !mysql.HasNoDefaultValueFlag(col.Flag) {
		defaultValue = col.DefaultValue
	}

	extra := ""
	if mysql.HasAutoIncrementFlag(col.Flag) {
		extra = "auto_increment"
	} else if mysql.HasOnUpdateNowFlag(col.Flag) {
		extra = "on update CURRENT_TIMESTAMP"
	}

	return &ColDesc{
		Field:        name.O,
		Type:         col.GetTypeDesc(),
		Collation:    col.Collate,
		Null:         nullFlag,
		Key:          keyFlag,
		DefaultValue: defaultValue,
		Extra:        extra,
		Privileges:   defaultPrivileges,
		Comment:      "",
	}
}

func detachIndexFilterConditions(conditions []expression.Expression, indexColumns []*model.IndexColumn, table *model.TableInfo) ([]expression.Expression, []expression.Expression) {
	var pKName model.CIStr
	if table.PKIsHandle {
		for _, colInfo := range table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				pKName = colInfo.Name
				break
			}
		}
	}
	var indexConditions, tableConditions []expression.Expression
	for _, cond := range conditions {
		if checkIndexCondition(cond, indexColumns, pKName) {
			indexConditions = append(indexConditions, cond)
		} else {
			tableConditions = append(tableConditions, cond)
		}
	}
	return indexConditions, tableConditions
}

func (r *refiner) buildTableRange(p *TableScan) {
	var pkHandleColumn *model.ColumnInfo
	for _, colInfo := range p.Table.Columns {
		if mysql.HasPriKeyFlag(colInfo.Flag) && p.Table.PKIsHandle {
			pkHandleColumn = colInfo
		}
	}
	if pkHandleColumn == nil {
		return
	}
	rb := rangeBuilder{}
	rangePoints := fullRange
	checker := conditionChecker{pkName: pkHandleColumn.Name, tableName: p.Table.Name}
	for _, cond := range r.conditions {
		if checker.check(cond) {
			rangePoints = rb.intersection(rangePoints, rb.build(cond))
		}
	}
	p.Ranges = rb.buildTableRanges(rangePoints)
	r.err = rb.err
}

// IndexToProto converts a model.IndexInfo to a tipb.IndexInfo.
func IndexToProto(t *model.TableInfo, idx *model.IndexInfo) *tipb.IndexInfo {
	pi := &tipb.IndexInfo{
		TableId: t.ID,
		IndexId: idx.ID,
		Unique:  idx.Unique,
	}
	cols := make([]*tipb.ColumnInfo, 0, len(idx.Columns)+1)
	for _, c := range idx.Columns {
		cols = append(cols, columnToProto(t.Columns[c.Offset]))
	}
	if t.PKIsHandle {
		// Coprocessor needs to know PKHandle column info, so we need to append it.
		for _, col := range t.Columns {
			if mysql.HasPriKeyFlag(col.Flag) {
				colPB := columnToProto(col)
				colPB.PkHandle = true
				cols = append(cols, colPB)
				break
			}
		}
	}
	pi.Columns = cols
	return pi
}

func (e *ShowExec) fetchShowCreateTable() error {
	tb, err := e.getTable()
	if err != nil {
		return errors.Trace(err)
	}

	// TODO: let the result more like MySQL.
	var buf bytes.Buffer
	buf.WriteString(fmt.Sprintf("CREATE TABLE `%s` (\n", tb.Meta().Name.O))
	var pkCol *table.Column
	for i, col := range tb.Cols() {
		buf.WriteString(fmt.Sprintf("  `%s` %s", col.Name.O, col.GetTypeDesc()))
		if mysql.HasAutoIncrementFlag(col.Flag) {
			buf.WriteString(" NOT NULL AUTO_INCREMENT")
		} else {
			if mysql.HasNotNullFlag(col.Flag) {
				buf.WriteString(" NOT NULL")
			}
			if !mysql.HasNoDefaultValueFlag(col.Flag) {
				switch col.DefaultValue {
				case nil:
					buf.WriteString(" DEFAULT NULL")
				case "CURRENT_TIMESTAMP":
					buf.WriteString(" DEFAULT CURRENT_TIMESTAMP")
				default:
					buf.WriteString(fmt.Sprintf(" DEFAULT '%v'", col.DefaultValue))
				}
			}
			if mysql.HasOnUpdateNowFlag(col.Flag) {
				buf.WriteString(" ON UPDATE CURRENT_TIMESTAMP")
			}
		}
		if len(col.Comment) > 0 {
			buf.WriteString(fmt.Sprintf(" COMMENT '%s'", col.Comment))
		}
		if i != len(tb.Cols())-1 {
			buf.WriteString(",\n")
		}
		if tb.Meta().PKIsHandle && mysql.HasPriKeyFlag(col.Flag) {
			pkCol = col
		}
	}

	if pkCol != nil {
		// If PKIsHanle, pk info is not in tb.Indices(). We should handle it here.
		buf.WriteString(",\n")
		buf.WriteString(fmt.Sprintf(" PRIMARY KEY (`%s`)", pkCol.Name.O))
	}

	if len(tb.Indices()) > 0 || len(tb.Meta().ForeignKeys) > 0 {
		buf.WriteString(",\n")
	}

	for i, idx := range tb.Indices() {
		idxInfo := idx.Meta()
		if idxInfo.Primary {
			buf.WriteString("  PRIMARY KEY ")
		} else if idxInfo.Unique {
			buf.WriteString(fmt.Sprintf("  UNIQUE KEY `%s` ", idxInfo.Name.O))
		} else {
			buf.WriteString(fmt.Sprintf("  KEY `%s` ", idxInfo.Name.O))
		}

		cols := make([]string, 0, len(idxInfo.Columns))
		for _, c := range idxInfo.Columns {
			cols = append(cols, c.Name.O)
		}
		buf.WriteString(fmt.Sprintf("(`%s`)", strings.Join(cols, "`,`")))
		if i != len(tb.Indices())-1 {
			buf.WriteString(",\n")
		}
	}

	if len(tb.Indices()) > 0 && len(tb.Meta().ForeignKeys) > 0 {
		buf.WriteString(",\n")
	}

	for _, fk := range tb.Meta().ForeignKeys {
		if fk.State != model.StatePublic {
			continue
		}

		cols := make([]string, 0, len(fk.Cols))
		for _, c := range fk.Cols {
			cols = append(cols, c.O)
		}

		refCols := make([]string, 0, len(fk.RefCols))
		for _, c := range fk.Cols {
			refCols = append(refCols, c.O)
		}

		buf.WriteString(fmt.Sprintf("  CONSTRAINT `%s` FOREIGN KEY (`%s`)", fk.Name.O, strings.Join(cols, "`,`")))
		buf.WriteString(fmt.Sprintf(" REFERENCES `%s` (`%s`)", fk.RefTable.O, strings.Join(refCols, "`,`")))

		if ast.ReferOptionType(fk.OnDelete) != ast.ReferOptionNoOption {
			buf.WriteString(fmt.Sprintf(" ON DELETE %s", ast.ReferOptionType(fk.OnDelete)))
		}

		if ast.ReferOptionType(fk.OnUpdate) != ast.ReferOptionNoOption {
//.........这里部分代码省略.........

func dataForStatisticsInTable(schema *model.DBInfo, table *model.TableInfo) [][]types.Datum {
	rows := [][]types.Datum{}
	if table.PKIsHandle {
		for _, col := range table.Columns {
			if mysql.HasPriKeyFlag(col.Flag) {
				record := types.MakeDatums(
					catalogVal,    // TABLE_CATALOG
					schema.Name.O, // TABLE_SCHEMA
					table.Name.O,  // TABLE_NAME
					"0",           // NON_UNIQUE
					schema.Name.O, // INDEX_SCHEMA
					"PRIMARY",     // INDEX_NAME
					1,             // SEQ_IN_INDEX
					col.Name.O,    // COLUMN_NAME
					"A",           // COLLATION
					0,             // CARDINALITY
					nil,           // SUB_PART
					nil,           // PACKED
					"",            // NULLABLE
					"BTREE",       // INDEX_TYPE
					"",            // COMMENT
					"",            // INDEX_COMMENT
				)
				rows = append(rows, record)
			}
		}
	}
	nameToCol := make(map[string]*model.ColumnInfo, len(table.Columns))
	for _, c := range table.Columns {
		nameToCol[c.Name.L] = c
	}
	for _, index := range table.Indices {
		nonUnique := "1"
		if index.Unique {
			nonUnique = "0"
		}
		for i, key := range index.Columns {
			col := nameToCol[key.Name.L]
			nullable := "YES"
			if mysql.HasNotNullFlag(col.Flag) {
				nullable = ""
			}
			record := types.MakeDatums(
				catalogVal,    // TABLE_CATALOG
				schema.Name.O, // TABLE_SCHEMA
				table.Name.O,  // TABLE_NAME
				nonUnique,     // NON_UNIQUE
				schema.Name.O, // INDEX_SCHEMA
				index.Name.O,  // INDEX_NAME
				i+1,           // SEQ_IN_INDEX
				key.Name.O,    // COLUMN_NAME
				"A",           // COLLATION
				0,             // CARDINALITY
				nil,           // SUB_PART
				nil,           // PACKED
				nullable,      // NULLABLE
				"BTREE",       // INDEX_TYPE
				"",            // COMMENT
				"",            // INDEX_COMMENT
			)
			rows = append(rows, record)
		}
	}
	return rows
}

func (p *DataSource) handleTableScan(prop requiredProperty) (*physicalPlanInfo, *physicalPlanInfo, error) {
	table := p.Table
	var resultPlan PhysicalPlan
	ts := &PhysicalTableScan{
		Table:       p.Table,
		Columns:     p.Columns,
		TableAsName: p.TableAsName,
		DBName:      p.DBName,
	}
	ts.SetSchema(p.GetSchema())
	resultPlan = ts
	if sel, ok := p.GetParentByIndex(0).(*Selection); ok {
		newSel := *sel
		conds := make([]expression.Expression, 0, len(sel.Conditions))
		for _, cond := range sel.Conditions {
			conds = append(conds, cond.DeepCopy())
		}
		ts.AccessCondition, newSel.Conditions = detachTableScanConditions(conds, table)
		err := buildNewTableRange(ts)
		if err != nil {
			return nil, nil, errors.Trace(err)
		}
		if len(newSel.Conditions) > 0 {
			newSel.SetChildren(ts)
			resultPlan = &newSel
		}
	} else {
		ts.Ranges = []TableRange{{math.MinInt64, math.MaxInt64}}
	}

	statsTbl := p.statisticTable
	rowCount := uint64(statsTbl.Count)
	if table.PKIsHandle {
		for i, colInfo := range ts.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				ts.pkCol = p.GetSchema()[i]
				break
			}
		}
		var offset int
		for _, colInfo := range table.Columns {
			if mysql.HasPriKeyFlag(colInfo.Flag) {
				offset = colInfo.Offset
				break
			}
		}
		rowCount = 0
		for _, rg := range ts.Ranges {
			var cnt int64
			var err error
			if rg.LowVal == math.MinInt64 && rg.HighVal == math.MaxInt64 {
				cnt = statsTbl.Count
			} else if rg.LowVal == math.MinInt64 {
				cnt, err = statsTbl.Columns[offset].LessRowCount(types.NewDatum(rg.HighVal))
			} else if rg.HighVal == math.MaxInt64 {
				cnt, err = statsTbl.Columns[offset].GreaterRowCount(types.NewDatum(rg.LowVal))
			} else {
				cnt, err = statsTbl.Columns[offset].BetweenRowCount(types.NewDatum(rg.LowVal), types.NewDatum(rg.HighVal))
			}
			if err != nil {
				return nil, nil, errors.Trace(err)
			}
			rowCount += uint64(cnt)
		}
	}
	rowCounts := []uint64{rowCount}
	return resultPlan.matchProperty(prop, rowCounts), resultPlan.matchProperty(nil, rowCounts), nil
}