From 8021f859f5b3a4ebf144f5fc4c8a588ffac4cfc5 Mon Sep 17 00:00:00 2001 From: Bharat Kathi Date: Thu, 6 Aug 2026 20:45:45 -0700 Subject: [PATCH] perf(gr26): batch signal inserts into a single round trip --- gr26/service/signal.go | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/gr26/service/signal.go b/gr26/service/signal.go index 0009f056..7c3b020f 100644 --- a/gr26/service/signal.go +++ b/gr26/service/signal.go @@ -14,9 +14,12 @@ import ( // produced_at is MATERIALIZED from timestamp; created_at defaults to insert // wall clock and acts as the ReplacingMergeTree version. Both omitted here. -const insertSignalSQL = `INSERT INTO signal (id, timestamp, vehicle_id, name, value, raw_value) VALUES (?, ?, ?, ?, ?, ?)` +const insertSignalSQL = `INSERT INTO signal (id, timestamp, vehicle_id, name, value, raw_value)` func CreateSignals(signals []mapache.Signal) error { + if len(signals) == 0 { + return nil + } // Stamp CreatedAt locally before any MQTT publish or CH insert so // downstream consumers (live cache eviction, SSE Last-Event-ID) have // a reliable wall-clock anchor independent of producer/CAN-frame clock @@ -39,13 +42,18 @@ func CreateSignals(signals []mapache.Signal) error { if !config.ClickhouseEnabled() { return nil } - ctx := database.InsertCtx(context.Background()) + // One buffered block insert per call — a single round trip regardless of + // signal count, which matters with CH ~25ms away from foundry. async_insert + // (InsertCtx) doesn't apply to native block inserts, so plain ctx here. + batch, err := database.Conn.PrepareBatch(context.Background(), insertSignalSQL) + if err != nil { + return err + } + defer batch.Close() for _, s := range signals { - if err := database.Conn.Exec(ctx, insertSignalSQL, - s.ID, int64(s.Timestamp), s.VehicleID, s.Name, s.Value, int64(s.RawValue), - ); err != nil { + if err := batch.Append(s.ID, int64(s.Timestamp), s.VehicleID, s.Name, s.Value, int64(s.RawValue)); err != nil { return err } } - return nil + return batch.Send() }