Merge branch 'DATACASS-32' of https://github.com/shvid/spring-data-cassandra into DATACASS-32

src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java

@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.config;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
@@ -26,6 +27,8 @@ import org.springframework.core.type.filter.AnnotationTypeFilter;
 import org.springframework.data.annotation.Persistent;
 import org.springframework.data.cassandra.convert.CassandraConverter;
 import org.springframework.data.cassandra.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.CassandraAdminOperations;
+import org.springframework.data.cassandra.core.CassandraAdminTemplate;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.CassandraTemplate;
 import org.springframework.data.cassandra.core.Keyspace;
@@ -46,7 +49,13 @@ import com.datastax.driver.core.Session;
  * @author Alex Shvid
  */
 @Configuration
-public abstract class AbstractCassandraConfiguration {
+public abstract class AbstractCassandraConfiguration implements BeanClassLoaderAware {
+
+	/**
+	 * Used by CassandraTemplate and CassandraAdminTemplate
+	 */
+
+	private ClassLoader beanClassLoader;
 
 	/**
 	 * Return the name of the keyspace to connect to.
@@ -118,7 +127,22 @@ public abstract class AbstractCassandraConfiguration {
 	 */
 	@Bean
 	public CassandraOperations cassandraTemplate() throws Exception {
-		return new CassandraTemplate(keyspace());
+		CassandraTemplate template = new CassandraTemplate(keyspace());
+		template.setBeanClassLoader(beanClassLoader);
+		return template;
+	}
+
+	/**
+	 * Creates a {@link CassandraAdminTemplate}.
+	 * 
+	 * @return
+	 * @throws Exception
+	 */
+	@Bean
+	public CassandraAdminOperations cassandraAdminTemplate() throws Exception {
+		CassandraAdminTemplate adminTemplate = new CassandraAdminTemplate(keyspace());
+		adminTemplate.setBeanClassLoader(beanClassLoader);
+		return adminTemplate;
 	}
 
 	/**
@@ -170,4 +194,12 @@ public abstract class AbstractCassandraConfiguration {
 		return initialEntitySet;
 	}
 
+	/**
+	 * Bean ClassLoader Aware for CassandraTemplate/CassandraAdminTemplate
+	 */
+
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = classLoader;
+	}
+
 }

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

@@ -5,6 +5,7 @@ import java.util.Map;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
@@ -23,7 +24,7 @@ import com.datastax.driver.core.TableMetadata;
 /**
  * Default implementation of {@link CassandraAdminOperations}.
  */
-public class CassandraAdminTemplate implements CassandraAdminOperations {
+public class CassandraAdminTemplate implements CassandraAdminOperations, BeanClassLoaderAware {
 
 	private static Logger log = LoggerFactory.getLogger(CassandraAdminTemplate.class);
 
@@ -239,4 +240,9 @@ public class CassandraAdminTemplate implements CassandraAdminOperations {
 		}
 		return entity.getTable();
 	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = classLoader;
+	}
 }

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

@@ -27,6 +27,7 @@ import java.util.Set;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -58,7 +59,7 @@ import com.datastax.driver.core.querybuilder.Batch;
  * @author Alex Shvid
  * @author David Webb
  */
