Skip to content

Replace DuckLake with authoritative Parquet storage #207

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vishr wants to merge 19 commits into
base: main
Choose a base branch
from
Open

Replace DuckLake with authoritative Parquet storage #207

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
793e067
feat(storage)!: replace DuckLake telemetry
vishr Aug 26, 2026
45bcae2
fix(storage): harden telemetry persistence
vishr Aug 26, 2026
d382a30
fix(storage): harden maintenance concurrency
vishr Aug 26, 2026
5a7a5fb
fix(storage): make compaction recovery safe
vishr Aug 26, 2026
f1463b1
fix(storage): close recovery and query gaps
vishr Aug 26, 2026
2cad290
fix(storage): close remaining hot/cold and durability review gaps
vishr Aug 27, 2026
05a77f9
fix(storage): keep telemetry replayable and bound cold reads
vishr Aug 27, 2026
6b97e89
fix(storage): narrow WAL quarantine and validate segment blocks
vishr Aug 27, 2026
1f05228
fix(storage): bound segment decoders and drop the rollup key limit
vishr Aug 27, 2026
e6b4b58
fix(storage): enforce durable ingest guarantees
vishr Aug 27, 2026
a221291
feat(storage)!: finalize telemetry engine
vishr Aug 27, 2026
f18a19d
perf(storage)!: bound hot index memory
vishr Aug 27, 2026
2bf8822
chore(bench): keep storage POC local
vishr Aug 27, 2026
526fd87
refactor(storage)!: make Parquet authoritative
vishr Aug 27, 2026
d87e8a3
fix(storage): close Parquet concurrency gaps
vishr Aug 27, 2026
8f205d6
fix(storage): bound Parquet publication
vishr Aug 27, 2026
3746d98
fix(storage): isolate Parquet maintenance
vishr Aug 28, 2026
895c238
fix(storage): bound Parquet publication
vishr Aug 28, 2026
fbc5604
fix(storage): guarantee maintenance progress
vishr Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,15 +22,16 @@ executable, including the React client.

![Fanout architecture](docs/diagrams/architecture.svg)

Telemetry lands over OTLP/gRPC or OTLP/HTTP, is batched into DuckLake/Parquet,
and is read back through a DuckDB query kernel that also maintains service,
Telemetry lands over OTLP/gRPC or OTLP/HTTP, is durably committed as atomic
Parquet batches with persistent trace indexes, and is read back through a
DuckDB query kernel that also maintains service,
endpoint, and edge rollups. The browser client, an in-process agent, and any
external MCP host all reach the same typed observability contract rather than
issuing raw SQL.

Every write to the telemetry catalog — ingest flush, rollups, and background
maintenance alike — passes through a single write gate that holds one catalog
write in flight at a time:
Independent ingest batches encode in parallel. Rollup-cache writes are
serialized inside DuckDB, while retention and compaction atomically swap
immutable Parquet directories behind active readers:

![Fanout persistence](docs/diagrams/persistence.svg)

Expand All @@ -53,7 +54,7 @@ what separates it from its neighbours.
| If you use | Where Fanout differs |
| --- | --- |
| **Grafana with Loki, Tempo, and Mimir** | That stack keeps a service and a query language per signal, plus object storage underneath. Fanout keeps one process, one data directory, and one typed contract across all three signals, at the cost of the horizontal scale those components are built for. |
| **SigNoz** | Both are OTLP-native and self-hosted. SigNoz composes a collector, ClickHouse, and query services; Fanout compiles ingest, storage, query, alerting, and the browser client into one binary, with DuckLake/Parquet on local disk instead of a database cluster. |
| **SigNoz** | Both are OTLP-native and self-hosted. SigNoz composes a collector, ClickHouse, and query services; Fanout compiles ingest, indexed storage, DuckDB queries, alerting, and the browser client into one binary, with open Parquet on local disk instead of a database cluster. |
| **Jaeger** | Jaeger covers traces and expects a storage backend you run separately. Fanout ingests traces, logs, and metrics into the same store, with nothing else to deploy. |
| **Prometheus with Grafana** | Prometheus pulls metrics and is excellent at them. Fanout accepts pushed OTLP for all three signals and is built around investigating a specific incident rather than maintaining long-range metric series. |
| **Datadog**, **Honeycomb**, **Grafana Cloud** | Those are managed services: someone else runs the storage, the scaling, and the upgrades, and your telemetry leaves your network to get there. Fanout is a binary you run, on data that stays on your disk. |
Expand Down
1,204 changes: 1,057 additions & 147 deletions THIRD_PARTY_NOTICES
View file
Open in desktop

