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