Add support for Cassandra Vector search.

Closes #1504

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

@@ -59,6 +59,7 @@ import org.springframework.data.cassandra.core.query.Update.RemoveOp;
 import org.springframework.data.cassandra.core.query.Update.SetAtIndexOp;
 import org.springframework.data.cassandra.core.query.Update.SetAtKeyOp;
 import org.springframework.data.cassandra.core.query.Update.SetOp;
+import org.springframework.data.cassandra.core.query.VectorSort;
 import org.springframework.data.convert.EntityWriter;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentProperty;
@@ -72,6 +73,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 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;
@@ -688,24 +690,30 @@ public class StatementFactory {
 	private StatementBuilder<Select> createSelectAndOrder(List<Selector> selectors, CassandraPersistentEntity<?> entity,
 			CqlIdentifier from, Filter filter, Sort sort) {
 
-		Select select;
+		StatementBuilder<Select> builder = StatementBuilder.of((Select) QueryBuilder.selectFrom(from),
+				cassandraConverter.getCodecRegistry());
 
-		if (selectors.isEmpty()) {
-			select = QueryBuilder.selectFrom(getKeyspace(entity, from), from).all();
-		} else {
+		builder.bind((statement, factory) -> {
 
-			List<com.datastax.oss.driver.api.querybuilder.select.Selector> mappedSelectors = new ArrayList<>(
-					selectors.size());
-			for (Selector selector : selectors) {
-				com.datastax.oss.driver.api.querybuilder.select.Selector orElseGet = selector.getAlias()
-						.map(it -> getSelection(selector).as(it)).orElseGet(() -> getSelection(selector));
-				mappedSelectors.add(orElseGet);
+			Select select;
+
+			if (selectors.isEmpty()) {
+				select = QueryBuilder.selectFrom(getKeyspace(entity, from), from).all();
+			} else {
+
+				List<com.datastax.oss.driver.api.querybuilder.select.Selector> mappedSelectors = new ArrayList<>(
+						selectors.size());
+				for (Selector selector : selectors) {
+					com.datastax.oss.driver.api.querybuilder.select.Selector orElseGet = selector.getAlias()
+							.map(it -> getSelection(selector, factory).as(it)).orElseGet(() -> getSelection(selector, factory));
+					mappedSelectors.add(orElseGet);
+				}
+
+				select = QueryBuilder.selectFrom(getKeyspace(entity, from), from).selectors(mappedSelectors);
 			}
 
-			select = QueryBuilder.selectFrom(getKeyspace(entity, from), from).selectors(mappedSelectors);
-		}
-
-		StatementBuilder<Select> builder = StatementBuilder.of(select, cassandraConverter.getCodecRegistry());
+			return select;
+		});
 
 		builder.bind((statement, factory) -> {
 			return statement.where(getRelations(filter, factory));
@@ -713,13 +721,23 @@ public class StatementFactory {
 
 		if (sort.isSorted()) {
 
-			builder.apply((statement) -> {
+			builder.bind((statement, factory) -> {
 
 				Select statementToUse = statement;
 
-				for (Sort.Order order : sort) {
-					statementToUse = statementToUse.orderBy(order.getProperty(),
-							order.isAscending() ? ClusteringOrder.ASC : ClusteringOrder.DESC);
+				if (sort instanceof VectorSort vs) {
+
+					for (Sort.Order order : sort) {
+
+						Object vector = vs.getVector();
+						statementToUse = statementToUse.orderByAnnOf(order.getProperty(), (CqlVector<?>) vector);
+					}
+				} else {
+
+					for (Sort.Order order : sort) {
+						statementToUse = statementToUse.orderBy(order.getProperty(),
+								order.isAscending() ? ClusteringOrder.ASC : ClusteringOrder.DESC);
+					}
 				}
 
 				return statementToUse;
@@ -730,25 +748,33 @@ public class StatementFactory {
 	}
 
 	private static List<Relation> getRelations(Filter filter, TermFactory factory) {
+
 		List<Relation> relations = new ArrayList<>();
+
 		for (CriteriaDefinition criteriaDefinition : filter) {
 			relations.add(toClause(criteriaDefinition, factory));
 		}
+
 		return relations;
 	}
 
-	private static com.datastax.oss.driver.api.querybuilder.select.Selector getSelection(Selector selector) {
+	private static com.datastax.oss.driver.api.querybuilder.select.Selector getSelection(Selector selector,
+			TermFactory factory) {
 
 		if (selector instanceof FunctionCall) {
 
 			com.datastax.oss.driver.api.querybuilder.select.Selector[] arguments = ((FunctionCall) selector).getParameters()
 					.stream().map(param -> {
 
-						if (param instanceof ColumnSelector) {
+						if (param instanceof ColumnSelector s) {
 
-							return com.datastax.oss.driver.api.querybuilder.select.Selector
-									.column(((ColumnSelector) param).getExpression());
+							return com.datastax.oss.driver.api.querybuilder.select.Selector.column(s.getExpression());
 						}
+
+						if (param instanceof CqlIdentifier i) {
+							return com.datastax.oss.driver.api.querybuilder.select.Selector.column(i.toString());
+						}
+
 						return new SimpleSelector(param.toString());
 
 					}).toArray(com.datastax.oss.driver.api.querybuilder.select.Selector[]::new);

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

@@ -20,6 +20,7 @@ import java.net.InetAddress;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
@@ -30,10 +31,12 @@ import org.springframework.core.convert.converter.ConverterFactory;
 import org.springframework.data.cassandra.core.cql.converter.RowToListConverter;
 import org.springframework.data.cassandra.core.cql.converter.RowToMapConverter;
 import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.domain.Vector;
 import org.springframework.util.Assert;
 import org.springframework.util.NumberUtils;
 
 import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 
 /**
  * Wrapper class to contain useful converters for the usage with Cassandra.
@@ -41,7 +44,7 @@ import com.datastax.oss.driver.api.core.cql.Row;
  * @author Mark Paluch
  * @since 1.5
  */
-abstract class CassandraConverters {
+public abstract class CassandraConverters {
 
 	/**
 	 * Private constructor to prevent instantiation.
@@ -66,6 +69,17 @@ abstract class CassandraConverters {
 		converters.add(RowToStringConverter.INSTANCE);
 		converters.add(RowToUuidConverter.INSTANCE);
 
+		converters.add(VectorToFloatArrayConverter.INSTANCE);
+		converters.add(VectorToDoubleArrayConverter.INSTANCE);
+		converters.add(VectorToFloatListConverter.INSTANCE);
+
+		converters.add(FloatArrayToVectorConverter.INSTANCE);
+		converters.add(DoubleArrayToVectorConverter.INSTANCE);
+		converters.add(NumberListToVectorConverter.INSTANCE);
+
+		converters.add(VectorToCqlVectorConverter.INSTANCE);
+		converters.add(CqlVectorToVectorConverter.INSTANCE);
+
 		return converters;
 	}
 
@@ -222,4 +236,128 @@ abstract class CassandraConverters {
 			return row.getLocalDate(0);
 		}
 	}
+
+	@ReadingConverter
+	public enum DoubleArrayToVectorConverter implements Converter<double[], CqlVector<Double>> {
+
+		INSTANCE;
+
+		@Override
+		public CqlVector<Double> convert(double[] source) {
+
+			Double[] converted = new Double[source.length];
+			for (int i = 0; i < converted.length; i++) {
+				converted[i] = source[i];
+			}
+			return CqlVector.newInstance(converted);
+		}
+	}
+
+	public enum CqlVectorToVectorConverter implements Converter<CqlVector<?>, Vector> {
+
+		INSTANCE;
+
+		@Override
+		public Vector convert(CqlVector<?> source) {
+			return CassandraVector.of(source);
+		}
+	}
+
+	public enum VectorToCqlVectorConverter implements Converter<Vector, CqlVector<?>> {
+
+		INSTANCE;
+
+		@Override
+		public CqlVector<?> convert(Vector source) {
+
+			if (source instanceof CassandraVector cv) {
+				return cv.getSource();
+			}
+
+			if (source.getType() == Float.class || source.getType() == Float.TYPE) {
+
+				float[] floatArray = source.toFloatArray();
+				List<Float> boxed = new ArrayList<>(floatArray.length);
+
+				for (float v : floatArray) {
+					boxed.add(v);
+				}
+				return CqlVector.newInstance(boxed);
+			}
+
+			return CqlVector.newInstance(Arrays.stream(source.toDoubleArray()).boxed().toList());
+		}
+	}
+
+	@ReadingConverter
+	public enum FloatArrayToVectorConverter implements Converter<float[], CqlVector<Float>> {
+
+		INSTANCE;
+
+		@Override
+		public CqlVector<Float> convert(float[] source) {
+
+			Float[] converted = new Float[source.length];
+			for (int i = 0; i < converted.length; i++) {
+				converted[i] = source[i];
+			}
+			return CqlVector.newInstance(converted);
+		}
+	}
+
+	@ReadingConverter
+	public enum NumberListToVectorConverter implements Converter<List<Number>, CqlVector<Number>> {
+
+		INSTANCE;
+
+		@Override
+		public CqlVector<Number> convert(List<Number> source) {
+			return CqlVector.newInstance(source);
+		}
+	}
+
+	@ReadingConverter
+	public enum VectorToFloatArrayConverter implements Converter<CqlVector<Number>, float[]> {
+
+		INSTANCE;
+
+		@Override
+		public float[] convert(CqlVector<Number> source) {
+			float[] array = new float[source.size()];
+			for (int i = 0; i < array.length; i++) {
+				array[i] = source.get(i).floatValue();
+			}
+			return array;
+		}
+	}
+
+	@ReadingConverter
+	public enum VectorToDoubleArrayConverter implements Converter<CqlVector<Number>, double[]> {
+
+		INSTANCE;
+
+		@Override
+		public double[] convert(CqlVector<Number> source) {
+			double[] array = new double[source.size()];
+			for (int i = 0; i < array.length; i++) {
+				array[i] = source.get(i).doubleValue();
+			}
+			return array;
+		}
+	}
+
+	@ReadingConverter
+	public enum VectorToFloatListConverter implements Converter<CqlVector<Number>, List<Float>> {
+
+		INSTANCE;
+
+		@Override
+		public List<Float> convert(CqlVector<Number> source) {
+			List<Float> values = new ArrayList<>(source.size());
+			for (int i = 0; i < source.size(); i++) {
+				values.add(source.get(i).floatValue());
+			}
+			return values;
+		}
+	}
 }

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

@@ -109,7 +109,6 @@ public class CassandraCustomConversions extends org.springframework.data.convert
 
 		CassandraConverterConfiguration(List<?> converters) {
 			super(STORE_CONVERSIONS, converters, getConverterFilter());
-
 		}
 
 		CassandraConverterConfiguration(List<?> userConverters, PropertyValueConversions propertyValueConversions) {

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

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core.convert;
+
+import org.springframework.data.domain.Vector;
+import org.springframework.util.Assert;
+
+import com.datastax.oss.driver.api.core.data.CqlVector;
+
+/**
+ * Vector implementation for Cassandra's {@link CqlVector}.
+ *
+ * @author Mark Paluch
+ * @since 4.5
+ */
+public class CassandraVector implements Vector {
+
+	private final CqlVector<?> cqlVector;
+
+	private CassandraVector(CqlVector<?> cqlVector) {
+		this.cqlVector = cqlVector;
+	}
+
+	/**
+	 * Creates a new CassandraVector for the given {@link CqlVector}.
+	 *
+	 * @param cqlVector must not be {@literal null}.
+	 * @return a new CassandraVector for the given {@link CqlVector}
+	 */
+	public static CassandraVector of(CqlVector<?> cqlVector) {
+
+		Assert.notNull(cqlVector, "CqlVector must not be null");
+
+		return new CassandraVector(cqlVector);
+	}
+
+	@Override
+	public Class<? extends Number> getType() {
+
+		if (!cqlVector.isEmpty()) {
+
+			Object o = cqlVector.get(0);
+
+			if (o instanceof Float) {
+				return Float.class;
+			}
+
+			if (o instanceof Double) {
+				return Double.class;
+			}
+		}
+
+		return Number.class;
+	}
+
+	@Override
+	public CqlVector<?> getSource() {
+		return cqlVector;
+	}
+
+	@Override
+	public int size() {
+		return cqlVector.size();
+	}
+
+	@Override
+	public float[] toFloatArray() {
+
+		float[] v = new float[cqlVector.size()];
+		for (int i = 0; i < cqlVector.size(); i++) {
+			v[i] = ((Number) cqlVector.get(i)).floatValue();
+		}
+
+		return v;
+	}
+
+	@Override
+	public double[] toDoubleArray() {
+
+		double[] v = new double[cqlVector.size()];
+		for (int i = 0; i < cqlVector.size(); i++) {
+			v[i] = ((Number) cqlVector.get(i)).doubleValue();
+		}
+
+		return v;
+	}
+
+	@Override
+	public String toString() {
+		return cqlVector.toString();
+	}
+
+}

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

@@ -38,6 +38,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraType;
 import org.springframework.data.cassandra.core.mapping.CassandraType.Name;
 import org.springframework.data.cassandra.core.mapping.Frozen;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
+import org.springframework.data.cassandra.core.mapping.VectorType;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.convert.PropertyValueConversions;
 import org.springframework.data.convert.PropertyValueConverter;
@@ -141,6 +142,10 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 						property.getName(), property.getType(), property.getOwner().getName(), annotation.typeArguments().length));
 			}
 
+			if (annotation.type() == Name.VECTOR) {
+				return resolve(property.getRequiredAnnotation(VectorType.class));
+			}
+
 			return resolve(annotation);
 		}
 
@@ -250,6 +255,12 @@ class DefaultColumnTypeResolver implements ColumnTypeResolver {
 		}
 	}
 
+	public CassandraColumnType resolve(VectorType annotation) {
+
+		DataType subtype = CassandraSimpleTypeHolder.getRequiredDataTypeFor(annotation.subtype());
+		return createCassandraTypeDescriptor(DataTypes.vectorOf(subtype, annotation.dimensions()));
+	}
+
 	@Override
 	public CassandraColumnType resolve(CassandraType annotation) {
 

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

@@ -31,17 +31,19 @@ import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecifica
 import org.springframework.data.cassandra.core.cql.keyspace.SpecificationBuilder;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
 import org.springframework.data.cassandra.core.mapping.Indexed;
-import org.springframework.data.cassandra.core.mapping.SAIIndexed;
 import org.springframework.data.cassandra.core.mapping.SASI;
 import org.springframework.data.cassandra.core.mapping.SASI.NonTokenizingAnalyzed;
 import org.springframework.data.cassandra.core.mapping.SASI.Normalization;
 import org.springframework.data.cassandra.core.mapping.SASI.StandardAnalyzed;
+import org.springframework.data.cassandra.core.mapping.SaiIndexed;
+import org.springframework.data.domain.Vector;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 
 /**
  * Factory to create {@link org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification} based on
@@ -52,7 +54,7 @@ import com.datastax.oss.driver.api.core.CqlIdentifier;
  * @since 2.0
  * @see Indexed
  * @see SASI
- * @see SAIIndexed
+ * @see SaiIndexed
  */
 @SuppressWarnings("unchecked")
 class IndexSpecificationFactory {
@@ -98,10 +100,10 @@ class IndexSpecificationFactory {
 			indexes.add(createIndexSpecification(keyspace, property.getRequiredAnnotation(SASI.class), property));
 		}
 
-		if (property.isAnnotationPresent(SAIIndexed.class)) {
+		if (property.isAnnotationPresent(SaiIndexed.class)) {
 
 			CreateIndexSpecification index = createIndexSpecification(keyspace,
-					property.getRequiredAnnotation(SAIIndexed.class), property);
+					property.getRequiredAnnotation(SaiIndexed.class), property);
 
 			if (property.isMapLike()) {
 				index.entries();
@@ -114,7 +116,7 @@ class IndexSpecificationFactory {
 
 			indexes.addAll(createTypeAnnotatedIndexes(Indexed.class, property,
 					indexed -> createIndexSpecification(keyspace, indexed, property)));
-			indexes.addAll(createTypeAnnotatedIndexes(SAIIndexed.class, property,
+			indexes.addAll(createTypeAnnotatedIndexes(SaiIndexed.class, property,
 					indexed -> createIndexSpecification(keyspace, indexed, property)));
 		}
 
@@ -207,7 +209,7 @@ class IndexSpecificationFactory {
 	}
 
 	private static CreateIndexSpecification createIndexSpecification(@Nullable CqlIdentifier keyspace,
-			SAIIndexed annotation, CassandraPersistentProperty property) {
+			SaiIndexed annotation, CassandraPersistentProperty property) {
 
 		CreateIndexSpecification index;
 
@@ -217,14 +219,20 @@ class IndexSpecificationFactory {
 			index = SpecificationBuilder.createIndex(keyspace, null);
 		}
 
-		index.using("sai") //
-				.columnName(property.getRequiredColumnName())
-				.withOption("case_sensitive", Boolean.toString(annotation.caseSensitive()))
-				.withOption("normalize", Boolean.toString(annotation.normalize()))
-				.withOption("ascii", Boolean.toString(annotation.ascii()))
-				.withOption("similarity_function", annotation.similarityFunction().name());
+		CreateIndexSpecification sai = index.using("sai") //
+				.columnName(property.getRequiredColumnName());
 
-		return index;
+		if (isVector(property.getType())) {
+			return sai.withOption("similarity_function", annotation.similarityFunction().name());
+		} else {
+			return sai.withOption("case_sensitive", Boolean.toString(annotation.caseSensitive()))
+					.withOption("normalize", Boolean.toString(annotation.normalize()))
+					.withOption("ascii", Boolean.toString(annotation.ascii()));
+		}
+	}
+
+	private static boolean isVector(Class<?> type) {
+		return type.equals(CqlVector.class) || Vector.class.isAssignableFrom(type);
 	}
 
 	interface CreateIndexConfigurer<T extends Annotation> extends BiConsumer<T, CreateIndexSpecification> {}

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

@@ -162,6 +162,8 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 				this::getCustomConversions);
 		this.embeddedEntityOperations = new EmbeddedEntityOperations(mappingContext);
 		this.spELContext = new SpELContext(RowReaderPropertyAccessor.INSTANCE);
+
+		getCustomConversions().registerConvertersIn((DefaultConversionService) getConversionService());
 	}
 
 	/**
@@ -972,7 +974,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return getConversionService().convert(value, resolvedTargetType);
 		}
 
-		if (value instanceof Collection) {
+		if (value instanceof Collection && columnType.isCollectionLike()) {
 			return writeCollectionInternal((Collection<Object>) value, columnType);
 		}
 
@@ -1015,7 +1017,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 		}
 
-		if (getCustomConversions().isSimpleType(value.getClass())) {
+		if (getCustomConversions().isSimpleType(value.getClass()) || getCustomConversions().isSimpleType(requestedTargetType)) {
 			return getPotentiallyConvertedSimpleValue(value, requestedTargetType);
 		}
 

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

@@ -37,6 +37,7 @@ import org.springframework.data.cassandra.core.query.Criteria;
 import org.springframework.data.cassandra.core.query.CriteriaDefinition;
 import org.springframework.data.cassandra.core.query.CriteriaDefinition.Predicate;
 import org.springframework.data.cassandra.core.query.Filter;
+import org.springframework.data.cassandra.core.query.VectorSort;
 import org.springframework.data.convert.PropertyValueConverter;
 import org.springframework.data.convert.ValueConversionContext;
 import org.springframework.data.domain.Sort;
@@ -135,7 +136,7 @@ public class QueryMapper {
 
 			Predicate predicate = criteriaDefinition.getPredicate();
 			Object value = predicate.getValue();
-			Object mappedValue = value != null ? getMappedValue(field, predicate, value) : null;
+			Object mappedValue = value != null ? getMappedValue(field, predicate.getOperator(), value) : null;
 			Predicate mappedPredicate = new Predicate(predicate.getOperator(), mappedValue);
 
 			result.add(Criteria.of(field.getMappedKey(), mappedPredicate));
@@ -145,7 +146,7 @@ public class QueryMapper {
 	}
 
 	@Nullable
-	private Object getMappedValue(Field field, Predicate predicate, Object value) {
+	private Object getMappedValue(Field field, CriteriaDefinition.Operator operator, Object value) {
 
 		if (field.getProperty().isPresent()
 				&& field.getProperty().filter(it -> converter.getCustomConversions().hasValueConverter(it)).isPresent()) {
@@ -173,7 +174,7 @@ public class QueryMapper {
 			return valueConverter.write(value, conversionContext);
 		}
 
-		ColumnType typeDescriptor = getColumnType(field, value, ColumnTypeTransformer.of(field, predicate.getOperator()));
+		ColumnType typeDescriptor = getColumnType(field, value, ColumnTypeTransformer.of(field, operator));
 		return getConverter().convertToColumnType(value, typeDescriptor);
 	}
 
@@ -204,7 +205,7 @@ public class QueryMapper {
 				List<CqlIdentifier> mappedColumnNames = getCqlIdentifier(column, field);
 
 				for (CqlIdentifier mappedColumnName : mappedColumnNames) {
-					selectors.add(getMappedSelector(selector, mappedColumnName));
+					selectors.add(getMappedSelector(selector, mappedColumnName, field));
 				}
 			});
 		}
@@ -230,7 +231,7 @@ public class QueryMapper {
 		});
 	}
 
-	private Selector getMappedSelector(Selector selector, CqlIdentifier cqlIdentifier) {
+	private Selector getMappedSelector(Selector selector, CqlIdentifier cqlIdentifier, Field field) {
 
 		if (selector instanceof ColumnSelector) {
 
@@ -248,8 +249,24 @@ public class QueryMapper {
 			FunctionCall mappedFunctionCall = FunctionCall.from(functionCall.getExpression(),
 					functionCall.getParameters().stream().map(obj -> {
 
-						if (obj instanceof Selector) {
-							return getMappedSelector((Selector) obj, cqlIdentifier);
+						if (obj instanceof Selector sel) {
+							return getMappedSelector(sel, cqlIdentifier, field);
+						}
+
+						if (obj instanceof ColumnName cn) {
+
+							CqlIdentifier identifier = cn.getCqlIdentifier().or(() -> cn.getColumnName().map(CqlIdentifier::fromCql))
+									.orElseGet(() -> CqlIdentifier.fromCql(cn.toCql()));
+
+							return getMappedSelector(ColumnSelector.from(cn), identifier, field);
+						}
+
+						if (obj instanceof CqlIdentifier identifier) {
+							return getMappedSelector(ColumnSelector.from(identifier), identifier, field);
+						}
+
+						if (field.getProperty().isPresent()) {
+							return getMappedValue(field, CriteriaDefinition.Operators.EQ, obj);
 						}
 
 						return obj;
@@ -321,12 +338,18 @@ public class QueryMapper {
 
 		List<Order> mappedOrders = new ArrayList<>();
 
+		Object vector = sort instanceof VectorSort vs ? vs.getVector() : null;
+
 		for (Order order : sort) {
 
 			ColumnName columnName = ColumnName.from(order.getProperty());
 
 			Field field = createPropertyField(entity, columnName);
 
+			if (vector != null) {
+				vector = getMappedValue(field, CriteriaDefinition.Operators.EQ, vector);
+			}
+
 			List<CqlIdentifier> mappedColumnNames = getCqlIdentifier(columnName, field);
 
 			if (mappedColumnNames.isEmpty()) {
@@ -338,7 +361,7 @@ public class QueryMapper {
 			}
 		}
 
-		return Sort.by(mappedOrders);
+		return vector != null ? new VectorSort(mappedOrders, vector) : Sort.by(mappedOrders);
 	}
 
 	private List<CqlIdentifier> getCqlIdentifier(ColumnName column, Field field) {

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

@@ -29,6 +29,7 @@ import org.springframework.data.cassandra.core.mapping.EmbeddedEntityOperations;
 import org.springframework.data.cassandra.core.mapping.Indexed;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.domain.Vector;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.lang.NonNull;
@@ -36,6 +37,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.detach.AttachmentPoint;
 import com.datastax.oss.driver.api.core.type.DataType;
@@ -197,6 +199,13 @@ public class SchemaFactory {
 			return typeResolver.resolve(property).getDataType();
 		} catch (MappingException e) {
 
+			if (isVector(property.getType())) {
+
+				throw new MappingException(String.format(
+						"Cannot resolve DataType for type [%s] for property [%s] in entity [%s]; Annotate the vector property with @VectorType(…)",
+						property.getType(), property.getName(), property.getOwner().getName()), e);
+			}
+
 			throw new MappingException(String.format(
 					"Cannot resolve DataType for type [%s] for property [%s] in entity [%s]; Consider registering a Converter or annotating the property with @CassandraType",
 					property.getType(), property.getName(), property.getOwner().getName()), e);
@@ -262,10 +271,12 @@ public class SchemaFactory {
 		List<CreateIndexSpecification> indexes = new ArrayList<>();
 
 		for (CassandraPersistentProperty property : entity) {
+
 			if (property.isCompositePrimaryKey()) {
 				CassandraPersistentEntity<?> pkEntity = mappingContext.getRequiredPersistentEntity(property);
 				indexes.addAll(getCreateIndexSpecificationsFor(pkEntity, pkEntity.getKeyspace(), pkEntity.getTableName()));
 			}
+
 			if (property.isEmbedded()) {
 
 				if (property.isAnnotationPresent(Indexed.class)) {
@@ -321,6 +332,10 @@ public class SchemaFactory {
 		return specification;
 	}
 
+	private static boolean isVector(Class<?> type) {
+		return type.equals(CqlVector.class) || Vector.class.isAssignableFrom(type);
+	}
+
 	enum ShallowUserTypeResolver implements UserTypeResolver {
 		INSTANCE;
 

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

@@ -25,6 +25,7 @@ import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.VectorType;
 
 /**
  * Object to configure a CQL column specification.
@@ -222,15 +223,22 @@ public class ColumnSpecification {
 	}
 
 	public StringBuilder toCql(StringBuilder cql) {
-		return cql.append(name.asCql(true)).append(" ").append(type.asCql(true, true));
+		return cql.append(name.asCql(true)).append(" ").append(renderType());
+	}
+
+	private String renderType() {
+
+		// TODO 'org.apache.cassandra.db.marshal.VectorType(1536)'
+		if (type instanceof VectorType vt) {
+			return "vector<%s, %d>".formatted(vt.getElementType().asCql(true, true), vt.getDimensions());
+		}
+
+		return type.asCql(true, true);
 	}
 
 	@Override
 	public String toString() {
-		return toCql(new StringBuilder()).append(" /* ")
-				.append("keyType=").append(keyType).append(", ")
-				.append("ordering=").append(ordering).append(", ")
-				.append("isStatic=").append(isStatic)
-				.append(" */ ").toString();
+		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/mapping/CassandraSimpleTypeHolder.java

@@ -31,6 +31,7 @@ import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.data.CqlDuration;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 import com.datastax.oss.driver.api.core.data.TupleValue;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.type.DataType;
@@ -85,6 +86,7 @@ public class CassandraSimpleTypeHolder extends SimpleTypeHolder {
 		simpleTypes.add(TupleValue.class);
 		simpleTypes.add(UdtValue.class);
 		simpleTypes.add(CqlDuration.class);
+		simpleTypes.add(CqlVector.class);
 
 		classToDataType = Collections.unmodifiableMap(classToDataType(codecRegistry, primitiveWrappers));
 		nameToDataType = Collections.unmodifiableMap(nameToDataType());

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

@@ -73,6 +73,13 @@ public @interface CassandraType {
 	 * @since 3.0
 	 */
 	enum Name {
-		ASCII, BIGINT, BLOB, BOOLEAN, COUNTER, DECIMAL, DOUBLE, FLOAT, INT, TIMESTAMP, UUID, VARCHAR, TEXT, VARINT, TIMEUUID, INET, DATE, TIME, SMALLINT, TINYINT, DURATION, LIST, MAP, SET, UDT, TUPLE;
+		ASCII, BIGINT, BLOB, BOOLEAN, COUNTER, DECIMAL, DOUBLE, FLOAT, INT, TIMESTAMP, UUID, VARCHAR, TEXT, VARINT, TIMEUUID, INET, DATE, TIME, SMALLINT, TINYINT, DURATION, LIST, MAP, SET, UDT,
+
+		/**
+		 * Using vector types requires the usage of {@link VectorType}.
+		 *
+		 * @since 4.5
+		 */
+		VECTOR, TUPLE;
 	}
 }

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

@@ -30,12 +30,13 @@ import java.lang.annotation.Target;
  * <li>non-frozen user-defined type (UDT)</li>
  * </ul>
  * <p>
- * The following columns of a {@link Table} type can be annotated with {@link SAIIndexed}:
+ * The following columns of a {@link Table} type can be annotated with {@link SaiIndexed}:
  * <ul>
  * <li>Scalar data types</li>
  * <li>Frozen user-defined types</li>
  * <li>Collection types</li>
  * <li>Map type</li>
+ * <li>Vector types</li>
  * </ul>
  * <p>
  * Map types distinguish between entry, key or value-level indexing.
@@ -50,11 +51,12 @@ import java.lang.annotation.Target;
  * </pre>
  *
  * @author Mark Paluch
+ * @since 4.5
  */
 @Documented
 @Retention(value = RetentionPolicy.RUNTIME)
 @Target(value = { ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE, ElementType.TYPE_USE })
-public @interface SAIIndexed {
+public @interface SaiIndexed {
 
 	/**
 	 * The name of the index. If {@literal null} or empty, then the index name will be generated by Cassandra and will be
@@ -84,14 +86,9 @@ public @interface SAIIndexed {
 
 	/**
 	 * Vector search relies on computing the similarity or distance between vectors to identify relevant matches. The
-	 * similarity function is used to compute the similarity between two vectors.
+	 * similarity function is used to compute the similarity between two vectors. When using SAI indexes with vector data
+	 * types, then only the similarity function is used with index options.
 	 */
 	SimilarityFunction similarityFunction() default SimilarityFunction.COSINE;
 
-	/**
-	 * Enumeration of similarity functions.
-	 */
-	enum SimilarityFunction {
-		COSINE, DOT_PRODUCT, EUCLIDEAN
-	}
 }

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core.mapping;
+
+/**
+ * Enumeration of similarity functions.
+ *
+ * @since 4.5
+ */
+public enum SimilarityFunction {
+
+	COSINE, DOT_PRODUCT, EUCLIDEAN
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core.mapping;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Specific variant of {@code CassandraType(VECTOR)} allowing specification of a Cassandra vector type alongside with
+ * its subtype and number of dimensions.
+ *
+ * @author Mark Paluch
+ * @since 4.5
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target(value = { ElementType.ANNOTATION_TYPE, ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER })
+@CassandraType(type = CassandraType.Name.VECTOR)
+public @interface VectorType {
+
+	/**
+	 * @return Vector subtype, defaults to float.
+	 */
+	CassandraType.Name subtype() default CassandraType.Name.FLOAT;
+
+	/**
+	 * @return number of dimensions.
+	 */
+	int dimensions();
+
+}

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

@@ -20,16 +20,22 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
+import java.util.function.Function;
 
+import org.springframework.data.cassandra.core.convert.CassandraVector;
+import org.springframework.data.cassandra.core.mapping.SimilarityFunction;
+import org.springframework.data.domain.Vector;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 
 /**
  * Value object to abstract column names involved in a CQL query. Columns can be constructed from an array of names and
@@ -128,7 +134,7 @@ public class Columns implements Iterable<ColumnName> {
 	 * @return a new {@link Columns} object containing all column definitions and the TTL for {@code columnName}.
 	 */
 	public Columns ttl(String columnName) {
-		return select(columnName, FunctionCall.from("TTL", ColumnSelector.from(columnName)));
+		return select(columnName, SelectorBuilder::ttl);
 	}
 
 	/**
@@ -139,7 +145,7 @@ public class Columns implements Iterable<ColumnName> {
 	 * @return a new {@link Columns} object containing all column definitions and the TTL for {@code columnName}.
 	 */
 	public Columns ttl(CqlIdentifier columnName) {
-		return select(columnName, FunctionCall.from("TTL", ColumnSelector.from(columnName)));
+		return select(columnName, SelectorBuilder::ttl);
 	}
 
 	/**
@@ -150,13 +156,35 @@ public class Columns implements Iterable<ColumnName> {
 	 * @return a new {@link Columns} object containing all column definitions and the selected {@code columnName}.
 	 */
 	public Columns select(String columnName, Selector selector) {
+		return select(ColumnName.from(columnName), selector);
+	}
 
-		Assert.notNull(columnName, "Column name must not be null");
+	/**
+	 * Include column {@code columnName} with a built {@link Selector}. This column selection overrides an existing
+	 * selection for the column name.
+	 *
+	 * @param columnName must not be {@literal null}.
+	 * @return a new {@link Columns} object containing all column definitions and the selected {@code columnName}.
+	 * @since 4.5
+	 */
+	public Columns select(CqlIdentifier columnName, Function<SelectorBuilder, Selector> builder) {
 
-		Map<ColumnName, Selector> result = new LinkedHashMap<>(this.columns);
-		result.put(ColumnName.from(columnName), selector);
+		ColumnName from = ColumnName.from(columnName);
+		return select(from, builder.apply(new DefaultSelectorBuilder(from)));
+	}
 
-		return new Columns(result);
+	/**
+	 * Include column {@code columnName} with a built {@link Selector}. This column selection overrides an existing
+	 * selection for the column name.
+	 *
+	 * @param columnName must not be {@literal null}.
+	 * @return a new {@link Columns} object containing all column definitions and the selected {@code columnName}.
+	 * @since 4.5
+	 */
+	public Columns select(String columnName, Function<SelectorBuilder, Selector> builder) {
+
+		ColumnName from = ColumnName.from(columnName);
+		return select(from, builder.apply(new DefaultSelectorBuilder(from)));
 	}
 
 	/**
@@ -167,11 +195,20 @@ public class Columns implements Iterable<ColumnName> {
 	 * @return a new {@link Columns} object containing all column definitions and the selected {@code columnName}.
 	 */
 	public Columns select(CqlIdentifier columnName, Selector selector) {
+		return select(ColumnName.from(columnName), selector);
+	}
 
-		Assert.notNull(columnName, "Column name must not be null");
+	/**
+	 * Include column {@code columnName} with {@link Selector}. This column selection overrides an existing selection for
+	 * the column name.
+	 *
+	 * @param columnName must not be {@literal null}.
+	 * @return a new {@link Columns} object containing all column definitions and the selected {@code columnName}.
+	 */
+	private Columns select(ColumnName columnName, Selector selector) {
 
 		Map<ColumnName, Selector> result = new LinkedHashMap<>(this.columns);
-		result.put(ColumnName.from(columnName), selector);
+		result.put(columnName, selector);
 
 		return new Columns(result);
 	}
@@ -244,7 +281,7 @@ public class Columns implements Iterable<ColumnName> {
 		Iterator<Entry<ColumnName, Selector>> iterator = this.columns.entrySet().iterator();
 		StringBuilder builder = toString(iterator);
 
-		if (builder.length() == 0) {
+		if (builder.isEmpty()) {
 			return "*";
 		}
 
@@ -275,15 +312,36 @@ public class Columns implements Iterable<ColumnName> {
 	}
 
 	/**
-	 * Strategy interface to render a column selection.
+	 * Strategy interface to render a column or function selection.
 	 *
 	 * @author Mark Paluch
 	 */
 	public interface Selector {
 
+		/**
+		 * Apply the given {@code alias} to the current {@link Selector} expression.
+		 *
+		 * @param alias
+		 * @return the aliased {@code Selector} expression.
+		 * @since 4.5
+		 */
+		default Selector as(String alias) {
+			return as(CqlIdentifier.fromCql(alias));
+		}
+
+		/**
+		 * Apply the given {@code alias} to the current {@link Selector} expression.
+		 *
+		 * @param alias
+		 * @return the aliased {@code Selector} expression.
+		 * @since 4.5
+		 */
+		Selector as(CqlIdentifier alias);
+
 		String getExpression();
 
 		Optional<CqlIdentifier> getAlias();
+
 	}
 
 	/**
@@ -334,40 +392,27 @@ public class Columns implements Iterable<ColumnName> {
 			return from(ColumnName.from(columnName));
 		}
 
-		/**
-		 * Create a {@link ColumnSelector} for the current {@link #getExpression() expression} aliased as {@code alias}.
-		 *
-		 * @param alias must not be {@literal null} or empty.
-		 * @return the aliased {@link ColumnSelector}.
-		 */
-		public ColumnSelector as(String alias) {
-			return as(CqlIdentifier.fromCql(alias));
-		}
-
 		/**
 		 * Create a {@link ColumnSelector} for the current {@link #getExpression() expression} aliased as {@code alias}.
 		 *
 		 * @param alias must not be {@literal null}.
 		 * @return the aliased {@link ColumnSelector}.
 		 */
+		@Override
 		public ColumnSelector as(CqlIdentifier alias) {
 			return new ColumnSelector(columnName, alias);
 		}
 
+		@Override
 		public Optional<CqlIdentifier> getAlias() {
 			return alias;
 		}
 
+		@Override
 		public String getExpression() {
 			return columnName.toCql();
 		}
 
-		@Override
-		public String toString() {
-			return getAlias().map(cqlIdentifier -> String.format("%s AS %s", getExpression(), cqlIdentifier))
-					.orElseGet(this::getExpression);
-		}
-
 		@Override
 		public boolean equals(@Nullable Object o) {
 			if (this == o) {
@@ -389,6 +434,12 @@ public class Columns implements Iterable<ColumnName> {
 			result = 31 * result + ObjectUtils.nullSafeHashCode(alias);
 			return result;
 		}
+
+		@Override
+		public String toString() {
+			return getAlias().map(cqlIdentifier -> String.format("%s AS %s", getExpression(), cqlIdentifier))
+					.orElseGet(this::getExpression);
+		}
 	}
 
 	/**
@@ -418,22 +469,13 @@ public class Columns implements Iterable<ColumnName> {
 			return new FunctionCall(expression, Arrays.asList(params));
 		}
 
-		/**
-		 * Create a {@link FunctionCall} for the current {@link #getExpression() expression} aliased as {@code alias}.
-		 *
-		 * @param alias must not be {@literal null} or empty.
-		 * @return the aliased {@link ColumnSelector}.
-		 */
-		public FunctionCall as(String alias) {
-			return as(CqlIdentifier.fromCql(alias));
-		}
-
 		/**
 		 * Create a {@link FunctionCall} for the current {@link #getExpression() expression} aliased as {@code alias}.
 		 *
 		 * @param alias must not be {@literal null}.
 		 * @return the aliased {@link ColumnSelector}.
 		 */
+		@Override
 		public FunctionCall as(CqlIdentifier alias) {
 			return new FunctionCall(expression, params, alias);
 		}
@@ -452,16 +494,6 @@ public class Columns implements Iterable<ColumnName> {
 			return params;
 		}
 
-		@Override
-		public String toString() {
-
-			String parameters = StringUtils.collectionToDelimitedString(getParameters(), ", ");
-
-			return getAlias()
-					.map(cqlIdentifier -> String.format("%s(%s) AS %s", getExpression(), parameters, cqlIdentifier))
-					.orElseGet(() -> String.format("%s(%s)", getExpression(), parameters));
-		}
-
 		@Override
 		public boolean equals(@Nullable Object o) {
 			if (this == o) {
@@ -487,5 +519,149 @@ public class Columns implements Iterable<ColumnName> {
 			result = 31 * result + ObjectUtils.nullSafeHashCode(alias);
 			return result;
 		}
+
+		@Override
+		public String toString() {
+
+			String parameters = StringUtils.collectionToDelimitedString(getParameters(), ", ");
+
+			return getAlias().map(cqlIdentifier -> String.format("%s(%s) AS %s", getExpression(), parameters, cqlIdentifier))
+					.orElseGet(() -> String.format("%s(%s)", getExpression(), parameters));
+		}
+
 	}
+
+	/**
+	 * Entrypoint to build a {@link Selector}.
+	 *
+	 * @since 4.5
+	 */
+	public interface SelectorBuilder {
+
+		/**
+		 * Include the column in the selection.
+		 *
+		 * @return column selector for the used column name.
+		 */
+		Selector column();
+
+		/**
+		 * Return the time to live for the column in the selection.
+		 *
+		 * @return TTL function selector for the used column name.
+		 */
+		Selector ttl();
+
+		/**
+		 * Return a builder for a similarity function using the given {@link CqlVector}.
+		 *
+		 * @param vector
+		 * @return builder to build a similarity function.
+		 */
+		SimilarityBuilder similarity(CqlVector<?> vector);
+
+		/**
+		 * Return a builder for a similarity function using the given {@link Vector}.
+		 *
+		 * @param vector
+		 * @return builder to build a similarity function.
+		 */
+		SimilarityBuilder similarity(Vector vector);
+
+	}
+
+	/**
+	 * Builder for similarity functions.
+	 *
+	 * @since 4.5
+	 */
+	public interface SimilarityBuilder {
+
+		/**
+		 * Return the Cosine similarity function for the column in the selection based on the previously defined vector.
+		 *
+		 * @return cosine similarity function selector.
+		 */
+		Selector cosine();
+
+		/**
+		 * Return the Euclidean similarity function for the column in the selection based on the previously defined vector.
+		 *
+		 * @return euclidean similarity function selector.
+		 */
+		Selector euclidean();
+
+		/**
+		 * Return the Dot-Product similarity function for the column in the selection based on the previously defined
+		 * vector.
+		 *
+		 * @return dot-product similarity function selector.
+		 */
+		Selector dotProduct();
+
+		/**
+		 * Return a similarity function using {@link SimilarityFunction} for the column in the selection based on the
+		 * previously defined vector.
+		 *
+		 * @param similarityFunction must not be {@literal null}.
+		 * @return similarity function selector.
+		 */
+		Selector using(SimilarityFunction similarityFunction);
+
+	}
+
+	static class DefaultSelectorBuilder implements SelectorBuilder {
+
+		private final ColumnName columnName;
+
+		DefaultSelectorBuilder(ColumnName columnName) {
+			this.columnName = columnName;
+		}
+
+		@Override
+		public Selector column() {
+			return new ColumnSelector(columnName);
+		}
+
+		@Override
+		public Selector ttl() {
+			return FunctionCall.from("TTL", ColumnSelector.from(columnName));
+		}
+
+		@Override
+		public SimilarityBuilder similarity(CqlVector<?> vector) {
+			return similarity(CassandraVector.of(vector));
+		}
+
+		@Override
+		public SimilarityBuilder similarity(Vector vector) {
+
+			Assert.notNull(vector, "Vector must not be null");
+
+			return new SimilarityBuilder() {
+				@Override
+				public Selector cosine() {
+					return using(SimilarityFunction.COSINE);
+				}
+
+				@Override
+				public Selector euclidean() {
+					return using(SimilarityFunction.EUCLIDEAN);
+				}
+
+				@Override
+				public Selector dotProduct() {
+					return using(SimilarityFunction.DOT_PRODUCT);
+				}
+
+				@Override
+				public Selector using(SimilarityFunction similarityFunction) {
+					return FunctionCall.from("similarity_" + similarityFunction.name().toLowerCase(Locale.ROOT), columnName,
+							vector);
+				}
+			};
+		}
+
+	}
+
 }

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

@@ -26,6 +26,7 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.PageRequest;
@@ -86,6 +87,16 @@ public class Query implements Filter {
 		return EMPTY;
 	}
 
+	/**
+	 * Static factory method to create a {@link Query} for the given column selection.
+	 *
+	 * @return a new {@link Query} selecting {@link Columns}.
+	 * @since 4.5
+	 */
+	public static Query select(Columns columns) {
+		return EMPTY.columns(columns);
+	}
+
 	/**
 	 * Static factory method to create a {@link Query} using the provided {@link CriteriaDefinition}.
 	 *
@@ -182,7 +193,17 @@ public class Query implements Filter {
 			}
 		}
 
-		return new Query(this.criteriaDefinitions, this.columns, this.sort.and(sort), this.scrollPosition,
+		Sort sortToUse = this.sort;
+		if (this.sort.isUnsorted()) {
+			sortToUse = sort;
+		} else {
+			if (sortToUse instanceof VectorSort || sort instanceof VectorSort) {
+				throw new InvalidDataAccessApiUsageException("Cannot concatenate multiple VectorSort instances");
+			}
+			sortToUse = this.sort.and(sort);
+		}
+
+		return new Query(this.criteriaDefinitions, this.columns, sortToUse, this.scrollPosition,
 				this.queryOptions, this.limit, this.allowFiltering);
 	}
 

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core.query;
+
+import java.io.Serial;
+import java.util.List;
+
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Vector;
+import org.springframework.util.Assert;
+
+import com.datastax.oss.driver.api.core.data.CqlVector;
+
+/**
+ * Sort option for queries that applies vector sorting.
+ *
+ * @author Mark Paluch
+ * @since 4.5
+ */
+public class VectorSort extends Sort {
+
+	private static final @Serial long serialVersionUID = 1L;
+
+	private final Object vector;
+
+	public VectorSort(String column, Object vector) {
+		super(List.of(Order.by(column)));
+		this.vector = vector;
+	}
+
+	public VectorSort(List<Order> orders, Object vector) {
+		super(orders);
+		Assert.isTrue(orders.size() == 1, "Orders must contain a single element");
+		this.vector = vector;
+	}
+
+	/**
+	 * Creates a new {@link VectorSort} for the given attributes with the default sort direction.
+	 */
+	public static VectorSort ann(String column, CqlVector<? extends Number> vector) {
+		return new VectorSort(column, vector);
+	}
+
+	/**
+	 * Creates a new {@link VectorSort} for the given attributes with the default sort direction.
+	 */
+	public static VectorSort ann(String column, float... vector) {
+		return new VectorSort(column, Vector.of(vector));
+	}
+
+	/**
+	 * Creates a new {@link VectorSort} for the given attributes with the default sort direction.
+	 */
+	public static VectorSort ann(String column, double... vector) {
+		return new VectorSort(column, Vector.of(vector));
+	}
+
+	/**
+	 * Creates a new {@link VectorSort} for the given attributes with the default sort direction.
+	 */
+	public static VectorSort ann(String column, Vector vector) {
+		return new VectorSort(column, vector);
+	}
+
+	public Object getVector() {
+		return vector;
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParametersParameterAccessor.java

@@ -52,6 +52,7 @@ public class CassandraParametersParameterAccessor extends ParametersParameterAcc
 	@Override
 	public DataType getDataType(int index) {
 
+		// TODO: Vector
 		CassandraType cassandraType = findCassandraType(index);
 
 		return (cassandraType != null ? CassandraSimpleTypeHolder.getDataTypeFor(cassandraType.type())

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

@@ -47,6 +47,7 @@ import org.springframework.data.cassandra.core.mapping.PrimaryKeyClass;
 import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
 import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
+import org.springframework.data.cassandra.core.mapping.VectorType;
 import org.springframework.data.convert.CustomConversions;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -188,17 +189,18 @@ class CassandraPersistentEntitySchemaCreatorUnitTests extends CassandraPersisten
 		verify(operations).execute("CREATE INDEX ON indexedentity (firstname);");
 	}
 
-	@Test // DATACASS-213
-	void foo() {
+	@Test // GH-1504
+	void shouldCreateTableWithVector() {
 
-		context.getPersistentEntity(Person.class);
+		context.getPersistentEntity(TableWithVector.class);
 
 		CassandraPersistentEntitySchemaCreator schemaCreator = new CassandraPersistentEntitySchemaCreator(context,
 				adminOperations);
 
 		schemaCreator.createTables(false);
 
-		// verify(operations).execute("CREATE INDEX ON indexedentity (firstname);");
+		verify(operations).execute(
+				"CREATE TABLE tablewithvector (id text, comment text, comment_vs vector<float, 1536>, PRIMARY KEY (id));");
 	}
 
 	private void verifyTypesGetCreatedInOrderFor(String... typenames) {
@@ -253,4 +255,12 @@ class CassandraPersistentEntitySchemaCreatorUnitTests extends CassandraPersisten
 
 		int age;
 	}
+
+	@Table
+	public static class TableWithVector {
+
+		@PrimaryKey String id;
+		String comment;
+		@VectorType(dimensions = 1536) float[] comment_vs;
+	}
 }

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

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2016-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
+import org.springframework.data.cassandra.core.cql.CqlTemplate;
+import org.springframework.data.cassandra.core.mapping.SaiIndexed;
+import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
+import org.springframework.data.cassandra.core.mapping.Table;
+import org.springframework.data.cassandra.core.mapping.VectorType;
+import org.springframework.data.cassandra.core.query.Columns;
+import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.cassandra.core.query.VectorSort;
+import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
+import org.springframework.data.cassandra.support.CassandraVersion;
+import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTests;
+import org.springframework.data.domain.Vector;
+import org.springframework.data.util.Version;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+
+/**
+ * Integration tests for {@link CassandraTemplate} using Vector Search.
+ *
+ * @author Mark Paluch
+ */
+class CassandraVectorSearchIntegrationTests extends AbstractKeyspaceCreatingIntegrationTests {
+
+	private static final Version CASSANDRA_5 = Version.parse("5.0");
+
+	private Version cassandraVersion;
+
+	private CassandraTemplate template;
+
+	@BeforeEach
+	void setUp() {
+
+		MappingCassandraConverter converter = new MappingCassandraConverter();
+		converter.setUserTypeResolver(new SimpleUserTypeResolver(session, CqlIdentifier.fromCql(keyspace)));
+		converter.afterPropertiesSet();
+
+		cassandraVersion = CassandraVersion.get(session);
+
+		template = new CassandraTemplate(new CqlTemplate(session), converter);
+
+		prepareTemplate(template);
+
+		SchemaTestUtils.potentiallyCreateTableFor(Comments.class, template);
+	}
+
+	/**
+	 * Post-process the {@link CassandraTemplate} before running the tests.
+	 *
+	 * @param template
+	 */
+	void prepareTemplate(CassandraTemplate template) {
+
+	}
+
+	@Test // GH-1504
+	void shouldQueryVector() {
+
+		assertThat(cassandraVersion.isGreaterThanOrEqualTo(CASSANDRA_5)).isTrue();
+
+		Comments one = new Comments();
+		one.setId(UUID.randomUUID());
+		one.setVector(Vector.of(0.45f, 0.09f, 0.01f, 0.2f, 0.11f));
+		one.setComment("Raining too hard should have postponed");
+
+		Comments two = new Comments();
+		two.setId(UUID.randomUUID());
+		two.setVector(Vector.of(0.99f, 0.5f, 0.99f, 0.1f, 0.34f));
+		two.setComment("Second rest stop was out of water");
+
+		Comments three = new Comments();
+		three.setId(UUID.randomUUID());
+		three.setVector(Vector.of(0.9f, 0.54f, 0.12f, 0.1f, 0.95f));
+		three.setComment("LATE RIDERS SHOULD NOT DELAY THE START");
+
+		template.insert(one);
+		template.insert(two);
+		template.insert(three);
+
+		Vector vector = Vector.of(0.2f, 0.15f, 0.3f, 0.2f, 0.05f);
+
+		Columns columns = Columns.empty().include("comment").select("vector",
+				it -> it.similarity(vector).cosine().as("similarity"));
+		Query query = Query.select(columns).limit(3).sort(VectorSort.ann("vector", vector));
+
+		List<CommentSearch> result = template.query(Comments.class).as(CommentSearch.class).matching(query).all();
+
+		assertThat(result).hasSize(3);
+		for (CommentSearch commentSearch : result) {
+			assertThat(commentSearch.similarity).isNotCloseTo(0f, offset(0.1f));
+		}
+	}
+
+	static class CommentSearch {
+
+		String comment;
+
+		float similarity;
+
+		@Override
+		public String toString() {
+			return "CommentSearch{" + "comment='" + comment + '\'' + ", similarity=" + similarity + '}';
+		}
+	}
+
+	@Table
+	static class Comments {
+
+		@Id UUID id;
+		String comment;
+
+		@VectorType(dimensions = 5)
+		@SaiIndexed Vector vector;
+
+		public UUID getId() {
+			return id;
+		}
+
+		public void setId(UUID id) {
+			this.id = id;
+		}
+
+		public String getComment() {
+			return comment;
+		}
+
+		public void setComment(String comment) {
+			this.comment = comment;
+		}
+
+		public Vector getVector() {
+			return vector;
+		}
+
+		public void setVector(Vector vector) {
+			this.vector = vector;
+		}
+	}
+
+}

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

@@ -38,17 +38,21 @@ import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder.ParameterHandling;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.VectorType;
 import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.core.query.Update;
+import org.springframework.data.cassandra.core.query.VectorSort;
 import org.springframework.data.cassandra.domain.Group;
 import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Vector;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 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.codec.TypeCodec;
@@ -210,8 +214,7 @@ class StatementFactoryUnitTests {
 	@Test // GH-1172
 	void shouldMapSelectInQueryAsInlineValue() {
 
-		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")),
-				groupEntity);
+		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")), groupEntity);
 
 		assertThat(select.build(ParameterHandling.INLINE).getQuery()).isEqualTo("SELECT * FROM group WHERE foo IN ('bar')");
 
@@ -224,8 +227,7 @@ class StatementFactoryUnitTests {
 	@Test // GH-1172
 	void shouldMapSelectInQueryAsByIndexValue() {
 
-		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")),
-				groupEntity);
+		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")), groupEntity);
 		SimpleStatement statement = select.build(ParameterHandling.BY_INDEX);
 
 		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM group WHERE foo IN ?");
@@ -241,8 +243,7 @@ class StatementFactoryUnitTests {
 	@Test // GH-1172
 	void shouldMapSelectInQueryAsByNamedValue() {
 
-		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")),
-				groupEntity);
+		StatementBuilder<Select> select = statementFactory.select(Query.query(where("foo").in("bar")), groupEntity);
 		SimpleStatement statement = select.build(ParameterHandling.BY_NAME);
 
 		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM group WHERE foo IN :p0");
@@ -710,8 +711,7 @@ class StatementFactoryUnitTests {
 	@Test // DATACASS-569
 	void shouldCreateSetUpdateIfExists() {
 
-		Query query = Query.query(where("foo").is("bar"))
-				.queryOptions(UpdateOptions.builder().withIfExists().build());
+		Query query = Query.query(where("foo").is("bar")).queryOptions(UpdateOptions.builder().withIfExists().build());
 
 		StatementBuilder<com.datastax.oss.driver.api.querybuilder.update.Update> update = statementFactory.update(query,
 				Update.empty().set("firstName", "baz"), personEntity);
@@ -914,6 +914,56 @@ class StatementFactoryUnitTests {
 				.isEqualTo("SELECT count(1) FROM group WHERE foo='bar'");
 	}
 
+	@Test // GH-1504
+	void shouldUpdateCqlVector() {
+
+		WithVector withVector = new WithVector();
+		withVector.id = "foo";
+		withVector.vector = CqlVector.newInstance(1.2f, 1.3f);
+		withVector.array = new float[] { 2.2f, 2.3f };
+		withVector.list = Arrays.asList(3.2f, 3.3f);
+
+		SimpleStatement statement = statementFactory.update(withVector, WriteOptions.empty())
+				.build(ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo("UPDATE withvector SET vector=:p0, array=:p1, list=:p2 WHERE id=:p3");
+
+		assertThat(statement.getNamedValues().get(CqlIdentifier.fromCql("p0"))).isInstanceOf(CqlVector.class).hasToString("[1.2, 1.3]");
+		assertThat(statement.getNamedValues().get(CqlIdentifier.fromCql("p1"))).isInstanceOf(CqlVector.class).hasToString("[2.2, 2.3]");
+		assertThat(statement.getNamedValues().get(CqlIdentifier.fromCql("p2"))).isInstanceOf(CqlVector.class).hasToString("[3.2, 3.3]");
+	}
+
+	@Test // GH-1504
+	void shouldQueryVector() {
+
+		Query query = Query.empty()
+				.sort(VectorSort.ann("vector", Vector.of(1.2f, 1.3f)));
+
+		SimpleStatement statement = statementFactory.select(query, converter.getMappingContext().getRequiredPersistentEntity(WithVector.class))
+				.build(ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM withvector ORDER BY vector ANN OF [1.2, 1.3]");
+		assertThat(statement.getNamedValues()).isEmpty();
+	}
+
+	@Test // GH-1504
+	void shouldRenderSimilaritySelector() {
+
+		Vector vector = Vector.of(0.2f, 0.15f, 0.3f, 0.2f, 0.05f);
+
+		Columns columns = Columns.empty().include("comment").select("vector", it -> it.similarity(vector).cosine());
+
+		Query query = Query.select(columns).sort(VectorSort.ann("vector", Vector.of(1.2f, 1.3f)));
+
+		SimpleStatement statement = statementFactory
+				.select(query, converter.getMappingContext().getRequiredPersistentEntity(WithVector.class))
+				.build(ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo(
+				"SELECT comment,similarity_cosine(vector,[0.2, 0.15, 0.3, 0.2, 0.05]) FROM withvector ORDER BY vector ANN OF [1.2, 1.3]");
+		assertThat(statement.getNamedValues()).isEmpty();
+	}
+
 	@SuppressWarnings("unused")
 	static class Person {
 
@@ -933,4 +983,13 @@ class StatementFactoryUnitTests {
 	record MyString(String value) {
 
 	}
+
+	static class WithVector {
+
+		@Id String id;
+
+		@VectorType(dimensions = 12) CqlVector<Number> vector;
+		@VectorType(dimensions = 12) float[] array;
+		@VectorType(dimensions = 12) List<Float> list;
+	}
 }

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

@@ -30,13 +30,15 @@ import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentE
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.Indexed;
 import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
-import org.springframework.data.cassandra.core.mapping.SAIIndexed;
 import org.springframework.data.cassandra.core.mapping.SASI;
 import org.springframework.data.cassandra.core.mapping.SASI.NonTokenizingAnalyzed;
 import org.springframework.data.cassandra.core.mapping.SASI.Normalization;
 import org.springframework.data.cassandra.core.mapping.SASI.StandardAnalyzed;
+import org.springframework.data.cassandra.core.mapping.SaiIndexed;
+import org.springframework.data.cassandra.core.mapping.SimilarityFunction;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 
 /**
  * Unit tests for {@link IndexSpecificationFactory}.
@@ -98,22 +100,36 @@ class IndexSpecificationFactoryUnitTests {
 		assertThat(simpleSasi.getUsing()).isEqualTo("sai");
 		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
 		assertThat(simpleSasi.getOptions()).containsEntry("case_sensitive", "true").containsEntry("normalize", "false")
-				.containsEntry("ascii", "false").containsEntry("similarity_function", "COSINE");
+				.containsEntry("ascii", "false").doesNotContainKey("similarity_function");
 	}
 
 	@Test // GH-1505
 	void createSaiIndexShouldApplyIndexOptions() {
 
-		CreateIndexSpecification simpleSasi = createIndexFor(IndexedType.class, "customSai");
+		CreateIndexSpecification simpleSai = createIndexFor(IndexedType.class, "customSai");
 
-		assertThat(simpleSasi.getName()).isEqualTo(CqlIdentifier.fromInternal("foo"));
-		assertThat(simpleSasi.getColumnName()).isEqualTo(CqlIdentifier.fromInternal("customsai"));
-		assertThat(simpleSasi.getTableName()).isNull();
-		assertThat(simpleSasi.isCustom()).isFalse();
-		assertThat(simpleSasi.getUsing()).isEqualTo("sai");
-		assertThat(simpleSasi.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
-		assertThat(simpleSasi.getOptions()).containsEntry("case_sensitive", "false").containsEntry("normalize", "true")
-				.containsEntry("ascii", "true").containsEntry("similarity_function", "EUCLIDEAN");
+		assertThat(simpleSai.getName()).isEqualTo(CqlIdentifier.fromInternal("foo"));
+		assertThat(simpleSai.getColumnName()).isEqualTo(CqlIdentifier.fromInternal("customsai"));
+		assertThat(simpleSai.getTableName()).isNull();
+		assertThat(simpleSai.isCustom()).isFalse();
+		assertThat(simpleSai.getUsing()).isEqualTo("sai");
+		assertThat(simpleSai.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(simpleSai.getOptions()).containsEntry("case_sensitive", "false").containsEntry("normalize", "true")
+				.containsEntry("ascii", "true").doesNotContainKey("similarity_function");
+	}
+
+	@Test // GH-1504
+	void createSaiIndexShouldApplyVectorIndexOptions() {
+
+		CreateIndexSpecification vector = createIndexFor(IndexedType.class, "someVector");
+
+		assertThat(vector.getName()).isNull();
+		assertThat(vector.getColumnName()).isEqualTo(CqlIdentifier.fromInternal("somevector"));
+		assertThat(vector.getTableName()).isNull();
+		assertThat(vector.isCustom()).isFalse();
+		assertThat(vector.getUsing()).isEqualTo("sai");
+		assertThat(vector.getColumnFunction()).isEqualTo(ColumnFunction.NONE);
+		assertThat(vector.getOptions()).hasSize(1).containsEntry("similarity_function", "COSINE");
 	}
 
 	@Test // GH-1505
@@ -236,10 +252,12 @@ class IndexSpecificationFactoryUnitTests {
 		@NonTokenizingAnalyzed(caseSensitive = false,
 				normalization = Normalization.LOWERCASE) String sasiNontokenizingLowercase;
 
-		@SAIIndexed String simpleSai;
+		@SaiIndexed String simpleSai;
 
-		@SAIIndexed(value = "foo", caseSensitive = false, normalize = true, ascii = true,
-				similarityFunction = SAIIndexed.SimilarityFunction.EUCLIDEAN) String customSai;
+		@SaiIndexed CqlVector<?> someVector;
+
+		@SaiIndexed(value = "foo", caseSensitive = false, normalize = true, ascii = true,
+				similarityFunction = SimilarityFunction.EUCLIDEAN) String customSai;
 	}
 
 	@AccessType(Type.PROPERTY)
@@ -255,7 +273,7 @@ class IndexSpecificationFactoryUnitTests {
 	@AccessType(Type.PROPERTY)
 	private static class SaiIndexedMapKeyProperty {
 
-		public Map<@SAIIndexed String, String> getEntries() {
+		public Map<@SaiIndexed String, String> getEntries() {
 			return null;
 		}
 

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

@@ -1823,4 +1823,5 @@ public class MappingCassandraConverterUnitTests {
 			this.number = number;
 		}
 	}
+
 }

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

@@ -21,6 +21,7 @@ import static org.springframework.data.domain.Sort.Order.*;
 
 import java.time.LocalDate;
 import java.time.LocalTime;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Currency;
@@ -35,7 +36,10 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
+
 import org.springframework.data.annotation.Id;
+import org.springframework.data.cassandra.core.StatementFactory;
+import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.Column;
@@ -45,6 +49,7 @@ import org.springframework.data.cassandra.core.mapping.PrimaryKeyColumn;
 import org.springframework.data.cassandra.core.mapping.Tuple;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
+import org.springframework.data.cassandra.core.mapping.VectorType;
 import org.springframework.data.cassandra.core.query.ColumnName;
 import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Columns.Selector;
@@ -63,6 +68,8 @@ import org.springframework.data.domain.Sort.Order;
 import org.springframework.lang.Nullable;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 import com.datastax.oss.driver.api.core.data.TupleValue;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.type.DataTypes;
@@ -78,6 +85,8 @@ public class QueryMapperUnitTests {
 
 	private final CassandraMappingContext mappingContext = new CassandraMappingContext();
 
+	private MappingCassandraConverter cassandraConverter;
+
 	private CassandraPersistentEntity<?> personPersistentEntity;
 
 	private QueryMapper queryMapper;
@@ -98,7 +107,7 @@ public class QueryMapperUnitTests {
 		mappingContext.setCustomConversions(customConversions);
 		mappingContext.setUserTypeResolver(userTypeResolver);
 
-		MappingCassandraConverter cassandraConverter = new MappingCassandraConverter(mappingContext);
+		cassandraConverter = new MappingCassandraConverter(mappingContext);
 
 		cassandraConverter.setCustomConversions(customConversions);
 		cassandraConverter.afterPropertiesSet();
@@ -431,6 +440,47 @@ public class QueryMapperUnitTests {
 		assertThat(mappedObject.iterator().next().getPredicate().getValue()).isEqualTo(42L);
 	}
 
+	@Test //
+	void shouldConvertVectorValues() {
+
+		Filter filter = Filter.from(Criteria.where("array").is(new float[] { 1.1f, 2.2f }));
+
+		Filter mappedObject = this.queryMapper.getMappedObject(filter,
+				this.mappingContext.getRequiredPersistentEntity(WithVector.class));
+
+		assertThat(mappedObject.iterator().next().getColumnName()).isEqualTo(ColumnName.from("array"));
+		assertThat(mappedObject.iterator().next().getPredicate().getValue()).isEqualTo(new float[] { 1.1f, 2.2f });
+	}
+
+	@Test // GH-1504
+	void shouldConvertVectorValuesFromList() {
+
+		Filter filter = Filter.from(Criteria.where("list").is(Arrays.asList(1.1f, 2.2f)));
+
+		Filter mappedObject = this.queryMapper.getMappedObject(filter,
+				this.mappingContext.getRequiredPersistentEntity(WithVector.class));
+
+		assertThat(mappedObject.iterator().next().getColumnName()).isEqualTo(ColumnName.from("list"));
+		assertThat(mappedObject.iterator().next().getPredicate().getValue()).isEqualTo(CqlVector.newInstance(1.1f, 2.2f));
+	}
+
+	@Test // GH-1504
+	void shouldConvertVectorSelectorFunction() {
+
+		Columns columns = Columns.empty();
+		Columns.FunctionCall similarity = Columns.FunctionCall.from("similarity_cosine", CqlIdentifier.fromCql("array"), Arrays.asList(1.1f, 2.2f));
+
+		Query query = Query.empty().columns(columns.select("array", similarity));
+
+		StatementFactory factory = new StatementFactory(queryMapper, new UpdateMapper(cassandraConverter));
+
+		SimpleStatement statement = factory.select(query, this.mappingContext.getRequiredPersistentEntity(WithVector.class))
+				.build(StatementBuilder.ParameterHandling.BY_NAME);
+
+		assertThat(statement.getQuery()).isEqualTo("SELECT similarity_cosine(array,[1.1, 2.2]) FROM withvector");
+		assertThat(statement.getNamedValues()).isEmpty();
+	}
+
 	@SuppressWarnings("unused")
 	static class Person {
 
@@ -540,4 +590,10 @@ public class QueryMapperUnitTests {
 			this.age = age;
 		}
 	}
+
+	static class WithVector {
+
+		@VectorType(dimensions = 12) float[] array;
+		@VectorType(dimensions = 12) List<Float> list;
+	}
 }

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

@@ -44,11 +44,13 @@ import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecifica
 import org.springframework.data.cassandra.core.mapping.*;
 import org.springframework.data.cassandra.domain.AllPossibleTypes;
 import org.springframework.data.cassandra.support.UserDefinedTypeBuilder;
+import org.springframework.data.domain.Vector;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.util.StringUtils;
 
 import com.datastax.driver.core.UDTValue;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.data.CqlVector;
 import com.datastax.oss.driver.api.core.data.TupleValue;
 import com.datastax.oss.driver.api.core.data.UdtValue;
 import com.datastax.oss.driver.api.core.type.DataType;
@@ -966,4 +968,53 @@ public class SchemaFactoryUnitTests {
 
 		int age;
 	}
+
+	@Test // GH-1504
+	void shouldCreateTableWithVector() {
+
+		CassandraPersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(TypeWithVector.class);
+
+		CreateTableSpecification tableSpecification = schemaFactory.getCreateTableSpecificationFor(persistentEntity);
+
+		ColumnSpecification cqlvector = tableSpecification.getColumns().get(1);
+		assertThat(cqlvector.getName().toString()).isEqualTo("cqlvector");
+		assertThat(cqlvector.getType()).isEqualTo(DataTypes.vectorOf(DataTypes.FLOAT, 5));
+	}
+
+	@Test // GH-1504
+	void shouldFailOnMissingCqlVectorAnnotation() {
+
+		assertThatExceptionOfType(MappingException.class)
+				.isThrownBy(() -> schemaFactory.getCreateTableSpecificationFor(
+						mappingContext.getRequiredPersistentEntity(TypeWithCqlVectorMissingAnnotation.class)))
+				.withMessageContaining("@VectorType");
+
+		assertThatExceptionOfType(MappingException.class)
+				.isThrownBy(() -> schemaFactory.getCreateTableSpecificationFor(
+						mappingContext.getRequiredPersistentEntity(TypeWithVectorMissingAnnotation.class)))
+				.withMessageContaining("@VectorType");
+	}
+
+	@Table
+	private static class TypeWithVector {
+		@Id String id;
+
+		@VectorType(dimensions = 5) CqlVector<Float> cqlVector;
+	}
+
+	@Table
+	private static class TypeWithCqlVectorMissingAnnotation {
+
+		@Id String id;
+
+		CqlVector<Float> cqlVector;
+	}
+
+	@Table
+	private static class TypeWithVectorMissingAnnotation {
+
+		@Id String id;
+
+		Vector cqlVector;
+	}
 }

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

@@ -36,7 +36,7 @@ class CassandraSimpleTypeHolderUnitTests {
 	@Test // DATACASS-488
 	void shouldResolveTypeNamesForAllPrimaryTypes() {
 
-		EnumSet<Name> excluded = EnumSet.of(Name.MAP, Name.SET, Name.LIST, Name.UDT, Name.TUPLE);
+		EnumSet<Name> excluded = EnumSet.of(Name.MAP, Name.SET, Name.LIST, Name.UDT, Name.TUPLE, Name.VECTOR);
 
 		for (Name name : Name.values()) {
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SchemaTestUtils.java

@@ -15,13 +15,14 @@
  */
 package org.springframework.data.cassandra.repository.support;
 
+import java.util.List;
 import java.util.Optional;
 
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.convert.SchemaFactory;
 import org.springframework.data.cassandra.core.cql.SessionCallback;
 import org.springframework.data.cassandra.core.cql.generator.CqlGenerator;
-import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGenerator;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateIndexSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
 import org.springframework.data.cassandra.core.cql.keyspace.CreateUserTypeSpecification;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -79,8 +80,12 @@ public class SchemaTestUtils {
 					.flatMap(it -> it.getTable(persistentEntity.getTableName()));
 
 			if (!table.isPresent()) {
+
 				CreateTableSpecification tableSpecification = schemaFactory.getCreateTableSpecificationFor(persistentEntity);
-				operations.getCqlOperations().execute(new CreateTableCqlGenerator(tableSpecification).toCql());
+				operations.getCqlOperations().execute(CqlGenerator.toCql(tableSpecification));
+
+				List<CreateIndexSpecification> indexes = schemaFactory.getCreateIndexSpecificationsFor(persistentEntity);
+				indexes.forEach(index -> operations.getCqlOperations().execute(CqlGenerator.toCql(index)));
 			}
 			return null;
 		});

src/main/antora/modules/ROOT/examples/VectorSearchExample.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2020-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+// tag::file[]
+package org.springframework.data.cassandra.example;
+
+import org.springframework.data.cassandra.core.mapping.PrimaryKey;
+import org.springframework.data.cassandra.core.mapping.Table;
+
+// tag::class[]
+@Table
+class Comments {
+
+	@Id UUID id;
+	String comment;
+
+	@VectorType(dimensions = 5)
+	@SaiIndexed Vector vector;
+}
+
+class CommentSearch {
+
+	String comment;
+
+	float similarity;
+}
+// end::class[]
+
+// end::file[]

src/main/antora/modules/ROOT/examples/mapping/PersonWithIndexes.java

@@ -34,6 +34,10 @@ class PersonWithIndexes {
   @StandardAnalyzed
   private String names;
 
+  @SaiIndexed
+  @VectorType(dimensions = 1536) // required for table generation
+  private Vector vector;
+
   @Indexed("indexed_map")
   private Map<String, String> entries;
 

src/main/antora/modules/ROOT/pages/cassandra/template.adoc

@@ -293,9 +293,56 @@ TIP: You can directly apply xref:repositories/projections.adoc[] to resulting do
 
 The terminating methods (`first()`, `one()`, `all()`, and `stream()`) handle switching between retrieving a single entity and retrieving multiple entities as `List` or `Stream` and similar operations.
 
-WARNING: The new fluent template API methods (that is, `query(..)`, `insert(..)`, `update(..)`, and `delete(..)`) use effectively thread-safe supporting objects to compose the CQL statement.
+WARNING: The new fluent template API methods (that is, `query(…)`, `insert(…)`, `update(…)`, and `delete(…)`) use effectively thread-safe supporting objects to compose the CQL statement.
 However, it comes at the added cost of additional young-gen JVM heap overhead, since the design is based on final fields for the various CQL statement components and construction on mutation.
-You should be careful when possibly inserting or deleting a large number of objects (such as inside of a loop, for instance).
+You should be careful when possibly inserting or deleting a large number of objects (such as inside a loop, for instance).
+
+=== Vector Search Queries
+
+Projections are the foundation for querying data returned in a different form than the entity.
+While Cassandra's primary use-case follows a key-value model – storing what you retrieve – Vector search is different.
+Running Vector search queries typically results in an aggregation or report-like result set.
+A typical query would return some form of content (such as a `text` column) along with its score (or distance) to the actual vector.
+
+Consider the following domain model:
+
+.Vector Search model
+====
+[source,java]
+----
+include::example$VectorSearchExample.java[tags=class]
+----
+====
+
+`Comments` is the domain type defining a vector column and the `comment` column.
+
+Running a Vector Search requires usage of `ANN` sorting and would typically define a similarity function to determine its distance from the given vector.
+
+.Using Vector Search
+====
+[source,java,indent=0,subs="verbatim,quotes",role="primary"]
+----
+Columns columns = Columns.from("comment")                                                <1>
+                     .select("vector", builder -> builder.similarity(vector)
+                                                         .cosine().as("similarity"));    <2>
+
+Query query = Query.select(columns)
+                   .limit(3)
+                   .sort(VectorSort.ann("vector", vector));                              <3>
+
+template.query(Comments.class)
+        .as(CommentSearch.class)                                                         <4>
+        .matching(query)
+        .all();
+----
+
+<1> Select the columns to query.
+<2> Include a function call to `similarity_cosine(vector, […])`. `Columns.select(…)` uses a selector builder customizer to configure the actual selection.
+Make sure to declare an alias to map the result onto `CommentSearch.similarity` as result mapping uses column names.
+<3> Use the `VectorSort.ann(…)` to define the sort order in comparison to the given vector.
+<4> Define the target type to map the result to.
+The `CommentSearch` type is result projection type defining columns that map the result.
+====
 
 [[cassandra-template.save-update-remove]]
 == Saving, Updating, and Removing Rows
@@ -556,7 +603,7 @@ template.save(daenerys);
 template.save(tmp); // throws OptimisticLockingFailureException                       <4>
 ----
 
-<1> Intially insert document. `version` is set to `0`.
+<1> Initially insert document. `version` is set to `0`.
 <2> Load the just inserted document. `version` is still `0`.
 <3> Update the document with `version = 0`.
 Set the `lastname` and bump `version` to `1`.

src/main/antora/modules/ROOT/pages/object-mapping.adoc

@@ -79,6 +79,9 @@ The following table maps Spring Data types to Cassandra types:
 | `UDTValue`, mapped User-Defined Types
 | user type
 
+| `Vector`, `CqlVector`
+| `vector<…>`
+
 | `java.util.Map<K, V>`
 | `map`
 
@@ -374,7 +377,7 @@ Opposite to this behavior `USE_EMPTY` tries to create a new instance using eithe
 ----
 public class User {
 
-	@PrimaryKey("user_id")
+    @PrimaryKey("user_id")
     private String userId;
 
     @Embedded(onEmpty = USE_NULL) <1>
@@ -448,8 +451,12 @@ Properties of the embedded object are flattened into the structure of its parent
 Describes the index to be created at session initialization.
 * `@SASI`: Applied at the field level.
 Allows SASI index creation during session initialization.
+* `@SaiIndexed`: Applied at the field level.
+Allows to define SAI (storage-attached indexes) index creation during session initialization.
 * `@CassandraType`: Applied at the field level to specify a Cassandra data type.
 Types are derived from the property declaration by default.
+* `@VectorType`: Applied at the field level to specify the Cassandra vector type.
+This annotation is required when using schema generation.
 * `@Frozen`: Applied at the field level to class-types and parametrized types.
 Declares a frozen UDT column or frozen collection like `List<@Frozen UserDefinedPersonType>`.
 * `@UserDefinedType`: Applied at the type level to specify a Cassandra User-defined Data Type (UDT).
@@ -500,7 +507,7 @@ include::example$mapping/Coordinates.java[tags=class]
 [[mapping.index-creation]]
 ==== Index Creation
 
-You can annotate particular entity properties with `@Indexed` or `@SASI` if you wish to create secondary indexes on application startup.
+You can annotate particular entity properties with `@Indexed`, `@SaiIndexed`, or `@SASI` if you wish to create secondary indexes on application startup.
 Index creation creates simple secondary indexes for scalar types, user-defined types, and collection types.
 
 You can configure a SASI Index to apply an analyzer, such as `StandardAnalyzer` or `NonTokenizingAnalyzer` (by using
@@ -520,7 +527,7 @@ include::example$mapping/PersonWithIndexes.java[tags=class]
 
 [NOTE]
 ====
-The `@Indexed` annotation can be applied to single properties of embedded entities or along side with the `@Embedded` annotation, in which case all properties of the embedded are indexed.
+The `@Indexed` annotation can be applied to single properties of embedded entities or along with the `@Embedded` annotation, in which case all properties of the embedded are indexed.
 ====
 
 CAUTION: Index creation on session initialization may have a severe performance impact on application startup.