testing coming along

src/main/java/org/springframework/cassandra/core/cql/generator/AlterTableCqlGenerator.java

@@ -16,7 +16,7 @@ import org.springframework.cassandra.core.keyspace.Option;
  * 
  * @author Matthew T. Adams
  */
-public class AlterTableCqlGenerator extends AbstractTableOperationCqlGenerator<AlterTableSpecification> {
+public class AlterTableCqlGenerator extends TableOptionsCqlGenerator<AlterTableSpecification> {
 
 	public AlterTableCqlGenerator(AlterTableSpecification specification) {
 		super(specification);

src/main/java/org/springframework/cassandra/core/cql/generator/CreateTableCqlGenerator.java

@@ -17,7 +17,7 @@ import org.springframework.cassandra.core.keyspace.Option;
  * 
  * @author Matthew T. Adams
  */
-public class CreateTableCqlGenerator extends AbstractTableOperationCqlGenerator<CreateTableSpecification> {
+public class CreateTableCqlGenerator extends TableCqlGenerator<CreateTableSpecification> {
 
 	public CreateTableCqlGenerator(CreateTableSpecification specification) {
 		super(specification);
@@ -110,7 +110,7 @@ public class CreateTableCqlGenerator extends AbstractTableOperationCqlGenerator<
 			clustering.append(")");
 		}
 
-		boolean parenthesize = partitionKeys.size() + primaryKeys.size() > 1;
+		boolean parenthesize = true;// partitionKeys.size() + primaryKeys.size() > 1;
 
 		cql.append(parenthesize ? "(" : "");
 		cql.append(partitions);
@@ -118,6 +118,8 @@ public class CreateTableCqlGenerator extends AbstractTableOperationCqlGenerator<
 		cql.append(primaries);
 		cql.append(parenthesize ? ")" : "");
 		// end primary key clause
+
+		cql.append(")");
 		// end columns
 
 		// begin options

src/main/java/org/springframework/cassandra/core/cql/generator/DropTableCqlGenerator.java

@@ -3,32 +3,20 @@ package org.springframework.cassandra.core.cql.generator;
 import static org.springframework.cassandra.core.cql.CqlStringUtils.noNull;
 
 import org.springframework.cassandra.core.keyspace.DropTableSpecification;
-import org.springframework.util.Assert;
 
 /**
  * CQL generator for generating a <code>DROP TABLE</code> statement.
  * 
  * @author Matthew T. Adams
  */
-public class DropTableCqlGenerator {
-
-	protected DropTableSpecification specification;
+public class DropTableCqlGenerator extends TableNameCqlGenerator<DropTableSpecification> {
 
 	public DropTableCqlGenerator(DropTableSpecification specification) {
-		setSpecification(specification);
-	}
-
-	protected void setSpecification(DropTableSpecification specification) {
-		Assert.notNull(specification);
-		this.specification = specification;
+		super(specification);
 	}
 
 	public StringBuilder toCql(StringBuilder cql) {
-		return noNull(cql).append("DROP TABLE ").append(specification.getIfExists() ? "IF EXISTS " : "")
-				.append(specification.getNameAsIdentifier());
-	}
-
-	public String toCql() {
-		return toCql(null).toString();
+		return noNull(cql).append("DROP TABLE ").append(spec().getIfExists() ? "IF EXISTS " : "")
+				.append(spec().getNameAsIdentifier()).append(";");
 	}
 }

src/main/java/org/springframework/cassandra/core/cql/generator/TableCqlGenerator.java

@@ -6,9 +6,8 @@ import static org.springframework.cassandra.core.cql.CqlStringUtils.singleQuote;
 
 import java.util.Map;
 
-import org.springframework.cassandra.core.keyspace.AbstractTableSpecification;
 import org.springframework.cassandra.core.keyspace.Option;
-import org.springframework.util.Assert;
+import org.springframework.cassandra.core.keyspace.TableSpecification;
 
 /**
  * Base class that contains behavior common to CQL generation for table operations.
@@ -16,31 +15,16 @@ import org.springframework.util.Assert;
  * @author Matthew T. Adams
  * @param T The subtype of this class for which this is a CQL generator.
  */
-public abstract class AbstractTableOperationCqlGenerator<T extends AbstractTableSpecification<T>> {
+public abstract class TableCqlGenerator<T extends TableSpecification<T>> extends
+		TableOptionsCqlGenerator<TableSpecification<T>> {
 
-	public abstract StringBuilder toCql(StringBuilder cql);
-
-	private AbstractTableSpecification<T> specification;
-
-	public AbstractTableOperationCqlGenerator(AbstractTableSpecification<T> specification) {
-		setSpecification(specification);
-	}
-
-	protected void setSpecification(AbstractTableSpecification<T> specification) {
-		Assert.notNull(specification);
-		this.specification = specification;
+	public TableCqlGenerator(TableSpecification<T> specification) {
+		super(specification);
 	}
 
 	@SuppressWarnings("unchecked")
-	public T getSpecification() {
-		return (T) specification;
-	}
-
-	/**
-	 * Convenient synonymous method of {@link #getSpecification()}.
-	 */
 	protected T spec() {
-		return getSpecification();
+		return (T) getSpecification();
 	}
 
 	protected StringBuilder optionValueMap(Map<Option, Object> valueMap, StringBuilder cql) {
@@ -78,8 +62,4 @@ public abstract class AbstractTableOperationCqlGenerator<T extends AbstractTable
 
 		return cql;
 	}
-
-	public String toCql() {
-		return toCql(null).toString();
-	}
 }

src/main/java/org/springframework/cassandra/core/cql/generator/TableNameCqlGenerator.java

@@ -0,0 +1,36 @@
+package org.springframework.cassandra.core.cql.generator;
+
+import org.springframework.cassandra.core.keyspace.TableNameSpecification;
+import org.springframework.util.Assert;
+
+public abstract class TableNameCqlGenerator<T extends TableNameSpecification<T>> {
+
+	public abstract StringBuilder toCql(StringBuilder cql);
+
+	private TableNameSpecification<T> specification;
+
+	public TableNameCqlGenerator(TableNameSpecification<T> specification) {
+		setSpecification(specification);
+	}
+
+	protected void setSpecification(TableNameSpecification<T> specification) {
+		Assert.notNull(specification);
+		this.specification = specification;
+	}
+
+	@SuppressWarnings("unchecked")
+	public T getSpecification() {
+		return (T) specification;
+	}
+
+	/**
+	 * Convenient synonymous method of {@link #getSpecification()}.
+	 */
+	protected T spec() {
+		return getSpecification();
+	}
+
+	public String toCql() {
+		return toCql(null).toString();
+	}
+}

src/main/java/org/springframework/cassandra/core/cql/generator/TableOptionsCqlGenerator.java

@@ -0,0 +1,65 @@
+package org.springframework.cassandra.core.cql.generator;
+
+import static org.springframework.cassandra.core.cql.CqlStringUtils.escapeSingle;
+import static org.springframework.cassandra.core.cql.CqlStringUtils.noNull;
+import static org.springframework.cassandra.core.cql.CqlStringUtils.singleQuote;
+
+import java.util.Map;
+
+import org.springframework.cassandra.core.keyspace.Option;
+import org.springframework.cassandra.core.keyspace.TableOptionsSpecification;
+
+/**
+ * Base class that contains behavior common to CQL generation for table operations.
+ * 
+ * @author Matthew T. Adams
+ * @param T The subtype of this class for which this is a CQL generator.
+ */
+public abstract class TableOptionsCqlGenerator<T extends TableOptionsSpecification<T>> extends
+		TableNameCqlGenerator<TableOptionsSpecification<T>> {
+
+	public TableOptionsCqlGenerator(TableOptionsSpecification<T> specification) {
+		super(specification);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected T spec() {
+		return (T) getSpecification();
+	}
+
+	protected StringBuilder optionValueMap(Map<Option, Object> valueMap, StringBuilder cql) {
+		cql = noNull(cql);
+
+		if (valueMap == null || valueMap.isEmpty()) {
+			return cql;
+		}
+		// else option value is a non-empty map
+
+		// append { 'name' : 'value', ... }
+		cql.append("{ ");
+		boolean mapFirst = true;
+		for (Map.Entry<Option, Object> entry : valueMap.entrySet()) {
+			if (mapFirst) {
+				mapFirst = false;
+			} else {
+				cql.append(", ");
+			}
+
+			Option option = entry.getKey();
+			cql.append(singleQuote(option.getName())); // entries in map keys are always quoted
+			cql.append(" : ");
+			Object entryValue = entry.getValue();
+			entryValue = entryValue == null ? "" : entryValue.toString();
+			if (option.escapesValue()) {
+				entryValue = escapeSingle(entryValue);
+			}
+			if (option.quotesValue()) {
+				entryValue = singleQuote(entryValue);
+			}
+			cql.append(entryValue);
+		}
+		cql.append(" }");
+
+		return cql;
+	}
+}

src/main/java/org/springframework/cassandra/core/keyspace/AlterTableSpecification.java

@@ -6,25 +6,45 @@ import java.util.List;
 
 import com.datastax.driver.core.DataType;
 
-public class AlterTableSpecification extends AbstractTableSpecification<AlterTableSpecification> {
+/**
+ * Builder class to construct an <code>ALTER TABLE</code> specification.
+ * 
+ * @author Matthew T. Adams
+ */
+public class AlterTableSpecification extends TableOptionsSpecification<AlterTableSpecification> {
 
+	/**
+	 * The list of column changes.
+	 */
 	private List<ColumnChangeSpecification> changes = new ArrayList<ColumnChangeSpecification>();
 
+	/**
+	 * Adds a <code>DROP</code> to the list of column changes.
+	 */
 	public AlterTableSpecification drop(String column) {
 		changes.add(new DropColumnSpecification(column));
 		return this;
 	}
 
+	/**
+	 * Adds an <code>ADD</code> to the list of column changes.
+	 */
 	public AlterTableSpecification add(String column, DataType type) {
 		changes.add(new AddColumnSpecification(column, type));
 		return this;
 	}
 
+	/**
+	 * Adds an <code>ALTER</code> to the list of column changes.
+	 */
 	public AlterTableSpecification alter(String column, DataType type) {
 		changes.add(new AlterColumnSpecification(column, type));
 		return this;
 	}
 
+	/**
+	 * Returns an unmodifiable list of column changes.
+	 */
 	public List<ColumnChangeSpecification> getChanges() {
 		return Collections.unmodifiableList(changes);
 	}

src/main/java/org/springframework/cassandra/core/keyspace/CreateTableSpecification.java

@@ -1,27 +1,13 @@
 package org.springframework.cassandra.core.keyspace;
 
-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.Collections;
-import java.util.List;
-
-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.
+ * Builder class to construct a <code>CREATE TABLE</code> specification.
  * 
  * @author Matthew T. Adams
  */
-public class CreateTableSpecification extends AbstractTableSpecification<CreateTableSpecification> {
+public class CreateTableSpecification extends TableSpecification<CreateTableSpecification> {
 
 	private boolean ifNotExists = false;
-	private List<ColumnSpecification> columns = new ArrayList<ColumnSpecification>();
 
 	/**
 	 * Causes the inclusion of an <code>IF NOT EXISTS</code> clause.
@@ -42,36 +28,7 @@ public class CreateTableSpecification extends AbstractTableSpecification<CreateT
 		return this;
 	}
 
-	public CreateTableSpecification column(String name, DataType type) {
-		return column(name, type, null, null);
-	}
-
-	public CreateTableSpecification partitionKeyColumn(String name, DataType type) {
-		return column(name, type, PARTITION, null);
-	}
-
-	public CreateTableSpecification primaryKeyColumn(String name, DataType type) {
-		return primaryKeyColumn(name, type, ASCENDING);
-	}
-
-	public CreateTableSpecification primaryKeyColumn(String name, DataType type, Ordering order) {
-		return column(name, type, PRIMARY, order);
-	}
-
-	protected CreateTableSpecification column(String name, DataType type, KeyType keyType, Ordering ordering) {
-		columns().add(new ColumnSpecification().name(name).type(type).keyType(keyType).ordering(ordering));
-		return this;
-	}
-
-	protected List<ColumnSpecification> columns() {
-		return columns == null ? columns = new ArrayList<ColumnSpecification>() : columns;
-	}
-
 	public boolean getIfNotExists() {
 		return ifNotExists;
 	}
-
-	public List<ColumnSpecification> getColumns() {
-		return Collections.unmodifiableList(columns);
-	}
 }

src/main/java/org/springframework/cassandra/core/keyspace/DefaultTableDescriptor.java

@@ -0,0 +1,17 @@
+package org.springframework.cassandra.core.keyspace;
+
+/**
+ * Convenient default implementation of {@link TableDescriptor} as an extension of {@link TableSpecification} that
+ * doesn't require the use of generics.
+ * 
+ * @author Matthew T. Adams
+ */
+public class DefaultTableDescriptor extends TableSpecification<DefaultTableDescriptor> {
+
+	/**
+	 * Factory method to produce a new {@link DefaultTableDescriptor}. Convenient if imported statically.
+	 */
+	public static DefaultTableDescriptor table() {
+		return new DefaultTableDescriptor();
+	}
+}

src/main/java/org/springframework/cassandra/core/keyspace/DropTableSpecification.java

@@ -1,19 +1,14 @@
 package org.springframework.cassandra.core.keyspace;
 
-import static org.springframework.cassandra.core.cql.CqlStringUtils.checkIdentifier;
-import static org.springframework.cassandra.core.cql.CqlStringUtils.identifize;
+/**
+ * Builder class that supports the construction of <code>DROP TABLE</code> specifications.
+ * 
+ * @author Matthew T. Adams
+ */
+public class DropTableSpecification extends TableNameSpecification<DropTableSpecification> {
 
-public class DropTableSpecification {
-
-	private String name;
 	private boolean ifExists;
 
-	public DropTableSpecification name(String name) {
-		checkIdentifier(name);
-		this.name = name;
-		return this;
-	}
-
 	public DropTableSpecification ifExists() {
 		return ifExists(true);
 	}
@@ -26,8 +21,4 @@ public class DropTableSpecification {
 	public boolean getIfExists() {
 		return ifExists;
 	}
-
-	public String getNameAsIdentifier() {
-		return identifize(name);
-	}
 }

src/main/java/org/springframework/cassandra/core/keyspace/TableDescriptor.java

@@ -0,0 +1,52 @@
+package org.springframework.cassandra.core.keyspace;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Describes a table.
+ * 
+ * @author Matthew T. Adams
+ */
+public interface TableDescriptor {
+
+	/**
+	 * Returns the name of the table.
+	 */
+	String getName();
+
+	/**
+	 * Returns the name of the table as an identifer or quoted identifier as appropriate.
+	 */
+	String getNameAsIdentifier();
+
+	/**
+	 * Returns an unmodifiable {@link List} of {@link ColumnSpecification}s.
+	 */
+	List<ColumnSpecification> getColumns();
+
+	/**
+	 * Returns an unmodifiable list of all partition key columns.
+	 */
+	public List<ColumnSpecification> getPartitionKeyColumns();
+
+	/**
+	 * Returns an unmodifiable list of all primary key columns that are not also partition key columns.
+	 */
+	public List<ColumnSpecification> getPrimaryKeyColumns();
+
+	/**
+	 * Returns an unmodifiable list of all partition and primary key columns.
+	 */
+	public List<ColumnSpecification> getKeyColumns();
+
+	/**
+	 * Returns an unmodifiable list of all non-key columns.
+	 */
+	public List<ColumnSpecification> getNonKeyColumns();
+
+	/**
+	 * Returns an unmodifiable {@link Map} of table options.
+	 */
+	Map<String, Object> getOptions();
+}

src/main/java/org/springframework/cassandra/core/keyspace/TableNameSpecification.java

@@ -0,0 +1,38 @@
+package org.springframework.cassandra.core.keyspace;
+
+import static org.springframework.cassandra.core.cql.CqlStringUtils.checkIdentifier;
+import static org.springframework.cassandra.core.cql.CqlStringUtils.identifize;
+
+/**
+ * Abstract builder class to support the construction of table specifications.
+ * 
+ * @author Matthew T. Adams
+ * @param <T> The subtype of the {@link TableNameSpecification}
+ */
+public abstract class TableNameSpecification<T extends TableNameSpecification<T>> {
+
+	/**
+	 * The name of the table.
+	 */
+	private String name;
+
+	/**
+	 * Sets the table name.
+	 * 
+	 * @return this
+	 */
+	@SuppressWarnings("unchecked")
+	public T name(String name) {
+		checkIdentifier(name);
+		this.name = name;
+		return (T) this;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public String getNameAsIdentifier() {
+		return identifize(name);
+	}
+}

src/main/java/org/springframework/cassandra/core/keyspace/TableOption.java

@@ -122,9 +122,13 @@ public enum TableOption implements Option {
 			this.value = value;
 		}
 
-		public String toString() {
+		public String getValue() {
 			return value;
 		}
+
+		public String toString() {
+			return getValue();
+		}
 	}
 
 	/**

src/main/java/org/springframework/cassandra/core/keyspace/TableOptionsSpecification.java

@@ -1,8 +1,6 @@
 package org.springframework.cassandra.core.keyspace;
 
-import static org.springframework.cassandra.core.cql.CqlStringUtils.checkIdentifier;
 import static org.springframework.cassandra.core.cql.CqlStringUtils.escapeSingle;
-import static org.springframework.cassandra.core.cql.CqlStringUtils.identifize;
 import static org.springframework.cassandra.core.cql.CqlStringUtils.singleQuote;
 
 import java.util.Collections;
@@ -12,39 +10,26 @@ import java.util.Map;
 import org.springframework.cassandra.core.cql.CqlStringUtils;
 
 /**
- * Base class that contains behavior common to table operations.
+ * Abstract builder class to support the construction of table specifications that have table options, that is, those
+ * options normally specified by <code>WITH ... AND ...</code>.
+ * <p/>
+ * It is important to note that although this class depends on {@link TableOption} for convenient and typesafe use, it
+ * ultimately stores its options in a <code>Map<String,Object></code> for flexibility. This means that
+ * {@link #with(TableOption)} and {@link #with(TableOption, Object)} delegate to
+ * {@link #with(String, Object, boolean, boolean)}. This design allows the API to support new Cassandra options as they
+ * are introduced without having to update the code immediately.
  * 
  * @author Matthew T. Adams
- * @param T The subtype of AbstractTableBuilder.
+ * @param <T> The subtype of the {@link TableOptionsSpecification}.
  */
-public abstract class AbstractTableSpecification<T extends AbstractTableSpecification<T>> {
-
-	private String name;
+public abstract class TableOptionsSpecification<T extends TableOptionsSpecification<T>> extends
+		TableNameSpecification<TableOptionsSpecification<T>> {
 
 	protected Map<String, Object> options = new LinkedHashMap<String, Object>();
 
-	/**
-	 * Sets the table name.
-	 * 
-	 * @return this
-	 */
 	@SuppressWarnings("unchecked")
 	public T name(String name) {
-		setName(name);
-		return (T) this;
-	}
-
-	public void setName(String name) {
-		checkIdentifier(name);
-		this.name = name;
-	}
-
-	public String getName() {
-		return name;
-	}
-
-	public String getNameAsIdentifier() {
-		return identifize(name);
+		return (T) super.name(name);
 	}
 
 	/**

src/main/java/org/springframework/cassandra/core/keyspace/TableSpecification.java

@@ -0,0 +1,162 @@
+package org.springframework.cassandra.core.keyspace;
+
+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.Collections;
+import java.util.List;
+
+import org.springframework.data.cassandra.mapping.KeyType;
+import org.springframework.data.cassandra.mapping.Ordering;
+
+import com.datastax.driver.core.DataType;
+
+/**
+ * Builder class to support the construction of table specifications that have columns. This class can also be used as a
+ * standalone {@link TableDescriptor}, independent of {@link CreateTableSpecification}.
+ * 
+ * @author Matthew T. Adams
+ */
+public class TableSpecification<T> extends TableOptionsSpecification<TableSpecification<T>> implements TableDescriptor {
+
+	/**
+	 * List of all columns.
+	 */
+	private List<ColumnSpecification> columns = new ArrayList<ColumnSpecification>();
+
+	/**
+	 * List of only those columns that comprise the partition key.
+	 */
+	private List<ColumnSpecification> partitionKeyColumns = new ArrayList<ColumnSpecification>();
+
+	/**
+	 * List of only those columns that comprise the primary key that are not also part of the partition key.
+	 */
+	private List<ColumnSpecification> primaryKeyColumns = new ArrayList<ColumnSpecification>();
+
+	/**
+	 * List of only those columns that are not partition or primary key columns.
+	 */
+	private List<ColumnSpecification> nonKeyColumns = new ArrayList<ColumnSpecification>();
+
+	/**
+	 * Adds the given non-key column to the table. Must be specified after all primary key columns.
+	 * 
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes.
+	 * @param type The data type of the column.
+	 */
+	public T column(String name, DataType type) {
+		return column(name, type, null, null);
+	}
+
+	/**
+	 * Adds the given partition key column to the table. Must be specified before any other columns.
+	 * 
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes.
+	 * @param type The data type of the column.
+	 * @return this
+	 */
+	public T partitionKeyColumn(String name, DataType type) {
+		return column(name, type, PARTITION, null);
+	}
+
+	/**
+	 * Adds the given primary key column to the table with ascending ordering. Must be specified after all partition key
+	 * columns and before any non-key columns.
+	 * 
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes.
+	 * @param type The data type of the column.
+	 * @return this
+	 */
+	public T primaryKeyColumn(String name, DataType type) {
+		return primaryKeyColumn(name, type, ASCENDING);
+	}
+
+	/**
+	 * Adds the given primary key column to the table with the given ordering (<code>null</code> meaning ascending). Must
+	 * be specified after all partition key columns and before any non-key columns.
+	 * 
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes.
+	 * @param type The data type of the column.
+	 * @return this
+	 */
+	public T primaryKeyColumn(String name, DataType type, Ordering ordering) {
+		return column(name, type, PRIMARY, ordering);
+	}
+
+	/**
+	 * Adds the given info as a new column to the table. Partition key columns must precede primary key columns, which
+	 * must precede non-key columns.
+	 * 
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes.
+	 * @param type The data type of the column.
+	 * @param keyType Indicates key type. Null means that the column is not a key column.
+	 * @param ordering If the given {@link KeyType} is {@link KeyType#PRIMARY}, then the given ordering is used, else
+	 *          ignored.
+	 * @return this
+	 */
+	@SuppressWarnings("unchecked")
+	protected T column(String name, DataType type, KeyType keyType, Ordering ordering) {
+
+		ColumnSpecification column = new ColumnSpecification().name(name).type(type).keyType(keyType)
+				.ordering(keyType == PRIMARY ? ordering : null);
+
+		columns.add(column);
+
+		if (keyType == KeyType.PARTITION) {
+			partitionKeyColumns.add(column);
+		}
+
+		if (keyType == KeyType.PRIMARY) {
+			primaryKeyColumns.add(column);
+		}
+
+		if (keyType == null) {
+			nonKeyColumns.add(column);
+		}
+
+		return (T) this;
+	}
+
+	/**
+	 * Returns an unmodifiable list of all columns.
+	 */
+	public List<ColumnSpecification> getColumns() {
+		return Collections.unmodifiableList(columns);
+	}
+
+	/**
+	 * Returns an unmodifiable list of all partition key columns.
+	 */
+	public List<ColumnSpecification> getPartitionKeyColumns() {
+		return Collections.unmodifiableList(partitionKeyColumns);
+	}
+
+	/**
+	 * Returns an unmodifiable list of all primary key columns that are not also partition key columns.
+	 */
+	public List<ColumnSpecification> getPrimaryKeyColumns() {
+		return Collections.unmodifiableList(primaryKeyColumns);
+	}
+
+	/**
+	 * Returns an unmodifiable list of all primary key columns that are not also partition key columns.
+	 */
+	public List<ColumnSpecification> getKeyColumns() {
+
+		ArrayList<ColumnSpecification> keyColumns = new ArrayList<ColumnSpecification>();
+		keyColumns.addAll(partitionKeyColumns);
+		keyColumns.addAll(primaryKeyColumns);
+
+		return Collections.unmodifiableList(keyColumns);
+	}
+
+	/**
+	 * Returns an unmodifiable list of all non-key columns.
+	 */
+	public List<ColumnSpecification> getNonKeyColumns() {
+		return Collections.unmodifiableList(nonKeyColumns);
+	}
+}

src/main/java/org/springframework/cassandra/core/util/MapBuilder.java

@@ -1,4 +1,4 @@
-package org.springframework.cassandra.core.keyspace;
+package org.springframework.cassandra.core.util;
 
 import java.util.Collection;
 import java.util.LinkedHashMap;

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

@@ -144,7 +144,7 @@ public class CassandraKeyspaceFactoryBean implements FactoryBean<Keyspace>, Init
 			// drop the old keyspace if needed
 			if (keyspaceExists && (keyspaceAttributes.isCreate() || keyspaceAttributes.isCreateDrop())) {
 				log.info("Drop keyspace " + keyspace + " on afterPropertiesSet");
-				session.execute("DROP KEYSPACE " + keyspace);
+				session.execute("DROP KEYSPACE " + keyspace + ";");
 				keyspaceExists = false;
 			}