Correctly apply TableOptions through CassandraAdminTemplate.createTable(…).

Closes: #359
Original pull request: #1385

Author: mipo256

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

@@ -55,6 +55,7 @@ public interface CassandraAdminOperations extends CassandraOperations {
 	void createTable(boolean ifNotExists, CqlIdentifier tableName, Class<?> entityClass,
 			Map<String, Object> optionsByName);
 
+
 	/**
 	 * Drops a table based on the given {@link Class entity type}. The name of the table is derived from either the simple
 	 * name of the {@link Class entity class} or name of the table specified with the {@link Table} mapping annotation.

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

@@ -29,8 +29,10 @@
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.DropTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.DropUserTypeSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.TableOption;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
@@ -44,6 +46,7 @@
  * @author Fabio J. Mendes
  * @author John Blum
  * @author Vagif Zeynalov
+ * @author Mikhail Polivakha
  */
 public class CassandraAdminTemplate extends CassandraTemplate implements CassandraAdminOperations {
 
@@ -108,13 +111,23 @@ public SchemaFactory getSchemaFactory() {
 	 * @see org.springframework.data.cassandra.core.CassandraAdminOperations#createTable(boolean, org.springframework.data.cassandra.core.cql.CqlIdentifier, java.lang.Class, java.util.Map)
 	 */
 	@Override
-	public void createTable(boolean ifNotExists, CqlIdentifier tableName, Class<?> entityClass,
-			Map<String, Object> optionsByName) {
-
+	public void createTable(boolean ifNotExists, CqlIdentifier tableName, Class<?> entityClass, Map<String, Object> optionsByName) {
 		CassandraPersistentEntity<?> entity = getConverter().getMappingContext().getRequiredPersistentEntity(entityClass);
 
 		CreateTableSpecification createTableSpecification = this.schemaFactory
-				.getCreateTableSpecificationFor(entity, tableName).ifNotExists(ifNotExists);
+				.getCreateTableSpecificationFor(entity, tableName)
+				.ifNotExists(ifNotExists);
+
+		if (!CollectionUtils.isEmpty(optionsByName)) {
+			optionsByName.forEach((key, value) -> {
+				TableOption tableOption = TableOption.valueOfIgnoreCase(key);
+				if (tableOption.requiresValue()) {
+					createTableSpecification.with(tableOption, value);
+				} else {
+					createTableSpecification.with(tableOption);
+				}
+			});
+		}
 
 		getCqlOperations().execute(CreateTableCqlGenerator.toCql(createTableSpecification));
 	}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java

@@ -196,7 +196,7 @@ interface AdaptibleEntity<T> extends Entity<T> {
 		StatementBuilder<Delete> appendVersionCondition(StatementBuilder<Delete> delete);
 
 		/**
-		 * Initializes the version property of the of the current entity if available.
+		 * Initializes the version property of the current entity if available.
 		 *
 		 * @return the entity with the version property updated if available.
 		 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java

@@ -295,6 +295,7 @@ StatementBuilder<RegularInsert> insert(Object objectToInsert, WriteOptions optio
 		Assert.notNull(persistentEntity, "CassandraPersistentEntity must not be null");
 
 		boolean insertNulls;
+
 		if (options instanceof InsertOptions) {
 
 			InsertOptions insertOptions = (InsertOptions) options;

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java

@@ -298,12 +298,12 @@ public ColumnTypeResolver getColumnTypeResolver() {
 	@SuppressWarnings("unchecked")
 	private <S> ConvertingPropertyAccessor<S> newConvertingPropertyAccessor(S source,
 			CassandraPersistentEntity<?> entity) {
-
 		PersistentPropertyAccessor<S> propertyAccessor = source instanceof PersistentPropertyAccessor
-				? (PersistentPropertyAccessor<S>) source
-				: entity.getPropertyAccessor(source);
+						? (PersistentPropertyAccessor<S>) source
+						: entity.getPropertyAccessor(source);
 
 		return new ConvertingPropertyAccessor<>(propertyAccessor, getConversionService());
+
 	}
 
 	private <S> CassandraPersistentEntityParameterValueProvider newParameterValueProvider(ConversionContext context,

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/SchemaFactory.java

@@ -147,7 +147,7 @@ public CreateTableSpecification getCreateTableSpecificationFor(CassandraPersiste
 			if (property.isCompositePrimaryKey()) {
 
 				CassandraPersistentEntity<?> primaryKeyEntity = mappingContext
-						.getRequiredPersistentEntity(property.getRawType());
+								.getRequiredPersistentEntity(property.getRawType());
 
 				for (CassandraPersistentProperty primaryKeyProperty : primaryKeyEntity) {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/keyspace/TableOption.java

@@ -18,6 +18,7 @@
 import java.util.Map;
 
 import org.springframework.lang.Nullable;
+import org.springframework.util.StringUtils;
 
 /**
  * Enumeration that represents all known table options. If a table option is not listed here, but is supported by
@@ -26,6 +27,7 @@
  *
  * @author Matthew T. Adams
  * @author Mark Paluch
+ * @author Mikhail Polivakha
  * @see CompactionOption
  * @see CompressionOption
  * @see CachingOption
@@ -84,6 +86,15 @@ public enum TableOption implements Option {
 		this.delegate = new DefaultOption(name, type, requiresValue, escapesValue, quotesValue);
 	}
 
+	public static TableOption valueOfIgnoreCase(String optionName) {
+		for (TableOption value : values()) {
+			if (value.getName().equalsIgnoreCase(optionName)) {
+				return value;
+			}
+		}
+		throw new IllegalArgumentException(String.format("Unable to recognize specified Table option '%s'", optionName));
+	}
+
 	@Override
 	public Class<?> getType() {
 		return this.delegate.getType();

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

@@ -296,7 +296,7 @@ protected T column(CqlIdentifier name, DataType type, Optional<PrimaryKeyType> o
 
 		this.columns.add(column);
 
-		if (!optionalKeyType.isPresent()) {
+		if (optionalKeyType.isEmpty()) {
 			this.nonKeyColumns.add(column);
 		}
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraAdminTemplateIntegrationTests.java

@@ -17,13 +17,19 @@
 
 import static org.assertj.core.api.Assertions.*;
 
+import java.time.LocalDate;
 import java.util.Collection;
+import java.util.Map;
 
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.cql.generator.DropTableCqlGenerator;
 import org.springframework.data.cassandra.core.cql.keyspace.DropTableSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.TableOption;
+import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.cassandra.domain.User;
 import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTests;
 
@@ -36,6 +42,7 @@
  * Integration tests for {@link CassandraAdminTemplate}.
  *
  * @author Mark Paluch
+ * @author Mikhail Polivakha
  */
 class CassandraAdminTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTests {
 
@@ -59,6 +66,27 @@ private KeyspaceMetadata getKeyspaceMetadata() {
 		return getSession().getKeyspace().flatMap(metadata::getKeyspace).get();
 	}
 
+	@Test
+	void givenAdminTemplate_whenCreateTableWithOptions_ThenCreatedTableContainsTheseOptions() {
+		cassandraAdminTemplate.createTable(
+				true,
+				CqlIdentifier.fromCql("someTable"),
+				SomeTable.class,
+				Map.of(
+					TableOption.COMMENT.getName(), "This is comment for table",
+					TableOption.BLOOM_FILTER_FP_CHANCE.getName(), "0.3"
+				)
+		);
+
+		TableMetadata someTable = getKeyspaceMetadata().getTables().values().stream().findFirst().orElse(null);
+
+		Assertions.assertThat(someTable).isNotNull();
+		Assertions.assertThat(someTable.getOptions().get(CqlIdentifier.fromCql(TableOption.COMMENT.getName())))
+				.isEqualTo("This is comment for table");
+		Assertions.assertThat(someTable.getOptions().get(CqlIdentifier.fromCql(TableOption.BLOOM_FILTER_FP_CHANCE.getName())))
+				.isEqualTo(0.3);
+	}
+
 	@Test // DATACASS-173
 	void testCreateTables() {
 
@@ -85,4 +113,13 @@ void testDropTable() {
 
 		assertThat(getKeyspaceMetadata().getTables()).hasSize(0);
 	}
+
+	@Table("someTable")
+	private static class SomeTable {
+
+		@Id
+		private String name;
+		private Integer number;
+		private LocalDate createdAt;
+	}
 }