postgresql - 使用 scalikejdbc 流式传输大表

Question

我有一个返回大量数据的查询，我想使用提取大小将这些数据从 postgres 流式传输到 scala。我的代码看起来像这样

  override def getFilteredData(filters: Seq[Filter]): StreamReadySQL[myObject] =  {
      sql"""  QUERY
        """.stripMargin
           .fetchSize(2000)
           .map(MyObject(_))
           .iterator()
  }

在其他课程中，我这样称呼它

  val productPublisher: DatabasePublisher[RuleData] = {
    DB readOnlyStream {
      repositories.productRepository.getFilteredData(trigger.filters)
    }
  }

  Source.fromPublisher(publisher).mapAsync(Parallelism) { batch =>
    //do something
    Future.successful(().asRight[ApplicationError])
  }.runWith(Sink.head[Either[ApplicationError, Unit]])

使用此代码，上面的源将只返回最后 n 个获取的条目，如何从可以迭代所有数据的流中获取迭代器？