DATACASS-155 - Support TIMESTAMP in WriteOptions.

We now allow configuration of the write timestamp for INSERT/UPDATE/DELETE statements.

Original pull request: #122.

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

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.concurrent.TimeUnit;
 
 import lombok.EqualsAndHashCode;
@@ -40,9 +41,9 @@ public class InsertOptions extends WriteOptions {
 	private boolean ifNotExists;
 
 	private InsertOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy,
-			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, boolean ifNotExists) {
+			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp, boolean ifNotExists) {
 
-		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl);
+		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl, timestamp);
 
 		this.ifNotExists = ifNotExists;
 	}
@@ -152,6 +153,16 @@ public class InsertOptions extends WriteOptions {
 			return (InsertOptionsBuilder) super.ttl(ttl);
 		}
 
+		@Override
+		public InsertOptionsBuilder timestamp(long timestamp) {
+			return (InsertOptionsBuilder) super.timestamp(timestamp);
+		}
+
+		@Override
+		public InsertOptionsBuilder timestamp(Instant timestamp) {
+			return (InsertOptionsBuilder) super.timestamp(timestamp);
+		}
+
 		/**
 		 * Use light-weight transactions by applying {@code IF NOT EXISTS}.
 		 *
@@ -181,7 +192,7 @@ public class InsertOptions extends WriteOptions {
 		 */
 		public InsertOptions build() {
 			return new InsertOptions(this.consistencyLevel, this.retryPolicy, this.tracing,
-					this.fetchSize, this.readTimeout, this.ttl, this.ifNotExists);
+					this.fetchSize, this.readTimeout, this.ttl, this.timestamp, this.ifNotExists);
 		}
 	}
 }

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

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.concurrent.TimeUnit;
 
 import lombok.EqualsAndHashCode;
@@ -40,9 +41,9 @@ public class UpdateOptions extends WriteOptions {
 	private boolean ifExists;
 
 	private UpdateOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy,
-			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, boolean ifExists) {
+			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp, boolean ifExists) {
 
-		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl);
+		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl, timestamp);
 
 		this.ifExists = ifExists;
 	}
@@ -152,6 +153,16 @@ public class UpdateOptions extends WriteOptions {
 			return (UpdateOptionsBuilder) super.ttl(ttl);
 		}
 
+		@Override
+		public UpdateOptionsBuilder timestamp(long timestamp) {
+			return (UpdateOptionsBuilder) super.timestamp(timestamp);
+		}
+
+		@Override
+		public UpdateOptionsBuilder timestamp(Instant timestamp) {
+			return (UpdateOptionsBuilder) super.timestamp(timestamp);
+		}
+
 		/**
 		 * Use light-weight transactions by applying {@code IF EXISTS}.
 		 *
@@ -181,7 +192,7 @@ public class UpdateOptions extends WriteOptions {
 		 */
 		public UpdateOptions build() {
 			return new UpdateOptions(this.consistencyLevel, this.retryPolicy, this.tracing,
-					this.fetchSize, this.readTimeout, this.ttl, this.ifExists);
+					this.fetchSize, this.readTimeout, this.ttl, this.timestamp, this.ifExists);
 		}
 	}
 }

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

@@ -106,6 +106,9 @@ public abstract class QueryOptionsUtil {
 		if (!writeOptions.getTtl().isNegative()) {
 			insert.using(QueryBuilder.ttl(Math.toIntExact(writeOptions.getTtl().getSeconds())));
 		}
+		if (writeOptions.getTimestamp() != null) {
+			insert.using(QueryBuilder.timestamp(writeOptions.getTimestamp()));
+		}
 
 		return insert;
 	}
@@ -126,6 +129,9 @@ public abstract class QueryOptionsUtil {
 		if (!writeOptions.getTtl().isNegative()) {
 			update.using(QueryBuilder.ttl(Math.toIntExact(writeOptions.getTtl().getSeconds())));
 		}
+		if (writeOptions.getTimestamp() != null) {
+			update.using(QueryBuilder.timestamp(writeOptions.getTimestamp()));
+		}
 
 		return update;
 	}

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

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core.cql;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.concurrent.TimeUnit;
 
 import lombok.EqualsAndHashCode;
