func TestParseExpressions(t *testing.T) {

	for _, test := range parseTests {
		exprTree, err := expr.ParseExpression(test.qlText)
		//u.Infof("After Parse:  %v", err)
		switch {
		case err == nil && !test.ok:
			t.Errorf("%q: 1 expected error; got none", test.name)
			continue
		case err != nil && test.ok:
			t.Errorf("%q: 2 unexpected error: %v", test.name, err)
			continue
		case err != nil && !test.ok:
			// expected error, got one
			if *VerboseTests {
				u.Infof("%s: %s\n\t%s", test.name, test.qlText, err)
			}
			continue
		}
		var result string
		result = exprTree.Root.StringAST()
		if result != test.result {
			t.Errorf("reslen: %v vs %v", len(result), len(test.result))
			t.Errorf("\n%s \n\t'%v'\nexpected\n\t'%v'", test.name, result, test.result)
		}
	}
}

func (m *DeletionScanner) Run(context *expr.Context) error {
	defer context.Recover()
	defer close(m.msgOutCh)

	u.Infof("In Delete Scanner expr %#v", m.sql.Where)
	select {
	case <-m.SigChan():
		return nil
	default:
		if m.sql.Where != nil {
			// Hm, how do i evaluate here?  Do i need a special Vm?
			//return fmt.Errorf("Not implemented delete vm")
			deletedCt, err := m.db.DeleteExpression(m.sql.Where)
			if err != nil {
				u.Errorf("Could not put values: %v", err)
				return err
			}
			m.deleted = deletedCt
			vals := make([]driver.Value, 2)
			vals[0] = int64(0)
			vals[1] = int64(deletedCt)
			m.msgOutCh <- &datasource.SqlDriverMessage{vals, 1}
		}
	}
	return nil
}

// toFloat64 convert all reflect.Value-s into float64.
func ToFloat64(v reflect.Value) float64 {
	if v.Kind() == reflect.Interface {
		v = v.Elem()
	}
	switch v.Kind() {
	case reflect.Float32, reflect.Float64:
		return v.Float()
	case reflect.Int16, reflect.Int8, reflect.Int, reflect.Int32, reflect.Int64:
		return float64(v.Int())
	case reflect.String:
		s := v.String()
		var f float64
		var err error
		if strings.HasPrefix(s, "0x") {
			f, err = strconv.ParseFloat(s, 64)
		} else {
			f, err = strconv.ParseFloat(s, 64)
		}
		if err == nil {
			return float64(f)
		}
	case reflect.Slice:
		// Should we grab first one?
		item1 := v.Index(0)
		u.Infof("is slice of strings?: %T", v, item1)
	default:
		u.Warnf("Cannot convert type?  %v", v.Kind())
	}
	return math.NaN()
}

func walkUnary(ctx expr.EvalContext, node *expr.UnaryNode) (value.Value, bool) {

	a, ok := Eval(ctx, node.Arg)
	if !ok {
		u.Infof("whoops, %#v", node)
		return a, false
	}
	switch node.Operator.T {
	case lex.TokenNegate:
		switch argVal := a.(type) {
		case value.BoolValue:
			//u.Infof("found urnary bool:  res=%v   expr=%v", !argVal.v, node.StringAST())
			return value.NewBoolValue(!argVal.Val()), true
		default:
			//u.Errorf("urnary type not implementedUnknonwn node type:  %T", argVal)
			panic(ErrUnknownNodeType)
		}
	case lex.TokenMinus:
		if an, aok := a.(value.NumericValue); aok {
			return value.NewNumberValue(-an.Float()), true
		}
	default:
		u.Warnf("urnary not implemented:   %#v", node)
	}

	return value.NewNilValue(), false
}

func (m *Source) LoadConn() error {

	//u.Debugf("LoadConn() nil?%v", m.Conn == nil)
	if m.Conn != nil {
		return nil
	}
	if m.DataSource == nil {
		// Not all sources require a source, ie literal queries
		// and some, information schema, or fully qualifyied schema queries
		// requires schema switching
		if m.IsSchemaQuery() && m.ctx != nil {
			m.ctx.Schema = m.ctx.Schema.InfoSchema
			u.Infof("switching to info schema")
			if err := m.load(); err != nil {
				u.Errorf("could not load schema? %v", err)
				return err
			}
			if m.DataSource == nil {
				return u.LogErrorf("could not load info schema source %v", m.Stmt)
			}
		} else {
			u.Debugf("return bc no datasource ctx=nil?%v schema?%v", m.ctx == nil, m.IsSchemaQuery())
			return nil
		}
	}
	source, err := m.DataSource.Open(m.Stmt.SourceName())
	if err != nil {
		return err
	}
	m.Conn = source
	return nil
}

