diff --git a/README.adoc b/README.adoc
index c03cddb..1033deb 100644
--- a/README.adoc
+++ b/README.adoc
@@ -8,7 +8,7 @@ The state of R2DBC is incubating to evaluate how an reactive integration could l
 
 == This is NOT an ORM
 
-Spring Data R2DBC does not try to be an ORM. 
+Spring Data R2DBC does not try to be an ORM.
 Instead it is more of a construction kit for your personal reactive relational data access component that you can define the way you like or need it.
 
 == Maven Coordinates
@@ -64,12 +64,13 @@ Mono<Integer> count = databaseClient.execute()
 
 Flux<Map<String, Object>> rows = databaseClient.execute()
 				.sql("SELECT id, name, manual FROM legoset")
-				.fetch().all();
+				.fetch()
+				.all();
 
 Flux<Long> result = db.execute()
 				.sql("SELECT txid_current();")
-				.exchange()
-				.flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all());
+				.map((r, md) -> r.get(0, Long.class))
+				.all();
 ----
 
 === Examples selecting data
@@ -100,14 +101,13 @@ Flux<Integer> ids = databaseClient.insert()
 				.value("id", 42055)
 				.value("name", "Description")
 				.nullValue("manual", Integer.class)
-				.exchange() //
-				.flatMapMany(it -> it.extract((r, m) -> r.get("id", Integer.class)).all())
+				.map((r, m) -> r.get("id", Integer.class)
+				.all();
 
-Flux<Integer> ids = databaseClient.insert()
+Mono<Void> completion = databaseClient.insert()
 				.into(LegoSet.class)
 				.using(legoSet)
-				.exchange()
-				.flatMapMany(it -> it.extract((r, m) -> r.get("id", Integer.class)).all())
+				.then();
 ----
 
 === Examples using reactive repositories
@@ -162,7 +162,7 @@ Flux<Integer> rowsUpdated = databaseClient.inTransaction(db -> {
 [source,java]
 ----
 Flux<Long> txId = databaseClient.execute().sql("SELECT txid_current();").exchange()
-				.flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all());
+				.flatMapMany(it -> it.map((r, md) -> r.get(0, Long.class)).all());
 
 Mono<Void> then = databaseClient.enableTransactionSynchronization(databaseClient.beginTransaction() //
 				.thenMany(txId)) //
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
index 4e49dcb..2ad6c5a 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DatabaseClient.java
@@ -156,17 +156,26 @@ public interface DatabaseClient {
 		 */
 		<R> TypedExecuteSpec<R> as(Class<R> resultType);
 
+		/**
+		 * Configure a result mapping {@link java.util.function.BiFunction function}.
+		 *
+		 * @param mappingFunction must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return
+		 */
+		<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
+
 		/**
 		 * Perform the SQL call and retrieve the result.
 		 */
 		FetchSpec<Map<String, Object>> fetch();
 
 		/**
-		 * Perform the SQL request and return a {@link SqlResult}.
+		 * Perform the SQL call and return a {@link Mono} that completes without result on statement completion.
 		 *
-		 * @return a {@code Mono} for the result
+		 * @return a {@link Mono} ignoring its payload (actively dropping).
 		 */
-		Mono<SqlResult<Map<String, Object>>> exchange();
+		Mono<Void> then();
 	}
 
 	/**
@@ -183,17 +192,26 @@ public interface DatabaseClient {
 		 */
 		<R> TypedExecuteSpec<R> as(Class<R> resultType);
 
+		/**
+		 * Configure a result mapping {@link java.util.function.BiFunction function}.
+		 *
+		 * @param mappingFunction must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return
+		 */
+		<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
+
 		/**
 		 * Perform the SQL call and retrieve the result.
 		 */
 		FetchSpec<T> fetch();
 
 		/**
-		 * Perform the SQL request and return a {@link SqlResult}.
+		 * Perform the SQL call and return a {@link Mono} that completes without result on statement completion.
 		 *
-		 * @return a {@code Mono} for the result
+		 * @return a {@link Mono} ignoring its payload (actively dropping).
 		 */
-		Mono<SqlResult<T>> exchange();
+		Mono<Void> then();
 	}
 
 	/**
@@ -229,7 +247,7 @@ public interface DatabaseClient {
 		 * @param table must not be {@literal null} or empty.
 		 * @return
 		 */
-		GenericInsertSpec into(String table);
+		GenericInsertSpec<Map<String, Object>> into(String table);
 
 		/**
 		 * Specify the target table to insert to using the {@link Class entity class}.
@@ -254,17 +272,19 @@ public interface DatabaseClient {
 		 */
 		<R> TypedSelectSpec<R> as(Class<R> resultType);
 
+		/**
+		 * Configure a result mapping {@link java.util.function.BiFunction function}.
+		 *
+		 * @param mappingFunction must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return
+		 */
+		<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
+
 		/**
 		 * Perform the SQL call and retrieve the result.
 		 */
 		FetchSpec<Map<String, Object>> fetch();
-
-		/**
-		 * Perform the SQL request and return a {@link SqlResult}.
-		 *
-		 * @return a {@code Mono} for the result
-		 */
-		Mono<SqlResult<Map<String, Object>>> exchange();
 	}
 
 	/**
@@ -279,28 +299,21 @@ public interface DatabaseClient {
 		 * @param resultType must not be {@literal null}.
 		 * @param <R> result type.
 		 */
