Skip to main content

kui_channel/
lib.rs

1//! A data channel (ADR 0065): bulk data the core owns and a handle names,
2//! written once and read by index from anywhere that holds the handle — an
3//! extension's fill, another Lua state, a worker thread — so a hundred
4//! thousand rows cross into a component as one small value instead of a
5//! copy every time they change.
6//!
7//! A channel is a table of named, typed columns of one length, or a run of
8//! bytes. It is immutable per revision: [`Channel::publish`] makes the next
9//! revision under the same identity, and a reader holding an older one
10//! keeps it until it lets go. Its handle is a value —
11//! `{"$channel": id, "rev": rev}` — so it rides in props, params and
12//! messages like any other, and [`Channel::resolve`] turns one back into the
13//! latest live revision of that identity.
14//!
15//! A crate of its own (ADR 0075, decision 3), on `kui-value` alone, so
16//! what has no kui core — the recording's reader — makes channels too;
17//! `kui-core` re-exports it as `kui_core::channel`. With the `lua`
18//! feature, [`lua`] is a channel as a script holds it.
19//!
20//! ```rust
21//! use kui_channel::{Cell, Channel, Column};
22//!
23//! let ch = Channel::table(vec![
24//!     ("name".into(), Column::strs(["build", "test", "deploy"])),
25//!     ("ms".into(), Column::F64(vec![12.5, 30.0, 4.0])),
26//! ])
27//! .unwrap();
28//! assert_eq!(ch.len(), 3);
29//! let handle = ch.handle();
30//! let again = Channel::resolve(&handle).unwrap();
31//! let ms = again.col("ms").unwrap();
32//! assert_eq!(again.get(1, ms), Cell::F64(30.0));
33//! ```
34
35use std::sync::atomic::{AtomicU64, Ordering};
36use std::sync::{Arc, LazyLock, Mutex, OnceLock, Weak};
37
38use rustc_hash::FxHashMap;
39
40use kui_value::Value;
41
42mod lanes;
43#[cfg(feature = "lua")]
44pub mod lua;
45
46pub use lanes::{Lanes, Run};
47
48/// The key a handle's identity rides under.
49pub const HANDLE_KEY: &str = "$channel";
50/// The key its revision rides under.
51pub const REV_KEY: &str = "rev";
52
53/// One column: every row's value of one field, all of one type.
54#[derive(Clone, Debug, PartialEq)]
55pub enum Column {
56    F64(Vec<f64>),
57    I64(Vec<i64>),
58    Bool(Vec<bool>),
59    Str(Strs),
60}
61
62impl Column {
63    /// A string column from anything that yields strings.
64    pub fn strs<S: AsRef<str>>(items: impl IntoIterator<Item = S>) -> Column {
65        Column::Str(Strs::from_iter(items))
66    }
67
68    pub fn len(&self) -> usize {
69        match self {
70            Column::F64(v) => v.len(),
71            Column::I64(v) => v.len(),
72            Column::Bool(v) => v.len(),
73            Column::Str(s) => s.len(),
74        }
75    }
76
77    pub fn is_empty(&self) -> bool {
78        self.len() == 0
79    }
80
81    /// The column's type, as the bindings name it.
82    pub fn kind(&self) -> &'static str {
83        match self {
84            Column::F64(_) => "number",
85            Column::I64(_) => "integer",
86            Column::Bool(_) => "boolean",
87            Column::Str(_) => "string",
88        }
89    }
90
91    /// An empty column of this one's kind.
92    fn empty_like(&self) -> Column {
93        match self {
94            Column::F64(_) => Column::F64(Vec::new()),
95            Column::I64(_) => Column::I64(Vec::new()),
96            Column::Bool(_) => Column::Bool(Vec::new()),
97            Column::Str(_) => Column::Str(Strs::default()),
98        }
99    }
100
101    /// `other`'s rows `range` after this column's, of the same kind.
102    fn extend_rows(&mut self, other: &Column, range: std::ops::Range<usize>) {
103        match (self, other) {
104            (Column::F64(a), Column::F64(b)) => a.extend_from_slice(&b[range]),
105            (Column::I64(a), Column::I64(b)) => a.extend_from_slice(&b[range]),
106            (Column::Bool(a), Column::Bool(b)) => a.extend_from_slice(&b[range]),
107            (Column::Str(a), Column::Str(b)) => {
108                for i in range {
109                    a.push(b.get(i).unwrap_or_default());
110                }
111            }
112            _ => unreachable!("runs of one column are of one kind"),
113        }
114    }
115
116    /// Row `i`'s cell; `Cell::Null` past the end.
117    pub fn get(&self, i: usize) -> Cell<'_> {
118        match self {
119            Column::F64(v) => v.get(i).map_or(Cell::Null, |x| Cell::F64(*x)),
120            Column::I64(v) => v.get(i).map_or(Cell::Null, |x| Cell::I64(*x)),
121            Column::Bool(v) => v.get(i).map_or(Cell::Null, |x| Cell::Bool(*x)),
122            Column::Str(s) => s.get(i).map_or(Cell::Null, Cell::Str),
123        }
124    }
125}
126
127/// A string column: one buffer and where each string ends, so a hundred
128/// thousand names are two allocations.
129#[derive(Clone, Debug, Default, PartialEq)]
130pub struct Strs {
131    bytes: String,
132    ends: Vec<u32>,
133}
134
135impl Strs {
136    pub fn push(&mut self, s: &str) {
137        self.bytes.push_str(s);
138        self.ends.push(self.bytes.len() as u32);
139    }
140
141    pub fn len(&self) -> usize {
142        self.ends.len()
143    }
144
145    pub fn is_empty(&self) -> bool {
146        self.ends.is_empty()
147    }
148
149    pub fn get(&self, i: usize) -> Option<&str> {
150        let end = *self.ends.get(i)? as usize;
151        let start = if i == 0 { 0 } else { self.ends[i - 1] as usize };
152        self.bytes.get(start..end)
153    }
154
155    pub fn iter(&self) -> impl Iterator<Item = &str> {
156        (0..self.len()).filter_map(|i| self.get(i))
157    }
158}
159
160impl<S: AsRef<str>> FromIterator<S> for Strs {
161    fn from_iter<I: IntoIterator<Item = S>>(items: I) -> Self {
162        let mut out = Strs::default();
163        for s in items {
164            out.push(s.as_ref());
165        }
166        out
167    }
168}
169
170/// One cell, borrowed from its channel.
171#[derive(Clone, Copy, Debug, PartialEq)]
172pub enum Cell<'a> {
173    Null,
174    F64(f64),
175    I64(i64),
176    Bool(bool),
177    Str(&'a str),
178}
179
180impl Cell<'_> {
181    /// The cell as a value.
182    pub fn to_value(self) -> Value {
183        match self {
184            Cell::Null => Value::Null,
185            Cell::F64(x) => Value::Float(x),
186            Cell::I64(x) => Value::Int(x),
187            Cell::Bool(x) => Value::Bool(x),
188            Cell::Str(s) => Value::str(s),
189        }
190    }
191
192    /// A number cell as an `f64`.
193    pub fn as_f64(self) -> Option<f64> {
194        match self {
195            Cell::F64(x) => Some(x),
196            Cell::I64(x) => Some(x as f64),
197            _ => None,
198        }
199    }
200}
201
202/// What a channel holds.
203#[derive(Clone, Debug, PartialEq)]
204pub enum Data {
205    /// Named columns of one length.
206    Table {
207        names: Vec<String>,
208        cols: Vec<Column>,
209    },
210    /// A run of bytes.
211    Bytes(Vec<u8>),
212}
213
214/// Why a channel could not be made.
215#[derive(Clone, Debug, PartialEq)]
216pub struct ChannelError(pub String);
217
218impl std::fmt::Display for ChannelError {
219    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
220        f.write_str(&self.0)
221    }
222}
223
224impl std::error::Error for ChannelError {}
225
226/// A run of a table's rows, shared between the revisions that hold it:
227/// an append adds one, so a revision costs its new rows and not its old
228/// ones (backlog C147). Its rows in the channel are `start..`, its own
229/// from `skip` on — a cap drops the oldest from the front.
230#[derive(Clone)]
231struct Chunk {
232    start: usize,
233    skip: usize,
234    cols: Arc<Vec<Column>>,
235}
236
237impl Chunk {
238    fn rows(&self) -> usize {
239        self.cols.first().map_or(0, Column::len) - self.skip
240    }
241}
242
243enum Body {
244    /// The columns' names, the runs of rows, and each column whole, made
245    /// once a revision when a reader asks for that one whole and there is
246    /// more than one run (or the first is cut). One at a time: a table's
247    /// filters, sorts and takes read run by run and join none (backlog
248    /// RG172), so a column read whole costs that column, not the table.
249    Table {
250        names: Vec<String>,
251        chunks: Vec<Chunk>,
252        flat: Vec<OnceLock<Column>>,
253        data: OnceLock<Data>,
254    },
255    Bytes(Data),
256}
257
258struct Inner {
259    id: u64,
260    rev: u64,
261    body: Body,
262    len: usize,
263}
264
265impl Inner {
266    /// Column `c` whole.
267    fn whole(&self, c: usize) -> Option<&Column> {
268        let Body::Table { chunks, flat, .. } = &self.body else {
269            return None;
270        };
271        let slot = flat.get(c)?;
272        if let [one] = chunks.as_slice()
273            && one.skip == 0
274        {
275            return one.cols.get(c);
276        }
277        Some(slot.get_or_init(|| {
278            let mut out = chunks[0].cols[c].empty_like();
279            for ch in chunks {
280                out.extend_rows(&ch.cols[c], ch.skip..ch.skip + ch.rows());
281            }
282            out
283        }))
284    }
285
286    /// The runs a table's rows are in, empty for bytes.
287    fn chunks(&self) -> &[Chunk] {
288        match &self.body {
289            Body::Table { chunks, .. } => chunks,
290            Body::Bytes(_) => &[],
291        }
292    }
293
294    /// Row `row`'s run and its index in that run's columns.
295    fn locate(&self, row: usize) -> (usize, usize) {
296        let chunks = self.chunks();
297        let k = match chunks {
298            [_] => 0,
299            many => many.partition_point(|c| c.start <= row) - 1,
300        };
301        (k, row - chunks[k].start + chunks[k].skip)
302    }
303
304    /// Calls `f` with each row's run and index — of `within`, in its order,
305    /// or of every row in row order, a run at a time with no search.
306    fn each_row(&self, within: Option<&[usize]>, mut f: impl FnMut(usize, usize, usize)) {
307        match within {
308            Some(rows) => {
309                for &r in rows {
310                    let (k, i) = self.locate(r);
311                    f(r, k, i);
312                }
313            }
314            None => {
315                for (k, ch) in self.chunks().iter().enumerate() {
316                    for i in ch.skip..ch.skip + ch.rows() {
317                        f(ch.start + i - ch.skip, k, i);
318                    }
319                }
320            }
321        }
322    }
323}
324
325/// A slot a revision fills when column `c` is first read whole.
326fn whole_slots(n: usize) -> Vec<OnceLock<Column>> {
327    (0..n).map(|_| OnceLock::new()).collect()
328}
329
330/// One revision of a channel, held: cheap to clone, and the data stays as
331/// long as some clone (here or in another binding) does.
332#[derive(Clone)]
333pub struct Channel(Arc<Inner>);
334
335impl std::fmt::Debug for Channel {
336    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
337        f.debug_struct("Channel")
338            .field("id", &self.0.id)
339            .field("rev", &self.0.rev)
340            .field("len", &self.0.len)
341            .finish()
342    }
343}
344
345impl PartialEq for Channel {
346    /// The same identity at the same revision: what a replay compares.
347    fn eq(&self, other: &Self) -> bool {
348        self.0.id == other.0.id && self.0.rev == other.0.rev
349    }
350}
351
352/// Every identity's latest revision, weakly: a channel nobody holds is
353/// gone, and its handle resolves to nothing.
354static REGISTRY: LazyLock<Mutex<FxHashMap<u64, Weak<Inner>>>> =
355    LazyLock::new(|| Mutex::new(FxHashMap::default()));
356static NEXT_ID: AtomicU64 = AtomicU64::new(1);
357
358fn registry() -> std::sync::MutexGuard<'static, FxHashMap<u64, Weak<Inner>>> {
359    REGISTRY.lock().unwrap_or_else(|e| e.into_inner())
360}
361
362fn table_len(names: &[String], cols: &[Column]) -> Result<usize, ChannelError> {
363    if names.len() != cols.len() {
364        return Err(ChannelError(format!(
365            "{} names for {} columns",
366            names.len(),
367            cols.len()
368        )));
369    }
370    let mut seen = std::collections::HashSet::new();
371    for n in names {
372        if !seen.insert(n.as_str()) {
373            return Err(ChannelError(format!("column `{n}` is named twice")));
374        }
375    }
376    let len = cols.first().map_or(0, Column::len);
377    if let Some((i, c)) = cols.iter().enumerate().find(|(_, c)| c.len() != len) {
378        return Err(ChannelError(format!(
379            "column `{}` has {} rows and `{}` has {len}: every column of a channel is one length",
380            names[i],
381            c.len(),
382            names[0]
383        )));
384    }
385    Ok(len)
386}
387
388/// At most this many runs before a reader by row searches them: past it
389/// the runs merge however their sizes stand.
390const MAX_CHUNKS: usize = 48;
391
392/// Merges the last two runs while the last is as long as the one before
393/// it, so a table appended to a row at a time holds about log₂ of its rows
394/// in runs and each row is copied about that many times in all.
395fn settle(chunks: &mut Vec<Chunk>) {
396    while chunks.len() >= 2 {
397        let n = chunks.len();
398        let (a, b) = (&chunks[n - 2], &chunks[n - 1]);
399        if b.rows() < a.rows() && n <= MAX_CHUNKS {
400            break;
401        }
402        let cols = (0..a.cols.len())
403            .map(|c| {
404                let mut out = a.cols[c].empty_like();
405                out.extend_rows(&a.cols[c], a.skip..a.skip + a.rows());
406                out.extend_rows(&b.cols[c], b.skip..b.skip + b.rows());
407                out
408            })
409            .collect();
410        let start = a.start;
411        chunks.truncate(n - 2);
412        chunks.push(Chunk {
413            start,
414            skip: 0,
415            cols: Arc::new(cols),
416        });
417    }
418}
419
420impl Channel {
421    fn register(id: u64, rev: u64, body: Body, len: usize) -> Channel {
422        let inner = Arc::new(Inner { id, rev, body, len });
423        let mut reg = registry();
424        // A sweep of the gone, now and then: the map holds weak entries,
425        // and one per identity ever made would grow with a busy writer.
426        if reg.len() > 256 && reg.len().is_power_of_two() {
427            reg.retain(|_, w| w.strong_count() > 0);
428        }
429        reg.insert(id, Arc::downgrade(&inner));
430        Channel(inner)
431    }
432
433    fn table_body(names: Vec<String>, cols: Vec<Column>) -> Body {
434        Body::Table {
435            flat: whole_slots(names.len()),
436            names,
437            chunks: vec![Chunk {
438                start: 0,
439                skip: 0,
440                cols: Arc::new(cols),
441            }],
442            data: OnceLock::new(),
443        }
444    }
445
446    /// A table of named columns of one length.
447    pub fn table(cols: Vec<(String, Column)>) -> Result<Channel, ChannelError> {
448        let (names, cols): (Vec<String>, Vec<Column>) = cols.into_iter().unzip();
449        let len = table_len(&names, &cols)?;
450        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
451        Ok(Self::register(id, 1, Self::table_body(names, cols), len))
452    }
453
454    /// Restores detached contents at a saved revision under a fresh identity.
455    /// Live channels and their readers cannot alias the restored history.
456    pub fn from_snapshot(data: Data, revision: u64) -> Result<Channel, ChannelError> {
457        let (body, len) = match data {
458            Data::Table { names, cols } => {
459                let len = table_len(&names, &cols)?;
460                (Self::table_body(names, cols), len)
461            }
462            Data::Bytes(bytes) => {
463                let len = bytes.len();
464                (Body::Bytes(Data::Bytes(bytes)), len)
465            }
466        };
467        Ok(Self::register(
468            NEXT_ID.fetch_add(1, Ordering::Relaxed),
469            revision.max(1),
470            body,
471            len,
472        ))
473    }
474
475    /// A run of bytes.
476    pub fn bytes(bytes: Vec<u8>) -> Channel {
477        let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
478        let len = bytes.len();
479        Self::register(id, 1, Body::Bytes(Data::Bytes(bytes)), len)
480    }
481
482    /// The next revision under this identity, holding `cols`: a handle to
483    /// the identity resolves to it from now on, and this revision stays
484    /// readable for as long as it is held.
485    pub fn publish(&self, cols: Vec<(String, Column)>) -> Result<Channel, ChannelError> {
486        let (names, cols): (Vec<String>, Vec<Column>) = cols.into_iter().unzip();
487        let len = table_len(&names, &cols)?;
488        Ok(Self::register(
489            self.0.id,
490            self.0.rev + 1,
491            Self::table_body(names, cols),
492            len,
493        ))
494    }
495
496    /// [`Self::publish`] for bytes.
497    pub fn publish_bytes(&self, bytes: Vec<u8>) -> Channel {
498        let len = bytes.len();
499        Self::register(
500            self.0.id,
501            self.0.rev + 1,
502            Body::Bytes(Data::Bytes(bytes)),
503            len,
504        )
505    }
506
507    /// The next revision under this identity: this one's rows and `rows`
508    /// after them (backlog C147). `rows` names every column this table
509    /// has, each of the same kind — integers widen into a number column,
510    /// and numbers appended to a column of integers widen the column, its
511    /// rows so far read as numbers from this revision on (backlog RG173):
512    /// a log of milliseconds that began on whole ones takes its first
513    /// fraction — all of one length. With `cap`, the oldest rows go past it, so a log
514    /// keeps its last `cap` lines. The rows this revision holds are shared
515    /// with the next, not copied: an append costs its own rows.
516    pub fn append(
517        &self,
518        rows: Vec<(String, Column)>,
519        cap: Option<usize>,
520    ) -> Result<Channel, ChannelError> {
521        let Body::Table { names, chunks, .. } = &self.0.body else {
522            return Err(ChannelError("a channel of bytes takes no rows".into()));
523        };
524        if rows.len() != names.len() {
525            return Err(ChannelError(format!(
526                "{} columns appended to a channel of {}: {}",
527                rows.len(),
528                names.len(),
529                names.join(", ")
530            )));
531        }
532        let mut given: Vec<Option<Column>> = vec![None; names.len()];
533        let mut widen = vec![false; names.len()];
534        let mut n = None;
535        for (name, col) in rows {
536            let Some(i) = names.iter().position(|x| *x == name) else {
537                return Err(ChannelError(format!(
538                    "the channel has no column `{name}`; it has {}",
539                    names.join(", ")
540                )));
541            };
542            if given[i].is_some() {
543                return Err(ChannelError(format!("column `{name}` is appended twice")));
544            }
545            let have = &chunks[0].cols[i];
546            let col = match (have, col) {
547                (Column::F64(_), Column::I64(v)) => {
548                    Column::F64(v.into_iter().map(|x| x as f64).collect())
549                }
550                (Column::I64(_), c @ Column::F64(_)) if !c.is_empty() => {
551                    widen[i] = true;
552                    c
553                }
554                (h, c) if h.kind() == c.kind() => c,
555                // No rows have no kind: a binding's empty list reads as
556                // numbers, which a string column takes as none of its own.
557                (h, c) if c.is_empty() => h.empty_like(),
558                (h, c) => {
559                    return Err(ChannelError(format!(
560                        "column `{name}` holds {}s; {}s were appended to it",
561                        h.kind(),
562                        c.kind()
563                    )));
564                }
565            };
566            match n {
567                None => n = Some(col.len()),
568                Some(n) if n != col.len() => {
569                    return Err(ChannelError(format!(
570                        "column `{name}` appends {} rows where another appends {n}",
571                        col.len()
572                    )));
573                }
574                Some(_) => {}
575            }
576            given[i] = Some(col);
577        }
578        let added = n.unwrap_or(0);
579        let mut chunks = chunks.clone();
580        // The runs so far, their widened columns as numbers: new runs,
581        // since the old revision keeps its integers. Once a column's life.
582        if widen.contains(&true) {
583            for ch in &mut chunks {
584                let mut cols = (*ch.cols).clone();
585                for (c, w) in cols.iter_mut().zip(&widen) {
586                    if *w && let Column::I64(v) = c {
587                        *c = Column::F64(v.iter().map(|&x| x as f64).collect());
588                    }
589                }
590                ch.cols = Arc::new(cols);
591            }
592        }
593        let mut len = self.0.len;
594        if added > 0 {
595            chunks.push(Chunk {
596                start: len,
597                skip: 0,
598                cols: Arc::new(given.into_iter().map(Option::unwrap).collect()),
599            });
600            len += added;
601            settle(&mut chunks);
602        }
603        if let Some(cap) = cap
604            && len > cap
605        {
606            let mut drop = len - cap;
607            len = cap;
608            while drop > 0 && !chunks.is_empty() {
609                let rows = chunks[0].rows();
610                if rows <= drop && chunks.len() > 1 {
611                    chunks.remove(0);
612                    drop -= rows;
613                } else {
614                    chunks[0].skip += drop.min(rows);
615                    drop = 0;
616                }
617            }
618            let mut start = 0;
619            for c in &mut chunks {
620                c.start = start;
621                start += c.rows();
622            }
623        }
624        Ok(Self::register(
625            self.0.id,
626            self.0.rev + 1,
627            Body::Table {
628                names: names.clone(),
629                chunks,
630                flat: whole_slots(names.len()),
631                data: OnceLock::new(),
632            },
633            len,
634        ))
635    }
636
637    /// The next revision under this identity, holding `source`'s rows as
638    /// they stand — shared, not copied: a view showing what it views
639    /// unfiltered, its handle staying the view's.
640    pub fn publish_from(&self, source: &Channel) -> Channel {
641        let body = match &source.0.body {
642            Body::Table { names, chunks, .. } => Body::Table {
643                names: names.clone(),
644                chunks: chunks.clone(),
645                flat: whole_slots(names.len()),
646                data: OnceLock::new(),
647            },
648            Body::Bytes(d) => Body::Bytes(d.clone()),
649        };
650        Self::register(self.0.id, self.0.rev + 1, body, source.0.len)
651    }
652
653    pub fn id(&self) -> u64 {
654        self.0.id
655    }
656
657    pub fn rev(&self) -> u64 {
658        self.0.rev
659    }
660
661    /// Rows for a table, bytes for bytes.
662    pub fn len(&self) -> usize {
663        self.0.len
664    }
665
666    pub fn is_empty(&self) -> bool {
667        self.0.len == 0
668    }
669
670    /// What it holds, whole: a table appended to is put together once,
671    /// on the first ask.
672    pub fn data(&self) -> &Data {
673        match &self.0.body {
674            Body::Bytes(d) => d,
675            Body::Table { names, data, .. } => data.get_or_init(|| Data::Table {
676                names: names.clone(),
677                cols: (0..names.len())
678                    .map(|c| self.0.whole(c).cloned().expect("a named column"))
679                    .collect(),
680            }),
681        }
682    }
683
684    /// The column names, in order; empty for bytes.
685    pub fn names(&self) -> &[String] {
686        match &self.0.body {
687            Body::Table { names, .. } => names,
688            Body::Bytes(_) => &[],
689        }
690    }
691
692    /// The index of the column called `name`.
693    pub fn col(&self, name: &str) -> Option<usize> {
694        self.names().iter().position(|n| n == name)
695    }
696
697    /// The column at `i`, whole: a table appended to puts it together
698    /// once a revision, on the first ask — that column alone;
699    /// [`Self::get`] reads a row without, and the filters, sorts, takes
700    /// and reductions read run by run.
701    pub fn column(&self, i: usize) -> Option<&Column> {
702        self.0.whole(i)
703    }
704
705    /// The column called `name`.
706    pub fn column_named(&self, name: &str) -> Option<&Column> {
707        self.column(self.col(name)?)
708    }
709
710    /// Row `row` of column `col`; `Cell::Null` out of range.
711    pub fn get(&self, row: usize, col: usize) -> Cell<'_> {
712        let Body::Table { chunks, .. } = &self.0.body else {
713            return Cell::Null;
714        };
715        if row >= self.0.len {
716            return Cell::Null;
717        }
718        let at = match chunks.as_slice() {
719            [one] => one,
720            many => &many[many.partition_point(|c| c.start <= row) - 1],
721        };
722        at.cols
723            .get(col)
724            .map_or(Cell::Null, |c| c.get(row - at.start + at.skip))
725    }
726
727    /// The bytes, for a bytes channel.
728    pub fn as_bytes(&self) -> Option<&[u8]> {
729        match &self.0.body {
730            Body::Bytes(Data::Bytes(b)) => Some(b),
731            _ => None,
732        }
733    }
734
735    /// The handle: `{"$channel": id, "rev": rev}`, a value that rides
736    /// anywhere a value does.
737    pub fn handle(&self) -> Value {
738        handle_value(self.0.id, self.0.rev)
739    }
740
741    /// The latest live revision of the identity `v` names, if `v` is a
742    /// handle and some side still holds one.
743    pub fn resolve(v: &Value) -> Option<Channel> {
744        let (id, _) = handle_of(v)?;
745        registry().get(&id)?.upgrade().map(Channel)
746    }
747}
748
749/// What [`Channel::filter`] keeps a row for (backlog C148).
750#[derive(Clone, Debug, PartialEq)]
751pub enum Match {
752    /// The cell equals this: a number to a number column (an integer to
753    /// an integer one exactly), a string to a string, a boolean to a
754    /// boolean.
755    Eq(Value),
756    /// A string cell holds `text`; `case` false matches it whatever the
757    /// letters' case.
758    Contains { text: String, case: bool },
759    /// A number cell is at least `min` and at most `max`, either left
760    /// open; NaN is in no range.
761    Range { min: Option<f64>, max: Option<f64> },
762}
763
764/// `needle` in `hay`, ASCII letters of either case alike.
765fn contains_ascii_blind(hay: &str, needle: &str) -> bool {
766    let (h, n) = (hay.as_bytes(), needle.as_bytes());
767    n.is_empty() || h.windows(n.len()).any(|w| w.eq_ignore_ascii_case(n))
768}
769
770impl Channel {
771    /// Column `col` of the first run: the column's kind, and its rows
772    /// when the table is one run. No column `col` is an error.
773    pub(crate) fn column_index(&self, col: usize) -> Result<&Column, ChannelError> {
774        self.0
775            .chunks()
776            .first()
777            .and_then(|ch| ch.cols.get(col))
778            .ok_or_else(|| {
779                ChannelError(format!(
780                    "the channel has no column {col}; it has {}",
781                    self.names().len()
782                ))
783            })
784    }
785
786    /// Column `col`'s numbers over rows `from..to`, integers widened,
787    /// read from the runs that hold them: what a reduction or the lanes
788    /// read without the column joined whole. Column `col` is numbers or
789    /// integers.
790    pub(crate) fn numbers_in(&self, col: usize, from: usize, to: usize) -> Vec<f64> {
791        let mut out = Vec::with_capacity(to.saturating_sub(from));
792        for ch in self.0.chunks() {
793            let (a, z) = (from.max(ch.start), to.min(ch.start + ch.rows()));
794            if a >= z {
795                continue;
796            }
797            let (i, j) = (a - ch.start + ch.skip, z - ch.start + ch.skip);
798            match &ch.cols[col] {
799                Column::F64(v) => out.extend_from_slice(&v[i..j]),
800                Column::I64(v) => out.extend(v[i..j].iter().map(|&x| x as f64)),
801                _ => {}
802            }
803        }
804        out
805    }
806
807    /// Each row's cell `f` reads from its run's column `col` — of
808    /// `within`, or every row — with the row: what a sort orders.
809    fn keyed<'a, T>(
810        &'a self,
811        col: usize,
812        within: Option<&[usize]>,
813        f: impl Fn(&'a Column, usize) -> T,
814    ) -> Vec<(T, usize)> {
815        let chunks = self.0.chunks();
816        let mut out = Vec::with_capacity(within.map_or(self.len(), <[usize]>::len));
817        self.0
818            .each_row(within, |r, k, i| out.push((f(&chunks[k].cols[col], i), r)));
819        out
820    }
821
822    /// `within`, when given, holds rows of this channel.
823    fn check_rows(&self, within: Option<&[usize]>) -> Result<(), ChannelError> {
824        let len = self.len();
825        match within.and_then(|rows| rows.iter().find(|&&r| r >= len)) {
826            Some(r) => Err(ChannelError(format!(
827                "row {r} is past the channel's {len} rows"
828            ))),
829            None => Ok(()),
830        }
831    }
832
833    /// The rows whose cell in column `col` matches `m`, in row order —
834    /// all of them, or those of `within` in its order (backlog C148): a
835    /// filter a worker or a table builds a view from without the rows
836    /// crossing into a binding. A match of the wrong kind for the column
837    /// is an error.
838    pub fn filter(
839        &self,
840        col: usize,
841        m: &Match,
842        within: Option<&[usize]>,
843    ) -> Result<Vec<usize>, ChannelError> {
844        let kind = self.column_index(col)?;
845        self.check_rows(within)?;
846        let lower = match m {
847            Match::Contains { text, case: false } if !text.is_ascii() => Some(text.to_lowercase()),
848            _ => None,
849        };
850        // A test a run: the runs of one column are of one kind, so the
851        // first decides whether the match is one this column takes.
852        let mut tests = Vec::with_capacity(self.0.chunks().len());
853        for ch in self.0.chunks() {
854            let Some(t) = matcher(&ch.cols[col], m, lower.as_deref()) else {
855                let what = match m {
856                    Match::Eq(_) => "that value",
857                    Match::Contains { .. } => "a string search",
858                    Match::Range { .. } => "a range",
859                };
860                return Err(ChannelError(format!(
861                    "column `{}` holds {}s; {what} matches none",
862                    self.names()[col],
863                    kind.kind()
864                )));
865            };
866            tests.push(t);
867        }
868        let mut out = Vec::new();
869        self.0.each_row(within, |r, k, i| {
870            if tests[k](i) {
871                out.push(r);
872            }
873        });
874        Ok(out)
875    }
876
877    /// The rows — all, or `within`'s — ordered by column `col`, least
878    /// first or with `descending` greatest first (backlog C148). Rows of
879    /// one value keep their order; NaN goes last either way; strings
880    /// compare by their bytes, `false` before `true`.
881    pub fn sort(
882        &self,
883        col: usize,
884        descending: bool,
885        within: Option<&[usize]>,
886    ) -> Result<Vec<usize>, ChannelError> {
887        use std::cmp::Ordering as O;
888        let kind = self.column_index(col)?;
889        self.check_rows(within)?;
890        let flip = |o: O| if descending { o.reverse() } else { o };
891        // Each row's key read from its run, then a stable sort of the
892        // keys: the rows of one value keep the order they came in.
893        fn rows<T>(keyed: Vec<(T, usize)>) -> Vec<usize> {
894            keyed.into_iter().map(|(_, r)| r).collect()
895        }
896        Ok(match kind {
897            Column::F64(_) => {
898                let mut k = self.keyed(col, within, |c, i| match c {
899                    Column::F64(v) => v[i],
900                    _ => f64::NAN,
901                });
902                k.sort_by(|a, b| match (a.0.is_nan(), b.0.is_nan()) {
903                    (false, false) => flip(a.0.total_cmp(&b.0)),
904                    (x, y) => x.cmp(&y),
905                });
906                rows(k)
907            }
908            Column::I64(_) => {
909                let mut k = self.keyed(col, within, |c, i| match c {
910                    Column::I64(v) => v[i],
911                    _ => 0,
912                });
913                k.sort_by(|a, b| flip(a.0.cmp(&b.0)));
914                rows(k)
915            }
916            Column::Bool(_) => {
917                let mut k = self.keyed(col, within, |c, i| match c {
918                    Column::Bool(v) => v[i],
919                    _ => false,
920                });
921                k.sort_by(|a, b| flip(a.0.cmp(&b.0)));
922                rows(k)
923            }
924            Column::Str(_) => {
925                let mut k = self.keyed(col, within, |c, i| match c {
926                    Column::Str(v) => v.get(i),
927                    _ => None,
928                });
929                k.sort_by(|a, b| flip(a.0.cmp(&b.0)));
930                rows(k)
931            }
932        })
933    }
934
935    /// The columns of `rows`, in that order, under this channel's names:
936    /// what [`Self::take`] makes a channel of, and what a view publishes
937    /// as the next revision of its own.
938    pub fn take_columns(&self, rows: &[usize]) -> Result<Vec<(String, Column)>, ChannelError> {
939        self.check_rows(Some(rows))?;
940        if self.names().is_empty() {
941            return Ok(Vec::new());
942        }
943        // Each row found in its run once, for every column: the rows taken
944        // are copied, the table is not joined (backlog RG172).
945        let at: Vec<(usize, usize)> = rows.iter().map(|&r| self.0.locate(r)).collect();
946        let chunks = self.0.chunks();
947        Ok(self
948            .names()
949            .iter()
950            .enumerate()
951            .map(|(c, name)| {
952                let cell = |&(k, i): &(usize, usize)| chunks[k].cols[c].get(i);
953                let col = match &chunks[0].cols[c] {
954                    Column::F64(_) => Column::F64(
955                        at.iter()
956                            .map(|p| cell(p).as_f64().unwrap_or(f64::NAN))
957                            .collect(),
958                    ),
959                    Column::I64(_) => Column::I64(
960                        at.iter()
961                            .map(|p| match cell(p) {
962                                Cell::I64(x) => x,
963                                _ => 0,
964                            })
965                            .collect(),
966                    ),
967                    Column::Bool(_) => Column::Bool(
968                        at.iter()
969                            .map(|p| matches!(cell(p), Cell::Bool(true)))
970                            .collect(),
971                    ),
972                    Column::Str(_) => Column::Str(
973                        at.iter()
974                            .map(|p| match cell(p) {
975                                Cell::Str(s) => s,
976                                _ => "",
977                            })
978                            .collect(),
979                    ),
980                };
981                (name.clone(), col)
982            })
983            .collect())
984    }
985
986    /// A new channel of `rows`, in that order (backlog C148): a filtered
987    /// or sorted view, made in Rust.
988    pub fn take(&self, rows: &[usize]) -> Result<Channel, ChannelError> {
989        if self.as_bytes().is_some() {
990            return Err(ChannelError(
991                "a channel of bytes has no rows to take".into(),
992            ));
993        }
994        Channel::table(self.take_columns(rows)?)
995    }
996}
997
998/// A filter's test of one run's column `column` by its index there;
999/// `None` for a match of a kind the column is not. `lower` is a
1000/// case-blind search's text lowercased, when it is not ASCII.
1001type Test<'a> = Box<dyn Fn(usize) -> bool + 'a>;
1002
1003fn matcher<'a>(column: &'a Column, m: &'a Match, lower: Option<&'a str>) -> Option<Test<'a>> {
1004    Some(match (column, m) {
1005        (Column::F64(v), Match::Eq(x)) => {
1006            let x = x.as_float()?;
1007            Box::new(move |i| v[i] == x)
1008        }
1009        (Column::I64(v), Match::Eq(Value::Int(x))) => Box::new(move |i| v[i] == *x),
1010        (Column::I64(v), Match::Eq(Value::Float(x))) => Box::new(move |i| v[i] as f64 == *x),
1011        (Column::Bool(v), Match::Eq(Value::Bool(x))) => Box::new(move |i| v[i] == *x),
1012        (Column::Str(v), Match::Eq(Value::Str(x))) => {
1013            Box::new(move |i| v.get(i) == Some(x.as_str()))
1014        }
1015        (Column::Str(v), Match::Contains { text, case: true }) => {
1016            Box::new(move |i| v.get(i).is_some_and(|s| s.contains(text.as_str())))
1017        }
1018        (Column::Str(v), Match::Contains { text, case: false }) if text.is_ascii() => {
1019            Box::new(move |i| v.get(i).is_some_and(|s| contains_ascii_blind(s, text)))
1020        }
1021        (Column::Str(v), Match::Contains { case: false, .. }) => {
1022            let text = lower?;
1023            Box::new(move |i| v.get(i).is_some_and(|s| s.to_lowercase().contains(text)))
1024        }
1025        (Column::F64(v), Match::Range { min, max }) => {
1026            let (lo, hi) = (
1027                min.unwrap_or(f64::NEG_INFINITY),
1028                max.unwrap_or(f64::INFINITY),
1029            );
1030            Box::new(move |i| v[i] >= lo && v[i] <= hi)
1031        }
1032        (Column::I64(v), Match::Range { min, max }) => {
1033            let (lo, hi) = (
1034                min.unwrap_or(f64::NEG_INFINITY),
1035                max.unwrap_or(f64::INFINITY),
1036            );
1037            Box::new(move |i| {
1038                let x = v[i] as f64;
1039                x >= lo && x <= hi
1040            })
1041        }
1042        _ => return None,
1043    })
1044}
1045
1046/// One bucket of a [`Channel::reduce`]: the least and greatest value
1047/// among its rows, the first and the last, and the row each was read
1048/// from — a row of the channel, not of the range. A NaN is no value: it
1049/// is skipped, and a bucket of nothing but NaN reads NaN for all four,
1050/// every row its own first, so a line drawn through it breaks there.
1051/// Ties go to the earliest row.
1052#[derive(Clone, Copy, Debug, PartialEq)]
1053pub struct Bucket {
1054    pub min: f64,
1055    pub max: f64,
1056    pub first: f64,
1057    pub last: f64,
1058    pub min_row: usize,
1059    pub max_row: usize,
1060    pub first_row: usize,
1061    pub last_row: usize,
1062}
1063
1064impl Channel {
1065    /// Column `col` over rows `from..to` in `buckets` buckets of as near
1066    /// one size as whole rows allow, each one's [`Bucket`] (backlog C95):
1067    /// what a chart draws a hundred thousand points as — per pixel column
1068    /// the first, least, greatest and last, in row order, which is the
1069    /// line the rows would draw — without the rows crossing into the
1070    /// binding. A range with fewer rows than `buckets` gets a bucket a
1071    /// row; `to` past the end stops at it, and an empty range (`from` at
1072    /// or past `to` or the end, or no buckets) gets none. The column is
1073    /// numbers or integers (widened); any other, or no column `col`, is
1074    /// an error.
1075    pub fn reduce(
1076        &self,
1077        col: usize,
1078        from: usize,
1079        to: usize,
1080        buckets: usize,
1081    ) -> Result<Vec<Bucket>, ChannelError> {
1082        let mut out = Vec::new();
1083        self.reduce_into(col, from, to, buckets, &mut out)?;
1084        Ok(out)
1085    }
1086
1087    /// [`Self::reduce`] into `out`, cleared first: a chart that reduces
1088    /// every frame keeps one buffer.
1089    pub fn reduce_into(
1090        &self,
1091        col: usize,
1092        from: usize,
1093        to: usize,
1094        buckets: usize,
1095        out: &mut Vec<Bucket>,
1096    ) -> Result<(), ChannelError> {
1097        out.clear();
1098        let column = self.column_index(col)?;
1099        let to = to.min(self.len());
1100        let n = to.saturating_sub(from);
1101        let k = buckets.min(n);
1102        match column {
1103            // Checked before the range, so a mistake does not wait for
1104            // the data to show.
1105            Column::F64(_) | Column::I64(_) if k == 0 => {}
1106            Column::F64(_) | Column::I64(_) => {
1107                // A range inside one run reads in place; one across runs
1108                // reads its own rows out of them, not the column whole.
1109                let (r, i) = self.0.locate(from);
1110                let ch = &self.0.chunks()[r];
1111                let one = to <= ch.start + ch.rows();
1112                let read;
1113                let v: &[f64] = match &ch.cols[col] {
1114                    Column::F64(v) if one => &v[i..i + n],
1115                    Column::I64(v) if one => {
1116                        reduce_rows(&v[i..i + n], from, k, out, |x| x as f64);
1117                        return Ok(());
1118                    }
1119                    _ => {
1120                        read = self.numbers_in(col, from, to);
1121                        &read
1122                    }
1123                };
1124                reduce_rows(v, from, k, out, |x| x)
1125            }
1126            other => return Err(not_numbers(&self.names()[col], other)),
1127        }
1128        Ok(())
1129    }
1130}
1131
1132fn not_numbers(name: &str, c: &Column) -> ChannelError {
1133    ChannelError(format!(
1134        "column `{name}` holds {}s; reduce reads a column of numbers",
1135        c.kind()
1136    ))
1137}
1138
1139/// The buckets of `rows`, which begin at row `base`: `k` of them, bucket
1140/// `b` holding rows `b·n/k ..(b+1)·n/k`. One pass per bucket, its first
1141/// and last found from its ends and its extremes from the first on, NaN
1142/// failing every comparison and so skipped by them.
1143fn reduce_rows<T: Copy>(
1144    rows: &[T],
1145    base: usize,
1146    k: usize,
1147    out: &mut Vec<Bucket>,
1148    f: impl Fn(T) -> f64,
1149) {
1150    out.reserve(k);
1151    let n = rows.len() as u128;
1152    let edge = |b: usize| (b as u128 * n / k as u128) as usize;
1153    for b in 0..k {
1154        let (a, z) = (edge(b), edge(b + 1));
1155        let bucket = &rows[a..z];
1156        let Some(i) = bucket.iter().position(|&x| !f(x).is_nan()) else {
1157            let row = base + a;
1158            out.push(Bucket {
1159                min: f64::NAN,
1160                max: f64::NAN,
1161                first: f64::NAN,
1162                last: f64::NAN,
1163                min_row: row,
1164                max_row: row,
1165                first_row: row,
1166                last_row: row,
1167            });
1168            continue;
1169        };
1170        let j = bucket.iter().rposition(|&x| !f(x).is_nan()).unwrap_or(i);
1171        let first = f(bucket[i]);
1172        let ((min, lo), (max, hi)) = extremes(&bucket[i..], &f);
1173        let (lo, hi) = (lo + i, hi + i);
1174        out.push(Bucket {
1175            min,
1176            max,
1177            first,
1178            last: f(bucket[j]),
1179            min_row: base + a + lo,
1180            max_row: base + a + hi,
1181            first_row: base + a + i,
1182            last_row: base + a + j,
1183        });
1184    }
1185}
1186
1187/// The least and greatest of `rows`, which starts with a number, and
1188/// where each is first found. Four lanes, each a run of every fourth
1189/// row, so the comparisons of one do not wait on another's: one lane was
1190/// a comparison's latency a row, 1 ns, the four a quarter of it. A NaN
1191/// fails every comparison and is passed over; a lane that has seen only
1192/// NaN holds the first row's number, which is no harm, since that number
1193/// is in the run.
1194#[inline]
1195fn extremes<T: Copy>(rows: &[T], f: &impl Fn(T) -> f64) -> ((f64, usize), (f64, usize)) {
1196    let x0 = f(rows[0]);
1197    let mut min = [x0; 4];
1198    let mut max = [x0; 4];
1199    let mut lo = [0usize; 4];
1200    let mut hi = [0usize; 4];
1201    let (quads, rest) = rows.as_chunks::<4>();
1202    for (q, quad) in quads.iter().enumerate() {
1203        for l in 0..4 {
1204            let x = f(quad[l]);
1205            let at = q * 4 + l;
1206            if x < min[l] {
1207                min[l] = x;
1208                lo[l] = at;
1209            }
1210            if x > max[l] {
1211                max[l] = x;
1212                hi[l] = at;
1213            }
1214        }
1215    }
1216    let mut least = (min[0], lo[0]);
1217    let mut most = (max[0], hi[0]);
1218    // The lanes' answers, ties to the earliest row as one pass gives.
1219    for l in 1..4 {
1220        if min[l] < least.0 || (min[l] == least.0 && lo[l] < least.1) {
1221            least = (min[l], lo[l]);
1222        }
1223        if max[l] > most.0 || (max[l] == most.0 && hi[l] < most.1) {
1224            most = (max[l], hi[l]);
1225        }
1226    }
1227    for (r, &x) in rest.iter().enumerate() {
1228        let x = f(x);
1229        let at = quads.len() * 4 + r;
1230        if x < least.0 {
1231            least = (x, at);
1232        }
1233        if x > most.0 {
1234            most = (x, at);
1235        }
1236    }
1237    (least, most)
1238}
1239
1240/// A handle's value for an identity and revision.
1241pub fn handle_value(id: u64, rev: u64) -> Value {
1242    Value::map([
1243        (HANDLE_KEY, Value::Int(id as i64)),
1244        (REV_KEY, Value::Int(rev as i64)),
1245    ])
1246}
1247
1248/// The identity and revision a handle names, or `None` for any other value.
1249pub fn handle_of(v: &Value) -> Option<(u64, u64)> {
1250    let Value::Map(entries) = v else {
1251        return None;
1252    };
1253    let mut id = None;
1254    let mut rev = 0;
1255    for (k, x) in entries {
1256        match (k.as_str(), x) {
1257            (HANDLE_KEY, Value::Int(i)) => id = Some(*i as u64),
1258            (REV_KEY, Value::Int(r)) => rev = *r as u64,
1259            _ => return None,
1260        }
1261    }
1262    id.map(|id| (id, rev))
1263}
1264
1265/// Whether `v` is a channel handle.
1266pub fn is_handle(v: &Value) -> bool {
1267    handle_of(v).is_some()
1268}
1269
1270/// A column from a list of values all of one kind — numbers (integers
1271/// stay integers only when every one is), booleans or strings — what the
1272/// bindings build from their own arrays. `None` names the first value that
1273/// breaks the kind.
1274pub fn column_of(values: &[Value]) -> Result<Column, ChannelError> {
1275    let Some(first) = values.first() else {
1276        return Ok(Column::F64(Vec::new()));
1277    };
1278    let bad = |i: usize, want: &str| {
1279        Err(ChannelError(format!(
1280            "row {} is {}, in a column of {want}s",
1281            i + 1,
1282            match &values[i] {
1283                Value::Null => "nothing",
1284                Value::Bool(_) => "a boolean",
1285                Value::Int(_) | Value::Float(_) => "a number",
1286                Value::Str(_) => "a string",
1287                Value::List(_) => "a list",
1288                Value::Map(_) => "a map",
1289            }
1290        )))
1291    };
1292    match first {
1293        Value::Int(_) | Value::Float(_) => {
1294            if values.iter().all(|v| matches!(v, Value::Int(_))) {
1295                Ok(Column::I64(
1296                    values.iter().map(|v| v.as_int().unwrap_or(0)).collect(),
1297                ))
1298            } else {
1299                let mut out = Vec::with_capacity(values.len());
1300                for (i, v) in values.iter().enumerate() {
1301                    match v {
1302                        Value::Int(x) => out.push(*x as f64),
1303                        Value::Float(x) => out.push(*x),
1304                        _ => return bad(i, "number"),
1305                    }
1306                }
1307                Ok(Column::F64(out))
1308            }
1309        }
1310        Value::Bool(_) => {
1311            let mut out = Vec::with_capacity(values.len());
1312            for (i, v) in values.iter().enumerate() {
1313                match v {
1314                    Value::Bool(b) => out.push(*b),
1315                    _ => return bad(i, "boolean"),
1316                }
1317            }
1318            Ok(Column::Bool(out))
1319        }
1320        Value::Str(_) => {
1321            let mut out = Strs::default();
1322            for (i, v) in values.iter().enumerate() {
1323                match v {
1324                    Value::Str(s) => out.push(s),
1325                    _ => return bad(i, "string"),
1326                }
1327            }
1328            Ok(Column::Str(out))
1329        }
1330        _ => bad(0, "number, boolean or string"),
1331    }
1332}
1333
1334#[cfg(test)]
1335mod tests {
1336    use super::*;
1337
1338    #[test]
1339    fn a_reduction_finds_each_buckets_extremes_and_ends() {
1340        let ch = Channel::table(vec![(
1341            "y".into(),
1342            Column::F64(vec![3.0, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0]),
1343        )])
1344        .unwrap();
1345        let r = ch.reduce(0, 0, 8, 2).unwrap();
1346        assert_eq!(r.len(), 2);
1347        assert_eq!(
1348            (r[0].min, r[0].min_row),
1349            (1.0, 1),
1350            "ties go to the earliest row"
1351        );
1352        assert_eq!((r[0].max, r[0].max_row), (4.0, 2));
1353        assert_eq!(
1354            (r[0].first, r[0].first_row, r[0].last, r[0].last_row),
1355            (3.0, 0, 1.0, 3)
1356        );
1357        assert_eq!(
1358            (r[1].min, r[1].min_row, r[1].max, r[1].max_row),
1359            (2.0, 6, 9.0, 5)
1360        );
1361        // Rows are the channel's, not the range's.
1362        let r = ch.reduce(0, 4, 8, 1).unwrap();
1363        assert_eq!((r[0].first_row, r[0].max_row, r[0].last_row), (4, 5, 7));
1364        // Integers widen.
1365        let ints = Channel::table(vec![("n".into(), Column::I64(vec![5, -2, 7]))]).unwrap();
1366        let r = ints.reduce(0, 0, 3, 1).unwrap();
1367        assert_eq!((r[0].min, r[0].max), (-2.0, 7.0));
1368    }
1369
1370    #[test]
1371    fn a_reduction_at_its_edges() {
1372        let ch = Channel::table(vec![
1373            ("y".into(), Column::F64(vec![1.0, 2.0, 3.0, 4.0, 5.0])),
1374            ("s".into(), Column::strs(["a", "b", "c", "d", "e"])),
1375        ])
1376        .unwrap();
1377        // Empty ranges and no buckets: nothing.
1378        assert!(ch.reduce(0, 3, 3, 10).unwrap().is_empty());
1379        assert!(ch.reduce(0, 4, 2, 10).unwrap().is_empty());
1380        assert!(ch.reduce(0, 9, 20, 10).unwrap().is_empty());
1381        assert!(ch.reduce(0, 0, 5, 0).unwrap().is_empty());
1382        // More buckets than rows: a bucket a row, `to` stopped at the end.
1383        let r = ch.reduce(0, 1, 99, 10).unwrap();
1384        assert_eq!(r.len(), 4);
1385        for (i, b) in r.iter().enumerate() {
1386            assert_eq!((b.first_row, b.last_row), (i + 1, i + 1));
1387            assert_eq!((b.min, b.max, b.last), (b.first, b.first, b.first));
1388        }
1389        // Buckets of near one size: 5 rows in 2 are 2 and 3.
1390        let r = ch.reduce(0, 0, 5, 2).unwrap();
1391        assert_eq!((r[0].last_row, r[1].first_row), (1, 2));
1392        // A column that is not numbers, or none, is an error, even empty.
1393        assert!(ch.reduce(1, 0, 5, 2).unwrap_err().0.contains("strings"));
1394        assert!(ch.reduce(1, 0, 0, 2).is_err());
1395        assert!(ch.reduce(7, 0, 5, 2).is_err());
1396    }
1397
1398    /// The four lanes answer what one pass would, ties and NaN included,
1399    /// over lengths that leave every remainder and values drawn from a
1400    /// few, so ties are common.
1401    #[test]
1402    fn the_lanes_agree_with_one_pass() {
1403        let mut seed = 0x2545_f491u32;
1404        for len in 1..40usize {
1405            for _ in 0..20 {
1406                let rows: Vec<f64> = (0..len)
1407                    .map(|i| {
1408                        seed ^= seed << 13;
1409                        seed ^= seed >> 17;
1410                        seed ^= seed << 5;
1411                        if i > 0 && seed.is_multiple_of(5) {
1412                            f64::NAN
1413                        } else {
1414                            (seed % 7) as f64
1415                        }
1416                    })
1417                    .collect();
1418                let (mut min, mut max) = ((rows[0], 0), (rows[0], 0));
1419                for (at, &x) in rows.iter().enumerate() {
1420                    if x < min.0 {
1421                        min = (x, at);
1422                    }
1423                    if x > max.0 {
1424                        max = (x, at);
1425                    }
1426                }
1427                assert_eq!(extremes(&rows, &|x| x), (min, max), "{rows:?}");
1428            }
1429        }
1430    }
1431
1432    #[test]
1433    fn a_reduction_skips_nan() {
1434        let nan = f64::NAN;
1435        let ch = Channel::table(vec![(
1436            "y".into(),
1437            Column::F64(vec![nan, 2.0, nan, -1.0, nan, nan, nan, nan, 7.0]),
1438        )])
1439        .unwrap();
1440        let r = ch.reduce(0, 0, 9, 3).unwrap();
1441        assert_eq!(
1442            (r[0].first, r[0].first_row, r[0].last, r[0].last_row),
1443            (2.0, 1, 2.0, 1)
1444        );
1445        assert_eq!(
1446            (r[1].min, r[1].min_row, r[1].max, r[1].max_row),
1447            (-1.0, 3, -1.0, 3)
1448        );
1449        assert_eq!((r[2].first_row, r[2].first), (8, 7.0));
1450        // A bucket of nothing but NaN reads NaN, at its first row.
1451        let r = ch.reduce(0, 4, 8, 2).unwrap();
1452        assert!(
1453            r.iter()
1454                .all(|b| b.min.is_nan() && b.max.is_nan() && b.first.is_nan() && b.last.is_nan())
1455        );
1456        assert_eq!((r[0].min_row, r[1].last_row), (4, 6));
1457    }
1458
1459    #[test]
1460    fn a_table_reads_by_row_and_column() {
1461        let ch = Channel::table(vec![
1462            ("a".into(), Column::I64(vec![1, 2, 3])),
1463            ("b".into(), Column::strs(["x", "", "zz"])),
1464            ("c".into(), Column::Bool(vec![true, false, true])),
1465        ])
1466        .unwrap();
1467        assert_eq!(ch.len(), 3);
1468        assert_eq!(ch.get(2, 0), Cell::I64(3));
1469        assert_eq!(ch.get(1, 1), Cell::Str(""));
1470        assert_eq!(ch.get(2, 1), Cell::Str("zz"));
1471        assert_eq!(ch.get(3, 1), Cell::Null);
1472        assert_eq!(ch.column_named("c").unwrap().kind(), "boolean");
1473    }
1474
1475    #[test]
1476    fn columns_of_two_lengths_or_one_name_twice_are_refused() {
1477        assert!(
1478            Channel::table(vec![
1479                ("a".into(), Column::I64(vec![1])),
1480                ("b".into(), Column::I64(vec![1, 2])),
1481            ])
1482            .is_err()
1483        );
1484        assert!(
1485            Channel::table(vec![
1486                ("a".into(), Column::I64(vec![1])),
1487                ("a".into(), Column::I64(vec![2])),
1488            ])
1489            .is_err()
1490        );
1491    }
1492
1493    #[test]
1494    fn a_handle_resolves_while_held_and_follows_the_latest_revision() {
1495        let first = Channel::table(vec![("n".into(), Column::F64(vec![1.0]))]).unwrap();
1496        let h = first.handle();
1497        assert_eq!(Channel::resolve(&h), Some(first.clone()));
1498        let second = first
1499            .publish(vec![("n".into(), Column::F64(vec![1.0, 2.0]))])
1500            .unwrap();
1501        assert_eq!(second.id(), first.id());
1502        assert_eq!(second.rev(), 2);
1503        // The old handle names the identity; it resolves to the latest.
1504        assert_eq!(Channel::resolve(&h).unwrap().len(), 2);
1505        // The old revision stays readable while held.
1506        assert_eq!(first.len(), 1);
1507        let h2 = second.handle();
1508        drop(first);
1509        drop(second);
1510        assert!(Channel::resolve(&h2).is_none(), "nobody holds it");
1511        assert!(handle_of(&Value::map([("$channel", Value::str("x"))])).is_none());
1512    }
1513
1514    #[test]
1515    fn a_column_is_built_from_values_of_one_kind() {
1516        let ints = column_of(&[Value::Int(1), Value::Int(2)]).unwrap();
1517        assert_eq!(ints, Column::I64(vec![1, 2]));
1518        let mixed = column_of(&[Value::Int(1), Value::Float(2.5)]).unwrap();
1519        assert_eq!(mixed, Column::F64(vec![1.0, 2.5]));
1520        let err = column_of(&[Value::str("a"), Value::Int(1)]).unwrap_err();
1521        assert!(err.0.contains("row 2 is a number"), "{err}");
1522    }
1523
1524    fn log(n: usize) -> Channel {
1525        Channel::table(vec![
1526            ("line".into(), Column::strs((0..n).map(|i| format!("l{i}")))),
1527            ("ms".into(), Column::F64((0..n).map(|i| i as f64).collect())),
1528        ])
1529        .unwrap()
1530    }
1531
1532    fn rows(lines: &[&str], ms: &[i64]) -> Vec<(String, Column)> {
1533        vec![
1534            ("ms".into(), Column::I64(ms.to_vec())),
1535            ("line".into(), Column::strs(lines)),
1536        ]
1537    }
1538
1539    /// An append is the next revision with the rows after this one's —
1540    /// columns by name, integers widened into a number column — and the
1541    /// old revision keeps its rows (backlog C147).
1542    #[test]
1543    fn an_append_adds_rows_and_the_old_revision_keeps_its_own() {
1544        let a = log(2);
1545        let b = a.append(rows(&["x", "y"], &[10, 11]), None).unwrap();
1546        assert_eq!((b.id(), b.rev(), b.len()), (a.id(), 2, 4));
1547        assert_eq!(b.get(3, 0), Cell::Str("y"));
1548        assert_eq!(b.get(2, 1), Cell::F64(10.0));
1549        assert_eq!(a.len(), 2);
1550        assert_eq!(a.get(2, 0), Cell::Null);
1551        assert_eq!(Channel::resolve(&a.handle()).unwrap().len(), 4);
1552        let whole = b.column_named("line").unwrap();
1553        assert_eq!(whole, &Column::strs(["l0", "l1", "x", "y"]));
1554    }
1555
1556    /// A row at a time, past a cap: the last `cap` rows, read by row and
1557    /// whole alike, and the runs few.
1558    #[test]
1559    fn a_capped_log_keeps_its_last_rows() {
1560        let mut ch = log(0);
1561        for i in 0..1000i64 {
1562            ch = ch
1563                .append(rows(&[&format!("n{i}")], &[i]), Some(300))
1564                .unwrap();
1565            if let Body::Table { chunks, .. } = &ch.0.body {
1566                assert!(chunks.len() <= 12, "{} runs at row {i}", chunks.len());
1567            }
1568        }
1569        assert_eq!(ch.len(), 300);
1570        assert_eq!(ch.get(0, 0), Cell::Str("n700"));
1571        assert_eq!(ch.get(299, 1), Cell::F64(999.0));
1572        let ms = ch.column_named("ms").unwrap();
1573        assert_eq!(ms.len(), 300);
1574        assert_eq!(ms.get(150), Cell::F64(850.0));
1575        for r in [0, 1, 77, 150, 298, 299] {
1576            assert_eq!(ch.get(r, 1), ms.get(r), "row {r}");
1577        }
1578    }
1579
1580    #[test]
1581    fn an_append_of_the_wrong_shape_is_refused() {
1582        let a = log(1);
1583        let e = |r: Result<Channel, ChannelError>| r.unwrap_err().0;
1584        assert!(
1585            e(a.append(vec![("ms".into(), Column::F64(vec![1.0]))], None)).contains("1 columns")
1586        );
1587        assert!(e(a.append(rows(&["x"], &[1, 2]), None)).contains("rows"));
1588        let bools = vec![
1589            ("ms".into(), Column::Bool(vec![true])),
1590            ("line".into(), Column::strs(["x"])),
1591        ];
1592        assert!(e(a.append(bools, None)).contains("booleans"));
1593        // Numbers onto a column of integers widen it, the rows so far read
1594        // as numbers in the new revision and as integers in the old one
1595        // (RG173); an append of integers after reads as numbers too.
1596        let ints = Channel::table(vec![("t".into(), Column::I64(vec![1, 2]))]).unwrap();
1597        let wide = ints
1598            .append(vec![("t".into(), Column::F64(vec![2.5]))], None)
1599            .unwrap();
1600        assert_eq!(
1601            (wide.get(0, 0), wide.get(2, 0)),
1602            (Cell::F64(1.0), Cell::F64(2.5))
1603        );
1604        assert_eq!(
1605            ints.get(0, 0),
1606            Cell::I64(1),
1607            "the old revision keeps its own"
1608        );
1609        let more = wide
1610            .append(vec![("t".into(), Column::I64(vec![4]))], None)
1611            .unwrap();
1612        assert_eq!(more.column(0), Some(&Column::F64(vec![1.0, 2.0, 2.5, 4.0])));
1613        // No rows have no kind: an empty list, numbers to a binding, is
1614        // taken by the string column.
1615        let none = vec![
1616            ("ms".into(), Column::F64(Vec::new())),
1617            ("line".into(), Column::F64(Vec::new())),
1618        ];
1619        let same = a.append(none, None).unwrap();
1620        assert_eq!((same.len(), same.get(0, 0)), (1, a.get(0, 0)));
1621        assert!(
1622            Channel::bytes(vec![1])
1623                .append(rows(&["x"], &[1]), None)
1624                .is_err()
1625        );
1626    }
1627
1628    /// A revision of several runs is read run by run: every filter, sort,
1629    /// take, reduction and lane reads what the same rows in one run read,
1630    /// and none of them joins a column whole (backlog RG172) — Live's
1631    /// filtered view took every column of a million spans at each poll.
1632    #[test]
1633    fn readers_go_run_by_run_and_join_nothing() {
1634        let mut ch = log(3);
1635        for i in 0..40i64 {
1636            let names: Vec<String> = (0..=i % 3).map(|k| format!("l{i}-{k}")).collect();
1637            let names: Vec<&str> = names.iter().map(String::as_str).collect();
1638            let ms: Vec<i64> = (0..names.len() as i64)
1639                .map(|k| (i * 7 + k * 13) % 50)
1640                .collect();
1641            ch = ch.append(rows(&names, &ms), Some(70)).unwrap();
1642        }
1643        let runs = match &ch.0.body {
1644            Body::Table { chunks, .. } => chunks.len(),
1645            Body::Bytes(_) => 0,
1646        };
1647        assert!(runs > 1, "{runs} runs");
1648        let one = Channel::table(
1649            ch.names()
1650                .iter()
1651                .enumerate()
1652                .map(|(c, n)| {
1653                    (n.clone(), {
1654                        let mut out = ch.0.chunks()[0].cols[c].empty_like();
1655                        for k in ch.0.chunks() {
1656                            out.extend_rows(&k.cols[c], k.skip..k.skip + k.rows());
1657                        }
1658                        out
1659                    })
1660                })
1661                .collect(),
1662        )
1663        .unwrap();
1664        let range = Match::Range {
1665            min: Some(10.0),
1666            max: Some(30.0),
1667        };
1668        let text = Match::Contains {
1669            text: "L1".into(),
1670            case: false,
1671        };
1672        let within: Vec<usize> = (0..ch.len()).rev().step_by(3).collect();
1673        for w in [None, Some(within.as_slice())] {
1674            assert_eq!(ch.filter(1, &range, w), one.filter(1, &range, w));
1675            assert_eq!(ch.filter(0, &text, w), one.filter(0, &text, w));
1676            assert_eq!(ch.sort(1, true, w), one.sort(1, true, w));
1677            assert_eq!(ch.sort(0, false, w), one.sort(0, false, w));
1678        }
1679        let picked = ch.sort(1, false, None).unwrap();
1680        assert_eq!(ch.take_columns(&picked), one.take_columns(&picked));
1681        assert_eq!(ch.reduce(1, 5, 66, 7), one.reduce(1, 5, 66, 7));
1682        let (a, b) = (
1683            ch.lanes(1, None, None).unwrap(),
1684            one.lanes(1, None, None).unwrap(),
1685        );
1686        assert_eq!(
1687            (a.count(), a.len(), a.extent()),
1688            (b.count(), b.len(), b.extent())
1689        );
1690        if let Body::Table { flat, .. } = &ch.0.body {
1691            assert!(
1692                flat.iter().all(|s| s.get().is_none()),
1693                "a column was joined whole"
1694            );
1695        }
1696    }
1697
1698    /// Filters by value, by a string's part and by a range, chained
1699    /// through `within`; a sort that keeps ties in order and NaN last; and
1700    /// a take of the rows they chose (backlog C148).
1701    #[test]
1702    fn a_filter_a_sort_and_a_take() {
1703        let ch = Channel::table(vec![
1704            (
1705                "name".into(),
1706                Column::strs(["api/get", "db", "API/put", "api/del", "x"]),
1707            ),
1708            ("ms".into(), Column::F64(vec![5.0, 1.0, f64::NAN, 5.0, 9.0])),
1709            ("n".into(), Column::I64(vec![1, 2, 3, 4, 5])),
1710        ])
1711        .unwrap();
1712        let api = Match::Contains {
1713            text: "api".into(),
1714            case: false,
1715        };
1716        assert_eq!(ch.filter(0, &api, None).unwrap(), [0, 2, 3]);
1717        let exact = Match::Contains {
1718            text: "api".into(),
1719            case: true,
1720        };
1721        assert_eq!(ch.filter(0, &exact, None).unwrap(), [0, 3]);
1722        let fast = Match::Range {
1723            min: None,
1724            max: Some(5.0),
1725        };
1726        let both = ch.filter(1, &fast, Some(&[0, 2, 3])).unwrap();
1727        assert_eq!(both, [0, 3], "NaN is in no range");
1728        assert_eq!(ch.filter(2, &Match::Eq(Value::Int(4)), None).unwrap(), [3]);
1729        assert_eq!(
1730            ch.filter(1, &Match::Eq(Value::Int(5)), None).unwrap(),
1731            [0, 3]
1732        );
1733        assert!(ch.filter(1, &api, None).is_err(), "a search on numbers");
1734        assert!(
1735            ch.filter(0, &fast, Some(&[9])).is_err(),
1736            "a row past the end"
1737        );
1738
1739        assert_eq!(ch.sort(1, false, None).unwrap(), [1, 0, 3, 4, 2]);
1740        assert_eq!(ch.sort(1, true, None).unwrap(), [4, 0, 3, 1, 2]);
1741        assert_eq!(ch.sort(0, false, Some(&[3, 0])).unwrap(), [3, 0]);
1742
1743        let view = ch.take(&[3, 0]).unwrap();
1744        assert_ne!(view.id(), ch.id());
1745        assert_eq!(view.len(), 2);
1746        assert_eq!(view.get(0, 0), Cell::Str("api/del"));
1747        assert_eq!(view.get(1, 2), Cell::I64(1));
1748        assert!(ch.take(&[5]).is_err());
1749
1750        // A view of the whole, its rows shared and its identity its own.
1751        let shown = view.publish_from(&ch);
1752        assert_eq!((shown.id(), shown.rev(), shown.len()), (view.id(), 2, 5));
1753        assert_eq!(shown.get(2, 0), Cell::Str("API/put"));
1754    }
1755}