// sub-select not implemented in exec yet
func testSubselect(t *testing.T) {
	sqlText := `
		select 
	        user_id, email
	    FROM users
	    WHERE user_id in 
	        (select user_id from orders)
    `
	job, err := BuildSqlJob(rtConf, "mockcsv", sqlText)
	assert.Tf(t, err == nil, "no error %v", err)

	//writeCtx := NewContextSimple()
	msgs := make([]datasource.Message, 0)
	go func() {
		outChan := job.DrainChan()
		for msg := range outChan {
			u.Infof("msg: %v", msg)
			msgs = append(msgs, msg)
		}
	}()
	err = job.Run()
	time.Sleep(time.Millisecond * 30)
	assert.Tf(t, err == nil, "no error %v", err)
	assert.Tf(t, len(msgs) == 1, "should have filtered out 2 messages")

}

// 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 TestWithJson(t *testing.T) {
	// This is obviously not exactly sql standard
	// but is nice for proxy's
	sql := `
		SELECT id, name FROM user
		WITH {
			"key":"value2"
			,"keyint":45,
			"keyfloat":55.5, 
			"keybool": true,
			"keyarraymixed":["a",2,"b"],
			"keyarrayobj":[
				{"hello":"value","age":55},
				{"hello":"value","age":55}
			],
			"keyobj":{"hello":"value","age":55},
			"keyobjnested":{
				"hello":"value",
				"array":[
					"a",
					2,
					"b"
				]
			}
		}
		`
	req, err := ParseSql(sql)
	assert.Tf(t, err == nil && req != nil, "Must parse: %s  \n\t%v", sql, err)
	sel, ok := req.(*SqlSelect)
	assert.Tf(t, ok, "is SqlSelect: %T", req)
	assert.Tf(t, len(sel.From) == 1 && sel.From[0].Name == "user", "has 1 from: %v", sel.From)
	assert.Tf(t, len(sel.With) == 8, "has with: %v", sel.With)
	assert.Tf(t, len(sel.With.Helper("keyobj")) == 2, "has 2obj keys: %v", sel.With.Helper("keyobj"))
	u.Infof("sel.With:  \n%s", sel.With.PrettyJson())
}

func (m *SourceSchema) Table(tableName string) (*Table, error) {
	tbl, ok := m.tableMap[tableName]
	if ok && tbl != nil {
		return tbl, nil
	} else if ok && tbl == nil {
		u.Infof("try to get table from source schema %v", tableName)
		if sourceTable, ok := m.DS.(SchemaProvider); ok {
			tbl, err := sourceTable.Table(tableName)
			if err == nil {
				m.AddTable(tbl)
			}
			return tbl, err
		}
	}
	if tbl != nil && !tbl.Current() {
		// What?
		if sourceTable, ok := m.DS.(SchemaProvider); ok {
			tbl, err := sourceTable.Table(tableName)
			if err == nil {
				m.AddTable(tbl)
			}
			return tbl, err
		}
	}
	return nil, fmt.Errorf("Could not find that table: %v", tableName)
}

// 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
}

func RunKafkaConsumer(msgChan chan *loges.LineEvent, partitionstr, topic, kafkaHost string, offset, maxMsgCt uint64, maxSize uint) {
	var broker *kafka.BrokerConsumer

	u.Infof("Connecting to host=%s topic=%s part=%s", kafkaHost, topic, partitionstr)
	parts := strings.Split(partitionstr, ",")
	if len(parts) > 1 {
		tps := kafka.NewTopicPartitions(topic, partitionstr, offset, uint32(maxSize))
		broker = kafka.NewMultiConsumer(kafkaHost, tps)
	} else {
		partition, _ := strconv.Atoi(partitionstr)
		broker = kafka.NewBrokerConsumer(kafkaHost, topic, partition, offset, uint32(maxSize))
	}

	var msgCt int
	done := make(chan bool, 1)
	kafkaMsgs := make(chan *kafka.Message)
	go broker.ConsumeOnChannel(kafkaMsgs, 1000, done)
	for msg := range kafkaMsgs {
		if msg != nil {
			msgCt++
			if uint64(msgCt) > maxMsgCt {
				panic("ending")
			}
			//msg.Print()
			msgChan <- &loges.LineEvent{Data: msg.Payload(), Offset: msg.Offset(), Item: msg}
		} else {
			u.Error("No kafka message?")
			break
		}
	}
}

