View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  
20  package org.apache.hadoop.hbase.regionserver;
21  
22  import java.io.IOException;
23  import java.util.Comparator;
24  import java.util.HashSet;
25  import java.util.List;
26  import java.util.PriorityQueue;
27  import java.util.Set;
28  
29  import org.apache.hadoop.hbase.Cell;
30  import org.apache.hadoop.hbase.CellComparator;
31  import org.apache.hadoop.hbase.classification.InterfaceAudience;
32  import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
33  
34  /**
35   * Implements a heap merge across any number of KeyValueScanners.
36   * <p>
37   * Implements KeyValueScanner itself.
38   * <p>
39   * This class is used at the Region level to merge across Stores
40   * and at the Store level to merge across the memstore and StoreFiles.
41   * <p>
42   * In the Region case, we also need InternalScanner.next(List), so this class
43   * also implements InternalScanner.  WARNING: As is, if you try to use this
44   * as an InternalScanner at the Store level, you will get runtime exceptions.
45   */
46  @InterfaceAudience.Private
47  public class KeyValueHeap extends NonReversedNonLazyKeyValueScanner
48      implements KeyValueScanner, InternalScanner {
49    protected PriorityQueue<KeyValueScanner> heap = null;
50  
51    /**
52     * The current sub-scanner, i.e. the one that contains the next key/value
53     * to return to the client. This scanner is NOT included in {@link #heap}
54     * (but we frequently add it back to the heap and pull the new winner out).
55     * We maintain an invariant that the current sub-scanner has already done
56     * a real seek, and that current.peek() is always a real key/value (or null)
57     * except for the fake last-key-on-row-column supplied by the multi-column
58     * Bloom filter optimization, which is OK to propagate to StoreScanner. In
59     * order to ensure that, always use {@link #pollRealKV()} to update current.
60     */
61    protected KeyValueScanner current = null;
62  
63    protected KVScannerComparator comparator;
64  
65    protected Set<KeyValueScanner> scannersForDelayedClose = new HashSet<KeyValueScanner>();
66  
67    /**
68     * Constructor.  This KeyValueHeap will handle closing of passed in
69     * KeyValueScanners.
70     * @param scanners
71     * @param comparator
72     */
73    public KeyValueHeap(List<? extends KeyValueScanner> scanners,
74        CellComparator comparator) throws IOException {
75      this(scanners, new KVScannerComparator(comparator));
76    }
77  
78    /**
79     * Constructor.
80     * @param scanners
81     * @param comparator
82     * @throws IOException
83     */
84    KeyValueHeap(List<? extends KeyValueScanner> scanners,
85        KVScannerComparator comparator) throws IOException {
86      this.comparator = comparator;
87      if (!scanners.isEmpty()) {
88        this.heap = new PriorityQueue<KeyValueScanner>(scanners.size(),
89            this.comparator);
90        for (KeyValueScanner scanner : scanners) {
91          if (scanner.peek() != null) {
92            this.heap.add(scanner);
93          } else {
94            this.scannersForDelayedClose.add(scanner);
95          }
96        }
97        this.current = pollRealKV();
98      }
99    }
100 
101   public Cell peek() {
102     if (this.current == null) {
103       return null;
104     }
105     return this.current.peek();
106   }
107 
108   public Cell next()  throws IOException {
109     if(this.current == null) {
110       return null;
111     }
112     Cell kvReturn = this.current.next();
113     Cell kvNext = this.current.peek();
114     if (kvNext == null) {
115       this.scannersForDelayedClose.add(this.current);
116       this.current = null;
117       this.current = pollRealKV();
118     } else {
119       KeyValueScanner topScanner = this.heap.peek();
120       // no need to add current back to the heap if it is the only scanner left
121       if (topScanner != null && this.comparator.compare(kvNext, topScanner.peek()) >= 0) {
122         this.heap.add(this.current);
123         this.current = null;
124         this.current = pollRealKV();
125       }
126     }
127     return kvReturn;
128   }
129 
130   /**
131    * Gets the next row of keys from the top-most scanner.
132    * <p>
133    * This method takes care of updating the heap.
134    * <p>
135    * This can ONLY be called when you are using Scanners that implement InternalScanner as well as
136    * KeyValueScanner (a {@link StoreScanner}).
137    * @param result
138    * @return true if more rows exist after this one, false if scanner is done
139    */
140   @Override
141   public boolean next(List<Cell> result) throws IOException {
142     return next(result, NoLimitScannerContext.getInstance());
143   }
144 
145   @Override
146   public boolean next(List<Cell> result, ScannerContext scannerContext) throws IOException {
147     if (this.current == null) {
148       return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
149     }
150     InternalScanner currentAsInternal = (InternalScanner)this.current;
151     boolean moreCells = currentAsInternal.next(result, scannerContext);
152     Cell pee = this.current.peek();
153 
154     /*
155      * By definition, any InternalScanner must return false only when it has no
156      * further rows to be fetched. So, we can close a scanner if it returns
157      * false. All existing implementations seem to be fine with this. It is much
158      * more efficient to close scanners which are not needed than keep them in
159      * the heap. This is also required for certain optimizations.
160      */
161 
162     if (pee == null || !moreCells) {
163       this.scannersForDelayedClose.add(this.current);
164     } else {
165       this.heap.add(this.current);
166     }
167     this.current = null;
168     this.current = pollRealKV();
169     if (this.current == null) {
170       moreCells = scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
171     }
172     return moreCells;
173   }
174 
175   protected static class KVScannerComparator implements Comparator<KeyValueScanner> {
176     protected CellComparator kvComparator;
177     /**
178      * Constructor
179      * @param kvComparator
180      */
181     public KVScannerComparator(CellComparator kvComparator) {
182       this.kvComparator = kvComparator;
183     }
184     public int compare(KeyValueScanner left, KeyValueScanner right) {
185       int comparison = compare(left.peek(), right.peek());
186       if (comparison != 0) {
187         return comparison;
188       } else {
189         // Since both the keys are exactly the same, we break the tie in favor
190         // of the key which came latest.
191         long leftSequenceID = left.getSequenceID();
192         long rightSequenceID = right.getSequenceID();
193         if (leftSequenceID > rightSequenceID) {
194           return -1;
195         } else if (leftSequenceID < rightSequenceID) {
196           return 1;
197         } else {
198           return 0;
199         }
200       }
201     }
202     /**
203      * Compares two KeyValue
204      * @param left
205      * @param right
206      * @return less than 0 if left is smaller, 0 if equal etc..
207      */
208     public int compare(Cell left, Cell right) {
209       return this.kvComparator.compare(left, right);
210     }
211     /**
212      * @return KVComparator
213      */
214     public CellComparator getComparator() {
215       return this.kvComparator;
216     }
217   }
218 
219   public void close() {
220     for (KeyValueScanner scanner : this.scannersForDelayedClose) {
221       scanner.close();
222     }
223     this.scannersForDelayedClose.clear();
224     if (this.current != null) {
225       this.current.close();
226     }
227     if (this.heap != null) {
228       KeyValueScanner scanner;
229       while ((scanner = this.heap.poll()) != null) {
230         scanner.close();
231       }
232     }
233   }
234 
235   /**
236    * Seeks all scanners at or below the specified seek key.  If we earlied-out
237    * of a row, we may end up skipping values that were never reached yet.
238    * Rather than iterating down, we want to give the opportunity to re-seek.
239    * <p>
240    * As individual scanners may run past their ends, those scanners are
241    * automatically closed and removed from the heap.
242    * <p>
243    * This function (and {@link #reseek(Cell)}) does not do multi-column
244    * Bloom filter and lazy-seek optimizations. To enable those, call
245    * {@link #requestSeek(Cell, boolean, boolean)}.
246    * @param seekKey KeyValue to seek at or after
247    * @return true if KeyValues exist at or after specified key, false if not
248    * @throws IOException
249    */
250   @Override
251   public boolean seek(Cell seekKey) throws IOException {
252     return generalizedSeek(false,    // This is not a lazy seek
253                            seekKey,
254                            false,    // forward (false: this is not a reseek)
255                            false);   // Not using Bloom filters
256   }
257 
258   /**
259    * This function is identical to the {@link #seek(Cell)} function except
260    * that scanner.seek(seekKey) is changed to scanner.reseek(seekKey).
261    */
262   @Override
263   public boolean reseek(Cell seekKey) throws IOException {
264     return generalizedSeek(false,    // This is not a lazy seek
265                            seekKey,
266                            true,     // forward (true because this is reseek)
267                            false);   // Not using Bloom filters
268   }
269 
270   /**
271    * {@inheritDoc}
272    */
273   @Override
274   public boolean requestSeek(Cell key, boolean forward,
275       boolean useBloom) throws IOException {
276     return generalizedSeek(true, key, forward, useBloom);
277   }
278 
279   /**
280    * @param isLazy whether we are trying to seek to exactly the given row/col.
281    *          Enables Bloom filter and most-recent-file-first optimizations for
282    *          multi-column get/scan queries.
283    * @param seekKey key to seek to
284    * @param forward whether to seek forward (also known as reseek)
285    * @param useBloom whether to optimize seeks using Bloom filters
286    */
287   private boolean generalizedSeek(boolean isLazy, Cell seekKey,
288       boolean forward, boolean useBloom) throws IOException {
289     if (!isLazy && useBloom) {
290       throw new IllegalArgumentException("Multi-column Bloom filter " +
291           "optimization requires a lazy seek");
292     }
293 
294     if (current == null) {
295       return false;
296     }
297     heap.add(current);
298     current = null;
299 
300     KeyValueScanner scanner;
301     while ((scanner = heap.poll()) != null) {
302       Cell topKey = scanner.peek();
303       if (comparator.getComparator().compare(seekKey, topKey) <= 0) {
304         // Top KeyValue is at-or-after Seek KeyValue. We only know that all
305         // scanners are at or after seekKey (because fake keys of
306         // scanners where a lazy-seek operation has been done are not greater
307         // than their real next keys) but we still need to enforce our
308         // invariant that the top scanner has done a real seek. This way
309         // StoreScanner and RegionScanner do not have to worry about fake keys.
310         heap.add(scanner);
311         current = pollRealKV();
312         return current != null;
313       }
314 
315       boolean seekResult;
316       if (isLazy && heap.size() > 0) {
317         // If there is only one scanner left, we don't do lazy seek.
318         seekResult = scanner.requestSeek(seekKey, forward, useBloom);
319       } else {
320         seekResult = NonLazyKeyValueScanner.doRealSeek(
321             scanner, seekKey, forward);
322       }
323 
324       if (!seekResult) {
325         this.scannersForDelayedClose.add(scanner);
326       } else {
327         heap.add(scanner);
328       }
329     }
330 
331     // Heap is returning empty, scanner is done
332     return false;
333   }
334 
335   /**
336    * Fetches the top sub-scanner from the priority queue, ensuring that a real
337    * seek has been done on it. Works by fetching the top sub-scanner, and if it
338    * has not done a real seek, making it do so (which will modify its top KV),
339    * putting it back, and repeating this until success. Relies on the fact that
340    * on a lazy seek we set the current key of a StoreFileScanner to a KV that
341    * is not greater than the real next KV to be read from that file, so the
342    * scanner that bubbles up to the top of the heap will have global next KV in
343    * this scanner heap if (1) it has done a real seek and (2) its KV is the top
344    * among all top KVs (some of which are fake) in the scanner heap.
345    */
346   protected KeyValueScanner pollRealKV() throws IOException {
347     KeyValueScanner kvScanner = heap.poll();
348     if (kvScanner == null) {
349       return null;
350     }
351 
352     while (kvScanner != null && !kvScanner.realSeekDone()) {
353       if (kvScanner.peek() != null) {
354         try {
355           kvScanner.enforceSeek();
356         } catch (IOException ioe) {
357           // Add the item to delayed close set in case it is leak from close
358           this.scannersForDelayedClose.add(kvScanner);
359           throw ioe;
360         }
361         Cell curKV = kvScanner.peek();
362         if (curKV != null) {
363           KeyValueScanner nextEarliestScanner = heap.peek();
364           if (nextEarliestScanner == null) {
365             // The heap is empty. Return the only possible scanner.
366             return kvScanner;
367           }
368 
369           // Compare the current scanner to the next scanner. We try to avoid
370           // putting the current one back into the heap if possible.
371           Cell nextKV = nextEarliestScanner.peek();
372           if (nextKV == null || comparator.compare(curKV, nextKV) < 0) {
373             // We already have the scanner with the earliest KV, so return it.
374             return kvScanner;
375           }
376 
377           // Otherwise, put the scanner back into the heap and let it compete
378           // against all other scanners (both those that have done a "real
379           // seek" and a "lazy seek").
380           heap.add(kvScanner);
381         } else {
382           // Close the scanner because we did a real seek and found out there
383           // are no more KVs.
384           this.scannersForDelayedClose.add(kvScanner);
385         }
386       } else {
387         // Close the scanner because it has already run out of KVs even before
388         // we had to do a real seek on it.
389         this.scannersForDelayedClose.add(kvScanner);
390       }
391       kvScanner = heap.poll();
392     }
393 
394     return kvScanner;
395   }
396 
397   /**
398    * @return the current Heap
399    */
400   public PriorityQueue<KeyValueScanner> getHeap() {
401     return this.heap;
402   }
403 
404   @Override
405   public long getSequenceID() {
406     return 0;
407   }
408 
409   KeyValueScanner getCurrentForTesting() {
410     return current;
411   }
412 
413   @Override
414   public Cell getNextIndexedKey() {
415     // here we return the next index key from the top scanner
416     return current == null ? null : current.getNextIndexedKey();
417   }
418 
419   @Override
420   public void shipped() throws IOException {
421     for (KeyValueScanner scanner : this.scannersForDelayedClose) {
422       scanner.close(); // There wont be further fetch of Cells from these scanners. Just close.
423     }
424     this.scannersForDelayedClose.clear();
425     if (this.current != null) {
426       this.current.shipped();
427     }
428     if (this.heap != null) {
429       for (KeyValueScanner scanner : this.heap) {
430         scanner.shipped();
431       }
432     }
433   }
434 }