Large diffs are not rendered by default.

87 changes: 43 additions & 44 deletions cmd/bench/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,8 +81,7 @@ type config struct {
maxQueryP95 float64
// backfillHours, when >0, spreads each event's timestamp uniformly over the
// last N hours (instead of "now"). Used to PRE-SEED a multi-hour dataset so
// the lake spans several hour partitions — required to exercise within-day
// (hour-partition) pruning, which a same-hour run can't.
// time-window queries and rollups cover more than the current hour.
backfillHours float64
// seed makes the synthetic workload reproducible: same seed, same services,
// endpoints, attributes, and error placement. Two runs are only comparable if
Expand Down Expand Up @@ -557,7 +556,7 @@ func (g *generator) outCtx(ctx context.Context) context.Context {

// eventTime returns the timestamp for an emitted event: now(), or — when
// backfillHours>0 — a time spread uniformly over the last N hours so a pre-seed
// run populates multiple hour partitions (to exercise within-day pruning).
// run exercises multi-hour time-window queries and rollups.
func (g *generator) eventTime(rng *rand.Rand) time.Time {
if g.cfg.backfillHours <= 0 {
return time.Now()
Expand Down Expand Up @@ -910,43 +909,43 @@ type latencyReport struct {
}

type serverReport struct {
BaselineAvailable bool `json:"baseline_available"`
ProcessStartTime float64 `json:"process_start_time_seconds"`
ProcessRestarted bool `json:"process_restarted"`
IngestRowsStart float64 `json:"ingest_rows_start"`
IngestRowsEnd float64 `json:"ingest_rows_end"`
IngestRowsDelta float64 `json:"ingest_rows_delta"`
RowsDroppedStart float64 `json:"rows_dropped_start"`
RowsDroppedEnd float64 `json:"rows_dropped_end"`
RowsDroppedDelta float64 `json:"rows_dropped_delta"`
LakePartitionsStart float64 `json:"lake_partitions_start"`
LakePartitions float64 `json:"lake_partitions"`
LakePartitionsDelta float64 `json:"lake_partitions_delta"`
LakeSizeBytesStart float64 `json:"lake_size_bytes_start"`
LakeSizeBytes float64 `json:"lake_size_bytes"`
LakeSizeBytesDelta float64 `json:"lake_size_bytes_delta"`
LakeGrowthBytesPerSec float64 `json:"lake_growth_bytes_per_sec"`
IngestQueueDepth float64 `json:"ingest_queue_depth"`
AvgRollupMs float64 `json:"avg_rollup_ms"`
AvgFlushMs float64 `json:"avg_flush_ms"`
AvgQueryMs float64 `json:"avg_query_ms"`
CPUSecondsStart float64 `json:"cpu_seconds_start"`
CPUSecondsEnd float64 `json:"cpu_seconds_end"`
CPUSecondsDelta float64 `json:"cpu_seconds_delta"`
CPUCores float64 `json:"cpu_cores"`
RSSBytes float64 `json:"rss_bytes"`
HeapAllocBytes float64 `json:"heap_alloc_bytes"`
AllocBytesStart float64 `json:"alloc_bytes_start"`
AllocBytesEnd float64 `json:"alloc_bytes_end"`
AllocBytesDelta float64 `json:"alloc_bytes_delta"`
AllocBytesPerSec float64 `json:"alloc_bytes_per_sec"`
GCPauseSecondsStart float64 `json:"gc_pause_seconds_start"`
GCPauseSecondsEnd float64 `json:"gc_pause_seconds_end"`
GCPauseSecondsDelta float64 `json:"gc_pause_seconds_delta"`
WriteGateWaitMs map[string]distributionReport `json:"write_gate_wait_ms,omitempty"`
WriteGateHoldMs map[string]distributionReport `json:"write_gate_hold_ms,omitempty"`
DuckLakeOperations map[string]backgroundOperationReport `json:"ducklake_operations,omitempty"`
Rollups map[string]rollupReport `json:"rollups,omitempty"`
BaselineAvailable bool `json:"baseline_available"`
ProcessStartTime float64 `json:"process_start_time_seconds"`
ProcessRestarted bool `json:"process_restarted"`
IngestRowsStart float64 `json:"ingest_rows_start"`
IngestRowsEnd float64 `json:"ingest_rows_end"`
IngestRowsDelta float64 `json:"ingest_rows_delta"`
RowsDroppedStart float64 `json:"rows_dropped_start"`
RowsDroppedEnd float64 `json:"rows_dropped_end"`
RowsDroppedDelta float64 `json:"rows_dropped_delta"`
ParquetFilesStart float64 `json:"parquet_files_start"`
ParquetFiles float64 `json:"parquet_files"`
ParquetFilesDelta float64 `json:"parquet_files_delta"`
ParquetSizeBytesStart float64 `json:"parquet_size_bytes_start"`
ParquetSizeBytes float64 `json:"parquet_size_bytes"`
ParquetSizeBytesDelta float64 `json:"parquet_size_bytes_delta"`
ParquetGrowthBytesPerSec float64 `json:"parquet_growth_bytes_per_sec"`
IngestQueueDepth float64 `json:"ingest_queue_depth"`
AvgRollupMs float64 `json:"avg_rollup_ms"`
AvgFlushMs float64 `json:"avg_flush_ms"`
AvgQueryMs float64 `json:"avg_query_ms"`
CPUSecondsStart float64 `json:"cpu_seconds_start"`
CPUSecondsEnd float64 `json:"cpu_seconds_end"`
CPUSecondsDelta float64 `json:"cpu_seconds_delta"`
CPUCores float64 `json:"cpu_cores"`
RSSBytes float64 `json:"rss_bytes"`
HeapAllocBytes float64 `json:"heap_alloc_bytes"`
AllocBytesStart float64 `json:"alloc_bytes_start"`
AllocBytesEnd float64 `json:"alloc_bytes_end"`
AllocBytesDelta float64 `json:"alloc_bytes_delta"`
AllocBytesPerSec float64 `json:"alloc_bytes_per_sec"`
GCPauseSecondsStart float64 `json:"gc_pause_seconds_start"`
GCPauseSecondsEnd float64 `json:"gc_pause_seconds_end"`
GCPauseSecondsDelta float64 `json:"gc_pause_seconds_delta"`
WriteGateWaitMs map[string]distributionReport `json:"write_gate_wait_ms,omitempty"`
WriteGateHoldMs map[string]distributionReport `json:"write_gate_hold_ms,omitempty"`
TelemetryOperations map[string]backgroundOperationReport `json:"telemetry_operations,omitempty"`
Rollups map[string]rollupReport `json:"rollups,omitempty"`
}

func printReport(r report) {
Expand Down Expand Up @@ -982,11 +981,11 @@ func printReport(r report) {
s := r.Server
fmt.Printf("server (Δ over run):\n")
fmt.Printf(" rows accepted=%.0f dropped=%.0f\n", s.IngestRowsDelta, s.RowsDroppedDelta)
fmt.Printf(" lake_partitions=%.0f lake_size=%.1fMB ingest_queue_depth=%.0f\n",
s.LakePartitions, s.LakeSizeBytes/(1<<20), s.IngestQueueDepth)
fmt.Printf(" parquet_files=%.0f parquet_size=%.1fMB ingest_queue_depth=%.0f\n",
s.ParquetFiles, s.ParquetSizeBytes/(1<<20), s.IngestQueueDepth)
fmt.Printf(" avg rollup=%.1fms flush=%.1fms query=%.1fms\n", s.AvgRollupMs, s.AvgFlushMs, s.AvgQueryMs)
fmt.Printf(" cpu=%.2f core(s) rss=%.1fMB alloc=%.1fMB/s lake_growth=%.1fMB\n",
s.CPUCores, s.RSSBytes/(1<<20), s.AllocBytesPerSec/(1<<20), s.LakeSizeBytesDelta/(1<<20))
fmt.Printf(" cpu=%.2f core(s) rss=%.1fMB alloc=%.1fMB/s parquet_growth=%.1fMB\n",
s.CPUCores, s.RSSBytes/(1<<20), s.AllocBytesPerSec/(1<<20), s.ParquetSizeBytesDelta/(1<<20))
}
if r.Passed {
fmt.Printf("verdict PASS\n")
Expand Down
90 changes: 45 additions & 45 deletions cmd/bench/metrics_report.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,8 +292,8 @@ func serverDelta(base, final *metricSnapshot, durationSeconds float64) *serverRe
}
return round2(value / durationSeconds)
}
lakePartitionsStart := base.total("fanout_lake_partitions")
lakeSizeStart := base.total("fanout_lake_size_bytes")
parquetFilesStart := base.total("fanout_parquet_files")
parquetSizeStart := base.total("fanout_parquet_size_bytes")
cpuSeconds := delta("process_cpu_seconds_total")
allocBytes := delta("go_memstats_alloc_bytes_total")
// process_start_time_seconds is constant for the life of a process, so a
Expand All @@ -302,43 +302,43 @@ func serverDelta(base, final *metricSnapshot, durationSeconds float64) *serverRe
startTimeBefore := base.total("process_start_time_seconds")
startTimeAfter := final.total("process_start_time_seconds")
return &serverReport{
BaselineAvailable: baselineAvailable,
ProcessStartTime: startTimeAfter,
ProcessRestarted: baselineAvailable && startTimeBefore > 0 && startTimeAfter != startTimeBefore,
IngestRowsStart: base.total("fanout_ingest_rows_total"),
IngestRowsEnd: final.total("fanout_ingest_rows_total"),
IngestRowsDelta: delta("fanout_ingest_rows_total"),
RowsDroppedStart: base.total("fanout_rows_dropped_total"),
RowsDroppedEnd: final.total("fanout_rows_dropped_total"),
RowsDroppedDelta: delta("fanout_rows_dropped_total"),
LakePartitionsStart: lakePartitionsStart,
LakePartitions: final.total("fanout_lake_partitions"),
LakePartitionsDelta: final.total("fanout_lake_partitions") - lakePartitionsStart,
LakeSizeBytesStart: lakeSizeStart,
LakeSizeBytes: final.total("fanout_lake_size_bytes"),
LakeSizeBytesDelta: final.total("fanout_lake_size_bytes") - lakeSizeStart,
LakeGrowthBytesPerSec: rate(final.total("fanout_lake_size_bytes") - lakeSizeStart),
IngestQueueDepth: final.total("fanout_ingest_queue_depth"),
AvgRollupMs: averageDurationMs(base, final, "fanout_rollup_duration_seconds"),
AvgFlushMs: averageDurationMs(base, final, "fanout_flush_duration_seconds"),
AvgQueryMs: averageDurationMs(base, final, "fanout_query_duration_seconds"),
CPUSecondsStart: round4(base.total("process_cpu_seconds_total")),
CPUSecondsEnd: round4(final.total("process_cpu_seconds_total")),
CPUSecondsDelta: round4(cpuSeconds),
CPUCores: perSecond(cpuSeconds, durationSeconds),
RSSBytes: final.total("process_resident_memory_bytes"),
HeapAllocBytes: final.total("go_memstats_heap_alloc_bytes"),
AllocBytesStart: base.total("go_memstats_alloc_bytes_total"),
AllocBytesEnd: final.total("go_memstats_alloc_bytes_total"),
AllocBytesDelta: allocBytes,
AllocBytesPerSec: rate(allocBytes),
GCPauseSecondsStart: round4(base.total("go_gc_duration_seconds_sum")),
GCPauseSecondsEnd: round4(final.total("go_gc_duration_seconds_sum")),
GCPauseSecondsDelta: round4(delta("go_gc_duration_seconds_sum")),
WriteGateWaitMs: histogramReports(base, final, "fanout_write_gate_wait_seconds", "operation"),
WriteGateHoldMs: histogramReports(base, final, "fanout_write_gate_hold_seconds", "operation"),
DuckLakeOperations: backgroundReports(base, final),
Rollups: rollupReports(base, final),
BaselineAvailable: baselineAvailable,
ProcessStartTime: startTimeAfter,
ProcessRestarted: baselineAvailable && startTimeBefore > 0 && startTimeAfter != startTimeBefore,
IngestRowsStart: base.total("fanout_ingest_rows_total"),
IngestRowsEnd: final.total("fanout_ingest_rows_total"),
IngestRowsDelta: delta("fanout_ingest_rows_total"),
RowsDroppedStart: base.total("fanout_rows_dropped_total"),
RowsDroppedEnd: final.total("fanout_rows_dropped_total"),
RowsDroppedDelta: delta("fanout_rows_dropped_total"),
ParquetFilesStart: parquetFilesStart,
ParquetFiles: final.total("fanout_parquet_files"),
ParquetFilesDelta: final.total("fanout_parquet_files") - parquetFilesStart,
ParquetSizeBytesStart: parquetSizeStart,
ParquetSizeBytes: final.total("fanout_parquet_size_bytes"),
ParquetSizeBytesDelta: final.total("fanout_parquet_size_bytes") - parquetSizeStart,
ParquetGrowthBytesPerSec: rate(final.total("fanout_parquet_size_bytes") - parquetSizeStart),
IngestQueueDepth: final.total("fanout_ingest_queue_depth"),
AvgRollupMs: averageDurationMs(base, final, "fanout_rollup_duration_seconds"),
AvgFlushMs: averageDurationMs(base, final, "fanout_flush_duration_seconds"),
AvgQueryMs: averageDurationMs(base, final, "fanout_query_duration_seconds"),
CPUSecondsStart: round4(base.total("process_cpu_seconds_total")),
CPUSecondsEnd: round4(final.total("process_cpu_seconds_total")),
CPUSecondsDelta: round4(cpuSeconds),
CPUCores: perSecond(cpuSeconds, durationSeconds),
RSSBytes: final.total("process_resident_memory_bytes"),
HeapAllocBytes: final.total("go_memstats_heap_alloc_bytes"),
AllocBytesStart: base.total("go_memstats_alloc_bytes_total"),
AllocBytesEnd: final.total("go_memstats_alloc_bytes_total"),
AllocBytesDelta: allocBytes,
AllocBytesPerSec: rate(allocBytes),
GCPauseSecondsStart: round4(base.total("go_gc_duration_seconds_sum")),
GCPauseSecondsEnd: round4(final.total("go_gc_duration_seconds_sum")),
GCPauseSecondsDelta: round4(delta("go_gc_duration_seconds_sum")),
WriteGateWaitMs: histogramReports(base, final, "fanout_write_gate_wait_seconds", "operation"),
WriteGateHoldMs: histogramReports(base, final, "fanout_write_gate_hold_seconds", "operation"),
TelemetryOperations: backgroundReports(base, final),
Rollups: rollupReports(base, final),
}
}

Expand Down Expand Up @@ -425,10 +425,10 @@ func histogramDelta(base, final *metricSnapshot, name string, filters map[string

func backgroundReports(base, final *metricSnapshot) map[string]backgroundOperationReport {
operations := unionStrings(
base.labelValues("fanout_ducklake_operation_total", "operation", nil),
final.labelValues("fanout_ducklake_operation_total", "operation", nil),
base.labelValues("fanout_ducklake_operation_duration_seconds_count", "operation", nil),
final.labelValues("fanout_ducklake_operation_duration_seconds_count", "operation", nil),
base.labelValues("fanout_telemetry_operation_total", "operation", nil),
final.labelValues("fanout_telemetry_operation_total", "operation", nil),
base.labelValues("fanout_telemetry_operation_duration_seconds_count", "operation", nil),
final.labelValues("fanout_telemetry_operation_duration_seconds_count", "operation", nil),
)
if len(operations) == 0 {
return nil
Expand All @@ -437,8 +437,8 @@ func backgroundReports(base, final *metricSnapshot) map[string]backgroundOperati
for _, operation := range operations {
filters := map[string]string{"operation": operation}
reports[operation] = backgroundOperationReport{
DurationMs: histogramDelta(base, final, "fanout_ducklake_operation_duration_seconds", filters),
Outcomes: counterOutcomes(base, final, "fanout_ducklake_operation_total", "result", filters),
DurationMs: histogramDelta(base, final, "fanout_telemetry_operation_duration_seconds", filters),
Outcomes: counterOutcomes(base, final, "fanout_telemetry_operation_total", "result", filters),
}
}
return reports
Expand Down
Loading
Loading