Do not prefix unsafe orer by expressions with table prefix.

Such expressions now get passed on unchanged.

This also deprecates org.springframework.data.r2dbc.query.QueryMapper.getMappedObject(Sort, RelationalPersistentEntity<?>).
It was only used in tests and translates an Order into another Order, which sounds wrong.

Closes #1512
Original pull request: #1513

Author: schauder

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

@@ -93,6 +93,8 @@ public List<OrderByField> getMappedSort(Table table, Sort sort, @Nullable Relati
 
 		for (Sort.Order order : sort) {
 
+			SqlSort.validate(order);
+
 			OrderByField simpleOrderByField = createSimpleOrderByField(table, entity, order);
 			OrderByField orderBy = simpleOrderByField
 					.withNullHandling(order.getNullHandling());
@@ -105,7 +107,9 @@ public List<OrderByField> getMappedSort(Table table, Sort sort, @Nullable Relati
 
 	private OrderByField createSimpleOrderByField(Table table, RelationalPersistentEntity<?> entity, Sort.Order order) {
 
-		SqlSort.validate(order);
+		if (order instanceof SqlSort.SqlOrder sqlOrder && sqlOrder.isUnsafe()) {
+			return OrderByField.from(Expressions.just(sqlOrder.getProperty()));
+		}
 
 		Field field = createPropertyField(entity, SqlIdentifier.unquoted(order.getProperty()), this.mappingContext);
 		return OrderByField.from(table.column(field.getMappedColumnName()));

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

@@ -430,7 +430,7 @@ public void shouldMapSortWithUnsafeExpression() {
 
 		assertThat(fields) //
 				.extracting(Objects::toString) //
-				.containsExactly("tbl." + unsafeExpression + " ASC");
+				.containsExactly( unsafeExpression + " ASC");
 	}
 
 	private Condition map(Criteria criteria) {

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/query/QueryMapper.java

@@ -22,6 +22,7 @@
 import java.util.Map;
 import java.util.regex.Pattern;
 
+import org.jetbrains.annotations.NotNull;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyPath;
@@ -50,6 +51,8 @@
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
+import static org.springframework.data.relational.core.sql.Expressions.*;
+
 /**
  * Maps {@link CriteriaDefinition} and {@link Sort} objects considering mapping metadata and dialect-specific
  * conversion.
@@ -102,7 +105,9 @@ public String toSql(SqlIdentifier identifier) {
 	 * @param sort must not be {@literal null}.
 	 * @param entity related {@link RelationalPersistentEntity}, can be {@literal null}.
 	 * @return
+	 * @deprecated without replacement.
 	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	public Sort getMappedObject(Sort sort, @Nullable RelationalPersistentEntity<?> entity) {
 
 		if (entity == null) {
@@ -137,6 +142,8 @@ public List<OrderByField> getMappedSort(Table table, Sort sort, @Nullable Relati
 
 		for (Sort.Order order : sort) {
 
+			SqlSort.validate(order);
+
 			OrderByField simpleOrderByField = createSimpleOrderByField(table, entity, order);
 			OrderByField orderBy = simpleOrderByField
 					.withNullHandling(order.getNullHandling());
@@ -147,9 +154,12 @@ public List<OrderByField> getMappedSort(Table table, Sort sort, @Nullable Relati
 
 	}
 
+
 	private OrderByField createSimpleOrderByField(Table table, RelationalPersistentEntity<?> entity, Sort.Order order) {
 
-		SqlSort.validate(order);
+		if (order instanceof SqlSort.SqlOrder sqlOrder && sqlOrder.isUnsafe()) {
+			return OrderByField.from(Expressions.just(sqlOrder.getProperty()));
+		}
 
 		Field field = createPropertyField(entity, SqlIdentifier.unquoted(order.getProperty()), this.mappingContext);
 		return OrderByField.from(table.column(field.getMappedColumnName()));

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/query/QueryMapperUnitTests.java

@@ -39,6 +39,7 @@
 import org.springframework.data.relational.core.sql.Functions;
 import org.springframework.data.relational.core.sql.OrderByField;
 import org.springframework.data.relational.core.sql.Table;
+import org.springframework.data.relational.domain.SqlSort;
 import org.springframework.r2dbc.core.Parameter;
 import org.springframework.r2dbc.core.binding.BindMarkersFactory;
 import org.springframework.r2dbc.core.binding.BindTarget;
@@ -440,6 +441,19 @@ public void shouldMapSortWithUnknownField() {
 				.containsExactly("tbl.unknownField DESC");
 	}
 
+	@Test // GH-1512
+	public void shouldTablePrefixUnsafeOrderExpression() {
+
+		Sort sort = Sort.by(SqlSort.SqlOrder.desc("unknownField").withUnsafe());
+
+		List<OrderByField> fields = mapper.getMappedSort(Table.create("tbl"), sort,
+				mapper.getMappingContext().getRequiredPersistentEntity(Person.class));
+
+		assertThat(fields) //
+				.extracting(Objects::toString) //
+				.containsExactly("unknownField DESC");
+	}
+
 	@Test // GH-1507
 	public void shouldMapSortWithAllowedSpecialCharacters() {