linkrs_query/executor/streaming/operators/spec/source.rs
1//! Immutable configuration for source operators.
2
3use std::sync::Arc;
4
5use crate::executor::streaming::slot::SlotLayout;
6use crate::storage::{ScanPredicate, ShardSlice};
7use linkrs_core::types::expr::ContextualExpression;
8use linkrs_core::Value;
9
10/// A predicate that has been validated and bound to an index schema.
11///
12/// Created at plan-build time from the logical plan's `IndexLimit` / filter.
13#[derive(Debug, Clone)]
14pub enum BoundIndexPredicate {
15 /// Exact equality: `column = value`
16 Equal { column: String, value: Value },
17 /// Range scan: `column BETWEEN begin AND end`
18 Range {
19 column: String,
20 begin: Option<Value>,
21 end: Option<Value>,
22 include_begin: bool,
23 include_end: bool,
24 },
25 /// Prefix scan (for string columns): `column STARTS WITH prefix`
26 Prefix { column: String, prefix: Value },
27 /// Full scan (no predicate, all entries)
28 Full,
29}
30
31/// Describes which columns the index should return.
32///
33/// A covering index can satisfy the projection directly; a non-covering
34/// index requires back-to-table fetches.
35#[derive(Debug, Clone)]
36pub enum IndexProjection {
37 /// Return only the row ID (back-to-table required).
38 RowIdOnly,
39 /// Return specific columns from the index (covering if the index
40 /// includes all of them).
41 Columns(Vec<String>),
42 /// Return all indexed columns.
43 AllColumns,
44}
45
46/// Immutable config for source operators.
47///
48/// Mutable state (`cursor`, `buffer`, `current_index`, `partition_id`,
49/// `partition_range`) lives in [`SourceState`](super::super::state::SourceState).
50#[derive(Debug, Clone)]
51pub enum SourceSpec {
52 ScanVertices {
53 rows: Vec<Vec<Value>>,
54 col_names: Vec<String>,
55 },
56 /// Recursive-CTE working-table scan.
57 ///
58 /// Reads the rows published by the enclosing fixpoint operator under the
59 /// mangled `cte_name` (see `crate::cte`). Produced when a step pattern
60 /// labels a node with the CTE name; never touches storage.
61 CteScan {
62 cte_name: String,
63 col_names: Vec<String>,
64 },
65 /// Standalone DML values — evaluated once per execution in the source
66 /// operator so that volatile expressions (e.g. `now()`) are resolved at
67 /// execution time, not at plan build time.
68 StandaloneValues {
69 values: Vec<Vec<ContextualExpression>>,
70 col_names: Vec<String>,
71 },
72 StorageScanVertices {
73 space_name: String,
74 limit: Option<usize>,
75 col_names: Vec<String>,
76 projected_properties: Vec<String>,
77 /// Identity mode: the scan variable is never consumed as a whole
78 /// entity downstream, so the entity column carries lightweight
79 /// `Value::VertexId` instead of boxed vertices.
80 identity_only: bool,
81 /// Rowless mode (set with `identity_only` by the planner when the
82 /// direct consumer is column-capable): skip the row view entirely
83 /// and emit typed columns only. Valid only when the columnar path
84 /// is on; the scan falls back to rows when it is off.
85 skip_rows: bool,
86 /// Scan predicates pushed into the storage layer (pure pre-filter;
87 /// the original filter still runs on top).
88 predicate: Vec<ScanPredicate>,
89 /// Tag-restricted scan: only rows of this tag are scanned at the
90 /// storage layer (the matching `contains(labels(v), ...)` residual
91 /// conjunct may then be elided).
92 tag: Option<String>,
93 /// Optional vertex-id range restricting the scan to one partition.
94 /// Set only on per-partition copies produced by the partitioned
95 /// physical-plan builder; `None` means a full scan.
96 partition_range: Option<std::ops::Range<i64>>,
97 /// Optional shard slice dividing a partitioned scan's enumeration
98 /// across partitions. Set together with a cleared `partition_range`
99 /// on per-partition copies whose union must cover the table without
100 /// repeating full-table walks; `None` enumerates the whole table.
101 shard_slice: Option<ShardSlice>,
102 },
103 ScanEdges {
104 rows: Vec<Vec<Value>>,
105 col_names: Vec<String>,
106 },
107 StorageScanEdges {
108 space_name: String,
109 limit: Option<usize>,
110 edge_type: Option<String>,
111 col_names: Vec<String>,
112 projected_properties: Vec<String>,
113 /// Identity mode: the scan variable is never consumed as a whole
114 /// entity downstream, so the entity column carries lightweight
115 /// `Value::EdgeHeader` instead of boxed edges.
116 identity_only: bool,
117 /// Scan predicates pushed into the storage layer (pure pre-filter;
118 /// the original filter still runs on top).
119 predicate: Vec<ScanPredicate>,
120 /// Optional shard slice dividing a partitioned edge scan's source
121 /// enumeration across partitions. Set on per-partition copies whose
122 /// union must cover every edge without repeating full-table walks;
123 /// `None` walks the whole source space.
124 shard_slice: Option<ShardSlice>,
125 },
126 GetVertices {
127 space_name: String,
128 tag: String,
129 vertex_ids: Option<Vec<Value>>,
130 projected_properties: Vec<String>,
131 /// Entity variable name for the fetched vertex (e.g. the tag name).
132 col_names: Vec<String>,
133 },
134 GetEdges {
135 space_name: String,
136 edge_type: Option<String>,
137 src: Option<String>,
138 dst: Option<String>,
139 rank: i64,
140 /// Property names to keep on the materialized edge; empty reads all.
141 projected_properties: Vec<String>,
142 },
143 GetNeighbors {
144 space_name: String,
145 tag: String,
146 direction: String,
147 projected_properties: Vec<String>,
148 },
149 /// Index scan with typed predicate and projection.
150 ///
151 /// The predicate and projection are validated at build time against
152 /// the index schema. Stale row IDs are skipped, not treated as EOF.
153 IndexScan {
154 space_name: String,
155 index_name: String,
156 index_id: u64,
157 /// Tag owning the indexed rows; covering vertex rows are rebuilt with it.
158 tag: String,
159 predicate: Box<BoundIndexPredicate>,
160 projection: IndexProjection,
161 residual_filter: Option<linkrs_core::types::expr::Expression>,
162 output_layout: Arc<SlotLayout>,
163 },
164 /// Produces one singleton row from which correlated apply can pull.
165 /// `col_names` mirrors the outer (left) layout so the emitted chunk slots
166 /// align with the correlation frame read at runtime.
167 Argument { col_names: Vec<String> },
168 /// Unary property retrieval: reads properties from the input entity IDs.
169 ///
170 /// Unlike the source-variant GetProp (which is a zero-input stub),
171 /// this unary variant takes an input child and produces one output
172 /// row per input row with additional property columns.
173 GetProp {
174 space_name: String,
175 /// Slot number of the entity (vertex or edge) column in the input chunk.
176 entity_slot: usize,
177 /// Property names to read.
178 prop_names: Vec<String>,
179 /// Whether the entity is a vertex or edge.
180 is_vertex: bool,
181 /// Output layout (input columns + new property columns).
182 output_layout: Arc<SlotLayout>,
183 },
184 /// Produces one zero-column row on first `next()`, then `None`.
185 /// Used as the seed for command-type operators (DDL, DML, etc.)
186 /// that require a source but have no real input.
187 Start,
188}