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

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

@@ -20,56 +20,60 @@ import java.util.Map;
 import com.datastax.driver.core.TableMetadata;
 
 /**
- * @author David Webb
+ * Operations for managing a Cassandra keyspace.
  * 
+ * @author David Webb
+ * @author Matthew T. Adams
  */
 public interface CassandraAdminOperations {
 
 	/**
-	 * Get the Table Meta Data from Cassandra
+	 * Get the given table's metadata.
 	 * 
-	 * @param entityClass
-	 * @param tableName
-	 * @return
+	 * @param tableName The name of the table.
 	 */
-	TableMetadata getTableMetadata(Class<?> entityClass, String tableName);
+	TableMetadata getTableMetadata(String tableName);
 
 	/**
-	 * Create a table with the name and fields indicated by the entity class
+	 * Create a table with the name given and fields corresponding to the given class. If the table already exists and
+	 * parameter <code>ifNotExists</code> is {@literal true}, this is a no-op and {@literal false} is returned. If the
+	 * table doesn't exist, parameter <code>ifNotExists</code> is ignored, the table is created and {@literal true} is
+	 * returned.
 	 * 
-	 * @param ifNotExists
-	 * @param tableName
-	 * @param entityClass
-	 * @param optionsByName
+	 * @param ifNotExists If true, will only create the table if it doesn't exist, else the create operation will be
+	 *          ignored and the method will return {@literal false}.
+	 * @param tableName The name of the table.
+	 * @param entityClass The class whose fields determine the columns created.
+	 * @param optionsByName Table options, given by the string option name and the appropriate option value.
+	 * @return Returns true if a table was created, false if not.
 	 */
-	void createTable(boolean ifNotExists, String tableName, Class<?> entityClass, Map<String, Object> optionsByName);
+	boolean createTable(boolean ifNotExists, String tableName, Class<?> entityClass, Map<String, Object> optionsByName);
 
 	/**
-	 * Alter table with the name and fields indicated by the entity class
+	 * Add columns to the given table from the given class. If parameter dropRemovedAttributColumns is true, then this
+	 * effectively becomes a synchronization operation between the class's fields and the existing table's columns.
 	 * 
-	 * @param entityClass class that determines metadata of the table to create/drop.
-	 * @param tableName explicit name of the table
+	 * @param tableName The name of the existing table.
+	 * @param entityClass The class whose fields determine the columns added.
+	 * @param dropRemovedAttributeColumns Whether to drop columns that exist on the table but that don't have
+	 *          corresponding fields in the class. If true, this effectively becomes a synchronziation operation.
 	 */
 	void alterTable(String tableName, Class<?> entityClass, boolean dropRemovedAttributeColumns);
 
 	/**
-	 * @param tableName
-	 * @param entityClass
+	 * Drops the existing table with the given name and creates a new one; basically a {@link #dropTable(String)} followed
+	 * by a {@link #createTable(boolean, String, Class, Map)}.
+	 * 
+	 * @param tableName The name of the table.
+	 * @param entityClass The class whose fields determine the new table's columns.
+	 * @param optionsByName Table options, given by the string option name and the appropriate option value.
 	 */
-	void replaceTable(String tableName, Class<?> entityClass);
+	void replaceTable(String tableName, Class<?> entityClass, Map<String, Object> optionsByName);
 
 	/**
-	 * Alter table with the name and fields indicated by the entity class
+	 * Drops the named table.
 	 * 
-	 * @param entityClass class that determines metadata of the table to create/drop.
-	 */
-	void dropTable(Class<?> entityClass);
-
-	/**
-	 * Alter table with the name and fields indicated by the entity class
-	 * 
-	 * @param tableName explicit name of the table.
+	 * @param tableName The name of the table.
 	 */
 	void dropTable(String tableName);
-
 }

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

@@ -9,6 +9,7 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.cassandra.convert.CassandraConverter;
+import org.springframework.data.cassandra.core.exceptions.CassandraTableExistsException;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.util.CqlUtils;
@@ -20,16 +21,16 @@ import com.datastax.driver.core.Session;
 import com.datastax.driver.core.TableMetadata;
 
 /**
- *
+ * Default implementation of {@link CassandraAdminOperations}.
  */