-public class CassandraTemplate implements CassandraOperations {
+public class CassandraTemplate implements CassandraOperations, BeanClassLoaderAware {
 
 	/**
 	 * Simple {@link RowCallback} that will transform {@link Row} into the given target type using the given
@@ -79,8 +80,8 @@ public class CassandraTemplate implements CassandraOperations {
 		}
 
 		@Override
-		public T doWith(Row object) {
-			T source = reader.read(type, object);
+		public T doWith(Row row) {
+			T source = reader.read(type, row);
 			return source;
 		}
 	}

src/main/java/org/springframework/data/cassandra/cql/CqlBuilder.java

@@ -1,8 +0,0 @@
-package org.springframework.data.cassandra.cql;
-
-public class CqlBuilder {
-
-	public static CreateTable createTable() {
-		return new CreateTable();
-	}
-}

src/main/java/org/springframework/data/cassandra/cql/CqlStringUtils.java

@@ -0,0 +1,92 @@
+package org.springframework.data.cassandra.cql;
+
+import java.util.regex.Pattern;
+
+public class CqlStringUtils {
+
+	protected static final String SINGLE_QUOTE = "\'";
+	protected static final String DOUBLE_SINGLE_QUOTE = "\'\'";
+	protected static final String DOUBLE_QUOTE = "\"";
+	protected static final String DOUBLE_DOUBLE_QUOTE = "\"\"";
+
+	/**
+	 * Helper {@link StringBuilder} factory method. If given a non-<code>null</code> argument, returns that, else returns
+	 * a new {@link StringBuilder}. Intended to be imported statically by other classes in the builder's fluent API
+	 * implementation.
+	 * 
+	 * @param sb
+	 * @return The given {@link StringBuilder} if not null, else a new one.
+	 * 
+	 * @author Matthew T. Adams
+	 */
+	public static StringBuilder ensureNotNull(StringBuilder sb) {
+		return sb == null ? new StringBuilder() : sb;
+	}
+
+	public static final String IDENTIFIER_REGEX = "[a-zA-Z0-9_]*";
+	public static final Pattern IDENTIFIER_PATTERN = Pattern.compile(IDENTIFIER_REGEX);
+
+	public static boolean isIdentifier(CharSequence chars) {
+		return IDENTIFIER_PATTERN.matcher(chars).matches();
+	}
+
+	public static final String QUOTED_IDENTIFIER_REGEX = "([a-zA-Z0-9_]|'{2}+|\"{2}+)*";
+	public static final Pattern QUOTED_IDENTIFIER_PATTERN = Pattern.compile(IDENTIFIER_REGEX);
+
+	public static boolean isQuotedIdentifier(CharSequence chars) {
+		return QUOTED_IDENTIFIER_PATTERN.matcher(chars).matches();
+	}
+
+	public static void checkQuotedIdentifier(CharSequence chars) {
+		if (!CqlStringUtils.isQuotedIdentifier(chars)) {
+			throw new IllegalArgumentException("[" + chars + "] is not a valid CQL quoted identifier");
+		}
+	}
+
+	/**
+	 * Trims then escapes the given {@link CharSequence}. Given <code>null</code>, returns <code>null</code>.
+	 */
+	public static String scrub(Object thing) {
+		return thing == null ? (String) null : escape(thing.toString().trim());
+	}
+
+	/**
+	 * Doubles single quote characters and doubles double quote characters (' -&gt; '' and " -&gt; ""). Given
+	 * <code>null</code>, returns <code>null</code>.
+	 */
+	public static String escape(Object thing) {
+		return escapeDouble(escapeSingle(thing));
+	}
+
+	/**
+	 * Doubles single quote characters (' -&gt; ''). Given <code>null</code>, returns <code>null</code>.
+	 */
+	public static String escapeSingle(Object things) {
+		return things == null ? (String) null : things.toString().replace(SINGLE_QUOTE, DOUBLE_SINGLE_QUOTE);
+	}
+
+	/**
+	 * Doubles double quote characters (" -&gt; ""). Given <code>null</code>, returns <code>null</code>.
+	 */
+	public static String escapeDouble(Object things) {
+		return things == null ? (String) null : things.toString().replace(DOUBLE_QUOTE, DOUBLE_SINGLE_QUOTE);
+	}
+
+	/**
+	 * Surrounds given object's {@link Object#toString()} with single quotes. Given <code>null</code>, returns
+	 * <code>null</code>.
+	 */
+	public static String singleQuote(Object thing) {
+		return thing == null ? (String) null : new StringBuilder().append(SINGLE_QUOTE).append(thing).append(SINGLE_QUOTE)
+				.toString();
+	}
+
+	/**
+	 * Surrounds given object's {@link Object#toString()} with double quotes. Given <code>null</code>, returns
+	 * <code>null</code>.
+	 */
+	public static String doubleQuote(Object thing) {
+		return thing == null ? (String) null : new StringBuilder().append(DOUBLE_QUOTE).append(thing).append(DOUBLE_QUOTE)
+				.toString();
+	}
+}

src/main/java/org/springframework/data/cassandra/cql/CreateTable.java

@@ -1,328 +0,0 @@
-package org.springframework.data.cassandra.cql;
-
-import static org.springframework.data.cassandra.cql.CreateTable.Column.Key.PARTITION;
-import static org.springframework.data.cassandra.cql.CreateTable.Column.Key.PRIMARY;
-import static org.springframework.data.cassandra.cql.CreateTable.Column.Order.ASCENDING;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.springframework.data.cassandra.cql.CreateTable.Column.Key;
-import org.springframework.data.cassandra.cql.CreateTable.Column.Order;
-import org.springframework.util.Assert;
-
-public class CreateTable {
-
-	protected static StringBuilder ensure(StringBuilder sb) {
-		return sb == null ? new StringBuilder() : sb;
-	}
-
-	public static class Column {
-
-		public enum Key {
-			PARTITION, PRIMARY
-		}
-
-		public enum Order {
-			ASCENDING("ASC"), DESCENDING("DESC");
-
-			private String cql;
-
-			private Order(String cql) {
-				this.cql = cql;
-			}
-
-			public String cql() {
-				return cql;
-			}
-		}
-
-		private String name;
-		private String type;
-		private Key key;
-		private Order order = ASCENDING;
-
-		public Column name(String name) {
-			Assert.hasLength(name);
-			this.name = name;
-			return this;
-		}
-
-		public Column type(String type) {
-			Assert.hasLength(type);
-			this.type = type;
-			return this;
-		}
-
-		public Column key(Key key) {
-			return key(key, ASCENDING);
-		}
-
-		public Column key(Key key, Order order) {
-			this.key = key;
-			this.order = order;
-			return this;
-		}
-
-		public void assertValid() {
-			// TODO
-		}
-
-		public StringBuilder cql(StringBuilder cql) {
-			return (cql = ensure(cql)).append(name).append(" ").append(type);
-		}
-
-		@Override
-		public String toString() {
-			return cql(null).toString();
-		}
-	}
-
-	private boolean ifNotExists = false;
-	private String name;
-	private List<Column> columns = new ArrayList<Column>();
-	private Map<String, Object> options = new HashMap<String, Object>();
-
-	public CreateTable ifNotExists() {
-		return ifNotExists(true);
-	}
-
-	public CreateTable ifNotExists(boolean ifNotExists) {
-		this.ifNotExists = ifNotExists;
-		return this;
-	}
-
-	public CreateTable name(String name) {
-		Assert.hasLength(name);
-		this.name = name;
-		return this;
-	}
-
-	public CreateTable options(Map<String, Object> options) {
-		this.options = options;
-		return this;
-	}
-
-	public CreateTable option(String name, Object value) {
-		options().put(name, value);
-		return this;
-	}
-
-	public CreateTable columns(List<Column> columns) {
-		columns().addAll(columns);
-		return this;
-	}
-
-	public CreateTable column(String name, String type) {
-		return column(name, type, null, null);
-	}
-
-	public CreateTable partition(String name, String type) {
-		return partition(name, type, null);
-	}
-
-	public CreateTable partition(String name, String type, Order order) {
-		return column(name, type, PARTITION, order);
-	}
-
-	public CreateTable primary(String name, String type) {
-		return primary(name, type, null);
-	}
-
-	public CreateTable primary(String name, String type, Order order) {
-		return column(name, type, PRIMARY, order);
-	}
-
-	public CreateTable column(String name, String type, Key key, Order order) {
-		columns().add(new Column().name(name).type(type).key(key, order));
-		return this;
-	}
-
-	protected List<Column> columns() {
-		return columns == null ? columns = new ArrayList<Column>() : columns;
-	}
-
-	protected Map<String, Object> options() {
-		return options == null ? options = new HashMap<String, Object>() : options;
-	}
-
-	public String cql() {
-		return cql(true);
-	}
-
-	protected String cql(boolean validate) {
-		if (validate) {
-			assertValid();
-		}
-
-		StringBuilder cql = new StringBuilder();
-
-		preamble(cql);
-		columnsAndOptions(cql);
-
-		cql.append(";");
-
-		return cql.toString();
-	}
-
-	protected StringBuilder preamble(StringBuilder cql) {
-		return (cql = ensure(cql)).append("CREATE TABLE ").append(ifNotExists ? "IF NOT EXISTS " : "").append(name);
-	}
-
-	@SuppressWarnings("unchecked")
-	protected StringBuilder columnsAndOptions(StringBuilder cql) {
-
-		cql = ensure(cql);
-
-		// begin columns
-		cql.append(" (");
-
-		List<Column> partitionKeys = new ArrayList<Column>();
-		List<Column> primaryKeys = new ArrayList<Column>();
-		for (Column col : columns) {
-			col.cql(cql).append(", ");
-
-			if (col.key == PARTITION) {
-				partitionKeys.add(col);
-			} else if (col.key == PRIMARY) {
-				primaryKeys.add(col);
-			}
-		}
-
-		// begin primary key clause
-		cql.append("PRIMARY KEY ");
-		StringBuilder partitions = new StringBuilder();
-		StringBuilder primaries = new StringBuilder();
-
-		if (partitionKeys.size() > 1) {
-			partitions.append("(");
-		}
-
-		StringBuilder clustering = null;
-
-		boolean clusteringFirst = true;
-		boolean first = true;
-		for (Column col : partitionKeys) {
-			if (first) {
-				first = false;
-			} else {
-				partitions.append(", ");
-			}
-			partitions.append(col.name);
-
-			if (col.order != null) { // then ordering specified
-				if (clustering == null) { // then initialize clustering clause
-					clustering = new StringBuilder().append("CLUSTERING ORDER BY (");
-				}
-				if (clusteringFirst) {
-					clusteringFirst = false;
-				} else {
-					clustering.append(", ");
-				}
-				clustering.append(col.name).append(" ").append(col.order.cql());
-			}
-		}
-		if (clustering != null) { // then end clustering option
-			clustering.append(")");
-		}
-		if (partitionKeys.size() > 1) {
-			partitions.append(")");
-		}
-
-		first = true;
-		for (Column col : primaryKeys) {
-			if (first) {
-				first = false;
-			} else {
-				primaries.append(", ");
-			}
-			primaries.append(col.name);
-		}
-		boolean parenthesize = partitionKeys.size() + primaryKeys.size() > 1;
-
-		cql.append(parenthesize ? "(" : "");
-		cql.append(partitions);
-		cql.append(primaryKeys.size() > 0 ? ", " : "");
-		cql.append(primaries);
-		cql.append(parenthesize ? ")" : "");
-		// end primary key clause
-		// end columns
-
-		// begin options
-		// begin option clause
-		if (clustering != null || !options.isEmpty()) {
-			// option preamble
-			first = true;
-
-			cql.append(" WITH ");
-
-			if (clustering != null) {
-				cql.append(clustering);
-				first = false;
-			}
-			if (!options.isEmpty()) {
-				for (String name : options.keySet()) {
-					// append AND if we're not on first option
-					if (first) {
-						first = false;
-					} else {
-						cql.append(" AND ");
-					}
-
-					// append <name> = <value>
-					cql.append(name);
-
-					Object value = options.get(name);
-					if (value == null) { // then assume string-only, valueless option like "COMPACT STORAGE"
-						continue;
-					}
-					if (value instanceof CharSequence) { // then value is a string
-						cql.append(" = '").append(value.toString()).append("'");
-						continue; // end string option
-					}
-
-					Map<String, Object> valueMap = null;
-					if ((value instanceof Map) && !(valueMap = (Map<String, Object>) value).isEmpty()) {
-						// then option value is a non-empty map
-
-						// append { 'name' : 'value', ... }
-						cql.append(" = { ");
-						boolean mapFirst = true;
-						for (Map.Entry<String, Object> entry : valueMap.entrySet()) {
-							if (mapFirst) {
-								mapFirst = false;
-							} else {
-								cql.append(", ");
-							}
-
-							cql.append("'").append(entry.getKey()).append("'"); // 'name'
-							cql.append(" : ");
-							Object entryValue = entry.getValue();
-							cql.append("'").append(entryValue == null ? "" : entryValue.toString()).append("'"); // 'value'
-						}
-						cql.append(" } ");
-
-						continue; // end non-empty value map
-					}
-
-					// else not a string, so just use unquoted string version of value
-					cql.append(value.toString());
-				}
-			}
-		}
-		// end options
-
-		return cql;
-	}
-
-	public void assertValid() {
-		// TODO
-	}
-
-	@Override
-	public String toString() {
-		return cql(false);
-	}
-}

src/main/java/org/springframework/data/cassandra/cql/builder/ColumnBuilder.java

@@ -0,0 +1,166 @@
+package org.springframework.data.cassandra.cql.builder;
+
+import static org.springframework.data.cassandra.cql.CqlStringUtils.checkQuotedIdentifier;
+import static org.springframework.data.cassandra.cql.CqlStringUtils.ensureNotNull;
+import static org.springframework.data.cassandra.mapping.KeyType.PARTITION;
+import static org.springframework.data.cassandra.mapping.KeyType.PRIMARY;
+import static org.springframework.data.cassandra.mapping.Ordering.ASCENDING;
+
+import org.springframework.data.cassandra.cql.CqlStringUtils;
+import org.springframework.data.cassandra.mapping.KeyType;
+import org.springframework.data.cassandra.mapping.Ordering;
+
+import com.datastax.driver.core.DataType;
+
+/**
+ * Builder class to help construct CQL statements that involve column manipulation. Not threadsafe.
+ * <p/>
+ * Use {@link #name(String)} and {@link #type(String)} to set the name and type of the column, respectively. To specify
+ * a <code>PRIMARY KEY</code> column, use {@link #primary()} or {@link #primary(Ordering)}. To specify that the
+ * <code>PRIMARY KEY</code> column is or is part of the partition key, use {@link #partition()} instead of
+ * {@link #primary()} or {@link #primary(Ordering)}.
+ * 
+ * @author Matthew T. Adams
+ */
+public class ColumnBuilder {
+
+	/**
+	 * Default ordering of primary key fields is {@link Ordering#ASCENDING}.
+	 */
+	public static final Ordering DFAULT_ORDERING = ASCENDING;
+
+	private String name;
+	private DataType type;
+	private KeyType keyType;
+	private Ordering ordering;
+
+	/**
+	 * Sets the column's name. Quotes are not escaped.
+	 * 
+	 * @see CqlStringUtils#escape(CharSequence)
+	 * @see CqlStringUtils#scrub(CharSequence)
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder name(String name) {
+		checkQuotedIdentifier(name);
+		this.name = name;
+		return this;
+	}
+
+	/**
+	 * Sets the column's type.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder type(DataType type) {
+		this.type = type;
+		return this;
+	}
+
+	/**
+	 * Identifies this column as a primary key column that is also part of a partition key. Sets the column's
+	 * {@link #keyType} to {@link KeyType#PARTITION} and its {@link #ordering} to <code>null</code>.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder partition() {
+		return partition(true);
+	}
+
+	/**
+	 * Toggles the identification of this column as a primary key column that also is or is part of a partition key. Sets
+	 * {@link #ordering} to <code>null</code> and, if the given boolean is <code>true</code>, then sets the column's
+	 * {@link #keyType} to {@link KeyType#PARTITION}, else sets it to <code>null</code>.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder partition(boolean partition) {
+		this.keyType = partition ? PARTITION : null;
+		this.ordering = null;
+		return this;
+	}
+
+	/**
+	 * Identifies this column as a primary key column with default ordering. Sets the column's {@link #keyType} to
+	 * {@link KeyType#PRIMARY} and its {@link #ordering} to {@link #DFAULT_ORDERING}.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder primary() {
+		return primary(DFAULT_ORDERING);
+	}
+
+	/**
+	 * Identifies this column as a primary key column with the given ordering. Sets the column's {@link #keyType} to
+	 * {@link KeyType#PRIMARY} and its {@link #ordering} to the given {@link Ordering}.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder primary(Ordering order) {
+		return primary(order, true);
+	}
+
+	/**
+	 * Toggles the identification of this column as a primary key column. If the given boolean is <code>true</code>, then
+	 * sets the column's {@link #keyType} to {@link KeyType#PARTITION} and {@link #ordering} to the given {@link Ordering}
+	 * , else sets both {@link #keyType} and {@link #ordering} to <code>null</code>.
+	 * 
+	 * @return this
+	 */
+	public ColumnBuilder primary(Ordering order, boolean primary) {
+		this.keyType = primary ? PRIMARY : null;
+		this.ordering = primary ? order : null;
+		return this;
+	}
+
+	/**
+	 * Sets the column's {@link #keyType}.
+	 * 
+	 * @return this
+	 */
+	/* package */ColumnBuilder keyType(KeyType keyType) {
+		this.keyType = keyType;
+		return this;
+	}
+
+	/**
+	 * Sets the column's {@link #ordering}.
+	 * 
+	 * @return this
+	 */
+	/* package */ColumnBuilder ordering(Ordering ordering) {
+		this.ordering = ordering;
+		return this;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public DataType getType() {
+		return type;
+	}
+
+	public KeyType getKeyType() {
+		return keyType;
+	}
+
+	public Ordering getOrdering() {
+		return ordering;
+	}
+
+	public String toCql() {
+		return toCql(null).toString();
+	}
+
+	public StringBuilder toCql(StringBuilder cql) {
+		return (cql = ensureNotNull(cql)).append(name).append(" ").append(type);
+	}
+
+	@Override
+	public String toString() {
+		return toCql(null).append(" /* keyType=").append(keyType).append(", ordering=").append(ordering).append(" */ ")
+				.toString();
+	}
+}

src/main/java/org/springframework/data/cassandra/cql/builder/CqlBuilder.java

@@ -0,0 +1,11 @@
+package org.springframework.data.cassandra.cql.builder;
+
+public class CqlBuilder {
+
+	/**
+	 * Entry point into the {@link CqlBuilder}'s fluent API to create a table. Convenient if imported statically.
+	 */
+	public static CreateTableBuilder createTable() {
+		return new CreateTableBuilder();
+	}
+}

src/main/java/org/springframework/data/cassandra/cql/builder/CreateTableBuilder.java

@@ -0,0 +1,338 @@
+package org.springframework.data.cassandra.cql.builder;
+
+import static org.springframework.data.cassandra.cql.CqlStringUtils.checkQuotedIdentifier;
+import static org.springframework.data.cassandra.cql.CqlStringUtils.ensureNotNull;
+import static org.springframework.data.cassandra.cql.CqlStringUtils.escapeSingle;
+import static org.springframework.data.cassandra.cql.CqlStringUtils.singleQuote;
+import static org.springframework.data.cassandra.mapping.KeyType.PARTITION;
+import static org.springframework.data.cassandra.mapping.KeyType.PRIMARY;
+import static org.springframework.data.cassandra.mapping.Ordering.ASCENDING;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.data.cassandra.cql.CqlStringUtils;
+import org.springframework.data.cassandra.mapping.KeyType;
+import org.springframework.data.cassandra.mapping.Ordering;
+
+import com.datastax.driver.core.DataType;
+
+/**
+ * Builder class to construct CQL for a <code>CREATE TABLE</code> statement. Not threadsafe.
+ * 
+ * @author Matthew T. Adams
+ */
+public class CreateTableBuilder {
+
+	private boolean ifNotExists = false;
+	private String name;
+	private List<ColumnBuilder> columns = new ArrayList<ColumnBuilder>();
+	private Map<String, Object> options = new HashMap<String, Object>();
+
+	/**
+	 * Causes the inclusion of an <code>IF NOT EXISTS</code> clause.
+	 * 
+	 * @return this
+	 */
+	public CreateTableBuilder ifNotExists() {
+		return ifNotExists(true);
+	}
+
+	/**
+	 * Toggles the inclusion of an <code>IF NOT EXISTS</code> clause.
+	 * 
+	 * @return this
+	 */
+	public CreateTableBuilder ifNotExists(boolean ifNotExists) {
+		this.ifNotExists = ifNotExists;
+		return this;
+	}
+
+	/**
+	 * Sets the table name. Quotes are not escaped.
+	 * 
+	 * @see CqlStringUtils#escape(CharSequence)
+	 * @see CqlStringUtils#scrub(CharSequence)
+	 * 
+	 * @return this
+	 */
+	public CreateTableBuilder name(String name) {
+		checkQuotedIdentifier(name);
+		this.name = name;
+		return this;
+	}
+
+	/**
+	 * Adds the given single-string option with no value to this table's options. Convenient overload of
+	 * <code>with(string, null, false, false)</code>.
+	 * 
+	 * @param singleStringOption
+	 * @return
+	 */
+	public CreateTableBuilder with(String string) {
+		return with(string, null, false, false);
+	}
+
+	/**
+	 * Adds the given single-quote-escaped then single-quoted option value by name to this table's options. Convenient
+	 * overload of <code>with(name, value, true, true)</code>
+	 * 
+	 * @see #with(String, Object, boolean, boolean)
+	 * @return this
+	 */
+	public CreateTableBuilder withQuoted(String name, Object value) {
+		return with(name, value, true, true);
+	}
+
+	/**
+	 * Adds the given option value by name with no quoting or escaping to this table's options. Convenient overload of
+	 * <code>with(name, value, false, false)</code>
+	 * 
+	 * @see #with(String, Object, boolean, boolean)
+	 * @return this
+	 */
+	public CreateTableBuilder withUnquoted(String name, Object value) {
+		return with(name, value, false, false);
+	}
+
+	public CreateTableBuilder with(String name, Map<String, Object> valueMap) {
+		return with(name, valueMap, false, false);
+	}
+
+	/**
+	 * Adds the given option by name to this table's options.
+	 * <p/>
+	 * Options that have <code>null</code> values are considered single string options where the name of the option is the
+	 * string to be used. Otherwise, the result of {@link Object#toString()} is considered to be the value of the option
+	 * with the given name. The value, after conversion to string, may have embedded single quotes escaped according to
+	 * parameter <code>escape</code> and may be single-quoted according to parameter <code>quote</code>.
+	 * 
+	 * @param name The name of the option
+	 * @param value The value of the option. If <code>null</code>, the value is ignored and the option is considered to be
+	 *          composed of only the name, otherwise the value's {@link Object#toString()} value is used.
+	 * @param escape Whether to escape the value via {@link CqlStringUtils#escapeSingle(Object)}. Ignored if given value
+	 *          is an instance of a {@link Map}.
+	 * @param quote Whether to quote the value via {@link CqlStringUtils#singleQuote(Object)}. Ignored if given value is
+	 *          an instance of a {@link Map}.
+	 * @return this
+	 */
+	public CreateTableBuilder with(String name, Object value, boolean escape, boolean quote) {
+		if (!(value instanceof Map)) {
+			if (escape) {
+				value = escapeSingle(value);
+			}
+			if (quote) {
+				value = singleQuote(value);
+			}
+		}
+		options().put(name, value);
+		return this;
+	}
+
+	public CreateTableBuilder column(String name, DataType type) {
+		return column(name, type, null, null);
+	}
+
+	public CreateTableBuilder partitionKeyColumn(String name, DataType type) {
+		return column(name, type, PARTITION, null);
+	}
+
+	public CreateTableBuilder primaryKeyColumn(String name, DataType type) {
+		return primaryKeyColumn(name, type, ASCENDING);
+	}
+
+	public CreateTableBuilder primaryKeyColumn(String name, DataType type, Ordering order) {
+		return column(name, type, PRIMARY, order);
+	}
+
+	protected CreateTableBuilder column(String name, DataType type, KeyType keyType, Ordering ordering) {
+		columns().add(new ColumnBuilder().name(name).type(type).keyType(keyType).ordering(ordering));
+		return this;
+	}
+
+	/**
+	 * Convenient method that calls <code>with("COMPACT STORAGE", null)</code>.
+	 * 
+	 * @see #with(String, Object)
+	 * @return this
+	 */
+	public CreateTableBuilder withCompactStorage() {
+		return with("COMPACT STORAGE");
+	}
+
+	protected List<ColumnBuilder> columns() {
+		return columns == null ? columns = new ArrayList<ColumnBuilder>() : columns;
+	}
+
+	protected Map<String, Object> options() {
+		return options == null ? options = new HashMap<String, Object>() : options;
+	}
+
+	public String toCql() {
+
+		StringBuilder cql = new StringBuilder();
+
+		preambleCql(cql);
+		columnsAndOptionsCql(cql);
+
+		cql.append(";");
+
+		return cql.toString();
+	}
+
+	protected StringBuilder preambleCql(StringBuilder cql) {
+		return (cql = ensureNotNull(cql)).append("CREATE TABLE ").append(ifNotExists ? "IF NOT EXISTS " : "").append(name);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected StringBuilder columnsAndOptionsCql(StringBuilder cql) {
+
+		cql = ensureNotNull(cql);
+
+		// begin columns
+		cql.append(" (");
+
+		List<ColumnBuilder> partitionKeys = new ArrayList<ColumnBuilder>();
+		List<ColumnBuilder> primaryKeys = new ArrayList<ColumnBuilder>();
+		for (ColumnBuilder col : columns) {
+			col.toCql(cql).append(", ");
+
+			if (col.getKeyType() == PARTITION) {
+				partitionKeys.add(col);
+			} else if (col.getKeyType() == PRIMARY) {
+				primaryKeys.add(col);
+			}
+		}
+
+		// begin primary key clause
+		cql.append("PRIMARY KEY ");
+		StringBuilder partitions = new StringBuilder();
+		StringBuilder primaries = new StringBuilder();
+
+		if (partitionKeys.size() > 1) {
+			partitions.append("(");
+		}
+
+		boolean first = true;
+		for (ColumnBuilder col : partitionKeys) {
+			if (first) {
+				first = false;
+			} else {
+				partitions.append(", ");
+			}
+			partitions.append(col.getName());
+
+		}
+		if (partitionKeys.size() > 1) {
+			partitions.append(")");
+		}
+
+		StringBuilder clustering = null;
+		boolean clusteringFirst = true;
+		first = true;
+		for (ColumnBuilder col : primaryKeys) {
+			if (first) {
+				first = false;
+			} else {
+				primaries.append(", ");
+			}
+			primaries.append(col.getName());
+
+			if (col.getOrdering() != null) { // then ordering specified
+				if (clustering == null) { // then initialize clustering clause
+					clustering = new StringBuilder().append("CLUSTERING ORDER BY (");
+				}
+				if (clusteringFirst) {
+					clusteringFirst = false;
+				} else {
+					clustering.append(", ");
+				}
+				clustering.append(col.getName()).append(" ").append(col.getOrdering().cql());
+			}
+		}
+		if (clustering != null) { // then end clustering option
+			clustering.append(")");
+		}
+
+		boolean parenthesize = partitionKeys.size() + primaryKeys.size() > 1;
+
+		cql.append(parenthesize ? "(" : "");
+		cql.append(partitions);
+		cql.append(primaryKeys.size() > 0 ? ", " : "");
+		cql.append(primaries);
+		cql.append(parenthesize ? ")" : "");
+		// end primary key clause
+		// end columns
+
+		// begin options
+		// begin option clause
+		if (clustering != null || !options.isEmpty()) {
+			// option preamble
+			first = true;
+
+			cql.append(" WITH ");
+
+			if (clustering != null) {
+				cql.append(clustering);
+				first = false;
+			}
+			if (!options.isEmpty()) {
+				for (String name : options.keySet()) {
+					// append AND if we're not on first option
+					if (first) {
+						first = false;
+					} else {
+						cql.append(" AND ");
+					}
+
+					// append <name> = <value>
+					cql.append(name);
+
+					Object value = options.get(name);
+					if (value == null) { // then assume string-only, valueless option like "COMPACT STORAGE"
+						continue;
+					}
+
+					cql.append(" = ");
+
+					Map<String, Object> valueMap = null;
+					if ((value instanceof Map) && !(valueMap = (Map<String, Object>) value).isEmpty()) {
+						// then option value is a non-empty map
+
+						// append { 'name' : 'value', ... }
+						cql.append("{ ");
+						boolean mapFirst = true;
+						for (Map.Entry<String, Object> entry : valueMap.entrySet()) {
+							if (mapFirst) {
+								mapFirst = false;
+							} else {
+								cql.append(", ");
+							}
+
+							cql.append(singleQuote(entry.getKey())); // 'name'
+							cql.append(" : ");
+							Object entryValue = entry.getValue();
+							cql.append(singleQuote(entryValue == null ? "" : entryValue.toString())); // 'value'
+						}
+						cql.append(" }");
+
+						continue; // end non-empty value map
+					}
+
+					// else just use value as string
+					cql.append(value.toString());
+				}
+			}
+		}
+		// end options
+
+		return cql;
+	}
+
+	@Override
+	public String toString() {
+		return toCql();
+	}
+}

src/main/java/org/springframework/data/cassandra/mapping/KeyType.java

@@ -0,0 +1,19 @@
+package org.springframework.data.cassandra.mapping;
+
+/**
+ * Values representing primary key column types.
+ * 
+ * @author Matthew T. Adams
+ */
+public enum KeyType {
+
+	/**
+	 * Used for a column that is a primary key that also is or is part of the partition key.
+	 */
+	PARTITION,
+
+	/**
+	 * Use for a primary key column that is not part of the partition key and, therefore, may also be ordered.
+	 */
+	PRIMARY
+}

src/main/java/org/springframework/data/cassandra/mapping/Ordering.java

@@ -0,0 +1,32 @@
+package org.springframework.data.cassandra.mapping;
+
+/**
+ * Enum for Cassandra primary key column ordering.
+ * 
+ * @author Matthew T. Adams
+ */
+public enum Ordering {
+
+	/**
+	 * Ascending Cassandra column ordering.
+	 */
+	ASCENDING("ASC"),
+
+	/**
+	 * Descending Cassandra column ordering.
+	 */
+	DESCENDING("DESC");
+
+	private String cql;
+
+	private Ordering(String cql) {
+		this.cql = cql;
+	}
+
+	/**
+	 * Returns the CQL keyword of this {@link Ordering}.
+	 */
+	public String cql() {
+		return cql;
+	}
+}

src/main/java/org/springframework/data/cassandra/mapping/RowId.java

@@ -31,5 +31,4 @@ import org.springframework.data.annotation.Id;
 @Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
 @Id
 public @interface RowId {
-
 }

src/test/java/org/springframework/data/cassandra/cql/CreateTableBuilderTest.java

@@ -1,28 +1,28 @@
 package org.springframework.data.cassandra.cql;
 
-import static org.springframework.data.cassandra.cql.CqlBuilder.createTable;
+import static org.springframework.data.cassandra.cql.builder.CqlBuilder.createTable;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.junit.Test;
-import org.springframework.data.cassandra.cql.CreateTable.Column.Order;
+import org.springframework.data.cassandra.cql.builder.CreateTableBuilder;
 
-public class CreateTableTest {
+import com.datastax.driver.core.DataType;
+
+public class CreateTableBuilderTest {
 
 	@Test
 	public void createTableTest() {
 		String name = "mytable";
-		String type0 = "text";
+		DataType type0 = DataType.text();
 		String partition0 = "partitionKey0";
 		String partition1 = "partitionKey1";
 		String primary0 = "primary0";
-		String type1 = "text";
+		DataType type1 = DataType.text();
 		String column1 = "column1";
-		String type2 = "text";
+		DataType type2 = DataType.bigint();
 		String column2 = "column2";
-		Object value1 = null;
-		String option1 = "COMPACT STORAGE";
 		Object value2 = "this is a comment";
 		String option2 = "comment";
 		Object value3 = "0.00075";
@@ -31,12 +31,12 @@ public class CreateTableTest {
 		value4.put("class", "LeveledCompactionStrategy");
 		String option4 = "compaction";
 
-		CreateTable builder = createTable().ifNotExists().name(name).partition(partition0, type0, Order.ASCENDING)
-				.partition(partition1, type0, Order.DESCENDING).primary(primary0, type0).column(column1, type1)
-				.column(column2, type2).option(option1, value1).option(option2, value2).option(option3, value3)
-				.option(option4, value4);
+		CreateTableBuilder builder = createTable().ifNotExists().name(name).partitionKeyColumn(partition0, type0)
+				.partitionKeyColumn(partition1, type0).primaryKeyColumn(primary0, type0).column(column1, type1)
+				.column(column2, type2).withQuoted(option2, value2).withUnquoted(option3, value3).with(option4, value4)
+				.withCompactStorage();
 
-		String cql = builder.cql();
+		String cql = builder.toCql();
 		System.out.println(cql);
 	}
 }