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
//! Immutable configuration for source operators.
use std::sync::Arc;
use crate::executor::streaming::slot::SlotLayout;
use crate::storage::{ScanPredicate, ShardSlice};
use linkrs_core::types::expr::ContextualExpression;
use linkrs_core::Value;
/// A predicate that has been validated and bound to an index schema.
///
/// Created at plan-build time from the logical plan's `IndexLimit` / filter.
#[derive(Debug, Clone)]
pub enum BoundIndexPredicate {
/// Exact equality: `column = value`
Equal { column: String, value: Value },
/// Range scan: `column BETWEEN begin AND end`
Range {
column: String,
begin: Option<Value>,
end: Option<Value>,
include_begin: bool,
include_end: bool,
},
/// Prefix scan (for string columns): `column STARTS WITH prefix`
Prefix { column: String, prefix: Value },
/// Full scan (no predicate, all entries)
Full,
}
/// Describes which columns the index should return.
///
/// A covering index can satisfy the projection directly; a non-covering
/// index requires back-to-table fetches.
#[derive(Debug, Clone)]
pub enum IndexProjection {
/// Return only the row ID (back-to-table required).
RowIdOnly,
/// Return specific columns from the index (covering if the index
/// includes all of them).
Columns(Vec<String>),
/// Return all indexed columns.
AllColumns,
}
/// Immutable config for source operators.
///
/// Mutable state (`cursor`, `buffer`, `current_index`, `partition_id`,
/// `partition_range`) lives in [`SourceState`](super::super::state::SourceState).
#[derive(Debug, Clone)]
pub enum SourceSpec {
ScanVertices {
rows: Vec<Vec<Value>>,
col_names: Vec<String>,
},
/// Recursive-CTE working-table scan.
///
/// Reads the rows published by the enclosing fixpoint operator under the
/// mangled `cte_name` (see `crate::cte`). Produced when a step pattern
/// labels a node with the CTE name; never touches storage.
CteScan {
cte_name: String,
col_names: Vec<String>,
},
/// Standalone DML values — evaluated once per execution in the source
/// operator so that volatile expressions (e.g. `now()`) are resolved at
/// execution time, not at plan build time.
StandaloneValues {
values: Vec<Vec<ContextualExpression>>,
col_names: Vec<String>,
},
StorageScanVertices {
space_name: String,
limit: Option<usize>,
col_names: Vec<String>,
projected_properties: Vec<String>,
/// Identity mode: the scan variable is never consumed as a whole
/// entity downstream, so the entity column carries lightweight
/// `Value::VertexId` instead of boxed vertices.
identity_only: bool,
/// Rowless mode (set with `identity_only` by the planner when the
/// direct consumer is column-capable): skip the row view entirely
/// and emit typed columns only. Valid only when the columnar path
/// is on; the scan falls back to rows when it is off.
skip_rows: bool,
/// Scan predicates pushed into the storage layer (pure pre-filter;
/// the original filter still runs on top).
predicate: Vec<ScanPredicate>,
/// Tag-restricted scan: only rows of this tag are scanned at the
/// storage layer (the matching `contains(labels(v), ...)` residual
/// conjunct may then be elided).
tag: Option<String>,
/// Optional vertex-id range restricting the scan to one partition.
/// Set only on per-partition copies produced by the partitioned
/// physical-plan builder; `None` means a full scan.
partition_range: Option<std::ops::Range<i64>>,
/// Optional shard slice dividing a partitioned scan's enumeration
/// across partitions. Set together with a cleared `partition_range`
/// on per-partition copies whose union must cover the table without
/// repeating full-table walks; `None` enumerates the whole table.
shard_slice: Option<ShardSlice>,
},
ScanEdges {
rows: Vec<Vec<Value>>,
col_names: Vec<String>,
},
StorageScanEdges {
space_name: String,
limit: Option<usize>,
edge_type: Option<String>,
col_names: Vec<String>,
projected_properties: Vec<String>,
/// Identity mode: the scan variable is never consumed as a whole
/// entity downstream, so the entity column carries lightweight
/// `Value::EdgeHeader` instead of boxed edges.
identity_only: bool,
/// Scan predicates pushed into the storage layer (pure pre-filter;
/// the original filter still runs on top).
predicate: Vec<ScanPredicate>,
/// Optional shard slice dividing a partitioned edge scan's source
/// enumeration across partitions. Set on per-partition copies whose
/// union must cover every edge without repeating full-table walks;
/// `None` walks the whole source space.
shard_slice: Option<ShardSlice>,
},
GetVertices {
space_name: String,
tag: String,
vertex_ids: Option<Vec<Value>>,
projected_properties: Vec<String>,
/// Entity variable name for the fetched vertex (e.g. the tag name).
col_names: Vec<String>,
},
GetEdges {
space_name: String,
edge_type: Option<String>,
src: Option<String>,
dst: Option<String>,
rank: i64,
/// Property names to keep on the materialized edge; empty reads all.
projected_properties: Vec<String>,
},
GetNeighbors {
space_name: String,
tag: String,
direction: String,
projected_properties: Vec<String>,
},
/// Index scan with typed predicate and projection.
///
/// The predicate and projection are validated at build time against
/// the index schema. Stale row IDs are skipped, not treated as EOF.
IndexScan {
space_name: String,
index_name: String,
index_id: u64,
/// Tag owning the indexed rows; covering vertex rows are rebuilt with it.
tag: String,
predicate: Box<BoundIndexPredicate>,
projection: IndexProjection,
residual_filter: Option<linkrs_core::types::expr::Expression>,
output_layout: Arc<SlotLayout>,
},
/// Produces one singleton row from which correlated apply can pull.
/// `col_names` mirrors the outer (left) layout so the emitted chunk slots
/// align with the correlation frame read at runtime.
Argument { col_names: Vec<String> },
/// Unary property retrieval: reads properties from the input entity IDs.
///
/// Unlike the source-variant GetProp (which is a zero-input stub),
/// this unary variant takes an input child and produces one output
/// row per input row with additional property columns.
GetProp {
space_name: String,
/// Slot number of the entity (vertex or edge) column in the input chunk.
entity_slot: usize,
/// Property names to read.
prop_names: Vec<String>,
/// Whether the entity is a vertex or edge.
is_vertex: bool,
/// Output layout (input columns + new property columns).
output_layout: Arc<SlotLayout>,
},
/// Produces one zero-column row on first `next()`, then `None`.
/// Used as the seed for command-type operators (DDL, DML, etc.)
/// that require a source but have no real input.
Start,
}