icelake

Stream records into Apache Iceberg tables on your own object storage.
A Go library and a small CLI; standard readers stay on the read path.

icelake accepts recursive records, durably stages them in SQLite, writes bounded uncompressed Parquet files by default, and commits those files to Apache Iceberg v2 tables in an S3-compatible bucket. Set ZSTDLevel from 1 through 22 to enable ZSTD compression. The table schema is Apache iceberg.Schema itself: flat and nested tables use one model, with permanent IDs on struct fields, list elements, map keys, and map values.

• Every successful insert is durable locally before it returns.
• Every row receives an immutable UUIDv7 _icelake_record_id that survives replay and is written to Iceberg.
• All Iceberg v1/v2 primitive types and recursive struct, list, and map combinations are supported.
• Optional Iceberg sort orders physically order each new file; files contain bounded, independently prunable Parquet row groups.
• FlushMaxBytes targets encoded Parquet object size. Large files are built incrementally on disk instead of in one whole-file memory buffer.
• Local-only mode writes the same Parquet spool without a bucket. Reopen the same state in bucket mode to upload and commit it.
• An optional ClickHouse raw mirror follows durable staging within seconds and repairs itself from authoritative Iceberg files after an outage. It never blocks canonical lake progress.
• DuckDB, PyIceberg, Spark, and other standard readers query the resulting open formats directly.

Install a signed release with mise:

mise use -g github:gvkhna/icelake

Static Linux and macOS archives for amd64 and arm64 are available from the releases page.

For an embedded Go application:

go get github.com/gvkhna/icelake/v2

A portable declaration wraps each table's native Iceberg schema and optional native sort order. Field ID 1 is always the required UUID identifier generated by icelake; input rows omit it.

{
  "tables": [
    {
      "namespace": "market",
      "table": "fills",
      "schema": {
        "type": "struct",
        "schema-id": 0,
        "identifier-field-ids": [1],
        "fields": [
          {"id": 1, "name": "_icelake_record_id", "required": true, "type": "uuid"},
          {"id": 2, "name": "symbol", "required": true, "type": "string"},
          {"id": 3, "name": "price", "required": true, "type": "decimal(18, 9)"},
          {"id": 4, "name": "venue_timestamp", "required": true, "type": "timestamptz"}
        ]
      },
      "sort-order": {
        "order-id": 1,
        "fields": [
          {"source-id": 2, "transform": "identity", "direction": "asc", "null-order": "nulls-last"},
          {"source-id": 4, "transform": "identity", "direction": "asc", "null-order": "nulls-last"}
        ]
      }
    }
  ]
}

Start locally with no bucket:

export ICELAKE_DATA_DIR="$PWD/icelake-data"
export ICELAKE_SCHEMA_FILE="$PWD/schema.json"
export ICELAKE_LOCAL_ONLY=true
mkdir -p "$ICELAKE_DATA_DIR"

printf '%s\n' \
  '{"table":"market.fills","row":{"symbol":"ABC","price":"1.234567890","venue_timestamp":"2026-08-10T12:00:00Z"}}' \
  | icelake run -f

The input is an NDJSON sequence of {table,row} envelopes, a deterministic CBOR sequence, or both encodings interleaved. Decimals and temporal values use lossless text forms; nested structs are objects, lists are arrays, and Iceberg maps are arrays of {"key":...,"value":...} entries.

For bucket mode, omit ICELAKE_LOCAL_ONLY and set:

export ICELAKE_ENDPOINT=https://ACCOUNT.r2.cloudflarestorage.com
export ICELAKE_BUCKET=my-bucket
export ICELAKE_PREFIX=warehouse
export ICELAKE_ACCESS_KEY_ID=...
export ICELAKE_SECRET_ACCESS_KEY=...

icelake run backgrounds itself by default; -f keeps it attached. icelake sync asks a running daemon to flush and reconcile now, or temporarily takes exclusive ownership and performs the same work when no daemon is running. icelake check inspects canonical safety and mirror freshness separately. icelake usage is the complete command manual.

package main

import (
    "context"
    "log"
    "os"
    "time"

    "github.com/apache/iceberg-go"
    "github.com/gvkhna/icelake/v2"
)

