Added schema generation support for static columns.

Closes #978
Original pull request: #1108.

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

@@ -174,6 +174,8 @@ public class SchemaFactory {
 					specification.partitionKeyColumn(property.getRequiredColumnName(), type);
 				} else if (property.isClusterKeyColumn()) {
 					specification.clusteredKeyColumn(property.getRequiredColumnName(), type, property.getPrimaryKeyOrdering());
+				} else if (property.isStaticColumn()) {
+					specification.staticColumn(property.getRequiredColumnName(), type);
 				} else {
 					specification.column(property.getRequiredColumnName(), type);
 				}

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

@@ -77,7 +77,11 @@ public class CreateTableCqlGenerator extends TableOptionsCqlGenerator<TableSpeci
 		List<ColumnSpecification> partitionKeys = new ArrayList<>();
 		List<ColumnSpecification> clusterKeys = new ArrayList<>();
 		for (ColumnSpecification col : spec().getColumns()) {
-			col.toCql(cql).append(", ");
+			if (col.isStatic()) {
+				col.toCql(cql).append(" static, ");
+			} else {
+				col.toCql(cql).append(", ");
+			}
 
 			if (col.getKeyType() == PARTITIONED) {
 				partitionKeys.add(col);

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

@@ -29,10 +29,10 @@ import com.datastax.oss.driver.api.core.type.DataType;
 /**
  * Object to configure a CQL column specification.
  * <p/>
- * Use {@link #name(String)} and {@link #type(String)} to set the name and type of the column, respectively. To specify
+ * Use {@link #name(String)} and {@link #type(DataType)} to set the name and type of the column, respectively. To specify
  * a clustered {@code PRIMARY KEY} column, use {@link #clustered()} or {@link #clustered(Ordering)}. To specify that the
  * {@code PRIMARY KEY} column is or is part of the partition key, use {@link #partitioned()} instead of
- * {@link #clustered()} or {@link #clustered(Ordering)}.
+ * {@link #clustered()} or {@link #clustered(Ordering)}. To specify {@code STATIC} column, use {@link #staticColumn()}.
  *
  * @author Matthew T. Adams
  * @author Alex Shvid
@@ -53,6 +53,8 @@ public class ColumnSpecification {
 
 	private @Nullable Ordering ordering;
 
+	private boolean isStatic;
+
 	private ColumnSpecification(CqlIdentifier name) {
 		this.name = name;
 	}
@@ -178,6 +180,19 @@ public class ColumnSpecification {
 		return this;
 	}
 
+	/**
+	 * Identifies this column as a static column. Sets the column's {@link #isStatic} to {@literal true}.
+	 *
+	 * @return this
+	 * @since 3.2
+	 */
+	public ColumnSpecification staticColumn() {
+
+		this.isStatic = true;
+
+		return this;
+	}
+
 	public CqlIdentifier getName() {
 		return name;
 	}
@@ -197,6 +212,10 @@ public class ColumnSpecification {
 		return ordering;
 	}
 
+	public boolean isStatic() {
+		return isStatic;
+	}
+
 	public String toCql() {
 		return toCql(new StringBuilder()).toString();
 	}
@@ -211,7 +230,10 @@ public class ColumnSpecification {
 	 */
 	@Override
 	public String toString() {
-		return toCql(new StringBuilder()).append(" /* keyType=").append(keyType).append(", ordering=").append(ordering)
+		return toCql(new StringBuilder()).append(" /* ")
+				.append("keyType=").append(keyType).append(", ")
+				.append("ordering=").append(ordering).append(", ")
+				.append("isStatic=").append(isStatic)
 				.append(" */ ").toString();
 	}
 }

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

@@ -58,6 +58,11 @@ public interface TableDescriptor {
 	 */
 	List<ColumnSpecification> getNonKeyColumns();
 
+	/**
+	 * Returns an unmodifiable list of static columns.
+	 */
+	List<ColumnSpecification> getStaticColumns();
+
 	/**
 	 * Returns an unmodifiable {@link Map} of table options.
 	 */

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

@@ -59,6 +59,11 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 */
 	private List<ColumnSpecification> nonKeyColumns = new ArrayList<>();
 
+	/**
+	 * List of static columns.
+	 */
+	private List<ColumnSpecification> staticColumns = new ArrayList<>();
+
 	protected TableSpecification(CqlIdentifier name) {
 		super(name);
 	}
@@ -82,7 +87,31 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @param type The data type of the column, must not be {@literal null}.
 	 */
 	public T column(CqlIdentifier name, DataType type) {
-		return column(name, type, Optional.empty(), Optional.empty());
+		return column(name, type, Optional.empty(), Optional.empty(), false);
+	}
+
+	/**
+	 * Adds the given static column to the table. Must be specified after all primary key columns.
+	 *
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes,
+	 *          must not be {@literal null}.
+	 * @param type The data type of the column, must not be {@literal null}.
+	 * @since 3.2
+	 */
+	public T staticColumn(String name, DataType type) {
+		return staticColumn(CqlIdentifier.fromCql(name), type);
+	}
+
+	/**
+	 * Adds the given static column to the table. Must be specified after all primary key columns.
+	 *
+	 * @param name The column name; must be a valid unquoted or quoted identifier without the surrounding double quotes,
+	 *          must not be {@literal null}.
+	 * @param type The data type of the column, must not be {@literal null}.
+	 * @since 3.2
+	 */
+	public T staticColumn(CqlIdentifier name, DataType type) {
+		return column(name, type, Optional.empty(), Optional.empty(), true);
 	}
 
 	/**
@@ -106,7 +135,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @return this
 	 */
 	public T partitionKeyColumn(CqlIdentifier name, DataType type) {
-		return column(name, type, Optional.of(PARTITIONED), Optional.empty());
+		return column(name, type, Optional.of(PARTITIONED), Optional.empty(), false);
 	}
 
 	/**
@@ -137,7 +166,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 
 		Assert.notNull(ordering, "Ordering must not be null");
 
-		return column(CqlIdentifier.fromCql(name), type, Optional.of(CLUSTERED), Optional.of(ordering));
+		return column(CqlIdentifier.fromCql(name), type, Optional.of(CLUSTERED), Optional.of(ordering), false);
 	}
 
 	/**
@@ -167,7 +196,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 
 		Assert.notNull(ordering, "Ordering must not be null");
 
-		return column(name, type, Optional.of(CLUSTERED), Optional.of(ordering));
+		return column(name, type, Optional.of(CLUSTERED), Optional.of(ordering), false);
 	}
 
 	/**
@@ -181,7 +210,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	 * @return this
 	 */
 	public T clusteredKeyColumn(CqlIdentifier name, DataType type, Optional<Ordering> ordering) {
-		return column(name, type, Optional.of(CLUSTERED), ordering);
+		return column(name, type, Optional.of(CLUSTERED), ordering, false);
 	}
 
 	/**
@@ -215,7 +244,7 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 		Assert.notNull(keyType, "PrimaryKeyType must not be null");
 		Assert.notNull(ordering, "Ordering must not be null");
 
-		return column(CqlIdentifier.fromCql(name), type, Optional.of(keyType), Optional.of(ordering));
+		return column(CqlIdentifier.fromCql(name), type, Optional.of(keyType), Optional.of(ordering), false);
 	}
 
 	/**
@@ -235,17 +264,18 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 		Assert.notNull(keyType, "PrimaryKeyType must not be null");
 		Assert.notNull(ordering, "Ordering must not be null");
 
-		return column(CqlIdentifier.fromCql(name), type, Optional.of(keyType), ordering);
+		return column(CqlIdentifier.fromCql(name), type, Optional.of(keyType), ordering, false);
 	}
 
 	@SuppressWarnings("unchecked")
 	protected T column(CqlIdentifier name, DataType type, Optional<PrimaryKeyType> optionalKeyType,
-			Optional<Ordering> optionalOrdering) {
+			Optional<Ordering> optionalOrdering, boolean isStatic) {
 
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(type, "DataType must not be null");
 		Assert.notNull(optionalKeyType, "PrimaryKeyType must not be null");
 		Assert.notNull(optionalOrdering, "Ordering must not be null");
+		Assert.isTrue(!(optionalKeyType.isPresent() && isStatic),"PrimaryKey must not be static");
 
 		ColumnSpecification column = ColumnSpecification.name(name).type(type);
 
@@ -269,6 +299,11 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 			this.nonKeyColumns.add(column);
 		}
 
+		if (isStatic) {
+			column.staticColumn();
+			this.staticColumns.add(column);
+		}
+
 		return (T) this;
 	}
 
@@ -317,4 +352,14 @@ public class TableSpecification<T> extends TableOptionsSpecification<TableSpecif
 	public List<ColumnSpecification> getNonKeyColumns() {
 		return Collections.unmodifiableList(this.nonKeyColumns);
 	}
+
+	/**
+	 * Returns an unmodifiable list of static columns.
+	 *
+	 * @since 3.2
+	 */
+	@Override
+	public List<ColumnSpecification> getStaticColumns() {
+		return Collections.unmodifiableList(this.staticColumns);
+	}
 }

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

@@ -176,6 +176,16 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return isAnnotationPresent(PrimaryKeyColumn.class);
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isStaticColumn()
+	 */
+	@Override
+	public boolean isStaticColumn() {
+		Column annotation = findAnnotation(Column.class);
+
+		return annotation != null && annotation.isStatic();
+	}
+
 	@Nullable
 	private CqlIdentifier determineColumnName() {
 

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

@@ -135,6 +135,14 @@ public class BasicCassandraPersistentTupleProperty extends BasicCassandraPersist
 		return false;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isStaticColumn()
+	 */
+	@Override
+	public boolean isStaticColumn() {
+		return false;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isEmbedded()
 	 */

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

@@ -34,6 +34,7 @@ public class CachingCassandraPersistentProperty extends BasicCassandraPersistent
 	private final boolean isPartitionKeyColumn;
 	private final boolean isPrimaryKeyColumn;
 	private final boolean isEmbedded;
+	private final boolean isStaticColumn;
 
 	public CachingCassandraPersistentProperty(Property property, CassandraPersistentEntity<?> owner,
 			SimpleTypeHolder simpleTypeHolder) {
@@ -45,6 +46,7 @@ public class CachingCassandraPersistentProperty extends BasicCassandraPersistent
 		isPartitionKeyColumn = super.isPartitionKeyColumn();
 		isPrimaryKeyColumn = super.isPrimaryKeyColumn();
 		isEmbedded = super.isEmbedded();
+		isStaticColumn = super.isStaticColumn();
 	}
 
 	/* (non-Javadoc)
@@ -88,6 +90,14 @@ public class CachingCassandraPersistentProperty extends BasicCassandraPersistent
 		return isPrimaryKeyColumn;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentProperty#isStaticColumn()
+	 */
+	@Override
+	public boolean isStaticColumn() {
+		return isStaticColumn;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentProperty#isEmbedded()
 	 */

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

@@ -159,6 +159,12 @@ public interface CassandraPersistentProperty
 	 */
 	boolean isPrimaryKeyColumn();
 
+	/**
+	 * Whether the property is a static column.
+	 * @since 3.2
+	 */
+	boolean isStaticColumn();
+
 	/**
 	 * @return {@literal true} if the property should be embedded.
 	 * @since 3.0

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

@@ -53,6 +53,12 @@ public @interface Column {
 	 */
 	String value() default "";
 
+	/**
+	 * Whether the column is static.
+	 * Default is {@literal false}.
+	 */
+	boolean isStatic() default false;
+
 	/**
 	 * Whether to cause the column name to be force-quoted.
 	 *

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

@@ -404,6 +404,11 @@ public class EmbeddedEntityOperations {
 			return delegate.isEmbedded();
 		}
 
+		@Override
+		public boolean isStaticColumn() {
+			return delegate.isStaticColumn();
+		}
+
 		@Override
 		@org.springframework.lang.Nullable
 		public AnnotatedType findAnnotatedType(Class<? extends Annotation> annotationType) {

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

@@ -875,4 +875,53 @@ public class SchemaFactoryUnitTests {
 		assertThat(aegolastname.getColumnName()).isEqualTo(CqlIdentifier.fromCql("aegolastname"));
 	}
 
+	@Table
+	@Data
+	static class TypeWithStatic {
+
+		@Id String id;
+		@Column(isStatic = true) String name;
+	}
+
+	@Table
+	@Data
+	static class TypeWithStaticTuple {
+
+		@Id String id;
+
+		@Column(isStatic = true) Address address;
+	}
+
+	@Tuple
+	static class Address {
+		@Element(0) String street;
+		@Element(1) int number;
+	}
+
+	@Test // DATACASS-812
+	void createdTableSpecificationShouldConsiderStaticColumns() {
+
+		CassandraPersistentEntity<?> persistentEntity = ctx
+				.getRequiredPersistentEntity(TypeWithStatic.class);
+
+		CreateTableSpecification tableSpecification = schemaFactory.getCreateTableSpecificationFor(persistentEntity);
+
+		ColumnSpecification name = tableSpecification.getStaticColumns().get(0);
+		assertThat(name.getName().toString()).isEqualTo("name");
+		assertThat(name.isStatic()).isTrue();
+	}
+
+	@Test // DATACASS-812
+	void createdTableSpecificationShouldConsiderStaticForTypedTupleColumns() {
+
+		CassandraPersistentEntity<?> persistentEntity = ctx
+				.getRequiredPersistentEntity(TypeWithStaticTuple.class);
+
+		CreateTableSpecification tableSpecification = schemaFactory.getCreateTableSpecificationFor(persistentEntity);
+
+		ColumnSpecification name = tableSpecification.getStaticColumns().get(0);
+		assertThat(name.getName().toString()).isEqualTo("address");
+		assertThat(name.isStatic()).isTrue();
+	}
+
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/generator/CreateTableCqlGeneratorUnitTests.java

@@ -188,6 +188,21 @@ class CreateTableCqlGeneratorUnitTests {
 				+ "WITH CLUSTERING ORDER BY (date_of_birth ASC) AND COMPACT STORAGE;");
 	}
 
+	@Test // DATACASS-812
+	void createTableWithStaticColumns() {
+
+		CreateTableSpecification table = CreateTableSpecification.createTable("person")
+				.partitionKeyColumn("id", DataTypes.ASCII)
+				.clusteredKeyColumn("date_of_birth", DataTypes.DATE, Ordering.ASCENDING)
+				.column("name", DataTypes.ASCII)
+				.staticColumn("country", DataTypes.ASCII);
+
+		assertThat(toCql(table)).isEqualTo("CREATE TABLE person ("
+				+ "id ascii, date_of_birth date, name ascii, country ascii static, "
+				+ "PRIMARY KEY (id, date_of_birth)) "
+				+ "WITH CLUSTERING ORDER BY (date_of_birth ASC);");
+	}
+
 	/**
 	 * Asserts that the preamble is first & correctly formatted in the given CQL string.
 	 */