Skip to main content

How to Stream Large Query Results

In this guide, we read a large sales table as a stream of row chunks with Frag#queryStream and Frag#queryChunked. By the end, you will have a report of copies sold per title and a CSV export of every sale, both computed while only a small batch of rows is held in memory at a time.

We start from a table of 12,000 sales and add one capability at a time:

  • Aggregating the whole table with runFold, one chunk at a time
  • Choosing the chunk size
  • Exporting every row to a CSV file through a Sink
  • Filtering and reshaping rows on their way to the file
  • Keeping the stream inside the block that owns the connection
  • Handling a bad query or a row that can't be decoded

This guide builds on Build Your First Database Repository on SQLite. We use JdbcTransactor, Table, Repo and sql"..." the way that tutorial introduces them and don't explain them again. The streaming operators we use (map, filter, flatMap, grouped, runFold) are the ones the tutorial "Build Your First Streaming Pipeline" teaches.

The Problem​

Frag#query returns a List: it reads every row of the result into memory before it gives you the first one. That is fine for a page of results, but a report over a whole table holds the entire table on the heap, and an export of a table that doesn't fit in memory fails with an OutOfMemoryError halfway through.

Without a streaming API, the usual way out is a hand-written JDBC cursor loop:

import java.sql.DriverManager
import scala.collection.mutable

val connection = DriverManager.getConnection("jdbc:sqlite:bookstore.db")
try {
  val statement = connection.prepareStatement("SELECT title, quantity FROM sale")
  try {
    statement.setFetchSize(500)
    val rows = statement.executeQuery()
    try {
      val copiesPerTitle = mutable.Map.empty[String, Int]
      while (rows.next()) {
        val title    = rows.getString("title")
        val quantity = rows.getInt("quantity")
        copiesPerTitle(title) = copiesPerTitle.getOrElse(title, 0) + quantity
      }
      println(copiesPerTitle)
    } finally rows.close()
  } finally statement.close()
} finally connection.close()

This keeps memory flat, but every report repeats the same three nested try/finally blocks, decodes columns by hand, and mixes the reading, the decoding and the aggregation in one loop. Writing a second consumer, such as a file export, means copying the whole loop again.

In this guide, we solve this by asking Frag for a Stream of decoded rows and building each consumer from stream operators.

Prerequisites​

Add the sql module, the streams module and the SQLite JDBC driver to your build.sbt:

// Scala 3.9 and newer, with scalacOptions += "-experimental"; on Scala 3.3 to 3.8 the artifacts are "zio-blocks-sql" and "zio-blocks-streams"
libraryDependencies += "dev.zio"   %% "zio-blocks-next-sql"     % "0.0.56"
libraryDependencies += "dev.zio"   %% "zio-blocks-next-streams" % "0.0.56"
libraryDependencies += "org.xerial" % "sqlite-jdbc"             % "3.53.4.0"

The sql module already depends on streams; listing it makes the dependency explicit, since the code imports zio.blocks.streams directly. The sql module is published for Scala 3 only, and JdbcTransactor runs on the JVM.

We use these imports throughout the guide:

import java.nio.file.Files
import java.time.LocalDate
import scala.util.Try

import zio.blocks.chunk.Chunk
import zio.blocks.schema.Schema
import zio.blocks.streams.*
import zio.blocks.sql.*

The Sales Table​

Our bookstore records one row per sale: which book, how many copies, and on which day. The sales data is fictional; the titles are real Scala books:

case class Sale(id: Long, title: String, quantity: Int, soldOn: LocalDate)

object Sale {
  given Schema[Sale] = Schema.derived
}

val saleTable = Table.derived[Sale]
val sales     = Repo.derived[Sale, Long]

val titles = Vector("Zionomicon", "Programming in Scala", "Functional Programming in Scala")

We store the table in a temporary SQLite file. jdbc:sqlite::memory: would not work here: SQLite gives every new connection its own empty in-memory database, and each connect block opens a new connection. Then we create the table and insert 12,000 sales in one transaction:

