Iterating Results - nMoncho/helenus GitHub Wiki

After executing a query we can iterate its results in different ways, it's important to keep in mind that, just like the Cassandra Java Driver results can only be iterated once. If the result is short enough, it can be converted to a Scala collection and iterated multiple times.

For paging results, please read Pagination.

Synchronous Results

When executing a query synchronously we get a PagingIterable[T], where T is the output element type. By default T is a Row, but can be transformed to any other type with a RowMapper.

PagingIterable works like a Java Iterable, so it can be traversed, and converted, in the same manner. Mind that PagingIterable pages results in a blocking fashion, fetching new pages under the hood as results are consumed.

Helenus also provides some extension methods to make iterating over the results easier:

Dequeue elements

PagingIterable has a method named one that allows us to dequeue elements, one at a time, from the result set. This method returns null when all rows have been exhausted. Helenus provides nextOption wrapping the same result with an Option to avoid propagating nulls.

val hotels = "SELECT * FROM hotels".toCQL.prepareUnit
// hotels: internal.cql.ScalaPreparedStatementUnit[Row] = net.nmoncho.helenus.internal.cql.ScalaPreparedStatementUnit@4b08933f

val resultSet = hotels.execute()
// resultSet: PagingIterable[Row] = com.datastax.oss.driver.internal.core.PagingIterableWrapper@2717634b

val first = resultSet.nextOption()
// first: Option[Row] = Some(
//   value = com.datastax.oss.driver.internal.core.cql.DefaultRow@18c70b1d
// )

val second = resultSet.nextOption()
// second: Option[Row] = None

val third = resultSet.nextOption()
// third: Option[Row] = None

Transform to a Scala Iterator

A PagingIterable can produce a Java Iterator with its iterator method. We can covert this iterator into a Scala iterator with one of the conversion method provided by the Scala library.

We can also do this in one step with the iter extension method:

val iterator = hotels.execute().iter
// iterator: Iterator[Row] = empty iterator

val firstIt = iterator.nextOption()
// firstIt: Option[Row] = Some(
//   value = com.datastax.oss.driver.internal.core.cql.DefaultRow@57f364a1
// )

val secondIt = resultSet.nextOption()
// secondIt: Option[Row] = None

val thirdIt = resultSet.nextOption()
// thirdIt: Option[Row] = None

Transform to a Scala Collection

A PagingIterable can be converted to a Scala Collection with the to extension method:

val allHotels = hotels.execute().to(List)
// allHotels: List[Row] = List(
//   com.datastax.oss.driver.internal.core.cql.DefaultRow@5f641e4f
// )

This conversion method should only be used when we know that the result set won't bring too many results, as it will fetch all results eagerly.

Asynchronous Results

Unlike the synchronous API, the Cassandra Java Driver asynchronous API exposes a pagination API with its currentPage, hasMorePage, and fetchNextPage methods.

Helenus tries to align this API with its synchronous counterpart using extension methods:

Dequeue elements

We can also consume one element at a time using nextOption, for which we get an optional next element, and the AsyncPagingIterable we should use on our next invocation of nextOption. The purpose behind this is to avoid dealing a reference of AsyncPagingIterable that mutates when the next page has to be fetched.

Transform to a Scala Iterator

Using the currPage extension method we can transform the current page to a Scala Iterator.

Using the iter extension method we can emulate the synchronous API, where new pages are fetched under the hood. A new page will only be fetched when all the elements for the current page have been consumed.

Reactive Results

Iterating results reactively will probably be done either with Akka or Pekko.