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

DATACASS-114 - Added WriteOptions with TTL support in CqlOperations and

Browse Source 
CassandraOperations.
This commit is contained in:
David T Webb
2014-03-21 12:35:12 -04:00
parent dd8a89da57
commit 482661268d
6 changed files with 162 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java
View File

@@ -973,7 +973,7 @@ public interface CqlOperations {
* @param rowIterator Implementation to provide the Object[] to be bound to the CQL.
* @param options The Query Options Object
*/
void ingest(String cql, RowIterator rowIterator, QueryOptions options);
void ingest(String cql, RowIterator rowIterator, WriteOptions options);
/**
* This is an operation designed for high performance writes. The cql is used to create a PreparedStatement once, then
@@ -1001,7 +1001,7 @@ public interface CqlOperations {
* @param rows List of List<?> with data to bind to the CQL.
* @param options The Query Options Object
*/
void ingest(String cql, List<List<?>> rows, QueryOptions options);
void ingest(String cql, List<List<?>> rows, WriteOptions options);
/**
* This is an operation designed for high performance writes. The cql is used to create a PreparedStatement once, then
@@ -1041,7 +1041,7 @@ public interface CqlOperations {
* @param rows Object array of Object array of values to bind to the CQL.
* @param options The Query Options Object
*/
void ingest(String cql, Object[][] rows, QueryOptions options);
void ingest(String cql, Object[][] rows, WriteOptions options);
/**
* Delete all rows in the table

84
spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
View File

@@ -116,6 +116,84 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return q;
}
/**
* Add common {@link Query} options for Insert queries.
*
* @param q
* @param options
* @return the {@link Query} given.
*/
public static Query addWriteOptions(Insert q, WriteOptions options) {
if (options == null) {
return q;
}
if (options.getConsistencyLevel() != null) {
q.setConsistencyLevel(ConsistencyLevelResolver.resolve(options.getConsistencyLevel()));
}
if (options.getRetryPolicy() != null) {
q.setRetryPolicy(RetryPolicyResolver.resolve(options.getRetryPolicy()));
}
if (options.getTtl() != null) {
q.using(QueryBuilder.ttl(options.getTtl()));
}
return q;
}
/**
* Add common {@link Query} options for Update queries.
*
* @param q
* @param options
* @return the {@link Query} given.
*/
public static Query addWriteOptions(Update q, WriteOptions options) {
if (options == null) {
return q;
}
if (options.getConsistencyLevel() != null) {
q.setConsistencyLevel(ConsistencyLevelResolver.resolve(options.getConsistencyLevel()));
}
if (options.getRetryPolicy() != null) {
q.setRetryPolicy(RetryPolicyResolver.resolve(options.getRetryPolicy()));
}
if (options.getTtl() != null) {
q.using(QueryBuilder.ttl(options.getTtl()));
}
return q;
}
/**
* Add common {@link Query} options for Batch queries.
*
* @param q
* @param options
* @return the {@link Query} given.
*/
public static Query addWriteOptions(Batch q, WriteOptions options) {
if (options == null) {
return q;
}
if (options.getConsistencyLevel() != null) {
q.setConsistencyLevel(ConsistencyLevelResolver.resolve(options.getConsistencyLevel()));
}
if (options.getRetryPolicy() != null) {
q.setRetryPolicy(RetryPolicyResolver.resolve(options.getRetryPolicy()));
}
if (options.getTtl() != null) {
q.using(QueryBuilder.ttl(options.getTtl()));
}
return q;
}
/**
* Add common Query options for all types of queries.
*
@@ -743,7 +821,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void ingest(String cql, RowIterator rowIterator, QueryOptions options) {
public void ingest(String cql, RowIterator rowIterator, WriteOptions options) {
PreparedStatement preparedStatement = getSession().prepare(cql);
addPreparedStatementOptions(preparedStatement, options);
@@ -759,7 +837,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void ingest(String cql, final List<List<?>> rows, QueryOptions options) {
public void ingest(String cql, final List<List<?>> rows, WriteOptions options) {
Assert.notNull(rows);
Assert.notEmpty(rows);
@@ -788,7 +866,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void ingest(String cql, final Object[][] rows, QueryOptions options) {
public void ingest(String cql, final Object[][] rows, WriteOptions options) {
ingest(cql, new RowIterator() {

40
spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2013-2014 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
/**
* @author David Webb
*
*/
public class WriteOptions extends QueryOptions {
private Integer ttl;
/**
* @return Returns the ttl.
*/
public Integer getTtl() {
return ttl;
}
/**
* @param ttl The ttl to set.
*/
public void setTtl(Integer ttl) {
this.ttl = ttl;
}
}

14
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java → spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java
View File

@@ -47,6 +47,7 @@ import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.cassandra.core.RowIterator;
import org.springframework.cassandra.core.RowMapper;
import org.springframework.cassandra.core.SessionCallback;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.keyspace.CreateTableSpecification;
import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
import org.springframework.dao.DataAccessException;
@@ -75,11 +76,11 @@ import static org.junit.Assert.assertTrue;
* @author David Webb
*
*/
public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
private static CqlOperations cqlTemplate;
private static Logger log = LoggerFactory.getLogger(CassandraOperationsTest.class);
private static Logger log = LoggerFactory.getLogger(CQLOperationsTest.class);
/*
* Objects used for test data
@@ -98,7 +99,7 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
"cassandraOperationsTest-cql-dataload.cql", this.keyspace), CASSANDRA_CONFIG, CASSANDRA_HOST,
CASSANDRA_NATIVE_PORT);
public CassandraOperationsTest() {
public CQLOperationsTest() {
super();
// TODO clear = true;
}
@@ -144,7 +145,7 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
@Override
public Collection<MyHost> mapHosts(Set<Host> host) throws DriverException {
List<MyHost> list = new LinkedList<CassandraOperationsTest.MyHost>();
List<MyHost> list = new LinkedList<CQLOperationsTest.MyHost>();
for (Host h : host) {
MyHost mh = new MyHost();
@@ -170,6 +171,9 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
@SuppressWarnings("unchecked")
public void ingestionTestListOfList() {
WriteOptions options = new WriteOptions();
options.setTtl(360);
String cql = "insert into book (isbn, title, author, pages) values (?, ?, ?, ?)";
List<List<?>> values = new LinkedList<List<?>>();
@@ -178,7 +182,7 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
values.add(new LinkedList<Object>(CollectionUtils.arrayToList(o2)));
values.add(new LinkedList<Object>(CollectionUtils.arrayToList(o3)));
cqlTemplate.ingest(cql, values);
cqlTemplate.ingest(cql, values, options);
// Assert that the rows were inserted into Cassandra
Book b1 = getBook((String) o1[0]);

17
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java
View File

@@ -19,6 +19,7 @@ import java.util.List;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.convert.CassandraConverter;
@@ -77,7 +78,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> T insert(T entity, QueryOptions options);
<T> T insert(T entity, WriteOptions options);
/**
* Insert the given list of objects to the table by annotation table name.
@@ -93,7 +94,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> List<T> insert(List<T> entities, QueryOptions options);
<T> List<T> insert(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -108,7 +109,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> T insertAsynchronously(T entity, QueryOptions options);
<T> T insertAsynchronously(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -123,7 +124,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> List<T> insertAsynchronously(List<T> entities, QueryOptions options);
<T> List<T> insertAsynchronously(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -138,7 +139,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> T update(T entity, QueryOptions options);
<T> T update(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -153,7 +154,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> List<T> update(List<T> entities, QueryOptions options);
<T> List<T> update(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -168,7 +169,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> T updateAsynchronously(T entity, QueryOptions options);
<T> T updateAsynchronously(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
@@ -183,7 +184,7 @@ public interface CassandraOperations extends CqlOperations {
* @param options
* @return
*/
<T> List<T> updateAsynchronously(List<T> entities, QueryOptions options);
<T> List<T> updateAsynchronously(List<T> entities, WriteOptions options);
/**
* Remove the given object from the table by id.

37
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
import org.springframework.cassandra.core.CqlTemplate;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.SessionCallback;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.cassandra.core.util.CollectionUtils;
import org.springframework.dao.DataAccessException;
@@ -66,7 +67,8 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
/**
* Default Constructor for wiring in the required components later
*/
public CassandraTemplate() {}
public CassandraTemplate() {
}
/**
* Constructor if only session and converter are known at time of Template Creation
@@ -188,7 +190,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> List<T> insert(List<T> entities, QueryOptions options) {
public <T> List<T> insert(List<T> entities, WriteOptions options) {
return batchInsert(entities, options, false);
}
@@ -198,7 +200,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> T insert(T entity, QueryOptions options) {
public <T> T insert(T entity, WriteOptions options) {
return insert(entity, options, false);
}
@@ -208,7 +210,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> List<T> insertAsynchronously(List<T> entities, QueryOptions options) {
public <T> List<T> insertAsynchronously(List<T> entities, WriteOptions options) {
return batchInsert(entities, options, true);
}
@@ -218,7 +220,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> T insertAsynchronously(T entity, QueryOptions options) {
public <T> T insertAsynchronously(T entity, WriteOptions options) {
return insert(entity, options, true);
}
@@ -350,7 +352,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> List<T> update(List<T> entities, QueryOptions options) {
public <T> List<T> update(List<T> entities, WriteOptions options) {
return batchUpdate(entities, options, false);
}
@@ -360,7 +362,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> T update(T entity, QueryOptions options) {
public <T> T update(T entity, WriteOptions options) {
return update(entity, options, false);
}
@@ -370,7 +372,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> List<T> updateAsynchronously(List<T> entities, QueryOptions options) {
public <T> List<T> updateAsynchronously(List<T> entities, WriteOptions options) {
return batchUpdate(entities, options, true);
}
@@ -380,7 +382,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
@Override
public <T> T updateAsynchronously(T entity, QueryOptions options) {
public <T> T updateAsynchronously(T entity, WriteOptions options) {
return update(entity, options, true);
}
@@ -459,13 +461,14 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
}
}
protected <T> T insert(T entity, QueryOptions options, boolean asynchronously) {
protected <T> T insert(T entity, WriteOptions options, boolean asynchronously) {
Assert.notNull(entity);
Insert insert = createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
String query = insert.getQueryString();
logger.debug(query);
if (asynchronously) {
executeAsynchronously(query);
@@ -476,7 +479,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
return entity;
}
protected <T> List<T> batchInsert(List<T> entities, QueryOptions options, boolean asychronously) {
protected <T> List<T> batchInsert(List<T> entities, WriteOptions options, boolean asychronously) {
Assert.notEmpty(entities);
@@ -504,7 +507,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param updateAsychronously
* @return
*/
protected <T> List<T> batchUpdate(List<T> entities, QueryOptions options, boolean asychronously) {
protected <T> List<T> batchUpdate(List<T> entities, WriteOptions options, boolean asychronously) {
Assert.notEmpty(entities);
@@ -554,7 +557,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param updateAsychronously
* @return
*/
protected <T> T update(T entity, QueryOptions options, boolean asychronously) {
protected <T> T update(T entity, WriteOptions options, boolean asychronously) {
Assert.notNull(entity);
@@ -581,7 +584,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static Insert createInsertQuery(String tableName, Object objectToSave, QueryOptions options,
public static Insert createInsertQuery(String tableName, Object objectToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
Insert q = QueryBuilder.insertInto(tableName);
@@ -608,7 +611,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static Update toUpdateQuery(String tableName, Object objectToSave, QueryOptions options,
public static Update toUpdateQuery(String tableName, Object objectToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
Update q = QueryBuilder.update(tableName);
@@ -635,7 +638,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static <T> Batch toUpdateBatchQuery(String tableName, List<T> objectsToSave, QueryOptions options,
public static <T> Batch toUpdateBatchQuery(String tableName, List<T> objectsToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
/*
@@ -667,7 +670,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static <T> Batch createInsertBatchQuery(String tableName, List<T> entities, QueryOptions options,
public static <T> Batch createInsertBatchQuery(String tableName, List<T> entities, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
Batch batch = QueryBuilder.batch();