@@ -40,6 +41,7 @@ public class WriteOptions extends QueryOptions {
 	private static final WriteOptions EMPTY = new WriteOptionsBuilder().build();
 
 	private final Duration ttl;
+	private final @Nullable Long timestamp;
 
 	/**
 	 * Creates new {@link WriteOptions} for the given {@link ConsistencyLevel} and {@link RetryPolicy}.
@@ -68,14 +70,16 @@ public class WriteOptions extends QueryOptions {
 		super(consistencyLevel, retryPolicy);
 
 		this.ttl = ttl == null ? Duration.ofMillis(-1) : Duration.ofSeconds(ttl);
+		this.timestamp = null;
 	}
 
 	protected WriteOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy,
-			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl) {
+			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl, Long timestamp) {
 
 		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout);
 
 		this.ttl = ttl;
+		this.timestamp = timestamp;
 	}
 
 	/**
@@ -116,6 +120,13 @@ public class WriteOptions extends QueryOptions {
 		return this.ttl;
 	}
 
+	/**
+	 * @return mutation timestamp in microseconds.
+	 */
+	public Long getTimestamp() {
+		return this.timestamp;
+	}
+
 	/**
 	 * Builder for {@link WriteOptions}.
 	 *
@@ -125,6 +136,7 @@ public class WriteOptions extends QueryOptions {
 	public static class WriteOptionsBuilder extends QueryOptionsBuilder {
 
 		protected Duration ttl = Duration.ofMillis(-1);
+		protected Long timestamp = null;
 
 		protected WriteOptionsBuilder() {}
 
@@ -133,6 +145,7 @@ public class WriteOptions extends QueryOptions {
 			super(writeOptions);
 
 			this.ttl = writeOptions.ttl;
+			this.timestamp = writeOptions.timestamp;
 		}
 
 		/*
@@ -236,6 +249,30 @@ public class WriteOptions extends QueryOptions {
 			return this;
 		}
 
+		/**
+		 * Sets the timestamp of write operations.
+		 *
+		 * @param timestamp mutation timestamp in microseconds.
+		 * @return {@code this} {@link WriteOptionsBuilder}
+		 */
+		public WriteOptionsBuilder timestamp(long timestamp) {
+			Assert.isTrue(timestamp >= 0, "Timestamp must be greater than equal to zero");
+			this.timestamp = timestamp;
+			return this;
+		}
+
+		/**
+		 * Sets the timestamp of write operations.
+		 *
+		 * @param timestamp mutation date time.
+		 * @return {@code this} {@link WriteOptionsBuilder}
+		 */
+		public WriteOptionsBuilder timestamp(Instant timestamp) {
+			Assert.notNull(timestamp, "Timestamp must not be null");
+			this.timestamp = TimeUnit.MILLISECONDS.toMicros(timestamp.toEpochMilli());
+			return this;
+		}
+
 		/**
 		 * Builds a new {@link WriteOptions} with the configured values.
 		 *
@@ -243,7 +280,7 @@ public class WriteOptions extends QueryOptions {
 		 */
 		public WriteOptions build() {
 			return new WriteOptions(this.consistencyLevel, this.retryPolicy, this.tracing,
-					this.fetchSize, this.readTimeout, this.ttl);
+					this.fetchSize, this.readTimeout, this.ttl, this.timestamp);
 		}
 	}
 }

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

@@ -19,6 +19,9 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assume.assumeTrue;
 import static org.springframework.data.cassandra.core.query.Criteria.where;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
@@ -191,6 +194,24 @@ public class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingI
 		assertThat(count).isEqualTo(1L);
 	}
 
+	@Test // DATACASS-155
+	public void shouldNotOverrideLaterMutation() {
+		Instant now = LocalDateTime.now().atZone( ZoneId.systemDefault() ).toInstant();
+		User user = new User("heisenberg", "Walter", "White");
+		template.insert(user);
+
+		// more recent mutation
+		user.setFirstname("John");
+		template.update(user);
+
+		// previous mutation
+		user.setFirstname("Greg");
+		template.update(user, UpdateOptions.builder().timestamp(now.minusSeconds(10)).build());
+
+		User loaded = template.selectOneById(user.getId(), User.class);
+		assertThat(loaded.getFirstname()).isEqualTo("John");
+	}
+
 	@Test // DATACASS-512
 	public void shouldInsertEntityAndCountByQuery() {
 

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

@@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 
 import org.junit.Test;
 
@@ -30,13 +33,16 @@ public class InsertOptionsUnitTests {
 
 	@Test // DATACASS-250
 	public void shouldConfigureInsertOptions() {
+		Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC);
 
 		InsertOptions insertOptions = InsertOptions.builder()
 				.ttl(10)
+				.timestamp(now)
 				.withIfNotExists()
 				.build();
 
 		assertThat(insertOptions.getTtl()).isEqualTo(Duration.ofSeconds(10));
+		assertThat(insertOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(insertOptions.isIfNotExists()).isTrue();
 	}
 
@@ -45,14 +51,16 @@ public class InsertOptionsUnitTests {
 
 		InsertOptions insertOptions = InsertOptions.builder()
 				.ttl(10)
+				.timestamp(1519222753)
 				.withIfNotExists()
 				.build();
 
-		InsertOptions mutated = insertOptions.mutate().ttl(Duration.ofSeconds(5)).build();
+		InsertOptions mutated = insertOptions.mutate().ttl(Duration.ofSeconds(5)).timestamp(1519200753).build();
 
 		assertThat(mutated).isNotNull();
 		assertThat(mutated).isNotSameAs(insertOptions);
 		assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(5));
+		assertThat(mutated.getTimestamp()).isEqualTo(1519200753);
 		assertThat(mutated.isIfNotExists()).isTrue();
 	}
 }

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

