private void getFlowableQueryResults(final TaskContext taskContext,
                                         final SqlBuilder sql,
                                         final Function<ResultSet, Val[]> resultSetMapper,
                                         final ValuesConsumer valuesConsumer) {
        long count = 0;

        try (final Connection connection = sqlStatisticsDbConnProvider.getConnection()) {
            //settings ot prevent mysql from reading the whole resultset into memory
            //see https://github.com/ontop/ontop/wiki/WorkingWithMySQL
            //Also needs 'useCursorFetch=true' on the jdbc connect string
            try (final PreparedStatement preparedStatement = connection.prepareStatement(
                    sql.toString(),
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY,
                    ResultSet.CLOSE_CURSORS_AT_COMMIT)) {

                final int fetchSize = searchConfig.getFetchSize();
                LOGGER.debug("Setting fetch size to {}", fetchSize);
                preparedStatement.setFetchSize(fetchSize);

                PreparedStatementUtil.setArguments(preparedStatement, sql.getArgs());
                LAMBDA_LOGGER.debug(() -> String.format("Created preparedStatement %s", preparedStatement));

                final String message = String.format("Executing query %s", sql);
                taskContext.info(() -> message);
                LAMBDA_LOGGER.debug(() -> message);

                try (final ResultSet resultSet = preparedStatement.executeQuery()) {

                    // TODO prob needs to change in 6.1
                    while (resultSet.next() &&
                           !Thread.currentThread().isInterrupted()) {
                        LOGGER.trace("Adding result");
                        final Val[] values = resultSetMapper.apply(resultSet);
                        valuesConsumer.accept(values);
                        count++;
                    }

                    if (Thread.currentThread().isInterrupted()) {
                        LOGGER.debug("Task is terminated/interrupted, calling onComplete");
                    } else {
                        LOGGER.debug("End of resultSet, calling onComplete");
                    }

                } catch (SQLException e) {
                    throw new RuntimeException(String.format("Error executing query %s, %s",
                            preparedStatement, e.getMessage()), e);
                }

            } catch (SQLException e) {
                throw new RuntimeException(String.format("Error preparing statement for sql [%s]", sql), e);
            }

        } catch (final SQLException e) {
            throw new RuntimeException("Error getting connection", e);
        }
    }
Build a mapper function that will only extract the columns of interest from the resultSet row.
Assumes something external to the returned function will advance the resultSet
stroom-statistics/stroom-statistics-impl-sql/src/main/java/stroom/statistics/impl/sql/search/StatisticsSearchServiceImpl.java