1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

pub mod array;
mod format;
mod named;
pub mod scalar;
mod sink;
mod stream;
pub mod tuple;

use std::sync::Arc;

pub use format::{RowBatchBuilder, RowFormat, RowFormatView, RowViewIter};
pub use named::NamedRowFormat;
pub use sink::RowSink;
pub use stream::RowStream;

use crate::Time;
use datafusion::arrow::{array::ArrayRef, datatypes::Field};

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Row<R>(pub Time, pub R);

impl<R: RowFormat> RowFormat for Row<R> {
    const COLUMNS: usize = R::COLUMNS + 1;
    type Builder = RowBuilder<R::Builder>;
    type View = RowView<R>;

    fn builder(fields: &[Arc<Field>]) -> crate::Result<Self::Builder> {
        if fields.len() != Self::COLUMNS {
            return Err(crate::Error::ColumnCount(Self::COLUMNS, fields.len()));
        }
        let time = Time::builder(&fields[..1])?;
        let values = R::builder(&fields[1..])?;
        Ok(RowBuilder { time, values })
    }

    fn view(rows: usize, fields: &[Arc<Field>], arrays: &[ArrayRef]) -> crate::Result<Self::View> {
        if arrays.len() != Self::COLUMNS {
            return Err(crate::Error::ColumnCount(Self::COLUMNS, arrays.len()));
        }
        let time = Time::view(rows, &fields[..1], &arrays[..1])?;
        let values = R::view(rows, &fields[1..], &arrays[1..])?;
        debug_assert_eq!(values.len(), rows);

        Ok(RowView { time, values })
    }
}

impl<R: NamedRowFormat> NamedRowFormat for Row<R> {
    fn name(&self, i: usize) -> String {
        match i {
            0 => "time".to_string(),
            _ => self.1.name(i - 1),
        }
    }

    fn data_type(&self, i: usize) -> crate::TensorType {
        match i {
            0 => crate::TensorType::Timestamp,
            _ => self.1.data_type(i - 1),
        }
    }

    fn row_shape(&self, i: usize) -> crate::shape::Dyn {
        match i {
            0 => crate::shape::Dyn::from([]),
            _ => self.1.row_shape(i - 1),
        }
    }
}

#[derive(Debug, Clone)]
pub struct RowBuilder<R> {
    time: scalar::ScalarBuilder<Time>,
    values: R,
}

impl<R> RowBatchBuilder<Row<R>> for RowBuilder<R::Builder>
where
    R: RowFormat,
{
    #[inline]
    fn len(&self) -> usize {
        self.time.len()
    }

    fn push(&mut self, row: Row<R>) {
        self.time.push(row.0);
        self.values.push(row.1);
    }

    fn build_columns(&mut self) -> crate::Result<Vec<ArrayRef>> {
        let mut cols = Vec::with_capacity(<Row<R> as RowFormat>::COLUMNS);
        cols.extend(self.time.build_columns()?);
        cols.extend(self.values.build_columns()?);

        Ok(cols)
    }
}

#[derive(Debug, Clone)]
pub struct RowView<R: RowFormat> {
    time: scalar::ScalarRowView<Time>,
    values: R::View,
}

impl<R: RowFormat> RowFormatView<Row<R>> for RowView<R> {
    fn len(&self) -> usize {
        self.time.len()
    }

    fn row(&self, i: usize) -> Row<R> {
        Row(self.time.row(i), self.values.row(i))
    }

    unsafe fn row_unchecked(&self, i: usize) -> Row<R> {
        Row(self.time.row_unchecked(i), self.values.row_unchecked(i))
    }
}

impl<R: RowFormat> IntoIterator for RowView<R> {
    type Item = Row<R>;
    type IntoIter = RowViewIter<Row<R>, Self>;

    fn into_iter(self) -> Self::IntoIter {
        RowViewIter::new(self)
    }
}