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
128
129
130
131
132
133
134
135
136
137
138
139
140

use {
    super::{
        context::FilterContext,
        evaluate::{evaluate, Evaluated},
    },
    crate::{
        ast::{Assignment, ColumnDef, ColumnOption, ColumnOptionDef},
        data::{schema::ColumnDefExt, Row, Value},
        result::Result,
        store::GStore,
    },
    futures::stream::{self, TryStreamExt},
    serde::Serialize,
    std::{fmt::Debug, rc::Rc},
    thiserror::Error,
};

#[derive(Error, Serialize, Debug, PartialEq)]
pub enum UpdateError {
    #[error("column not found {0}")]
    ColumnNotFound(String),

    #[error("update on primary key is not supported: {0}")]
    UpdateOnPrimaryKeyNotSupported(String),

    #[error("conflict on schema, row data does not fit to schema")]
    ConflictOnSchema,
}

pub struct Update<'a> {
    storage: &'a dyn GStore,
    table_name: &'a str,
    fields: &'a [Assignment],
    column_defs: &'a [ColumnDef],
}

impl<'a> Update<'a> {
    pub fn new(
        storage: &'a dyn GStore,
        table_name: &'a str,
        fields: &'a [Assignment],
        column_defs: &'a [ColumnDef],
    ) -> Result<Self> {
        for assignment in fields.iter() {
            let Assignment { id, .. } = assignment;

            if column_defs.iter().all(|col_def| &col_def.name != id) {
                return Err(UpdateError::ColumnNotFound(id.to_owned()).into());
            } else if column_defs.iter().any(|ColumnDef { name, options, .. }| {
                if name != id {
                    return false;
                }

                options.iter().any(|ColumnOptionDef { option, .. }| {
                    option == &ColumnOption::Unique { is_primary: true }
                })
            }) {
                return Err(UpdateError::UpdateOnPrimaryKeyNotSupported(id.to_owned()).into());
            }
        }

        Ok(Self {
            storage,
            table_name,
            fields,
            column_defs,
        })
    }

    async fn find(&self, row: &Row, column_def: &ColumnDef) -> Result<Option<Value>> {
        let all_columns = Rc::from(self.all_columns());
        let context = FilterContext::new(self.table_name, Rc::clone(&all_columns), Some(row), None);
        let context = Some(Rc::new(context));

        match self
            .fields
            .iter()
            .find(|assignment| assignment.id == column_def.name)
        {
            None => Ok(None),
            Some(assignment) => {
                let Assignment { value, .. } = &assignment;
                let ColumnDef { data_type, .. } = column_def;
                let nullable = column_def.is_nullable();

                let value = match evaluate(self.storage, context, None, value).await? {
                    Evaluated::Literal(v) => Value::try_from_literal(data_type, &v)?,
                    Evaluated::Value(v) => {
                        v.validate_type(data_type)?;
                        v.into_owned()
                    }
                };

                value.validate_null(nullable)?;

                Ok(Some(value))
            }
        }
    }

    pub async fn apply(&self, row: Row) -> Result<Row> {
        let Row(values) = &row;

        let values = values.clone().into_iter().enumerate().map(|(i, value)| {
            self.column_defs
                .get(i)
                .map(|col_def| (col_def, value))
                .ok_or_else(|| UpdateError::ConflictOnSchema.into())
        });

        stream::iter(values)
            .and_then(|(col_def, value)| {
                let row = &row;

                async move {
                    self.find(row, col_def)
                        .await
                        .transpose()
                        .unwrap_or(Ok(value))
                }
            })
            .try_collect::<Vec<_>>()
            .await
            .map(Row)
    }

    pub fn all_columns(&self) -> Vec<String> {
        self.column_defs
            .iter()
            .map(|col_def| col_def.name.to_owned())
            .collect()
    }

    pub fn columns_to_update(&self) -> Vec<String> {
        self.fields
            .iter()
            .map(|assignment| assignment.id.to_owned())
            .collect()
    }
}