package wshub import ( "fmt" "net/http" "sort" "strconv" "strings" "time" "github.com/parquet-go/parquet-go" ) // ExportSample is one row of the binary export: a single stored sample, in // long ("tidy") form, keyed by source and signal with its own timestamp. // // Keeping each signal's samples as its own rows — rather than resampling onto a // shared time grid — is what makes the export hole-free: per-signal streams of // different lengths export exactly as stored, nothing is fabricated, and // nothing is dropped. type ExportSample struct { Source string `parquet:"source"` Signal string `parquet:"signal"` Time float64 `parquet:"time"` Value float64 `parquet:"value"` } // exportChunkRows bounds each batched write and, via MaxRowsPerRowGroup, the // size of each parquet row group: memory stays bounded however large the // export is, because a finished row group is flushed to the HTTP stream. const exportChunkRows = 65536 // exportWriteBuffer is the parquet writer's output buffer: larger than the // 32KiB default means fewer writes on the HTTP stream for a multi-GB export. const exportWriteBuffer = 1 << 20 // HandleExport serves GET /api/export?t0=..&t1=..[&signals=a,b] as a Parquet // file containing every stored sample of the named signals in [t0, t1]. // // Unlike /api/zoom there is no decimation: the file holds the full contents of // the rings. At rates above the ring budget those contents are min/max buckets // (the finest resolution the hub retains); at lower rates they are verbatim. func (h *Hub) HandleExport(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() t0, err0 := strconv.ParseFloat(q.Get("t0"), 64) t1, err1 := strconv.ParseFloat(q.Get("t1"), 64) if err0 != nil || err1 != nil || t1 <= t0 { http.Error(w, "invalid t0/t1", http.StatusBadRequest) return } var keys []string if s := strings.TrimSpace(q.Get("signals")); s != "" { keys = strings.Split(s, ",") for i := range keys { keys[i] = strings.TrimSpace(keys[i]) } } // Snapshot the rings we will read. A signal removed mid-export must not // silently drop rows from the file. h.ringsMu.RLock() refs := make(map[string]*sigRing) if keys == nil { for k, rb := range h.rings { refs[k] = rb } } else { for _, k := range keys { if rb, ok := h.rings[k]; ok { refs[k] = rb } } } h.ringsMu.RUnlock() if len(refs) == 0 { http.Error(w, "no signals", http.StatusNotFound) return } // Deterministic column order. names := make([]string, 0, len(refs)) for k := range refs { names = append(names, k) } sort.Strings(names) w.Header().Set("Content-Type", "application/vnd.apache.parquet") w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"signals_%d.parquet\"", time.Now().Unix())) writer := parquet.NewGenericWriter[ExportSample](w, parquet.MaxRowsPerRowGroup(exportChunkRows), parquet.WriteBufferSize(exportWriteBuffer), ) batch := make([]ExportSample, 0, exportChunkRows) for _, key := range names { st, sv := refs[key].slice(t0, t1) colon := strings.IndexByte(key, ':') source, signal := key, key if colon >= 0 { source = key[:colon] signal = key[colon+1:] } for i := range st { batch = append(batch, ExportSample{Source: source, Signal: signal, Time: st[i], Value: sv[i]}) if len(batch) >= exportChunkRows { if _, err := writer.Write(batch); err != nil { // Client went away or the stream broke; stop writing. return } batch = batch[:0] } } } if len(batch) > 0 { _, _ = writer.Write(batch) } _ = writer.Close() }