sdf-metadata 0.14.0

metadata definition for core sdf
Documentation 
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

use anyhow::Result;

use crate::{
    metadata::io::topic::KVSchemaType,
    util::{
        sdf_types_map::SdfTypesMap, validation_error::ValidationError,
        validation_failure::ValidationFailure,
    },
    wit::{
        dataflow::PostTransforms,
        operator::{OperatorType, StepInvocation},
    },
};

impl PostTransforms {
    pub fn operators(&self) -> Vec<(StepInvocation, OperatorType)> {
        match self {
            PostTransforms::AssignTimestamp(window) => window.operators(),
            PostTransforms::Partition(partition) => partition.operators(),
        }
    }
    pub fn output_type(&self, input_type: KVSchemaType) -> Result<KVSchemaType, ValidationError> {
        match self {
            PostTransforms::AssignTimestamp(window) => window.output_type(input_type),
            PostTransforms::Partition(partition) => partition.output_type(input_type),
        }
    }

    pub fn validate(
        &self,
        types: &SdfTypesMap,
        expected_input_type: &KVSchemaType,
    ) -> Result<(), ValidationFailure> {
        let mut errors = ValidationFailure::new();

        match self {
            PostTransforms::AssignTimestamp(window) => {
                if let Err(window_error) =
                    window.validate(types, expected_input_type, "transforms block")
                {
                    errors.concat_with_context("Window", &window_error)
                }
            }
            PostTransforms::Partition(partition) => {
                if let Err(partition_error) =
                    partition.validate(types, expected_input_type, "transforms block")
                {
                    errors.concat_with_context("Partition", &partition_error)
                }
            }
        }

        if errors.any() {
            Err(errors)
        } else {
            Ok(())
        }
    }

    #[cfg(feature = "parser")]
    pub fn update_inline_operators(&mut self) -> Result<()> {
        match self {
            PostTransforms::AssignTimestamp(window) => window.update_inline_operators(),
            PostTransforms::Partition(partition) => partition.update_inline_operators(),
        }
    }
}

#[cfg(test)]
mod test {
    use crate::wit::operator::{
        PartitionOperator, TransformOperator, PostTransforms, StepInvocation, Transforms, Window,
    };

    #[test]
    fn test_operators() {
        let post_transforms = PostTransforms::AssignTimestamp(Window {
            partition: Some(PartitionOperator {
                assign_key: StepInvocation {
                    uses: "my-assign".into(),
                    ..Default::default()
                },
                transforms: Transforms {
                    steps: vec![TransformOperator::Map(StepInvocation {
                        uses: "my-map".into(),
                        ..Default::default()
                    })],
                },
                update_state: None,
            }),
            flush: Some(StepInvocation {
                uses: "my-flush".into(),
                ..Default::default()
            }),
            assign_timestamp: StepInvocation {
                uses: "my-assign-timestamp".into(),
                ..Default::default()
            },
            ..Default::default()
        });

        let op = post_transforms.operators();
        assert_eq!(op.len(), 4);
        assert_eq!(op[0].0.uses, "my-assign-timestamp");
        assert_eq!(op[1].0.uses, "my-assign");
        assert_eq!(op[2].0.uses, "my-map");
        assert_eq!(op[3].0.uses, "my-flush");
    }
}