1use 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
48pub const HANDLE_KEY: &str = "$channel";
50pub const REV_KEY: &str = "rev";
52
53#[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 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 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 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 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 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#[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#[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 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 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#[derive(Clone, Debug, PartialEq)]
204pub enum Data {
205 Table {
207 names: Vec<String>,
208 cols: Vec<Column>,
209 },
210 Bytes(Vec<u8>),
212}
213
214#[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#[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 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 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 fn chunks(&self) -> &[Chunk] {
288 match &self.body {
289 Body::Table { chunks, .. } => chunks,
290 Body::Bytes(_) => &[],
291 }
292 }
293
294 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 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
325fn whole_slots(n: usize) -> Vec<OnceLock<Column>> {
327 (0..n).map(|_| OnceLock::new()).collect()
328}
329
330#[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 fn eq(&self, other: &Self) -> bool {
348 self.0.id == other.0.id && self.0.rev == other.0.rev
349 }
350}
351
352static 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
388const MAX_CHUNKS: usize = 48;
391
392fn 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 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 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 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 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 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 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 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 (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 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 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 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 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 pub fn names(&self) -> &[String] {
686 match &self.0.body {
687 Body::Table { names, .. } => names,
688 Body::Bytes(_) => &[],
689 }
690 }
691
692 pub fn col(&self, name: &str) -> Option<usize> {
694 self.names().iter().position(|n| n == name)
695 }
696
697 pub fn column(&self, i: usize) -> Option<&Column> {
702 self.0.whole(i)
703 }
704
705 pub fn column_named(&self, name: &str) -> Option<&Column> {
707 self.column(self.col(name)?)
708 }
709
710 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 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 pub fn handle(&self) -> Value {
738 handle_value(self.0.id, self.0.rev)
739 }
740
741 pub fn resolve(v: &Value) -> Option<Channel> {
744 let (id, _) = handle_of(v)?;
745 registry().get(&id)?.upgrade().map(Channel)
746 }
747}
748
749#[derive(Clone, Debug, PartialEq)]
751pub enum Match {
752 Eq(Value),
756 Contains { text: String, case: bool },
759 Range { min: Option<f64>, max: Option<f64> },
762}
763
764fn 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 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 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 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 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 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 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 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 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 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 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 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
998type 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#[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 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 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 Column::F64(_) | Column::I64(_) if k == 0 => {}
1106 Column::F64(_) | Column::I64(_) => {
1107 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
1139fn 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#[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 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
1240pub 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
1248pub 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
1265pub fn is_handle(v: &Value) -> bool {
1267 handle_of(v).is_some()
1268}
1269
1270pub 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 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 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 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 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 let r = ch.reduce(0, 0, 5, 2).unwrap();
1391 assert_eq!((r[0].last_row, r[1].first_row), (1, 2));
1392 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 #[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 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 assert_eq!(Channel::resolve(&h).unwrap().len(), 2);
1505 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 #[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 #[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 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 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 #[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 #[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 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}