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.filter;
019
020import java.util.ArrayList;
021import java.util.Collections;
022import java.util.List;
023import java.util.Objects;
024import org.apache.hadoop.hbase.Cell;
025import org.apache.hadoop.hbase.CellUtil;
026import org.apache.hadoop.hbase.HConstants;
027import org.apache.hadoop.hbase.PrivateCellUtil;
028import org.apache.hadoop.hbase.client.ClientUtil;
029import org.apache.hadoop.hbase.exceptions.DeserializationException;
030import org.apache.hadoop.hbase.util.Bytes;
031import org.apache.yetus.audience.InterfaceAudience;
032
033import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
034import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
035
036import org.apache.hadoop.hbase.shaded.protobuf.generated.FilterProtos;
037
038/**
039 * Filter to support scan multiple row key ranges. It can construct the row key ranges from the
040 * passed list which can be accessed by each region server. HBase is quite efficient when scanning
041 * only one small row key range. If user needs to specify multiple row key ranges in one scan, the
042 * typical solutions are: 1. through FilterList which is a list of row key Filters, 2. using the SQL
043 * layer over HBase to join with two table, such as hive, phoenix etc. However, both solutions are
044 * inefficient. Both of them can't utilize the range info to perform fast forwarding during scan
045 * which is quite time consuming. If the number of ranges are quite big (e.g. millions), join is a
046 * proper solution though it is slow. However, there are cases that user wants to specify a small
047 * number of ranges to scan (e.g. <1000 ranges). Both solutions can't provide satisfactory
048 * performance in such case. MultiRowRangeFilter is to support such usec ase (scan multiple row key
049 * ranges), which can construct the row key ranges from user specified list and perform
050 * fast-forwarding during scan. Thus, the scan will be quite efficient.
051 */
052@InterfaceAudience.Public
053public class MultiRowRangeFilter extends FilterBase {
054
055  private static final int ROW_BEFORE_FIRST_RANGE = -1;
056
057  private final List<RowRange> rangeList;
058  private final RangeIteration ranges;
059
060  private boolean done = false;
061  private int index;
062  private BasicRowRange range;
063  private ReturnCode currentReturnCode;
064
065  /**
066   * @param list A list of <code>RowRange</code>
067   */
068  public MultiRowRangeFilter(List<RowRange> list) {
069    // We don't use rangeList anywhere else, but keeping it lets us pay a little
070    // memory to avoid touching the serialization logic.
071    this.rangeList = Collections.unmodifiableList(sortAndMerge(list));
072    this.ranges = new RangeIteration(rangeList);
073  }
074
075  /**
076   * Constructor for creating a <code>MultiRowRangeFilter</code> from multiple rowkey prefixes. As
077   * <code>MultiRowRangeFilter</code> javadoc says (See the solution 1 of the first statement), if
078   * you try to create a filter list that scans row keys corresponding to given prefixes (e.g.,
079   * <code>FilterList</code> composed of multiple <code>PrefixFilter</code>s), this constructor
080   * provides a way to avoid creating an inefficient one.
081   * @param rowKeyPrefixes the array of byte array
082   */
083  public MultiRowRangeFilter(byte[][] rowKeyPrefixes) {
084    this(createRangeListFromRowKeyPrefixes(rowKeyPrefixes));
085  }
086
087  private static List<RowRange> createRangeListFromRowKeyPrefixes(byte[][] rowKeyPrefixes) {
088    if (rowKeyPrefixes == null) {
089      throw new IllegalArgumentException("Invalid rowkey prefixes");
090    }
091
092    List<RowRange> list = new ArrayList<>();
093    for (byte[] rowKeyPrefix : rowKeyPrefixes) {
094      byte[] stopRow = ClientUtil.calculateTheClosestNextRowKeyForPrefix(rowKeyPrefix);
095      list.add(new RowRange(rowKeyPrefix, true, stopRow, false));
096    }
097    return list;
098  }
099
100  public List<RowRange> getRowRanges() {
101    // Used by hbase-rest
102    return this.rangeList;
103  }
104
105  @Override
106  public boolean filterAllRemaining() {
107    return done;
108  }
109
110  @Override
111  public boolean filterRowKey(Cell firstRowCell) {
112    if (filterAllRemaining()) return true;
113
114    // N.b. We can only do this after we're iterating over records. If we try to do
115    // it before, the Scan (and this filter) may not yet be fully initialized. This is a
116    // wart on Filter and something that'd be nice to clean up (like CP's in HBase2.0)
117    if (!ranges.isInitialized()) {
118      ranges.initialize(isReversed());
119    }
120
121    // If it is the first time of running, calculate the current range index for
122    // the row key. If index is out of bound which happens when the start row
123    // user sets is after the largest stop row of the ranges, stop the scan.
124    // If row key is after the current range, find the next range and update index.
125    byte[] rowArr = firstRowCell.getRowArray();
126    int length = firstRowCell.getRowLength();
127    int offset = firstRowCell.getRowOffset();
128    if (!ranges.hasFoundFirstRange() || !range.contains(rowArr, offset, length)) {
129      byte[] rowkey = CellUtil.cloneRow(firstRowCell);
130      index = ranges.getNextRangeIndex(rowkey);
131      if (ranges.isIterationComplete(index)) {
132        done = true;
133        currentReturnCode = ReturnCode.NEXT_ROW;
134        return false;
135      }
136      if (index != ROW_BEFORE_FIRST_RANGE) {
137        range = ranges.get(index);
138      } else {
139        range = ranges.get(0);
140      }
141      if (ranges.isExclusive()) {
142        ranges.resetExclusive();
143        currentReturnCode = ReturnCode.NEXT_ROW;
144        return false;
145      }
146      if (!ranges.hasFoundFirstRange()) {
147        if (index != ROW_BEFORE_FIRST_RANGE) {
148          currentReturnCode = ReturnCode.INCLUDE;
149        } else {
150          currentReturnCode = ReturnCode.SEEK_NEXT_USING_HINT;
151        }
152        ranges.setFoundFirstRange();
153      } else {
154        if (range.contains(rowArr, offset, length)) {
155          currentReturnCode = ReturnCode.INCLUDE;
156        } else {
157          currentReturnCode = ReturnCode.SEEK_NEXT_USING_HINT;
158        }
159      }
160    } else {
161      currentReturnCode = ReturnCode.INCLUDE;
162    }
163    return false;
164  }
165
166  @Override
167  public ReturnCode filterCell(final Cell ignored) {
168    return currentReturnCode;
169  }
170
171  @Override
172  public Cell getNextCellHint(Cell currentKV) {
173    // skip to the next range's start row
174    // #getComparisonData lets us avoid the `if (reversed)` branch
175    byte[] comparisonData = range.getComparisonData();
176    return PrivateCellUtil.createFirstOnRow(comparisonData, 0, (short) comparisonData.length);
177  }
178
179  /**
180   * @return The filter serialized using pb
181   */
182  @Override
183  public byte[] toByteArray() {
184    FilterProtos.MultiRowRangeFilter.Builder builder =
185      FilterProtos.MultiRowRangeFilter.newBuilder();
186    for (RowRange range : rangeList) {
187      if (range != null) {
188        FilterProtos.RowRange.Builder rangebuilder = FilterProtos.RowRange.newBuilder();
189        if (range.startRow != null)
190          rangebuilder.setStartRow(UnsafeByteOperations.unsafeWrap(range.startRow));
191        rangebuilder.setStartRowInclusive(range.startRowInclusive);
192        if (range.stopRow != null)
193          rangebuilder.setStopRow(UnsafeByteOperations.unsafeWrap(range.stopRow));
194        rangebuilder.setStopRowInclusive(range.stopRowInclusive);
195        builder.addRowRangeList(rangebuilder.build());
196      }
197    }
198    return builder.build().toByteArray();
199  }
200
201  /**
202   * @param pbBytes A pb serialized instance
203   * @return An instance of MultiRowRangeFilter
204   * @throws org.apache.hadoop.hbase.exceptions.DeserializationException
205   */
206  public static MultiRowRangeFilter parseFrom(final byte[] pbBytes)
207    throws DeserializationException {
208    FilterProtos.MultiRowRangeFilter proto;
209    try {
210      proto = FilterProtos.MultiRowRangeFilter.parseFrom(pbBytes);
211    } catch (InvalidProtocolBufferException e) {
212      throw new DeserializationException(e);
213    }
214    int length = proto.getRowRangeListCount();
215    List<FilterProtos.RowRange> rangeProtos = proto.getRowRangeListList();
216    List<RowRange> rangeList = new ArrayList<>(length);
217    for (FilterProtos.RowRange rangeProto : rangeProtos) {
218      RowRange range =
219        new RowRange(rangeProto.hasStartRow() ? rangeProto.getStartRow().toByteArray() : null,
220          rangeProto.getStartRowInclusive(),
221          rangeProto.hasStopRow() ? rangeProto.getStopRow().toByteArray() : null,
222          rangeProto.getStopRowInclusive());
223      rangeList.add(range);
224    }
225    return new MultiRowRangeFilter(rangeList);
226  }
227
228  /**
229   * @param o the filter to compare
230   * @return true if and only if the fields of the filter that are serialized are equal to the
231   *         corresponding fields in other. Used for testing.
232   */
233  @Override
234  boolean areSerializedFieldsEqual(Filter o) {
235    if (o == this) return true;
236    if (!(o instanceof MultiRowRangeFilter)) return false;
237
238    MultiRowRangeFilter other = (MultiRowRangeFilter) o;
239    if (this.rangeList.size() != other.rangeList.size()) return false;
240    for (int i = 0; i < rangeList.size(); ++i) {
241      RowRange thisRange = this.rangeList.get(i);
242      RowRange otherRange = other.rangeList.get(i);
243      if (
244        !(Bytes.equals(thisRange.startRow, otherRange.startRow)
245          && Bytes.equals(thisRange.stopRow, otherRange.stopRow)
246          && (thisRange.startRowInclusive == otherRange.startRowInclusive)
247          && (thisRange.stopRowInclusive == otherRange.stopRowInclusive))
248      ) {
249        return false;
250      }
251    }
252    return true;
253  }
254
255  /**
256   * sort the ranges and if the ranges with overlap, then merge them.
257   * @param ranges the list of ranges to sort and merge.
258   * @return the ranges after sort and merge.
259   */
260  public static List<RowRange> sortAndMerge(List<RowRange> ranges) {
261    if (ranges.isEmpty()) {
262      throw new IllegalArgumentException("No ranges found.");
263    }
264    List<RowRange> invalidRanges = new ArrayList<>();
265    List<RowRange> newRanges = new ArrayList<>(ranges.size());
266    Collections.sort(ranges);
267    if (ranges.get(0).isValid()) {
268      if (ranges.size() == 1) {
269        newRanges.add(ranges.get(0));
270      }
271    } else {
272      invalidRanges.add(ranges.get(0));
273    }
274
275    byte[] lastStartRow = ranges.get(0).startRow;
276    boolean lastStartRowInclusive = ranges.get(0).startRowInclusive;
277    byte[] lastStopRow = ranges.get(0).stopRow;
278    boolean lastStopRowInclusive = ranges.get(0).stopRowInclusive;
279    int i = 1;
280    for (; i < ranges.size(); i++) {
281      RowRange range = ranges.get(i);
282      if (!range.isValid()) {
283        invalidRanges.add(range);
284      }
285      if (Bytes.equals(lastStopRow, HConstants.EMPTY_BYTE_ARRAY)) {
286        newRanges.add(
287          new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow, lastStopRowInclusive));
288        break;
289      }
290      // with overlap in the ranges
291      if (
292        (Bytes.compareTo(lastStopRow, range.startRow) > 0)
293          || (Bytes.compareTo(lastStopRow, range.startRow) == 0
294            && !(lastStopRowInclusive == false && range.isStartRowInclusive() == false))
295      ) {
296        if (Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
297          newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, range.stopRow,
298            range.stopRowInclusive));
299          break;
300        }
301        // if first range contains second range, ignore the second range
302        if (Bytes.compareTo(lastStopRow, range.stopRow) >= 0) {
303          if ((Bytes.compareTo(lastStopRow, range.stopRow) == 0)) {
304            if (lastStopRowInclusive == true || range.stopRowInclusive == true) {
305              lastStopRowInclusive = true;
306            }
307          }
308          if ((i + 1) == ranges.size()) {
309            newRanges.add(
310              new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow, lastStopRowInclusive));
311          }
312        } else {
313          lastStopRow = range.stopRow;
314          lastStopRowInclusive = range.stopRowInclusive;
315          if ((i + 1) < ranges.size()) {
316            i++;
317            range = ranges.get(i);
318            if (!range.isValid()) {
319              invalidRanges.add(range);
320            }
321          } else {
322            newRanges.add(
323              new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow, lastStopRowInclusive));
324            break;
325          }
326          while (
327            (Bytes.compareTo(lastStopRow, range.startRow) > 0)
328              || (Bytes.compareTo(lastStopRow, range.startRow) == 0
329                && (lastStopRowInclusive == true || range.startRowInclusive == true))
330          ) {
331            if (Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
332              break;
333            }
334            // if this first range contain second range, ignore the second range
335            if (Bytes.compareTo(lastStopRow, range.stopRow) >= 0) {
336              if (lastStopRowInclusive == true || range.stopRowInclusive == true) {
337                lastStopRowInclusive = true;
338              }
339              i++;
340              if (i < ranges.size()) {
341                range = ranges.get(i);
342                if (!range.isValid()) {
343                  invalidRanges.add(range);
344                }
345              } else {
346                break;
347              }
348            } else {
349              lastStopRow = range.stopRow;
350              lastStopRowInclusive = range.stopRowInclusive;
351              i++;
352              if (i < ranges.size()) {
353                range = ranges.get(i);
354                if (!range.isValid()) {
355                  invalidRanges.add(range);
356                }
357              } else {
358                break;
359              }
360            }
361          }
362          if (Bytes.equals(range.stopRow, HConstants.EMPTY_BYTE_ARRAY)) {
363            if (
364              (Bytes.compareTo(lastStopRow, range.startRow) < 0)
365                || (Bytes.compareTo(lastStopRow, range.startRow) == 0
366                  && lastStopRowInclusive == false && range.startRowInclusive == false)
367            ) {
368              newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow,
369                lastStopRowInclusive));
370              newRanges.add(range);
371            } else {
372              newRanges.add(new RowRange(lastStartRow, lastStartRowInclusive, range.stopRow,
373                range.stopRowInclusive));
374              break;
375            }
376          }
377          newRanges.add(
378            new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow, lastStopRowInclusive));
379          if ((i + 1) == ranges.size()) {
380            newRanges.add(range);
381          }
382          lastStartRow = range.startRow;
383          lastStartRowInclusive = range.startRowInclusive;
384          lastStopRow = range.stopRow;
385          lastStopRowInclusive = range.stopRowInclusive;
386        }
387      } else {
388        newRanges.add(
389          new RowRange(lastStartRow, lastStartRowInclusive, lastStopRow, lastStopRowInclusive));
390        if ((i + 1) == ranges.size()) {
391          newRanges.add(range);
392        }
393        lastStartRow = range.startRow;
394        lastStartRowInclusive = range.startRowInclusive;
395        lastStopRow = range.stopRow;
396        lastStopRowInclusive = range.stopRowInclusive;
397      }
398    }
399    // check the remaining ranges
400    for (int j = i; j < ranges.size(); j++) {
401      if (!ranges.get(j).isValid()) {
402        invalidRanges.add(ranges.get(j));
403      }
404    }
405    // if invalid range exists, throw the exception
406    if (invalidRanges.size() != 0) {
407      throwExceptionForInvalidRanges(invalidRanges, true);
408    }
409    // If no valid ranges found, throw the exception
410    if (newRanges.isEmpty()) {
411      throw new IllegalArgumentException("No valid ranges found.");
412    }
413    return newRanges;
414  }
415
416  private static void throwExceptionForInvalidRanges(List<RowRange> invalidRanges,
417    boolean details) {
418    StringBuilder sb = new StringBuilder();
419    sb.append(invalidRanges.size()).append(" invaild ranges.\n");
420    if (details) {
421      for (RowRange range : invalidRanges) {
422        sb.append("Invalid range: start row => " + Bytes.toString(range.startRow) + ", stop row => "
423          + Bytes.toString(range.stopRow)).append('\n');
424      }
425    }
426    throw new IllegalArgumentException(sb.toString());
427  }
428
429  private static abstract class BasicRowRange implements Comparable<BasicRowRange> {
430    protected byte[] startRow;
431    protected boolean startRowInclusive = true;
432    protected byte[] stopRow;
433    protected boolean stopRowInclusive = false;
434
435    public BasicRowRange() {
436    }
437
438    /**
439     * If the startRow is empty or null, set it to HConstants.EMPTY_BYTE_ARRAY, means begin at the
440     * start row of the table. If the stopRow is empty or null, set it to
441     * HConstants.EMPTY_BYTE_ARRAY, means end of the last row of table.
442     */
443    public BasicRowRange(String startRow, boolean startRowInclusive, String stopRow,
444      boolean stopRowInclusive) {
445      this(
446        (startRow == null || startRow.isEmpty())
447          ? HConstants.EMPTY_BYTE_ARRAY
448          : Bytes.toBytes(startRow),
449        startRowInclusive,
450        (stopRow == null || stopRow.isEmpty())
451          ? HConstants.EMPTY_BYTE_ARRAY
452          : Bytes.toBytes(stopRow),
453        stopRowInclusive);
454    }
455
456    public BasicRowRange(byte[] startRow, boolean startRowInclusive, byte[] stopRow,
457      boolean stopRowInclusive) {
458      this.startRow = (startRow == null) ? HConstants.EMPTY_BYTE_ARRAY : startRow;
459      this.startRowInclusive = startRowInclusive;
460      this.stopRow = (stopRow == null) ? HConstants.EMPTY_BYTE_ARRAY : stopRow;
461      this.stopRowInclusive = stopRowInclusive;
462    }
463
464    public byte[] getStartRow() {
465      return startRow;
466    }
467
468    public byte[] getStopRow() {
469      return stopRow;
470    }
471
472    /**
473     * @return if start row is inclusive.
474     */
475    public boolean isStartRowInclusive() {
476      return startRowInclusive;
477    }
478
479    /**
480     * @return if stop row is inclusive.
481     */
482    public boolean isStopRowInclusive() {
483      return stopRowInclusive;
484    }
485
486    public boolean contains(byte[] row) {
487      return contains(row, 0, row.length);
488    }
489
490    public boolean contains(byte[] buffer, int offset, int length) {
491      if (startRowInclusive) {
492        if (stopRowInclusive) {
493          return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) >= 0
494            && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
495              || Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) <= 0);
496        } else {
497          return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) >= 0
498            && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
499              || Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) < 0);
500        }
501      } else {
502        if (stopRowInclusive) {
503          return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) > 0
504            && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
505              || Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) <= 0);
506        } else {
507          return Bytes.compareTo(buffer, offset, length, startRow, 0, startRow.length) > 0
508            && (Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
509              || Bytes.compareTo(buffer, offset, length, stopRow, 0, stopRow.length) < 0);
510        }
511      }
512    }
513
514    public boolean isValid() {
515      return Bytes.equals(startRow, HConstants.EMPTY_BYTE_ARRAY)
516        || Bytes.equals(stopRow, HConstants.EMPTY_BYTE_ARRAY)
517        || Bytes.compareTo(startRow, stopRow) < 0
518        || (Bytes.compareTo(startRow, stopRow) == 0 && stopRowInclusive == true);
519    }
520
521    @Override
522    public boolean equals(Object obj) {
523      if (!(obj instanceof BasicRowRange)) {
524        return false;
525      }
526      if (this == obj) {
527        return true;
528      }
529      BasicRowRange rr = (BasicRowRange) obj;
530      return Bytes.equals(this.stopRow, rr.getStopRow())
531        && Bytes.equals(this.startRow, this.getStartRow())
532        && this.startRowInclusive == rr.isStartRowInclusive()
533        && this.stopRowInclusive == rr.isStopRowInclusive();
534    }
535
536    @Override
537    public int hashCode() {
538      return Objects.hash(Bytes.hashCode(this.stopRow), Bytes.hashCode(this.startRow),
539        this.startRowInclusive, this.stopRowInclusive);
540    }
541
542    /**
543     * Returns the data to be used to compare {@code this} to another object.
544     */
545    public abstract byte[] getComparisonData();
546
547    /**
548     * Returns whether the bounding row used for binary-search is inclusive or not. For forward
549     * scans, we would check the starRow, but we would check the stopRow for the reverse scan case.
550     */
551    public abstract boolean isSearchRowInclusive();
552
553    @Override
554    public int compareTo(BasicRowRange other) {
555      byte[] left;
556      byte[] right;
557      if (isAscendingOrder()) {
558        left = this.getComparisonData();
559        right = other.getComparisonData();
560      } else {
561        left = other.getComparisonData();
562        right = this.getComparisonData();
563      }
564      return Bytes.compareTo(left, right);
565    }
566
567    public abstract boolean isAscendingOrder();
568  }
569
570  /**
571   * Internal RowRange that reverses the sort-order to handle reverse scans.
572   */
573  @InterfaceAudience.Private
574  private static class ReversedRowRange extends BasicRowRange {
575    public ReversedRowRange(byte[] startRow, boolean startRowInclusive, byte[] stopRow,
576      boolean stopRowInclusive) {
577      super(startRow, startRowInclusive, stopRow, stopRowInclusive);
578    }
579
580    @Override
581    public byte[] getComparisonData() {
582      return this.stopRow;
583    }
584
585    @Override
586    public boolean isSearchRowInclusive() {
587      return this.stopRowInclusive;
588    }
589
590    @Override
591    public boolean isAscendingOrder() {
592      return false;
593    }
594  }
595
596  @InterfaceAudience.Public
597  public static class RowRange extends BasicRowRange {
598    public RowRange() {
599    }
600
601    /**
602     * If the startRow is empty or null, set it to HConstants.EMPTY_BYTE_ARRAY, means begin at the
603     * start row of the table. If the stopRow is empty or null, set it to
604     * HConstants.EMPTY_BYTE_ARRAY, means end of the last row of table.
605     */
606    public RowRange(String startRow, boolean startRowInclusive, String stopRow,
607      boolean stopRowInclusive) {
608      super(startRow, startRowInclusive, stopRow, stopRowInclusive);
609    }
610
611    public RowRange(byte[] startRow, boolean startRowInclusive, byte[] stopRow,
612      boolean stopRowInclusive) {
613      super(startRow, startRowInclusive, stopRow, stopRowInclusive);
614    }
615
616    @Override
617    public byte[] getComparisonData() {
618      return startRow;
619    }
620
621    @Override
622    public boolean isSearchRowInclusive() {
623      return startRowInclusive;
624    }
625
626    @Override
627    public boolean isAscendingOrder() {
628      return true;
629    }
630  }
631
632  /**
633   * Abstraction over the ranges of rows to return from this filter, regardless of forward or
634   * reverse scans being used. This Filter can use this class, agnostic of iteration direction, as
635   * the same algorithm can be applied in both cases.
636   */
637  @InterfaceAudience.Private
638  private static class RangeIteration {
639    private boolean exclusive = false;
640    private boolean initialized = false;
641    private boolean foundFirstRange = false;
642    private boolean reversed = false;
643    private final List<RowRange> sortedAndMergedRanges;
644    private List<? extends BasicRowRange> ranges;
645
646    public RangeIteration(List<RowRange> sortedAndMergedRanges) {
647      this.sortedAndMergedRanges = sortedAndMergedRanges;
648    }
649
650    void initialize(boolean reversed) {
651      // Avoid double initialization
652      assert !this.initialized;
653      this.reversed = reversed;
654      if (reversed) {
655        // If we are doing a reverse scan, we can reverse the ranges (both the elements in
656        // the list as well as their start/stop key), and use the same algorithm.
657        this.ranges = flipAndReverseRanges(sortedAndMergedRanges);
658      } else {
659        this.ranges = sortedAndMergedRanges;
660      }
661      this.initialized = true;
662    }
663
664    /**
665     * Rebuilds the sorted ranges (by startKey) into an equivalent sorted list of ranges, only by
666     * stopKey instead. Descending order and the ReversedRowRange compareTo implementation make sure
667     * that we can use Collections.binarySearch().
668     */
669    static List<ReversedRowRange> flipAndReverseRanges(List<RowRange> ranges) {
670      List<ReversedRowRange> flippedRanges = new ArrayList<>(ranges.size());
671      for (int i = ranges.size() - 1; i >= 0; i--) {
672        RowRange origRange = ranges.get(i);
673        ReversedRowRange newRowRange = new ReversedRowRange(origRange.startRow,
674          origRange.startRowInclusive, origRange.stopRow, origRange.isStopRowInclusive());
675        flippedRanges.add(newRowRange);
676      }
677      return flippedRanges;
678    }
679
680    /**
681     * Calculates the position where the given rowkey fits in the ranges list.
682     * @param rowKey the row key to calculate
683     * @return index the position of the row key
684     */
685    public int getNextRangeIndex(byte[] rowKey) {
686      BasicRowRange temp;
687      if (reversed) {
688        temp = new ReversedRowRange(null, true, rowKey, true);
689      } else {
690        temp = new RowRange(rowKey, true, null, true);
691      }
692      // Because we make sure that `ranges` has the correct natural ordering (given it containing
693      // RowRange or ReverseRowRange objects). This keeps us from having to have two different
694      // implementations below.
695      final int index = Collections.binarySearch(ranges, temp);
696      if (index < 0) {
697        int insertionPosition = -index - 1;
698        // check if the row key in the range before the insertion position
699        if (insertionPosition != 0 && ranges.get(insertionPosition - 1).contains(rowKey)) {
700          return insertionPosition - 1;
701        }
702        // check if the row key is before the first range
703        if (insertionPosition == 0 && !ranges.get(insertionPosition).contains(rowKey)) {
704          return ROW_BEFORE_FIRST_RANGE;
705        }
706        if (!foundFirstRange) {
707          foundFirstRange = true;
708        }
709        return insertionPosition;
710      }
711      // the row key equals one of the start keys, and the the range exclude the start key
712      if (ranges.get(index).isSearchRowInclusive() == false) {
713        exclusive = true;
714      }
715      return index;
716    }
717
718    /**
719     * Sets {@link #foundFirstRange} to {@code true}, indicating that we found a matching row range.
720     */
721    public void setFoundFirstRange() {
722      this.foundFirstRange = true;
723    }
724
725    /**
726     * Gets the RowRange at the given offset.
727     */
728    @SuppressWarnings("unchecked")
729    public <T extends BasicRowRange> T get(int i) {
730      return (T) ranges.get(i);
731    }
732
733    /**
734     * Returns true if the first matching row range was found.
735     */
736    public boolean hasFoundFirstRange() {
737      return foundFirstRange;
738    }
739
740    /**
741     * Returns true if the current range's key is exclusive
742     */
743    public boolean isExclusive() {
744      return exclusive;
745    }
746
747    /**
748     * Resets the exclusive flag.
749     */
750    public void resetExclusive() {
751      exclusive = false;
752    }
753
754    /**
755     * Returns true if this class has been initialized by calling {@link #initialize(boolean)}.
756     */
757    public boolean isInitialized() {
758      return initialized;
759    }
760
761    /**
762     * Returns true if we exhausted searching all row ranges.
763     */
764    public boolean isIterationComplete(int index) {
765      return index >= ranges.size();
766    }
767  }
768
769  @Override
770  public boolean equals(Object obj) {
771    return obj instanceof Filter && areSerializedFieldsEqual((Filter) obj);
772  }
773
774  @Override
775  public int hashCode() {
776    return Objects.hash(this.ranges);
777  }
778}