How to insert in Cassandra into Composite Columns using Astyanax client?

Question

I am trying to insert into Composite Columns in Cassandra using Astyanax client. Below is the column family I have created in which all the columns are Composite Columns. I am not sure how should I insert into below Column Family with each columns is a Composite Column.

create column family USER_DATA
with key_validation_class = 'UTF8Type'
and comparator = 'CompositeType(UTF8Type,UTF8Type,DateType)'
and default_validation_class = 'UTF8Type'
and gc_grace = 86400
and column_metadata = [ {column_name : 'lmd', validation_class : DateType}];

Below is a simple example which I have created that will insert into Cassandra but its not for Composite Columns. attributesMap is the map which will have the data that I am trying to insert, key will be column name and the actual value will be composite value of that column.

String s1 = "Hello";
String S2 = "World";
long s3 = System.currentTimeMillis();

// insert the above three values into a composite column

clientDao.upsertAttributes("123", attributesMap, columnFamily);

And Below is my DAOImpl code which has upsertAttributes method which I was using to insert into Cassandra only for single column value. Is there any way, I can modify the same to start using Composite Column behavior using Astyanax?

/**
 * Performs an upsert of the specified attributes for the specified id.
 */
public void upsertAttributes(final String rowKey, final Map<String, String> attributes, final String columnFamily) {

    try {
        MutationBatch m = CassandraAstyanaxConnection.getInstance().getKeyspace().prepareMutationBatch();

        ColumnListMutation<String> mutation = m.withRow(CassandraAstyanaxConnection.getInstance().getEmp_cf(), rowKey);
        for (Map.Entry<String, String> entry : attributes.entrySet()) {
            mutation = mutation.putColumn(entry.getKey(), entry.getValue(), null);
        }

        mutation.putColumn("lmd", System.currentTimeMillis());

        m.setConsistencyLevel(ConsistencyLevel.CL_ONE).execute();

    } catch (ConnectionException e) {
            // log here
    } catch (Exception e) {
            // log here
    }
}

Below is the class where I am making Cassandra Connection using Astyanax client-

public class CassandraAstyanaxConnection {

    private AstyanaxContext<Keyspace> context;
    private Keyspace keyspace;
    private ColumnFamily<String, String> emp_cf;


    private static class ConnectionHolder {
        static final CassandraAstyanaxConnection connection = new CassandraAstyanaxConnection();
    }

    public static CassandraAstyanaxConnection getInstance() {
        return ConnectionHolder.connection;
    }

    /**
     * Creating Cassandra connection using Astyanax client
     *
     */
    private CassandraAstyanaxConnection() {

        context = new AstyanaxContext.Builder()
        .forCluster(Constants.CLUSTER)
        .forKeyspace(Constants.KEYSPACE)
        .withConnectionPoolConfiguration(new ConnectionPoolConfigurationImpl("MyConnectionPool")
            .setPort(9160)
            .setMaxConnsPerHost(1000)
            .setSeeds("host1:portnumber")
        )
        .withAstyanaxConfiguration(new AstyanaxConfigurationImpl()
            .setCqlVersion("3.0.0")
            .setTargetCassandraVersion("1.2")
            .setConnectionPoolType(ConnectionPoolType.ROUND_ROBIN)
            .setDiscoveryType(NodeDiscoveryType.RING_DESCRIBE))
        .withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
        .buildKeyspace(ThriftFamilyFactory.getInstance());

        context.start();
        keyspace = context.getEntity();

        emp_cf = ColumnFamily.newColumnFamily(Constants.COLUMN_FAMILY, StringSerializer.get(), StringSerializer.get());
    }

    /**
     * returns the keyspace
     * 
     * @return
     */
    public Keyspace getKeyspace() {
        return keyspace;
    }

    public ColumnFamily<String, String> getEmp_cf() {
        return emp_cf;
    }
}

Any simple example to insert into Composite Columns using Astyanax will help me understand better. I am not able to find any example related to that. Can anybody help me with this?

Answer 1

We have only done this with Composite keys, but I believe the principal remains the same. The ColumnListMutation class has a putColumn method which can take an arbitrary type and it's serializer. This requires a fixed type (not an arbitrary Map) although you could create a serializer that took a Map (there is a MapSerializer but I have not used it). We use the AnnotatedCompositeSerializer instead of writing a Serializer from scratch. So the code would be something like this (I have not tested this, but it should be enough to get you started):

public static class ComplexType {
  @Component(ordinal = 0)
  String val1;
  @Component(ordinal = 1)
  String val2;
  @Component(ordinal = 2)
  int timestamp;

  // get/set methods left out...
}

private static final AnnotateCompositeSerializer<ComplexType> complexTypeSerializer = new AnnotatedCompositeSerializer<>(ComplexType.class);

public void upsertAttributes(final String rowKey, final Map<String, String> attributes,  final String columnFamily) {

  try {
      MutationBatch m = CassandraAstyanaxConnection.getInstance().getKeyspace().prepareMutationBatch();

      ComplexType ct = createComplexTypeFromAttributes(attributes);

      ColumnListMutation<String> mutation = m.withRow(CassandraAstyanaxConnection.getInstance().getEmp_cf(), rowKey);
    mutation
          .putColumn("lmd", ct, complexTypeSerializer, null);

    m.setConsistencyLevel(ConsistencyLevel.CL_ONE).execute();

} catch (ConnectionException e) {
        // log here
} catch (Exception e) {
        // log here
}
}