Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Query Entities over POST (5.7.2, resource 6.23): the Query body read as
3 : //! the GET twin reads its URI, then paged, shaped and joined.
4 :
5 : use super::*;
6 :
7 : // ---------- POST /entityOperations/query (6.23) ----------
8 :
9 440 : pub async fn batch_query(
10 440 : State(st): State<AppState>,
11 440 : CleanParams(params): CleanParams,
12 440 : headers: HeaderMap,
13 440 : body: Bytes,
14 440 : ) -> Response {
15 440 : match batch_query_inner(&st, ¶ms, &headers, &body).await {
16 204 : Ok(r) => r,
17 236 : Err(e) => e.into_response(),
18 : }
19 440 : }
20 :
21 440 : async fn batch_query_inner(
22 440 : st: &AppState,
23 440 : params: &HashMap<String, String>,
24 440 : headers: &HeaderMap,
25 440 : body: &[u8],
26 440 : ) -> ApiResult<Response> {
27 440 : let tenant = tenant_from(headers)?;
28 440 : check_params(
29 440 : params,
30 440 : &["limit", "offset", "count", "options", "format", "local"],
31 0 : )?;
32 : // POST query IS Query Entities: geo+json is a valid Accept here (6.3.15)
33 440 : let accept = parse_accept_geo(headers)?;
34 440 : let parsed = parse_body(st, headers, body, BodyKind::Standard).await?;
35 440 : let q = parsed.object(NgsiError::BadRequestData(
36 440 : "query body must be an object".into(),
37 440 : ))?;
38 440 : if q.get("type").and_then(Value::as_str) != Some("Query") {
39 8 : return Err(NgsiError::BadRequestData("body type must be Query (5.2.23)".into()).into());
40 432 : }
41 : // Convert Query members into virtual params reusing the GET filter path.
42 432 : let mut vp: HashMap<String, String> = HashMap::new();
43 432 : query_doc_params(q, false, &mut vp)?;
44 248 : if let Some(l) = params.get("local") {
45 0 : vp.insert("local".into(), l.clone());
46 248 : }
47 : // ADR-0020: the engine is shown every member of the Query body
48 248 : let named = crate::entities::ReadMembers::with_ctx(st, &vp, &parsed.ctx);
49 248 : let filter = gate!(
50 : st, &tenant, headers, "5.7.2",
51 : q: named.q.as_ref(),
52 : scope_q: vp.get("scopeQ").map(String::as_str),
53 : types: &named.types,
54 : attrs: &named.attrs,
55 : geo: named.geo.as_ref(),
56 : )
57 248 : .await?;
58 246 : refuse_too_wide(&vp)?;
59 : // ADR-0020: the engine's narrowing joins the Query's own filters here,
60 : // after the "too wide" judgement — which is about what the client asked
61 : // for. Everything below reads the narrowed query, the forwarded one
62 : // included.
63 : // what the client's own Query reads, before the engine's q joins it
64 : //
65 226 : let narrow = crate::repr::narrowing_reading(&filter, &vp, &parsed.ctx)?;
66 226 : let vp = filter.narrow_params(&vp, &parsed.ctx)?;
67 226 : let fed = fed_matches(st, &tenant, headers, &parsed.ctx, &vp).await?;
68 222 : let mut matches =
69 226 : crate::entities::filter_entities_fed(st, &tenant, &vp, &parsed.ctx, fed, Some(&narrow))
70 226 : .await?;
71 222 : let mut page_params = params.clone();
72 222 : page_params.extend(vp.clone());
73 : // 5.2.43 ordering: same 4.23 keys as the GET twin, applied pre-pagination
74 222 : if let Some(spec) = page_params.get("orderBy") {
75 34 : crate::repr::order_as_shown(&mut matches, Some(&narrow), |m| {
76 34 : crate::paging::order_entities(m, spec, &page_params, &parsed.ctx)
77 34 : })?;
78 188 : }
79 210 : let (page, count_hdr, _links) = paginate(
80 210 : st,
81 210 : &page_params,
82 210 : matches,
83 210 : "/ngsi-ld/v1/entityOperations/query",
84 0 : )?;
85 : // body members (pick/omit/attrs/lang/datasetId) shape the representation
86 : // exactly like their 6.3.7 query-parameter twins
87 210 : let mut repr = parse_repr(&page_params, &parsed.ctx)?;
88 206 : crate::repr::narrow_repr(&mut repr, &filter);
89 206 : let join = crate::entities::parse_join(&vp)?;
90 204 : crate::entities::check_linked_projection(&repr, &join)?;
91 204 : let (rendered_from, mut payload): (Vec<&Value>, Vec<Value>) =
92 204 : shaped_page(&page, &repr, &parsed.ctx).into_iter().unzip();
93 204 : if let Some((mode, level)) = &join {
94 : // 4.5.23.1 bounds the WIDTH of the retrieval per REQUEST, so one
95 : // allowance is spent across the whole page, exactly as the GET twin
96 : // spends it. Minting a fresh allowance per payload Entity multiplies
97 : // the ceiling by the page size, which is the request the ceiling
98 : // exists to bound: a page of densely linked Entities.
99 18 : let scope = JoinScope {
100 18 : st,
101 18 : tenant: &tenant,
102 18 : ctx: &parsed.ctx,
103 18 : repr: &repr,
104 18 : level: *level,
105 18 : held: crate::entities::contained_by(&page_params),
106 18 : };
107 18 : match mode.as_str() {
108 18 : "inline" => {
109 8 : let remote = crate::federation::active(&page_params).then_some(headers);
110 8 : scope.inline(&mut payload, &rendered_from, remote).await;
111 : }
112 10 : "flat" => payload.extend(scope.flat(&page).await),
113 0 : _ => {}
114 : }
115 186 : }
116 204 : let out = if accept == Accept::GeoJson {
117 4 : crate::repr::to_geojson_collection(payload, None)
118 : } else {
119 200 : Value::Array(payload)
120 : };
121 : // 6.3.6: the json answer's Link names the @context it was compacted in
122 204 : let served = crate::contexts::linkable(st, &tenant, headers, &parsed.ctx, accept).await?;
123 204 : let mut resp = respond_prefer(StatusCode::OK, out, &served, accept, &tenant, headers);
124 204 : attach_paging(&mut resp, count_hdr, &[]);
125 204 : filter.mark(&st.host_alias, resp.headers_mut());
126 204 : Ok(resp)
127 440 : }
128 :
129 : /// Each page Entity in the requested representation, beside the Entity it
130 : /// was rendered from; one a `pick` leaves empty is dropped.
131 204 : fn shaped_page<'p>(
132 204 : page: &'p [Value],
133 204 : repr: &crate::repr::Repr,
134 204 : ctx: &antares_jsonld::Context,
135 204 : ) -> Vec<(&'p Value, Value)> {
136 204 : page.iter()
137 224 : .filter_map(|doc| {
138 224 : let shaped = apply(doc, repr);
139 224 : if repr.pick.is_some() && shaped.as_object().is_some_and(|o| o.is_empty()) {
140 0 : return None;
141 224 : }
142 224 : Some((doc, crate::repr::compact_for(repr, &shaped, ctx)))
143 224 : })
144 204 : .collect()
145 204 : }
146 :
147 : /// 5.7.2.4: the Query must name what it reads, and every filter in it must
148 : /// parse.
149 246 : fn refuse_too_wide(vp: &HashMap<String, String>) -> ApiResult<()> {
150 : // 5.7.2.4 (p. 201): "At least one of the following input data shall be
151 : // provided: a) selector of Entity Types; b) list of Attribute names,
152 : // including at least one non-system Attribute; c) NGSI-LD Query,
153 : // including at least one non-system Attribute; d) NGSI-LD GeoQuery;
154 : // e) local scope. If none of the above is provided, then an error of type
155 : // BadRequestData shall be raised (too wide query)." Query Entities is ONE
156 : // operation: the resource that carries the Query in a body answers to the
157 : // same behaviour clause as the one that carries it in the URI, and
158 : // without this a bare `{"type":"Query"}` reads the whole tenant and fans
159 : // the filterless query out to every matching registration.
160 246 : let q_ast = vp.get("q").map(|q| antares_ql::parse_q(q)).transpose()?;
161 : // 5.7.2.4: "if ... the query, geoquery or context source filter are not
162 : // syntactically valid (as per the referred clauses 4.9 and 4.10) an error
163 : // of type BadRequestData shall be raised." Registration matching parses
164 : // the csf again and drops what it cannot parse, which widens the fan-out
165 : // to the Context Sources the filter was there to exclude — so it is
166 : // refused here, the way the twin carrying the Query in the URI refuses it.
167 242 : if let Some(csf) = vp.get("csf") {
168 4 : antares_ql::parse_q(csf)?;
169 238 : }
170 238 : if !crate::entities::qualifies_non_wide(vp, q_ast.as_ref()) {
171 12 : return Err(NgsiError::BadRequestData(
172 12 : "query needs at least one of type, attrs, q, georel (5.7.2.4)".into(),
173 12 : )
174 12 : .into());
175 226 : }
176 226 : Ok(())
177 246 : }
178 :
179 : /// The matches the registrations add, unless this request came round the
180 : /// federation loop.
181 226 : async fn fed_matches(
182 226 : st: &AppState,
183 226 : tenant: &antares_model::TenantId,
184 226 : headers: &HeaderMap,
185 226 : ctx: &antares_jsonld::Context,
186 226 : vp: &HashMap<String, String>,
187 226 : ) -> ApiResult<Vec<(bool, Value)>> {
188 : Ok(
189 226 : if crate::federation::active(vp)
190 226 : && !crate::federation::via_loop(
191 226 : headers,
192 226 : &crate::federation::alias_for(&st.host_alias, tenant),
193 226 : )
194 : {
195 : // 6.3.17 scopes NGSILD-Warning to GET /entities(/{id}) — collected
196 : // here for the log only, never emitted on entityOperations/query
197 226 : let mut warnings = Vec::new();
198 226 : let fed =
199 226 : crate::federation::fed_query(st, tenant, headers, ctx, vp, &mut warnings).await?;
200 226 : for w in &warnings {
201 0 : tracing::debug!("distributed query warning (batch query): {w}");
202 : }
203 226 : fed
204 : } else {
205 0 : Vec::new()
206 : },
207 : )
208 226 : }
209 :
210 : /// The 4.5.23 join of one page: what both join modes walk with.
211 : struct JoinScope<'a> {
212 : st: &'a AppState,
213 : tenant: &'a antares_model::TenantId,
214 : ctx: &'a antares_jsonld::Context,
215 : repr: &'a crate::repr::Repr,
216 : level: usize,
217 : held: Vec<String>,
218 : }
219 :
220 : impl JoinScope<'_> {
221 : /// 4.5.23 inline join: the linked Entities inside each payload Entity,
222 : /// each beside the Entity it was rendered from; `remote` carries the
223 : /// request headers when the query is federated.
224 8 : async fn inline(&self, payload: &mut [Value], stored: &[&Value], remote: Option<&HeaderMap>) {
225 8 : let mut budget = crate::repr::MAX_JOIN_LOOKUPS;
226 16 : for (p, stored) in payload.iter_mut().zip(stored) {
227 16 : crate::repr::inline_join_beyond(
228 16 : self.st,
229 16 : self.tenant,
230 16 : self.ctx,
231 16 : self.repr,
232 16 : p,
233 16 : stored,
234 16 : self.level,
235 16 : &self.held,
236 16 : &mut budget,
237 16 : remote,
238 16 : )
239 16 : .await;
240 : }
241 8 : }
242 :
243 : /// 4.5.23 flat join: the linked Entities the page does not hold, to follow it.
244 10 : async fn flat(&self, page: &[Value]) -> Vec<Value> {
245 10 : let (st, tenant, ctx, repr, level, held) = (
246 10 : self.st,
247 10 : self.tenant,
248 10 : self.ctx,
249 10 : self.repr,
250 10 : self.level,
251 10 : &self.held[..],
252 10 : );
253 10 : let mut budget = crate::repr::MAX_JOIN_LOOKUPS;
254 10 : let mut linked = std::collections::BTreeMap::new();
255 20 : for doc in page {
256 20 : crate::repr::collect_flat_beyond(
257 20 : st,
258 20 : tenant,
259 20 : ctx,
260 20 : repr,
261 20 : doc,
262 20 : level,
263 20 : &mut linked,
264 20 : held,
265 20 : &mut budget,
266 20 : )
267 20 : .await;
268 : }
269 20 : let page_ids: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
270 10 : linked
271 10 : .into_iter()
272 2016 : .filter(|(id, _)| !page_ids.contains(&id.as_str()))
273 2016 : .map(|(_, (ldoc, lrepr))| crate::repr::compact_for(&lrepr, &apply(&ldoc, &lrepr), ctx))
274 10 : .collect()
275 10 : }
276 : }
|
