Skip to main content

laddu_data/data/
event.rs

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/// Immutable columnar batch of events sharing one schema.
236#[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    /// Validates column counts and lengths and constructs a batch.
258    ///
259    /// # Errors
260    ///
261    /// Returns [`LadduDataError`] when column counts do not match `schema` or
262    /// column and weight lengths are inconsistent.
263    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    /// Collects owned row events into a columnar batch.
287    ///
288    /// # Errors
289    ///
290    /// Returns [`LadduDataError`] when an event has the wrong number of values
291    /// or weighted and unweighted events are mixed.
292    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    /// Returns the shared logical schema.
302    pub fn schema(&self) -> &Arc<Schema> {
303        &self.schema
304    }
305
306    /// Returns the number of rows.
307    pub fn len(&self) -> usize {
308        self.len
309    }
310
311    /// Returns the logical payload bytes per event represented by this batch.
312    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    /// Returns the retained column payload size in bytes.
321    ///
322    /// Shared schema metadata, allocation headers, and other owners of shared
323    /// columns are not included.
324    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    /// Returns whether the batch contains no rows.
334    pub fn is_empty(&self) -> bool {
335        self.len == 0
336    }
337
338    /// Returns a four-momentum column by index.
339    pub fn vec4_column(&self, index: usize) -> &[RealVec4] {
340        &self.parts.p4s[index]
341    }
342
343    /// Returns a scalar column by index.
344    pub fn scalar_column(&self, index: usize) -> &[f64] {
345        &self.parts.scalars[index]
346    }
347
348    /// Returns the optional explicit weight column.
349    pub fn weights_column(&self) -> Option<&[f64]> {
350        self.parts.weights.as_slice()
351    }
352
353    /// Returns a four-momentum column by logical name.
354    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    /// Returns a scalar column by logical name.
360    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    /// Returns one four-momentum cell.
366    pub fn p4_at(&self, col: usize, row: usize) -> RealVec4 {
367        self.parts.p4s[col][row]
368    }
369
370    /// Returns a four-momentum column through a schema-resolved binding.
371    ///
372    /// Schema compatibility is checked once, so callers can safely reuse the
373    /// returned slice for every row in a packing loop.
374    ///
375    /// # Errors
376    ///
377    /// Returns [`LadduDataError::Schema`] when the binding belongs to another
378    /// schema.
379    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    /// Returns one scalar cell.
389    pub fn scalar_at(&self, col: usize, row: usize) -> f64 {
390        self.parts.scalars[col][row]
391    }
392
393    /// Returns a scalar column through a schema-resolved binding.
394    ///
395    /// Schema compatibility is checked once, so callers can safely reuse the
396    /// returned slice for every row in a packing loop.
397    ///
398    /// # Errors
399    ///
400    /// Returns [`LadduDataError::Schema`] when the binding belongs to another
401    /// schema.
402    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    /// Returns the explicit row weight, or one when weights are absent.
412    pub fn weights_at(&self, row: usize) -> f64 {
413        self.parts.weights.at(row)
414    }
415
416    /// Returns a borrowed view of one row.
417    pub fn event(&self, row: usize) -> BatchEvent<'_> {
418        BatchEvent { batch: self, row }
419    }
420
421    /// Iterates over borrowed event views.
422    pub fn iter(&self) -> impl Iterator<Item = BatchEvent<'_>> {
423        (0..self.len()).map(|i| self.event(i))
424    }
425
426    /// Copies selected rows into a new batch in the requested order.
427    ///
428    /// # Panics
429    ///
430    /// Panics when a selected row is outside the batch.
431    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    /// Copies rows satisfying `keep` into a new batch.
441    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    /// Returns a batch sharing value columns with newly computed weights.
451    ///
452    /// # Panics
453    ///
454    /// Panics if an internal batch invariant is violated while rebuilding the
455    /// batch.
456    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    /// Copies the half-open row range `start..end` into a new batch.
465    ///
466    /// # Panics
467    ///
468    /// Panics when `start > end` or `end` exceeds the batch length.
469    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    /// Concatenates schema-compatible batches.
486    ///
487    /// # Errors
488    ///
489    /// Returns [`LadduDataError`] when `batches` is empty or contains
490    /// incompatible schemas.
491    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/// Borrowed view of one row in an [`EventBatch`].
555#[derive(Copy, Clone, Debug)]
556pub struct BatchEvent<'a> {
557    batch: &'a EventBatch,
558    row: usize,
559}
560
561impl<'a> BatchEvent<'a> {
562    /// Returns the row index.
563    pub fn row(&self) -> usize {
564        self.row
565    }
566
567    /// Returns the backing batch.
568    pub fn batch(&self) -> &'a EventBatch {
569        self.batch
570    }
571
572    /// Returns a four-momentum value by column index.
573    pub fn p4(&self, col: usize) -> RealVec4 {
574        self.batch.p4_at(col, self.row)
575    }
576
577    /// Returns a scalar value by column index.
578    pub fn scalar(&self, col: usize) -> f64 {
579        self.batch.scalar_at(col, self.row)
580    }
581
582    /// Returns the row weight, defaulting to one.
583    pub fn weight(&self) -> f64 {
584        self.batch.weights_at(self.row)
585    }
586
587    /// Returns a four-momentum value by logical name.
588    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    /// Returns a scalar value by logical name.
594    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/// Borrowed event view with a possibly transformed weight.
601#[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    /// Returns the row index in the backing batch.
610    pub fn row(&self) -> usize {
611        self.row
612    }
613
614    /// Returns a four-momentum value by column index.
615    pub fn p4(&self, col: usize) -> RealVec4 {
616        self.batch.p4_at(col, self.row)
617    }
618
619    /// Returns a scalar value by column index.
620    pub fn scalar(&self, col: usize) -> f64 {
621        self.batch.scalar_at(col, self.row)
622    }
623
624    /// Returns this view's effective weight.
625    pub fn weight(&self) -> f64 {
626        self.weight
627    }
628
629    /// Returns a four-momentum value by logical name.
630    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    /// Returns a scalar value by logical name.
636    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/// Owned row-oriented event used while constructing batches.
643#[derive(Clone, Debug)]
644pub struct OwnedEvent {
645    /// Four-momentum values in schema order.
646    pub p4s: Vec<RealVec4>,
647    /// Scalar values in schema order.
648    pub scalars: Vec<f64>,
649    /// Optional explicit event weight.
650    pub weight: Option<f64>,
651}
652
653impl OwnedEvent {
654    /// Creates an unweighted owned event.
655    pub fn new(p4s: Vec<RealVec4>, scalars: Vec<f64>) -> Self {
656        Self {
657            p4s,
658            scalars,
659            weight: None,
660        }
661    }
662
663    /// Creates an owned event with an explicit weight.
664    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
673/// Shared checked assembly for row-oriented batch producers.
674pub(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
791/// Incremental builder for a columnar [`EventBatch`].
792pub struct EventBatchBuilder {
793    assembler: BatchAssembler,
794}
795
796impl EventBatchBuilder {
797    /// Creates an empty builder.
798    pub fn new(schema: Arc<Schema>) -> Self {
799        Self::with_capacity(schema, 0)
800    }
801
802    /// Creates an empty builder with per-column capacity.
803    pub fn with_capacity(schema: Arc<Schema>, capacity: usize) -> Self {
804        Self {
805            assembler: BatchAssembler::new(schema, capacity),
806        }
807    }
808
809    /// Appends an unweighted event from ordered values.
810    ///
811    /// # Errors
812    ///
813    /// Returns [`LadduDataError`] when value counts do not match the schema or
814    /// the builder already contains weighted events.
815    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    /// Appends a weighted event from ordered values.
827    ///
828    /// # Errors
829    ///
830    /// Returns [`LadduDataError`] when value counts do not match the schema or
831    /// the builder already contains unweighted events.
832    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    /// Validates and appends one owned event.
850    ///
851    /// # Errors
852    ///
853    /// Returns [`LadduDataError`] when the event shape does not match the
854    /// schema or its weight presence differs from prior events.
855    pub fn push_event(&mut self, event: OwnedEvent) -> LadduDataResult<&mut Self> {
856        self.assembler.push_owned(event)?;
857        Ok(self)
858    }
859
860    /// Appends all owned events from an iterator.
861    ///
862    /// # Errors
863    ///
864    /// Returns the first [`LadduDataError`] produced by an event whose shape or
865    /// weight presence is incompatible with the builder.
866    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    /// Finalizes the builder into an immutable batch.
878    ///
879    /// # Errors
880    ///
881    /// Returns [`LadduDataError`] if the accumulated columns or weights have
882    /// inconsistent lengths.
883    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}