val dbFile     = Files.createTempFile("bookstore-sales", ".db")
val transactor = JdbcTransactor.fromUrl(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

val saleCount = 12000

transactor.transact {
  saleTable.createTable(SqlDialect.SQLite).update
  // In every 7 sales, 3 are of Zionomicon and 2 of each other title
  sales.insertBatch((1 to saleCount).map { i =>
    Sale(i.toLong, titles(i % 7 % 3), 1 + i % 4, LocalDate.of(2026, 1, 1).plusDays((i % 365).toLong))
  })
}

We also name the query we stream from in most sections. A Frag is only SQL text and parameters, so defining it touches no connection:

val allSales = sql"SELECT id, title, quantity, sold_on FROM sale ORDER BY id"

Aggregate a Large Table One Chunk at a Time​

Frag#queryStream[A] returns a Stream[Throwable, Chunk[A]]. Each element of the stream is a Chunk of up to 64 decoded rows, and a terminal operation such as runFold pulls the chunks one after another and returns an Either. We add up the copies sold per title, and also record how many chunks arrived and how large the biggest one was:

case class Report(copiesPerTitle: Map[String, Int], rows: Int, chunks: Int, largestChunk: Int)

val emptyReport = Report(Map.empty, 0, 0, 0)

def addChunk(report: Report, chunk: Chunk[Sale]): Report = {
  val copies = chunk.foldLeft(report.copiesPerTitle) { (totals, sale) =>
    totals.updated(sale.title, totals.getOrElse(sale.title, 0) + sale.quantity)
  }
  Report(copies, report.rows + chunk.size, report.chunks + 1, report.largestChunk.max(chunk.size))
}

val report: Either[Throwable, Report] =
  transactor.connect {
    allSales.queryStream[Sale].runFold(emptyReport)(addChunk)
  }

Then we print it:

report.foreach { r =>
  r.copiesPerTitle.toList.sorted.foreach((title, copies) => println(s"$title: $copies copies"))
  println(s"${r.rows} rows in ${r.chunks} chunks, at most ${r.largestChunk} rows per chunk")
}
// Functional Programming in Scala: 8569 copies
// Programming in Scala: 8572 copies
// Zionomicon: 12859 copies
// 12000 rows in 188 chunks, at most 64 rows per chunk

The fold saw all 12,000 rows, but each step only received one chunk of at most 64 of them. The fold keeps the running totals and drops each chunk once it has added it in, so the rows the stream holds at any moment are bounded by the chunk size, not by the size of the table.

Choose a Chunk Size​

queryStream is queryChunked with the default chunk size of 64. When you want larger or smaller batches, call Frag#queryChunked[A](chunkSize) instead. We run the same fold with chunks of 1,000 rows:

val bigChunks: Either[Throwable, Report] =
  transactor.connect {
    allSales.queryChunked[Sale](1000).runFold(emptyReport)(addChunk)
  }

The totals don't change, but the batches do:

bigChunks.foreach(r => println(s"${r.rows} rows in ${r.chunks} chunks, at most ${r.largestChunk} rows per chunk"))
// 12000 rows in 12 chunks, at most 1000 rows per chunk

The chunk size trades memory for per-chunk overhead. Larger chunks mean fewer steps through the stream and suit consumers that work best in batches, such as a batch insert into another table or a buffered write. Smaller chunks keep fewer rows in memory and deliver the first rows sooner, which matters when each row is large or when the consumer is slow. The chunk size bounds only the rows the stream holds; how many rows the JDBC driver fetches from the server per round trip is the driver's own setting.

The size must be at least 1. queryChunked checks it as soon as it is called, before any stream exists, so a bad size throws instead of returning a Left:

println(Try(transactor.connect(allSales.queryChunked[Sale](0))))
// Failure(java.lang.IllegalArgumentException: requirement failed: queryChunked: chunkSize must be >= 1, got 0)

Export Every Row to a CSV File​

To export the table, we write each chunk to a file as it arrives. Sink.foreach builds a Sink that runs a function for every element, and Stream#run(sink) drains the stream into it. We open a buffered writer, write a header line, and turn every sale into one CSV line:

def csvLine(sale: Sale): String =
  s"${sale.id},${sale.title},${sale.quantity},${sale.soldOn}"

val exportFile = Files.createTempFile("sales-export", ".csv")

val exported: Either[Throwable, Unit] = {
  val writer = Files.newBufferedWriter(exportFile)
  try {
    writer.write("id,title,quantity,sold_on")
    writer.newLine()
    transactor.connect {
      allSales.queryStream[Sale].run(Sink.foreach[Chunk[Sale]] { chunk =>
        chunk.foreach { sale =>
          writer.write(csvLine(sale))
          writer.newLine()
        }
      })
    }
  } finally writer.close()
}

Sink.foreach doesn't close the writer, so we close it in finally, whether the export succeeds or fails. Then we check the file:

println(exported)
// Right(())
println(s"Lines in the file: ${Files.readAllLines(exportFile).size}")
// Lines in the file: 12001
Files.readAllLines(exportFile).subList(0, 4).forEach(line => println(line))
// id,title,quantity,sold_on
// 1,Programming in Scala,2,2026-01-02
// 2,Functional Programming in Scala,3,2026-01-03
// 3,Zionomicon,4,2026-01-04

The file has the header and one line per sale. None of our titles contains a comma or a quote, so csvLine doesn't quote its fields; for real data, quote any field that may contain them.

Filter and Reshape Rows on the Way Out​

Stream operators work between the query and the sink, so the file receives only the rows and the columns we want. To export only Zionomicon sales as date,copies lines, we first flatten the chunks into single rows with flatMap(Stream.fromChunk(_)), then filter, map each sale to its line, and batch the lines again with grouped so the writer gets 500 lines per call:

val zionomiconFile = Files.createTempFile("zionomicon-sales", ".csv")

val zionomiconBatches: Either[Throwable, Long] = {
  val writer = Files.newBufferedWriter(zionomiconFile)
  try
    transactor.connect {
      allSales
        .queryChunked[Sale](1000)
        .flatMap(chunk => Stream.fromChunk(chunk))
        .filter(_.title == "Zionomicon")
        .map(sale => s"${sale.soldOn},${sale.quantity}")
        .grouped(500)
        .tapEach(lines => lines.foreach { line => writer.write(line); writer.newLine() })
        .count
    }
  finally writer.close()
}

tapEach writes each batch and passes it on, and count returns how many batches went through:

println(s"Batches written: $zionomiconBatches")
// Batches written: Right(11)
println(s"Lines in the file: ${Files.readAllLines(zionomiconFile).size}")
// Lines in the file: 5142
Files.readAllLines(zionomiconFile).subList(0, 3).forEach(line => println(line))
// 2026-01-04,4
// 2026-01-07,3
// 2026-01-08,4

The file has one line per Zionomicon sale. grouped(500) emits full batches of 500 lines and a smaller last batch for the rest.

warning

A known bug: after a flatMap over objects such as rows, a map to a primitive (Int, Long, Double, Boolean, Char) followed by grouped, sliding, take, takeWhile or scan throws a ClassCastException from blocking terminals such as runCollect or count. For example, .flatMap(chunk => Stream.fromChunk(chunk)).map(_.quantity).grouped(500) fails. Keep the elements as objects or strings until after the stateful operator, as above, or map to the primitive inside the batches (.grouped(500).map(_.map(_.quantity))).

Keep the Stream Inside the Connection Block​

A query stream is lazy. queryStream only prepares a description of the work; the statement runs and the result set opens on the first pull, and both close when the stream is fully drained or closed early. The connection itself belongs to the connect (or transact) block that provided the DbCon, and that block closes it when it returns.

So a stream that leaves its block can't be read any more. Here we return the stream itself from connect and run it afterwards:

val escaped: Stream[Throwable, Chunk[Sale]] =
  transactor.connect(allSales.queryStream[Sale])

The first pull happens after the connection is closed:

println(escaped.runFold(0)((rows, chunk) => rows + chunk.size))
// Left(java.sql.SQLException: database connection closed)

The failure is a Left from the terminal operation, not an exception. The fix is to run the terminal operation inside the block and return its result instead of the stream:

println(transactor.connect(allSales.queryStream[Sale].runFold(0)((rows, chunk) => rows + chunk.size)))
// Right(12000)

Because the stream is a description, every terminal operation runs the query again. If you need two results from the same rows, compute both in one fold, as the report above does, rather than running the stream twice.

caution

Consume the stream on the connection that opened it. Calling connect or transact again inside an open block, for example to write each chunk to another table, opens a second connection while the first one still holds the result set open. On SQLite the write then waits for the read's lock and fails with SQLITE_BUSY; with a connection pool on PostgreSQL, enough nested calls exhaust the pool and time out. Do the writes through the same DbCon, inside the same block.

Handle Query and Decoding Errors​

A query that fails surfaces as a Left from the terminal operation, carrying the driver's exception. A misspelled column fails when the statement runs on the first pull:

val badColumn = transactor.connect {
  sql"SELECT id, titel, quantity, sold_on FROM sale".queryStream[Sale].runFold(0)((rows, chunk) => rows + chunk.size)
}

We match on the result:

badColumn match {
  case Left(error)  => println(s"Query failed: ${error.getMessage}")
  case Right(rows)  => println(s"Read $rows rows")
}
// Query failed: [SQLITE_ERROR] SQL error or missing database (no such column: titel)

A row that can't be decoded fails the same way, at the row where decoding fails. Here the query hands the title column to the soldOn field, and a book title isn't a date:

val badRow = transactor.connect {
  sql"SELECT id, title, quantity, title AS sold_on FROM sale".queryStream[Sale].runFold(0)((rows, chunk) => rows + chunk.size)
}

The first row already fails:

badRow.left.foreach(error => println(s"${error.getClass.getSimpleName}: ${error.getMessage}"))
// DateTimeParseException: Text 'Programming in Scala' could not be parsed at index 0

Both errors arrive as Throwable values, so you decide in one place whether to log them, retry, or turn them into a domain error with Either#left.map.

warning

A known bug: a query that returns fewer columns than the row type has (for example SELECT id, title FROM sale decoded as Sale) throws a java.sql.SQLException out of the terminal operation instead of returning a Left. Wrap such calls in Try if the column list comes from outside your code.

Putting It Together​

The complete example creates the table, prints the per-title report, writes the CSV export, and deletes its temporary files:

sql-examples/src/main/scala/streamlargequeryresults/CompleteExample.scala
package streamlargequeryresults

import java.nio.file.Files

import zio.blocks.chunk.Chunk
import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Complete Example
 *
 * Creates the table, prints the per-title report, writes the CSV export, and
 * deletes its temporary files.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.CompleteExample"
 */
object CompleteExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val report: Either[Throwable, Report] =
        transactor.connect {
          allSales.queryStream[Sale].runFold(emptyReport)(addChunk)
        }

      report.foreach { r =>
        r.copiesPerTitle.toList.sorted.foreach((title, copies) => println(s"$title: $copies copies"))
        println(s"${r.rows} rows in ${r.chunks} chunks, at most ${r.largestChunk} rows per chunk")
      }

      val exportFile = Files.createTempFile("sales-export", ".csv")
      try {
        val exported: Either[Throwable, Unit] = {
          val writer = Files.newBufferedWriter(exportFile)
          try {
            writer.write("id,title,quantity,sold_on")
            writer.newLine()
            transactor.connect {
              allSales
                .queryStream[Sale]
                .run(Sink.foreach[Chunk[Sale]] { chunk =>
                  chunk.foreach { sale =>
                    writer.write(csvLine(sale))
                    writer.newLine()
                  }
                })
            }
          } finally writer.close()
        }

        println(exported)
        println(s"Lines in the file: ${Files.readAllLines(exportFile).size}")
        Files.readAllLines(exportFile).subList(0, 4).forEach(line => println(line))
      } finally Files.deleteIfExists(exportFile)
    }
}

