Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! PgStore slice three: the temporal store over
3 : //! `attr_instances` ROWS. The 0002
4 : //! bridge doc is gone — `temporal_entities` holds only the small `meta`
5 : //! document; every instance lives as a row, reads RECONSTRUCT the doc shape
6 : //! the API layer consumes (so window()/aggregation/presentation are
7 : //! untouched), and writes are deltas, never a full-history rewrite.
8 : //!
9 : //! What this buys: the hypertable/partition
10 : //! machinery acts on the data queries actually read; retention shortens
11 : //! query results; instance pruning and entity paging run in SQL with the
12 : //! `(tenant_id, entity_id, attr_id, observed_at DESC)` index.
13 :
14 : use antares_model::TenantId;
15 : use serde_json::Value;
16 : use sqlx::postgres::PgPool;
17 : use sqlx::Row;
18 :
19 : pub struct PgTemporalStore {
20 : pool: PgPool,
21 : /// Set when this instance serves only the temporal seam: it never holds
22 : /// the entities, so the append guard must not look for them here.
23 : pub temporal_only: bool,
24 : }
25 :
26 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
27 : /// certain Entity, Property or Relationship shall become invalid" — an
28 : /// expired temporal entity is absent from every read, ahead of the retention
29 : /// sweep. One shared literal so `query`, its count fallback and `get_range`
30 : /// can never disagree.
31 : ///
32 : /// The stamp is jsonb TEXT, so it goes through `try_timestamptz`
33 : /// (0001_init.sql): a bare cast RAISES on anything it cannot parse, and
34 : /// these reads are
35 : /// tenant-wide — one bad stamp would take down the whole tenant's temporal
36 : /// API rather than hide one entity. An unusable stamp reads as no expiry, the
37 : /// same direction the memory arm takes in `filter::expired_at`.
38 : const NOT_EXPIRED: &str =
39 : "(try_timestamptz(m.meta->>'expiresAt') IS NULL OR try_timestamptz(m.meta->>'expiresAt') > now())";
40 :
41 : /// The temporal tables carry the same four extracted columns as `entities`,
42 : /// so they are read by the same function (`super::entity::types_scopes_stamps`)
43 : /// rather than a second copy of it.
44 : use super::entity::types_scopes_stamps as extract;
45 :
46 : /// Entity-doc members that are NOT temporal attributes. `scope` may itself be
47 : /// instance-shaped (deletion instances, 020_19/20) — it lives in `meta`
48 : /// verbatim either way, exactly as the bridge kept it unpruned.
49 : const DOC_META: &[&str] = antares_model::ENTITY_META_KEYS;
50 :
51 : /// The meta-only document stored in `temporal_entities.meta`.
52 399 : fn meta_of(doc: &Value) -> Value {
53 399 : let mut m = serde_json::Map::new();
54 399 : if let Some(obj) = doc.as_object() {
55 3192 : for k in DOC_META {
56 3192 : if let Some(v) = obj.get(*k) {
57 1608 : m.insert((*k).to_owned(), v.clone());
58 1608 : }
59 : }
60 0 : }
61 399 : Value::Object(m)
62 399 : }
63 :
64 : /// Decompose a doc's attribute arrays into row JSON for the multi-row
65 : /// INSERT. The `observed_at` string is derived ONLY from the instance
66 : /// document, so decomposing a reconstructed doc reproduces byte-identical
67 : /// keys (what the mutate diff relies on).
68 410 : fn decompose(doc: &Value) -> Vec<Value> {
69 410 : let mut rows: Vec<Value> = Vec::new();
70 410 : if let Some(obj) = doc.as_object() {
71 1168 : for (attr, instances) in obj {
72 1168 : if DOC_META.contains(&attr.as_str()) {
73 654 : continue;
74 514 : }
75 514 : let Some(arr) = instances.as_array() else {
76 4 : continue;
77 : };
78 607 : for i in arr {
79 3957 : let s = |k: &str| i.get(k).and_then(Value::as_str);
80 : // 4.6.3: a comma seconds-fraction is legal in a request and
81 : // these stamps go straight into `::timestamptz` casts, which
82 : // refuse it — and the raise lands in the temporal drain,
83 : // which absorbs it, so the whole request's history would be
84 : // lost with a 2xx already returned. Timestamps only:
85 : // `datasetId` is a URI, where a comma is an ordinary
86 : // character and rewriting it would corrupt the id.
87 2743 : let ts = |k: &str| s(k).map(antares_store::filter::canonical_datetime);
88 607 : let Some(instance_id) = s("instanceId") else {
89 0 : continue; // stamped by the API layer; belt only
90 : };
91 : // observed_at falls back through the instance's own
92 : // timestamps — deletion instances carry only deletedAt and
93 : // must NOT collapse onto the epoch (retention would reap them)
94 607 : let observed = ts("observedAt")
95 607 : .or_else(|| ts("modifiedAt"))
96 607 : .or_else(|| ts("deletedAt"))
97 607 : .or_else(|| ts("createdAt"))
98 607 : .unwrap_or(std::borrow::Cow::Borrowed("1970-01-01T00:00:00Z"));
99 607 : rows.push(serde_json::json!({
100 607 : "attr_id": attr,
101 607 : "instance_id": instance_id,
102 607 : "dataset_id": s("datasetId"),
103 607 : "observed_at": observed,
104 607 : "created_at": ts("createdAt").unwrap_or_else(|| observed.clone()),
105 607 : "modified_at": ts("modifiedAt").unwrap_or(observed),
106 607 : "deleted_at": ts("deletedAt"),
107 607 : "data": i,
108 : }));
109 : }
110 : }
111 0 : }
112 410 : rows
113 410 : }
114 :
115 : /// Multi-row instance upsert inside the caller's transaction.
116 389 : async fn insert_rows(
117 389 : tx: &mut sqlx::PgConnection,
118 389 : tenant: &TenantId,
119 389 : entity_id: &str,
120 389 : rows: Vec<Value>,
121 389 : ) -> Result<(), sqlx::Error> {
122 389 : if rows.is_empty() {
123 9 : return Ok(());
124 380 : }
125 : // geo_value: extracted per instance when the value LOOKS like a GeoJSON
126 : // geometry; try_geomfromgeojson (0001_init.sql) maps anything PostGIS rejects to
127 : // NULL, which the S3 prefilter treats as "reaches the evaluator".
128 380 : sqlx::query(
129 380 : "INSERT INTO attr_instances
130 380 : (tenant_id, entity_id, attr_id, instance_id, dataset_id, observed_at,
131 380 : created_at, modified_at, deleted_at, data, geo_value)
132 380 : SELECT $1, $2, e->>'attr_id', e->>'instance_id', e->>'dataset_id',
133 380 : (e->>'observed_at')::timestamptz, (e->>'created_at')::timestamptz,
134 380 : (e->>'modified_at')::timestamptz, (e->>'deleted_at')::timestamptz,
135 380 : e->'data',
136 380 : CASE WHEN jsonb_typeof(e->'data'->'value') = 'object'
137 380 : AND e->'data'->'value'->>'type' IN
138 380 : ('Point','MultiPoint','LineString','MultiLineString',
139 380 : 'Polygon','MultiPolygon')
140 380 : THEN try_geomfromgeojson((e->'data'->'value')::text) END
141 380 : FROM jsonb_array_elements($3::jsonb) AS e
142 380 : ON CONFLICT (tenant_id, entity_id, attr_id, instance_id, observed_at)
143 380 : DO UPDATE SET data = EXCLUDED.data, modified_at = EXCLUDED.modified_at,
144 380 : dataset_id = EXCLUDED.dataset_id,
145 380 : deleted_at = EXCLUDED.deleted_at,
146 380 : geo_value = EXCLUDED.geo_value",
147 380 : )
148 380 : .bind(tenant.as_str())
149 380 : .bind(entity_id)
150 380 : .bind(Value::Array(rows))
151 380 : .execute(&mut *tx)
152 380 : .await?;
153 380 : Ok(())
154 389 : }
155 :
156 : // TemporalFilter lives in `store::filter`; re-exported for path compat.
157 : pub use crate::store::filter::{TemporalFilter, TemporalOutcome};
158 :
159 : /// The correlated subquery reconstructing the attribute object for the meta
160 : /// row aliased `m`, with the 4.11 range and the lastN RANK() cap applied over
161 : /// the rows (byte-exact against the API window: predicates and ordering run
162 : /// on the instance JSON with COLLATE "C", never on the partition column).
163 : /// Returns the SQL fragment + its text binds, numbered from `first_bind`;
164 : /// `None` when a range is present but outside the compiler's exact subset —
165 : /// the caller then reconstructs unpruned and the window stays the arbiter.
166 380 : fn attr_object_expr(f: &TemporalFilter<'_>, first_bind: usize) -> Option<(String, Vec<String>)> {
167 380 : let (range_and, mut binds) = window_sql(f, first_bind)?;
168 378 : let tp = first_bind;
169 378 : let expr = match f.last_n {
170 4 : Some(n) => {
171 4 : let n_bind = first_bind + binds.len();
172 4 : binds.push(n.to_string());
173 : // 4.11 lastN keeps the N most recent INSTANTS, so the rank orders
174 : // on the same canonical key the window compares on — raw bytes
175 : // put "…00.000Z" after "…00Z" and kept the wrong N.
176 4 : let order_key = crate::compile::temporal::dt_key_sql(&format!("(ai.data ->> ${tp})"));
177 4 : format!(
178 : "COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (\
179 : SELECT s.attr_id, jsonb_agg(s.data ORDER BY s.created_at, s.observed_at, s.instance_id) AS insts \
180 : FROM (SELECT ai.*, rank() OVER (PARTITION BY ai.attr_id, ai.data ->> 'datasetId' \
181 : ORDER BY {order_key} DESC NULLS LAST) AS rk \
182 : FROM attr_instances ai \
183 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_id = m.id{range_and}) s \
184 : WHERE s.rk <= ${n_bind}::bigint GROUP BY s.attr_id) g), '{{}}'::jsonb)"
185 : )
186 : }
187 : None => {
188 : // No range and no lastN: nothing names the timeproperty, so the
189 : // bind `window_sql` reserved for it is not a parameter of the
190 : // statement the caller assembles. Handing it back anyway would
191 : // make every plain Retrieve Temporal Evolution bind one argument
192 : // more than its SQL declares.
193 374 : if f.range.is_none() {
194 127 : binds.clear();
195 247 : }
196 374 : format!(
197 : "COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (\
198 : SELECT ai.attr_id, jsonb_agg(ai.data ORDER BY ai.created_at, ai.observed_at, ai.instance_id) AS insts \
199 : FROM attr_instances ai \
200 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_id = m.id{range_and} \
201 : GROUP BY ai.attr_id) g), '{{}}'::jsonb)"
202 : )
203 : }
204 : };
205 378 : Some((expr, binds))
206 380 : }
207 :
208 : /// The 4.11 window over the instance rows `ai` of the meta row `m`: the
209 : /// ` AND …` fragment plus its text binds — $first_bind is always the
210 : /// timeproperty (predicate member / order key), the range binds follow.
211 : /// `None` when the range is outside the compiler's exact subset.
212 392 : fn window_sql(f: &TemporalFilter<'_>, first_bind: usize) -> Option<(String, Vec<String>)> {
213 392 : let mut binds = vec![f.timeproperty.to_owned()];
214 392 : let mut range_and = String::new();
215 392 : if let Some(r) = &f.range {
216 260 : let c = crate::compile::temporal::compile_instance_range(r, "ai.data", first_bind)?;
217 258 : range_and = format!(" AND {}", c.sql);
218 258 : debug_assert_eq!(c.binds[0], f.timeproperty);
219 258 : binds.extend(c.binds.into_iter().skip(1));
220 : // widened COLUMN bound on the SAME binds ($first_bind+1 = timeAt):
221 : // lets the (tenant, entity, attr, observed_at) btree serve the range;
222 : // the byte-exact text predicate above still decides membership
223 258 : if let Some(cb) = crate::compile::temporal::column_range_bound(r, "ai", first_bind + 1) {
224 258 : range_and.push_str(&format!(" AND {cb}"));
225 258 : }
226 132 : }
227 390 : Some((range_and, binds))
228 392 : }
229 :
230 : /// Timestamp text in the shape the API's aggregated rows carry.
231 : const BUCKET_TS: &str = r#"to_char({} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"')"#;
232 :
233 : /// 4.5.19 / 5.7.4.4 aggregated temporal representation computed in SQL for
234 : /// the meta row `m`: per attribute the bucket matrix of every requested
235 : /// method as `[value, start, end]` rows, in one object
236 : /// `{"bad": <any windowed value non-numeric>, "attrs": {iri: {"type":
237 : /// "Property", method: rows…}}}`. Only the numeric class (numbers and
238 : /// booleans, Table 4.5.19.1-1) is computed here; `bad` tells the caller to
239 : /// reconstruct instead so the API keeps every other class and the
240 : /// eligibility errors. Buckets: `period_secs` wide from the anchor (the
241 : /// request's timeAt, else the attribute's first instant); no period = one
242 : /// bucket spanning the query's whole time range (4.5.19.1 PT0S), with the
243 : /// edge 4.11 leaves open closed by the attribute's own first or last
244 : /// instant — the API's own rule.
245 12 : fn aggregate_expr(
246 12 : f: &TemporalFilter<'_>,
247 12 : agg: &crate::store::filter::Aggregate<'_>,
248 12 : first_bind: usize,
249 12 : ) -> Option<(String, Vec<String>)> {
250 12 : let (range_and, mut binds) = window_sql(f, first_bind)?;
251 12 : let tp = first_bind;
252 12 : let col = match f.timeproperty {
253 12 : "observedAt" => "observed_at",
254 0 : "createdAt" => "created_at",
255 0 : "modifiedAt" => "modified_at",
256 0 : _ => return None,
257 : };
258 12 : let anchor = match agg.anchor {
259 11 : Some(a) => {
260 11 : binds.push(antares_store::filter::canonical_datetime(a).into_owned());
261 11 : format!("${}::timestamptz", first_bind + binds.len() - 1)
262 : }
263 1 : None => format!("min(ai.{col}) OVER (PARTITION BY ai.attr_id)"),
264 : };
265 12 : let (bs, be) = match agg.period_secs {
266 : None => {
267 : // 4.5.19.1: a zero duration "is interpreted as a duration
268 : // spanning the whole time range specified by the temporal
269 : // query". `before` names only the range's end and `after` only
270 : // its start (4.11), so the data closes the other edge.
271 9 : let (qs, qe) = match &f.range {
272 8 : Some(r) if r.timerel == "before" => (None, Some(r.time_at)),
273 7 : Some(r) if r.timerel == "between" => (Some(r.time_at), r.end_time_at),
274 6 : Some(r) if r.timerel == "after" => (Some(r.time_at), None),
275 1 : _ => (None, None),
276 : };
277 9 : let start = match qs {
278 7 : Some(v) => {
279 7 : binds.push(antares_store::filter::canonical_datetime(v).into_owned());
280 7 : format!("${}::timestamptz", first_bind + binds.len() - 1)
281 : }
282 2 : None => "s0.first_ts".to_owned(),
283 : };
284 9 : let end = match qe {
285 2 : Some(v) => {
286 2 : binds.push(antares_store::filter::canonical_datetime(v).into_owned());
287 2 : format!("${}::timestamptz", first_bind + binds.len() - 1)
288 : }
289 7 : None => "s0.last + interval '1 second'".to_owned(),
290 : };
291 9 : (start, end)
292 : }
293 3 : Some(sc) => {
294 3 : binds.push(sc.to_string());
295 3 : let n = first_bind + binds.len() - 1;
296 3 : let start = format!("date_bin(make_interval(secs => ${n}::bigint), s0.ts, s0.anchor)");
297 3 : (
298 3 : start.clone(),
299 3 : format!("{start} + make_interval(secs => ${n}::bigint)"),
300 3 : )
301 : }
302 : };
303 12 : let mut aggs = String::new();
304 12 : let mut rows = String::new();
305 12 : let mut pairs = String::new();
306 29 : for (i, m) in agg.methods.iter().enumerate() {
307 29 : let sql = match m.as_str() {
308 29 : "totalCount" => "count(*)",
309 21 : "distinctCount" => "count(DISTINCT s.v)",
310 21 : "min" => "min(s.v)",
311 19 : "max" => "max(s.v)",
312 12 : "sum" => "sum(s.v)",
313 12 : "avg" => "avg(s.v)",
314 0 : "stddev" => "stddev_pop(s.v)",
315 0 : "sumsq" => "sum(s.v * s.v)",
316 0 : _ => return None,
317 : };
318 29 : aggs.push_str(&format!(", {sql} AS m{i}"));
319 29 : rows.push_str(&format!(
320 29 : ", jsonb_agg(jsonb_build_array(g.m{i}, {}, {}) ORDER BY g.bs) AS r{i}",
321 29 : BUCKET_TS.replace("{}", "g.bs"),
322 29 : BUCKET_TS.replace("{}", "g.be")
323 29 : ));
324 : // method names are the allowlist above, never request text
325 29 : pairs.push_str(&format!(", '{m}', b.r{i}"));
326 : }
327 12 : let expr = format!(
328 : "(SELECT jsonb_build_object('bad', bool_or(b.bad), 'attrs', \
329 : jsonb_object_agg(b.attr_id, jsonb_build_object('type', 'Property'{pairs}))) \
330 : FROM (SELECT g.attr_id, bool_or(g.bad) AS bad{rows} \
331 : FROM (SELECT s.attr_id, s.bs, s.be, bool_or(s.bad) AS bad{aggs} \
332 : FROM (SELECT s0.attr_id, s0.v, s0.bad, {bs} AS bs, {be} AS be \
333 : FROM (SELECT ai.attr_id, \
334 : CASE WHEN jsonb_typeof(ai.data -> 'value') = 'boolean' \
335 : THEN (CASE WHEN (ai.data ->> 'value')::boolean THEN 1 ELSE 0 END)::float8 \
336 : WHEN jsonb_typeof(ai.data -> 'value') = 'number' \
337 : THEN (ai.data ->> 'value')::float8 END AS v, \
338 : jsonb_typeof(ai.data -> 'value') IS DISTINCT FROM 'number' \
339 : AND jsonb_typeof(ai.data -> 'value') IS DISTINCT FROM 'boolean' AS bad, \
340 : ai.{col} AS ts, {anchor} AS anchor, \
341 : min(ai.{col}) OVER (PARTITION BY ai.attr_id) AS first_ts, \
342 : max(ai.{col}) OVER (PARTITION BY ai.attr_id) AS last \
343 : FROM attr_instances ai \
344 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_id = m.id \
345 : AND jsonb_typeof(ai.data -> ${tp}) = 'string'{range_and}) s0) s \
346 : GROUP BY s.attr_id, s.bs, s.be) g \
347 : GROUP BY g.attr_id) b)"
348 : );
349 12 : Some((expr, binds))
350 12 : }
351 :
352 : impl PgTemporalStore {
353 59 : pub fn new(pool: PgPool) -> Self {
354 59 : Self {
355 59 : pool,
356 59 : temporal_only: false,
357 59 : }
358 59 : }
359 :
360 : /// `false` when the id already exists (create semantics, like the memory
361 : /// store's `create`).
362 : ///
363 : /// 4.22: an expired one does NOT exist, so it is dropped here — history
364 : /// included — and the insert below is then an ordinary create. Leaving it
365 : /// in place would make `create` report a conflict for an id every read
366 : /// calls absent, and `AnyStore::upsert` reads that `false` as "already
367 : /// there" and falls through to `mutate`, which refuses an expired entity
368 : /// too: the upsert would write nothing and report success.
369 140 : pub async fn create(
370 140 : &self,
371 140 : tenant: &TenantId,
372 140 : id: &str,
373 140 : doc: &Value,
374 140 : ) -> Result<bool, sqlx::Error> {
375 140 : let (types, scopes, created, modified) = extract(doc);
376 140 : let meta = meta_of(doc);
377 140 : let rows = decompose(doc);
378 140 : let mut tx = super::begin(&self.pool).await?;
379 140 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
380 140 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
381 140 : let reaped = sqlx::query(sqlx::AssertSqlSafe(format!(
382 140 : "DELETE FROM temporal_entities m \
383 140 : WHERE m.tenant_id = $1 AND m.id = $2 AND NOT {NOT_EXPIRED}"
384 140 : )))
385 140 : .bind(tenant.as_str())
386 140 : .bind(id)
387 140 : .execute(&mut *tx)
388 140 : .await?
389 140 : .rows_affected();
390 140 : if reaped == 1 {
391 1 : sqlx::query("DELETE FROM attr_instances WHERE tenant_id = $1 AND entity_id = $2")
392 1 : .bind(tenant.as_str())
393 1 : .bind(id)
394 1 : .execute(&mut *tx)
395 1 : .await?;
396 139 : }
397 140 : let n = sqlx::query(
398 140 : "INSERT INTO temporal_entities
399 140 : (tenant_id, id, types, scopes, meta, created_at, modified_at)
400 140 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz)
401 140 : ON CONFLICT (tenant_id, id) DO NOTHING",
402 140 : )
403 140 : .bind(tenant.as_str())
404 140 : .bind(id)
405 140 : .bind(&types)
406 140 : .bind(&scopes)
407 140 : .bind(&meta)
408 140 : .bind(&created)
409 140 : .bind(&modified)
410 140 : .execute(&mut *tx)
411 140 : .await?
412 140 : .rows_affected();
413 140 : if n == 1 {
414 133 : insert_rows(&mut tx, tenant, id, rows).await?;
415 7 : }
416 140 : tx.commit().await?;
417 140 : Ok(n == 1)
418 140 : }
419 :
420 : /// Append-only fast path (auto-recording, 5.6.12 adds): NO reconstruction,
421 : /// no history read — a shell meta insert (first touch) plus one multi-row
422 : /// instance upsert. This is the write the old full-resync made O(history).
423 : ///
424 : /// Conditional on the entity still existing. 5.6.6 Delete Entity removes
425 : /// the entity and the temporal evolution recorded for it, in that order and
426 : /// in two transactions; an auto-recording append that overlaps the delete
427 : /// would otherwise commit history for an entity that is gone, and no later
428 : /// delete would ever clean it. The `FOR KEY SHARE` lock is what makes the
429 : /// check hold: it lets concurrent updates of the same entity through and
430 : /// makes a concurrent DELETE wait until this append has committed.
431 248 : pub async fn append(
432 248 : &self,
433 248 : tenant: &TenantId,
434 248 : id: &str,
435 248 : shell: &Value,
436 248 : additions: &Value,
437 248 : ) -> Result<(), sqlx::Error> {
438 248 : let (types, scopes, created, modified) = extract(shell);
439 248 : let meta = meta_of(shell);
440 248 : let rows = decompose(additions);
441 248 : let mut tx = super::begin(&self.pool).await?;
442 248 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
443 248 : if !self.temporal_only {
444 248 : let live = sqlx::query(
445 248 : "SELECT 1 FROM entities WHERE tenant_id = $1 AND id = $2 FOR KEY SHARE",
446 248 : )
447 248 : .bind(tenant.as_str())
448 248 : .bind(id)
449 248 : .fetch_optional(&mut *tx)
450 248 : .await?;
451 248 : if live.is_none() {
452 1 : tx.commit().await?;
453 1 : return Ok(());
454 247 : }
455 0 : }
456 : // DO NOTHING froze types/scopes at first touch — an
457 : // entity gaining a type stayed invisible to type-filtered
458 : // temporal queries forever. The shell carries the CURRENT
459 : // entity, so refresh on change; the IS DISTINCT FROM guard keeps
460 : // the common no-change append from churning the meta row.
461 247 : sqlx::query(
462 247 : "INSERT INTO temporal_entities
463 247 : (tenant_id, id, types, scopes, meta, created_at, modified_at)
464 247 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz)
465 247 : ON CONFLICT (tenant_id, id) DO UPDATE SET
466 247 : types = EXCLUDED.types,
467 247 : scopes = EXCLUDED.scopes,
468 247 : meta = EXCLUDED.meta,
469 247 : modified_at = EXCLUDED.modified_at
470 247 : WHERE temporal_entities.types IS DISTINCT FROM EXCLUDED.types
471 247 : OR temporal_entities.scopes IS DISTINCT FROM EXCLUDED.scopes
472 247 : OR temporal_entities.meta IS DISTINCT FROM EXCLUDED.meta",
473 247 : )
474 247 : .bind(tenant.as_str())
475 247 : .bind(id)
476 247 : .bind(&types)
477 247 : .bind(&scopes)
478 247 : .bind(&meta)
479 247 : .bind(&created)
480 247 : .bind(&modified)
481 247 : .execute(&mut *tx)
482 247 : .await?;
483 247 : insert_rows(&mut tx, tenant, id, rows).await?;
484 247 : tx.commit().await?;
485 247 : Ok(())
486 248 : }
487 :
488 107 : pub async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
489 107 : self.get_range(tenant, id, &TemporalFilter::default()).await
490 107 : }
491 :
492 : /// 5.6.16 Delete Temporal Evolution: `true` when it was there, `false`
493 : /// for "no existing Entity whose id (URI) is equivalent held locally",
494 : /// which the caller answers as ResourceNotFound. 4.22 decides what "held
495 : /// locally" means, the same way `get_range` and `query` decide it.
496 2555 : pub async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, sqlx::Error> {
497 2555 : let mut tx = super::begin(&self.pool).await?;
498 2555 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
499 2555 : let n = sqlx::query(sqlx::AssertSqlSafe(format!(
500 2555 : "DELETE FROM temporal_entities m \
501 2555 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED}"
502 2555 : )))
503 2555 : .bind(tenant.as_str())
504 2555 : .bind(id)
505 2555 : .execute(&mut *tx)
506 2555 : .await?
507 2555 : .rows_affected();
508 : // no FK: a partitioned table cannot be the referencing side of a
509 : // cascade from temporal_entities — clean the instances explicitly,
510 : // and only for the row this call actually removed. An expired
511 : // entity is refused above and keeps its history until the 4.22
512 : // reap or a create replaces it; wiping it here would destroy the
513 : // history behind a 404.
514 2555 : if n == 1 {
515 293 : sqlx::query("DELETE FROM attr_instances WHERE tenant_id = $1 AND entity_id = $2")
516 293 : .bind(tenant.as_str())
517 293 : .bind(id)
518 293 : .execute(&mut *tx)
519 293 : .await?;
520 2262 : }
521 2555 : tx.commit().await?;
522 2555 : Ok(n == 1)
523 2555 : }
524 :
525 : /// Temporal query: entity narrowing (ids/types/attrs) in the WHERE,
526 : /// instance pruning (range + lastN cap) in the reconstruction, and —
527 : /// when the caller passes a page — entity qualification + LIMIT/OFFSET
528 : /// in SQL, so a temporal query no longer materializes the whole tenant.
529 248 : pub async fn query(
530 248 : &self,
531 248 : tenant: &TenantId,
532 248 : f: &TemporalFilter<'_>,
533 248 : ) -> Result<TemporalOutcome, sqlx::Error> {
534 248 : self.query_inner(tenant, f, f.aggregate.is_some()).await
535 248 : }
536 :
537 : /// `push_agg`: compute the filter's 4.5.19 aggregation in SQL; a row
538 : /// whose windowed values are not all numeric makes the whole query fall
539 : /// back to instance reconstruction (one extra round trip, only then).
540 249 : async fn query_inner(
541 249 : &self,
542 249 : tenant: &TenantId,
543 249 : f: &TemporalFilter<'_>,
544 249 : push_agg: bool,
545 249 : ) -> Result<TemporalOutcome, sqlx::Error> {
546 : enum B {
547 : Text(String),
548 : Arr(Vec<String>),
549 : Num(i64),
550 : Float(f64),
551 : }
552 249 : let mut binds: Vec<B> = vec![B::Text(tenant.as_str().to_owned())];
553 : // 4.22: an expired ENTITY is invalid on temporal reads too — filter it
554 : // in SQL (no bind) so paging/totals stay exact. Expired instances are
555 : // stripped at the read boundary (any.rs). Literal, applies to the
556 : // fallback count query too.
557 249 : let mut wheres = vec!["m.tenant_id = $1".to_owned(), NOT_EXPIRED.to_owned()];
558 249 : if let Some(ids) = f.ids {
559 29 : binds.push(B::Arr(ids.iter().map(|s| s.to_string()).collect()));
560 29 : wheres.push(format!("m.id = ANY(${})", binds.len()));
561 220 : }
562 249 : if let Some(types) = f.types {
563 168 : // overlap: entity has ANY of the wanted types (flat OR list)
564 168 : binds.push(B::Arr(types.to_vec()));
565 168 : wheres.push(format!("m.types && ${}", binds.len()));
566 168 : }
567 249 : if let Some(attrs) = f.attrs {
568 2 : binds.push(B::Arr(attrs.to_vec()));
569 2 : wheres.push(format!(
570 2 : "EXISTS (SELECT 1 FROM attr_instances x WHERE x.tenant_id = m.tenant_id \
571 2 : AND x.entity_id = m.id AND x.attr_id = ANY(${}))",
572 2 : binds.len()
573 2 : ));
574 247 : }
575 : // page pushdown: only when the caller passed one AND the range (if
576 : // any) compiles — SQL then also applies the evaluator's entity-
577 : // qualification rule (≥1 instance, in-window when ranged). WHERE
578 : // binds come FIRST so the fallback count query (offset past the end)
579 : // can reuse them with identical numbering.
580 : // None: no range to compile. Some(None): a range that does not
581 : // compile, which is the case page pushdown must not take — the SQL
582 : // would qualify entities on a window it cannot express.
583 249 : let compiled_range = f.range.as_ref().map(|r| {
584 242 : crate::compile::temporal::compile_instance_range(r, "ai.data", binds.len() + 1)
585 242 : });
586 249 : let mut paged = false;
587 249 : if f.page.is_some() && !matches!(compiled_range, Some(None)) {
588 98 : let mut qual = "EXISTS (SELECT 1 FROM attr_instances ai WHERE \
589 98 : ai.tenant_id = m.tenant_id AND ai.entity_id = m.id"
590 98 : .to_owned();
591 98 : if let (Some(r), Some(Some(c))) = (&f.range, compiled_range) {
592 98 : let n = binds.len() + 1;
593 242 : for b in c.binds {
594 242 : binds.push(B::Text(b));
595 242 : }
596 98 : qual.push_str(&format!(" AND {}", c.sql));
597 : // index-serving widened bound on the same binds ($n+1 = timeAt)
598 98 : if let Some(cb) = crate::compile::temporal::column_range_bound(r, "ai", n + 1) {
599 98 : qual.push_str(&format!(" AND {cb}"));
600 98 : }
601 0 : }
602 98 : qual.push(')');
603 98 : wheres.push(qual);
604 98 : paged = true;
605 151 : }
606 : // 5.7.4.4 S2 superset prefilter: entities with no windowed instance
607 : // satisfying the compilable part of q= are never reconstructed. The
608 : // API arbiter re-evaluates q on every row that comes back, so this
609 : // can only narrow (compile::qprefilter invariant), never decide.
610 249 : if let Some(qn) = f.q {
611 136 : if let Some(c) = crate::compile::qprefilter::compile_prefilter(
612 136 : qn,
613 136 : f.range.as_ref(),
614 136 : "m",
615 136 : binds.len() + 1,
616 136 : f.expand,
617 136 : ) {
618 1658 : for b in c.binds {
619 1658 : binds.push(B::Text(b));
620 1658 : }
621 122 : wheres.push(c.sql);
622 14 : }
623 113 : }
624 : // 5.7.4.4 S3 superset prefilter: entities with no windowed instance
625 : // of the geoproperty possibly satisfying the geoquery are never
626 : // reconstructed. NULL geo_value (rows with no extracted geometry) always
627 : // survives; GeoQuery::matches stays the arbiter.
628 249 : if let Some((spec, iri)) = f.geo {
629 25 : let attr_bind = binds.len() + 1;
630 25 : let mut win_binds: Vec<String> = Vec::new();
631 25 : let mut window = String::new();
632 25 : if let Some(r) = &f.range {
633 25 : if let Some(cb) =
634 25 : crate::compile::temporal::column_range_bound(r, "gi", attr_bind + 1)
635 : {
636 : // 4.6.3 lets a request spell the seconds fraction with a
637 : // comma; the bound lands in a `::timestamptz` cast, which
638 : // takes only the point form. Every other range bind is
639 : // canonicalized for exactly this reason.
640 25 : win_binds
641 25 : .push(antares_store::filter::canonical_datetime(r.time_at).into_owned());
642 25 : if r.timerel == "between" {
643 22 : if let Some(e) = r.end_time_at {
644 22 : win_binds
645 22 : .push(antares_store::filter::canonical_datetime(e).into_owned());
646 22 : }
647 3 : }
648 25 : window = format!(" AND {cb}");
649 0 : }
650 0 : }
651 25 : if let Some(c) = crate::compile::geo::compile_geo_instance(
652 25 : spec,
653 25 : "gi.geo_value",
654 25 : attr_bind + 1 + win_binds.len(),
655 25 : ) {
656 25 : binds.push(B::Text(iri.to_owned()));
657 47 : for b in win_binds {
658 47 : binds.push(B::Text(b));
659 47 : }
660 25 : for b in c.geo_binds {
661 25 : binds.push(B::Text(b));
662 25 : }
663 25 : for n in c.num_binds {
664 15 : binds.push(B::Float(n));
665 15 : }
666 25 : wheres.push(format!(
667 : "EXISTS (SELECT 1 FROM attr_instances gi \
668 : WHERE gi.tenant_id = m.tenant_id AND gi.entity_id = m.id \
669 : AND gi.attr_id = ${attr_bind}{window} AND {})",
670 : c.sql
671 : ));
672 0 : }
673 224 : }
674 249 : let n_where = binds.len();
675 249 : let where_sql = wheres.join(" AND ");
676 249 : let agg = if push_agg {
677 12 : f.aggregate
678 12 : .as_ref()
679 12 : .and_then(|a| aggregate_expr(f, a, n_where + 1))
680 : } else {
681 237 : None
682 : };
683 249 : let aggregated = agg.is_some();
684 249 : let (attr_expr, extra) = match agg {
685 12 : Some((e, b)) => (format!("jsonb_build_object('$agg', {e})"), b),
686 237 : None => match attr_object_expr(f, n_where + 1) {
687 237 : Some(v) => v,
688 : // refused range shape: reconstruct unpruned, window arbitrates.
689 : // The default filter has neither range nor lastN, the two
690 : // shapes the compiler can refuse: pinned by
691 : // `the_unpruned_fallback_always_compiles`.
692 : #[allow(clippy::expect_used)]
693 0 : None => attr_object_expr(&TemporalFilter::default(), n_where + 1)
694 0 : .expect("no range/lastN always compiles"),
695 : },
696 : };
697 249 : binds.extend(extra.into_iter().map(B::Text));
698 249 : let mut select_total = String::new();
699 249 : let mut tail = " ORDER BY m.id".to_owned();
700 : // `paged` is set only inside `if f.page.is_some()`, so binding the
701 : // two together says that in the type system instead of in a comment.
702 249 : if let (true, Some(page)) = (paged, f.page.as_ref()) {
703 98 : select_total = ", count(*) OVER () AS total".into();
704 98 : binds.push(B::Num(page.limit));
705 98 : tail.push_str(&format!(" LIMIT ${}", binds.len()));
706 98 : binds.push(B::Num(page.offset));
707 98 : tail.push_str(&format!(" OFFSET ${}", binds.len()));
708 151 : } else {
709 151 : // No page pushed down: the caller still has to filter, so the only
710 151 : // bound on this statement is the safety ceiling — without it a
711 151 : // bare `?timerel=…` reconstructs every temporal entity of the
712 151 : // tenant into one Vec.
713 151 : binds.push(B::Num(super::entity::MAX_UNDECIDED_ROWS));
714 151 : tail.push_str(&format!(" LIMIT ${}", binds.len()));
715 151 : }
716 249 : let sql = format!(
717 : "SELECT m.meta || {attr_expr}{select_total} FROM temporal_entities m \
718 : WHERE {where_sql}{tail}"
719 : );
720 249 : let first: Result<Option<TemporalOutcome>, sqlx::Error> = async {
721 249 : let mut tx = super::begin(&self.pool).await?;
722 249 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
723 : // `sql` is compiler literals + $n placeholders only.
724 249 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone()));
725 3494 : for b in &binds {
726 3494 : qy = match b {
727 2933 : B::Text(s) => qy.bind(s),
728 199 : B::Arr(v) => qy.bind(v),
729 347 : B::Num(n) => qy.bind(n),
730 15 : B::Float(x) => qy.bind(x),
731 : };
732 : }
733 249 : let rows = qy.fetch_all(&mut *tx).await?;
734 249 : let mut total = if paged {
735 98 : rows.first().map(|r| r.get::<i64, _>(1))
736 : } else {
737 151 : None
738 : };
739 249 : if paged && total.is_none() {
740 : // offset past the end: the window function came back with the
741 : // page, which is empty — count separately with the WHERE binds
742 30 : let count_sql =
743 30 : format!("SELECT count(*) FROM temporal_entities m WHERE {where_sql}");
744 30 : let mut cq = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql));
745 238 : for b in binds.iter().take(n_where) {
746 238 : cq = match b {
747 226 : B::Text(s) => cq.bind(s),
748 12 : B::Arr(v) => cq.bind(v),
749 0 : B::Num(n) => cq.bind(n),
750 0 : B::Float(x) => cq.bind(x),
751 : };
752 : }
753 30 : total = Some(cq.fetch_one(&mut *tx).await?);
754 219 : }
755 249 : tx.commit().await?;
756 249 : super::entity::check_ceiling(paged, rows.len(), super::entity::MAX_UNDECIDED_ROWS)?;
757 481 : let mut docs: Vec<Value> = rows.into_iter().map(|r| r.get::<Value, _>(0)).collect();
758 249 : if aggregated {
759 14 : for d in &mut docs {
760 14 : let Some(o) = d.as_object_mut() else { continue };
761 14 : let Some(agg) = o.remove("$agg") else {
762 0 : continue;
763 : };
764 14 : if agg.get("bad").and_then(Value::as_bool) == Some(true) {
765 1 : return Ok(None);
766 13 : }
767 13 : if let Some(attrs) = agg.get("attrs").and_then(Value::as_object) {
768 21 : for (k, v) in attrs {
769 21 : o.insert(k.clone(), v.clone());
770 21 : }
771 0 : }
772 : }
773 237 : }
774 248 : Ok(Some(TemporalOutcome {
775 248 : rows: docs,
776 248 : paged,
777 248 : total,
778 248 : aggregated,
779 248 : }))
780 249 : }
781 249 : .await;
782 249 : match first? {
783 248 : Some(out) => Ok(out),
784 : // the fallback re-runs the same statement without the pushed
785 : // aggregation; boxed because an async fn cannot recurse inline
786 1 : None => Box::pin(self.query_inner(tenant, f, false)).await,
787 : }
788 249 : }
789 :
790 : /// Single-entity fetch with the same instance pruning (Retrieve
791 : /// Temporal Evolution, 5.7.3). `None` = entity absent.
792 129 : pub async fn get_range(
793 129 : &self,
794 129 : tenant: &TenantId,
795 129 : id: &str,
796 129 : f: &TemporalFilter<'_>,
797 129 : ) -> Result<Option<Value>, sqlx::Error> {
798 : // Same unpruned fallback as `query_inner`, and compiling for the
799 : // same reason: no range and no lastN is the shape that never refuses.
800 : #[allow(clippy::expect_used)]
801 129 : let (attr_expr, binds) = match attr_object_expr(f, 3) {
802 129 : Some(v) => v,
803 0 : None => attr_object_expr(&TemporalFilter::default(), 3)
804 0 : .expect("no range/lastN always compiles"),
805 : };
806 : // 4.22: an expired entity is invalid → None (404), same as get().
807 129 : let sql = format!(
808 : "SELECT m.meta || {attr_expr} FROM temporal_entities m \
809 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED}"
810 : );
811 129 : let mut tx = super::begin(&self.pool).await?;
812 129 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
813 129 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
814 129 : .bind(tenant.as_str())
815 129 : .bind(id);
816 129 : for b in &binds {
817 34 : qy = qy.bind(b);
818 34 : }
819 129 : let row = qy.fetch_optional(&mut *tx).await?;
820 129 : tx.commit().await?;
821 129 : Ok(row.map(|r| r.get::<Value, _>(0)))
822 129 : }
823 :
824 6 : pub async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, sqlx::Error> {
825 6 : Ok(self.query(tenant, &TemporalFilter::default()).await?.rows)
826 6 : }
827 :
828 : /// Row-locked read-modify-write over the RECONSTRUCTED doc, written back
829 : /// as a DELTA: rows are diffed by (attr, instanceId) and only moved,
830 : /// changed or removed instances touch the table — never a full-history
831 : /// rewrite (the old resync's O(history) write amplification, and the
832 : /// thing that fought hypertable compression).
833 16 : pub async fn mutate<T, E>(
834 16 : &self,
835 16 : tenant: &TenantId,
836 16 : id: &str,
837 16 : f: impl FnOnce(&mut Value) -> Result<T, E>,
838 16 : ) -> Result<Option<Result<T, E>>, sqlx::Error> {
839 16 : let mut tx = super::begin(&self.pool).await?;
840 16 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
841 : // the meta row is the serialization point (FOR UPDATE). 4.22
842 : // qualifies it: 5.6.12 to 5.6.15 reach the history through here,
843 : // and none of them may modify an entity every read calls absent.
844 16 : let row = sqlx::query(sqlx::AssertSqlSafe(format!(
845 16 : "SELECT m.meta FROM temporal_entities m \
846 16 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED} FOR UPDATE"
847 16 : )))
848 16 : .bind(tenant.as_str())
849 16 : .bind(id)
850 16 : .fetch_optional(&mut *tx)
851 16 : .await?;
852 16 : let Some(row) = row else {
853 5 : tx.commit().await?;
854 5 : return Ok(None);
855 : };
856 11 : let meta: Value = row.get(0);
857 : // reconstruct the full doc inside the same transaction
858 11 : let attrs_row = sqlx::query(
859 11 : "SELECT COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (
860 11 : SELECT ai.attr_id,
861 11 : jsonb_agg(ai.data ORDER BY ai.created_at, ai.observed_at, ai.instance_id) AS insts
862 11 : FROM attr_instances ai
863 11 : WHERE ai.tenant_id = $1 AND ai.entity_id = $2
864 11 : GROUP BY ai.attr_id) g), '{}'::jsonb)",
865 11 : )
866 11 : .bind(tenant.as_str())
867 11 : .bind(id)
868 11 : .fetch_one(&mut *tx)
869 11 : .await?;
870 11 : let attrs: Value = attrs_row.get(0);
871 11 : let mut doc = meta;
872 11 : if let (Some(d), Some(a)) = (doc.as_object_mut(), attrs.as_object()) {
873 11 : for (k, v) in a {
874 9 : d.insert(k.clone(), v.clone());
875 9 : }
876 0 : }
877 11 : let before_rows = decompose(&doc);
878 11 : match f(&mut doc) {
879 9 : Ok(t) => {
880 9 : let (types, scopes, _created, modified) = extract(&doc);
881 9 : let after_rows = decompose(&doc);
882 : // diff by logical identity (attr, instanceId); a changed
883 : // observed_at moves the physical row → delete + insert
884 54 : let key = |r: &Value| -> (String, String) {
885 54 : (
886 54 : r["attr_id"].as_str().unwrap_or("").to_owned(),
887 54 : r["instance_id"].as_str().unwrap_or("").to_owned(),
888 54 : )
889 54 : };
890 9 : let old: std::collections::HashMap<(String, String), &Value> =
891 14 : before_rows.iter().map(|r| (key(r), r)).collect();
892 9 : let new: std::collections::HashMap<(String, String), &Value> =
893 20 : after_rows.iter().map(|r| (key(r), r)).collect();
894 9 : let mut deletes: Vec<Value> = Vec::new();
895 14 : for (k, o) in &old {
896 14 : match new.get(k) {
897 0 : None => deletes.push(serde_json::json!({"a": k.0, "i": k.1})),
898 14 : Some(n) if n["observed_at"] != o["observed_at"] => {
899 0 : deletes.push(serde_json::json!({"a": k.0, "i": k.1}))
900 : }
901 14 : _ => {}
902 : }
903 : }
904 9 : let upserts: Vec<Value> = after_rows
905 9 : .iter()
906 20 : .filter(|r| old.get(&key(r)).is_none_or(|o| *o != *r))
907 9 : .cloned()
908 9 : .collect();
909 9 : if !deletes.is_empty() {
910 0 : sqlx::query(
911 0 : "DELETE FROM attr_instances t
912 0 : USING jsonb_array_elements($3::jsonb) AS e
913 0 : WHERE t.tenant_id = $1 AND t.entity_id = $2
914 0 : AND t.attr_id = e->>'a' AND t.instance_id = e->>'i'",
915 0 : )
916 0 : .bind(tenant.as_str())
917 0 : .bind(id)
918 0 : .bind(Value::Array(deletes))
919 0 : .execute(&mut *tx)
920 0 : .await?;
921 9 : }
922 9 : insert_rows(&mut tx, tenant, id, upserts).await?;
923 9 : sqlx::query(
924 9 : "UPDATE temporal_entities SET meta = $3, types = $4, scopes = $5,
925 9 : modified_at = $6::timestamptz
926 9 : WHERE tenant_id = $1 AND id = $2",
927 9 : )
928 9 : .bind(tenant.as_str())
929 9 : .bind(id)
930 9 : .bind(meta_of(&doc))
931 9 : .bind(&types)
932 9 : .bind(&scopes)
933 9 : .bind(&modified)
934 9 : .execute(&mut *tx)
935 9 : .await?;
936 9 : tx.commit().await?;
937 9 : Ok(Some(Ok(t)))
938 : }
939 2 : Err(e) => {
940 2 : tx.rollback().await?;
941 2 : Ok(Some(Err(e)))
942 : }
943 : }
944 16 : }
945 : }
946 :
947 : #[cfg(test)]
948 : mod tests {
949 : use super::*;
950 : use crate::compile::temporal::InstanceRange;
951 :
952 : /// Two callers fall back to `attr_object_expr(&TemporalFilter::default())`
953 : /// when a range is outside the compiler's exact subset, and unwrap it:
954 : /// the default filter has no range and no `lastN`, and `None` is returned
955 : /// only for a range the compiler refuses. That is the reconstruct-unpruned
956 : /// path, so if it ever stopped compiling the broker would panic on
957 : /// precisely the queries the fallback exists to serve. Pinned here rather
958 : /// than argued in a comment.
959 : #[test]
960 2 : fn the_unpruned_fallback_always_compiles() {
961 2 : let f = TemporalFilter::default();
962 2 : assert!(f.range.is_none(), "the default filter carries no range");
963 2 : assert!(f.last_n.is_none(), "the default filter carries no lastN");
964 8 : for first_bind in [1, 3, 4, 17] {
965 8 : let built = attr_object_expr(&f, first_bind);
966 8 : let (sql, binds) = built.unwrap_or_else(|| {
967 0 : panic!("the unpruned fallback failed to compile at bind {first_bind}")
968 : });
969 8 : assert!(!sql.is_empty(), "empty SQL at bind {first_bind}");
970 8 : assert!(
971 8 : !sql.contains('$'),
972 : "the unpruned fragment names no placeholder: {sql}"
973 : );
974 8 : assert!(
975 8 : binds.is_empty(),
976 : "a fragment that names no placeholder must hand back no bind: {binds:?}"
977 : );
978 : }
979 2 : }
980 :
981 2 : fn between<'a>() -> TemporalFilter<'a> {
982 2 : TemporalFilter {
983 2 : range: Some(InstanceRange {
984 2 : timerel: "between",
985 2 : time_at: "2026-01-01T00:00:00Z",
986 2 : end_time_at: Some("2026-02-01T00:00:00Z"),
987 2 : timeproperty: "observedAt",
988 2 : }),
989 2 : ..Default::default()
990 2 : }
991 2 : }
992 :
993 : /// The temporal tables extract their four columns with the same function
994 : /// the entity table uses, so what holds there holds here: `type` and
995 : /// `scope` read as arrays whichever shape the document spells them in,
996 : /// and 4.6.3's comma seconds-fraction is canonicalised. A stamp that
997 : /// reaches a `::timestamptz` bind with its comma intact is rejected by
998 : /// Postgres, and the append fails on a document the API accepted.
999 : #[test]
1000 2 : fn extract_canonicalises_the_stamps_and_takes_type_in_either_shape() {
1001 2 : let (types, scopes, created, modified) = extract(&serde_json::json!({
1002 2 : "id": "urn:x", "type": "T", "scope": ["/a", "/b"],
1003 2 : "createdAt": "2026-01-01T00:00:00,500Z",
1004 2 : "modifiedAt": "2026-01-01T00:00:01.500Z"}));
1005 2 : assert_eq!(types, ["T"]);
1006 2 : assert_eq!(scopes.expect("scopes"), ["/a", "/b"]);
1007 2 : assert_eq!(
1008 : created, "2026-01-01T00:00:00.500Z",
1009 : "the comma is the fraction"
1010 : );
1011 2 : assert_eq!(modified, "2026-01-01T00:00:01.500Z");
1012 2 : let (types, scopes, created, _) = extract(&serde_json::json!({"id": "urn:x"}));
1013 2 : assert!(types.is_empty(), "no type is no types, never a panic");
1014 2 : assert!(
1015 2 : scopes.is_none(),
1016 : "an absent scope stays absent, not an empty list"
1017 : );
1018 2 : assert_eq!(
1019 : created, "1970-01-01T00:00:00Z",
1020 : "the belt stamp is usable as a bind"
1021 : );
1022 2 : }
1023 :
1024 : #[test]
1025 2 : fn meta_of_keeps_only_meta_members() {
1026 2 : let doc = serde_json::json!({
1027 2 : "id": "urn:x", "type": ["T"], "createdAt": "c", "modifiedAt": "m",
1028 2 : "https://a/attr": [{"instanceId": "i1"}]
1029 : });
1030 2 : let m = meta_of(&doc);
1031 2 : assert!(m.get("id").is_some() && m.get("https://a/attr").is_none());
1032 2 : }
1033 :
1034 : #[test]
1035 2 : fn range_binds_are_numbered_from_first_bind() {
1036 2 : let (expr, binds) = attr_object_expr(&between(), 2).expect("compiles");
1037 2 : assert_eq!(
1038 : binds,
1039 2 : vec!["observedAt", "2026-01-01T00:00:00Z", "2026-02-01T00:00:00Z"]
1040 : );
1041 2 : assert!(
1042 2 : expr.contains("$2") && expr.contains("$3") && expr.contains("$4"),
1043 : "{expr}"
1044 : );
1045 : // predicates run on the instance JSON, never the partition column
1046 2 : assert!(expr.contains("ai.data ->>"), "{expr}");
1047 2 : }
1048 :
1049 : #[test]
1050 2 : fn last_n_caps_with_rank_per_attr_and_dataset_never_row_number() {
1051 2 : let f = TemporalFilter {
1052 2 : last_n: Some(5),
1053 2 : ..Default::default()
1054 2 : };
1055 2 : let (expr, binds) = attr_object_expr(&f, 1).expect("compiles");
1056 : // RANK keeps timestamp ties; ROW_NUMBER would cut an instance the
1057 : // API-side per-attr lastN still wants (the tie-break divergence bug)
1058 2 : assert!(expr.contains("rank() OVER"), "{expr}");
1059 2 : assert!(!expr.contains("row_number"), "{expr}");
1060 2 : assert!(
1061 2 : expr.contains("PARTITION BY ai.attr_id, ai.data ->> 'datasetId'"),
1062 : "{expr}"
1063 : );
1064 2 : assert!(expr.contains("COLLATE \"C\" DESC NULLS LAST"), "{expr}");
1065 2 : assert_eq!(binds, vec!["observedAt", "5"]);
1066 2 : }
1067 :
1068 : #[test]
1069 2 : fn refused_range_shape_refuses_the_pruning() {
1070 2 : let f = TemporalFilter {
1071 2 : range: Some(InstanceRange {
1072 2 : timerel: "since", // not a 4.11 relation
1073 2 : time_at: "t",
1074 2 : end_time_at: None,
1075 2 : timeproperty: "observedAt",
1076 2 : }),
1077 2 : last_n: Some(3),
1078 2 : ..Default::default()
1079 2 : };
1080 : // half-pruning would silently skip the range: refuse instead
1081 2 : assert!(attr_object_expr(&f, 1).is_none());
1082 2 : }
1083 :
1084 : #[test]
1085 2 : fn decompose_never_parks_deletion_instances_on_the_epoch() {
1086 2 : let doc = serde_json::json!({
1087 2 : "id": "urn:x",
1088 2 : "https://a/attr": [{
1089 2 : "type": "Property", "value": "urn:ngsi-ld:null",
1090 2 : "instanceId": "urn:ngsi-ld:Instance:d1",
1091 2 : "deletedAt": "2026-08-08T10:00:00Z"
1092 : }]
1093 : });
1094 2 : let rows = decompose(&doc);
1095 2 : assert_eq!(rows.len(), 1);
1096 2 : assert_eq!(rows[0]["observed_at"], "2026-08-08T10:00:00Z");
1097 2 : }
1098 : }
|
