Introduce RowDocumentExtractor.

Transform the tabular structure into a graph of RowDocument associated with nested documents and lists.

Add container license acceptance for updated container images.

See #1446
See #1450
See #1445
Original pull request: #1572

ci/accept-third-party-license.sh

@@ -4,6 +4,7 @@
   echo "mcr.microsoft.com/mssql/server:2022-CU5-ubuntu-20.04"
   echo "ibmcom/db2:11.5.7.0a"
   echo "harbor-repo.vmware.com/mcr-proxy-cache/mssql/server:2019-CU16-ubuntu-20.04"
+  echo "harbor-repo.vmware.com/mcr-proxy-cache/mssql/server:2022-CU5-ubuntu-20.04"
   echo "harbor-repo.vmware.com/dockerhub-proxy-cache/ibmcom/db2:11.5.7.0a"
 } > spring-data-jdbc/src/test/resources/container-license-acceptance.txt
 
@@ -11,5 +12,6 @@
   echo "mcr.microsoft.com/mssql/server:2022-latest"
   echo "ibmcom/db2:11.5.7.0a"
   echo "harbor-repo.vmware.com/mcr-proxy-cache/mssql/server:2022-latest"
+  echo "harbor-repo.vmware.com/mcr-proxy-cache/mssql/server:2022-CU5-ubuntu-20.04"
   echo "harbor-repo.vmware.com/dockerhub-proxy-cache/ibmcom/db2:11.5.7.0a"
 } > spring-data-r2dbc/src/test/resources/container-license-acceptance.txt

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

@@ -68,8 +68,8 @@ class AggregateResultSetExtractor<T> implements ResultSetExtractor<Iterable<T>>
 	 *          column of the {@link ResultSet} that holds the data for that
 	 *          {@link org.springframework.data.relational.core.mapping.AggregatePath}.
 	 */
