在 Scala 中功能性地处理数据库游标答案

【问题标题】：Functionally processing a database cursor in Scala在 Scala 中功能性地处理数据库游标
【发布时间】：2012-10-08 16:55:03
【问题描述】：

当我需要使用 JDBC 驱动程序从 PostgreSQL 数据库中读取数百万个数据库行时，我总是使用游标，否则我会得到 OutOfMemoryError。这是我使用的模式（伪代码）：

begin transaction
execute("declare cursor...")
while (true) {
  boolean processedSomeRows = false
  resultSet = executeQuery("fetch forward...")
  while (resultSet.next()) {
    processedSomeRows = true
    ...
  }
  if (!processedSomeRows) break
}
close cursor
commit

这是我想出的在 Scala 中实现的更具“功能性”的等价物：

begin transaction
execute("declare cursor...")

@tailrec
def loop(resultSet: ResultSet,
         processed: Boolean): Boolean = {
  if (!resultSet.next()) processed
  else {
    // Process current result set row
    loop(resultSet, true)
  }
}

while (loop(executeQuery("fetch forward..."), false))
  ; //Empty loop

close cursor
commit

我知道这是人为的，但有没有更好的方法而不诉诸可变性？如果我尝试在 Haskell 中执行此操作，我可能会想出一个涉及 monad 的解决方案，但我不想让我的思绪落入那些“曲折的小段落，都一样”，因为它可能永远不会回来......

【问题讨论】：

我的直接反应是，无论你的“处理当前结果集”函数是什么，一个 Iterator[ResultSet] 提供一个 .map() 将是一个更清洁的解决方案的一部分......你有任何理由拒绝这种方法？
我通常使用Iterators 或Streams 来处理ResultSets，但我想不出一种方法来停止迭代（上面的processed 参数）。
在 Iterator 上使用 takeWhile，当您放入其中的任何函数变为 false 时，它将停止绘制新值。这个问题实际上是关于如何在 Iterator 完成后执行清理代码吗？
我必须考虑一下（使用takeWhile），看看我是否可以终止外部while 循环。

标签： database scala functional-programming cursor

【解决方案1】：

这是我想出的 Scala 解决方案：

@tailrec
def processCursor(query: => ResultSet)(process: ResultSet => Unit) {
  @tailrec
  def loop(resultSet: ResultSet,
            processed: Boolean): Boolean = {
    if (!resultSet.next()) processed
    else {
      process
      loop(resultSet, true)
    }
  }
  if (loop(query, false)) processCursor(query)(process)
}

这样称呼它：

begin transaction
execute("declare cursor...")

processCursor(statement.executeQuery("fetch forward...")) {
  resultSet =>
  // process current row of the ResultSet
}

close cursor
commit

如何改进？

【讨论】：

【解决方案2】：

基于@Ralph 答案的替代方式：

  def processCursor[T](resultSet: ResultSet)(process: ResultSet => T) = {
    @tailrec
    def loop(seq: Seq[T], resultSet: ResultSet): Seq[T] = {
      if (resultSet.next()) loop(seq :+ process(resultSet), resultSet)
      else seq
    }
    loop(Seq.empty, resultSet)
  }

【讨论】：