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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
// SPDX-License-Identifier: BUSL-1.1
//! Core row-ingest path: per-row value coercion, ON CONFLICT DO UPDATE merge
//! resolution, and the row-level `MutationEngine` insert call.
use nodedb_types::columnar::ColumnarSchema;
use nodedb_types::columnar::schema::{TS_SYSTEM, TS_VALID_FROM, TS_VALID_UNTIL};
use nodedb_types::surrogate::Surrogate;
use nodedb_types::value::Value;
use crate::bridge::envelope::{ErrorCode, Response};
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::handlers::upsert::apply_on_conflict_updates;
use crate::data::executor::task::ExecutionTask;
use nodedb_physical::physical_plan::ColumnarInsertIntent;
use nodedb_physical::physical_plan::document::UpdateValue;
use super::schema::{ndb_field_to_value, row_values_to_object};
/// Parameters for [`CoreLoop::insert_columnar_rows`].
pub(in crate::data::executor) struct RowIngestParams<'a> {
pub engine_key: &'a (nodedb_types::DatabaseId, crate::types::TenantId, String),
pub schema: &'a ColumnarSchema,
pub bitemporal: bool,
pub intent: ColumnarInsertIntent,
pub on_conflict_updates: &'a [(String, UpdateValue)],
pub surrogates: &'a [Surrogate],
pub ndb_rows: &'a [nodedb_types::Value],
}
impl CoreLoop {
/// Insert each row in `params.ndb_rows` into the columnar engine at
/// `params.engine_key`, applying intent-specific ON CONFLICT semantics
/// (upsert-overwrite for `Insert` and `Put`, silent skip for
/// `InsertIfAbsent`, merge-via-`apply_on_conflict_updates` for `Put`
/// with non-empty `on_conflict_updates`).
///
/// Returns the accepted row count, or `Err(Response)` on the first
/// unrecoverable error (short-circuits the remaining rows).
pub(in crate::data::executor) fn insert_columnar_rows(
&mut self,
task: &ExecutionTask,
params: RowIngestParams<'_>,
) -> Result<u64, Response> {
let RowIngestParams {
engine_key,
schema,
bitemporal,
intent,
on_conflict_updates,
surrogates,
ndb_rows,
} = params;
let mut accepted = 0u64;
for (row_idx, row) in ndb_rows.iter().enumerate() {
let obj = match row {
nodedb_types::Value::Object(m) => m,
_ => continue,
};
// Build Value slice in schema order. For bitemporal
// collections, the three reserved columns are auto-populated
// when absent from the user payload: `_ts_system` is always
// clamped to the current wall-clock time (clients cannot
// forge system time), `_ts_valid_from` / `_ts_valid_until`
// default to the open interval `[i64::MIN, i64::MAX)` if
// missing.
let sys_now = if bitemporal {
self.bitemporal_now_ms()
} else {
0
};
let values: Vec<Value> = match schema
.columns
.iter()
.map(|col| match col.name.as_str() {
TS_SYSTEM if bitemporal => Ok(Value::Integer(sys_now)),
TS_VALID_FROM if bitemporal => Ok(match obj.get(TS_VALID_FROM) {
Some(Value::Integer(i)) => Value::Integer(*i),
_ => Value::Integer(i64::MIN),
}),
TS_VALID_UNTIL if bitemporal => Ok(match obj.get(TS_VALID_UNTIL) {
Some(Value::Integer(i)) => Value::Integer(*i),
_ => Value::Integer(i64::MAX),
}),
_ => ndb_field_to_value(obj.get(&col.name), &col.column_type),
})
.collect::<Result<Vec<Value>, crate::Error>>()
{
Ok(v) => v,
Err(e) => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: format!("columnar insert coercion: {e}"),
},
));
}
};
// Resolve the actual row to write (merged for ON CONFLICT DO
// UPDATE, plain otherwise). This runs before the mutable
// engine borrow needed by the insert call.
let final_values: Vec<Value> = match intent {
ColumnarInsertIntent::Put if !on_conflict_updates.is_empty() => {
let pk_bytes = {
let engine = match self.columnar_engines.get(engine_key) {
Some(e) => e,
None => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: "columnar engine vanished during insert".into(),
},
));
}
};
match engine.encode_pk_from_row(&values) {
Ok(b) => b,
Err(e) => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: format!("columnar insert: pk encode failed: {e}"),
},
));
}
}
};
let prior_row = self
.columnar_engines
.get(engine_key)
.and_then(|e| e.lookup_memtable_row_by_pk(&pk_bytes))
.or_else(|| self.read_flushed_row_by_pk(engine_key, &pk_bytes));
match prior_row {
None => values,
Some(prior) => {
let existing_val = row_values_to_object(schema, &prior);
let excluded_val = row_values_to_object(schema, &values);
let merged = apply_on_conflict_updates(
existing_val,
&excluded_val,
on_conflict_updates,
);
let merged_obj = match merged {
nodedb_types::Value::Object(m) => m,
_ => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: "merged ON CONFLICT value was not an object"
.into(),
},
));
}
};
match schema
.columns
.iter()
.map(|col| {
ndb_field_to_value(merged_obj.get(&col.name), &col.column_type)
})
.collect::<Result<Vec<Value>, crate::Error>>()
{
Ok(v) => v,
Err(e) => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: format!("columnar ON CONFLICT coercion: {e}"),
},
));
}
}
}
}
}
_ => values,
};
let engine = match self.columnar_engines.get_mut(engine_key) {
Some(e) => e,
None => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: "columnar engine vanished during insert".into(),
},
));
}
};
let row_surrogate = surrogates.get(row_idx).copied();
let result = match intent {
ColumnarInsertIntent::InsertIfAbsent => engine.insert_if_absent(&final_values),
ColumnarInsertIntent::Insert | ColumnarInsertIntent::Put => match row_surrogate {
Some(s) => engine.insert_with_surrogate(&final_values, s),
None => engine.insert(&final_values),
},
};
match result {
Ok(_) => accepted += 1,
Err(e) => {
return Err(self.response_error(
task,
ErrorCode::Internal {
detail: format!("columnar insert failed: {e}"),
},
));
}
}
}
Ok(accepted)
}
}