Use bind markers for BY_INDEX and BY_NAME IN queries.

Closes #1172
Original pull request: #1178.

Author: samueldlightfoot

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()));
 				}
 

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() {
 

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