From e121b441881202b8f83ede5c8b4fb4c56d79106e Mon Sep 17 00:00:00 2001
From: samueldlightfoot <samueldlightfoot@gmail.com>
Date: Mon, 11 Oct 2021 17:52:14 +0100
Subject: [PATCH] Use bindMarkers for BY_INDEX and BY_NAME IN query generation.

See #1172
---
 .../data/cassandra/core/StatementFactory.java        | 12 ++++++++++++
 .../cassandra/core/StatementFactoryUnitTests.java    | 11 +++++++++++
 .../query/PartTreeCassandraQueryUnitTests.java       |  4 ++--
 3 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
index f9d649b35f..8972e6711c 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
@@ -69,6 +69,7 @@
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.metadata.schema.ClusteringOrder;
+import com.datastax.oss.driver.api.querybuilder.BindMarker;
 import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
 import com.datastax.oss.driver.api.querybuilder.condition.Condition;
 import com.datastax.oss.driver.api.querybuilder.condition.ConditionBuilder;
@@ -91,6 +92,7 @@
  *
  * @author Mark Paluch
  * @author John Blum
+ * @author Sam Lightfoot
  * @see com.datastax.oss.driver.api.core.cql.Statement
  * @see org.springframework.data.cassandra.core.query.Query
  * @see org.springframework.data.cassandra.core.query.Update
@@ -966,6 +968,11 @@ private static Relation toClause(CriteriaDefinition criteriaDefinition, TermFact
 
 				if (predicate.getValue() instanceof List
 						|| (predicate.getValue() != null && predicate.getValue().getClass().isArray())) {
+					Term term = factory.create(predicate.getValue());
+					if (term instanceof BindMarker) {
+						return column.in((BindMarker) term);
+					}
+
 					return column.in(toLiterals(predicate.getValue()));
 				}
 
@@ -1031,6 +1038,11 @@ private static Condition toCondition(CriteriaDefinition criteriaDefinition, Term
 
 				if (predicate.getValue() instanceof List
 						|| (predicate.getValue() != null && predicate.getValue().getClass().isArray())) {
+					Term term = factory.create(predicate.getValue());
+					if (term instanceof BindMarker) {
+						return column.in((BindMarker) term);
+					}
+
 					return column.in(toLiterals(predicate.getValue()));
 				}
 
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
index de97e671b7..80cbf2e5dd 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
@@ -53,6 +53,7 @@
  * Unit tests for {@link StatementFactory}.
  *
  * @author Mark Paluch
+ * @author Sam Lightfoot
  */
 class StatementFactoryUnitTests {
 
@@ -170,6 +171,16 @@ void shouldMapSelectQueryWithLimitAndAllowFiltering() {
 				.isEqualTo("SELECT * FROM group LIMIT 10 ALLOW FILTERING");
 	}
 
+	@Test
+	void shouldMapSelectInQuery() {
+
+		Query query = Query.query(Criteria.where("foo").in("bar"));
+
+		StatementBuilder<Select> select = statementFactory.select(query, groupEntity);
+
+		assertThat(select.build(ParameterHandling.INLINE).getQuery()).isEqualTo("SELECT * FROM group WHERE foo IN ('bar')");
+	}
+
 	@Test // DATACASS-343
 	void shouldMapDeleteQueryWithColumns() {
 
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
index bf4197e45b..2d370bb1b2 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
@@ -159,7 +159,7 @@ void shouldDeriveFieldInCollectionQuery() {
 		String query = deriveQueryFromMethod(Repo.class, "findByFirstnameIn", new Class[] { Collection.class },
 				Arrays.asList("Hank", "Walter")).getQuery();
 
-		assertThat(query).isEqualTo("SELECT * FROM person WHERE firstname IN ('Hank','Walter')");
+		assertThat(query).isEqualTo("SELECT * FROM person WHERE firstname IN ?");
 	}
 
 	@Test // DATACASS-172
@@ -186,7 +186,7 @@ void shouldDeriveUdtInCollectionQuery() {
 		String query = deriveQueryFromMethod(Repo.class, "findByMainAddressIn", new Class[] { Collection.class },
 				Collections.singleton(udtValue)).getQuery();
 
-		assertThat(query).isEqualTo("SELECT * FROM person WHERE mainaddress IN ({city:NULL,country:NULL})");
+		assertThat(query).isEqualTo("SELECT * FROM person WHERE mainaddress IN ?");
 	}
 
 	@Test // DATACASS-343