You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ec...@apache.org on 2013/02/25 23:50:29 UTC
svn commit: r1449950 [19/35] - in /hbase/trunk: ./ hbase-client/
hbase-client/src/ hbase-client/src/main/ hbase-client/src/main/java/
hbase-client/src/main/java/org/ hbase-client/src/main/java/org/apache/
hbase-client/src/main/java/org/apache/hadoop/ h...
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueExcludeFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueExcludeFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueExcludeFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueExcludeFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,181 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A {@link Filter} that checks a single column value, but does not emit the
+ * tested column. This will enable a performance boost over
+ * {@link SingleColumnValueFilter}, if the tested column value is not actually
+ * needed as input (besides for the filtering itself).
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class SingleColumnValueExcludeFilter extends SingleColumnValueFilter {
+
+ /**
+ * Constructor for binary compare of the value of a single column. If the
+ * column is found and the condition passes, all columns of the row will be
+ * emitted; except for the tested column value. If the column is not found or
+ * the condition fails, the row will not be emitted.
+ *
+ * @param family name of column family
+ * @param qualifier name of column qualifier
+ * @param compareOp operator
+ * @param value value to compare column values against
+ */
+ public SingleColumnValueExcludeFilter(byte[] family, byte[] qualifier,
+ CompareOp compareOp, byte[] value) {
+ super(family, qualifier, compareOp, value);
+ }
+
+ /**
+ * Constructor for binary compare of the value of a single column. If the
+ * column is found and the condition passes, all columns of the row will be
+ * emitted; except for the tested column value. If the condition fails, the
+ * row will not be emitted.
+ * <p>
+ * Use the filterIfColumnMissing flag to set whether the rest of the columns
+ * in a row will be emitted if the specified column to check is not found in
+ * the row.
+ *
+ * @param family name of column family
+ * @param qualifier name of column qualifier
+ * @param compareOp operator
+ * @param comparator Comparator to use.
+ */
+ public SingleColumnValueExcludeFilter(byte[] family, byte[] qualifier,
+ CompareOp compareOp, ByteArrayComparable comparator) {
+ super(family, qualifier, compareOp, comparator);
+ }
+
+ /**
+ * Constructor for protobuf deserialization only.
+ * @param family
+ * @param qualifier
+ * @param compareOp
+ * @param comparator
+ * @param filterIfMissing
+ * @param latestVersionOnly
+ */
+ protected SingleColumnValueExcludeFilter(final byte[] family, final byte[] qualifier,
+ final CompareOp compareOp, ByteArrayComparable comparator, final boolean filterIfMissing,
+ final boolean latestVersionOnly) {
+ super(family, qualifier, compareOp, comparator, filterIfMissing, latestVersionOnly);
+ }
+
+ // We cleaned result row in FilterRow to be consistent with scanning process.
+ public boolean hasFilterRow() {
+ return true;
+ }
+
+ // Here we remove from row all key values from testing column
+ public void filterRow(List<KeyValue> kvs) {
+ Iterator it = kvs.iterator();
+ while (it.hasNext()) {
+ KeyValue kv = (KeyValue)it.next();
+ // If the current column is actually the tested column,
+ // we will skip it instead.
+ if (kv.matchingColumn(this.columnFamily, this.columnQualifier)) {
+ it.remove();
+ }
+ }
+ }
+
+ public static Filter createFilterFromArguments(ArrayList<byte []> filterArguments) {
+ SingleColumnValueFilter tempFilter = (SingleColumnValueFilter)
+ SingleColumnValueFilter.createFilterFromArguments(filterArguments);
+ SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter (
+ tempFilter.getFamily(), tempFilter.getQualifier(),
+ tempFilter.getOperator(), tempFilter.getComparator());
+
+ if (filterArguments.size() == 6) {
+ filter.setFilterIfMissing(tempFilter.getFilterIfMissing());
+ filter.setLatestVersionOnly(tempFilter.getLatestVersionOnly());
+ }
+ return filter;
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ FilterProtos.SingleColumnValueExcludeFilter.Builder builder =
+ FilterProtos.SingleColumnValueExcludeFilter.newBuilder();
+ builder.setSingleColumnValueFilter(super.convert());
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link SingleColumnValueExcludeFilter} instance
+ * @return An instance of {@link SingleColumnValueExcludeFilter} made from <code>bytes</code>
+ * @throws DeserializationException
+ * @see #toByteArray
+ */
+ public static SingleColumnValueExcludeFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.SingleColumnValueExcludeFilter proto;
+ try {
+ proto = FilterProtos.SingleColumnValueExcludeFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+
+ FilterProtos.SingleColumnValueFilter parentProto = proto.getSingleColumnValueFilter();
+ final CompareOp compareOp =
+ CompareOp.valueOf(parentProto.getCompareOp().name());
+ final ByteArrayComparable comparator;
+ try {
+ comparator = ProtobufUtil.toComparator(parentProto.getComparator());
+ } catch (IOException ioe) {
+ throw new DeserializationException(ioe);
+ }
+
+ return new SingleColumnValueExcludeFilter(parentProto.hasColumnFamily() ? parentProto
+ .getColumnFamily().toByteArray() : null, parentProto.hasColumnQualifier() ? parentProto
+ .getColumnQualifier().toByteArray() : null, compareOp, comparator, parentProto
+ .getFilterIfMissing(), parentProto.getLatestVersionOnly());
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof SingleColumnValueExcludeFilter)) return false;
+
+ return super.areSerializedFieldsEqual(o);
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SingleColumnValueFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,388 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.CompareType;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+/**
+ * This filter is used to filter cells based on value. It takes a {@link CompareFilter.CompareOp}
+ * operator (equal, greater, not equal, etc), and either a byte [] value or
+ * a ByteArrayComparable.
+ * <p>
+ * If we have a byte [] value then we just do a lexicographic compare. For
+ * example, if passed value is 'b' and cell has 'a' and the compare operator
+ * is LESS, then we will filter out this cell (return true). If this is not
+ * sufficient (eg you want to deserialize a long and then compare it to a fixed
+ * long value), then you can pass in your own comparator instead.
+ * <p>
+ * You must also specify a family and qualifier. Only the value of this column
+ * will be tested. When using this filter on a {@link Scan} with specified
+ * inputs, the column to be tested should also be added as input (otherwise
+ * the filter will regard the column as missing).
+ * <p>
+ * To prevent the entire row from being emitted if the column is not found
+ * on a row, use {@link #setFilterIfMissing}.
+ * Otherwise, if the column is found, the entire row will be emitted only if
+ * the value passes. If the value fails, the row will be filtered out.
+ * <p>
+ * In order to test values of previous versions (timestamps), set
+ * {@link #setLatestVersionOnly} to false. The default is true, meaning that
+ * only the latest version's value is tested and all previous versions are ignored.
+ * <p>
+ * To filter based on the value of all scanned columns, use {@link ValueFilter}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class SingleColumnValueFilter extends FilterBase {
+ static final Log LOG = LogFactory.getLog(SingleColumnValueFilter.class);
+
+ protected byte [] columnFamily;
+ protected byte [] columnQualifier;
+ protected CompareOp compareOp;
+ protected ByteArrayComparable comparator;
+ protected boolean foundColumn = false;
+ protected boolean matchedColumn = false;
+ protected boolean filterIfMissing = false;
+ protected boolean latestVersionOnly = true;
+
+ /**
+ * Constructor for binary compare of the value of a single column. If the
+ * column is found and the condition passes, all columns of the row will be
+ * emitted. If the condition fails, the row will not be emitted.
+ * <p>
+ * Use the filterIfColumnMissing flag to set whether the rest of the columns
+ * in a row will be emitted if the specified column to check is not found in
+ * the row.
+ *
+ * @param family name of column family
+ * @param qualifier name of column qualifier
+ * @param compareOp operator
+ * @param value value to compare column values against
+ */
+ public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
+ final CompareOp compareOp, final byte[] value) {
+ this(family, qualifier, compareOp, new BinaryComparator(value));
+ }
+
+ /**
+ * Constructor for binary compare of the value of a single column. If the
+ * column is found and the condition passes, all columns of the row will be
+ * emitted. If the condition fails, the row will not be emitted.
+ * <p>
+ * Use the filterIfColumnMissing flag to set whether the rest of the columns
+ * in a row will be emitted if the specified column to check is not found in
+ * the row.
+ *
+ * @param family name of column family
+ * @param qualifier name of column qualifier
+ * @param compareOp operator
+ * @param comparator Comparator to use.
+ */
+ public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
+ final CompareOp compareOp, final ByteArrayComparable comparator) {
+ this.columnFamily = family;
+ this.columnQualifier = qualifier;
+ this.compareOp = compareOp;
+ this.comparator = comparator;
+ }
+
+ /**
+ * Constructor for protobuf deserialization only.
+ * @param family
+ * @param qualifier
+ * @param compareOp
+ * @param comparator
+ * @param filterIfMissing
+ * @param latestVersionOnly
+ */
+ protected SingleColumnValueFilter(final byte[] family, final byte[] qualifier,
+ final CompareOp compareOp, ByteArrayComparable comparator, final boolean filterIfMissing,
+ final boolean latestVersionOnly) {
+ this(family, qualifier, compareOp, comparator);
+ this.filterIfMissing = filterIfMissing;
+ this.latestVersionOnly = latestVersionOnly;
+ }
+
+ /**
+ * @return operator
+ */
+ public CompareOp getOperator() {
+ return compareOp;
+ }
+
+ /**
+ * @return the comparator
+ */
+ public ByteArrayComparable getComparator() {
+ return comparator;
+ }
+
+ /**
+ * @return the family
+ */
+ public byte[] getFamily() {
+ return columnFamily;
+ }
+
+ /**
+ * @return the qualifier
+ */
+ public byte[] getQualifier() {
+ return columnQualifier;
+ }
+
+ public ReturnCode filterKeyValue(KeyValue keyValue) {
+ // System.out.println("REMOVE KEY=" + keyValue.toString() + ", value=" + Bytes.toString(keyValue.getValue()));
+ if (this.matchedColumn) {
+ // We already found and matched the single column, all keys now pass
+ return ReturnCode.INCLUDE;
+ } else if (this.latestVersionOnly && this.foundColumn) {
+ // We found but did not match the single column, skip to next row
+ return ReturnCode.NEXT_ROW;
+ }
+ if (!keyValue.matchingColumn(this.columnFamily, this.columnQualifier)) {
+ return ReturnCode.INCLUDE;
+ }
+ foundColumn = true;
+ if (filterColumnValue(keyValue.getBuffer(),
+ keyValue.getValueOffset(), keyValue.getValueLength())) {
+ return this.latestVersionOnly? ReturnCode.NEXT_ROW: ReturnCode.INCLUDE;
+ }
+ this.matchedColumn = true;
+ return ReturnCode.INCLUDE;
+ }
+
+ private boolean filterColumnValue(final byte [] data, final int offset,
+ final int length) {
+ int compareResult = this.comparator.compareTo(data, offset, length);
+ switch (this.compareOp) {
+ case LESS:
+ return compareResult <= 0;
+ case LESS_OR_EQUAL:
+ return compareResult < 0;
+ case EQUAL:
+ return compareResult != 0;
+ case NOT_EQUAL:
+ return compareResult == 0;
+ case GREATER_OR_EQUAL:
+ return compareResult > 0;
+ case GREATER:
+ return compareResult >= 0;
+ default:
+ throw new RuntimeException("Unknown Compare op " + compareOp.name());
+ }
+ }
+
+ public boolean filterRow() {
+ // If column was found, return false if it was matched, true if it was not
+ // If column not found, return true if we filter if missing, false if not
+ return this.foundColumn? !this.matchedColumn: this.filterIfMissing;
+ }
+
+ public boolean hasFilterRow() {
+ return true;
+ }
+
+ public void reset() {
+ foundColumn = false;
+ matchedColumn = false;
+ }
+
+ /**
+ * Get whether entire row should be filtered if column is not found.
+ * @return true if row should be skipped if column not found, false if row
+ * should be let through anyways
+ */
+ public boolean getFilterIfMissing() {
+ return filterIfMissing;
+ }
+
+ /**
+ * Set whether entire row should be filtered if column is not found.
+ * <p>
+ * If true, the entire row will be skipped if the column is not found.
+ * <p>
+ * If false, the row will pass if the column is not found. This is default.
+ * @param filterIfMissing flag
+ */
+ public void setFilterIfMissing(boolean filterIfMissing) {
+ this.filterIfMissing = filterIfMissing;
+ }
+
+ /**
+ * Get whether only the latest version of the column value should be compared.
+ * If true, the row will be returned if only the latest version of the column
+ * value matches. If false, the row will be returned if any version of the
+ * column value matches. The default is true.
+ * @return return value
+ */
+ public boolean getLatestVersionOnly() {
+ return latestVersionOnly;
+ }
+
+ /**
+ * Set whether only the latest version of the column value should be compared.
+ * If true, the row will be returned if only the latest version of the column
+ * value matches. If false, the row will be returned if any version of the
+ * column value matches. The default is true.
+ * @param latestVersionOnly flag
+ */
+ public void setLatestVersionOnly(boolean latestVersionOnly) {
+ this.latestVersionOnly = latestVersionOnly;
+ }
+
+ public static Filter createFilterFromArguments(ArrayList<byte []> filterArguments) {
+ Preconditions.checkArgument(filterArguments.size() == 4 || filterArguments.size() == 6,
+ "Expected 4 or 6 but got: %s", filterArguments.size());
+ byte [] family = ParseFilter.removeQuotesFromByteArray(filterArguments.get(0));
+ byte [] qualifier = ParseFilter.removeQuotesFromByteArray(filterArguments.get(1));
+ CompareOp compareOp = ParseFilter.createCompareOp(filterArguments.get(2));
+ ByteArrayComparable comparator = ParseFilter.createComparator(
+ ParseFilter.removeQuotesFromByteArray(filterArguments.get(3)));
+
+ if (comparator instanceof RegexStringComparator ||
+ comparator instanceof SubstringComparator) {
+ if (compareOp != CompareOp.EQUAL &&
+ compareOp != CompareOp.NOT_EQUAL) {
+ throw new IllegalArgumentException ("A regexstring comparator and substring comparator " +
+ "can only be used with EQUAL and NOT_EQUAL");
+ }
+ }
+
+ SingleColumnValueFilter filter = new SingleColumnValueFilter(family, qualifier,
+ compareOp, comparator);
+
+ if (filterArguments.size() == 6) {
+ boolean filterIfMissing = ParseFilter.convertByteArrayToBoolean(filterArguments.get(4));
+ boolean latestVersionOnly = ParseFilter.convertByteArrayToBoolean(filterArguments.get(5));
+ filter.setFilterIfMissing(filterIfMissing);
+ filter.setLatestVersionOnly(latestVersionOnly);
+ }
+ return filter;
+ }
+
+ FilterProtos.SingleColumnValueFilter convert() {
+ FilterProtos.SingleColumnValueFilter.Builder builder =
+ FilterProtos.SingleColumnValueFilter.newBuilder();
+ if (this.columnFamily != null) {
+ builder.setColumnFamily(ByteString.copyFrom(this.columnFamily));
+ }
+ if (this.columnQualifier != null) {
+ builder.setColumnQualifier(ByteString.copyFrom(this.columnQualifier));
+ }
+ HBaseProtos.CompareType compareOp = CompareType.valueOf(this.compareOp.name());
+ builder.setCompareOp(compareOp);
+ builder.setComparator(ProtobufUtil.toComparator(this.comparator));
+ builder.setFilterIfMissing(this.filterIfMissing);
+ builder.setLatestVersionOnly(this.latestVersionOnly);
+
+ return builder.build();
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ return convert().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link SingleColumnValueFilter} instance
+ * @return An instance of {@link SingleColumnValueFilter} made from <code>bytes</code>
+ * @throws org.apache.hadoop.hbase.exceptions.DeserializationException
+ * @see #toByteArray
+ */
+ public static SingleColumnValueFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.SingleColumnValueFilter proto;
+ try {
+ proto = FilterProtos.SingleColumnValueFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+
+ final CompareOp compareOp =
+ CompareOp.valueOf(proto.getCompareOp().name());
+ final ByteArrayComparable comparator;
+ try {
+ comparator = ProtobufUtil.toComparator(proto.getComparator());
+ } catch (IOException ioe) {
+ throw new DeserializationException(ioe);
+ }
+
+ return new SingleColumnValueFilter(proto.hasColumnFamily() ? proto.getColumnFamily()
+ .toByteArray() : null, proto.hasColumnQualifier() ? proto.getColumnQualifier()
+ .toByteArray() : null, compareOp, comparator, proto.getFilterIfMissing(), proto
+ .getLatestVersionOnly());
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof SingleColumnValueFilter)) return false;
+
+ SingleColumnValueFilter other = (SingleColumnValueFilter)o;
+ return Bytes.equals(this.getFamily(), other.getFamily())
+ && Bytes.equals(this.getQualifier(), other.getQualifier())
+ && this.compareOp.equals(other.compareOp)
+ && this.getComparator().areSerializedFieldsEqual(other.getComparator())
+ && this.getFilterIfMissing() == other.getFilterIfMissing()
+ && this.getLatestVersionOnly() == other.getLatestVersionOnly();
+ }
+
+ /**
+ * The only CF this filter needs is given column family. So, it's the only essential
+ * column in whole scan. If filterIfMissing == false, all families are essential,
+ * because of possibility of skipping the rows without any data in filtered CF.
+ */
+ public boolean isFamilyEssential(byte[] name) {
+ return !this.filterIfMissing || Bytes.equals(name, this.columnFamily);
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s (%s, %s, %s, %s)",
+ this.getClass().getSimpleName(), Bytes.toStringBinary(this.columnFamily),
+ Bytes.toStringBinary(this.columnQualifier), this.compareOp.name(),
+ Bytes.toStringBinary(this.comparator.getValue()));
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SkipFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SkipFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SkipFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SkipFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,145 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+
+import java.io.IOException;
+
+/**
+ * A wrapper filter that filters an entire row if any of the KeyValue checks do
+ * not pass.
+ * <p>
+ * For example, if all columns in a row represent weights of different things,
+ * with the values being the actual weights, and we want to filter out the
+ * entire row if any of its weights are zero. In this case, we want to prevent
+ * rows from being emitted if a single key is filtered. Combine this filter
+ * with a {@link ValueFilter}:
+ * <p>
+ * <pre>
+ * scan.setFilter(new SkipFilter(new ValueFilter(CompareOp.NOT_EQUAL,
+ * new BinaryComparator(Bytes.toBytes(0))));
+ * </code>
+ * Any row which contained a column whose value was 0 will be filtered out
+ * (since ValueFilter will not pass that KeyValue).
+ * Without this filter, the other non-zero valued columns in the row would still
+ * be emitted.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class SkipFilter extends FilterBase {
+ private boolean filterRow = false;
+ private Filter filter;
+
+ public SkipFilter(Filter filter) {
+ this.filter = filter;
+ }
+
+ public Filter getFilter() {
+ return filter;
+ }
+
+ public void reset() {
+ filter.reset();
+ filterRow = false;
+ }
+
+ private void changeFR(boolean value) {
+ filterRow = filterRow || value;
+ }
+
+ public ReturnCode filterKeyValue(KeyValue v) {
+ ReturnCode c = filter.filterKeyValue(v);
+ changeFR(c != ReturnCode.INCLUDE);
+ return c;
+ }
+
+ @Override
+ public KeyValue transform(KeyValue v) {
+ return filter.transform(v);
+ }
+
+ public boolean filterRow() {
+ return filterRow;
+ }
+
+ public boolean hasFilterRow() {
+ return true;
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ FilterProtos.SkipFilter.Builder builder =
+ FilterProtos.SkipFilter.newBuilder();
+ builder.setFilter(ProtobufUtil.toFilter(this.filter));
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link SkipFilter} instance
+ * @return An instance of {@link SkipFilter} made from <code>bytes</code>
+ * @throws DeserializationException
+ * @see #toByteArray
+ */
+ public static SkipFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.SkipFilter proto;
+ try {
+ proto = FilterProtos.SkipFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+ try {
+ return new SkipFilter(ProtobufUtil.toFilter(proto.getFilter()));
+ } catch (IOException ioe) {
+ throw new DeserializationException(ioe);
+ }
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof SkipFilter)) return false;
+
+ SkipFilter other = (SkipFilter)o;
+ return getFilter().areSerializedFieldsEqual(other.getFilter());
+ }
+
+ public boolean isFamilyEssential(byte[] name) {
+ return filter.isFamilyEssential(name);
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + " " + this.filter.toString();
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SubstringComparator.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SubstringComparator.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SubstringComparator.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/SubstringComparator.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,111 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.protobuf.generated.ComparatorProtos;
+import org.apache.hadoop.hbase.util.Bytes;
+
+
+/**
+ * This comparator is for use with SingleColumnValueFilter, for filtering based on
+ * the value of a given column. Use it to test if a given substring appears
+ * in a cell value in the column. The comparison is case insensitive.
+ * <p>
+ * Only EQUAL or NOT_EQUAL tests are valid with this comparator.
+ * <p>
+ * For example:
+ * <p>
+ * <pre>
+ * SingleColumnValueFilter scvf =
+ * new SingleColumnValueFilter("col", CompareOp.EQUAL,
+ * new SubstringComparator("substr"));
+ * </pre>
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class SubstringComparator extends ByteArrayComparable {
+
+ private String substr;
+
+ /**
+ * Constructor
+ * @param substr the substring
+ */
+ public SubstringComparator(String substr) {
+ super(Bytes.toBytes(substr.toLowerCase()));
+ this.substr = substr.toLowerCase();
+ }
+
+ @Override
+ public byte[] getValue() {
+ return Bytes.toBytes(substr);
+ }
+
+ @Override
+ public int compareTo(byte[] value, int offset, int length) {
+ return Bytes.toString(value, offset, length).toLowerCase().contains(substr) ? 0
+ : 1;
+ }
+
+ /**
+ * @return The comparator serialized using pb
+ */
+ public byte [] toByteArray() {
+ ComparatorProtos.SubstringComparator.Builder builder =
+ ComparatorProtos.SubstringComparator.newBuilder();
+ builder.setSubstr(this.substr);
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link SubstringComparator} instance
+ * @return An instance of {@link SubstringComparator} made from <code>bytes</code>
+ * @throws DeserializationException
+ * @see #toByteArray
+ */
+ public static SubstringComparator parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ ComparatorProtos.SubstringComparator proto;
+ try {
+ proto = ComparatorProtos.SubstringComparator.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+ return new SubstringComparator(proto.getSubstr());
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the comparator that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(ByteArrayComparable other) {
+ if (other == this) return true;
+ if (!(other instanceof SubstringComparator)) return false;
+
+ SubstringComparator comparator = (SubstringComparator)other;
+ return super.areSerializedFieldsEqual(comparator)
+ && this.substr.equals(comparator.substr);
+ }
+
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/TimestampsFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/TimestampsFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/TimestampsFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/TimestampsFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,174 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeSet;
+
+/**
+ * Filter that returns only cells whose timestamp (version) is
+ * in the specified list of timestamps (versions).
+ * <p>
+ * Note: Use of this filter overrides any time range/time stamp
+ * options specified using {@link org.apache.hadoop.hbase.client.Get#setTimeRange(long, long)},
+ * {@link org.apache.hadoop.hbase.client.Scan#setTimeRange(long, long)}, {@link org.apache.hadoop.hbase.client.Get#setTimeStamp(long)},
+ * or {@link org.apache.hadoop.hbase.client.Scan#setTimeStamp(long)}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class TimestampsFilter extends FilterBase {
+
+ TreeSet<Long> timestamps;
+ private static final int MAX_LOG_TIMESTAMPS = 5;
+
+ // Used during scans to hint the scan to stop early
+ // once the timestamps fall below the minTimeStamp.
+ long minTimeStamp = Long.MAX_VALUE;
+
+ /**
+ * Constructor for filter that retains only those
+ * cells whose timestamp (version) is in the specified
+ * list of timestamps.
+ *
+ * @param timestamps
+ */
+ public TimestampsFilter(List<Long> timestamps) {
+ for (Long timestamp : timestamps) {
+ Preconditions.checkArgument(timestamp >= 0, "must be positive %s", timestamp);
+ }
+ this.timestamps = new TreeSet<Long>(timestamps);
+ init();
+ }
+
+ /**
+ * @return the list of timestamps
+ */
+ public List<Long> getTimestamps() {
+ List<Long> list = new ArrayList<Long>(timestamps.size());
+ list.addAll(timestamps);
+ return list;
+ }
+
+ private void init() {
+ if (this.timestamps.size() > 0) {
+ minTimeStamp = this.timestamps.first();
+ }
+ }
+
+ /**
+ * Gets the minimum timestamp requested by filter.
+ * @return minimum timestamp requested by filter.
+ */
+ public long getMin() {
+ return minTimeStamp;
+ }
+
+ @Override
+ public ReturnCode filterKeyValue(KeyValue v) {
+ if (this.timestamps.contains(v.getTimestamp())) {
+ return ReturnCode.INCLUDE;
+ } else if (v.getTimestamp() < minTimeStamp) {
+ // The remaining versions of this column are guaranteed
+ // to be lesser than all of the other values.
+ return ReturnCode.NEXT_COL;
+ }
+ return ReturnCode.SKIP;
+ }
+
+ public static Filter createFilterFromArguments(ArrayList<byte []> filterArguments) {
+ ArrayList<Long> timestamps = new ArrayList<Long>();
+ for (int i = 0; i<filterArguments.size(); i++) {
+ long timestamp = ParseFilter.convertByteArrayToLong(filterArguments.get(i));
+ timestamps.add(timestamp);
+ }
+ return new TimestampsFilter(timestamps);
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ FilterProtos.TimestampsFilter.Builder builder =
+ FilterProtos.TimestampsFilter.newBuilder();
+ builder.addAllTimestamps(this.timestamps);
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link TimestampsFilter} instance
+ * @return An instance of {@link TimestampsFilter} made from <code>bytes</code>
+ * @throws DeserializationException
+ * @see #toByteArray
+ */
+ public static TimestampsFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.TimestampsFilter proto;
+ try {
+ proto = FilterProtos.TimestampsFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+ return new TimestampsFilter(proto.getTimestampsList());
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof TimestampsFilter)) return false;
+
+ TimestampsFilter other = (TimestampsFilter)o;
+ return this.getTimestamps().equals(other.getTimestamps());
+ }
+
+ @Override
+ public String toString() {
+ return toString(MAX_LOG_TIMESTAMPS);
+ }
+
+ protected String toString(int maxTimestamps) {
+ StringBuilder tsList = new StringBuilder();
+
+ int count = 0;
+ for (Long ts : this.timestamps) {
+ if (count >= maxTimestamps) {
+ break;
+ }
+ ++count;
+ tsList.append(ts.toString());
+ if (count < this.timestamps.size() && count < maxTimestamps) {
+ tsList.append(", ");
+ }
+ }
+
+ return String.format("%s (%d/%d): [%s]", this.getClass().getSimpleName(),
+ count, this.timestamps.size(), tsList.toString());
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/ValueFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/ValueFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/ValueFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/ValueFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,124 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+/**
+ * This filter is used to filter based on column value. It takes an
+ * operator (equal, greater, not equal, etc) and a byte [] comparator for the
+ * cell value.
+ * <p>
+ * This filter can be wrapped with {@link WhileMatchFilter} and {@link SkipFilter}
+ * to add more control.
+ * <p>
+ * Multiple filters can be combined using {@link FilterList}.
+ * <p>
+ * To test the value of a single qualifier when scanning multiple qualifiers,
+ * use {@link SingleColumnValueFilter}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class ValueFilter extends CompareFilter {
+
+ /**
+ * Constructor.
+ * @param valueCompareOp the compare op for value matching
+ * @param valueComparator the comparator for value matching
+ */
+ public ValueFilter(final CompareOp valueCompareOp,
+ final ByteArrayComparable valueComparator) {
+ super(valueCompareOp, valueComparator);
+ }
+
+ @Override
+ public ReturnCode filterKeyValue(KeyValue v) {
+ if (doCompare(this.compareOp, this.comparator, v.getBuffer(),
+ v.getValueOffset(), v.getValueLength())) {
+ return ReturnCode.SKIP;
+ }
+ return ReturnCode.INCLUDE;
+ }
+
+ public static Filter createFilterFromArguments(ArrayList<byte []> filterArguments) {
+ ArrayList arguments = CompareFilter.extractArguments(filterArguments);
+ CompareOp compareOp = (CompareOp)arguments.get(0);
+ ByteArrayComparable comparator = (ByteArrayComparable)arguments.get(1);
+ return new ValueFilter(compareOp, comparator);
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ FilterProtos.ValueFilter.Builder builder =
+ FilterProtos.ValueFilter.newBuilder();
+ builder.setCompareFilter(super.convert());
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link ValueFilter} instance
+ * @return An instance of {@link ValueFilter} made from <code>bytes</code>
+ * @throws DeserializationException
+ * @see #toByteArray
+ */
+ public static ValueFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.ValueFilter proto;
+ try {
+ proto = FilterProtos.ValueFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+ final CompareOp valueCompareOp =
+ CompareOp.valueOf(proto.getCompareFilter().getCompareOp().name());
+ ByteArrayComparable valueComparator = null;
+ try {
+ if (proto.getCompareFilter().hasComparator()) {
+ valueComparator = ProtobufUtil.toComparator(proto.getCompareFilter().getComparator());
+ }
+ } catch (IOException ioe) {
+ throw new DeserializationException(ioe);
+ }
+ return new ValueFilter(valueCompareOp,valueComparator);
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof ValueFilter)) return false;
+
+ return super.areSerializedFieldsEqual(o);
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/WhileMatchFilter.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/WhileMatchFilter.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/WhileMatchFilter.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/WhileMatchFilter.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,145 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.filter;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
+
+import java.io.IOException;
+
+/**
+ * A wrapper filter that returns true from {@link #filterAllRemaining()} as soon
+ * as the wrapped filters {@link Filter#filterRowKey(byte[], int, int)},
+ * {@link Filter#filterKeyValue(org.apache.hadoop.hbase.KeyValue)},
+ * {@link org.apache.hadoop.hbase.filter.Filter#filterRow()} or
+ * {@link org.apache.hadoop.hbase.filter.Filter#filterAllRemaining()} methods
+ * returns true.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class WhileMatchFilter extends FilterBase {
+ private boolean filterAllRemaining = false;
+ private Filter filter;
+
+ public WhileMatchFilter(Filter filter) {
+ this.filter = filter;
+ }
+
+ public Filter getFilter() {
+ return filter;
+ }
+
+ public void reset() {
+ this.filter.reset();
+ }
+
+ private void changeFAR(boolean value) {
+ filterAllRemaining = filterAllRemaining || value;
+ }
+
+ public boolean filterAllRemaining() {
+ return this.filterAllRemaining || this.filter.filterAllRemaining();
+ }
+
+ public boolean filterRowKey(byte[] buffer, int offset, int length) {
+ boolean value = filter.filterRowKey(buffer, offset, length);
+ changeFAR(value);
+ return value;
+ }
+
+ public ReturnCode filterKeyValue(KeyValue v) {
+ ReturnCode c = filter.filterKeyValue(v);
+ changeFAR(c != ReturnCode.INCLUDE);
+ return c;
+ }
+
+ @Override
+ public KeyValue transform(KeyValue v) {
+ return filter.transform(v);
+ }
+
+ public boolean filterRow() {
+ boolean filterRow = this.filter.filterRow();
+ changeFAR(filterRow);
+ return filterRow;
+ }
+
+ public boolean hasFilterRow() {
+ return true;
+ }
+
+ /**
+ * @return The filter serialized using pb
+ */
+ public byte [] toByteArray() {
+ FilterProtos.WhileMatchFilter.Builder builder =
+ FilterProtos.WhileMatchFilter.newBuilder();
+ builder.setFilter(ProtobufUtil.toFilter(this.filter));
+ return builder.build().toByteArray();
+ }
+
+ /**
+ * @param pbBytes A pb serialized {@link WhileMatchFilter} instance
+ * @return An instance of {@link WhileMatchFilter} made from <code>bytes</code>
+ * @throws org.apache.hadoop.hbase.exceptions.DeserializationException
+ * @see #toByteArray
+ */
+ public static WhileMatchFilter parseFrom(final byte [] pbBytes)
+ throws DeserializationException {
+ FilterProtos.WhileMatchFilter proto;
+ try {
+ proto = FilterProtos.WhileMatchFilter.parseFrom(pbBytes);
+ } catch (InvalidProtocolBufferException e) {
+ throw new DeserializationException(e);
+ }
+ try {
+ return new WhileMatchFilter(ProtobufUtil.toFilter(proto.getFilter()));
+ } catch (IOException ioe) {
+ throw new DeserializationException(ioe);
+ }
+ }
+
+ /**
+ * @param other
+ * @return true if and only if the fields of the filter that are serialized
+ * are equal to the corresponding fields in other. Used for testing.
+ */
+ boolean areSerializedFieldsEqual(Filter o) {
+ if (o == this) return true;
+ if (!(o instanceof WhileMatchFilter)) return false;
+
+ WhileMatchFilter other = (WhileMatchFilter)o;
+ return getFilter().areSerializedFieldsEqual(other.getFilter());
+ }
+
+ public boolean isFamilyEssential(byte[] name) {
+ return filter.isFamilyEssential(name);
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + " " + this.filter.toString();
+ }
+}
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/package-info.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/package-info.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/package-info.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/filter/package-info.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,34 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Provides row-level filters applied to HRegion scan results during calls to
+ * {@link org.apache.hadoop.hbase.client.ResultScanner#next()}.
+
+<p>
+Filters run the extent of a table unless you wrap your filter in a
+{@link org.apache.hadoop.hbase.filter.WhileMatchFilter}.
+The latter returns as soon as the filter stops matching.
+</p>
+<p>Do not rely on filters carrying state across rows; its not reliable in current
+hbase as we have no handlers in place for when regions split, close or server
+crashes.
+</p>
+*/
+package org.apache.hadoop.hbase.filter;
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcCallback.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcCallback.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcCallback.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcCallback.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.ipc;
+
+import com.google.protobuf.RpcCallback;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+
+/**
+ * Simple {@link RpcCallback} implementation providing a
+ * {@link java.util.concurrent.Future}-like {@link BlockingRpcCallback#get()} method, which
+ * will block util the instance's {@link BlockingRpcCallback#run(Object)} method has been called.
+ * {@code R} is the RPC response type that will be passed to the {@link #run(Object)} method.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class BlockingRpcCallback<R> implements RpcCallback<R> {
+ private R result;
+ private boolean resultSet = false;
+
+ /**
+ * Called on completion of the RPC call with the response object, or {@code null} in the case of
+ * an error.
+ * @param parameter the response object or {@code null} if an error occurred
+ */
+ @Override
+ public void run(R parameter) {
+ synchronized (this) {
+ result = parameter;
+ resultSet = true;
+ this.notify();
+ }
+ }
+
+ /**
+ * Returns the parameter passed to {@link #run(Object)} or {@code null} if a null value was
+ * passed. When used asynchronously, this method will block until the {@link #run(Object)}
+ * method has been called.
+ * @return the response object or {@code null} if no response was passed
+ */
+ public synchronized R get() throws IOException {
+ while (!resultSet) {
+ try {
+ this.wait();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ InterruptedIOException exception = new InterruptedIOException(ie.getMessage());
+ exception.initCause(ie);
+ throw exception;
+ }
+ }
+ return result;
+ }
+}
\ No newline at end of file
Added: hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/CoprocessorRpcChannel.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/CoprocessorRpcChannel.java?rev=1449950&view=auto
==============================================================================
--- hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/CoprocessorRpcChannel.java (added)
+++ hbase/trunk/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/CoprocessorRpcChannel.java Mon Feb 25 22:50:17 2013
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.hadoop.hbase.ipc;
+
+import com.google.protobuf.BlockingRpcChannel;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.Message;
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcChannel;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.protobuf.ResponseConverter;
+
+import java.io.IOException;
+
+/**
+ * Base class which provides clients with an RPC connection to
+ * call coprocessor endpoint {@link Service}s
+ */
+@InterfaceAudience.Private
+public abstract class CoprocessorRpcChannel implements RpcChannel, BlockingRpcChannel {
+ private static Log LOG = LogFactory.getLog(CoprocessorRpcChannel.class);
+
+ @Override
+ public void callMethod(Descriptors.MethodDescriptor method,
+ RpcController controller,
+ Message request, Message responsePrototype,
+ RpcCallback<Message> callback) {
+ Message response = null;
+ try {
+ response = callExecService(method, request, responsePrototype);
+ } catch (IOException ioe) {
+ LOG.warn("Call failed on IOException", ioe);
+ ResponseConverter.setControllerException(controller, ioe);
+ }
+ if (callback != null) {
+ callback.run(response);
+ }
+ }
+
+ @Override
+ public Message callBlockingMethod(Descriptors.MethodDescriptor method,
+ RpcController controller,
+ Message request, Message responsePrototype)
+ throws ServiceException {
+ try {
+ return callExecService(method, request, responsePrototype);
+ } catch (IOException ioe) {
+ throw new ServiceException("Error calling method "+method.getFullName(), ioe);
+ }
+ }
+
+ protected abstract Message callExecService(Descriptors.MethodDescriptor method,
+ Message request, Message responsePrototype) throws IOException;
+}