func mergeUvMsgs(lmsgs, rmsgs []datasource.Message, lcols, rcols map[string]*expr.Column) []*datasource.ContextUrlValues {
	out := make([]*datasource.ContextUrlValues, 0)
	for _, lm := range lmsgs {
		switch lmt := lm.Body().(type) {
		case *datasource.ContextUrlValues:
			for _, rm := range rmsgs {
				switch rmt := rm.Body().(type) {
				case *datasource.ContextUrlValues:
					// for k, val := range rmt.Data {
					// 	u.Debugf("k=%v v=%v", k, val)
					// }
					newMsg := datasource.NewContextUrlValues(url.Values{})
					newMsg = reAlias(newMsg, lmt.Data, lcols)
					newMsg = reAlias(newMsg, rmt.Data, rcols)
					//u.Debugf("pre:  %#v", lmt.Data)
					//u.Debugf("post:  %#v", newMsg.Data)
					out = append(out, newMsg)
				default:
					u.Warnf("uknown type: %T", rm)
				}
			}
		default:
			u.Warnf("uknown type: %T   %T", lmt, lm)
		}
	}
	return out
}

// Run Command
func (m *Command) Run() error {
	//defer m.Ctx.Recover()
	defer close(m.msgOutCh)

	if m.Ctx.Session == nil {
		u.Warnf("no session?")
		return fmt.Errorf("no session?")
	}

	writeContext, ok := m.Ctx.Session.(expr.ContextWriter)
	if !ok || writeContext == nil {
		u.Warnf("not ok? %T", m.Ctx.Session)
		return fmt.Errorf("No write context?")
	}

	//u.Debugf("running set? %v", m.p.Stmt.String())
	for _, col := range m.p.Stmt.Columns {
		err := evalSetExpression(col, m.Ctx.Session, col.Expr)
		if err != nil {
			u.Warnf("Could not evaluate %s", col.Expr, err)
			return err
		}
	}
	// for k, v := range m.Ctx.Session.Row() {
	// 	u.Infof("%p session? %s: %v", m.Ctx.Session, k, v.Value())
	// }
	return nil
}

// Infer Value type from Node
func ValueTypeFromNode(n Node) value.ValueType {
	switch nt := n.(type) {
	case *FuncNode:
	case *StringNode:
		return value.StringType
	case *IdentityNode:
		// ??
		return value.StringType
	case *NumberNode:
		return value.NumberType
	case *BinaryNode:
		switch nt.Operator.T {
		case lex.TokenLogicAnd, lex.TokenLogicOr:
			return value.BoolType
		case lex.TokenMultiply, lex.TokenMinus, lex.TokenAdd, lex.TokenDivide:
			return value.NumberType
		case lex.TokenModulus:
			return value.IntType
		default:
			u.Warnf("NoValueType? %T", n)
		}
	case nil:
		return value.UnknownType
	default:
		u.Warnf("NoValueType? %T", n)
	}
	return value.UnknownType
}

func (m *PlannerDefault) WalkDelete(p *Delete) error {
	u.Debugf("VisitDelete %+v", p.Stmt)
	conn, err := m.Ctx.Schema.Open(p.Stmt.Table)
	if err != nil {
		u.Warnf("%p no schema for %q err=%v", m.Ctx.Schema, p.Stmt.Table, err)
		return err
	}

	mutatorSource, hasMutator := conn.(schema.ConnMutation)
	if hasMutator {
		mutator, err := mutatorSource.CreateMutator(m.Ctx)
		if err != nil {
			u.Warnf("%p could not create mutator for %q err=%v", m.Ctx.Schema, p.Stmt.Table, err)
			//return nil, err
		} else {
			p.Source = mutator
			return nil
		}
	}

	deleteDs, isDelete := conn.(schema.ConnDeletion)
	if !isDelete {
		return fmt.Errorf("%T does not implement required schema.Deletion for deletions", conn)
	}
	p.Source = deleteDs
	return nil
}

