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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
//! Streaming and bulk insert operations for SDBQL executor.
//!
//! This module contains:
//! - try_streaming_bulk_insert: Optimized bulk insert for large ranges
//! - log_mutation: Log single mutation for replication
//! - log_mutations_async: Async batch mutation logging
use serde_json::Value;
use super::super::types::Context;
use super::super::QueryExecutor;
use crate::error::{DbError, DbResult};
use crate::sdbql::ast::*;
use crate::sync::log::LogEntry;
use crate::sync::protocol::Operation;
impl<'a> QueryExecutor<'a> {
pub(super) fn try_streaming_bulk_insert(
&self,
query: &Query,
initial_bindings: &Context,
) -> DbResult<Option<(Vec<Value>, usize)>> {
// Check pattern: exactly 2 body clauses (FOR + INSERT), no sort/limit/filter
if query.body_clauses.len() != 2
|| query.sort_clause.is_some()
|| query.limit_clause.is_some()
{
return Ok(None);
}
// First clause must be FOR with range expression
let for_clause = match &query.body_clauses[0] {
BodyClause::For(fc) => fc,
_ => return Ok(None),
};
// Second clause must be INSERT
let insert_clause = match &query.body_clauses[1] {
BodyClause::Insert(ic) => ic,
_ => return Ok(None),
};
// This path returns the loop index for each row, so it only serves a
// RETURN of exactly that (or no RETURN), and it knows nothing of
// OPTIONS or NEW / OLD.
let returns_index = match &query.return_clause {
None => true,
Some(rc) => {
!rc.distinct
&& matches!(&rc.expression, Expression::Variable(v) if v == &for_clause.variable)
}
};
if !returns_index
|| insert_clause.options != MutationOptions::default()
|| insert_clause.binds_new
|| insert_clause.binds_old
|| !query.post_limit_lets.is_empty()
{
return Ok(None);
}
// FOR must have a range expression
let range_expr = match &for_clause.source_expression {
Some(Expression::Range(start, end)) => (start, end),
_ => return Ok(None),
};
// Evaluate range bounds
let start_val = self.evaluate_expr_with_context(range_expr.0, initial_bindings)?;
let end_val = self.evaluate_expr_with_context(range_expr.1, initial_bindings)?;
let start = match &start_val {
Value::Number(n) => n.as_i64().or_else(|| n.as_f64().map(|f| f as i64)),
_ => None,
};
let end = match &end_val {
Value::Number(n) => n.as_i64().or_else(|| n.as_f64().map(|f| f as i64)),
_ => None,
};
let (start, end) = match (start, end) {
(Some(s), Some(e)) => (s, e),
_ => return Ok(None),
};
// Only use streaming for large ranges (>5000 items)
const STREAMING_THRESHOLD: i64 = 5_000;
const BATCH_SIZE: i64 = 5_000;
// i128: `end - start` overflows i64 for `-9223372036854775808..9223372036854775807`.
let total_count = (end as i128 - start as i128 + 1).max(0);
if total_count < STREAMING_THRESHOLD as i128 {
return Ok(None); // Use normal path for small ranges
}
// With a RETURN every produced index is kept until the end, so the
// range is bounded by the row ceiling like any other row set. This
// path reads the range straight from the AST, so the evaluator's own
// range cap never applied to it.
if query.return_clause.is_some() && total_count > self.max_intermediate_rows() as i128 {
return Err(DbError::ExecutionError(format!(
"Query exceeded the intermediate row limit ({} > {}). Add a \
FILTER or LIMIT, or raise SOLIDB_MAX_INTERMEDIATE_ROWS.",
total_count,
self.max_intermediate_rows()
)));
}
tracing::info!(
"STREAMING INSERT: Processing {} documents in batches of {}",
total_count,
BATCH_SIZE
);
// Get collection once
let collection = self.get_collection_for_write(&insert_clause.collection)?;
// Disable streaming bulk insert for sharded collections (fall back to generic path for routing)
if let Some(config) = collection.get_shard_config() {
if config.num_shards > 0 {
tracing::debug!(
"Streaming insert disabled for sharded collection: {}",
insert_clause.collection
);
return Ok(None);
}
}
let var_name = &for_clause.variable;
let mut all_results: Vec<Value> = Vec::new();
let mut current = start;
let total_start = std::time::Instant::now();
while current <= end {
// Deadline and result growth, once per batch.
self.check_budget(all_results.len())?;
let batch_end = current.saturating_add(BATCH_SIZE - 1).min(end);
let batch_size = (batch_end - current + 1) as usize;
// Build documents for this batch
let mut documents = Vec::with_capacity(batch_size);
for i in current..=batch_end {
let mut ctx = initial_bindings.clone();
ctx.insert(var_name.clone(), Value::Number(serde_json::Number::from(i)));
let doc_value = self.evaluate_expr_with_context(&insert_clause.document, &ctx)?;
documents.push(doc_value);
}
// Batch insert
let inserted_docs = collection.insert_batch(documents)?;
// Handle RETURN clause if present
if query.return_clause.is_some() {
for i in current..=batch_end {
all_results.push(Value::Number(serde_json::Number::from(i)));
}
}
// Log to replication asynchronously
self.log_mutations_async(&insert_clause.collection, Operation::Insert, &inserted_docs);
// `insert_batch` already wrote the index entries atomically with
// the documents (audit D5); re-indexing here would race writers.
current = batch_end + 1;
// Throttled flush of stats (max 1 per second)
collection.flush_stats_throttled();
// Log progress for very large inserts
if total_count > 100_000 && (current - start) % 100_000 == 0 {
tracing::info!(
"STREAMING INSERT: Processed {}/{} documents",
current - start,
total_count
);
}
}
let elapsed = total_start.elapsed();
tracing::info!(
"STREAMING INSERT: Completed {} documents in {:?} ({:.0} docs/sec)",
total_count,
elapsed,
total_count as f64 / elapsed.as_secs_f64()
);
// Final flush to ensure count is persisted
collection.flush_stats();
Ok(Some((all_results, total_count as usize)))
}
/// Log a mutation to the replication service
pub(super) fn log_mutation(
&self,
collection: &str,
operation: Operation,
key: &str,
data: Option<&Value>,
) {
if let (Some(repl), Some(ref db)) = (&self.replication, &self.database) {
let entry = LogEntry {
sequence: 0,
node_id: "".to_string(),
database: db.clone(),
collection: collection.to_string(),
operation,
key: key.to_string(),
data: data.and_then(|v| serde_json::to_vec(v).ok()),
timestamp: chrono::Utc::now().timestamp_millis() as u64,
origin_sequence: None,
};
let _ = repl.append(entry);
}
}
/// Log multiple mutations asynchronously in a background thread
/// Used for bulk INSERT operations to avoid blocking the response
/// Log multiple mutations asynchronously in a background thread
/// Used for bulk INSERT operations to avoid blocking the response
pub(super) fn log_mutations_async(
&self,
collection: &str,
operation: Operation,
docs: &[crate::storage::Document],
) {
// Clone the replication service if available
let repl_clone = self.replication.cloned();
let db_clone = self.database.clone();
if let (Some(repl), Some(db)) = (repl_clone, db_clone) {
let collection = collection.to_string();
// Serialize documents upfront
let entries: Vec<LogEntry> = docs
.iter()
.map(|doc| LogEntry {
sequence: 0,
node_id: "".to_string(),
database: db.clone(),
collection: collection.clone(),
operation, // Operation is Copy
key: doc.key.clone(),
data: serde_json::to_vec(&doc.to_value()).ok(),
timestamp: chrono::Utc::now().timestamp_millis() as u64,
origin_sequence: None,
})
.collect();
let count = entries.len();
tracing::debug!(
"INSERT: Starting async replication logging for {} docs",
count
);
// Execute replication logging synchronously (RocksDB is fast)
// We use the cloned ReplicationLog reference which points to the same DB
let start = std::time::Instant::now();
let _ = repl.append_batch(entries);
let elapsed = start.elapsed();
tracing::debug!(
"INSERT: Replication logging of {} docs completed in {:?}",
count,
elapsed
);
}
}
}