001/**
002 * Copyright The Apache Software Foundation
003 *
004 * Licensed to the Apache Software Foundation (ASF) under one or more
005 * contributor license agreements. See the NOTICE file distributed with this
006 * work for additional information regarding copyright ownership. The ASF
007 * licenses this file to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance with the License.
009 * You may obtain a copy of the License at
010 *
011 * http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
015 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
016 * License for the specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.hadoop.hbase.regionserver;
020
021import java.io.IOException;
022import java.util.List;
023
024import org.apache.commons.lang3.NotImplementedException;
025import org.apache.hadoop.hbase.HConstants;
026import org.apache.yetus.audience.InterfaceAudience;
027import org.apache.hadoop.hbase.Cell;
028import org.apache.hadoop.hbase.CellComparator;
029import org.apache.hadoop.hbase.CellUtil;
030
031/**
032 * ReversedKeyValueHeap is used for supporting reversed scanning. Compared with
033 * KeyValueHeap, its scanner comparator is a little different (see
034 * ReversedKVScannerComparator), all seek is backward seek(see
035 * {@link KeyValueScanner#backwardSeek}), and it will jump to the previous row
036 * if it is already at the end of one row when calling next().
037 */
038@InterfaceAudience.Private
039public class ReversedKeyValueHeap extends KeyValueHeap {
040
041  /**
042   * @param scanners
043   * @param comparator
044   * @throws IOException
045   */
046  public ReversedKeyValueHeap(List<? extends KeyValueScanner> scanners,
047      CellComparator comparator) throws IOException {
048    super(scanners, new ReversedKVScannerComparator(comparator));
049  }
050
051  @Override
052  public boolean seek(Cell seekKey) throws IOException {
053    throw new IllegalStateException(
054        "seek cannot be called on ReversedKeyValueHeap");
055  }
056
057  @Override
058  public boolean reseek(Cell seekKey) throws IOException {
059    throw new IllegalStateException(
060        "reseek cannot be called on ReversedKeyValueHeap");
061  }
062
063  @Override
064  public boolean requestSeek(Cell key, boolean forward, boolean useBloom)
065      throws IOException {
066    throw new IllegalStateException(
067        "requestSeek cannot be called on ReversedKeyValueHeap");
068  }
069
070  @Override
071  public boolean seekToPreviousRow(Cell seekKey) throws IOException {
072    if (current == null) {
073      return false;
074    }
075    heap.add(current);
076    current = null;
077
078    KeyValueScanner scanner;
079    while ((scanner = heap.poll()) != null) {
080      Cell topKey = scanner.peek();
081      if (comparator.getComparator().compareRows(topKey, seekKey) < 0) {
082        // Row of Top KeyValue is before Seek row.
083        heap.add(scanner);
084        current = pollRealKV();
085        return current != null;
086      }
087
088      if (!scanner.seekToPreviousRow(seekKey)) {
089        this.scannersForDelayedClose.add(scanner);
090      } else {
091        heap.add(scanner);
092      }
093    }
094
095    // Heap is returning empty, scanner is done
096    return false;
097  }
098
099  @Override
100  public boolean backwardSeek(Cell seekKey) throws IOException {
101    if (current == null) {
102      return false;
103    }
104    heap.add(current);
105    current = null;
106
107    KeyValueScanner scanner;
108    while ((scanner = heap.poll()) != null) {
109      Cell topKey = scanner.peek();
110      if ((CellUtil.matchingRows(seekKey, topKey) && comparator
111          .getComparator().compare(seekKey, topKey) <= 0)
112          || comparator.getComparator().compareRows(seekKey, topKey) > 0) {
113        heap.add(scanner);
114        current = pollRealKV();
115        return current != null;
116      }
117      if (!scanner.backwardSeek(seekKey)) {
118        this.scannersForDelayedClose.add(scanner);
119      } else {
120        heap.add(scanner);
121      }
122    }
123    return false;
124  }
125
126  @Override
127  public Cell next() throws IOException {
128    if (this.current == null) {
129      return null;
130    }
131    Cell kvReturn = this.current.next();
132    Cell kvNext = this.current.peek();
133    if (kvNext == null
134        || this.comparator.kvComparator.compareRows(kvNext, kvReturn) > 0) {
135      if (this.current.seekToPreviousRow(kvReturn)) {
136        this.heap.add(this.current);
137      } else {
138        this.scannersForDelayedClose.add(this.current);
139      }
140      this.current = null;
141      this.current = pollRealKV();
142    } else {
143      KeyValueScanner topScanner = this.heap.peek();
144      if (topScanner != null
145          && this.comparator.compare(this.current, topScanner) > 0) {
146        this.heap.add(this.current);
147        this.current = null;
148        this.current = pollRealKV();
149      }
150    }
151    return kvReturn;
152  }
153
154  /**
155   * In ReversedKVScannerComparator, we compare the row of scanners' peek values
156   * first, sort bigger one before the smaller one. Then compare the KeyValue if
157   * they have the equal row, sort smaller one before the bigger one
158   */
159  private static class ReversedKVScannerComparator extends
160      KVScannerComparator {
161
162    /**
163     * Constructor
164     * @param kvComparator
165     */
166    public ReversedKVScannerComparator(CellComparator kvComparator) {
167      super(kvComparator);
168    }
169
170    @Override
171    public int compare(KeyValueScanner left, KeyValueScanner right) {
172      int rowComparison = compareRows(left.peek(), right.peek());
173      if (rowComparison != 0) {
174        return -rowComparison;
175      }
176      return super.compare(left, right);
177    }
178
179    /**
180     * Compares rows of two KeyValue
181     * @param left
182     * @param right
183     * @return less than 0 if left is smaller, 0 if equal etc..
184     */
185    public int compareRows(Cell left, Cell right) {
186      return super.kvComparator.compareRows(left, right);
187    }
188  }
189
190  @Override
191  public boolean seekToLastRow() throws IOException {
192    throw new NotImplementedException(HConstants.NOT_IMPLEMENTED);
193  }
194}