Current section
Files
Jump to
Current section
Files
livebooks/04_building_storage_systems.livemd
<!-- livebook:{"persist_outputs":true} -->
# Building Storage Systems with ExCodecs
```elixir
# Use this install to work with the source code
# Mix.install(
# [
# {:ex_codecs, path: Path.join(__DIR__, "..")},
# {:rustler, "~> 0.36"},
# {:jason, "~> 1.4"},
# {:kino, "~> 0.14"},
# {:kino_vega_lite, "~> 0.1.13"}
# ],
# config: [rustler_precompiled: [force_build: [ex_codecs: true]]]
# )
Mix.install( [
{:ex_codecs, "~> 0.2.3"},
{:jason, "~> 1.4"},
{:kino, "~> 0.14"},
{:kino_vega_lite, "~> 0.1.13"}
])
```
## Livebook Series
| # | Livebook |
| --- | -------------------------------------------------------------- |
| 01 | [Introduction](01_introduction.livemd) |
| 02 | [Compression Fundamentals](02_compression_fundamentals.livemd) |
| 03 | [Codec Comparison](03_codec_comparison.livemd) |
| 04 | **Building Storage Systems** (you are here) |
| 05 | [Zarr-Style Workloads](05_zarr_style_workloads.livemd) |
| 06 | [Spatial Codecs](06_spatial_codecs.livemd) |
## Practical Usage Patterns
### The Basic Pattern
```elixir
data = "Important data to store"
{:ok, compressed} = ExCodecs.encode(:zstd, data)
{:ok, recovered} = ExCodecs.decode(:zstd, compressed)
IO.puts("Round-trip OK: #{recovered == data}")
```
<!-- livebook:{"output":true} -->
```
Round-trip OK: true
```
<!-- livebook:{"output":true} -->
```
:ok
```
### Choosing a Codec at Runtime
```elixir
defmodule StorageConfig do
def codec_for_content_type("application/json"), do: :zstd
def codec_for_content_type("application/octet-stream"), do: :lz4
def codec_for_content_type("application/x-array"), do: :blosc2
def codec_for_content_type(_), do: :zstd
end
content_type = "application/json"
codec = StorageConfig.codec_for_content_type(content_type)
IO.puts("Selected codec: #{inspect(codec)}")
IO.puts("Supported: #{ExCodecs.supports?(codec)}")
```
<!-- livebook:{"output":true} -->
```
Selected codec: :zstd
Supported: true
```
<!-- livebook:{"output":true} -->
```
:ok
```
## Building a Compressed Storage Layer
```elixir
defmodule CompressedStore do
@moduledoc """
A simple key-value store with transparent compression.
"""
use Agent
defstruct [:codec, :opts, :table]
def start_link(opts \\ []) do
codec = Keyword.get(opts, :codec, :zstd)
codec_opts = Keyword.get(opts, :opts, [])
name = Keyword.get(opts, :name, __MODULE__)
Agent.start_link(fn ->
%__MODULE__{codec: codec, opts: codec_opts, table: %{}}
end, name: name)
end
def put(store \\ __MODULE__, key, value) do
Agent.get_and_update(store, fn state ->
case ExCodecs.encode(state.codec, value, state.opts) do
{:ok, compressed} ->
entry = %{
compressed: compressed,
original_size: byte_size(value),
codec: state.codec
}
{{:ok, byte_size(compressed)}, %{state | table: Map.put(state.table, key, entry)}}
{:error, err} ->
{{:error, err}, state}
end
end)
end
def get(store \\ __MODULE__, key) do
Agent.get(store, fn state ->
case Map.get(state.table, key) do
nil -> {:error, :not_found}
entry -> ExCodecs.decode(entry.codec, entry.compressed)
end
end)
end
def info(store \\ __MODULE__) do
Agent.get(store, fn state ->
entries = Map.values(state.table)
total_original = Enum.sum(Enum.map(entries, & &1.original_size))
total_compressed = Enum.sum(Enum.map(entries, &byte_size(&1.compressed)))
%{
codec: state.codec,
entry_count: length(entries),
total_original_bytes: total_original,
total_compressed_bytes: total_compressed,
savings_pct: Float.round(100 * (1 - total_compressed / max(total_original, 1)), 1)
}
end)
end
def delete(store \\ __MODULE__, key) do
Agent.get_and_update(store, fn state ->
{Map.get(state.table, key), %{state | table: Map.delete(state.table, key)}}
end)
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, CompressedStore, <<70, 79, 82, 49, 0, 0, 36, ...>>, ...}
```
### Using the Compressed Store
```elixir
case CompressedStore.start_link(codec: :zstd, opts: [level: 3]) do
{:ok, _pid} -> :ok
{:error, {:already_started, _pid}} -> :ok
end
CompressedStore.put(:doc1, String.duplicate("Hello, World! ", 1000))
CompressedStore.put(:doc2, :crypto.strong_rand_bytes(4096) |> Base.encode64())
CompressedStore.put(:doc3, Jason.encode!(for i <- 1..100, do: %{id: i, value: :rand.uniform(1000)}))
{:ok, doc1} = CompressedStore.get(:doc1)
IO.puts("Retrieved doc1 (#{byte_size(doc1)} bytes)")
info = CompressedStore.info()
IO.puts("\nStorage Summary:")
IO.puts(" Codec: #{inspect(info.codec)}")
IO.puts(" Entries: #{info.entry_count}")
IO.puts(" Original: #{info.total_original_bytes} bytes")
IO.puts(" Compressed: #{info.total_compressed_bytes} bytes")
IO.puts(" Savings: #{info.savings_pct}%")
```
<!-- livebook:{"output":true} -->
```
Retrieved doc1 (14000 bytes)
Storage Summary:
Codec: :zstd
Entries: 3
Original: 21645 bytes
Compressed: 4591 bytes
Savings: 78.8%
```
<!-- livebook:{"output":true} -->
```
:ok
```
## Multi-Codec Store
```elixir
defmodule MultiCodecStore do
@moduledoc """
A store that selects the best codec per value.
"""
use Agent
@codecs [:lz4, :snappy, :zstd, :bzip2]
def start_link(opts \\ []) do
name = Keyword.get(opts, :name, __MODULE__)
Agent.start_link(fn -> %{} end, name: name)
end
def put(store \\ __MODULE__, key, value) do
Agent.get_and_update(store, fn state ->
results = for codec <- @codecs do
{:ok, enc} = ExCodecs.encode(codec, value)
{codec, enc, byte_size(enc)}
end
{best_codec, best_enc, best_size} =
Enum.min_by(results, fn {_, _, size} -> size end)
entry = %{
codec: best_codec,
compressed: best_enc,
original_size: byte_size(value),
compressed_size: best_size
}
{{:ok, entry}, Map.put(state, key, entry)}
end)
end
def get(store \\ __MODULE__, key) do
Agent.get(store, fn state ->
case Map.get(state, key) do
nil -> {:error, :not_found}
entry -> ExCodecs.decode(entry.codec, entry.compressed)
end
end)
end
def stats(store \\ __MODULE__) do
Agent.get(store, fn state ->
state
|> Map.values()
|> Enum.group_by(& &1.codec)
|> Enum.map(fn {codec, entries} ->
{codec, length(entries),
Enum.sum(Enum.map(entries, & &1.original_size)),
Enum.sum(Enum.map(entries, & &1.compressed_size))}
end)
|> Enum.sort_by(&elem(&1, 1))
end)
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, MultiCodecStore, <<70, 79, 82, 49, 0, 0, 27, ...>>, ...}
```
### Using the Multi-Codec Store
```elixir
case MultiCodecStore.start_link() do
{:ok, _pid} -> :ok
{:error, {:already_started, _pid}} -> :ok
end
datasets = %{
repetitive: String.duplicate("abcabcabc", 5000),
semi_random: (for _ <- 1..10000, into: <<>>, do: <<:rand.uniform(255)>>),
floats: (for i <- 1..4096, into: <<>>, do: <<i * 0.1::float-size(64)-little>>)
}
for {name, data} <- datasets do
{:ok, entry} = MultiCodecStore.put(name, data)
IO.puts("#{name}: best codec = #{inspect(entry.codec)}, " <>
"#{entry.original_size} -> #{entry.compressed_size} bytes " <>
"(#{Float.round(100 * entry.compressed_size / entry.original_size, 1)}%)")
end
IO.puts("\nCodec distribution:")
for {codec, count, orig, comp} <- MultiCodecStore.stats() do
IO.puts(" #{inspect(codec)}: #{count} entries, #{orig} -> #{comp} bytes")
end
```
<!-- livebook:{"output":true} -->
```
floats: best codec = :zstd, 32768 -> 4801 bytes (14.7%)
repetitive: best codec = :zstd, 45000 -> 21 bytes (0.0%)
semi_random: best codec = :snappy, 10000 -> 10005 bytes (100.0%)
Codec distribution:
:snappy: 1 entries, 10000 -> 10005 bytes
:zstd: 2 entries, 77768 -> 4822 bytes
```
<!-- livebook:{"output":true} -->
```
[:ok, :ok]
```
## Chunked Compression for Large Files
```elixir
defmodule ChunkedCompressor do
@moduledoc """
Compresses large data in fixed-size chunks to limit peak memory usage.
The container stores the chunk count, the original byte size, and an offset
table so individual chunks can be decompressed on demand (random access)
without re-decoding the whole blob.
"""
@default_chunk_size 64 * 1024
def compress(data, codec \\ :zstd, opts \\ [], chunk_size \\ @default_chunk_size) do
chunks = chunk_data(data, chunk_size)
compressed_chunks =
Enum.map(chunks, fn chunk ->
{:ok, enc} = ExCodecs.encode(codec, chunk, opts)
enc
end)
offsets =
compressed_chunks
|> Enum.scan(0, fn chunk, acc -> acc + byte_size(chunk) end)
|> Enum.drop(-1)
|> List.insert_at(0, 0)
header = <<
length(compressed_chunks)::unsigned-big-32,
byte_size(data)::unsigned-big-64
>>
offset_data = for offset <- offsets, into: <<>>, do: <<offset::unsigned-big-32>>
{:ok, header <> offset_data <> IO.iodata_to_binary(compressed_chunks)}
end
def decompress(blob, codec \\ :zstd) do
<<num_chunks::unsigned-big-32, original_size::unsigned-big-64, rest::binary>> = blob
offsets_size = num_chunks * 4
<<offsets::binary-size(^offsets_size), chunks_binary::binary>> = rest
offsets_list = for <<offset::unsigned-big-32 <- offsets>>, do: offset
offsets_list = offsets_list ++ [byte_size(chunks_binary)]
chunks =
Enum.chunk_every(offsets_list, 2, 1, :discard)
|> Enum.map(fn [start_off, end_off] ->
binary_part(chunks_binary, start_off, end_off - start_off)
end)
decompressed =
Enum.map(chunks, fn chunk ->
{:ok, dec} = ExCodecs.decode(codec, chunk)
dec
end)
result = IO.iodata_to_binary(decompressed)
if byte_size(result) == original_size do
{:ok, result}
else
{:error, :size_mismatch}
end
end
@doc """
Decompresses a single chunk by 0-based index, using the offset table.
"""
def read_chunk(blob, index, codec \\ :zstd) do
<<num_chunks::unsigned-big-32, _original_size::unsigned-big-64, rest::binary>> = blob
if index < 0 or index >= num_chunks do
{:error, :out_of_range}
else
offsets_size = num_chunks * 4
<<offsets::binary-size(^offsets_size), chunks_binary::binary>> = rest
offsets_list = for <<offset::unsigned-big-32 <- offsets>>, do: offset
offsets_list = offsets_list ++ [byte_size(chunks_binary)]
[start_off, end_off] = Enum.at(Enum.chunk_every(offsets_list, 2, 1, :discard), index)
chunk = binary_part(chunks_binary, start_off, end_off - start_off)
ExCodecs.decode(codec, chunk)
end
end
defp chunk_data(data, chunk_size) when is_integer(chunk_size) and chunk_size > 0 do
case data do
<<chunk::binary-size(^chunk_size), rest::binary>> -> [chunk | chunk_data(rest, chunk_size)]
<<>> -> []
remainder when is_binary(remainder) -> [remainder]
end
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, ChunkedCompressor, <<70, 79, 82, 49, 0, 0, 32, ...>>, ...}
```
```elixir
large_data = String.duplicate("This is a chunk of data that repeats. ", 10000)
{:ok, compressed} = ChunkedCompressor.compress(large_data, :zstd)
{:ok, recovered} = ChunkedCompressor.decompress(compressed, :zstd)
<<num_chunks::unsigned-big-32, _::unsigned-big-64, _::binary>> = compressed
{:ok, first_chunk} = ChunkedCompressor.read_chunk(compressed, 0, :zstd)
IO.puts("Original: #{byte_size(large_data)} bytes")
IO.puts("Compressed: #{byte_size(compressed)} bytes")
IO.puts("Ratio: #{Float.round(100 * byte_size(compressed) / byte_size(large_data), 1)}%")
IO.puts("Chunks: #{num_chunks}")
IO.puts("Round-trip: #{recovered == large_data}")
IO.puts("Random access chunk 0: #{byte_size(first_chunk)} bytes (matches: #{first_chunk == binary_part(large_data, 0, 64 * 1024)})")
```
<!-- livebook:{"output":true} -->
```
Original: 380000 bytes
Compressed: 384 bytes
Ratio: 0.1%
Chunks: 6
Round-trip: true
Random access chunk 0: 65536 bytes (matches: true)
```
<!-- livebook:{"output":true} -->
```
:ok
```
## Error Handling Best Practices
```elixir
case ExCodecs.encode(:zstd, "some data") do
{:ok, compressed} ->
{:ok, compressed}
{:error, %ExCodecs.Error{reason: :unsupported_codec}} ->
{:error, :codec_missing}
{:error, %ExCodecs.Error{reason: :codec_unavailable}} ->
{:error, :codec_unavailable}
{:error, %ExCodecs.Error{reason: :nif_not_loaded}} ->
{:error, :nif_not_loaded}
{:error, %ExCodecs.Error{reason: :invalid_data} = err} ->
IO.puts("Invalid data: #{err.message}")
{:error, :bad_input}
{:error, %ExCodecs.Error{reason: :compression_failed} = err} ->
IO.puts("Compression failed: #{err.message}")
{:error, :compress_failed}
{:error, %ExCodecs.Error{} = err} ->
IO.puts("Structured error #{err.reason}: #{err.message}")
{:error, err.reason}
end
```
<!-- livebook:{"output":true} -->
```
{:ok, <<40, 181, 47, 253, 32, 9, 73, 0, 0, 115, 111, 109, 101, 32, 100, 97, 116, 97>>}
```
```elixir
# Exercise the error branches with real failures.
results =
[
{:ok, _} = ExCodecs.encode(:zstd, "ok"),
{:error, %ExCodecs.Error{reason: :invalid_options}} =
ExCodecs.encode(:zstd, "x", level: 99),
{:error, %ExCodecs.Error{reason: :decompression_failed}} =
ExCodecs.decode(:zstd, "not a zstd frame"),
{:error, %ExCodecs.Error{reason: :unsupported_codec}} =
ExCodecs.encode(:nonexistent, "x")
]
for {tag, {status, _} = result} <- Enum.zip([:ok, :bad_level, :bad_frame, :bad_codec], results) do
IO.puts("#{String.pad_trailing("#{tag}", 12)} -> #{inspect(result)}")
end
```
<!-- livebook:{"output":true} -->
```
warning: variable "status" is unused (if the variable is not meant to be used, prefix it with an underscore)
└─ work/ex_codecs/livebooks/04_building_storage_systems.livemd#cell:pruxa6uyvvgiepiz:13
ok -> {:ok, <<40, 181, 47, 253, 32, 2, 17, 0, 0, 111, 107>>}
bad_level -> {:error, %ExCodecs.Error{reason: :invalid_options, message: "Level must be an integer between 1 and 22", codec: nil, details: nil}}
bad_frame -> {:error, %ExCodecs.Error{reason: :decompression_failed, message: "Decompression failed", codec: :zstd, details: nil}}
bad_codec -> {:error, %ExCodecs.Error{reason: :unsupported_codec, message: "The specified codec is not supported", codec: :nonexistent, details: nil}}
```
<!-- livebook:{"output":true} -->
```
[:ok, :ok, :ok, :ok]
```
```elixir
# Bound decompression for untrusted payloads (default is 256 MiB).
# Classic bomb shape: 64 KiB of zeros → tiny compressed input.
bomb_raw = :binary.copy(<<0>>, 65_536)
{:ok, bomb} = ExCodecs.encode(:zstd, bomb_raw)
{:error, %ExCodecs.Error{reason: :output_limit_exceeded}} =
ExCodecs.decode(:zstd, bomb, max_output_size: 1_024)
{:ok, ^bomb_raw} = ExCodecs.decode(:zstd, bomb, max_output_size: 65_536)
IO.puts("bomb compressed size: #{byte_size(bomb)} bytes (#{Float.round(65_536 / byte_size(bomb), 1)}x)")
```
<!-- livebook:{"output":true} -->
```
bomb compressed size: 11 bytes (5957.8x)
```
<!-- livebook:{"output":true} -->
```
:ok
```
> See [Compression Fundamentals](02_compression_fundamentals.livemd) for the
> same pattern across all compression codecs.
```elixir
defmodule SafeCompress do
def call(codec, data, opts \\ []) do
with {:check_codec, true} <- {:check_codec, ExCodecs.supports?(codec)},
{:check_binary, true} <- {:check_binary, is_binary(data)},
{:ok, compressed} <- ExCodecs.encode(codec, data, opts),
{:ok, ^data} <- ExCodecs.decode(codec, compressed) do
{:ok, compressed}
else
{:check_codec, false} -> {:error, :unsupported_codec}
{:check_binary, false} -> {:error, :not_binary}
{:error, err} -> {:error, err}
end
end
end
{:ok, result} = SafeCompress.call(:zstd, "verify round-trip")
IO.puts("Safe compress OK: #{is_binary(result)}")
```
<!-- livebook:{"output":true} -->
```
Safe compress OK: true
```
<!-- livebook:{"output":true} -->
```
:ok
```
```elixir
defmodule CompressWithFallback do
def call(data, codecs \\ [:zstd, :lz4, :snappy]) do
codecs
|> Enum.reduce_while({:error, :no_codec_available}, fn codec, _acc ->
case ExCodecs.encode(codec, data) do
{:ok, compressed} -> {:halt, {:ok, {codec, compressed}}}
{:error, err} -> {:cont, {:error, {codec, err.reason}}}
end
end)
end
end
{:ok, {used_codec, compressed}} = CompressWithFallback.call("fallback test data")
IO.puts("Used codec: #{inspect(used_codec)}")
IO.puts("Compressed: #{byte_size(compressed)} bytes")
```
<!-- livebook:{"output":true} -->
```
Used codec: :zstd
Compressed: 27 bytes
```
<!-- livebook:{"output":true} -->
```
:ok
```
## ETS Integration
```elixir
defmodule CompressedETS do
@moduledoc """
ETS-backed storage with transparent compression.
"""
def new(name \\ :compressed_store) do
if :ets.whereis(name) != :undefined, do: :ets.delete(name)
:ets.new(name, [:set, :public, :named_table])
name
end
def insert(table, key, value, codec \\ :zstd, opts \\ []) do
{:ok, compressed} = ExCodecs.encode(codec, value, opts)
:ets.insert(table, {key, compressed, codec, byte_size(value)})
{:ok, byte_size(compressed)}
end
def lookup(table, key) do
case :ets.lookup(table, key) do
[{^key, compressed, codec, _orig_size}] ->
ExCodecs.decode(codec, compressed)
[] ->
{:error, :not_found}
end
end
def stats(table) do
entries = :ets.tab2list(table)
total_orig = Enum.sum(Enum.map(entries, fn {_, _, _, orig} -> orig end))
total_comp = Enum.sum(Enum.map(entries, fn {_, comp, _, _} -> byte_size(comp) end))
%{
entries: length(entries),
original_bytes: total_orig,
compressed_bytes: total_comp,
savings_pct: Float.round(100 * (1 - total_comp / max(total_orig, 1)), 1)
}
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, CompressedETS, <<70, 79, 82, 49, 0, 0, 20, ...>>, ...}
```
```elixir
table = CompressedETS.new()
for i <- 1..100 do
data = String.duplicate("item-#{rem(i, 10)} ", 500)
CompressedETS.insert(table, "key_#{i}", data)
end
stats = CompressedETS.stats(table)
IO.puts("ETS Compressed Store:")
IO.puts(" Entries: #{stats.entries}")
IO.puts(" Original: #{stats.original_bytes} bytes")
IO.puts(" Compressed: #{stats.compressed_bytes} bytes")
IO.puts(" Savings: #{stats.savings_pct}%")
{:ok, val} = CompressedETS.lookup(table, "key_1")
IO.puts("\nRetrieved #{byte_size(val)} bytes successfully")
:ets.delete(table)
```
<!-- livebook:{"output":true} -->
```
ETS Compressed Store:
Entries: 100
Original: 350000 bytes
Compressed: 2400 bytes
Savings: 99.3%
Retrieved 3500 bytes successfully
```
<!-- livebook:{"output":true} -->
```
true
```
## File Storage Integration
```elixir
defmodule CompressedFileIO do
@moduledoc """
File I/O with transparent compression.
"""
@magic "EXCD"
def write(path, data, codec \\ :zstd, opts \\ []) do
{:ok, compressed} = ExCodecs.encode(codec, data, opts)
header = <<
@magic::binary,
codec_atom_to_bin(codec)::binary,
byte_size(data)::unsigned-big-64,
byte_size(compressed)::unsigned-big-64
>>
File.write(path, header <> compressed)
end
def read(path) do
case File.read(path) do
{:ok, <<@magic::binary, codec_bin::binary-size(8), _orig_size::unsigned-big-64, _comp_size::unsigned-big-64, compressed::binary>>} ->
codec = bin_to_codec_atom(codec_bin)
ExCodecs.decode(codec, compressed)
{:ok, _} ->
{:error, :invalid_format}
{:error, reason} ->
{:error, reason}
end
end
defp codec_atom_to_bin(atom) do
atom |> Atom.to_string() |> String.pad_trailing(8)
end
defp bin_to_codec_atom(bin) do
case bin |> String.trim_trailing() do
"zstd" -> :zstd
"lz4" -> :lz4
"snappy" -> :snappy
"bzip2" -> :bzip2
"blosc2" -> :blosc2
other -> {:error, {:unknown_codec, other}}
end
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, CompressedFileIO, <<70, 79, 82, 49, 0, 0, 16, ...>>, ...}
```
```elixir
test_data = String.duplicate("Compressed file storage example. ", 2000)
path = Path.join(System.tmp_dir!(), "ex_codecs_demo.dat")
:ok = CompressedFileIO.write(path, test_data, :zstd, level: 9)
{:ok, recovered} = CompressedFileIO.read(path)
IO.puts("Wrote and read #{byte_size(test_data)} bytes")
IO.puts("Round-trip OK: #{recovered == test_data}")
IO.puts("File size: #{File.stat!(path).size} bytes")
File.rm(path)
```
<!-- livebook:{"output":true} -->
```
Wrote and read 66000 bytes
Round-trip OK: true
File size: 83 bytes
```
<!-- livebook:{"output":true} -->
```
:ok
```
## Key Takeaways
1. **Wrap encode/decode** in your own modules to centralize codec choice and error handling
2. **Always handle errors** — codecs can fail on invalid input, memory pressure, etc.
3. **Store the codec name** alongside compressed data so you know how to decompress
4. **Use ETS** for compressed in-memory caches — significant memory savings
5. **Chunk large files** — enables random access, parallel processing, and bounded memory
6. **Try multiple codecs** for mixed workloads — different data compresses best with different algorithms
## Navigation
**Previous:** [Codec Comparison](03_codec_comparison.livemd) ·
**Next:** [Zarr-Style Workloads](05_zarr_style_workloads.livemd)