1. Overview
Iterating through the ResultSet is a common approach to retrieving data from a JDBC query. However, in some cases, we may prefer to work with a stream of records instead.
In this article, we’ll explore a few approaches to processing a ResultSet using the Stream API.
2. Using Spliterators
We’ll start with a pure JDK approach, using Spliterators to create a stream.
First, let’s define a model for our entity:
public record CityRecord(String city, String country) {
}
In our CityRecord we store the information about the city and its country.
Next, let’s create a repository that interacts with the database and returns a stream of our CityRecord instances:
public class JDBCStreamAPIRepository {
private static final String QUERY = "SELECT name, country FROM cities";
private final Logger logger = LoggerFactory.getLogger(JDBCStreamAPIRepository.class);
public Stream<CityRecord> getCitiesStreamUsingSpliterator(Connection connection)
throws SQLException {
PreparedStatement preparedStatement = connection.prepareStatement(QUERY);
connection.setAutoCommit(false);
preparedStatement.setFetchSize(5000);
ResultSet resultSet = preparedStatement.executeQuery();
return StreamSupport.stream(new Spliterators.AbstractSpliterator<CityRecord>(
Long.MAX_VALUE, Spliterator.ORDERED) {
@Override
public boolean tryAdvance(Consumer<? super CityRecord> action) {
try {
if(!resultSet.next()) return false;
action.accept(createCityRecord(resultSet));
return true;
} catch(SQLException ex) {
throw new RuntimeException(ex);
}
}
}, false);
}
private CityRecord createCityRecord(ResultSet resultSet) throws SQLException {
return new CityRecord(resultSet.getString(1), resultSet.getString(2));
}
}
We’ve created a PreparedStatement to retrieve all the items from the cities table, specifying the fetch size to control memory consumption. We used an AbstractSpliterator to generate a stream, where new records are produced as long as the ResultSet has more values. Additionally, we mapped each row to a CityRecord using the createCityRecord method.
Finally, let’s write a test for our repository:
public class JDBCResultSetWithStreamAPIUnitTest {
private static Connection connection = null;
private static final String JDBC_URL = "jdbc:h2:mem:testDatabase";
private static final String USERNAME = "dbUser";
private static final String PASSWORD = "dbPassword";
JDBCStreamAPIRepository jdbcStreamAPIRepository = new JDBCStreamAPIRepository();
@BeforeEach
void setup() throws Exception {
connection = DriverManager.getConnection(JDBC_URL, USERNAME, PASSWORD);
initialDataSetup();
}
private void initialDataSetup() throws SQLException {
Statement statement = connection.createStatement();
String ddlQuery = "CREATE TABLE cities (name VARCHAR(50), country VARCHAR(50))";
statement.execute(ddlQuery);
List<String> sqlQueryList = Arrays.asList(
"INSERT INTO cities VALUES ('London', 'United Kingdom')",
"INSERT INTO cities VALUES ('Sydney', 'Australia')",
"INSERT INTO cities VALUES ('Bucharest', 'Romania')"
);
for (String query : sqlQueryList) {
statement.execute(query);
}
}
@Test
void givenJDBCStreamAPIRepository_whenGetCitiesStreamUsingSpliterator_thenExpectedRecordsShouldBeReturned() throws SQLException {
Stream<CityRecord> cityRecords = jdbcStreamAPIRepository
.getCitiesStreamUsingSpliterator(connection);
List<CityRecord> cities = cityRecords.toList();
assertThat(cities)
.containsExactly(
new CityRecord("London", "United Kingdom"),
new CityRecord("Sydney", "Australia"),
new CityRecord("Bucharest", "Romania"));
}
We establish a connection to the H2 database and, before the test, prepare the cities table with a few entries. Finally, we verify that our repository returns all the expected items from the table as a stream.
3. Using JOOQ
JOOQ is a popular library for working with relational databases. It already provides methods to retrieve a stream of records from a ResultSet.
Let’s begin by adding the necessary dependencies:
<dependency>
<groupId>org.jooq</groupId>
<artifactId>jooq</artifactId>
<version>3.19.11</version>
</dependency>
Next, let’s add a new method to our JDBCStreamAPIRepository:
public Stream<CityRecord> getCitiesStreamUsingJOOQ(Connection connection)
throws SQLException {
PreparedStatement preparedStatement = connection.prepareStatement(QUERY);
connection.setAutoCommit(false);
preparedStatement.setFetchSize(5000);
ResultSet resultSet = preparedStatement.executeQuery();
return DSL.using(connection)
.fetchStream(resultSet)
.map(r -> new CityRecord(r.get("NAME", String.class),
r.get("COUNTRY", String.class)))];
}
We used the fetchStream() method from the ResultQuery class to build a stream of records from the ResultSet. Additionally, we map JOOQ records to CityRecord instances before returning them from the method.
Let’s call our new method and verify that it behaves correctly:
@Test
void givenJDBCStreamAPIRepository_whenGetCitiesStreamUsingJOOQ_thenExpectedRecordsShouldBeReturned() throws SQLException {
Stream<CityRecord> cityRecords = jdbcStreamAPIRepository
.getCitiesStreamUsingJOOQ(connection);
List<CityRecord> cities = cityRecords.toList();
assertThat(cities)
.containsExactly(
new CityRecord("London", "United Kingdom"),
new CityRecord("Sydney", "Australia"),
new CityRecord("Bucharest", "Romania"));
}
As expected, we retrieved all the city records from the database in the stream.
4. Using jdbc-stream
Alternatively, we can create a stream from the ResultSet using a lightweight library called jdbc-stream.
Let’s add its dependencies:
<dependency>
<groupId>com.github.juliomarcopineda</groupId>
<artifactId>jdbc-stream</artifactId>
<version>0.1.1</version>
</dependency>
Now, let’s add a new method to our JDBCStreamAPIRepository:
public Stream<CityRecord> getCitiesStreamUsingJdbcStream(Connection connection)
throws SQLException {
PreparedStatement preparedStatement = connection.prepareStatement(QUERY);
connection.setAutoCommit(false);
preparedStatement.setFetchSize(5000);
ResultSet resultSet = preparedStatement.executeQuery();
return JdbcStream.stream(resultSet)
.map(r -> {
try {
return createCityRecord(resultSet);
} catch (SQLException e) {
throw new RuntimeException(e);
}
});
}
We’ve used JdbcStream to build a stream from our ResultSet. Under the hood, it uses Spliterators and builds the stream with the same logic as our own implementation.
Now, we’ll check how our new repository method works:
@Test
void givenJDBCStreamAPIRepository_whenGetCitiesStreamUsingJdbcStream_thenExpectedRecordsShouldBeReturned() throws SQLException {
Stream<CityRecord> cityRecords = jdbcStreamAPIRepository
.getCitiesStreamUsingJdbcStream(connection);
List<CityRecord> cities = cityRecords.toList();
assertThat(cities)
.containsExactly(
new CityRecord("London", "United Kingdom"),
new CityRecord("Sydney", "Australia"),
new CityRecord("Bucharest", "Romania"));
}
We’ve obtained all the expected items using the jdbc-stream library.
5. Close Resources
When working with JDBC, we must close all the resources we use to avoid connection leaks. The common practice is to use the try-with-resources syntax around Connection, PreparedStatement, and ResultSet. However, this approach isn’t suitable when using streams. If we return a stream from a repository method, all our resources will already be closed, and any operations on the stream won’t be able to access them.
To avoid this issue, we need to close all our resources using the stream’s onClose() method. Additionally, we must ensure that the stream is closed after we finish working with it.
Let’s modify our repository method to include the resource-closing logic:
public Stream<CityRecord> getCitiesStreamUsingJdbcStream(Connection connection)
throws SQLException {
PreparedStatement preparedStatement = connection.prepareStatement(QUERY);
connection.setAutoCommit(false);
preparedStatement.setFetchSize(5000);
ResultSet resultSet = preparedStatement.executeQuery();
return JdbcStream.stream(resultSet)
.map(r -> {
try {
return createCityRecord(resultSet);
} catch (SQLException e) {
throw new RuntimeException(e);
}
})
.onClose(() -> closeResources(connection, resultSet, preparedStatement));
}
private void closeResources(Connection connection, ResultSet resultSet, PreparedStatement preparedStatement) {
try {
resultSet.close();
preparedStatement.close();
connection.close();
logger.info("Resources closed");
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
We’ve added the closeResources() method and attached it to the onClose() stream handler.
Now, let’s modify our client code to ensure that the stream is closed after use:
@Test
void givenJDBCStreamAPIRepository_whenGetCitiesStreamUsingJdbcStream_thenExpectedRecordsShouldBeReturned() throws SQLException {
Stream<CityRecord> cityRecords = jdbcStreamAPIRepository
.getCitiesStreamUsingJdbcStream(connection);
List<CityRecord> cities = cityRecords.toList();
cityRecords.close();
assertThat(cities)
.containsExactly(
new CityRecord("London", "United Kingdom"),
new CityRecord("Sydney", "Australia"),
new CityRecord("Bucharest", "Romania"));
}
Here, we close the stream after all the items have been processed. Additionally, we can observe a log message indicating that all resources have been closed:
[main] INFO com.baeldung.resultset.streams.JDBCStreamAPIRepository -- Resources closed
6. Conclusion
In this article, we explored several options for manipulating ResultSets using the Stream API. This approach is particularly useful when dealing with large datasets that cannot be loaded into memory all at once. Additionally, if we follow a functional style in our applications, a streaming repository will align well with our logic.
The code backing this article is available on GitHub. Once you're
logged in as a Baeldung Pro Member, start learning and coding on the project.
