func (m *JobBuilder) VisitJoin(from *expr.SqlSource) (expr.Task, error) {
	u.Debugf("VisitJoin %s", from.Source)
	//u.Debugf("from.Name:'%v' : %v", from.Name, from.Source.String())
	source := m.schema.Conn(from.SourceName())
	//u.Debugf("left source: %T", source)
	// Must provider either Scanner, SourcePlanner, Seeker interfaces
	if sourcePlan, ok := source.(datasource.SourcePlanner); ok {
		//  This is flawed, visitor pattern would have you pass in a object which implements interface
		//    but is one of many different objects that implement that interface so that the
		//    Accept() method calls the apppropriate method
		u.Warnf("SourcePlanner????")
		scanner, err := sourcePlan.Accept(NewSourcePlan(from))
		if err == nil {
			return NewSourceJoin(from, scanner), nil
		}
		u.Errorf("Could not source plan for %v  %T %#v", from.Name, source, source)
	}

	scanner, ok := source.(datasource.Scanner)
	if !ok {
		u.Errorf("Could not create scanner for %v  %T %#v", from.Name, source, source)
		return nil, fmt.Errorf("Must Implement Scanner")
	}
	if err := buildColIndex(scanner, from); err != nil {
		return nil, err
	}
	return NewSourceJoin(from, scanner), nil
}

// Get connection for given Database
//
//  @db      database name
//
func (m *RuntimeSchema) Conn(db string) SourceConn {

	if m.connInfo == "" {
		//u.Debugf("RuntimeConfig.Conn(db='%v')   // connInfo='%v'", db, m.connInfo)
		if source := m.Sources.Get(strings.ToLower(db)); source != nil {
			//u.Debugf("found source: db=%s   %T", db, source)
			conn, err := source.Open(db)
			if err != nil {
				u.Errorf("could not open data source: %v  %v", db, err)
				return nil
			}
			//u.Infof("source: %T  %#v", conn, conn)
			return conn
		} else {
			u.Errorf("DataSource(%s) was not found", db)
		}
	} else {
		//u.Debugf("No Conn? RuntimeConfig.Conn(db='%v')   // connInfo='%v'", db, m.connInfo)
		// We have connection info, likely sq/driver
		source := m.DataSource(m.connInfo)
		//u.Infof("source=%v    about to call Conn() db='%v'", source, db)
		conn, err := source.Open(db)

		if err != nil {
			u.Errorf("could not open data source: %v  %v", db, err)
			return nil
		}
		return conn
	}
	return nil
}

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
}

func (m *Upsert) insertRows(ctx *Context, rows [][]*expr.ValueColumn) error {

	for _, row := range rows {
		//u.Infof("In Insert Scanner iter %#v", row)
		select {
		case <-m.SigChan():
			return nil
		default:
			vals := make([]driver.Value, len(row))
			for x, val := range row {
				if val.Expr != nil {
					exprVal, ok := vm.Eval(nil, val.Expr)
					if !ok {
						u.Errorf("Could not evaluate: %v", val.Expr)
						return fmt.Errorf("Could not evaluate expression: %v", val.Expr)
					}
					vals[x] = exprVal.Value()
				} else {
					vals[x] = val.Value.Value()
				}

				//u.Debugf("%d col: %v   vals:%v", x, val, vals[x])
			}
			if _, err := m.db.Put(ctx, nil, vals); err != nil {
				u.Errorf("Could not put values: %v", err)
				return err
			}
			// continue
		}
	}
	return nil
}

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
}

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
}

// A scanner to read from data source
func NewSourceJoin(leftFrom, rightFrom *expr.SqlSource, conf *RuntimeConfig) (*SourceJoin, error) {
	m := &SourceJoin{
		TaskBase: NewTaskBase("SourceJoin"),
	}
	m.TaskBase.TaskType = m.Type()

	m.leftStmt = leftFrom
	m.rightStmt = rightFrom

	source := conf.Conn(leftFrom.Name)
	u.Debugf("source: %T", source)
	// Must provider either Scanner, and or Seeker interfaces
	if scanner, ok := source.(datasource.Scanner); !ok {
		u.Errorf("Could not create scanner for %v  %T %#v", leftFrom.Name, source, source)
		return nil, fmt.Errorf("Must Implement Scanner")
	} else {
		m.leftSource = scanner
	}

	source2 := conf.Conn(rightFrom.Name)
	u.Debugf("source right: %T", source2)
	// Must provider either Scanner, and or Seeker interfaces
	if scanner, ok := source2.(datasource.Scanner); !ok {
		u.Errorf("Could not create scanner for %v  %T %#v", leftFrom.Name, source2, source2)
		return nil, fmt.Errorf("Must Implement Scanner")
	} else {
		m.rightSource = scanner
	}

	return m, nil
}

