1use std::fmt;
2use std::sync::Arc;
3
4use laddu_physics::vectors::RealVec4;
5
6use crate::{
7 BatchLayout, LadduDataError, LadduDataResult,
8 schema::{P4Binding, Precision, ScalarBinding, Schema},
9};
10
11#[derive(Clone, Debug)]
12struct BatchParts {
13 p4s: Arc<[Arc<[RealVec4]>]>,
14 scalars: Arc<[Arc<[f64]>]>,
15 weights: Weights,
16}
17
18#[derive(Clone, Debug)]
19enum Weights {
20 ImplicitUnit,
21 Explicit(Arc<[f64]>),
22}
23
24impl Weights {
25 fn from_option(weights: Option<Arc<[f64]>>) -> Self {
26 match weights {
27 Some(weights) => Self::Explicit(weights),
28 None => Self::ImplicitUnit,
29 }
30 }
31
32 fn as_slice(&self) -> Option<&[f64]> {
33 match self {
34 Self::ImplicitUnit => None,
35 Self::Explicit(weights) => Some(weights),
36 }
37 }
38
39 fn at(&self, row: usize) -> f64 {
40 self.as_slice().map_or(1.0, |weights| weights[row])
41 }
42
43 fn is_explicit(&self) -> bool {
44 matches!(self, Self::Explicit(_))
45 }
46
47 fn select(&self, rows: &[usize]) -> Self {
48 match self {
49 Self::ImplicitUnit => Self::ImplicitUnit,
50 Self::Explicit(weights) => {
51 let selected: Arc<[f64]> = rows.iter().map(|&row| weights[row]).collect();
52 Self::Explicit(selected)
53 }
54 }
55 }
56
57 fn slice(&self, start: usize, end: usize) -> Self {
58 match self {
59 Self::ImplicitUnit => Self::ImplicitUnit,
60 Self::Explicit(weights) => Self::Explicit(Arc::from(&weights[start..end])),
61 }
62 }
63
64 fn reweight<F>(&self, len: usize, f: F) -> Self
65 where
66 F: Fn(usize, f64) -> f64,
67 {
68 let weights: Arc<[f64]> = (0..len).map(|i| f(i, self.at(i))).collect();
69 Self::Explicit(weights)
70 }
71}
72
73impl BatchParts {
74 fn from_columns(p4s: Vec<Arc<[RealVec4]>>, scalars: Vec<Arc<[f64]>>, weights: Weights) -> Self {
75 Self {
76 p4s: p4s.into(),
77 scalars: scalars.into(),
78 weights,
79 }
80 }
81
82 fn validate(&self, schema: &Schema, expected_len: Option<usize>) -> LadduDataResult<usize> {
83 if self.p4s.len() != schema.n_p4s() {
84 return Err(LadduDataError::Schema(
85 "wrong number of vec4 columns".into(),
86 ));
87 }
88
89 if self.scalars.len() != schema.n_scalars() {
90 return Err(LadduDataError::Schema(
91 "wrong number of scalar columns".into(),
92 ));
93 }
94
95 let len = infer_len(&self.p4s, &self.scalars, self.weights.as_slice())?;
96 if let Some(expected_len) = expected_len {
97 let has_columns =
98 !self.p4s.is_empty() || !self.scalars.is_empty() || self.weights.is_explicit();
99 if has_columns && len != expected_len {
100 return Err(LadduDataError::Schema("inconsistent batch length".into()));
101 }
102 return Ok(expected_len);
103 }
104 Ok(len)
105 }
106
107 fn select(&self, rows: &[usize]) -> Self {
108 let p4s = self
109 .p4s
110 .iter()
111 .map(|col| rows.iter().map(|&i| col[i]).collect())
112 .collect();
113 let scalars = self
114 .scalars
115 .iter()
116 .map(|col| rows.iter().map(|&i| col[i]).collect())
117 .collect();
118
119 Self {
120 p4s,
121 scalars,
122 weights: self.weights.select(rows),
123 }
124 }
125
126 fn slice(&self, start: usize, end: usize) -> Self {
127 let p4s = self
128 .p4s
129 .iter()
130 .map(|col| Arc::<[RealVec4]>::from(&col[start..end]))
131 .collect();
132 let scalars = self
133 .scalars
134 .iter()
135 .map(|col| Arc::<[f64]>::from(&col[start..end]))
136 .collect();
137
138 Self {
139 p4s,
140 scalars,
141 weights: self.weights.slice(start, end),
142 }
143 }
144
145 fn reweight<F>(&self, len: usize, f: F) -> Self
146 where
147 F: Fn(usize, f64) -> f64,
148 {
149 Self {
150 p4s: Arc::clone(&self.p4s),
151 scalars: Arc::clone(&self.scalars),
152 weights: self.weights.reweight(len, f),
153 }
154 }
155
156 fn concat(batches: &[(&Self, usize)]) -> Self {
157 let len: usize = batches.iter().map(|(_, len)| *len).sum();
158 let n_p4s = batches.first().map_or(0, |(batch, _)| batch.p4s.len());
159 let n_scalars = batches.first().map_or(0, |(batch, _)| batch.scalars.len());
160
161 let mut p4s = Vec::with_capacity(n_p4s);
162 for col in 0..n_p4s {
163 let mut out = Vec::with_capacity(len);
164 for (batch, _) in batches {
165 out.extend_from_slice(&batch.p4s[col]);
166 }
167 p4s.push(Arc::from(out));
168 }
169
170 let mut scalars = Vec::with_capacity(n_scalars);
171 for col in 0..n_scalars {
172 let mut out = Vec::with_capacity(len);
173 for (batch, _) in batches {
174 out.extend_from_slice(&batch.scalars[col]);
175 }
176 scalars.push(Arc::from(out));
177 }
178
179 let weights = if batches.iter().any(|(batch, _)| batch.weights.is_explicit()) {
180 let mut out = Vec::with_capacity(len);
181 for (batch, batch_len) in batches {
182 for row in 0..*batch_len {
183 out.push(batch.weights.at(row));
184 }
185 }
186 Weights::Explicit(Arc::from(out))
187 } else {
188 Weights::ImplicitUnit
189 };
190
191 Self::from_columns(p4s, scalars, weights)
192 }
193}
194
195#[derive(Default)]
196enum WeightAssembler {
197 #[default]
198 ImplicitUnit,
199 Explicit(Vec<f64>),
200}
201
202impl WeightAssembler {
203 fn push(&mut self, weight: Option<f64>, len: usize) -> LadduDataResult<()> {
204 match self {
205 Self::Explicit(weights) => match weight {
206 Some(weight) => weights.push(weight),
207 None => {
208 return Err(LadduDataError::InvalidArgument(
209 "cannot mix weighted and unweighted events in one batch",
210 ));
211 }
212 },
213 Self::ImplicitUnit => match weight {
214 Some(weight) if len == 0 => *self = Self::Explicit(vec![weight]),
215 Some(_) => {
216 return Err(LadduDataError::InvalidArgument(
217 "cannot mix unweighted and weighted events in one batch",
218 ));
219 }
220 None => {}
221 },
222 }
223
224 Ok(())
225 }
226
227 fn finish(self) -> Weights {
228 match self {
229 Self::ImplicitUnit => Weights::ImplicitUnit,
230 Self::Explicit(weights) => Weights::Explicit(Arc::from(weights)),
231 }
232 }
233}
234
235#[derive(Clone)]
237pub struct EventBatch {
238 schema: Arc<Schema>,
239 len: usize,
240 parts: BatchParts,
241}
242
243impl fmt::Debug for EventBatch {
244 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
245 formatter
246 .debug_struct("EventBatch")
247 .field("schema", &self.schema)
248 .field("len", &self.len)
249 .field("p4s", &self.parts.p4s)
250 .field("scalars", &self.parts.scalars)
251 .field("weights", &self.parts.weights.as_slice())
252 .finish()
253 }
254}
255
256impl EventBatch {
257 pub fn new(
264 schema: Arc<Schema>,
265 p4s: Vec<Arc<[RealVec4]>>,
266 scalars: Vec<Arc<[f64]>>,
267 weights: Option<Arc<[f64]>>,
268 ) -> LadduDataResult<Self> {
269 BatchAssembler::from_columns(schema, p4s, scalars, weights)
270 }
271
272 fn from_parts(schema: Arc<Schema>, parts: BatchParts) -> LadduDataResult<Self> {
273 let len = parts.validate(&schema, None)?;
274 Ok(Self { schema, len, parts })
275 }
276
277 fn from_parts_with_len(
278 schema: Arc<Schema>,
279 parts: BatchParts,
280 expected_len: usize,
281 ) -> LadduDataResult<Self> {
282 let len = parts.validate(&schema, Some(expected_len))?;
283 Ok(Self { schema, len, parts })
284 }
285
286 pub fn from_events<I>(schema: Arc<Schema>, events: I) -> LadduDataResult<Self>
293 where
294 I: IntoIterator<Item = OwnedEvent>,
295 {
296 let mut builder = EventBatchBuilder::new(schema);
297 builder.extend(events)?;
298 builder.finish()
299 }
300
301 pub fn schema(&self) -> &Arc<Schema> {
303 &self.schema
304 }
305
306 pub fn len(&self) -> usize {
308 self.len
309 }
310
311 pub fn bytes_per_event(&self) -> usize {
313 BatchLayout::from_batch(self)
314 .bytes_per_event(Precision::F64)
315 .ok()
316 .and_then(|bytes| usize::try_from(bytes).ok())
317 .unwrap_or(usize::MAX)
318 }
319
320 pub fn resident_bytes(&self) -> usize {
325 BatchLayout::from_batch(self)
326 .footprint(Precision::F64)
327 .and_then(|footprint| footprint.checked_peak_bytes(self.len))
328 .ok()
329 .and_then(|bytes| usize::try_from(bytes).ok())
330 .unwrap_or(usize::MAX)
331 }
332
333 pub fn is_empty(&self) -> bool {
335 self.len == 0
336 }
337
338 pub fn vec4_column(&self, index: usize) -> &[RealVec4] {
340 &self.parts.p4s[index]
341 }
342
343 pub fn scalar_column(&self, index: usize) -> &[f64] {
345 &self.parts.scalars[index]
346 }
347
348 pub fn weights_column(&self) -> Option<&[f64]> {
350 self.parts.weights.as_slice()
351 }
352
353 pub fn vec4_column_named(&self, name: &str) -> Option<&[RealVec4]> {
355 let i = self.schema.p4_index(name)?;
356 Some(self.vec4_column(i))
357 }
358
359 pub fn scalar_column_named(&self, name: &str) -> Option<&[f64]> {
361 let i = self.schema.scalar_index(name)?;
362 Some(self.scalar_column(i))
363 }
364
365 pub fn p4_at(&self, col: usize, row: usize) -> RealVec4 {
367 self.parts.p4s[col][row]
368 }
369
370 pub fn p4_column_bound(&self, binding: &P4Binding) -> LadduDataResult<&[RealVec4]> {
380 if !binding.matches(&self.schema) {
381 return Err(LadduDataError::Schema(
382 "column binding belongs to a different schema".into(),
383 ));
384 }
385 Ok(self.vec4_column(binding.index()))
386 }
387
388 pub fn scalar_at(&self, col: usize, row: usize) -> f64 {
390 self.parts.scalars[col][row]
391 }
392
393 pub fn scalar_column_bound(&self, binding: &ScalarBinding) -> LadduDataResult<&[f64]> {
403 if !binding.matches(&self.schema) {
404 return Err(LadduDataError::Schema(
405 "column binding belongs to a different schema".into(),
406 ));
407 }
408 Ok(self.scalar_column(binding.index()))
409 }
410
411 pub fn weights_at(&self, row: usize) -> f64 {
413 self.parts.weights.at(row)
414 }
415
416 pub fn event(&self, row: usize) -> BatchEvent<'_> {
418 BatchEvent { batch: self, row }
419 }
420
421 pub fn iter(&self) -> impl Iterator<Item = BatchEvent<'_>> {
423 (0..self.len()).map(|i| self.event(i))
424 }
425
426 pub fn select(&self, rows: &[usize]) -> Self {
432 Self::from_parts_with_len(
433 Arc::clone(&self.schema),
434 self.parts.select(rows),
435 rows.len(),
436 )
437 .expect("select preserves EventBatch invariants")
438 }
439
440 pub fn filter<F>(&self, keep: F) -> Self
442 where
443 F: Fn(BatchEvent<'_>) -> bool,
444 {
445 let rows: Vec<usize> = (0..self.len).filter(|&i| keep(self.event(i))).collect();
446
447 self.select(&rows)
448 }
449
450 pub fn reweight<F>(&self, f: F) -> Self
457 where
458 F: Fn(usize, f64) -> f64,
459 {
460 Self::from_parts(Arc::clone(&self.schema), self.parts.reweight(self.len, f))
461 .expect("reweight preserves EventBatch invariants")
462 }
463
464 pub fn slice(&self, start: usize, end: usize) -> Self {
470 assert!(start <= end);
471 assert!(end <= self.len);
472
473 if start == 0 && end == self.len {
474 return self.clone();
475 }
476
477 Self::from_parts_with_len(
478 Arc::clone(&self.schema),
479 self.parts.slice(start, end),
480 end - start,
481 )
482 .expect("slice preserves EventBatch invariants")
483 }
484
485 pub fn concat(batches: &[Self]) -> LadduDataResult<Self> {
492 if batches.is_empty() {
493 return Err(LadduDataError::InvalidArgument(
494 "cannot concatenate zero batches",
495 ));
496 }
497
498 let schema = Arc::clone(&batches[0].schema);
499
500 for batch in batches {
501 if schema != batch.schema {
502 return Err(LadduDataError::Schema(
503 "cannot concatenate batches with different schemas".into(),
504 ));
505 }
506 }
507
508 let parts = batches
509 .iter()
510 .map(|batch| (&batch.parts, batch.len))
511 .collect::<Vec<_>>();
512 let len = batches.iter().map(|batch| batch.len).sum();
513 Self::from_parts_with_len(schema, BatchParts::concat(&parts), len)
514 }
515}
516
517fn infer_len(
518 vec4s: &[Arc<[RealVec4]>],
519 scalars: &[Arc<[f64]>],
520 weight: Option<&[f64]>,
521) -> LadduDataResult<usize> {
522 let len = vec4s
523 .first()
524 .map(|c| c.len())
525 .or_else(|| scalars.first().map(|c| c.len()))
526 .or_else(|| weight.map(|w| w.len()))
527 .unwrap_or(0);
528
529 for col in vec4s {
530 if col.len() != len {
531 return Err(LadduDataError::Schema(
532 "inconsistent vec4 column length".into(),
533 ));
534 }
535 }
536
537 for col in scalars {
538 if col.len() != len {
539 return Err(LadduDataError::Schema(
540 "inconsistent scalar column length".into(),
541 ));
542 }
543 }
544
545 if let Some(w) = weight
546 && w.len() != len
547 {
548 return Err(LadduDataError::Schema("inconsistent weight length".into()));
549 }
550
551 Ok(len)
552}
553
554#[derive(Copy, Clone, Debug)]
556pub struct BatchEvent<'a> {
557 batch: &'a EventBatch,
558 row: usize,
559}
560
561impl<'a> BatchEvent<'a> {
562 pub fn row(&self) -> usize {
564 self.row
565 }
566
567 pub fn batch(&self) -> &'a EventBatch {
569 self.batch
570 }
571
572 pub fn p4(&self, col: usize) -> RealVec4 {
574 self.batch.p4_at(col, self.row)
575 }
576
577 pub fn scalar(&self, col: usize) -> f64 {
579 self.batch.scalar_at(col, self.row)
580 }
581
582 pub fn weight(&self) -> f64 {
584 self.batch.weights_at(self.row)
585 }
586
587 pub fn p4_named(&self, name: &str) -> Option<RealVec4> {
589 let col = self.batch.schema.p4_index(name)?;
590 Some(self.p4(col))
591 }
592
593 pub fn scalar_named(&self, name: &str) -> Option<f64> {
595 let col = self.batch.schema.scalar_index(name)?;
596 Some(self.scalar(col))
597 }
598}
599
600#[derive(Copy, Clone, Debug)]
602pub struct Event<'a> {
603 pub(super) batch: &'a EventBatch,
604 pub(super) row: usize,
605 pub(super) weight: f64,
606}
607
608impl<'a> Event<'a> {
609 pub fn row(&self) -> usize {
611 self.row
612 }
613
614 pub fn p4(&self, col: usize) -> RealVec4 {
616 self.batch.p4_at(col, self.row)
617 }
618
619 pub fn scalar(&self, col: usize) -> f64 {
621 self.batch.scalar_at(col, self.row)
622 }
623
624 pub fn weight(&self) -> f64 {
626 self.weight
627 }
628
629 pub fn p4_named(&self, name: &str) -> Option<RealVec4> {
631 let col = self.batch.schema.p4_index(name)?;
632 Some(self.p4(col))
633 }
634
635 pub fn scalar_named(&self, name: &str) -> Option<f64> {
637 let col = self.batch.schema.scalar_index(name)?;
638 Some(self.scalar(col))
639 }
640}
641
642#[derive(Clone, Debug)]
644pub struct OwnedEvent {
645 pub p4s: Vec<RealVec4>,
647 pub scalars: Vec<f64>,
649 pub weight: Option<f64>,
651}
652
653impl OwnedEvent {
654 pub fn new(p4s: Vec<RealVec4>, scalars: Vec<f64>) -> Self {
656 Self {
657 p4s,
658 scalars,
659 weight: None,
660 }
661 }
662
663 pub fn weighted(p4s: Vec<RealVec4>, scalars: Vec<f64>, weight: f64) -> Self {
665 Self {
666 p4s,
667 scalars,
668 weight: Some(weight),
669 }
670 }
671}
672
673pub(crate) struct BatchAssembler {
675 schema: Arc<Schema>,
676 p4s: Vec<Vec<RealVec4>>,
677 scalars: Vec<Vec<f64>>,
678 weights: WeightAssembler,
679 len: usize,
680}
681
682impl BatchAssembler {
683 pub(crate) fn new(schema: Arc<Schema>, capacity: usize) -> Self {
684 let p4s = (0..schema.n_p4s())
685 .map(|_| Vec::with_capacity(capacity))
686 .collect();
687 let scalars = (0..schema.n_scalars())
688 .map(|_| Vec::with_capacity(capacity))
689 .collect();
690
691 Self {
692 schema,
693 p4s,
694 scalars,
695 weights: WeightAssembler::default(),
696 len: 0,
697 }
698 }
699
700 pub(crate) fn with_weight_mode(
701 schema: Arc<Schema>,
702 capacity: usize,
703 explicit_weights: bool,
704 ) -> Self {
705 let mut assembler = Self::new(schema, capacity);
706 if explicit_weights {
707 assembler.weights = WeightAssembler::Explicit(Vec::with_capacity(capacity));
708 }
709 assembler
710 }
711
712 pub(crate) fn from_columns(
713 schema: Arc<Schema>,
714 p4s: Vec<Arc<[RealVec4]>>,
715 scalars: Vec<Arc<[f64]>>,
716 weights: Option<Arc<[f64]>>,
717 ) -> LadduDataResult<EventBatch> {
718 EventBatch::from_parts(
719 schema,
720 BatchParts::from_columns(p4s, scalars, Weights::from_option(weights)),
721 )
722 }
723
724 fn push_owned(&mut self, event: OwnedEvent) -> LadduDataResult<()> {
725 if event.p4s.len() != self.schema.n_p4s() {
726 return Err(LadduDataError::Schema(
727 "wrong number of event vec4 values".into(),
728 ));
729 }
730
731 if event.scalars.len() != self.schema.n_scalars() {
732 return Err(LadduDataError::Schema(
733 "wrong number of event scalar values".into(),
734 ));
735 }
736
737 self.weights.push(event.weight, self.len)?;
738
739 for (col, value) in event.p4s.into_iter().enumerate() {
740 self.p4s[col].push(value);
741 }
742 for (col, value) in event.scalars.into_iter().enumerate() {
743 self.scalars[col].push(value);
744 }
745
746 self.len += 1;
747 Ok(())
748 }
749
750 pub(crate) fn push_borrowed(
751 &mut self,
752 event: Event<'_>,
753 explicit_weight: bool,
754 ) -> LadduDataResult<()> {
755 if event.batch.schema().n_p4s() != self.schema.n_p4s() {
756 return Err(LadduDataError::Schema(
757 "wrong number of event vec4 values".into(),
758 ));
759 }
760
761 if event.batch.schema().n_scalars() != self.schema.n_scalars() {
762 return Err(LadduDataError::Schema(
763 "wrong number of event scalar values".into(),
764 ));
765 }
766
767 self.weights
768 .push(explicit_weight.then_some(event.weight()), self.len)?;
769
770 for col in 0..self.schema.n_p4s() {
771 self.p4s[col].push(event.p4(col));
772 }
773 for col in 0..self.schema.n_scalars() {
774 self.scalars[col].push(event.scalar(col));
775 }
776
777 self.len += 1;
778 Ok(())
779 }
780
781 pub(crate) fn finish(self) -> LadduDataResult<EventBatch> {
782 let parts = BatchParts::from_columns(
783 self.p4s.into_iter().map(Arc::from).collect(),
784 self.scalars.into_iter().map(Arc::from).collect(),
785 self.weights.finish(),
786 );
787 EventBatch::from_parts_with_len(self.schema, parts, self.len)
788 }
789}
790
791pub struct EventBatchBuilder {
793 assembler: BatchAssembler,
794}
795
796impl EventBatchBuilder {
797 pub fn new(schema: Arc<Schema>) -> Self {
799 Self::with_capacity(schema, 0)
800 }
801
802 pub fn with_capacity(schema: Arc<Schema>, capacity: usize) -> Self {
804 Self {
805 assembler: BatchAssembler::new(schema, capacity),
806 }
807 }
808
809 pub fn push<P, S>(&mut self, p4s: P, scalars: S) -> LadduDataResult<&mut Self>
816 where
817 P: IntoIterator<Item = RealVec4>,
818 S: IntoIterator<Item = f64>,
819 {
820 self.push_event(OwnedEvent::new(
821 p4s.into_iter().collect(),
822 scalars.into_iter().collect(),
823 ))
824 }
825
826 pub fn push_weighted<P, S>(
833 &mut self,
834 p4s: P,
835 scalars: S,
836 weight: f64,
837 ) -> LadduDataResult<&mut Self>
838 where
839 P: IntoIterator<Item = RealVec4>,
840 S: IntoIterator<Item = f64>,
841 {
842 self.push_event(OwnedEvent::weighted(
843 p4s.into_iter().collect(),
844 scalars.into_iter().collect(),
845 weight,
846 ))
847 }
848
849 pub fn push_event(&mut self, event: OwnedEvent) -> LadduDataResult<&mut Self> {
856 self.assembler.push_owned(event)?;
857 Ok(self)
858 }
859
860 pub fn extend<I>(&mut self, events: I) -> LadduDataResult<&mut Self>
867 where
868 I: IntoIterator<Item = OwnedEvent>,
869 {
870 for event in events {
871 self.push_event(event)?;
872 }
873
874 Ok(self)
875 }
876
877 pub fn finish(self) -> LadduDataResult<EventBatch> {
884 self.assembler.finish()
885 }
886}
887
888#[cfg(test)]
889mod tests {
890 use super::*;
891
892 fn v(x: f64) -> RealVec4 {
893 RealVec4 {
894 e: x + 0.3,
895 px: x,
896 py: x + 0.1,
897 pz: x + 0.2,
898 }
899 }
900
901 fn schema_with_weight() -> Arc<Schema> {
902 Arc::new(Schema::new(["p"], ["x"], true).unwrap())
903 }
904
905 fn weighted_batch(start: usize, len: usize) -> EventBatch {
906 let schema = schema_with_weight();
907
908 let events = (start..start + len)
909 .map(|i| OwnedEvent::weighted(vec![v(i as f64)], vec![i as f64], 10.0 + i as f64));
910
911 EventBatch::from_events(schema, events).unwrap()
912 }
913
914 fn scalar_values(batch: &EventBatch) -> Vec<f64> {
915 batch.scalar_column(0).to_vec()
916 }
917
918 #[test]
919 fn bound_columns_reuse_schema_resolution_for_row_access() {
920 let batch = weighted_batch(3, 2);
921 let scalar = batch.schema().bind_scalar("x").unwrap();
922 let p4 = batch.schema().bind_p4("p").unwrap();
923
924 assert_eq!(batch.scalar_column_bound(&scalar).unwrap(), &[3.0, 4.0]);
925 assert_eq!(batch.p4_column_bound(&p4).unwrap(), &[v(3.0), v(4.0)]);
926
927 let other_schema = Arc::new(Schema::new(["other"], ["x"], true).unwrap());
928 let other = other_schema.bind_scalar("x").unwrap();
929 assert!(matches!(
930 batch.scalar_column_bound(&other),
931 Err(LadduDataError::Schema(message)) if message == "column binding belongs to a different schema"
932 ));
933 }
934
935 #[test]
936 fn event_batch_rejects_shape_mismatches_and_builder_rejects_mixed_weights() {
937 let schema = schema_with_weight();
938
939 let bad_vec4_count = EventBatch::new(
940 Arc::clone(&schema),
941 vec![],
942 vec![Arc::from([1.0, 2.0])],
943 Some(Arc::from([1.0, 2.0])),
944 );
945
946 assert!(matches!(bad_vec4_count, Err(LadduDataError::Schema(_))));
947
948 let bad_lengths = EventBatch::new(
949 Arc::clone(&schema),
950 vec![Arc::from([v(1.0), v(2.0)])],
951 vec![Arc::from([1.0])],
952 Some(Arc::from([1.0, 2.0])),
953 );
954
955 assert!(matches!(bad_lengths, Err(LadduDataError::Schema(_))));
956
957 let mut builder = EventBatchBuilder::new(schema);
958 builder.push([v(1.0)], [1.0]).unwrap();
959
960 let mixed = builder.push_weighted([v(2.0)], [2.0], 2.0);
961 assert!(matches!(mixed, Err(LadduDataError::InvalidArgument(_))));
962 }
963
964 #[test]
965 fn select_slice_filter_reweight_and_concat_preserve_columns_and_weight_semantics() {
966 let weighted = weighted_batch(0, 4);
967 let selected = weighted.select(&[3, 1]);
968
969 assert_eq!(scalar_values(&selected), vec![3.0, 1.0]);
970 assert_eq!(selected.weights_column().unwrap(), &[13.0, 11.0]);
971 assert_eq!(selected.p4_at(0, 0).px, 3.0);
972 assert_eq!(selected.p4_at(0, 1).e, 1.3);
973
974 let sliced = weighted.slice(1, 3);
975 assert_eq!(scalar_values(&sliced), vec![1.0, 2.0]);
976 assert_eq!(sliced.weights_column().unwrap(), &[11.0, 12.0]);
977
978 let filtered = weighted.filter(|ev| ev.scalar(0) >= 2.0);
979 assert_eq!(scalar_values(&filtered), vec![2.0, 3.0]);
980
981 let reweighted = filtered.reweight(|i, w| w + 100.0 + i as f64);
982 assert_eq!(reweighted.weights_column().unwrap(), &[112.0, 114.0]);
983
984 let schema = schema_with_weight();
985
986 let unweighted_with_weight_schema = EventBatch::from_events(
987 Arc::clone(&schema),
988 [
989 OwnedEvent::new(vec![v(100.0)], vec![100.0]),
990 OwnedEvent::new(vec![v(101.0)], vec![101.0]),
991 ],
992 )
993 .unwrap();
994
995 let weighted_tail = EventBatch::from_events(
996 schema,
997 [
998 OwnedEvent::weighted(vec![v(200.0)], vec![200.0], 5.0),
999 OwnedEvent::weighted(vec![v(201.0)], vec![201.0], 6.0),
1000 ],
1001 )
1002 .unwrap();
1003
1004 let concatenated =
1005 EventBatch::concat(&[unweighted_with_weight_schema, weighted_tail]).unwrap();
1006
1007 assert_eq!(
1008 scalar_values(&concatenated),
1009 vec![100.0, 101.0, 200.0, 201.0]
1010 );
1011 assert_eq!(
1012 concatenated.weights_column().unwrap(),
1013 &[1.0, 1.0, 5.0, 6.0]
1014 );
1015 }
1016
1017 #[test]
1018 fn implicit_unit_weights_survive_assembly_and_row_transforms() {
1019 let schema = Arc::new(Schema::new(["p"], ["x"], false).unwrap());
1020 let batch = EventBatch::from_events(
1021 Arc::clone(&schema),
1022 (0..3).map(|i| OwnedEvent::new(vec![v(i as f64)], vec![i as f64])),
1023 )
1024 .unwrap();
1025
1026 assert!(batch.weights_column().is_none());
1027 assert_eq!(batch.weights_at(2), 1.0);
1028
1029 let selected = batch.select(&[2, 0]);
1030 let sliced = batch.slice(1, 3);
1031 let filtered = batch.filter(|event| event.scalar(0) > 0.0);
1032 let concatenated = EventBatch::concat(&[selected, sliced]).unwrap();
1033
1034 assert!(filtered.weights_column().is_none());
1035 assert!(concatenated.weights_column().is_none());
1036 assert_eq!(concatenated.weights_at(3), 1.0);
1037
1038 let reweighted = batch.reweight(|row, weight| weight + row as f64);
1039 assert_eq!(reweighted.weights_column().unwrap(), &[1.0, 2.0, 3.0]);
1040 }
1041
1042 #[test]
1043 fn shared_assembler_preserves_weight_mode_and_rejects_transitions() {
1044 let schema = Arc::new(Schema::new(["p"], ["x"], true).unwrap());
1045 let source = EventBatch::from_events(
1046 Arc::clone(&schema),
1047 [
1048 OwnedEvent::weighted(vec![v(1.0)], vec![1.0], 2.0),
1049 OwnedEvent::weighted(vec![v(2.0)], vec![2.0], 3.0),
1050 ],
1051 )
1052 .unwrap();
1053
1054 let mut explicit = BatchAssembler::new(Arc::clone(&schema), 2);
1055 let first = Event {
1056 batch: &source,
1057 row: 0,
1058 weight: source.weights_at(0),
1059 };
1060 explicit.push_borrowed(first, true).unwrap();
1061 let second = Event {
1062 batch: &source,
1063 row: 1,
1064 weight: source.weights_at(1),
1065 };
1066 let transition = explicit.push_borrowed(second, false);
1067 assert!(matches!(
1068 transition,
1069 Err(LadduDataError::InvalidArgument(_))
1070 ));
1071 let explicit = explicit.finish().unwrap();
1072 assert_eq!(explicit.len(), 1);
1073 assert_eq!(explicit.weights_column().unwrap(), &[2.0]);
1074
1075 let unweighted = EventBatch::from_events(
1076 Arc::clone(&schema),
1077 [OwnedEvent::new(vec![v(3.0)], vec![3.0])],
1078 )
1079 .unwrap();
1080 let mut implicit = BatchAssembler::new(schema, 1);
1081 let event = Event {
1082 batch: &unweighted,
1083 row: 0,
1084 weight: unweighted.weights_at(0),
1085 };
1086 implicit.push_borrowed(event, false).unwrap();
1087 let implicit = implicit.finish().unwrap();
1088 assert!(implicit.weights_column().is_none());
1089 assert_eq!(implicit.weights_at(0), 1.0);
1090 }
1091
1092 #[test]
1093 fn assembly_table_covers_column_shapes_and_observable_sharing() {
1094 let cases = [
1095 (
1096 Arc::new(Schema::new(Vec::<&str>::new(), Vec::<&str>::new(), false).unwrap()),
1097 false,
1098 ),
1099 (
1100 Arc::new(Schema::new(["p"], Vec::<&str>::new(), false).unwrap()),
1101 false,
1102 ),
1103 (
1104 Arc::new(Schema::new(Vec::<&str>::new(), ["x"], false).unwrap()),
1105 false,
1106 ),
1107 (
1108 Arc::new(Schema::new(Vec::<&str>::new(), Vec::<&str>::new(), true).unwrap()),
1109 true,
1110 ),
1111 (Arc::new(Schema::new(["p"], ["x"], true).unwrap()), true),
1112 ];
1113
1114 for (schema, weighted) in cases {
1115 let events = (0..2).map(|i| {
1116 let p4s = if schema.n_p4s() == 0 {
1117 Vec::new()
1118 } else {
1119 vec![v(i as f64)]
1120 };
1121 let scalars = if schema.n_scalars() == 0 {
1122 Vec::new()
1123 } else {
1124 vec![i as f64]
1125 };
1126 if weighted {
1127 OwnedEvent::weighted(p4s, scalars, 2.0 + i as f64)
1128 } else {
1129 OwnedEvent::new(p4s, scalars)
1130 }
1131 });
1132 let batch = EventBatch::from_events(Arc::clone(&schema), events).unwrap();
1133 assert_eq!(batch.len(), 2);
1134 assert_eq!(batch.weights_column().is_some(), weighted);
1135
1136 let selected = batch.select(&(0..batch.len()).collect::<Vec<_>>());
1137 assert_eq!(selected.len(), batch.len());
1138 for col in 0..schema.n_p4s() {
1139 assert_eq!(selected.vec4_column(col), batch.vec4_column(col));
1140 }
1141 for col in 0..schema.n_scalars() {
1142 assert_eq!(selected.scalar_column(col), batch.scalar_column(col));
1143 }
1144 assert_eq!(selected.weights_column(), batch.weights_column());
1145
1146 let concatenated = EventBatch::concat(&[batch.slice(0, 1), batch.slice(1, 2)]).unwrap();
1147 assert_eq!(concatenated.len(), batch.len());
1148 for col in 0..schema.n_p4s() {
1149 assert_eq!(concatenated.vec4_column(col), batch.vec4_column(col));
1150 }
1151 for col in 0..schema.n_scalars() {
1152 assert_eq!(concatenated.scalar_column(col), batch.scalar_column(col));
1153 }
1154 assert_eq!(concatenated.weights_column(), batch.weights_column());
1155
1156 if schema.n_p4s() > 0 || schema.n_scalars() > 0 {
1157 let reweighted = batch.reweight(|_, weight| weight + 1.0);
1158 if schema.n_p4s() > 0 {
1159 assert_eq!(
1160 reweighted.vec4_column(0).as_ptr(),
1161 batch.vec4_column(0).as_ptr()
1162 );
1163 }
1164 if schema.n_scalars() > 0 {
1165 assert_eq!(
1166 reweighted.scalar_column(0).as_ptr(),
1167 batch.scalar_column(0).as_ptr()
1168 );
1169 }
1170 }
1171 }
1172 }
1173}