Каков правильный способ работы с результатами slick 3.0.0 и Postgresql?

Я пытаюсь понять, как работать с гладкой потоковой передачей. Я использую slick 3.0.0 с драйвером postgres

Ситуация следующая: сервер должен предоставить клиентские последовательности данных, разделенных на куски, ограниченные размером (в байтах). Итак, я написал следующий гладкий запрос:

val sequences = TableQuery[Sequences]
def find(userId: Long, timestamp: Long) = sequences.filter(s ⇒ s.userId === userId && s.timestamp > timestamp).sortBy(_.timestamp.asc).result
val seq = db.stream(find(0L, 0L))

Я объединил seq с akka-streams Source, написал пользовательский PushPullStage, который ограничивает размер данных (в байтах) и заканчивается вверх по потоку, когда он достигает ограничения по размеру. Он работает отлично. Проблема в том, что когда я просматриваю журналы postgres, я вижу такой запрос select * from sequences where user_id = 0 and timestamp > 0 order by timestamp;

Итак, на первый взгляд кажется, что много запросов (и ненужных) запросов на базу данных происходит, только для использования нескольких байтов в каждом запросе. Каков правильный способ создания потоковой передачи с помощью Slick, чтобы минимизировать запросы к базам данных и наилучшим образом использовать данные, передаваемые в каждом запросе?

Ответы

Ответ 1

"Правильный способ" для потоковой передачи с помощью Slick и Postgres включает в себя три вещи:

  • Должен использовать db.stream()

  • Должен отключить autoCommit в JDBC-драйвере. Один из способов - выполнить запрос в транзакции путем суффикса .transactionally.

  • Должно установить fetchSize как что-то другое, чем 0, или postgres будет толкать весь resultSet клиенту за один раз.

Пример:

DB.stream(
  find(0L, 0L)
    .transactionally
    .withStatementParameters(fetchSize = 1000)
).foreach(println)

Полезные ссылки:

https://github.com/slick/slick/issues/1038

https://github.com/slick/slick/issues/809