func (m *JobExecutor) WalkJoin(p *plan.JoinMerge) (Task, error) {
	execTask := NewTaskParallel(m.Ctx)
	//u.Debugf("join.Left: %#v    \nright:%#v", p.Left, p.Right)
	l, err := m.WalkPlanAll(p.Left)
	if err != nil {
		u.Errorf("whoops %T  %v", l, err)
		return nil, err
	}
	err = execTask.Add(l)
	if err != nil {
		u.Errorf("whoops %T  %v", l, err)
		return nil, err
	}
	r, err := m.WalkPlanAll(p.Right)
	if err != nil {
		return nil, err
	}
	err = execTask.Add(r)
	if err != nil {
		return nil, err
	}

	jm := NewJoinNaiveMerge(m.Ctx, l.(TaskRunner), r.(TaskRunner), p)
	err = execTask.Add(jm)
	if err != nil {
		return nil, err
	}
	return execTask, nil
}

// Delete using a Where Expression
func (m *dbConn) DeleteExpression(where expr.Node) (int, error) {
	//return 0, fmt.Errorf("not implemented")
	evaluator := vm.Evaluator(where)
	var deletedKeys []schema.Key
	txn := m.db.Txn(true)
	iter, err := txn.Get(m.md.tbl.Name, m.md.primaryIndex)
	if err != nil {
		txn.Abort()
		u.Errorf("could not get values %v", err)
		return 0, err
	}
deleteLoop:
	for {
		item := iter.Next()
		if item == nil {
			break
		}

		msg, ok := item.(datasource.SqlDriverMessage)
		if !ok {
			u.Warnf("wat?  %T   %#v", item, item)
			err = fmt.Errorf("unexpected message type %T", item)
			break
		}
		whereValue, ok := evaluator(msg.ToMsgMap(m.md.tbl.FieldPositions))
		if !ok {
			u.Debugf("could not evaluate where: %v", msg)
		}
		switch whereVal := whereValue.(type) {
		case value.BoolValue:
			if whereVal.Val() == false {
				//this means do NOT delete
			} else {
				// Delete!
				if err = txn.Delete(m.md.tbl.Name, msg); err != nil {
					u.Errorf("could not delete %v", err)
					break deleteLoop
				}
				indexVal := msg.Vals[0]
				deletedKeys = append(deletedKeys, schema.NewKeyUint(makeId(indexVal)))
			}
		case nil:
			// ??
			u.Warnf("this should be fine, couldn't evaluate so don't delete %v", msg)
		default:
			if whereVal.Nil() {
				// Doesn't match, so don't delete
			} else {
				u.Warnf("unknown where eval result? %T", whereVal)
			}
		}
	}
	if err != nil {
		txn.Abort()
		return 0, err
	}
	txn.Commit()
	return len(deletedKeys), nil
}

func SourceFromPB(pb *PlanPb, ctx *Context) (*Source, error) {
	m := Source{
		SourcePb: pb.Source,
		ctx:      ctx,
	}
	if len(pb.Source.Custom) > 0 {
		m.Custom = make(u.JsonHelper)
		if err := json.Unmarshal(pb.Source.Custom, &m.Custom); err != nil {
			u.Errorf("Could not unmarshall custom data %v", err)
		}
		//u.Debugf("custom %v", m.Custom)
	}
	if pb.Source.Projection != nil {
		m.Proj = rel.ProjectionFromPb(pb.Source.Projection)
	}
	if pb.Source.SqlSource != nil {
		m.Stmt = rel.SqlSourceFromPb(pb.Source.SqlSource)
	}
	m.PlanBase = NewPlanBase(pb.Parallel)
	if len(pb.Children) > 0 {
		m.tasks = make([]Task, len(pb.Children))
		for i, pbt := range pb.Children {
			childPlan, err := SelectTaskFromTaskPb(pbt, ctx, m.Stmt.Source)
			if err != nil {
				u.Errorf("%T not implemented? %v", pbt, err)
				return nil, err
			}
			m.tasks[i] = childPlan
		}
	}

	err := m.load()
	if err != nil {
		u.Errorf("could not load? %v", err)
		return nil, err
	}
	if m.Conn == nil {
		err = m.LoadConn()
		if err != nil {
			u.Errorf("conn error? %v", err)
			return nil, err
		}
		if m.Conn == nil {
			if m.Stmt != nil {
				if m.Stmt.IsLiteral() {
					// this is fine
				} else {
					u.Warnf("no data source and not literal query? %s", m.Stmt.String())
					return nil, ErrNoDataSource
				}
			} else {
				//u.Warnf("hm  no conn, no stmt?....")
				//return nil, ErrNoDataSource
			}
		}
	}

	return &m, nil
}

