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
// SPDX-License-Identifier: Apache-2.0
//! Columnar engine base operations dispatched to the Data Plane.
//!
//! `ColumnarOp` is the base for all columnar-profile collections:
//! - **Plain columnar**: analytics collections without time semantics.
//! - **Timeseries**: extends with `time_range` + `bucket_interval_ms` (via `TimeseriesOp`).
//! - **Spatial**: extends with R-tree + OGC predicates (via `SpatialOp`).
//!
//! All profiles share the same `ColumnarMemtable` → `SegmentWriter` infrastructure.
use nodedb_types::{Surrogate, SurrogateBitmap, SystemTimeScope};
/// Intent carried on `ColumnarOp::Insert` — see enum docs.
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
serde::Serialize,
serde::Deserialize,
zerompk::ToMessagePack,
zerompk::FromMessagePack,
)]
pub enum ColumnarInsertIntent {
/// Plain `INSERT`. On columnar, duplicate PK is **not** an error —
/// the prior row is tombstoned via the segment's delete bitmap and
/// the new row is appended. Cross-engine SQL consistency is kept on
/// the read side (`SELECT WHERE pk = X` returns one row) rather than
/// the insert-error side, matching ClickHouse's append + dedup model.
Insert,
/// `INSERT ... ON CONFLICT DO NOTHING`: silent no-op on duplicate.
InsertIfAbsent,
/// `UPSERT` / `INSERT ... ON CONFLICT (pk) DO UPDATE`. Behaves like
/// `Insert` when `on_conflict_updates` is empty (overwrite); when
/// non-empty, the handler reads the existing row, applies the
/// assignments (with `EXCLUDED.col` bound to the incoming row), and
/// writes the merged result.
Put,
}
/// Base columnar physical operations.
#[derive(
Debug,
Clone,
PartialEq,
serde::Serialize,
serde::Deserialize,
zerompk::ToMessagePack,
zerompk::FromMessagePack,
)]
pub enum ColumnarOp {
/// Read rows from columnar memtable + segments.
///
/// Applies filters, projects columns, respects limit.
/// No time-range semantics — that's `TimeseriesOp::Scan`.
Scan {
collection: String,
projection: Vec<String>,
limit: usize,
filters: Vec<u8>,
rls_filters: Vec<u8>,
/// ORDER BY clause: `(field_name, ascending)` pairs, applied
/// against matching rows before the `limit` is enforced. Empty
/// for scans with no ORDER BY.
sort_keys: Vec<(String, bool)>,
/// System-time selection. `Current` = current-state read; `AsOf(ms)` =
/// rows whose `_ts_system` is ≤ ms; `AllVersions` = every `_ts_system`
/// row ordered ascending (audit log), system-time column projected.
/// Only meaningful for collections created `WITH BITEMPORAL`.
#[serde(default)]
system_time: SystemTimeScope,
/// Bitemporal valid-time predicate: keep rows whose
/// `[_ts_valid_from, _ts_valid_until)` interval contains this
/// point. `None` = no valid-time filter.
#[serde(default)]
valid_at_ms: Option<i64>,
/// Optional surrogate prefilter injected by a cross-engine sub-plan.
/// When present, the scan skips rows whose surrogate is absent from
/// this bitmap. `None` = no prefilter; full collection is scanned.
#[serde(default)]
prefilter: Option<SurrogateBitmap>,
/// MessagePack-serialized `Vec<ComputedColumn>` for scalar projection
/// expressions (e.g. JSON arrow operators). Empty when no computed
/// columns are present in the query.
#[serde(default)]
computed_columns: Vec<u8>,
},
/// Insert rows into a columnar memtable.
///
/// Accepts JSON or MessagePack payload. The memtable is created on
/// first insert with schema inferred from the payload.
///
/// `intent` distinguishes plain `INSERT` (upsert-semantics on columnar:
/// duplicate PK tombstones the prior row and appends the new one) from
/// `ON CONFLICT DO NOTHING` (`InsertIfAbsent` — silent skip on dup) and
/// `UPSERT` / `ON CONFLICT (pk) DO UPDATE` (`Put` — optionally with
/// per-row merges in `on_conflict_updates`).
Insert {
collection: String,
/// Row data. Format determined by `format` field.
payload: Vec<u8>,
/// "json" for JSON array of objects, "msgpack" for MessagePack,
/// "ilp" for InfluxDB Line Protocol (delegated to timeseries path).
format: String,
/// INSERT / INSERT IF ABSENT / UPSERT distinction. See
/// `ColumnarInsertIntent` for semantics per variant.
intent: ColumnarInsertIntent,
/// `ON CONFLICT (pk) DO UPDATE SET field = expr` assignments.
/// Carried only when `intent == Put`; empty for `Insert`,
/// `InsertIfAbsent`, and plain `UPSERT` (whole-row overwrite).
/// Each `UpdateValue` is either a literal msgpack bytes payload
/// or a `SqlExpr` the handler evaluates against the existing row
/// plus the would-be-inserted row (with `EXCLUDED.col` resolution).
on_conflict_updates: Vec<(String, super::document::UpdateValue)>,
/// Per-row stable cross-engine identities, parallel to the rows
/// in `payload`. CP-side assigner populates this in row order
/// before dispatch. `vec![]` only in test fixtures (and length
/// must equal the row count when populated).
surrogates: Vec<Surrogate>,
/// MessagePack-serialized `ColumnarSchema` from the DDL catalog.
/// When non-empty, the Data Plane uses this schema to initialize the
/// memtable engine instead of inferring the schema from the payload.
/// This is required for columns whose SQL type is ambiguous at the
/// Value level (e.g. JSON arrives as `Value::String` but must be
/// stored in `ColumnData::Json`, not `ColumnData::String`).
/// Empty `vec![]` only in legacy test fixtures that do not carry schema.
#[serde(default)]
schema_bytes: Vec<u8>,
/// Sync provenance: identifies the originating peer and sequence for idempotency.
#[serde(default)]
provenance: Option<nodedb_types::sync::wire::SyncProvenance>,
/// WAL record LSN for deduplication. Set by the WAL catch-up task so the
/// Data Plane can skip records that have already been ingested or flushed
/// to disk. `None` for live ingest (always accepted).
#[serde(default)]
wal_lsn: Option<u64>,
},
/// Update rows matching filter predicates.
///
/// Uses `MutationEngine` for plain/spatial profiles.
/// `updates` is a list of (field_name, json_value_bytes) pairs.
Update {
collection: String,
/// Serialized `Vec<ScanFilter>` (MessagePack).
filters: Vec<u8>,
/// Field assignments: `(column_name, json_value_bytes)`.
updates: Vec<(String, Vec<u8>)>,
},
/// Delete rows matching filter predicates.
///
/// Uses `MutationEngine` for plain/spatial profiles.
Delete {
collection: String,
/// Serialized `Vec<ScanFilter>` (MessagePack).
filters: Vec<u8>,
},
/// Cursor-paginated raw scan for the clone materializer.
///
/// Returns `(surrogate, row_value_bytes)` pairs plus next-cursor in one
/// payload. Honors `system_as_of_ms` so the materializer reads source
/// as-of the clone's `as_of_lsn` for bitemporal collections.
MaterializeScan {
collection: String,
cursor: Vec<u8>,
count: usize,
system_as_of_ms: Option<i64>,
},
}