Running the Examples​

All the code in this guide has runnable companions in the sql-examples project. To get them, clone the repository and change into the examples project:

git clone https://github.com/zio/zio-blocks.git
cd zio-blocks/sql-examples

Each example creates its own temporary database and files and deletes them when it finishes, so you can run them in any order. To compile all the examples without running them:

sbt compile

The Shared Setup​

The Sale row type, the 12,000 seeded sales and the temporary database every example uses are in this file:

sql-examples/src/main/scala/streamlargequeryresults/Sales.scala
sql-examples/src/main/scala/streamlargequeryresults/Sales.scala
package streamlargequeryresults

import java.nio.file.Files
import java.time.LocalDate

import zio.blocks.chunk.Chunk
import zio.blocks.schema.Schema
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results: shared setup
 *
 * The Sale row type with its Schema, the Table and Repo for it, the query the
 * examples stream from, the report helpers, and withSalesDb, which seeds 12,000
 * sales into a temporary SQLite file and deletes the file when the block ends.
 */
case class Sale(id: Long, title: String, quantity: Int, soldOn: LocalDate)

object Sale {
  given Schema[Sale] = Schema.derived
}

val saleTable = Table.derived[Sale]
val sales     = Repo.derived[Sale, Long]

val titles = Vector("Zionomicon", "Programming in Scala", "Functional Programming in Scala")