func encodeExpected(
	t *testing.T, label string, val interface{}, wantStr string, wantErr error,
) {
	var buf bytes.Buffer
	enc := NewEncoder(&buf)
	err := enc.Encode(val)
	if err != wantErr {
		if wantErr != nil {
			if wantErr == errAnything && err != nil {
				return
			}
			t.Errorf("%s: want Encode error %v, got %v", label, wantErr, err)
		} else {
			t.Errorf("%s: Encode failed: %s", label, err)
		}
	}
	if err != nil {
		return
	}
	if got := buf.String(); wantStr != got {
		u.Debugf("\n\n%s wanted: \n%s\ngot: \n%s", label, wantStr, got)
		for pos, r := range wantStr {
			if len(got)-1 <= pos {
				u.Warnf("len mismatch? %v vs %v", len(got), len(wantStr))
			} else if r != rune(got[pos]) {
				u.Warnf("mismatch at position: %v   %s!=%s", pos, string(r), string(got[pos]))
				break
			}
		}
		t.Fatalf("%s: want\n-----\n%q\n-----\nbut got\n-----\n%q\n-----\n", label, wantStr, got)
	}
}

func joinValue(ctx *Context, node expr.Node, msg datasource.Message) (string, bool) {

	if msg == nil {
		u.Warnf("got nil message?")
	}
	if msgReader, ok := msg.Body().(expr.ContextReader); ok {

		joinVal, ok := vm.Eval(msgReader, node)
		//u.Debugf("msg: %#v", msgReader)
		//u.Infof("evaluating: ok?%v T:%T result=%v node expr:%v", ok, joinVal, joinVal.ToString(), node.StringAST())
		if !ok {
			u.Errorf("could not evaluate: %v", msg)
			return "", false
		}
		switch val := joinVal.(type) {
		case value.StringValue:
			return val.Val(), true
		default:
			u.Warnf("unknown type? %T", joinVal)
		}
	} else {
		u.Errorf("could not convert to message reader: %T", msg.Body())
	}
	return "", false
}

func (m *JobExecutor) WalkSource(p *plan.Source) (Task, error) {
	//u.Debugf("%p NewSource? %p", m, p)
	if len(p.Static) > 0 {
		//u.Warnf("found static source")
		static := membtree.NewStaticData("static")
		static.SetColumns(p.Cols)
		_, err := static.Put(nil, nil, p.Static)
		if err != nil {
			u.Errorf("Could not put %v", err)
		}
		return NewSourceScanner(m.Ctx, p, static), nil
	} else if p.Conn == nil {
		u.Warnf("no conn? %T", p.DataSource)
		if p.DataSource == nil {
			u.Warnf("no datasource")
			return nil, fmt.Errorf("missing data source")
		}
		source, err := p.DataSource.Open(p.Stmt.SourceName())
		if err != nil {
			return nil, err
		}
		p.Conn = source
		//u.Debugf("setting p.Conn %p %T", p.Conn, p.Conn)
	}

	e, hasSourceExec := p.Conn.(ExecutorSource)
	if hasSourceExec {
		return e.WalkExecSource(p)
	}
	return NewSource(m.Ctx, p)
}

// Test to make sure we get what we expect.
func expect(t *testing.T, lx *lexer, items []item) {
	for i := 0; i < len(items); i++ {
		item := lx.nextItem()
		if item.typ == itemEOF {
			break
		} else if item.typ == itemError {
			t.Fatal(item.val)
		}
		if item != items[i] {
			//u.Debugf("\n\n%s wanted: \n%s\ngot: \n%s", label, wantStr, got)
			wantStr := items[i].val
			got := item.val
			for pos, r := range wantStr {
				if len(got)-1 < pos {
					u.Warnf("len mismatch? %v vs %v", len(got), len(wantStr))
				} else if r != rune(got[pos]) {
					u.Warnf("mismatch at position: %v   %q!=%q", pos, string(r), string(got[pos]))
					break
				} else {
					//u.Debugf("match at position: %v   %q=%q", pos, string(r), string(got[pos]))
				}
			}
			t.Fatalf("Testing: '%s'\nExpected %q, received %q\n",
				lx.input, items[i], item)
		}
	}
}

func keyFromWhere(wh expr.Node) datasource.Key {
	switch n := wh.(type) {
	case *expr.SqlWhere:
		return keyFromWhere(n.Expr)
	case *expr.BinaryNode:
		if len(n.Args) != 2 {
			u.Warnf("need more args? %#v", n.Args)
			return nil
		}
		in, ok := n.Args[0].(*expr.IdentityNode)
		if !ok {
			u.Warnf("not identity? %T", n.Args[0])
			return nil
		}
		// This only allows for    identity = value
		// NOT:      identity = expr(identity, arg)
		//
		switch valT := n.Args[1].(type) {
		case *expr.NumberNode:
			return datasource.NewKeyCol(in.Text, valT.Float64)
		case *expr.StringNode:
			return datasource.NewKeyCol(in.Text, valT.Text)
		//case *expr.FuncNode:
		default:
			u.Warnf("not supported arg? %#v", valT)
		}
	default:
		u.Warnf("not supported node type? %#v", n)
	}
	return nil
}

