core/rawdb: add freezer's AncientBytes method

Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
jsvisa 2025-07-02 01:02:38 +08:00 committed by Gary Rong
parent 659342a523
commit 1459c2ed24
4 changed files with 133 additions and 0 deletions

View file

@ -202,6 +202,14 @@ func (f *Freezer) AncientRange(kind string, start, count, maxBytes uint64) ([][]
return nil, errUnknownTable
}
// RetrieveBytes retrieves a byte range [offset:offset+length] from the specified ancient item.
func (f *Freezer) AncientBytes(kind string, number uint64, offset, length uint64) ([]byte, error) {
if table := f.tables[kind]; table != nil {
return table.RetrieveBytes(number, offset, length)
}
return nil, errUnknownTable
}
// Ancients returns the length of the frozen items.
func (f *Freezer) Ancients() (uint64, error) {
return f.frozen.Load(), nil

View file

@ -126,6 +126,13 @@ func (f *resettableFreezer) AncientRange(kind string, start, count, maxBytes uin
return f.freezer.AncientRange(kind, start, count, maxBytes)
}
// RetrieveBytes retrieves a byte range [offset:offset+length] from the specified ancient item.
func (f *resettableFreezer) AncientBytes(kind string, number uint64, offset, length uint64) ([]byte, error) {
f.lock.RLock()
defer f.lock.RUnlock()
return f.freezer.AncientBytes(kind, number, offset, length)
}
// Ancients returns the length of the frozen items.
func (f *resettableFreezer) Ancients() (uint64, error) {
f.lock.RLock()

View file

@ -1235,3 +1235,65 @@ func (t *freezerTable) dumpIndex(w io.Writer, start, stop int64) {
}
fmt.Fprintf(w, "|--------------------------|\n")
}
// RetrieveBytes retrieves a byte range [offset:offset+length] from the specified ancient item.
func (t *freezerTable) RetrieveBytes(item, offset, length uint64) ([]byte, error) {
t.lock.RLock()
defer t.lock.RUnlock()
if t.index == nil || t.head == nil || t.metadata.file == nil {
return nil, errClosed
}
items := t.items.Load()
hidden := t.itemHidden.Load()
if items <= item || hidden > item {
return nil, errOutOfBounds
}
// Read the index entries for the item and the next item,
// this method will return two indices or an error.
indices, err := t.getIndices(item, 1)
if err != nil {
return nil, err
}
index0 := indices[0]
index1 := indices[1]
startOffset, endOffset, fileId := index0.bounds(index1)
itemSize := endOffset - startOffset
dataFile, exist := t.files[fileId]
if !exist {
return nil, fmt.Errorf("missing data file %d", fileId)
}
readBytes := int(itemSize)
if t.config.noSnappy {
// Range check before reading
if offset > uint64(itemSize) || offset+length > uint64(itemSize) {
return nil, fmt.Errorf("requested range out of bounds: item size %d, offset %d, length %d", itemSize, offset, length)
}
startOffset += uint32(offset)
readBytes = int(length)
}
buf := make([]byte, readBytes)
_, err = dataFile.ReadAt(buf, int64(startOffset))
if err != nil {
return nil, err
}
// If not compressed, read only the requested bytes
if t.config.noSnappy {
return buf, nil
}
// If compressed, read the full item, decompress, then slice
data, err := snappy.Decode(nil, buf)
if err != nil {
return nil, err
}
if offset > uint64(len(data)) || offset+length > uint64(len(data)) {
return nil, fmt.Errorf("requested range out of bounds after decompression: item size %d, offset %d, length %d", len(data), offset, length)
}
return data[offset : offset+length], nil
}

View file

@ -1571,3 +1571,59 @@ func TestTailTruncationCrash(t *testing.T) {
t.Fatalf("Unexpected index flush offset, want: %d, got: %d", 26*indexEntrySize, f.metadata.flushOffset)
}
}
func TestFreezerAncientBytes(t *testing.T) {
t.Parallel()
types := []struct {
name string
config freezerTableConfig
}{
{"uncompressed", freezerTableConfig{noSnappy: true}},
{"compressed", freezerTableConfig{noSnappy: false}},
}
for _, typ := range types {
t.Run(typ.name, func(t *testing.T) {
f, err := newTable(os.TempDir(), fmt.Sprintf("ancientbytes-%s-%d", typ.name, rand.Uint64()), metrics.NewMeter(), metrics.NewMeter(), metrics.NewGauge(), 1000, typ.config, false)
if err != nil {
t.Fatal(err)
}
defer f.Close()
for i := 0; i < 10; i++ {
data := getChunk(100, i)
batch := f.newBatch()
require.NoError(t, batch.AppendRaw(uint64(i), data))
require.NoError(t, batch.commit())
}
for i := 0; i < 10; i++ {
full, err := f.Retrieve(uint64(i))
require.NoError(t, err)
// Full read
got, err := f.RetrieveBytes(uint64(i), 0, uint64(len(full)))
require.NoError(t, err)
if !bytes.Equal(got, full) {
t.Fatalf("full read mismatch for entry %d", i)
}
// Middle slice
got, err = f.RetrieveBytes(uint64(i), 10, 50)
require.NoError(t, err)
if !bytes.Equal(got, full[10:60]) {
t.Fatalf("middle slice mismatch for entry %d", i)
}
// Single byte
got, err = f.RetrieveBytes(uint64(i), 99, 1)
require.NoError(t, err)
if !bytes.Equal(got, full[99:100]) {
t.Fatalf("single byte mismatch for entry %d", i)
}
// Out of bounds
_, err = f.RetrieveBytes(uint64(i), 100, 1)
if err == nil {
t.Fatalf("expected error for out-of-bounds read for entry %d", i)
}
}
})
}
}