pipeline/etl/
field.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::ops::{Deref, DerefMut};
16use std::str::FromStr;
17
18use snafu::OptionExt;
19
20use crate::error::{EmptyInputFieldSnafu, Error, MissingInputFieldSnafu, Result};
21
22/// Raw processor-defined inputs and outputs
23#[derive(Debug, Default, Clone)]
24pub struct Field {
25    input_field: String,
26    target_field: Option<String>,
27}
28
29impl FromStr for Field {
30    type Err = Error;
31
32    fn from_str(s: &str) -> Result<Self> {
33        let mut parts = s.split(',');
34        let input_field = parts
35            .next()
36            .context(MissingInputFieldSnafu)?
37            .trim()
38            .to_string();
39        let target_field = parts.next().map(|x| x.trim().to_string());
40
41        if input_field.is_empty() {
42            return EmptyInputFieldSnafu.fail();
43        }
44
45        Ok(Field {
46            input_field,
47            target_field,
48        })
49    }
50}
51
52impl Field {
53    /// Create a new field with the given input and target fields.
54    pub(crate) fn new(input_field: impl Into<String>, target_field: Option<String>) -> Self {
55        Field {
56            input_field: input_field.into(),
57            target_field,
58        }
59    }
60
61    /// Get the input field.
62    pub(crate) fn input_field(&self) -> &str {
63        &self.input_field
64    }
65
66    /// Get the target field.
67    pub(crate) fn target_field(&self) -> Option<&str> {
68        self.target_field.as_deref()
69    }
70
71    /// Get the target field or the input field if the target field is not set.
72    pub(crate) fn target_or_input_field(&self) -> &str {
73        self.target_field.as_deref().unwrap_or(&self.input_field)
74    }
75
76    pub(crate) fn set_target_field(&mut self, target_field: Option<String>) {
77        self.target_field = target_field;
78    }
79}
80
81/// A collection of fields.
82#[derive(Debug, Default, Clone)]
83pub struct Fields(Vec<Field>);
84
85impl Fields {
86    pub(crate) fn new(fields: Vec<Field>) -> Self {
87        Fields(fields)
88    }
89
90    pub(crate) fn one(field: Field) -> Self {
91        Fields(vec![field])
92    }
93}
94
95impl Deref for Fields {
96    type Target = Vec<Field>;
97
98    fn deref(&self) -> &Self::Target {
99        &self.0
100    }
101}
102
103impl DerefMut for Fields {
104    fn deref_mut(&mut self) -> &mut Self::Target {
105        &mut self.0
106    }
107}
108
109impl IntoIterator for Fields {
110    type Item = Field;
111    type IntoIter = std::vec::IntoIter<Field>;
112
113    fn into_iter(self) -> Self::IntoIter {
114        self.0.into_iter()
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use crate::etl::field::Field;
121
122    #[test]
123    fn test_parse_field() {
124        let field: Result<Field, _> = " ".parse();
125        assert!(field.is_err());
126
127        let field: Result<Field, _> = ",".parse();
128        assert!(field.is_err());
129
130        let field: Result<Field, _> = ",field".parse();
131        assert!(field.is_err());
132
133        let cases = [
134            // ("field", "field", None, None),
135            ("field, target_field", "field", Some("target_field")),
136            ("field", "field", None),
137        ];
138
139        for (s, field, target_field) in cases.into_iter() {
140            let f: Field = s.parse().unwrap();
141            assert_eq!(f.input_field(), field, "{s}");
142            assert_eq!(f.target_field(), target_field, "{s}");
143        }
144    }
145}
pipeline/etl/field.rs

pipeline/etl/
field.rs