// MultiNode evaluator
//
//     A   IN   (b,c,d)
//
func walkMulti(ctx expr.EvalContext, node *expr.MultiArgNode) (value.Value, bool) {

	a, aok := Eval(ctx, node.Args[0])
	//u.Infof("multi:  %T:%v  %v", a, a, node.Operator)
	if !aok {
		u.Infof("Could not evaluate args, %#v", node.Args[0])
		return value.BoolValueFalse, false
	}
	switch node.Operator.T {
	case lex.TokenIN:
		for i := 1; i < len(node.Args); i++ {
			v, ok := Eval(ctx, node.Args[i])
			if ok {
				//u.Debugf("in? %v %v", a, v)
				if eq, err := value.Equal(a, v); eq && err == nil {
					return value.NewBoolValue(true), true
				}
			} else {
				u.Warnf("could not evaluate arg: %v", node.Args[i])
			}
		}
		return value.NewBoolValue(false), true
	default:
		u.Warnf("tri node walk not implemented:   %#v", node)
	}

	return value.NewNilValue(), false
}

func (c *FuncNode) Check() error {

	if len(c.Args) < len(c.F.Args) && !c.F.VariadicArgs {
		return fmt.Errorf("parse: not enough arguments for %s  supplied:%d  f.Args:%v", c.Name, len(c.Args), len(c.F.Args))
	} else if (len(c.Args) >= len(c.F.Args)) && c.F.VariadicArgs {
		// ok
	} else if len(c.Args) > len(c.F.Args) {
		u.Warnf("lenc.Args >= len(c.F.Args?  %v", (len(c.Args) >= len(c.F.Args)))
		err := fmt.Errorf("parse: too many arguments for %s want:%v got:%v   %#v", c.Name, len(c.F.Args), len(c.Args), c.Args)
		u.Errorf("funcNode.Check(): %v", err)
		return err
	}
	for i, a := range c.Args {

		if ne, isNodeExpr := a.(Node); isNodeExpr {
			if err := ne.Check(); err != nil {
				return err
			}
		} else if _, isValue := a.(value.Value); isValue {
			// TODO: we need to check co-ercion here, ie which Args can be converted to what types
			if nodeVal, ok := a.(NodeValueType); ok {
				// For Env Variables, we need to Check those (On Definition?)
				if c.F.Args[i].Kind() != nodeVal.Type().Kind() {
					u.Errorf("error in parse Check(): %v", a)
					return fmt.Errorf("parse: expected %v, got %v    ", nodeVal.Type().Kind(), c.F.Args[i].Kind())
				}
			}
		} else {
			u.Warnf("Unknown type for func arg %T", a)
			return fmt.Errorf("Unknown type for func arg %T", a)
		}
	}
	return nil
}

// Create a source schema from given named source
//  we will find Source for that name and introspect
func createSchema(sourceName string) (*schema.Schema, bool) {

	sourceName = strings.ToLower(sourceName)
	ss := schema.NewSchemaSource(sourceName, sourceName)

	ds := registry.Get(sourceName)
	if ds == nil {
		parts := strings.SplitN(sourceName, "://", 2)
		//u.Infof("parts: %d   %v", len(parts), parts)
		if len(parts) == 2 {
			ds = registry.Get(parts[0])
			if ds == nil {
				//return &qlbConn{schema: s, connInfo: parts[1]}, nil
				u.Warnf("not able to find schema %q", sourceName)
				return nil, false
			}
		} else {
			//u.WarnT(7)
			u.Warnf("not able to find schema %q", sourceName)
			return nil, false
		}
	}

	u.Infof("reg p:%p source=%q  ds %#v tables:%v", registry, sourceName, ds, ds.Tables())
	ss.DS = ds
	schema := schema.NewSchema(sourceName)
	ss.Schema = schema
	u.Debugf("schema:%p ss:%p createSchema(%q) NEW ", schema, ss, sourceName)

	loadSchema(ss)

	return schema, true
}

