func organizeClausesByBinding(clauses []*semantic.GraphClause) map[string][]*semantic.GraphClause {
	clausesByBinding := map[string][]*semantic.GraphClause{}
	for _, cls := range clauses {
		for b := range cls.BindingsMap() {
			clausesByBinding[b] = append(clausesByBinding[b], cls)
		}
	}
	return clausesByBinding
}
organizeClausesByBinding takes the graph clauses received as input and organize them in a map
on which the keys are the bindings of these clauses.
package planner
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/google/badwolf/bql/lexer"
	"github.com/google/badwolf/bql/planner/filter"
	"github.com/google/badwolf/bql/planner/tracer"
	"github.com/google/badwolf/bql/semantic"
	"github.com/google/badwolf/bql/table"
	"github.com/google/badwolf/storage"
	"github.com/google/badwolf/triple"
	"github.com/google/badwolf/triple/literal"
	"github.com/google/badwolf/triple/predicate"
	"golang.org/x/sync/errgroup"
)
type Executor interface {
	Execute(ctx context.Context) (*table.Table, error)
	String(ctx context.Context) string
	Type() string
}
type createPlan struct {
	stm    *semantic.Statement
	store  storage.Store
	tracer io.Writer
}
func (p *createPlan) Type() string {
	return "CREATE"
}
func (p *createPlan) Execute(ctx context.Context) (*table.Table, error) {
	t, err := table.New([]string{})
	if err != nil {
		return nil, err
	}
	errs := []string{}
	for _, gName := range p.stm.GraphNames() {
		gNameCopy := gName  
		tracer.V(2).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{fmt.Sprintf("Creating new graph %q", gNameCopy)},
			}
		})
		if _, err := p.store.NewGraph(ctx, gName); err != nil {
			errs = append(errs, err.Error())
		}
	}
	if len(errs) > 0 {
		return nil, errors.New(strings.Join(errs, "; "))
	}
	return t, nil
}
func (p *createPlan) String(ctx context.Context) string {
	return fmt.Sprintf("CREATE plan:\n\nstore(%q).NewGraph(_, %v)", p.store.Name(nil), p.stm.Graphs())
}
type dropPlan struct {
	stm    *semantic.Statement
	store  storage.Store
	tracer io.Writer
}
func (p *dropPlan) Type() string {
	return "DROP"
}
func (p *dropPlan) Execute(ctx context.Context) (*table.Table, error) {
	t, err := table.New([]string{})
	if err != nil {
		return nil, err
	}
	errs := []string{}
	for _, gName := range p.stm.GraphNames() {
		gNameCopy := gName  
		tracer.V(2).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{fmt.Sprintf("Deleting graph %q", gNameCopy)},
			}
		})
		if err := p.store.DeleteGraph(ctx, gName); err != nil {
			errs = append(errs, err.Error())
		}
	}
	if len(errs) > 0 {
		return nil, errors.New(strings.Join(errs, "; "))
	}
	return t, nil
}
func (p *dropPlan) String(ctx context.Context) string {
	return fmt.Sprintf("DROP plan:\n\nstore(%q).DeleteGraph(_, %v)", p.store.Name(nil), p.stm.Graphs())
}
type insertPlan struct {
	stm    *semantic.Statement
	store  storage.Store
	tracer io.Writer
}
func (p *insertPlan) Type() string {
	return "INSERT"
}
type updater func(storage.Graph, []*triple.Triple) error
func update(ctx context.Context, ts []*triple.Triple, gbs []string, store storage.Store, f updater) error {
	var (
		mu   sync.Mutex
		wg   sync.WaitGroup
		errs []string
	)
	appendError := func(err error) {
		mu.Lock()
		defer mu.Unlock()
		errs = append(errs, err.Error())
	}
	for _, graphBinding := range gbs {
		wg.Add(1)
		go func(graph string) {
			defer wg.Done()
			g, err := store.Graph(ctx, graph)
			if err != nil {
				appendError(err)
				return
			}
			err = f(g, ts)
			if err != nil {
				appendError(err)
			}
		}(graphBinding)
	}
	wg.Wait()
	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "; "))
	}
	return nil
}
func (p *insertPlan) Execute(ctx context.Context) (*table.Table, error) {
	t, err := table.New([]string{})
	if err != nil {
		return nil, err
	}
	return t, update(ctx, p.stm.Data(), p.stm.OutputGraphNames(), p.store, func(g storage.Graph, d []*triple.Triple) error {
		gID := g.ID(ctx)
		nTrpls := len(d)
		tracer.V(2).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{fmt.Sprintf("Inserting %d triples to graph %q", nTrpls, gID)},
			}
		})
		return g.AddTriples(ctx, d)
	})
}
func (p *insertPlan) String(ctx context.Context) string {
	b := bytes.NewBufferString("INSERT plan:\n\n")
	for _, g := range p.stm.OutputGraphs() {
		b.WriteString(fmt.Sprintf("store(%q).Graph(%q).AddTriples(_, data)\n", p.store.Name(nil), g))
	}
	b.WriteString("where data:\n")
	for _, t := range p.stm.Data() {
		b.WriteString("\t")
		b.WriteString(t.String())
		b.WriteString("\n")
	}
	return b.String()
}
type deletePlan struct {
	stm    *semantic.Statement
	store  storage.Store
	tracer io.Writer
}
func (p *deletePlan) Type() string {
	return "DELETE"
}
func (p *deletePlan) Execute(ctx context.Context) (*table.Table, error) {
	t, err := table.New([]string{})
	if err != nil {
		return nil, err
	}
	return t, update(ctx, p.stm.Data(), p.stm.InputGraphNames(), p.store, func(g storage.Graph, d []*triple.Triple) error {
		gID := g.ID(ctx)
		nTrpls := len(d)
		tracer.V(2).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{fmt.Sprintf("Removing %d triples from graph %q", nTrpls, gID)},
			}
		})
		return g.RemoveTriples(ctx, d)
	})
}
func (p *deletePlan) String(ctx context.Context) string {
	b := bytes.NewBufferString("DELETE plan:\n\n")
	for _, g := range p.stm.InputGraphs() {
		b.WriteString(fmt.Sprintf("store(%q).Graph(%q).RemoveTriples(_, data)\n", p.store.Name(nil), g))
	}
	b.WriteString("where data:\n")
	for _, t := range p.stm.Data() {
		b.WriteString("\t")
		b.WriteString(t.String())
		b.WriteString("\n")
	}
	return b.String()
}
type queryPlan struct {
	stm   *semantic.Statement
	store storage.Store
	bndgs     []string
	grfsNames []string
	grfs      []storage.Graph
	clauses   []*semantic.GraphClause
	filters   []*semantic.FilterClause
	tbl       *table.Table
	chanSize  int
	tracer    io.Writer
}
func (p *queryPlan) Type() string {
	return "SELECT"
}
func newQueryPlan(ctx context.Context, store storage.Store, stm *semantic.Statement, chanSize int, w io.Writer) (*queryPlan, error) {
	bs := []string{}
	for _, b := range stm.Bindings() {
		bs = append(bs, b)
	}
	t, err := table.New([]string{})
	if err != nil {
		return nil, err
	}
	return &queryPlan{
		stm:       stm,
		store:     store,
		bndgs:     bs,
		grfsNames: stm.InputGraphNames(),
		clauses:   stm.GraphPatternClauses(),
		filters:   stm.FilterClauses(),
		tbl:       t,
		chanSize:  chanSize,
		tracer:    w,
	}, nil
}
func (p *queryPlan) processClause(ctx context.Context, cls *semantic.GraphClause, lo *storage.LookupOptions) (bool, error) {
	if cls.Specificity() == 3 {
		tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{"Clause is fully specified"},
			}
		})
		if cls.Optional && !cls.HasAlias() {
			tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
				return &tracer.Arguments{
					Msgs: []string{fmt.Sprintf("Processing optional clause of specificity 3: %v", cls)},
				}
			})
			return false, nil
		}
		t, err := triple.New(cls.S, cls.P, cls.O)
		if err != nil {
			return false, err
		}
		b, tbl, err := simpleExist(ctx, p.grfs, cls, t, p.tracer)
		if err != nil {
			return false, err
		}
		if err := p.tbl.AppendTable(tbl); err != nil {
			return b, err
		}
		return b, nil
	}
	exist, total := 0, 0
	var existing []string
	for _, b := range cls.Bindings() {
		total++
		if p.tbl.HasBinding(b) {
			exist++
			existing = append(existing, b)
		}
	}
	if exist == 0 {
		tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
			return &tracer.Arguments{
				Msgs: []string{fmt.Sprintf("None of the clause binding exist %v/%v", cls.Bindings(), existing)},
			}
		})
		stmLimit := int64(0)
		if len(p.stm.GraphPatternClauses()) == 1 && len(p.stm.GroupBy()) == 0 && len(p.stm.HavingExpression()) == 0 {
			stmLimit = p.stm.Limit()
		}
		tbl, err := simpleFetch(ctx, p.grfs, cls, lo, stmLimit, p.chanSize, p.tracer)
		if err != nil {
			return true, err
		}
		if len(p.tbl.Bindings()) > 0 {
			if cls.Optional {
				tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
					return &tracer.Arguments{
						Msgs: []string{fmt.Sprintf("Processing optional clause of disjoint bindings: %v", cls)},
					}
				})
				return false, p.tbl.LeftOptionalJoin(tbl)
			}
			return false, p.tbl.DotProduct(tbl)
		}
		return false, p.tbl.AppendTable(tbl)
	}
	tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
		return &tracer.Arguments{
			Msgs: []string{fmt.Sprintf("Some clause binding exist %v/%v", cls.Bindings(), existing)},
		}
	})
	return false, p.specifyClauseWithTable(ctx, cls, lo)
}
func getBoundValueForComponent(r table.Row, bs []string) *table.Cell {
	var cs []*table.Cell
	for _, b := range bs {
		if v, ok := r[b]; ok {
			cs = append(cs, v)
		}
	}
	if len(cs) == 1 || len(cs) == 2 && reflect.DeepEqual(cs[0], cs[1]) {
		return cs[0]
	}
	return nil
}
func (p *queryPlan) addSpecifiedData(ctx context.Context, r table.Row, cls *semantic.GraphClause, lo *storage.LookupOptions) error {
	if cls.S == nil {
		v := getBoundValueForComponent(r, []string{cls.SBinding, cls.SAlias})
		if v != nil {
			if v.N != nil {
				cls.S = v.N
			}
		}
	}
	if cls.P == nil && cls.PID != "" && cls.PAnchorBinding != "" {
		v := r[cls.PAnchorBinding]
		if v != nil && v.T != nil {
			p, err := predicate.NewTemporal(cls.PID, *v.T)
			if err != nil {
				return err
			}
			cls.P = p
		}
	}
	if cls.P == nil {
		v := getBoundValueForComponent(r, []string{cls.PBinding, cls.PAlias})
		if v != nil {
			if v.P != nil {
				cls.P = v.P
			}
		}
		nlo, err := updateTimeBoundsForRow(lo, cls, r)
		if err != nil {
			return err
		}
		lo = nlo
	}
	if cls.O == nil && cls.OID != "" && cls.OAnchorBinding != "" {
		v := r[cls.OAnchorBinding]
		if v != nil && v.T != nil {
			p, err := predicate.NewTemporal(cls.OID, *v.T)
			if err != nil {
				return err
			}
			cls.O = triple.NewPredicateObject(p)
		}
	}
	if cls.O == nil {
		v := getBoundValueForComponent(r, []string{cls.OBinding, cls.OAlias})
		if v != nil {
			o, err := cellToObject(v)
			if err == nil {
				cls.O = o
			}
		}
		nlo, err := updateTimeBoundsForRow(lo, cls, r)
		if err != nil {
			return err
		}
		lo = nlo
	}
	tracer.V(3).Trace(p.tracer, func() *tracer.Arguments {
		return &tracer.Arguments{
			Msgs: []string{fmt.Sprintf("Corrected clause: %v", cls)},
		}
	})
	stmLimit := int64(0)
	if len(p.stm.GraphPatternClauses()) == 1 && len(p.stm.GroupBy()) == 0 && len(p.stm.HavingExpression()) == 0 {
		stmLimit = p.stm.Limit()
	}
	tbl, err := simpleFetch(ctx, p.grfs, cls, lo, stmLimit, p.chanSize, p.tracer)
	if err != nil {
		return err
	}
	p.tbl.AddBindings(tbl.Bindings())
	if tbl.NumRows() == 0 && cls.Optional {
		nr := make(table.Row)
		for _, k := range tbl.Bindings() {
			if _, ok := r[k]; !ok {
				nr[k] = &table.Cell{}
			}
		}
		p.tbl.AddRow(table.MergeRows([]table.Row{r, nr}))
		return nil
	}
	for _, nr := range tbl.Rows() {
		p.tbl.AddRow(table.MergeRows([]table.Row{r, nr}))
	}
	return nil
}
func (p *queryPlan) specifyClauseWithTable(ctx context.Context, cls *semantic.GraphClause, lo *storage.LookupOptions) error {
	rws := p.tbl.Rows()
	p.tbl.Truncate()
	grp, gCtx := errgroup.WithContext(ctx)
	for _, tmpRow := range rws {
		r := tmpRow
		grp.Go(func() error {
			var tmpCls = *cls
			return p.addSpecifiedData(gCtx, r, &tmpCls, lo)
		})
	}
	return grp.Wait()
}
func cellToObject(c *table.Cell) (*triple.Object, error) {
	if c == nil {
		return nil, errors.New("cannot create an object out of and empty cell")
	}
	if c.N != nil {
		return triple.NewNodeObject(c.N), nil
	}
	if c.P != nil {
		return triple.NewPredicateObject(c.P), nil
	}
	if c.L != nil {
		return triple.NewLiteralObject(c.L), nil
	}
	if c.S != nil {
		l, err := literal.DefaultBuilder().Parse(fmt.Sprintf(`"%s"^^type:string`, *c.S))
		if err != nil {
			return nil, err
		}
		return triple.NewLiteralObject(l), nil
	}
	return nil, fmt.Errorf("invalid cell %v", c)
}
func (p *queryPlan) filterOnExistence(ctx context.Context, cls *semantic.GraphClause, lo *storage.LookupOptions) error {
	data := p.tbl.Rows()
	p.tbl.Truncate()
	ocls := *cls
	grp, gCtx := errgroup.WithContext(ctx)
	for _, tmp := range data {
		if gCtx.Err() != nil {
			break
		}
		r := tmp
		cls := ocls
		grp.Go(func() error {
			sbj, prd, obj := cls.S, cls.P, cls.O
			if sbj == nil && p.tbl.HasBinding(cls.SBinding) {
				v, ok := r[cls.SBinding]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.SBinding)
				}
				if v.N == nil {
					return fmt.Errorf("binding %q requires a node, got %+v instead", cls.SBinding, v)
				}
				sbj = v.N
			}
			if sbj == nil && p.tbl.HasBinding(cls.SAlias) {
				v, ok := r[cls.SAlias]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.SAlias)
				}
				if v.N == nil {
					return fmt.Errorf("binding %q requires a node, got %+v instead", cls.SAlias, v)
				}
				sbj = v.N
			}
			if prd == nil && p.tbl.HasBinding(cls.PBinding) {
				v, ok := r[cls.PBinding]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.PBinding)
				}
				if v.P == nil {
					return fmt.Errorf("binding %q requires a predicate, got %+v instead", cls.PBinding, v)
				}
				prd = v.P
			}
			if prd == nil && p.tbl.HasBinding(cls.PAlias) {
				v, ok := r[cls.PAlias]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.SAlias)
				}
				if v.N == nil {
					return fmt.Errorf("binding %q requires a predicate, got %+v instead", cls.SAlias, v)
				}
				prd = v.P
			}
			if obj == nil && p.tbl.HasBinding(cls.OBinding) {
				v, ok := r[cls.OBinding]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.OBinding)
				}
				co, err := cellToObject(v)
				if err != nil {
					return err
				}
				obj = co
			}
			if obj == nil && p.tbl.HasBinding(cls.OAlias) {
				v, ok := r[cls.OAlias]
				if !ok {
					return fmt.Errorf("row %+v misses binding %q", r, cls.OAlias)
				}
				if v.N == nil {
					return fmt.Errorf("binding %q requires a object, got %+v instead", cls.OAlias, v)
				}
				co, err := cellToObject(v)
				if err != nil {
					return err
				}
				obj = co
			}
			if sbj == nil || prd == nil || obj == nil {
				return fmt.Errorf("failed to fully specify clause %v for row %+v", cls, r)
			}
			exist := false
			for _, g := range p.stm.InputGraphs() {
				gID := g.ID(gCtx)
				t, err := triple.New(sbj, prd, obj)
				if err != nil {
					return err
				}
				tracer.V(2).Trace(p.tracer, func() *tracer.Arguments {
					return &tracer.Arguments{
						Msgs: []string{fmt.Sprintf("g.Exist(%v), graph: %s", t, gID)},
					}
				})
				b, err := g.Exist(gCtx, t)
				if err != nil {
					return err
				}
				exist = exist || b
				if exist || gCtx.Err() != nil {
					break
				}
			}
			if exist {
				p.tbl.AddRow(r)
			}
			return nil
		})
	}
	return grp.Wait()
}