val saleCount = 12000

val allSales = sql"SELECT id, title, quantity, sold_on FROM sale ORDER BY id"

case class Report(copiesPerTitle: Map[String, Int], rows: Int, chunks: Int, largestChunk: Int)

val emptyReport = Report(Map.empty, 0, 0, 0)

def addChunk(report: Report, chunk: Chunk[Sale]): Report = {
  val copies = chunk.foldLeft(report.copiesPerTitle) { (totals, sale) =>
    totals.updated(sale.title, totals.getOrElse(sale.title, 0) + sale.quantity)
  }
  Report(copies, report.rows + chunk.size, report.chunks + 1, report.largestChunk.max(chunk.size))
}

def csvLine(sale: Sale): String =
  s"${sale.id},${sale.title},${sale.quantity},${sale.soldOn}"

def withSalesDb[A](f: JdbcTransactor => A): A = {
  val dbFile     = Files.createTempFile("bookstore-sales", ".db")
  val transactor = JdbcTransactor.fromUrl(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)
  try {
    transactor.transact {
      saleTable.createTable(SqlDialect.SQLite).update
      // In every 7 sales, 3 are of Zionomicon and 2 of each other title
      sales.insertBatch((1 to saleCount).map { i =>
        Sale(i.toLong, titles(i % 7 % 3), 1 + i % 4, LocalDate.of(2026, 1, 1).plusDays((i % 365).toLong))
      })
    }
    f(transactor)
  } finally Files.deleteIfExists(dbFile)
}

