Misc bug fixes. Descriptions in commits.

datasource/context.go

@@ -209,6 +209,9 @@ func (m *ContextSimple) Commit(rowInfo []expr.SchemaInfo, row expr.RowWriter) er
 	return nil
 }
 func (m *ContextSimple) Delete(row map[string]value.Value) error {
+	for k, _ := range row {
+		delete(m.Data, k)
+	}
 	return nil
 }
 

datasource/files/filesource.go

@@ -10,7 +10,8 @@ import (
 	"time"
 
 	u "github.com/araddon/gou"
-	"github.com/dchest/siphash"
+
+	hash "github.com/aviddiviner/go-murmur"
 	"github.com/lytics/cloudstorage"
 	"golang.org/x/net/context"
 	"google.golang.org/api/iterator"
@@ -46,7 +47,7 @@ type FileReaderIterator interface {
 type Partitioner func(uint64, *FileInfo) int
 
 func SipPartitioner(partitionCt uint64, fi *FileInfo) int {
-	hashU64 := siphash.Hash(0, 1, []byte(fi.Name))
+    hashU64 := hash.MurmurHash64A([]byte(fi.Name), 1)
 	return int(hashU64 % partitionCt)
 }
 

datasource/membtree/btree.go

@@ -7,7 +7,7 @@ import (
 	"fmt"
 
 	u "github.com/araddon/gou"
-	"github.com/dchest/siphash"
+	hash "github.com/aviddiviner/go-murmur"
 	"github.com/google/btree"
 	"golang.org/x/net/context"
 
@@ -78,14 +78,14 @@ func makeId(dv driver.Value) uint64 {
 	case int64:
 		return uint64(vt)
 	case []byte:
-		return siphash.Hash(0, 1, vt)
+		return hash.MurmurHash64A(vt, 1)
 		// iv, err := strconv.ParseUint(string(vt), 10, 64)
 		// if err != nil {
 		// 	u.Warnf("could not create id: %v  for %v", err, dv)
 		// }
 		// return iv
 	case string:
-		return siphash.Hash(0, 1, []byte(vt))
+		return hash.MurmurHash64A([]byte(vt), 1)
 		// iv, err := strconv.ParseUint(vt, 10, 64)
 		// if err != nil {
 		// 	u.Warnf("could not create id: %v  for %v", err, dv)

datasource/memdb/index.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	u "github.com/araddon/gou"
-	"github.com/dchest/siphash"
+	hash "github.com/aviddiviner/go-murmur"
 	"github.com/hashicorp/go-memdb"
 
 	"github.com/araddon/qlbridge/datasource"
@@ -25,9 +25,9 @@ func makeId(dv driver.Value) uint64 {
 	case int64:
 		return uint64(vt)
 	case []byte:
-		return siphash.Hash(456729, 1111581582, vt)
+		return hash.MurmurHash64A(vt, 1111581582)
 	case string:
-		return siphash.Hash(456729, 1111581582, []byte(vt))
+		return hash.MurmurHash64A([]byte(vt), 1111581582)
 		//by := append(make([]byte,0,8), byte(r), byte(r>>8), byte(r>>16), byte(r>>24), byte(r>>32), byte(r>>40), byte(r>>48), byte(r>>56))
 	case datasource.KeyCol:
 		return makeId(vt.Val)

datasource/sqlite/conn.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 
 	u "github.com/araddon/gou"
-	"github.com/dchest/siphash"
+	"github.com/aviddiviner/go-murmur"
 	"github.com/google/btree"
 	"golang.org/x/net/context"
 	// Import driver for sqlite
@@ -323,9 +323,9 @@ func MakeId(dv driver.Value) uint64 {
 	case int64:
 		return uint64(vt)
 	case []byte:
-		return siphash.Hash(456729, 1111581582, vt)
+		return go-murmur.MurmurHash64A(vt, 1111581582)
 	case string:
-		return siphash.Hash(456729, 1111581582, []byte(vt))
+		return go-murmur.MurmurHash64A([]byte(vt), 1111581582)
 		//by := append(make([]byte,0,8), byte(r), byte(r>>8), byte(r>>16), byte(r>>24), byte(r>>32), byte(r>>40), byte(r>>48), byte(r>>56))
 	case datasource.KeyCol:
 		return MakeId(vt.Val)

exec/exec.go

@@ -8,11 +8,18 @@ package exec
 
 import (
 	"fmt"
-
+	"database/sql/driver"
 	"github.com/araddon/qlbridge/plan"
 	"github.com/araddon/qlbridge/schema"
 )
 
+const (
+    JOINMERGE_MAKER = "UseJoinMerge"
+    WHERE_MAKER = "UseWhere"
+    GROUPBY_MAKER = "UseGroupBy"
+    PROJECTION_MAKER = "UseProjection"
+)
+
 var (
 	// ErrShuttingDown already shutting down error
 	ErrShuttingDown = fmt.Errorf("Received Shutdown Signal")
@@ -105,6 +112,7 @@ type (
 		WalkHaving(p *plan.Having) (Task, error)
 		WalkGroupBy(p *plan.GroupBy) (Task, error)
 		WalkOrder(p *plan.Order) (Task, error)
+		WalkInto(p *plan.Into) (Task, error)
 		WalkProjection(p *plan.Projection) (Task, error)
 		// Other Statements
 		WalkCommand(p *plan.Command) (Task, error)
@@ -122,4 +130,18 @@ type (
 		// WalkExecSource given our plan, turn that into a Task.
 		WalkExecSource(p *plan.Source) (Task, error)
 	}
+
+	// SinkMaker Sink Factory
+	SinkMaker func(ctx *plan.Context, dest string, params map[string]interface{}) (Sink, error)
+
+	// Sinks are execution tasks used to direct query result set output to a destination.
+	Sink interface {
+		Open(ctx *plan.Context, destination string, params map[string]interface{}) error
+		Next(dest []driver.Value, colIndex map[string]int) error
+		Cleanup() error
+		Close() error
+	}
+
+	// JoinMergeMaker Factory
+	JoinMergeMaker func(ctx *plan.Context, l, r TaskRunner, p *plan.JoinMerge) TaskRunner
 )

exec/executor.go

@@ -65,17 +65,21 @@ func BuildSqlJobPlanned(planner plan.Planner, executor Executor, ctx *plan.Conte
 	if ctx.Raw == "" {
 		return nil, fmt.Errorf("no sql provided")
 	}
-	stmt, err := rel.ParseSql(ctx.Raw)
-	if err != nil {
-		u.Debugf("could not parse sql : %v", err)
-		return nil, err
-	}
-	if stmt == nil {
-		return nil, fmt.Errorf("Not statement for parse? %v", ctx.Raw)
+	var err error
+	var pln plan.Task
+	var stmt rel.SqlStatement
+	if ctx.Stmt == nil {  // Prepared statement
+		stmt, err = rel.ParseSql(ctx.Raw)
+		if err != nil {
+			u.Debugf("could not parse sql : %v", err)
+			return nil, err
+		}
+		if stmt == nil {
+			return nil, fmt.Errorf("Not statement for parse? %v", ctx.Raw)
+		}
+		ctx.Stmt = stmt
 	}
-	ctx.Stmt = stmt
-
-	pln, err := plan.WalkStmt(ctx, stmt, planner)
+	pln, err = plan.WalkStmt(ctx, ctx.Stmt, planner)
 
 	if err != nil {
 		return nil, err
@@ -197,6 +201,7 @@ func (m *JobExecutor) WalkSource(p *plan.Source) (Task, error) {
 	}
 	return NewSource(m.Ctx, p)
 }
+
 func (m *JobExecutor) WalkSourceExec(p *plan.Source) (Task, error) {
 
 	if p.Conn == nil {
@@ -219,22 +224,71 @@ func (m *JobExecutor) WalkSourceExec(p *plan.Source) (Task, error) {
 	return nil, fmt.Errorf("%T Must Implement Scanner for %q", p.Conn, p.Stmt.String())
 }
 func (m *JobExecutor) WalkWhere(p *plan.Where) (Task, error) {
-	return NewWhere(m.Ctx, p), nil
+
+    var tr TaskRunner
+	tr = NewWhere(m.Ctx, p)
+    if m.Ctx.Session != nil {
+        if v, ok := m.Ctx.Session.Get(WHERE_MAKER); ok {
+            //if factory, ok2 := v.Value().(JoinMergeMaker); !ok2 {
+            if factory, ok2 := v.Value().(func(ctx *plan.Context, p *plan.Where) TaskRunner); !ok2 {
+                return nil, fmt.Errorf("Cannot cast [%T] to WhereMaker factory.", v.Value)
+            } else {
+                tr = factory(m.Ctx, p)
+            }
+        }
+    }
+    return tr, nil
 }
 func (m *JobExecutor) WalkHaving(p *plan.Having) (Task, error) {
 	return NewHaving(m.Ctx, p), nil
 }
 func (m *JobExecutor) WalkGroupBy(p *plan.GroupBy) (Task, error) {
-	return NewGroupBy(m.Ctx, p), nil
+
+    var tr TaskRunner
+	tr = NewGroupBy(m.Ctx, p)
+    if m.Ctx.Session != nil {
+        if v, ok := m.Ctx.Session.Get(GROUPBY_MAKER); ok {
+            //if factory, ok2 := v.Value().(JoinMergeMaker); !ok2 {
+            if factory, ok2 := v.Value().(func(ctx *plan.Context, p *plan.GroupBy) TaskRunner); !ok2 {
+                return nil, fmt.Errorf("Cannot cast [%T] to GroupByMaker factory.", v.Value)
+            } else {
+                tr = factory(m.Ctx, p)
+            }
+        }
+    }
+    return tr, nil
 }
 func (m *JobExecutor) WalkOrder(p *plan.Order) (Task, error) {
 	return NewOrder(m.Ctx, p), nil
 }
+func (m *JobExecutor) WalkInto(p *plan.Into) (Task, error) {
+	return NewInto(m.Ctx, p), nil
+}
 func (m *JobExecutor) WalkProjection(p *plan.Projection) (Task, error) {
-	return NewProjection(m.Ctx, p), nil
+    var tr TaskRunner
+	tr = NewProjection(m.Ctx, p)
+    if m.Ctx.Session != nil {
+        if v, ok := m.Ctx.Session.Get(PROJECTION_MAKER); ok {
+            //if factory, ok2 := v.Value().(JoinMergeMaker); !ok2 {
+            if factory, ok2 := v.Value().(func(ctx *plan.Context, p *plan.Projection) TaskRunner); !ok2 {
+                return nil, fmt.Errorf("Cannot cast [%T] to ProjectionMaker factory.", v.Value)
+            } else {
+                tr = factory(m.Ctx, p)
+            }
+        }
+    }
+    return tr, nil
 }
 func (m *JobExecutor) WalkJoin(p *plan.JoinMerge) (Task, error) {
-	execTask := NewTaskParallel(m.Ctx)
+
+    // If the left task is already parallelized then must be a multi table join.  
+    // No need to parallelize subsequent join tasks.
+    var execTask TaskRunner
+    if p.Left.IsParallel() {
+	    execTask = NewTaskSequential(m.Ctx)
+    } else {
+	    execTask = NewTaskParallel(m.Ctx)
+    }
 	//u.Debugf("join.Left: %#v    \nright:%#v", p.Left, p.Right)
 	l, err := m.WalkPlanAll(p.Left)
 	if err != nil {
@@ -255,7 +309,21 @@ func (m *JobExecutor) WalkJoin(p *plan.JoinMerge) (Task, error) {
 		return nil, err
 	}
 
-	jm := NewJoinNaiveMerge(m.Ctx, l.(TaskRunner), r.(TaskRunner), p)
+
+    var jm TaskRunner
+	jm = NewJoinNaiveMerge(m.Ctx, l.(TaskRunner), r.(TaskRunner), p)
+    if m.Ctx.Session != nil {
+        if v, ok := m.Ctx.Session.Get(JOINMERGE_MAKER); ok {
+            //if factory, ok2 := v.Value().(JoinMergeMaker); !ok2 {
+            if factory, ok2 := v.Value().(func(ctx *plan.Context, l, r TaskRunner, 
+                    p *plan.JoinMerge) TaskRunner); !ok2 {
+                return nil, fmt.Errorf("Cannot cast [%T] to JoinMergeMaker factory.", v.Value)
+            } else {
+                jm = factory(m.Ctx, l.(TaskRunner), r.(TaskRunner), p)
+            }
+        }
+    }
+
 	err = execTask.Add(jm)
 	if err != nil {
 		return nil, err
@@ -273,7 +341,7 @@ func (m *JobExecutor) WalkPlanAll(p plan.Task) (Task, error) {
 	}
 	if len(p.Children()) > 0 {
 		dagRoot := m.NewTask(p)
-		//u.Debugf("sequential?%v  parallel?%v", p.IsSequential(), p.IsParallel())
+		//u.Debugf("%p sequential?%v  parallel?%v", p, p.IsSequential(), p.IsParallel())
 		err = dagRoot.Add(root)
 		if err != nil {
 			u.Errorf("Could not add root: %v", err)
@@ -296,6 +364,8 @@ func (m *JobExecutor) WalkPlanTask(p plan.Task) (Task, error) {
 		return m.Executor.WalkGroupBy(p)
 	case *plan.Order:
 		return m.Executor.WalkOrder(p)
+	case *plan.Into:
+		return m.Executor.WalkInto(p)
 	case *plan.Projection:
 		return m.Executor.WalkProjection(p)
 	case *plan.JoinMerge: