Sketch: JDBC ResultSet -> Java Stream

LazyJdbcConnection.java

public class LazyJdbcConnection extends JdbcConnection {
    public LazyJdbcConnection(JdbcTemplate jdbcTemplate) {
        super(jdbcTemplate);
    }

    public <T> T executeQueryLazily(final String sqlQuery, final LazyResultSetExtractor<T> resultSetExtractor) throws SQLException {
        log.debug("Executing query: {}", sqlQuery);

        if (jdbcTemplate.getDataSource() == null) {
            throw new SQLException("Data source not available via JdbcTemplate");
        }
        var connection = jdbcTemplate.getDataSource().getConnection();

        var pstmt = connection.prepareStatement(sqlQuery);
        var rs = pstmt.executeQuery();

        resultSetExtractor.onFinish(() -> {
            JdbcUtils.closeResultSet(rs);
            JdbcUtils.closeStatement(pstmt);
            JdbcUtils.closeConnection(connection);
        });
        return resultSetExtractor.extractData(rs);
    }
}

LazyResultSetExtractor.java

/**
 * Promises to run all actions registered via {@link LazyResultSetExtractor#onFinish(Runnable)},
 * in the order in which they were registered,
 * after it finished consuming the {@code ResultSet}.
 *
 * <em>Note:</em> This may happen at any point during the lifetime of an instance,
 * not necessarily before {@link LazyResultSetExtractor#extractData(ResultSet)} returns!
 * (As would be the case for regular {@code ResultSetExtractor}s.)
 * @param <T>
 */
public interface LazyResultSetExtractor<T> extends ResultSetExtractor<T> {
    void onFinish(Runnable action);
}

RecordExtractor.java

public class RecordExtractor<PrimaryKey> implements LazyResultSetExtractor<Stream<ValueOrError<GenericRecord, SQLException>>> {
    @Override @Nonnull
    public Stream<ValueOrError<GenericRecord, SQLException>> extractData(@Nonnull final ResultSet rs) {
        return StreamSupport.stream(new ResultSetSpliterator(rs), false);
    }
}

ResultSetSpliterator.java

@RequiredArgsConstructor
class ResultSetSpliterator implements Spliterator<ValueOrError<GenericRecord, SQLException>> {

    private final ResultSet resultSet;

    @Override
    public boolean tryAdvance(Consumer<? super ValueOrError<GenericRecord, SQLException>> action) {
        Objects.requireNonNull(action); // as per specification of `tryAdvance`

        boolean hasRows;
        try {
            hasRows = resultSet.next();
        } catch (SQLException e) {
            // TODO: make error handling injectable
            log.debug("Stopped record extraction with exception", e);
            // NB: According to the doc of ResultSet#next, this is a valid way of implementing a ResultSet. Meh.
            hasRows = false;
        }

        if (!hasRows) {
            extractionFinished(); // TODO: generalize; can contain stuff like closing ResultSet, statement,
            return false;
        }

        try {
            action.accept(new ValueOrError<>(generateRecord(resultSet))); // TODO: abstract from generateRecord
            return true;
        } catch (SQLException e) {
            action.accept(new ValueOrError<>(e));
            return true;
        } catch (Throwable t) {
            log.error("Unexpected error while extracting result set -- aborting", t);
            action.accept(new ValueOrError<>(new Aborted("Unexpected error", t)));
            try {
                resultSet.afterLast(); // skip to the end
            } catch (SQLException e) {
                log.warn("Could not skip to the end of the result set", e);
                // ¯\_(ツ)_/¯
            }
            return true;
        }
    }

    @Override
    public Spliterator<ValueOrError<GenericRecord, SQLException>> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return 0;
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL | IMMUTABLE;
    }
}