func (m *filterQLParser) parseFilterClause(depth int, negate bool) (*FilterExpr, error) {

	fe := NewFilterExpr()
	fe.Negate = negate
	//u.Debugf("%d filterclause? negate?%v  cur=%v", depth, negate, m.Cur())

	switch m.Cur().T {
	case lex.TokenInclude:
		// embed/include a named filter

		m.Next()
		//u.Infof("type %v", m.Cur())
		if m.Cur().T != lex.TokenIdentity && m.Cur().T != lex.TokenValue {
			return nil, fmt.Errorf("Expected identity for Include but got %v", m.Cur())
		}
		fe.Include = m.Cur().V
		m.Next()

	case lex.TokenUdfExpr:
		// we have a udf/functional expression filter
		tree := expr.NewTree(m.filterTokenPager)
		if err := m.parseNode(tree); err != nil {
			u.Errorf("could not parse: %v", err)
			return nil, err
		}
		fe.Expr = tree.Root

	case lex.TokenIdentity, lex.TokenLike, lex.TokenExists, lex.TokenBetween,
		lex.TokenIN, lex.TokenIntersects, lex.TokenValue, lex.TokenContains:

		if m.Cur().T == lex.TokenIdentity {
			if strings.ToLower(m.Cur().V) == "include" {
				// TODO:  this is a bug in lexer ...
				// embed/include a named filter
				m.Next()
				if m.Cur().T != lex.TokenIdentity && m.Cur().T != lex.TokenValue {
					return nil, fmt.Errorf("Expected identity for Include but got %v", m.Cur())
				}
				fe.Include = m.Cur().V
				m.Next()
				return fe, nil
			}
		}

		tree := expr.NewTree(m.filterTokenPager)
		if err := m.parseNode(tree); err != nil {
			u.Errorf("could not parse: %v", err)
			return nil, err
		}
		fe.Expr = tree.Root
		if !m.fs.HasDateMath {
			m.fs.HasDateMath = expr.HasDateMath(fe.Expr)
		}
	default:
		return nil, fmt.Errorf("Expected clause but got %v", m.Cur())
	}
	return fe, 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
}

func (m *Projection) loadFinal(ctx *Context, isFinal bool) error {

	//u.Debugf("creating plan.Projection final %s", m.Stmt.String())

	m.Proj = rel.NewProjection()

	for _, from := range m.Stmt.From {

		fromName := strings.ToLower(from.SourceName())
		tbl, err := ctx.Schema.Table(fromName)
		if err != nil {
			u.Errorf("could not get table: %v", err)
			return err
		} else if tbl == nil {
			u.Errorf("unexepcted nil table? %v", from.Name)
			return fmt.Errorf("Table not found %q", from.Name)
		} else {

			//u.Debugf("getting cols? %v   cols=%v", from.ColumnPositions())
			for _, col := range from.Source.Columns {
				//_, right, _ := col.LeftRight()
				//u.Infof("col %#v", col)
				if col.Star {
					for _, f := range tbl.Fields {
						m.Proj.AddColumnShort(f.Name, f.Type)
					}
				} else {
					if schemaCol, ok := tbl.FieldMap[col.SourceField]; ok {
						if isFinal {
							if col.InFinalProjection() {
								//u.Debugf("in plan final %s", col.As)
								m.Proj.AddColumnShort(col.As, schemaCol.Type)
							}
						} else {
							//u.Debugf("not final %s", col.As)
							m.Proj.AddColumnShort(col.As, schemaCol.Type)
						}
						//u.Debugf("projection: %p add col: %v %v", m.Proj, col.As, schemaCol.Type.String())
					} else {
						//u.Warnf("schema col not found: final?%v col: %#v", isFinal, col)
						if isFinal {
							if col.InFinalProjection() {
								m.Proj.AddColumnShort(col.As, value.StringType)
							}
						} else {
							m.Proj.AddColumnShort(col.As, value.StringType)
						}
					}
				}

			}
		}
	}
	return nil
}

