Custom ColumnRangeFilter implementation for HBase

ColumnRangeFilter.java

/**
 * Copyright (C) 2014 VCNC Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package kr.co.vcnc.hadoop.hbase.filter;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
 
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.util.Bytes;
 
public class ColumnRangeFilter extends FilterBase {
    private byte[] startQualifier;
    private byte[] endQualifier;
 
    private boolean startInclusive;
    private boolean endInclusive;
 
    public ColumnRangeFilter() {
        super();
    }
 
    public ColumnRangeFilter(byte[] startColumn, byte[] endColumn) {
        this(startColumn, true, endColumn, false);
    }
 
    public ColumnRangeFilter(byte[] startColumn, boolean startInclusive,
            byte[] endColumn, boolean endInclusive) {
        super();
        this.startQualifier = startColumn;
        this.endQualifier = endColumn;
        this.startInclusive = startInclusive;
        this.endInclusive = endInclusive;
    }
 
    @Override
    public ReturnCode filterKeyValue(KeyValue kv) {
        if (kv.getBuffer() == null) {
            return ReturnCode.INCLUDE;
        } else {
            return filterColumn(kv.getBuffer(), kv.getQualifierOffset(), kv.getQualifierLength());
        }
    }
 
    private ReturnCode filterColumn(byte[] buffer, int qualifierOffset, int qualifierLength) {
        int startCmp = 1;
        if (startQualifier != null) {
            startCmp = Bytes.compareTo(buffer, qualifierOffset,
                    qualifierLength, startQualifier, 0, startQualifier.length);
        }
        int endCmp = -1;
        if (endQualifier != null) {
            endCmp = Bytes.compareTo(buffer, qualifierOffset, qualifierLength,
                    endQualifier, 0, endQualifier.length);
        }
 
        if (startCmp < 0) {
            return ReturnCode.SEEK_NEXT_USING_HINT;
        } else if (((startInclusive && startCmp >= 0) || (!startInclusive && startCmp > 0))
                && ((endInclusive && endCmp <= 0) || (!endInclusive && endCmp < 0))) {
            return ReturnCode.INCLUDE;
        } else {
            if (startCmp == 0) {
                return ReturnCode.NEXT_COL;
            } else {
                return ReturnCode.NEXT_ROW;
            }
        }
    }
 
    @Override
    public void readFields(DataInput in) throws IOException {
        this.startQualifier = Bytes.readByteArray(in);
        this.endQualifier = Bytes.readByteArray(in);
        this.startInclusive = in.readBoolean();
        this.endInclusive = in.readBoolean();
    }
 
    @Override
    public void write(DataOutput out) throws IOException {
        Bytes.writeByteArray(out, startQualifier);
        Bytes.writeByteArray(out, endQualifier);
        out.writeBoolean(startInclusive);
        out.writeBoolean(endInclusive);
    }
 
    @Override
    public KeyValue getNextKeyHint(KeyValue kv) {
        return KeyValue.createFirstOnRow(kv.getBuffer(), kv.getRowOffset(),
                kv.getRowLength(), kv.getBuffer(), kv.getFamilyOffset(),
                kv.getFamilyLength(), startQualifier, 0,
                startQualifier != null ? startQualifier.length : 0);
    }
}