Files
wehub-resource-sync 5357c39144
Fuzzer / Run Fuzzer (push) Has been cancelled
Race tests / Go race tests (ubuntu-22.04) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:01:40 +08:00

909 lines
30 KiB
Go

// Copyright 2024 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package nbs
import (
"bufio"
"context"
"crypto/sha512"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"math/bits"
"os"
"sync/atomic"
"github.com/dolthub/gozstd"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/dolthub/dolt/go/libraries/doltcore/dconfig"
"github.com/dolthub/dolt/go/libraries/utils/dynassert"
"github.com/dolthub/dolt/go/store/chunks"
"github.com/dolthub/dolt/go/store/hash"
)
// reconstructHashFromPrefixAndSuffix creates a hash from a prefix and suffix
func reconstructHashFromPrefixAndSuffix(prefix uint64, suffix [hash.SuffixLen]byte) hash.Hash {
var h hash.Hash
binary.BigEndian.PutUint64(h[:hash.PrefixLen], prefix)
copy(h[hash.PrefixLen:], suffix[:])
return h
}
// archiveReader is a reader for the archive format. We use primitive type slices where possible. These are read directly
// from disk into memory for speed. The downside is complexity on the read path, but it's all constant time.
type archiveReader struct {
reader tableReaderAt
indexReader archiveIndexReader // Memory-mapped or fallback index reader
dictCache *lru.TwoQueueCache[uint32, *DecompBundle]
footer archiveFooter
}
type suffix [hash.SuffixLen]byte
type archiveFooter struct {
fileSignature string
indexSize uint64
fileSize uint64 // Not actually part of the footer, but necessary for calculating offsets.
byteSpanCount uint32
chunkCount uint32
metadataSize uint32
dataCheckSum sha512Sum
indexCheckSum sha512Sum
metaCheckSum sha512Sum
hash hash.Hash
formatVersion byte
}
// actualFooterSize returns the footer size, in bytes for a specific archive. Due to the evolution of the archive format,
// the footer size expanded in format version 3, so we need to calculate the footer size when calculating offsets
// for this instance.
func (f *archiveFooter) actualFooterSize() uint64 {
if f.formatVersion < archiveVersionGiantIndexSupport {
// Version 1 and 2 archives have a smaller footer.
return archiveFooterSize - 4
}
return archiveFooterSize
}
// dataSpan returns the span of the data section of the archive. This is used during conjoin.
func (f *archiveFooter) dataSpan() byteSpan {
return byteSpan{offset: 0, length: f.fileSize - f.actualFooterSize() - uint64(f.metadataSize) - uint64(f.indexSize)}
}
// totalIndexSpan returns the span of the entire index section of the archive.
func (f *archiveFooter) totalIndexSpan() byteSpan {
return byteSpan{offset: f.fileSize - f.actualFooterSize() - uint64(f.metadataSize) - uint64(f.indexSize), length: uint64(f.indexSize)}
}
// indexByteOffsetSpan returns the span of the byte offsets section of the index. This is the first part of the index
func (f *archiveFooter) indexByteOffsetSpan() byteSpan {
totalIdx := f.totalIndexSpan()
return byteSpan{offset: totalIdx.offset, length: uint64(f.byteSpanCount) * uint64Size}
}
// indexPrefixSpan returns the span of the prefix section of the index. This is the second part of the index.
func (f *archiveFooter) indexPrefixSpan() byteSpan {
// Prefix starts after the byte spans. Length is uint64 * chunk count.
offs := f.indexByteOffsetSpan()
return byteSpan{offs.offset + offs.length, uint64(f.chunkCount) * uint64Size}
}
// indexChunkRefSpan returns the span of the chunk reference section of the index. This is the third part of the index.
func (f *archiveFooter) indexChunkRefSpan() byteSpan {
// chunk refs starts after the prefix. Length is (uint32 + uint32) * chunk count.
prefixes := f.indexPrefixSpan()
chLen := uint64(f.chunkCount) * (uint32Size + uint32Size)
return byteSpan{prefixes.offset + prefixes.length, chLen}
}
// indexSuffixSpan returns the span of the suffix section of the index. This is the fourth part of the index.
func (f *archiveFooter) indexSuffixSpan() byteSpan {
suffixLen := uint64(f.chunkCount) * hash.SuffixLen
chunkRefs := f.indexChunkRefSpan()
return byteSpan{chunkRefs.offset + chunkRefs.length, suffixLen}
}
// metadataSpan returns the span of the metadata section of the archive.
func (f *archiveFooter) metadataSpan() byteSpan {
return byteSpan{offset: f.fileSize - f.actualFooterSize() - uint64(f.metadataSize), length: uint64(f.metadataSize)}
}
func newArchiveMetadata(ctx context.Context, reader tableReaderAt, name hash.Hash, fileSize uint64, q MemoryQuotaProvider, stats *Stats) (*ArchiveMetadata, error) {
aRdr, err := newArchiveReader(ctx, reader, name, fileSize, q, stats)
if err != nil {
return nil, err
}
if aRdr.footer.formatVersion > archiveFormatVersionMax {
return nil, ErrInvalidFormatVersion
}
metaSpan := aRdr.footer.metadataSpan()
metaRdr := newSectionReader(ctx, reader, int64(metaSpan.offset), int64(metaSpan.length), stats)
// Read the data into a byte slice
metaData := make([]byte, metaSpan.length)
_, err = metaRdr.Read(metaData)
if err != nil {
return nil, err
}
var result map[string]string
// Unmarshal the JSON data into the map. TODO - use json tags.
err = json.Unmarshal(metaData, &result)
if err != nil {
return nil, err
}
snappyChunks := 0
snappyBytes := uint64(0)
zStdChunks := 0
zStdBytes := uint64(0)
seenDictIds := map[uint32]bool{}
dictionaryBytes := uint64(0)
idx := 0
for idx < int(aRdr.footer.chunkCount) {
dictId, dataId := aRdr.getChunkRef(idx)
bs := aRdr.getByteSpanByID(dataId)
if dictId != 0 {
zStdBytes += bs.length
zStdChunks += 1
if !seenDictIds[dictId] {
seenDictIds[dictId] = true
bs := aRdr.getByteSpanByID(dictId)
dictionaryBytes += bs.length
}
} else {
snappyBytes += bs.length
snappyChunks += 1
}
idx += 1
}
return &ArchiveMetadata{
formatVersion: int(aRdr.footer.formatVersion),
snappyChunkCount: snappyChunks,
snappyBytes: snappyBytes,
zStdChunkCount: zStdChunks,
zStdBytes: zStdBytes,
dictionaryCount: len(seenDictIds),
dictionaryBytes: dictionaryBytes,
originalTableFileId: result[amdkOriginTableFile],
}, nil
}
func newArchiveReaderFromFooter(ctx context.Context, reader tableReaderAt, name hash.Hash, fileSz uint64, footer []byte, q MemoryQuotaProvider, stats *Stats) (archiveReader, error) {
if uint64(len(footer)) != archiveFooterSize {
return archiveReader{}, errors.New("runtime error: invalid footer.")
}
ftr, err := buildArchiveFooter(name, fileSz, footer)
if err != nil {
return archiveReader{}, err
}
return buildArchiveReader(ctx, reader, ftr, q, stats)
}
func newArchiveReader(ctx context.Context, reader tableReaderAt, name hash.Hash, fileSize uint64, q MemoryQuotaProvider, stats *Stats) (archiveReader, error) {
footer, err := loadFooter(ctx, reader, name, fileSize, stats)
if err != nil {
return archiveReader{}, fmt.Errorf("Failed to loadFooter: %w", err)
}
return buildArchiveReader(ctx, reader, footer, q, stats)
}
func buildArchiveReader(ctx context.Context, reader tableReaderAt, footer archiveFooter, q MemoryQuotaProvider, stats *Stats) (archiveReader, error) {
dictCache, err := lru.New2Q[uint32, *DecompBundle](256)
if err != nil {
return archiveReader{}, err
}
var indexRdr archiveIndexReader
// Try to use memory mapping if the reader is a file
if fileReader, ok := reader.(*fileReaderAt); ok && fileReader.mmapIndexes {
indexRdr, err = newMmapIndexReader(fileReader.f, footer)
if err != nil {
return archiveReader{}, err
}
} else {
if _, isSet := os.LookupEnv(dconfig.EnvAssertNoInMemoryArchiveIndex); isSet {
return archiveReader{}, fmt.Errorf("attempted to load archive index into memory but %s was set", dconfig.EnvAssertNoInMemoryArchiveIndex)
}
indexRdr, err = newInMemoryArchiveIndexReader(ctx, reader, footer, q, stats)
if err != nil {
return archiveReader{}, err
}
}
return archiveReader{
reader: reader,
indexReader: indexRdr,
footer: footer,
dictCache: dictCache,
}, nil
}
// newInMemoryArchiveIndexReader implements the original index loading logic for non-file readers
func newInMemoryArchiveIndexReader(ctx context.Context, reader tableReaderAt, footer archiveFooter, q MemoryQuotaProvider, stats *Stats) (archiveIndexReader, error) {
byteOffSpan := footer.indexByteOffsetSpan()
secRdr := newSectionReader(ctx, reader, int64(byteOffSpan.offset), int64(byteOffSpan.length), stats)
byteSpans, err := q.AcquireQuotaUint64Slice(ctx, int(footer.byteSpanCount)+1)
if err != nil {
return nil, fmt.Errorf("Failed to allocate byteSpans uint64 slice: %w", err)
}
bytesSoFar := len(byteSpans) * uint64Size
byteSpans[0] = 0 // Null byteSpan to simplify logic.
err = binary.Read(secRdr, binary.BigEndian, byteSpans[1:])
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to read byte spans: %w", err)
}
prefixSpan := footer.indexPrefixSpan()
prefixRdr := newSectionReader(ctx, reader, int64(prefixSpan.offset), int64(prefixSpan.length), stats)
prefixes, err := q.AcquireQuotaUint64Slice(ctx, int(footer.chunkCount))
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to allocate prefixes uint32 slice: %w", err)
}
bytesSoFar += len(prefixes) * uint64Size
err = binary.Read(prefixRdr, binary.BigEndian, prefixes[:])
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to read prefixes: %w", err)
}
chunkRefSpan := footer.indexChunkRefSpan()
chunkRdr := newSectionReader(ctx, reader, int64(chunkRefSpan.offset), int64(chunkRefSpan.length), stats)
chnks, err := q.AcquireQuotaUint32Slice(ctx, int(footer.chunkCount)*2)
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to allocate chunks uint32 slice: %w", err)
}
bytesSoFar += len(chnks) * uint32Size
err = binary.Read(chunkRdr, binary.BigEndian, chnks[:])
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to read chunk references: %w", err)
}
suffixSpan := footer.indexSuffixSpan()
sufRdr := newSectionReader(ctx, reader, int64(suffixSpan.offset), int64(suffixSpan.length), stats)
suffixes, err := q.AcquireQuotaByteSlice(ctx, int(suffixSpan.length))
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, fmt.Errorf("Failed to allocate suffixes byte slice: %w", err)
}
bytesSoFar += len(suffixes)
_, err = io.ReadFull(sufRdr, suffixes)
if err != nil {
q.ReleaseQuotaBytes(bytesSoFar)
return nil, err
}
ret := &inMemoryArchiveIndexReader{
prefixes: prefixes,
spanIndex: byteSpans,
chunkRefs: chnks,
suffixes: suffixes,
q: q,
}
ret.refCnt.Add(1)
return ret, nil
}
// inMemoryArchiveIndexReader provides the original in-memory index implementation as a fallback
type inMemoryArchiveIndexReader struct {
prefixes []uint64
spanIndex []uint64
chunkRefs []uint32
suffixes []byte
refCnt atomic.Int32
q MemoryQuotaProvider
}
func (f *inMemoryArchiveIndexReader) getNumChunks() uint32 {
return uint32(len(f.prefixes))
}
func (f *inMemoryArchiveIndexReader) getSpanIndex(idx uint32) uint64 {
if idx >= uint32(len(f.spanIndex)) {
return 0
}
return f.spanIndex[idx]
}
func (f *inMemoryArchiveIndexReader) getPrefix(idx uint32) uint64 {
if idx >= uint32(len(f.prefixes)) {
return 0
}
return f.prefixes[idx]
}
func (f *inMemoryArchiveIndexReader) searchPrefix(prefix uint64) int32 {
return int32(prollyBinSearch(f.prefixes, prefix))
}
func (f *inMemoryArchiveIndexReader) getChunkRef(idx uint32) (dict, data uint32) {
if idx < 0 || idx*2+1 >= uint32(len(f.chunkRefs)) {
return 0, 0
}
return f.chunkRefs[idx*2], f.chunkRefs[idx*2+1]
}
func (f *inMemoryArchiveIndexReader) getSuffix(idx uint32) suffix {
if idx >= f.getNumChunks() {
return suffix{}
}
start := uint64(idx) * hash.SuffixLen
return suffix(f.suffixes[start : start+hash.SuffixLen])
}
func (f *inMemoryArchiveIndexReader) clone() (archiveIndexReader, error) {
if !dynassert.Assert(f.refCnt.Add(1) > 1, "attempt to clone a closed inMemoryArchiveIndexReader") {
// Restore previous refcnt, despite being in a weird state...
f.refCnt.Add(-1)
// Just return ourselves in this weird state I guess...
return f, nil
}
return f, nil
}
func (f *inMemoryArchiveIndexReader) Close() error {
cnt := f.refCnt.Add(-1)
dynassert.Assert(cnt >= 0, "invalid cnt on inMemoryArchiveIndexReader. closed more times than cloned?")
// No need to restore count which was over closed. We already incorrectly
// released the bytes and can't necessarily do anything about it.
if cnt == 0 {
numBytes := len(f.chunkRefs)*uint32Size + len(f.prefixes)*uint64Size + len(f.spanIndex)*uint64Size + len(f.suffixes)
f.q.ReleaseQuotaBytes(numBytes)
}
return nil
}
// clone returns a new archiveReader with a new (provided) reader. All other fields are immutable or thread safe,
// so they are copied.
func (ar *archiveReader) clone() (archiveReader, error) {
reader, err := ar.reader.clone()
if err != nil {
return archiveReader{}, err
}
indexReader, err := ar.indexReader.clone()
if err != nil {
return archiveReader{}, err
}
return archiveReader{
reader: reader,
indexReader: indexReader,
footer: ar.footer,
dictCache: ar.dictCache, // cache is thread safe.
}, nil
}
type readerAtWithStatsBridge struct {
reader ReaderAtWithStats
ctx context.Context
stats *Stats
}
func (r readerAtWithStatsBridge) ReadAt(p []byte, off int64) (int, error) {
return r.reader.ReadAtWithStats(r.ctx, p, off, r.stats)
}
func newSectionReader(ctx context.Context, rd ReaderAtWithStats, off, len int64, stats *Stats) *io.SectionReader {
return io.NewSectionReader(readerAtWithStatsBridge{rd, ctx, stats}, off, len)
}
func loadFooter(ctx context.Context, reader ReaderAtWithStats, name hash.Hash, fileSize uint64, stats *Stats) (f archiveFooter, err error) {
section := newSectionReader(ctx, reader, int64(fileSize-archiveFooterSize), int64(archiveFooterSize), stats)
buf := make([]byte, archiveFooterSize)
_, err = io.ReadFull(section, buf)
if err != nil {
return
}
return buildArchiveFooter(name, fileSize, buf)
}
func buildArchiveFooter(name hash.Hash, fileSize uint64, buf []byte) (f archiveFooter, err error) {
f.formatVersion = buf[afrVersionOffset]
f.fileSignature = string(buf[afrSigOffset:])
// Verify File Signature
if f.fileSignature != archiveFileSignature {
err = ErrInvalidFileSignature
return
}
// Verify Format Version. 1,2,3 supported.
if f.formatVersion > archiveFormatVersionMax {
err = ErrInvalidFormatVersion
return
}
smallFooter := false
if f.formatVersion < archiveVersionGiantIndexSupport {
smallFooter = true
}
if smallFooter {
// Version 1 and 2 archives have a smaller footer. Ignore the first 4 bytes.
if afrIndexLenOffset != 0 {
// Future proofing for the event where we need to extend the footer with additional fields. This is intended
// to blow up in development if we try and change it.
panic("runtime error: afrIndexChkSumOffset must be 0.")
}
f.indexSize = uint64(binary.BigEndian.Uint32(buf[4 : 4+uint32Size]))
} else {
f.indexSize = binary.BigEndian.Uint64(buf[afrIndexLenOffset : afrIndexLenOffset+uint64Size])
}
f.byteSpanCount = binary.BigEndian.Uint32(buf[afrByteSpanOffset : afrByteSpanOffset+uint32Size])
f.chunkCount = binary.BigEndian.Uint32(buf[afrChunkCountOffset : afrChunkCountOffset+uint32Size])
f.metadataSize = binary.BigEndian.Uint32(buf[afrMetaLenOffset : afrMetaLenOffset+uint32Size])
f.dataCheckSum = sha512Sum(buf[afrDataChkSumOffset : afrDataChkSumOffset+sha512.Size])
f.indexCheckSum = sha512Sum(buf[afrIndexChkSumOffset : afrIndexChkSumOffset+sha512.Size])
f.metaCheckSum = sha512Sum(buf[afrMetaChkSumOffset : afrMetaChkSumOffset+sha512.Size])
f.fileSize = fileSize
f.hash = name
return
}
// search returns the index of the hash in the archive. If the hash is not found, -1 is returned.
func (ar *archiveReader) search(hash hash.Hash) int {
prefix := hash.Prefix()
possibleMatch := ar.indexReader.searchPrefix(prefix)
targetSfx := hash.Suffix()
if possibleMatch < 0 || uint32(possibleMatch) >= ar.footer.chunkCount {
return -1
}
for idx := uint32(possibleMatch); idx < ar.footer.chunkCount && ar.indexReader.getPrefix(idx) == prefix; idx++ {
if ar.indexReader.getSuffix(idx) == suffix(targetSfx) {
return int(idx)
}
}
return -1
}
func (ar *archiveReader) has(hash hash.Hash) bool {
return ar.search(hash) >= 0
}
// get returns the decompressed data for the given hash. If the hash is not found, nil is returned (not an error)
func (ar *archiveReader) get(ctx context.Context, hash hash.Hash, stats *Stats) ([]byte, error) {
dict, data, err := ar.getRaw(ctx, hash, stats)
if err != nil || data == nil {
return nil, err
}
if dict == nil {
if ar.footer.formatVersion >= archiveVersionSnappySupport {
// Snappy compression format. The data is compressed with a checksum at the end.
cc, err := NewCompressedChunk(hash, data)
if err != nil {
return nil, err
}
chk, err := cc.ToChunk()
if err != nil {
return nil, err
}
return chk.Data(), nil
}
return nil, errors.New("runtime error: unable to get archived chunk. dictionary is nil")
}
var result []byte
result, err = gozstd.DecompressDict(nil, data, dict.dDict)
if err != nil {
return nil, err
}
return result, nil
}
// getAsToChunker returns the chunk which is has not been decompressed. Similar to get, but with a different return type.
// If the hash is not found, a ToChunker instance with IsEmpty() == true is returned (no error)
func (ar *archiveReader) getAsToChunker(ctx context.Context, h hash.Hash, stats *Stats) (ToChunker, error) {
dict, data, err := ar.getRaw(ctx, h, stats)
if err != nil {
return nil, err
}
if data == nil {
return CompressedChunk{}, nil
}
if dict == nil {
if ar.footer.formatVersion >= archiveVersionSnappySupport {
cc, err := NewCompressedChunk(h, data)
if err != nil {
return nil, err
}
return cc, nil
}
return nil, errors.New("runtime error: unable to get archived chunk. dictionary is nil")
}
return &ArchiveToChunker{dict, data, h}, nil
}
func (ar *archiveReader) count() uint32 {
return ar.footer.chunkCount
}
func (ar *archiveReader) close() error {
// Join the errors so the reader still closes and releases its file when the index close fails.
return errors.Join(ar.indexReader.Close(), ar.reader.Close())
}
// readByteSpan reads the byte span from the archive. This allocates a new byte slice and returns it to the caller.
func (ar *archiveReader) readByteSpan(ctx context.Context, bs byteSpan, stats *Stats) ([]byte, error) {
buff := make([]byte, bs.length)
_, err := ar.reader.ReadAtWithStats(ctx, buff[:], int64(bs.offset), stats)
if err != nil {
return nil, err
}
return buff, nil
}
// getRaw returns the raw data for the given hash. If the hash is not found, nil is returned for both output, and no error.
//
// The data is returned still compressed:
// Format Version 1: Only zStd compression is supported. The data returned requires the dictionary to be decompressed.
// Format Version 2: The compression format of the data is:
// - zStd when a dictionary is returned. The data is decompressed with the dictionary.
// - Snappy compression when no dictionary is returned. The data has a checksum 32 bit checksum at the end. This
// format matches the noms format.
func (ar *archiveReader) getRaw(ctx context.Context, hash hash.Hash, stats *Stats) (dict *DecompBundle, data []byte, err error) {
idx := ar.search(hash)
if idx < 0 {
return nil, nil, nil
}
dictId, dataId := ar.getChunkRef(idx)
if dictId != 0 {
if cached, cacheHit := ar.dictCache.Get(dictId); cacheHit {
dict = cached
} else {
byteSpan := ar.getByteSpanByID(dictId)
dictBytes, err := ar.readByteSpan(ctx, byteSpan, stats)
if err != nil {
return nil, nil, err
}
dict, err = NewDecompBundle(dictBytes)
if err != nil {
return nil, nil, err
}
ar.dictCache.Add(dictId, dict)
}
}
byteSpan := ar.getByteSpanByID(dataId)
data, err = ar.readByteSpan(ctx, byteSpan, stats)
if err != nil {
return nil, nil, err
}
return
}
// getChunkRef returns the dictionary and data references for the chunk at the given index. Assumes good input!
func (ar *archiveReader) getChunkRef(idx int) (dict, data uint32) {
return ar.indexReader.getChunkRef(uint32(idx))
}
// getByteSpanByID returns the byte span for the chunk at the given index. Assumes good input!
func (ar *archiveReader) getByteSpanByID(id uint32) byteSpan {
if id == 0 {
return byteSpan{}
}
// This works because spanIndex[0] == 0. See initialization.
offset := ar.indexReader.getSpanIndex(id - 1)
length := ar.indexReader.getSpanIndex(id) - offset
return byteSpan{offset: offset, length: length}
}
// getSuffixByID returns the suffix for the chunk at the given index. Assumes good input!
func (ar *archiveReader) getSuffixByID(id uint64) suffix {
return ar.indexReader.getSuffix(uint32(id))
}
func (ar *archiveReader) getMetadata(ctx context.Context, stats *Stats) ([]byte, error) {
return ar.readByteSpan(ctx, ar.footer.metadataSpan(), stats)
}
type bridgeReaderAt struct {
rdr ReaderAtWithStats
ctx context.Context
stats *Stats
}
func (r *bridgeReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
return r.rdr.ReadAtWithStats(r.ctx, p, off, r.stats)
}
func (ar *archiveReader) iterate(ctx context.Context, cb func(chunks.Chunk) error, stats *Stats) error {
// Build reverse indexes for dictionary and data ByteSpans
// dictReverseIndex: Dictionary ByteSpan ID -> struct{} - indicates that we expect that span to be a dictionary.
// dataReverseIndex: Data ByteSpan ID -> chunk ref index - indicates that we expect that span to be a data chunk,
// and the value is the index into the chunkRefs slice where the chunk reference is stored.
dictReverseIndex := make(map[uint32]struct{})
dataReverseIndex := make(map[uint32]uint32)
for chunkRefIdx := uint32(0); chunkRefIdx < ar.footer.chunkCount; chunkRefIdx++ {
dictId, dataId := ar.getChunkRef(int(chunkRefIdx))
if dictId != 0 {
dictReverseIndex[dictId] = struct{}{}
}
dataReverseIndex[dataId] = chunkRefIdx
}
dataSpan := ar.footer.dataSpan()
dataReader := io.NewSectionReader(&bridgeReaderAt{
rdr: ar.reader,
ctx: ctx,
stats: stats,
}, int64(dataSpan.offset), int64(dataSpan.length))
bufReader := bufio.NewReader(dataReader)
byteSpanCounter := uint32(1)
buf := make([]byte, 4*1024*1024)
loadedDictionaries := make(map[uint32]*gozstd.DDict)
for byteSpanCounter <= ar.footer.byteSpanCount {
if ctx.Err() != nil {
return context.Cause(ctx)
}
span := ar.getByteSpanByID(byteSpanCounter)
for cap(buf) < int(span.length) {
buf = append(buf, make([]byte, cap(buf))...)
}
_, err := io.ReadFull(bufReader, buf[:span.length])
if err != nil {
return fmt.Errorf("error reading archive file: %w", err)
}
spanData := buf[:span.length]
if _, exists := dictReverseIndex[byteSpanCounter]; exists {
dict, err := NewDecompBundle(spanData)
if err != nil {
return fmt.Errorf("Failure creating dictionary from bytes: %w", err)
}
loadedDictionaries[byteSpanCounter] = dict.dDict
} else if chunkId, exists := dataReverseIndex[byteSpanCounter]; exists {
dictId, dataId := ar.getChunkRef(int(chunkId))
if byteSpanCounter != dataId {
panic("Reverse Index incorrect: ByteSpan ID does not match data ID in chunk reference")
}
// Reconstruct the hash for this chunk
prefix := ar.indexReader.getPrefix(chunkId)
suffix := ar.indexReader.getSuffix(chunkId)
h := reconstructHashFromPrefixAndSuffix(prefix, suffix)
var chunkData []byte
if dictId == 0 {
// Snappy compression (no dictionary)
if ar.footer.formatVersion >= archiveVersionSnappySupport {
cc, err := NewCompressedChunk(h, spanData)
if err != nil {
return err
}
chk, err := cc.ToChunk()
if err != nil {
return err
}
chunkData = chk.Data()
} else {
return errors.New("runtime error: no dictionary for old format version")
}
} else {
dict, ok := loadedDictionaries[dictId]
if !ok {
panic("Reverse Index incomplete: Dictionary ID not found in loaded dictionaries")
}
chunkData, err = gozstd.DecompressDict(nil, spanData, dict)
if err != nil {
return fmt.Errorf("error decompressing span: %d, %v, %w", byteSpanCounter, span, err)
}
}
chk := chunks.NewChunkWithHash(h, chunkData)
err = cb(chk)
if err != nil {
return err
}
} else {
panic("Reverse Index incomplete: ByteSpan ID not found in either dictionary or data reverse index")
}
byteSpanCounter++
}
return nil
}
func (ar *archiveReader) tolerantIterate(ctx context.Context, cb func(chunks.Chunk), errCb func(error), stats *Stats) {
dictReverseIndex := make(map[uint32]struct{})
dataReverseIndex := make(map[uint32]uint32)
for chunkRefIdx := uint32(0); chunkRefIdx < ar.footer.chunkCount; chunkRefIdx++ {
dictId, dataId := ar.getChunkRef(int(chunkRefIdx))
if dictId != 0 {
dictReverseIndex[dictId] = struct{}{}
}
dataReverseIndex[dataId] = chunkRefIdx
}
dataSpan := ar.footer.dataSpan()
dataReader := io.NewSectionReader(&bridgeReaderAt{
rdr: ar.reader,
ctx: ctx,
stats: stats,
}, int64(dataSpan.offset), int64(dataSpan.length))
bufReader := bufio.NewReader(dataReader)
byteSpanCounter := uint32(1)
buf := make([]byte, 4*1024*1024)
loadedDictionaries := make(map[uint32]*gozstd.DDict)
failedDictionaries := make(map[uint32]struct{})
for byteSpanCounter <= ar.footer.byteSpanCount {
if ctx.Err() != nil {
return
}
span := ar.getByteSpanByID(byteSpanCounter)
for cap(buf) < int(span.length) {
buf = append(buf, make([]byte, cap(buf))...)
}
_, readErr := io.ReadFull(bufReader, buf[:span.length])
if readErr != nil {
// Stream position is unknown; cannot safely continue sequential read.
errCb(fmt.Errorf("error reading archive file at span %d: %w", byteSpanCounter, readErr))
return
}
spanData := buf[:span.length]
if _, exists := dictReverseIndex[byteSpanCounter]; exists {
dict, err := NewDecompBundle(spanData)
if err != nil {
errCb(fmt.Errorf("failure loading archive dictionary span %d: %w", byteSpanCounter, err))
failedDictionaries[byteSpanCounter] = struct{}{}
} else {
loadedDictionaries[byteSpanCounter] = dict.dDict
}
} else if chunkId, exists := dataReverseIndex[byteSpanCounter]; exists {
dictId, dataId := ar.getChunkRef(int(chunkId))
if byteSpanCounter != dataId {
panic("Reverse Index incorrect: ByteSpan ID does not match data ID in chunk reference")
}
prefix := ar.indexReader.getPrefix(chunkId)
suffix := ar.indexReader.getSuffix(chunkId)
h := reconstructHashFromPrefixAndSuffix(prefix, suffix)
var chunkData []byte
chunkOk := true
if dictId == 0 {
if ar.footer.formatVersion >= archiveVersionSnappySupport {
cc, err := NewCompressedChunk(h, spanData)
if err != nil {
errCb(fmt.Errorf("chunk %s: %w", h.String(), err))
chunkOk = false
} else {
chk, err := cc.ToChunk()
if err != nil {
errCb(fmt.Errorf("chunk %s: decompress error: %w", h.String(), err))
chunkOk = false
} else {
chunkData = chk.Data()
}
}
} else {
errCb(fmt.Errorf("chunk %s: no dictionary for old format version", h.String()))
chunkOk = false
}
} else {
if _, failed := failedDictionaries[dictId]; failed {
errCb(fmt.Errorf("chunk %s: skipped due to failed dictionary span %d", h.String(), dictId))
chunkOk = false
} else {
dict, ok := loadedDictionaries[dictId]
if !ok {
errCb(fmt.Errorf("chunk %s: dictionary span %d not loaded", h.String(), dictId))
chunkOk = false
} else {
var decompErr error
chunkData, decompErr = gozstd.DecompressDict(nil, spanData, dict)
if decompErr != nil {
errCb(fmt.Errorf("chunk %s: decompression error: %w", h.String(), decompErr))
chunkOk = false
}
}
}
}
if chunkOk {
cb(chunks.NewChunkWithHash(h, chunkData))
}
} else {
errCb(fmt.Errorf("archive span %d not found in dictionary or data reverse index", byteSpanCounter))
}
byteSpanCounter++
}
}
// prollyBinSearch is a search that returns the _best_ index of the target in the input slice. If the target exists,
// one or more times, the index of the first instance is returned. If the target does not exist, the index which it
// would be inserted at is returned.
//
// A strong requirement for the proper behavior of this function is to have a sorted and well distributed slice where the
// values are not dense. Crypto hashes are a good example of this.
//
// For our purposes where we are just trying to get the index, we must compare the resulting index to our target to
// determine if it is a match.
func prollyBinSearch(slice []uint64, target uint64) int {
items := len(slice)
if items == 0 {
return 0
}
lft, rht := 0, items
lo, hi := slice[lft], slice[rht-1]
if target > hi {
return rht
}
if lo >= target {
return lft
}
for lft < rht {
valRangeSz := hi - lo
idxRangeSz := uint64(rht - lft - 1)
shiftedTgt := target - lo
mhi, mlo := bits.Mul64(shiftedTgt, idxRangeSz)
dU64, _ := bits.Div64(mhi, mlo, valRangeSz)
idx := int(dU64) + lft
if slice[idx] < target {
lft = idx + 1
// No need to update lo if i == items, since this loop will be ending.
if lft < items {
lo = slice[lft]
// Interpolation doesn't like lo >= target, so if we're already there, just return |i|.
if lo >= target {
return lft
}
}
} else {
rht = idx
hi = slice[rht]
}
}
return lft
}