Merge pull request #473 from travisturner/inverse-calls

Determine if a Call is inverse or not to correctly map to slices.

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

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))

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])
+		}
+	})
+
+}