вроде ответом nda я не нарушу:
генерю сиды на основе данных

Дамп не катит?)

не, там другие юзкейсы

мб не стоит в этом случае использовать ассоциации, а заджоинить всё что нужно в запрос и его стримить?

возможно

только еще стоит учитывать что Flow не работает, с ектовским стримом. Решается всё самописным Stream.recource

Можно сделать цикл с постоянно увеличивающимся оффсетом

и если не ошибаюсь ещё repo.stream еще и в транзакцию нужно пихать

так и делаю

лучше курсор

def all_stream(query, opts \\ []) do
       chunk_size  = Keyword.get(opts, :chunk_size, 500)
       offset      = Keyword.get(opts, :offset, 0)

       Stream.resource(
         fn -> {query, offset} end,

         fn {query, last_id} ->
           list = query
             |> where([e], field(e, ^@all_stream_pk_id) > ^last_id)
             |> limit(^chunk_size)
             |> order_by([e], asc: field(e, ^@all_stream_pk_id))
             |> all

           case List.last(list) do
             %{@all_stream_pk_id => id}  -> {[list], {query, id}}
             nil                         -> {:halt, {query, last_id}}
           end
         end,

         fn _ -> [] end
       )
     end

Это ещё вопрос. На офсетах можно распараллелить, плюс в случае ошибки отловить и повторить

ну @all_stream_pk_id убери там под себя

а у оффсета O(n)

Хотя распараллелится или нет - я не уверен

Может там на чтение тоже лочит

разпаралелится)

А при генерации сидов время вообще не важно. Это не бизнесс процесс

тогда и параллелить не нужно

Ну как бы да, но типа почему бы и нет?)