Added fine tuning of query via internal result chunking

Author: Alex Golubov

README.md

@@ -62,5 +62,35 @@ Sqlitex.Server.query(Golf.DB,
                       ORDER BY g.played_at DESC LIMIT 10")
 ```
 
+# Configuration
+
+Sqlitex uses the Erlang library [esqlite](https://github.com/mmzeeman/esqlite)
+which accepts a timeout parameter for almost all interactions with the database.
+The default value for this timeout is 5000 ms. Many functions in Sqlitex accept
+a `:db_timeout` option that is passed on to the esqlite calls and also defaults
+to 5000 ms. If required, this default value can be overridden globally with the
+following in your `config.exs`:
+
+```
+config :sqlitex, db_timeout: 10_000 # or other positive integer number of ms
+```
+
+Another esqlite parameter is :db_chunk_size.
+This is a count of rows to read from native sqlite and send to erlang process in one bulk.
+For example, the table `mytable` has 1000 rows. We make the query to get all rows with `db_chunk_size: 500` parameter:
+```
+Sqlitex.query(db, "select * from mytable", db_chunk_size: 500)
+```
+in this case all rows will be passed from native sqlite OS thread to the erlang process in two passes.
+Each pass will contain 500 rows.  
+This parameter decrease overhead of transmitting rows from native OS sqlite thread to the erlang process by
+chunking list of result rows.  
+Please, decrease this value if rows are heavy. Default value is 5000.  
+If you in doubt what to do with this parameter, please, do nothing. Default value is ok.
+```
+config :sqlitex, db_chunk_size: 500 # if most of the database rows are heavy
+```
+
+
 # Looking for Ecto?
 Check out the [SQLite Ecto2 adapter](https://github.com/Sqlite-Ecto/sqlite_ecto2)

lib/sqlitex.ex

@@ -37,24 +37,38 @@ defmodule Sqlitex do
   ```
   config :sqlitex, db_timeout: 10_000 # or other positive integer number of ms
   ```
+
+  Another esqlite parameter is :db_chunk_size.
+  This is a count of rows to read from native sqlite and send to erlang process in one bulk.
+  For example, the table `mytable` has 1000 rows. We make the query to get all rows with `db_chunk_size: 500` parameter:
+  ```
+  Sqlitex.query(db, "select * from mytable", db_chunk_size: 500)
+  ```
+  in this case all rows will be passed from native sqlite OS thread to the erlang process in two passes.
+  Each pass will contain 500 rows.  
+  This parameter decrease overhead of transmitting rows from native OS sqlite thread to the erlang process by
+  chunking list of result rows.  
+  Please, decrease this value if rows are heavy. Default value is 5000.  
+  If you in doubt what to do with this parameter, please, do nothing. Default value is ok.
+  ```
+  config :sqlitex, db_chunk_size: 500 # if most of the database rows are heavy
+  ```
   """
 
   alias Sqlitex.Config
 
   @spec close(connection) :: :ok
   @spec close(connection, Keyword.t) :: :ok
   def close(db, opts \\ []) do
-    timeout = Keyword.get(opts, :db_timeout, Config.db_timeout())
-    :esqlite3.close(db, timeout)
+    :esqlite3.close(db, Config.db_timeout(opts))
   end
 
   @spec open(charlist | String.t) :: {:ok, connection} | {:error, {atom, charlist}}
   @spec open(charlist | String.t, Keyword.t) :: {:ok, connection} | {:error, {atom, charlist}}
   def open(path, opts \\ [])
   def open(path, opts) when is_binary(path), do: open(string_to_charlist(path), opts)
   def open(path, opts) do
-    timeout = Keyword.get(opts, :db_timeout, Config.db_timeout())
-    :esqlite3.open(path, timeout)
+    :esqlite3.open(path, Config.db_timeout(opts))
   end
 
   def with_db(path, fun, opts \\ []) do
@@ -78,8 +92,7 @@ defmodule Sqlitex do
   """
   @spec set_update_hook(connection, pid, Keyword.t()) :: :ok | {:error, term()}
   def set_update_hook(db, pid, opts \\ []) do
-    timeout = Keyword.get(opts, :db_timeout, Config.db_timeout())
-    :esqlite3.set_update_hook(pid, db, timeout)
+    :esqlite3.set_update_hook(pid, db, Config.db_timeout(opts))
   end
 
   @doc """
@@ -92,8 +105,7 @@ defmodule Sqlitex do
   @spec exec(connection, string_or_charlist) :: :ok | sqlite_error
   @spec exec(connection, string_or_charlist, Keyword.t) :: :ok | sqlite_error
   def exec(db, sql, opts \\ []) do
-    timeout = Keyword.get(opts, :db_timeout, Config.db_timeout())
-    :esqlite3.exec(sql, db, timeout)
+    :esqlite3.exec(sql, db, Config.db_timeout(opts))
   end
 
   @doc "A shortcut to `Sqlitex.Query.query/3`"

lib/sqlitex/config.ex

@@ -1,7 +1,22 @@
 defmodule Sqlitex.Config do
   @moduledoc false
 
-  def db_timeout do
-    Application.get_env(:sqlitex, :db_timeout, 5_000)
+  @default_call_timeout 5_000
+  @default_db_timeout 5_000
+  @default_db_chunk_size 5_000
+
+  def call_timeout(opts \\ []) do
+    Keyword.get(opts, :call_timeout,
+      Keyword.get(opts, :timeout, # backward compatibility with the :timeout parameter
+        Application.get_env(:sqlitex, :call_timeout, @default_call_timeout)))
+  end
+
+  def db_timeout(opts \\ []) do
+    Keyword.get(opts, :db_timeout, Application.get_env(:sqlitex, :db_timeout, @default_db_timeout))
   end
+
+  def  db_chunk_size(opts \\ []) do
+    Keyword.get(opts, :db_chunk_size, Application.get_env(:sqlitex, :db_chunk_size, @default_db_chunk_size))
+  end
+
 end

lib/sqlitex/query.ex

@@ -24,6 +24,8 @@ defmodule Sqlitex.Query do
   * `into` - The collection to put results into.  This defaults to a list.
   * `db_timeout` - The timeout (in ms) to apply to each of the underlying SQLite operations. Defaults
     to `Application.get_env(:sqlitex, :db_timeout)` or `5000` ms if not configured.
+  * `db_chunk_size` - The number of rows to read from native sqlite and send to erlang process in one bulk.
+    Defaults to `Application.get_env(:sqlitex, :db_chunk_size)` or `5000` ms if not configured.
 
   ## Returns
   * {:ok, [results...]} on success
@@ -39,7 +41,7 @@ defmodule Sqlitex.Query do
   def query(db, sql, opts \\ []) do
     with {:ok, stmt} <- Statement.prepare(db, sql, opts),
          {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), opts),
-         {:ok, res} <- Statement.fetch_all(stmt, Keyword.get(opts, :db_timeout, 5_000), Keyword.get(opts, :into, [])),
+         {:ok, res} <- Statement.fetch_all(stmt, opts),
     do: {:ok, res}
   end
 
@@ -73,6 +75,8 @@ defmodule Sqlitex.Query do
     to bind as a list.
   * `db_timeout` - The timeout (in ms) to apply to each of the underlying SQLite operations. Defaults
     to `Application.get_env(:sqlitex, :db_timeout)` or `5000` ms if not configured.
+  * `db_chunk_size` - The number of rows to read from native sqlite and send to erlang process in one bulk.
+    Defaults to `Application.get_env(:sqlitex, :db_chunk_size)` or `5000` ms if not configured.
 
   ## Returns
   * {:ok, %{rows: [[1, 2], [2, 3]], columns: [:a, :b], types: [:INTEGER, :INTEGER]}} on success
@@ -84,7 +88,7 @@ defmodule Sqlitex.Query do
   def query_rows(db, sql, opts \\ []) do
     with {:ok, stmt} <- Statement.prepare(db, sql, opts),
          {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), opts),
-         {:ok, rows} <- Statement.fetch_all(stmt, Keyword.get(opts, :db_timeout, 5_000), :raw_list),
+         {:ok, rows} <- Statement.fetch_all(stmt, Keyword.put(opts, :into, :raw_list)),
     do: {:ok, %{rows: rows, columns: stmt.column_names, types: stmt.column_types}}
   end
 

lib/sqlitex/server.ex

@@ -58,87 +58,107 @@ defmodule Sqlitex.Server do
   - `stmt_cache_size: (positive_integer)` to override the default limit (20) of statements
     that are cached when calling `prepare/3`.
   - `db_timeout: (positive_integer)` to override `:esqlite3`'s default timeout of 5000 ms for
-    interactions with the database. This can also be set in `config.exs` as
-    `config :sqlitex, db_timeout: 5_000`.
+    interactions with the database. This can also be set in `config.exs` as `config :sqlitex, db_timeout: 5_000`.
+  - `db_chunk_size: (positive_integer)` to override `:esqlite3`'s default chunk_size of 5000 rows
+    to read from native sqlite and send to erlang process in one bulk.
+    This can also be set in `config.exs` as `config :sqlitex, db_chunk_size: 5_000`.
   """
   def start_link(db_path, opts \\ []) do
     stmt_cache_size = Keyword.get(opts, :stmt_cache_size, 20)
-    timeout = Keyword.get(opts, :db_timeout, Config.db_timeout())
-    GenServer.start_link(__MODULE__, {db_path, stmt_cache_size, timeout}, opts)
+    config = [
+      db_timeout: Config.db_timeout(opts),
+      db_chunk_size: Config.db_chunk_size(opts)
+    ]
+    GenServer.start_link(__MODULE__, {db_path, stmt_cache_size, config}, opts)
   end
 
   ## GenServer callbacks
 
-  def init({db_path, stmt_cache_size, timeout})
+  def init({db_path, stmt_cache_size, config})
     when is_integer(stmt_cache_size) and stmt_cache_size > 0
   do
-    case Sqlitex.open(db_path, [db_timeout: timeout]) do
-      {:ok, db} -> {:ok, {db, __MODULE__.StatementCache.new(db, stmt_cache_size), timeout}}
+    case Sqlitex.open(db_path, config) do
+      {:ok, db} -> {:ok, {db, __MODULE__.StatementCache.new(db, stmt_cache_size), config}}
       {:error, reason} -> {:stop, reason}
     end
   end
 
-  def handle_call({:exec, sql}, _from, {db, stmt_cache, timeout}) do
-    result = Sqlitex.exec(db, sql, [db_timeout: timeout])
-    {:reply, result, {db, stmt_cache, timeout}}
+  def handle_call({:exec, sql}, _from, {db, stmt_cache, config}) do
+    result = Sqlitex.exec(db, sql, config)
+    {:reply, result, {db, stmt_cache, config}}
   end
 
-  def handle_call({:query, sql, opts}, _from, {db, stmt_cache, timeout}) do
-    case query_impl(sql, opts, stmt_cache, timeout) do
-      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, timeout}}
-      err -> {:reply, err, {db, stmt_cache, timeout}}
+  def handle_call({:query, sql, opts}, _from, {db, stmt_cache, config}) do
+    case query_impl(sql, stmt_cache, Keyword.merge(config, opts)) do
+      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, config}}
+      err -> {:reply, err, {db, stmt_cache, config}}
     end
   end
 
-  def handle_call({:query_rows, sql, opts}, _from, {db, stmt_cache, timeout}) do
-    case query_rows_impl(sql, opts, stmt_cache, timeout) do
-      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, timeout}}
-      err -> {:reply, err, {db, stmt_cache, timeout}}
+  def handle_call({:query_rows, sql, opts}, _from, {db, stmt_cache, config}) do
+    case query_rows_impl(sql, stmt_cache, Keyword.merge(config, opts)) do
+      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, config}}
+      err -> {:reply, err, {db, stmt_cache, config}}
     end
   end
 
-  def handle_call({:prepare, sql}, _from, {db, stmt_cache, timeout}) do
-    case prepare_impl(sql, stmt_cache, timeout) do
-      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, timeout}}
-      err -> {:reply, err, {db, stmt_cache, timeout}}
+  def handle_call({:prepare, sql}, _from, {db, stmt_cache, config}) do
+    case prepare_impl(sql, stmt_cache, config) do
+      {:ok, result, new_cache} -> {:reply, {:ok, result}, {db, new_cache, config}}
+      err -> {:reply, err, {db, stmt_cache, config}}
     end
   end
 
-  def handle_call({:create_table, name, table_opts, cols}, _from, {db, stmt_cache, timeout}) do
-    result = Sqlitex.create_table(db, name, table_opts, cols, [db_timeout: timeout])
-    {:reply, result, {db, stmt_cache, timeout}}
+  def handle_call({:create_table, name, table_opts, cols}, _from, {db, stmt_cache, config}) do
+    result = Sqlitex.create_table(db, name, table_opts, cols, config)
+    {:reply, result, {db, stmt_cache, config}}
   end
 
-  def handle_call({:set_update_hook, pid, opts}, _from, {db, stmt_cache, timeout}) do
-    result = Sqlitex.set_update_hook(db, pid, opts)
-    {:reply, result, {db, stmt_cache, timeout}}
+  def handle_call({:set_update_hook, pid, opts}, _from, {db, stmt_cache, config}) do
+    result = Sqlitex.set_update_hook(db, pid, Keyword.merge(config, opts))
+    {:reply, result, {db, stmt_cache, config}}
   end
 
-  def handle_cast(:stop, {db, stmt_cache, timeout}) do
-    {:stop, :normal, {db, stmt_cache, timeout}}
+  def handle_cast(:stop, {db, stmt_cache, config}) do
+    {:stop, :normal, {db, stmt_cache, config}}
   end
 
-  def terminate(_reason, {db, _stmt_cache, _timeout}) do
-    Sqlitex.close(db)
+  def terminate(_reason, {db, _stmt_cache, config}) do
+    Sqlitex.close(db, config)
     :ok
   end
 
   ## Public API
 
+  @doc """
+  Same as `Sqlitex.exec/3` but using the shared db connections saved in the GenServer state.
+
+  Returns the results otherwise.
+  """
   def exec(pid, sql, opts \\ []) do
-    GenServer.call(pid, {:exec, sql}, timeout(opts))
+    GenServer.call(pid, {:exec, sql}, Config.call_timeout(opts))
   end
 
+  @doc """
+  Same as `Sqlitex.Query.query/3` but using the shared db connections saved in the GenServer state.
+
+  Returns the results otherwise.
+  """
   def query(pid, sql, opts \\ []) do
-    GenServer.call(pid, {:query, sql, opts}, timeout(opts))
+    GenServer.call(pid, {:query, sql, opts}, Config.call_timeout(opts))
   end
 
+  @doc """
+  Same as `Sqlitex.Query.query_rows/3` but using the shared db connections saved in the GenServer state.
+
+  Returns the results otherwise.
+  """
   def query_rows(pid, sql, opts \\ []) do
-    GenServer.call(pid, {:query_rows, sql, opts}, timeout(opts))
+    GenServer.call(pid, {:query_rows, sql, opts}, Config.call_timeout(opts))
   end
 
   def set_update_hook(server_pid, notification_pid, opts \\ []) do
-    GenServer.call(server_pid, {:set_update_hook, notification_pid, opts}, timeout(opts))
+    GenServer.call(server_pid, {:set_update_hook, notification_pid, opts}, Config.call_timeout(opts))
   end
 
   @doc """
@@ -160,7 +180,7 @@ defmodule Sqlitex.Server do
   could not be prepared.
   """
   def prepare(pid, sql, opts \\ []) do
-    GenServer.call(pid, {:prepare, sql}, timeout(opts))
+    GenServer.call(pid, {:prepare, sql}, Config.call_timeout(opts))
   end
 
   def create_table(pid, name, table_opts \\ [], cols) do
@@ -173,30 +193,24 @@ defmodule Sqlitex.Server do
 
   ## Helpers
 
-  defp query_impl(sql, opts, stmt_cache, db_timeout) do
-    db_opts = [db_timeout: db_timeout]
-
-    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, db_opts),
-         {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), db_opts),
-         {:ok, rows} <- Statement.fetch_all(stmt, Keyword.get(opts, :db_timeout, 5_000), Keyword.get(opts, :into, [])),
+  defp query_impl(sql, stmt_cache, opts) do
+    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, opts),
+         {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), opts),
+         {:ok, rows} <- Statement.fetch_all(stmt, opts),
     do: {:ok, rows, new_cache}
   end
 
-  defp query_rows_impl(sql, opts, stmt_cache, db_timeout) do
-    db_opts = [db_timeout: db_timeout]
-
-    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, db_opts),
-         {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), db_opts),
-         {:ok, rows} <- Statement.fetch_all(stmt, Keyword.get(opts, :db_timeout, 5_000), :raw_list),
+  defp query_rows_impl(sql, stmt_cache, opts) do
+    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, opts),
+         {:ok, stmt} <- Statement.bind_values(stmt, Keyword.get(opts, :bind, []), opts),
+         {:ok, rows} <- Statement.fetch_all(stmt, Keyword.put(opts, :into, :raw_list)),
     do: {:ok,
          %{rows: rows, columns: stmt.column_names, types: stmt.column_types},
          new_cache}
   end
 
-  defp prepare_impl(sql, stmt_cache, db_timeout) do
-    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, [db_timeout: db_timeout]),
+  defp prepare_impl(sql, stmt_cache, opts) do
+    with {%Cache{} = new_cache, stmt} <- Cache.prepare(stmt_cache, sql, opts),
     do: {:ok, %{columns: stmt.column_names, types: stmt.column_types}, new_cache}
   end
-
-  defp timeout(kwopts), do: Keyword.get(kwopts, :timeout, 5000)
 end