-	AggregateResultSetExtractor(RelationalPersistentEntity<T> rootEntity,
-			JdbcConverter converter, PathToColumnMapping pathToColumn) {
+	AggregateResultSetExtractor(RelationalPersistentEntity<T> rootEntity, JdbcConverter converter,
+			PathToColumnMapping pathToColumn) {
 
 		Assert.notNull(rootEntity, "rootEntity must not be null");
 		Assert.notNull(converter, "converter must not be null");
@@ -126,6 +126,8 @@ class AggregateResultSetExtractor<T> implements ResultSetExtractor<Iterable<T>>
 		return instance;
 	}
 
+
+
 	/**
 	 * A {@link Reader} is responsible for reading a single entity or collection of entities from a set of columns
 	 *

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

@@ -0,0 +1,203 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.convert;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.springframework.dao.DataRetrievalFailureException;
+import org.springframework.data.jdbc.core.convert.RowDocumentExtractorSupport.AggregateContext;
+import org.springframework.data.jdbc.core.convert.RowDocumentExtractorSupport.RowDocumentSink;
+import org.springframework.data.jdbc.core.convert.RowDocumentExtractorSupport.TabularResultAdapter;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.domain.RowDocument;
+import org.springframework.jdbc.support.JdbcUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+
+/**
+ * {@link ResultSet}-driven extractor to extract {@link RowDocument documents}.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+class ResultSetRowDocumentExtractor {
+
+	private final RelationalMappingContext context;
+	private final PathToColumnMapping propertyToColumn;
+
+	ResultSetRowDocumentExtractor(RelationalMappingContext context, PathToColumnMapping propertyToColumn) {
+		this.context = context;
+		this.propertyToColumn = propertyToColumn;
+	}
+
+	/**
+	 * Adapter to extract values and column metadata from a {@link ResultSet}.
+	 */
+	enum ResultSetAdapter implements TabularResultAdapter<ResultSet> {
+		INSTANCE;
+
+		@Override
+		public Object getObject(ResultSet row, int index) {
+			try {
+				return JdbcUtils.getResultSetValue(row, index);
+			} catch (SQLException e) {
+				throw new DataRetrievalFailureException("Cannot retrieve column " + index + " from ResultSet", e);
+			}
+		}
+
+		@Override
+		public Map<String, Integer> getColumnMap(ResultSet result) {
+
+			try {
+				ResultSetMetaData metaData = result.getMetaData();
+				Map<String, Integer> columns = new LinkedCaseInsensitiveMap<>(metaData.getColumnCount());
+
+				for (int i = 0; i < metaData.getColumnCount(); i++) {
+					columns.put(metaData.getColumnLabel(i + 1), i + 1);
+				}
+				return columns;
+			} catch (SQLException e) {
+				throw new DataRetrievalFailureException("Cannot retrieve ColumnMap from ResultSet", e);
+			}
+		}
+	}
+
+	/**
+	 * Reads the next {@link RowDocument} from the {@link ResultSet}. The result set can be pristine (i.e.
+	 * {@link ResultSet#isBeforeFirst()}) or pointing already at a row.
+	 *
+	 * @param entity entity defining the document structure.
+	 * @param resultSet the result set to consume.
+	 * @return a {@link RowDocument}.
+	 * @throws SQLException if thrown by the JDBC API.
+	 * @throws IllegalStateException if the {@link ResultSet#isAfterLast() fully consumed}.
+	 */
+	public RowDocument extractNextDocument(Class<?> entity, ResultSet resultSet) throws SQLException {
+		return extractNextDocument(context.getRequiredPersistentEntity(entity), resultSet);
+	}
+
+	/**
+	 * Reads the next {@link RowDocument} from the {@link ResultSet}. The result set can be pristine (i.e.
+	 * {@link ResultSet#isBeforeFirst()}) or pointing already at a row.
+	 *
+	 * @param entity entity defining the document structure.
+	 * @param resultSet the result set to consume.
+	 * @return a {@link RowDocument}.
+	 * @throws SQLException if thrown by the JDBC API.
+	 * @throws IllegalStateException if the {@link ResultSet#isAfterLast() fully consumed}.
+	 */
+	public RowDocument extractNextDocument(RelationalPersistentEntity<?> entity, ResultSet resultSet)
+			throws SQLException {
+
+		Iterator<RowDocument> iterator = iterate(entity, resultSet);
+
+		if (!iterator.hasNext()) {
+			throw new IllegalStateException("ResultSet is fully consumed");
+		}
+
+		return iterator.next();
+	}
+
+	/**
+	 * Obtain a {@link Iterator} to retrieve {@link RowDocument documents} from a {@link ResultSet}.
+	 *
+	 * @param entity the entity to determine the document structure.
+	 * @param rs the input result set.
+	 * @return an iterator to consume the {@link ResultSet} as RowDocuments.
+	 * @throws SQLException if thrown by the JDBC API.
+	 */
+	public Iterator<RowDocument> iterate(RelationalPersistentEntity<?> entity, ResultSet rs) throws SQLException {
+		return new RowDocumentIterator(entity, rs);
+	}
+
+	/**
+	 * Iterator implementation that advances through the {@link ResultSet} and feeds its input into a
+	 * {@link org.springframework.data.jdbc.core.convert.RowDocumentExtractorSupport.RowDocumentSink}.
+	 */
+	private class RowDocumentIterator implements Iterator<RowDocument> {
+
+		private final ResultSet resultSet;
+		private final AggregatePath rootPath;
+		private final RelationalPersistentEntity<?> rootEntity;
+		private final Integer identifierIndex;
+		private final AggregateContext<ResultSet> aggregateContext;
+
+		private final boolean initiallyConsumed;
+		private boolean hasNext;
+
+		RowDocumentIterator(RelationalPersistentEntity<?> entity, ResultSet resultSet) throws SQLException {
+
+			ResultSetAdapter adapter = ResultSetAdapter.INSTANCE;
+
+			if (resultSet.isBeforeFirst()) {
+				hasNext = resultSet.next();
+			}
+
+			this.initiallyConsumed = resultSet.isAfterLast();
+			this.rootPath = context.getAggregatePath(entity);
+			this.rootEntity = entity;
+
+			String idColumn = propertyToColumn.column(rootPath.append(entity.getRequiredIdProperty()));
+			Map<String, Integer> columns = adapter.getColumnMap(resultSet);
+			this.aggregateContext = new AggregateContext<>(adapter, context, propertyToColumn, columns);
+
+			this.resultSet = resultSet;
+			this.identifierIndex = columns.get(idColumn);
+		}
+
+		@Override
+		public boolean hasNext() {
+
+			if (initiallyConsumed) {
+				return false;
+			}
+
+			return hasNext;
+		}
+
+		@Override
+		@Nullable
+		public RowDocument next() {
+
+			RowDocumentSink<ResultSet> reader = new RowDocumentSink<>(aggregateContext, rootEntity, rootPath);
+			Object key = ResultSetAdapter.INSTANCE.getObject(resultSet, identifierIndex);
+
+			try {
+				do {
+					Object nextKey = ResultSetAdapter.INSTANCE.getObject(resultSet, identifierIndex);
+
+					if (nextKey != null && !nextKey.equals(key)) {
+						break;
+					}
+
+					reader.accept(resultSet);
+					hasNext = resultSet.next();
+				} while (hasNext);
+			} catch (SQLException e) {
+				throw new DataRetrievalFailureException("Cannot advance ResultSet", e);
+			}
+
+			return reader.getResult();
+		}
+	}
+
+}

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

