Help for JDBC with Java streams

We wanted to use JDBC with Java Streams, but encountered several difficulties. Fortunately we found solutions with rather small bits of code.

Checked exceptions

The main obstacle was the JDBC API throwing SQLException for all API methods used in our code. SQLException is a checked exception, so must be declared in our method signatures, or caught otherwise. However the Streams API only accepts methods which declare to throw no checked exceptions, so something simple like this will not compile:

stream(results).
        map(row -> row.getString("label")).  // checked exception
        forEach(this::processLabel);

The call to ResultSet.getString(String) throws a checked exception. The usual approach is to wrap the call, and handle the exception in the wrapping method:

method String streamGetLabel(final ResultSet results) {
    try {
        return results.getString("label");
    } catch (final SQLException e) {
        throw new UncheckedIOException(e);
    }
}

(Here UncheckedIOException is an unchecked exception wrapper we wrote for SQLException, similar to UncheckedIOException in the JDK for IOException.)

Then the stream becomes:

stream(results).
        map(this::streamGetLabel).
        forEach(this::processLabel);

This is OK, however needing to write a wrapper method for each time we wanted to use JDBC in a stream became tedious.

Solution

First we wrote a SAM (more on SAM interfaces/classes) interface as a lookalike for the JDK Function interface: this is what Stream.map(Function) wants.The lookalike is different in that it throws SQLException:

@FunctionalInterface
public interface SQLFunction<T, R> {
    R apply(final T t) throws SQLException;

    // Other default methods - no more abstract methods
}

Then we used this in a closed Function implementation to wrap and delegate to the lookalike, and throw UncheckedSQLException if the lookalike throws SQLException:

@RequiredArgsConstructor(staticName = "applyUnchecked")
public final class UncheckedSQLFunction<T, R>
        implements Function<T, R> {
    private final SQLFunction<T, R> wrapped;

    @Override
    public R apply(final T t) {
        try {
            return wrapped.apply(t);
        } catch (final SQLException e) {
            throw new UncheckedSQLException(e);
        }
    }
}

(Here we use the excellent Lombok library to generate our constructor, and give us a static convenience method, "applyUnchecked".)

Finally some static importing, and our example streams use becomes:

stream(results).
        map(applyUnchecked(row -> row.getString("label"))).
        forEach(this::processLabel);

Or with more help:

stream(results).
        map(getString("label")).
        forEach(this::processLabel);

We wrote similar lookalikes and wrappers for Predicate and Consumer. It would be easy enough to write them for other Java functional interfaces, such as BiFunction.

Streaming result sets

The next difficulty we tackled was how to loop over ResultSet, and use them with Streams.

(A side note: ResultSet is not a set but a list: rows are ordered, and they can duplicate each other in their column data. However, they were named after the SQL concept of sets, not the Java one.)

Fundamentally, a ResultSet is not an iterator, but is close:

Iterator	ResultSet	Returns
hasNext()	next()	boolean
next()	this	ResultSet	(Yes, the ResultSet itself is the equivalent)

Solution

To provide a ResultSet as an iterator:

final List<String> values = new ArrayList<>();
        for (final ResultSet row : iterable(results)) {
            values.add(row.getString("value"));
        }

Moreso, to provide one as a stream:

final List<String> values = stream(results).
                map(getString("value")).
                collect(toList());

Any SQL failures are thrown as unchecked exceptions. The stream has the characteristics: immutable, nonnull, and ordered.

Transactions

We found JDBC transactions to be tricky. Fundamentally they are tied to a connection; there is no proper nesting. (To simulate nesting, use separate connections. Even then, there is no guarantee of ordering from the database engine.) And they have a baroque API, relying on diddling of the "auto-commit" setting with care needed to restore its setting after the transaction concludes. Several bugs ensued before we switched to using small helper interfaces and methods.

Further, some programming languages do not distinguish void from other return types (e.g., Unit type): Java is not one of them. Likewise for user vs primitive types (Boolean vs boolean). Hence, there are separate transaction blocks for consumers, functions, and predicates.

Solution

One example explains them all. Consider functions and the SQLFunction lookalike interface:

@RequiredArgsConstructor(staticName = "applyTransacted")
public final class TransactedFunction<T, R>
        implements SQLFunction<T, R> {
    private final Connection connection;
    private final SQLFunction<T, R> wrapped;

    @Override
    public R apply(final T in)
            throws SQLException {
        connection.setAutoCommit(false);
        try {
            final R out = wrapped.apply(in);
            connection.commit();
            return out;
        } catch (final SQLException e) {
            connection.rollback();
            throw e;
        } finally {
            connection.setAutoCommit(true);
        }
    }
}

(The pattern is the same for other collection operations.)

With a helper and static importing:

final Integer value = applyTransacted(connection, in -> 0).apply("string");

Or when the transaction fails:

applyTransacted(connection, in -> {
        throw new SQLException("Something went wrong");
    }).apply("string");

Some convenience

Many places in these examples are improved with helper functions, or for transactions, with currying (similar to the builder pattern). Hence, the wide use of Lombok static constructors. Transactions are another example as they need a Connection for begin/commit/rollback.

Solution

A simple helper curries connection for transactions:

@RequiredArgsConstructor(staticName = "with")
public final class WithConnection {
    private final Connection connection;

    public <T> Predicate<T> testTransacted(final SQLPredicate<T> wrapped) {
        return UncheckedSQLPredicate.testUnchecked(
                TransactedPredicate.<T>testTransacted(connection, wrapped));
    }

    public <T, R> Function<T, R> applyTransacted(
            final SQLFunction<T, R> wrapped) {
        return UncheckedSQLFunction.applyUnchecked(
                TransactedFunction.<T, R>applyTransacted(connection,
                        wrapped));
    }

