001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver;
019
020import java.io.IOException;
021import java.util.ArrayList;
022import java.util.Comparator;
023import java.util.List;
024import java.util.PriorityQueue;
025import java.util.function.IntConsumer;
026import org.apache.hadoop.hbase.Cell;
027import org.apache.hadoop.hbase.CellComparator;
028import org.apache.hadoop.hbase.ExtendedCell;
029import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
030import org.apache.yetus.audience.InterfaceAudience;
031import org.slf4j.Logger;
032import org.slf4j.LoggerFactory;
033
034/**
035 * Implements a heap merge across any number of KeyValueScanners.
036 * <p>
037 * Implements KeyValueScanner itself.
038 * <p>
039 * This class is used at the Region level to merge across Stores and at the Store level to merge
040 * across the memstore and StoreFiles.
041 * <p>
042 * In the Region case, we also need InternalScanner.next(List), so this class also implements
043 * InternalScanner. WARNING: As is, if you try to use this as an InternalScanner at the Store level,
044 * you will get runtime exceptions.
045 */
046@InterfaceAudience.Private
047public class KeyValueHeap extends NonReversedNonLazyKeyValueScanner
048  implements KeyValueScanner, InternalScanner {
049  private static final Logger LOG = LoggerFactory.getLogger(KeyValueHeap.class);
050  protected PriorityQueue<KeyValueScanner> heap = null;
051  // Holds the scanners when a ever a eager close() happens. All such eagerly closed
052  // scans are collected and when the final scanner.close() happens will perform the
053  // actual close.
054  protected List<KeyValueScanner> scannersForDelayedClose = null;
055
056  /**
057   * The current sub-scanner, i.e. the one that contains the next key/value to return to the client.
058   * This scanner is NOT included in {@link #heap} (but we frequently add it back to the heap and
059   * pull the new winner out). We maintain an invariant that the current sub-scanner has already
060   * done a real seek, and that current.peek() is always a real key/value (or null) except for the
061   * fake last-key-on-row-column supplied by the multi-column Bloom filter optimization, which is OK
062   * to propagate to StoreScanner. In order to ensure that, always use {@link #pollRealKV()} to
063   * update current.
064   */
065  protected KeyValueScanner current = null;
066
067  protected KVScannerComparator comparator;
068
069  /**
070   * Constructor. This KeyValueHeap will handle closing of passed in KeyValueScanners.
071   */
072  public KeyValueHeap(List<? extends KeyValueScanner> scanners, CellComparator comparator)
073    throws IOException {
074    this(scanners, new KVScannerComparator(comparator));
075  }
076
077  /**
078   * Constructor.
079   */
080  KeyValueHeap(List<? extends KeyValueScanner> scanners, KVScannerComparator comparator)
081    throws IOException {
082    this.comparator = comparator;
083    this.scannersForDelayedClose = new ArrayList<>(scanners.size());
084    if (!scanners.isEmpty()) {
085      this.heap = new PriorityQueue<>(scanners.size(), this.comparator);
086      for (KeyValueScanner scanner : scanners) {
087        if (scanner.peek() != null) {
088          this.heap.add(scanner);
089        } else {
090          this.scannersForDelayedClose.add(scanner);
091        }
092      }
093      this.current = pollRealKV();
094    }
095  }
096
097  @Override
098  public ExtendedCell peek() {
099    if (this.current == null) {
100      return null;
101    }
102    return this.current.peek();
103  }
104
105  boolean isLatestCellFromMemstore() {
106    return !this.current.isFileScanner();
107  }
108
109  @Override
110  public void recordBlockSize(IntConsumer blockSizeConsumer) {
111    this.current.recordBlockSize(blockSizeConsumer);
112  }
113
114  @Override
115  public ExtendedCell next() throws IOException {
116    if (this.current == null) {
117      return null;
118    }
119    ExtendedCell kvReturn = this.current.next();
120    ExtendedCell kvNext = this.current.peek();
121    if (kvNext == null) {
122      this.scannersForDelayedClose.add(this.current);
123      this.current = null;
124      this.current = pollRealKV();
125    } else {
126      KeyValueScanner topScanner = this.heap.peek();
127      // no need to add current back to the heap if it is the only scanner left
128      if (topScanner != null && this.comparator.compare(kvNext, topScanner.peek()) >= 0) {
129        this.heap.add(this.current);
130        this.current = null;
131        this.current = pollRealKV();
132      }
133    }
134    return kvReturn;
135  }
136
137  /**
138   * Gets the next row of keys from the top-most scanner.
139   * <p>
140   * This method takes care of updating the heap.
141   * <p>
142   * This can ONLY be called when you are using Scanners that implement InternalScanner as well as
143   * KeyValueScanner (a {@link StoreScanner}).
144   * @return true if more rows exist after this one, false if scanner is done
145   */
146  @Override
147  public boolean next(List<Cell> result, ScannerContext scannerContext) throws IOException {
148    if (this.current == null) {
149      return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
150    }
151    InternalScanner currentAsInternal = (InternalScanner) this.current;
152    boolean moreCells = currentAsInternal.next(result, scannerContext);
153    Cell pee = this.current.peek();
154
155    /*
156     * By definition, any InternalScanner must return false only when it has no further rows to be
157     * fetched. So, we can close a scanner if it returns false. All existing implementations seem to
158     * be fine with this. It is much more efficient to close scanners which are not needed than keep
159     * them in the heap. This is also required for certain optimizations.
160     */
161
162    if (pee == null || !moreCells) {
163      // add the scanner that is to be closed
164      this.scannersForDelayedClose.add(this.current);
165    } else {
166      this.heap.add(this.current);
167    }
168    this.current = null;
169    this.current = pollRealKV();
170    if (this.current == null) {
171      moreCells = scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
172    }
173    return moreCells;
174  }
175
176  protected static class KVScannerComparator implements Comparator<KeyValueScanner> {
177    protected CellComparator kvComparator;
178
179    /**
180     * Constructor
181     */
182    public KVScannerComparator(CellComparator kvComparator) {
183      this.kvComparator = kvComparator;
184    }
185
186    @Override
187    public int compare(KeyValueScanner left, KeyValueScanner right) {
188      int comparison = compare(left.peek(), right.peek());
189      if (comparison != 0) {
190        return comparison;
191      } else {
192        // Since both the keys are exactly the same, we break the tie in favor of higher ordered
193        // scanner since it'll have newer data. Since higher value should come first, we reverse
194        // sort here.
195        return Long.compare(right.getScannerOrder(), left.getScannerOrder());
196      }
197    }
198
199    /**
200     * Compares two KeyValue
201     * @return less than 0 if left is smaller, 0 if equal etc..
202     */
203    public int compare(Cell left, Cell right) {
204      return this.kvComparator.compare(left, right);
205    }
206
207    /**
208     *     */
209    public CellComparator getComparator() {
210      return this.kvComparator;
211    }
212  }
213
214  @Override
215  public void close() {
216    for (KeyValueScanner scanner : this.scannersForDelayedClose) {
217      scanner.close();
218    }
219    this.scannersForDelayedClose.clear();
220    if (this.current != null) {
221      this.current.close();
222    }
223    if (this.heap != null) {
224      // Order of closing the scanners shouldn't matter here, so simply iterate and close them.
225      for (KeyValueScanner scanner : heap) {
226        scanner.close();
227      }
228    }
229  }
230
231  /**
232   * Seeks all scanners at or below the specified seek key. If we earlied-out of a row, we may end
233   * up skipping values that were never reached yet. Rather than iterating down, we want to give the
234   * opportunity to re-seek.
235   * <p>
236   * As individual scanners may run past their ends, those scanners are automatically closed and
237   * removed from the heap.
238   * <p>
239   * This function (and {@link #reseek(ExtendedCell)}) does not do multi-column Bloom filter and
240   * lazy-seek optimizations. To enable those, call
241   * {@link #requestSeek(ExtendedCell, boolean, boolean)}.
242   * @param seekKey KeyValue to seek at or after
243   * @return true if KeyValues exist at or after specified key, false if not
244   */
245  @Override
246  public boolean seek(ExtendedCell seekKey) throws IOException {
247    return generalizedSeek(false, // This is not a lazy seek
248      seekKey, false, // forward (false: this is not a reseek)
249      false); // Not using Bloom filters
250  }
251
252  /**
253   * This function is identical to the {@link #seek(ExtendedCell)} function except that
254   * scanner.seek(seekKey) is changed to scanner.reseek(seekKey).
255   */
256  @Override
257  public boolean reseek(ExtendedCell seekKey) throws IOException {
258    return generalizedSeek(false, // This is not a lazy seek
259      seekKey, true, // forward (true because this is reseek)
260      false); // Not using Bloom filters
261  }
262
263  /**
264   * {@inheritDoc}
265   */
266  @Override
267  public boolean requestSeek(ExtendedCell key, boolean forward, boolean useBloom)
268    throws IOException {
269    return generalizedSeek(true, key, forward, useBloom);
270  }
271
272  /**
273   * @param isLazy   whether we are trying to seek to exactly the given row/col. Enables Bloom
274   *                 filter and most-recent-file-first optimizations for multi-column get/scan
275   *                 queries.
276   * @param seekKey  key to seek to
277   * @param forward  whether to seek forward (also known as reseek)
278   * @param useBloom whether to optimize seeks using Bloom filters
279   */
280  private boolean generalizedSeek(boolean isLazy, ExtendedCell seekKey, boolean forward,
281    boolean useBloom) throws IOException {
282    if (!isLazy && useBloom) {
283      throw new IllegalArgumentException(
284        "Multi-column Bloom filter " + "optimization requires a lazy seek");
285    }
286
287    if (current == null) {
288      return false;
289    }
290
291    KeyValueScanner scanner = current;
292    try {
293      while (scanner != null) {
294        Cell topKey = scanner.peek();
295        if (comparator.getComparator().compare(seekKey, topKey) <= 0) {
296          // Top KeyValue is at-or-after Seek KeyValue. We only know that all
297          // scanners are at or after seekKey (because fake keys of
298          // scanners where a lazy-seek operation has been done are not greater
299          // than their real next keys) but we still need to enforce our
300          // invariant that the top scanner has done a real seek. This way
301          // StoreScanner and RegionScanner do not have to worry about fake
302          // keys.
303          heap.add(scanner);
304          scanner = null;
305          current = pollRealKV();
306          return current != null;
307        }
308
309        boolean seekResult;
310        if (isLazy && heap.size() > 0) {
311          // If there is only one scanner left, we don't do lazy seek.
312          seekResult = scanner.requestSeek(seekKey, forward, useBloom);
313        } else {
314          seekResult = NonLazyKeyValueScanner.doRealSeek(scanner, seekKey, forward);
315        }
316
317        if (!seekResult) {
318          this.scannersForDelayedClose.add(scanner);
319        } else {
320          heap.add(scanner);
321        }
322        scanner = heap.poll();
323        if (scanner == null) {
324          current = null;
325        }
326      }
327    } catch (Exception e) {
328      if (scanner != null) {
329        try {
330          scanner.close();
331        } catch (Exception ce) {
332          LOG.warn("close KeyValueScanner error", ce);
333        }
334      }
335      throw e;
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 seek has been done on
344   * it. Works by fetching the top sub-scanner, and if it has not done a real seek, making it do so
345   * (which will modify its top KV), putting it back, and repeating this until success. Relies on
346   * the fact that on a lazy seek we set the current key of a StoreFileScanner to a KV that is not
347   * greater than the real next KV to be read from that file, so the scanner that bubbles up to the
348   * top of the heap will have global next KV in this scanner heap if (1) it has done a real seek
349   * and (2) its KV is the top among all top KVs (some of which are fake) in the scanner heap.
350   */
351  protected KeyValueScanner pollRealKV() throws IOException {
352    KeyValueScanner kvScanner = heap.poll();
353    if (kvScanner == null) {
354      return null;
355    }
356
357    while (kvScanner != null && !kvScanner.realSeekDone()) {
358      if (kvScanner.peek() != null) {
359        try {
360          kvScanner.enforceSeek();
361        } catch (IOException ioe) {
362          // Add the item to delayed close set in case it is leak from close
363          this.scannersForDelayedClose.add(kvScanner);
364          throw ioe;
365        }
366        Cell curKV = kvScanner.peek();
367        if (curKV != null) {
368          KeyValueScanner nextEarliestScanner = heap.peek();
369          if (nextEarliestScanner == null) {
370            // The heap is empty. Return the only possible scanner.
371            return kvScanner;
372          }
373
374          // Compare the current scanner to the next scanner. We try to avoid
375          // putting the current one back into the heap if possible.
376          Cell nextKV = nextEarliestScanner.peek();
377          if (nextKV == null || comparator.compare(curKV, nextKV) < 0) {
378            // We already have the scanner with the earliest KV, so return it.
379            return kvScanner;
380          }
381
382          // Otherwise, put the scanner back into the heap and let it compete
383          // against all other scanners (both those that have done a "real
384          // seek" and a "lazy seek").
385          heap.add(kvScanner);
386        } else {
387          // Close the scanner because we did a real seek and found out there
388          // are no more KVs.
389          this.scannersForDelayedClose.add(kvScanner);
390        }
391      } else {
392        // Close the scanner because it has already run out of KVs even before
393        // we had to do a real seek on it.
394        this.scannersForDelayedClose.add(kvScanner);
395      }
396      kvScanner = heap.poll();
397    }
398
399    return kvScanner;
400  }
401
402  /** Returns the current Heap */
403  public PriorityQueue<KeyValueScanner> getHeap() {
404    return this.heap;
405  }
406
407  KeyValueScanner getCurrentForTesting() {
408    return current;
409  }
410
411  @Override
412  public ExtendedCell getNextIndexedKey() {
413    // here we return the next index key from the top scanner
414    return current == null ? null : current.getNextIndexedKey();
415  }
416
417  @Override
418  public void shipped() throws IOException {
419    for (KeyValueScanner scanner : this.scannersForDelayedClose) {
420      scanner.close(); // There wont be further fetch of Cells from these scanners. Just close.
421    }
422    this.scannersForDelayedClose.clear();
423    if (this.current != null) {
424      this.current.shipped();
425    }
426    if (this.heap != null) {
427      for (KeyValueScanner scanner : this.heap) {
428        scanner.shipped();
429      }
430    }
431  }
432}