@@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 
 import org.junit.Test;
 
@@ -30,13 +33,16 @@ public class UpdateOptionsUnitTests {
 
 	@Test // DATACASS-250
 	public void shouldConfigureUpdateOptions() {
+		Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC);
 
 		UpdateOptions updateOptions = UpdateOptions.builder()
 				.ttl(10)
+				.timestamp(now)
 				.withIfExists()
 				.build();
 
 		assertThat(updateOptions.getTtl()).isEqualTo(Duration.ofSeconds(10));
+		assertThat(updateOptions.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(updateOptions.isIfExists()).isTrue();
 	}
 
@@ -45,14 +51,16 @@ public class UpdateOptionsUnitTests {
 
 		UpdateOptions updateOptions = UpdateOptions.builder()
 				.ttl(10)
+				.timestamp(1519222753)
 				.withIfExists()
 				.build();
 
-		UpdateOptions mutated = updateOptions.mutate().ttl(20).build();
+		UpdateOptions mutated = updateOptions.mutate().ttl(20).timestamp(1519000753).build();
 
 		assertThat(mutated).isNotNull();
 		assertThat(mutated).isNotSameAs(updateOptions);
 		assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(20));
+		assertThat(mutated.getTimestamp()).isEqualTo(1519000753);
 		assertThat(mutated.isIfExists()).isTrue();
 	}
 }

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

@@ -18,6 +18,9 @@ package org.springframework.data.cassandra.core.cql;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 
 import org.junit.Test;
 
@@ -38,6 +41,7 @@ public class WriteOptionsUnitTests {
 		WriteOptions writeOptions = WriteOptions.builder()
 				.consistencyLevel(com.datastax.driver.core.ConsistencyLevel.ANY)
 				.ttl(123)
+				.timestamp(1519000753)
 				.retryPolicy(FallthroughRetryPolicy.INSTANCE)
 				.readTimeout(1)
 				.fetchSize(10)
@@ -45,6 +49,7 @@ public class WriteOptionsUnitTests {
 				.build();
 
 		assertThat(writeOptions.getTtl()).isEqualTo(Duration.ofSeconds(123));
+		assertThat(writeOptions.getTimestamp()).isEqualTo(1519000753);
 		assertThat(writeOptions.getRetryPolicy()).isEqualTo(FallthroughRetryPolicy.INSTANCE);
 		assertThat(writeOptions.getConsistencyLevel()).isEqualTo(ConsistencyLevel.ANY);
 		assertThat(writeOptions.getReadTimeout()).isEqualTo(Duration.ofMillis(1));
@@ -80,10 +85,12 @@ public class WriteOptionsUnitTests {
 
 	@Test // DATACASS-56
 	public void buildWriteOptionsMutate() {
+		Instant now = LocalDateTime.now().toInstant(ZoneOffset.UTC);
 
 		WriteOptions writeOptions = WriteOptions.builder()
 				.consistencyLevel(com.datastax.driver.core.ConsistencyLevel.ANY)
 				.ttl(123)
+				.timestamp(now)
 				.retryPolicy(FallthroughRetryPolicy.INSTANCE)
 				.readTimeout(1)
 				.fetchSize(10)
@@ -95,6 +102,7 @@ public class WriteOptionsUnitTests {
 		assertThat(mutated).isNotNull();
 		assertThat(mutated).isNotSameAs(writeOptions);
 		assertThat(mutated.getTtl()).isEqualTo(Duration.ofSeconds(123));
+		assertThat(mutated.getTimestamp()).isEqualTo(now.toEpochMilli() * 1000);
 		assertThat(mutated.getRetryPolicy()).isEqualTo(DowngradingConsistencyRetryPolicy.INSTANCE);
 		assertThat(mutated.getConsistencyLevel()).isEqualTo(ConsistencyLevel.ANY);
 		assertThat(mutated.getReadTimeout()).isEqualTo(Duration.ofMillis(1));