linkrs_query/planning/statements/pattern_planner/
path.rs1use std::sync::Arc;
4
5use crate::metadata::MetadataContext;
6use crate::parser::ast::pattern::{PathElement, Pattern, RepetitionType};
7use crate::planning::plan::core::nodes::base::plan_node_traits::PlanNode;
8use crate::planning::plan::core::nodes::LoopNode;
9use crate::planning::plan::logical::logical_nodes::control_flow::LogicalLoopNode;
10use crate::planning::plan::logical::LogicalNodeEnum;
11use crate::planning::plan::SubPlan;
12use crate::planning::planner::PlannerError;
13use crate::planning::statements::plan_combiner;
14use linkrs_core::types::expr::expression_context::ExpressionAnalysisContext;
15
16use super::edge::{plan_pattern_edge, plan_pattern_edge_with_input};
17use super::node::plan_pattern_node;
18use super::{plan_pattern, PlanningContext};
19
20pub fn plan_path_pattern(
21 pattern: &Pattern,
22 ctx: &PlanningContext,
23) -> Result<SubPlan, PlannerError> {
24 match pattern {
25 Pattern::Path(path) => {
26 if path.elements.is_empty() {
27 return Err(PlannerError::PlanGenerationFailed(
28 "empty path model".to_string(),
29 ));
30 }
31
32 let mut plan = SubPlan::new(None, None);
33 let mut prev_node_alias: Option<String> = None;
34 let mut is_first_node = true;
35 let mut is_first_edge = true;
36
37 let elements: Vec<_> = path.elements.iter().collect();
38 let mut i = 0;
39
40 while i < elements.len() {
41 match elements[i] {
42 PathElement::Node(node) => {
43 if is_first_node {
44 let node_plan = plan_pattern_node(
45 node,
46 ctx.space_id,
47 ctx.space_name,
48 ctx.enable_index_optimization,
49 ctx.metadata_context,
50 ctx.expr_context,
51 ctx.where_expression,
52 )?;
53 plan = if let Some(existing_root) = plan.root.take() {
54 plan_combiner::cross_join_plans(
55 SubPlan {
56 root: Some(existing_root),
57 tail: plan.tail,
58 logical_root: plan.logical_root.take(),
59 },
60 node_plan,
61 )?
62 } else {
63 node_plan
64 };
65 let node_alias =
66 node.variable.clone().unwrap_or_else(|| "n".to_string());
67 prev_node_alias = Some(node_alias);
68 is_first_node = false;
69 } else {
70 let node_alias =
71 node.variable.clone().unwrap_or_else(|| "n".to_string());
72 prev_node_alias = Some(node_alias);
73 }
74 i += 1;
75 }
76 PathElement::Edge(edge) => {
77 if prev_node_alias.is_none() {
78 return Err(PlannerError::PlanGenerationFailed(
79 "The edge pattern must follow the node pattern".to_string(),
80 ));
81 }
82
83 let input_alias = prev_node_alias.as_deref().unwrap();
84
85 let (dst_var, dst_labels) = if i + 1 < elements.len() {
86 if let PathElement::Node(next_node) = elements[i + 1] {
87 (next_node.variable.as_deref(), next_node.labels.clone())
88 } else {
89 (None, Vec::new())
90 }
91 } else {
92 (None, Vec::new())
93 };
94
95 let edge_plan = plan_pattern_edge_with_input(
96 edge,
97 ctx.space_id,
98 input_alias,
99 dst_var,
100 &dst_labels,
101 ctx.expr_context,
102 ctx.optional,
103 )?;
104
105 plan = if let Some(existing_root) = plan.root.take() {
106 if is_first_edge {
107 plan_combiner::connect_node_to_edge_expansion(
108 SubPlan {
109 root: Some(existing_root),
110 tail: plan.tail,
111 logical_root: plan.logical_root.take(),
112 },
113 edge_plan,
114 input_alias,
115 )?
116 } else {
117 plan_combiner::join_edge_expansions(
118 SubPlan {
119 root: Some(existing_root),
120 tail: plan.tail,
121 logical_root: plan.logical_root.take(),
122 },
123 edge_plan,
124 input_alias,
125 )?
126 }
127 } else {
128 edge_plan
129 };
130
131 is_first_edge = false;
132
133 prev_node_alias = dst_var.map(|s| s.to_string());
134 i += 1;
135 }
136 PathElement::Alternative(patterns) => {
137 let alt_plan = plan_alternative_patterns(patterns, ctx)?;
138 plan = if let Some(existing_root) = plan.root.take() {
139 plan_combiner::cross_join_plans(
140 SubPlan {
141 root: Some(existing_root),
142 tail: plan.tail,
143 logical_root: plan.logical_root.take(),
144 },
145 alt_plan,
146 )?
147 } else {
148 alt_plan
149 };
150 i += 1;
151 }
152 PathElement::Optional(elem) => {
153 let opt_plan = plan_optional_element(elem, ctx)?;
154 plan = if let Some(existing_root) = plan.root.take() {
155 plan_combiner::left_join_plans(
156 SubPlan {
157 root: Some(existing_root),
158 tail: plan.tail,
159 logical_root: plan.logical_root.take(),
160 },
161 opt_plan,
162 )?
163 } else {
164 opt_plan
165 };
166 i += 1;
167 }
168 PathElement::Repeated(elem, rep_type) => {
169 let rep_plan = plan_repeated_element(
170 elem,
171 *rep_type,
172 ctx.space_id,
173 ctx.space_name,
174 ctx.expr_context,
175 ctx.enable_index_optimization,
176 ctx.metadata_context,
177 )?;
178 plan = if let Some(existing_root) = plan.root.take() {
179 plan_combiner::cross_join_plans(
180 SubPlan {
181 root: Some(existing_root),
182 tail: plan.tail,
183 logical_root: plan.logical_root.take(),
184 },
185 rep_plan,
186 )?
187 } else {
188 rep_plan
189 };
190 i += 1;
191 }
192 PathElement::Recursive(_) => {
193 return Err(PlannerError::UnsupportedOperation(
194 "Recursive comprehension with variable binding and projection is rejected; rewrite with plain variable-length traversal".to_string(),
195 ));
196 }
197 PathElement::CreateEdge(_) => {
198 return Err(PlannerError::UnsupportedOperation(
199 "CREATE edge patterns have no traversal plan".to_string(),
200 ));
201 }
202 }
203 }
204
205 Ok(plan)
206 }
207 _ => plan_pattern(pattern, ctx),
208 }
209}
210pub fn plan_alternative_patterns(
211 patterns: &[Pattern],
212 ctx: &PlanningContext,
213) -> Result<SubPlan, PlannerError> {
214 if patterns.is_empty() {
215 return Err(PlannerError::PlanGenerationFailed(
216 "The alternative path cannot be empty".to_string(),
217 ));
218 }
219
220 let mut plan = plan_pattern(&patterns[0], ctx)?;
221
222 for pattern in patterns.iter().skip(1) {
223 let pattern_plan = plan_pattern(pattern, ctx)?;
224 plan = plan_combiner::union_plans(plan, pattern_plan)?;
225 }
226
227 Ok(plan)
228}
229
230pub fn plan_optional_element(
231 element: &PathElement,
232 ctx: &PlanningContext,
233) -> Result<SubPlan, PlannerError> {
234 let opt_plan = match element {
235 PathElement::Node(node) => plan_pattern_node(
236 node,
237 ctx.space_id,
238 ctx.space_name,
239 ctx.enable_index_optimization,
240 ctx.metadata_context,
241 ctx.expr_context,
242 ctx.where_expression,
243 )?,
244 PathElement::Edge(edge) => {
245 plan_pattern_edge(edge, ctx.space_id, ctx.space_name, ctx.expr_context, false)?
246 }
247 _ => {
248 return Err(PlannerError::PlanGenerationFailed(
249 "Optional paths do not support nested complex patterns".to_string(),
250 ));
251 }
252 };
253
254 Ok(opt_plan)
255}
256
257pub fn plan_repeated_element(
258 element: &PathElement,
259 rep_type: RepetitionType,
260 space_id: u64,
261 space_name: &str,
262 expr_context: &Option<Arc<ExpressionAnalysisContext>>,
263 enable_index_optimization: bool,
264 metadata_context: &Option<MetadataContext>,
265) -> Result<SubPlan, PlannerError> {
266 let base_plan = match element {
267 PathElement::Node(node) => plan_pattern_node(
268 node,
269 space_id,
270 space_name,
271 enable_index_optimization,
272 metadata_context,
273 expr_context,
274 None,
275 )?,
276 PathElement::Edge(edge) => {
277 plan_pattern_edge(edge, space_id, space_name, expr_context, false)?
278 }
279 _ => {
280 return Err(PlannerError::PlanGenerationFailed(
281 "Repeated paths do not support nested complex patterns".to_string(),
282 ));
283 }
284 };
285
286 let condition_str = match rep_type {
287 RepetitionType::ZeroOrMore => "loop_count >= 0".to_string(),
288 RepetitionType::OneOrMore => "loop_count >= 1".to_string(),
289 RepetitionType::ZeroOrOne => "loop_count <= 1".to_string(),
290 RepetitionType::Exactly(n) => format!("loop_count == {}", n),
291 RepetitionType::Range(min, max) => {
292 format!("loop_count >= {} && loop_count <= {}", min, max)
293 }
294 };
295
296 let expr_ctx = expr_context.as_ref().ok_or_else(|| {
297 PlannerError::PlanGenerationFailed("Expression context is unavailable".to_string())
298 })?;
299 let expr_meta = linkrs_core::types::expr::ExpressionMeta::new(
300 linkrs_core::Expression::Variable(condition_str),
301 );
302 let id = expr_ctx.register_expression(expr_meta);
303 let ctx_expr = linkrs_core::types::ContextualExpression::new(id, expr_ctx.clone());
304
305 let mut loop_node = LoopNode::new(-1, ctx_expr.clone());
306
307 if let Some(base_root) = &base_plan.root {
308 loop_node.set_body(base_root.clone());
309 }
310
311 let logical_root = base_plan
313 .logical_root()
314 .cloned()
315 .map(|body| LogicalNodeEnum::Loop(LogicalLoopNode::new_with_body(ctx_expr, body)));
316
317 Ok(SubPlan {
318 root: Some(loop_node.into_enum()),
319 tail: base_plan.tail,
320 logical_root,
321 })
322}