DATACASS-96 - resolved merge conflicts

spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlIdentifier.java

@@ -1,132 +0,0 @@
-/*
- * Copyright 2013-2014 the original author or authors.
- * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- * 
- * http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.cassandra.core;
-
-import java.util.regex.Pattern;
-
-import org.springframework.cassandra.core.cql.CqlStringUtils;
-
-/**
- * This encapsulates the logic for CQL identifiers.
- * 
- * @author John McPeek
- * 
- */
-public class CqlIdentifier {
-	public static final String UNQUOTED_IDENTIFIER_REGEX = "[a-zA-Z_][a-zA-Z0-9_]*";
-	public static final Pattern UNQUOTED_IDENTIFIER_PATTERN = Pattern.compile(UNQUOTED_IDENTIFIER_REGEX);
-	public static final String QUOTED_IDENTIFIER_REGEX = "[a-zA-Z_]([a-zA-Z0-9_]|\"{2}+)*";
-	public static final Pattern QUOTED_IDENTIFIER_PATTERN = Pattern.compile(QUOTED_IDENTIFIER_REGEX);
-
-	private String name;
-	private boolean quoted;
-
-	public CqlIdentifier(String identifier) {
-		this(identifier, false);
-	}
-
-	/**
-	 * Renders the given string as a legal Cassandra identifier.
-	 * <ul>
-	 * <li>If the given identifier is a legal quoted identifier or forceQuote is true, it is set encased in double quotes.
-	 * </li>
-	 * <li>If the given identifier is a legal unquoted identifier, it is set unchanged.</li>
-	 * <li>If the given identifier is illegal, an {@link IllegalArgumentException} is thrown.</li>
-	 * </ul>
-	 */
-	public CqlIdentifier(String name, boolean forceQuoting) {
-		if (isUnquotedIdentifier(name) && forceQuoting == false) {
-			this.name = name;
-		} else if (isQuotedIdentifier(name)) {
-			this.name = name;
-			quoted = true;
-		} else {
-			throw new IllegalArgumentException("[" + name + "] is not a valid CQL quoted or unquoted identifier");
-		}
-	}
-
-	public String toCql() {
-		String id = quoted ? CqlStringUtils.doubleQuote(name) : name;
-		return id;
-	}
-
-	public StringBuilder toCql(StringBuilder sb) {
-		return sb.append(toCql());
-	}
-
-	@Override
-	public String toString() {
-		return toCql();
-	}
-
-	public String getName() {
-		return name;
-	}
-
-	public boolean isQuoted() {
-		return quoted;
-	}
-
-	public static CqlIdentifier cqlId(String identifier) {
-		CqlIdentifier id = new CqlIdentifier(identifier);
-		return id;
-	}
-
-	public static CqlIdentifier quotedCqlId(String identifier) {
-		CqlIdentifier id = new CqlIdentifier(identifier, true);
-		return id;
-	}
-
-	public static boolean isIdentifier(CharSequence chars) {
-		return isUnquotedIdentifier(chars) || isQuotedIdentifier(chars);
-	}
-
-	public static boolean isUnquotedIdentifier(CharSequence chars) {
-		return UNQUOTED_IDENTIFIER_PATTERN.matcher(chars).matches();
-	}
-
-	public static boolean isQuotedIdentifier(CharSequence chars) {
-		return QUOTED_IDENTIFIER_PATTERN.matcher(chars).matches();
-	}
-
-	@Override
-	public int hashCode() {
-		final int prime = 31;
-		int result = 1;
-		result = prime * result + ((name == null) ? 0 : name.hashCode());
-		result = prime * result + (quoted ? 1231 : 1237);
-		return result;
-	}
-
-	@Override
-	public boolean equals(Object obj) {
-		if (this == obj)
-			return true;
-		if (obj == null)
-			return false;
-		if (getClass() != obj.getClass())
-			return false;
-		CqlIdentifier other = (CqlIdentifier) obj;
-		if (name == null) {
-			if (other.name != null)
-				return false;
-		} else if (!name.equals(other.name))
-			return false;
-		if (quoted != other.quoted)
-			return false;
-		return true;
-	}
-}

spring-cassandra/src/main/java/org/springframework/cassandra/core/ReservedKeyword.java

@@ -0,0 +1,41 @@
+package org.springframework.cassandra.core;
+
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * CQL keywords.
+ * 
+ * @see <a
+ *      href="http://cassandra.apache.org/doc/cql3/CQL.html#appendixA">http://cassandra.apache.org/doc/cql3/CQL.html#appendixA</a>
+ * 
+ * @author Matthew T. Adams
+ */
+public enum ReservedKeyword {
+	ADD, ALTER, AND, ANY, APPLY, ASC, AUTHORIZE, BATCH, BEGIN, BY, COLUMNFAMILY, CREATE, DELETE, DESC, DROP, EACH_QUORUM, FROM, GRANT, IN, INDEX, INSERT, INTO, KEYSPACE, LIMIT, LOCAL_ONE, LOCAL_QUORUM, MODIFY, NORECURSIVE, OF, ON, ONE, ORDER, PRIMARY, QUORUM, REVOKE, SCHEMA, SELECT, SET, TABLE, THREE, TOKEN, TRUNCATE, TWO, UPDATE, USE, USING, WHERE, WITH;
+
+	/**
+	 * @see ReservedKeyword#isReserved(String)
+	 */
+	public static boolean isReserved(CharSequence candidate) {
+		Assert.notNull(candidate);
+		return isReserved(candidate.toString());
+	}
+
+	/**
+	 * Returns whether the given string is a CQL reserved keyword. This comparison is done regardless of case.
+	 */
+	public static boolean isReserved(String candidate) {
+
+		if (!StringUtils.hasText(candidate)) {
+			return false;
+		}
+
+		try {
+			Enum.valueOf(ReservedKeyword.class, candidate.toUpperCase());
+			return true;
+		} catch (IllegalArgumentException x) {
+			return false;
+		}
+	}
+}

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/CqlConstantType.java

@@ -0,0 +1,41 @@
+package org.springframework.cassandra.core.cql;
+
+import static org.springframework.cassandra.core.cql.CqlConstantType.Regex.*;
+import java.util.regex.Pattern;
+
+public enum CqlConstantType {
+
+	STRING(STRING_PATTERN), INTEGER(INTEGER_PATTERN), FLOAT(FLOAT_PATTERN), BOOLEAN(BOOLEAN_PATTERN), UUID(UUID_PATTERN), BLOB(
+			BLOB_PATTERN);
+
+	private Pattern pattern;
+
+	private CqlConstantType(Pattern pattern) {
+		this.pattern = pattern;
+	}
+
+	public boolean matches(CharSequence candidate) {
+		return pattern.matcher(candidate).matches();
+	}
+
+	public static class Regex {
+
+		public static final String STRING_REGEX = "'((?:[^']+|'')*)'";
+		public static final Pattern STRING_PATTERN = Pattern.compile(STRING_REGEX);
+
+		public static final String INTEGER_REGEX = "\\-?[0-9]+";
+		public static final Pattern INTEGER_PATTERN = Pattern.compile(INTEGER_REGEX);
+
+		public static final String FLOAT_REGEX = "(\\-?[0-9]+(\\.[0-9]*)?([eE][+-]?[0-9+])?)|NaN|Infinity";
+		public static final Pattern FLOAT_PATTERN = Pattern.compile(FLOAT_REGEX);
+
+		public static final String BOOLEAN_REGEX = "(?i)true|false";
+		public static final Pattern BOOLEAN_PATTERN = Pattern.compile(BOOLEAN_REGEX);
+
+		public static final String UUID_REGEX = "(?i)[0-9a-f]{8}+\\-[0-9a-f]{4}+\\-[0-9a-f]{4}+\\-[0-9a-f]{4}+\\-[0-9a-f]{12}+";
+		public static final Pattern UUID_PATTERN = Pattern.compile(UUID_REGEX);
+
+		public static final String BLOB_REGEX = "(?i)0[x](0-9a-f)+";
+		public static final Pattern BLOB_PATTERN = Pattern.compile(BLOB_REGEX);
+	}
+}

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/CqlIdentifier.java

