sstable: lazy load the index block in single level iterator

This commit implements the index block lazy loading in single-level iterator as default behavior.
The index block loading is now deferred until first access. At the same time, it leads to semantic
changes as the file reading ordering changes because of this.

Here is a quick overview of the change.

Before (eager loading)

1. Iterator Construction:
   1.1 the constructor calls readTopLevelIndexBlock()
   1.2 readTopLevelIndexBlock() calls readIndexBlock()
   1.3 readIndexBlock(() calls blockReader.Read()
   1.4 blockReader.Read() calls objstorage ReadHandle ReadAt()
   1.5 ReadAt() loads the index block
   1.6 construction completes

2. First Positioning (e.g., First(), SeekGE(), SeekPrefixGE()):
   2.1 the positioning function uses the already-loaded index via PI(&i.index)
   2.2 file opened for data read
   2.3 ReadAt() loads the data block

After (lazy loading)

1. Iterator Construction:
   1.1 the constructor sets indexLoaded = false to defer loading
   1.2 construction completes (no I/O)

2. First Positioning (e.g., First(), SeekGE(), SeekPrefixGE()):
   2.1 the positioning function calls the new indexIter() wrapper
   2.2 indexIter() calls ensureIndexLoaded()
   2.3 ensureIndexLoaded() calls ReadAt()
   2.4 ReadAt() loads the index block
   2.5 uses the newly-loaded index for positioning
   2.6 ReadAt() loads the data block

As seen above, the I/O operation is now deferred. While the overall behavior is transparent to
users, the internal semantic is different now. And the data-driven tests have to change
accordingly. See the order of events change in file: checkpoint, cleaner, and event_listener.

What's interesting (optimized) is the change observed in flushable_ingest, which demonstrate
the I/O saving thanks to the optimization.

```
-# When the key doesn't pass the bloom filter, we should see only two block
-# reads.
+# When the key doesn't pass the bloom filter, we should see only one block
+# read due to lazy loading optimization - bloom filter is checked first,
+# and if it rejects the key, we can avoid loading the index block entirely.
 get with-fs-logging
 small-00001-does-not-exist
 ----
-read-at(158, 41): 000004.sst
 read-at(199, 74): 000004.sst
```

implements #3248

Author: xxmplus

sstable/reader_iter_single_lvl.go

@@ -165,6 +165,9 @@ type singleLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIte
 	useFilterBlock         bool
 	lastBloomFilterMatched bool
 
+	// Lazy loading flag
+	indexLoaded bool
+
 	transforms IterTransforms
 
 	// All fields above this field are cleared when resetting the iterator for reuse.
@@ -214,14 +217,9 @@ func newColumnBlockSingleLevelIterator(
 		i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 	}
 	i.data.InitOnce(r.keySchema, r.Comparer, &i.internalValueConstructor)
-	indexH, err := r.readTopLevelIndexBlock(ctx, i.readEnv.Block, i.indexFilterRH)
-	if err == nil {
-		err = i.index.InitHandle(r.Comparer, indexH, opts.Transforms)
-	}
-	if err != nil {
-		_ = i.Close()
-		return nil, err
-	}
+
+	// Use lazy loading by default - index will be loaded on first access
+	i.indexLoaded = false
 	return i, nil
 }
 
@@ -254,14 +252,8 @@ func newRowBlockSingleLevelIterator(
 		i.data.SetHasValuePrefix(true)
 	}
 
-	indexH, err := r.readTopLevelIndexBlock(ctx, i.readEnv.Block, i.indexFilterRH)
-	if err == nil {
-		err = i.index.InitHandle(r.Comparer, indexH, opts.Transforms)
-	}
-	if err != nil {
-		_ = i.Close()
-		return nil, err
-	}
+	// Use lazy loading by default - index will be loaded on first access
+	i.indexLoaded = false
 	return i, nil
 }
 