-		<R> TypedSelectSpec<R> as(Class<R> resultType);
+		<R> FetchSpec<R> as(Class<R> resultType);
 
 		/**
-		 * Configure a result mapping {@link java.util.function.Function}.
+		 * Configure a result mapping {@link java.util.function.BiFunction function}.
 		 *
 		 * @param mappingFunction must not be {@literal null}.
 		 * @param <R> result type.
 		 * @return
 		 */
-		<R> TypedSelectSpec<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
+		<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
 
 		/**
 		 * Perform the SQL call and retrieve the result.
 		 */
 		FetchSpec<T> fetch();
-
-		/**
-		 * Perform the SQL request and return a {@link SqlResult}.
-		 *
-		 * @return a {@code Mono} for the result
-		 */
-		Mono<SqlResult<T>> exchange();
 	}
 
 	/**
@@ -332,8 +345,10 @@ public interface DatabaseClient {
 
 	/**
 	 * Contract for specifying {@code INSERT} options leading to the exchange.
+	 *
+	 * @param <T> Result type of tabular insert results.
 	 */
-	interface GenericInsertSpec extends InsertSpec {
+	interface GenericInsertSpec<T> extends InsertSpec<T> {
 
 		/**
 		 * Specify a field and non-{@literal null} value to insert.
@@ -341,7 +356,7 @@ public interface DatabaseClient {
 		 * @param field must not be {@literal null} or empty.
 		 * @param value must not be {@literal null}
 		 */
-		GenericInsertSpec value(String field, Object value);
+		GenericInsertSpec<T> value(String field, Object value);
 
 		/**
 		 * Specify a {@literal null} value to insert.
@@ -349,7 +364,7 @@ public interface DatabaseClient {
 		 * @param field must not be {@literal null} or empty.
 		 * @param type must not be {@literal null}.
 		 */
-		GenericInsertSpec nullValue(String field, Class<?> type);
+		GenericInsertSpec<T> nullValue(String field, Class<?> type);
 	}
 
 	/**
@@ -363,7 +378,7 @@ public interface DatabaseClient {
 		 * @param objectToInsert
 		 * @return
 		 */
-		InsertSpec using(T objectToInsert);
+		InsertSpec<Map<String, Object>> using(T objectToInsert);
 
 		/**
 		 * Use the given {@code tableName} as insert target.
@@ -374,30 +389,43 @@ public interface DatabaseClient {
 		TypedInsertSpec<T> table(String tableName);
 
 		/**
-		 * Insert the given {@link Publisher} to insert one or more objects.
+		 * Insert the given {@link Publisher} to insert one or more objects. Inserts only a single object when calling
+		 * {@link FetchSpec#one()} or {@link FetchSpec#first()}.
 		 *
 		 * @param objectToInsert
 		 * @return
+		 * @see InsertSpec#fetch()
 		 */
-		InsertSpec using(Publisher<T> objectToInsert);
+		InsertSpec<Map<String, Object>> using(Publisher<T> objectToInsert);
 	}
 
 	/**
 	 * Contract for specifying {@code INSERT} options leading to the exchange.
+	 *
+	 * @param <T> Result type of tabular insert results.
 	 */