// Infer Value type from Node
func ValueTypeFromNode(n Node) value.ValueType {
	switch nt := n.(type) {
	case *FuncNode:
		return value.UnknownType
	case *StringNode:
		return value.StringType
	case *IdentityNode:
		// ??
		return value.StringType
	case *NumberNode:
		return value.NumberType
	case *BinaryNode:
		switch nt.Operator.T {
		case lex.TokenLogicAnd, lex.TokenLogicOr, lex.TokenEqual, lex.TokenEqualEqual:
			return value.BoolType
		case lex.TokenMultiply, lex.TokenMinus, lex.TokenAdd, lex.TokenDivide:
			return value.NumberType
		case lex.TokenModulus:
			return value.IntType
		case lex.TokenLT, lex.TokenLE, lex.TokenGT, lex.TokenGE:
			return value.BoolType
		default:
			//u.LogTracef(u.WARN, "hello")
			u.Warnf("NoValueType? %T  %#v", n, n)
		}
	case nil:
		return value.UnknownType
	default:
		u.Warnf("NoValueType? %T", n)
	}
	return value.UnknownType
}

func upsertSource(ctx *Context, table string) (schema.ConnUpsert, error) {

	conn, err := ctx.Schema.Open(table)
	if err != nil {
		u.Warnf("%p no schema for %q err=%v", ctx.Schema, table, err)
		return nil, err
	}

	mutatorSource, hasMutator := conn.(schema.ConnMutation)
	if hasMutator {
		mutator, err := mutatorSource.CreateMutator(ctx)
		if err != nil {
			u.Warnf("%p could not create mutator for %q err=%v", ctx.Schema, table, err)
			//return nil, err
		} else {
			return mutator, nil
		}
	}

	upsertDs, isUpsert := conn.(schema.ConnUpsert)
	if !isUpsert {
		return nil, fmt.Errorf("%T does not implement required schema.Upsert for upserts", conn)
	}
	return upsertDs, nil
}

func rewriteNode(from *SqlSource, isLeft bool, node Node) Node {
	switch nt := node.(type) {
	case *IdentityNode:
		if left, right, ok := nt.LeftRight(); ok {
			//u.Debugf("rewriteNode   isLeft?%v from.Name:%v l:%v  r:%v", isLeft, from.alias, left, right)
			if left == from.alias {
				in := IdentityNode{Text: right}
				//u.Warnf("nice, found it! in = %v", in)
				return &in
			}
		}
	case *BinaryNode:
		switch nt.Operator.T {
		case lex.TokenAnd, lex.TokenLogicAnd, lex.TokenLogicOr:
			n1 := rewriteNode(from, isLeft, nt.Args[0])
			n2 := rewriteNode(from, isLeft, nt.Args[1])
			return &BinaryNode{Operator: nt.Operator, Args: [2]Node{n1, n2}}
		case lex.TokenEqual, lex.TokenEqualEqual:
			n := rewriteNode(from, isLeft, nt.Args[0])
			if n != nil {
				return n
			}
			n = rewriteNode(from, isLeft, nt.Args[1])
			if n != nil {
				return n
			}
			u.Warnf("Could not find node: %#v", node)
		default:
			u.Warnf("un-implemented op: %v", nt.Operator)
		}
	default:
		u.Warnf("%T node types are not suppored yet for join rewrite", node)
	}
	return nil
}

func (m *DataSources) Get(sourceType string) DataSource {
	if source, ok := m.sources[strings.ToLower(sourceType)]; ok {
		return source
	}
	if len(m.sources) == 1 {
		for _, src := range m.sources {
			return src
		}
	}
	u.Warnf("what are we getting? %v", sourceType)
	if len(m.tableSources) == 0 {
		for _, src := range m.sources {
			tbls := src.Tables()
			for _, tbl := range tbls {
				if _, ok := m.tableSources[tbl]; ok {
					u.Warnf("table names must be unique across sources %v", tbl)
				} else {
					m.tableSources[tbl] = src
				}
			}
		}
	}
	if src, ok := m.tableSources[sourceType]; ok {
		return src
	}
	return nil
}

