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.FileNotFoundException;
23  import java.io.IOException;
24  import java.util.ArrayList;
25  import java.util.Collection;
26  import java.util.List;
27  import java.util.concurrent.atomic.AtomicLong;
28  
29  import org.apache.commons.logging.Log;
30  import org.apache.commons.logging.LogFactory;
31  import org.apache.hadoop.hbase.classification.InterfaceAudience;
32  import org.apache.hadoop.hbase.Cell;
33  import org.apache.hadoop.hbase.CellUtil;
34  import org.apache.hadoop.hbase.HConstants;
35  import org.apache.hadoop.hbase.KeyValue;
36  import org.apache.hadoop.hbase.KeyValueUtil;
37  import org.apache.hadoop.hbase.client.Scan;
38  import org.apache.hadoop.hbase.io.TimeRange;
39  import org.apache.hadoop.hbase.io.hfile.HFileScanner;
40  import org.apache.hadoop.hbase.regionserver.StoreFile.Reader;
41  
42  /**
43   * KeyValueScanner adaptor over the Reader.  It also provides hooks into
44   * bloom filter things.
45   */
46  @InterfaceAudience.LimitedPrivate("Coprocessor")
47  public class StoreFileScanner implements KeyValueScanner {
48    private static final Log LOG = LogFactory.getLog(HStore.class);
49  
50    // the reader it comes from:
51    private final StoreFile.Reader reader;
52    private final HFileScanner hfs;
53    private Cell cur = null;
54    private boolean closed = false;
55  
56    private boolean realSeekDone;
57    private boolean delayedReseek;
58    private Cell delayedSeekKV;
59  
60    private boolean enforceMVCC = false;
61    private boolean hasMVCCInfo = false;
62    // A flag represents whether could stop skipping KeyValues for MVCC
63    // if have encountered the next row. Only used for reversed scan
64    private boolean stopSkippingKVsIfNextRow = false;
65  
66    private static AtomicLong seekCount;
67  
68    private ScanQueryMatcher matcher;
69  
70    private long readPt;
71  
72    /**
73     * Implements a {@link KeyValueScanner} on top of the specified {@link HFileScanner}
74     * @param hfs HFile scanner
75     */
76    public StoreFileScanner(StoreFile.Reader reader, HFileScanner hfs, boolean useMVCC,
77        boolean hasMVCC, long readPt) {
78      this.readPt = readPt;
79      this.reader = reader;
80      this.hfs = hfs;
81      this.enforceMVCC = useMVCC;
82      this.hasMVCCInfo = hasMVCC;
83      this.reader.incrementRefCount();
84    }
85  
86    boolean isPrimaryReplica() {
87      return reader.isPrimaryReplicaReader();
88    }
89  
90    /**
91     * Return an array of scanners corresponding to the given
92     * set of store files.
93     */
94    public static List<StoreFileScanner> getScannersForStoreFiles(
95        Collection<StoreFile> files,
96        boolean cacheBlocks,
97        boolean usePread, long readPt) throws IOException {
98      return getScannersForStoreFiles(files, cacheBlocks,
99                                     usePread, false, false, readPt);
100   }
101 
102   /**
103    * Return an array of scanners corresponding to the given set of store files.
104    */
105   public static List<StoreFileScanner> getScannersForStoreFiles(
106       Collection<StoreFile> files, boolean cacheBlocks, boolean usePread,
107       boolean isCompaction, boolean useDropBehind, long readPt) throws IOException {
108     return getScannersForStoreFiles(files, cacheBlocks, usePread, isCompaction,
109         useDropBehind, null, readPt);
110   }
111 
112   /**
113    * Return an array of scanners corresponding to the given set of store files,
114    * And set the ScanQueryMatcher for each store file scanner for further
115    * optimization
116    */
117   public static List<StoreFileScanner> getScannersForStoreFiles(
118       Collection<StoreFile> files, boolean cacheBlocks, boolean usePread,
119       boolean isCompaction, boolean canUseDrop,
120       ScanQueryMatcher matcher, long readPt, boolean isPrimaryReplica) throws IOException {
121     List<StoreFileScanner> scanners = new ArrayList<StoreFileScanner>(
122         files.size());
123     boolean succ = false;
124     try {
125       for (StoreFile file : files) {
126         StoreFile.Reader r = file.createReader(canUseDrop);
127         r.setReplicaStoreFile(isPrimaryReplica);
128         StoreFileScanner scanner = r.getStoreFileScanner(cacheBlocks, usePread,
129             isCompaction, readPt);
130         scanner.setScanQueryMatcher(matcher);
131         scanners.add(scanner);
132       }
133       succ = true;
134     } finally {
135       if (!succ) {
136         for (StoreFileScanner scanner : scanners) {
137           scanner.close();
138         }
139       }
140     }
141     return scanners;
142   }
143 
144   public static List<StoreFileScanner> getScannersForStoreFiles(
145     Collection<StoreFile> files, boolean cacheBlocks, boolean usePread,
146     boolean isCompaction, boolean canUseDrop,
147     ScanQueryMatcher matcher, long readPt) throws IOException {
148     return getScannersForStoreFiles(files, cacheBlocks, usePread, isCompaction, canUseDrop,
149       matcher, readPt, true);
150   }
151 
152   public String toString() {
153     return "StoreFileScanner[" + hfs.toString() + ", cur=" + cur + "]";
154   }
155 
156   public Cell peek() {
157     return cur;
158   }
159 
160   public Cell next() throws IOException {
161     Cell retKey = cur;
162 
163     try {
164       // only seek if we aren't at the end. cur == null implies 'end'.
165       if (cur != null) {
166         hfs.next();
167         setCurrentCell(hfs.getKeyValue());
168         if (hasMVCCInfo || this.reader.isBulkLoaded()) {
169           skipKVsNewerThanReadpoint();
170         }
171       }
172     } catch (FileNotFoundException e) {
173       throw e;
174     } catch(IOException e) {
175       throw new IOException("Could not iterate " + this, e);
176     }
177     return retKey;
178   }
179 
180   public boolean seek(Cell key) throws IOException {
181     if (seekCount != null) seekCount.incrementAndGet();
182 
183     try {
184       try {
185         if(!seekAtOrAfter(hfs, key)) {
186           this.cur = null;
187           return false;
188         }
189 
190         setCurrentCell(hfs.getKeyValue());
191 
192         if (!hasMVCCInfo && this.reader.isBulkLoaded()) {
193           return skipKVsNewerThanReadpoint();
194         } else {
195           return !hasMVCCInfo ? true : skipKVsNewerThanReadpoint();
196         }
197       } finally {
198         realSeekDone = true;
199       }
200     } catch (FileNotFoundException e) {
201       throw e;
202     } catch (IOException ioe) {
203       throw new IOException("Could not seek " + this + " to key " + key, ioe);
204     }
205   }
206 
207   public boolean reseek(Cell key) throws IOException {
208     if (seekCount != null) seekCount.incrementAndGet();
209 
210     try {
211       try {
212         if (!reseekAtOrAfter(hfs, key)) {
213           this.cur = null;
214           return false;
215         }
216         setCurrentCell(hfs.getKeyValue());
217 
218         if (!hasMVCCInfo && this.reader.isBulkLoaded()) {
219           return skipKVsNewerThanReadpoint();
220         } else {
221           return !hasMVCCInfo ? true : skipKVsNewerThanReadpoint();
222         }
223       } finally {
224         realSeekDone = true;
225       }
226     } catch (FileNotFoundException e) {
227       throw e;
228     } catch (IOException ioe) {
229       throw new IOException("Could not reseek " + this + " to key " + key,
230           ioe);
231     }
232   }
233 
234   protected void setCurrentCell(Cell newVal) throws IOException {
235     this.cur = newVal;
236     if (this.cur != null && this.reader.isBulkLoaded()) {
237       CellUtil.setSequenceId(cur, this.reader.getSequenceID());
238     }
239   }
240 
241   protected boolean skipKVsNewerThanReadpoint() throws IOException {
242     // We want to ignore all key-values that are newer than our current
243     // readPoint
244     Cell startKV = cur;
245     while(enforceMVCC
246         && cur != null
247         && (cur.getMvccVersion() > readPt)) {
248       boolean hasNext = hfs.next();
249       setCurrentCell(hfs.getKeyValue());
250       if (hasNext && this.stopSkippingKVsIfNextRow
251           && getComparator().compareRows(cur.getRowArray(), cur.getRowOffset(),
252               cur.getRowLength(), startKV.getRowArray(), startKV.getRowOffset(),
253               startKV.getRowLength()) > 0) {
254         return false;
255       }
256     }
257 
258     if (cur == null) {
259       return false;
260     }
261 
262     return true;
263   }
264 
265   public void close() {
266     cur = null;
267     if (closed) return;    
268     if (this.reader != null) {
269       this.reader.decrementRefCount();
270     }
271     closed = true;
272     this.hfs.close();
273   }
274 
275   /**
276    *
277    * @param s
278    * @param k
279    * @return false if not found or if k is after the end.
280    * @throws IOException
281    */
282   public static boolean seekAtOrAfter(HFileScanner s, Cell k)
283   throws IOException {
284     int result = s.seekTo(k);
285     if(result < 0) {
286       if (result == HConstants.INDEX_KEY_MAGIC) {
287         // using faked key
288         return true;
289       }
290       // Passed KV is smaller than first KV in file, work from start of file
291       return s.seekTo();
292     } else if(result > 0) {
293       // Passed KV is larger than current KV in file, if there is a next
294       // it is the "after", if not then this scanner is done.
295       return s.next();
296     }
297     // Seeked to the exact key
298     return true;
299   }
300 
301   static boolean reseekAtOrAfter(HFileScanner s, Cell k)
302   throws IOException {
303     //This function is similar to seekAtOrAfter function
304     int result = s.reseekTo(k);
305     if (result <= 0) {
306       if (result == HConstants.INDEX_KEY_MAGIC) {
307         // using faked key
308         return true;
309       }
310       // If up to now scanner is not seeked yet, this means passed KV is smaller
311       // than first KV in file, and it is the first time we seek on this file.
312       // So we also need to work from the start of file.
313       if (!s.isSeeked()) {
314         return  s.seekTo();
315       }
316       return true;
317     }
318     // passed KV is larger than current KV in file, if there is a next
319     // it is after, if not then this scanner is done.
320     return s.next();
321   }
322 
323   @Override
324   public long getSequenceID() {
325     return reader.getSequenceID();
326   }
327 
328   /**
329    * Pretend we have done a seek but don't do it yet, if possible. The hope is
330    * that we find requested columns in more recent files and won't have to seek
331    * in older files. Creates a fake key/value with the given row/column and the
332    * highest (most recent) possible timestamp we might get from this file. When
333    * users of such "lazy scanner" need to know the next KV precisely (e.g. when
334    * this scanner is at the top of the heap), they run {@link #enforceSeek()}.
335    * <p>
336    * Note that this function does guarantee that the current KV of this scanner
337    * will be advanced to at least the given KV. Because of this, it does have
338    * to do a real seek in cases when the seek timestamp is older than the
339    * highest timestamp of the file, e.g. when we are trying to seek to the next
340    * row/column and use OLDEST_TIMESTAMP in the seek key.
341    */
342   @Override
343   public boolean requestSeek(Cell kv, boolean forward, boolean useBloom)
344       throws IOException {
345     if (kv.getFamilyLength() == 0) {
346       useBloom = false;
347     }
348 
349     boolean haveToSeek = true;
350     if (useBloom) {
351       // check ROWCOL Bloom filter first.
352       if (reader.getBloomFilterType() == BloomType.ROWCOL) {
353         haveToSeek = reader.passesGeneralBloomFilter(kv.getRowArray(),
354             kv.getRowOffset(), kv.getRowLength(), kv.getQualifierArray(),
355             kv.getQualifierOffset(), kv.getQualifierLength());
356       } else if (this.matcher != null && !matcher.hasNullColumnInQuery() &&
357           ((CellUtil.isDeleteFamily(kv) || CellUtil.isDeleteFamilyVersion(kv)))) {
358         // if there is no such delete family kv in the store file,
359         // then no need to seek.
360         haveToSeek = reader.passesDeleteFamilyBloomFilter(kv.getRowArray(),
361             kv.getRowOffset(), kv.getRowLength());
362       }
363     }
364 
365     delayedReseek = forward;
366     delayedSeekKV = kv;
367 
368     if (haveToSeek) {
369       // This row/column might be in this store file (or we did not use the
370       // Bloom filter), so we still need to seek.
371       realSeekDone = false;
372       long maxTimestampInFile = reader.getMaxTimestamp();
373       long seekTimestamp = kv.getTimestamp();
374       if (seekTimestamp > maxTimestampInFile) {
375         // Create a fake key that is not greater than the real next key.
376         // (Lower timestamps correspond to higher KVs.)
377         // To understand this better, consider that we are asked to seek to
378         // a higher timestamp than the max timestamp in this file. We know that
379         // the next point when we have to consider this file again is when we
380         // pass the max timestamp of this file (with the same row/column).
381         setCurrentCell(KeyValueUtil.createFirstOnRowColTS(kv, maxTimestampInFile));
382       } else {
383         // This will be the case e.g. when we need to seek to the next
384         // row/column, and we don't know exactly what they are, so we set the
385         // seek key's timestamp to OLDEST_TIMESTAMP to skip the rest of this
386         // row/column.
387         enforceSeek();
388       }
389       return cur != null;
390     }
391 
392     // Multi-column Bloom filter optimization.
393     // Create a fake key/value, so that this scanner only bubbles up to the top
394     // of the KeyValueHeap in StoreScanner after we scanned this row/column in
395     // all other store files. The query matcher will then just skip this fake
396     // key/value and the store scanner will progress to the next column. This
397     // is obviously not a "real real" seek, but unlike the fake KV earlier in
398     // this method, we want this to be propagated to ScanQueryMatcher.
399     setCurrentCell(KeyValueUtil.createLastOnRowCol(kv));
400 
401     realSeekDone = true;
402     return true;
403   }
404 
405   Reader getReader() {
406     return reader;
407   }
408 
409   KeyValue.KVComparator getComparator() {
410     return reader.getComparator();
411   }
412 
413   @Override
414   public boolean realSeekDone() {
415     return realSeekDone;
416   }
417 
418   @Override
419   public void enforceSeek() throws IOException {
420     if (realSeekDone)
421       return;
422 
423     if (delayedReseek) {
424       reseek(delayedSeekKV);
425     } else {
426       seek(delayedSeekKV);
427     }
428   }
429 
430   public void setScanQueryMatcher(ScanQueryMatcher matcher) {
431     this.matcher = matcher;
432   }
433 
434   @Override
435   public boolean isFileScanner() {
436     return true;
437   }
438 
439   // Test methods
440 
441   static final long getSeekCount() {
442     return seekCount.get();
443   }
444   static final void instrument() {
445     seekCount = new AtomicLong();
446   }
447 
448   @Override
449   public boolean shouldUseScanner(Scan scan, Store store, long oldestUnexpiredTS) {
450     // if the file has no entries, no need to validate or create a scanner.
451     byte[] cf = store.getFamily().getName();
452     TimeRange timeRange = scan.getColumnFamilyTimeRange().get(cf);
453     if (timeRange == null) {
454       timeRange = scan.getTimeRange();
455     }
456     return reader.passesTimerangeFilter(timeRange, oldestUnexpiredTS) && reader
457         .passesKeyRangeFilter(scan) && reader.passesBloomFilter(scan, scan.getFamilyMap().get(cf));
458   }
459 
460   @Override
461   @SuppressWarnings("deprecation")
462   public boolean seekToPreviousRow(Cell originalKey) throws IOException {
463     try {
464       try {
465         boolean keepSeeking = false;
466         Cell key = originalKey;
467         do {
468           KeyValue seekKey = KeyValueUtil.createFirstOnRow(key.getRowArray(), key.getRowOffset(),
469               key.getRowLength());
470           if (seekCount != null) seekCount.incrementAndGet();
471           if (!hfs.seekBefore(seekKey.getBuffer(), seekKey.getKeyOffset(),
472               seekKey.getKeyLength())) {
473             this.cur = null;
474             return false;
475           }
476           KeyValue firstKeyOfPreviousRow = KeyValueUtil.createFirstOnRow(hfs.getKeyValue()
477               .getRowArray(), hfs.getKeyValue().getRowOffset(), hfs.getKeyValue().getRowLength());
478 
479           if (seekCount != null) seekCount.incrementAndGet();
480           if (!seekAtOrAfter(hfs, firstKeyOfPreviousRow)) {
481             close();
482             return false;
483           }
484 
485           setCurrentCell(hfs.getKeyValue());
486           this.stopSkippingKVsIfNextRow = true;
487           boolean resultOfSkipKVs;
488           try {
489             resultOfSkipKVs = skipKVsNewerThanReadpoint();
490           } finally {
491             this.stopSkippingKVsIfNextRow = false;
492           }
493           if (!resultOfSkipKVs
494               || getComparator().compareRows(cur, firstKeyOfPreviousRow) > 0) {
495             keepSeeking = true;
496             key = firstKeyOfPreviousRow;
497             continue;
498           } else {
499             keepSeeking = false;
500           }
501         } while (keepSeeking);
502         return true;
503       } finally {
504         realSeekDone = true;
505       }
506     } catch (FileNotFoundException e) {
507       throw e;
508     } catch (IOException ioe) {
509       throw new IOException("Could not seekToPreviousRow " + this + " to key "
510           + originalKey, ioe);
511     }
512   }
513 
514   @Override
515   public boolean seekToLastRow() throws IOException {
516     byte[] lastRow = reader.getLastRowKey();
517     if (lastRow == null) {
518       return false;
519     }
520     KeyValue seekKey = KeyValueUtil.createFirstOnRow(lastRow);
521     if (seek(seekKey)) {
522       return true;
523     } else {
524       return seekToPreviousRow(seekKey);
525     }
526   }
527 
528   @Override
529   public boolean backwardSeek(Cell key) throws IOException {
530     seek(key);
531     if (cur == null
532         || getComparator().compareRows(cur.getRowArray(), cur.getRowOffset(),
533             cur.getRowLength(), key.getRowArray(), key.getRowOffset(),
534             key.getRowLength()) > 0) {
535       return seekToPreviousRow(key);
536     }
537     return true;
538   }
539 
540   @Override
541   public Cell getNextIndexedKey() {
542     return hfs.getNextIndexedKey();
543   }
544 }