import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;public class test2 {
    public static void main(String[] args) throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException, UnsupportedEncodingException {
        TTransport tr = new TSocket("localhost", 9160);
        try {            
            TProtocol proto = new TBinaryProtocol(tr);
            Cassandra.Client client = new Cassandra.Client(proto);
            tr.open();
            
            try{
                System.out.println("::ColumnOrSuperColumn get(keyspace, key, column_path, consistency_level)");
                //ColumnOrSuperColumn get(keyspace, key, column_path, consistency_level)
                //通过 column_path 参数获取 Column 或 SuperColumn 
                //如果查询结果为空将引发 NotFoundException异常. 
                //(这是唯一一个能够抛NotFoundException异常的方法)
                ColumnPath columnPath = new ColumnPath();
                columnPath.setColumn("name".getBytes());
                columnPath.setColumn_family("Standard1");
                //columnPath.setSuper_column("super_column".getBytes()); //如果是SuperColumn
                ColumnOrSuperColumn result_get = client.get("Keyspace1", "3", columnPath, ConsistencyLevel.ONE);
                System.out.println( new String(result_get.column.name, "UTF-8") + " -> " + new String(result_get.column.value, "UTF-8"));
            }catch(NotFoundException e){
                e.printStackTrace();        
            }
            
            System.out.println("::list<ColumnOrSuperColumn> get_slice(keyspace, key, column_parent, predicate, consistency_level)");
            //list<ColumnOrSuperColumn> get_slice(keyspace, key, column_parent, predicate, consistency_level)
            //根据 column_parent返回一组Column 在 predicate参数中可以设定Column的条件
            //(column_parent可以是一个ColumnFamily 名称或者是一个 ColumnFamily/SuperColumn) 
            
            ColumnParent column_parent = new ColumnParent();
            column_parent.setColumn_family("Standard1");
            //columnPath.setSuper_column("super_column".getBytes()); //如果是SuperColumn
            
            SlicePredicate predicate = new SlicePredicate();
            
//            List<byte[]> column_names = new ArrayList<byte[]>();
//            column_names.add("name".getBytes());
//            column_names.add("title".getBytes());    
//            column_names.add("not exists column".getBytes());    
//            predicate.setColumn_names(column_names); //使用指定的column列表
        
            SliceRange sliceRange = new SliceRange();
            sliceRange.setStart("z4".getBytes()); //是Column 的范围
            sliceRange.setFinish("a1".getBytes());
            sliceRange.setReversed(true); //必须与上面start和finish一致;如果为true finish小于start
            sliceRange.setCount(3); //Column数量
            predicate.setSlice_range(sliceRange); //不能和ColumnNames参数同时指定
            
            List<ColumnOrSuperColumn> result_get_slice = client.get_slice("Keyspace1", "3", column_parent, predicate, ConsistencyLevel.ONE);
            for(ColumnOrSuperColumn item:result_get_slice){
                //item.super_column.columns //如果是SuperColumn
                System.out.println( new String(item.column.name, "UTF-8") + " -> " + new String(item.column.value, "UTF-8"));                
            }
            
            System.out.println("::map<string,list<ColumnOrSuperColumn>> multiget_slice(keyspace, keys, column_parent, predicate, consistency_level)");
//            map<string,list<ColumnOrSuperColumn>> multiget_slice(keyspace, keys, column_parent, predicate, consistency_level) 
//                和上面函数get_slice基本相同，只是上面的函数get_slice查询一个key的结果，这个根据list<string>进行查询
            List<String> keys = new ArrayList<String>();
            keys.add("1");
            keys.add("3");
            keys.add("6");
            
            Map<String,List<ColumnOrSuperColumn>> multiget_slice_result = client.multiget_slice("Keyspace1",keys,column_parent,predicate,ConsistencyLevel.ONE);
            for(String key:multiget_slice_result.keySet()){ //遍历Key
                System.out.println("key -> " + key);
                for(ColumnOrSuperColumn item:multiget_slice_result.get(key)){           
                    System.out.println( new String(item.column.name, "UTF-8") + " -> " + new String(item.column.value, "UTF-8"));
                }
            }
            
            System.out.println("::i32 get_count(keyspace, key, column_parent, consistency_level)");
            //计算指定参数 column_parent中的 Column 个数
            //The method is not O(1). It takes all the columns from disk to calculate the answer. The only benefit of the method is that you do not need to pull all the columns over Thrift interface to count them.
            int get_count_result = client.get_count("Keyspace1", "3", column_parent, ConsistencyLevel.ONE);
            System.out.println("get_count_result -> " + get_count_result);
            
            System.out.println("::list<KeySlice> get_range_slices(keyspace, column_parent, predicate, range, consistency_level)");
            //list<KeySlice> get_range_slices(keyspace, column_parent, predicate, range, consistency_level) 
            //是上面所有查询函数综合：指定Column、Key的范围进行查询，其实和multiget_slice很像
            //Replaces get_range_slice. Returns a list of slices for the keys within the specified KeyRange. 
            //Unlike get_key_range, this applies the given predicate to all keys in the range, not just those with undeleted matching data. 
            //This method is only allowed when using an order-preserving partitioner.
            
            KeyRange keyRange = new KeyRange();
            keyRange.setStart_key("1000");
            keyRange.setEnd_key("5000");             
            keyRange.setCount(10);
            //并没有返回key在1000到5000的值 ？？？
            
            SlicePredicate predicate2 = new SlicePredicate();
            List<byte[]> column_names = new ArrayList<byte[]>();
            column_names.add("title".getBytes());
            predicate2.setColumn_names(column_names);
            
            List<KeySlice> get_range_slices_result = client.get_range_slices("Keyspace1", column_parent, predicate2, keyRange, ConsistencyLevel.ONE);
            
            for(KeySlice keySlice:get_range_slices_result){
                System.out.println("key -> " + keySlice.key);
                for(ColumnOrSuperColumn item:keySlice.columns){           
                    System.out.println( new String(item.column.name, "UTF-8") + " -> " + new String(item.column.value, "UTF-8"));
                }
            }            
            
            System.out.println("::insert(keyspace, key, column_path, value, timestamp, consistency_level)");
            //insert(keyspace, key, column_path, value, timestamp, consistency_level)
            //Insert a Column consisting of (column_path.column, value, timestamp) at the given column_path.
            //column_family and optional column_path.super_column. 
            //Note that column_path.column is here required, since a SuperColumn cannot directly contain binary values -- 
            //it can only contain sub-Columns. 
            long timestamp = System.currentTimeMillis();
            ColumnPath columnPath = new ColumnPath();
            columnPath.setColumn_family("Standard1");
            columnPath.setColumn("name".getBytes());            
            client.insert("Keyspace1", "999322", columnPath, "Rick".getBytes(), timestamp, ConsistencyLevel.ONE);
            System.out.println("Insert ok");
            //batch_mutate(keyspace, mutation_map, consistency_level) 
            //Executes the specified mutations on the keyspace. mutation_map is a map<string, map<string, list<Mutation>>>; 
            //the outer map maps the key to the inner map, which maps the column family to the Mutation; 
            //这个参数有点复杂，可以这样看： map<key : string, map<column_family : string, list<Mutation>>>. 
            //To be more specific, the outer map key is a row key, the inner map key is the column family name.
            //A Mutation specifies either columns to insert or columns to delete. See Mutation and Deletion above for more details. 
            Map<String, Map<String, List<Mutation>>> mutation_map = new HashMap<String, Map<String, List<Mutation>>>();
            Map<String, List<Mutation>> mutations = new HashMap<String, List<Mutation>>();
            List<Mutation> mutas = new ArrayList<Mutation>();
            
            Mutation mutation1 = new Mutation();            
            ColumnOrSuperColumn column_or_supercolumn = new ColumnOrSuperColumn();
            Column column = new Column();
            column.setName("name".getBytes());
            column.setValue("Richard".getBytes());
            column.setTimestamp(System.currentTimeMillis());
            column_or_supercolumn.setColumn(column);            
            mutation1.setColumn_or_supercolumn(column_or_supercolumn);
            
            Mutation mutation2 = new Mutation();            
            ColumnOrSuperColumn column_or_supercolumn2 = new ColumnOrSuperColumn();
            Column column2 = new Column();
            column2.setName("email".getBytes());
            column2.setValue("richard@gmail.com".getBytes());
            column2.setTimestamp(System.currentTimeMillis());
            column_or_supercolumn2.setColumn(column2);            
            mutation2.setColumn_or_supercolumn(column_or_supercolumn2);
            mutas.add(mutation2);
            mutas.add(mutation1);
            
            mutations.put("Standard1", mutas);    //Standard1 是column_family
            mutation_map.put("12315", mutations); //12315  是key 可以同时插入多个Key、在不同的columnFamily、多个Column
            
            client.batch_mutate("Keyspace1", mutation_map, ConsistencyLevel.ONE);
            System.out.println("batch_mutate insert done.");
            
            mutation_map.clear();
            mutations.clear();
            mutas.clear();
            Mutation mutation3 = new Mutation();
            
            Deletion deletion1 = new Deletion();
            deletion1.setTimestamp(System.currentTimeMillis());
            SlicePredicate predicate3 = new SlicePredicate();
            predicate3.column_names = new ArrayList<byte[]>();
            predicate3.column_names.add("email".getBytes());
            predicate3.column_names.add("name".getBytes());            
            deletion1.setPredicate(predicate3);            
            //deletion1.setSuper_column(super_column)
            mutation3.setDeletion(deletion1);
            mutas.add(mutation3);//重复
            
            mutations.put("Standard1", mutas);
            mutation_map.put("12315", mutations);
            
            client.batch_mutate("Keyspace1", mutation_map, ConsistencyLevel.ONE);
            System.out.println("batch_mutate deletion done.");
        } catch (TTransportException e) {
            e.printStackTrace();
        }finally{
            tr.close();
        }
    }

}