-	interface InsertSpec {
+	interface InsertSpec<T> {
 
 		/**
-		 * Perform the SQL call.
+		 * Configure a result mapping {@link java.util.function.BiFunction function}.
+		 *
+		 * @param mappingFunction must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return
+		 */
+		<R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
+
+		/**
+		 * Perform the SQL call and retrieve the result.
+		 */
+		FetchSpec<T> fetch();
+
+		/**
+		 * Perform the SQL call and return a {@link Mono} that completes without result on statement completion.
+		 *
+		 * @return a {@link Mono} ignoring its payload (actively dropping).
 		 */
 		Mono<Void> then();
-
-		/**
-		 * Perform the SQL request and return a {@link SqlResult}.
-		 *
-		 * @return a {@code Mono} for the result
-		 */
-		Mono<SqlResult<Map<String, Object>>> exchange();
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
index a6545a9..400e7eb 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
@@ -41,6 +41,7 @@ import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -220,6 +221,15 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, typeToRead);
 	}
 
+	/**
+	 * Customization hook.
+	 */
+	protected <T> DefaultTypedExecuteSpec<T> createTypedExecuteSpec(Map<Integer, SettableValue> byIndex,
+			Map<String, SettableValue> byName, Supplier<String> sqlSupplier,
+			BiFunction<Row, RowMetadata, T> mappingFunction) {
+		return new DefaultTypedExecuteSpec<>(byIndex, byName, sqlSupplier, mappingFunction);
+	}
+
 	/**
 	 * Customization hook.
 	 */
@@ -295,6 +305,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			this.sqlSupplier = sqlSupplier;
 		}
 
+		ExecuteSpecSupport(ExecuteSpecSupport other) {
+
+			this.byIndex = other.byIndex;
+			this.byName = other.byName;
+			this.sqlSupplier = other.sqlSupplier;
+		}
+
 		protected String getSql() {
 
 			String sql = sqlSupplier.get();
@@ -396,14 +413,22 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType);
 		}
 
+		@Override
+		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
+
+			Assert.notNull(mappingFunction, "Mapping function must not be null!");
+
+			return exchange(getSql(), mappingFunction);
+		}
+
 		@Override
 		public FetchSpec<Map<String, Object>> fetch() {
 			return exchange(getSql(), ColumnMapRowMapper.INSTANCE);
 		}
 
 		@Override
-		public Mono<SqlResult<Map<String, Object>>> exchange() {
-			return Mono.just(exchange(getSql(), ColumnMapRowMapper.INSTANCE));
+		public Mono<Void> then() {
+			return fetch().rowsUpdated().then();
 		}
 
 		@Override
@@ -456,6 +481,14 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead);
 		}
 
+		DefaultTypedExecuteSpec(Map<Integer, SettableValue> byIndex, Map<String, SettableValue> byName,
+				Supplier<String> sqlSupplier, BiFunction<Row, RowMetadata, T> mappingFunction) {
+
+			super(byIndex, byName, sqlSupplier);
+			this.typeToRead = null;
+			this.mappingFunction = mappingFunction;
+		}
+
 		@Override
 		public <R> TypedExecuteSpec<R> as(Class<R> resultType) {
 
@@ -464,14 +497,22 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return createTypedExecuteSpec(this.byIndex, this.byName, this.sqlSupplier, resultType);
 		}
 
+		@Override
+		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
+
+			Assert.notNull(mappingFunction, "Mapping function must not be null!");
+
+			return exchange(getSql(), mappingFunction);
+		}
+
 		@Override
 		public FetchSpec<T> fetch() {
 			return exchange(getSql(), mappingFunction);
 		}
 
 		@Override
