wip: Added all overrides in CassandraOperations

Added QueryOptions to Operations, Template and CqlUtils.

src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core;
 
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.data.cassandra.convert.CassandraConverter;
 
@@ -25,6 +26,10 @@ import com.datastax.driver.core.ResultSetFuture;
 /**
  * @author Alex Shvid
  */
+/**
+ * @author David Webb
+ * 
+ */
 public interface CassandraOperations {
 
 	/**
@@ -92,6 +97,22 @@ public interface CassandraOperations {
 	 */
 	<T> T insert(T entity, String tableName);
 
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> T insert(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> T insert(T entity, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Insert the given list of objects to the table by annotation table name.
 	 * 
@@ -109,6 +130,22 @@ public interface CassandraOperations {
 	 */
 	<T> List<T> insert(List<T> entities, String tableName);
 
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> List<T> insert(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> List<T> insert(List<T> entities, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Insert the given object to the table by id.
 	 * 
@@ -116,6 +153,29 @@ public interface CassandraOperations {
 	 */
 	<T> T insertAsynchronously(T entity);
 
+	/**
+	 * Insert the given object to the table by id.
+	 * 
+	 * @param object
+	 */
+	<T> T insertAsynchronously(T entity, String tableName);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> T insertAsynchronously(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> T insertAsynchronously(T entity, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Insert the given object to the table by id.
 	 * 
@@ -128,14 +188,23 @@ public interface CassandraOperations {
 	 * 
 	 * @param object
 	 */
-	<T> T insertAsynchronously(T entity, String tableName);
+	<T> List<T> insertAsynchronously(List<T> entities, String tableName);
 
 	/**
-	 * Insert the given object to the table by id.
-	 * 
-	 * @param object
+	 * @param entities
+	 * @param tableName
+	 * @param options
+	 * @return
 	 */
-	<T> List<T> insertAsynchronously(List<T> entities, String tableName);
+	<T> List<T> insertAsynchronously(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> List<T> insertAsynchronously(List<T> entities, String tableName, Map<String, Object> optionsByName);
 
 	/**
 	 * Insert the given object to the table by id.
@@ -144,6 +213,29 @@ public interface CassandraOperations {
 	 */
 	<T> T update(T entity);
 
+	/**
+	 * Insert the given object to the table by id.
+	 * 
+	 * @param object
+	 */
+	<T> T update(T entity, String tableName);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> T update(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> T update(T entity, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Insert the given object to the table by id.
 	 * 
@@ -156,14 +248,23 @@ public interface CassandraOperations {
 	 * 
 	 * @param object
 	 */
-	<T> T update(T entity, String tableName);
+	<T> List<T> update(List<T> entities, String tableName);
 
 	/**
-	 * Insert the given object to the table by id.
-	 * 
-	 * @param object
+	 * @param entities
+	 * @param tableName
+	 * @param options
+	 * @return
 	 */
-	<T> List<T> update(List<T> entities, String tableName);
+	<T> List<T> update(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> List<T> update(List<T> entities, String tableName, Map<String, Object> optionsByName);
 
 	/**
 	 * Insert the given object to the table by id.
@@ -177,14 +278,30 @@ public interface CassandraOperations {
 	 * 
 	 * @param object
 	 */
-	<T> List<T> updateAsynchronously(List<T> entities);
+	<T> T updateAsynchronously(T entity, String tableName);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> T updateAsynchronously(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> T updateAsynchronously(T entity, String tableName, Map<String, Object> optionsByName);
 
 	/**
 	 * Insert the given object to the table by id.
 	 * 
 	 * @param object
 	 */
-	<T> T updateAsynchronously(T entity, String tableName);
+	<T> List<T> updateAsynchronously(List<T> entities);
 
 	/**
 	 * Insert the given object to the table by id.
@@ -193,6 +310,22 @@ public interface CassandraOperations {
 	 */
 	<T> List<T> updateAsynchronously(List<T> entities, String tableName);
 
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param options
+	 * @return
+	 */
+	<T> List<T> updateAsynchronously(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 * @return
+	 */
+	<T> List<T> updateAsynchronously(List<T> entities, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Remove the given object from the table by id.
 	 * 
@@ -200,6 +333,28 @@ public interface CassandraOperations {
 	 */
 	<T> void delete(T entity);
 
+	/**
+	 * Removes the given object from the given table.
+	 * 
+	 * @param object
+	 * @param table must not be {@literal null} or empty.
+	 */
+	<T> void delete(T entity, String tableName);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 */
+	<T> void delete(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 */
+	<T> void delete(T entity, String tableName, Map<String, Object> optionsByName);
+
 	/**
 	 * Remove the given object from the table by id.
 	 * 
@@ -213,15 +368,21 @@ public interface CassandraOperations {
 	 * @param object
 	 * @param table must not be {@literal null} or empty.
 	 */
-	<T> void delete(T entity, String tableName);
+	<T> void delete(List<T> entities, String tableName);
 
 	/**
-	 * Removes the given object from the given table.
-	 * 
-	 * @param object
-	 * @param table must not be {@literal null} or empty.
+	 * @param entities
+	 * @param tableName
+	 * @param options
 	 */
-	<T> void delete(List<T> entities, String tableName);
+	<T> void delete(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 */
+	<T> void delete(List<T> entities, String tableName, Map<String, Object> optionsByName);
 
 	/**
 	 * Remove the given object from the table by id.
@@ -230,6 +391,28 @@ public interface CassandraOperations {
 	 */
 	<T> void deleteAsychronously(T entity);
 
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param options
+	 */
+	<T> void deleteAsychronously(T entity, String tableName, QueryOptions options);
+
+	/**
+	 * @param entity
+	 * @param tableName
+	 * @param optionsByName
+	 */
+	<T> void deleteAsychronously(T entity, String tableName, Map<String, Object> optionsByName);
+
+	/**
+	 * Removes the given object from the given table.
+	 * 
+	 * @param object
+	 * @param table must not be {@literal null} or empty.
+	 */
+	<T> void deleteAsychronously(T entity, String tableName);
+
 	/**
 	 * Remove the given object from the table by id.
 	 * 
@@ -243,15 +426,21 @@ public interface CassandraOperations {
 	 * @param object
 	 * @param table must not be {@literal null} or empty.
 	 */
-	<T> void deleteAsychronously(T entity, String tableName);
+	<T> void deleteAsychronously(List<T> entities, String tableName);
 
 	/**
-	 * Removes the given object from the given table.
-	 * 
-	 * @param object
-	 * @param table must not be {@literal null} or empty.
+	 * @param entities
+	 * @param tableName
+	 * @param options
 	 */
-	<T> void deleteAsychronously(List<T> entities, String tableName);
+	<T> void deleteAsychronously(List<T> entities, String tableName, QueryOptions options);
+
+	/**
+	 * @param entities
+	 * @param tableName
+	 * @param optionsByName
+	 */
+	<T> void deleteAsychronously(List<T> entities, String tableName, Map<String, Object> optionsByName);
 
 	/**
 	 * Returns the underlying {@link CassandraConverter}.

src/main/java/org/springframework/data/cassandra/core/ConsistencyLevel.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2011-2013 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.data.cassandra.core;
+
+/**
+ * Generic Consistency Levels associated with Cassandra.
+ * 
+ * @author David Webb
+ * 
+ */
+public enum ConsistencyLevel {
+
+	ANY, ONE, TWO, THREE, QUOROM, LOCAL_QUOROM, EACH_QUOROM, ALL
+
+}

src/main/java/org/springframework/data/cassandra/core/ConsistencyLevelResolver.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2011-2013 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.data.cassandra.core;
+
+/**
+ * Determine driver consistency level based on ConsistencyLevel
+ * 
+ * @author David Webb
+ * 
+ */
+public final class ConsistencyLevelResolver {
+
+	/**
+	 * No instances allowed
+	 */
+	private ConsistencyLevelResolver() {
+	}
+
+	/**
+	 * Decode the generic spring data cassandra enum to the type required by the DataStax Driver.
+	 * 
+	 * @param level
+	 * @return The DataStax Driver Consistency Level.
+	 */
+	public static com.datastax.driver.core.ConsistencyLevel resolve(ConsistencyLevel level) {
+
+		com.datastax.driver.core.ConsistencyLevel resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ONE;
+
+		/*
+		 * Determine the driver level based on our enum
+		 */
+		switch (level) {
+		case ONE:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ONE;
+			break;
+		case ALL:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ALL;
+			break;
+		case ANY:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ANY;
+			break;
+		case EACH_QUOROM:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
+			break;
+		case LOCAL_QUOROM:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM;
+			break;
+		case QUOROM:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.QUORUM;
+			break;
+		case THREE:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.THREE;
+			break;
+		case TWO:
+			resolvedLevel = com.datastax.driver.core.ConsistencyLevel.TWO;
+			break;
+		default:
+			break;
+		}
+
+		return resolvedLevel;
+
+	}
+
+}

src/main/java/org/springframework/data/cassandra/core/QueryOptions.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2011-2013 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.data.cassandra.core;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Contains Query Options for Cassnadra queries. This controls the Consistency Tuning and Retry Policy for a Query.
+ * 
+ * @author David Webb
+ * 
+ */
+public class QueryOptions {
+
+	private ConsistencyLevel consistencyLevel;
+	private RetryPolicy retryPolicy;
+	private Integer ttl;
+
+	/**
+	 * Create a Map of all these options.
+	 */
+	public Map<String, Object> toMap() {
+
+		Map<String, Object> m = new HashMap<String, Object>();
+
+		if (getConsistencyLevel() != null) {
+			m.put(QueryOptionMapKeys.CONSISTENCY_LEVEL, getConsistencyLevel());
+		}
+		if (getRetryPolicy() != null) {
+			m.put(QueryOptionMapKeys.RETRY_POLICY, getRetryPolicy());
+		}
+		if (getTtl() != null) {
+			m.put(QueryOptionMapKeys.TTL, getTtl());
+		}
+
+		return m;
+	}
+
+	/**
+	 * @return Returns the consistencyLevel.
+	 */
+	public ConsistencyLevel getConsistencyLevel() {
+		return consistencyLevel;
+	}
+
+	/**
+	 * @param consistencyLevel The consistencyLevel to set.
+	 */
+	public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
+		this.consistencyLevel = consistencyLevel;
+	}
+
+	/**
+	 * @return Returns the retryPolicy.
+	 */
+	public RetryPolicy getRetryPolicy() {
+		return retryPolicy;
+	}
+
+	/**
+	 * @param retryPolicy The retryPolicy to set.
+	 */
+	public void setRetryPolicy(RetryPolicy retryPolicy) {
+		this.retryPolicy = retryPolicy;
+	}
+
+	/**
+	 * @return Returns the ttl.
+	 */
+	public Integer getTtl() {
+		return ttl;
+	}
+
+	/**
+	 * @param ttl The ttl to set.
+	 */
+	public void setTtl(Integer ttl) {
+		this.ttl = ttl;
+	}
+
+	/**
+	 * Constants for looking up Map Elements by Key
+	 * 
+	 * @author David Webb
+	 * 
+	 */
+	public static interface QueryOptionMapKeys {
+		public final String CONSISTENCY_LEVEL = "ConsistencyLevel";
+		public final String RETRY_POLICY = "RetryPolicy";
+		public final String TTL = "TTL";
+	}
+}

src/main/java/org/springframework/data/cassandra/core/RetryPolicy.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2011-2013 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.data.cassandra.core;
+
+/**
+ * Retry Policies associated with Cassandra.
+ * 
+ * @author David Webb
+ * 
+ */
+public enum RetryPolicy {
+
+	DEFAULT, DOWNGRADING_CONSISTENCY, FALLTHROUGH, LOGGING
+
+}

src/main/java/org/springframework/data/cassandra/core/RetryPolicyResolver.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2011-2013 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.data.cassandra.core;
+
+import com.datastax.driver.core.policies.DefaultRetryPolicy;
+import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
+import com.datastax.driver.core.policies.FallthroughRetryPolicy;
+
+/**
+ * Determine driver query retry policy
+ * 
+ * @author David Webb
+ * 
+ */
+public final class RetryPolicyResolver {
+
+	/**
+	 * No instances allowed
+	 */
+	private RetryPolicyResolver() {
+	}
+
+	/**
+	 * Decode the generic spring data cassandra enum to the type required by the DataStax Driver.
+	 * 
+	 * @param level
+	 * @return The DataStax Driver Consistency Level.
+	 */
+	public static com.datastax.driver.core.policies.RetryPolicy resolve(RetryPolicy policy) {
+
+		com.datastax.driver.core.policies.RetryPolicy resolvedPolicy = DefaultRetryPolicy.INSTANCE;
+
+		/*
+		 * Determine the driver level based on our enum
+		 */
+		switch (policy) {
+		case DEFAULT:
+			resolvedPolicy = DefaultRetryPolicy.INSTANCE;
+			break;
+		case DOWNGRADING_CONSISTENCY:
+			resolvedPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
+			break;
+		case FALLTHROUGH:
+			resolvedPolicy = FallthroughRetryPolicy.INSTANCE;
+			break;
+		default:
+			resolvedPolicy = DefaultRetryPolicy.INSTANCE;
+			break;
+		}
+
+		return resolvedPolicy;
+
+	}
+}

src/main/java/org/springframework/data/cassandra/util/CqlUtils.java

@@ -3,10 +3,16 @@ 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;
@@ -188,8 +194,6 @@ public abstract class CqlUtils {
 			}
 		});
 
-		// System.out.println("CQL=" + table.asCQLQuery());
-
 		return result;
 	}
 
@@ -200,6 +204,7 @@ public abstract class CqlUtils {
 	 * @param tableName
 	 * @param entity
 	 * @param objectToSave
+	 * @param optionsByName
 	 * @param mappingContext
 	 * @param beanClassLoader
 	 * 
@@ -207,7 +212,7 @@ public abstract class CqlUtils {
 	 * @throws EntityWriterException
 	 */
 	public static Query toInsertQuery(String keyspaceName, String tableName, final Object objectToSave,
-			CassandraPersistentEntity<?> entity) throws EntityWriterException {
+			CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
 
 		final Insert q = QueryBuilder.insertInto(keyspaceName, tableName);
 		final Exception innerException = new Exception();
@@ -242,6 +247,18 @@ public abstract class CqlUtils {
 			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;
 
 	}
@@ -260,7 +277,8 @@ public abstract class CqlUtils {
 	 * @throws EntityWriterException
 	 */
 	public static <T> Batch toInsertBatchQuery(final String keyspaceName, final String tableName,
-			final List<T> objectsToSave, CassandraPersistentEntity<?> entity) throws EntityWriterException {
+			final List<T> objectsToSave, CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName)
+			throws EntityWriterException {
 
 		/*
 		 * Return variable is a Batch statement
@@ -271,10 +289,12 @@ public abstract class CqlUtils {
 
 		for (final T objectToSave : objectsToSave) {
 
-			queries.add(toInsertQuery(keyspaceName, tableName, objectToSave, entity));
+			queries.add(toInsertQuery(keyspaceName, tableName, objectToSave, entity, optionsByName));
 
 		}
 
+		addQueryOptions(b, optionsByName);
+
 		return b;
 
 	}
@@ -288,7 +308,7 @@ public abstract class CqlUtils {
 	 * @throws EntityWriterException
 	 */
 	public static Query toDeleteQuery(String keyspace, String tableName, final Object objectToRemove,
-			CassandraPersistentEntity<?> entity) throws EntityWriterException {
+			CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
 
 		final Delete.Selection ds = QueryBuilder.delete();
 		final Delete q = ds.from(keyspace, tableName);
@@ -328,55 +348,12 @@ public abstract class CqlUtils {
 			throw new EntityWriterException("Failed to convert Persistent Entity to CQL/Query", innerException.getCause());
 		}
 
+		addQueryOptions(q, optionsByName);
+
 		return q;
 
 	}
 
-	/**
-	 * Generate the CQL for insert
-	 * 
-	 * @param tableName
-	 * @param entity
-	 * @return
-	 */
-	public static String toInsertCQL(String tableName, final CassandraPersistentEntity<?> entity) {
-
-		final StringBuilder str = new StringBuilder();
-		str.append("INSERT INTO ");
-		str.append(tableName);
-		str.append(" (");
-
-		final List<String> cols = 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();
-				cols.add(columnName);
-
-				str.append(columnName);
-
-			}
-		});
-
-		str.append(") VALUES (");
-
-		for (int i = 0; i < cols.size(); i++) {
-			if (i > 0) {
-				str.append(", ");
-			}
-			str.append("?");
-		}
-
-		str.append(")");
-
-		return str.toString();
-	}
-
 	/**
 	 * @param dataType
 	 * @return
@@ -423,7 +400,7 @@ public abstract class CqlUtils {
 	 * @throws EntityWriterException
 	 */
 	public static <T> Batch toDeleteBatchQuery(String keyspaceName, String tableName, List<T> entities,
-			CassandraPersistentEntity<?> entity) throws EntityWriterException {
+			CassandraPersistentEntity<?> entity, Map<String, Object> optionsByName) throws EntityWriterException {
 
 		/*
 		 * Return variable is a Batch statement
@@ -434,12 +411,40 @@ public abstract class CqlUtils {
 
 		for (final T objectToSave : entities) {
 
-			queries.add(toDeleteQuery(keyspaceName, tableName, objectToSave, entity));
+			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)));
+		}
+
+	}
+
 }