Align with R2DBC 0.9 API changes

Adopt to R2DBC Parameter type, deprecate our own one in favor of the R2DBC type.
Add support for extensible transaction definitions and support to consume Readable and result segments.
Return Long instead of Integer in DatabaseClient update in preparation for R2DBC 1.0 changes.

build.gradle

@@ -346,7 +346,7 @@ configure([rootProject] + javaProjects) { project ->
 			// "https://junit.org/junit5/docs/5.8.2/api/",
 			"https://www.reactive-streams.org/reactive-streams-1.0.3-javadoc/",
 			"https://javadoc.io/static/io.rsocket/rsocket-core/1.1.1/",
-			"https://r2dbc.io/spec/0.8.5.RELEASE/api/",
+			"https://r2dbc.io/spec/0.9.1.RELEASE/api/",
 			// The external Javadoc link for JSR 305 must come last to ensure that types from
 			// JSR 250 (such as @PostConstruct) are still supported. This is due to the fact
 			// that JSR 250 and JSR 305 both define types in javax.annotation, which results

spring-r2dbc/src/main/java/org/springframework/r2dbc/connection/R2dbcTransactionManager.java

@@ -21,8 +21,10 @@ import java.time.Duration;
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.IsolationLevel;
+import io.r2dbc.spi.Option;
 import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.InitializingBean;
@@ -47,7 +49,7 @@ import org.springframework.util.Assert;
  * <p><b>Note: The {@code ConnectionFactory} that this transaction manager
  * operates on needs to return independent {@code Connection}s.</b>
  * The {@code Connection}s may come from a pool (the typical case), but the
- * {@code ConnectionFactory} must not return scoped scoped {@code Connection}s
+ * {@code ConnectionFactory} must not return scoped {@code Connection}s
  * or the like. This transaction manager will associate {@code Connection}
  * with context-bound transactions itself, according to the specified propagation
  * behavior. It assumes that a separate, independent {@code Connection} can
@@ -72,6 +74,11 @@ import org.springframework.util.Assert;
  * synchronizations (if synchronization is generally active), assuming resources
  * operating on the underlying R2DBC {@code Connection}.
  *
+ * <p>Spring's {@code TransactionDefinition} attributes are carried forward to R2DBC drivers
+ * using extensible R2DBC {@link io.r2dbc.spi.TransactionDefinition}. Subclasses may
+ * override {@link #createTransactionDefinition(TransactionDefinition)} to customize
+ * transaction definitions for vendor-specific attributes.
+ *
  * @author Mark Paluch
  * @since 5.3
  * @see ConnectionFactoryUtils#getConnection(ConnectionFactory)
@@ -203,7 +210,8 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager
 			}
 
 			return connectionMono.flatMap(con -> {
-				return prepareTransactionalConnection(con, definition, transaction).then(Mono.from(con.beginTransaction()))
+				return prepareTransactionalConnection(con, definition, transaction)
+						.then(Mono.from(doBegin(definition, con)))
 						.doOnSuccess(v -> {
 							txObject.getConnectionHolder().setTransactionActive(true);
 							Duration timeout = determineTimeout(definition);
@@ -230,6 +238,31 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager
 		}).then();
 	}
 
+	private Publisher<Void> doBegin(TransactionDefinition definition, Connection con) {
+		io.r2dbc.spi.TransactionDefinition transactionDefinition = createTransactionDefinition(definition);
+		if (logger.isDebugEnabled()) {
+			logger.debug("Starting R2DBC transaction on Connection [" + con + "] using [" + transactionDefinition + "]");
+		}
+		return con.beginTransaction(transactionDefinition);
+	}
+
+	/**
+	 * Determine the transaction definition from our {@code TransactionDefinition}.
+	 * Can be overridden to wrap the R2DBC {@code TransactionDefinition} to adjust or
+	 * enhance transaction attributes.
+	 * @param definition the transaction definition
+	 * @return the actual transaction definition to use
+	 * @since 6.0
+	 * @see io.r2dbc.spi.TransactionDefinition
+	 */
+	protected io.r2dbc.spi.TransactionDefinition createTransactionDefinition(TransactionDefinition definition) {
+		// Apply specific isolation level, if any.
+		IsolationLevel isolationLevelToUse = resolveIsolationLevel(definition.getIsolationLevel());
+		return new ExtendedTransactionDefinition(definition.getName(), definition.isReadOnly(),
+				definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ? isolationLevelToUse : null,
+				determineTimeout(definition));
+	}
+
 	/**
 	 * Determine the actual timeout to use for the given definition.
 	 * Will fall back to this manager's default timeout if the
@@ -375,21 +408,6 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager
 					.then();
 		}
 
-		// Apply specific isolation level, if any.
-		IsolationLevel isolationLevelToUse = resolveIsolationLevel(definition.getIsolationLevel());
-		if (isolationLevelToUse != null && definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
-
-			if (logger.isDebugEnabled()) {
-				logger.debug("Changing isolation level of R2DBC Connection [" + con + "] to " + isolationLevelToUse.asSql());
-			}
-			IsolationLevel currentIsolation = con.getTransactionIsolationLevel();
-			if (!currentIsolation.asSql().equalsIgnoreCase(isolationLevelToUse.asSql())) {
-
-				txObject.setPreviousIsolationLevel(currentIsolation);
-				prepare = prepare.then(Mono.from(con.setTransactionIsolationLevel(isolationLevelToUse)));
-			}
-		}
-
 		// Switch to manual commit if necessary. This is very expensive in some R2DBC drivers,
 		// so we don't want to do it unnecessarily (for example if we've explicitly
 		// configured the connection pool to set it already).
@@ -436,6 +454,62 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager
 	}
 
 
+	/**
+	 * Extended R2DBC transaction definition object providing transaction attributes
+	 * to R2DBC drivers when starting a transaction.
+	 */
+	private record ExtendedTransactionDefinition(@Nullable String transactionName,
+												boolean readOnly,
+												@Nullable IsolationLevel isolationLevel,
+												Duration lockWaitTimeout) implements io.r2dbc.spi.TransactionDefinition {
+
+		private ExtendedTransactionDefinition(@Nullable String transactionName, boolean readOnly,
+				@Nullable IsolationLevel isolationLevel, Duration lockWaitTimeout) {
+			this.transactionName = transactionName;
+			this.readOnly = readOnly;
+			this.isolationLevel = isolationLevel;
+			this.lockWaitTimeout = lockWaitTimeout;
+		}
+
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public <T> T getAttribute(Option<T> option) {
+			return (T) doGetValue(option);
+		}
+
+		@Nullable
+		private Object doGetValue(Option<?> option) {
+			if (io.r2dbc.spi.TransactionDefinition.ISOLATION_LEVEL.equals(option)) {
+				return this.isolationLevel;
+			}
+			if (io.r2dbc.spi.TransactionDefinition.NAME.equals(option)) {
+				return this.transactionName;
+			}
+			if (io.r2dbc.spi.TransactionDefinition.READ_ONLY.equals(option)) {
+				return this.readOnly;
+			}
+			if (io.r2dbc.spi.TransactionDefinition.LOCK_WAIT_TIMEOUT.equals(option)
+				&& !this.lockWaitTimeout.isZero()) {
+				return this.lockWaitTimeout;
+			}
+			return null;
+		}
+
+		@Override
+		public String toString() {
+			StringBuilder sb = new StringBuilder();
+			sb.append(getClass().getSimpleName());
+			sb.append(" [transactionName='").append(this.transactionName).append('\'');
+			sb.append(", readOnly=").append(this.readOnly);
+			sb.append(", isolationLevel=").append(this.isolationLevel);
+			sb.append(", lockWaitTimeout=").append(this.lockWaitTimeout);
+			sb.append(']');
+			return sb.toString();
+		}
+	}
+
+
 	/**
 	 * ConnectionFactory transaction object, representing a ConnectionHolder.
 	 * Used as transaction object by R2dbcTransactionManager.

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/BindParameterSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -16,7 +16,7 @@
 
 package org.springframework.r2dbc.core;
 
-import org.springframework.lang.Nullable;
+import io.r2dbc.spi.Parameter;
 
 /**
  * Interface that defines common functionality for objects
@@ -44,24 +44,13 @@ interface BindParameterSource {
 	boolean hasValue(String paramName);
 
 	/**
-	 * Return the parameter value for the requested named parameter.
+	 * Return the parameter for the requested named parameter.
 	 * @param paramName the name of the parameter
-	 * @return the value of the specified parameter (can be {@code null})
+	 * @return the specified parameter
 	 * @throws IllegalArgumentException if there is no value
 	 * for the requested parameter
 	 */
-	@Nullable
+	Parameter getValue(String paramName) throws IllegalArgumentException;
-	Object getValue(String paramName) throws IllegalArgumentException;
-
-	/**
-	 * Determine the type for the specified named parameter.
-	 * @param paramName the name of the parameter
-	 * @return the type of the specified parameter, or
-	 * {@link Object#getClass()} if not known.
-	 */
-	default Class<?> getType(String paramName) {
-		return Object.class;
-	}
 
 	/**
 	 * Return the parameter names of the underlying parameter source.

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/ColumnMapRowMapper.java

@@ -16,7 +16,7 @@
 
 package org.springframework.r2dbc.core;
 
-import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.function.BiFunction;
 
@@ -55,12 +55,12 @@ public class ColumnMapRowMapper implements BiFunction<Row, RowMetadata, Map<Stri
 	@SuppressWarnings("deprecation")  // getColumnNames() is deprecated as of R2DBC 0.9
 	@Override
 	public Map<String, Object> apply(Row row, RowMetadata rowMetadata) {
-		Collection<String> columns = rowMetadata.getColumnNames();
+		List<? extends ColumnMetadata> columns = rowMetadata.getColumnMetadatas();
 		int columnCount = columns.size();
 		Map<String, Object> mapOfColValues = createColumnMap(columnCount);
 		int index = 0;
-		for (String column : columns) {
+		for (ColumnMetadata column : columns) {
-			String key = getColumnKey(column);
+			String key = getColumnKey(column.getName());
 			Object obj = getColumnValue(row, index++);
 			mapOfColValues.put(key, obj);
 		}

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/DatabaseClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -20,12 +20,17 @@ import java.util.Map;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Readable;
+import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
 import io.r2dbc.spi.Statement;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.r2dbc.core.binding.BindMarkersFactory;
@@ -157,9 +162,9 @@ public interface DatabaseClient extends ConnectionAccessor {
 
 		/**
 		 * Bind a non-{@code null} value to a parameter identified by its
-		 * {@code index}. {@code value} can be either a scalar value or {@link Parameter}.
+		 * {@code index}. {@code value} can be either a scalar value or {@link io.r2dbc.spi.Parameter}.
 		 * @param index zero based index to bind the parameter to
-		 * @param value either a scalar value or {@link Parameter}
+		 * @param value either a scalar value or {@link io.r2dbc.spi.Parameter}
 		 */
 		GenericExecuteSpec bind(int index, Object value);
 
@@ -213,14 +218,12 @@ public interface DatabaseClient extends ConnectionAccessor {
 
 		/**
 		 * Configure a result mapping {@link Function function} and enter the execution stage.
-		 * @param mappingFunction a function that maps from {@link Row} to the result type
+		 * @param mappingFunction a function that maps from {@link Readable} to the result type
 		 * @param <R> the result type
 		 * @return a {@link FetchSpec} for configuration what to fetch
+		 * @since 6.0
 		 */
-		default <R> RowsFetchSpec<R> map(Function<Row, R> mappingFunction) {
+		<R> RowsFetchSpec<R> map(Function<? super Readable, R> mappingFunction);
-			Assert.notNull(mappingFunction, "Mapping function must not be null");
-			return map((row, rowMetadata) -> mappingFunction.apply(row));
-		}
 
 		/**
 		 * Configure a result mapping {@link BiFunction function} and enter the execution stage.
@@ -231,6 +234,17 @@ public interface DatabaseClient extends ConnectionAccessor {
 		 */
 		<R> RowsFetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
 
+		/**
+		 * Perform the SQL call and apply {@link BiFunction function} to the {@link  Result}.
+		 * @param mappingFunction a function that maps from {@link Result} into a result publisher
+		 * @param <R> the result type
+		 * @return a {@link Flux} emitting mapped elements
+		 * @since 6.0
+		 * @see Result#filter(Predicate)
+		 * @see Result#flatMap(Function)
+		 */
+		<R> Flux<R> flatMap(Function<Result, Publisher<R>> mappingFunction);
+
 		/**
 		 * Perform the SQL call and retrieve the result by entering the execution stage.
 		 */

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/DefaultDatabaseClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -32,7 +32,10 @@ import java.util.stream.Collectors;
 
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Parameter;
+import io.r2dbc.spi.Parameters;
 import io.r2dbc.spi.R2dbcException;
+import io.r2dbc.spi.Readable;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Row;
 import io.r2dbc.spi.RowMetadata;
@@ -188,11 +191,12 @@ class DefaultDatabaseClient implements DatabaseClient {
 				new CloseSuppressingInvocationHandler(con));
 	}
 
-	private static Mono<Integer> sumRowsUpdated(
+	private static Mono<Long> sumRowsUpdated(
 			Function<Connection, Flux<Result>> resultFunction, Connection it) {
 		return resultFunction.apply(it)
 				.flatMap(Result::getRowsUpdated)
-				.collect(Collectors.summingInt(Integer::intValue));
+				.cast(Number.class)
+				.collect(Collectors.summingLong(Number::longValue));
 	}
 
 	/**
@@ -243,17 +247,21 @@ class DefaultDatabaseClient implements DatabaseClient {
 		}
 
 		@Override
+		@SuppressWarnings("deprecation")
 		public DefaultGenericExecuteSpec bind(int index, Object value) {
 			assertNotPreparedOperation();
 			Assert.notNull(value, () -> String.format(
 					"Value at index %d must not be null. Use bindNull(…) instead.", index));
 
 			Map<Integer, Parameter> byIndex = new LinkedHashMap<>(this.byIndex);
-			if (value instanceof Parameter) {
+			if (value instanceof Parameter p) {
-				byIndex.put(index, (Parameter) value);
+				byIndex.put(index, p);
+			}
+			else if (value instanceof org.springframework.r2dbc.core.Parameter p) {
+				byIndex.put(index, p.hasValue() ? Parameters.in(p.getValue()) : Parameters.in(p.getType()));
 			}
 			else {
-				byIndex.put(index, Parameter.fromOrEmpty(value, value.getClass()));
+				byIndex.put(index, Parameters.in(value));
 			}
 
 			return new DefaultGenericExecuteSpec(byIndex, this.byName, this.sqlSupplier, this.filterFunction);
@@ -264,12 +272,13 @@ class DefaultDatabaseClient implements DatabaseClient {
 			assertNotPreparedOperation();
 
 			Map<Integer, Parameter> byIndex = new LinkedHashMap<>(this.byIndex);
-			byIndex.put(index, Parameter.empty(type));
+			byIndex.put(index, Parameters.in(type));
 
 			return new DefaultGenericExecuteSpec(byIndex, this.byName, this.sqlSupplier, this.filterFunction);
 		}
 
 		@Override
+		@SuppressWarnings("deprecation")
 		public DefaultGenericExecuteSpec bind(String name, Object value) {
 			assertNotPreparedOperation();
 
@@ -278,11 +287,14 @@ class DefaultDatabaseClient implements DatabaseClient {
 					"Value for parameter %s must not be null. Use bindNull(…) instead.", name));
 
 			Map<String, Parameter> byName = new LinkedHashMap<>(this.byName);
-			if (value instanceof Parameter) {
+			if (value instanceof Parameter p) {
-				byName.put(name, (Parameter) value);
+				byName.put(name, p);
+			}
+			else if (value instanceof org.springframework.r2dbc.core.Parameter p) {
+				byName.put(name, p.hasValue() ? Parameters.in(p.getValue()) : Parameters.in(p.getType()));
 			}
 			else {
-				byName.put(name, Parameter.fromOrEmpty(value, value.getClass()));
+				byName.put(name, Parameters.in(value));
 			}
 
 			return new DefaultGenericExecuteSpec(this.byIndex, byName, this.sqlSupplier, this.filterFunction);
@@ -294,7 +306,7 @@ class DefaultDatabaseClient implements DatabaseClient {
 			Assert.hasText(name, "Parameter name must not be null or empty!");
 
 			Map<String, Parameter> byName = new LinkedHashMap<>(this.byName);
-			byName.put(name, Parameter.empty(type));
+			byName.put(name, Parameters.in(type));
 
 			return new DefaultGenericExecuteSpec(this.byIndex, byName, this.sqlSupplier, this.filterFunction);
 		}
@@ -306,15 +318,27 @@ class DefaultDatabaseClient implements DatabaseClient {
 					this.byIndex, this.byName, this.sqlSupplier, this.filterFunction.andThen(filter));
 		}
 
+		@Override
+		public <R> FetchSpec<R> map(Function<? super Readable, R> mappingFunction) {
+			Assert.notNull(mappingFunction, "Mapping function must not be null");
+			return execute(this.sqlSupplier, result -> result.map(mappingFunction));
+		}
+
 		@Override
 		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
 			Assert.notNull(mappingFunction, "Mapping function must not be null");
-			return execute(this.sqlSupplier, mappingFunction);
+			return execute(this.sqlSupplier, result -> result.map(mappingFunction));
+		}
+
+		@Override
+		public <R> Flux<R> flatMap(Function<Result, Publisher<R>> mappingFunction) {
+			Assert.notNull(mappingFunction, "Mapping function must not be null");
+			return flatMap(this.sqlSupplier, mappingFunction);
 		}
 
 		@Override
 		public FetchSpec<Map<String, Object>> fetch() {
-			return execute(this.sqlSupplier, ColumnMapRowMapper.INSTANCE);
+			return execute(this.sqlSupplier, result -> result.map(ColumnMapRowMapper.INSTANCE));
 		}
 
 		@Override
@@ -322,7 +346,7 @@ class DefaultDatabaseClient implements DatabaseClient {
 			return fetch().rowsUpdated().then();
 		}
 
-		private <T> FetchSpec<T> execute(Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
+		private ResultFunction getResultFunction(Supplier<String> sqlSupplier) {
 			String sql = getRequiredSql(sqlSupplier);
 			Function<Connection, Statement> statementFunction = connection -> {
 				if (logger.isDebugEnabled()) {
@@ -376,11 +400,25 @@ class DefaultDatabaseClient implements DatabaseClient {
 				.cast(Result.class).checkpoint("SQL \"" + sql + "\" [DatabaseClient]");
 			};
 
+			return new ResultFunction(resultFunction, sql);
+		}
+
+		private <T> FetchSpec<T> execute(Supplier<String> sqlSupplier, Function<Result, Publisher<T>> resultAdapter) {
+			ResultFunction resultHandler = getResultFunction(sqlSupplier);
+
 			return new DefaultFetchSpec<>(
-					DefaultDatabaseClient.this, sql,
+					DefaultDatabaseClient.this, resultHandler.sql(),
-					new ConnectionFunction<>(sql, resultFunction),
+					new ConnectionFunction<>(resultHandler.sql(), resultHandler.function()),
-					new ConnectionFunction<>(sql, connection -> sumRowsUpdated(resultFunction, connection)),
+					new ConnectionFunction<>(resultHandler.sql(), connection -> sumRowsUpdated(resultHandler.function(), connection)),
-					mappingFunction);
+					resultAdapter);
+		}
+
+		private <T> Flux<T> flatMap(Supplier<String> sqlSupplier, Function<Result, Publisher<T>> mappingFunction) {
+			ResultFunction resultHandler = getResultFunction(sqlSupplier);
+			ConnectionFunction<Flux<T>> connectionFunction = new ConnectionFunction<>(resultHandler.sql(), cx -> resultHandler.function()
+					.apply(cx)
+					.flatMap(mappingFunction));
+			return inConnectionMany(connectionFunction);
 		}
 
 		private MapBindParameterSource retrieveParameters(String sql, List<String> parameterNames,
@@ -424,27 +462,11 @@ class DefaultDatabaseClient implements DatabaseClient {
 		}
 
 		private void bindByName(Statement statement, Map<String, Parameter> byName) {
-			byName.forEach((name, parameter) -> {
+			byName.forEach(statement::bind);
-				Object value = parameter.getValue();
-				if (value != null) {
-					statement.bind(name, value);
-				}
-				else {
-					statement.bindNull(name, parameter.getType());
-				}
-			});
 		}
 
 		private void bindByIndex(Statement statement, Map<Integer, Parameter> byIndex) {
-			byIndex.forEach((i, parameter) -> {
+			byIndex.forEach(statement::bind);
-				Object value = parameter.getValue();
-				if (value != null) {
-					statement.bind(i, value);
-				}
-				else {
-					statement.bindNull(i, parameter.getType());
-				}
-			});
 		}
 
 		private String getRequiredSql(Supplier<String> sqlSupplier) {
@@ -452,6 +474,9 @@ class DefaultDatabaseClient implements DatabaseClient {
 			Assert.state(StringUtils.hasText(sql), "SQL returned by SQL supplier must not be empty!");
 			return sql;
 		}
+
+		record ResultFunction(Function<Connection, Flux<Result>> function, String sql){}
+
 	}
 
 

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/DefaultFetchSpec.java

@@ -16,13 +16,11 @@
 
 package org.springframework.r2dbc.core;
 
-import java.util.function.BiFunction;
 import java.util.function.Function;
 
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.Result;
-import io.r2dbc.spi.Row;
+import org.reactivestreams.Publisher;
-import io.r2dbc.spi.RowMetadata;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -43,21 +41,21 @@ class DefaultFetchSpec<T> implements FetchSpec<T> {
 
 	private final Function<Connection, Flux<Result>> resultFunction;
 
-	private final Function<Connection, Mono<Integer>> updatedRowsFunction;
+	private final Function<Connection, Mono<Long>> updatedRowsFunction;
 
-	private final BiFunction<Row, RowMetadata, T> mappingFunction;
+	private final Function<Result, Publisher<T>> resultAdapter;
 
 
 	DefaultFetchSpec(ConnectionAccessor connectionAccessor, String sql,
 			Function<Connection, Flux<Result>> resultFunction,
-			Function<Connection, Mono<Integer>> updatedRowsFunction,
+			Function<Connection, Mono<Long>> updatedRowsFunction,
-			BiFunction<Row, RowMetadata, T> mappingFunction) {
+			Function<Result, Publisher<T>> resultAdapter) {
 
 		this.sql = sql;
 		this.connectionAccessor = connectionAccessor;
 		this.resultFunction = resultFunction;
 		this.updatedRowsFunction = updatedRowsFunction;
-		this.mappingFunction = mappingFunction;
+		this.resultAdapter = resultAdapter;
 	}
 
 
@@ -86,11 +84,11 @@ class DefaultFetchSpec<T> implements FetchSpec<T> {
 	public Flux<T> all() {
 		return this.connectionAccessor.inConnectionMany(new ConnectionFunction<>(this.sql,
 				connection -> this.resultFunction.apply(connection)
-						.flatMap(result -> result.map(this.mappingFunction))));
+						.flatMap(this.resultAdapter)));
 	}
 
 	@Override
-	public Mono<Integer> rowsUpdated() {
+	public Mono<Long> rowsUpdated() {
 		return this.connectionAccessor.inConnection(this.updatedRowsFunction);
 	}
 

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/MapBindParameterSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -19,6 +19,9 @@ package org.springframework.r2dbc.core;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import io.r2dbc.spi.Parameter;
+import io.r2dbc.spi.Parameters;
+
 import org.springframework.util.Assert;
 
 /**
@@ -61,7 +64,7 @@ class MapBindParameterSource implements BindParameterSource {
 	MapBindParameterSource addValue(String paramName, Object value) {
 		Assert.notNull(paramName, "Parameter name must not be null");
 		Assert.notNull(value, "Value must not be null");
-		this.values.put(paramName, Parameter.fromOrEmpty(value, value.getClass()));
+		this.values.put(paramName, Parameters.in(value));
 		return this;
 	}
 
@@ -72,21 +75,11 @@ class MapBindParameterSource implements BindParameterSource {
 	}
 
 	@Override
-	public Class<?> getType(String paramName) {
+	public Parameter getValue(String paramName) throws IllegalArgumentException {
-		Assert.notNull(paramName, "Parameter name must not be null");
-		Parameter parameter = this.values.get(paramName);
-		if (parameter != null) {
-			return parameter.getType();
-		}
-		return Object.class;
-	}
-
-	@Override
-	public Object getValue(String paramName) throws IllegalArgumentException {
 		if (!hasValue(paramName)) {
 			throw new IllegalArgumentException("No value registered for key '" + paramName + "'");
 		}
-		return this.values.get(paramName).getValue();
+		return this.values.get(paramName);
 	}
 
 	@Override

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/NamedParameterUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -26,6 +26,8 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.TreeMap;
 
+import io.r2dbc.spi.Parameter;
+
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.lang.Nullable;
 import org.springframework.r2dbc.core.binding.BindMarker;
@@ -291,10 +293,10 @@ abstract class NamedParameterUtils {
 			actualSql.append(originalSql, lastIndex, startIndex);
 			NamedParameters.NamedParameter marker = markerHolder.getOrCreate(paramName);
 			if (paramSource.hasValue(paramName)) {
-				Object value = paramSource.getValue(paramName);
+				Parameter parameter = paramSource.getValue(paramName);
-				if (value instanceof Collection) {
+				if (parameter.getValue() instanceof Collection<?> c) {
 
-					Iterator<?> entryIter = ((Collection<?>) value).iterator();
+					Iterator<?> entryIter = c.iterator();
 					int k = 0;
 					int counter = 0;
 					while (entryIter.hasNext()) {
@@ -508,14 +510,14 @@ abstract class NamedParameterUtils {
 		}
 
 		@SuppressWarnings("unchecked")
-		public void bind(BindTarget target, String identifier, Object value) {
+		public void bind(BindTarget target, String identifier, Parameter parameter) {
 			List<BindMarker> bindMarkers = getBindMarkers(identifier);
 			if (bindMarkers == null) {
-				target.bind(identifier, value);
+				target.bind(identifier, parameter);
 				return;
 			}
-			if (value instanceof Collection) {
+			if (parameter.getValue() instanceof Collection) {
-				Collection<Object> collection = (Collection<Object>) value;
+				Collection<Object> collection = (Collection<Object>) parameter.getValue();
 				Iterator<Object> iterator = collection.iterator();
 				Iterator<BindMarker> markers = bindMarkers.iterator();
 				while (iterator.hasNext()) {
@@ -532,7 +534,7 @@ abstract class NamedParameterUtils {
 			}
 			else {
 				for (BindMarker bindMarker : bindMarkers) {
-					bindMarker.bind(target, value);
+					bindMarker.bind(target, parameter);
 				}
 			}
 		}
@@ -544,14 +546,14 @@ abstract class NamedParameterUtils {
 			markers.next().bind(target, valueToBind);
 		}
 
-		public void bindNull(BindTarget target, String identifier, Class<?> valueType) {
+		public void bindNull(BindTarget target, String identifier, Parameter parameter) {
 			List<BindMarker> bindMarkers = getBindMarkers(identifier);
 			if (bindMarkers == null) {
-				target.bindNull(identifier, valueType);
+				target.bind(identifier, parameter);
 				return;
 			}
 			for (BindMarker bindMarker : bindMarkers) {
-				bindMarker.bindNull(target, valueType);
+				bindMarker.bind(target, parameter);
 			}
 		}
 
@@ -576,12 +578,12 @@ abstract class NamedParameterUtils {
 		@Override
 		public void bindTo(BindTarget target) {
 			for (String namedParameter : this.parameterSource.getParameterNames()) {
-				Object value = this.parameterSource.getValue(namedParameter);
+				Parameter parameter = this.parameterSource.getValue(namedParameter);
-				if (value == null) {
+				if (parameter.getValue() == null) {
-					bindNull(target, namedParameter, this.parameterSource.getType(namedParameter));
+					bindNull(target, namedParameter, parameter);
 				}
 				else {
-					bind(target, namedParameter, value);
+					bind(target, namedParameter, parameter);
 				}
 			}
 		}

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/Parameter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -29,7 +29,9 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Paluch
  * @author Juergen Hoeller
  * @since 5.3
+ * @deprecated since 6.0, use {@code io.r2dbc.spi.Parameter} instead.
  */
+@Deprecated(since = "6.0")
 public final class Parameter {
 
 	@Nullable

spring-r2dbc/src/main/java/org/springframework/r2dbc/core/UpdatedRowsFetchSpec.java

@@ -30,6 +30,6 @@ public interface UpdatedRowsFetchSpec {
 	 * Get the number of updated rows.
 	 * @return a Mono emitting the number of updated rows
 	 */
-	Mono<Integer> rowsUpdated();
+	Mono<Long> rowsUpdated();
 
 }

spring-r2dbc/src/main/kotlin/org/springframework/r2dbc/core/DatabaseClientExtensions.kt

@@ -16,6 +16,7 @@
 
 package org.springframework.r2dbc.core
 
+import io.r2dbc.spi.Parameters
 import kotlinx.coroutines.reactor.awaitSingleOrNull
 
 /**
@@ -35,7 +36,7 @@ suspend fun DatabaseClient.GenericExecuteSpec.await() {
  * @author Ibanga Enoobong Ime
  */
 @Suppress("EXTENSION_SHADOWED_BY_MEMBER")
-inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.bind(index: Int, value: T?) = bind(index, Parameter.fromOrEmpty(value, T::class.java))
+inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.bind(index: Int, value: T?) = bind(index, if (value != null) Parameters.`in`(value) else Parameters.`in`(T::class.java))
 
 /**
  * Extension for [DatabaseClient.GenericExecuteSpec.bind] providing a variant leveraging reified type parameters
@@ -44,4 +45,4 @@ inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.bind(index: Int,
  * @author Ibanga Enoobong Ime
  */
 @Suppress("EXTENSION_SHADOWED_BY_MEMBER")
-inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.bind(name: String, value: T?) = bind(name, Parameter.fromOrEmpty(value, T::class.java))
+inline fun <reified T : Any> DatabaseClient.GenericExecuteSpec.bind(name: String, value: T?) = bind(name, if (value != null) Parameters.`in`(value) else Parameters.`in`(T::class.java))

spring-r2dbc/src/main/kotlin/org/springframework/r2dbc/core/UpdatedRowsFetchSpecExtensions.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -23,5 +23,5 @@ import kotlinx.coroutines.reactive.awaitSingle
  *
  * @author Fred Montariol
  */
-suspend fun UpdatedRowsFetchSpec.awaitRowsUpdated(): Int =
+suspend fun UpdatedRowsFetchSpec.awaitRowsUpdated(): Long =
 		rowsUpdated().awaitSingle()

spring-r2dbc/src/test/java/org/springframework/r2dbc/connection/R2dbcTransactionManagerUnitTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.r2dbc.connection;
 
+import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import io.r2dbc.spi.Connection;
@@ -25,6 +26,7 @@ import io.r2dbc.spi.R2dbcBadGrammarException;
 import io.r2dbc.spi.Statement;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -65,7 +67,7 @@ class R2dbcTransactionManagerUnitTests {
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	void before() {
 		when(connectionFactoryMock.create()).thenReturn((Mono) Mono.just(connectionMock));
-		when(connectionMock.beginTransaction()).thenReturn(Mono.empty());
+		when(connectionMock.beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class))).thenReturn(Mono.empty());
 		when(connectionMock.close()).thenReturn(Mono.empty());
 		tm = new R2dbcTransactionManager(connectionFactoryMock);
 	}
@@ -91,7 +93,7 @@ class R2dbcTransactionManagerUnitTests {
 
 		assertThat(commits).hasValue(1);
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).commitTransaction();
 		verify(connectionMock).close();
 		verifyNoMoreInteractions(connectionMock);
@@ -125,13 +127,13 @@ class R2dbcTransactionManagerUnitTests {
 	}
 
 	@Test
-	void appliesIsolationLevel() {
+	void appliesTransactionDefinition() {
 		when(connectionMock.commitTransaction()).thenReturn(Mono.empty());
-		when(connectionMock.getTransactionIsolationLevel()).thenReturn(
-				IsolationLevel.READ_COMMITTED);
-		when(connectionMock.setTransactionIsolationLevel(any())).thenReturn(Mono.empty());
 
 		DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
+		definition.setName("my-transaction");
+		definition.setTimeout(10);
+		definition.setReadOnly(true);
 		definition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE);
 
 		TransactionalOperator operator = TransactionalOperator.create(tm, definition);
@@ -142,12 +144,17 @@ class R2dbcTransactionManagerUnitTests {
 				.expectNextCount(1)
 				.verifyComplete();
 
-		verify(connectionMock).beginTransaction();
+		ArgumentCaptor<io.r2dbc.spi.TransactionDefinition> txCaptor = ArgumentCaptor.forClass(io.r2dbc.spi.TransactionDefinition.class);
-		verify(connectionMock).setTransactionIsolationLevel(
+		verify(connectionMock).beginTransaction(txCaptor.capture());
-				IsolationLevel.READ_COMMITTED);
+		verify(connectionMock, never()).setTransactionIsolationLevel(any());
-		verify(connectionMock).setTransactionIsolationLevel(IsolationLevel.SERIALIZABLE);
 		verify(connectionMock).commitTransaction();
 		verify(connectionMock).close();
+
+		io.r2dbc.spi.TransactionDefinition def = txCaptor.getValue();
+		assertThat(def.getAttribute(io.r2dbc.spi.TransactionDefinition.NAME)).isEqualTo("my-transaction");
+		assertThat(def.getAttribute(io.r2dbc.spi.TransactionDefinition.LOCK_WAIT_TIMEOUT)).isEqualTo(Duration.ofSeconds(10));
+		assertThat(def.getAttribute(io.r2dbc.spi.TransactionDefinition.READ_ONLY)).isEqualTo(true);
+		assertThat(def.getAttribute(io.r2dbc.spi.TransactionDefinition.ISOLATION_LEVEL)).isEqualTo(IsolationLevel.SERIALIZABLE);
 	}
 
 	@Test
@@ -167,7 +174,7 @@ class R2dbcTransactionManagerUnitTests {
 				.expectNextCount(1)
 				.verifyComplete();
 
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock, never()).setTransactionIsolationLevel(any());
 		verify(connectionMock).commitTransaction();
 	}
@@ -187,7 +194,7 @@ class R2dbcTransactionManagerUnitTests {
 				.expectNextCount(1)
 				.verifyComplete();
 
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock, never()).setAutoCommit(anyBoolean());
 		verify(connectionMock).commitTransaction();
 	}
@@ -208,7 +215,7 @@ class R2dbcTransactionManagerUnitTests {
 				.expectNextCount(1)
 				.verifyComplete();
 
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).setAutoCommit(false);
 		verify(connectionMock).setAutoCommit(true);
 		verify(connectionMock).commitTransaction();
@@ -236,7 +243,7 @@ class R2dbcTransactionManagerUnitTests {
 				.verifyComplete();
 
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).createStatement("SET TRANSACTION READ ONLY");
 		verify(connectionMock).commitTransaction();
 		verify(connectionMock).close();
@@ -258,7 +265,7 @@ class R2dbcTransactionManagerUnitTests {
 				.verifyError(IllegalTransactionStateException.class);
 
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).createStatement("foo");
 		verify(connectionMock).commitTransaction();
 		verify(connectionMock).close();
@@ -288,7 +295,7 @@ class R2dbcTransactionManagerUnitTests {
 		assertThat(commits).hasValue(0);
 		assertThat(rollbacks).hasValue(1);
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).rollbackTransaction();
 		verify(connectionMock).close();
 		verifyNoMoreInteractions(connectionMock);
@@ -311,7 +318,7 @@ class R2dbcTransactionManagerUnitTests {
 				.verifyError(IllegalTransactionStateException.class);
 
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).createStatement("foo");
 		verify(connectionMock, never()).commitTransaction();
 		verify(connectionMock).rollbackTransaction();
@@ -341,7 +348,7 @@ class R2dbcTransactionManagerUnitTests {
 				.verifyComplete();
 
 		verify(connectionMock).isAutoCommit();
-		verify(connectionMock).beginTransaction();
+		verify(connectionMock).beginTransaction(any(io.r2dbc.spi.TransactionDefinition.class));
 		verify(connectionMock).rollbackTransaction();
 		verify(connectionMock).close();
 		verifyNoMoreInteractions(connectionMock);

spring-r2dbc/src/test/java/org/springframework/r2dbc/connection/TransactionAwareConnectionFactoryProxyUnitTests.java

@@ -23,6 +23,7 @@ import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.Wrapped;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -130,7 +131,7 @@ class TransactionAwareConnectionFactoryProxyUnitTests {
 
 	@Test
 	void shouldEmitBoundConnection() {
-		when(connectionMock1.beginTransaction()).thenReturn(Mono.empty());
+		when(connectionMock1.beginTransaction(ArgumentMatchers.any())).thenReturn(Mono.empty());
 		when(connectionMock1.commitTransaction()).thenReturn(Mono.empty());
 		when(connectionMock1.close()).thenReturn(Mono.empty());
 

spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractDatabaseClientIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -81,7 +81,7 @@ public abstract class AbstractDatabaseClientIntegrationTests {
 				.bindNull("manual", Integer.class)
 				.fetch().rowsUpdated()
 				.as(StepVerifier::create)
-				.expectNext(1)
+				.expectNext(1L)
 				.verifyComplete();
 
 		databaseClient.sql("SELECT id FROM legoset")
@@ -123,7 +123,7 @@ public abstract class AbstractDatabaseClientIntegrationTests {
 				.bindNull("manual", Integer.class)
 				.fetch().rowsUpdated()
 				.as(StepVerifier::create)
-				.expectNext(1)
+				.expectNext(1L)
 				.verifyComplete();
 
 		databaseClient.sql("SELECT id FROM legoset")

spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java

@@ -109,15 +109,15 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests  {
 
 	@Test
 	public void executeInsertInTransaction() {
-		Flux<Integer> integerFlux = databaseClient
+		Flux<Long> longFlux = databaseClient
 				.sql(getInsertIntoLegosetStatement())
 				.bind(0, 42055)
 				.bind(1, "SCHAUFELRADBAGGER")
 				.bindNull(2, Integer.class)
 				.fetch().rowsUpdated().flux().as(rxtx::transactional);
 
-		integerFlux.as(StepVerifier::create)
+		longFlux.as(StepVerifier::create)
-				.expectNext(1)
+				.expectNext(1L)
 				.verifyComplete();
 
 		databaseClient

spring-r2dbc/src/test/java/org/springframework/r2dbc/core/DefaultDatabaseClientUnitTests.java

@@ -20,6 +20,7 @@ import java.util.Arrays;
 
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Parameters;
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Statement;
 import io.r2dbc.spi.test.MockColumnMetadata;
@@ -137,12 +138,12 @@ class DefaultDatabaseClientUnitTests {
 		databaseClient.sql("SELECT * FROM table WHERE key = $1").bindNull(0,
 				String.class).then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bindNull(0, String.class);
+		verify(statement).bind(0, Parameters.in(String.class));
 
 		databaseClient.sql("SELECT * FROM table WHERE key = $1").bindNull("$1",
 				String.class).then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bindNull("$1", String.class);
+		verify(statement).bind("$1", Parameters.in(String.class));
 	}
 
 	@Test
@@ -155,13 +156,13 @@ class DefaultDatabaseClientUnitTests {
 				Parameter.empty(String.class)).then().as(
 						StepVerifier::create).verifyComplete();
 
-		verify(statement).bindNull(0, String.class);
+		verify(statement).bind(0, Parameters.in(String.class));
 
 		databaseClient.sql("SELECT * FROM table WHERE key = $1").bind("$1",
 				Parameter.empty(String.class)).then().as(
 						StepVerifier::create).verifyComplete();
 
-		verify(statement).bindNull("$1", String.class);
+		verify(statement).bind("$1", Parameters.in(String.class));
 	}
 
 	@Test
@@ -173,7 +174,7 @@ class DefaultDatabaseClientUnitTests {
 		databaseClient.sql("SELECT * FROM table WHERE key = :key").bindNull("key",
 				String.class).then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bindNull(0, String.class);
+		verify(statement).bind(0, Parameters.in(String.class));
 	}
 
 	@Test
@@ -204,12 +205,12 @@ class DefaultDatabaseClientUnitTests {
 		databaseClient.sql("SELECT * FROM table WHERE key = $1").bind(0,
 				Parameter.from("foo")).then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bind(0, "foo");
+		verify(statement).bind(0, Parameters.in("foo"));
 
 		databaseClient.sql("SELECT * FROM table WHERE key = $1").bind("$1",
 				"foo").then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bind("$1", "foo");
+		verify(statement).bind("$1", Parameters.in("foo"));
 	}
 
 	@Test
@@ -221,7 +222,7 @@ class DefaultDatabaseClientUnitTests {
 		databaseClient.sql("SELECT * FROM table WHERE key = :key").bind("key",
 				"foo").then().as(StepVerifier::create).verifyComplete();
 
-		verify(statement).bind(0, "foo");
+		verify(statement).bind(0, Parameters.in("foo"));
 	}
 
 	@Test

spring-r2dbc/src/test/java/org/springframework/r2dbc/core/NamedParameterUtilsUnitTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -22,6 +22,7 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import io.r2dbc.spi.Parameters;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.r2dbc.core.binding.BindMarkersFactory;
@@ -294,7 +295,7 @@ public class NamedParameterUtilsUnitTests {
 
 		PreparedOperation<String> operation = NamedParameterUtils.substituteNamedParameters(
 				sql, factory, new MapBindParameterSource(
-						Collections.singletonMap("id", Parameter.from("foo"))));
+						Collections.singletonMap("id", Parameters.in("foo"))));
 
 		assertThat(operation.toQuery()).isEqualTo(
 				"SELECT * FROM person where name = $0 or lastname = $0");
@@ -307,7 +308,7 @@ public class NamedParameterUtilsUnitTests {
 			@Override
 			public void bind(int index, Object value) {
 				assertThat(index).isEqualTo(0);
-				assertThat(value).isEqualTo("foo");
+				assertThat(value).isEqualTo(Parameters.in("foo"));
 			}
 			@Override
 			public void bindNull(String identifier, Class<?> type) {
@@ -330,7 +331,7 @@ public class NamedParameterUtilsUnitTests {
 
 		PreparedOperation<String> operation = NamedParameterUtils.substituteNamedParameters(
 				sql, factory, new MapBindParameterSource(Collections.singletonMap("ids",
-						Parameter.from(Arrays.asList("foo", "bar", "baz")))));
+						Parameters.in(Arrays.asList("foo", "bar", "baz")))));
 
 		assertThat(operation.toQuery()).isEqualTo(
 				"SELECT * FROM person where name IN ($0, $1, $2) or lastname IN ($0, $1, $2)");
@@ -369,7 +370,7 @@ public class NamedParameterUtilsUnitTests {
 
 		PreparedOperation<String> operation = NamedParameterUtils.substituteNamedParameters(
 				sql, factory, new MapBindParameterSource(
-						Collections.singletonMap("id", Parameter.from("foo"))));
+						Collections.singletonMap("id", Parameters.in("foo"))));
 
 		assertThat(operation.toQuery()).isEqualTo(
 				"SELECT * FROM person where name = ? or lastname = ?");
@@ -395,7 +396,7 @@ public class NamedParameterUtilsUnitTests {
 			}
 		});
 
-		assertThat(bindValues).hasSize(2).containsEntry(0, "foo").containsEntry(1, "foo");
+		assertThat(bindValues).hasSize(2).containsEntry(0, Parameters.in("foo")).containsEntry(1, Parameters.in("foo"));
 	}
 
 	@Test
@@ -406,7 +407,7 @@ public class NamedParameterUtilsUnitTests {
 
 		PreparedOperation<String> operation = NamedParameterUtils.substituteNamedParameters(
 				sql, factory, new MapBindParameterSource(
-						Collections.singletonMap("id", Parameter.empty(String.class))));
+						Collections.singletonMap("id", Parameters.in(String.class))));
 
 		assertThat(operation.toQuery()).isEqualTo(
 				"SELECT * FROM person where name = $0 or lastname = $0");
@@ -418,7 +419,8 @@ public class NamedParameterUtilsUnitTests {
 			}
 			@Override
 			public void bind(int index, Object value) {
-				throw new UnsupportedOperationException();
+				assertThat(index).isEqualTo(0);
+				assertThat(value).isEqualTo(Parameters.in(String.class));
 			}
 			@Override
 			public void bindNull(String identifier, Class<?> type) {
@@ -426,8 +428,7 @@ public class NamedParameterUtilsUnitTests {
 			}
 			@Override
 			public void bindNull(int index, Class<?> type) {
-				assertThat(index).isEqualTo(0);
+				throw new UnsupportedOperationException();
-				assertThat(type).isEqualTo(String.class);
 			}
 		});
 	}