@@ -449,7 +441,7 @@ func (i *singleLevelIterator[I, PI, P, PD]) SetContext(ctx context.Context) {
 // unpositioned. If unsuccessful, it sets i.err to any error encountered, which
 // may be nil if we have simply exhausted the entire table.
 func (i *singleLevelIterator[I, PI, P, PD]) loadDataBlock(dir int8) loadBlockResult {
-	if !PI(&i.index).Valid() {
+	if i.err != nil || !PI(&i.index).Valid() {
 		// Ensure the data block iterator is invalidated even if loading of the block
 		// fails.
 		PD(&i.data).Invalidate()
@@ -521,6 +513,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) ReadValueBlock(
 // apprioriate bound, depending on the iteration direction, and returns either
 // `blockIntersects` or `blockExcluded`.
 func (i *singleLevelIterator[I, PI, D, PD]) resolveMaybeExcluded(dir int8) intersectsResult {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return blockExcluded
+		}
+	}
 	// TODO(jackson): We could first try comparing to top-level index block's
 	// key, and if within bounds avoid per-data block key comparisons.
 
@@ -682,6 +680,13 @@ func (i *singleLevelIterator[I, PI, D, PD]) SeekGE(
 func (i *singleLevelIterator[I, PI, D, PD]) seekGEHelper(
 	key []byte, boundsCmp int, flags base.SeekGEFlags,
 ) *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
+
 	// Invariant: trySeekUsingNext => !i.data.isDataInvalidated() && i.exhaustedBounds != +1
 
 	// SeekGE performs various step-instead-of-seeking optimizations: eg enabled
@@ -818,6 +823,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) seekPrefixGE(
 			flags = flags.DisableTrySeekUsingNext()
 		}
 		i.lastBloomFilterMatched = false
+
 		// Check prefix bloom filter.
 		var mayContain bool
 		mayContain, i.err = i.bloomFilterMayContain(prefix)
@@ -922,6 +928,13 @@ func (i *singleLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV
 	i.boundsCmp = 0
 	i.positionedUsingLatestBounds = true
 
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
+
 	indexOk := PI(&i.index).SeekGE(key)
 	// We can have multiple internal keys with the same user key as the seek
 	// key. In that case, we want the last (greatest) internal key.
@@ -993,6 +1006,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV
 func (i *singleLevelIterator[I, PI, D, PD]) SeekLT(
 	key []byte, flags base.SeekLTFlags,
 ) *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	if i.readEnv.Virtual != nil {
 		// Might have to fix upper bound since virtual sstable bounds are not
 		// known to callers of SeekLT.
@@ -1116,6 +1135,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) First() *base.InternalKV {
 // index file. For the latter, one cannot make any claims about absolute
 // positioning.
 func (i *singleLevelIterator[I, PI, D, PD]) firstInternal() *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	i.exhaustedBounds = 0
 	i.err = nil // clear cached iteration error
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
@@ -1180,6 +1205,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) Last() *base.InternalKV {
 // index file. For the latter, one cannot make any claims about absolute
 // positioning.
 func (i *singleLevelIterator[I, PI, D, PD]) lastInternal() *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	i.exhaustedBounds = 0
 	i.err = nil // clear cached iteration error
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
@@ -1249,6 +1280,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) Next() *base.InternalKV {
 
 // NextPrefix implements (base.InternalIterator).NextPrefix.
 func (i *singleLevelIterator[I, PI, D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	if i.exhaustedBounds == +1 {
 		panic("NextPrefix called even though exhausted upper bound")
 	}
@@ -1343,6 +1380,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) Prev() *base.InternalKV {
 }
 
 func (i *singleLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	for {
 		if !PI(&i.index).Next() {
 			PD(&i.data).Invalidate()
@@ -1421,6 +1464,12 @@ func (i *singleLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 }
 
 func (i *singleLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
+	if !i.indexLoaded {
+		if err := i.ensureIndexLoaded(); err != nil {
+			i.err = err
+			return nil
+		}
+	}
 	for {
 		if !PI(&i.index).Prev() {
 			PD(&i.data).Invalidate()
@@ -1514,6 +1563,8 @@ func (i *singleLevelIterator[I, PI, D, PD]) closeInternal() error {
 	}
 	var err error
 	err = firstError(err, PD(&i.data).Close())
+	// Always close index iterator unconditionally to avoid BufferPool panic
+	// Even if lazy loading wasn't used, the index might have been initialized
 	err = firstError(err, PI(&i.index).Close())
 	if i.indexFilterRH != nil {
 		err = firstError(err, i.indexFilterRH.Close())
@@ -1532,6 +1583,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) closeInternal() error {
 		err = firstError(err, i.vbRH.Close())
 		i.vbRH = nil
 	}
+	i.indexLoaded = false
 	return err
 }
 
@@ -1546,3 +1598,16 @@ func (i *singleLevelIterator[I, PI, D, PD]) String() string {
 func (i *singleLevelIterator[I, PI, D, PD]) DebugTree(tp treeprinter.Node) {
 	tp.Childf("%T(%p) fileNum=%s", i, i, i.String())
 }
+
+func (i *singleLevelIterator[I, PI, D, PD]) ensureIndexLoaded() error {
+	indexH, err := i.reader.readTopLevelIndexBlock(i.ctx, i.readEnv.Block, i.indexFilterRH)
+	if err == nil {
+		err = PI(&i.index).InitHandle(i.reader.Comparer, indexH, i.transforms)
+	}
+	if err != nil {
+		return err
+	}
+
+	i.indexLoaded = true
+	return nil
+}

sstable/reader_iter_test.go

@@ -53,15 +53,20 @@ func TestIteratorErrorOnInit(t *testing.T) {
 	defer toggle.Off()
 
 	var stats base.InternalIteratorStats
-	for k := 0; k < 20; k++ {
+	for range 20 {
 		if rand.IntN(2) == 0 {
-			_, err := newRowBlockSingleLevelIterator(context.Background(), r, IterOptions{
+			iter, err := newRowBlockSingleLevelIterator(context.Background(), r, IterOptions{
 				Transforms:           NoTransforms,
 				FilterBlockSizeLimit: NeverUseFilterBlock,
 				Env:                  ReadEnv{Block: block.ReadEnv{Stats: &stats, BufferPool: &pool}},
 				ReaderProvider:       MakeTrivialReaderProvider(r),
 			})
-			require.Error(t, err)
+			// Single-level iterators use lazy loading - creation succeeds but first access fails
+			require.NoError(t, err)
+			// Error should surface when trying to use the iterator
+			_ = iter.First()
+			require.Error(t, iter.Error())
+			iter.Close()
 		} else {
 			_, err := newRowBlockTwoLevelIterator(context.Background(), r, IterOptions{
 				Transforms:           NoTransforms,

sstable/reader_iter_two_lvl.go

@@ -78,6 +78,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) loadSecondLevelIndexBlock(dir int8) loa
 		i.secondLevel.err = err
 		return loadBlockFailed
 	}
+	i.secondLevel.indexLoaded = true
 	return loadBlockOK
 }
 

testdata/checkpoint

@@ -261,16 +261,16 @@ open: db/000007.sst (options: *vfs.randomReadsOption)
 read-at(700, 61): db/000007.sst
 read-at(649, 51): db/000007.sst
 read-at(132, 517): db/000007.sst
-read-at(91, 41): db/000005.sst
 open: db/000005.sst (options: *vfs.sequentialReadsOption)
+read-at(91, 41): db/000005.sst
 read-at(0, 91): db/000005.sst
-read-at(91, 41): db/000007.sst
 open: db/000007.sst (options: *vfs.sequentialReadsOption)
+read-at(91, 41): db/000007.sst
 read-at(0, 91): db/000007.sst
 create: db/000010.sst
 close: db/000005.sst
-read-at(95, 41): db/000009.sst
 open: db/000009.sst (options: *vfs.sequentialReadsOption)
+read-at(95, 41): db/000009.sst
 read-at(0, 95): db/000009.sst
 close: db/000007.sst
 close: db/000009.sst

testdata/cleaner

@@ -82,13 +82,13 @@ open: db/000007.sst (options: *vfs.randomReadsOption)
 read-at(501, 61): db/000007.sst
 read-at(472, 37): db/000007.sst
 read-at(53, 419): db/000007.sst
-read-at(52, 27): db/000005.sst
 open: db/000005.sst (options: *vfs.sequentialReadsOption)
+read-at(52, 27): db/000005.sst
 read-at(0, 52): db/000005.sst
 create: db/000008.sst
 close: db/000005.sst
-read-at(26, 27): db/000007.sst
 open: db/000007.sst (options: *vfs.sequentialReadsOption)
+read-at(26, 27): db/000007.sst
 read-at(0, 26): db/000007.sst
 close: db/000007.sst
 sync-data: db/000008.sst

testdata/event_listener

@@ -145,11 +145,11 @@ open: db/000008.sst (options: *vfs.randomReadsOption)
 read-at(686, 61): db/000008.sst
 read-at(636, 50): db/000008.sst
 read-at(119, 517): db/000008.sst
-read-at(78, 41): db/000005.sst
 open: db/000005.sst (options: *vfs.sequentialReadsOption)
+read-at(78, 41): db/000005.sst
 read-at(0, 78): db/000005.sst
-read-at(78, 41): db/000008.sst
 open: db/000008.sst (options: *vfs.sequentialReadsOption)
+read-at(78, 41): db/000008.sst
 read-at(0, 78): db/000008.sst
 close: db/000008.sst
 close: db/000005.sst

testdata/flushable_ingest

@@ -701,17 +701,17 @@ allowFlush
 get with-fs-logging
 small-00001
 ----
-read-at(158, 41): 000004.sst
 read-at(199, 74): 000004.sst
+read-at(158, 41): 000004.sst
 read-at(0, 158): 000004.sst
 small-00001:val-00001
 
-# When the key doesn't pass the bloom filter, we should see only two block
-# reads.
+# When the key doesn't pass the bloom filter, we should see only one block
+# read due to lazy loading optimization - bloom filter is checked first,
+# and if it rejects the key, we can avoid loading the index block entirely.
 get with-fs-logging
 small-00001-does-not-exist
 ----
-read-at(158, 41): 000004.sst
 read-at(199, 74): 000004.sst
 small-00001-does-not-exist: pebble: not found