@@ -0,0 +1,485 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.convert;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.data.relational.domain.RowDocument;
+import org.springframework.lang.Nullable;
+
+/**
+ * Support class for {@code ResultSet}-driven extractor implementations extracting {@link RowDocument documents} from
+ * flexible input streams.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+abstract class RowDocumentExtractorSupport {
+
+	/**
+	 * Result adapter to obtain values and column metadata.
+	 *
+	 * @param <RS>
+	 */
+	interface TabularResultAdapter<RS> {
+
+		/**
+		 * Read a value from the row input at {@code index}.
+		 *
+		 * @param row the row to read from.
+		 * @param index the column index.
+		 * @return the column value. Can be {@code null}.
+		 */
+		@Nullable
+		Object getObject(RS row, int index);
+
+		/**
+		 * Retrieve a column name to column index map for access by column name.
+		 *
+		 * @param result the result set to read from.
+		 * @return column name to column index map.
+		 */
+		Map<String, Integer> getColumnMap(RS result);
+	}
+
+	/**
+	 * Reading context encapsulating value reading and column handling.
+	 *
+	 * @param <RS>
+	 */
+	protected static class AggregateContext<RS> {
+
+		private final TabularResultAdapter<RS> adapter;
+		final RelationalMappingContext context;
+		final PathToColumnMapping propertyToColumn;
+		private final Map<String, Integer> columnMap;
+
+		protected AggregateContext(TabularResultAdapter<RS> adapter, RelationalMappingContext context,
+				PathToColumnMapping propertyToColumn, Map<String, Integer> columnMap) {
+			this.adapter = adapter;
+			this.context = context;
+			this.propertyToColumn = propertyToColumn;
+			this.columnMap = columnMap;
+		}
+
+		public RelationalPersistentEntity<?> getRequiredPersistentEntity(RelationalPersistentProperty property) {
+			return context.getRequiredPersistentEntity(property);
+		}
+
+		public String getColumnName(AggregatePath path) {
+			return propertyToColumn.column(path);
+		}
+
+		public String getKeyColumnName(AggregatePath path) {
+			return propertyToColumn.keyColumn(path);
+		}
+
+		public boolean containsColumn(String columnName) {
+			return columnMap.containsKey(columnName);
+		}
+
+		@Nullable
+		public Object getObject(RS row, String columnName) {
+			return adapter.getObject(row, columnMap.get(columnName));
+		}
+
+		/**
+		 * Collect the value for {@link AggregatePath} from the {@code row} and add it under {@link SqlIdentifier} to the
+		 * {@link RowDocument}.
+		 */
+		void collectValue(RS row, AggregatePath source, RowDocument document, SqlIdentifier targetName) {
+
+			String columnLabel = propertyToColumn.column(source);
+			Integer index = columnMap.get(columnLabel);
+
+			if (index == null) {
+				return;
+			}
+
+			Object resultSetValue = adapter.getObject(row, index);
+			if (resultSetValue == null) {
+				return;
+			}
+
+			document.put(targetName.getReference(), resultSetValue);
+		}
+
+	}
+
+	/**
+	 * Sink abstraction for tabular result sets that represent an aggregate including all of its nested entities. Reading
+	 * is driven by the results and readers receive a feed of rows to extract the data they are looking for.
+	 * <p>
+	 * Sinks aim to produce a {@link #getResult() result}. Based on the inputs, results may be {@link #hasResult()
+	 * present} or absent.
+	 */
+	protected abstract static class TabularSink<RS> {
+
+		/**
+		 * Accept a row of data and process their results to form potentially a {@link #getResult() result}.
+		 *
+		 * @param row the row to read from.
+		 */
+		abstract void accept(RS row);
+
+		/**
+		 * @return {@code true} if the sink has produced a result.
+		 */
+		abstract boolean hasResult();
+
+		/**
+		 * Retrieve the sink result if present.
+		 *
+		 * @return the sink result.
+		 */
+		@Nullable
+		abstract Object getResult();
+
+		/**
+		 * Reset the sink to prepare for the next result.
+		 */
+		abstract void reset();
+	}
+
+	/**
+	 * Entity-driven sink to form a {@link RowDocument document} representing the underlying entities properties.
+	 *
+	 * @param <RS>
+	 */
+	protected static class RowDocumentSink<RS> extends TabularSink<RS> {
+
+		private final AggregateContext<RS> aggregateContext;
+		private final RelationalPersistentEntity<?> entity;
+		private final AggregatePath basePath;
+
+		private RowDocument result;
+		private final Map<RelationalPersistentProperty, TabularSink<RS>> readerState = new LinkedHashMap<>();
+
+		public RowDocumentSink(AggregateContext<RS> aggregateContext, RelationalPersistentEntity<?> entity,
+				AggregatePath basePath) {
+			this.aggregateContext = aggregateContext;
+			this.entity = entity;
+			this.basePath = basePath;
+		}
+
+		@Override
+		void accept(RS row) {
+
+			boolean first = result == null;
+
+			if (first) {
+				RowDocument document = new RowDocument();
+				readFirstRow(row, document);
+				this.result = document;
+			}
+
+			for (TabularSink<RS> reader : readerState.values()) {
+				reader.accept(row);
+			}
+		}
+
+		/**
+		 * First row contains the root aggregate and all headers for nested collections/maps/entities.
+		 */
+		private void readFirstRow(RS row, RowDocument document) {
+
+			for (RelationalPersistentProperty property : entity) {
+
+				AggregatePath path = basePath.append(property);
+
+				if (property.isQualified()) {
+					readerState.put(property, new ContainerSink<>(aggregateContext, property, path));
+					continue;
+				}
+
+				if (property.isEmbedded()) {
+					collectEmbeddedValues(row, document, property, path);
+					continue;
+				}
+
+				if (property.isEntity()) {
+					readerState.put(property,
+							new RowDocumentSink<>(aggregateContext, aggregateContext.getRequiredPersistentEntity(property), path));
+					continue;
+				}
+
+				aggregateContext.collectValue(row, path, document, property.getColumnName());
+			}
+		}
+
+		/**
+		 * Read properties of embedded from the result set and store them under their column names
+		 */
+		private void collectEmbeddedValues(RS row, RowDocument document, RelationalPersistentProperty property,
+				AggregatePath path) {
+
+			RelationalPersistentEntity<?> embeddedHolder = aggregateContext.getRequiredPersistentEntity(property);
+			for (RelationalPersistentProperty embeddedProperty : embeddedHolder) {
+
+				if (embeddedProperty.isQualified() || embeddedProperty.isCollectionLike() || embeddedProperty.isEntity()) {
+					// hell, no!
+					throw new UnsupportedOperationException("Reading maps and collections into embeddable isn't supported yet");
+				}
+
+				AggregatePath nested = path.append(embeddedProperty);
+				aggregateContext.collectValue(row, nested, document, nested.getColumnInfo().name());
+			}
+		}
+
+		@Override
+		boolean hasResult() {
+
+			if (result == null) {
+				return false;
+			}
+
+			for (TabularSink<?> value : readerState.values()) {
+				if (value.hasResult()) {
+					return true;
+				}
+			}
+
+			return !result.isEmpty();
+		}
+
+		@Override
+		RowDocument getResult() {
+
+			readerState.forEach((property, reader) -> {
+
+				if (reader.hasResult()) {
+					result.put(property.getColumnName().getReference(), reader.getResult());
+				}
+			});
+
+			return result;
+		}
+
+		@Override
+		void reset() {
+			result = null;
+			readerState.clear();
+		}
+	}
+
+	/**
+	 * Sink using a single column to retrieve values from.
+	 *
+	 * @param <RS>
+	 */
+	private static class SingleColumnSink<RS> extends TabularSink<RS> {
+
+		private final AggregateContext<RS> aggregateContext;
+		private final String columnName;
+
+		private @Nullable Object value;
+
+		public SingleColumnSink(AggregateContext<RS> aggregateContext, AggregatePath path) {
+			this.aggregateContext = aggregateContext;
+			this.columnName = path.getColumnInfo().name().getReference();
+		}
+
+		@Override
+		void accept(RS row) {
+
+			if (aggregateContext.containsColumn(columnName)) {
+				value = aggregateContext.getObject(row, columnName);
+			} else {
+				value = null;
+			}
+		}
+
+		@Override
+		boolean hasResult() {
+			return value != null;
+		}
+
+		@Override
+		Object getResult() {
+			return getValue();
+		}
+
+		@Nullable
+		public Object getValue() {
+			return value;
+		}
+
+		@Override
+		void reset() {
+			value = null;
+		}
+	}
+
+	/**
+	 * A sink that aggregates multiple values in a {@link CollectionContainer container} such as List or Map. Inner values
+	 * are determined by the value type while the key type is expected to be a simple type such a string or a number.
+	 *
+	 * @param <RS>
+	 */
+	private static class ContainerSink<RS> extends TabularSink<RS> {
+
+		private final String keyColumn;
+		private final AggregateContext<RS> aggregateContext;
+
+		private Object key;
+		private boolean hasResult = false;
+
+		private final TabularSink<RS> componentReader;
+		private final CollectionContainer container;
+
+		public ContainerSink(AggregateContext<RS> aggregateContext, RelationalPersistentProperty property,
+				AggregatePath path) {
+
+			this.aggregateContext = aggregateContext;
+			this.keyColumn = aggregateContext.getKeyColumnName(path);
+			this.componentReader = property.isEntity()
+					? new RowDocumentSink<>(aggregateContext, aggregateContext.getRequiredPersistentEntity(property), path)
+					: new SingleColumnSink<>(aggregateContext, path);
+
+			this.container = property.isMap() ? new MapContainer() : new ListContainer();
+		}
+
+		@Override
+		void accept(RS row) {
+
+			if (!aggregateContext.containsColumn(keyColumn)) {
+				return;
+			}
+
+			Object key = aggregateContext.getObject(row, keyColumn);
+			if (key == null && !hasResult) {
+				return;
+			}
+
+			boolean keyChange = key != null && !key.equals(this.key);
+
+			if (!hasResult) {
+				hasResult = true;
+			}
+
+			if (keyChange) {
+				if (componentReader.hasResult()) {
+					container.add(this.key, componentReader.getResult());
+					componentReader.reset();
+				}
+			}
+
+			if (key != null) {
+				this.key = key;
+			}
+
+			this.componentReader.accept(row);
+		}
+
+		@Override
+		public boolean hasResult() {
+			return hasResult;
+		}
+
+		@Override
+		public Object getResult() {
+
+			if (componentReader.hasResult()) {
+				container.add(this.key, componentReader.getResult());
+				componentReader.reset();
+			}
+
+			return container.get();
+		}
+
+		@Override
+		void reset() {
+			hasResult = false;
+		}
+	}
+
+	/**
+	 * Base class defining method signatures to add values to a container that can hold multiple values, such as a List or
+	 * Map.
+	 */
+	private abstract static class CollectionContainer {
+
+		/**
+		 * Append the value.
+		 *
+		 * @param key the entry key/index.
+		 * @param value the entry value, can be {@literal null}.
+		 */
+		abstract void add(Object key, @Nullable Object value);
+
+		/**
+		 * Return the container holding the values that were previously added.
+		 *
+		 * @return the container holding the values that were previously added.
+		 */
+		abstract Object get();
+	}
+
+	// TODO: Are we 0 or 1 based?
+	private static class ListContainer extends CollectionContainer {
+
+		private final Map<Number, Object> list = new TreeMap<>(Comparator.comparing(Number::longValue));
+
+		@Override
+		public void add(Object key, @Nullable Object value) {
+			list.put(((Number) key).intValue() - 1, value);
+		}
+
+		@Override
+		public List<Object> get() {
+
+			List<Object> result = new ArrayList<>(list.size());
+
+			// TODO: How do we go about padding? Should we insert null values?
+			list.forEach((index, o) -> {
+
+				while (result.size() < index.intValue()) {
+					result.add(null);
+				}
+
+				result.add(o);
+			});
+
+			return result;
+		}
+	}
+
+	private static class MapContainer extends CollectionContainer {
+
+		private final Map<Object, Object> map = new LinkedHashMap<>();
+
+		@Override
+		public void add(Object key, @Nullable Object value) {
+			map.put(key, value);
+		}
+
+		@Override
+		public Map<Object, Object> get() {
+			return new LinkedHashMap<>(map);
+		}
+	}
+
+}

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

