Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

File rename part 2

Browse Source
This commit is contained in:
David Webb
2013-11-14 15:11:20 +00:00
parent f8d144de2e
commit 84c10608fe
2 changed files with 536 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
src/main/java/org/springframework/data/cassandra/core/CassandraKeyspaceFactoryBean.java
View File

@@ -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<Keyspace>,
InitializingBean, DisposableBean, BeanClassLoaderAware, PersistenceExceptionTranslator {
public class CassandraKeyspaceFactoryBean implements FactoryBean<Keyspace>, 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<? extends CassandraPersistentEntity<?>, 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<String> alter = CQLUtils.alterTable(useTableName, entity, table);
List<String> 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();

450
src/main/java/org/springframework/data/cassandra/util/CqlUtils.java Normal file
View File

@@ -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<String> ids = new ArrayList<String>();
final List<String> idColumns = new ArrayList<String>();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
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<String> createIndexes(final String tableName, final CassandraPersistentEntity<?> entity) {
final List<String> result = new ArrayList<String>();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
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<String> alterTable(final String tableName, final CassandraPersistentEntity<?> entity,
final TableMetadata table) {
final List<String> result = new ArrayList<String>();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
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<String, Object> optionsByName) throws EntityWriterException {
final Insert q = QueryBuilder.insertInto(keyspaceName, tableName);
final Exception innerException = new Exception();
entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
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 <T> Batch toInsertBatchQuery(final String keyspaceName, final String tableName,
final List<T> objectsToSave, CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName)
throws EntityWriterException {
/*
* Return variable is a Batch statement
*/
final Batch b = QueryBuilder.batch();
List<Query> queries = new ArrayList<Query>();
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<String, Object> 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<CassandraPersistentProperty>() {
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 <T> Batch toDeleteBatchQuery(String keyspaceName, String tableName, List<T> entities,
CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
/*
* Return variable is a Batch statement
*/
final Batch b = QueryBuilder.batch();
List<Query> queries = new ArrayList<Query>();
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<String, Object> 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)));
}
}
}