refactor(aura-ingest): chunk-direct M1 transpose, drop the AoS intermediate

load_m1_window collected every bar into an intermediate Vec<M1Parsed> (a full
AoS materialization, un-presized so it grew by reallocation) before transpose_m1
copied it again into the SoA columns — ~2x peak load memory plus realloc churn.

Transpose each chunk straight into the columns instead, via a new shared
M1Columns::extend_from_bars (reserves per chunk for amortized growth). One fewer
full copy, halved peak. Behaviour-preserving: the resulting M1Columns is
byte-identical and transpose_m1's / load_m1_window's signatures are unchanged.
Pinned by a new parity test that needs no live DataServer.

closes #78
This commit is contained in:
2026-06-17 09:59:58 +02:00
parent 3de00e23e0
commit 6390093f93
+49 -14
View File
@@ -62,6 +62,31 @@ impl M1Columns {
}
}
/// Append one chunk of AoS bars, transposing into these SoA columns and
/// normalizing time ms->epoch-ns at the boundary (C3). Reserves per chunk so
/// repeated appends grow amortized; the push order is preserved (the
/// chronological merge order C3 relies on). Lets `load_m1_window` transpose
/// chunk-by-chunk without an intermediate AoS buffer.
fn extend_from_bars(&mut self, bars: &[M1Parsed]) {
let n = bars.len();
self.ts.reserve(n);
self.open.reserve(n);
self.high.reserve(n);
self.low.reserve(n);
self.close.reserve(n);
self.spread.reserve(n);
self.volume.reserve(n);
for b in bars {
self.ts.push(unix_ms_to_epoch_ns(b.time_ms));
self.open.push(b.open);
self.high.push(b.high);
self.low.push(b.low);
self.close.push(b.close);
self.spread.push(b.spread);
self.volume.push(b.volume);
}
}
/// The close column as an engine source stream: each normalized timestamp
/// zipped with its close as a [`Scalar::f64`]. Ascending in timestamp iff
/// the bars were (the C3 ingestion precondition `Harness::run` relies on).
@@ -81,21 +106,14 @@ impl M1Columns {
/// columns (C1) — it reads no clock and no external state.
pub fn transpose_m1(bars: &[M1Parsed]) -> M1Columns {
let mut c = M1Columns::with_capacity(bars.len());
for b in bars {
c.ts.push(unix_ms_to_epoch_ns(b.time_ms));
c.open.push(b.open);
c.high.push(b.high);
c.low.push(b.low);
c.close.push(b.close);
c.spread.push(b.spread);
c.volume.push(b.volume);
}
c.extend_from_bars(bars);
c
}
/// Drain a data-server M1 window into transposed SoA columns. Reads the
/// chronological chunk iterator to exhaustion into one AoS buffer, then
/// transposes once at the boundary (C3 — one merge point, not per chunk).
/// chronological chunk iterator to exhaustion, transposing each chunk directly
/// into the SoA columns at the boundary (C3 — one merge point; no intermediate
/// AoS buffer, so peak memory is the columns alone, not columns + an AoS copy).
///
/// Each bound is an `Option<i64>` of inclusive Unix-ms (data-server's window
/// contract): `None` means unbounded on that side (data-server skips the
@@ -122,12 +140,15 @@ pub fn load_m1_window(
to_ms: Option<i64>,
) -> Option<M1Columns> {
let mut it = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
let mut bars: Vec<M1Parsed> = Vec::new();
// Transpose each chunk straight into the SoA columns — no intermediate AoS
// buffer (halves peak load memory, drops one full copy). The per-chunk push
// order is the chronological merge order (C3).
let mut cols = M1Columns::with_capacity(0);
// M1Parsed is Copy; &Arc<[M1Parsed]> derefs to &[M1Parsed] in arg position.
while let Some(chunk) = it.next_chunk() {
bars.extend_from_slice(&chunk);
cols.extend_from_bars(&chunk);
}
Some(transpose_m1(&bars))
Some(cols)
}
/// Which base column of an M1 bar a source streams (C7: a composite window is a
@@ -301,6 +322,20 @@ mod tests {
assert!(c.volume.is_empty());
}
#[test]
fn chunked_accumulation_equals_single_transpose() {
// `load_m1_window` now transposes chunk-by-chunk into the SoA columns
// (no intermediate AoS buffer). That must be byte-identical to one
// transpose over the concatenation — the path it replaced. This pins the
// shared `extend_from_bars` logic without needing a live DataServer.
let all = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
let single = transpose_m1(&all);
let mut chunked = M1Columns::with_capacity(0);
chunked.extend_from_bars(&all[0..2]); // first chunk
chunked.extend_from_bars(&all[2..3]); // second chunk
assert_eq!(chunked, single);
}
#[test]
fn close_stream_zips_ts_with_close_in_order() {
// distinct close per bar so order is observable; ts ascending.