diff --git a/fb/BlockOffset.go b/fb/BlockOffset.go index 6ef437e25..c4db1d266 100644 --- a/fb/BlockOffset.go +++ b/fb/BlockOffset.go @@ -84,8 +84,72 @@ func (rcv *BlockOffset) MutateLen(n uint32) bool { return rcv._tab.MutateUint32Slot(8, n) } +// MinVersion is the smallest key version (timestamp) among the keys in this +// block. Stored in an appended flatbuffer field (vtable slot 3, offset 10) so +// that blocks written without it decode to 0. +func (rcv *BlockOffset) MinVersion() uint64 { + o := flatbuffers.UOffsetT(rcv._tab.Offset(10)) + if o != 0 { + return rcv._tab.GetUint64(o + rcv._tab.Pos) + } + return 0 +} + +func (rcv *BlockOffset) MutateMinVersion(n uint64) bool { + return rcv._tab.MutateUint64Slot(10, n) +} + +// MaxVersion is the largest key version (timestamp) among the keys in this +// block. Stored in an appended flatbuffer field (vtable slot 4, offset 12). +func (rcv *BlockOffset) MaxVersion() uint64 { + o := flatbuffers.UOffsetT(rcv._tab.Offset(12)) + if o != 0 { + return rcv._tab.GetUint64(o + rcv._tab.Pos) + } + return 0 +} + +func (rcv *BlockOffset) MutateMaxVersion(n uint64) bool { + return rcv._tab.MutateUint64Slot(12, n) +} + +// VersionRangePresent reports whether this block carries a [min,max] version +// range (i.e. it was produced by version-range-aware code). A block written by +// older code returns false and must never be pruned. +func (rcv *BlockOffset) VersionRangePresent() bool { + return rcv._tab.Offset(10) != 0 && rcv._tab.Offset(12) != 0 +} + +// UncompressedLen is the uncompressed byte length of this block. Stored in an +// appended flatbuffer field (vtable slot 5, offset 14); 0 when absent. +func (rcv *BlockOffset) UncompressedLen() uint32 { + o := flatbuffers.UOffsetT(rcv._tab.Offset(14)) + if o != 0 { + return rcv._tab.GetUint32(o + rcv._tab.Pos) + } + return 0 +} + +func (rcv *BlockOffset) MutateUncompressedLen(n uint32) bool { + return rcv._tab.MutateUint32Slot(14, n) +} + +// KeyCount is the number of entries in this block. Stored in an appended +// flatbuffer field (vtable slot 6, offset 16); 0 when absent. +func (rcv *BlockOffset) KeyCount() uint32 { + o := flatbuffers.UOffsetT(rcv._tab.Offset(16)) + if o != 0 { + return rcv._tab.GetUint32(o + rcv._tab.Pos) + } + return 0 +} + +func (rcv *BlockOffset) MutateKeyCount(n uint32) bool { + return rcv._tab.MutateUint32Slot(16, n) +} + func BlockOffsetStart(builder *flatbuffers.Builder) { - builder.StartObject(3) + builder.StartObject(7) } func BlockOffsetAddKey(builder *flatbuffers.Builder, key flatbuffers.UOffsetT) { builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(key), 0) @@ -99,6 +163,18 @@ func BlockOffsetAddOffset(builder *flatbuffers.Builder, offset uint32) { func BlockOffsetAddLen(builder *flatbuffers.Builder, len uint32) { builder.PrependUint32Slot(2, len, 0) } +func BlockOffsetAddMinVersion(builder *flatbuffers.Builder, minVersion uint64) { + builder.PrependUint64Slot(3, minVersion, 0) +} +func BlockOffsetAddMaxVersion(builder *flatbuffers.Builder, maxVersion uint64) { + builder.PrependUint64Slot(4, maxVersion, 0) +} +func BlockOffsetAddUncompressedLen(builder *flatbuffers.Builder, uncompressedLen uint32) { + builder.PrependUint32Slot(5, uncompressedLen, 0) +} +func BlockOffsetAddKeyCount(builder *flatbuffers.Builder, keyCount uint32) { + builder.PrependUint32Slot(6, keyCount, 0) +} func BlockOffsetEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT { return builder.EndObject() } diff --git a/fb/TableIndex.go b/fb/TableIndex.go index 7b4074b57..4d67a1baf 100644 --- a/fb/TableIndex.go +++ b/fb/TableIndex.go @@ -140,8 +140,30 @@ func (rcv *TableIndex) MutateStaleDataSize(n uint32) bool { return rcv._tab.MutateUint32Slot(16, n) } +// MinVersion is the smallest key version (timestamp) across all keys in this +// table. It is stored in an appended flatbuffer field (vtable slot 7) so that +// tables written without it (or by older code) decode to 0 here. Use +// MinVersionPresent to distinguish a genuine 0 from an absent field. +func (rcv *TableIndex) MinVersion() uint64 { + o := flatbuffers.UOffsetT(rcv._tab.Offset(18)) + if o != 0 { + return rcv._tab.GetUint64(o + rcv._tab.Pos) + } + return 0 +} + +// MinVersionPresent reports whether the min_version field was written to this +// table index (i.e. it was produced by version-range-aware code). +func (rcv *TableIndex) MinVersionPresent() bool { + return rcv._tab.Offset(18) != 0 +} + +func (rcv *TableIndex) MutateMinVersion(n uint64) bool { + return rcv._tab.MutateUint64Slot(18, n) +} + func TableIndexStart(builder *flatbuffers.Builder) { - builder.StartObject(7) + builder.StartObject(8) } func TableIndexAddOffsets(builder *flatbuffers.Builder, offsets flatbuffers.UOffsetT) { builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(offsets), 0) @@ -170,6 +192,9 @@ func TableIndexAddOnDiskSize(builder *flatbuffers.Builder, onDiskSize uint32) { func TableIndexAddStaleDataSize(builder *flatbuffers.Builder, staleDataSize uint32) { builder.PrependUint32Slot(6, staleDataSize, 0) } +func TableIndexAddMinVersion(builder *flatbuffers.Builder, minVersion uint64) { + builder.PrependUint64Slot(7, minVersion, 0) +} func TableIndexEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT { return builder.EndObject() } diff --git a/fb/flatbuffer.fbs b/fb/flatbuffer.fbs index dce714b57..31b6b9fbc 100644 --- a/fb/flatbuffer.fbs +++ b/fb/flatbuffer.fbs @@ -13,12 +13,26 @@ table TableIndex { uncompressed_size:uint32; on_disk_size:uint32; stale_data_size:uint32; + // min_version is the smallest version (timestamp) across all keys in this + // table. Appended at the end for forward/backward compatibility: old readers + // ignore it, and new readers see the default (0) and treat the table range as + // unknown when reading tables written without it. Presence is detected via the + // vtable (Offset(18) != 0), so a genuine min_version of 0 is distinguishable + // from an absent field only when paired with a non-zero max_version. + min_version:uint64; } table BlockOffset { key:[ubyte]; offset:uint; len:uint; + // Appended fields for per-block pruning and exact decompress sizing. Forward/ + // backward compatible: old readers ignore them, new readers see 0 when reading + // tables written without them. + min_version:uint64; + max_version:uint64; + uncompressed_len:uint32; + key_count:uint32; } root_type TableIndex; diff --git a/iterator.go b/iterator.go index f57cfa4c9..fca61bfef 100644 --- a/iterator.go +++ b/iterator.go @@ -318,6 +318,43 @@ type IteratorOptions struct { prefixIsKey bool // If set, use the prefix for bloom filter lookup. Prefix []byte // Only iterate over this given prefix. SinceTs uint64 // Only read data that has version > SinceTs. + // UntilTs, when non-zero, is an inclusive upper bound on versions: only read + // data that has version <= UntilTs. Together with SinceTs it defines a + // version window (SinceTs, UntilTs]. It is a performance/scoping hint that + // lets storage skip whole blocks/SSTables outside the window; the per-entry + // version filter remains authoritative. Zero means no upper bound. + UntilTs uint64 +} + +// versionWindow returns the inclusive [lower, upper] version range that this +// iterator can possibly surface, and whether storage-level skipping should be +// applied. readTs is the transaction read timestamp (the natural upper bound). +// +// lower MUST equal the smallest version the authoritative per-entry filter would +// KEEP, so that pruning never drops in-window data. That filter only excludes +// v <= SinceTs when SinceTs > 0; when SinceTs == 0 it excludes nothing below, so +// version 0 is in-window for an UntilTs-only scan. Hence: +// - SinceTs == 0 -> lower = 0 (version 0 is kept; its block must not be pruned) +// - 0 < SinceTs < MaxUint64 -> lower = SinceTs + 1 +// - SinceTs == MaxUint64 -> empty window (the per-entry filter keeps nothing) +func (opt *IteratorOptions) versionWindow(readTs uint64) (lower, upper uint64, enabled bool) { + enabled = opt.SinceTs > 0 || opt.UntilTs != 0 + if opt.SinceTs == math.MaxUint64 { + // The per-entry filter (v > SinceTs) keeps nothing. Make the window empty + // (lower=MaxUint64, upper=0) so max SinceTs. + } + upper = readTs + if opt.UntilTs != 0 && opt.UntilTs < upper { + upper = opt.UntilTs + } + return lower, upper, enabled } func (opt *IteratorOptions) compareToPrefix(key []byte) int { @@ -334,6 +371,17 @@ func (opt *IteratorOptions) pickTable(t table.TableInterface) bool { if t.MaxVersion() < opt.SinceTs { return false } + // Ignore this table if its [min,max] version range provably falls entirely + // above the requested upper bound. Tables without version-range metadata + // return false here and are never skipped. The window bounds must match the + // authoritative per-entry filter (see versionWindow); in particular lower is + // 0 when SinceTs==0 so version-0 data is never pruned. + if opt.UntilTs != 0 { + lower, upper, _ := opt.versionWindow(math.MaxUint64) + if t.OutsideVersionRange(lower, upper) { + return false + } + } if len(opt.Prefix) == 0 { return true } @@ -355,12 +403,19 @@ func (opt *IteratorOptions) pickTable(t table.TableInterface) bool { // that the tables are sorted in the right order. func (opt *IteratorOptions) pickTables(all []*table.Table) []*table.Table { filterTables := func(tables []*table.Table) []*table.Table { - if opt.SinceTs > 0 { + if opt.SinceTs > 0 || opt.UntilTs != 0 { + // Window bounds aligned with the authoritative per-entry filter; lower + // is 0 when SinceTs==0 so version-0 data is never pruned. + lower, upper, _ := opt.versionWindow(math.MaxUint64) tmp := tables[:0] for _, t := range tables { if t.MaxVersion() < opt.SinceTs { continue } + // Drop tables whose version range falls entirely above the upper bound. + if opt.UntilTs != 0 && t.OutsideVersionRange(lower, upper) { + continue + } tmp = append(tmp, t) } tables = tmp @@ -625,8 +680,10 @@ func (it *Iterator) parseItem() bool { // Skip any versions which are beyond the readTs. version := y.ParseTs(key) - // Ignore everything that is above the readTs and below or at the sinceTs. - if version > it.readTs || (it.opt.SinceTs > 0 && version <= it.opt.SinceTs) { + // Ignore everything that is above the readTs and below or at the sinceTs, + // and (when set) anything strictly above the UntilTs upper bound. + if version > it.readTs || (it.opt.SinceTs > 0 && version <= it.opt.SinceTs) || + (it.opt.UntilTs != 0 && version > it.opt.UntilTs) { mi.Next() return false } @@ -684,7 +741,15 @@ FILL: // Reverse direction. nextTs := y.ParseTs(mi.Key()) mik := y.ParseKey(mi.Key()) - if nextTs <= it.readTs && bytes.Equal(mik, item.key) { + // A newer version is only a valid candidate if it is still inside the requested + // version window: <= readTs and, when UntilTs is set, <= UntilTs. Without the + // UntilTs check the walk-up climbs past the upper bound and returns a version + // > UntilTs (the forward path drops those at the per-entry filter above), so a + // reverse non-AllVersions scan would return a different version than the forward + // scan. SinceTs needs no re-check: candidates climb in increasing ts, so any + // version newer than the in-window entry we started from is also > SinceTs. + if nextTs <= it.readTs && (it.opt.UntilTs == 0 || nextTs <= it.opt.UntilTs) && + bytes.Equal(mik, item.key) { // This is a valid potential candidate. goto FILL } diff --git a/iterator_test.go b/iterator_test.go index 1b36cf660..6f4e16c91 100644 --- a/iterator_test.go +++ b/iterator_test.go @@ -31,6 +31,10 @@ func (tm *tableMock) Biggest() []byte { return tm.right } func (tm *tableMock) DoesNotHave(hash uint32) bool { return false } func (tm *tableMock) MaxVersion() uint64 { return math.MaxUint64 } +// OutsideVersionRange: the mock has no version-range metadata, so it is never +// skipped (matches the behavior of older real tables). +func (tm *tableMock) OutsideVersionRange(lower, upper uint64) bool { return false } + func TestPickTables(t *testing.T) { opt := DefaultIteratorOptions @@ -150,6 +154,265 @@ func TestIterateSinceTs(t *testing.T) { }) } +// TestIterateVersionWindow asserts that a version-bounded scan using UntilTs (an +// upper bound complementing SinceTs) returns IDENTICAL results to an unbounded +// AllVersions scan filtered to the same window, across multi-version data +// including deletes, after the data has been flushed to SSTables. This exercises +// both the table-level and block-level version-range skipping on the read path. +func TestIterateVersionWindow(t *testing.T) { + dir, err := os.MkdirTemp("", "badger-test") + require.NoError(t, err) + defer removeDir(dir) + + opts := getTestOptions(dir) + db, err := OpenManaged(opts) + require.NoError(t, err) + + const nkeys = 200 + bkey := func(i int) []byte { return []byte(fmt.Sprintf("key%05d", i)) } + + // Write three version bands far apart so whole tables/blocks fall outside + // typical windows. Some keys get deleted at a later version. + type rec struct { + key string + version uint64 + deleted bool + } + var written []rec + bands := []uint64{10, 1000, 100000} + for _, base := range bands { + for i := 0; i < nkeys; i++ { + ver := base + uint64(i%50) + txn := db.NewTransactionAt(ver-1, true) + if i%7 == 0 { + require.NoError(t, txn.Delete(bkey(i))) + written = append(written, rec{key: string(bkey(i)), version: ver, deleted: true}) + } else { + require.NoError(t, txn.Set(bkey(i), []byte(fmt.Sprintf("v%d", ver)))) + written = append(written, rec{key: string(bkey(i)), version: ver}) + } + require.NoError(t, txn.CommitAt(ver, nil)) + } + } + // Flatten to push everything into SSTables so the storage-level skip applies. + require.NoError(t, db.Flatten(2)) + + // reference returns the set of (key,version) pairs (incl. deletes) within the + // window, matching what an AllVersions+InternalAccess iterator should surface. + reference := func(lower, upper uint64) map[string]bool { + m := map[string]bool{} + for _, r := range written { + if r.version > lower && r.version <= upper { + m[fmt.Sprintf("%s@%d", r.key, r.version)] = true + } + } + return m + } + + readWindow := func(sinceTs, untilTs, readTs uint64) map[string]bool { + iopt := DefaultIteratorOptions + iopt.AllVersions = true + iopt.SinceTs = sinceTs + iopt.UntilTs = untilTs + got := map[string]bool{} + txn := db.NewTransactionAt(readTs, false) + defer txn.Discard() + it := txn.NewIterator(iopt) + defer it.Close() + for it.Rewind(); it.Valid(); it.Next() { + item := it.Item() + got[fmt.Sprintf("%s@%d", string(item.KeyCopy(nil)), item.Version())] = true + } + return got + } + + readTs := db.MaxVersion() + windows := [][2]uint64{ + {0, math.MaxUint64}, // full + {0, 500}, // only band 1 + {900, 1100}, // only band 2 + {50000, 200000}, // only band 3 + {500, 900}, // gap => empty + {0, 1049}, // bands 1 and 2 + } + for _, w := range windows { + sinceTs, untilTs := w[0], w[1] + t.Run(fmt.Sprintf("window_%d_%d", sinceTs, untilTs), func(t *testing.T) { + want := reference(sinceTs, untilTs) + got := readWindow(sinceTs, untilTs, readTs) + require.Equal(t, want, got, "bounded window scan must match manual filter") + }) + } + + require.NoError(t, db.Close()) +} + +// TestUntilTsReverseReturnsInWindowVersion pins the reverse-iteration upper-bound +// bug: a non-AllVersions scan must return the newest version <= UntilTs for a key, +// identically in forward and reverse. Versions 3,4,7 with UntilTs=4 must yield v4; +// before the fix the reverse walk-up climbed past UntilTs and returned v7. The +// existing window test uses AllVersions=true, which returns before the reverse +// walk-up, so it cannot catch this. +func TestUntilTsReverseReturnsInWindowVersion(t *testing.T) { + dir, err := os.MkdirTemp("", "badger-test") + require.NoError(t, err) + defer removeDir(dir) + + db, err := OpenManaged(getTestOptions(dir)) + require.NoError(t, err) + defer db.Close() + + key := []byte("k") + for _, v := range []uint64{3, 4, 7} { + txn := db.NewTransactionAt(v-1, true) + require.NoError(t, txn.Set(key, []byte(fmt.Sprintf("v%d", v)))) + require.NoError(t, txn.CommitAt(v, nil)) + } + require.NoError(t, db.Flatten(2)) + + scan := func(reverse bool) uint64 { + iopt := DefaultIteratorOptions + iopt.Reverse = reverse + iopt.UntilTs = 4 // window (0, 4]; newest in-window version is 4 + txn := db.NewTransactionAt(10, false) + defer txn.Discard() + it := txn.NewIterator(iopt) + defer it.Close() + var got uint64 + n := 0 + for it.Rewind(); it.Valid(); it.Next() { + got = it.Item().Version() + n++ + } + require.Equalf(t, 1, n, "reverse=%v: expected exactly one key", reverse) + return got + } + + require.Equal(t, uint64(4), scan(false), "forward: newest version <= UntilTs") + require.Equal(t, uint64(4), scan(true), "reverse: newest version <= UntilTs (must match forward)") +} + +// TestUntilTsLowerBoundKeepsSmallestVersion is an end-to-end regression test for +// the version-window lower-bound bug. badger rejects CommitTs==0, so the smallest +// storable version is 1; we write a key at version 1, flush it to an SSTable, and +// run an UntilTs-only scan (SinceTs=0, UntilTs=1). With the lower bound correctly +// set to 0 (not SinceTs+1=1), the entry's block/table is not pruned and the key +// is returned. (The old code used lower=1, which is fine for a v=1 entry but wrong +// for a v=0 one; this test pins the boundary for the smallest publicly storable +// version, and the unit/table tests cover v=0 directly.) +func TestUntilTsLowerBoundKeepsSmallestVersion(t *testing.T) { + dir, err := os.MkdirTemp("", "badger-test") + require.NoError(t, err) + defer removeDir(dir) + + db, err := OpenManaged(getTestOptions(dir)) + require.NoError(t, err) + + txn := db.NewTransactionAt(0, true) + require.NoError(t, txn.Set([]byte("k"), []byte("v1"))) + require.NoError(t, txn.CommitAt(1, nil)) + require.NoError(t, db.Flatten(1)) + + iopt := DefaultIteratorOptions + iopt.AllVersions = true + iopt.SinceTs = 0 + iopt.UntilTs = 1 + + rtxn := db.NewTransactionAt(math.MaxUint64, false) + defer rtxn.Discard() + it := rtxn.NewIterator(iopt) + defer it.Close() + + var keys []string + for it.Rewind(); it.Valid(); it.Next() { + item := it.Item() + keys = append(keys, fmt.Sprintf("%s@%d", string(item.KeyCopy(nil)), item.Version())) + } + require.Equal(t, []string{"k@1"}, keys, "smallest-version entry must survive an UntilTs-only scan") + require.NoError(t, db.Close()) +} + +// TestVersionWindowLowerBound guards the boundary alignment between the storage +// prune window and the authoritative per-entry filter. The per-entry filter only +// excludes v <= SinceTs when SinceTs > 0, so for an UntilTs-only scan version 0 +// is in-window and lower MUST be 0 (not 1) or a maxVersion==0 block gets wrongly +// pruned. +func TestVersionWindowLowerBound(t *testing.T) { + cases := []struct { + sinceTs, untilTs, readTs uint64 + wantLower, wantUpper uint64 + wantEnabled bool + }{ + // UntilTs-only scan: lower must be 0 so version-0 data stays in-window. + {sinceTs: 0, untilTs: 100, readTs: math.MaxUint64, wantLower: 0, wantUpper: 100, wantEnabled: true}, + // SinceTs set: lower excludes v <= SinceTs. + {sinceTs: 10, untilTs: 0, readTs: math.MaxUint64, wantLower: 11, wantUpper: math.MaxUint64, wantEnabled: true}, + {sinceTs: 10, untilTs: 100, readTs: math.MaxUint64, wantLower: 11, wantUpper: 100, wantEnabled: true}, + // readTs caps the upper bound when smaller than UntilTs. + {sinceTs: 0, untilTs: 100, readTs: 50, wantLower: 0, wantUpper: 50, wantEnabled: true}, + // No bounds: disabled. + {sinceTs: 0, untilTs: 0, readTs: math.MaxUint64, wantLower: 0, wantUpper: math.MaxUint64, wantEnabled: false}, + // SinceTs at the max: empty window, no overflow. + {sinceTs: math.MaxUint64, untilTs: 0, readTs: math.MaxUint64, wantLower: math.MaxUint64, wantUpper: 0, wantEnabled: true}, + } + for _, c := range cases { + opt := IteratorOptions{SinceTs: c.sinceTs, UntilTs: c.untilTs} + lower, upper, enabled := opt.versionWindow(c.readTs) + require.Equal(t, c.wantLower, lower, "lower for since=%d until=%d", c.sinceTs, c.untilTs) + require.Equal(t, c.wantUpper, upper, "upper for since=%d until=%d", c.sinceTs, c.untilTs) + require.Equal(t, c.wantEnabled, enabled, "enabled for since=%d until=%d", c.sinceTs, c.untilTs) + } +} + +// TestUntilTsKeepsVersionZeroTable asserts the table picker does NOT prune a +// table whose entire version range is [0,0] on an UntilTs-only scan: version 0 +// is in-window because SinceTs==0. +func TestUntilTsKeepsVersionZeroTable(t *testing.T) { + opt := DefaultIteratorOptions + opt.SinceTs = 0 + opt.UntilTs = 100 + zeroTable := &prunableTableMock{minV: 0, maxV: 0, hasRange: true} + require.True(t, opt.pickTable(zeroTable), + "version-0 table must NOT be pruned on an UntilTs-only scan") +} + +// TestUntilTsPrunesTables asserts that the table picker actually drops SSTables +// whose version range lies entirely above the UntilTs upper bound. +func TestUntilTsPrunesTables(t *testing.T) { + mk := func(minV, maxV uint64, hasRange bool) *prunableTableMock { + return &prunableTableMock{minV: minV, maxV: maxV, hasRange: hasRange} + } + opt := DefaultIteratorOptions + opt.UntilTs = 100 + + // Table entirely above the window => pruned. + require.False(t, opt.pickTable(mk(500, 600, true))) + // Table intersecting the window => kept. + require.True(t, opt.pickTable(mk(50, 120, true))) + // Table below window (still has versions <= 100) => kept. + require.True(t, opt.pickTable(mk(1, 50, true))) + // Old-format table without a range => never pruned, even if it "looks" above. + require.True(t, opt.pickTable(mk(500, 600, false))) +} + +// prunableTableMock implements table.TableInterface with a controllable version +// range, for testing the UntilTs table-pruning predicate. +type prunableTableMock struct { + minV, maxV uint64 + hasRange bool +} + +func (m *prunableTableMock) Smallest() []byte { return y.KeyWithTs([]byte("a"), 1) } +func (m *prunableTableMock) Biggest() []byte { return y.KeyWithTs([]byte("z"), 1) } +func (m *prunableTableMock) DoesNotHave(hash uint32) bool { return false } +func (m *prunableTableMock) MaxVersion() uint64 { return m.maxV } +func (m *prunableTableMock) OutsideVersionRange(lower, upper uint64) bool { + if !m.hasRange { + return false + } + return m.maxV < lower || m.minV > upper +} + func TestIterateSinceTsWithPendingWrites(t *testing.T) { // The pending entries still have version=0. Even IteratorOptions.SinceTs is 0, the entries // should be visible. diff --git a/level_handler.go b/level_handler.go index 0c58e1fef..2476b1c78 100644 --- a/level_handler.go +++ b/level_handler.go @@ -7,6 +7,7 @@ package badger import ( "fmt" + "math" "sort" "sync" @@ -313,14 +314,18 @@ func (s *levelHandler) appendIterators(iters []y.Iterator, opt *IteratorOptions) out = append(out, t) } } - return appendIteratorsReversed(iters, out, topt) + return appendIteratorsReversed(iters, out, topt, opt) } tables := opt.pickTables(s.tables) if len(tables) == 0 { return iters } - return append(iters, table.NewConcatIterator(tables, topt)) + ci := table.NewConcatIterator(tables, topt) + if lower, upper, enabled := opt.versionWindow(math.MaxUint64); enabled { + ci.SetVersionBounds(lower, upper) + } + return append(iters, ci) } type levelHandlerRLocked struct{} diff --git a/levels.go b/levels.go index eed99dc95..a97c8663f 100644 --- a/levels.go +++ b/levels.go @@ -911,7 +911,7 @@ func (s *levelsController) compactBuildTables( var iters []y.Iterator switch { case lev == 0: - iters = appendIteratorsReversed(iters, topTables, table.NOCACHE) + iters = appendIteratorsReversed(iters, topTables, table.NOCACHE, nil) case len(topTables) > 0: y.AssertTrue(len(topTables) == 1) iters = []y.Iterator{topTables[0].NewIterator(table.NOCACHE)} @@ -1645,10 +1645,20 @@ func (s *levelsController) get(key []byte, maxVs y.ValueStruct, startLevel int) return maxVs, nil } -func appendIteratorsReversed(out []y.Iterator, th []*table.Table, opt int) []y.Iterator { +func appendIteratorsReversed( + out []y.Iterator, th []*table.Table, topt int, opt *IteratorOptions) []y.Iterator { + var lower, upper uint64 + var vEnabled bool + if opt != nil { + lower, upper, vEnabled = opt.versionWindow(math.MaxUint64) + } for i := len(th) - 1; i >= 0; i-- { // This will increment the reference of the table handler. - out = append(out, th[i].NewIterator(opt)) + it := th[i].NewIterator(topt) + if vEnabled { + it.SetVersionBounds(lower, upper) + } + out = append(out, it) } return out } diff --git a/table/builder.go b/table/builder.go index e493692ab..3e4ee96ef 100644 --- a/table/builder.go +++ b/table/builder.go @@ -62,6 +62,13 @@ type bblock struct { baseKey []byte // Base key for the current block. entryOffsets []uint32 // Offsets of entries present in current block. end int // Points to the end offset of the block. + + // minVersion/maxVersion are a conservative [min,max] over the versions + // (timestamps) of every key added to this block, including tombstones. + // minVersion is initialized to MaxUint64 so the first key sets the range. + // They are emitted into the BlockOffset for version-range scan pruning. + minVersion uint64 + maxVersion uint64 } // Builder is used in building a table. @@ -72,12 +79,18 @@ type Builder struct { compressedSize atomic.Uint32 uncompressedSize atomic.Uint32 - lenOffsets uint32 - keyHashes []uint32 // Used for building the bloomfilter. - opts *Options - maxVersion uint64 - onDiskSize uint32 - staleDataSize int + lenOffsets uint32 + keyHashes []uint32 // Used for building the bloomfilter. + opts *Options + maxVersion uint64 + minVersion uint64 // Smallest key version across the whole table (MaxUint64 until first key). + onDiskSize uint32 + + // omitVersionRange, when true, suppresses emitting the appended version-range + // fields, reproducing the on-disk layout written by code predating this + // feature. Test-only; production always emits the range. + omitVersionRange bool + staleDataSize int // Used to concurrently compress/encrypt blocks. wg sync.WaitGroup @@ -120,12 +133,14 @@ func NewTableBuilder(opts Options) *Builder { sz = maxAllocatorInitialSz } b := &Builder{ - alloc: opts.AllocPool.Get(sz, "TableBuilder"), - opts: &opts, + alloc: opts.AllocPool.Get(sz, "TableBuilder"), + opts: &opts, + minVersion: math.MaxUint64, } b.alloc.Tag = "Builder" b.curBlock = &bblock{ - data: b.alloc.Allocate(opts.BlockSize + padding), + data: b.alloc.Allocate(opts.BlockSize + padding), + minVersion: math.MaxUint64, } b.opts.tableCapacity = uint64(float64(b.opts.TableSize) * 0.95) @@ -209,9 +224,22 @@ func (b *Builder) keyDiff(newKey []byte) []byte { func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint32) { b.keyHashes = append(b.keyHashes, y.Hash(y.ParseKey(key))) - if version := y.ParseTs(key); version > b.maxVersion { + version := y.ParseTs(key) + if version > b.maxVersion { b.maxVersion = version } + if version < b.minVersion { + b.minVersion = version + } + // Track a conservative per-block [min,max] over every key's version, + // including tombstones (their version is part of the key). The first key in + // a block sets the range since minVersion starts at MaxUint64. + if version > b.curBlock.maxVersion { + b.curBlock.maxVersion = version + } + if version < b.curBlock.minVersion { + b.curBlock.minVersion = version + } // diffKey stores the difference of key with baseKey. var diffKey []byte @@ -342,7 +370,8 @@ func (b *Builder) addInternal(key []byte, value y.ValueStruct, valueLen uint32, b.finishBlock() // Create a new block and start writing. b.curBlock = &bblock{ - data: b.alloc.Allocate(b.opts.BlockSize + padding), + data: b.alloc.Allocate(b.opts.BlockSize + padding), + minVersion: math.MaxUint64, } } b.addHelper(key, value, valueLen) @@ -542,10 +571,19 @@ func (b *Builder) buildIndex(bloom []byte) ([]byte, uint32) { bfoff = builder.CreateByteVector(bloom) } b.onDiskSize += dataSize + // Normalize the table-wide min version: if no keys were added, minVersion is + // still MaxUint64; emit 0 so we don't advertise a bogus range. + minVersion := b.minVersion + if minVersion == math.MaxUint64 { + minVersion = 0 + } fb.TableIndexStart(builder) fb.TableIndexAddOffsets(builder, boEnd) fb.TableIndexAddBloomFilter(builder, bfoff) fb.TableIndexAddMaxVersion(builder, b.maxVersion) + if !b.omitVersionRange { + fb.TableIndexAddMinVersion(builder, minVersion) + } fb.TableIndexAddUncompressedSize(builder, b.uncompressedSize.Load()) fb.TableIndexAddKeyCount(builder, uint32(len(b.keyHashes))) fb.TableIndexAddOnDiskSize(builder, b.onDiskSize) @@ -584,5 +622,17 @@ func (b *Builder) writeBlockOffset( fb.BlockOffsetAddKey(builder, k) fb.BlockOffsetAddOffset(builder, startOffset) fb.BlockOffsetAddLen(builder, uint32(bl.end)) + if b.omitVersionRange { + return fb.BlockOffsetEnd(builder) + } + // Per-block version range for scan pruning, plus near-free decompress sizing. + // A finished block always has at least one key, so [minVersion,maxVersion] is + // a valid conservative range. Note: when minVersion==0 (and/or maxVersion==0) + // the field is omitted by flatbuffers (value==default), and the reader will + // treat the block range as absent and never prune it — a safe fallback. + fb.BlockOffsetAddMinVersion(builder, bl.minVersion) + fb.BlockOffsetAddMaxVersion(builder, bl.maxVersion) + fb.BlockOffsetAddUncompressedLen(builder, uint32(bl.end)) + fb.BlockOffsetAddKeyCount(builder, uint32(len(bl.entryOffsets))) return fb.BlockOffsetEnd(builder) } diff --git a/table/iterator.go b/table/iterator.go index 772f6396e..6fcdd282e 100644 --- a/table/iterator.go +++ b/table/iterator.go @@ -171,6 +171,15 @@ type Iterator struct { // Internally, Iterator is bidirectional. However, we only expose the // unidirectional functionality for now. opt int // Valid options are REVERSED and NOCACHE. + + // vLower/vUpper define an inclusive version window [vLower,vUpper]. When + // vEnabled is true, blocks whose [min,max] version range provably does not + // intersect the window are skipped without decompressing them. This is a + // performance prune only; the caller's per-entry version filter stays + // authoritative. Blocks lacking version-range metadata are never skipped. + vEnabled bool + vLower uint64 + vUpper uint64 } // NewIterator returns a new iterator of the Table @@ -180,6 +189,34 @@ func (t *Table) NewIterator(opt int) *Iterator { return ti } +// SetVersionBounds restricts iteration to keys whose version v satisfies +// lower <= v <= upper, enabling whole-block skipping when a block's recorded +// [min,max] version range does not intersect the window. It must be called +// before the iterator is positioned. Blocks without version-range metadata +// (e.g. from older tables) are never skipped. +func (itr *Iterator) SetVersionBounds(lower, upper uint64) { + itr.vEnabled = true + itr.vLower = lower + itr.vUpper = upper +} + +// blockOutsideWindow reports whether block bpos provably contains no key in the +// active version window and may be skipped. Always false when version bounds are +// not set or the block lacks version-range metadata. +func (itr *Iterator) blockOutsideWindow(bpos int) bool { + if !itr.vEnabled { + return false + } + var ko fb.BlockOffset + if !itr.t.offsets(&ko, bpos) { + return false + } + if !ko.VersionRangePresent() { + return false + } + return ko.MaxVersion() < itr.vLower || ko.MinVersion() > itr.vUpper +} + // Close closes the iterator (and it must be called). func (itr *Iterator) Close() error { itr.bi.Close() @@ -207,6 +244,14 @@ func (itr *Iterator) seekToFirst() { return } itr.bpos = 0 + // Skip leading blocks that are outside the version window. + for itr.bpos < numBlocks && itr.blockOutsideWindow(itr.bpos) { + itr.bpos++ + } + if itr.bpos >= numBlocks { + itr.err = io.EOF + return + } block, err := itr.t.block(itr.bpos, itr.useCache()) if err != nil { itr.err = err @@ -226,6 +271,14 @@ func (itr *Iterator) seekToLast() { return } itr.bpos = numBlocks - 1 + // Skip trailing blocks that are outside the version window. + for itr.bpos >= 0 && itr.blockOutsideWindow(itr.bpos) { + itr.bpos-- + } + if itr.bpos < 0 { + itr.err = io.EOF + return + } block, err := itr.t.block(itr.bpos, itr.useCache()) if err != nil { itr.err = err @@ -311,6 +364,13 @@ func (itr *Iterator) seekForPrev(key []byte) { func (itr *Iterator) next() { itr.err = nil + // Skip whole blocks that fall outside the version window without loading or + // decompressing them. Safe because such blocks contain no key in [vLower,vUpper]. + for itr.bpos < itr.t.offsetsLength() && itr.blockOutsideWindow(itr.bpos) { + itr.bpos++ + itr.bi.data = nil + } + if itr.bpos >= itr.t.offsetsLength() { itr.err = io.EOF return @@ -341,6 +401,13 @@ func (itr *Iterator) next() { func (itr *Iterator) prev() { itr.err = nil + + // Skip whole blocks that fall outside the version window (reverse direction). + for itr.bpos >= 0 && itr.blockOutsideWindow(itr.bpos) { + itr.bpos-- + itr.bi.data = nil + } + if itr.bpos < 0 { itr.err = io.EOF return @@ -429,6 +496,21 @@ type ConcatIterator struct { iters []*Iterator // Corresponds to tables. tables []*Table // Disregarding reversed, this is in ascending order. options int // Valid options are REVERSED and NOCACHE. + + // Version window propagated to each lazily-created table Iterator for + // per-block skipping. Disabled unless SetVersionBounds is called. + vEnabled bool + vLower uint64 + vUpper uint64 +} + +// SetVersionBounds restricts this concat iterator (and every table iterator it +// lazily creates) to keys whose version v satisfies lower <= v <= upper. It must +// be called before the iterator is positioned. +func (s *ConcatIterator) SetVersionBounds(lower, upper uint64) { + s.vEnabled = true + s.vLower = lower + s.vUpper = upper } // NewConcatIterator creates a new concatenated iterator @@ -457,7 +539,11 @@ func (s *ConcatIterator) setIdx(idx int) { return } if s.iters[idx] == nil { - s.iters[idx] = s.tables[idx].NewIterator(s.options) + it := s.tables[idx].NewIterator(s.options) + if s.vEnabled { + it.SetVersionBounds(s.vLower, s.vUpper) + } + s.iters[idx] = it } s.cur = s.iters[s.idx] } @@ -473,6 +559,27 @@ func (s *ConcatIterator) Rewind() { s.setIdx(len(s.iters) - 1) } s.cur.Rewind() + s.skipInvalidTables() +} + +// skipInvalidTables advances to the next table (in iteration direction) whose +// iterator is valid, after Rewind/Seek positioned us on one that is not. Version-range +// block skipping (SetVersionBounds) can leave a table iterator invalid when all its +// blocks fall outside the window; without this, the ConcatIterator — and hence its +// MergeIterator level — would look exhausted and strand later tables' in-window keys. +// Mirrors the empty-table loop in Next(). When version bounds are disabled this is a +// no-op, since a non-empty table is always valid after Rewind. +func (s *ConcatIterator) skipInvalidTables() { + for s.cur != nil && !s.cur.Valid() { + if s.options&REVERSED == 0 { + s.setIdx(s.idx + 1) + } else { + s.setIdx(s.idx - 1) + } + if s.cur != nil { + s.cur.Rewind() + } + } } // Valid implements y.Interface diff --git a/table/table.go b/table/table.go index cd6d86dc2..87638d585 100644 --- a/table/table.go +++ b/table/table.go @@ -81,6 +81,10 @@ type TableInterface interface { Biggest() []byte DoesNotHave(hash uint32) bool MaxVersion() uint64 + // OutsideVersionRange reports whether the table provably contains no key with + // a version v in [lower,upper]; returns false for tables without version-range + // metadata (never skip). + OutsideVersionRange(lower, upper uint64) bool } // Table represents a loaded table file with the info we have about it. @@ -110,6 +114,8 @@ type Table struct { type cheapIndex struct { MaxVersion uint64 + MinVersion uint64 + HasVersionRange bool // true if this table was written with a min_version field. KeyCount uint32 UncompressedSize uint32 OnDiskSize uint32 @@ -125,6 +131,28 @@ func (t *Table) offsetsLength() int { return t.cheapIndex().OffsetsLength } // MaxVersion returns the maximum version across all keys stored in this table. func (t *Table) MaxVersion() uint64 { return t.cheapIndex().MaxVersion } +// MinVersion returns the minimum version across all keys stored in this table. +// It is only meaningful when HasVersionRange reports true; otherwise it is 0. +func (t *Table) MinVersion() uint64 { return t.cheapIndex().MinVersion } + +// HasVersionRange reports whether this table was written with version-range +// metadata (a min_version field). Tables written by older code return false and +// must never be pruned on the basis of their version range. +func (t *Table) HasVersionRange() bool { return t.cheapIndex().HasVersionRange } + +// OutsideVersionRange reports whether this table provably contains NO key whose +// version v satisfies lower <= v <= upper, and can therefore be skipped entirely +// by a version-bounded scan. It returns false (do not skip) for tables without +// version-range metadata. The check is a conservative performance prune only; +// the per-entry version filter remains authoritative. +func (t *Table) OutsideVersionRange(lower, upper uint64) bool { + ci := t.cheapIndex() + if !ci.HasVersionRange { + return false + } + return ci.MaxVersion < lower || ci.MinVersion > upper +} + // BloomFilterSize returns the size of the bloom filter in bytes stored in memory. func (t *Table) BloomFilterSize() int { return t.cheapIndex().BloomFilterLength } @@ -459,6 +487,8 @@ func (t *Table) initIndex() (*fb.BlockOffset, error) { } t._cheap = &cheapIndex{ MaxVersion: index.MaxVersion(), + MinVersion: index.MinVersion(), + HasVersionRange: index.MinVersionPresent(), KeyCount: index.KeyCount(), UncompressedSize: index.UncompressedSize(), OnDiskSize: index.OnDiskSize(), diff --git a/table/version_range_test.go b/table/version_range_test.go new file mode 100644 index 000000000..c81c667ea --- /dev/null +++ b/table/version_range_test.go @@ -0,0 +1,381 @@ +/* + * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package table + +import ( + "fmt" + "math" + "math/rand" + "os" + "sort" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/dgraph-io/badger/v4/fb" + "github.com/dgraph-io/badger/v4/options" + "github.com/dgraph-io/badger/v4/y" +) + +// bitDelete mirrors badger's tombstone meta bit. The exact value is irrelevant +// to the table layer (the version lives in the key regardless); we only use it +// to assert that delete entries widen the block version range. +const testBitDelete = 1 << 0 + +type vkv struct { + key string + version uint64 + value string + delete bool +} + +// buildVersionTable builds a table from (key,version,value) tuples. Keys are +// encoded with their version via y.KeyWithTs and sorted in badger order so that +// blocks are populated deterministically. A tiny block size forces many blocks. +func buildVersionTable(t *testing.T, kvs []vkv, blockSize int) *Table { + opts := Options{ + Compression: options.ZSTD, + BlockSize: blockSize, + BloomFalsePositive: 0.01, + TableSize: 30 << 20, + } + b := NewTableBuilder(opts) + defer b.Close() + + encoded := make([][]byte, len(kvs)) + idx := make([]int, len(kvs)) + for i := range kvs { + encoded[i] = y.KeyWithTs([]byte(kvs[i].key), kvs[i].version) + idx[i] = i + } + sort.Slice(idx, func(a, b int) bool { + return y.CompareKeys(encoded[idx[a]], encoded[idx[b]]) < 0 + }) + + for _, i := range idx { + meta := byte('A') + if kvs[i].delete { + meta = testBitDelete + } + b.Add(encoded[i], y.ValueStruct{Value: []byte(kvs[i].value), Meta: meta}, 0) + } + + filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Uint32()) + tbl, err := CreateTable(filename, b) + require.NoError(t, err) + return tbl +} + +// scanAll returns all (key-with-ts) keys an iterator yields, applying an +// optional version window filter [lower,upper] at the per-entry level. This is +// the authoritative reference the pruned scan must match exactly. +func referenceKeys(kvs []vkv, lower, upper uint64) []string { + var out []string + for _, kv := range kvs { + if kv.version >= lower && kv.version <= upper { + out = append(out, string(y.KeyWithTs([]byte(kv.key), kv.version))) + } + } + sort.Slice(out, func(a, b int) bool { + return y.CompareKeys([]byte(out[a]), []byte(out[b])) < 0 + }) + return out +} + +// iterateKeys collects keys from a table iterator that fall within [lower,upper] +// (the per-entry filter the real merge iterator applies). When bound is true the +// iterator also has SetVersionBounds applied (storage-level block skipping). +func iterateKeys(it *Iterator, lower, upper uint64) []string { + var out []string + for it.Rewind(); it.Valid(); it.Next() { + v := y.ParseTs(it.Key()) + if v < lower || v > upper { + continue + } + out = append(out, string(it.Key())) + } + return out +} + +// buildOldFormatTable builds a table whose index omits the appended +// version-range fields, reproducing the on-disk format written by code that +// predates this feature. +func buildOldFormatTable(t *testing.T) *Table { + opts := Options{ + Compression: options.ZSTD, + BlockSize: 512, + BloomFalsePositive: 0.01, + TableSize: 30 << 20, + } + b := NewTableBuilder(opts) + b.omitVersionRange = true + defer b.Close() + + type pair struct{ k, v string } + pairs := make([]pair, 50) + for i := 0; i < 50; i++ { + pairs[i] = pair{k: fmt.Sprintf("key%04d", i), v: "v"} + } + sort.Slice(pairs, func(a, c int) bool { return pairs[a].k < pairs[c].k }) + for i, p := range pairs { + b.Add(y.KeyWithTs([]byte(p.k), uint64(10+i)), y.ValueStruct{Value: []byte(p.v), Meta: 'A'}, 0) + } + filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Uint32()) + tbl, err := CreateTable(filename, b) + require.NoError(t, err) + return tbl +} + +// TestBlockVersionRangeMetadata verifies the builder records a conservative +// per-block [min,max] over every key's version, including tombstones, and a +// table-wide min/max. +func TestBlockVersionRangeMetadata(t *testing.T) { + // Two version "bands": low keys at versions 1..5, high keys at 100..105, + // plus a delete at a high version to assert it widens the range. + var kvs []vkv + for i := 0; i < 40; i++ { + kvs = append(kvs, vkv{key: fmt.Sprintf("low%04d", i), version: uint64(1 + i%5), value: "v"}) + } + for i := 0; i < 40; i++ { + del := i == 7 // a tombstone at a high version + kvs = append(kvs, vkv{key: fmt.Sprintf("zzz%04d", i), version: uint64(100 + i%6), value: "v", delete: del}) + } + + tbl := buildVersionTable(t, kvs, 1024) // tiny blocks => many blocks + defer func() { require.NoError(t, tbl.DecrRef()) }() + + require.True(t, tbl.HasVersionRange(), "new table must advertise version range") + require.EqualValues(t, 1, tbl.MinVersion()) + require.EqualValues(t, 105, tbl.MaxVersion()) + + n := tbl.offsetsLength() + require.Greater(t, n, 1, "test needs multiple blocks; lower BlockSize") + + // Every block's recorded range must contain every version of every key in + // that block (conservative over-approximation, deletes included). + for i := 0; i < n; i++ { + var ko fb.BlockOffset + require.True(t, tbl.offsets(&ko, i)) + require.True(t, ko.VersionRangePresent(), "block %d missing version range", i) + bmin, bmax := ko.MinVersion(), ko.MaxVersion() + require.LessOrEqual(t, bmin, bmax) + + // Walk the actual entries in this block and confirm containment. + it := tbl.NewIterator(0) + it.bpos = i + block, err := tbl.block(i, true) + require.NoError(t, err) + it.bi.setBlock(block) + for it.bi.seekToFirst(); it.bi.Valid(); it.bi.next() { + v := y.ParseTs(it.bi.key) + require.GreaterOrEqual(t, v, bmin, "block %d min too high for key version", i) + require.LessOrEqual(t, v, bmax, "block %d max too low for key version", i) + } + require.NoError(t, it.Close()) + } +} + +// TestVersionBoundedScanIdenticalResults asserts a bounded scan with block +// skipping returns IDENTICAL results to an unbounded scan filtered per-entry, +// across randomized multi-version data including deletes, for many windows. +func TestVersionBoundedScanIdenticalResults(t *testing.T) { + rng := rand.New(rand.NewSource(42)) + var kvs []vkv + for i := 0; i < 500; i++ { + k := fmt.Sprintf("key%05d", i) + nver := 1 + rng.Intn(4) + for j := 0; j < nver; j++ { + ver := uint64(1 + rng.Intn(200)) + kvs = append(kvs, vkv{ + key: k, + version: ver, + value: fmt.Sprintf("v%d", ver), + delete: rng.Intn(5) == 0, + }) + } + } + + tbl := buildVersionTable(t, kvs, 512) + defer func() { require.NoError(t, tbl.DecrRef()) }() + + windows := [][2]uint64{ + {1, math.MaxUint64}, // full range + {50, 150}, + {1, 30}, + {180, 250}, // partially above data + {300, 400}, // entirely above data => empty + {100, 100}, // single version + } + + for _, w := range windows { + lower, upper := w[0], w[1] + t.Run(fmt.Sprintf("window_%d_%d", lower, upper), func(t *testing.T) { + want := referenceKeys(kvs, lower, upper) + + // Unbounded iterator, per-entry filtered (no storage skipping). + itUnbounded := tbl.NewIterator(0) + gotUnbounded := iterateKeys(itUnbounded, lower, upper) + require.NoError(t, itUnbounded.Close()) + + // Bounded iterator WITH block skipping enabled. + itBounded := tbl.NewIterator(0) + itBounded.SetVersionBounds(lower, upper) + gotBounded := iterateKeys(itBounded, lower, upper) + require.NoError(t, itBounded.Close()) + + require.Equal(t, want, gotUnbounded, "unbounded scan mismatch") + require.Equal(t, want, gotBounded, "bounded scan must equal unbounded filtered scan") + }) + } +} + +// TestBlocksProvablyPruned asserts that blocks whose version range lies entirely +// outside the window are actually skipped (not loaded). We build distinct version +// bands so that a window selecting only the high band must skip the low-band +// blocks, and verify via the storage-level predicate that those blocks are pruned. +func TestBlocksProvablyPruned(t *testing.T) { + var kvs []vkv + // Low band: keys aaa* at versions 1..3. + for i := 0; i < 60; i++ { + kvs = append(kvs, vkv{key: fmt.Sprintf("aaa%04d", i), version: uint64(1 + i%3), value: "v"}) + } + // High band: keys zzz* at versions 500..503 (incl. a delete). + for i := 0; i < 60; i++ { + kvs = append(kvs, vkv{key: fmt.Sprintf("zzz%04d", i), version: uint64(500 + i%4), value: "v", delete: i == 3}) + } + + tbl := buildVersionTable(t, kvs, 512) + defer func() { require.NoError(t, tbl.DecrRef()) }() + + n := tbl.offsetsLength() + require.Greater(t, n, 2) + + // Window selecting only the high band. + lower, upper := uint64(490), uint64(510) + it := tbl.NewIterator(0) + it.SetVersionBounds(lower, upper) + defer it.Close() + + prunable, kept := 0, 0 + for i := 0; i < n; i++ { + if it.blockOutsideWindow(i) { + prunable++ + } else { + kept++ + } + } + require.Greater(t, prunable, 0, "expected at least one low-band block to be prunable") + require.Greater(t, kept, 0, "expected at least one high-band block to be kept") + + // The bounded scan must return exactly the high-band entries in the window. + got := iterateKeys(it, lower, upper) + want := referenceKeys(kvs, lower, upper) + require.Equal(t, want, got) + // Sanity: all returned keys are high-band. + for _, k := range got { + require.Equal(t, "zzz", string(y.ParseKey([]byte(k)))[:3]) + } +} + +// TestVersionZeroBlockNotSkipped is a regression test for a lower-bound bug: +// on an UntilTs-only scan (window lower bound 0), a block whose entire version +// range is [0,0] must NOT be skipped, because version-0 keys are in-window. +// +// Note: when a block's min and max are both 0, the appended flatbuffer fields are +// omitted (value==default), so VersionRangePresent() is false and the block is +// conservatively never pruned. Either way the version-0 data must be returned. +func TestVersionZeroBlockNotSkipped(t *testing.T) { + var kvs []vkv + // A band of version-0 keys ... + for i := 0; i < 40; i++ { + kvs = append(kvs, vkv{key: fmt.Sprintf("aaa%04d", i), version: 0, value: "v"}) + } + // ... and a band of higher-version keys to force multiple blocks/bands. + for i := 0; i < 40; i++ { + kvs = append(kvs, vkv{key: fmt.Sprintf("zzz%04d", i), version: uint64(500 + i%4), value: "v"}) + } + tbl := buildVersionTable(t, kvs, 512) + defer func() { require.NoError(t, tbl.DecrRef()) }() + + // Window lower=0 (UntilTs-only scan semantics), upper=100. + it := tbl.NewIterator(0) + it.SetVersionBounds(0, 100) + defer it.Close() + + // The version-0 keys are in-window; the bounded scan must surface every one + // of them (and none of the high-version keys). + want := referenceKeys(kvs, 0, 100) + require.NotEmpty(t, want) + got := iterateKeys(it, 0, 100) + require.Equal(t, want, got, "version-0 block must not be skipped on a lower=0 window") +} + +// TestOldFormatTableNeverSkipped simulates a table written by older code that +// lacks version-range metadata: it must open fine, report HasVersionRange=false, +// and never be pruned by the block/table skip predicates. +func TestOldFormatTableNeverSkipped(t *testing.T) { + tbl := buildOldFormatTable(t) + defer func() { require.NoError(t, tbl.DecrRef()) }() + + require.False(t, tbl.HasVersionRange(), "old-format table must not advertise a version range") + // Table-level skip must never fire for an old table, even for a window that + // could not possibly intersect any plausible range. + require.False(t, tbl.OutsideVersionRange(1000, 2000)) + require.False(t, tbl.OutsideVersionRange(0, 0)) + + // Block-level skip must never fire either. + it := tbl.NewIterator(0) + defer it.Close() + it.SetVersionBounds(1000, 2000) + for i := 0; i < tbl.offsetsLength(); i++ { + var ko fb.BlockOffset + require.True(t, tbl.offsets(&ko, i)) + require.False(t, ko.VersionRangePresent(), "old block must not advertise a version range") + require.False(t, it.blockOutsideWindow(i), "old-format block must never be skipped") + } + + // And iteration still yields every key (skipping is disabled for old tables). + count := 0 + for it.Rewind(); it.Valid(); it.Next() { + count++ + } + require.Equal(t, 50, count) +} + +// TestConcatIteratorSkipsFullyPrunedFirstTable is a regression test for the +// ConcatIterator stranding later tables when the first table's blocks are all +// outside the version window. Table 1 (keys a1,a2) holds only old versions and is +// fully block-pruned by the window [5,10]; table 2 (key m1) holds an in-window +// version. A version-bounded ConcatIterator's Rewind must still surface table 2's key +// instead of treating the whole level as exhausted, which dropped in-window data +// before the fix. (Seek can't hit this: seekFrom does not version-skip blocks, and the +// binary search guarantees the chosen table holds a key >= the seek key.) +func TestConcatIteratorSkipsFullyPrunedFirstTable(t *testing.T) { + lower, upper := uint64(5), uint64(10) + + t1 := buildVersionTable(t, []vkv{ + {key: "a1", version: 1, value: "x"}, + {key: "a2", version: 2, value: "x"}, + }, 512) + defer func() { require.NoError(t, t1.DecrRef()) }() + t2 := buildVersionTable(t, []vkv{ + {key: "m1", version: 7, value: "y"}, + }, 512) + defer func() { require.NoError(t, t2.DecrRef()) }() + + require.Less(t, t1.MaxVersion(), lower, "table1 must lie entirely below the window") + + ci := NewConcatIterator([]*Table{t1, t2}, 0) + ci.SetVersionBounds(lower, upper) + defer func() { require.NoError(t, ci.Close()) }() + + var got []string + for ci.Rewind(); ci.Valid(); ci.Next() { + got = append(got, string(y.ParseKey(ci.Key()))) + } + require.Equal(t, []string{"m1"}, got, + "Rewind must surface the in-window key in table 2 despite table 1 being fully pruned") +}