-		public Mono<SqlResult<T>> exchange() {
-			return Mono.just(exchange(getSql(), mappingFunction));
+		public Mono<Void> then() {
+			return fetch().rowsUpdated().then();
 		}
 
 		@Override
@@ -603,10 +644,21 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 		@Override
 		public <R> TypedSelectSpec<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "Result type must not be null!");
+
 			return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, resultType,
 					dataAccessStrategy.getRowMapper(resultType));
 		}
 
+		@Override
+		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
+
+			Assert.notNull(mappingFunction, "Mapping function must not be null!");
+
+			return exchange(mappingFunction);
+		}
+
 		@Override
 		public DefaultGenericSelectSpec project(String... selectedFields) {
 			return (DefaultGenericSelectSpec) super.project(selectedFields);
@@ -627,11 +679,6 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return exchange(ColumnMapRowMapper.INSTANCE);
 		}
 
-		@Override
-		public Mono<SqlResult<Map<String, Object>>> exchange() {
-			return Mono.just(exchange(ColumnMapRowMapper.INSTANCE));
-		}
-
 		private <R> SqlResult<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
 
 			Set<String> columns;
@@ -660,7 +707,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	@SuppressWarnings("unchecked")
 	private class DefaultTypedSelectSpec<T> extends DefaultSelectSpecSupport implements TypedSelectSpec<T> {
 
-		private final Class<?> typeToRead;
+		private final @Nullable Class<T> typeToRead;
 		private final BiFunction<Row, RowMetadata, T> mappingFunction;
 
 		DefaultTypedSelectSpec(Class<T> typeToRead) {
@@ -671,7 +718,12 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			this.mappingFunction = dataAccessStrategy.getRowMapper(typeToRead);
 		}
 
-		DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page, Class<?> typeToRead,
+		DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page,
+				BiFunction<Row, RowMetadata, T> mappingFunction) {
+			this(table, projectedFields, sort, page, null, mappingFunction);
+		}
+
+		DefaultTypedSelectSpec(String table, List<String> projectedFields, Sort sort, Pageable page, Class<T> typeToRead,
 				BiFunction<Row, RowMetadata, T> mappingFunction) {
 			super(table, projectedFields, sort, page);
 			this.typeToRead = typeToRead;
@@ -679,20 +731,19 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public <R> TypedSelectSpec<R> as(Class<R> resultType) {
+		public <R> FetchSpec<R> as(Class<R> resultType) {
 
 			Assert.notNull(resultType, "Result type must not be null!");
 
-			return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead,
-					dataAccessStrategy.getRowMapper(resultType));
+			return exchange(dataAccessStrategy.getRowMapper(resultType));
 		}
 
 		@Override
-		public <R> TypedSelectSpec<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
+		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
 
 			Assert.notNull(mappingFunction, "Mapping function must not be null!");
 
-			return new DefaultTypedSelectSpec<>(table, projectedFields, sort, page, typeToRead, mappingFunction);
+			return exchange(mappingFunction);
 		}
 
 		@Override
@@ -711,15 +762,10 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 		}
 
 		@Override