func (m *MockCsvSource) loadTable(tableName string) error {

	csvRaw, ok := m.raw[tableName]
	if !ok {
		return schema.ErrNotFound
	}
	sr := strings.NewReader(csvRaw)
	u.Debugf("mockcsv:%p load mockcsv: %q  data:%v", m, tableName, csvRaw)
	csvSource, _ := datasource.NewCsvSource(tableName, 0, sr, make(<-chan bool, 1))
	tbl := membtree.NewStaticData(tableName)
	u.Infof("loaded columns %v", csvSource.Columns())
	tbl.SetColumns(csvSource.Columns())
	//u.Infof("set index col for %v: %v -- %v", tableName, 0, csvSource.Columns()[0])
	m.tables[tableName] = tbl

	// Now we are going to page through the Csv rows and Put into
	//  Static Data Source, ie copy into memory btree structure
	for {
		msg := csvSource.Next()
		if msg == nil {
			//u.Infof("table:%v  len=%v", tableName, tbl.Length())
			return nil
		}
		dm, ok := msg.Body().(*datasource.SqlDriverMessageMap)
		if !ok {
			return fmt.Errorf("Expected *datasource.SqlDriverMessageMap but got %T", msg.Body())
		}

		// We don't know the Key
		tbl.Put(nil, nil, dm.Values())
	}
	return nil
}

func TestPlans(t *testing.T) {
	for _, pt := range planTests {
		ctx := td.TestContext(pt.q)
		u.Infof("running %s for plan check", pt.q)
		p := selectPlan(t, ctx)
		assert.T(t, p != nil)

		u.Infof("%#v", ctx.Projection)
		u.Infof("cols %#v", ctx.Projection)
		if pt.cols > 0 {
			// ensure our projection has these columns
			assert.Tf(t, len(ctx.Projection.Proj.Columns) == pt.cols,
				"expected %d cols got %v  %#v", pt.cols, len(ctx.Projection.Proj.Columns), ctx.Projection.Proj)
		}

	}
}

func LoadTestDataOnce() {
	loadData.Do(func() {
		// Load in a "csv file" into our mock data store
		u.Infof("about to load table?")
		mockcsv.LoadTable("users", `user_id,email,interests,reg_date,referral_count
9Ip1aKbeZe2njCDM,"[email protected]","fishing","2012-10-17T17:29:39.738Z",82
hT2impsOPUREcVPc,"[email protected]","swimming","2009-12-11T19:53:31.547Z",12
hT2impsabc345c,"not_an_email","swimming","2009-12-11T19:53:31.547Z",12`)

		u.Infof("after load users table")
		mockcsv.LoadTable("orders", `order_id,user_id,item_id,price,order_date,item_count
1,9Ip1aKbeZe2njCDM,1,22.50,"2012-12-24T17:29:39.738Z",82
2,9Ip1aKbeZe2njCDM,2,37.50,"2013-10-24T17:29:39.738Z",82
3,abcabcabc,1,22.50,"2013-10-24T17:29:39.738Z",82
`)
	})

}

func NewGraphiteRunner(addr string) *GraphiteRunner {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		u.Errorf("Failed to connect to graphite/carbon: %+v", err)
	} else {
		u.Infof("Connected graphite to %v", addr)
	}
	return &GraphiteRunner{conn: conn}
}

func valueArray(pg TokenPager) (value.Value, error) {

	u.Debugf("valueArray cur:%v peek:%v", pg.Cur().V, pg.Peek().V)
	vals := make([]value.Value, 0)
arrayLoop:
	for {
		tok := pg.Next() // consume token
		u.Infof("valueArray() consumed token?: %v", tok)
		switch tok.T {
		case lex.TokenComma:
			// continue
		case lex.TokenRightParenthesis:
			u.Warnf("found right paren  %v cur: %v", tok, pg.Cur())
			break arrayLoop
		case lex.TokenEOF, lex.TokenEOS, lex.TokenFrom, lex.TokenAs:
			u.Debugf("return: %v", tok)
			break arrayLoop
		case lex.TokenValue:
			vals = append(vals, value.NewStringValue(tok.V))
		case lex.TokenFloat, lex.TokenInteger:
			fv, err := strconv.ParseFloat(tok.V, 64)
			if err == nil {
				vals = append(vals, value.NewNumberValue(fv))
			}
			return value.NilValueVal, err
		default:
			return value.NilValueVal, fmt.Errorf("Could not recognize token: %v", tok)
		}

		tok = pg.Next()
		switch tok.T {
		case lex.TokenComma:
			// fine, consume the comma
		case lex.TokenRightBracket:
			u.Warnf("right bracket: %v", tok)
			break arrayLoop
		default:
			u.Warnf("unrecognized token: %v", tok)
			return value.NilValueVal, fmt.Errorf("unrecognized token %v", tok)
		}
	}
	u.Infof("returning array: %v", vals)
	return value.NewSliceValues(vals), nil
}

