extend filter to multi-column gets

src/main/java/com/yahoo/omid/client/ColumnFamilyAndQuantifier.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
+ *
+ * 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. See accompanying LICENSE file.
+ */
+
+package com.yahoo.omid.client;
+
+//A wrapper for both column family and the qualifier
+//Make it easier to be used in maps and hash maps
+public class ColumnFamilyAndQuantifier {
+   protected byte[] family;
+   protected byte[] qualifier;
+   protected Integer hash = null;
+   ColumnFamilyAndQuantifier(byte[] f, byte[] q) {
+      family = f;
+      qualifier = q;
+   }
+   @Override
+      public boolean equals(Object o) {
+         if (o instanceof ColumnFamilyAndQuantifier) {
+            ColumnFamilyAndQuantifier other = (ColumnFamilyAndQuantifier) o;
+            if (family.length != other.family.length || qualifier.length != other.qualifier.length)
+               return false;
+            for (int i = 0; i < family.length; i++)
+               if (family[i] != other.family[i])
+                  return false;
+            for (int i = 0; i < qualifier.length; i++)
+               if (qualifier[i] != other.qualifier[i])
+                  return false;
+            return true;
+         }
+         return false;
+      }
+   @Override
+      public int hashCode() {
+         if (hash != null)
+            return hash;
+         int h = 0;
+         h = computeHash(h, family);
+         h = computeHash(h, qualifier);
+         hash = h;
+         return h;
+      }
+   private int computeHash(int hash, byte[] larray) {
+      hash += larray.length;
+      for (int i = 0; i < larray.length; i++) {
+         hash += larray[i];
+      }
+      return hash;
+   }
+}

src/main/java/com/yahoo/omid/client/MinVersionsFilter.java

@@ -27,6 +27,8 @@
 
 /**
  * Filter that sets both minTimestamp and minVersions
+ * It assumes a single row but works with more than one column family/qualifier
+ * It also assumes the following order <column family, column qualifier, timestamp>
  * @maysam
  */
 public class MinVersionsFilter extends FilterBase {
@@ -36,8 +38,9 @@ public class MinVersionsFilter extends FilterBase {
    long endTime = Long.MAX_VALUE;
    int minVersions;
 
-   int includedVersions;
-
+   //keep track of included versions for each column qualifier of each column family
+   int includedVersionsForLastColumn;
+   ColumnFamilyAndQuantifier lastColumn;
 
    /**
     * Used during deserialization. Do not use otherwise.
@@ -54,16 +57,32 @@ public MinVersionsFilter(long startTime, long endTime, int minVersions) {
    }
 
    private void init() {
-      includedVersions = 0;
+      includedVersionsForLastColumn = 0;
+      lastColumn = null;
+   }
+
+   private int getIncludedVersions(ColumnFamilyAndQuantifier column) {
+      if (lastColumn == null || !lastColumn.equals(column)) {
+         lastColumn = column;
+         includedVersionsForLastColumn = 0;
+      }
+      return includedVersionsForLastColumn;
+   }
+
+   private void setIncludedVersions(ColumnFamilyAndQuantifier column, int versions) {
+      includedVersionsForLastColumn = versions;
    }
 
    @Override
       public ReturnCode filterKeyValue(KeyValue v) {
          long version = v.getTimestamp();
          if (version >= endTime)
             return ReturnCode.SKIP;
+         ColumnFamilyAndQuantifier column = new ColumnFamilyAndQuantifier(v.getFamily(), v.getQualifier());
+         int includedVersions = getIncludedVersions(column);
          if (includedVersions < minVersions || version > startTime) {
             includedVersions++;
+            setIncludedVersions(column, includedVersions);
             return ReturnCode.INCLUDE;
          }
          return ReturnCode.NEXT_COL;
@@ -84,3 +103,4 @@ public void write(DataOutput out) throws IOException {
          out.writeInt(this.minVersions);
       }
 }
+

src/main/java/com/yahoo/omid/client/MinVersionsSingleColumnFilter.java

@@ -0,0 +1,88 @@
+/**
+ * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
+ *
+ * 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. See accompanying LICENSE file.
+ */
+
+package com.yahoo.omid.client;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+import java.util.TreeSet;
+
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.filter.*;
+
+/**
+ * Filter that sets both minTimestamp and minVersions
+ * Assumes that there is only one column in the output
+ * @maysam
+ */
+public class MinVersionsSingleColumnFilter extends FilterBase {
+
+   //read at least minVersions and go till reach startTime
+   long startTime = 0;
+   long endTime = Long.MAX_VALUE;
+   int minVersions;
+
+   int includedVersions;
+
+
+   /**
+    * Used during deserialization. Do not use otherwise.
+    */
+   public MinVersionsSingleColumnFilter() {
+      super();
+   }
+
+   public MinVersionsSingleColumnFilter(long startTime, long endTime, int minVersions) {
+      this.startTime = startTime;
+      this.endTime = endTime;
+      this.minVersions = minVersions;
+      init();
+   }
+
+   private void init() {
+      includedVersions = 0;
+   }
+
+   @Override
+      public ReturnCode filterKeyValue(KeyValue v) {
+         long version = v.getTimestamp();
+         if (version >= endTime)
+            return ReturnCode.SKIP;
+         if (includedVersions < minVersions || version > startTime) {
+            includedVersions++;
+            return ReturnCode.INCLUDE;
+         }
+         return ReturnCode.NEXT_COL;
+      }
+
+   @Override
+      public void readFields(DataInput in) throws IOException {
+         this.startTime = in.readLong();
+         this.endTime = in.readLong();
+         this.minVersions = in.readInt();
+         init();
+      }
+
+   @Override
+      public void write(DataOutput out) throws IOException {
+         out.writeLong(this.startTime);
+         out.writeLong(this.endTime);
+         out.writeInt(this.minVersions);
+      }
+}
+