radon/src/executor/merge_engine.go at v1.0.7 · radondb/radon · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
/*
 * Radon
 *
 * Copyright 2018 The Radon Authors.
 * Code is licensed under the GPLv3.
 *
 */

package executor

import (
	"backend"
	"planner"
	"xcontext"

	"github.com/xelabs/go-mysqlstack/sqlparser"
	querypb "github.com/xelabs/go-mysqlstack/sqlparser/depends/query"
	"github.com/xelabs/go-mysqlstack/xlog"
)

var (
	_ PlanEngine = &MergeEngine{}
)

// MergeEngine represents merge executor.
type MergeEngine struct {
	log  *xlog.Log
	node *planner.MergeNode
	txn  backend.Transaction
}

// NewMergeEngine creates the new merge executor.
func NewMergeEngine(log *xlog.Log, node *planner.MergeNode, txn backend.Transaction) *MergeEngine {
	return &MergeEngine{
		log:  log,
		node: node,
		txn:  txn,
	}
}

// execute used to execute the executor.
func (m *MergeEngine) execute(ctx *xcontext.ResultContext) error {
	var err error

	reqCtx := xcontext.NewRequestContext()
	reqCtx.Mode = m.node.ReqMode
	reqCtx.TxnMode = xcontext.TxnRead
	if reqCtx.Mode == xcontext.ReqNormal {
		reqCtx.Querys = m.node.Querys
	} else {
		buf := sqlparser.NewTrackedBuffer(nil)
		m.node.Sel.Format(buf)
		reqCtx.RawQuery = buf.String()
	}

	if ctx.Results, err = m.txn.Execute(reqCtx); err != nil {
		return err
	}
	return execSubPlan(m.log, m.node, ctx)
}

// execBindVars used to execute querys with bindvas.
func (m *MergeEngine) execBindVars(ctx *xcontext.ResultContext, bindVars map[string]*querypb.BindVariable, wantfields bool) error {
	var query string
	var err error

	querys := m.node.Querys
	for i, p := range m.node.ParsedQuerys {
		query, err = p.GenerateQuery(bindVars, nil)
		if err != nil {
			return err
		}
		querys[i].Query = query
	}

	reqCtx := xcontext.NewRequestContext()
	reqCtx.Mode = xcontext.ReqNormal
	reqCtx.TxnMode = xcontext.TxnRead
	reqCtx.Querys = querys

	if ctx.Results, err = m.txn.Execute(reqCtx); err != nil {
		return err
	}
	return execSubPlan(m.log, m.node, ctx)
}

// getFields fetches the field info.
func (m *MergeEngine) getFields(ctx *xcontext.ResultContext, bindVars map[string]*querypb.BindVariable) error {
	var err error

	query := m.node.Querys[len(m.node.Querys)-1]
	buf := sqlparser.NewTrackedBuffer(nil)
	sqlparser.FormatImpossibleQuery(buf, m.node.Sel)
	query.Query = buf.String()

	reqCtx := xcontext.NewRequestContext()
	reqCtx.Mode = xcontext.ReqNormal
	reqCtx.TxnMode = xcontext.TxnRead
	reqCtx.Querys = []xcontext.QueryTuple{query}

	if ctx.Results, err = m.txn.Execute(reqCtx); err != nil {
		return err
	}
	return nil
}