    public <T> Consumer<T> acceptTransacted(final SQLConsumer<T> wrapped) {
        return UncheckedSQLConsumer.acceptUnchecked(
                TransactedConsumer.<T>acceptTransacted(connection, wrapped));
    }
}

Example use:

final Optional<Integer> value = Stream.of(0).
        filter(with(connection).testTransacted(in -> true)).
        findFirst();

(Yes, we might also describe the code as partial application. The "object-oriented" implementation confuses matters with the hidden this reference.)

Conclusion

There is nothing we did that was difficult or complex: simple one-liner interfaces, simple wrapper implementations of Java funcional interfaces, some rote JDBC best practices. The main difficulty was conceptual: seeing the duplication of many, small wrapper methods, and pulling out their commonality. This is a good pattern to keep in mind throughout your code.

UPDATE:

And the source: Java helpers for JDBC and Streams.

Tracking Java 8 stream count

I had a coding problem to fail when a Java 8 stream was empty. One way is illustrated below in example1. Another approach was interesting, shown in example2

public final class Streamy {
    private static <T, E extends RuntimeException> void example1(
            final Stream<T> items,
            final Consumer<? super T> process,
            final Supplier<E> thrown)
            throws E {
        if (0 == items.
                peek(process).
                count())
            throw thrown.get();
    }

    private static <T, E extends RuntimeException> void example2(
            final Stream<T> items,
            final Consumer<? super T> process,
            final Supplier<E> thrown)
            throws E {
        final AtomicBoolean foundSome = new AtomicBoolean();
        try (final Stream<T> stream = items) {
            stream.
                    onClose(() -> {
                        if (!foundSome.get())
                            throw thrown.get();
                    }).
                    peek(__ -> foundSome.set(true)).
                    forEach(process);
        }
    }

    public static void main(final String... args) {
        example1(Stream.of(), out::println, RuntimeException::new);
        example2(Stream.of(), out::println, RuntimeException::new);
    }
}

Comparing them I find:

• Using count() is shorter and more clear
• Using onClose() is more expressive

I found it odd to use peek() in example1 to execute the real purpose of the code, and was happy to discover onClose() though disappointed to need try-with-resources for it to run.

It was unfortunate that the more expressive approach (peek() for side effect, forEach() for processing, onClose for post-processing check) was also harder to understand.

Overcoming Java 8 streams

Java 8 streams provide functional and LINQ-like features in a fluent API. But streams are not without drawbacks:

• Referenced methods and lambdas cannot throw checked exceptions
• Controlling the threads used, especially for parallel streams, is awkward
• Streams are not designed for extension

Overcoming these drawbacks requires a "look-a-like" API. For example, implementing java.util.stream.Stream does not help: none of the existing methods throw checked exceptions, and none of the existing stream factory helpers would return your implementation with new methods.

So I wrote my own, copying the existing stream API, updating the methods to throw checked exceptions:

hm.binkley.util.stream.CheckedStream ('develop' branch for now)

From the javadoc:

CheckedStream is a throwing Stream look-a-like with control over thread pool. It cannot be a Stream as it takes throwing versions of suppliers, functions and consumers. Otherwise it is a faithful reproduction.

Write this:

   long beanCount() throws SomeException, OtherException {
       checked(Stream.of(1, 2, 3)).
           map(this::someThrowingFunction).
           peek(That::oldBean).
           count();
   }

not this:

   long beanCount() throws SomeException, OtherException {
       try {
           Stream.of(1, 2, 3).
               map(i -> {
                   try {
                       someThrowingFunction(i);
                   } catch (final SomeException e) {
                       throw new RuntimeException(e);
                   }
               }).
               peek(i -> {
                   try {
                       That.oldBean(i);
                   } catch (final OtherException e) {
                       throw new RuntimeException(e);
                   }
               }).
               count();
       } catch (final RuntimeException e) {
           final Throwable x = e.getCause();
           if (x instanceof SomeException)
               throw (SomeException) x;
           if (x instanceof OtherException)
               throw (OtherException) x;
           throw e;
       }
   }

"Intentional" exceptions (checked exceptions plus CancellationException) have "scrubbed" stacktraces: frames from framework/glue packages are removed before the intentional exception is rethrown to calling code. Scrubbed stacktraces are much easier to understand, the framework and glue code having been removed.

To see the unscrubbed stacktrace, set the system property "hm.binkley.util.stream.CheckedStream.debug" to "true".

Controlling the thread pool used by Stream is a challenge. Deep in the implementation, it checks if being run in a ForkJoinTask, and uses that thread if so, otherwise using the common pool. So with CheckedStream write this:

       checked(stream, new ForkJoinPool()).
           map(currentThread()).
           forEach(System.out::println);

not this:

       try {
           new ForkJoinPool().submit(() -> stream.
                   map(currentThread()).
                   forEach(System.out::println)).
               get();
       } catch (final ExecutionException e) {
           final Throwable x = e.getCause();
           if (x instanceof Error)
               throw (Error) x;
           if (x instanceof RuntimeException)
               // Much tricker when stream functions throw runtime
               throw (RuntimeException) x;
           throw new Error(e); // We have no checked exceptions in this example
       }

Care is taken to respect lazy and terminal operations in using thread pools. Changing thread pool or thread mode mid-stream is supported, and are "immediate" operations: they terminate the existing stream, and start a new one with the changes:

stream.sequential().
    filter(this::someFilter).
    parallel(threads). // Existing lazy operations terminated
    map(this:someMapper).
    forEach(System.out::println);

Immediate operations ensure stream methods are run in the correct threading context.

I hope you'll agree: CheckedStream is nicer to use, especially with existing code using checked exceptions.

Suggestions, bug fixes, improvements welcome!