@@ -0,0 +1,202 @@
+package org.springframework.cassandra.core.cql;
+
+import java.util.regex.Pattern;
+
+import org.springframework.cassandra.core.ReservedKeyword;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.TableMetadata;
+
+/**
+ * This encapsulates the logic for CQL quoted and unquoted identifiers.
+ * <p/>
+ * CQL identifiers, when unquoted, are converted to lower case. When quoted, they are returned as-is with no lower
+ * casing and encased in double quotes. To render, use any of the methods {@link #toCql()},
+ * {@link #toCql(StringBuilder)}, or {@link #toString()}.
+ * 
+ * @see #CqlIdentifier(String)
+ * @see #CqlIdentifier(String, boolean)
+ * @see #toCql()
+ * @see #toCql(StringBuilder)
+ * @see #toString()
+ * 
+ * @author John McPeek
+ * @author Matthew T. Adams
+ */
+public final class CqlIdentifier implements Comparable<CqlIdentifier> {
+
+	public static final String UNQUOTED_REGEX = "(?i)[a-z][\\w]*";
+	public static final Pattern UNQUOTED = Pattern.compile(UNQUOTED_REGEX);
+
+	public static final String QUOTED_REGEX = "(?i)[a-z]([\\w]*(\"\")+[\\w]*)+";
+	public static final Pattern QUOTED = Pattern.compile(QUOTED_REGEX);
+
+	/**
+	 * Factory method for {@link CqlIdentifier}. Convenient if imported statically.
+	 * 
+	 * @see #CqlIdentifier(String)
+	 */
+	public static CqlIdentifier cqlId(CharSequence identifier) {
+		return new CqlIdentifier(identifier);
+	}
+
+	/**
+	 * Factory method for a force-quoted {@link CqlIdentifier}. Convenient if imported statically.
+	 * 
+	 * @see #CqlIdentifier(String, boolean)
+	 */
+	public static CqlIdentifier quotedCqlId(CharSequence identifier) {
+		return new CqlIdentifier(identifier, true);
+	}
+
+	/**
+	 * Returns <code>true</code> if the given {@link CharSequence} is a legal unquoted identifier.
+	 */
+	public static boolean isUnquotedIdentifier(CharSequence chars) {
+		return UNQUOTED.matcher(chars).matches() && !ReservedKeyword.isReserved(chars);
+	}
+
+	/**
+	 * Returns <code>true</code> if the given {@link CharSequence} is a legal unquoted identifier.
+	 */
+	public static boolean isQuotedIdentifier(CharSequence chars) {
+		return QUOTED.matcher(chars).matches() || ReservedKeyword.isReserved(chars);
+	}
+
+	private String identifier;
+	private String unquoted;
+	private boolean quoted;
+
+	/**
+	 * Creates a new {@link CqlIdentifier} without force-quoting it. It may end up quoted, depending on its value.
+	 * 
+	 * @see #cqlId(String)
+	 */
+	public CqlIdentifier(CharSequence identifier) {
+		this(identifier, false);
+	}
+
+	/**
+	 * Creates a new CQL identifier, optionally force-quoting it. Force-quoting can be used to preserve identifier case.
+	 * <ul>
+	 * <li>If the given identifier is a legal quoted identifier or <code>forceQuote</code> is <code>true</code>,
+	 * {@link #isQuoted()} will return <code>true</code> and the identifier will be quoted when rendered.</li>
+	 * <li>If the given identifier is a legal unquoted identifier, {@link #isQuoted()} will return <code>false</code>,
+	 * plus the name will be converted to lower case and rendered as such.</li>
+	 * <li>If the given identifier is illegal, an {@link IllegalArgumentException} is thrown.</li>
+	 * </ul>
+	 * 
+	 * @see #cqlId(String)
+	 * @see #quotedCqlId(String)
+	 */
+	public CqlIdentifier(CharSequence identifier, boolean forceQuote) {
+		setIdentifier(identifier, forceQuote);
+	}
+
+	/**
+	 * Tests & sets the given identifier.
+	 */
+	private void setIdentifier(CharSequence identifier, boolean forceQuoting) {
+
+		Assert.notNull(identifier);
+
+		String string = identifier.toString();
+		Assert.hasText(string);
+
+		if (forceQuoting || isQuotedIdentifier(string)) {
+			this.unquoted = string;
+			this.identifier = "\"" + string + "\"";
+			quoted = true;
+		} else if (isUnquotedIdentifier(string)) {
+			this.identifier = this.unquoted = string.toLowerCase();
+		} else {
+			throw new IllegalArgumentException(String.format(
+					"given string [%s] is not a valid quoted or unquoted identifier", identifier));
+		}
+	}
+
+	/**
+	 * Returns the identifier <em>without</em> encasing quotes, regardless of the value of {@link #isQuoted()}. For
+	 * example, if {@link #isQuoted()} is <code>true</code>, then this value will be the same as {@link #toCql()} and
+	 * {@link #toString()}.
+	 * <p/>
+	 * This is needed, for example, to get the correct {@link TableMetadata} from
+	 * {@link KeyspaceMetadata#getTable(String)}: the given string must <em>not</em> be quoted.
+	 */
+	public String getUnquoted() {
+		return unquoted;
+	}
+
+	/**
+	 * Renders this identifier appropriately.
+	 */
+	public String toCql() {
+		return identifier;
+	}
+
+	/**
+	 * Appends the rendering of this identifier to the given {@link StringBuilder}, then returns that
+	 * {@link StringBuilder}. If <code>null</code> is given, a new {@link StringBuilder} is created, appended to, and
+	 * returned.
+	 */
+	public StringBuilder toCql(StringBuilder sb) {
+		sb = sb == null ? new StringBuilder() : sb;
+		return sb.append(toCql());
+	}
+
+	/**
+	 * Alias for {@link #toCql()}.
+	 */
+	@Override
+	public String toString() {
+		return toCql();
+	}
+
+	/**
+	 * Whether or not this identifier is quoted.
+	 */
+	public boolean isQuoted() {
+		return quoted;
+	}
+
+	@Override
+	public int hashCode() {
+		return ((Boolean) quoted).hashCode() ^ identifier.hashCode();
+	}
+
+	/**
+	 * Compares this {@link CqlIdentifier} to the given object. Note that if a {@link CharSequence} is given, a new
+	 * {@link CqlIdentifier} is created from it and compared, such that a {@link CharSequence} can be effectively equal to
+	 * a {@link CqlIdentifier}.
+	 */
+	@Override
+	public boolean equals(Object that) {
+		if (this == that) {
+			return true;
+		}
+		if (that == null) {
+			return false;
+		}
+		if (!(that instanceof CqlIdentifier) && !(that instanceof CharSequence)) {
+			return false;
+		}
+
+		CqlIdentifier other = (that instanceof CqlIdentifier) ? (CqlIdentifier) that : cqlId((CharSequence) that);
+
+		return this.quoted == other.quoted && this.identifier.equals(other.identifier);
+	}
+
+	/**
+	 * Unquoted identifiers sort before quoted ones. Otherwise, they compare according to their identifiers.
+	 */
+	@Override
+	public int compareTo(CqlIdentifier that) {
+
+		int comparison = ((Boolean) this.quoted).compareTo(that.quoted);
+		if (comparison != 0) {
+			return comparison;
+		}
+		return this.identifier.compareTo(that.identifier);
+	}
+}

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/CqlStringUtils.java

@@ -15,8 +15,6 @@
  */
 package org.springframework.cassandra.core.cql;
 
