Specs

iterate_parallel_scan(
  instance(),
  table_name(),
  index_name(),
  fun :: (term() -> term()),
  options()
) :: term()

A simple wrapper of stream_parallel_scan/4 to take care OTSSessionExpired error with retry, make parallel scan as a stream that applies the given function to the complete result of scan query.

In general, recommend to use this function for the common use case of parallel scan.

Options

• :scan_query, required, the main option to use query.
  • :query, required, bind to the query functions, the same as query option of search/5.
  • :limit, optional, the limited size of query, defaults to 2000, the maximum value of limit is 2000.
• :columns_to_get, optional, fetch the special fields, by default it returns all fields of the search index, here are available options:
  • :all_from_index, return all attribute column fields of search index;
  • :none, do not return any attribute column fields;
  • ["field1", "field2"], specifies the expected return attribute column fields.
• :timeout, optional, the :timeout option of Task.async_stream/3, defaults to :infinity.

Example

def iterate_stream(stream) do
  Enum.map(stream, fn
    {:ok, response} ->
      response
    {:error, error} ->
      error
  end)
end

iterate_parallel_scan(
  "table",
  "index",
  &iterate_stream/1,
  scan_query: [
    query: match_query("is_actived", "true"),
    limit: 1000
  ],
  columns_to_get: ["is_actived", "name", "score"]
)

Specs

iterate_parallel_scan(
  instance(),
  table_name(),
  index_name(),
  mod :: module(),
  fun :: atom(),
  args :: [term()],
  options()
) :: term()

A simple wrapper of stream_parallel_scan/4 to take care OTSSessionExpired error with retry, make parallel scan as a stream that applies the given function from module with the list of arguments args to the complete result of scan query.

In general, recommend to use this function for the common use case of parallel scan.

Options

Please see options of iterate_parallel_scan/5.

Example

defmodule StreamHandler do
  def iterate_stream(stream) do
    Enum.map(stream, fn
      {:ok, response} ->
        response
      {:error, error} ->
        error
    end)
  end
end

iterate_parallel_scan(
  "table",
  "index",
  StreamHandler,
  :iterate_stream,
  [],
  scan_query: [
    query: match_query("is_actived", "true"),
    limit: 1000
  ],
  columns_to_get: ["field1", "field2"]
)