1 В избранное 0 Ответвления 0

OSCHINA-MIRROR/hanchuanchuan-goInception

Присоединиться к Gitlife
Откройте для себя и примите участие в публичных проектах с открытым исходным кодом с участием более 10 миллионов разработчиков. Приватные репозитории также полностью бесплатны :)
Присоединиться бесплатно
Клонировать/Скачать
plan_to_pb.go 6 КБ
Копировать Редактировать Исходные данные Просмотреть построчно История
hanchuanchuan Отправлено 6 лет назад 27f3c5a
// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.
package core
import (
"github.com/hanchuanchuan/goInception/expression"
"github.com/hanchuanchuan/goInception/expression/aggregation"
"github.com/hanchuanchuan/goInception/model"
"github.com/hanchuanchuan/goInception/sessionctx"
"github.com/hanchuanchuan/goInception/table"
"github.com/hanchuanchuan/goInception/tablecodec"
"github.com/hanchuanchuan/goInception/util/ranger"
"github.com/pingcap/errors" // ToPB implements PhysicalPlan ToPB interface.
"github.com/pingcap/tipb/go-tipb"
)
func (p *basePhysicalPlan) ToPB(_ sessionctx.Context) (*tipb.Executor, error) {
return nil, errors.Errorf("plan %s fails converts to PB", p.basePlan.ExplainID())
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalHashAgg) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
sc := ctx.GetSessionVars().StmtCtx
client := ctx.GetClient()
aggExec := &tipb.Aggregation{
GroupBy: expression.ExpressionsToPBList(sc, p.GroupByItems, client),
}
for _, aggFunc := range p.AggFuncs {
aggExec.AggFunc = append(aggExec.AggFunc, aggregation.AggFuncToPBExpr(sc, client, aggFunc))
}
return &tipb.Executor{Tp: tipb.ExecType_TypeAggregation, Aggregation: aggExec}, nil
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalStreamAgg) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
sc := ctx.GetSessionVars().StmtCtx
client := ctx.GetClient()
aggExec := &tipb.Aggregation{
GroupBy: expression.ExpressionsToPBList(sc, p.GroupByItems, client),
}
for _, aggFunc := range p.AggFuncs {
aggExec.AggFunc = append(aggExec.AggFunc, aggregation.AggFuncToPBExpr(sc, client, aggFunc))
}
return &tipb.Executor{Tp: tipb.ExecType_TypeStreamAgg, Aggregation: aggExec}, nil
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalSelection) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
sc := ctx.GetSessionVars().StmtCtx
client := ctx.GetClient()
selExec := &tipb.Selection{
Conditions: expression.ExpressionsToPBList(sc, p.Conditions, client),
}
return &tipb.Executor{Tp: tipb.ExecType_TypeSelection, Selection: selExec}, nil
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalTopN) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
sc := ctx.GetSessionVars().StmtCtx
client := ctx.GetClient()
topNExec := &tipb.TopN{
Limit: p.Count,
}
for _, item := range p.ByItems {
topNExec.OrderBy = append(topNExec.OrderBy, expression.SortByItemToPB(sc, client, item.Expr, item.Desc))
}
return &tipb.Executor{Tp: tipb.ExecType_TypeTopN, TopN: topNExec}, nil
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalLimit) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
limitExec := &tipb.Limit{
Limit: p.Count,
}
return &tipb.Executor{Tp: tipb.ExecType_TypeLimit, Limit: limitExec}, nil
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalTableScan) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
columns := p.Columns
tsExec := &tipb.TableScan{
TableId: p.Table.ID,
Columns: model.ColumnsToProto(columns, p.Table.PKIsHandle),
Desc: p.Desc,
}
err := SetPBColumnsDefaultValue(ctx, tsExec.Columns, p.Columns)
return &tipb.Executor{Tp: tipb.ExecType_TypeTableScan, TblScan: tsExec}, errors.Trace(err)
}
// checkCoverIndex checks whether we can pass unique info to TiKV. We should push it if and only if the length of
// range and index are equal.
func checkCoverIndex(idx *model.IndexInfo, ranges []*ranger.Range) bool {
// If the index is (c1, c2) but the query range only contains c1, it is not a unique get.
if !idx.Unique {
return false
}
for _, rg := range ranges {
if len(rg.LowVal) != len(idx.Columns) {
return false
}
}
return true
}
// ToPB implements PhysicalPlan ToPB interface.
func (p *PhysicalIndexScan) ToPB(ctx sessionctx.Context) (*tipb.Executor, error) {
columns := make([]*model.ColumnInfo, 0, p.schema.Len())
tableColumns := p.Table.Cols()
for _, col := range p.schema.Columns {
if col.ID == model.ExtraHandleID {
columns = append(columns, model.NewExtraHandleColInfo())
} else {
columns = append(columns, model.FindColumnInfo(tableColumns, col.ColName.L))
}
}
idxExec := &tipb.IndexScan{
TableId: p.Table.ID,
IndexId: p.Index.ID,
Columns: model.ColumnsToProto(columns, p.Table.PKIsHandle),
Desc: p.Desc,
}
unique := checkCoverIndex(p.Index, p.Ranges)
idxExec.Unique = &unique
return &tipb.Executor{Tp: tipb.ExecType_TypeIndexScan, IdxScan: idxExec}, nil
}
// SetPBColumnsDefaultValue sets the default values of tipb.ColumnInfos.
func SetPBColumnsDefaultValue(ctx sessionctx.Context, pbColumns []*tipb.ColumnInfo, columns []*model.ColumnInfo) error {
for i, c := range columns {
if c.OriginDefaultValue == nil {
continue
}
sessVars := ctx.GetSessionVars()
originStrict := sessVars.StrictSQLMode
sessVars.StrictSQLMode = false
d, err := table.GetColOriginDefaultValue(ctx, c)
sessVars.StrictSQLMode = originStrict
if err != nil {
return errors.Trace(err)
}
pbColumns[i].DefaultVal, err = tablecodec.EncodeValue(ctx.GetSessionVars().StmtCtx, d)
if err != nil {
return errors.Trace(err)
}
}
return nil
}
// SupportStreaming returns true if a pushed down operation supports using coprocessor streaming API.
// Note that this function handle pushed down physical plan only! It's called in constructDAGReq.
// Some plans are difficult (if possible) to implement streaming, and some are pointless to do so.
// TODO: Support more kinds of physical plan.
func SupportStreaming(p PhysicalPlan) bool {
switch p.(type) {
case *PhysicalTableScan, *PhysicalIndexScan, *PhysicalSelection:
return true
}
return false
}

Опубликовать ( 0 )

Вы можете оставить комментарий после Вход в систему

1
https://gitlife.ru/oschina-mirror/hanchuanchuan-goInception.git
git@gitlife.ru:oschina-mirror/hanchuanchuan-goInception.git
oschina-mirror
hanchuanchuan-goInception
hanchuanchuan-goInception
v1.2.3