diff --git a/andnot_compact_test.go b/andnot_compact_test.go new file mode 100644 index 0000000..409f2fb --- /dev/null +++ b/andnot_compact_test.go @@ -0,0 +1,336 @@ +package sroar + +import ( + "math/rand" + "testing" +) + +// AndNot results whose bitmap containers drop below the array threshold must be +// down-converted: semantically identical, materially smaller. +func TestAndNotCompactsSparseResults(t *testing.T) { + // dense a: every container becomes a bitmap container + a := NewBitmap() + for x := uint64(0); x < 1_000_000; x++ { + a.Set(x) + } + // b removes all but ~500 scattered values per container + b := NewBitmap() + rng := rand.New(rand.NewSource(3)) + survivors := map[uint64]struct{}{} + for i := 0; i < 8000; i++ { + survivors[uint64(rng.Int63n(1_000_000))] = struct{}{} + } + for x := uint64(0); x < 1_000_000; x++ { + if _, ok := survivors[x]; !ok { + b.Set(x) + } + } + + res := AndNot(a, b) + + // semantic equality against per-element reference + for x := uint64(0); x < 1_000_000; x++ { + _, want := survivors[x] + if got := res.Contains(x); got != want { + t.Fatalf("mismatch at %d: got %v want %v", x, got, want) + } + } + if res.GetCardinality() != len(survivors) { + t.Fatalf("cardinality %d want %d", res.GetCardinality(), len(survivors)) + } + + // size: ~8k survivors as array containers should be a few bytes/value, far + // below the ~16 bitmap containers (8KB each) the uncompacted result kept. + gotBytes := len(res.ToBuffer()) + uncompacted := 16 * 8192 + if gotBytes >= uncompacted/4 { + t.Fatalf("result not compacted: %d bytes (uncompacted would be ~%d)", gotBytes, uncompacted) + } +} + +// The boundary: results at/above the threshold must stay bitmap containers and +// results just below must convert, both with exact semantics. +func TestAndNotCompactThreshold(t *testing.T) { + for _, keep := range []int{andNotCompactThreshold - 1, andNotCompactThreshold, 4200} { + a := NewBitmap() + for x := uint64(0); x < 65536; x++ { + a.Set(x) // one full bitmap container + } + b := NewBitmap() + for x := uint64(keep); x < 65536; x++ { + b.Set(x) // remove all but the first `keep` values + } + res := AndNot(a, b) + if got := res.GetCardinality(); got != keep { + t.Fatalf("keep=%d: cardinality %d", keep, got) + } + for x := uint64(0); x < 65536; x++ { + want := x < uint64(keep) + if res.Contains(x) != want { + t.Fatalf("keep=%d: mismatch at %d", keep, x) + } + } + } +} + +// Mixed container shapes through the two-pass sizing: matched array/bitmap pairs +// in all four combinations, unmatched containers, and containers emptied by the +// subtraction — all against a brute-force reference. +func TestAndNotTwoPassMixedShapes(t *testing.T) { + rng := rand.New(rand.NewSource(17)) + a, b := NewBitmap(), NewBitmap() + // container 0: a bitmap, b bitmap (partial overlap) + for x := uint64(0); x < 65536; x++ { + a.Set(x) + if x%3 != 0 { + b.Set(x) + } + } + // container 1: a array, b array + for i := 0; i < 900; i++ { + x := uint64(1<<16) + uint64(rng.Int63n(65536)) + a.Set(x) + if i%2 != 0 { + b.Set(x) + } + } + // container 2: a array unmatched in b + for i := 0; i < 500; i++ { + a.Set(uint64(2<<16) + uint64(rng.Int63n(65536))) + } + // container 3: a bitmap fully erased by b + for x := uint64(3 << 16); x < 4<<16; x++ { + a.Set(x) + b.Set(x) + } + // container 4: b-only (must not appear) + for i := 0; i < 100; i++ { + b.Set(uint64(4<<16) + uint64(rng.Int63n(65536))) + } + + want := NewBitmap() + for _, x := range a.ToArray() { + if !b.Contains(x) { + want.Set(x) + } + } + + res := AndNot(a, b) + if res.GetCardinality() != want.GetCardinality() { + t.Fatalf("cardinality %d want %d", res.GetCardinality(), want.GetCardinality()) + } + for _, x := range want.ToArray() { + if !res.Contains(x) { + t.Fatalf("missing %d", x) + } + } + for x := uint64(0); x < 5<<16; x += 7 { + if res.Contains(x) != want.Contains(x) { + t.Fatalf("mismatch at %d", x) + } + } +} + +// Every container offset in an AndNot result must stay a multiple of 4 uint16s: +// uint16To64SliceUnsafe builds *uint64 views over container payloads, so a +// misaligned compacted container would be UB (and a SIGBUS on strict-alignment +// targets) for every operation that later touches res. +func TestAndNotResultAlignment(t *testing.T) { + checkAligned := func(t *testing.T, res *Bitmap) { + t.Helper() + for i := 0; i < res.keys.numKeys(); i++ { + off := res.keys.val(i) + if off%4 != 0 { + t.Fatalf("container %d (key %d) at misaligned offset %d", i, res.keys.key(i), off) + } + c := res.getContainer(off) + if len(c)%4 != 0 { + t.Fatalf("container %d has unpadded size %d", i, len(c)) + } + } + } + + // the shape that put a compacted container ahead of a bitmap container + a, b := NewBitmap(), NewBitmap() + for x := uint64(0); x < 131072; x++ { + a.Set(x) + } + for x := uint64(100); x < 65536; x++ { + b.Set(x) + } + res := AndNot(a, b) + checkAligned(t, res) + // exercise the unsafe uint64 views over the result + if got := res.Maximum(); got != 131071 { + t.Fatalf("Maximum on compacted result: %d", got) + } + if !res.Intersects(a) { + t.Fatal("Intersects on compacted result") + } + res.AndNot(b) // in-place op over the compacted arena + + // fuzz: random container mixes, all results must stay aligned + rng := rand.New(rand.NewSource(7)) + for trial := 0; trial < 30; trial++ { + a, b := NewBitmap(), NewBitmap() + for c := 0; c < 6; c++ { + base := uint64(c) << 16 + an := rng.Intn(60000) + 1 + for i := 0; i < an; i++ { + a.Set(base + uint64(rng.Intn(65536))) + } + if rng.Intn(3) > 0 { + bn := rng.Intn(65000) + 1 + for i := 0; i < bn; i++ { + b.Set(base + uint64(rng.Intn(65536))) + } + } + } + res := AndNot(a, b) + checkAligned(t, res) + } +} + +// Sparse bitmap containers of a with no matching key in b must be compacted +// exactly like matched ones — the space win must not depend on b's key set. +func TestAndNotCompactsUnmatchedContainers(t *testing.T) { + a := NewBitmap() + for x := uint64(0); x < 3000; x++ { + a.Set(x) // key-0 container converts to bitmap + } + for x := uint64(500); x < 3000; x++ { + a.Remove(x) // 500 remain; containers never down-convert, stays bitmap + } + for x := uint64(1 << 16); x < 1<<16+5000; x++ { + a.Set(x) // second container so b can match something else + } + if off, ok := a.keys.getValue(0); !ok || a.getContainer(off)[indexType] != typeBitmap { + t.Fatal("fixture: key-0 container must be a sparse bitmap container") + } + b := NewBitmap() + for x := uint64(1 << 16); x < 1<<16+2500; x++ { + b.Set(x) // matches only container 1; container 0 is unmatched + } + res := AndNot(a, b) + off, found := res.keys.getValue(0) + if !found { + t.Fatal("unmatched container missing from result") + } + if c := res.getContainer(off); c[indexType] != typeArray { + t.Fatalf("unmatched sparse bitmap container not compacted (type %d, size %d)", c[indexType], len(c)) + } + for x := uint64(0); x < 500; x++ { + if !res.Contains(x) { + t.Fatalf("missing %d", x) + } + } + if res.Contains(600) { + t.Fatal("false positive") + } +} + +// Two-pass sizing must grow res exactly once: allocations per AndNot call stay +// constant regardless of container count (res struct + initial data + one +// reservation + scratch + pass-1 counts), instead of one doubling per +// ~container. +func TestAndNotAllocsBounded(t *testing.T) { + a := NewBitmap() + for x := uint64(0); x < 2_000_000; x++ { + a.Set(x) // ~30 full bitmap containers + } + b := NewBitmap() + for x := uint64(0); x < 2_000_000; x += 3 { + b.Set(x) // remove a third: results stay bitmap containers + } + allocs := testing.AllocsPerRun(10, func() { + res := AndNot(a, b) + if res.GetCardinality() == 0 { + t.Fatal("unexpected empty result") + } + }) + // NewBitmap (struct+data), one exact reservation, one 8KB scratch, the + // pass-1 counts, plus small constant slack. + if allocs > 8 { + t.Fatalf("AndNot allocations not bounded: %.0f per call", allocs) + } +} + +// Array-container results, matched and unmatched, are emitted at their exact +// compacted size: capacity slack from the source container's doubling growth +// must not survive into the result arena. +func TestAndNotShrinksArrayContainers(t *testing.T) { + const base = uint64(1) << 16 + for _, matched := range []bool{false, true} { + a := NewBitmap() + for x := base; x < base+2000; x++ { + a.Set(x) // array container grown by doubling + } + for x := base + 10; x < base+2000; x++ { + a.Remove(x) // 10 values remain in an oversized container + } + b := NewBitmap() + b.Set(5 << 16) + want := 10 + if matched { + b.Set(base + 5) + want = 9 + } + res := AndNot(a, b) + off, found := res.keys.getValue(base) + if !found { + t.Fatal("container missing") + } + c := res.getContainer(off) + if c[indexType] != typeArray || getCardinality(c) != want { + t.Fatalf("matched=%v: type %d cardinality %d", matched, c[indexType], getCardinality(c)) + } + if len(c) != compactedArraySize(want) { + t.Fatalf("matched=%v: container size %d, want %d", matched, len(c), compactedArraySize(want)) + } + } +} + +// Heavy-shrink AndNot: dense minuend, subtrahend that removes almost everything — +// the regime where exact sizing and compaction matter most. +func BenchmarkAndNotHeavyShrink(b *testing.B) { + a := NewBitmap() + for x := uint64(0); x < 2_000_000; x++ { + a.Set(x) + } + sub := NewBitmap() + rng := rand.New(rand.NewSource(9)) + for x := uint64(0); x < 2_000_000; x++ { + if rng.Intn(100) != 0 { // keep ~1% + sub.Set(x) + } + } + b.ReportAllocs() + b.ResetTimer() + var s int + for i := 0; i < b.N; i++ { + s += AndNot(a, sub).GetCardinality() + } + andNotSink = s +} + +var andNotSink int + +// Dense AndNot: results stay bitmap containers — the regime where the sizing +// pass short-circuits and pass 2 builds containers directly inside res. +func BenchmarkAndNotDense(b *testing.B) { + a := NewBitmap() + for x := uint64(0); x < 2_000_000; x++ { + a.Set(x) + } + sub := NewBitmap() + for x := uint64(0); x < 2_000_000; x += 3 { + sub.Set(x) + } + b.ReportAllocs() + b.ResetTimer() + var s int + for i := 0; i < b.N; i++ { + s += AndNot(a, sub).GetCardinality() + } + andNotSink = s +} diff --git a/bitmap_opt.go b/bitmap_opt.go index 450336b..879c9e1 100644 --- a/bitmap_opt.go +++ b/bitmap_opt.go @@ -210,6 +210,13 @@ func andContainersInRange(a, b *Bitmap, ai, aj int, optBuf []uint16, useGallopB } } +// andNotCompactThreshold is the result cardinality below which AndNot writes a +// surviving bitmap container out as an array container. Set below the 4096 +// encoding break-even on purpose: the conversion is only worth losing the O(1) +// bit-test probe when the array is small enough to stay cache-resident and the +// space win is large (8KB -> at most ~2KB). +const andNotCompactThreshold = 1024 + func AndNot(a, b *Bitmap) *Bitmap { res := NewBitmap() if a.IsEmpty() { @@ -219,69 +226,89 @@ func AndNot(a, b *Bitmap) *Bitmap { return a.Clone() } - buf := make([]uint16, maxContainerSize) - andNotContainers(a, b, res, buf) + andNotContainers(a, b, res) return res } -func andNotContainers(a, b, res *Bitmap, optBuf []uint16) { - ai, an := 0, a.keys.numKeys() - bi, bn := 0, b.keys.numKeys() - - akToAc := map[uint64][]uint16{} - sizeContainers := 0 - newKeys := 0 - - for ai < an && bi < bn { +// andNotWalk drives one ordered pass over a's keys, pairing each with b's +// matching container (bc == nil for unmatched keys) and skipping containers +// that are empty by their cardinality header. Both andNotContainers passes use +// it, so the sizing and materialization walks cannot disagree. +func andNotWalk(a, b *Bitmap, visit func(ak uint64, ac, bc []uint16)) { + an := a.keys.numKeys() + bn := b.keys.numKeys() + for ai, bi := 0, 0; ai < an; ai++ { ak := a.keys.key(ai) - bk := b.keys.key(bi) - if ak == bk { - off := a.keys.val(ai) - ac := a.getContainer(off) - off = b.keys.val(bi) - bc := b.getContainer(off) - if c := containerAndNotAlt(ac, bc, optBuf, 0); len(c) > 0 && getCardinality(c) > 0 { - // create a new container and update the key offset to this container. - offset := res.newContainerNoClr(uint16(len(c))) - copy(res.data[offset:], c) - res.setKey(ak, offset) - } - ai++ - bi++ - } else if ak < bk { - off := a.keys.val(ai) - ac := a.getContainer(off) - if getCardinality(ac) > 0 { - akToAc[ak] = ac - sizeContainers += len(ac) - newKeys++ - } - ai++ - } else { + for bi < bn && b.keys.key(bi) < ak { bi++ } - } - for ; ai < an; ai++ { - offset := a.keys.val(ai) - ac := a.getContainer(offset) - if getCardinality(ac) > 0 { - ak := a.keys.key(ai) - akToAc[ak] = ac - sizeContainers += len(ac) - newKeys++ + ac := a.getContainer(a.keys.val(ai)) + if getCardinality(ac) == 0 { + continue + } + var bc []uint16 + if bi < bn && b.keys.key(bi) == ak { + bc = b.getContainer(b.keys.val(bi)) + bi++ } + visit(ak, ac, bc) } +} - if sizeContainers > 0 { - res.expandConditionally(newKeys, sizeContainers) +// compactedArraySize is the arena footprint of an array container of +// cardinality n: header, values, one spare slot, padded to a multiple of 4 +// uint16s so every later container keeps the 8-byte alignment the +// uint16To64SliceUnsafe views rely on. A full 4096-value array drops the spare +// rather than exceed maxContainerSize. +func compactedArraySize(n int) int { + return min((int(startIdx)+n+1+3)&^3, maxContainerSize) +} - for ak, ac := range akToAc { - // create a new container and update the key offset to this container. - offset := res.newContainerNoClr(uint16(len(ac))) - copy(res.data[offset:], ac) - res.setKey(ak, offset) +// andNotResultSize returns the uint16s the result for source ac occupies in +// res at cardinality n. A bitmap source at/above the threshold stays a +// maxContainerSize bitmap, so andNotResultCard's clamp there is harmless. +func andNotResultSize(ac []uint16, n int) int { + if ac[indexType] == typeBitmap && n >= andNotCompactThreshold { + return maxContainerSize + } + return compactedArraySize(n) +} + +func andNotContainers(a, b, res *Bitmap) { + // Count each pair's result cardinality and total the container sizes, then + // grow res once. The counts are kept so the materialization below does not + // recompute them. + counts := make([]uint16, 0, a.keys.numKeys()) + newKeys, sizeContainers := 0, 0 + andNotWalk(a, b, func(ak uint64, ac, bc []uint16) { + n := andNotResultCard(ac, bc) + counts = append(counts, uint16(n)) + if n <= 0 { + return } + newKeys++ + sizeContainers += andNotResultSize(ac, n) + }) + if newKeys == 0 { + return } + res.expandConditionally(newKeys, sizeContainers) + + // Materialize each result: reserve a slot of the counted size and let + // containerAndNotAlt fill it. Sizing and filling both go through + // andNotResultSize, so the slot always fits exactly. + idx := 0 + andNotWalk(a, b, func(ak uint64, ac, bc []uint16) { + n := int(counts[idx]) + idx++ + if n == 0 { + return + } + sz := andNotResultSize(ac, n) + offset := res.newContainerNoClr(uint16(sz)) + containerAndNotAlt(ac, bc, res.data[offset:offset+uint64(sz)], 0) + res.setKey(ak, offset) + }) } // AndNot performs in-place AND-NOT of ra and bm (ra &^= bm). diff --git a/container.go b/container.go index 86e0321..98cb2a1 100644 --- a/container.go +++ b/container.go @@ -80,6 +80,14 @@ func setCardinality(data []uint16, c int) { } } +// setArrayHeader stamps c's header as an array container of the given +// cardinality, sized to c's full length. +func setArrayHeader(c []uint16, card int) { + c[indexSize] = uint16(len(c)) + c[indexType] = typeArray + setCardinality(c, card) +} + func zeroOutContainer(c []uint16) { c[indexCardinality] = 0 c[indexCardinality+1] = 0 @@ -237,9 +245,7 @@ func (c array) andArray(other array) []uint16 { // Truncate out to how many values were found. out = out[:startIdx+num+1] - out[indexType] = typeArray - out[indexSize] = uint16(len(out)) - setCardinality(out, int(num)) + setArrayHeader(out, int(num)) return out } @@ -254,9 +260,7 @@ func (c array) andNotArray(other array, buf []uint16) []uint16 { // Truncate out to how many values were found. out = out[:startIdx+num+1] - out[indexType] = typeArray - out[indexSize] = uint16(len(out)) - setCardinality(out, int(num)) + setArrayHeader(out, int(num)) return out } @@ -278,9 +282,7 @@ func (c array) orArray(other array, buf []uint16, runMode int) []uint16 { // The output would be of typeArray. out := buf[:int(startIdx)+max] num := union2by2(c.all(), other.all(), out[startIdx:]) - out[indexType] = typeArray - out[indexSize] = uint16(len(out)) - setCardinality(out, num) + setArrayHeader(out, num) return out } @@ -288,7 +290,6 @@ var tmp = make([]uint16, 8192) func (c array) andBitmap(other bitmap) []uint16 { out := make([]uint16, int(startIdx)+getCardinality(c)+2) // some extra space. - out[indexType] = typeArray pos := startIdx for _, x := range c.all() { @@ -298,8 +299,7 @@ func (c array) andBitmap(other bitmap) []uint16 { // Ensure we have at least one empty slot at the end. res := out[:pos+1] - res[indexSize] = uint16(len(res)) - setCardinality(res, int(pos-startIdx)) + setArrayHeader(res, int(pos-startIdx)) return res } @@ -588,18 +588,87 @@ func (b bitmap) orArray(other array, buf []uint16, runMode int) []uint16 { } func (b bitmap) all() []uint16 { - var res []uint16 - data := b[startIdx:] - for idx := uint16(0); idx < uint16(len(data)); idx++ { - x := data[idx] - // TODO: This could potentially be optimized. - for pos := uint16(0); pos < 16; pos++ { - if x&bitmapMask[pos] > 0 { - res = append(res, (idx<<4)|pos) + res := make([]uint16, getCardinality(b)) + return res[:b.intoArray(res)] +} + +// intoArray writes b's set values into out in ascending order and returns the +// number written. Stops when out is full, so a cardinality header that +// understates the popcount degrades deterministically rather than overrunning. +func (b bitmap) intoArray(out []uint16) int { + idx := 0 + for w, word := range b[startIdx:] { + if word == 0 { + continue + } + base := uint16(w) << 4 + // bitmapMask[pos] is 1<<(15-pos), so the smallest pos is the highest bit: + // LeadingZeros16 yields values in ascending order. + for word != 0 { + if idx == len(out) { + return idx } + pos := uint16(bits.LeadingZeros16(word)) + out[idx] = base | pos + idx++ + word &^= 1 << (15 - pos) } } - return res + return idx +} + +// andNotBitmapIntoArray writes the ascending survivors of (b &^ other) into out +// and returns the count, subtracting word by word with no intermediate bitmap. +func (b bitmap) andNotBitmapIntoArray(other bitmap, out []uint16) int { + odata := other[startIdx:] + idx := 0 + for w, word := range b[startIdx:] { + if word &^= odata[w]; word == 0 { + continue + } + base := uint16(w) << 4 + for word != 0 { + if idx == len(out) { + return idx + } + pos := uint16(bits.LeadingZeros16(word)) + out[idx] = base | pos + idx++ + word &^= 1 << (15 - pos) + } + } + return idx +} + +// andNotArrayIntoArray writes the ascending survivors of (b &^ other) into out +// and returns the count, enumerating b's bits while a two-pointer walks other's +// sorted values to skip the removed ones. +func (b bitmap) andNotArrayIntoArray(other array, out []uint16) int { + ovals := other.all() + idx, j := 0, 0 + for w, word := range b[startIdx:] { + if word == 0 { + continue + } + base := uint16(w) << 4 + for word != 0 { + if idx == len(out) { + return idx + } + pos := uint16(bits.LeadingZeros16(word)) + word &^= 1 << (15 - pos) + val := base | pos + for j < len(ovals) && ovals[j] < val { + j++ + } + if j < len(ovals) && ovals[j] == val { + continue // removed by the array subtrahend + } + out[idx] = val + idx++ + } + } + return idx } // TODO: It can be optimized. diff --git a/container_opt.go b/container_opt.go index 4de0e8e..9d74e57 100644 --- a/container_opt.go +++ b/container_opt.go @@ -14,6 +14,81 @@ func init() { setCardinality(emptyArrayContainer, 0) } +// andNotResultCard returns the cardinality of (ac &^ bc) without materializing +// or allocating, clamped at andNotCompactThreshold (above it the exact count is +// unused, as the source stays a maxContainerSize bitmap). A nil or empty bc +// leaves ac unchanged. +// +// Bitmap sources are counted from the cardinality header, not by recounting +// bits: bc removes at most bcCard values, so acCard-bcCard >= andNotCompactThreshold +// proves a dense result without any counting. +func andNotResultCard(ac, bc []uint16) int { + acCard := getCardinality(ac) + bcCard := 0 + if bc != nil { + bcCard = getCardinality(bc) + } + if bcCard == 0 { + // nothing to subtract: result is ac unchanged + if ac[indexType] == typeBitmap { + // clamp: a full bitmap's cardinality overflows the caller's uint16, + // and the exact value above the threshold is unused + return min(acCard, andNotCompactThreshold) + } + // an array result stays an array sized to this exact count, so no clamp + return acCard + } + at, bt := ac[indexType], bc[indexType] + switch { + case at == typeArray && bt == typeArray: + return acCard - intersection2by2Cardinality(array(ac).all(), array(bc).all()) + case at == typeArray && bt == typeBitmap: + n := acCard + for _, x := range array(ac).all() { + if bitmap(bc).has(x) { + n-- + } + } + return n + case at == typeBitmap && bt == typeArray: + if acCard-bcCard >= andNotCompactThreshold { + return andNotCompactThreshold // dense by headers alone + } + // Start from acCard and drop each array element present in ac. n only + // falls, so n-remaining is its floor; once that reaches the threshold + // the result is dense no matter the rest. + n := acCard + remaining := bcCard + for _, x := range array(bc).all() { + remaining-- + if bitmap(ac).has(x) { + n-- + } + if n-remaining >= andNotCompactThreshold { + return andNotCompactThreshold + } + } + // the final iteration's floor check was n >= threshold, so here n < threshold + return n + case at == typeBitmap && bt == typeBitmap: + if acCard-bcCard >= andNotCompactThreshold { + return andNotCompactThreshold // dense by headers alone + } + a64 := uint16To64SliceUnsafe(ac[startIdx:]) + b64 := uint16To64SliceUnsafe(bc[startIdx:]) + n := 0 + for i := range a64 { + n += bits.OnesCount64(a64[i] &^ b64[i]) + if n >= andNotCompactThreshold { + return andNotCompactThreshold // dense: exact count not needed + } + } + return n + default: + panic("andNotResultCard: We should not reach here") + } +} + func containerAndAlt(ac, bc []uint16, optBuf []uint16, runMode int) []uint16 { at := ac[indexType] bt := bc[indexType] @@ -340,195 +415,144 @@ func (b bitmap) intersectsBitmap(other bitmap) bool { return false } -func containerAndNotAlt(ac, bc []uint16, optBuf []uint16, runMode int) []uint16 { - at := ac[indexType] - bt := bc[indexType] - - if at == typeArray && bt == typeArray { - left := array(ac) - right := array(bc) - return left.andNotArrayAlt(right, optBuf, runMode) - } - if at == typeArray && bt == typeBitmap { - left := array(ac) - right := bitmap(bc) - return left.andNotBitmapAlt(right, optBuf, runMode) - } - if at == typeBitmap && bt == typeArray { - left := bitmap(ac) - right := array(bc) - return left.andNotArrayAlt(right, optBuf, runMode) - } - if at == typeBitmap && bt == typeBitmap { - left := bitmap(ac) - right := bitmap(bc) - return left.andNotBitmapAlt(right, optBuf, runMode) - } - panic("containerAnd: We should not reach here") -} - -func (c array) andNotArrayAlt(other array, optBuf []uint16, runMode int) []uint16 { - cnum := getCardinality(c) - onum := getCardinality(other) - - if cnum == 0 { - if runMode&runInline == 0 { - return emptyArrayContainer - } - // do nothing, array already empty - return nil - } - if onum == 0 { - if runMode&runInline == 0 { - return resizeArray(c, optBuf) - } - // do nothing, nothing to remove - return nil - } - - // merge - out := c +// containerAndNotAlt computes ac &^ bc. A nil bc means nothing to subtract. +// +// Inline (runMode&runInline): subtracts bc from ac in place. +// +// Non-inline: writes the result into dst, whose length selects the output form — +// a full bitmap when len(dst) == maxContainerSize, otherwise a compact array. +// The result is produced straight into dst with no workspace: an array source +// subtracts into the slot; a dense bitmap is copied in and subtracted in place; +// a sparse bitmap has its survivors enumerated into the slot. A compact array's +// spare tail is left uninitialized — array reads never look past the cardinality. +func containerAndNotAlt(ac, bc, dst []uint16, runMode int) []uint16 { if runMode&runInline == 0 { - out = optBuf - if out == nil { - out = make([]uint16, roundSize(startIdx+uint16(cnum))) + out := dst[startIdx:] + var w int + switch ac[indexType] { + case typeArray: + a := array(ac) + switch { + case bc == nil: + w = copy(out, a.all()) + case bc[indexType] == typeArray: + w = a.andNotArrayIntoArray(array(bc), out) + case bc[indexType] == typeBitmap: + w = a.andNotBitmapIntoArray(bitmap(bc), out) + default: + panic("containerAndNot: unknown bc container type") + } + case typeBitmap: + if len(dst) == maxContainerSize { + // dense result stays a bitmap: subtract in place inside the slot + copy(dst, ac) + containerAndNotAlt(dst, bc, nil, runInline) + return dst + } + // sparse result: enumerate the survivors into the slot as an array + b := bitmap(ac) + switch { + case bc == nil: + w = b.intoArray(out) + case bc[indexType] == typeArray: + w = b.andNotArrayIntoArray(array(bc), out) + case bc[indexType] == typeBitmap: + w = b.andNotBitmapIntoArray(bitmap(bc), out) + default: + panic("containerAndNot: unknown bc container type") + } + default: + panic("containerAndNot: unknown ac container type") } + setArrayHeader(dst, w) + return dst } - setc := c.all() - seto := other.all() - num := difference(setc, seto, out[startIdx:]) - lastIdx := startIdx + uint16(num) - if runMode&runInline == 0 { - return bufAsArray(out, lastIdx) + if bc == nil { + return nil // subtracting nothing leaves ac unchanged + } + at := ac[indexType] + bt := bc[indexType] + switch { + case at == typeArray && bt == typeArray: + array(ac).andNotArrayAlt(array(bc)) + case at == typeArray && bt == typeBitmap: + array(ac).andNotBitmapAlt(bitmap(bc)) + case at == typeBitmap && bt == typeArray: + bitmap(ac).andNotArrayAlt(array(bc)) + case at == typeBitmap && bt == typeBitmap: + bitmap(ac).andNotBitmapAlt(bitmap(bc)) + default: + panic("containerAndNot: We should not reach here") } - setCardinality(c, num) return nil } -func (c array) andNotBitmapAlt(other bitmap, optBuf []uint16, runMode int) []uint16 { - cnum := getCardinality(c) - onum := getCardinality(other) +// The four andNot*Alt methods subtract other from the receiver in place. An +// empty receiver or an empty other leaves the receiver untouched. - if cnum == 0 { - if runMode&runInline == 0 { - return emptyArrayContainer - } - // do nothing, array already empty - return nil - } - if onum == 0 { - if runMode&runInline == 0 { - return resizeArray(c, optBuf) - } - // do nothing, nothing to remove - return nil +func (c array) andNotArrayAlt(other array) { + if getCardinality(c) == 0 || getCardinality(other) == 0 { + return } + setCardinality(c, c.andNotArrayIntoArray(other, c[startIdx:])) +} - // merge - out := c - if runMode&runInline == 0 { - out = optBuf - if out == nil { - out = make([]uint16, roundSize(startIdx+uint16(cnum))) - } +func (c array) andNotBitmapAlt(other bitmap) { + if getCardinality(c) == 0 || getCardinality(other) == 0 { + return } + setCardinality(c, c.andNotBitmapIntoArray(other, c[startIdx:])) +} - lastIdx := startIdx +// andNotArrayIntoArray writes the ascending values of (c &^ other) into out, +// returning the count. Array minus array is exactly a set difference. +func (c array) andNotArrayIntoArray(other array, out []uint16) int { + return difference(c.all(), other.all(), out) +} + +// andNotBitmapIntoArray writes the ascending values of (c &^ other) into out +// and returns the count, keeping the values not present in other. +func (c array) andNotBitmapIntoArray(other bitmap, out []uint16) int { + w := 0 for _, x := range c.all() { if !other.has(x) { - out[lastIdx] = x - lastIdx++ + out[w] = x + w++ } } - - if runMode&runInline == 0 { - return bufAsArray(out, lastIdx) - } - setCardinality(c, int(lastIdx-startIdx)) - return nil + return w } -func (b bitmap) andNotArrayAlt(other array, optBuf []uint16, runMode int) []uint16 { +func (b bitmap) andNotArrayAlt(other array) { bnum := getCardinality(b) - onum := getCardinality(other) - - if bnum == 0 { - if runMode&runInline == 0 { - return emptyArrayContainer - } - // do nothing, array already empty - return nil - } - if onum == 0 { - if runMode&runInline == 0 { - return b - } - // do nothing, nothing to remove - return nil - } - - // merge - out := b - if runMode&runInline == 0 { - out = copyBitmap(b, optBuf) + if bnum == 0 || getCardinality(other) == 0 { + return } - delnum := 0 for _, x := range other.all() { idx := x >> 4 pos := x & 0xF - if has := out[startIdx+idx]&bitmapMask[pos] > 0; has { - out[startIdx+idx] ^= bitmapMask[pos] + if b[startIdx+idx]&bitmapMask[pos] > 0 { + b[startIdx+idx] ^= bitmapMask[pos] delnum++ } } - setCardinality(out, bnum-delnum) - - if runMode&runInline == 0 { - return out - } - return nil + setCardinality(b, bnum-delnum) } -func (b bitmap) andNotBitmapAlt(other bitmap, optBuf []uint16, runMode int) []uint16 { - bnum := getCardinality(b) - onum := getCardinality(other) - - if bnum == 0 { - if runMode&runInline == 0 { - return emptyArrayContainer - } - // do nothing, array already empty - return nil - } - if onum == 0 { - if runMode&runInline == 0 { - return b - } - // do nothing, nothing to remove - return nil - } - - // merge - out := b - if runMode&runInline == 0 { - out = copyBitmap(b, optBuf) +func (b bitmap) andNotBitmapAlt(other bitmap) { + if getCardinality(b) == 0 || getCardinality(other) == 0 { + return } - - dst64 := uint16To64SliceUnsafe(out[startIdx:]) + dst64 := uint16To64SliceUnsafe(b[startIdx:]) src64 := uint16To64SliceUnsafe(other[startIdx:]) var num int for i := range dst64 { dst64[i] &^= src64[i] num += bits.OnesCount64(dst64[i]) } - setCardinality(out, num) - - if runMode&runInline == 0 { - return out - } - return nil + setCardinality(b, num) } func containerOrAlt(ac, bc []uint16, buf []uint16, runMode int) []uint16 { @@ -804,9 +828,7 @@ func resizeArray(c array, out []uint16) []uint16 { } else { out = out[:size] } - out[indexType] = typeArray - out[indexSize] = uint16(len(out)) - setCardinality(out, cnum) + setArrayHeader(out, cnum) copy(out[startIdx:], c[startIdx:lastIdx]) return out } @@ -821,9 +843,7 @@ func copyBitmap(b bitmap, out []uint16) []uint16 { func bufAsArray(buf []uint16, lastIdx uint16) []uint16 { out := buf[:roundSize(lastIdx)] - out[indexType] = typeArray - out[indexSize] = uint16(len(out)) - setCardinality(out, int(lastIdx-startIdx)) + setArrayHeader(out, int(lastIdx-startIdx)) return out }