pipeline

Documentation

Index

• type Pipeline
  • func New(tableID uint64, attrs []string, ins vm.Instructions, reg *process.WaitRegister) *Pipeline
  • func NewMerge(ins vm.Instructions, reg *process.WaitRegister) *Pipeline
  • func (p *Pipeline) Cleanup(proc *process.Process, pipelineFailed bool, err error)
  • func (p *Pipeline) ConstRun(bat *batch.Batch, proc *process.Process) (end bool, err error)
  • func (p *Pipeline) MergeRun(proc *process.Process) (end bool, err error)
  • func (p *Pipeline) Run(r engine.Reader, topValueMsgTag int32, proc *process.Process) (end bool, err error)
  • func (p *Pipeline) String() string

Types

type Pipeline

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline contains the information associated with a pipeline in a query execution plan. A query execution plan may contains one or more pipelines. As an example:

 CREATE TABLE order
 (
       order_id    INT,
       uid          INT,
       item_id      INT,
       year         INT,
       nation       VARCHAR(100)
 );

 CREATE TABLE customer
 (
       uid          INT,
       nation       VARCHAR(100),
       city         VARCHAR(100)
 );

 CREATE TABLE supplier
 (
       item_id      INT,
       nation       VARCHAR(100),
       city         VARCHAR(100)
 );

	SELECT c.city, s.city, sum(o.revenue) AS revenue
 FROM customer c, order o, supplier s
 WHERE o.uid = c.uid
 AND o.item_id = s.item_id
 AND c.nation = 'CHINA'
 AND s.nation = 'CHINA'
 AND o.year >= 1992 and o.year <= 1997
 GROUP BY c.city, s.city, o.year
 ORDER BY o.year asc, revenue desc;

 AST PLAN:
    order
      |
    group
      |
    filter
      |
    join
    /  \
   s   join
       /  \
      l   c

In this example, a possible pipeline is as follows:

pipeline: o ⨝ c ⨝ s

-> σ(c.nation = 'CHINA' ∧  o.year >= 1992 ∧  o.year <= 1997 ∧  s.nation = 'CHINA')
-> γ([c.city, s.city, o.year, sum(o.revenue) as revenue], c.city, s.city, o.year)
-> τ(o.year asc, revenue desc)
-> π(c.city, s.city, revenue)

func New

func New(tableID uint64, attrs []string, ins vm.Instructions, reg *process.WaitRegister) *Pipeline

func NewMerge

func NewMerge(ins vm.Instructions, reg *process.WaitRegister) *Pipeline

func (*Pipeline) Cleanup

func (p *Pipeline) Cleanup(proc *process.Process, pipelineFailed bool, err error)

Cleanup do memory release work for whole pipeline. we deliver the error because some operator may need to know what the error it is.

func (*Pipeline) ConstRun

func (p *Pipeline) ConstRun(bat *batch.Batch, proc *process.Process) (end bool, err error)

func (*Pipeline) MergeRun

func (p *Pipeline) MergeRun(proc *process.Process) (end bool, err error)

func (*Pipeline) Run

func (p *Pipeline) Run(r engine.Reader, topValueMsgTag int32, proc *process.Process) (end bool, err error)

func (*Pipeline) String

func (p *Pipeline) String() string