View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.filter;
19  
20  import java.io.IOException;
21  import java.util.ArrayList;
22  import java.util.Collections;
23  import java.util.List;
24  
25  import org.apache.hadoop.hbase.Cell;
26  import org.apache.hadoop.hbase.HConstants;
27  import org.apache.hadoop.hbase.KeyValueUtil;
28  import org.apache.hadoop.hbase.classification.InterfaceAudience;
29  import org.apache.hadoop.hbase.classification.InterfaceStability;
30  import org.apache.hadoop.hbase.exceptions.DeserializationException;
31  import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
32  import org.apache.hadoop.hbase.util.ByteStringer;
33  import org.apache.hadoop.hbase.util.Bytes;
34  
35  import com.google.protobuf.InvalidProtocolBufferException;
36  
37  /**
38   * Filter to support scan multiple row key ranges. It can construct the row key ranges from the
39   * passed list which can be accessed by each region server.
40   *
41   * HBase is quite efficient when scanning only one small row key range. If user needs to specify
42   * multiple row key ranges in one scan, the typical solutions are: 1. through FilterList which is a
43   * list of row key Filters, 2. using the SQL layer over HBase to join with two table, such as hive,
44   * phoenix etc. However, both solutions are inefficient. Both of them can't utilize the range info
45   * to perform fast forwarding during scan which is quite time consuming. If the number of ranges
46   * are quite big (e.g. millions), join is a proper solution though it is slow. However, there are
47   * cases that user wants to specify a small number of ranges to scan (e.g. <1000 ranges). Both
48   * solutions can't provide satisfactory performance in such case. MultiRowRangeFilter is to support
49   * such usec ase (scan multiple row key ranges), which can construct the row key ranges from user
50   * specified list and perform fast-forwarding during scan. Thus, the scan will be quite efficient.
51   */
52  @InterfaceAudience.Public
53  @InterfaceStability.Evolving
54  public class MultiRowRangeFilter extends FilterBase {
55  
56    private List<RowRange> rangeList;
57  
58    private static final int ROW_BEFORE_FIRST_RANGE = -1;
59    private boolean EXCLUSIVE = false;
60    private boolean done = false;
61    private boolean initialized = false;
62    private int index;
63    private RowRange range;
64    private ReturnCode currentReturnCode;
65  
66    /**
67     * @param list A list of <code>RowRange</code>
68     * @throws java.io.IOException
69     *           throw an exception if the range list is not in an natural order or any
70     *           <code>RowRange</code> is invalid
71     */
72    public MultiRowRangeFilter(List<RowRange> list) throws IOException {
73      this.rangeList = sortAndMerge(list);
74    }
75  
76    @Override
77    public boolean filterAllRemaining() {
78      return done;
79    }
80  
81    public List<RowRange> getRowRanges() {
82      return this.rangeList;
83    }
84  
85    @Override
86    public boolean filterRowKey(byte[] buffer, int offset, int length) {
87      // If it is the first time of running, calculate the current range index for
88      // the row key. If index is out of bound which happens when the start row
89      // user sets is after the largest stop row of the ranges, stop the scan.
90      // If row key is after the current range, find the next range and update index.
91      if (!initialized || !range.contains(buffer, offset, length)) {
92        byte[] rowkey = new byte[length];
93        System.arraycopy(buffer, offset, rowkey, 0, length);
94        index = getNextRangeIndex(rowkey);
95        if (index >= rangeList.size()) {
96          done = true;
97          currentReturnCode = ReturnCode.NEXT_ROW;
98          return false;
99        }
100       if(index != ROW_BEFORE_FIRST_RANGE) {
101         range = rangeList.get(index);
102       } else {
103         range = rangeList.get(0);
104       }
105       if (EXCLUSIVE) {
106         EXCLUSIVE = false;
107         currentReturnCode = ReturnCode.NEXT_ROW;
108         return false;
109       }
110       if (!initialized) {
111         if(index != ROW_BEFORE_FIRST_RANGE) {
112           currentReturnCode = ReturnCode.INCLUDE;
113         } else {
114           currentReturnCode = ReturnCode.SEEK_NEXT_USING_HINT;
115         }
116         initialized = true;
117       } else {
118         if (range.contains(buffer, offset, length)) {
119           currentReturnCode = ReturnCode.INCLUDE;
120         } else {
121           currentReturnCode = ReturnCode.SEEK_NEXT_USING_HINT;
122         }
123       }
124     } else {
125       currentReturnCode = ReturnCode.INCLUDE;
126     }
127     return false;
128   }
129 
130   @Override
131   public ReturnCode filterKeyValue(Cell ignored) {
132     return currentReturnCode;
133   }
134 
135   @Override
136   public Cell getNextCellHint(Cell currentKV) {
137     // skip to the next range's start row
138     return KeyValueUtil.createFirstOnRow(range.startRow);
139   }
140 
141   /**
142    * @return The filter serialized using pb
143    */
144   @Override
145   public byte[] toByteArray() {
146     FilterProtos.MultiRowRangeFilter.Builder builder = FilterProtos.MultiRowRangeFilter
147         .newBuilder();
148     for (RowRange range : rangeList) {
149       if (range != null) {
150         FilterProtos.RowRange.Builder rangebuilder = FilterProtos.RowRange.newBuilder();
151         if (range.startRow != null)
152           rangebuilder.setStartRow(ByteStringer.wrap(range.startRow));
153         rangebuilder.setStartRowInclusive(range.startRowInclusive);
154         if (range.stopRow != null)
155           rangebuilder.setStopRow(ByteStringer.wrap(range.stopRow));
156         rangebuilder.setStopRowInclusive(range.stopRowInclusive);
157         builder.addRowRangeList(rangebuilder.build());
158       }
159     }
160     return builder.build().toByteArray();
161   }
162 
163   /**
164    * @param pbBytes A pb serialized instance
165    * @return An instance of MultiRowRangeFilter
166    * @throws org.apache.hadoop.hbase.exceptions.DeserializationException
167    */
168   public static MultiRowRangeFilter parseFrom(final byte[] pbBytes)
169       throws DeserializationException {
170     FilterProtos.MultiRowRangeFilter proto;
171     try {
172       proto = FilterProtos.MultiRowRangeFilter.parseFrom(pbBytes);
173     } catch (InvalidProtocolBufferException e) {
174       throw new DeserializationException(e);
175     }
176     int length = proto.getRowRangeListCount();
177     List<FilterProtos.RowRange> rangeProtos = proto.getRowRangeListList();
178     List<RowRange> rangeList = new ArrayList<RowRange>(length);
179     for (FilterProtos.RowRange rangeProto : rangeProtos) {
180       RowRange range = new RowRange(rangeProto.hasStartRow() ? rangeProto.getStartRow()
181           .toByteArray() : null, rangeProto.getStartRowInclusive(), rangeProto.hasStopRow() ?
182               rangeProto.getStopRow().toByteArray() : null, rangeProto.getStopRowInclusive());
183       rangeList.add(range);
184     }
185     try {
186       return new MultiRowRangeFilter(rangeList);
187     } catch (IOException e) {
188       throw new DeserializationException("Fail to instantiate the MultiRowRangeFilter", e);
189     }
190   }
191 
192   /**
193    * @param o the filter to compare
194    * @return true if and only if the fields of the filter that are serialized are equal to the
195    *         corresponding fields in other. Used for testing.
196    */
197   @Override
198   boolean areSerializedFieldsEqual(Filter o) {
199     if (o == this)
200       return true;
201     if (!(o instanceof MultiRowRangeFilter))
202       return false;
203 
204     MultiRowRangeFilter other = (MultiRowRangeFilter) o;
205     if (this.rangeList.size() != other.rangeList.size())
206       return false;
207     for (int i = 0; i < rangeList.size(); ++i) {
208       RowRange thisRange = this.rangeList.get(i);
209       RowRange otherRange = other.rangeList.get(i);
210       if (!(Bytes.equals(thisRange.startRow, otherRange.startRow) && Bytes.equals(
211           thisRange.stopRow, otherRange.stopRow) && (thisRange.startRowInclusive ==
212           otherRange.startRowInclusive) && (thisRange.stopRowInclusive ==
213           otherRange.stopRowInclusive))) {
214         return false;
215       }
216     }
217     return true;
218   }
219 
220   /**
221    * calculate the position where the row key in the ranges list.
222    *
223    * @param rowKey the row key to calculate
224    * @return index the position of the row key
225    */
226   private int getNextRangeIndex(byte[] rowKey) {
227     RowRange temp = new RowRange(rowKey, true, null, true);
228     int index = Collections.binarySearch(rangeList, temp);
229     if (index < 0) {
230       int insertionPosition = -index - 1;
231       // check if the row key in the range before the insertion position
232       if (insertionPosition != 0 && rangeList.get(insertionPosition - 1).contains(rowKey)) {
233         return insertionPosition - 1;
234       }
235       // check if the row key is before the first range
236       if (insertionPosition == 0 && !rangeList.get(insertionPosition).contains(rowKey)) {
237         return ROW_BEFORE_FIRST_RANGE;
238       }
239       if (!initialized) {
240         initialized = true;
241       }
242       return insertionPosition;
243     }
244     // the row key equals one of the start keys, and the the range exclude the start key
245     if(rangeList.get(index).startRowInclusive == false) {
246       EXCLUSIVE = true;
247     }
248     return index;
249   }
250 
251   /**
252    * sort the ranges and if the ranges with overlap, then merge them.
253    *
254    * @param ranges the list of ranges to sort and merge.
255    * @return the ranges after sort and merge.
256    */
257   public static List<RowRange> sortAndMerge(List<RowRange> ranges) {
258     if (ranges.size() == 0) {
259       throw new IllegalArgumentException("No ranges found.");
260     }
261     List<RowRange> invalidRanges = new ArrayList<RowRange>();
262     List<RowRange> newRanges = new ArrayList<RowRange>(ranges.size());
263     Collections.sort(ranges);
264     if(ranges.get(0).isValid()) {
265       if (ranges.size() == 1) {
266         newRanges.add(ranges.get(0));
267       }
268     } else {
269       invalidRanges.add(ranges.get(0));
270     }
271 
272     byte[] lastStartRow = ranges.get(0).startRow;
273     boolean lastStartRowInclusive = ranges.get(0).startRowInclusive;
274     byte[] lastStopRow = ranges.get(0).stopRow;
275     boolean lastStopRowInclusive = ranges.get(0).stopRowInclusive;
276     int i = 1;
277     for (; i < ranges.size(); i++) {
278       RowRange range = ranges.get(i);
279       if (!range.isValid()) {
280         invalidRanges.add(range);
281       }
282       if(Bytes.equals(lastStopRow, HConstants.EMPTY_BYTE_ARRAY)) {
283         newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
284             lastStopRowInclusive));
285         break;
286       }
287       // with overlap in the ranges
288       if ((Bytes.compareTo(lastStopRow, range.startRow) > 0) ||
289           (Bytes.compareTo(lastStopRow, range.startRow) == 0 && !(lastStopRowInclusive == false &&
290           range.isStartRowInclusive() == false))) {
291         if(Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
292           newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, range.stopRow,
293               range.stopRowInclusive));
294           break;
295         }
296         // if first range contains second range, ignore the second range
297         if (Bytes.compareTo(lastStopRow, range.stopRow) >= 0) {
298           if((Bytes.compareTo(lastStopRow, range.stopRow) == 0)) {
299             if(lastStopRowInclusive == true || range.stopRowInclusive == true) {
300               lastStopRowInclusive = true;
301             }
302           }
303           if ((i + 1) == ranges.size()) {
304             newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
305                 lastStopRowInclusive));
306           }
307         } else {
308           lastStopRow = range.stopRow;
309           lastStopRowInclusive = range.stopRowInclusive;
310           if ((i + 1) < ranges.size()) {
311             i++;
312             range = ranges.get(i);
313             if (!range.isValid()) {
314               invalidRanges.add(range);
315             }
316           } else {
317             newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
318                 lastStopRowInclusive));
319             break;
320           }
321           while ((Bytes.compareTo(lastStopRow, range.startRow) > 0) ||
322               (Bytes.compareTo(lastStopRow, range.startRow) == 0 &&
323               (lastStopRowInclusive == true || range.startRowInclusive==true))) {
324             if(Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
325               break;
326             }
327             // if this first range contain second range, ignore the second range
328             if (Bytes.compareTo(lastStopRow, range.stopRow) >= 0) {
329               if(lastStopRowInclusive == true || range.stopRowInclusive == true) {
330                 lastStopRowInclusive = true;
331               }
332               i++;
333               if (i < ranges.size()) {
334                 range = ranges.get(i);
335                 if (!range.isValid()) {
336                   invalidRanges.add(range);
337                 }
338               } else {
339                 break;
340               }
341             } else {
342               lastStopRow = range.stopRow;
343               lastStopRowInclusive = range.stopRowInclusive;
344               i++;
345               if (i < ranges.size()) {
346                 range = ranges.get(i);
347                 if (!range.isValid()) {
348                   invalidRanges.add(range);
349                 }
350               } else {
351                 break;
352               }
353             }
354           }
355           if(Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
356             if((Bytes.compareTo(lastStopRow, range.startRow) < 0) ||
357                 (Bytes.compareTo(lastStopRow, range.startRow) == 0 &&
358                 lastStopRowInclusive == false && range.startRowInclusive == false)) {
359               newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
360                   lastStopRowInclusive));
361               newRanges.add(range);
362             } else {
363               newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, range.stopRow,
364                   range.stopRowInclusive));
365               break;
366             }
367           }
368           newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
369               lastStopRowInclusive));
370           if ((i + 1) == ranges.size()) {
371             newRanges.add(range);
372           }
373           lastStartRow = range.startRow;
374           lastStartRowInclusive = range.startRowInclusive;
375           lastStopRow = range.stopRow;
376           lastStopRowInclusive = range.stopRowInclusive;
377         }
378       } else {
379         newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
380             lastStopRowInclusive));
381         if ((i + 1) == ranges.size()) {
382           newRanges.add(range);
383         }
384         lastStartRow = range.startRow;
385         lastStartRowInclusive = range.startRowInclusive;
386         lastStopRow = range.stopRow;
387         lastStopRowInclusive = range.stopRowInclusive;
388       }
389     }
390     // check the remaining ranges
391     for(int j=i; j < ranges.size(); j++) {
392       if(!ranges.get(j).isValid()) {
393         invalidRanges.add(ranges.get(j));
394       }
395     }
396     // if invalid range exists, throw the exception
397     if (invalidRanges.size() != 0) {
398       throwExceptionForInvalidRanges(invalidRanges, true);
399     }
400     // If no valid ranges found, throw the exception
401     if(newRanges.size() == 0) {
402       throw new IllegalArgumentException("No valid ranges found.");
403     }
404     return newRanges;
405   }
406 
407   private static void throwExceptionForInvalidRanges(List<RowRange> invalidRanges,
408       boolean details) {
409     StringBuilder sb = new StringBuilder();
410     sb.append(invalidRanges.size()).append(" invaild ranges.\n");
411     if (details) {
412       for (RowRange range : invalidRanges) {
413         sb.append(
414             "Invalid range: start row => " + Bytes.toString(range.startRow) + ", stop row => "
415                 + Bytes.toString(range.stopRow)).append('\n');
416       }
417     }
418     throw new IllegalArgumentException(sb.toString());
419   }
420 
421   @InterfaceAudience.Public
422   @InterfaceStability.Evolving
423   public static class RowRange implements Comparable<RowRange> {
424     private byte[] startRow;
425     private boolean startRowInclusive = true;
426     private byte[] stopRow;
427     private boolean stopRowInclusive = false;
428 
429     public RowRange() {
430     }
431     /**
432      * If the startRow is empty or null, set it to HConstants.EMPTY_BYTE_ARRAY, means begin at the
433      * start row of the table. If the stopRow is empty or null, set it to
434      * HConstants.EMPTY_BYTE_ARRAY, means end of the last row of table.
435      */
436     public RowRange(String startRow, boolean startRowInclusive, String stopRow,
437         boolean stopRowInclusive) {
438       this((startRow == null || startRow.isEmpty()) ? HConstants.EMPTY_BYTE_ARRAY :
439         Bytes.toBytes(startRow), startRowInclusive,
440         (stopRow == null || stopRow.isEmpty()) ? HConstants.EMPTY_BYTE_ARRAY :
441         Bytes.toBytes(stopRow), stopRowInclusive);
442     }
443 
444     public RowRange(byte[] startRow,  boolean startRowInclusive, byte[] stopRow,
445         boolean stopRowInclusive) {
446       this.startRow = (startRow == null) ? HConstants.EMPTY_BYTE_ARRAY : startRow;
447       this.startRowInclusive = startRowInclusive;
448       this.stopRow = (stopRow == null) ? HConstants.EMPTY_BYTE_ARRAY :stopRow;
449       this.stopRowInclusive = stopRowInclusive;
450     }
451 
452     public byte[] getStartRow() {
453       return startRow;
454     }
455 
456     public byte[] getStopRow() {
457       return stopRow;
458     }
459 
460     /**
461      * @return if start row is inclusive.
462      */
463     public boolean isStartRowInclusive() {
464       return startRowInclusive;
465     }
466 
467     /**
468      * @return if stop row is inclusive.
469      */
470     public boolean isStopRowInclusive() {
471       return stopRowInclusive;
472     }
473 
474     public boolean contains(byte[] row) {
475       return contains(row, 0, row.length);
476     }
477 
478     public boolean contains(byte[] buffer, int offset, int length) {
479       if(startRowInclusive) {
480         if(stopRowInclusive) {
481           return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) >= 0
482               && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY) ||
483                   Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) <= 0);
484         } else {
485           return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) >= 0
486               && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY) ||
487                   Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) < 0);
488         }
489       } else {
490         if(stopRowInclusive) {
491           return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) > 0
492               && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY) ||
493                   Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) <= 0);
494         } else {
495           return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) > 0
496               && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY) ||
497                   Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) < 0);
498         }
499       }
500     }
501 
502     @Override
503     @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="EQ_COMPARETO_USE_OBJECT_EQUALS",
504       justification="This compareTo is not of this Object, but of referenced RowRange")
505     public int compareTo(RowRange other) {
506       return Bytes.compareTo(this.startRow, other.startRow);
507     }
508 
509     public boolean isValid() {
510       return Bytes.equals(startRow, HConstants.EMPTY_BYTE_ARRAY)
511           || Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
512           || Bytes.compareTo(startRow, stopRow) < 0
513           || (Bytes.compareTo(startRow, stopRow) == 0 && stopRowInclusive == true);
514     }
515   }
516 }