-		public FetchSpec<T> fetch() {
+		public SqlResult<T> fetch() {
 			return exchange(mappingFunction);
 		}
 
-		@Override
-		public Mono<SqlResult<T>> exchange() {
-			return Mono.just(exchange(mappingFunction));
-		}
-
 		private <R> SqlResult<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
 
 			List<String> columns;
@@ -749,13 +795,13 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	class DefaultInsertIntoSpec implements InsertIntoSpec {
 
 		@Override
-		public GenericInsertSpec into(String table) {
-			return new DefaultGenericInsertSpec(table, Collections.emptyMap());
+		public GenericInsertSpec<Map<String, Object>> into(String table) {
+			return new DefaultGenericInsertSpec<>(table, Collections.emptyMap(), ColumnMapRowMapper.INSTANCE);
 		}
 
 		@Override
 		public <T> TypedInsertSpec<T> into(Class<T> table) {
-			return new DefaultTypedInsertSpec<>(table);
+			return new DefaultTypedInsertSpec<>(table, ColumnMapRowMapper.INSTANCE);
 		}
 	}
 
@@ -763,10 +809,11 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	 * Default implementation of {@link DatabaseClient.GenericInsertSpec}.
 	 */
 	@RequiredArgsConstructor
-	class DefaultGenericInsertSpec implements GenericInsertSpec {
+	class DefaultGenericInsertSpec<T> implements GenericInsertSpec<T> {
 
 		private final String table;
 		private final Map<String, SettableValue> byName;
+		private final BiFunction<Row, RowMetadata, T> mappingFunction;
 
 		@Override
 		public GenericInsertSpec value(String field, Object value) {
@@ -776,7 +823,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
 			byName.put(field, new SettableValue(field, value, null));
 
-			return new DefaultGenericInsertSpec(this.table, byName);
+			return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction);
 		}
 
 		@Override
@@ -787,20 +834,28 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			Map<String, SettableValue> byName = new LinkedHashMap<>(this.byName);
 			byName.put(field, new SettableValue(field, null, type));
 
-			return new DefaultGenericInsertSpec(this.table, byName);
+			return new DefaultGenericInsertSpec<>(this.table, byName, this.mappingFunction);
+		}
+
+		@Override
+		public <R> FetchSpec<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
+
+			Assert.notNull(mappingFunction, "Mapping function must not be null!");
+
+			return exchange(mappingFunction);
+		}
+
+		@Override
+		public FetchSpec<T> fetch() {
+			return exchange(this.mappingFunction);
 		}
 
 		@Override
 		public Mono<Void> then() {
-			return exchange((row, md) -> row).all().then();
+			return fetch().rowsUpdated().then();
 		}
 
-		@Override
-		public Mono<SqlResult<Map<String, Object>>> exchange() {
-			return Mono.just(exchange(ColumnMapRowMapper.INSTANCE));
-		}
-
-		private <T> SqlResult<T> exchange(BiFunction<Row, RowMetadata, T> mappingFunction) {
+		private <R> SqlResult<R> exchange(BiFunction<Row, RowMetadata, R> mappingFunction) {
 
 			if (byName.isEmpty()) {
 				throw new IllegalStateException("Insert fields is empty!");
@@ -809,7 +864,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			BindableOperation bindableInsert = dataAccessStrategy.insertAndReturnGeneratedKeys(table, byName.keySet());
 
 			String sql = bindableInsert.toQuery();
-			Function<Connection, Statement> insertFunction = it -> {
+			Function<Connection, Statement<?>> insertFunction = it -> {
 
 				if (logger.isDebugEnabled()) {
 					logger.debug("Executing SQL statement [" + sql + "]");
@@ -837,17 +892,19 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 	 * Default implementation of {@link DatabaseClient.TypedInsertSpec}.
 	 */
 	@RequiredArgsConstructor
-	class DefaultTypedInsertSpec<T> implements TypedInsertSpec<T>, InsertSpec {
+	class DefaultTypedInsertSpec<T, R> implements TypedInsertSpec<T>, InsertSpec<R> {
 
 		private final Class<?> typeToInsert;
 		private final String table;
 		private final Publisher<T> objectToInsert;
+		private final BiFunction<Row, RowMetadata, R> mappingFunction;
 
-		DefaultTypedInsertSpec(Class<?> typeToInsert) {
+		DefaultTypedInsertSpec(Class<?> typeToInsert, BiFunction<Row, RowMetadata, R> mappingFunction) {
 
 			this.typeToInsert = typeToInsert;
 			this.table = dataAccessStrategy.getTableName(typeToInsert);
 			this.objectToInsert = Mono.empty();
+			this.mappingFunction = mappingFunction;
 		}
 
 		@Override
@@ -855,7 +912,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.hasText(tableName, "Table name must not be null or empty!");
 
-			return new DefaultTypedInsertSpec<>(typeToInsert, tableName, objectToInsert);
+			return new DefaultTypedInsertSpec<>(typeToInsert, tableName, objectToInsert, this.mappingFunction);
 		}
 
 		@Override
@@ -863,7 +920,7 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.notNull(objectToInsert, "Object to insert must not be null!");
 
-			return new DefaultTypedInsertSpec<>(typeToInsert, table, Mono.just(objectToInsert));
+			return new DefaultTypedInsertSpec<>(typeToInsert, table, Mono.just(objectToInsert), this.mappingFunction);
 		}
 
 		@Override
@@ -871,7 +928,20 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 
 			Assert.notNull(objectToInsert, "Publisher to insert must not be null!");
 
-			return new DefaultTypedInsertSpec<>(typeToInsert, table, objectToInsert);
+			return new DefaultTypedInsertSpec<>(typeToInsert, table, objectToInsert, this.mappingFunction);
+		}
+
+		@Override
+		public <MR> FetchSpec<MR> map(BiFunction<Row, RowMetadata, MR> mappingFunction) {
+
+			Assert.notNull(mappingFunction, "Mapping function must not be null!");
+
+			return exchange(mappingFunction);
+		}
+
+		@Override
+		public FetchSpec<R> fetch() {
+			return exchange(this.mappingFunction);
 		}
 
 		@Override
@@ -879,12 +949,33 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return Mono.from(objectToInsert).flatMapMany(toInsert -> exchange(toInsert, (row, md) -> row).all()).then();
 		}
 
-		@Override
-		public Mono<SqlResult<Map<String, Object>>> exchange() {
-			return Mono.from(objectToInsert).map(toInsert -> exchange(toInsert, ColumnMapRowMapper.INSTANCE));
+		private <MR> FetchSpec<MR> exchange(BiFunction<Row, RowMetadata, MR> mappingFunction) {
+
+			return new FetchSpec<MR>() {
+				@Override
+				public Mono<MR> one() {
+					return Mono.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).one());
+				}
+
+				@Override
+				public Mono<MR> first() {
+					return Mono.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).first());
+				}
+
+				@Override
+				public Flux<MR> all() {
+					return Flux.from(objectToInsert).flatMap(toInsert -> exchange(toInsert, mappingFunction).all());
+				}
+
+				@Override
+				public Mono<Integer> rowsUpdated() {
+					return Mono.from(objectToInsert).flatMapMany(toInsert -> exchange(toInsert, mappingFunction).rowsUpdated())
+							.collect(Collectors.summingInt(Integer::intValue));
+				}
+			};
 		}
 
-		private <R> SqlResult<R> exchange(Object toInsert, BiFunction<Row, RowMetadata, R> mappingFunction) {
+		private <MR> SqlResult<MR> exchange(Object toInsert, BiFunction<Row, RowMetadata, MR> mappingFunction) {
 
 			List<SettableValue> insertValues = dataAccessStrategy.getValuesToInsert(toInsert);
 			Set<String> columns = new LinkedHashSet<>();
@@ -919,7 +1010,8 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
 			return new DefaultSqlResult<>(DefaultDatabaseClient.this, //
 					sql, //
 					resultFunction, //
-					it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated).next(), //
+					it -> resultFunction.apply(it).flatMap(Result::getRowsUpdated)
+							.collect(Collectors.summingInt(Integer::intValue)), //
 					mappingFunction);
 		}
 	}
diff --git a/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java b/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
index 378b473..44f1165 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/DefaultSqlResult.java
@@ -34,6 +34,33 @@ import org.springframework.jdbc.core.SqlProvider;
  */
 class DefaultSqlResult<T> implements SqlResult<T> {
 
+	private final static SqlResult<?> EMPTY = new SqlResult<Object>() {
+		@Override
+		public <R> SqlResult<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
+			return DefaultSqlResult.empty();
+		}
+
+		@Override
+		public Mono<Object> one() {
+			return Mono.empty();
+		}
+
+		@Override
+		public Mono<Object> first() {
+			return Mono.empty();
+		}
+
+		@Override
+		public Flux<Object> all() {
+			return Flux.empty();
+		}
+
+		@Override
+		public Mono<Integer> rowsUpdated() {
+			return Mono.empty();
+		}
+	};
+
 	private final ConnectionAccessor connectionAccessor;
 	private final String sql;
 	private final Function<Connection, Flux<Result>> resultFunction;
@@ -71,11 +98,22 @@ class DefaultSqlResult<T> implements SqlResult<T> {
 		});
 	}
 
