Line data Source code
1 : // Copyright 2011 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package sstable
6 :
7 : import (
8 : "bytes"
9 : "cmp"
10 : "context"
11 : "encoding/binary"
12 : "io"
13 : "os"
14 : "slices"
15 : "time"
16 :
17 : "github.com/cespare/xxhash/v2"
18 : "github.com/cockroachdb/errors"
19 : "github.com/cockroachdb/pebble/internal/base"
20 : "github.com/cockroachdb/pebble/internal/bytealloc"
21 : "github.com/cockroachdb/pebble/internal/cache"
22 : "github.com/cockroachdb/pebble/internal/crc"
23 : "github.com/cockroachdb/pebble/internal/invariants"
24 : "github.com/cockroachdb/pebble/internal/keyspan"
25 : "github.com/cockroachdb/pebble/internal/private"
26 : "github.com/cockroachdb/pebble/objstorage"
27 : "github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
28 : )
29 :
30 : var errCorruptIndexEntry = base.CorruptionErrorf("pebble/table: corrupt index entry")
31 : var errReaderClosed = errors.New("pebble/table: reader is closed")
32 :
33 : // decodeBlockHandle returns the block handle encoded at the start of src, as
34 : // well as the number of bytes it occupies. It returns zero if given invalid
35 : // input. A block handle for a data block or a first/lower level index block
36 : // should not be decoded using decodeBlockHandle since the caller may validate
37 : // that the number of bytes decoded is equal to the length of src, which will
38 : // be false if the properties are not decoded. In those cases the caller
39 : // should use decodeBlockHandleWithProperties.
40 2 : func decodeBlockHandle(src []byte) (BlockHandle, int) {
41 2 : offset, n := binary.Uvarint(src)
42 2 : length, m := binary.Uvarint(src[n:])
43 2 : if n == 0 || m == 0 {
44 0 : return BlockHandle{}, 0
45 0 : }
46 2 : return BlockHandle{offset, length}, n + m
47 : }
48 :
49 : // decodeBlockHandleWithProperties returns the block handle and properties
50 : // encoded in src. src needs to be exactly the length that was encoded. This
51 : // method must be used for data block and first/lower level index blocks. The
52 : // properties in the block handle point to the bytes in src.
53 2 : func decodeBlockHandleWithProperties(src []byte) (BlockHandleWithProperties, error) {
54 2 : bh, n := decodeBlockHandle(src)
55 2 : if n == 0 {
56 0 : return BlockHandleWithProperties{}, errors.Errorf("invalid BlockHandle")
57 0 : }
58 2 : return BlockHandleWithProperties{
59 2 : BlockHandle: bh,
60 2 : Props: src[n:],
61 2 : }, nil
62 : }
63 :
64 2 : func encodeBlockHandle(dst []byte, b BlockHandle) int {
65 2 : n := binary.PutUvarint(dst, b.Offset)
66 2 : m := binary.PutUvarint(dst[n:], b.Length)
67 2 : return n + m
68 2 : }
69 :
70 2 : func encodeBlockHandleWithProperties(dst []byte, b BlockHandleWithProperties) []byte {
71 2 : n := encodeBlockHandle(dst, b.BlockHandle)
72 2 : dst = append(dst[:n], b.Props...)
73 2 : return dst
74 2 : }
75 :
76 : // block is a []byte that holds a sequence of key/value pairs plus an index
77 : // over those pairs.
78 : type block []byte
79 :
80 : type loadBlockResult int8
81 :
82 : const (
83 : loadBlockOK loadBlockResult = iota
84 : // Could be due to error or because no block left to load.
85 : loadBlockFailed
86 : loadBlockIrrelevant
87 : )
88 :
89 : type blockTransform func([]byte) ([]byte, error)
90 :
91 : // ReaderOption provide an interface to do work on Reader while it is being
92 : // opened.
93 : type ReaderOption interface {
94 : // readerApply is called on the reader during opening in order to set internal
95 : // parameters.
96 : readerApply(*Reader)
97 : }
98 :
99 : // Comparers is a map from comparer name to comparer. It is used for debugging
100 : // tools which may be used on multiple databases configured with different
101 : // comparers. Comparers implements the OpenOption interface and can be passed
102 : // as a parameter to NewReader.
103 : type Comparers map[string]*Comparer
104 :
105 1 : func (c Comparers) readerApply(r *Reader) {
106 1 : if r.Compare != nil || r.Properties.ComparerName == "" {
107 1 : return
108 1 : }
109 1 : if comparer, ok := c[r.Properties.ComparerName]; ok {
110 1 : r.Compare = comparer.Compare
111 1 : r.FormatKey = comparer.FormatKey
112 1 : r.Split = comparer.Split
113 1 : }
114 : }
115 :
116 : // Mergers is a map from merger name to merger. It is used for debugging tools
117 : // which may be used on multiple databases configured with different
118 : // mergers. Mergers implements the OpenOption interface and can be passed as
119 : // a parameter to NewReader.
120 : type Mergers map[string]*Merger
121 :
122 1 : func (m Mergers) readerApply(r *Reader) {
123 1 : if r.mergerOK || r.Properties.MergerName == "" {
124 1 : return
125 1 : }
126 1 : _, r.mergerOK = m[r.Properties.MergerName]
127 : }
128 :
129 : // cacheOpts is a Reader open option for specifying the cache ID and sstable file
130 : // number. If not specified, a unique cache ID will be used.
131 : type cacheOpts struct {
132 : cacheID uint64
133 : fileNum base.DiskFileNum
134 : }
135 :
136 : // Marker function to indicate the option should be applied before reading the
137 : // sstable properties and, in the write path, before writing the default
138 : // sstable properties.
139 0 : func (c *cacheOpts) preApply() {}
140 :
141 2 : func (c *cacheOpts) readerApply(r *Reader) {
142 2 : if r.cacheID == 0 {
143 2 : r.cacheID = c.cacheID
144 2 : }
145 2 : if r.fileNum.FileNum() == 0 {
146 2 : r.fileNum = c.fileNum
147 2 : }
148 : }
149 :
150 2 : func (c *cacheOpts) writerApply(w *Writer) {
151 2 : if w.cacheID == 0 {
152 2 : w.cacheID = c.cacheID
153 2 : }
154 2 : if w.fileNum.FileNum() == 0 {
155 2 : w.fileNum = c.fileNum
156 2 : }
157 : }
158 :
159 : // rawTombstonesOpt is a Reader open option for specifying that range
160 : // tombstones returned by Reader.NewRangeDelIter() should not be
161 : // fragmented. Used by debug tools to get a raw view of the tombstones
162 : // contained in an sstable.
163 : type rawTombstonesOpt struct{}
164 :
165 0 : func (rawTombstonesOpt) preApply() {}
166 :
167 1 : func (rawTombstonesOpt) readerApply(r *Reader) {
168 1 : r.rawTombstones = true
169 1 : }
170 :
171 2 : func init() {
172 2 : private.SSTableCacheOpts = func(cacheID uint64, fileNum base.DiskFileNum) interface{} {
173 2 : return &cacheOpts{cacheID, fileNum}
174 2 : }
175 2 : private.SSTableRawTombstonesOpt = rawTombstonesOpt{}
176 : }
177 :
178 : // CommonReader abstracts functionality over a Reader or a VirtualReader. This
179 : // can be used by code which doesn't care to distinguish between a reader and a
180 : // virtual reader.
181 : type CommonReader interface {
182 : NewRawRangeKeyIter() (keyspan.FragmentIterator, error)
183 : NewRawRangeDelIter() (keyspan.FragmentIterator, error)
184 : NewIterWithBlockPropertyFiltersAndContextEtc(
185 : ctx context.Context, lower, upper []byte,
186 : filterer *BlockPropertiesFilterer,
187 : hideObsoletePoints, useFilterBlock bool,
188 : stats *base.InternalIteratorStats,
189 : categoryAndQoS CategoryAndQoS,
190 : statsCollector *CategoryStatsCollector,
191 : rp ReaderProvider,
192 : ) (Iterator, error)
193 : NewCompactionIter(
194 : bytesIterated *uint64,
195 : categoryAndQoS CategoryAndQoS,
196 : statsCollector *CategoryStatsCollector,
197 : rp ReaderProvider,
198 : bufferPool *BufferPool,
199 : ) (Iterator, error)
200 : EstimateDiskUsage(start, end []byte) (uint64, error)
201 : CommonProperties() *CommonProperties
202 : }
203 :
204 : // Reader is a table reader.
205 : type Reader struct {
206 : readable objstorage.Readable
207 : cacheID uint64
208 : fileNum base.DiskFileNum
209 : err error
210 : indexBH BlockHandle
211 : filterBH BlockHandle
212 : rangeDelBH BlockHandle
213 : rangeKeyBH BlockHandle
214 : rangeDelTransform blockTransform
215 : valueBIH valueBlocksIndexHandle
216 : propertiesBH BlockHandle
217 : metaIndexBH BlockHandle
218 : footerBH BlockHandle
219 : opts ReaderOptions
220 : Compare Compare
221 : FormatKey base.FormatKey
222 : Split Split
223 : tableFilter *tableFilterReader
224 : // Keep types that are not multiples of 8 bytes at the end and with
225 : // decreasing size.
226 : Properties Properties
227 : tableFormat TableFormat
228 : rawTombstones bool
229 : mergerOK bool
230 : checksumType ChecksumType
231 : // metaBufferPool is a buffer pool used exclusively when opening a table and
232 : // loading its meta blocks. metaBufferPoolAlloc is used to batch-allocate
233 : // the BufferPool.pool slice as a part of the Reader allocation. It's
234 : // capacity 3 to accommodate the meta block (1), and both the compressed
235 : // properties block (1) and decompressed properties block (1)
236 : // simultaneously.
237 : metaBufferPool BufferPool
238 : metaBufferPoolAlloc [3]allocedBuffer
239 : }
240 :
241 : // Close implements DB.Close, as documented in the pebble package.
242 2 : func (r *Reader) Close() error {
243 2 : r.opts.Cache.Unref()
244 2 :
245 2 : if r.readable != nil {
246 2 : r.err = firstError(r.err, r.readable.Close())
247 2 : r.readable = nil
248 2 : }
249 :
250 2 : if r.err != nil {
251 1 : return r.err
252 1 : }
253 : // Make any future calls to Get, NewIter or Close return an error.
254 2 : r.err = errReaderClosed
255 2 : return nil
256 : }
257 :
258 : // NewIterWithBlockPropertyFilters returns an iterator for the contents of the
259 : // table. If an error occurs, NewIterWithBlockPropertyFilters cleans up after
260 : // itself and returns a nil iterator.
261 : func (r *Reader) NewIterWithBlockPropertyFilters(
262 : lower, upper []byte,
263 : filterer *BlockPropertiesFilterer,
264 : useFilterBlock bool,
265 : stats *base.InternalIteratorStats,
266 : categoryAndQoS CategoryAndQoS,
267 : statsCollector *CategoryStatsCollector,
268 : rp ReaderProvider,
269 2 : ) (Iterator, error) {
270 2 : return r.newIterWithBlockPropertyFiltersAndContext(
271 2 : context.Background(), lower, upper, filterer, false, useFilterBlock, stats,
272 2 : categoryAndQoS, statsCollector, rp, nil)
273 2 : }
274 :
275 : // NewIterWithBlockPropertyFiltersAndContextEtc is similar to
276 : // NewIterWithBlockPropertyFilters and additionally accepts a context for
277 : // tracing.
278 : //
279 : // If hideObsoletePoints, the callee assumes that filterer already includes
280 : // obsoleteKeyBlockPropertyFilter. The caller can satisfy this contract by
281 : // first calling TryAddBlockPropertyFilterForHideObsoletePoints.
282 : func (r *Reader) NewIterWithBlockPropertyFiltersAndContextEtc(
283 : ctx context.Context,
284 : lower, upper []byte,
285 : filterer *BlockPropertiesFilterer,
286 : hideObsoletePoints, useFilterBlock bool,
287 : stats *base.InternalIteratorStats,
288 : categoryAndQoS CategoryAndQoS,
289 : statsCollector *CategoryStatsCollector,
290 : rp ReaderProvider,
291 2 : ) (Iterator, error) {
292 2 : return r.newIterWithBlockPropertyFiltersAndContext(
293 2 : ctx, lower, upper, filterer, hideObsoletePoints, useFilterBlock, stats, categoryAndQoS,
294 2 : statsCollector, rp, nil)
295 2 : }
296 :
297 : // TryAddBlockPropertyFilterForHideObsoletePoints is expected to be called
298 : // before the call to NewIterWithBlockPropertyFiltersAndContextEtc, to get the
299 : // value of hideObsoletePoints and potentially add a block property filter.
300 : func (r *Reader) TryAddBlockPropertyFilterForHideObsoletePoints(
301 : snapshotForHideObsoletePoints uint64,
302 : fileLargestSeqNum uint64,
303 : pointKeyFilters []BlockPropertyFilter,
304 2 : ) (hideObsoletePoints bool, filters []BlockPropertyFilter) {
305 2 : hideObsoletePoints = r.tableFormat >= TableFormatPebblev4 &&
306 2 : snapshotForHideObsoletePoints > fileLargestSeqNum
307 2 : if hideObsoletePoints {
308 2 : pointKeyFilters = append(pointKeyFilters, obsoleteKeyBlockPropertyFilter{})
309 2 : }
310 2 : return hideObsoletePoints, pointKeyFilters
311 : }
312 :
313 : func (r *Reader) newIterWithBlockPropertyFiltersAndContext(
314 : ctx context.Context,
315 : lower, upper []byte,
316 : filterer *BlockPropertiesFilterer,
317 : hideObsoletePoints bool,
318 : useFilterBlock bool,
319 : stats *base.InternalIteratorStats,
320 : categoryAndQoS CategoryAndQoS,
321 : statsCollector *CategoryStatsCollector,
322 : rp ReaderProvider,
323 : v *virtualState,
324 2 : ) (Iterator, error) {
325 2 : // NB: pebble.tableCache wraps the returned iterator with one which performs
326 2 : // reference counting on the Reader, preventing the Reader from being closed
327 2 : // until the final iterator closes.
328 2 : if r.Properties.IndexType == twoLevelIndex {
329 2 : i := twoLevelIterPool.Get().(*twoLevelIterator)
330 2 : err := i.init(ctx, r, v, lower, upper, filterer, useFilterBlock, hideObsoletePoints, stats,
331 2 : categoryAndQoS, statsCollector, rp, nil /* bufferPool */)
332 2 : if err != nil {
333 1 : return nil, err
334 1 : }
335 2 : return i, nil
336 : }
337 :
338 2 : i := singleLevelIterPool.Get().(*singleLevelIterator)
339 2 : err := i.init(ctx, r, v, lower, upper, filterer, useFilterBlock, hideObsoletePoints, stats,
340 2 : categoryAndQoS, statsCollector, rp, nil /* bufferPool */)
341 2 : if err != nil {
342 1 : return nil, err
343 1 : }
344 2 : return i, nil
345 : }
346 :
347 : // NewIter returns an iterator for the contents of the table. If an error
348 : // occurs, NewIter cleans up after itself and returns a nil iterator. NewIter
349 : // must only be used when the Reader is guaranteed to outlive any LazyValues
350 : // returned from the iter.
351 2 : func (r *Reader) NewIter(lower, upper []byte) (Iterator, error) {
352 2 : return r.NewIterWithBlockPropertyFilters(
353 2 : lower, upper, nil, true /* useFilterBlock */, nil, /* stats */
354 2 : CategoryAndQoS{}, nil /*statsCollector */, TrivialReaderProvider{Reader: r})
355 2 : }
356 :
357 : // NewCompactionIter returns an iterator similar to NewIter but it also increments
358 : // the number of bytes iterated. If an error occurs, NewCompactionIter cleans up
359 : // after itself and returns a nil iterator.
360 : func (r *Reader) NewCompactionIter(
361 : bytesIterated *uint64,
362 : categoryAndQoS CategoryAndQoS,
363 : statsCollector *CategoryStatsCollector,
364 : rp ReaderProvider,
365 : bufferPool *BufferPool,
366 2 : ) (Iterator, error) {
367 2 : return r.newCompactionIter(bytesIterated, categoryAndQoS, statsCollector, rp, nil, bufferPool)
368 2 : }
369 :
370 : func (r *Reader) newCompactionIter(
371 : bytesIterated *uint64,
372 : categoryAndQoS CategoryAndQoS,
373 : statsCollector *CategoryStatsCollector,
374 : rp ReaderProvider,
375 : v *virtualState,
376 : bufferPool *BufferPool,
377 2 : ) (Iterator, error) {
378 2 : if r.Properties.IndexType == twoLevelIndex {
379 2 : i := twoLevelIterPool.Get().(*twoLevelIterator)
380 2 : err := i.init(
381 2 : context.Background(),
382 2 : r, v, nil /* lower */, nil /* upper */, nil,
383 2 : false /* useFilter */, v != nil && v.isForeign, /* hideObsoletePoints */
384 2 : nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
385 2 : )
386 2 : if err != nil {
387 0 : return nil, err
388 0 : }
389 2 : i.setupForCompaction()
390 2 : return &twoLevelCompactionIterator{
391 2 : twoLevelIterator: i,
392 2 : bytesIterated: bytesIterated,
393 2 : }, nil
394 : }
395 2 : i := singleLevelIterPool.Get().(*singleLevelIterator)
396 2 : err := i.init(
397 2 : context.Background(), r, v, nil /* lower */, nil, /* upper */
398 2 : nil, false /* useFilter */, v != nil && v.isForeign, /* hideObsoletePoints */
399 2 : nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
400 2 : )
401 2 : if err != nil {
402 0 : return nil, err
403 0 : }
404 2 : i.setupForCompaction()
405 2 : return &compactionIterator{
406 2 : singleLevelIterator: i,
407 2 : bytesIterated: bytesIterated,
408 2 : }, nil
409 : }
410 :
411 : // NewRawRangeDelIter returns an internal iterator for the contents of the
412 : // range-del block for the table. Returns nil if the table does not contain
413 : // any range deletions.
414 : //
415 : // TODO(sumeer): plumb context.Context since this path is relevant in the user-facing
416 : // iterator. Add WithContext methods since the existing ones are public.
417 2 : func (r *Reader) NewRawRangeDelIter() (keyspan.FragmentIterator, error) {
418 2 : if r.rangeDelBH.Length == 0 {
419 2 : return nil, nil
420 2 : }
421 2 : h, err := r.readRangeDel(nil /* stats */, nil /* iterStats */)
422 2 : if err != nil {
423 1 : return nil, err
424 1 : }
425 2 : i := &fragmentBlockIter{elideSameSeqnum: true}
426 2 : if err := i.blockIter.initHandle(r.Compare, h, r.Properties.GlobalSeqNum, false); err != nil {
427 0 : return nil, err
428 0 : }
429 2 : return i, nil
430 : }
431 :
432 : // NewRawRangeKeyIter returns an internal iterator for the contents of the
433 : // range-key block for the table. Returns nil if the table does not contain any
434 : // range keys.
435 : //
436 : // TODO(sumeer): plumb context.Context since this path is relevant in the user-facing
437 : // iterator. Add WithContext methods since the existing ones are public.
438 2 : func (r *Reader) NewRawRangeKeyIter() (keyspan.FragmentIterator, error) {
439 2 : if r.rangeKeyBH.Length == 0 {
440 2 : return nil, nil
441 2 : }
442 2 : h, err := r.readRangeKey(nil /* stats */, nil /* iterStats */)
443 2 : if err != nil {
444 1 : return nil, err
445 1 : }
446 2 : i := rangeKeyFragmentBlockIterPool.Get().(*rangeKeyFragmentBlockIter)
447 2 : if err := i.blockIter.initHandle(r.Compare, h, r.Properties.GlobalSeqNum, false); err != nil {
448 0 : return nil, err
449 0 : }
450 2 : return i, nil
451 : }
452 :
453 : type rangeKeyFragmentBlockIter struct {
454 : fragmentBlockIter
455 : }
456 :
457 2 : func (i *rangeKeyFragmentBlockIter) Close() error {
458 2 : err := i.fragmentBlockIter.Close()
459 2 : i.fragmentBlockIter = i.fragmentBlockIter.resetForReuse()
460 2 : rangeKeyFragmentBlockIterPool.Put(i)
461 2 : return err
462 2 : }
463 :
464 : func (r *Reader) readIndex(
465 : ctx context.Context, stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
466 2 : ) (bufferHandle, error) {
467 2 : ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
468 2 : return r.readBlock(ctx, r.indexBH, nil, nil, stats, iterStats, nil /* buffer pool */)
469 2 : }
470 :
471 : func (r *Reader) readFilter(
472 : ctx context.Context, stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
473 2 : ) (bufferHandle, error) {
474 2 : ctx = objiotracing.WithBlockType(ctx, objiotracing.FilterBlock)
475 2 : return r.readBlock(ctx, r.filterBH, nil /* transform */, nil /* readHandle */, stats, iterStats, nil /* buffer pool */)
476 2 : }
477 :
478 : func (r *Reader) readRangeDel(
479 : stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
480 2 : ) (bufferHandle, error) {
481 2 : ctx := objiotracing.WithBlockType(context.Background(), objiotracing.MetadataBlock)
482 2 : return r.readBlock(ctx, r.rangeDelBH, r.rangeDelTransform, nil /* readHandle */, stats, iterStats, nil /* buffer pool */)
483 2 : }
484 :
485 : func (r *Reader) readRangeKey(
486 : stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
487 2 : ) (bufferHandle, error) {
488 2 : ctx := objiotracing.WithBlockType(context.Background(), objiotracing.MetadataBlock)
489 2 : return r.readBlock(ctx, r.rangeKeyBH, nil /* transform */, nil /* readHandle */, stats, iterStats, nil /* buffer pool */)
490 2 : }
491 :
492 : func checkChecksum(
493 : checksumType ChecksumType, b []byte, bh BlockHandle, fileNum base.FileNum,
494 2 : ) error {
495 2 : expectedChecksum := binary.LittleEndian.Uint32(b[bh.Length+1:])
496 2 : var computedChecksum uint32
497 2 : switch checksumType {
498 2 : case ChecksumTypeCRC32c:
499 2 : computedChecksum = crc.New(b[:bh.Length+1]).Value()
500 1 : case ChecksumTypeXXHash64:
501 1 : computedChecksum = uint32(xxhash.Sum64(b[:bh.Length+1]))
502 0 : default:
503 0 : return errors.Errorf("unsupported checksum type: %d", checksumType)
504 : }
505 :
506 2 : if expectedChecksum != computedChecksum {
507 1 : return base.CorruptionErrorf(
508 1 : "pebble/table: invalid table %s (checksum mismatch at %d/%d)",
509 1 : errors.Safe(fileNum), errors.Safe(bh.Offset), errors.Safe(bh.Length))
510 1 : }
511 2 : return nil
512 : }
513 :
514 : type cacheValueOrBuf struct {
515 : // buf.Valid() returns true if backed by a BufferPool.
516 : buf Buf
517 : // v is non-nil if backed by the block cache.
518 : v *cache.Value
519 : }
520 :
521 2 : func (b cacheValueOrBuf) get() []byte {
522 2 : if b.buf.Valid() {
523 2 : return b.buf.p.pool[b.buf.i].b
524 2 : }
525 2 : return b.v.Buf()
526 : }
527 :
528 2 : func (b cacheValueOrBuf) release() {
529 2 : if b.buf.Valid() {
530 2 : b.buf.Release()
531 2 : } else {
532 2 : cache.Free(b.v)
533 2 : }
534 : }
535 :
536 2 : func (b cacheValueOrBuf) truncate(n int) {
537 2 : if b.buf.Valid() {
538 2 : b.buf.p.pool[b.buf.i].b = b.buf.p.pool[b.buf.i].b[:n]
539 2 : } else {
540 2 : b.v.Truncate(n)
541 2 : }
542 : }
543 :
544 : func (r *Reader) readBlock(
545 : ctx context.Context,
546 : bh BlockHandle,
547 : transform blockTransform,
548 : readHandle objstorage.ReadHandle,
549 : stats *base.InternalIteratorStats,
550 : iterStats *iterStatsAccumulator,
551 : bufferPool *BufferPool,
552 2 : ) (handle bufferHandle, _ error) {
553 2 : if h := r.opts.Cache.Get(r.cacheID, r.fileNum, bh.Offset); h.Get() != nil {
554 2 : // Cache hit.
555 2 : if readHandle != nil {
556 2 : readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+blockTrailerLen))
557 2 : }
558 2 : if stats != nil {
559 2 : stats.BlockBytes += bh.Length
560 2 : stats.BlockBytesInCache += bh.Length
561 2 : }
562 2 : if iterStats != nil {
563 2 : iterStats.reportStats(bh.Length, bh.Length)
564 2 : }
565 : // This block is already in the cache; return a handle to existing vlaue
566 : // in the cache.
567 2 : return bufferHandle{h: h}, nil
568 : }
569 :
570 : // Cache miss.
571 2 : var compressed cacheValueOrBuf
572 2 : if bufferPool != nil {
573 2 : compressed = cacheValueOrBuf{
574 2 : buf: bufferPool.Alloc(int(bh.Length + blockTrailerLen)),
575 2 : }
576 2 : } else {
577 2 : compressed = cacheValueOrBuf{
578 2 : v: cache.Alloc(int(bh.Length + blockTrailerLen)),
579 2 : }
580 2 : }
581 :
582 2 : readStartTime := time.Now()
583 2 : var err error
584 2 : if readHandle != nil {
585 2 : err = readHandle.ReadAt(ctx, compressed.get(), int64(bh.Offset))
586 2 : } else {
587 2 : err = r.readable.ReadAt(ctx, compressed.get(), int64(bh.Offset))
588 2 : }
589 2 : readDuration := time.Since(readStartTime)
590 2 : // TODO(sumeer): should the threshold be configurable.
591 2 : const slowReadTracingThreshold = 5 * time.Millisecond
592 2 : // The invariants.Enabled path is for deterministic testing.
593 2 : if invariants.Enabled {
594 2 : readDuration = slowReadTracingThreshold
595 2 : }
596 : // Call IsTracingEnabled to avoid the allocations of boxing integers into an
597 : // interface{}, unless necessary.
598 2 : if readDuration >= slowReadTracingThreshold && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
599 1 : r.opts.LoggerAndTracer.Eventf(ctx, "reading %d bytes took %s",
600 1 : int(bh.Length+blockTrailerLen), readDuration.String())
601 1 : }
602 2 : if stats != nil {
603 2 : stats.BlockReadDuration += readDuration
604 2 : }
605 2 : if err != nil {
606 1 : compressed.release()
607 1 : return bufferHandle{}, err
608 1 : }
609 2 : if err := checkChecksum(r.checksumType, compressed.get(), bh, r.fileNum.FileNum()); err != nil {
610 1 : compressed.release()
611 1 : return bufferHandle{}, err
612 1 : }
613 :
614 2 : typ := blockType(compressed.get()[bh.Length])
615 2 : compressed.truncate(int(bh.Length))
616 2 :
617 2 : var decompressed cacheValueOrBuf
618 2 : if typ == noCompressionBlockType {
619 2 : decompressed = compressed
620 2 : } else {
621 2 : // Decode the length of the decompressed value.
622 2 : decodedLen, prefixLen, err := decompressedLen(typ, compressed.get())
623 2 : if err != nil {
624 0 : compressed.release()
625 0 : return bufferHandle{}, err
626 0 : }
627 :
628 2 : if bufferPool != nil {
629 2 : decompressed = cacheValueOrBuf{buf: bufferPool.Alloc(decodedLen)}
630 2 : } else {
631 2 : decompressed = cacheValueOrBuf{v: cache.Alloc(decodedLen)}
632 2 : }
633 2 : if _, err := decompressInto(typ, compressed.get()[prefixLen:], decompressed.get()); err != nil {
634 0 : compressed.release()
635 0 : return bufferHandle{}, err
636 0 : }
637 2 : compressed.release()
638 : }
639 :
640 2 : if transform != nil {
641 1 : // Transforming blocks is very rare, so the extra copy of the
642 1 : // transformed data is not problematic.
643 1 : tmpTransformed, err := transform(decompressed.get())
644 1 : if err != nil {
645 0 : decompressed.release()
646 0 : return bufferHandle{}, err
647 0 : }
648 :
649 1 : var transformed cacheValueOrBuf
650 1 : if bufferPool != nil {
651 0 : transformed = cacheValueOrBuf{buf: bufferPool.Alloc(len(tmpTransformed))}
652 1 : } else {
653 1 : transformed = cacheValueOrBuf{v: cache.Alloc(len(tmpTransformed))}
654 1 : }
655 1 : copy(transformed.get(), tmpTransformed)
656 1 : decompressed.release()
657 1 : decompressed = transformed
658 : }
659 :
660 2 : if stats != nil {
661 2 : stats.BlockBytes += bh.Length
662 2 : }
663 2 : if iterStats != nil {
664 2 : iterStats.reportStats(bh.Length, 0)
665 2 : }
666 2 : if decompressed.buf.Valid() {
667 2 : return bufferHandle{b: decompressed.buf}, nil
668 2 : }
669 2 : h := r.opts.Cache.Set(r.cacheID, r.fileNum, bh.Offset, decompressed.v)
670 2 : return bufferHandle{h: h}, nil
671 : }
672 :
673 1 : func (r *Reader) transformRangeDelV1(b []byte) ([]byte, error) {
674 1 : // Convert v1 (RocksDB format) range-del blocks to v2 blocks on the fly. The
675 1 : // v1 format range-del blocks have unfragmented and unsorted range
676 1 : // tombstones. We need properly fragmented and sorted range tombstones in
677 1 : // order to serve from them directly.
678 1 : iter := &blockIter{}
679 1 : if err := iter.init(r.Compare, b, r.Properties.GlobalSeqNum, false); err != nil {
680 0 : return nil, err
681 0 : }
682 1 : var tombstones []keyspan.Span
683 1 : for key, value := iter.First(); key != nil; key, value = iter.Next() {
684 1 : t := keyspan.Span{
685 1 : Start: key.UserKey,
686 1 : End: value.InPlaceValue(),
687 1 : Keys: []keyspan.Key{{Trailer: key.Trailer}},
688 1 : }
689 1 : tombstones = append(tombstones, t)
690 1 : }
691 1 : keyspan.Sort(r.Compare, tombstones)
692 1 :
693 1 : // Fragment the tombstones, outputting them directly to a block writer.
694 1 : rangeDelBlock := blockWriter{
695 1 : restartInterval: 1,
696 1 : }
697 1 : frag := keyspan.Fragmenter{
698 1 : Cmp: r.Compare,
699 1 : Format: r.FormatKey,
700 1 : Emit: func(s keyspan.Span) {
701 1 : for _, k := range s.Keys {
702 1 : startIK := InternalKey{UserKey: s.Start, Trailer: k.Trailer}
703 1 : rangeDelBlock.add(startIK, s.End)
704 1 : }
705 : },
706 : }
707 1 : for i := range tombstones {
708 1 : frag.Add(tombstones[i])
709 1 : }
710 1 : frag.Finish()
711 1 :
712 1 : // Return the contents of the constructed v2 format range-del block.
713 1 : return rangeDelBlock.finish(), nil
714 : }
715 :
716 2 : func (r *Reader) readMetaindex(metaindexBH BlockHandle) error {
717 2 : // We use a BufferPool when reading metaindex blocks in order to avoid
718 2 : // populating the block cache with these blocks. In heavy-write workloads,
719 2 : // especially with high compaction concurrency, new tables may be created
720 2 : // frequently. Populating the block cache with these metaindex blocks adds
721 2 : // additional contention on the block cache mutexes (see #1997).
722 2 : // Additionally, these blocks are exceedingly unlikely to be read again
723 2 : // while they're still in the block cache except in misconfigurations with
724 2 : // excessive sstables counts or a table cache that's far too small.
725 2 : r.metaBufferPool.initPreallocated(r.metaBufferPoolAlloc[:0])
726 2 : // When we're finished, release the buffers we've allocated back to memory
727 2 : // allocator. We don't expect to use metaBufferPool again.
728 2 : defer r.metaBufferPool.Release()
729 2 :
730 2 : b, err := r.readBlock(
731 2 : context.Background(), metaindexBH, nil /* transform */, nil /* readHandle */, nil, /* stats */
732 2 : nil /* iterStats */, &r.metaBufferPool)
733 2 : if err != nil {
734 1 : return err
735 1 : }
736 2 : data := b.Get()
737 2 : defer b.Release()
738 2 :
739 2 : if uint64(len(data)) != metaindexBH.Length {
740 0 : return base.CorruptionErrorf("pebble/table: unexpected metaindex block size: %d vs %d",
741 0 : errors.Safe(len(data)), errors.Safe(metaindexBH.Length))
742 0 : }
743 :
744 2 : i, err := newRawBlockIter(bytes.Compare, data)
745 2 : if err != nil {
746 0 : return err
747 0 : }
748 :
749 2 : meta := map[string]BlockHandle{}
750 2 : for valid := i.First(); valid; valid = i.Next() {
751 2 : value := i.Value()
752 2 : if bytes.Equal(i.Key().UserKey, []byte(metaValueIndexName)) {
753 2 : vbih, n, err := decodeValueBlocksIndexHandle(i.Value())
754 2 : if err != nil {
755 0 : return err
756 0 : }
757 2 : if n == 0 || n != len(value) {
758 0 : return base.CorruptionErrorf("pebble/table: invalid table (bad value blocks index handle)")
759 0 : }
760 2 : r.valueBIH = vbih
761 2 : } else {
762 2 : bh, n := decodeBlockHandle(value)
763 2 : if n == 0 || n != len(value) {
764 0 : return base.CorruptionErrorf("pebble/table: invalid table (bad block handle)")
765 0 : }
766 2 : meta[string(i.Key().UserKey)] = bh
767 : }
768 : }
769 2 : if err := i.Close(); err != nil {
770 0 : return err
771 0 : }
772 :
773 2 : if bh, ok := meta[metaPropertiesName]; ok {
774 2 : b, err = r.readBlock(
775 2 : context.Background(), bh, nil /* transform */, nil /* readHandle */, nil, /* stats */
776 2 : nil /* iterStats */, nil /* buffer pool */)
777 2 : if err != nil {
778 1 : return err
779 1 : }
780 2 : r.propertiesBH = bh
781 2 : err := r.Properties.load(b.Get(), bh.Offset, r.opts.DeniedUserProperties)
782 2 : b.Release()
783 2 : if err != nil {
784 0 : return err
785 0 : }
786 : }
787 :
788 2 : if bh, ok := meta[metaRangeDelV2Name]; ok {
789 2 : r.rangeDelBH = bh
790 2 : } else if bh, ok := meta[metaRangeDelName]; ok {
791 1 : r.rangeDelBH = bh
792 1 : if !r.rawTombstones {
793 1 : r.rangeDelTransform = r.transformRangeDelV1
794 1 : }
795 : }
796 :
797 2 : if bh, ok := meta[metaRangeKeyName]; ok {
798 2 : r.rangeKeyBH = bh
799 2 : }
800 :
801 2 : for name, fp := range r.opts.Filters {
802 2 : types := []struct {
803 2 : ftype FilterType
804 2 : prefix string
805 2 : }{
806 2 : {TableFilter, "fullfilter."},
807 2 : }
808 2 : var done bool
809 2 : for _, t := range types {
810 2 : if bh, ok := meta[t.prefix+name]; ok {
811 2 : r.filterBH = bh
812 2 :
813 2 : switch t.ftype {
814 2 : case TableFilter:
815 2 : r.tableFilter = newTableFilterReader(fp)
816 0 : default:
817 0 : return base.CorruptionErrorf("unknown filter type: %v", errors.Safe(t.ftype))
818 : }
819 :
820 2 : done = true
821 2 : break
822 : }
823 : }
824 2 : if done {
825 2 : break
826 : }
827 : }
828 2 : return nil
829 : }
830 :
831 : // Layout returns the layout (block organization) for an sstable.
832 2 : func (r *Reader) Layout() (*Layout, error) {
833 2 : if r.err != nil {
834 0 : return nil, r.err
835 0 : }
836 :
837 2 : l := &Layout{
838 2 : Data: make([]BlockHandleWithProperties, 0, r.Properties.NumDataBlocks),
839 2 : Filter: r.filterBH,
840 2 : RangeDel: r.rangeDelBH,
841 2 : RangeKey: r.rangeKeyBH,
842 2 : ValueIndex: r.valueBIH.h,
843 2 : Properties: r.propertiesBH,
844 2 : MetaIndex: r.metaIndexBH,
845 2 : Footer: r.footerBH,
846 2 : Format: r.tableFormat,
847 2 : }
848 2 :
849 2 : indexH, err := r.readIndex(context.Background(), nil, nil)
850 2 : if err != nil {
851 1 : return nil, err
852 1 : }
853 2 : defer indexH.Release()
854 2 :
855 2 : var alloc bytealloc.A
856 2 :
857 2 : if r.Properties.IndexPartitions == 0 {
858 2 : l.Index = append(l.Index, r.indexBH)
859 2 : iter, _ := newBlockIter(r.Compare, indexH.Get())
860 2 : for key, value := iter.First(); key != nil; key, value = iter.Next() {
861 2 : dataBH, err := decodeBlockHandleWithProperties(value.InPlaceValue())
862 2 : if err != nil {
863 0 : return nil, errCorruptIndexEntry
864 0 : }
865 2 : if len(dataBH.Props) > 0 {
866 2 : alloc, dataBH.Props = alloc.Copy(dataBH.Props)
867 2 : }
868 2 : l.Data = append(l.Data, dataBH)
869 : }
870 2 : } else {
871 2 : l.TopIndex = r.indexBH
872 2 : topIter, _ := newBlockIter(r.Compare, indexH.Get())
873 2 : iter := &blockIter{}
874 2 : for key, value := topIter.First(); key != nil; key, value = topIter.Next() {
875 2 : indexBH, err := decodeBlockHandleWithProperties(value.InPlaceValue())
876 2 : if err != nil {
877 0 : return nil, errCorruptIndexEntry
878 0 : }
879 2 : l.Index = append(l.Index, indexBH.BlockHandle)
880 2 :
881 2 : subIndex, err := r.readBlock(context.Background(), indexBH.BlockHandle,
882 2 : nil /* transform */, nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
883 2 : if err != nil {
884 1 : return nil, err
885 1 : }
886 2 : if err := iter.init(r.Compare, subIndex.Get(), 0, /* globalSeqNum */
887 2 : false /* hideObsoletePoints */); err != nil {
888 0 : return nil, err
889 0 : }
890 2 : for key, value := iter.First(); key != nil; key, value = iter.Next() {
891 2 : dataBH, err := decodeBlockHandleWithProperties(value.InPlaceValue())
892 2 : if len(dataBH.Props) > 0 {
893 2 : alloc, dataBH.Props = alloc.Copy(dataBH.Props)
894 2 : }
895 2 : if err != nil {
896 0 : return nil, errCorruptIndexEntry
897 0 : }
898 2 : l.Data = append(l.Data, dataBH)
899 : }
900 2 : subIndex.Release()
901 2 : *iter = iter.resetForReuse()
902 : }
903 : }
904 2 : if r.valueBIH.h.Length != 0 {
905 2 : vbiH, err := r.readBlock(context.Background(), r.valueBIH.h, nil, nil, nil, nil, nil /* buffer pool */)
906 2 : if err != nil {
907 0 : return nil, err
908 0 : }
909 2 : defer vbiH.Release()
910 2 : vbiBlock := vbiH.Get()
911 2 : indexEntryLen := int(r.valueBIH.blockNumByteLength + r.valueBIH.blockOffsetByteLength +
912 2 : r.valueBIH.blockLengthByteLength)
913 2 : i := 0
914 2 : for len(vbiBlock) != 0 {
915 2 : if len(vbiBlock) < indexEntryLen {
916 0 : return nil, errors.Errorf(
917 0 : "remaining value index block %d does not contain a full entry of length %d",
918 0 : len(vbiBlock), indexEntryLen)
919 0 : }
920 2 : n := int(r.valueBIH.blockNumByteLength)
921 2 : bn := int(littleEndianGet(vbiBlock, n))
922 2 : if bn != i {
923 0 : return nil, errors.Errorf("unexpected block num %d, expected %d",
924 0 : bn, i)
925 0 : }
926 2 : i++
927 2 : vbiBlock = vbiBlock[n:]
928 2 : n = int(r.valueBIH.blockOffsetByteLength)
929 2 : blockOffset := littleEndianGet(vbiBlock, n)
930 2 : vbiBlock = vbiBlock[n:]
931 2 : n = int(r.valueBIH.blockLengthByteLength)
932 2 : blockLen := littleEndianGet(vbiBlock, n)
933 2 : vbiBlock = vbiBlock[n:]
934 2 : l.ValueBlock = append(l.ValueBlock, BlockHandle{Offset: blockOffset, Length: blockLen})
935 : }
936 : }
937 :
938 2 : return l, nil
939 : }
940 :
941 : // ValidateBlockChecksums validates the checksums for each block in the SSTable.
942 2 : func (r *Reader) ValidateBlockChecksums() error {
943 2 : // Pre-compute the BlockHandles for the underlying file.
944 2 : l, err := r.Layout()
945 2 : if err != nil {
946 1 : return err
947 1 : }
948 :
949 : // Construct the set of blocks to check. Note that the footer is not checked
950 : // as it is not a block with a checksum.
951 2 : blocks := make([]BlockHandle, len(l.Data))
952 2 : for i := range l.Data {
953 2 : blocks[i] = l.Data[i].BlockHandle
954 2 : }
955 2 : blocks = append(blocks, l.Index...)
956 2 : blocks = append(blocks, l.TopIndex, l.Filter, l.RangeDel, l.RangeKey, l.Properties, l.MetaIndex)
957 2 :
958 2 : // Sorting by offset ensures we are performing a sequential scan of the
959 2 : // file.
960 2 : slices.SortFunc(blocks, func(a, b BlockHandle) int {
961 2 : return cmp.Compare(a.Offset, b.Offset)
962 2 : })
963 :
964 : // Check all blocks sequentially. Make use of read-ahead, given we are
965 : // scanning the entire file from start to end.
966 2 : rh := r.readable.NewReadHandle(context.TODO())
967 2 : defer rh.Close()
968 2 :
969 2 : for _, bh := range blocks {
970 2 : // Certain blocks may not be present, in which case we skip them.
971 2 : if bh.Length == 0 {
972 2 : continue
973 : }
974 :
975 : // Read the block, which validates the checksum.
976 2 : h, err := r.readBlock(context.Background(), bh, nil, rh, nil, nil /* iterStats */, nil /* buffer pool */)
977 2 : if err != nil {
978 1 : return err
979 1 : }
980 2 : h.Release()
981 : }
982 :
983 2 : return nil
984 : }
985 :
986 : // CommonProperties implemented the CommonReader interface.
987 2 : func (r *Reader) CommonProperties() *CommonProperties {
988 2 : return &r.Properties.CommonProperties
989 2 : }
990 :
991 : // EstimateDiskUsage returns the total size of data blocks overlapping the range
992 : // `[start, end]`. Even if a data block partially overlaps, or we cannot
993 : // determine overlap due to abbreviated index keys, the full data block size is
994 : // included in the estimation.
995 : //
996 : // This function does not account for any metablock space usage. Assumes there
997 : // is at least partial overlap, i.e., `[start, end]` falls neither completely
998 : // before nor completely after the file's range.
999 : //
1000 : // Only blocks containing point keys are considered. Range deletion and range
1001 : // key blocks are not considered.
1002 : //
1003 : // TODO(ajkr): account for metablock space usage. Perhaps look at the fraction of
1004 : // data blocks overlapped and add that same fraction of the metadata blocks to the
1005 : // estimate.
1006 2 : func (r *Reader) EstimateDiskUsage(start, end []byte) (uint64, error) {
1007 2 : if r.err != nil {
1008 0 : return 0, r.err
1009 0 : }
1010 :
1011 2 : indexH, err := r.readIndex(context.Background(), nil, nil)
1012 2 : if err != nil {
1013 1 : return 0, err
1014 1 : }
1015 2 : defer indexH.Release()
1016 2 :
1017 2 : // Iterators over the bottom-level index blocks containing start and end.
1018 2 : // These may be different in case of partitioned index but will both point
1019 2 : // to the same blockIter over the single index in the unpartitioned case.
1020 2 : var startIdxIter, endIdxIter *blockIter
1021 2 : if r.Properties.IndexPartitions == 0 {
1022 2 : iter, err := newBlockIter(r.Compare, indexH.Get())
1023 2 : if err != nil {
1024 0 : return 0, err
1025 0 : }
1026 2 : startIdxIter = iter
1027 2 : endIdxIter = iter
1028 2 : } else {
1029 2 : topIter, err := newBlockIter(r.Compare, indexH.Get())
1030 2 : if err != nil {
1031 0 : return 0, err
1032 0 : }
1033 :
1034 2 : key, val := topIter.SeekGE(start, base.SeekGEFlagsNone)
1035 2 : if key == nil {
1036 1 : // The range falls completely after this file, or an error occurred.
1037 1 : return 0, topIter.Error()
1038 1 : }
1039 2 : startIdxBH, err := decodeBlockHandleWithProperties(val.InPlaceValue())
1040 2 : if err != nil {
1041 0 : return 0, errCorruptIndexEntry
1042 0 : }
1043 2 : startIdxBlock, err := r.readBlock(context.Background(), startIdxBH.BlockHandle,
1044 2 : nil /* transform */, nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
1045 2 : if err != nil {
1046 1 : return 0, err
1047 1 : }
1048 2 : defer startIdxBlock.Release()
1049 2 : startIdxIter, err = newBlockIter(r.Compare, startIdxBlock.Get())
1050 2 : if err != nil {
1051 0 : return 0, err
1052 0 : }
1053 :
1054 2 : key, val = topIter.SeekGE(end, base.SeekGEFlagsNone)
1055 2 : if key == nil {
1056 1 : if err := topIter.Error(); err != nil {
1057 0 : return 0, err
1058 0 : }
1059 2 : } else {
1060 2 : endIdxBH, err := decodeBlockHandleWithProperties(val.InPlaceValue())
1061 2 : if err != nil {
1062 0 : return 0, errCorruptIndexEntry
1063 0 : }
1064 2 : endIdxBlock, err := r.readBlock(context.Background(),
1065 2 : endIdxBH.BlockHandle, nil /* transform */, nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
1066 2 : if err != nil {
1067 1 : return 0, err
1068 1 : }
1069 2 : defer endIdxBlock.Release()
1070 2 : endIdxIter, err = newBlockIter(r.Compare, endIdxBlock.Get())
1071 2 : if err != nil {
1072 0 : return 0, err
1073 0 : }
1074 : }
1075 : }
1076 : // startIdxIter should not be nil at this point, while endIdxIter can be if the
1077 : // range spans past the end of the file.
1078 :
1079 2 : key, val := startIdxIter.SeekGE(start, base.SeekGEFlagsNone)
1080 2 : if key == nil {
1081 2 : // The range falls completely after this file, or an error occurred.
1082 2 : return 0, startIdxIter.Error()
1083 2 : }
1084 2 : startBH, err := decodeBlockHandleWithProperties(val.InPlaceValue())
1085 2 : if err != nil {
1086 0 : return 0, errCorruptIndexEntry
1087 0 : }
1088 :
1089 2 : includeInterpolatedValueBlocksSize := func(dataBlockSize uint64) uint64 {
1090 2 : // INVARIANT: r.Properties.DataSize > 0 since startIdxIter is not nil.
1091 2 : // Linearly interpolate what is stored in value blocks.
1092 2 : //
1093 2 : // TODO(sumeer): if we need more accuracy, without loading any data blocks
1094 2 : // (which contain the value handles, and which may also be insufficient if
1095 2 : // the values are in separate files), we will need to accumulate the
1096 2 : // logical size of the key-value pairs and store the cumulative value for
1097 2 : // each data block in the index block entry. This increases the size of
1098 2 : // the BlockHandle, so wait until this becomes necessary.
1099 2 : return dataBlockSize +
1100 2 : uint64((float64(dataBlockSize)/float64(r.Properties.DataSize))*
1101 2 : float64(r.Properties.ValueBlocksSize))
1102 2 : }
1103 2 : if endIdxIter == nil {
1104 1 : // The range spans beyond this file. Include data blocks through the last.
1105 1 : return includeInterpolatedValueBlocksSize(r.Properties.DataSize - startBH.Offset), nil
1106 1 : }
1107 2 : key, val = endIdxIter.SeekGE(end, base.SeekGEFlagsNone)
1108 2 : if key == nil {
1109 2 : if err := endIdxIter.Error(); err != nil {
1110 0 : return 0, err
1111 0 : }
1112 : // The range spans beyond this file. Include data blocks through the last.
1113 2 : return includeInterpolatedValueBlocksSize(r.Properties.DataSize - startBH.Offset), nil
1114 : }
1115 2 : endBH, err := decodeBlockHandleWithProperties(val.InPlaceValue())
1116 2 : if err != nil {
1117 0 : return 0, errCorruptIndexEntry
1118 0 : }
1119 2 : return includeInterpolatedValueBlocksSize(
1120 2 : endBH.Offset + endBH.Length + blockTrailerLen - startBH.Offset), nil
1121 : }
1122 :
1123 : // TableFormat returns the format version for the table.
1124 2 : func (r *Reader) TableFormat() (TableFormat, error) {
1125 2 : if r.err != nil {
1126 0 : return TableFormatUnspecified, r.err
1127 0 : }
1128 2 : return r.tableFormat, nil
1129 : }
1130 :
1131 : // NewReader returns a new table reader for the file. Closing the reader will
1132 : // close the file.
1133 2 : func NewReader(f objstorage.Readable, o ReaderOptions, extraOpts ...ReaderOption) (*Reader, error) {
1134 2 : o = o.ensureDefaults()
1135 2 : r := &Reader{
1136 2 : readable: f,
1137 2 : opts: o,
1138 2 : }
1139 2 : if r.opts.Cache == nil {
1140 1 : r.opts.Cache = cache.New(0)
1141 2 : } else {
1142 2 : r.opts.Cache.Ref()
1143 2 : }
1144 :
1145 2 : if f == nil {
1146 1 : r.err = errors.New("pebble/table: nil file")
1147 1 : return nil, r.Close()
1148 1 : }
1149 :
1150 : // Note that the extra options are applied twice. First here for pre-apply
1151 : // options, and then below for post-apply options. Pre and post refer to
1152 : // before and after reading the metaindex and properties.
1153 2 : type preApply interface{ preApply() }
1154 2 : for _, opt := range extraOpts {
1155 2 : if _, ok := opt.(preApply); ok {
1156 2 : opt.readerApply(r)
1157 2 : }
1158 : }
1159 2 : if r.cacheID == 0 {
1160 1 : r.cacheID = r.opts.Cache.NewID()
1161 1 : }
1162 :
1163 2 : footer, err := readFooter(f)
1164 2 : if err != nil {
1165 1 : r.err = err
1166 1 : return nil, r.Close()
1167 1 : }
1168 2 : r.checksumType = footer.checksum
1169 2 : r.tableFormat = footer.format
1170 2 : // Read the metaindex.
1171 2 : if err := r.readMetaindex(footer.metaindexBH); err != nil {
1172 1 : r.err = err
1173 1 : return nil, r.Close()
1174 1 : }
1175 2 : r.indexBH = footer.indexBH
1176 2 : r.metaIndexBH = footer.metaindexBH
1177 2 : r.footerBH = footer.footerBH
1178 2 :
1179 2 : if r.Properties.ComparerName == "" || o.Comparer.Name == r.Properties.ComparerName {
1180 2 : r.Compare = o.Comparer.Compare
1181 2 : r.FormatKey = o.Comparer.FormatKey
1182 2 : r.Split = o.Comparer.Split
1183 2 : }
1184 :
1185 2 : if o.MergerName == r.Properties.MergerName {
1186 2 : r.mergerOK = true
1187 2 : }
1188 :
1189 : // Apply the extra options again now that the comparer and merger names are
1190 : // known.
1191 2 : for _, opt := range extraOpts {
1192 2 : if _, ok := opt.(preApply); !ok {
1193 2 : opt.readerApply(r)
1194 2 : }
1195 : }
1196 :
1197 2 : if r.Compare == nil {
1198 1 : r.err = errors.Errorf("pebble/table: %d: unknown comparer %s",
1199 1 : errors.Safe(r.fileNum), errors.Safe(r.Properties.ComparerName))
1200 1 : }
1201 2 : if !r.mergerOK {
1202 1 : if name := r.Properties.MergerName; name != "" && name != "nullptr" {
1203 1 : r.err = errors.Errorf("pebble/table: %d: unknown merger %s",
1204 1 : errors.Safe(r.fileNum), errors.Safe(r.Properties.MergerName))
1205 1 : }
1206 : }
1207 2 : if r.err != nil {
1208 1 : return nil, r.Close()
1209 1 : }
1210 :
1211 2 : return r, nil
1212 : }
1213 :
1214 : // ReadableFile describes the smallest subset of vfs.File that is required for
1215 : // reading SSTs.
1216 : type ReadableFile interface {
1217 : io.ReaderAt
1218 : io.Closer
1219 : Stat() (os.FileInfo, error)
1220 : }
1221 :
1222 : // NewSimpleReadable wraps a ReadableFile in a objstorage.Readable
1223 : // implementation (which does not support read-ahead)
1224 2 : func NewSimpleReadable(r ReadableFile) (objstorage.Readable, error) {
1225 2 : info, err := r.Stat()
1226 2 : if err != nil {
1227 1 : return nil, err
1228 1 : }
1229 2 : res := &simpleReadable{
1230 2 : f: r,
1231 2 : size: info.Size(),
1232 2 : }
1233 2 : res.rh = objstorage.MakeNoopReadHandle(res)
1234 2 : return res, nil
1235 : }
1236 :
1237 : // simpleReadable wraps a ReadableFile to implement objstorage.Readable.
1238 : type simpleReadable struct {
1239 : f ReadableFile
1240 : size int64
1241 : rh objstorage.NoopReadHandle
1242 : }
1243 :
1244 : var _ objstorage.Readable = (*simpleReadable)(nil)
1245 :
1246 : // ReadAt is part of the objstorage.Readable interface.
1247 2 : func (s *simpleReadable) ReadAt(_ context.Context, p []byte, off int64) error {
1248 2 : n, err := s.f.ReadAt(p, off)
1249 2 : if invariants.Enabled && err == nil && n != len(p) {
1250 0 : panic("short read")
1251 : }
1252 2 : return err
1253 : }
1254 :
1255 : // Close is part of the objstorage.Readable interface.
1256 2 : func (s *simpleReadable) Close() error {
1257 2 : return s.f.Close()
1258 2 : }
1259 :
1260 : // Size is part of the objstorage.Readable interface.
1261 2 : func (s *simpleReadable) Size() int64 {
1262 2 : return s.size
1263 2 : }
1264 :
1265 : // NewReaddHandle is part of the objstorage.Readable interface.
1266 2 : func (s *simpleReadable) NewReadHandle(_ context.Context) objstorage.ReadHandle {
1267 2 : return &s.rh
1268 2 : }
|
