BitSliceIndexing: find min and max by bit plane, not by row

BitSliceIndexing/bsi.go

@@ -448,125 +448,60 @@ func compareValue(e *task, batch []uint32, resultsChan chan *roaring.Bitmap, wg
 	resultsChan <- results
 }
 
-// MinMax - Find minimum or maximum value.
+// MinMax - Find minimum or maximum value. The parallelism argument is
+// accepted for compatibility and no longer used; the search is driven by
+// the bit planes rather than by the rows, so there is nothing to divide.
 func (b *BSI) MinMax(parallelism int, op Operation, foundSet *roaring.Bitmap) int64 {
-
-	var n int = parallelism
-	if n == 0 {
-		n = runtime.NumCPU()
-	}
-
-	resultsChan := make(chan int64, n)
-
 	if foundSet == nil {
 		foundSet = b.eBM
 	}
-
-	card := foundSet.GetCardinality()
-	x := card / uint64(n)
-
-	remainder := card - (x * uint64(n))
-	var batch []uint32
-	var wg sync.WaitGroup
-	iter := foundSet.ManyIterator()
-	for i := 0; i < n; i++ {
-		if i == n-1 {
-			batch = make([]uint32, x+remainder)
+	candidates := roaring.And(foundSet, b.eBM)
+	if candidates.IsEmpty() {
+		if op == MAX {
+			return Min64BitSigned
+		}
+		return Max64BitSigned
+	}
+	return b.minMaxByPlanes(op, candidates)
+}
+
+// minMaxByPlanes narrows the candidate columns one bit plane at a time,
+// highest plane first, keeping only those that can still hold the answer.
+// A plane that would leave nothing is skipped, since every candidate then
+// agrees on that bit. One column survives, and only that one is decoded.
+func (b *BSI) minMaxByPlanes(op Operation, candidates *roaring.Bitmap) int64 {
+	j := b.BitCount() - 1
+	// The top plane is the sign bit only in a 64 bit wide BSI. A narrower
+	// one cannot represent a negative value, so every column is positive
+	// and the plain descent below is already correct.
+	if b.BitCount() == 64 {
+		var signed *roaring.Bitmap
+		if op == MIN {
+			signed = roaring.And(candidates, b.bA[j])
 		} else {
-			batch = make([]uint32, x)
+			signed = roaring.AndNot(candidates, b.bA[j])
 		}
-		iter.NextMany(batch)
-		wg.Add(1)
-		go b.minOrMax(op, batch, resultsChan, &wg)
-	}
-
-	wg.Wait()
-
-	close(resultsChan)
-	var minMax int64
-	if op == MAX {
-		minMax = Min64BitSigned
-	} else {
-		minMax = Max64BitSigned
-	}
-
-	for val := range resultsChan {
-		if (op == MAX && val > minMax) || (op == MIN && val < minMax) {
-			minMax = val
+		if !signed.IsEmpty() {
+			candidates = signed
 		}
+		j--
 	}
-	return minMax
-}
-
-func (b *BSI) minOrMax(op Operation, batch []uint32, resultsChan chan int64, wg *sync.WaitGroup) {
-
-	defer wg.Done()
-
-	x := b.BitCount()
-	var value int64 = Max64BitSigned
-	if op == MAX {
-		value = Min64BitSigned
-	}
-
-	for i := 0; i < len(batch); i++ {
-		cID := batch[i]
-		eq := true
-		lt, gt := false, false
-		j := b.BitCount() - 1
-		var cVal int64
-		valueIsNegative := uint64(value)&(1<<uint64(x-1)) > 0 && bits.Len64(uint64(value)) == 64
-		isNegative := false
-		if x == 64 {
-			isNegative = b.bA[j].Contains(cID)
-			if isNegative {
-				cVal |= 1 << uint64(j)
-			}
-			j--
-		}
-		compValue := value
-		if isNegative != valueIsNegative {
-			compValue = ^value + 1
-		}
-		for ; j >= 0; j-- {
-			sliceContainsBit := b.bA[j].Contains(cID)
-			if sliceContainsBit {
-				cVal |= 1 << uint64(j)
-			}
-			if uint64(compValue)&(1<<uint64(j)) > 0 {
-				// BIT in value is SET
-				if !sliceContainsBit {
-					if eq {
-						eq = false
-						if op == MAX && valueIsNegative && !isNegative {
-							gt = true
-							break
-						}
-						if op == MIN && (!valueIsNegative || (valueIsNegative == isNegative)) {
-							lt = true
-						}
-					}
-				}
-			} else {
-				// BIT in value is CLEAR
-				if sliceContainsBit {
-					if eq {
-						eq = false
-						if op == MIN && isNegative && !valueIsNegative {
-							lt = true
-						}
-						if op == MAX && (valueIsNegative || (valueIsNegative == isNegative)) {
-							gt = true
-						}
-					}
-				}
-			}
+	// Below the sign, two's complement orders the same way as unsigned, so
+	// the maximum keeps the columns with the bit set and the minimum keeps
+	// those without it.
+	for ; j >= 0; j-- {
+		var next *roaring.Bitmap
+		if op == MAX {
+			next = roaring.And(candidates, b.bA[j])
+		} else {
+			next = roaring.AndNot(candidates, b.bA[j])
 		}
-		if lt || gt {
-			value = cVal
+		if !next.IsEmpty() {
+			candidates = next
 		}
 	}
-
-	resultsChan <- value
+	value, _ := b.GetValue(uint64(candidates.Minimum()))
+	return value
 }
 
 // Sum all values contained within the foundSet.   As a convenience, the cardinality of the foundSet

BitSliceIndexing/bsi_minmax_benchmark_test.go

@@ -0,0 +1,38 @@
+package roaring
+
+import (
+	"fmt"
+	"testing"
+)
+
+var benchmarkMinMaxResult int64
+
+// BenchmarkMinMax covers both widths a BSI can take: a narrow one holding
+// only non-negative values, and the 64 bit wide one that a negative
+// declared minimum produces.
+func BenchmarkMinMax(b *testing.B) {
+	for _, shape := range []struct {
+		name               string
+		maxValue, minValue int64
+	}{
+		{"non-negative", 99, 0},
+		{"signed", 99, -1},
+	} {
+		for _, rows := range []int{100, 10000} {
+			bsi := NewBSI(shape.maxValue, shape.minValue)
+			for row := 0; row < rows; row++ {
+				value := int64(row % 99)
+				if shape.minValue < 0 {
+					value -= 50
+				}
+				bsi.SetValue(uint64(row), value)
+			}
+			b.Run(fmt.Sprintf("%s/planes%d/rows%d", shape.name, bsi.BitCount(), rows), func(b *testing.B) {
+				b.ReportAllocs()
+				for i := 0; i < b.N; i++ {
+					benchmarkMinMaxResult = bsi.MinMax(0, MAX, nil)
+				}
+			})
+		}
+	}
+}

BitSliceIndexing/bsi_test.go

@@ -647,3 +647,25 @@ func TestBatchEqualLargeQueryValues(t *testing.T) {
 		}
 	}
 }
+
+func TestMinMaxMixedSign(t *testing.T) {
+	// One negative and one positive value. The maximum is the positive one,
+	// and the minimum the negative one, whatever the parallelism argument.
+	bsi := NewBSI(99, -1)
+	bsi.SetValue(0, -45)
+	bsi.SetValue(1, 1)
+	for _, parallelism := range []int{0, 1, 3} {
+		assert.Equal(t, int64(1), bsi.MinMax(parallelism, MAX, nil))
+		assert.Equal(t, int64(-45), bsi.MinMax(parallelism, MIN, nil))
+	}
+
+	// Zero alongside negatives: zero is the maximum.
+	zeroAndNegatives := NewBSI(99, -1)
+	for column, value := range []int64{0, -46, -29} {
+		zeroAndNegatives.SetValue(uint64(column), value)
+	}
+	for _, parallelism := range []int{0, 1, 3} {
+		assert.Equal(t, int64(0), zeroAndNegatives.MinMax(parallelism, MAX, nil))
+		assert.Equal(t, int64(-46), zeroAndNegatives.MinMax(parallelism, MIN, nil))
+	}
+}