+	/**
+	 * Returns an empty {@link SqlResult}.
+	 *
+	 * @param <R>
+	 * @return
+	 */
+	@SuppressWarnings("unchecked")
+	public static <R> SqlResult<R> empty() {
+		return (SqlResult<R>) EMPTY;
+	}
+
 	/* (non-Javadoc)
-	 * @see org.springframework.data.jdbc.core.function.SqlResult#extract(java.util.function.BiFunction)
+	 * @see org.springframework.data.jdbc.core.function.SqlResult#map(java.util.function.BiFunction)
 	 */
 	@Override
-	public <R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction) {
+	public <R> SqlResult<R> map(BiFunction<Row, RowMetadata, R> mappingFunction) {
 		return new DefaultSqlResult<>(connectionAccessor, sql, resultFunction, updatedRowsFunction, mappingFunction);
 	}
 
diff --git a/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java b/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
index eabf54a..a0b3a0c 100644
--- a/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
+++ b/src/main/java/org/springframework/data/r2dbc/function/SqlResult.java
@@ -34,5 +34,5 @@ public interface SqlResult<T> extends FetchSpec<T> {
 	 * @param <R>
 	 * @return a new {@link SqlResult} with {@link BiFunction mapping function} applied.
 	 */
-	<R> SqlResult<R> extract(BiFunction<Row, RowMetadata, R> mappingFunction);
+	<R> SqlResult<R> map(BiFunction<Row, RowMetadata, R> mappingFunction);
 }
diff --git a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
index 2383034..fb1458b 100644
--- a/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
+++ b/src/main/java/org/springframework/data/r2dbc/repository/support/SimpleR2dbcRepository.java
@@ -32,7 +32,6 @@ import org.springframework.data.r2dbc.function.BindIdOperation;
 import org.springframework.data.r2dbc.function.BindableOperation;
 import org.springframework.data.r2dbc.function.DatabaseClient;
 import org.springframework.data.r2dbc.function.DatabaseClient.GenericExecuteSpec;
-import org.springframework.data.r2dbc.function.FetchSpec;
 import org.springframework.data.r2dbc.function.ReactiveDataAccessStrategy;
 import org.springframework.data.r2dbc.function.convert.MappingR2dbcConverter;
 import org.springframework.data.r2dbc.function.convert.SettableValue;
@@ -66,8 +65,8 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 			return databaseClient.insert() //
 					.into(entity.getJavaType()) //
 					.using(objectToSave) //
-					.exchange() //
-					.flatMap(it -> it.extract(converter.populateIdIfNecessary(objectToSave)).one());
+					.map(converter.populateIdIfNecessary(objectToSave)) //
+					.one();
 		}
 
 		Object id = entity.getRequiredId(objectToSave);
