diff --git a/executor.go b/executor.go index d1fd8b829..3179770e3 100644 --- a/executor.go +++ b/executor.go @@ -56,17 +56,41 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic opt = &ExecOptions{} } + // Don't bother calculating slices for query types that don't require it. + needsSlices := needsSlices(q.Calls) + + // MaxSlice can differ between inverse and standard views, so we need + // to send queries to different slices based on orientation. + var inverseSlices []uint64 + rowLabel := DefaultRowLabel + columnLabel := DefaultColumnLabel + // If slices aren't specified, then include all of them. if len(slices) == 0 { - if needsSlices(q.Calls) { + // Determine slices and inverseSlices for use in e.executeCall(). + if needsSlices { // Round up the number of slices. maxSlice := e.Holder.Index(index).MaxSlice() + maxInverseSlice := e.Holder.Index(index).MaxInverseSlice() // Generate a slices of all slices. slices = make([]uint64, maxSlice+1) for i := range slices { slices[i] = uint64(i) } + + // Generate a slices of all inverse slices. + inverseSlices = make([]uint64, maxInverseSlice+1) + for i := range inverseSlices { + inverseSlices[i] = uint64(i) + } + + // Fetch column label from index. + idx := e.Holder.Index(index) + if idx == nil { + return nil, ErrIndexNotFound + } + columnLabel = idx.ColumnLabel() } } @@ -78,6 +102,25 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic // Execute each call serially. results := make([]interface{}, 0, len(q.Calls)) for _, call := range q.Calls { + + if call.SupportsInverse() && needsSlices { + // Fetch frame & row label based on argument. + frame, _ := call.Args["frame"].(string) + if frame == "" { + frame = DefaultFrame + } + f := e.Holder.Frame(index, frame) + if f == nil { + return nil, ErrFrameNotFound + } + rowLabel = f.RowLabel() + + // If this call is to an inverse frame send to a different list of slices. + if call.IsInverse(rowLabel, columnLabel) { + slices = inverseSlices + } + } + v, err := e.executeCall(ctx, index, call, slices, opt) if err != nil { return nil, err diff --git a/pql/ast.go b/pql/ast.go index 6673bb638..57c41960a 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -156,6 +156,30 @@ func (c *Call) String() string { return buf.String() } +// SupportsInverse indicates that the call may be on an inverse frame. +func (c *Call) SupportsInverse() bool { + if c.Name == "Bitmap" { + return true + } + return false +} + +// IsInverse specifies if the call is for an inverse view. +// Return defaults to false unless absolutely sure of inversion. +func (c *Call) IsInverse(rowLabel, columnLabel string) bool { + if c.SupportsInverse() { + _, rowOK, rowErr := c.UintArg(rowLabel) + _, columnOK, columnErr := c.UintArg(columnLabel) + if rowErr != nil || columnErr != nil { + return false + } + if !rowOK && columnOK { + return true + } + } + return false +} + // CopyArgs returns a copy of m. func CopyArgs(m map[string]interface{}) map[string]interface{} { other := make(map[string]interface{}, len(m)) diff --git a/pql/ast_test.go b/pql/ast_test.go index 653bd5af5..57215938c 100644 --- a/pql/ast_test.go +++ b/pql/ast_test.go @@ -15,3 +15,69 @@ func TestCall_String(t *testing.T) { } }) } + +// Ensure call can be converted into a string. +func TestCall_SupportsInverse(t *testing.T) { + t.Run("Bitmap", func(t *testing.T) { + q, err := pql.ParseString(`Bitmap()`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].SupportsInverse() != true { + t.Fatalf("call should support inverse: %s", q.Calls[0]) + } + }) + t.Run("Count Bitmap", func(t *testing.T) { + q, err := pql.ParseString(`Count(Bitmap())`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].SupportsInverse() == true { + t.Fatalf("call should not support inverse: %s", q.Calls[0]) + } + }) + t.Run("Union Bitmaps", func(t *testing.T) { + q, err := pql.ParseString(`Union(Bitmap(), Bitmap())`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].SupportsInverse() == true { + t.Fatalf("call should not support inverse: %s", q.Calls[0]) + } + }) + +} + +// Ensure call is correctly determined to be against an inverse view. +func TestCall_IsInverse(t *testing.T) { + t.Run("Bitmap Row", func(t *testing.T) { + q, err := pql.ParseString(`Bitmap(frame="f", row=1)`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].IsInverse("row", "col") != false { + t.Fatalf("incorrect call inverse: %s", q.Calls[0]) + } + }) + t.Run("Bitmap Column", func(t *testing.T) { + q, err := pql.ParseString(`Bitmap(frame="f", col=1)`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].IsInverse("row", "col") != true { + t.Fatalf("incorrect call inverse: %s", q.Calls[0]) + } + }) + t.Run("Bitmap Column No Label", func(t *testing.T) { + q, err := pql.ParseString(`Bitmap(frame="f", col=1)`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].IsInverse("rowX", "colX") != false { + t.Fatalf("incorrect call inverse: %s", q.Calls[0]) + } + }) + t.Run("Count", func(t *testing.T) { + q, err := pql.ParseString(`Count(Bitmap(frame="f", col=1))`) + if err != nil { + t.Fatal(err) + } else if q.Calls[0].IsInverse("row", "col") != false { + t.Fatalf("incorrect call inverse: %s", q.Calls[0]) + } + }) + +}