func (s *Statement) NodeName() string {
	switch stmt := s.Stmt.(type) {
	case parser.CreateSourceStmt:
		return string(stmt.Name)
	case parser.CreateStreamAsSelectStmt:
		return string(stmt.Name)
	case parser.CreateStreamAsSelectUnionStmt:
		return string(stmt.Name)
	case parser.InsertIntoFromStmt:
		return string(stmt.Sink)
	}
	return ""
}
NodeName returns the name of node that the statement creates or manipulates.
If the statement doesn't create a node or the statement creates a sink, it
returns an empty string. Instead, INSERT INTO statements returns a name
of a sink.
package exp
import (
	"fmt"
	"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"
	"sort"
)
type Statement struct {
	Stmt interface{}
}
func NewStatement(s interface{}) (*Statement, error) {
	switch s.(type) {
	case parser.SelectStmt, parser.SelectUnionStmt,
		parser.UpdateSourceStmt, parser.UpdateSinkStmt, parser.UpdateStateStmt,
		parser.DropSourceStmt, parser.DropStreamStmt, parser.DropSinkStmt, parser.DropStateStmt,
		parser.SaveStateStmt, parser.LoadStateStmt, parser.LoadStateOrCreateStmt:
		return nil, fmt.Errorf("statement isn't supported by sensorbee exp command: %v", s)
	}
	return &Statement{
		Stmt: s,
	}, nil
}
func (s *Statement) Input() ([]string, error) {
	var (
		names []string
		err   error
	)
	switch stmt := s.Stmt.(type) {
	case parser.CreateStreamAsSelectStmt:
		names, err = inputFromSelect(&stmt.Select)
	case parser.CreateStreamAsSelectUnionStmt:
		for _, s := range stmt.Selects {
			ns, err := inputFromSelect(&s)
			if err != nil {
				return nil, err
			}
			names = append(names, ns...)
		}
	case parser.InsertIntoFromStmt:
		return []string{string(stmt.Input)}, nil
	}
	if err != nil {
		return nil, err
	}
	return uniqueInputNames(names), nil
}
func inputFromSelect(stmt *parser.SelectStmt) ([]string, error) {
	var names []string
	for _, rel := range stmt.Relations {
		switch rel.Type {
		case parser.ActualStream:
			names = append(names, rel.Name)
		case parser.UDSFStream:
			ctx := core.NewContext(nil)
			udfReg := udf.CopyGlobalUDFRegistry(ctx)
			params := make([]data.Value, len(rel.Params))
			for i, expr := range rel.Params {
				p, err := execution.EvaluateFoldable(expr, udfReg)
				if err != nil {
					return nil, err
				}
				params[i] = p
			}
			reg, err := udf.CopyGlobalUDSFCreatorRegistry()
			if err != nil {
				return nil, err
			}
			udsfc, err := reg.Lookup(rel.Name, len(params))
			if err != nil {
				return nil, err
			}
			decl := udf.NewUDSFDeclarer()
			udsf, err := udsfc.CreateUDSF(ctx, decl, params...)
			if err != nil {
				return nil, err
			}
			defer udsf.Terminate(ctx)
			for n := range decl.ListInputs() {
				names = append(names, n)
			}
		}
	}
	return names, nil
}
func uniqueInputNames(names []string) []string {
	if len(names) == 0 {
		return names
	}
	sort.Sort(sort.StringSlice(names))
	prev := ""
	k := 0
	for _, n := range names {
		if n == prev {
			continue
		}
		names[k] = n
		k++
		prev = n
	}
	return names[:k]
}