func main() {
    ctx := context.Background()
    schema := iceberg.NewSchemaWithIdentifiers(0, []int{1},
        iceberg.NestedField{ID: 1, Name: "_icelake_record_id", Required: true, Type: iceberg.PrimitiveTypes.UUID},
        iceberg.NestedField{ID: 2, Name: "symbol", Required: true, Type: iceberg.PrimitiveTypes.String},
        iceberg.NestedField{ID: 3, Name: "price", Required: true, Type: iceberg.DecimalTypeOf(18, 9)},
        iceberg.NestedField{ID: 4, Name: "venue_timestamp", Required: true, Type: iceberg.PrimitiveTypes.TimestampTz},
    )

    store, err := icelake.Open(ctx, icelake.Config{
        StagingPath:       "/var/lib/myservice/staging.db",
        CatalogPath:       "/var/lib/myservice/catalog.db",
        Endpoint:          "https://ACCOUNT.r2.cloudflarestorage.com",
        Bucket:            "my-bucket",
        WarehousePrefix:   "warehouse",
        AccessKeyID:       os.Getenv("R2_ACCESS_KEY_ID"),
        SecretAccessKey:   os.Getenv("R2_SECRET_ACCESS_KEY"),
        FlushMaxRecords:   50_000,
        FlushMaxBytes:     512 << 20,
        FlushInterval:     time.Hour,
        ZSTDLevel:         0, // uncompressed by default; set 1 through 22 for ZSTD
        StagingMaxRecords: 5_000_000,
        StagingMaxBytes:   4 << 30,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer store.Close(ctx)

    writer, err := icelake.OpenWriter(ctx, store, icelake.TableConfig{
        Namespace: "market",
        Table:     "fills",
        Schema:    schema,
    })
    if err != nil {
        log.Fatal(err)
    }

    if err := writer.InsertJSON(ctx, []byte(`{"symbol":"ABC","price":"1.234567890","venue_timestamp":"2026-08-10T12:00:00Z"}`)); err != nil {
        log.Fatal(err)
    }
    if err := writer.Flush(ctx); err != nil {
        log.Fatal(err)
    }
}

Insert, InsertBatch, InsertJSON, InsertJSONBatch, InsertCBOR, and InsertCBORBatch all use the same recursive schema path. An insert returns after the generated UUID and canonical row are atomically durable in staging. Writer.Flush waits for canonical Iceberg progress; Store.Synchronize also waits for the configured ClickHouse mirror watermark.

Generate the setup from the catalog rather than guessing an Iceberg metadata version:

icelake duckdb-init market.fills > icelake.duckdb.sql
duckdb -init icelake.duckdb.sql -c \
  'SELECT symbol, price FROM market.fills ORDER BY venue_timestamp LIMIT 10'

The generated setup installs and loads DuckDB's httpfs, iceberg, and cache_httpfs extensions, configures a durable cache, and refers to bucket credentials through getenv(...) rather than printing them.

export ICELAKE_CLICKHOUSE_ADDR=localhost:9000
export ICELAKE_CLICKHOUSE_USERNAME=icelake
export ICELAKE_CLICKHOUSE_PASSWORD=...
export ICELAKE_CLICKHOUSE_TTL='market.fills=720h@venue_timestamp' # optional

The daemon delivers accepted rows from durable staging on an independent seconds-scale cadence. Iceberg remains authoritative: a ClickHouse outage does not stop ingestion or Iceberg commits, and startup plus periodic reconciliation repairs retained rows from catalog-authoritative Parquet files. The stable raw view is <database>.<namespace>__<table>; incompatible layouts are rebuilt in a generation-specific physical table and switched only after verification.

Every mirrored row carries the same native UUID as Iceberg. Optional composites use explicit _present/_value tuples so null and empty remain different; maps use Array(Tuple(key,value)). TTL is projection policy and expired mirror rows are not restored. Store.MirrorStatuses reports healthy, behind, catching_up, rebuild_required, or schema_conflict without conflating mirror lag with canonical data loss.

• Schema and evolution
• Architecture
• Testing
• Release procedure
• Release history
• CLI manual
• Contributor rules

The toolchain is pinned by mise:

git clone https://github.com/gvkhna/icelake
cd icelake
mise install
GOFLAGS=-count=1 mise run check

The suite uses real MinIO and ClickHouse substrates plus independent DuckDB, PyIceberg, Iceberg-metadata, and Parquet-footer checks where each contract requires them.

MIT