func main() {

	if sqlText == "" {
		u.Errorf("You must provide a valid select query in argument:    --sql=\"select ...\"")
		return
	}

	// load all of our built-in functions
	builtins.LoadAllBuiltins()

	// Add a custom function to the VM to make available to SQL language
	expr.FuncAdd("email_is_valid", EmailIsValid)

	// Our file source of csv's is stdin
	stdIn, err := os.Open("/dev/stdin")
	if err != nil {
		u.Errorf("could not open stdin? %v", err)
		return
	}

	// We are registering the "csv" datasource, to show that
	// the backend/sources can be easily created/added.  This csv
	// reader is an example datasource that is very, very simple.
	exit := make(chan bool)
	src, _ := datasource.NewCsvSource("stdin", 0, stdIn, exit)
	datasource.Register("csv", src)

	db, err := sql.Open("qlbridge", "csv:///dev/stdin")
	if err != nil {
		panic(err.Error())
	}
	defer db.Close()

	rows, err := db.Query(sqlText)
	if err != nil {
		u.Errorf("could not execute query: %v", err)
		return
	}
	defer rows.Close()
	cols, _ := rows.Columns()

	// this is just stupid hijinx for getting pointers for unknown len columns
	readCols := make([]interface{}, len(cols))
	writeCols := make([]string, len(cols))
	for i, _ := range writeCols {
		readCols[i] = &writeCols[i]
	}
	fmt.Printf("\n\nScanning through CSV: (%v)\n\n", strings.Join(cols, ","))
	for rows.Next() {
		rows.Scan(readCols...)
		fmt.Println(strings.Join(writeCols, ", "))
	}
	fmt.Println("")
}

func (s *SearchDsl) Result(conn *Conn) (*SearchResult, error) {
	var retval SearchResult
	body, err := s.Bytes(conn)
	if err != nil {
		u.Errorf("%v", err)
		return nil, err
	}
	jsonErr := json.Unmarshal(body, &retval)
	if jsonErr != nil {
		u.Errorf("%v \n\t%s", jsonErr, string(body))
	}
	return &retval, jsonErr
}

func whereFilter(where expr.Node, task TaskRunner, cols map[string]*expr.Column) MessageHandler {
	out := task.MessageOut()
	evaluator := vm.Evaluator(where)
	return func(ctx *Context, msg datasource.Message) bool {
		// defer func() {
		// 	if r := recover(); r != nil {
		// 		u.Errorf("crap, %v", r)
		// 	}
		// }()
		var whereValue value.Value
		var ok bool

		switch mt := msg.(type) {
		case *datasource.SqlDriverMessage:
			//u.Debugf("WHERE:  T:%T  vals:%#v", msg, mt.Vals)
			//u.Debugf("cols:  %#v", cols)
			msgReader := datasource.NewValueContextWrapper(mt, cols)
			whereValue, ok = evaluator(msgReader)
		case *datasource.SqlDriverMessageMap:
			whereValue, ok = evaluator(mt)
		default:
			if msgReader, ok := msg.(expr.ContextReader); ok {
				whereValue, ok = evaluator(msgReader)
			} else {
				u.Errorf("could not convert to message reader: %T", msg)
			}
		}
		//u.Debugf("msg: %#v", msgReader)
		//u.Infof("evaluating: ok?%v  result=%v where expr:%v", ok, whereValue.ToString(), where.StringAST())
		if !ok {
			u.Errorf("could not evaluate: %v", msg)
			return false
		}
		switch whereVal := whereValue.(type) {
		case value.BoolValue:
			if whereVal.Val() == false {
				u.Debugf("Filtering out: T:%T   v:%#v", whereVal, whereVal)
				return true
			}
		default:
			u.Warnf("unknown type? %T", whereVal)
		}
		//u.Debug("about to send from where to forward: %#v", msg)
		select {
		case out <- msg:
			return true
		case <-task.SigChan():
			return false
		}
	}
}

