featurebase/sql3/planner/opfanout.go
pokeeffe-molecula a1fc6d04a1 introduce performance counters and system table fanout, plus refactor metrics (#2363)
* performance counters

* first cut of perf counters and system table fanout and a wire protocol
* significantly refactored prometheus support; removed statsd and exprvar

* removed node_id

* put dax subquery test back

* Change Translator.TranslateFieldIDs method to take a dax.TableKeyer

There are a bunch of other calls to the Translator interface methods
with currently take an `index string`, and those need to be converted to
dax.TableKeyer as well. But I need to review each call, because in at
least one place I noticed one being called with `result.Index` instead
of with the qtbl available. And I don't yet know how those could be
different.

Co-authored-by: Travis Turner <travis@molecula.com>
(cherry picked from commit 7f6ea0e6e5)
2023-01-19 21:35:02 +00:00

114 lines
2.9 KiB
Go

// Copyright 2022 Molecula Corp. All rights reserved.
package planner
import (
"context"
"fmt"
"github.com/featurebasedb/featurebase/v3/sql3"
"github.com/featurebasedb/featurebase/v3/sql3/planner/types"
)
// PlanOpFanout is a query fanout operator that will execute an operator across all cluster nodes
type PlanOpFanout struct {
planner *ExecutionPlanner
ChildOp types.PlanOperator
warnings []string
}
func NewPlanOpFanout(planner *ExecutionPlanner, child types.PlanOperator) *PlanOpFanout {
return &PlanOpFanout{
planner: planner,
ChildOp: child,
warnings: make([]string, 0),
}
}
func (p *PlanOpFanout) Schema() types.Schema {
return p.ChildOp.Schema()
}
func (p *PlanOpFanout) Iterator(ctx context.Context, row types.Row) (types.RowIterator, error) {
return newFanOutIterator(p.planner, p.ChildOp), nil
}
func (p *PlanOpFanout) WithChildren(children ...types.PlanOperator) (types.PlanOperator, error) {
if len(children) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of children '%d'", len(children))
}
return NewPlanOpFanout(p.planner, children[0]), nil
}
func (p *PlanOpFanout) Children() []types.PlanOperator {
return []types.PlanOperator{
p.ChildOp,
}
}
func (p *PlanOpFanout) Plan() map[string]interface{} {
result := make(map[string]interface{})
result["_op"] = fmt.Sprintf("%T", p)
ps := make([]string, 0)
for _, e := range p.Schema() {
ps = append(ps, fmt.Sprintf("'%s', '%s', '%s'", e.ColumnName, e.RelationName, e.Type.TypeDescription()))
}
result["_schema"] = ps
result["child"] = p.ChildOp.Plan()
return result
}
func (p *PlanOpFanout) String() string {
return ""
}
func (p *PlanOpFanout) AddWarning(warning string) {
p.warnings = append(p.warnings, warning)
}
func (p *PlanOpFanout) Warnings() []string {
return p.warnings
}
func (p *PlanOpFanout) Expressions() []types.PlanExpression {
return []types.PlanExpression{}
}
func (p *PlanOpFanout) WithUpdatedExpressions(exprs ...types.PlanExpression) (types.PlanOperator, error) {
if len(exprs) != 1 {
return nil, sql3.NewErrInternalf("unexpected number of exprs '%d'", len(exprs))
}
return NewPlanOpFilter(p.planner, exprs[0], p.ChildOp), nil
}
type fanOutIterator struct {
planner *ExecutionPlanner
childOp types.PlanOperator
rows types.Rows
}
func newFanOutIterator(planner *ExecutionPlanner, childOp types.PlanOperator) *fanOutIterator {
return &fanOutIterator{
planner: planner,
childOp: childOp,
}
}
func (i *fanOutIterator) Next(ctx context.Context) (types.Row, error) {
if i.rows == nil {
rows, err := i.planner.mapReducePlanOp(ctx, i.childOp, func(ctx context.Context, prev, v types.Rows) (types.Rows, error) {
return append(prev, v...), nil
})
if err != nil {
return nil, err
}
i.rows = rows
}
if len(i.rows) > 0 {
row := i.rows[0]
// Move to next result element.
i.rows = i.rows[1:]
return row, nil
}
return nil, types.ErrNoMoreRows
}