SY-3411: Fix CSV exporting with large amounts of data (#1758)

Performance improvement to the client-side CSV exporting:

Move main CSV exporting logic to the client, with an API that we can hopefully keep stable / mostly stable even when we move the heavy data processing work to the backend
Make CSV export return a ReadableStream to allow for processing off of the main thread for desktop + certain browsers
performance improvements, much faster export speeds now

Author: pjdotson

client/py/examples/multi-rate-calculated/simulated_daq.py

@@ -14,6 +14,7 @@
 demonstrate how to calculate derived values from these channels with different rates.
 """
 
+import random
 import time
 
 import numpy as np
@@ -71,21 +72,25 @@
     i = 0
     while rough_rate.wait():
         time = sy.TimeStamp.now()
-        time_2 = time + sy.TimeSpan.MILLISECOND * 3
+        time_2 = time + sy.TimeSpan.MICROSECOND * 3
         # Generate data to write to the first channel.
         data_to_write = {
             time_ch_1.key: [time, time_2],
             data_ch_1.key: [np.sin(i / 10), np.sin((i + 1) / 10)],
         }
 
-        # Only write to the second channel every third iteration, so its rate is 10Hz
-        # instead of 30Hz.
+        # Only write to the second channel every third iteration, so its rate is 10 Hz
+        # instead of 30 Hz.
         if i % 3 == 0:
-            # Generate timestamps at a different time to introduce intentional
-            # misalignment.
-            time = sy.TimeStamp.now()
-            time_2 = time + sy.TimeSpan.MILLISECOND
-            data_to_write[time_ch_2.key] = [time, time_2]
+            # Generate timestamps at a random time that is off by between -5 and +5
+            # nanoseconds using random.randint
+            second_time = time + sy.TimeSpan.NANOSECOND * random.randint(-5, 5)
+            second_time_2 = (
+                second_time
+                + sy.TimeSpan.MICROSECOND * 3
+                + sy.TimeSpan.NANOSECOND * random.randint(-5, 5)
+            )
+            data_to_write[time_ch_2.key] = [second_time, second_time_2]
             data_to_write[data_ch_2.key] = [np.sin(i / 100), np.sin((i + 1) / 100)]
 
         writer.write(data_to_write)

client/ts/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@synnaxlabs/client",
-  "version": "0.49.0",
+  "version": "0.49.2",
   "description": "The Synnax Client Library",
   "keywords": [
     "synnax",

client/ts/src/framer/client.ts

@@ -21,6 +21,7 @@ import { channel } from "@/channel";
 import { Deleter } from "@/framer/deleter";
 import { Frame } from "@/framer/frame";
 import { AUTO_SPAN, Iterator, type IteratorConfig } from "@/framer/iterator";
+import { Reader, type ReadRequest } from "@/framer/reader";
 import { openStreamer, type Streamer, type StreamerConfig } from "@/framer/streamer";
 import { Writer, type WriterConfig, WriterMode } from "@/framer/writer";
 import { ontology } from "@/ontology";
@@ -32,6 +33,7 @@ export class Client {
   private readonly streamClient: WebSocketClient;
   private readonly retriever: channel.Retriever;
   private readonly deleter: Deleter;
+  private readonly reader: Reader;
 
   constructor(
     stream: WebSocketClient,
@@ -41,6 +43,7 @@ export class Client {
     this.streamClient = stream;
     this.retriever = retriever;
     this.deleter = new Deleter(unary);
+    this.reader = new Reader(retriever, stream);
   }
 
   /**
@@ -140,15 +143,15 @@ export class Client {
   }
 
   async read(tr: CrudeTimeRange, channel: channel.KeyOrName): Promise<MultiSeries>;
-
   async read(tr: CrudeTimeRange, channels: channel.Params): Promise<Frame>;
-
+  async read(request: ReadRequest): Promise<ReadableStream<Uint8Array>>;
   async read(
-    tr: CrudeTimeRange,
-    channels: channel.Params,
-  ): Promise<MultiSeries | Frame> {
-    const { single } = channel.analyzeParams(channels);
-    const fr = await this.readFrame(tr, channels);
+    tr: CrudeTimeRange | ReadRequest,
+    channels?: channel.Params,
+  ): Promise<MultiSeries | Frame | ReadableStream<Uint8Array>> {
+    if (!("start" in tr)) return this.reader.read(tr);
+    const { single } = channel.analyzeParams(channels!);
+    const fr = await this.readFrame(tr, channels!);
     if (single) return fr.get(channels as channel.KeyOrName);
     return fr;
   }
@@ -199,13 +202,7 @@ export class Client {
     const { normalized, variant } = channel.analyzeParams(channels);
     const bounds = new TimeRange(timeRange);
     if (variant === "keys")
-      return await this.deleter.delete({
-        keys: normalized as channel.Key[],
-        bounds,
-      });
-    return await this.deleter.delete({
-      names: normalized as string[],
-      bounds,
-    });
+      return await this.deleter.delete({ keys: normalized as channel.Key[], bounds });
+    return await this.deleter.delete({ names: normalized as string[], bounds });
   }
 }