// Delete using a Where Expression
func (m *StaticDataSource) DeleteExpression(where expr.Node) (int, error) {
	//return 0, fmt.Errorf("not implemented")
	evaluator := vm.Evaluator(where)
	deletedKeys := make([]*Key, 0)
	m.bt.Ascend(func(a btree.Item) bool {
		di, ok := a.(*DriverItem)
		if !ok {
			u.Warnf("wat?  %T   %#v", a, a)
			return false
		}
		msgCtx := di.SqlDriverMessageMap
		whereValue, ok := evaluator(msgCtx)
		if !ok {
			u.Debugf("could not evaluate where: %v", msgCtx.Values())
			//return deletedCt, fmt.Errorf("Could not evaluate where clause")
			return true
		}
		switch whereVal := whereValue.(type) {
		case value.BoolValue:
			if whereVal.Val() == false {
				//this means do NOT delete
			} else {
				// Delete!
				indexVal := msgCtx.Values()[m.indexCol]
				deletedKeys = append(deletedKeys, NewKey(makeId(indexVal)))
			}
		case nil:
			// ??
		default:
			if whereVal.Nil() {
				// Doesn't match, so don't delete
			} else {
				u.Warnf("unknown type? %T", whereVal)
			}
		}
		return true
	})

	for _, deleteKey := range deletedKeys {
		//u.Debugf("calling delete: %v", deleteKey)
		if ct, err := m.Delete(deleteKey); err != nil {
			u.Errorf("Could not delete key: %v", deleteKey)
		} else if ct != 1 {
			u.Errorf("delete should have removed 1 key %v", deleteKey)
		}
	}
	return len(deletedKeys), nil
}

// Exec executes a query that doesn't return rows, such
// as an INSERT, UPDATE, DELETE
func (m *qlbStmt) Exec(args []driver.Value) (driver.Result, error) {
	var err error
	if len(args) > 0 {
		m.query, err = queryArgsConvert(m.query, args)
		if err != nil {
			return nil, err
		}
	}

	// Create a Job, which is Dag of Tasks that Run()
	ctx := plan.NewContext(m.query)
	ctx.Schema = m.conn.schema
	job, err := BuildSqlJob(ctx)
	if err != nil {
		return nil, err
	}
	m.job = job

	resultWriter := NewResultExecWriter(ctx)
	job.RootTask.Add(resultWriter)

	job.Setup()
	//u.Infof("in qlbdriver.Exec about to run")
	err = job.Run()
	//u.Debugf("After qlb driver.Run() in Exec()")
	if err != nil {
		u.Errorf("error on Query.Run(): %v", err)
		//resultWriter.ErrChan() <- err
		//job.Close()
	}
	return resultWriter.Result(), nil
}

// Open a go routine to run this source iteration until signal/complete
func SourceIterChannel(iter schema.Iterator, sigCh <-chan bool) <-chan schema.Message {

	out := make(chan schema.Message, 100)

	go func() {
		defer func() {
			if r := recover(); r != nil {
				u.Errorf("recover panic: %v", r)
			}
			// Can we safely close this?
			close(out)
		}()
		for item := iter.Next(); item != nil; item = iter.Next() {

			//u.Infof("In source Scanner iter %#v", item)
			select {
			case <-sigCh:
				u.Warnf("got signal quit")

				return
			case out <- item:
				// continue
			}
		}
	}()
	return out
}

// Run a Sql Job, by running to completion each task
func RunJob(conf *datasource.RuntimeSchema, tasks Tasks) error {

	//u.Debugf("in RunJob exec %v Recover?%v", len(tasks), conf.DisableRecover)
	ctx := new(Context)
	ctx.DisableRecover = conf.DisableRecover

	var wg sync.WaitGroup

	// start tasks in reverse order, so that by time
	// source starts up all downstreams have started
	for i := len(tasks) - 1; i >= 0; i-- {
		wg.Add(1)
		go func(taskId int) {
			if err := tasks[taskId].Run(ctx); err != nil {
				u.Errorf("%T.Run() errored %v", tasks[taskId], err)
				// TODO:  what do we do with this error?   send to error channel?
			}
			//u.Warnf("exiting taskId: %v %T", taskId, tasks[taskId])
			wg.Done()
		}(i)
	}

	wg.Wait()
	//u.Infof("RunJob(tasks) is completing")

	return nil
}