46

When I query a database and receive a (forward-only, read-only) ResultSet back, the ResultSet acts like a list of database rows.

I am trying to find some way to treat this ResultSet like a Scala Stream. This will allow such operations as filter, map, etc., while not consuming large amounts of RAM.

I implemented a tail-recursive method to extract the individual items, but this requires that all items be in memory at the same time, a problem if the ResultSet is very large:

// Iterate through the result set and gather all of the String values into a list
// then return that list
@tailrec
def loop(resultSet: ResultSet,
         accumulator: List[String] = List()): List[String] = {
  if (!resultSet.next) accumulator.reverse
  else {
    val value = resultSet.getString(1)
    loop(resultSet, value +: accumulator)
  }
}
3
  • Could you use an Iterable instead of a Stream to do what you want? Mar 9, 2012 at 17:19
  • 3
    Also a stream will retain the values in memory anyway so you wont actually save memory by the time you reach the end of the list. Jul 30, 2013 at 13:16
  • 1
    I think without a jdbc flag/option that makes jdbc itself stream the results, you still have one full copy of the data in memory, built by your jdbc api.
    – matanster
    Mar 8, 2016 at 20:57

10 Answers 10

78

I didn't test it, but why wouldn't it work?

new Iterator[String] {
  def hasNext = resultSet.next()
  def next() = resultSet.getString(1)
}.toStream
8
  • That looks perfect. I'll test it as soon as I get my database set up. I don't even think I need to convert it to a Stream. I can apply map, filter, etc. directly to it.
    – Ralph
    Mar 10, 2012 at 13:43
  • 1
    I would like to give you a second up-vote. I've added this code fragment to my Scala snippets library. It's quickly becoming one of my favorites.
    – Ralph
    Mar 22, 2012 at 11:56
  • 8
    It's a cool solution but I worry. I think the usual contract of Iterator is that hasNext is side-effect-free. It could be called any number of times between two calls to next. Is there something preventing this from becoming an issue? Jan 25, 2016 at 18:01
  • 1
    Didn't work for me with mysql-connector-java version 6. Not sure if I did anything wrong, but my ResultSet got closed on the second next() call, so I could only retrieve one result row. The only way it's not auto-closed before I got all rows seems to be using while (rs.next()) {...}, so I add items individually to a scala.collection.mutable.ListBuffer within the while. Doesn't seem pretty, but couldn't figure out any other way.
    – Nick
    Apr 7, 2017 at 11:16
  • 1
    @Nick Using new Iterator[String]{ ... }.toList instead of .toStream will fetch the entire set of results immediately, instead of just the first row.
    – steinar
    Aug 22, 2017 at 18:22
13

Utility function for @elbowich's answer:

def results[T](resultSet: ResultSet)(f: ResultSet => T) = {
  new Iterator[T] {
    def hasNext = resultSet.next()
    def next() = f(resultSet)
  }
}

Allows you to use type inference. E.g.:

stmt.execute("SELECT mystr, myint FROM mytable")

// Example 1:
val it = results(stmt.resultSet) {
  case rs => rs.getString(1) -> 100 * rs.getInt(2)
}
val m = it.toMap // Map[String, Int]

// Example 2:
val it = results(stmt.resultSet)(_.getString(1))
10

This sounds like a great opportunity for an implicit class. First define the implicit class somewhere:

import java.sql.ResultSet

object Implicits {

    implicit class ResultSetStream(resultSet: ResultSet) {

        def toStream: Stream[ResultSet] = {
            new Iterator[ResultSet] {
                def hasNext = resultSet.next()

                def next() = resultSet
            }.toStream
        }
    }
}

Next, simply import this implicit class wherever you have executed your query and defined the ResultSet object:

import com.company.Implicits._

Finally get the data out using the toStream method. For example, get all the ids as shown below:

val allIds = resultSet.toStream.map(result => result.getInt("id"))
2
  • Are you sure it works? It fails on DB2 with ResultSet being closed. If this worked in your case perhaps it depends on the specific database brand and/or configuration? Aug 31, 2018 at 9:32
  • It does but you can only use the stream as long as your connection remains open. If you close your connection, the stream will fail, as will the iterator. Sep 5, 2018 at 2:25
3

i needed something similar. Building on elbowich's very cool answer, I wrapped it a bit, and instead of the string, I return the result (so you can get any column)

def resultSetItr(resultSet: ResultSet): Stream[ResultSet] = {
    new Iterator[ResultSet] {
      def hasNext = resultSet.next()
      def next() = resultSet
    }.toStream
  }

I needed to access table metadata, but this will work for table rows (could do a stmt.executeQuery(sql) instead of md.getColumns):

 val md = connection.getMetaData()
 val columnItr = resultSetItr( md.getColumns(null, null, "MyTable", null))
      val columns = columnItr.map(col => {
        val columnType = col.getString("TYPE_NAME")
        val columnName = col.getString("COLUMN_NAME")
        val columnSize = col.getString("COLUMN_SIZE")
        new Column(columnName, columnType, columnSize.toInt, false)
      })
1
  • 1
    If you don't need to go back on the stream (e.g., forward iteration only), you can just use an iterator. This greatly reduces the memory overhead of using a stream (return an Iterator[ResultSet], and drop the toStream)
    – Greg
    Sep 15, 2014 at 17:32
2

Because ResultSet is just a mutable object being navigated by next, we need to define our own concept of a next row. We can do so with an input function as follows:

class ResultSetIterator[T](rs: ResultSet, nextRowFunc: ResultSet => T) 
extends Iterator[T] {

  private var nextVal: Option[T] = None

  override def hasNext: Boolean = {
    val ret = rs.next()
    if(ret) {
      nextVal = Some(nextRowFunc(rs))
    } else {
      nextVal = None
    }
    ret
  }

  override def next(): T = nextVal.getOrElse { 
    hasNext 
    nextVal.getOrElse( throw new ResultSetIteratorOutOfBoundsException 
  )}

  class ResultSetIteratorOutOfBoundsException extends Exception("ResultSetIterator reached end of list and next can no longer be called. hasNext should return false.")
}

EDIT: Translate to stream or something else as per above.

2
Iterator.continually(rs.next())
  .takeWhile(identity)
  .map(_ => Model(
      id = rs.getInt("id"),
      text = rs.getString("text")
   ))
1

Here is an alternative, similar to Sergey Alaev's and thoredge's solutions, for when we need a solution which honors the Iterator contract where hasNext is side-effect free.

Assuming a function f: ResultSet => T:

Iterator.unfold(resultSet.next()) { hasNext =>
  Option.when(hasNext)(f(resultSet), resultSet.next())
}

I've found it useful to have as map "extension method" on ResultSet.

implicit class ResultSetOps(resultSet: ResultSet) {
    def map[T](f: ResultSet => T): Iterator[T] = {
      Iterator.unfold(resultSet.next()) { hasNext =>
        Option.when(hasNext)(f(resultSet), resultSet.next())
      }
    }
  }
0

This implementation, although longer and clumsier it is in better correspondence with the ResultSet contract. The side-effect has been removed from hasNext(...) and moved into next().

new Iterator[String] {
  private var available = resultSet.next()
  override def hasNext: Boolean = available
  override def next(): String = {
    val string = resultSet.getString(1)
    available = resultSet.next()
    string
  }
}
0

I think most of above implementations has a nondeterministic hasNext method. Calling it two times will move cursor to the second row. I would advise to use something like that:

  new Iterator[ResultSet] {
    def hasNext = {
      !resultSet.isLast
    }
    def next() = {
      resultSet.next()
      resultSet
    }
  }
0

Another variant on the above, which works with Scala 2.12:

implicit class ResultSetOps(resultSet: ResultSet) {
 def map[T](f: ResultSet => T): Iterator[T] =
  Iterator.continually(resultSet).takeWhile(_.next()).map(f)
}

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.