private static Logger LOG = LoggerFactory.getLogger(ThriftKeyspaceImplTest.class);

    private static Keyspace                  keyspace;

    private static AstyanaxContext<Keyspace> keyspaceContext;

    private static ColumnFamily<String, String> CF_USER_INFO = ColumnFamily.newColumnFamily(

            "UserInfo", // Column Family Name

            StringSerializer.get(), // Key Serializer

            StringSerializer.get()); // Column Serializer

    private static ColumnFamily<Long, Long> CF_DELETE = ColumnFamily

            .newColumnFamily(

                    "delete",

                    LongSerializer.get(),

                    LongSerializer.get());

   

    private static ColumnFamily<Long, String> CF_USERS = ColumnFamily

            .newColumnFamily(

                    "users",

                    LongSerializer.get(),

                    StringSerializer.get());

    private static ColumnFamily<String, String> CF_TTL = ColumnFamily

            .newColumnFamily(

                    "ttl",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_CQL = ColumnFamily

            .newColumnFamily(

                    "cfcql",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_CALLBACK = ColumnFamily

            .newColumnFamily(

                    "cfcallback",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_STANDARD1 = ColumnFamily

            .newColumnFamily(

                    "Standard1",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, Long> CF_LONGCOLUMN = ColumnFamily

            .newColumnFamily(

                    "LongColumn1",

                    StringSerializer.get(),

                    LongSerializer.get());

    public static ColumnFamily<String, String> CF_STANDARD2 = ColumnFamily

            .newColumnFamily(

                    "Standard2",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_ALLROWS = ColumnFamily

            .newColumnFamily(

                    "AllRows",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_COUNTER1 = ColumnFamily

            .newColumnFamily(

                    "Counter1",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_NOT_DEFINED = ColumnFamily

            .newColumnFamily(

                    "NotDefined",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<String, String> CF_EMPTY = ColumnFamily

            .newColumnFamily(

                    "NotDefined",

                    StringSerializer.get(),

                    StringSerializer.get());

    public static ColumnFamily<Long, Long> ATOMIC_UPDATES = ColumnFamily

            .newColumnFamily(

                    "AtomicUpdates",

                    LongSerializer.get(),

                    LongSerializer.get());

    public static AnnotatedCompositeSerializer<MockCompositeType> M_SERIALIZER = new AnnotatedCompositeSerializer<MockCompositeType>(

            MockCompositeType.class);

   

    public static ColumnFamily<String, MockCompositeType> CF_COMPOSITE = ColumnFamily

            .newColumnFamily(

                    "CompositeColumn",

                    StringSerializer.get(),

                    M_SERIALIZER);

    public static ColumnFamily<ByteBuffer, ByteBuffer> CF_COMPOSITE_CSV = ColumnFamily

            .newColumnFamily(

                    "CompositeCsv",

                    ByteBufferSerializer.get(),

                    ByteBufferSerializer.get());

    public static ColumnFamily<MockCompositeType, String> CF_COMPOSITE_KEY = ColumnFamily

            .newColumnFamily(

                    "CompositeKey",

                    M_SERIALIZER,

                    StringSerializer.get());

    public static ColumnFamily<String, UUID> CF_TIME_UUID = ColumnFamily

            .newColumnFamily(

                    "TimeUUID1",

                    StringSerializer.get(),

                    TimeUUIDSerializer.get());

    public static AnnotatedCompositeSerializer<SessionEvent> SE_SERIALIZER = new AnnotatedCompositeSerializer<SessionEvent>(

            SessionEvent.class);

    public static ColumnFamily<String, SessionEvent> CF_CLICK_STREAM = ColumnFamily

            .newColumnFamily("ClickStream", StringSerializer.get(),

                    SE_SERIALIZER);

    private static final String SEEDS = "localhost:9160";

    private static final long   CASSANDRA_WAIT_TIME = 3000;

    private static String TEST_CLUSTER_NAME  = "cass_sandbox";

    private static String TEST_KEYSPACE_NAME = "AstyanaxUnitTests";

   

    @BeforeClass

    public static void setup() throws Exception {

        System.out.println("TESTING THRIFT KEYSPACE");

        SingletonEmbeddedCassandra.getInstance();

       

        Thread.sleep(CASSANDRA_WAIT_TIME);

       

        createKeyspace();

    }

    @AfterClass

    public static void teardown() throws Exception {

        if (keyspaceContext != null)

            keyspaceContext.shutdown();

       

        Thread.sleep(CASSANDRA_WAIT_TIME);

    }

    public static void createKeyspace() throws Exception {

        keyspaceContext = new AstyanaxContext.Builder()

                .forCluster(TEST_CLUSTER_NAME)

                .forKeyspace(TEST_KEYSPACE_NAME)

                .withAstyanaxConfiguration(

                        new AstyanaxConfigurationImpl()

                                .setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE)

                                .setConnectionPoolType(ConnectionPoolType.ROUND_ROBIN)

                                .setDiscoveryDelayInSeconds(60000))

                .withConnectionPoolConfiguration(

                        new ConnectionPoolConfigurationImpl(TEST_CLUSTER_NAME

                                + "_" + TEST_KEYSPACE_NAME)

                                .setSocketTimeout(30000)

                                .setMaxTimeoutWhenExhausted(2000)

                                .setMaxConnsPerHost(20)

                                .setInitConnsPerHost(10)

                                .setSeeds(SEEDS)

                                )

                .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())

                .buildKeyspace(ThriftFamilyFactory.getInstance());

        keyspaceContext.start();

       

        keyspace = keyspaceContext.getClient();

       

        try {

            keyspace.dropKeyspace();

        }

        catch (Exception e) {

            LOG.info(e.getMessage());

        }

       

        ImmutableMap<String, Object> ksOptions = ImmutableMap.<String, Object>builder()

                .put("strategy_options", ImmutableMap.<String, Object>builder()

                        .put("replication_factor", "1")

                        .build())

                .put("strategy_class",     "SimpleStrategy")

                .build();

       

        ImmutableMap<String, Object> NO_OPTIONS = ImmutableMap.of();

       

        Map<ColumnFamily, Map<String, Object>> cfs = ImmutableMap.<ColumnFamily, Map<String, Object>>builder()

                .put(CF_STANDARD1,

                     ImmutableMap.<String, Object>builder()

                         .put("column_metadata", ImmutableMap.<String, Object>builder()

                             .put("Index1", ImmutableMap.<String, Object>builder()

                                 .put("validation_class", "UTF8Type")

                                 .put("index_type",       "KEYS")

                                 .build())

                             .put("Index2", ImmutableMap.<String, Object>builder()

                                 .put("validation_class", "UTF8Type")

                                 .put("index_type",       "KEYS")

                                 .build())

                             .build())

                         .build())

                .put(CF_TTL,        NO_OPTIONS)

                .build();

        keyspace.createKeyspace(ksOptions, cfs);

       

        keyspace.createColumnFamily(CF_STANDARD2,  null);

        keyspace.createColumnFamily(CF_ALLROWS,  null);

        keyspace.createColumnFamily(CF_LONGCOLUMN, null);

        keyspace.createColumnFamily(CF_DELETE,     null);

        keyspace.createColumnFamily(ATOMIC_UPDATES,null);

        keyspace.createColumnFamily(CF_CQL,  null);

        keyspace.createColumnFamily(CF_CALLBACK,  null);

       

        keyspace.createColumnFamily(CF_COUNTER1, ImmutableMap.<String, Object>builder()

                .put("default_validation_class", "CounterColumnType")

                .build());

        keyspace.createColumnFamily(CF_CLICK_STREAM, ImmutableMap.<String, Object>builder()

                .put("comparator_type", "CompositeType(UTF8Type, TimeUUIDType)")

                .build());

        keyspace.createColumnFamily(CF_COMPOSITE_CSV, ImmutableMap.<String, Object>builder()

                .put("default_validation_class", "UTF8Type")

                .put("key_validation_class",     "UTF8Type")

                .put("comparator_type",          "CompositeType(UTF8Type, LongType)")

                .build());

        keyspace.createColumnFamily(CF_COMPOSITE, ImmutableMap.<String, Object>builder()

                .put("comparator_type", "CompositeType(AsciiType, IntegerType(reversed=true), IntegerType, BytesType, UTF8Type)")

                .build());

        keyspace.createColumnFamily(CF_COMPOSITE_KEY, ImmutableMap.<String, Object>builder()

                .put("key_validation_class", "BytesType")

                .build());

        keyspace.createColumnFamily(CF_TIME_UUID,         null);

        keyspace.createColumnFamily(CF_USER_INFO,         null);

        keyspace.createColumnFamily(CF_USERS, ImmutableMap.<String, Object>builder()

                .put("default_validation_class", "UTF8Type")

                .put("column_metadata", ImmutableMap.<String, Object>builder()

                        .put("firstname",  ImmutableMap.<String, Object>builder()

                                .put("validation_class", "UTF8Type")

                                .put("index_type",       "KEYS")

                                .build())

                        .put("lastname", ImmutableMap.<String, Object>builder()

                                .put("validation_class", "UTF8Type")

                                .put("index_type",       "KEYS")

                                .build())

                        .put("age", ImmutableMap.<String, Object>builder()

                                .put("validation_class", "LongType")

                                .put("index_type",       "KEYS")

                                .build())

                        .build())

                     .build());

       

        KeyspaceDefinition ki = keyspaceContext.getClient().describeKeyspace();

        System.out.println("Describe Keyspace: " + ki.getName());

        try {

            //

            // CF_Super :

            // 'A' :

            // 'a' :

            // 1 : 'Aa1',

            // 2 : 'Aa2',

            // 'b' :

            // ...

            // 'z' :

            // ...

            // 'B' :

            // ...

            //

            // CF_Standard :

            // 'A' :

            // 'a' : 1,

            // 'b' : 2,

            // ...

            // 'z' : 26,

            // 'B' :

            // ...

            //

            MutationBatch m;

            OperationResult<Void> result;

            m = keyspace.prepareMutationBatch();

            for (char keyName = 'A'; keyName <= 'Z'; keyName++) {

                String rowKey = Character.toString(keyName);

                ColumnListMutation<String> cfmStandard = m.withRow(

                        CF_STANDARD1, rowKey);

                for (char cName = 'a'; cName <= 'z'; cName++) {

                    cfmStandard.putColumn(Character.toString(cName),

                            (int) (cName - 'a') + 1, null);

                }

                cfmStandard

                        .putColumn("Index1", (int) (keyName - 'A') + 1, null);

                cfmStandard.putColumn("Index2", 42, null);

                m.execute();

            }

            m.withRow(CF_STANDARD1, "Prefixes").putColumn("Prefix1_a", 1, null)

                    .putColumn("Prefix1_b", 2, null)

                    .putColumn("prefix2_a", 3, null);

            result = m.execute();

            String rowKey = "A";

            ColumnListMutation<Long> cfmLong = m.withRow(CF_LONGCOLUMN, rowKey);

            for (Long l = -10L; l < 10L; l++) {

                cfmLong.putEmptyColumn(l, null);

            }

            cfmLong.putEmptyColumn(Long.MAX_VALUE, null);

            result = m.execute();

            m.withRow(CF_USER_INFO, "acct1234")

                .putColumn("firstname", "john", null)

                .putColumn("lastname", "smith", null)

                .putColumn("address", "555 Elm St", null)

                .putColumn("age", 30, null)

                .putEmptyColumn("empty");

            m.execute();

           

           

            // Inserts for CF_ALLROWS

            m = keyspace.prepareMutationBatch();

            for (char keyName = 'A'; keyName <= 'Z'; keyName++) {

                rowKey = Character.toString(keyName);

                ColumnListMutation<String> cfmStandard = m.withRow(

                        CF_ALLROWS, rowKey);

                for (char cName = 'a'; cName <= 'z'; cName++) {

                    cfmStandard.putColumn(Character.toString(cName),

                            (int) (cName - 'a') + 1, null);

                }

                m.execute();

            }

           

        } catch (Exception e) {

            System.out.println(e.getMessage());

            Assert.fail();

        }

    }

    @Test

    public void testMultiColumnDelete() throws Exception {

        MutationBatch mb = keyspace.prepareMutationBatch();

        mb.withRow(CF_DELETE, 1L)

            .setTimestamp(1).putEmptyColumn(1L, null)

            .setTimestamp(10).putEmptyColumn(2L, null)

            ;

        mb.execute();

       

        ColumnList<Long> result1 = keyspace.prepareQuery(CF_DELETE).getRow(1L).execute().getResult();

        Assert.assertEquals(2, result1.size());

        Assert.assertNotNull(result1.getColumnByName(1L));

        Assert.assertNotNull(result1.getColumnByName(2L));

       

        logColumnList("Insert", result1);

       

        mb = keyspace.prepareMutationBatch();

        mb.withRow(CF_DELETE,  1L)

            .setTimestamp(result1.getColumnByName(1L).getTimestamp()-1)

            .deleteColumn(1L)

            .setTimestamp(result1.getColumnByName(2L).getTimestamp()-1)

            .deleteColumn(2L)

            .putEmptyColumn(3L, null);

       

        mb.execute();

       

        result1 = keyspace.prepareQuery(CF_DELETE).getRow(1L).execute().getResult();

        logColumnList("Delete with older timestamp", result1);

        Assert.assertEquals(3, result1.size());

       

        LOG.info("Delete L2 with TS: " + (result1.getColumnByName(2L).getTimestamp()+1));

        mb.withRow(CF_DELETE,  1L)

            .setTimestamp(result1.getColumnByName(1L).getTimestamp()+1)

            .deleteColumn(1L)

            .setTimestamp(result1.getColumnByName(2L).getTimestamp()+1)

            .deleteColumn(2L);

        mb.execute();

       

        result1 = keyspace.prepareQuery(CF_DELETE).getRow(1L).execute().getResult();

        logColumnList("Delete with newer timestamp", result1);

        Assert.assertEquals(1, result1.size());

    }

   

    <T> void logColumnList(String label, ColumnList<T> cl) {

        LOG.info(">>>>>> " + label);

        for (Column<T> c : cl) {

            LOG.info(c.getName() + " " + c.getTimestamp());

        }

        LOG.info("<<<<<<");

    }

   

    @Test

    public void testCqlComposite() throws Exception {

        CqlStatementResult result = keyspace.prepareCqlStatement()

            .withCql("SELECT * FROM " + CF_COMPOSITE_CSV.getName())

            .execute()

            .getResult();

       

        result.getSchema();

        result.getRows(CF_COMPOSITE_CSV);

    }

   

    @Test

    public void testHasValue() throws Exception {

        ColumnList<String> response = keyspace.prepareQuery(CF_USER_INFO).getRow("acct1234").execute().getResult();

        Assert.assertEquals("firstname", response.getColumnByName("firstname").getName());

        Assert.assertEquals("firstname", response.getColumnByName("firstname").getName());

        Assert.assertEquals("john", response.getColumnByName("firstname").getStringValue());

        Assert.assertEquals("john", response.getColumnByName("firstname").getStringValue());

        Assert.assertEquals(true,  response.getColumnByName("firstname").hasValue());

        Assert.assertEquals(false, response.getColumnByName("empty").hasValue());

       

    }

   

    @Test

    public void getKeyspaceDefinition() throws Exception {

        KeyspaceDefinition def = keyspaceContext.getEntity().describeKeyspace();

        Collection<String> fieldNames = def.getFieldNames();

        LOG.info("Getting field names");

        for (String field : fieldNames) {

            LOG.info(field);

        }

        LOG.info(fieldNames.toString());

       

        for (FieldMetadata field : def.getFieldsMetadata()) {

            LOG.info(field.getName() + " = " + def.getFieldValue(field.getName()) + " (" + field.getType() + ")");

        }

       

        for (ColumnFamilyDefinition cfDef : def.getColumnFamilyList()) {