Fix ordering of sort-by properties.

We now apply sort-by properties individually on the built statement instead of collecting these into a Map first.

Resolves #1008

Author: mp911de

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

@@ -641,10 +641,14 @@ private static StatementBuilder<Select> createSelectAndOrder(List<Selector> sele
 
 			builder.apply((statement) -> {
 
-				Map<String, ClusteringOrder> ordering = sort.stream().collect(Collectors.toMap(Sort.Order::getProperty,
-						order -> order.isAscending() ? ClusteringOrder.ASC : ClusteringOrder.DESC));
+				Select statementToUse = statement;
 
-				return statement.orderBy(ordering);
+				for (Sort.Order order : sort) {
+					statementToUse = statementToUse.orderBy(order.getProperty(),
+							order.isAscending() ? ClusteringOrder.ASC : ClusteringOrder.DESC);
+				}
+
+				return statementToUse;
 			});
 		}
 

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

@@ -16,6 +16,7 @@
 package org.springframework.data.cassandra.core;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.domain.Sort.Direction.*;
 
 import java.time.Duration;
 import java.util.Collections;
@@ -131,16 +132,40 @@ public void shouldMapSelectQueryWithTtlColumns() {
 		assertThat(select.build(ParameterHandling.INLINE).getQuery()).isEqualTo("SELECT ttl(email) FROM group");
 	}
 
-	@Test // DATACASS-343
+	@Test // #1008
 	public void shouldMapSelectQueryWithSortLimitAndAllowFiltering() {
 
-		Query query = Query.empty().sort(Sort.by("id.hashPrefix")).limit(10).withAllowFiltering();
+		Query query = Query.empty().sort(Sort.by(DESC, "email", "age"));
+
+		StatementBuilder<Select> select = statementFactory.select(query,
+				converter.getMappingContext().getRequiredPersistentEntity(Group.class));
+
+		assertThat(select.build(ParameterHandling.INLINE).getQuery())
+				.isEqualTo("SELECT * FROM group ORDER BY email DESC,age DESC");
+	}
+
+	@Test // DATACASS-343
+	public void shouldMapSelectQueryWithSortByEmbeddedLimitAndAllowFiltering() {
+
+		Query query = Query.empty().sort(Sort.by(DESC, "id.hashPrefix", "id.username"));
+
+		StatementBuilder<Select> select = statementFactory.select(query,
+				converter.getMappingContext().getRequiredPersistentEntity(Group.class));
+
+		assertThat(select.build(ParameterHandling.INLINE).getQuery())
+				.isEqualTo("SELECT * FROM group ORDER BY hash_prefix DESC,username DESC");
+	}
+
+	@Test // DATACASS-343
+	public void shouldMapSelectQueryWithLimitAndAllowFiltering() {
+
+		Query query = Query.empty().limit(10).withAllowFiltering();
 
 		StatementBuilder<Select> select = statementFactory.select(query,
 				converter.getMappingContext().getRequiredPersistentEntity(Group.class));
 
 		assertThat(select.build(ParameterHandling.INLINE).getQuery())
-				.isEqualTo("SELECT * FROM group ORDER BY hash_prefix ASC LIMIT 10 ALLOW FILTERING");
+				.isEqualTo("SELECT * FROM group LIMIT 10 ALLOW FILTERING");
 	}
 
 	@Test // DATACASS-343