Snowflex is an Ecto adapter for Snowflake using Snowflake's SQL API.

This adapter implements the following Ecto behaviours:

• Ecto.Adapter - Core adapter functionality
• Ecto.Adapter.Queryable - Query execution and streaming
• Ecto.Adapter.Schema - Schema operations (insert, update, delete)

Requirements

If using the provided Snowflex.Transport.Http transport, the only currently supported authentication method is keypair.

You only need to supply your private key — the public-key fingerprint is derived from it automatically. You may still pass public_key_fingerprint explicitly if you prefer (see Snowflake's instructions for obtaining it).

Configuration

Configure the adapter in your application:

The adapter supports configurable transport implementations through the :transport option. By default, it uses Snowflex.Transport.Http for REST API communication with Snowflake.

config :my_app, MyApp.Repo,
  adapter: Snowflex,
  transport: Snowflex.Transport.Http,  # Optional, defaults to Http
  # Additional options passed to the transport
  account_name: "your-account",
  username: "your_username",
  private_key_path: "path/to/key.pem"  # Path to private key file
  # OR alternatively, provide the private key as a string:
  # private_key_from_string: "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
  # public_key_fingerprint is optional — derived from the private key when omitted.

You may supply other transports that conform to the Snowflex.Transport behaviour.

For additional configuration options of the provided Snowflex.Transport.Http transport, see it's documentation.

Query Tagging

All queries can be tagged for better observability and tracking in Snowflake. Tags are passed as options to any Repo function call:

# Tag a query with a UUID
MyRepo.all(query, query_tag: Ecto.UUID.generate())

# Tag a query with a custom identifier
iex> MyRepo.insert(changeset, query_tag: "user_registration_abc")

Query tags are visible in Snowflake's query history and can be used for:

• Tracking query origins
• Monitoring specific operations
• Debugging performance issues
• Auditing database access

Error Logging and Metadata

Snowflex automatically enriches error logs with contextual metadata to help with debugging and monitoring. When queries fail, detailed information about the error context is added to Logger metadata.

Available Metadata Keys

The following metadata keys are set when errors occur:

• :snowflex_account_name - Your Snowflake account identifier
• :snowflex_username - The username used for the connection
• :snowflex_query_id - Snowflake's unique query identifier
• :snowflex_statement - The SQL statement that failed
• :snowflex_warehouse - The default warehouse used for the query (if configured)
• :snowflex_role - The default role used for the query (if configured)
• :snowflex_database - The default database used for the query (if configured)
• :snowflex_schema - The default schema used for the query (if configured)

Configuring Logger

To capture this metadata in your logs, see Logger's Metadata Section for more information on how to configure Logger.

Example Error Log

With metadata configured, error logs will include context like:

snowflex_account_name=my_account snowflex_username=my_user snowflex_warehouse=MY_WH snowflex_database=MY_DB snowflex_statement=SELECT * FROM users [error] Snowflake query failed: [529] "Server too busy. Please retry."

This enriched logging makes it easier to:

• Track which account/warehouse/role errors are occurring in
• Correlate errors with specific SQL statements
• Debug issues across different Snowflake environments
• Monitor query failures in production

Type Support

The adapter supports the following type conversions:

From Snowflake to Ecto

• :integer - Integer values
• :decimal - Decimal values
• :float - Float values
• :date - Date values
• :time - Time values
• :utc_datetime - UTC datetime values
• :naive_datetime - Naive datetime values
• :binary - Binary data (as hex strings)
• :map and {:map, inner_type} - Stored as a VARIANT
• {:array, inner_type} - Stored as a VARIANT

From Ecto to Snowflake

• :binary - Binary data (as hex strings)
• :decimal - Decimal values
• :float - Float values
• :date - Date values
• :time - Time values
• :utc_datetime - UTC datetime values
• :naive_datetime - Naive datetime values
• :map and {:map, inner_type} - Stored as a VARIANT
• {:array, inner_type} - Stored as a VARIANT

Limitations/Considerations

Transactions

Snowflex does not support multi-statement transactions. The reason for this is the Snowflake SQL API does not support multi-request transactions. That is to say, all statements in a transaction must be sent in the same request. Because it is a common pattern to rely on the results of a previous statement in further downstream queries in the same transaction (e.g. Ecto.Multi), this limitation in the SQL API meant that we either needed to provide a potentially unintuitive use case, or just not support them at all.

Multiple Statements

Snowflex supports submitting multiple statements in the same query and will return the results of each statement packed into an array.

This can be useful when statements you want to execute need to occur inside of the same transaction (e.g. you need to leverage a temporary table) This is possible using both the query and query_many functions.

iex> Repo.query("SELECT 1; SELECT 2;")

{:ok,
 [
   %Snowflex.Result{rows: [[1]]},
   %Snowflex.Result{rows: [[2]]}
 ]
}

Streaming

Ecto.Repo.stream/2 streams a queryable lazily, fetching Snowflake result partitions on demand instead of gathering the full result set in memory. Because the connection must stay checked out for as long as the stream is being consumed — and Snowflake has no transactions — the enumeration must run inside Ecto.Repo.checkout/2 (the direct analogue of ecto_sql's requirement that streams run inside a transaction):

MyRepo.checkout(fn ->
  BigSchema
  |> where([b], b.inserted_at >= ^cutoff)
  |> MyRepo.stream()
  |> Stream.each(&process_row/1)
  |> Stream.run()
end, timeout: :timer.minutes(30))

For raw SQL, use Snowflex.stream_query/5 — the same cursor machinery with a function-scoped connection (no explicit checkout needed; it also reuses the connection when called inside checkout/2). Each element is a Snowflex.Result holding one partition of rows:

Snowflex.stream_query(MyRepo, "SELECT * FROM big_table", [], [timeout: :timer.minutes(30)], fn stream ->
  stream
  |> Stream.flat_map(fn %Snowflex.Result{rows: rows} -> rows || [] end)
  |> Stream.each(&process_row/1)
  |> Stream.run()
end)

When streaming with Snowflex.Transport.Http, keep in mind that Snowflake dictates the number of partitions returned. This is different than a normal TCP protocol like Postgrex, where the stream iterates one row at a time: partitions are fetched whole, so memory usage is bounded by the partition size rather than by a row count. Pass a :timeout that covers statement execution plus however long consumption takes, and prefer Ecto.Repo.all/2 when you want the entire result set anyway — the eager path fetches partitions in parallel and is faster for full materialization.

Asynchronous Statements

Ordinary queries occupy a connection for as long as the statement runs. When you only need to start a statement — a stored procedure, a long-running MERGE, a warehouse maintenance task — Snowflex.submit_async/4 submits it and returns Snowflake's statement handle as soon as the submission round-trip completes, so no pool slot is held while the statement runs:

{:ok, handle} = Snowflex.submit_async(MyRepo, "CALL rebuild_scenarios(?)", [scenario_id])

Later, retrieve the result with the handle. fetch_result/3 answers {:ok, :running} until the statement finishes, so you can poll with it alone:

case Snowflex.fetch_result(MyRepo, handle) do
  {:ok, :running} -> :still_running
  {:ok, %Snowflex.Result{rows: rows}} -> rows
  {:error, error} -> Logger.error("statement failed: #{Exception.message(error)}")
end

Rows decode exactly as they would from MyRepo.query/3. If you only care whether the statement finished and not about its rows, statement_status/3 returns {:ok, :running | :succeeded}. To stop a statement, cancel_statement/3 returns :ok:

{:ok, :running} = Snowflex.statement_status(MyRepo, handle)
:ok = Snowflex.cancel_statement(MyRepo, handle)

Three things to keep in mind. Nothing on the Elixir side waits on a submitted statement, so the only thing bounding it is Snowflake's own statement timeout — prefer statements that are safe to leave running unattended. Snowflake retains results for a limited window, so a handle fetched long after completion returns an error rather than rows. And fetch_result/3 materializes the entire result set (fetching partitions in parallel), so stream large results with Snowflex.stream_query/5 instead.

Migrations

Migrations are not currently supported by Snowflex.

Migration from ODBC

Previous versions of Snowflex relied on Erlang :odbc. While very stable and battle tested, has always suffered from the idiosyncrasies inherent in ODBC, as well as limitations on the Snowflake side.

This V1.0 implementation removes support for ODBC, and instead relies solely on Snowflake's SQL API.

If you want to progressively migrate to the Ecto implementation, these tips might be helpful:

• Remove all Snowflex.*_param() wrapped functions.
• In your Snowflake Repo, add a declaration similar to the following:

defmodule MyApp.Snowflake do
  use Ecto.Repo, otp_app: :my_app, adapter: Snowflex

  def sql_query(query) do
    execute(query)
  end

  def param_query(query, params \\ %{}) do
    execute(query, params)
  end

  defp execute(query, params \\ %{}) do
    query
    |> query_many!(params) # provided by `Ecto.Repo`
    |> process_results()
    |> unwrap_single_result()
  end

  defp process_results([]), do: []

   defp process_results([%Snowflex.Result{} = result | rest]),
     do: [unpack_snowflex_result(result) | process_results(rest)]
   defp process_results([other | rest]), do: [other | process_results(rest)]
   defp unpack_snowflex_result(%{columns: nil, num_rows: num_rows}), do: {:updated, num_rows}
   defp unpack_snowflex_result(%{columns: columns, rows: rows})
        when is_list(columns) and is_list(rows) do
     headers = Enum.map(columns, &(to_string(&1) |> String.downcase()))
     rows
     |> Enum.map(fn row ->
       Enum.zip(headers, row) |> Map.new()
     end)
   end
   # If there's just one result, unwrap it
   defp unwrap_single_result([result]), do: result
   defp unwrap_single_result(results), do: results
end

Any references to Snowflex for sql_query and param_query can then be replaced with MyApp.Snowflake.

To replace the previous functionality of cast_results, we would recommend that you leverage a schemaless changeset.

Testing in your Project

When running tests locally, it can often be helpful to avoid hitting Snowflake to avoid unnecessary compute/storage costs.

See Snowflex.MigrationGenerator for a strategy to use a local DB implementation when running unit tests, while still using Snowflake in dev/prod environments.