From 482661268d9bf2732222df29ec6f98805d2f9532 Mon Sep 17 00:00:00 2001 From: David T Webb Date: Fri, 21 Mar 2014 12:35:12 -0400 Subject: [PATCH] DATACASS-114 - Added WriteOptions with TTL support in CqlOperations and CassandraOperations. --- .../cassandra/core/CqlOperations.java | 6 +- .../cassandra/core/CqlTemplate.java | 84 ++++++++++++++++++- .../cassandra/core/WriteOptions.java | 40 +++++++++ ...ationsTest.java => CQLOperationsTest.java} | 14 ++-- .../cassandra/core/CassandraOperations.java | 17 ++-- .../cassandra/core/CassandraTemplate.java | 37 ++++---- 6 files changed, 162 insertions(+), 36 deletions(-) create mode 100644 spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java rename spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/{CassandraOperationsTest.java => CQLOperationsTest.java} (98%) diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java index 2cec14e5b..1bfb8e7e4 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java @@ -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> rows, QueryOptions options); + void ingest(String cql, 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 diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java index 6b15ce76b..42fa2a740 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java @@ -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> rows, QueryOptions options) { + public void ingest(String cql, final 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() { diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java b/spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java new file mode 100644 index 000000000..00467c8cd --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java @@ -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; + } + +} diff --git a/spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java b/spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java similarity index 98% rename from spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java rename to spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java index 3adee19c8..a359b89fa 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java @@ -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 mapHosts(Set host) throws DriverException { - List list = new LinkedList(); + List list = new LinkedList(); 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> values = new LinkedList>(); @@ -178,7 +182,7 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration values.add(new LinkedList(CollectionUtils.arrayToList(o2))); values.add(new LinkedList(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]); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java index 297bbbb36..fae5dcd09 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java @@ -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 insert(T entity, QueryOptions options); + 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 */ - List insert(List entities, QueryOptions options); + List insert(List 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 insertAsynchronously(T entity, QueryOptions options); + 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 */ - List insertAsynchronously(List entities, QueryOptions options); + List insertAsynchronously(List 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 update(T entity, QueryOptions options); + 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 */ - List update(List entities, QueryOptions options); + List update(List 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 updateAsynchronously(T entity, QueryOptions options); + 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 */ - List updateAsynchronously(List entities, QueryOptions options); + List updateAsynchronously(List entities, WriteOptions options); /** * Remove the given object from the table by id. diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index 372eac274..17f604318 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -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 List insert(List entities, QueryOptions options) { + public List insert(List entities, WriteOptions options) { return batchInsert(entities, options, false); } @@ -198,7 +200,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public T insert(T entity, QueryOptions options) { + public T insert(T entity, WriteOptions options) { return insert(entity, options, false); } @@ -208,7 +210,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public List insertAsynchronously(List entities, QueryOptions options) { + public List insertAsynchronously(List entities, WriteOptions options) { return batchInsert(entities, options, true); } @@ -218,7 +220,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public T insertAsynchronously(T entity, QueryOptions options) { + public T insertAsynchronously(T entity, WriteOptions options) { return insert(entity, options, true); } @@ -350,7 +352,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public List update(List entities, QueryOptions options) { + public List update(List entities, WriteOptions options) { return batchUpdate(entities, options, false); } @@ -360,7 +362,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public T update(T entity, QueryOptions options) { + public T update(T entity, WriteOptions options) { return update(entity, options, false); } @@ -370,7 +372,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public List updateAsynchronously(List entities, QueryOptions options) { + public List updateAsynchronously(List entities, WriteOptions options) { return batchUpdate(entities, options, true); } @@ -380,7 +382,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } @Override - public T updateAsynchronously(T entity, QueryOptions options) { + public T updateAsynchronously(T entity, WriteOptions options) { return update(entity, options, true); } @@ -459,13 +461,14 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation } } - protected T insert(T entity, QueryOptions options, boolean asynchronously) { + protected 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 List batchInsert(List entities, QueryOptions options, boolean asychronously) { + protected List batchInsert(List entities, WriteOptions options, boolean asychronously) { Assert.notEmpty(entities); @@ -504,7 +507,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation * @param updateAsychronously * @return */ - protected List batchUpdate(List entities, QueryOptions options, boolean asychronously) { + protected List batchUpdate(List entities, WriteOptions options, boolean asychronously) { Assert.notEmpty(entities); @@ -554,7 +557,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation * @param updateAsychronously * @return */ - protected T update(T entity, QueryOptions options, boolean asychronously) { + protected 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 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 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 Batch toUpdateBatchQuery(String tableName, List objectsToSave, QueryOptions options, + public static Batch toUpdateBatchQuery(String tableName, List objectsToSave, WriteOptions options, EntityWriter 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 Batch createInsertBatchQuery(String tableName, List entities, QueryOptions options, + public static Batch createInsertBatchQuery(String tableName, List entities, WriteOptions options, EntityWriter entityWriter) { Batch batch = QueryBuilder.batch();