func mergeValuesMsgs(lmsgs, rmsgs []datasource.Message, lcols, rcols []*expr.Column, cols map[string]*expr.Column) []*datasource.SqlDriverMessageMap {
	out := make([]*datasource.SqlDriverMessageMap, 0)
	//u.Infof("merge values: %v:%v", len(lcols), len(rcols))
	for _, lm := range lmsgs {
		switch lmt := lm.(type) {
		case *datasource.SqlDriverMessage:
			//u.Warnf("got sql driver message: %#v", lmt.Vals)
			for _, rm := range rmsgs {
				switch rmt := rm.(type) {
				case *datasource.SqlDriverMessage:
					// for k, val := range rmt.Vals {
					// 	u.Debugf("k=%v v=%v", k, val)
					// }
					newMsg := datasource.NewSqlDriverMessageMap()
					newMsg = reAlias2(newMsg, lmt.Vals, lcols)
					newMsg = reAlias2(newMsg, rmt.Vals, rcols)
					//u.Debugf("pre:  %#v", lmt.Vals)
					//u.Debugf("newMsg:  %#v", newMsg.Vals)
					out = append(out, newMsg)
				default:
					u.Warnf("uknown type: %T", rm)
				}
			}
		default:
			u.Warnf("uknown type: %T   %T", lmt, lm)
		}
	}
	return out
}

func evalSetExpression(col *rel.CommandColumn, ctx expr.ContextReadWriter, arg expr.Node) error {

	switch bn := arg.(type) {
	case *expr.BinaryNode:
		_, ok := bn.Args[0].(*expr.IdentityNode)
		if !ok {
			u.Warnf("expected identity but got %T in %s", bn.Args[0], arg.String())
			return fmt.Errorf("Expected identity but got %T", bn.Args[0])
		}
		rhv, ok := vm.Eval(ctx, bn.Args[1])
		if !ok {
			u.Warnf("expected right side value but got %T in %s", bn.Args[1], arg.String())
			return fmt.Errorf("Expected value but got %T", bn.Args[1])
		}
		//u.Infof(`writeContext.Put("%v",%v)`, col.Key(), rhv.Value())
		ctx.Put(col, ctx, rhv)
	case nil:
		// Special statements
		name := strings.ToLower(col.Name)
		switch {
		case strings.HasPrefix(name, "names ") || strings.HasPrefix(name, "character set"):
			// http://dev.mysql.com/doc/refman/5.7/en/charset-connection.html
			// hm, no idea what to do
			/*
				SET character_set_client = charset_name;
				SET character_set_results = charset_name;
				SET character_set_connection = charset_name;
			*/
		}
	default:
		u.Errorf("SET command only accepts binary nodes but got type:  %#v", arg)
		return fmt.Errorf("Un recognized command %T", arg)
	}
	return nil
}

func (m *CsvDataSource) Next() schema.Message {
	select {
	case <-m.exit:
		return nil
	default:
		for {
			row, err := m.csvr.Read()

			if err != nil {
				if err == io.EOF {
					return nil
				}
				u.Warnf("could not read row? %v", err)
				continue
			}
			m.rowct++
			if len(row) != len(m.headers) {
				u.Warnf("headers/cols dont match, dropping expected:%d got:%d   vals=", len(m.headers), len(row), row)
				continue
			}
			vals := make([]driver.Value, len(row))
			for i, val := range row {
				vals[i] = val
			}
			//u.Debugf("headers: %#v \n\trows:  %#v", m.headers, row)
			return NewSqlDriverMessageMap(m.rowct, vals, m.colindex)
		}
	}
}

func msgToRow(msg datasource.Message, cols []string, dest []driver.Value) error {

	//u.Debugf("msg? %v  %T \n%p %v", msg, msg, dest, dest)
	switch mt := msg.Body().(type) {
	case *datasource.ContextUrlValues:
		for i, key := range cols {
			if val, ok := mt.Get(key); ok && !val.Nil() {
				dest[i] = val.Value()
				//u.Infof("key=%v   val=%v", key, val)
			} else {
				u.Warnf("missing value? %v %T %v", key, val.Value(), val.Value())
			}
		}
		//u.Debugf("got msg in row result writer: %#v", mt)
	case *datasource.ContextSimple:
		for i, key := range cols {
			//u.Debugf("key=%v mt = nil? %v", key, mt)
			if val, ok := mt.Get(key); ok && val != nil && !val.Nil() {
				dest[i] = val.Value()
				//u.Infof("key=%v   val=%v", key, val)
			} else if val == nil {
				u.Errorf("could not evaluate? %v  %#v", key, mt)
			} else {
				u.Warnf("missing value? %v %T %v", key, val.Value(), val.Value())
			}
		}
		//u.Debugf("got msg in row result writer: %#v", mt)
	default:
		u.Errorf("unknown message type: %T", mt)
	}
	return nil
}