Counter super slice

core/src/main/java/me/prettyprint/cassandra/model/thrift/ThriftSuperSliceCounterQuery.java

@@ -0,0 +1,88 @@
+package me.prettyprint.cassandra.model.thrift;
+
+import me.prettyprint.cassandra.model.*;
+import me.prettyprint.cassandra.serializers.LongSerializer;
+import me.prettyprint.cassandra.service.KeyspaceService;
+import me.prettyprint.cassandra.utils.Assert;
+import me.prettyprint.hector.api.Keyspace;
+import me.prettyprint.hector.api.Serializer;
+import me.prettyprint.hector.api.beans.CounterSuperSlice;
+import me.prettyprint.hector.api.beans.SuperSlice;
+import me.prettyprint.hector.api.exceptions.HectorException;
+import me.prettyprint.hector.api.query.QueryResult;
+import me.prettyprint.hector.api.query.SuperSliceCounterQuery;
+import me.prettyprint.hector.api.query.SuperSliceQuery;
+import org.apache.cassandra.thrift.ColumnParent;
+import org.apache.cassandra.thrift.CounterSuperColumn;
+import org.apache.cassandra.thrift.SuperColumn;
+
+import java.util.List;
+
+
+/**
+ * A query for the thrift call get_slice.
+ * <p>
+ * Get a slice of super columns from a super column family.
+ *
+ * @author Ran Tavory
+ *
+ * @param <N>
+ */
+public final class ThriftSuperSliceCounterQuery<K, SN, N> extends
+    AbstractSliceQuery<K, SN, Long, CounterSuperSlice<SN, N>> implements SuperSliceCounterQuery<K, SN, N> {
+
+  private K key;
+  private final Serializer<N> nameSerializer;
+
+  public ThriftSuperSliceCounterQuery(Keyspace keyspace,
+                                      Serializer<K> keySerializer,
+                                      Serializer<SN> sNameSerializer,
+                                      Serializer<N> nameSerializer) {
+    super(keyspace, keySerializer, sNameSerializer, LongSerializer.get());
+    Assert.notNull(sNameSerializer, "sNameSerializer cannot be null");
+    this.nameSerializer = nameSerializer;
+  }
+
+  @Override
+  public SuperSliceCounterQuery<K, SN, N> setKey(K key) {
+    this.key = key;
+    return this;
+  }
+
+  @Override
+  public QueryResult<CounterSuperSlice<SN, N>> execute() {
+    return new QueryResultImpl<CounterSuperSlice<SN,N>>(keyspace.doExecute(
+        new KeyspaceOperationCallback<CounterSuperSlice<SN,N>>() {
+          @Override
+          public CounterSuperSlice<SN, N> doInKeyspace(KeyspaceService ks) throws HectorException {
+            ColumnParent columnParent = new ColumnParent(columnFamilyName);
+            List<CounterSuperColumn> thriftRet = ks.getCounterSuperSlice(keySerializer.toByteBuffer(key),
+                    columnParent, getPredicate());
+            return new CounterSuperSliceImpl<SN, N>(thriftRet, columnNameSerializer, nameSerializer);
+          }
+        }), this);
+  }
+
+  @Override
+  public String toString() {
+    return "SuperSliceQuery(" + key + "," + toStringInternal() + ")";
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+    public SuperSliceCounterQuery<K, SN, N> setColumnNames(SN... columnNames) {
+      return (SuperSliceCounterQuery<K, SN, N>) super.setColumnNames(columnNames);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public SuperSliceCounterQuery<K, SN, N> setRange(SN start, SN finish, boolean reversed, int count) {
+      return (SuperSliceCounterQuery<K, SN, N>) super.setRange(start, finish, reversed, count);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public SuperSliceCounterQuery<K, SN, N> setColumnFamily(String cf) {
+      return (SuperSliceCounterQuery<K, SN, N>) super.setColumnFamily(cf);
+    }
+}

core/src/main/java/me/prettyprint/cassandra/service/KeyspaceService.java

@@ -121,6 +121,15 @@ List<SuperColumn> getSuperSlice(ByteBuffer key, ColumnParent columnParent,
   List<SuperColumn> getSuperSlice(String key, ColumnParent columnParent,
           SlicePredicate predicate) throws HectorException;
 
+  /**
+   * Get the group of superColumn contained by columnParent.
+   */
+  List<CounterSuperColumn> getCounterSuperSlice(ByteBuffer key, ColumnParent columnParent,
+      SlicePredicate predicate) throws HectorException;
+
+  List<CounterSuperColumn> getCounterSuperSlice(String key, ColumnParent columnParent,
+          SlicePredicate predicate) throws HectorException;
+
   /**
    * Performs a get for columnPath in parallel on the given list of keys.
    *

core/src/main/java/me/prettyprint/cassandra/service/KeyspaceServiceImpl.java

@@ -423,6 +423,41 @@ public List<SuperColumn> execute(Cassandra.Client cassandra) throws HectorExcept
   }
 
 
+  @Override
+  public List<CounterSuperColumn> getCounterSuperSlice(String key, ColumnParent columnParent, SlicePredicate predicate)
+          throws HectorException {
+      return getCounterSuperSlice(StringSerializer.get().toByteBuffer(key), columnParent, predicate);
+  }
+
+
+  @Override
+  public List<CounterSuperColumn> getCounterSuperSlice(final ByteBuffer key, final ColumnParent columnParent,
+      final SlicePredicate predicate) throws HectorException {
+    Operation<List<CounterSuperColumn>> op = new Operation<List<CounterSuperColumn>>(OperationType.READ, failoverPolicy, keyspaceName, credentials) {
+
+      @Override
+      public List<CounterSuperColumn> execute(Cassandra.Client cassandra) throws HectorException {
+        try {
+          List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(key, columnParent,
+              predicate, getThriftCl(OperationType.READ));
+          if (cosclist == null) {
+            return null;
+          }
+          ArrayList<CounterSuperColumn> result = new ArrayList<CounterSuperColumn>(cosclist.size());
+          for (ColumnOrSuperColumn cosc : cosclist) {
+            result.add(cosc.getCounter_super_column());
+          }
+          return result;
+        } catch (Exception e) {
+          throw xtrans.translate(e);
+        }
+      }
+    };
+    operateWithFailover(op);
+    return op.getResult();
+  }
+
+
 
   @Override
   public void insert(final ByteBuffer key, final ColumnParent columnParent, final Column column) throws HectorException {

core/src/main/java/me/prettyprint/hector/api/factory/HFactory.java

@@ -348,6 +348,14 @@ public static <K, N> MultigetSliceCounterQuery<K, N> createMultigetSliceCounterQ
         nameSerializer);
   }
 
+  public static <K, SN, N> SuperSliceCounterQuery<K, SN, N> createSuperSliceCounterQuery(
+          Keyspace keyspace, Serializer<K> keySerializer,
+          Serializer<SN> sNameSerializer,
+          Serializer<N> nameSerializer) {
+      return new ThriftSuperSliceCounterQuery<K, SN, N>(keyspace, keySerializer, sNameSerializer, nameSerializer);
+  }
+
+
 
   public static <K, SN, N, V> SubColumnQuery<K, SN, N, V> createSubColumnQuery(
       Keyspace keyspace, Serializer<K> keySerializer,

core/src/main/java/me/prettyprint/hector/api/query/SuperSliceCounterQuery.java

@@ -0,0 +1,30 @@
+package me.prettyprint.hector.api.query;
+
+import me.prettyprint.hector.api.beans.CounterSuperSlice;
+import me.prettyprint.hector.api.beans.SuperSlice;
+
+import java.util.Collection;
+
+/**
+ * A query for the  call get_slice.
+ * <p>
+ * Get a slice of super columns from a super column family.
+ *
+ * @author Ran Tavory
+ *
+ * @param <SN> type of the super column name
+ * @param <N> type of the column name
+ */
+public interface SuperSliceCounterQuery<K, SN, N> extends Query<CounterSuperSlice<SN, N>> {
+
+  SuperSliceCounterQuery<K, SN, N> setKey(K key);
+
+  SuperSliceCounterQuery<K, SN, N> setColumnNames(SN... columnNames);
+
+  SuperSliceCounterQuery<K, SN, N> setRange(SN start, SN finish, boolean reversed, int count);
+
+  SuperSliceCounterQuery<K, SN, N> setColumnFamily(String cf);
+
+  Collection<SN> getColumnNames();
+
+}