func (tb *TopologyBuilder) RunEvalStmt(stmt *parser.EvalStmt) (data.Value, error) {
	if stmt.Input == nil {
		return execution.EvaluateFoldable(stmt.Expr, tb.Reg)
	}
	inputData, err := execution.EvaluateFoldable(*stmt.Input, tb.Reg)
	if err != nil {
		return nil, err
	}
	usedRelations := stmt.Expr.ReferencedRelations()
	if len(usedRelations) > 1 || (len(usedRelations) == 1 && !usedRelations[""]) {
		return nil, fmt.Errorf("stream prefixes cannot be used inside EVAL")
	}
	expr := stmt.Expr.RenameReferencedRelation("", "input")
	inputRow := data.Map{"input": inputData}
	return execution.EvaluateOnInput(expr, inputRow, tb.Reg)
}
RunEvalStmt evaluates the expression contained in the given EvalStmt
and returns the evaluation result.
package bql
import (
	"errors"
	"fmt"
	"math"
	"strings"
	"sync"
	"sync/atomic"
	"gopkg.in/sensorbee/sensorbee.v0/bql/execution"
	"gopkg.in/sensorbee/sensorbee.v0/bql/parser"
	"gopkg.in/sensorbee/sensorbee.v0/bql/udf"
	"gopkg.in/sensorbee/sensorbee.v0/core"
	"gopkg.in/sensorbee/sensorbee.v0/data"
)
type TopologyBuilder struct {
	topology       core.Topology
	Reg            udf.FunctionManager
	UDSFCreators   udf.UDSFCreatorRegistry
	UDSCreators    udf.UDSCreatorRegistry
	SourceCreators SourceCreatorRegistry
	SinkCreators   SinkCreatorRegistry
	UDSStorage     udf.UDSStorage
}
func NewTopologyBuilder(t core.Topology) (*TopologyBuilder, error) {
	udsfs, err := udf.CopyGlobalUDSFCreatorRegistry()
	if err != nil {
		return nil, err
	}
	udss, err := udf.CopyGlobalUDSCreatorRegistry()
	if err != nil {
		return nil, err
	}
	srcs, err := CopyGlobalSourceCreatorRegistry()
	if err != nil {
		return nil, err
	}
	if err := srcs.Register("node_statuses", createNodeStatusSourceCreator(t)); err != nil {
		return nil, err
	}
	if err := srcs.Register("edge_statuses", createEdgeStatusSourceCreator(t)); err != nil {
		return nil, err
	}
	sinks, err := CopyGlobalSinkCreatorRegistry()
	if err != nil {
		return nil, err
	}
	tb := &TopologyBuilder{
		topology:       t,
		Reg:            udf.CopyGlobalUDFRegistry(t.Context()),
		UDSFCreators:   udsfs,
		UDSCreators:    udss,
		SourceCreators: srcs,
		SinkCreators:   sinks,
		UDSStorage:     udf.NewInMemoryUDSStorage(),
	}
	return tb, nil
}
var (
	topologyBuilderTemporaryID int64
)
func topologyBuilderNextTemporaryID() int64 {
	return atomic.AddInt64(&topologyBuilderTemporaryID, 1)
}
func (tb *TopologyBuilder) Topology() core.Topology {
	return tb.topology
}
func (tb *TopologyBuilder) AddStmt(stmt interface{}) (core.Node, error) {
	switch stmt := stmt.(type) {
	case parser.CreateSourceStmt:
		paramsMap := tb.mkParamsMap(stmt.Params)
		creator, err := tb.SourceCreators.Lookup(string(stmt.Type))
		if err != nil {
			return nil, err
		}
		source, err := creator.CreateSource(tb.topology.Context(), &IOParams{
			TypeName: string(stmt.Type),
			Name:     string(stmt.Name),
		}, paramsMap)
		if err != nil {
			return nil, err
		}
		return tb.topology.AddSource(string(stmt.Name), source, &core.SourceConfig{
			PausedOnStartup: stmt.Paused == parser.Yes,
		})
	case parser.CreateStreamAsSelectStmt:
		return tb.createStreamAsSelectStmt(&stmt)
	case parser.CreateStreamAsSelectUnionStmt:
		names := make([]string, 0, len(stmt.Selects))
		nodes := make([]core.BoxNode, 0, len(stmt.Selects))
		removeTmpNodes := func() {
			for _, name := range names {
				tb.topology.Remove(name)
			}
		}
		for _, selStmt := range stmt.Selects {
			tmpName := fmt.Sprintf("sensorbee_tmp_%v", topologyBuilderNextTemporaryID())
			tmpStmt := parser.CreateStreamAsSelectStmt{
				parser.StreamIdentifier(tmpName),
				selStmt,
			}
			box, err := tb.AddStmt(tmpStmt)
			if err != nil {
				removeTmpNodes()
				return nil, err
			}
			names = append(names, tmpName)
			nodes = append(nodes, box.(core.BoxNode))
		}
		forwardBox := core.BoxFunc(func(ctx *core.Context, t *core.Tuple, w core.Writer) error {
			return w.Write(ctx, t)
		})
		node, err := tb.topology.AddBox(string(stmt.Name), forwardBox, nil)
		if err != nil {
			removeTmpNodes()
			return nil, err
		}
		for _, name := range names {
			if err := node.Input(name, nil); err != nil {
				removeTmpNodes()
				return nil, err
			}
		}
		for _, node := range nodes {
			node.StopOnDisconnect(core.Inbound | core.Outbound)
			node.RemoveOnStop()
		}
		node.StopOnDisconnect(core.Inbound)
		node.RemoveOnStop()
		return node, nil
	case parser.CreateSinkStmt:
		paramsMap := tb.mkParamsMap(stmt.Params)
		creator, err := tb.SinkCreators.Lookup(string(stmt.Type))
		if err != nil {
			return nil, err
		}
		sink, err := creator.CreateSink(tb.topology.Context(), &IOParams{
			TypeName: string(stmt.Type),
			Name:     string(stmt.Name),
		}, paramsMap)
		if err != nil {
			return nil, err
		}
		return tb.topology.AddSink(string(stmt.Name), sink, nil)
	case parser.CreateStateStmt:
		c, err := tb.UDSCreators.Lookup(string(stmt.Type))
		if err != nil {
			return nil, err
		}
		ctx := tb.topology.Context()
		s, err := c.CreateState(ctx, tb.mkParamsMap(stmt.Params))
		if err != nil {
			return nil, err
		}
		if err := ctx.SharedStates.Add(string(stmt.Name), string(stmt.Type), s); err != nil {
			return nil, err
		}
		return nil, nil
	case parser.UpdateStateStmt:
		ctx := tb.topology.Context()
		state, err := ctx.SharedStates.Get(string(stmt.Name))
		if err != nil {
			return nil, err
		}
		u, ok := state.(core.Updater)
		if !ok {
			return nil, fmt.Errorf("%s cannot be updated", string(stmt.Name))
		}
		return nil, u.Update(ctx, tb.mkParamsMap(stmt.Params))
	case parser.SaveStateStmt:
		return nil, tb.saveState(string(stmt.Name), stmt.Tag)
	case parser.LoadStateStmt:
		_, err := tb.loadState(string(stmt.Type), string(stmt.Name), stmt.Tag, tb.mkParamsMap(stmt.Params))
		return nil, err
	case parser.LoadStateOrCreateStmt:
		shouldCreate, err := tb.loadState(string(stmt.Type), string(stmt.Name), stmt.Tag, tb.mkParamsMap(stmt.LoadSpecs.Params))
		if shouldCreate {
			c := parser.CreateStateStmt{}
			c.Type = stmt.Type
			c.Name = stmt.Name
			c.Params = stmt.CreateSpecs.Params
			return tb.AddStmt(c)
		}
		return nil, err
	case parser.UpdateSourceStmt:
		src, err := tb.topology.Source(string(stmt.Name))
		if err != nil {
			return nil, err
		}
		u, ok := src.Source().(core.Updater)
		if !ok {
			return nil, fmt.Errorf("%s cannot be updated", string(stmt.Name))
		}
		return nil, u.Update(tb.topology.Context(), tb.mkParamsMap(stmt.Params))
	case parser.UpdateSinkStmt:
		sink, err := tb.topology.Sink(string(stmt.Name))
		if err != nil {
			return nil, err
		}
		u, ok := sink.Sink().(core.Updater)
		if !ok {
			return nil, fmt.Errorf("%s cannot be updated", string(stmt.Name))
		}
		return nil, u.Update(tb.topology.Context(), tb.mkParamsMap(stmt.Params))
	case parser.DropSourceStmt:
		_, err := tb.topology.Source(string(stmt.Source))
		if err != nil {
			return nil, err
		}
		return nil, tb.topology.Remove(string(stmt.Source))
	case parser.DropStreamStmt:
		_, err := tb.topology.Box(string(stmt.Stream))
		if err != nil {
			return nil, err
		}
		return nil, tb.topology.Remove(string(stmt.Stream))
	case parser.DropSinkStmt:
		_, err := tb.topology.Sink(string(stmt.Sink))
		if err != nil {
			return nil, err
		}
		return nil, tb.topology.Remove(string(stmt.Sink))
	case parser.DropStateStmt:
		ctx := tb.topology.Context()
		_, err := ctx.SharedStates.Get(string(stmt.State))
		if err != nil {
			return nil, err
		}
		_, err = ctx.SharedStates.Remove(string(stmt.State))
		return nil, err
	case parser.InsertIntoFromStmt:
		sink, err := tb.topology.Sink(string(stmt.Sink))
		if err != nil {
			return nil, err
		}
		if err := sink.Input(string(stmt.Input), nil); err != nil {
			return nil, err
		}
		return sink, nil
	case parser.PauseSourceStmt:
		src, err := tb.topology.Source(string(stmt.Source))
		if err != nil {
			return nil, err
		}
		if err := src.Pause(); err != nil {
			return nil, err
		}
		return src, nil
	case parser.ResumeSourceStmt:
		src, err := tb.topology.Source(string(stmt.Source))
		if err != nil {
			return nil, err
		}
		if err := src.Resume(); err != nil {
			return nil, err
		}
		return src, nil
	case parser.RewindSourceStmt:
		src, err := tb.topology.Source(string(stmt.Source))
		if err != nil {
			return nil, err
		}
		if err := src.Rewind(); err != nil {
			return nil, err
		}
		return src, nil
	}
	return nil, fmt.Errorf("statement of type %T is unimplemented", stmt)
}
type udsfBox struct {
	f udf.UDSF
}
var (
	_ core.StatefulBox = &udsfBox{}
)
func newUDSFBox(f udf.UDSF) *udsfBox {
	return &udsfBox{
		f: f,
	}
}
func (b *udsfBox) Init(ctx *core.Context) error {
	return nil
}
func (b *udsfBox) Process(ctx *core.Context, t *core.Tuple, w core.Writer) error {
	return b.f.Process(ctx, t, w)
}
func (b *udsfBox) Terminate(ctx *core.Context) error {
	return b.f.Terminate(ctx)
}
type udsfSource struct {
	f       udf.UDSF
	stopped core.AtomicFlag
}
var (
	_ core.Source = &udsfSource{}
)
func newUDSFSource(f udf.UDSF) *udsfSource {
	return &udsfSource{
		f: f,
	}
}
func (s *udsfSource) GenerateStream(ctx *core.Context, w core.Writer) error {
	return s.f.Process(ctx, core.NewTuple(data.Map{"b": data.True}),
		core.WriterFunc(func(ctx *core.Context, t *core.Tuple) error {
			if s.stopped.Enabled() {
				return core.ErrSourceStopped
			}
			return w.Write(ctx, t)
		}))
}
func (s *udsfSource) Stop(ctx *core.Context) error {
	s.stopped.Set(true)
	return s.f.Terminate(ctx)
}
func (tb *TopologyBuilder) createStreamAsSelectStmt(stmt *parser.CreateStreamAsSelectStmt) (core.Node, error) {
	outName := string(stmt.Name)
	box := NewBQLBox(&stmt.Select, tb.Reg)
	dbox, err := tb.topology.AddBox(outName, box, nil)
	if err != nil {
		return nil, err
	}
	for _, rel := range stmt.Select.Relations {
		if strings.ToLower(rel.Name) == strings.ToLower(outName) {
			return nil, fmt.Errorf("a stream '%v' contains a selfloop", outName)
		}
	}
	box.removeMe = func() { go tb.topology.Remove(outName) }
	removeNodes := true
	var temporaryNodes []string
	defer func() {
		if !removeNodes {
			return
		}
		for _, n := range temporaryNodes {
			tb.topology.Remove(n)
		}
		tb.topology.Remove(outName)
	}()
	connected := map[string]bool{}
	var pausedSources []core.SourceNode
	for _, rel := range stmt.Select.Relations {
		switch rel.Type {
		case parser.ActualStream:
			if connected[rel.Name] {
				continue
			}
			conf := &core.BoxInputConfig{
				InputName: rel.Name,
			}
			if rel.Capacity != parser.UnspecifiedCapacity {
				conf.Capacity = int(rel.Capacity)
			}
			if rel.Shedding == parser.DropOldest {
				conf.DropMode = core.DropOldest
			} else if rel.Shedding == parser.DropNewest {
				conf.DropMode = core.DropLatest
			} else if rel.Shedding == parser.Wait {
				conf.DropMode = core.DropNone
			}
			if err := dbox.Input(rel.Name, conf); err != nil {
				return nil, err
			}
			connected[rel.Name] = true
		case parser.UDSFStream:
			sn, name, err := tb.setUpUDSFStream(dbox, &rel)
			if err != nil {
				return nil, err
			}
			temporaryNodes = append(temporaryNodes, name)
			if sn != nil {
				pausedSources = append(pausedSources, sn)
			}
		default:
			return nil, fmt.Errorf("input stream of type %s not implemented",
				rel.Type)
		}
	}
	dbox.(core.BoxNode).StopOnDisconnect(core.Inbound)
	for _, sn := range pausedSources {
		if err := sn.Resume(); err != nil {
			return nil, err
		}
	}
	removeNodes = false
	return dbox, nil
}
func (tb *TopologyBuilder) setUpUDSFStream(subsequentBox core.BoxNode, rel *parser.AliasedStreamWindowAST) (core.SourceNode, string, error) {
	params := make([]data.Value, len(rel.Params))
	for i, expr := range rel.Params {
		p, err := execution.EvaluateFoldable(expr, tb.Reg)
		if err != nil {
			return nil, "", err
		}
		params[i] = p
	}
	udsfc, err := tb.UDSFCreators.Lookup(rel.Name, len(params))
	if err != nil {
		return nil, "", err
	}
	decl := udf.NewUDSFDeclarer()
	udsf, err := func() (f udf.UDSF, err error) {
		defer func() {
			if e := recover(); e != nil {
				if er, ok := e.(error); ok {
					err = er
				} else {
					err = fmt.Errorf("cannot create a UDSF: %v", e)
				}
			}
		}()
		return udsfc.CreateUDSF(tb.topology.Context(), decl, params...)
	}()
	if err != nil {
		return nil, "", err
	}
	temporaryName := fmt.Sprintf("sensorbee_tmp_udsf_%v", topologyBuilderNextTemporaryID())
	addInput := func() error {
		alias := rel.Alias
		if alias == "" {
			alias = rel.Name
		}
		conf := &core.BoxInputConfig{
			InputName: fmt.Sprintf("%s/%s", rel.Name, alias),
		}
		if rel.Capacity != parser.UnspecifiedCapacity {
			if rel.Capacity > math.MaxInt32 {
				return fmt.Errorf("specified buffer capacity %d is too large", rel.Capacity)
			} else if rel.Capacity < 0 {
				return fmt.Errorf("specified buffer capacity %d must not be negative", rel.Capacity)
			}
			conf.Capacity = int(rel.Capacity)
		}
		if rel.Shedding == parser.DropOldest {
			conf.DropMode = core.DropOldest
		} else if rel.Shedding == parser.DropNewest {
			conf.DropMode = core.DropLatest
		} else if rel.Shedding == parser.Wait {
			conf.DropMode = core.DropNone
		}
		return subsequentBox.Input(temporaryName, conf)
	}
	if len(decl.ListInputs()) == 0 {  
		sn, err := tb.topology.AddSource(temporaryName, newUDSFSource(udsf), &core.SourceConfig{
			PausedOnStartup: true,
		})
		if err != nil {
			return nil, "", err
		}
		if err := addInput(); err != nil {
			return nil, "", err
		}
		sn.StopOnDisconnect()
		sn.RemoveOnStop()
		return sn, temporaryName, nil
	}
	bn, err := tb.topology.AddBox(temporaryName, newUDSFBox(udsf), &core.BoxConfig{
	})
	if err != nil {
		return nil, "", err
	}
	for input, config := range decl.ListInputs() {
		if config.Capacity > math.MaxInt32 {
			return nil, "", fmt.Errorf(
				"specified buffer capacity %d is too large", config.Capacity)
		} else if config.Capacity < 0 {
			return nil, "", fmt.Errorf(
				"specified buffer capacity %d must not be negative", config.Capacity)
		}
		bc := &core.BoxInputConfig{
			InputName: config.InputName,
			Capacity:  config.Capacity,
			DropMode:  config.DropMode,
		}
		if err := bn.Input(input, bc); err != nil {
			return nil, "", err
		}
	}
	if err := addInput(); err != nil {
		return nil, "", err
	}
	bn.StopOnDisconnect(core.Inbound | core.Outbound)
	bn.RemoveOnStop()
	return nil, temporaryName, nil
}
func (tb *TopologyBuilder) mkParamsMap(params []parser.SourceSinkParamAST) data.Map {
	paramsMap := make(data.Map, len(params))
	for _, kv := range params {
		paramsMap[string(kv.Key)] = kv.Value
	}
	return paramsMap
}
type chanSink struct {
	m      sync.RWMutex
	ch     chan *core.Tuple
	closed bool
}
func newChanSink() (*chanSink, <-chan *core.Tuple) {
	c := make(chan *core.Tuple)
	return &chanSink{
		ch: c,
	}, c
}
func (s *chanSink) Write(ctx *core.Context, t *core.Tuple) error {
	s.m.RLock()
	defer s.m.RUnlock()
	if s.closed {
		return errors.New("the sink has already been closed")
	}
	s.ch <- t
	return nil
}
func (s *chanSink) Close(ctx *core.Context) error {
	go func() {
		for _ = range s.ch {
		}
	}()
	s.m.Lock()
	defer s.m.Unlock()
	if s.closed {
		return nil
	}
	s.closed = true
	close(s.ch)
	return nil
}
func (tb *TopologyBuilder) AddSelectStmt(stmt *parser.SelectStmt) (core.SinkNode, <-chan *core.Tuple, error) {
	tmpStmt := parser.SelectUnionStmt{[]parser.SelectStmt{*stmt}}
	return tb.AddSelectUnionStmt(&tmpStmt)
}
func (tb *TopologyBuilder) AddSelectUnionStmt(stmts *parser.SelectUnionStmt) (core.SinkNode, <-chan *core.Tuple, error) {
	sink, ch := newChanSink()
	tmpUnionNodeName := fmt.Sprintf("sensorbee_tmp_select_sink_%v", topologyBuilderNextTemporaryID())
	sn, err := tb.topology.AddSink(tmpUnionNodeName, sink, nil)
	if err != nil {
		sink.Close(tb.topology.Context())
		return nil, nil, err
	}
	names := make([]string, 0, len(stmts.Selects))
	for _, stmt := range stmts.Selects {
		createTmpBox := func(outputName string, stmt parser.SelectStmt) (core.Node, error) {
			sink, err := tb.topology.Sink(outputName)
			if err != nil {
				return nil, err
			}
			tmpName := fmt.Sprintf("sensorbee_tmp_%v", topologyBuilderNextTemporaryID())
			tmpStmt := parser.CreateStreamAsSelectStmt{
				parser.StreamIdentifier(tmpName),
				parser.SelectStmt{
					stmt.EmitterAST,
					stmt.ProjectionsAST,
					stmt.WindowedFromAST,
					stmt.FilterAST,
					stmt.GroupingAST,
					stmt.HavingAST,
				},
			}
			box, err := tb.AddStmt(tmpStmt)
			if err != nil {
				return nil, err
			}
			box.(core.BoxNode).RemoveOnStop()
			if err := sink.Input(tmpName, &core.SinkInputConfig{
				Capacity: 1,
			}); err != nil {
				tb.topology.Remove(tmpName)
				return nil, err
			}
			box.(core.BoxNode).StopOnDisconnect(core.Inbound | core.Outbound)
			return box, nil
		}
		node, err := createTmpBox(tmpUnionNodeName, stmt)
		if err != nil {
			for _, name := range names {
				tb.topology.Remove(name)
			}
			if err := sn.Stop(); err != nil {
				tb.topology.Context().ErrLog(err).WithField("node_type", core.NTSink).
					WithField("node_name", tmpUnionNodeName).Error("Cannot stop the temporary sink")
			}
			tb.topology.Remove(tmpUnionNodeName)
			return nil, nil, err
		}
		names = append(names, node.Name())
	}
	go func() {
		sn.RemoveOnStop()
		sn.StopOnDisconnect()
	}()
	return sn, ch, nil
}