Aggregate a Large Table One Chunk at a Time​

sql-examples/src/main/scala/streamlargequeryresults/Step1AggregateExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step1AggregateExample.scala
package streamlargequeryresults

import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 1: Aggregate a Large Table One Chunk
 * at a Time
 *
 * Folds the 12,000 sales into copies sold per title with runFold, one chunk at
 * a time.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step1AggregateExample"
 */
object Step1AggregateExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val report: Either[Throwable, Report] =
        transactor.connect {
          allSales.queryStream[Sale].runFold(emptyReport)(addChunk)
        }

      report.foreach { r =>
        r.copiesPerTitle.toList.sorted.foreach((title, copies) => println(s"$title: $copies copies"))
        println(s"${r.rows} rows in ${r.chunks} chunks, at most ${r.largestChunk} rows per chunk")
      }
    }
}

It prints the copies sold per title and the chunk statistics:

sbt "runMain streamlargequeryresults.Step1AggregateExample"

Choose a Chunk Size​

sql-examples/src/main/scala/streamlargequeryresults/Step2ChunkSizeExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step2ChunkSizeExample.scala
package streamlargequeryresults

import scala.util.Try

import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 2: Choose a Chunk Size
 *
 * Runs the same fold with the default chunk size of 64, with 1,000 and with 10,
 * and shows that a chunk size of 0 is rejected up front.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step2ChunkSizeExample"
 */
object Step2ChunkSizeExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      def show(label: String, report: Either[Throwable, Report]): Unit =
        report.foreach(r =>
          println(s"$label: ${r.rows} rows in ${r.chunks} chunks, at most ${r.largestChunk} rows per chunk")
        )

      val defaultChunks: Either[Throwable, Report] =
        transactor.connect {
          allSales.queryStream[Sale].runFold(emptyReport)(addChunk)
        }

      val bigChunks: Either[Throwable, Report] =
        transactor.connect {
          allSales.queryChunked[Sale](1000).runFold(emptyReport)(addChunk)
        }

      val smallChunks: Either[Throwable, Report] =
        transactor.connect {
          allSales.queryChunked[Sale](10).runFold(emptyReport)(addChunk)
        }

      show("Default (64)", defaultChunks)
      show("1000", bigChunks)
      show("10", smallChunks)

      println(Try(transactor.connect(allSales.queryChunked[Sale](0))))
    }
}

It reads the table with three chunk sizes and shows the rejected size 0:

sbt "runMain streamlargequeryresults.Step2ChunkSizeExample"

Export Every Row to a CSV File​

sql-examples/src/main/scala/streamlargequeryresults/Step3ExportExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step3ExportExample.scala
package streamlargequeryresults

import java.nio.file.Files

import zio.blocks.chunk.Chunk
import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 3: Export Every Row to a CSV File
 *
 * Writes every sale to a CSV file through Sink.foreach, then prints the line
 * count and the first lines of the file.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step3ExportExample"
 */
object Step3ExportExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val exportFile = Files.createTempFile("sales-export", ".csv")
      try {
        val exported: Either[Throwable, Unit] = {
          val writer = Files.newBufferedWriter(exportFile)
          try {
            writer.write("id,title,quantity,sold_on")
            writer.newLine()
            transactor.connect {
              allSales
                .queryStream[Sale]
                .run(Sink.foreach[Chunk[Sale]] { chunk =>
                  chunk.foreach { sale =>
                    writer.write(csvLine(sale))
                    writer.newLine()
                  }
                })
            }
          } finally writer.close()
        }

        println(exported)
        println(s"Lines in the file: ${Files.readAllLines(exportFile).size}")
        Files.readAllLines(exportFile).subList(0, 4).forEach(line => println(line))
      } finally Files.deleteIfExists(exportFile)
    }
}

