@@ -0,0 +1,1300 @@
+package com.hmsonline.storm.cassandra.trident;%0A%0Aimport java.util.List;%0Aimport java.util.Map;%0A%0Aimport backtype.storm.tuple.Values;%0A%0Aimport com.hmsonline.storm.cassandra.bolt.mapper.TridentColumnMapper;%0Aimport com.hmsonline.storm.cassandra.bolt.mapper.TridentTupleMapper;%0A%0Aimport storm.trident.operation.TridentCollector;%0Aimport storm.trident.state.BaseQueryFunction;%0Aimport storm.trident.tuple.TridentTuple;%0A%0Apublic class CassandraQuery extends BaseQueryFunction%3CCassandraState, Map%3C?,?%3E%3E %7B%0A    %0A    private TridentColumnMapper columnsMapper;%0A    private TridentTupleMapper tupleMapper;%0A    %0A    %0A    public CassandraQuery(String clientConfigKey, TridentTupleMapper tupleMapper,%0A            TridentColumnMapper columnMapper)%7B%0A        this.columnsMapper = columnMapper;%0A        this.tupleMapper = tupleMapper;%0A    %7D%0A%0A    @Override%0A    public List%3CMap%3C?,?%3E%3E batchRetrieve(CassandraState state, List%3CTridentTuple%3E args) %7B%0A        return state.batchRetrieve(args, this.tupleMapper);%0A    %7D%0A%0A    @Override%0A    public void execute(TridentTuple tuple, Map%3C?,?%3E result, TridentCollector collector) %7B%0A        List%3CValues%3E emitList = this.columnsMapper.mapToValues(this.tupleMapper.mapToRowKey(tuple), result, tuple);%0A        for(Values values : emitList)%7B%0A            collector.emit(values);%0A        %7D%0A    %7D%0A%0A%7D%0A

refactor CassandraState and add state query support.