diff --git a/src/main/java/org/springframework/data/cassandra/core/CassandraKeyspaceFactoryBean.java b/src/main/java/org/springframework/data/cassandra/core/CassandraKeyspaceFactoryBean.java index 11b63a8c1..7ffab1e38 100644 --- a/src/main/java/org/springframework/data/cassandra/core/CassandraKeyspaceFactoryBean.java +++ b/src/main/java/org/springframework/data/cassandra/core/CassandraKeyspaceFactoryBean.java @@ -34,7 +34,7 @@ import org.springframework.data.cassandra.convert.MappingCassandraConverter; import org.springframework.data.cassandra.mapping.CassandraMappingContext; import org.springframework.data.cassandra.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.mapping.CassandraPersistentProperty; -import org.springframework.data.cassandra.util.CQLUtils; +import org.springframework.data.cassandra.util.CqlUtils; import org.springframework.data.mapping.context.MappingContext; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -47,45 +47,43 @@ import com.datastax.driver.core.TableMetadata; import com.datastax.driver.core.exceptions.NoHostAvailableException; /** - * Convenient factory for configuring a Cassandra Session. - * Session is a thread safe singleton and created per a keyspace. - * So, it is enough to have one session per application. + * Convenient factory for configuring a Cassandra Session. Session is a thread safe singleton and created per a + * keyspace. So, it is enough to have one session per application. * * @author Alex Shvid */ -public class CassandraKeyspaceFactoryBean implements FactoryBean, -InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTranslator { +public class CassandraKeyspaceFactoryBean implements FactoryBean, InitializingBean, DisposableBean, + BeanClassLoaderAware, PersistenceExceptionTranslator { private static final Logger log = LoggerFactory.getLogger(CassandraKeyspaceFactoryBean.class); - + public static final String DEFAULT_REPLICATION_STRATEGY = "SimpleStrategy"; public static final int DEFAULT_REPLICATION_FACTOR = 1; - + private ClassLoader beanClassLoader; - + private Cluster cluster; private Session session; private String keyspace; - + private CassandraConverter converter; private MappingContext, CassandraPersistentProperty> mappingContext; - + private Keyspace keyspaceBean; - + private KeyspaceAttributes keyspaceAttributes; - + private final PersistenceExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator(); - - public void setBeanClassLoader(ClassLoader classLoader) { - this.beanClassLoader = classLoader; + + public void setBeanClassLoader(ClassLoader classLoader) { + this.beanClassLoader = classLoader; } - public Keyspace getObject() throws Exception { return keyspaceBean; } - + /* * (non-Javadoc) * @see org.springframework.beans.factory.FactoryBean#getObjectType() @@ -101,7 +99,7 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran public boolean isSingleton() { return true; } - + /* * (non-Javadoc) * @see org.springframework.dao.support.PersistenceExceptionTranslator#translateExceptionIfPossible(java.lang.RuntimeException) @@ -109,165 +107,162 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran public DataAccessException translateExceptionIfPossible(RuntimeException ex) { return exceptionTranslator.translateExceptionIfPossible(ex); } - + /* * (non-Javadoc) * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { - + if (this.converter == null) { this.converter = getDefaultCassandraConverter(); } this.mappingContext = this.converter.getMappingContext(); - if (cluster == null) { - throw new IllegalArgumentException( - "at least one cluster is required"); + throw new IllegalArgumentException("at least one cluster is required"); } Session session = null; session = cluster.connect(); - + if (StringUtils.hasText(keyspace)) { - + KeyspaceMetadata keyspaceMetadata = cluster.getMetadata().getKeyspace(keyspace.toLowerCase()); boolean keyspaceExists = keyspaceMetadata != null; boolean keyspaceCreated = false; - + if (keyspaceExists) { log.info("keyspace exists " + keyspaceMetadata.asCQLQuery()); } - + if (keyspaceAttributes == null) { keyspaceAttributes = new KeyspaceAttributes(); } - + // drop the old keyspace if needed if (keyspaceExists && (keyspaceAttributes.isCreate() || keyspaceAttributes.isCreateDrop())) { log.info("Drop keyspace " + keyspace + " on afterPropertiesSet"); session.execute("DROP KEYSPACE " + keyspace); keyspaceExists = false; - } - - // create the new keyspace if needed - if (!keyspaceExists && (keyspaceAttributes.isCreate() || keyspaceAttributes.isCreateDrop() || keyspaceAttributes.isUpdate())) { + } + + // create the new keyspace if needed + if (!keyspaceExists + && (keyspaceAttributes.isCreate() || keyspaceAttributes.isCreateDrop() || keyspaceAttributes.isUpdate())) { + + String query = String + .format( + "CREATE KEYSPACE %1$s WITH replication = { 'class' : '%2$s', 'replication_factor' : %3$d } AND DURABLE_WRITES = %4$b", + keyspace, keyspaceAttributes.getReplicationStrategy(), keyspaceAttributes.getReplicationFactor(), + keyspaceAttributes.isDurableWrites()); - String query = String.format("CREATE KEYSPACE %1$s WITH replication = { 'class' : '%2$s', 'replication_factor' : %3$d } AND DURABLE_WRITES = %4$b", - keyspace, - keyspaceAttributes.getReplicationStrategy(), - keyspaceAttributes.getReplicationFactor(), - keyspaceAttributes.isDurableWrites()); - log.info("Create keyspace " + keyspace + " on afterPropertiesSet " + query); - + session.execute(query); keyspaceCreated = true; } - + // update keyspace if needed if (keyspaceAttributes.isUpdate() && !keyspaceCreated) { - + if (compareKeyspaceAttributes(keyspaceAttributes, keyspaceMetadata) != null) { - - String query = String.format("ALTER KEYSPACE %1$s WITH replication = { 'class' : '%2$s', 'replication_factor' : %3$d } AND DURABLE_WRITES = %4$b", - keyspace, - keyspaceAttributes.getReplicationStrategy(), - keyspaceAttributes.getReplicationFactor(), - keyspaceAttributes.isDurableWrites()); - + + String query = String + .format( + "ALTER KEYSPACE %1$s WITH replication = { 'class' : '%2$s', 'replication_factor' : %3$d } AND DURABLE_WRITES = %4$b", + keyspace, keyspaceAttributes.getReplicationStrategy(), keyspaceAttributes.getReplicationFactor(), + keyspaceAttributes.isDurableWrites()); + log.info("Update keyspace " + keyspace + " on afterPropertiesSet " + query); session.execute(query); } - + } - + // validate keyspace if needed if (keyspaceAttributes.isValidate()) { - + if (!keyspaceExists) { throw new InvalidDataAccessApiUsageException("keyspace '" + keyspace + "' not found in the Cassandra"); } - + String errorField = compareKeyspaceAttributes(keyspaceAttributes, keyspaceMetadata); if (errorField != null) { - throw new InvalidDataAccessApiUsageException(errorField + " attribute is not much in the keyspace '" + keyspace + "'"); + throw new InvalidDataAccessApiUsageException(errorField + " attribute is not much in the keyspace '" + + keyspace + "'"); } - + } - + session.execute("USE " + keyspace); - + if (!CollectionUtils.isEmpty(keyspaceAttributes.getTables())) { - + for (TableAttributes tableAttributes : keyspaceAttributes.getTables()) { - + String entityClassName = tableAttributes.getEntity(); Class entityClass = ClassUtils.forName(entityClassName, this.beanClassLoader); CassandraPersistentEntity entity = determineEntity(entityClass); String useTableName = tableAttributes.getName() != null ? tableAttributes.getName() : entity.getTable(); - + if (keyspaceCreated) { createNewTable(session, useTableName, entity); - } - else if (keyspaceAttributes.isUpdate()) { + } else if (keyspaceAttributes.isUpdate()) { TableMetadata table = keyspaceMetadata.getTable(useTableName.toLowerCase()); if (table == null) { createNewTable(session, useTableName, entity); - } - else { + } else { // alter table columns - for (String cql : CQLUtils.alterTable(useTableName, entity, table)) { + for (String cql : CqlUtils.alterTable(useTableName, entity, table)) { log.info("Execute on keyspace " + keyspace + " CQL " + cql); session.execute(cql); } } - } - else if (keyspaceAttributes.isValidate()) { + } else if (keyspaceAttributes.isValidate()) { TableMetadata table = keyspaceMetadata.getTable(useTableName.toLowerCase()); if (table == null) { - throw new InvalidDataAccessApiUsageException("not found table " + useTableName + " for entity " + entityClassName); + throw new InvalidDataAccessApiUsageException("not found table " + useTableName + " for entity " + + entityClassName); } // validate columns - List alter = CQLUtils.alterTable(useTableName, entity, table); + List alter = CqlUtils.alterTable(useTableName, entity, table); if (!alter.isEmpty()) { - throw new InvalidDataAccessApiUsageException("invalid table " + useTableName + " for entity " + entityClassName + ". modify it by " + alter); + throw new InvalidDataAccessApiUsageException("invalid table " + useTableName + " for entity " + + entityClassName + ". modify it by " + alter); } } - - //System.out.println("tableAttributes, entityClass=" + entityClass + ", table = " + entity.getTable()); - + + // System.out.println("tableAttributes, entityClass=" + entityClass + ", table = " + entity.getTable()); + } } - - } - + + } + // initialize property this.session = session; - + this.keyspaceBean = new Keyspace(keyspace, session, converter); } - - private void createNewTable(Session session, String useTableName, - CassandraPersistentEntity entity) + private void createNewTable(Session session, String useTableName, CassandraPersistentEntity entity) throws NoHostAvailableException { - String cql = CQLUtils.createTable(useTableName, entity); + String cql = CqlUtils.createTable(useTableName, entity); log.info("Execute on keyspace " + keyspace + " CQL " + cql); session.execute(cql); - for (String indexCQL : CQLUtils.createIndexes(useTableName, entity)) { + for (String indexCQL : CqlUtils.createIndexes(useTableName, entity)) { log.info("Execute on keyspace " + keyspace + " CQL " + indexCQL); session.execute(indexCQL); } } - + /* * (non-Javadoc) * @see org.springframework.beans.factory.DisposableBean#destroy() */ public void destroy() throws Exception { - + if (StringUtils.hasText(keyspace) && keyspaceAttributes != null && keyspaceAttributes.isCreateDrop()) { log.info("Drop keyspace " + keyspace + " on destroy"); session.execute("USE system"); @@ -287,12 +282,13 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran public void setKeyspaceAttributes(KeyspaceAttributes keyspaceAttributes) { this.keyspaceAttributes = keyspaceAttributes; } - + public void setConverter(CassandraConverter converter) { this.converter = converter; } - private static String compareKeyspaceAttributes(KeyspaceAttributes keyspaceAttributes, KeyspaceMetadata keyspaceMetadata) { + private static String compareKeyspaceAttributes(KeyspaceAttributes keyspaceAttributes, + KeyspaceMetadata keyspaceMetadata) { if (keyspaceAttributes.isDurableWrites() != keyspaceMetadata.isDurableWrites()) { return "durableWrites"; } @@ -306,11 +302,10 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran if (keyspaceAttributes.getReplicationFactor() != replicationFactor) { return "replication_factor"; } - } - catch(NumberFormatException e) { + } catch (NumberFormatException e) { return "replication_factor"; } - + String attributesStrategy = keyspaceAttributes.getReplicationStrategy(); if (attributesStrategy.indexOf('.') == -1) { attributesStrategy = "org.apache.cassandra.locator." + attributesStrategy; @@ -321,7 +316,7 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran } return null; } - + CassandraPersistentEntity determineEntity(Class entityClass) { if (entityClass == null) { @@ -336,7 +331,7 @@ InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTran } return entity; } - + private static final CassandraConverter getDefaultCassandraConverter() { MappingCassandraConverter converter = new MappingCassandraConverter(new CassandraMappingContext()); converter.afterPropertiesSet(); diff --git a/src/main/java/org/springframework/data/cassandra/util/CqlUtils.java b/src/main/java/org/springframework/data/cassandra/util/CqlUtils.java new file mode 100644 index 000000000..65fc10a1f --- /dev/null +++ b/src/main/java/org/springframework/data/cassandra/util/CqlUtils.java @@ -0,0 +1,450 @@ +package org.springframework.data.cassandra.util; + +import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.cassandra.core.ConsistencyLevel; +import org.springframework.data.cassandra.core.ConsistencyLevelResolver; +import org.springframework.data.cassandra.core.QueryOptions; +import org.springframework.data.cassandra.core.RetryPolicy; +import org.springframework.data.cassandra.core.RetryPolicyResolver; +import org.springframework.data.cassandra.exception.EntityWriterException; +import org.springframework.data.cassandra.mapping.CassandraPersistentEntity; +import org.springframework.data.cassandra.mapping.CassandraPersistentProperty; +import org.springframework.data.mapping.PropertyHandler; + +import com.datastax.driver.core.ColumnMetadata; +import com.datastax.driver.core.DataType; +import com.datastax.driver.core.Query; +import com.datastax.driver.core.TableMetadata; +import com.datastax.driver.core.querybuilder.Batch; +import com.datastax.driver.core.querybuilder.Delete; +import com.datastax.driver.core.querybuilder.Delete.Where; +import com.datastax.driver.core.querybuilder.Insert; +import com.datastax.driver.core.querybuilder.QueryBuilder; + +/** + * + * Utilties to convert Cassandra Annotated objects to Queries and CQL. + * + * @author Alex Shvid + * @author David Webb + * + */ +public abstract class CqlUtils { + + private static Logger log = LoggerFactory.getLogger(CqlUtils.class); + + /** + * Generates the CQL String to create a table in Cassandra + * + * @param tableName + * @param entity + * @return The CQL that can be passed to session.execute() + */ + public static String createTable(String tableName, final CassandraPersistentEntity entity) { + + final StringBuilder str = new StringBuilder(); + str.append("CREATE TABLE "); + str.append(tableName); + str.append('('); + + final List ids = new ArrayList(); + final List idColumns = new ArrayList(); + + entity.doWithProperties(new PropertyHandler() { + public void doWithPersistentProperty(CassandraPersistentProperty prop) { + + if (str.charAt(str.length() - 1) != '(') { + str.append(','); + } + + String columnName = prop.getColumnName(); + + str.append(columnName); + str.append(' '); + + DataType dataType = prop.getDataType(); + + str.append(toCQL(dataType)); + + if (prop.isIdProperty()) { + ids.add(prop.getColumnName()); + } + + if (prop.isColumnId()) { + idColumns.add(prop.getColumnName()); + } + + } + + }); + + if (ids.isEmpty()) { + throw new InvalidDataAccessApiUsageException("not found primary ID in the entity " + entity.getType()); + } + + str.append(",PRIMARY KEY("); + + // if (ids.size() > 1) { + // str.append('('); + // } + + for (String id : ids) { + if (str.charAt(str.length() - 1) != '(') { + str.append(','); + } + str.append(id); + } + + // if (ids.size() > 1) { + // str.append(')'); + // } + + for (String id : idColumns) { + str.append(','); + str.append(id); + } + + str.append("));"); + + return str.toString(); + } + + /** + * Create the List of CQL for the indexes required for Cassandra mapped Table. + * + * @param tableName + * @param entity + * @return The list of CQL statements to run with session.execute() + */ + public static List createIndexes(final String tableName, final CassandraPersistentEntity entity) { + final List result = new ArrayList(); + + entity.doWithProperties(new PropertyHandler() { + public void doWithPersistentProperty(CassandraPersistentProperty prop) { + + if (prop.isIndexed()) { + + final StringBuilder str = new StringBuilder(); + str.append("CREATE INDEX ON "); + str.append(tableName); + str.append(" ("); + str.append(prop.getColumnName()); + str.append(");"); + + result.add(str.toString()); + } + + } + }); + + return result; + } + + /** + * Alter the table to refelct the entity annotations + * + * @param tableName + * @param entity + * @param table + * @return + */ + public static List alterTable(final String tableName, final CassandraPersistentEntity entity, + final TableMetadata table) { + final List result = new ArrayList(); + + entity.doWithProperties(new PropertyHandler() { + public void doWithPersistentProperty(CassandraPersistentProperty prop) { + + String columnName = prop.getColumnName(); + DataType columnDataType = prop.getDataType(); + ColumnMetadata columnMetadata = table.getColumn(columnName.toLowerCase()); + + if (columnMetadata != null && columnDataType.equals(columnMetadata.getType())) { + return; + } + + final StringBuilder str = new StringBuilder(); + str.append("ALTER TABLE "); + str.append(tableName); + if (columnMetadata == null) { + str.append(" ADD "); + } else { + str.append(" ALTER "); + } + + str.append(columnName); + str.append(' '); + + if (columnMetadata != null) { + str.append("TYPE "); + } + + str.append(toCQL(columnDataType)); + + str.append(';'); + result.add(str.toString()); + + } + }); + + return result; + } + + /** + * Generates a Query Object for an insert + * + * @param keyspaceName + * @param tableName + * @param entity + * @param objectToSave + * @param optionsByName + * @param mappingContext + * @param beanClassLoader + * + * @return The Query object to run with session.execute(); + * @throws EntityWriterException + */ + public static Query toInsertQuery(String keyspaceName, String tableName, final Object objectToSave, + CassandraPersistentEntity entity, Map optionsByName) throws EntityWriterException { + + final Insert q = QueryBuilder.insertInto(keyspaceName, tableName); + final Exception innerException = new Exception(); + + entity.doWithProperties(new PropertyHandler() { + public void doWithPersistentProperty(CassandraPersistentProperty prop) { + + /* + * See if the object has a value for that column, and if so, add it to the Query + */ + try { + + Object o = prop.getGetter().invoke(objectToSave, new Object[0]); + + log.info("Getter Invoke [" + prop.getColumnName() + " => " + o); + + if (o != null) { + q.value(prop.getColumnName(), o); + } + + } catch (IllegalAccessException e) { + innerException.initCause(e); + } catch (IllegalArgumentException e) { + innerException.initCause(e); + } catch (InvocationTargetException e) { + innerException.initCause(e); + } + } + }); + + if (innerException.getCause() != null) { + throw new EntityWriterException("Failed to convert Persistent Entity to CQL/Query", innerException.getCause()); + } + + /* + * Add Query Options + */ + addQueryOptions(q, optionsByName); + + /* + * Add TTL to Insert object + */ + if (optionsByName.get(QueryOptions.QueryOptionMapKeys.TTL) != null) { + q.using(QueryBuilder.ttl((Integer) optionsByName.get(QueryOptions.QueryOptionMapKeys.TTL))); + } + + return q; + + } + + /** + * Generates a Batch Object for multiple inserts + * + * @param keyspaceName + * @param tableName + * @param entity + * @param objectsToSave + * @param mappingContext + * @param beanClassLoader + * + * @return The Query object to run with session.execute(); + * @throws EntityWriterException + */ + public static Batch toInsertBatchQuery(final String keyspaceName, final String tableName, + final List objectsToSave, CassandraPersistentEntity entity, Map optionsByName) + throws EntityWriterException { + + /* + * Return variable is a Batch statement + */ + final Batch b = QueryBuilder.batch(); + + List queries = new ArrayList(); + + for (final T objectToSave : objectsToSave) { + + queries.add(toInsertQuery(keyspaceName, tableName, objectToSave, entity, optionsByName)); + + } + + addQueryOptions(b, optionsByName); + + return b; + + } + + /** + * @param keyspace + * @param tableName + * @param objectToRemove + * @param entity + * @return + * @throws EntityWriterException + */ + public static Query toDeleteQuery(String keyspace, String tableName, final Object objectToRemove, + CassandraPersistentEntity entity, Map optionsByName) throws EntityWriterException { + + final Delete.Selection ds = QueryBuilder.delete(); + final Delete q = ds.from(keyspace, tableName); + final Where w = q.where(); + + final Exception innerException = new Exception(); + + entity.doWithProperties(new PropertyHandler() { + public void doWithPersistentProperty(CassandraPersistentProperty prop) { + + /* + * See if the object has a value for that column, and if so, add it to the Query + */ + try { + + if (prop.isIdProperty()) { + Object o = (String) prop.getGetter().invoke(objectToRemove, new Object[0]); + + log.info("Getter Invoke [" + prop.getColumnName() + " => " + o); + + if (o != null) { + w.and(QueryBuilder.eq(prop.getColumnName(), o)); + } + } + + } catch (IllegalAccessException e) { + innerException.initCause(e); + } catch (IllegalArgumentException e) { + innerException.initCause(e); + } catch (InvocationTargetException e) { + innerException.initCause(e); + } + } + }); + + if (innerException.getCause() != null) { + throw new EntityWriterException("Failed to convert Persistent Entity to CQL/Query", innerException.getCause()); + } + + addQueryOptions(q, optionsByName); + + return q; + + } + + /** + * @param dataType + * @return + */ + public static String toCQL(DataType dataType) { + if (dataType.getTypeArguments().isEmpty()) { + return dataType.getName().name(); + } else { + StringBuilder str = new StringBuilder(); + str.append(dataType.getName().name()); + str.append('<'); + for (DataType argDataType : dataType.getTypeArguments()) { + if (str.charAt(str.length() - 1) != '<') { + str.append(','); + } + str.append(argDataType.getName().name()); + } + str.append('>'); + return str.toString(); + } + } + + /** + * @param tableName + * @return + */ + public static String dropTable(String tableName) { + + if (tableName == null) { + return null; + } + + StringBuilder str = new StringBuilder(); + str.append("DROP TABLE " + tableName + ";"); + return str.toString(); + } + + /** + * @param keyspace + * @param tableName + * @param entities + * @param cPEntity + * @return + * @throws EntityWriterException + */ + public static Batch toDeleteBatchQuery(String keyspaceName, String tableName, List entities, + CassandraPersistentEntity entity, Map optionsByName) throws EntityWriterException { + + /* + * Return variable is a Batch statement + */ + final Batch b = QueryBuilder.batch(); + + List queries = new ArrayList(); + + for (final T objectToSave : entities) { + + queries.add(toDeleteQuery(keyspaceName, tableName, objectToSave, entity, optionsByName)); + + } + + addQueryOptions(b, optionsByName); + + return b; + + } + + /** + * Add common Query options for all types of queries. + * + * @param q + * @param optionsByName + */ + private static void addQueryOptions(Query q, Map optionsByName) { + + if (optionsByName == null) { + return; + } + + /* + * Add Query Options + */ + if (optionsByName.get(QueryOptions.QueryOptionMapKeys.CONSISTENCY_LEVEL) != null) { + q.setConsistencyLevel(ConsistencyLevelResolver.resolve((ConsistencyLevel) optionsByName + .get(QueryOptions.QueryOptionMapKeys.CONSISTENCY_LEVEL))); + } + if (optionsByName.get(QueryOptions.QueryOptionMapKeys.RETRY_POLICY) != null) { + q.setRetryPolicy(RetryPolicyResolver.resolve((RetryPolicy) optionsByName + .get(QueryOptions.QueryOptionMapKeys.RETRY_POLICY))); + } + + } + +}