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
141
142
143
144
145
146
use anyhow::Result;
use crate::catalog::providers::TableProvider;
use crate::ctx::FrozenContext;
use crate::dbs::Statement;
use crate::doc::Document;
use crate::err::Error;
impl Document {
pub(super) async fn store_record_data(
&mut self,
ctx: &FrozenContext,
stm: &Statement<'_>,
) -> Result<()> {
// Check if changed
if !self.is_modified() {
return Ok(());
}
// Get the document table
let tb = self.doc_ctx.tb()?;
// Check if the table is DROP
if tb.drop {
return Ok(());
}
// Get the record id
let rid = self.id()?;
// Get the namespace id
let ns = self.doc_ctx.ns().namespace_id;
// Get the database id
let db = self.doc_ctx.db().database_id;
// Prep the doc
let doc = self.current.doc.clone().into_read_only();
// Match the statement type
match stm {
// This is a INSERT statement so try to insert the key.
// For INSERT statements we don't first check for the
// entry from the storage engine, so when we attempt
// to store the record value, we presume that the key
// does not exist. If the record value exists then we
// attempt to run the ON DUPLICATE KEY UPDATE clause but
// at this point the current document is not empty so we
// set and update the key, without checking if the key
// already exists in the storage engine.
Statement::Insert(_) if self.is_iteration_initial() => {
match ctx.tx().put_record(ns, db, &rid.table, &rid.key, doc).await {
// The key already exists, so return an error
Err(e) => {
if matches!(
e.downcast_ref(),
Some(Error::Kvs(crate::kvs::Error::TransactionKeyAlreadyExists))
) {
Err(anyhow::Error::new(Error::RecordExists {
record: rid.as_ref().to_owned(),
}))
} else {
Err(e)
}
}
// Return other values
x => x,
}
}
// This is a UPSERT statement so try to insert the key.
// For UPSERT statements we don't first check for the
// entry from the storage engine, so when we attempt
// to store the record value, we must ensure that the
// key does not exist. If the record value exists then we
// retry and attempt to update the record which exists.
Statement::Upsert(_) if self.is_iteration_initial() => {
match ctx.tx().put_record(ns, db, &rid.table, &rid.key, doc).await {
// The key already exists, so return an error
Err(e) => {
if matches!(
e.downcast_ref(),
Some(Error::Kvs(crate::kvs::Error::TransactionKeyAlreadyExists))
) {
Err(anyhow::Error::new(Error::RecordExists {
record: rid.as_ref().to_owned(),
}))
} else {
Err(e)
}
}
// Return other values
x => x,
}
}
// This is a CREATE statement so try to insert the key.
// For CREATE statements we don't first check for the
// entry from the storage engine, so when we attempt
// to store the record value, we must ensure that the
// key does not exist. If it already exists, then we
// return an error, and the statement fails.
Statement::Create(_) => {
match ctx.tx().put_record(ns, db, &rid.table, &rid.key, doc).await {
// The key already exists, so return an error
Err(e) => {
if matches!(
e.downcast_ref(),
Some(Error::Kvs(crate::kvs::Error::TransactionKeyAlreadyExists))
) {
Err(anyhow::Error::new(Error::RecordExists {
record: rid.as_ref().to_owned(),
}))
} else {
Err(e)
}
}
x => x,
}
}
// SECURITY: a RELATE that resolved to a new edge (no existing
// record loaded into `self.initial`) must NOT overwrite a
// pre-existing record at the same id. `Document::relate`
// chooses the create path based on `self.current.doc.is_nullish()`
// *before* the SET / CONTENT / MERGE clause is applied, so an
// attacker-controlled `id = edge:existing` would otherwise reach
// `set_record` and silently overwrite the existing edge under
// create permissions. Use the create-only `put_record` here too.
Statement::Relate(_) if self.initial.doc.as_ref().is_nullish() => {
match ctx.tx().put_record(ns, db, &rid.table, &rid.key, doc).await {
Err(e) => {
if matches!(
e.downcast_ref(),
Some(Error::Kvs(crate::kvs::Error::TransactionKeyAlreadyExists))
) {
Err(anyhow::Error::new(Error::RecordExists {
record: rid.as_ref().to_owned(),
}))
} else {
Err(e)
}
}
x => x,
}
}
// Let's update the stored value for the specified key
_ => ctx.tx().set_record(ns, db, &rid.table, &rid.key, doc).await,
}?;
// KV write succeeded; mark the document as mutated so the
// per-statement affected-row counter (bumped from
// `Document::process`) reflects this row.
self.mutated = true;
Ok(())
}
}