@@ -38,13 +38,14 @@ import org.springframework.data.relational.core.mapping.AggregatePath;
 import org.springframework.data.relational.core.mapping.DefaultNamingStrategy;
 import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
-import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.domain.RowDocument;
 
 /**
  * Unit tests for the {@link AggregateResultSetExtractor}.
  *
  * @author Jens Schauder
+ * @author Mark Paluch
  */
 public class AggregateResultSetExtractorUnitTests {
 
@@ -64,6 +65,7 @@ public class AggregateResultSetExtractorUnitTests {
 	};
 
 	AggregateResultSetExtractor<SimpleEntity> extractor = getExtractor(SimpleEntity.class);
+	ResultSetRowDocumentExtractor documentExtractor = new ResultSetRowDocumentExtractor(context, column);
 
 	@Test // GH-1446
 	void emptyResultSetYieldsEmptyResult() throws SQLException {
@@ -73,12 +75,18 @@ public class AggregateResultSetExtractorUnitTests {
 	}
 
 	@Test // GH-1446
-	void singleSimpleEntityGetsExtractedFromSingleRow() {
+	void singleSimpleEntityGetsExtractedFromSingleRow() throws SQLException {
 
 		ResultSet resultSet = ResultSetTestUtil.mockResultSet(asList(column("id1"), column("name")), //
 				1, "Alfred");
 		assertThat(extractor.extractData(resultSet)).extracting(e -> e.id1, e -> e.name)
 				.containsExactly(tuple(1L, "Alfred"));
+
+		resultSet.close();
+
+		RowDocument document = documentExtractor.extractNextDocument(SimpleEntity.class, resultSet);
+
+		assertThat(document).containsEntry("id1", 1).containsEntry("name", "Alfred");
 	}
 
 	@Test // GH-1446
@@ -136,20 +144,25 @@ public class AggregateResultSetExtractorUnitTests {
 
 	@NotNull
 	private <T> AggregateResultSetExtractor<T> getExtractor(Class<T> type) {
-		return (AggregateResultSetExtractor<T>) new AggregateResultSetExtractor<>(
-				(RelationalPersistentEntity<DummyRecord>) context.getPersistentEntity(type), converter, column);
+		return (AggregateResultSetExtractor<T>) new AggregateResultSetExtractor<>(context.getPersistentEntity(type),
+				converter, column);
 	}
 
 	@Nested
 	class EmbeddedReference {
 		@Test // GH-1446
-		void embeddedGetsExtractedFromSingleRow() {
+		void embeddedGetsExtractedFromSingleRow() throws SQLException {
 
 			ResultSet resultSet = ResultSetTestUtil.mockResultSet(asList(column("id1"), column("embeddedNullable.dummyName")), //
 					1, "Imani");
 
 			assertThat(extractor.extractData(resultSet)).extracting(e -> e.id1, e -> e.embeddedNullable.dummyName)
 					.containsExactly(tuple(1L, "Imani"));
+
+			resultSet.close();
+
+			RowDocument document = documentExtractor.extractNextDocument(SimpleEntity.class, resultSet);
+			assertThat(document).containsEntry("id1", 1).containsEntry("dummy_name", "Imani");
 		}
 
 		@Test // GH-1446
@@ -177,7 +190,7 @@ public class AggregateResultSetExtractorUnitTests {
 	@Nested
 	class ToOneRelationships {
 		@Test // GH-1446
-		void entityReferenceGetsExtractedFromSingleRow() {
+		void entityReferenceGetsExtractedFromSingleRow() throws SQLException {
 
 			ResultSet resultSet = ResultSetTestUtil.mockResultSet(
 					asList(column("id1"), column("dummy"), column("dummy.dummyName")), //
@@ -186,6 +199,13 @@ public class AggregateResultSetExtractorUnitTests {
 			assertThat(extractor.extractData(resultSet)) //
 					.extracting(e -> e.id1, e -> e.dummy.dummyName) //
 					.containsExactly(tuple(1L, "Dummy Alfred"));
+
+			resultSet.close();
+
+			RowDocument document = documentExtractor.extractNextDocument(SimpleEntity.class, resultSet);
+
+			assertThat(document).containsKey("dummy").containsEntry("dummy",
+					new RowDocument().append("dummy_name", "Dummy Alfred"));
 		}
 
 		@Test // GH-1446
@@ -321,35 +341,59 @@ public class AggregateResultSetExtractorUnitTests {
 	class Lists {
 
 		@Test // GH-1446
-		void extractSingleListReference() {
+		void extractSingleListReference() throws SQLException {
 
+			AggregateResultSetExtractor<WithList> extractor = getExtractor(WithList.class);
 			ResultSet resultSet = ResultSetTestUtil.mockResultSet(
-					asList(column("id1"), column("dummyList", KEY), column("dummyList.dummyName")), //
+					asList(column("id", WithList.class), column("people", KEY, WithList.class),
+							column("people.name", WithList.class)), //
 					1, 0, "Dummy Alfred", //
 					1, 1, "Dummy Berta", //
 					1, 2, "Dummy Carl");
 
-			Iterable<SimpleEntity> result = extractor.extractData(resultSet);
+			Iterable<WithList> result = extractor.extractData(resultSet);
 
-			assertThat(result).extracting(e -> e.id1).containsExactly(1L);
-			assertThat(result.iterator().next().dummyList).extracting(d -> d.dummyName) //
+			assertThat(result).extracting(e -> e.id).containsExactly(1L);
+			assertThat(result).flatExtracting(e -> e.people).extracting(e -> e.name) //
 					.containsExactly("Dummy Alfred", "Dummy Berta", "Dummy Carl");
+
+			resultSet.close();
+			RowDocument document = documentExtractor.extractNextDocument(WithList.class, resultSet);
+
+			assertThat(document).containsKey("people");
+			List<RowDocument> dummy_list = document.getList("people");
+			assertThat(dummy_list).hasSize(3).contains(new RowDocument().append("name", "Dummy Alfred"))
+					.contains(new RowDocument().append("name", "Dummy Berta"))
+					.contains(new RowDocument().append("name", "Dummy Carl"));
 		}
 
 		@Test // GH-1446
-		void extractSingleUnorderedListReference() {
+		void extractSingleUnorderedListReference() throws SQLException {
 
+			AggregateResultSetExtractor<WithList> extractor = getExtractor(WithList.class);
 			ResultSet resultSet = ResultSetTestUtil.mockResultSet(
-					asList(column("id1"), column("dummyList", KEY), column("dummyList.dummyName")), //
+					asList(column("id", WithList.class), column("people", KEY, WithList.class),
+							column("people.name", WithList.class)), //
 					1, 0, "Dummy Alfred", //
-					1, 2, "Dummy Carl", 1, 1, "Dummy Berta" //
+					1, 2, "Dummy Carl", //
+					1, 1, "Dummy Berta" //
 			);
 
-			Iterable<SimpleEntity> result = extractor.extractData(resultSet);
+			Iterable<WithList> result = extractor.extractData(resultSet);
 
-			assertThat(result).extracting(e -> e.id1).containsExactly(1L);
-			assertThat(result.iterator().next().dummyList).extracting(d -> d.dummyName) //
+			assertThat(result).extracting(e -> e.id).containsExactly(1L);
+			assertThat(result).flatExtracting(e -> e.people).extracting(e -> e.name) //
 					.containsExactly("Dummy Alfred", "Dummy Berta", "Dummy Carl");
+
+			resultSet.close();
+
+			RowDocument document = documentExtractor.extractNextDocument(WithList.class, resultSet);
+
+			assertThat(document).containsKey("people");
+			List<RowDocument> dummy_list = document.getList("people");
+			assertThat(dummy_list).hasSize(3).contains(new RowDocument().append("name", "Dummy Alfred"))
+					.contains(new RowDocument().append("name", "Dummy Berta"))
+					.contains(new RowDocument().append("name", "Dummy Carl"));
 		}
 
 		@Test // GH-1446
@@ -621,10 +665,18 @@ public class AggregateResultSetExtractorUnitTests {
 		return column(path, NORMAL);
 	}
 
+	private String column(String path, Class<?> entityType) {
+		return column(path, NORMAL, entityType);
+	}
+
 	private String column(String path, ColumnType columnType) {
+		return column(path, columnType, SimpleEntity.class);
+	}
+
+	private String column(String path, ColumnType columnType, Class<?> entityType) {
 
 		PersistentPropertyPath<RelationalPersistentProperty> propertyPath = context.getPersistentPropertyPath(path,
-				SimpleEntity.class);
+				entityType);
 
 		return column(context.getAggregatePath(propertyPath)) + (columnType == KEY ? "_key" : "");
 	}
@@ -637,6 +689,25 @@ public class AggregateResultSetExtractorUnitTests {
 		NORMAL, KEY
 	}
 
+	private static class Person {
+
+		String name;
+	}
+
+	private static class PersonWithId {
+
+		@Id Long id;
+		String name;
+	}
+
+	private static class WithList {
+
+		@Id long id;
+
+		List<Person> people;
+		List<PersonWithId> peopleWithIds;
+	}
+
 	private static class SimpleEntity {
 
 		@Id long id1;

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

@@ -15,21 +15,21 @@
  */
 package org.springframework.data.jdbc.core.convert;
 
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.springframework.util.Assert;
-import org.springframework.util.LinkedCaseInsensitiveMap;
+import static org.mockito.Mockito.*;
 
-import javax.naming.OperationNotSupportedException;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
-import static org.mockito.Mockito.*;
+import javax.naming.OperationNotSupportedException;
+
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.util.Assert;
+import org.springframework.util.LinkedCaseInsensitiveMap;
 
 /**
  * Utility for mocking ResultSets for tests.
@@ -55,7 +55,6 @@ class ResultSetTestUtil {
 		return mock(ResultSet.class, new ResultSetAnswer(columns, result));
 	}
 
-
 	private static List<Map<String, Object>> convertValues(List<String> columns, Object[] values) {
 
 		List<Map<String, Object>> result = new ArrayList<>();
@@ -90,6 +89,10 @@ class ResultSetTestUtil {
 		public Object answer(InvocationOnMock invocation) throws Throwable {
 
 			switch (invocation.getMethod().getName()) {
+				case "close" -> {
+					close();
+					return null;
+				}
 				case "next" -> {
 					return next();
 				}
@@ -111,7 +114,7 @@ class ResultSetTestUtil {
 					return this.toString();
 				}
 				case "findColumn" -> {
-					return isThereAColumnNamed(invocation.getArgument(0));
+					return findColumn(invocation.getArgument(0));
 				}
 				case "getMetaData" -> {
 					return new MockedMetaData();
@@ -120,10 +123,12 @@ class ResultSetTestUtil {
 			}
 		}
 
-		private int isThereAColumnNamed(String name) {
-			throw new UnsupportedOperationException("duh");
-//			Optional<Map<String, Object>> first = values.stream().filter(s -> s.equals(name)).findFirst();
-//			return (first.isPresent()) ? 1 : 0;
+		private int findColumn(String name) {
+			if (names.contains(name)) {
+				return names.indexOf(name) + 1;
+			}
+
+			return -1;
 		}
 
 		private boolean isAfterLast() {
@@ -145,6 +150,12 @@ class ResultSetTestUtil {
 			return rowMap.get(column);
 		}
 
+		private boolean close() {
+
+			index = -1;
+			return index < values.size();
+		}
+
 		private boolean next() {
 
 			index++;

spring-data-relational/src/main/java/org/springframework/data/relational/domain/RowDocument.java

@@ -0,0 +1,255 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.relational.domain;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Represents a tabular structure as document to enable hierarchical traversal of SQL results.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+public class RowDocument implements Map<String, Object> {
+
+	private final Map<String, Object> delegate;
+
+	public RowDocument() {
+		this.delegate = new LinkedCaseInsensitiveMap<>();
+	}
+
+	public RowDocument(Map<String, Object> map) {
+		this.delegate = new LinkedCaseInsensitiveMap<>();
+		this.delegate.putAll(delegate);
+	}
+
+	/**
+	 * Retrieve the value at {@code key} as {@link List}.
+	 *
+	 * @param key
+	 * @return the value or {@literal null}.
+	 * @throws ClassCastException if {@code key} holds a value that is not a {@link List}.
+	 */
+	@Nullable
+	public List getList(String key) {
+
+		Object item = get(key);
+		if (item instanceof List<?> || item == null) {
+			return (List) item;
+		}
+
+		throw new ClassCastException(String.format("Cannot cast element %s be cast to List", item));
+	}
+
+	/**
+	 * Retrieve the value at {@code key} as {@link Map}.
+	 *
+	 * @param key
+	 * @return the value or {@literal null}.
+	 * @throws ClassCastException if {@code key} holds a value that is not a {@link Map}.
+	 */
+	@Nullable
+	public Map<String, Object> getMap(String key) {
+
+		Object item = get(key);
+		if (item instanceof Map || item == null) {
+			return (Map<String, Object>) item;
+		}
+
+		throw new ClassCastException(String.format("Cannot cast element %s be cast to Map", item));
+	}
+
+	/**
+	 * Retrieve the value at {@code key} as {@link RowDocument}.
+	 *
+	 * @param key
+	 * @return the value or {@literal null}.
+	 * @throws ClassCastException if {@code key} holds a value that is not a {@link RowDocument}.
+	 */
+	public RowDocument getDocument(String key) {
+
+		Object item = get(key);
+		if (item instanceof RowDocument || item == null) {
+			return (RowDocument) item;
+		}
+
+		throw new ClassCastException(String.format("Cannot cast element %s be cast to RowDocument", item));
+	}
+
+	@Override
+	public int size() {
+		return delegate.size();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return delegate.isEmpty();
+	}
+
+	@Override
+	public boolean containsKey(Object key) {
+		return delegate.containsKey(key);
+	}
+
+	@Override
+	public boolean containsValue(Object value) {
+		return delegate.containsValue(value);
+	}
+
+	@Override
+	public Object get(Object key) {
+		return delegate.get(key);
+	}
+
+	@Nullable
+	@Override
+	public Object put(String key, Object value) {
+		return delegate.put(key, value);
+	}
+
+	/**
+	 * Appends a new entry (or overwrites an existing value at {@code key}).
+	 *
+	 * @param key
+	 * @param value
+	 * @return
+	 */
+	public RowDocument append(String key, Object value) {
+
+		put(key, value);
+		return this;
+	}
+
+	@Override
+	public Object remove(Object key) {
+		return delegate.remove(key);
+	}
+
+	@Override
+	public void putAll(Map<? extends String, ?> m) {
+		delegate.putAll(m);
+	}
+
+	@Override
+	public void clear() {
+		delegate.clear();
+	}
+
+	@Override
+	public Set<String> keySet() {
+		return delegate.keySet();
+	}
+
+	@Override
+	public Collection<Object> values() {
+		return delegate.values();
+	}
+
+	@Override
+	public Set<Entry<String, Object>> entrySet() {
+		return delegate.entrySet();
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o)
+			return true;
+		if (o == null || getClass() != o.getClass())
+			return false;
+
+		RowDocument that = (RowDocument) o;
+
+		return ObjectUtils.nullSafeEquals(delegate, that.delegate);
+	}
+
+	@Override
+	public int hashCode() {
+		return ObjectUtils.nullSafeHashCode(delegate);
+	}
+
+	@Override
+	public Object getOrDefault(Object key, Object defaultValue) {
+		return delegate.getOrDefault(key, defaultValue);
+	}
+
+	@Override
+	public void forEach(BiConsumer<? super String, ? super Object> action) {
+		delegate.forEach(action);
+	}
+
+	@Override
+	public void replaceAll(BiFunction<? super String, ? super Object, ?> function) {
+		delegate.replaceAll(function);
+	}
+
+	@Nullable
+	@Override
+	public Object putIfAbsent(String key, Object value) {
+		return delegate.putIfAbsent(key, value);
+	}
+
+	@Override
+	public boolean remove(Object key, Object value) {
+		return delegate.remove(key, value);
+	}
+
+	@Override
+	public boolean replace(String key, Object oldValue, Object newValue) {
+		return delegate.replace(key, oldValue, newValue);
+	}
+
+	@Nullable
+	@Override
+	public Object replace(String key, Object value) {
+		return delegate.replace(key, value);
+	}
+
+	@Override
+	public Object computeIfAbsent(String key, Function<? super String, ?> mappingFunction) {
+		return delegate.computeIfAbsent(key, mappingFunction);
+	}
+
+	@Override
+	public Object computeIfPresent(String key, BiFunction<? super String, ? super Object, ?> remappingFunction) {
+		return delegate.computeIfPresent(key, remappingFunction);
+	}
+
+	@Override
+	public Object compute(String key, BiFunction<? super String, ? super Object, ?> remappingFunction) {
+		return delegate.compute(key, remappingFunction);
+	}
+
+	@Override
+	public Object merge(String key, Object value, BiFunction<? super Object, ? super Object, ?> remappingFunction) {
+		return delegate.merge(key, value, remappingFunction);
+	}
+
+	@Override
+	public String toString() {
+		return getClass().getSimpleName() + delegate.toString();
+	}
+
+}