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