View Source Streaming, batches and transactions

Pages

query/2 and scan/2 return one %Dynamo.Page{}:

{:ok, page} = Repo.query(%Order{user_id: "u1"}, limit: 100)
page.items                # [%Order{}]
page.last_evaluated_key   # nil when done
page.count                # items in this page
Enum.map(page, & &1.total) # Page is Enumerable

{:ok, next} = Repo.query(%Order{user_id: "u1"}, limit: 100, start_key: page.last_evaluated_key)

Streams

Repo.stream/2 is a lazy Stream over all pages. Pass a struct to query or a schema module to scan.

Repo.stream(%Order{user_id: "u1"}, page_size: 200)
|> Stream.filter(&(&1.status == "open"))
|> Stream.map(&notify/1)
|> Stream.run()

Repo.stream(User)
|> Stream.chunk_every(25)
|> Enum.each(fn chunk -> Repo.batch_write!(Enum.map(chunk, &migrate/1)) end)

Errors inside a stream are raised as %Dynamo.Error{} – wrap in try or let the calling process crash and be restarted.

Parallel scan

segments: runs one task per segment under the repo's Task.Supervisor; max_concurrency: (default segments) bounds how many fetch at once. Items arrive interleaved; each task only fetches its next page after the previous one was consumed, so memory stays bounded by segments × page_size.

Repo.stream(User, segments: 8, max_concurrency: 4, page_size: 500)
|> Flow.from_enumerable(max_demand: 200)      # optional: Flow for CPU-heavy work
|> Flow.map(&expensive/1)
|> Enum.to_list()

Stopping early (Enum.take/2, an exception) stops the segment tasks.

query_all / scan_all

Convenience for small result sets. :limit bounds the total.

{:ok, orders} = Repo.query_all(%Order{user_id: "u1"}, limit: 500)

Batches

# BatchWriteItem: put structs and {:delete, struct}; chunked into 25 per request.
# UnprocessedItems are retried with backoff (:batch_retries, default 5).
{:ok, %{unprocessed: []}} = Repo.batch_write(new_users ++ Enum.map(stale, &{:delete, &1}))

# BatchGetItem: keys of one schema, chunked into 100; UnprocessedKeys retried.
{:ok, %{items: users, unprocessed_keys: []}} = Repo.batch_get(keys, consistent: true)

Any non-retryable error (:access_denied, :resource_not_found, …) aborts the batch and is returned. A non-empty unprocessed/unprocessed_keys after all retries means sustained throttling – back off at the application level.

Transactions

Repo.transaction(
  [
    {:check, %Account{id: "src"}, "#b >= :amt", names: %{"#b" => "balance"}, values: %{amt: 100}},
    {:update, %Account{id: "src"}, [balance: {:decrement, 100}]},
    {:update, %Account{id: "dst"}, [balance: {:increment, 100}]},
    {:put, %Transfer{id: "t1", amount: 100}, condition: "attribute_not_exists(pk)"}
  ],
  client_request_token: "transfer-t1"
)

Wait – DynamoDB rejects two operations on the same item in one transaction, and so does Dynamo.Request.transact_write/2 (before sending). Fold the check into the update's condition: instead:

{:update, %Account{id: "src"}, [balance: {:decrement, 100}], condition: "#b >= :amt", names: %{"#b" => "balance"}, values: %{amt: 100}}

Use client_request_token: for idempotency when retrying after a timeout – DynamoDB de-duplicates for 10 minutes. This matters because the client itself retries transport timeouts: a plain Repo.update(acct, balance: {:decrement, 50}) that times out may be applied twice. Either put it in a transaction with a token, or add a version guard:

Repo.update(acct, [balance: {:decrement, 50}, version: {:increment, 1}], condition: [version: acct.version])

On failure:

{:error, %Dynamo.Error{type: :transaction_canceled, cancellation_reasons: reasons}}
# reasons: one per operation, e.g. [%{code: "None"}, %{code: "ConditionalCheckFailed", message: "..."}]

Cost reminder

Scans read every item in the table regardless of filters. Streaming bounds memory, not read capacity. Prefer queries; reserve scans for migrations and exports.