func TestEngineInsert(t *testing.T) {

	// By "Loading" table we force it to exist in this non DDL mock store
	mockcsv.LoadTable("user_event", "id,user_id,event,date\n1,abcabcabc,signup,\"2012-12-24T17:29:39.738Z\"")
	sqlText := `
		INSERT into user_event (id, user_id, event, date)
		VALUES
			(uuid(), "9Ip1aKbeZe2njCDM", "logon", now())
	`
	job, err := BuildSqlJob(rtConf, "mockcsv", sqlText)
	assert.Tf(t, err == nil, "%v", err)

	err = job.Setup()
	assert.T(t, err == nil)
	err = job.Run()
	assert.T(t, err == nil)
	db, err := datasource.OpenConn("mockcsv", "user_event")
	assert.Tf(t, err == nil, "%v", err)
	gomap, ok := db.(*membtree.StaticDataSource)
	assert.T(t, ok, "Should be type StaticDataSource ", gomap)
	u.Infof("db:  %#v", gomap)
	assert.T(t, gomap.Length() == 2, "Should have inserted")

	// Now lets query it, we are going to use QLBridge Driver
	sqlText = `
		select id, user_id, event, date
	    FROM user_event
	    WHERE user_id = "9Ip1aKbeZe2njCDM"
    `
	sqlDb, err := sql.Open("qlbridge", "mockcsv")
	assert.Tf(t, err == nil, "no error: %v", err)
	assert.Tf(t, sqlDb != nil, "has conn: %v", sqlDb)
	defer func() { sqlDb.Close() }()

	rows, err := sqlDb.Query(sqlText)
	assert.Tf(t, err == nil, "error: %v", err)
	defer rows.Close()
	assert.Tf(t, rows != nil, "has results: %v", rows)
	cols, err := rows.Columns()
	assert.Tf(t, err == nil, "no error: %v", err)
	assert.Tf(t, len(cols) == 4, "4 cols: %v", cols)
	events := make([]*UserEvent, 0)
	for rows.Next() {
		var ue UserEvent
		err = rows.Scan(&ue.Id, &ue.UserId, &ue.Event, &ue.Date)
		assert.Tf(t, err == nil, "no error: %v", err)
		//u.Debugf("events=%+v", ue)
		events = append(events, &ue)
	}
	assert.Tf(t, rows.Err() == nil, "no error: %v", err)
	assert.Tf(t, len(events) == 1, "has 1 event row: %+v", events)

	ue1 := events[0]
	assert.T(t, ue1.Event == "logon")
	assert.T(t, ue1.UserId == "9Ip1aKbeZe2njCDM")
}

// First keyword was INSERT, REPLACE
func (m *Sqlbridge) parseSqlInsert() (*SqlInsert, error) {

	// insert into mytable (id, str) values (0, "a")
	req := NewSqlInsert()
	req.kw = m.Cur().T
	m.Next() // Consume Insert or Replace

	// INTO
	if m.Cur().T != lex.TokenInto {
		return nil, fmt.Errorf("expected INTO but got: %v", m.Cur())
	}
	m.Next() // Consume INTO

	// table name
	switch m.Cur().T {
	case lex.TokenTable:
		req.Table = m.Cur().V
		m.Next()
	default:
		return nil, fmt.Errorf("expected table name but got : %v", m.Cur().V)
	}

	// list of fields
	cols, err := m.parseFieldList()
	if err != nil {
		u.Error(err)
		return nil, err
	}
	req.Columns = cols

	m.Next() // left paren starts lisf of values
	switch m.Cur().T {
	case lex.TokenValues:
		m.Next() // Consume Values keyword
	case lex.TokenSelect:
		u.Infof("What is cur?%v", m.Cur())
		sel, err := m.parseSqlSelect()
		if err != nil {
			return nil, err
		}
		req.Select = sel
		return req, nil
	default:
		return nil, fmt.Errorf("expected values but got : %v", m.Cur().V)
	}

	//u.Debugf("found ?  %v", m.Cur())
	colVals, err := m.parseValueList()
	if err != nil {
		u.Error(err)
		return nil, err
	}
	req.Rows = colVals
	// we are good
	return req, nil
}

