Line data Source code
1 : // Copyright 2022 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 pebble
6 :
7 : import (
8 : "context"
9 :
10 : "github.com/cockroachdb/errors"
11 : "github.com/cockroachdb/pebble/internal/base"
12 : "github.com/cockroachdb/pebble/internal/keyspan"
13 : "github.com/cockroachdb/pebble/internal/manifest"
14 : "github.com/cockroachdb/pebble/sstable"
15 : )
16 :
17 : // NewExternalIter takes an input 2d array of sstable files which may overlap
18 : // across subarrays but not within a subarray (at least as far as points are
19 : // concerned; range keys are allowed to overlap arbitrarily even within a
20 : // subarray), and returns an Iterator over the merged contents of the sstables.
21 : // Input sstables may contain point keys, range keys, range deletions, etc. The
22 : // input files slice must be sorted in reverse chronological ordering. A key in a
23 : // file at a lower index subarray will shadow a key with an identical user key
24 : // contained within a file at a higher index subarray. Each subarray must be
25 : // sorted in internal key order, where lower index files contain keys that sort
26 : // left of files with higher indexes.
27 : //
28 : // Input sstables must only contain keys with the zero sequence number.
29 : //
30 : // Iterators constructed through NewExternalIter do not support all iterator
31 : // options, including block-property and table filters. NewExternalIter errors
32 : // if an incompatible option is set.
33 : func NewExternalIter(
34 : o *Options, iterOpts *IterOptions, files [][]sstable.ReadableFile,
35 1 : ) (it *Iterator, err error) {
36 1 : return NewExternalIterWithContext(context.Background(), o, iterOpts, files)
37 1 : }
38 :
39 : // NewExternalIterWithContext is like NewExternalIter, and additionally
40 : // accepts a context for tracing.
41 : func NewExternalIterWithContext(
42 : ctx context.Context, o *Options, iterOpts *IterOptions, files [][]sstable.ReadableFile,
43 1 : ) (it *Iterator, err error) {
44 1 : if iterOpts != nil {
45 1 : if err := validateExternalIterOpts(iterOpts); err != nil {
46 0 : return nil, err
47 0 : }
48 : }
49 :
50 1 : var readers [][]*sstable.Reader
51 1 :
52 1 : seqNumOffset := 0
53 1 : for _, levelFiles := range files {
54 1 : seqNumOffset += len(levelFiles)
55 1 : }
56 1 : for _, levelFiles := range files {
57 1 : seqNumOffset -= len(levelFiles)
58 1 : subReaders, err := openExternalTables(ctx, o, levelFiles, seqNumOffset, o.MakeReaderOptions())
59 1 : readers = append(readers, subReaders)
60 1 : if err != nil {
61 0 : // Close all the opened readers.
62 0 : for i := range readers {
63 0 : for j := range readers[i] {
64 0 : _ = readers[i][j].Close()
65 0 : }
66 : }
67 0 : return nil, err
68 : }
69 : }
70 :
71 1 : buf := iterAllocPool.Get().(*iterAlloc)
72 1 : dbi := &buf.dbi
73 1 : *dbi = Iterator{
74 1 : ctx: ctx,
75 1 : alloc: buf,
76 1 : merge: o.Merger.Merge,
77 1 : comparer: *o.Comparer,
78 1 : readState: nil,
79 1 : keyBuf: buf.keyBuf,
80 1 : prefixOrFullSeekKey: buf.prefixOrFullSeekKey,
81 1 : boundsBuf: buf.boundsBuf,
82 1 : batch: nil,
83 1 : // Add the readers to the Iterator so that Close closes them, and
84 1 : // SetOptions can re-construct iterators from them.
85 1 : externalReaders: readers,
86 1 : newIters: func(context.Context, *manifest.FileMetadata, *IterOptions,
87 1 : internalIterOpts, iterKinds) (iterSet, error) {
88 0 : // NB: External iterators are currently constructed without any
89 0 : // `levelIters`. newIters should never be called. When we support
90 0 : // organizing multiple non-overlapping files into a single level
91 0 : // (see TODO below), we'll need to adjust this tableNewIters
92 0 : // implementation to open iterators by looking up f in a map
93 0 : // of readers indexed by *fileMetadata.
94 0 : panic("unreachable")
95 : },
96 : seqNum: base.SeqNumMax,
97 : }
98 1 : if iterOpts != nil {
99 1 : dbi.opts = *iterOpts
100 1 : dbi.processBounds(iterOpts.LowerBound, iterOpts.UpperBound)
101 1 : }
102 1 : if err := finishInitializingExternal(ctx, dbi); err != nil {
103 0 : dbi.Close()
104 0 : return nil, err
105 0 : }
106 1 : return dbi, nil
107 : }
108 :
109 1 : func validateExternalIterOpts(iterOpts *IterOptions) error {
110 1 : switch {
111 0 : case iterOpts.PointKeyFilters != nil:
112 0 : return errors.Errorf("pebble: external iterator: PointKeyFilters unsupported")
113 0 : case iterOpts.RangeKeyFilters != nil:
114 0 : return errors.Errorf("pebble: external iterator: RangeKeyFilters unsupported")
115 0 : case iterOpts.OnlyReadGuaranteedDurable:
116 0 : return errors.Errorf("pebble: external iterator: OnlyReadGuaranteedDurable unsupported")
117 0 : case iterOpts.UseL6Filters:
118 0 : return errors.Errorf("pebble: external iterator: UseL6Filters unsupported")
119 : }
120 1 : return nil
121 : }
122 :
123 1 : func createExternalPointIter(ctx context.Context, it *Iterator) (topLevelIterator, error) {
124 1 : // TODO(jackson): In some instances we could generate fewer levels by using
125 1 : // L0Sublevels code to organize nonoverlapping files into the same level.
126 1 : // This would allow us to use levelIters and keep a smaller set of data and
127 1 : // files in-memory. However, it would also require us to identify the bounds
128 1 : // of all the files upfront.
129 1 :
130 1 : if !it.opts.pointKeys() {
131 0 : return emptyIter, nil
132 1 : } else if it.pointIter != nil {
133 1 : return it.pointIter, nil
134 1 : }
135 1 : mlevels := it.alloc.mlevels[:0]
136 1 :
137 1 : if len(it.externalReaders) > cap(mlevels) {
138 0 : mlevels = make([]mergingIterLevel, 0, len(it.externalReaders))
139 0 : }
140 : // We set a synthetic sequence number, with lower levels having higer numbers.
141 1 : seqNum := 0
142 1 : for _, readers := range it.externalReaders {
143 1 : seqNum += len(readers)
144 1 : }
145 1 : for _, readers := range it.externalReaders {
146 1 : for _, r := range readers {
147 1 : var (
148 1 : rangeDelIter keyspan.FragmentIterator
149 1 : pointIter internalIterator
150 1 : err error
151 1 : )
152 1 : // We could set hideObsoletePoints=true, since we are reading at
153 1 : // InternalKeySeqNumMax, but we don't bother since these sstables should
154 1 : // not have obsolete points (so the performance optimization is
155 1 : // unnecessary), and we don't want to bother constructing a
156 1 : // BlockPropertiesFilterer that includes obsoleteKeyBlockPropertyFilter.
157 1 : transforms := sstable.IterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
158 1 : seqNum--
159 1 : pointIter, err = r.NewPointIter(
160 1 : ctx, transforms, it.opts.LowerBound, it.opts.UpperBound, nil, /* BlockPropertiesFilterer */
161 1 : sstable.NeverUseFilterBlock,
162 1 : &it.stats.InternalStats, it.opts.CategoryAndQoS, nil,
163 1 : sstable.MakeTrivialReaderProvider(r))
164 1 : if err != nil {
165 0 : return nil, err
166 0 : }
167 1 : rangeDelIter, err = r.NewRawRangeDelIter(ctx, sstable.FragmentIterTransforms{
168 1 : SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum),
169 1 : })
170 1 : if err != nil {
171 0 : return nil, err
172 0 : }
173 1 : mlevels = append(mlevels, mergingIterLevel{
174 1 : iter: pointIter,
175 1 : rangeDelIter: rangeDelIter,
176 1 : })
177 : }
178 : }
179 :
180 1 : it.alloc.merging.init(&it.opts, &it.stats.InternalStats, it.comparer.Compare, it.comparer.Split, mlevels...)
181 1 : it.alloc.merging.snapshot = base.SeqNumMax
182 1 : if len(mlevels) <= cap(it.alloc.levelsPositioned) {
183 1 : it.alloc.merging.levelsPositioned = it.alloc.levelsPositioned[:len(mlevels)]
184 1 : }
185 1 : return &it.alloc.merging, nil
186 : }
187 :
188 1 : func finishInitializingExternal(ctx context.Context, it *Iterator) error {
189 1 : pointIter, err := createExternalPointIter(ctx, it)
190 1 : if err != nil {
191 0 : return err
192 0 : }
193 1 : it.pointIter = pointIter
194 1 : it.iter = it.pointIter
195 1 :
196 1 : if it.opts.rangeKeys() {
197 1 : it.rangeKeyMasking.init(it, &it.comparer)
198 1 : var rangeKeyIters []keyspan.FragmentIterator
199 1 : if it.rangeKey == nil {
200 1 : // We could take advantage of the lack of overlaps in range keys within
201 1 : // each slice in it.externalReaders, and generate keyspanimpl.LevelIters
202 1 : // out of those. However, since range keys are expected to be sparse to
203 1 : // begin with, the performance gain might not be significant enough to
204 1 : // warrant it.
205 1 : //
206 1 : // TODO(bilal): Explore adding a simpleRangeKeyLevelIter that does not
207 1 : // operate on FileMetadatas (similar to simpleLevelIter), and implements
208 1 : // this optimization.
209 1 : // We set a synthetic sequence number, with lower levels having higer numbers.
210 1 : seqNum := 0
211 1 : for _, readers := range it.externalReaders {
212 1 : seqNum += len(readers)
213 1 : }
214 1 : for _, readers := range it.externalReaders {
215 1 : for _, r := range readers {
216 1 : transforms := sstable.FragmentIterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
217 1 : seqNum--
218 1 : if rki, err := r.NewRawRangeKeyIter(ctx, transforms); err != nil {
219 0 : return err
220 1 : } else if rki != nil {
221 1 : rangeKeyIters = append(rangeKeyIters, rki)
222 1 : }
223 : }
224 : }
225 1 : if len(rangeKeyIters) > 0 {
226 1 : it.rangeKey = iterRangeKeyStateAllocPool.Get().(*iteratorRangeKeyState)
227 1 : it.rangeKey.init(it.comparer.Compare, it.comparer.Split, &it.opts)
228 1 : it.rangeKey.rangeKeyIter = it.rangeKey.iterConfig.Init(
229 1 : &it.comparer,
230 1 : base.SeqNumMax,
231 1 : it.opts.LowerBound, it.opts.UpperBound,
232 1 : &it.hasPrefix, &it.prefixOrFullSeekKey,
233 1 : false /* internalKeys */, &it.rangeKey.internal,
234 1 : )
235 1 : for i := range rangeKeyIters {
236 1 : it.rangeKey.iterConfig.AddLevel(rangeKeyIters[i])
237 1 : }
238 : }
239 : }
240 1 : if it.rangeKey != nil {
241 1 : it.rangeKey.iiter.Init(&it.comparer, it.iter, it.rangeKey.rangeKeyIter,
242 1 : keyspan.InterleavingIterOpts{
243 1 : Mask: &it.rangeKeyMasking,
244 1 : LowerBound: it.opts.LowerBound,
245 1 : UpperBound: it.opts.UpperBound,
246 1 : })
247 1 : it.iter = &it.rangeKey.iiter
248 1 : }
249 : }
250 1 : return nil
251 : }
252 :
253 : func openExternalTables(
254 : ctx context.Context,
255 : o *Options,
256 : files []sstable.ReadableFile,
257 : seqNumOffset int,
258 : readerOpts sstable.ReaderOptions,
259 1 : ) (readers []*sstable.Reader, err error) {
260 1 : readers = make([]*sstable.Reader, 0, len(files))
261 1 : for i := range files {
262 1 : readable, err := sstable.NewSimpleReadable(files[i])
263 1 : if err != nil {
264 0 : return readers, err
265 0 : }
266 1 : r, err := sstable.NewReader(ctx, readable, readerOpts)
267 1 : if err != nil {
268 0 : return readers, err
269 0 : }
270 1 : readers = append(readers, r)
271 : }
272 1 : return readers, err
273 : }
|