@@ -83,8 +82,7 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		update.bindId(wrapper, id);
 
 		return wrapper.getBoundOperation().as(entity.getJavaType()) //
-				.exchange() //
-				.flatMap(FetchSpec::rowsUpdated) //
+				.then() //
 				.thenReturn(objectToSave);
 	}
 
@@ -156,8 +154,9 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 		select.bindId(wrapper, id);
 
 		return wrapper.getBoundOperation().as(entity.getJavaType()) //
-				.exchange() //
-				.flatMap(it -> it.extract((r, md) -> r).first()).hasElement();
+				.map((r, md) -> r) //
+				.first() //
+				.hasElement();
 	}
 
 	/* (non-Javadoc)
@@ -220,8 +219,8 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 
 		return databaseClient.execute()
 				.sql(String.format("SELECT COUNT(%s) FROM %s", getIdColumnName(), entity.getTableName())) //
-				.exchange() //
-				.flatMap(it -> it.extract((r, md) -> r.get(0, Long.class)).first()) //
+				.map((r, md) -> r.get(0, Long.class)) //
+				.first() //
 				.defaultIfEmpty(0L);
 
 	}
@@ -312,7 +311,6 @@ public class SimpleR2dbcRepository<T, ID> implements ReactiveCrudRepository<T, I
 	public Mono<Void> deleteAll() {
 
 		return databaseClient.execute().sql(String.format("DELETE FROM %s", entity.getTableName())) //
-				.exchange() //
 				.then();
 	}
 
diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
index 3ebf5bc..bfe27ad 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractDatabaseClientIntegrationTests.java
@@ -20,6 +20,7 @@ import static org.springframework.data.domain.Sort.Order.*;
 
 import io.r2dbc.spi.ConnectionFactory;
 import lombok.Data;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Hooks;
 import reactor.test.StepVerifier;
 
@@ -108,6 +109,9 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 				.expectNext(1) //
 				.verifyComplete();
 
+		Flux<LegoSet> rows = databaseClient.select().from("legoset").orderBy(Sort.by(desc("id"))).as(LegoSet.class).fetch()
+				.all();
+
 		assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
 	}
 
@@ -160,8 +164,8 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 				.value("id", 42055) //
 				.value("name", "SCHAUFELRADBAGGER") //
 				.nullValue("manual", Integer.class) //
-				.exchange() //
-				.flatMapMany(FetchSpec::rowsUpdated) //
+				.fetch() //
+				.rowsUpdated() //
 				.as(StepVerifier::create) //
 				.expectNext(1).verifyComplete();
 
@@ -195,16 +199,18 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
 
 		databaseClient.insert().into(LegoSet.class)//
-				.using(legoSet).exchange() //
-				.flatMapMany(FetchSpec::rowsUpdated) //
+				.using(legoSet) //
+				.fetch() //
+				.rowsUpdated() //
 				.as(StepVerifier::create) //
-				.expectNext(1).verifyComplete();
+				.expectNext(1) //
+				.verifyComplete();
 
 		assertThat(jdbc.queryForMap("SELECT id, name, manual FROM legoset")).containsEntry("id", 42055);
 	}
 
 	@Test
-	public void select() {
+	public void selectAsMap() {
 
 		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
 
@@ -213,7 +219,8 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 		databaseClient.select().from(LegoSet.class) //
 				.project("id", "name", "manual") //
 				.orderBy(Sort.by("id")) //
-				.fetch().all() //
+				.fetch() //
+				.all() //
 				.as(StepVerifier::create) //
 				.assertNext(actual -> {
 					assertThat(actual.getId()).isEqualTo(42055);
@@ -222,6 +229,23 @@ public abstract class AbstractDatabaseClientIntegrationTests extends R2dbcIntegr
 				}).verifyComplete();
 	}
 
+	@Test
+	public void selectExtracting() {
+
+		jdbc.execute("INSERT INTO legoset (id, name, manual) VALUES(42055, 'SCHAUFELRADBAGGER', 12)");
+
+		DatabaseClient databaseClient = DatabaseClient.create(connectionFactory);
+
+		databaseClient.select().from("legoset") //
+				.project("id", "name", "manual") //
+				.orderBy(Sort.by("id")) //
+				.map((r, md) -> r.get("id", Integer.class)) //
+				.all() //
+				.as(StepVerifier::create) //
+				.expectNext(42055) //
+				.verifyComplete();
+	}
+
 	@Test
 	public void selectOrderByIdDesc() {
 
diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
index c50b305..a57c5f8 100644
--- a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
+++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java
@@ -148,8 +148,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
 		Queue<Long> transactionIds = new ArrayBlockingQueue<>(5);
 		TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
 
-		Flux<Long> txId = databaseClient.execute().sql(getCurrentTransactionIdStatement()).exchange()
-				.flatMapMany(it -> it.extract((r, md) -> r.get(0, Long.class)).all());
+		Flux<Long> txId = databaseClient.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0, Long.class)).all();
 
 		Mono<Void> then = databaseClient.enableTransactionSynchronization(databaseClient.beginTransaction() //
 				.thenMany(txId.concatWith(txId).doOnNext(transactionIds::add)) //
@@ -207,8 +206,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
 
 		Flux<Object> transactionIds = databaseClient.inTransaction(db -> {
 
-			Flux<Object> txId = db.execute().sql(getCurrentTransactionIdStatement()).exchange()
-					.flatMapMany(it -> it.extract((r, md) -> r.get(0)).all());
+			Flux<Object> txId = db.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0)).all();
 			return txId.concatWith(txId);
 		});