feat(table): per-block/table version-range metadata for scan skipping

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;

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<lower prunes every block, and avoid the
+		// SinceTs+1 overflow that would wrap lower to 0.
+		return math.MaxUint64, 0, enabled
+	}
+	if opt.SinceTs == 0 {
+		lower = 0 // version 0 is kept by the per-entry filter, so it must be in-window.
+	} else {
+		lower = opt.SinceTs + 1 // relevant versions satisfy v > 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
 	}