It writes sales.csv and prints its first lines:

sbt "runMain streamlargequeryresults.Step3ExportExample"

Filter and Reshape Rows on the Way Out​

sql-examples/src/main/scala/streamlargequeryresults/Step4TransformExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step4TransformExample.scala
package streamlargequeryresults

import java.nio.file.Files

import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 4: Filter and Reshape Rows on the
 * Way Out
 *
 * Exports only the Zionomicon sales as date,copies lines, written in batches of
 * 500 lines.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step4TransformExample"
 */
object Step4TransformExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val zionomiconFile = Files.createTempFile("zionomicon-sales", ".csv")
      try {
        val zionomiconBatches: Either[Throwable, Long] = {
          val writer = Files.newBufferedWriter(zionomiconFile)
          try
            transactor.connect {
              allSales
                .queryChunked[Sale](1000)
                .flatMap(chunk => Stream.fromChunk(chunk))
                .filter(_.title == "Zionomicon")
                .map(sale => s"${sale.soldOn},${sale.quantity}")
                .grouped(500)
                .tapEach(lines => lines.foreach { line => writer.write(line); writer.newLine() })
                .count
            }
          finally writer.close()
        }

        println(s"Batches written: $zionomiconBatches")
        println(s"Lines in the file: ${Files.readAllLines(zionomiconFile).size}")
        Files.readAllLines(zionomiconFile).subList(0, 3).forEach(line => println(line))
      } finally Files.deleteIfExists(zionomiconFile)
    }
}

It exports only the Zionomicon sales, in batches of 500 lines:

sbt "runMain streamlargequeryresults.Step4TransformExample"

Keep the Stream Inside the Connection Block​

sql-examples/src/main/scala/streamlargequeryresults/Step5ScopeExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step5ScopeExample.scala
package streamlargequeryresults

import zio.blocks.chunk.Chunk
import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 5: Keep the Stream Inside the
 * Connection Block
 *
 * Returns a stream out of its connect block and fails on the first pull, then
 * runs the terminal operation inside the block.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step5ScopeExample"
 */
object Step5ScopeExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val escaped: Stream[Throwable, Chunk[Sale]] =
        transactor.connect(allSales.queryStream[Sale])

      println(escaped.runFold(0)((rows, chunk) => rows + chunk.size))

      println(transactor.connect(allSales.queryStream[Sale].runFold(0)((rows, chunk) => rows + chunk.size)))
    }
}

It runs a stream outside its connect block, then inside it:

sbt "runMain streamlargequeryresults.Step5ScopeExample"

Handle Query and Decoding Errors​

sql-examples/src/main/scala/streamlargequeryresults/Step6ErrorsExample.scala
sql-examples/src/main/scala/streamlargequeryresults/Step6ErrorsExample.scala
package streamlargequeryresults

import zio.blocks.streams.*
import zio.blocks.sql.*

/**
 * How to Stream Large Query Results — Step 6: Handle Query and Decoding Errors
 *
 * Shows the Left produced by a misspelled column and by a row that can't be
 * decoded.
 *
 * Run with: cd sql-examples && sbt "runMain
 * streamlargequeryresults.Step6ErrorsExample"
 */
object Step6ErrorsExample {
  def main(args: Array[String]): Unit =
    withSalesDb { transactor =>
      val badColumn = transactor.connect {
        sql"SELECT id, titel, quantity, sold_on FROM sale"
          .queryStream[Sale]
          .runFold(0)((rows, chunk) => rows + chunk.size)
      }

      badColumn match {
        case Left(error) => println(s"Query failed: ${error.getMessage}")
        case Right(rows) => println(s"Read $rows rows")
      }

      val badRow = transactor.connect {
        sql"SELECT id, title, quantity, title AS sold_on FROM sale"
          .queryStream[Sale]
          .runFold(0)((rows, chunk) => rows + chunk.size)
      }

      badRow.left.foreach(error => println(s"${error.getClass.getSimpleName}: ${error.getMessage}"))
    }
}

It prints the Left for a misspelled column and for a row that can't be decoded:

sbt "runMain streamlargequeryresults.Step6ErrorsExample"

The Complete Example​

sbt "runMain streamlargequeryresults.CompleteExample"

Going Further​