WIP - REBASE AWAY

Author: macneale4

go/store/nbs/table_reader.go

@@ -505,9 +505,9 @@ func (tr tableReader) getManyAtOffsetsWithReadFunc(
 	offsetRecords offsetRecSlice,
 	stats *Stats,
 	readAtOffsets func(
-		ctx context.Context,
-		rb readBatch,
-		stats *Stats) error,
+	ctx context.Context,
+	rb readBatch,
+	stats *Stats) error,
 ) error {
 	batches := toReadBatches(offsetRecords, tr.blockSize)
 	for i := range batches {
@@ -765,36 +765,100 @@ func (tr tableReader) clone() (tableReader, error) {
 
 func (tr tableReader) iterateAllChunks(ctx context.Context, cb func(chunk chunks.Chunk), stats *Stats) error {
 	count := tr.idx.chunkCount()
-	for i := uint32(0); i < count; i++ {
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
+	if count == 0 {
+		return nil
+	}
 
-		var h hash.Hash
-		ie, err := tr.idx.indexEntry(i, &h)
+	// Build cumulative offsets array for efficient chunk location
+	cumulativeOffsets := make([]uint64, count+1)
+	cumulativeOffsets[0] = 0
+	for i := uint32(0); i < count; i++ {
+		ie, err := tr.idx.indexEntry(i, nil)
 		if err != nil {
 			return err
 		}
+		cumulativeOffsets[i+1] = cumulativeOffsets[i] + uint64(ie.Length())
+	}
 
-		res := make([]byte, ie.Length())
-		n, err := tr.r.ReadAtWithStats(ctx, res, int64(ie.Offset()), stats)
-		if err != nil {
-			return err
-		}
-		if uint32(n) != ie.Length() {
-			return errors.New("failed to read all data")
+	// Calculate total data size to determine where chunk records end
+	totalDataSize := cumulativeOffsets[count]
+
+	// Read data in 1MB chunks
+	const bufferSize = 1024 * 1024 // 1MB
+	currentOffset := uint64(0)
+	chunkCounter := uint32(0)
+
+	// Reuse buffer across reads
+	dataBlock := make([]byte, bufferSize)
+
+	for currentOffset < totalDataSize {
+		// Calculate how much data to read (up to 1MB or remaining data)
+		remainingData := totalDataSize - currentOffset
+		readSize := bufferSize
+		if remainingData < bufferSize {
+			readSize = int(remainingData)
+			dataBlock = dataBlock[:readSize] // Resize to remaining data
 		}
 
-		cchk, err := NewCompressedChunk(h, res)
+		_, err := tr.r.ReadAtWithStats(ctx, dataBlock, int64(currentOffset), stats)
 		if err != nil {
 			return err
 		}
-		chk, err := cchk.ToChunk()
-		if err != nil {
-			return err
+
+		// Process all chunks that are fully contained within this data block
+		blockStart := currentOffset
+		blockEnd := currentOffset + uint64(readSize)
+
+		for chunkCounter < count {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+
+			chunkStart := cumulativeOffsets[chunkCounter]
+			chunkEnd := cumulativeOffsets[chunkCounter+1]
+
+			// Check if this chunk extends beyond the current block
+			if chunkEnd > blockEnd {
+				// This chunk extends beyond current block, read next block
+				currentOffset = chunkStart
+				break
+			}
+
+			// Extract chunk data from buffer
+			bufferOffset := chunkStart - blockStart
+			chunkData := dataBlock[bufferOffset : bufferOffset+(chunkEnd-chunkStart)]
+
+			// Get the hash for this chunk
+			var h hash.Hash
+			_, err := tr.idx.indexEntry(chunkCounter, &h)
+			if err != nil {
+				return err
+			}
+
+			// Create compressed chunk and decompress
+			cchk, err := NewCompressedChunk(h, chunkData)
+			if err != nil {
+				return err
+			}
+			chk, err := cchk.ToChunk()
+			if err != nil {
+				return err
+			}
+
+			// Process the chunk
+			cb(chk)
+
+			chunkCounter++
 		}
 
-		cb(chk)
+		// If we processed all chunks in this block, move to next block
+		if chunkCounter < count && cumulativeOffsets[chunkCounter] >= blockEnd {
+			currentOffset = blockEnd
+		} else if chunkCounter >= count {
+			// All chunks processed
+			break
+		}
 	}
+
 	return nil
 }