-import java.util.regex.Pattern;
-
 import com.datastax.driver.core.DataType;
 
 public class CqlStringUtils {
@@ -33,9 +31,6 @@ public class CqlStringUtils {
 		return sb == null ? new StringBuilder() : sb;
 	}
 
-	public static final String UNESCAPED_DOUBLE_QUOTE_REGEX = "TODO";
-	public static final Pattern UNESCAPED_DOUBLE_QUOTE_PATTERN = Pattern.compile(UNESCAPED_DOUBLE_QUOTE_REGEX);
-
 	/**
 	 * Renders the given string as a legal Cassandra string column or table option value, by escaping single quotes and
 	 * encasing the result in single quotes. Given <code>null</code>, returns <code>null</code>.
@@ -51,15 +46,15 @@ public class CqlStringUtils {
 	/**
 	 * 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);
+	public static String escapeSingle(Object thing) {
+		return thing == null ? (String) null : thing.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_DOUBLE_QUOTE);
+	public static String escapeDouble(Object thing) {
+		return thing == null ? (String) null : thing.toString().replace(DOUBLE_QUOTE, DOUBLE_DOUBLE_QUOTE);
 	}
 
 	/**
@@ -116,4 +111,21 @@ public class CqlStringUtils {
 
 		return s.append(TYPE_PARAMETER_SUFFIX).toString();
 	}
+
+	public static String unquote(String s) {
+		return unquote(s, "\"");
+	}
+
+	public static String unquote(String s, String quoteChar) {
+		if (s == null) {
+			return s;
+		}
+		if (!s.startsWith(quoteChar) || !s.endsWith(quoteChar)) {
+			return s;
+		}
+		if (s.length() <= 2) {
+			return s;
+		}
+		return s.substring(1, s.length() - 1);
+	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/AddColumnCqlGenerator.java

@@ -30,8 +30,8 @@ public class AddColumnCqlGenerator extends ColumnChangeCqlGenerator<AddColumnSpe
 		super(specification);
 	}
 
+	@Override
 	public StringBuilder toCql(StringBuilder cql) {
-		return noNull(cql).append("ADD ").append(spec().getNameAsIdentifier()).append(" TYPE ")
-				.append(spec().getType().getName());
+		return noNull(cql).append("ADD ").append(spec().getName()).append(" TYPE ").append(spec().getType().getName());
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/AlterColumnCqlGenerator.java

@@ -31,7 +31,6 @@ public class AlterColumnCqlGenerator extends ColumnChangeCqlGenerator<AlterColum
 	}
 
 	public StringBuilder toCql(StringBuilder cql) {
-		return noNull(cql).append("ALTER ").append(spec().getNameAsIdentifier()).append(" TYPE ")
-				.append(spec().getType().getName());
+		return noNull(cql).append("ALTER ").append(spec().getName()).append(" TYPE ").append(spec().getType().getName());
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/AlterKeyspaceCqlGenerator.java

@@ -50,7 +50,7 @@ public class AlterKeyspaceCqlGenerator extends KeyspaceOptionsCqlGenerator<Alter
 	}
 
 	protected StringBuilder preambleCql(StringBuilder cql) {
-		return noNull(cql).append("ALTER KEYSPACE ").append(spec().getNameAsIdentifier()).append(" ");
+		return noNull(cql).append("ALTER KEYSPACE ").append(spec().getName()).append(" ");
 	}
 
 	@SuppressWarnings("unchecked")

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

@@ -56,7 +56,7 @@ public class AlterTableCqlGenerator extends TableOptionsCqlGenerator<AlterTableS
 	}
 
 	protected StringBuilder preambleCql(StringBuilder cql) {
-		return noNull(cql).append("ALTER TABLE ").append(spec().getNameAsIdentifier()).append(" ");
+		return noNull(cql).append("ALTER TABLE ").append(spec().getName()).append(" ");
 	}
 
 	protected StringBuilder changesCql(StringBuilder cql) {

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/CreateIndexCqlGenerator.java

@@ -43,8 +43,8 @@ public class CreateIndexCqlGenerator extends IndexNameCqlGenerator<CreateIndexSp
 
 		cql.append("CREATE").append(spec().isCustom() ? " CUSTOM" : "").append(" INDEX ")
 				.append(spec().getIfNotExists() ? "IF NOT EXISTS " : "")
-				.append(StringUtils.hasText(spec().getNameAsIdentifier()) ? spec().getNameAsIdentifier() : "").append(" ON ")
-				.append(spec().getTableNameAsIdentifier()).append(" (").append(spec().getColumnName()).append(")");
+				.append(spec().getName() == null ? "" : spec().getName()).append(" ON ")
+				.append(spec().getTableName()).append(" (").append(spec().getColumnName()).append(")");
 
 		if (spec().isCustom()) {
 			cql.append(" USING ").append(spec().getUsing());

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/CreateKeyspaceCqlGenerator.java

@@ -55,7 +55,7 @@ public class CreateKeyspaceCqlGenerator extends KeyspaceCqlGenerator<CreateKeysp
 
 	protected StringBuilder preambleCql(StringBuilder cql) {
 		return noNull(cql).append("CREATE KEYSPACE ").append(spec().getIfNotExists() ? "IF NOT EXISTS " : "")
-				.append(spec().getNameAsIdentifier());
+				.append(spec().getName());
 	}
 
 	@SuppressWarnings("unchecked")

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

@@ -58,7 +58,7 @@ public class CreateTableCqlGenerator extends TableCqlGenerator<CreateTableSpecif
 
 	protected StringBuilder preambleCql(StringBuilder cql) {
 		return noNull(cql).append("CREATE TABLE ").append(spec().getIfNotExists() ? "IF NOT EXISTS " : "")
-				.append(spec().getNameAsIdentifier());
+				.append(spec().getName());
 	}
 
 	@SuppressWarnings("unchecked")

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/DropColumnCqlGenerator.java

@@ -31,6 +31,6 @@ public class DropColumnCqlGenerator extends ColumnChangeCqlGenerator<DropColumnS
 	}
 
 	public StringBuilder toCql(StringBuilder cql) {
-		return noNull(cql).append("DROP ").append(spec().getNameAsIdentifier());
+		return noNull(cql).append("DROP ").append(spec().getName());
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/DropIndexCqlGenerator.java

@@ -39,6 +39,6 @@ public class DropIndexCqlGenerator extends IndexNameCqlGenerator<DropIndexSpecif
 	public StringBuilder toCql(StringBuilder cql) {
 		return noNull(cql).append("DROP INDEX ")
 		// .append(spec().getIfExists() ? "IF EXISTS " : "")
-				.append(spec().getNameAsIdentifier()).append(";");
+				.append(spec().getName()).append(";");
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/cql/generator/DropKeyspaceCqlGenerator.java

@@ -37,6 +37,6 @@ public class DropKeyspaceCqlGenerator extends KeyspaceNameCqlGenerator<DropKeysp
 	@Override
 	public StringBuilder toCql(StringBuilder cql) {
 		return noNull(cql).append("DROP KEYSPACE ").append(spec().getIfExists() ? "IF EXISTS " : "")
-				.append(spec().getNameAsIdentifier()).append(";");
+				.append(spec().getName()).append(";");
 	}
 }

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

@@ -38,6 +38,6 @@ public class DropTableCqlGenerator extends TableNameCqlGenerator<DropTableSpecif
 	public StringBuilder toCql(StringBuilder cql) {
 		return noNull(cql).append("DROP TABLE ")
 		// .append(spec().getIfExists() ? "IF EXISTS " : "")
-				.append(spec().getNameAsIdentifier()).append(";");
+				.append(spec().getName()).append(";");
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/AddColumnSpecification.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+
 import com.datastax.driver.core.DataType;
 
 public class AddColumnSpecification extends ColumnTypeChangeSpecification {
@@ -22,4 +24,8 @@ public class AddColumnSpecification extends ColumnTypeChangeSpecification {
 	public AddColumnSpecification(String name, DataType type) {
 		super(name, type);
 	}
+
+	public AddColumnSpecification(CqlIdentifier name, DataType type) {
+		super(name, type);
+	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/AlterColumnSpecification.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+
 import com.datastax.driver.core.DataType;
 
 public class AlterColumnSpecification extends ColumnTypeChangeSpecification {
@@ -22,4 +24,8 @@ public class AlterColumnSpecification extends ColumnTypeChangeSpecification {
 	public AlterColumnSpecification(String name, DataType type) {
 		super(name, type);
 	}
+
+	public AlterColumnSpecification(CqlIdentifier name, DataType type) {
+		super(name, type);
+	}
 }

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

@@ -28,6 +28,14 @@ import com.datastax.driver.core.DataType;
  */
 public class AlterTableSpecification extends TableOptionsSpecification<AlterTableSpecification> {
 
+	/**
+	 * Entry point into the {@link AlterTableSpecification}'s fluent API to alter a table. Convenient if imported
+	 * statically.
+	 */
+	public static AlterTableSpecification alterTable() {
+		return new AlterTableSpecification();
+	}
+
 	/**
 	 * The list of column changes.
 	 */
@@ -65,12 +73,4 @@ public class AlterTableSpecification extends TableOptionsSpecification<AlterTabl
 	public List<ColumnChangeSpecification> getChanges() {
 		return Collections.unmodifiableList(changes);
 	}
-
-	/**
-	 * Entry point into the {@link AlterTableSpecification}'s fluent API to alter a table. Convenient if imported
-	 * statically.
-	 */
-	public static AlterTableSpecification alterTable() {
-		return new AlterTableSpecification();
-	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/ColumnChangeSpecification.java

@@ -15,7 +15,10 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
-import org.springframework.cassandra.core.CqlIdentifier;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
  * Base class for column change specifications.
@@ -24,21 +27,22 @@ import org.springframework.cassandra.core.CqlIdentifier;
  */
 public abstract class ColumnChangeSpecification {
 
-	private CqlIdentifier identifier;
+	protected CqlIdentifier name;
 
-	public ColumnChangeSpecification(String name) {
+	protected ColumnChangeSpecification(String name) {
+		this(cqlId(name));
+	}
+
+	protected ColumnChangeSpecification(CqlIdentifier name) {
 		setName(name);
 	}
 
-	private void setName(String name) {
-		identifier = new CqlIdentifier(name);
+	protected void setName(CqlIdentifier name) {
+		Assert.notNull(name);
+		this.name = name;
 	}
 
-	public String getName() {
-		return identifier.getName();
-	}
-
-	public String getNameAsIdentifier() {
-		return identifier.toCql();
+	public CqlIdentifier getName() {
+		return name;
 	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/ColumnSpecification.java

@@ -18,11 +18,13 @@ package org.springframework.cassandra.core.keyspace;
 import static org.springframework.cassandra.core.Ordering.ASCENDING;
 import static org.springframework.cassandra.core.PrimaryKeyType.CLUSTERED;
 import static org.springframework.cassandra.core.PrimaryKeyType.PARTITIONED;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
 import static org.springframework.cassandra.core.cql.CqlStringUtils.noNull;
 
-import org.springframework.cassandra.core.CqlIdentifier;
 import org.springframework.cassandra.core.Ordering;
 import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 
 import com.datastax.driver.core.DataType;
 
@@ -44,7 +46,7 @@ public class ColumnSpecification {
 	 */
 	public static final Ordering DEFAULT_ORDERING = ASCENDING;
 
-	private CqlIdentifier identifier;
+	private CqlIdentifier name;
 	private DataType type; // TODO: determining if we should be coupling this to Datastax Java Driver type?
 	private PrimaryKeyType keyType;
 	private Ordering ordering;
@@ -55,7 +57,12 @@ public class ColumnSpecification {
 	 * @return this
 	 */
 	public ColumnSpecification name(String name) {
-		identifier = new CqlIdentifier(name);
+		return name(cqlId(name));
+	}
+
+	public ColumnSpecification name(CqlIdentifier name) {
+		Assert.notNull(name);
+		this.name = name;
 		return this;
 	}
 
@@ -130,7 +137,7 @@ public class ColumnSpecification {
 	 * 
 	 * @return this
 	 */
-	/* package */ColumnSpecification keyType(PrimaryKeyType keyType) {
+	ColumnSpecification keyType(PrimaryKeyType keyType) {
 		this.keyType = keyType;
 		return this;
 	}
@@ -140,17 +147,13 @@ public class ColumnSpecification {
 	 * 
 	 * @return this
 	 */
-	/* package */ColumnSpecification ordering(Ordering ordering) {
+	ColumnSpecification ordering(Ordering ordering) {
 		this.ordering = ordering;
 		return this;
 	}
 
-	public String getName() {
-		return identifier.getName();
-	}
-
-	public String getNameAsIdentifier() {
-		return identifier.toCql();
+	public CqlIdentifier getName() {
+		return name;
 	}
 
 	public DataType getType() {
@@ -170,7 +173,7 @@ public class ColumnSpecification {
 	}
 
 	public StringBuilder toCql(StringBuilder cql) {
-		return (cql = noNull(cql)).append(identifier).append(" ").append(type);
+		return (cql = noNull(cql)).append(name).append(" ").append(type);
 	}
 
 	@Override

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/ColumnTypeChangeSpecification.java

@@ -15,6 +15,9 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.util.Assert;
 
 import com.datastax.driver.core.DataType;
@@ -29,6 +32,10 @@ public abstract class ColumnTypeChangeSpecification extends ColumnChangeSpecific
 	private DataType type;
 
 	public ColumnTypeChangeSpecification(String name, DataType type) {
+		this(cqlId(name), type);
+	}
+
+	public ColumnTypeChangeSpecification(CqlIdentifier name, DataType type) {
 		super(name);
 		setType(type);
 	}

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/CreateIndexSpecification.java

@@ -15,7 +15,10 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
-import org.springframework.cassandra.core.CqlIdentifier;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -27,10 +30,18 @@ import org.springframework.util.StringUtils;
 public class CreateIndexSpecification extends IndexNameSpecification<CreateIndexSpecification> implements
 		IndexDescriptor {
 
+	/**
+	 * Entry point into the {@link CreateIndexSpecification}'s fluent API to create a index. Convenient if imported
+	 * statically.
+	 */
+	public static CreateIndexSpecification createIndex() {
+		return new CreateIndexSpecification();
+	}
+
 	private boolean ifNotExists = false;
 	private boolean custom = false;
-	private CqlIdentifier identifier;
-	private String columnName;
+	private CqlIdentifier tableName;
+	private CqlIdentifier columnName;
 	private String using;
 
 	/**
@@ -56,6 +67,7 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		return ifNotExists;
 	}
 
+	@Override
 	public boolean isCustom() {
 		return custom;
 	}
@@ -73,11 +85,13 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 		return this;
 	}
 
+	@Override
 	public String getUsing() {
 		return using;
 	}
 
-	public String getColumnName() {
+	@Override
+	public CqlIdentifier getColumnName() {
 		return columnName;
 	}
 
@@ -87,29 +101,27 @@ public class CreateIndexSpecification extends IndexNameSpecification<CreateIndex
 	 * @return this
 	 */
 	public CreateIndexSpecification tableName(String tableName) {
-		identifier = new CqlIdentifier(tableName);
+		return tableName(cqlId(tableName));
+	}
+
+	public CreateIndexSpecification tableName(CqlIdentifier tableName) {
+		Assert.notNull(tableName);
+		this.tableName = tableName;
 		return this;
 	}
 
-	public String getTableName() {
-		return identifier.getName();
-	}
-
-	public String getTableNameAsIdentifier() {
-		return identifier.toCql();
+	@Override
+	public CqlIdentifier getTableName() {
+		return tableName;
 	}
 
 	public CreateIndexSpecification columnName(String columnName) {
+		return columnName(cqlId(columnName));
+	}
+
+	public CreateIndexSpecification columnName(CqlIdentifier columnName) {
+		Assert.notNull(columnName);
 		this.columnName = columnName;
 		return this;
 	}
-
-	/**
-	 * Entry point into the {@link CreateIndexSpecification}'s fluent API to create a index. Convenient if imported
-	 * statically.
-	 */
-	public static CreateIndexSpecification createIndex() {
-		return new CreateIndexSpecification();
-	}
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/CreateKeyspaceSpecification.java

@@ -16,11 +16,20 @@
 package org.springframework.cassandra.core.keyspace;
 
 import org.springframework.cassandra.config.DataCenterReplication;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.cassandra.core.keyspace.KeyspaceOption.ReplicationStrategy;
 import org.springframework.cassandra.core.util.MapBuilder;
 
 public class CreateKeyspaceSpecification extends KeyspaceSpecification<CreateKeyspaceSpecification> {
 
+	/**
+	 * Entry point into the {@link CreateKeyspaceSpecification}'s fluent API to create a keyspace. Convenient if imported
+	 * statically.
+	 */
+	public static CreateKeyspaceSpecification createKeyspace() {
+		return new CreateKeyspaceSpecification();
+	}
+
 	private boolean ifNotExists = false;
 
 	/**
@@ -46,14 +55,6 @@ public class CreateKeyspaceSpecification extends KeyspaceSpecification<CreateKey
 		return ifNotExists;
 	}
 
-	/**
-	 * Entry point into the {@link CreateKeyspaceSpecification}'s fluent API to create a keyspace. Convenient if imported
-	 * statically.
-	 */
-	public static CreateKeyspaceSpecification createKeyspace() {
-		return new CreateKeyspaceSpecification();
-	}
-
 	public CreateKeyspaceSpecification withSimpleReplication() {
 		return withSimpleReplication(1);
 	}
@@ -86,6 +87,11 @@ public class CreateKeyspaceSpecification extends KeyspaceSpecification<CreateKey
 		return (CreateKeyspaceSpecification) super.name(name);
 	}
 
+	@Override
+	public CreateKeyspaceSpecification name(CqlIdentifier name) {
+		return (CreateKeyspaceSpecification) super.name(name);
+	}
+
 	@Override
 	public CreateKeyspaceSpecification with(KeyspaceOption option) {
 		return (CreateKeyspaceSpecification) super.with(option);

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

@@ -22,6 +22,14 @@ package org.springframework.cassandra.core.keyspace;
  */
 public class CreateTableSpecification extends TableSpecification<CreateTableSpecification> {
 
+	/**
+	 * Entry point into the {@link CreateTableSpecification}'s fluent API to create a table. Convenient if imported
+	 * statically.
+	 */
+	public static CreateTableSpecification createTable() {
+		return new CreateTableSpecification();
+	}
+
 	private boolean ifNotExists = false;
 
 	/**
@@ -52,14 +60,6 @@ public class CreateTableSpecification extends TableSpecification<CreateTableSpec
 		return (CreateTableSpecification) super.name(name);
 	}
 
-	/**
-	 * Entry point into the {@link CreateTableSpecification}'s fluent API to create a table. Convenient if imported
-	 * statically.
-	 */
-	public static CreateTableSpecification createTable() {
-		return new CreateTableSpecification();
-	}
-
 	@Override
 	public CreateTableSpecification with(TableOption option) {
 		return (CreateTableSpecification) super.with(option);

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/DropColumnSpecification.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+
 /**
  * A specification to drop a column.
  * 
@@ -22,7 +24,19 @@ package org.springframework.cassandra.core.keyspace;
  */
 public class DropColumnSpecification extends ColumnChangeSpecification {
 
+	public static DropColumnSpecification dropColumn(String name) {
+		return new DropColumnSpecification(name);
+	}
+
+	public static DropColumnSpecification dropColumn(CqlIdentifier name) {
+		return new DropColumnSpecification(name);
+	}
+
 	public DropColumnSpecification(String name) {
 		super(name);
 	}
+
+	public DropColumnSpecification(CqlIdentifier name) {
+		super(name);
+	}
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/DropKeyspaceSpecification.java

@@ -39,5 +39,4 @@ public class DropKeyspaceSpecification extends KeyspaceActionSpecification<DropK
 	public static DropKeyspaceSpecification dropKeyspace() {
 		return new DropKeyspaceSpecification();
 	}
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/IndexDescriptor.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 
 /**
  * Describes an index.
@@ -27,27 +28,16 @@ public interface IndexDescriptor {
 	/**
 	 * Returns the name of the index.
 	 */
-	String getName();
+	CqlIdentifier getName();
 
 	/**
 	 * Returns the table name for the index
 	 */
-	String getTableName();
+	CqlIdentifier getTableName();
 
-	/**
-	 * Returns the name of the index as an identifer or quoted identifier as appropriate.
-	 */
-	String getNameAsIdentifier();
-
-	/**
-	 * Returns the name of the table as an identifer or quoted identifier as appropriate.
-	 */
-	String getTableNameAsIdentifier();
-
-	String getColumnName();
+	CqlIdentifier getColumnName();
 
 	String getUsing();
 
 	boolean isCustom();
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/IndexNameSpecification.java

@@ -15,38 +15,43 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
-import org.springframework.cassandra.core.CqlIdentifier;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
- * Abstract builder class to support the construction of table specifications.
+ * Abstract builder class to support the construction of an index.
+ * 
+ * @param <T> The subtype of the {@link IndexNameSpecification}
  * 
  * @author David Webb
- * @param <T> The subtype of the {@link IndexNameSpecification}
+ * @author Matthew T. Adams
  */
 public abstract class IndexNameSpecification<T extends IndexNameSpecification<T>> {
 
 	/**
 	 * The name of the index.
 	 */
-	private CqlIdentifier identifier;
+	private CqlIdentifier name;
 
 	/**
 	 * Sets the index name.
 	 * 
 	 * @return this
 	 */
-	@SuppressWarnings("unchecked")
 	public T name(String name) {
-		identifier = new CqlIdentifier(name);
+		return name(cqlId(name));
+	}
+
+	@SuppressWarnings("unchecked")
+	public T name(CqlIdentifier name) {
+		Assert.notNull(name);
+		this.name = name;
 		return (T) this;
 	}
 
-	public String getName() {
-		return identifier.getName();
+	public CqlIdentifier getName() {
+		return name;
 	}
-
-	public String getNameAsIdentifier() {
-		return identifier.toCql();
-	}
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceActionSpecification.java

@@ -15,7 +15,10 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
-import org.springframework.cassandra.core.CqlIdentifier;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
  * Abstract builder class to support the construction of keyspace specifications.
@@ -29,7 +32,16 @@ public abstract class KeyspaceActionSpecification<T extends KeyspaceActionSpecif
 	/**
 	 * The name of the keyspace.
 	 */
-	private CqlIdentifier identifier;
+	private CqlIdentifier name;
+
+	/**
+	 * Sets the keyspace name.
+	 * 
+	 * @return this
+	 */
+	public T name(String name) {
+		return name(cqlId(name));
+	}
 
 	/**
 	 * Sets the keyspace name.
@@ -37,27 +49,14 @@ public abstract class KeyspaceActionSpecification<T extends KeyspaceActionSpecif
 	 * @return this
 	 */
 	@SuppressWarnings("unchecked")
-	public T name(String name) {
-		identifier = new CqlIdentifier(name);
+	public T name(CqlIdentifier name) {
+		Assert.notNull(name);
+		this.name = name;
 		return (T) this;
 	}
 
-	public String getName() {
-		return identifier.getName();
-	}
-
-	public String getNameAsIdentifier() {
-		return identifier.toCql();
-	}
-
-	/**
-	 * For debugging KeyspaceActionSprcifications
-	 */
-	@Override
-	public String toString() {
-		StringBuilder sb = new StringBuilder();
-		sb.append("Keyspace Action Specification {name: " + identifier + ", class: " + this.getClass() + "}");
-		return sb.toString();
+	public CqlIdentifier getName() {
+		return name;
 	}
 
 	/**
@@ -77,13 +76,12 @@ public abstract class KeyspaceActionSpecification<T extends KeyspaceActionSpecif
 		if (!(that instanceof KeyspaceActionSpecification)) {
 			return false;
 		}
-		KeyspaceActionSpecification<?> thatSpec = (KeyspaceActionSpecification<?>) that;
-		return this.identifier.equals(thatSpec.identifier) && this.getClass().equals(that.getClass());
+		KeyspaceActionSpecification<?> other = (KeyspaceActionSpecification<?>) that;
+		return this.name.equals(other.name) && this.getClass().equals(that.getClass());
 	}
 
 	@Override
 	public int hashCode() {
-		return this.identifier.hashCode() ^ this.getClass().hashCode();
+		return name.hashCode() ^ getClass().hashCode();
 	}
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceDescriptor.java

@@ -17,6 +17,8 @@ package org.springframework.cassandra.core.keyspace;
 
 import java.util.Map;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+
 /**
  * Describes a Keyspace.
  * 
@@ -27,12 +29,7 @@ public interface KeyspaceDescriptor {
 	/**
 	 * Returns the name of the table.
 	 */
-	String getName();
-
-	/**
-	 * Returns the name of the table as an identifier or quoted identifier as appropriate.
-	 */
-	String getNameAsIdentifier();
+	CqlIdentifier getName();
 
 	/**
 	 * Returns an unmodifiable {@link Map} of keyspace options.

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceOptionsSpecification.java

@@ -42,6 +42,7 @@ public abstract class KeyspaceOptionsSpecification<T extends KeyspaceOptionsSpec
 
 	protected Map<String, Object> options = new LinkedHashMap<String, Object>();
 
+	@Override
 	@SuppressWarnings("unchecked")
 	public T name(String name) {
 		return (T) super.name(name);
@@ -68,7 +69,7 @@ public abstract class KeyspaceOptionsSpecification<T extends KeyspaceOptionsSpec
 	 */
 	public T with(KeyspaceOption option, Object value) {
 		option.checkValue(value);
-		return (T) with(option.getName(), value, option.escapesValue(), option.quotesValue());
+		return with(option.getName(), value, option.escapesValue(), option.quotesValue());
 	}
 
 	/**
@@ -105,5 +106,4 @@ public abstract class KeyspaceOptionsSpecification<T extends KeyspaceOptionsSpec
 	public Map<String, Object> getOptions() {
 		return Collections.unmodifiableMap(options);
 	}
-
 }

spring-cassandra/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceSpecification.java

@@ -20,6 +20,7 @@ package org.springframework.cassandra.core.keyspace;
  * as a standalone {@link KeyspaceDescriptor}, independent of {@link CreateKeyspaceSpecification}.
  * 
  * @author John McPeek
+ * @author Matthew T. Adams
  */
 public class KeyspaceSpecification<T> extends KeyspaceOptionsSpecification<KeyspaceSpecification<T>> implements
 		KeyspaceDescriptor {

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

@@ -18,6 +18,8 @@ package org.springframework.cassandra.core.keyspace;
 import java.util.List;
 import java.util.Map;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+
 /**
  * Describes a table.
  * 
@@ -29,12 +31,7 @@ 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();
+	CqlIdentifier getName();
 
 	/**
 	 * Returns an unmodifiable {@link List} of {@link ColumnSpecification}s.

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

@@ -15,7 +15,10 @@
  */
 package org.springframework.cassandra.core.keyspace;
 
-import org.springframework.cassandra.core.CqlIdentifier;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+
+import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.util.Assert;
 
 /**
  * Abstract builder class to support the construction of table specifications.
@@ -28,24 +31,25 @@ public abstract class TableNameSpecification<T extends TableNameSpecification<T>
 	/**
 	 * The name of the table.
 	 */
-	private CqlIdentifier identifier;
+	private CqlIdentifier name;
 
 	/**
 	 * Sets the table name.
 	 * 
 	 * @return this
 	 */
-	@SuppressWarnings("unchecked")
 	public T name(String name) {
-		identifier = new CqlIdentifier(name);
+		return name(cqlId(name));
+	}
+
+	@SuppressWarnings("unchecked")
+	public T name(CqlIdentifier name) {
+		Assert.notNull(name);
+		this.name = name;
 		return (T) this;
 	}
 
-	public String getName() {
-		return identifier.getName();
-	}
-
-	public String getNameAsIdentifier() {
-		return identifier.toCql();
+	public CqlIdentifier getName() {
+		return name;
 	}
 }

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

@@ -22,8 +22,11 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.cassandra.core.cql.CqlStringUtils;
 
+import com.datastax.driver.core.DataType;
+
 /**
  * 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>.
@@ -42,11 +45,18 @@ public abstract class TableOptionsSpecification<T extends TableOptionsSpecificat
 
 	protected Map<String, Object> options = new LinkedHashMap<String, Object>();
 
+	@Override
 	@SuppressWarnings("unchecked")
 	public T name(String name) {
 		return (T) super.name(name);
 	}
 
+	@Override
+	@SuppressWarnings("unchecked")
+	public T name(CqlIdentifier name) {
+		return (T) super.name(name);
+	}
+
 	/**
 	 * Convenience method that calls <code>with(option, null)</code>.
 	 * 
@@ -68,7 +78,7 @@ public abstract class TableOptionsSpecification<T extends TableOptionsSpecificat
 	 */
 	public T with(TableOption option, Object value) {
 		option.checkValue(value);
-		return (T) with(option.getName(), value, option.escapesValue(), option.quotesValue());
+		return with(option.getName(), value, option.escapesValue(), option.quotesValue());
 	}
 
 	/**

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

@@ -17,6 +17,7 @@ package org.springframework.cassandra.core.keyspace;
 
 import static org.springframework.cassandra.core.PrimaryKeyType.CLUSTERED;
 import static org.springframework.cassandra.core.PrimaryKeyType.PARTITIONED;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -24,6 +25,7 @@ import java.util.List;
 
 import org.springframework.cassandra.core.Ordering;
 import org.springframework.cassandra.core.PrimaryKeyType;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 
 import com.datastax.driver.core.DataType;
 
@@ -63,6 +65,10 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @param type The data type of the column.
 	 */
 	public T column(String name, DataType type) {
+		return column(cqlId(name), type);
+	}
+
+	public T column(CqlIdentifier name, DataType type) {
 		return column(name, type, null, null);
 	}
 
@@ -74,6 +80,10 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @return this
 	 */
 	public T partitionKeyColumn(String name, DataType type) {
+		return partitionKeyColumn(cqlId(name), type);
+	}
+
+	public T partitionKeyColumn(CqlIdentifier name, DataType type) {
 		return column(name, type, PARTITIONED, null);
 	}
 
@@ -89,6 +99,10 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 		return clusteredKeyColumn(name, type, null);
 	}
 
+	public T clusteredKeyColumn(CqlIdentifier name, DataType type) {
+		return clusteredKeyColumn(name, type, null);
+	}
+
 	/**
 	 * 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.
@@ -98,6 +112,10 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @return this
 	 */
 	public T clusteredKeyColumn(String name, DataType type, Ordering ordering) {
+		return clusteredKeyColumn(cqlId(name), type, ordering);
+	}
+
+	public T clusteredKeyColumn(CqlIdentifier name, DataType type, Ordering ordering) {
 		return column(name, type, CLUSTERED, ordering);
 	}
 
@@ -112,8 +130,12 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 *          used, else ignored.
 	 * @return this
 	 */
-	@SuppressWarnings("unchecked")
 	protected T column(String name, DataType type, PrimaryKeyType keyType, Ordering ordering) {
+		return column(cqlId(name), type, keyType, ordering);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected T column(CqlIdentifier name, DataType type, PrimaryKeyType keyType, Ordering ordering) {
 
 		ColumnSpecification column = new ColumnSpecification().name(name).type(type).keyType(keyType)
 				.ordering(keyType == CLUSTERED ? ordering : null);
@@ -138,6 +160,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	/**
 	 * Returns an unmodifiable list of all columns.
 	 */
+	@Override
 	public List<ColumnSpecification> getColumns() {
 		return Collections.unmodifiableList(columns);
 	}
@@ -145,6 +168,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	/**
 	 * Returns an unmodifiable list of all partition key columns.
 	 */
+	@Override
 	public List<ColumnSpecification> getPartitionKeyColumns() {
 		return Collections.unmodifiableList(partitionKeyColumns);
 	}
@@ -152,6 +176,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	/**
 	 * Returns an unmodifiable list of all primary key columns that are not also partition key columns.
 	 */
+	@Override
 	public List<ColumnSpecification> getClusteredKeyColumns() {
 		return Collections.unmodifiableList(clusteredKeyColumns);
 	}
@@ -159,6 +184,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	/**
 	 * Returns an unmodifiable list of all primary key columns that are not also partition key columns.
 	 */
+	@Override
 	public List<ColumnSpecification> getPrimaryKeyColumns() {
 
 		ArrayList<ColumnSpecification> primaryKeyColumns = new ArrayList<ColumnSpecification>();
@@ -171,6 +197,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	/**
 	 * Returns an unmodifiable list of all non-key columns.
 	 */
+	@Override
 	public List<ColumnSpecification> getNonKeyColumns() {
 		return Collections.unmodifiableList(nonKeyColumns);
 	}

spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CqlIndexSpecificationAssertions.java

@@ -29,14 +29,14 @@ public class CqlIndexSpecificationAssertions {
 
 	public static void assertIndex(IndexDescriptor expected, String keyspace, Session session) {
 		IndexMetadata imd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
-				.getTable(expected.getTableName()).getColumn(expected.getColumnName()).getIndex();
+				.getTable(expected.getTableName().toCql()).getColumn(expected.getColumnName().toCql()).getIndex();
 
-		assertEquals(expected.getName().toLowerCase(), imd.getName().toLowerCase());
+		assertEquals(expected.getName(), imd.getName());
 	}
 
 	public static void assertNoIndex(IndexDescriptor expected, String keyspace, Session session) {
 		IndexMetadata imd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
-				.getTable(expected.getTableName()).getColumn(expected.getColumnName()).getIndex();
+				.getTable(expected.getTableName().toCql()).getColumn(expected.getColumnName().toCql()).getIndex();
 
 		assertNull(imd);
 	}

spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CqlTableSpecificationAssertions.java

@@ -42,9 +42,9 @@ public class CqlTableSpecificationAssertions {
 
 	public static void assertTable(TableDescriptor expected, String keyspace, Session session) {
 		TableMetadata tmd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
-				.getTable(expected.getName());
+				.getTable(expected.getName().getUnquoted()); // TODO: talk to Datastax about unquoting
 
-		assertEquals(expected.getName().toLowerCase(), tmd.getName().toLowerCase());
+		assertEquals(expected.getName().getUnquoted(), tmd.getName()); // TODO: talk to Datastax
 		assertPartitionKeyColumns(expected, tmd);
 		assertPrimaryKeyColumns(expected, tmd);
 		assertColumns(expected.getColumns(), tmd.getColumns());
@@ -53,7 +53,7 @@ public class CqlTableSpecificationAssertions {
 
 	public static void assertNoTable(DropTableSpecification expected, String keyspace, Session session) {
 		TableMetadata tmd = session.getCluster().getMetadata().getKeyspace(keyspace.toLowerCase())
-				.getTable(expected.getName());
+				.getTable(expected.getName().toCql());
 
 		assertNull(tmd);
 	}
@@ -171,7 +171,7 @@ public class CqlTableSpecificationAssertions {
 	}
 
 	public static void assertColumn(ColumnSpecification expected, ColumnMetadata actual) {
-		assertEquals(expected.getName().toLowerCase(), actual.getName().toLowerCase());
+		assertEquals(expected.getName().toCql(), actual.getName()); // TODO: expected.getName().getUnquoted()?
 		assertEquals(expected.getType(), actual.getType());
 	}
 }

spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/FunkyIdentifierIntegrationTest.java

@@ -0,0 +1,42 @@
+package org.springframework.cassandra.test.integration.core.cql.generator;
+
+import org.junit.Test;
+import org.springframework.cassandra.core.cql.generator.CreateTableCqlGenerator;
+import org.springframework.cassandra.core.keyspace.CreateTableSpecification;
+import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
+import org.springframework.cassandra.test.unit.core.cql.generator.CreateTableCqlGeneratorTests.FunkyTableNameTest;
+
+import com.datastax.driver.core.DataType;
+
+public class FunkyIdentifierIntegrationTest extends AbstractKeyspaceCreatingIntegrationTest {
+
+	public FunkyIdentifierIntegrationTest() {
+		super(randomKeyspaceName());
+	}
+
+	@Test
+	public void testFunkyTableName() {
+		for (String name : FunkyTableNameTest.FUNKY_LEGAL_NAMES) {
+			SESSION.execute(new CreateTableCqlGenerator(CreateTableSpecification.createTable().name(name)
+					.partitionKeyColumn("key", DataType.text())).toCql());
+		}
+	}
+
+	@Test
+	public void testFunkyColumnName() {
+		String table = "funky";
+		int i = 0;
+		for (String name : FunkyTableNameTest.FUNKY_LEGAL_NAMES) {
+			SESSION.execute(new CreateTableCqlGenerator(CreateTableSpecification.createTable().name(table + i++)
+					.partitionKeyColumn(name, DataType.text())).toCql());
+		}
+	}
+
+	@Test
+	public void testFunkyTableAndColumnName() {
+		for (String name : FunkyTableNameTest.FUNKY_LEGAL_NAMES) {
+			SESSION.execute(new CreateTableCqlGenerator(CreateTableSpecification.createTable().name(name)
+					.partitionKeyColumn(name, DataType.text())).toCql());
+		}
+	}
+}

spring-cassandra/src/test/java/org/springframework/cassandra/test/unit/core/cql/CqlIdentifierTest.java

@@ -15,21 +15,64 @@
  */
 package org.springframework.cassandra.test.unit.core.cql;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.springframework.cassandra.core.CqlIdentifier.isQuotedIdentifier;
-import static org.springframework.cassandra.core.CqlIdentifier.isUnquotedIdentifier;
+import static org.junit.Assert.*;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.quotedCqlId;
 
 import org.junit.Test;
+import org.springframework.cassandra.core.ReservedKeyword;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 
 public class CqlIdentifierTest {
 
 	@Test
-	public void testIsQuotedIdentifier() throws Exception {
-		assertFalse(isQuotedIdentifier("my\"id"));
-		assertTrue(isQuotedIdentifier("my\"\"id"));
-		assertFalse(isUnquotedIdentifier("my\"id"));
-		assertTrue(isUnquotedIdentifier("myid"));
+	public void testUnquotedIdentifiers() {
+
+		String[] ids = new String[] { "foo", "Foo", "FOO", "a_", "a1" };
+
+		for (String id : ids) {
+			CqlIdentifier cqlId = cqlId(id);
+			assertFalse(cqlId.isQuoted());
+			assertEquals(id.toLowerCase(), cqlId.toCql());
+		}
 	}
 
+	@Test
+	public void testForceQuotedIdentifiers() {
+
+		String[] ids = new String[] { "foo", "Foo", "FOO", "a_", "a1" };
+
+		for (String id : ids) {
+			CqlIdentifier cqlId = quotedCqlId(id);
+			assertTrue(cqlId.isQuoted());
+			assertEquals("\"" + id + "\"", cqlId.toCql());
+		}
+	}
+
+	@Test
+	public void testReservedWordsEndUpQuoted() {
+
+		for (ReservedKeyword id : ReservedKeyword.values()) {
+			CqlIdentifier cqlId = cqlId(id.name());
+			assertTrue(cqlId.isQuoted());
+			assertEquals("\"" + id.name() + "\"", cqlId.toCql());
+
+			cqlId = cqlId(id.name().toLowerCase());
+			assertTrue(cqlId.isQuoted());
+			assertEquals("\"" + id.name().toLowerCase() + "\"", cqlId.toCql());
+		}
+	}
+
+	@Test
+	public void testIllegals() {
+		String[] illegals = new String[] { null, "", "a ", "a a", "a\"", "a'", "a''", "\"\"", "''", "-", "a-", "_", "_a" };
+		for (String illegal : illegals) {
+			try {
+				cqlId(illegal);
+				fail(String.format("identifier [%s] should have caused IllegalArgumentException", illegal));
+			} catch (IllegalArgumentException x) {
+				// :)
+			}
+		}
+	}
 }

spring-cassandra/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/AlterTableCqlGeneratorTests.java

@@ -41,7 +41,6 @@ public class AlterTableCqlGeneratorTests {
 	 * Asserts that the preamble is first & correctly formatted in the given CQL string.
 	 */
 	public static void assertPreamble(String tableName, String cql) {
-		System.out.println("cql: " + cql);
 		assertTrue(cql.startsWith("ALTER TABLE " + tableName + " "));
 	}
 

spring-cassandra/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/CreateTableCqlGeneratorTests.java

@@ -16,13 +16,20 @@
 package org.springframework.cassandra.test.unit.core.cql.generator;
 
 import static org.junit.Assert.assertTrue;
+import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.cassandra.core.ReservedKeyword;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.cassandra.core.cql.generator.CreateTableCqlGenerator;
 import org.springframework.cassandra.core.keyspace.CreateTableSpecification;
 import org.springframework.cassandra.core.keyspace.Option;
@@ -40,7 +47,7 @@ public class CreateTableCqlGeneratorTests {
 	/**
 	 * Asserts that the preamble is first & correctly formatted in the given CQL string.
 	 */
-	public static void assertPreamble(String tableName, String cql) {
+	public static void assertPreamble(CqlIdentifier tableName, String cql) {
 		assertTrue(cql.startsWith("CREATE TABLE " + tableName + " "));
 	}
 
@@ -98,6 +105,7 @@ public class CreateTableCqlGeneratorTests {
 	public static abstract class CreateTableTest extends
 			TableOperationCqlGeneratorTest<CreateTableSpecification, CreateTableCqlGenerator> {
 
+		@Override
 		public CreateTableCqlGenerator generator() {
 			return new CreateTableCqlGenerator(specification);
 		}
@@ -105,12 +113,13 @@ public class CreateTableCqlGeneratorTests {
 
 	public static class BasicTest extends CreateTableTest {
 
-		public String name = "mytable";
+		public CqlIdentifier name = cqlId("mytable");
 		public DataType partitionKeyType0 = DataType.text();
-		public String partitionKey0 = "partitionKey0";
+		public CqlIdentifier partitionKey0 = cqlId("partitionKey0");
 		public DataType columnType1 = DataType.text();
 		public String column1 = "column1";
 
+		@Override
 		public CreateTableSpecification specification() {
 			return CreateTableSpecification.createTable().name(name).partitionKeyColumn(partitionKey0, partitionKeyType0)
 					.column(column1, columnType1);
@@ -122,18 +131,18 @@ public class CreateTableCqlGeneratorTests {
 
 			assertPreamble(name, cql);
 			assertColumns(String.format("%s %s, %s %s", partitionKey0, partitionKeyType0, column1, columnType1), cql);
-			assertPrimaryKey(partitionKey0, cql);
+			assertPrimaryKey(partitionKey0.toCql(), cql);
 		}
 	}
 
 	public static class CompositePartitionKeyTest extends CreateTableTest {
 
-		public String name = "composite_partition_key_table";
+		public CqlIdentifier name = cqlId("composite_partition_key_table");
 		public DataType partKeyType0 = DataType.text();
-		public String partKey0 = "partKey0";
+		public CqlIdentifier partKey0 = cqlId("partKey0");
 		public DataType partKeyType1 = DataType.text();
-		public String partKey1 = "partKey1";
-		public String column0 = "column0";
+		public CqlIdentifier partKey1 = cqlId("partKey1");
+		public CqlIdentifier column0 = cqlId("column0");
 		public DataType columnType0 = DataType.text();
 
 		@Override
@@ -162,19 +171,20 @@ public class CreateTableCqlGeneratorTests {
 	 */
 	public static class ReadRepairChanceTest extends CreateTableTest {
 
-		public String name = "mytable";
+		public CqlIdentifier name = cqlId("mytable");
 		public DataType partitionKeyType0 = DataType.text();
-		public String partitionKey0 = "partitionKey0";
+		public CqlIdentifier partitionKey0 = cqlId("partitionKey0");
 		public DataType partitionKeyType1 = DataType.timestamp();
-		public String partitionKey1 = "create_timestamp";
+		public CqlIdentifier partitionKey1 = cqlId("create_timestamp");
 		public DataType columnType1 = DataType.text();
-		public String column1 = "column1";
+		public CqlIdentifier column1 = cqlId("column1");
 		public Double readRepairChance = 0.5;
 
+		@Override
 		public CreateTableSpecification specification() {
-			return (CreateTableSpecification) CreateTableSpecification.createTable().name(name)
-					.partitionKeyColumn(partitionKey0, partitionKeyType0).partitionKeyColumn(partitionKey1, partitionKeyType1)
-					.column(column1, columnType1).with(TableOption.READ_REPAIR_CHANCE, readRepairChance);
+			return CreateTableSpecification.createTable().name(name).partitionKeyColumn(partitionKey0, partitionKeyType0)
+					.partitionKeyColumn(partitionKey1, partitionKeyType1).column(column1, columnType1)
+					.with(TableOption.READ_REPAIR_CHANCE, readRepairChance);
 		}
 
 		@Test
@@ -197,13 +207,13 @@ public class CreateTableCqlGeneratorTests {
 	 */
 	public static class MultipleOptionsTest extends CreateTableTest {
 
-		public String name = "timeseries_table";
+		public CqlIdentifier name = cqlId("timeseries_table");
 		public DataType partitionKeyType0 = DataType.timeuuid();
-		public String partitionKey0 = "tid";
+		public CqlIdentifier partitionKey0 = cqlId("tid");
 		public DataType partitionKeyType1 = DataType.timestamp();
-		public String partitionKey1 = "create_timestamp";
+		public CqlIdentifier partitionKey1 = cqlId("create_timestamp");
 		public DataType columnType1 = DataType.text();
-		public String column1 = "data_point";
+		public CqlIdentifier column1 = cqlId("data_point");
 		public Double readRepairChance = 0.5;
 		public Double dcLocalReadRepairChance = 0.7;
 		public Double bloomFilterFpChance = 0.001;
@@ -213,6 +223,7 @@ public class CreateTableCqlGeneratorTests {
 		public Map<Option, Object> compactionMap = new LinkedHashMap<Option, Object>();
 		public Map<Option, Object> compressionMap = new LinkedHashMap<Option, Object>();
 
+		@Override
 		public CreateTableSpecification specification() {
 
 			// Compaction
@@ -223,11 +234,11 @@ public class CreateTableCqlGeneratorTests {
 			compressionMap.put(CompressionOption.CHUNK_LENGTH_KB, 128);
 			compressionMap.put(CompressionOption.CRC_CHECK_CHANCE, 0.75);
 
-			return (CreateTableSpecification) CreateTableSpecification.createTable().name(name)
-					.partitionKeyColumn(partitionKey0, partitionKeyType0).partitionKeyColumn(partitionKey1, partitionKeyType1)
-					.column(column1, columnType1).with(TableOption.COMPACT_STORAGE)
-					.with(TableOption.READ_REPAIR_CHANCE, readRepairChance).with(TableOption.COMPACTION, compactionMap)
-					.with(TableOption.COMPRESSION, compressionMap).with(TableOption.BLOOM_FILTER_FP_CHANCE, bloomFilterFpChance)
+			return CreateTableSpecification.createTable().name(name).partitionKeyColumn(partitionKey0, partitionKeyType0)
+					.partitionKeyColumn(partitionKey1, partitionKeyType1).column(column1, columnType1)
+					.with(TableOption.COMPACT_STORAGE).with(TableOption.READ_REPAIR_CHANCE, readRepairChance)
+					.with(TableOption.COMPACTION, compactionMap).with(TableOption.COMPRESSION, compressionMap)
+					.with(TableOption.BLOOM_FILTER_FP_CHANCE, bloomFilterFpChance)
 					.with(TableOption.CACHING, CachingOption.KEYS_ONLY).with(TableOption.REPLICATE_ON_WRITE, replcateOnWrite)
 					.with(TableOption.COMMENT, comment).with(TableOption.DCLOCAL_READ_REPAIR_CHANCE, dcLocalReadRepairChance)
 					.with(TableOption.GC_GRACE_SECONDS, gcGraceSeconds);
@@ -256,4 +267,50 @@ public class CreateTableCqlGeneratorTests {
 		}
 	}
 
+	public static class FunkyTableNameTest {
+
+		public static final List<String> FUNKY_LEGAL_NAMES;
+
+		static {
+			List<String> funkies = new ArrayList<String>(Arrays.asList(new String[] { /* TODO */}));
+			// TODO: should these work? "a \"\" x", "a\"\"\"\"x", "a b"
+			for (ReservedKeyword funky : ReservedKeyword.values()) {
+				funkies.add(funky.name());
+			}
+			FUNKY_LEGAL_NAMES = Collections.unmodifiableList(funkies);
+		}
+
+		@Test
+		public void test() {
+			for (String name : FUNKY_LEGAL_NAMES) {
+				new TableNameTest(name).test();
+			}
+		}
+	}
+
+	/**
+	 * This class is supposed to be used by other test classes.
+	 */
+	public static class TableNameTest extends CreateTableTest {
+
+		public String tableName;
+
+		public TableNameTest(String tableName) {
+			this.tableName = tableName;
+		}
+
+		@Override
+		public CreateTableSpecification specification() {
+			return CreateTableSpecification.createTable().name(tableName).partitionKeyColumn(cqlId("pk"), DataType.text());
+		}
+
+		/**
+		 * There is no @Test annotation on this method on purpose! It's supposed to be called by another test class's @Test
+		 * method so that you can loop, calling this test method as many times as are necessary.
+		 */
+		public void test() {
+			prepare();
+			assertPreamble(cqlId(tableName), cql);
+		}
+	}
 }

spring-cassandra/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/TableOperationCqlGeneratorTest.java

@@ -34,7 +34,6 @@ public abstract class TableOperationCqlGeneratorTest<S extends TableNameSpecific
 
 	public abstract G generator();
 
-	public String tableName;
 	public S specification;
 	public G generator;
 	public String cql;

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/integration/mapping/CassandraCompositePrimaryKeyIntegrationTests.java

@@ -145,14 +145,14 @@ public class CassandraCompositePrimaryKeyIntegrationTests {
 		List<ColumnSpecification> partitionKeyColumns = spec.getPartitionKeyColumns();
 		assertEquals(1, partitionKeyColumns.size());
 		ColumnSpecification partitionKeyColumn = partitionKeyColumns.get(0);
-		assertEquals("z", partitionKeyColumn.getName());
+		assertEquals("z", partitionKeyColumn.getName().toCql());
 		assertEquals(PrimaryKeyType.PARTITIONED, partitionKeyColumn.getKeyType());
 		assertEquals(DataType.text(), partitionKeyColumn.getType());
 
 		List<ColumnSpecification> clusteredKeyColumns = spec.getClusteredKeyColumns();
 		assertEquals(1, clusteredKeyColumns.size());
 		ColumnSpecification clusteredKeyColumn = clusteredKeyColumns.get(0);
-		assertEquals("a", clusteredKeyColumn.getName());
+		assertEquals("a", clusteredKeyColumn.getName().toCql());
 		assertEquals(PrimaryKeyType.CLUSTERED, clusteredKeyColumn.getKeyType());
 		assertEquals(DataType.text(), partitionKeyColumn.getType());
 	}