func TestRunExpr(t *testing.T) {

	for _, test := range vmTests {

		if *NameMatch != "" && !strings.Contains(test.name, *NameMatch) {
			continue
		}

		//u.Debugf("about to parse: %v", test.qlText)
		exprVm, err := NewVm(test.qlText)

		//u.Infof("After Parse: %v  err=%v", test.qlText, err)
		switch {
		case err == nil && !test.parseok:
			t.Errorf("%q: 1 expected error; got none", test.name)
			continue
		case err != nil && test.parseok:
			t.Errorf("%q: 2 unexpected error: %v", test.name, err)
			continue
		case err != nil && !test.parseok:
			// expected error, got one
			if testing.Verbose() {
				u.Infof("%s: %s\n\t%s", test.name, test.qlText, err)
			}
			continue
		}

		writeContext := datasource.NewContextSimple()
		err = exprVm.Execute(writeContext, test.context)
		if exprVm.Tree != nil && exprVm.Tree.Root != nil {
			//Eval(writeContext, exprVm.Tree.Root)
		}

		results, _ := writeContext.Get("")
		//u.Infof("results:  %T %v  err=%v", results, results, err)
		if err != nil && test.evalok {
			t.Errorf("\n%s -- %v: \n\t%v\nexpected\n\t'%v'", test.name, test.qlText, results, test.result)
		}
		if test.result == nil && results != nil {
			t.Errorf("%s - should have nil result, but got: %v", test.name, results)
			continue
		}
		if test.result != nil && results == nil {
			t.Errorf("%s - should have non-nil result but was nil", test.name)
			continue
		}

		//u.Infof("results=%T   %#v", results, results)
		if test.result != nil && results.Value() != test.result {
			t.Fatalf("\n%s -- %v: \n\t%v--%T\nexpected\n\t%v--%T", test.name, test.qlText, results.Value(), results.Value(), test.result, test.result)
		} else if test.result == nil {
			// we expected nil
		}
	}
}

func (m *Upsert) updateValues(ctx *expr.Context) (int64, error) {

	select {
	case <-m.SigChan():
		return 0, nil
	default:
		// fall through
	}

	valmap := make(map[string]driver.Value, len(m.update.Values))
	for key, valcol := range m.update.Values {
		//u.Debugf("key:%v  val:%v", key, valcol)

		// TODO: #13  Need a way of expressing which layer (here, db) this expr should run in?
		//  - ie, run in backend datasource?   or here?  translate the expr to native language
		if valcol.Expr != nil {
			exprVal, ok := vm.Eval(nil, valcol.Expr)
			if !ok {
				u.Errorf("Could not evaluate: %s", valcol.Expr)
				return 0, fmt.Errorf("Could not evaluate expression: %v", valcol.Expr)
			}
			valmap[key] = exprVal.Value()
		} else {
			u.Debugf("%T  %v", valcol.Value.Value(), valcol.Value.Value())
			valmap[key] = valcol.Value.Value()
		}
		//u.Debugf("key:%v col: %v   vals:%v", key, valcol, valmap[key])
	}

	// if our backend source supports Where-Patches, ie update multiple
	dbpatch, ok := m.db.(datasource.PatchWhere)
	if ok {
		updated, err := dbpatch.PatchWhere(ctx, m.update.Where, valmap)
		u.Infof("patch: %v %v", updated, err)
		if err != nil {
			return updated, err
		}
		return updated, nil
	}

	// TODO:   If it does not implement Where Patch then we need to do a poly fill
	//      Do we have to recognize if the Where is on a primary key?
	// - for sources/queries that can't do partial updates we need to do a read first
	//u.Infof("does not implement PatchWhere")

	// Create a key from Where
	key := datasource.KeyFromWhere(m.update.Where)
	//u.Infof("key: %v", key)
	if _, err := m.db.Put(ctx, key, valmap); err != nil {
		u.Errorf("Could not put values: %v", err)
		return 0, err
	}
	u.Debugf("returning 1")
	return 1, nil
}