diff --git a/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java b/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
new file mode 100644
index 000000000..90fdba895
--- /dev/null
+++ b/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
@@ -0,0 +1,80 @@
+/*
+ * 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.cassandra.core;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.DriverException;
+
+/**
+ * Created a PreparedStatement and retrieved the PreparedStatement from cache if the statement has been prepared
+ * previously. In general, this creator should be used over the {@link SimplePreparedStatementCreator} as it provides
+ * better performance.
+ *
+ *
+ * There is overhead in Cassandra when Preparing a Statement. This is negligible on a single data center configuration,
+ * but when your cluster spans multiple data centers, preparing the same statement over and over again is not necessary
+ * and causes performance issues in high throughput use cases.
+ *
+ *
+ * @author David Webb
+ *
+ */
+public class CachedPreparedStatementCreator implements PreparedStatementCreator, CqlProvider {
+
+ private static Logger log = LoggerFactory.getLogger(CachedPreparedStatementCreator.class);
+
+ private final String cql;
+
+ private PreparedStatement cache;
+
+ /**
+ * Create a CachedPreparedStatementCreator from the provided CQL.
+ *
+ * @param cql
+ */
+ public CachedPreparedStatementCreator(String cql) {
+ Assert.notNull(cql, "CQL is required to create a PreparedStatement");
+ this.cql = cql;
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
+ */
+ @Override
+ public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+ if (cache == null) {
+ log.debug("PreparedStatement cache is null, preparing new Statement");
+ cache = session.prepare(getCql());
+ } else {
+ log.debug("Using cached PreparedStatement");
+ }
+ return cache;
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.CqlProvider#getCql()
+ */
+ @Override
+ public String getCql() {
+ return this.cql;
+ }
+
+}
diff --git a/src/main/java/org/springframework/cassandra/core/CqlParameter.java b/src/main/java/org/springframework/cassandra/core/CqlParameter.java
new file mode 100644
index 000000000..bfd5db193
--- /dev/null
+++ b/src/main/java/org/springframework/cassandra/core/CqlParameter.java
@@ -0,0 +1,139 @@
+/*
+ * 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.cassandra.core;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.DataType;
+
+/**
+ * @author David Webb
+ *
+ */
+public class CqlParameter {
+
+ /** The name of the parameter, if any */
+ private String name;
+
+ /** SQL type constant from {@link DataType} */
+ private final DataType type;
+
+ /** The scale to apply in case of a NUMERIC or DECIMAL type, if any */
+ private Integer scale;
+
+ /**
+ * Create a new anonymous CqlParameter, supplying the SQL type.
+ *
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ */
+ public CqlParameter(DataType type) {
+ this.type = type;
+ }
+
+ /**
+ * Create a new anonymous CqlParameter, supplying the SQL type.
+ *
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ * @param scale the number of digits after the decimal point
+ */
+ public CqlParameter(DataType type, int scale) {
+ this.type = type;
+ this.scale = scale;
+ }
+
+ /**
+ * Create a new CqlParameter, supplying name and SQL type.
+ *
+ * @param name name of the parameter, as used in input and output maps
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ */
+ public CqlParameter(String name, DataType type) {
+ this.name = name;
+ this.type = type;
+ }
+
+ /**
+ * Create a new CqlParameter, supplying name and SQL type.
+ *
+ * @param name name of the parameter, as used in input and output maps
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ * @param scale the number of digits after the decimal point (for DECIMAL and NUMERIC types)
+ */
+ public CqlParameter(String name, DataType type, int scale) {
+ this.name = name;
+ this.type = type;
+ this.scale = scale;
+ }
+
+ /**
+ * Copy constructor.
+ *
+ * @param otherParam the CqlParameter object to copy from
+ */
+ public CqlParameter(CqlParameter otherParam) {
+ Assert.notNull(otherParam, "CqlParameter object must not be null");
+ this.name = otherParam.name;
+ this.type = otherParam.type;
+ this.scale = otherParam.scale;
+ }
+
+ /**
+ * Return the name of the parameter.
+ */
+ public String getName() {
+ return this.name;
+ }
+
+ /**
+ * Return the SQL type of the parameter.
+ */
+ public DataType getType() {
+ return this.type;
+ }
+
+ /**
+ * Return the scale of the parameter, if any.
+ */
+ public Integer getScale() {
+ return this.scale;
+ }
+
+ /**
+ * Return whether this parameter holds input values that should be set before execution even if they are {@code null}.
+ *
+ * This implementation always returns {@code true}.
+ */
+ public boolean isInputValueProvided() {
+ return true;
+ }
+
+ /**
+ * Convert a list of JDBC types, as defined in {@code java.sql.Types}, to a List of CqlParameter objects as used in
+ * this package.
+ */
+ public static List sqlTypesToAnonymousParameterList(DataType[] types) {
+ List result = new LinkedList();
+ if (types != null) {
+ for (DataType type : types) {
+ result.add(new CqlParameter(type));
+ }
+ }
+ return result;
+ }
+}
diff --git a/src/main/java/org/springframework/cassandra/core/CqlParameterValue.java b/src/main/java/org/springframework/cassandra/core/CqlParameterValue.java
new file mode 100644
index 000000000..c2932815f
--- /dev/null
+++ b/src/main/java/org/springframework/cassandra/core/CqlParameterValue.java
@@ -0,0 +1,68 @@
+/*
+ * 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.cassandra.core;
+
+import com.datastax.driver.core.DataType;
+
+/**
+ * @author David Webb
+ *
+ */
+public class CqlParameterValue extends CqlParameter {
+
+ private final Object value;
+
+ /**
+ * Create a new CqlParameterValue, supplying the Cassandra DataType.
+ *
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ * @param value the value object
+ */
+ public CqlParameterValue(DataType type, Object value) {
+ super(type);
+ this.value = value;
+ }
+
+ /**
+ * Create a new CqlParameterValue, supplying the Cassandra DataType.
+ *
+ * @param type Cassandra Data Type of the parameter according to {@link DataType}
+ * @param scale the number of digits after the decimal point (for DECIMAL and NUMERIC types)
+ * @param value the value object
+ */
+ public CqlParameterValue(DataType type, int scale, Object value) {
+ super(type, scale);
+ this.value = value;
+ }
+
+ /**
+ * Create a new CqlParameterValue based on the given CqlParameter declaration.
+ *
+ * @param declaredParam the declared CqlParameter to define a value for
+ * @param value the value object
+ */
+ public CqlParameterValue(CqlParameter declaredParam, Object value) {
+ super(declaredParam);
+ this.value = value;
+ }
+
+ /**
+ * Return the value object that this parameter value holds.
+ */
+ public Object getValue() {
+ return this.value;
+ }
+}
diff --git a/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorFactory.java b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorFactory.java
new file mode 100644
index 000000000..974b0436e
--- /dev/null
+++ b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorFactory.java
@@ -0,0 +1,202 @@
+/*
+ * 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.cassandra.core;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.DriverException;
+
+/**
+ * @author David Webb
+ *
+ */
+public class PreparedStatementCreatorFactory {
+
+ /**
+ * The CQL, which won't change when the parameters change
+ */
+ private final String cql;
+
+ /** List of CqlParameter objects. May not be {@code null}. */
+ private final List declaredParameters;
+
+ /**
+ * Create a new factory.
+ */
+ public PreparedStatementCreatorFactory(String cql) {
+ this.cql = cql;
+ this.declaredParameters = new LinkedList();
+ }
+
+ /**
+ * Create a new factory with the given CQL and parameters.
+ *
+ * @param cql CQL
+ * @param declaredParameters list of {@link CqlParameter} objects
+ * @see CqlParameter
+ */
+ public PreparedStatementCreatorFactory(String cql, List declaredParameters) {
+ this.cql = cql;
+ this.declaredParameters = declaredParameters;
+ }
+
+ /**
+ * Return a new PreparedStatementBinder for the given parameters.
+ *
+ * @param params list of parameters (may be {@code null})
+ */
+ public PreparedStatementBinder newPreparedStatementBinder(List params) {
+ return new PreparedStatementCreatorImpl(params != null ? params : Collections.emptyList());
+ }
+
+ /**
+ * Return a new PreparedStatementBinder for the given parameters.
+ *
+ * @param params the parameter array (may be {@code null})
+ */
+ public PreparedStatementBinder newPreparedStatementBinder(Object[] params) {
+ return new PreparedStatementCreatorImpl(params != null ? Arrays.asList(params) : Collections.emptyList());
+ }
+
+ /**
+ * Return a new PreparedStatementCreator for the given parameters.
+ *
+ * @param params list of parameters (may be {@code null})
+ */
+ public PreparedStatementCreator newPreparedStatementCreator(List params) {
+ return new PreparedStatementCreatorImpl(params != null ? params : Collections.emptyList());
+ }
+
+ /**
+ * Return a new PreparedStatementCreator for the given parameters.
+ *
+ * @param params the parameter array (may be {@code null})
+ */
+ public PreparedStatementCreator newPreparedStatementCreator(Object[] params) {
+ return new PreparedStatementCreatorImpl(params != null ? Arrays.asList(params) : Collections.emptyList());
+ }
+
+ /**
+ * Return a new PreparedStatementCreator for the given parameters.
+ *
+ * @param sqlToUse the actual SQL statement to use (if different from the factory's, for example because of named
+ * parameter expanding)
+ * @param params the parameter array (may be {@code null})
+ */
+ public PreparedStatementCreator newPreparedStatementCreator(String sqlToUse, Object[] params) {
+ return new PreparedStatementCreatorImpl(sqlToUse, params != null ? Arrays.asList(params) : Collections.emptyList());
+ }
+
+ /**
+ * PreparedStatementCreator implementation returned by this class.
+ */
+ private class PreparedStatementCreatorImpl implements PreparedStatementCreator, PreparedStatementBinder, CqlProvider {
+
+ private final String actualCql;
+
+ private final List parameters;
+
+ public PreparedStatementCreatorImpl(List parameters) {
+ this(cql, parameters);
+ }
+
+ /**
+ * @param actualCql
+ * @param parameters
+ */
+ public PreparedStatementCreatorImpl(String actualCql, List parameters) {
+ this.actualCql = actualCql;
+ Assert.notNull(parameters, "Parameters List must not be null");
+ this.parameters = parameters;
+ if (this.parameters.size() != declaredParameters.size()) {
+ Set names = new HashSet();
+ for (int i = 0; i < parameters.size(); i++) {
+ Object param = parameters.get(i);
+ if (param instanceof CqlParameterValue) {
+ names.add(((CqlParameterValue) param).getName());
+ } else {
+ names.add("Parameter #" + i);
+ }
+ }
+ if (names.size() != declaredParameters.size()) {
+ throw new InvalidDataAccessApiUsageException("CQL [" + cql + "]: given " + names.size()
+ + " parameters but expected " + declaredParameters.size());
+ }
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
+ */
+ @Override
+ public PreparedStatement createPreparedStatement(Session session) throws DriverException {
+ return session.prepare(this.actualCql);
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.PreparedStatementBinder#bindValues(com.datastax.driver.core.PreparedStatement)
+ */
+ @Override
+ public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
+ if (this.parameters == null || this.parameters.size() == 0) {
+ return ps.bind();
+ }
+
+ // Test the type of the first value
+ Object v = this.parameters.get(0);
+ Object[] values;
+ if (v instanceof CqlParameterValue) {
+ LinkedList