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
//! Spatial-join executor — handles `Clause::SpatialJoin` produced by
//! `fuse_spatial_join` in the planner.
//!
//! Iterates probe-side nodes once, probes an R-tree built over container
//! bboxes, and emits only matching pairs. Replaces the 2-pattern cartesian
//! + WHERE contains() pipeline with O((N+M) log N + K) work.
use super::super::ast::{Predicate, SpatialProbeKind};
use super::*;
use petgraph::graph::NodeIndex;
use rstar::{RTree, RTreeObject, AABB};
use std::sync::Arc;
/// An R-tree entry: a container node's bbox tagged with its NodeIndex and
/// parsed geometry (Arc-shared with the per-query spatial cache).
struct IndexedContainer {
node_idx: NodeIndex,
geom: Arc<geo::Geometry<f64>>,
min_x: f64,
min_y: f64,
max_x: f64,
max_y: f64,
}
impl RTreeObject for IndexedContainer {
type Envelope = AABB<[f64; 2]>;
fn envelope(&self) -> Self::Envelope {
AABB::from_corners([self.min_x, self.min_y], [self.max_x, self.max_y])
}
}
impl<'a> CypherExecutor<'a> {
/// Execute a fused spatial-join clause. Produces one row per (container,
/// probe) pair where the probe's point lies inside the container's
/// geometry.
pub(super) fn execute_spatial_join(
&self,
container_var: &str,
probe_var: &str,
container_type: &str,
probe_type: &str,
probe_kind: SpatialProbeKind,
remainder: Option<&Predicate>,
) -> Result<ResultSet, String> {
// Fall back to a regular two-pattern MATCH + WHERE if either type has
// no nodes. This preserves correctness on empty subgraphs.
let container_indices = match self.graph.type_indices.get(container_type) {
Some(v) if !v.is_empty() => v,
_ => return Ok(ResultSet::new()),
};
let probe_indices = match self.graph.type_indices.get(probe_type) {
Some(v) if !v.is_empty() => v,
_ => return Ok(ResultSet::new()),
};
// Build the R-tree entries. Skip containers without parsed geometry
// or without a computable bbox.
let mut entries: Vec<IndexedContainer> = Vec::with_capacity(container_indices.len());
for idx in container_indices.iter() {
self.ensure_node_spatial_cached(idx);
let cache = self.spatial_node_cache.read().unwrap();
if let Some(Some(data)) = cache.get(&idx.index()) {
if let Some((geom, Some(bbox))) = &data.geometry {
entries.push(IndexedContainer {
node_idx: idx,
geom: Arc::clone(geom),
min_x: bbox.min().x,
min_y: bbox.min().y,
max_x: bbox.max().x,
max_y: bbox.max().y,
});
}
}
}
if entries.is_empty() {
return Ok(ResultSet::new());
}
let tree = RTree::<IndexedContainer>::bulk_load(entries);
let mut rows: Vec<ResultRow> = Vec::new();
for (probe_i, probe_idx) in probe_indices.iter().enumerate() {
self.ensure_node_spatial_cached(probe_idx);
let probe_point: Option<(f64, f64)> = {
let cache = self.spatial_node_cache.read().unwrap();
match cache.get(&probe_idx.index()) {
Some(Some(data)) => match probe_kind {
SpatialProbeKind::Location => data.location,
// Probe via the geometry centroid. Drops probes
// whose geometry is missing or whose centroid
// computation fails (degenerate WKT).
SpatialProbeKind::Centroid => {
data.geometry.as_ref().and_then(|(geom, _bbox)| {
crate::graph::features::spatial::geometry_centroid(geom).ok()
})
}
},
_ => None,
}
};
let (lat, lon) = match probe_point {
Some(pt) => pt,
None => continue,
};
// R-tree bbox probe: only containers whose bbox contains this point.
let env = AABB::from_point([lon, lat]);
for cand in tree.locate_in_envelope_intersecting(&env) {
let pt = geo::Point::new(lon, lat);
if crate::graph::features::spatial::geometry_contains_point(&cand.geom, &pt) {
let mut row = ResultRow::with_capacity(2, 0, 0);
row.node_bindings
.insert(container_var.to_string(), cand.node_idx);
row.node_bindings.insert(probe_var.to_string(), probe_idx);
rows.push(row);
}
}
if probe_i & 2047 == 0 {
self.check_deadline()?;
}
}
// Apply the residual WHERE predicate (e.g. the AND-remainder after
// `contains()` was stripped during planning).
if let Some(pred) = remainder {
let mut keep = Vec::with_capacity(rows.len());
for row in rows {
match self.evaluate_predicate(pred, &row) {
Ok(true) => keep.push(row),
Ok(false) => {}
Err(e) => return Err(e),
}
}
rows = keep;
}
Ok(ResultSet {
rows,
columns: Vec::new(),
lazy_return_items: None,
})
}
}