spring-r2dbc/src/test/kotlin/org/springframework/r2dbc/core/DatabaseClientExtensionsTests.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -19,6 +19,7 @@ package org.springframework.r2dbc.core
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import io.r2dbc.spi.Parameters
 import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.Test
 import reactor.core.publisher.Mono
@@ -42,7 +43,7 @@ class DatabaseClientExtensionsTests {
 		}
 
 		verify {
-			spec.bind(0, Parameter.fromOrEmpty("foo", String::class.java))
+			spec.bind(0, Parameters.`in`("foo"))
 		}
 	}
 
@@ -56,7 +57,7 @@ class DatabaseClientExtensionsTests {
 		}
 
 		verify {
-			spec.bind(0, Parameter.empty(String::class.java))
+			spec.bind(0, Parameters.`in`(String::class.java))
 		}
 	}
 
@@ -70,7 +71,7 @@ class DatabaseClientExtensionsTests {
 		}
 
 		verify {
-			spec.bind("field", Parameter.fromOrEmpty("foo", String::class.java))
+			spec.bind("field", Parameters.`in`("foo"))
 		}
 	}
 
@@ -84,7 +85,7 @@ class DatabaseClientExtensionsTests {
 		}
 
 		verify {
-			spec.bind("field", Parameter.empty(String::class.java))
+			spec.bind("field", Parameters.`in`(String::class.java))
 		}
 	}