-public class CassandraAdmin implements CassandraAdminOperations {
+public class CassandraAdminTemplate implements CassandraAdminOperations {
 
-	private static Logger log = LoggerFactory.getLogger(CassandraAdmin.class);
+	private static Logger log = LoggerFactory.getLogger(CassandraAdminTemplate.class);
 
-	private final Keyspace keyspace;
-	private final Session session;
-	private final CassandraConverter cassandraConverter;
-	private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
+	private Keyspace keyspace;
+	private Session session;
+	private CassandraConverter converter;
+	private MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
 
 	private final PersistenceExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();
 
@@ -40,18 +41,38 @@ public class CassandraAdmin implements CassandraAdminOperations {
 	 * 
 	 * @param keyspace must not be {@literal null}.
 	 */
-	public CassandraAdmin(Keyspace keyspace) {
+	public CassandraAdminTemplate(Keyspace keyspace) {
+		setKeyspace(keyspace);
+	}
+
+	protected CassandraAdminTemplate setKeyspace(Keyspace keyspace) {
+		Assert.notNull(keyspace);
 		this.keyspace = keyspace;
-		this.session = keyspace.getSession();
-		this.cassandraConverter = keyspace.getCassandraConverter();
-		this.mappingContext = this.cassandraConverter.getMappingContext();
+		return setSession(keyspace.getSession()).setCassandraConverter(keyspace.getCassandraConverter());
+	}
+
+	protected CassandraAdminTemplate setSession(Session session) {
+		Assert.notNull(session);
+		return this;
+	}
+
+	protected CassandraAdminTemplate setCassandraConverter(CassandraConverter converter) {
+		Assert.notNull(converter);
+		this.converter = converter;
+		return setMappingContext(converter.getMappingContext());
+	}
+
+	protected CassandraAdminTemplate setMappingContext(
+			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext) {
+		Assert.notNull(mappingContext);
+		return this;
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraAdminOperations#createTable(boolean, java.lang.String, java.lang.Class, java.util.Map)
 	 */
 	@Override
-	public void createTable(boolean ifNotExists, final String tableName, Class<?> entityClass,
+	public boolean createTable(boolean ifNotExists, final String tableName, Class<?> entityClass,
 			Map<String, Object> optionsByName) {
 
 		try {
@@ -59,25 +80,21 @@ public class CassandraAdmin implements CassandraAdminOperations {
 			final CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
 
 			execute(new SessionCallback<Object>() {
-
 				public Object doInSession(Session s) throws DataAccessException {
 
 					String cql = CqlUtils.createTable(tableName, entity);
-
 					log.info("CREATE TABLE CQL -> " + cql);
-
 					s.execute(cql);
-
 					return null;
-
 				}
 			});
+			return true;
 
-		} catch (LinkageError e) {
-			e.printStackTrace();
-		} finally {
+		} catch (CassandraTableExistsException ctex) {
+			return !ifNotExists;
+		} catch (RuntimeException x) {
+			throw tryToConvert(x);
 		}
-
 	}
 
 	/* (non-Javadoc)
@@ -86,16 +103,14 @@ public class CassandraAdmin implements CassandraAdminOperations {
 	@Override
 	public void alterTable(String tableName, Class<?> entityClass, boolean dropRemovedAttributeColumns) {
 		// TODO Auto-generated method stub
-
 	}
 
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraAdminOperations#replaceTable(java.lang.String, java.lang.Class)
 	 */
 	@Override
-	public void replaceTable(String tableName, Class<?> entityClass) {
-		// TODO Auto-generated method stub
-
+	public void replaceTable(String tableName, Class<?> entityClass, Map<String, Object> optionsByName) {
+		// TODO
 	}
 
 	/**
@@ -110,7 +125,7 @@ public class CassandraAdmin implements CassandraAdminOperations {
 
 		Assert.notNull(entity);
 
-		final TableMetadata tableMetadata = getTableMetadata(entityClass, tableName);
+		final TableMetadata tableMetadata = getTableMetadata(tableName);
 
 		final List<String> queryList = CqlUtils.alterTable(tableName, entity, tableMetadata);
 
@@ -133,7 +148,6 @@ public class CassandraAdmin implements CassandraAdminOperations {
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraOperations#dropTable(java.lang.Class)
 	 */
-	@Override
 	public void dropTable(Class<?> entityClass) {
 
 		final String tableName = determineTableName(entityClass);
@@ -170,31 +184,19 @@ public class CassandraAdmin implements CassandraAdminOperations {
 	 * @see org.springframework.data.cassandra.core.CassandraOperations#getTableMetadata(java.lang.Class)
 	 */
 	@Override
-	public TableMetadata getTableMetadata(Class<?> entityClass, String tableName) {
-
-		/*
-		 * Determine the table name if not provided
-		 */
-		if (tableName == null) {
-			tableName = determineTableName(entityClass);
-		}
+	public TableMetadata getTableMetadata(final String tableName) {
 
 		Assert.notNull(tableName);
 
-		final String metadataTableName = tableName;
-
 		return execute(new SessionCallback<TableMetadata>() {
 
 			public TableMetadata doInSession(Session s) throws DataAccessException {
 
 				log.info("Keyspace => " + keyspace.getKeyspace());
 
-				return s.getCluster().getMetadata().getKeyspace(keyspace.getKeyspace()).getTable(metadataTableName);
-
+				return s.getCluster().getMetadata().getKeyspace(keyspace.getKeyspace()).getTable(tableName);
 			}
-
 		});
-
 	}
 
 	/**
@@ -208,17 +210,15 @@ public class CassandraAdmin implements CassandraAdminOperations {
 		Assert.notNull(callback);
 
 		try {
-
 			return callback.doInSession(session);
-
-		} catch (DataAccessException e) {
-			throw potentiallyConvertRuntimeException(e);
+		} catch (RuntimeException x) {
+			throw tryToConvert(x);
 		}
 	}
 
-	private RuntimeException potentiallyConvertRuntimeException(RuntimeException ex) {
-		RuntimeException resolved = this.exceptionTranslator.translateExceptionIfPossible(ex);
-		return resolved == null ? ex : resolved;
+	protected RuntimeException tryToConvert(RuntimeException x) {
+		RuntimeException resolved = exceptionTranslator.translateExceptionIfPossible(x);
+		return resolved == null ? x : resolved;
 	}
 
 	/**

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

@@ -18,19 +18,19 @@ package org.springframework.data.cassandra.core;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
 import org.springframework.data.cassandra.core.exceptions.CassandraAuthenticationException;
+import org.springframework.data.cassandra.core.exceptions.CassandraConnectionFailureException;
+import org.springframework.data.cassandra.core.exceptions.CassandraInsufficientReplicasAvailableException;
 import org.springframework.data.cassandra.core.exceptions.CassandraInternalException;
 import org.springframework.data.cassandra.core.exceptions.CassandraInvalidConfigurationInQueryException;
 import org.springframework.data.cassandra.core.exceptions.CassandraInvalidQueryException;
-import org.springframework.data.cassandra.core.exceptions.CassandraTypeMismatchException;
 import org.springframework.data.cassandra.core.exceptions.CassandraKeyspaceExistsException;
-import org.springframework.data.cassandra.core.exceptions.CassandraConnectionFailureException;
-import org.springframework.data.cassandra.core.exceptions.CassandraReadTimeoutException;
 import org.springframework.data.cassandra.core.exceptions.CassandraQuerySyntaxException;
+import org.springframework.data.cassandra.core.exceptions.CassandraReadTimeoutException;
 import org.springframework.data.cassandra.core.exceptions.CassandraTableExistsException;
 import org.springframework.data.cassandra.core.exceptions.CassandraTraceRetrievalException;
 import org.springframework.data.cassandra.core.exceptions.CassandraTruncateException;
+import org.springframework.data.cassandra.core.exceptions.CassandraTypeMismatchException;
 import org.springframework.data.cassandra.core.exceptions.CassandraUnauthorizedException;
-import org.springframework.data.cassandra.core.exceptions.CassandraInsufficientReplicasAvailableException;
 import org.springframework.data.cassandra.core.exceptions.CassandraUncategorizedException;
 import org.springframework.data.cassandra.core.exceptions.CassandraWriteTimeoutException;
 
@@ -74,6 +74,10 @@ public class CassandraExceptionTranslator implements PersistenceExceptionTransla
 			return null;
 		}
 
+		if (x instanceof DataAccessException) {
+			return (DataAccessException) x;
+		}
+
 		// Remember: subclasses must come before superclasses, otherwise the
 		// superclass would match before the subclass!
 

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

@@ -0,0 +1,6 @@
+package org.springframework.data.cassandra.core;
+
+import org.springframework.core.convert.converter.Converter;
+
+public interface ClassNameToTableNameConverter extends Converter<String, String> {
+}

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

@@ -0,0 +1,6 @@
+package org.springframework.data.cassandra.core;
+
+import org.springframework.core.convert.converter.Converter;
+
+public interface ColumnNameToFieldNameConverter extends Converter<String, String> {
+}

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

@@ -0,0 +1,6 @@
+package org.springframework.data.cassandra.core;
+
+import org.springframework.core.convert.converter.Converter;
+
+public interface FieldNameToColumnNameConverter extends Converter<String, String> {
+}

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

@@ -0,0 +1,6 @@
+package org.springframework.data.cassandra.core;
+
+import org.springframework.core.convert.converter.Converter;
+
+public interface TableNameToClassNameConverter extends Converter<String, String> {
+}

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

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

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

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

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

@@ -0,0 +1,42 @@
+package org.springframework.data.cassandra.cql;
+
+import static org.springframework.data.cassandra.cql.CqlBuilder.createTable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+import org.springframework.data.cassandra.cql.CreateTable.Column.Order;
+
+public class CreateTableTest {
+
+	@Test
+	public void createTableTest() {
+		String name = "mytable";
+		String type0 = "text";
+		String partition0 = "partitionKey0";
+		String partition1 = "partitionKey1";
+		String primary0 = "primary0";
+		String type1 = "text";
+		String column1 = "column1";
+		String type2 = "text";
+		String column2 = "column2";
+		Object value1 = null;
+		String option1 = "COMPACT STORAGE";
+		Object value2 = "this is a comment";
+		String option2 = "comment";
+		Object value3 = "0.00075";
+		String option3 = "bloom_filter_fp_chance";
+		Map<String, Object> value4 = new HashMap<String, Object>();
+		value4.put("class", "LeveledCompactionStrategy");
+		String option4 = "compaction";
+
+		CreateTable builder = createTable().ifNotExists().name(name).partition(partition0, type0, Order.ASCENDING)
+				.partition(partition1, type0, Order.DESCENDING).primary(primary0, type0).column(column1, type1)
+				.column(column2, type2).option(option1, value1).option(option2, value2).option(option3, value3)
+				.option(option4, value4);
+
+		String cql = builder.cql();
+		System.out.println(cql);
+	}
+}