GH-3160: Handle element types correctly in CassandraFilterExpressionConverter.doValue

Fixes: 3160

https://github.com/spring-projects/spring-ai/issues/3160

When using a filter expression with IN operator on a collection field in
CassandraVectorStore.similaritySearch, a ClassCastException was thrown because
the code attempted to format individual collection elements using the collection's
codec instead of the element type's codec.

This fix modifies doValue to detect when we are formatting elements inside a
collection type and use the appropriate element type codec. While Cassandra
does not support using the IN operator directly on collection columns, this fix
ensures we generate syntactically correct CQL rather than throwing a Java
exception.

The change specifically addresses ListType collections by using the element type
codec for individual elements within the list.

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

vector-stores/spring-ai-cassandra-store/src/main/java/org/springframework/ai/vectorstore/cassandra/CassandraFilterExpressionConverter.java

@@ -23,7 +23,9 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
+import com.datastax.oss.driver.api.core.type.DataType;
 import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.ListType;
 import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
 import com.datastax.oss.driver.shaded.guava.common.base.Preconditions;
 
@@ -118,10 +120,19 @@ class CassandraFilterExpressionConverter extends AbstractFilterExpressionConvert
 	}
 
 	private void doValue(ColumnMetadata column, Object v, StringBuilder context) {
+
+		DataType dataType = column.getType();
+
+		// Check if we're handling an element inside a collection for an IN clause
+		if ((dataType instanceof ListType) && !(v instanceof Collection)) {
+			// Extract the element type from the collection type
+			dataType = ((ListType) dataType).getElementType();
+		}
+
 		if (DataTypes.SMALLINT.equals(column.getType())) {
 			v = ((Number) v).shortValue();
 		}
-		context.append(CodecRegistry.DEFAULT.codecFor(column.getType()).format(v));
+		context.append(CodecRegistry.DEFAULT.codecFor(dataType).format(v));
 	}
 
 	private Optional<ColumnMetadata> getColumn(String name) {

vector-stores/spring-ai-cassandra-store/src/test/java/org/springframework/ai/vectorstore/cassandra/CassandraVectorStoreIT.java

@@ -522,6 +522,61 @@ class CassandraVectorStoreIT extends BaseVectorStoreTests {
 		});
 	}
 
+	@Test
+	void searchWithCollectionFilter() {
+		this.contextRunner.run(context -> {
+			try (CassandraVectorStore store = createTestStore(context,
+					new SchemaColumn("currencies", DataTypes.listOf(DataTypes.TEXT), SchemaColumnTags.INDEXED))) {
+
+				// Create test documents with different currency lists
+				var btcDocument = new Document("BTC_doc", "Bitcoin document", Map.of("currencies", List.of("BTC")));
+				var ethDocument = new Document("ETH_doc", "Ethereum document", Map.of("currencies", List.of("ETH")));
+				var multiCurrencyDocument = new Document("MULTI_doc", "Multi-currency document",
+						Map.of("currencies", List.of("BTC", "ETH", "SOL")));
+
+				store.add(List.of(btcDocument, ethDocument, multiCurrencyDocument));
+
+				// Verify initial state
+				List<Document> results = store
+					.similaritySearch(SearchRequest.builder().query("document").topK(5).build());
+				assertThat(results).hasSize(3);
+
+				try {
+					// Test filtering with IN operator on a collection field
+					Filter.Expression filterExpression = new Filter.Expression(Filter.ExpressionType.IN,
+							new Filter.Key("currencies"), new Filter.Value(List.of("BTC")));
+
+					// Search using programmatic filter
+					store.similaritySearch(SearchRequest.builder()
+						.query("document")
+						.topK(5)
+						.similarityThresholdAll()
+						.filterExpression(filterExpression)
+						.build());
+
+					// If we get here without an exception, it means Cassandra
+					// unexpectedly accepted the query,
+					// which is surprising since Cassandra doesn't support the IN operator
+					// on collection columns.
+					// This would indicate a potential change in Cassandra's behavior.
+					Assertions.fail("Expected InvalidQueryException from Cassandra");
+				}
+				catch (InvalidQueryException e) {
+					// This is the expected outcome: Cassandra rejects the query with a
+					// specific error
+					// indicating that collection columns cannot be used with IN
+					// operators, which is
+					// a documented limitation of Cassandra's query language. Support for
+					// collection
+					// filtering via CONTAINS would be needed for this type of query to
+					// work.
+					assertThat(e.getMessage()).contains("Collection column 'currencies'");
+					assertThat(e.getMessage()).contains("cannot be restricted by a 'IN' relation");
+				}
+			}
+		});
+	}
+
 	@Test
 	void throwsExceptionOnInvalidIndexNameWithSchemaValidation() {
 		this.contextRunner.run(context -> {