Scala中的延迟分页（迭代器的流/迭代器？）

Question

我通过调用def readPage(pageNumber: Int): Iterator[Record]，一次一页从数据库 API 中顺序读取大量记录（每页记录数未知）

我正在尝试将这个 API 封装在 Stream[Iterator[Record]] 或 Iterator[Iterator[Record]] 之类的东西中，以一种功能性的方式，理想情况下没有可变状态，具有恒定的内存占用，以便我可以将其视为无限的页面流或迭代器序列，并从客户端抽象出分页。客户端可以对结果进行迭代，通过调用 next() 它将检索下一页 (Iterator[Record])。

在 Scala 中实现这一点最惯用和最有效的方法是什么。

编辑：需要一次一页地获取和处理记录，无法维护内存中所有页面的所有记录。如果一页失败，则抛出异常。大量页面/记录对于所有实际目的来说意味着无限。我想将其视为无限的页面流（或迭代器），每个页面都是有限数量记录的迭代器（例如，小于

我在 Monix 中查看了 BatchCursor，但它的用途不同。

编辑 2：这是当前版本，使用以下 Tomer 的 answer 作为起点，但使用 Stream 而不是 Iterator。 这允许根据 https://stackoverflow.com/a/10525539/165130 消除尾递归的需要，并且有 O(1) 时间用于流前置 #:: 操作（而如果我们通过 ++ 操作连接迭代器，它将是 O(n)）

注意：虽然流被延迟评估，Stream memoization 仍可能导致内存爆炸，并且内存管理得到tricky。从val 更改为def 以在下面的def pages = readAllPages 中定义Stream 似乎没有任何效果

def readAllPages(pageNumber: Int = 0): Stream[Iterator[Record]] = {
   val iter: Iterator[Record] = readPage(pageNumber)
   if (iter.isEmpty)
     Stream.empty
   else
    iter #:: readAllPages(pageNumber + 1)
} 
      
//usage
val pages = readAllPages
for{
    page<-pages
    record<-page
    if(isValid(record))
}
process(record)
 

编辑 3： Tomer 的第二个建议似乎是最好的，它的运行时间和内存占用与上述解决方案类似，但更简洁且容易出错。

val pages = Stream.from(1).map(readPage).takeWhile(_.nonEmpty)

注意：Stream.from(1) 创建一个从 1 开始并递增 1 的流，它在 API docs 中

Answer 1

你可以尝试实现这样的逻辑：

def readPage(pageNumber: Int): Iterator[Record] = ???

@tailrec
def readAllPages(pageNumber: Int): Iterator[Iterator[Record]] = {
  val iter = readPage(pageNumber)
  if (iter.nonEmpty) {
    // Compute on records
    // When finishing computing:
    Iterator(iter) ++ readAllPages(pageNumber + 1)
  } else {
    Iterator.empty
  }
}

readAllPages(0)

一个较短的选项是：

Stream.from(1).map(readPage).takeWhile(_.nonEmpty)