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, instance_id)` index (0017).
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 : keys: Keys,
25 : }
26 :
27 : /// The integer history keys' process state (docs/src/storage.md
28 : /// "Integer history keys").
29 : #[derive(Default)]
30 : struct Keys {
31 : /// `history_cutover`, cached once set: only a migration sets it, but a
32 : /// pod that started before a rolling deploy's migration sees it unset, so
33 : /// an unset value is read again on every write until it is set.
34 : /// A test store pins it with `PgTemporalStore::with_cutover`.
35 : cutover: tokio::sync::OnceCell<Option<chrono::DateTime<chrono::Utc>>>,
36 : /// (Tenant, IRI) pairs a committed `iri_ids` row holds. A write sends the
37 : /// dictionary insert only for an IRI missing here, so a warm process
38 : /// sends no extra statement. An IRI this process inserted is added only
39 : /// when a later write finds it stored: the inserting transaction may still
40 : /// roll back. The ids themselves come from `iri_no` inside each
41 : /// statement, so a stale entry fails a write (23502) instead of keying a
42 : /// row by an id that names nothing.
43 : known: std::sync::Mutex<Known>,
44 : /// Which writes keep short history: their rows go to the
45 : /// unlogged child of `attr_rows`, whatever the cut-over.
46 : short: std::sync::RwLock<std::sync::Arc<antares_store::ShortHistory>>,
47 : }
48 :
49 : /// Pairs one generation of [`Known`] holds before it ages. Two generations
50 : /// hold at most twice this: about 150 B per pair (two `String`s with a
51 : /// 60-byte IRI, the table slot), so ~30 MB per process at the bound.
52 : const KNOWN_GEN: usize = 100_000;
53 :
54 : /// The known (Tenant, IRI) set, bounded as a two-generation LRU: a pair
55 : /// found in `old` moves to `new`; when `new` is full it becomes `old` and
56 : /// the previous `old` is dropped. A working set under [`KNOWN_GEN`] pairs
57 : /// never misses; a pair aged out costs its next write one
58 : /// `INSERT ... ON CONFLICT DO NOTHING` round trip, after which it is known
59 : /// again.
60 : #[derive(Default)]
61 : struct Known {
62 : new: std::collections::HashSet<(String, String)>,
63 : old: std::collections::HashSet<(String, String)>,
64 : }
65 :
66 : impl Known {
67 184 : fn contains(&mut self, k: &(String, String)) -> bool {
68 184 : if self.new.contains(k) {
69 52 : return true;
70 132 : }
71 132 : match self.old.take(k) {
72 4 : Some(k) => {
73 4 : self.insert(k);
74 4 : true
75 : }
76 128 : None => false,
77 : }
78 184 : }
79 :
80 400018 : fn insert(&mut self, k: (String, String)) {
81 400018 : if self.new.len() >= KNOWN_GEN {
82 4 : self.old = std::mem::take(&mut self.new);
83 400014 : }
84 400018 : self.new.insert(k);
85 400018 : }
86 :
87 28 : fn retain(&mut self, mut keep: impl FnMut(&(String, String)) -> bool) {
88 28 : self.new.retain(&mut keep);
89 28 : self.old.retain(keep);
90 28 : }
91 : }
92 :
93 : impl Keys {
94 83 : fn known(&self) -> std::sync::MutexGuard<'_, Known> {
95 83 : self.known
96 83 : .lock()
97 83 : .unwrap_or_else(std::sync::PoisonError::into_inner)
98 83 : }
99 :
100 4814 : async fn cutover(
101 4814 : &self,
102 4814 : tx: &mut sqlx::PgConnection,
103 4814 : ) -> Result<Option<chrono::DateTime<chrono::Utc>>, sqlx::Error> {
104 4814 : if let Some(at) = self.cutover.get() {
105 28 : return Ok(*at);
106 4786 : }
107 4786 : let at = sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
108 4786 : "SELECT at FROM history_cutover",
109 4786 : )
110 4786 : .fetch_optional(&mut *tx)
111 4786 : .await
112 4786 : .map(Option::flatten)?;
113 4786 : if at.is_some() {
114 1 : // a racing writer may have set it first: both read the same row
115 1 : let _ = self.cutover.set(at);
116 4785 : }
117 4786 : Ok(at)
118 4814 : }
119 :
120 : /// Writes the `iri_ids` rows of `iris` the dictionary lacks, inside the
121 : /// caller's transaction (RLS checks its Tenant), in IRI order so two
122 : /// writers wait on each other's first sights in one order. `ensured`
123 : /// holds what this transaction already wrote or found.
124 41 : async fn ensure(
125 41 : &self,
126 41 : tx: &mut sqlx::PgConnection,
127 41 : tenant: &TenantId,
128 41 : mut iris: Vec<String>,
129 41 : ensured: &mut std::collections::HashSet<(String, String)>,
130 41 : ) -> Result<(), sqlx::Error> {
131 : {
132 41 : let mut known = self.known();
133 176 : iris.retain(|i| {
134 176 : let k = (tenant.as_str().to_owned(), i.clone());
135 176 : !known.contains(&k) && !ensured.contains(&k)
136 176 : });
137 : }
138 41 : if iris.is_empty() {
139 25 : return Ok(());
140 16 : }
141 16 : iris.sort_unstable();
142 16 : iris.dedup();
143 16 : let fresh: std::collections::HashSet<String> = sqlx::query_scalar(
144 16 : "INSERT INTO iri_ids (tenant_id, iri) SELECT $1, unnest($2::text[])
145 16 : ON CONFLICT DO NOTHING RETURNING iri",
146 16 : )
147 16 : .bind(tenant.as_str())
148 16 : .bind(&iris)
149 16 : .fetch_all(&mut *tx)
150 16 : .await?
151 16 : .into_iter()
152 16 : .collect();
153 16 : let mut known = self.known();
154 32 : for i in iris {
155 32 : let k = (tenant.as_str().to_owned(), i);
156 32 : if !fresh.contains(&k.1) {
157 12 : known.insert(k.clone());
158 20 : }
159 32 : ensured.insert(k);
160 : }
161 16 : Ok(())
162 41 : }
163 :
164 : /// Drops what this process knows of `tenant`'s dictionary: its purge, or
165 : /// a keyed write that found an IRI gone (another pod's purge).
166 26 : fn forget(&self, tenant: &TenantId) {
167 26 : self.known().retain(|(t, _)| t != tenant.as_str());
168 26 : }
169 :
170 : /// Whether a write of an Entity of `types` whose current state is
171 : /// `shell` keeps short history.
172 5013 : fn is_short(&self, tenant: &TenantId, types: &[String], shell: &Value) -> bool {
173 5013 : self.short
174 5013 : .read()
175 5013 : .unwrap_or_else(std::sync::PoisonError::into_inner)
176 5013 : .is_short(
177 5013 : tenant.as_str(),
178 5013 : types,
179 5013 : shell.get(antares_store::HISTORY_RETENTION).is_some(),
180 5013 : )
181 5013 : }
182 : }
183 :
184 : /// The dictionary IRIs a write of `doc` for Entity `id` keys its rows by:
185 : /// the Entity id and its Attribute names.
186 2 : fn iris_of(id: &str, doc: &Value) -> Vec<String> {
187 2 : let mut out = vec![id.to_owned()];
188 2 : if let Some(o) = doc.as_object() {
189 2 : out.extend(
190 2 : o.iter()
191 10 : .filter(|(k, v)| v.is_array() && !DOC_META.contains(&k.as_str()))
192 2 : .map(|(k, _)| k.clone()),
193 : );
194 0 : }
195 2 : out
196 2 : }
197 :
198 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
199 : /// certain Entity, Property or Relationship shall become invalid" — an
200 : /// expired temporal entity is absent from every read, ahead of the retention
201 : /// sweep. One shared literal so `query`, its count fallback and `get_range`
202 : /// can never disagree.
203 : ///
204 : /// The stamp is jsonb TEXT, so it goes through `try_timestamptz`
205 : /// (0001_init.sql): a bare cast RAISES on anything it cannot parse, and
206 : /// these reads are
207 : /// tenant-wide — one bad stamp would take down the whole tenant's temporal
208 : /// API rather than hide one entity. An unusable stamp reads as no expiry, the
209 : /// same direction the memory arm takes in `filter::expired_at`.
210 : const NOT_EXPIRED: &str =
211 : "(try_timestamptz(m.meta->>'expiresAt') IS NULL OR try_timestamptz(m.meta->>'expiresAt') > now())";
212 :
213 : /// The temporal tables carry the same four extracted columns as `entities`,
214 : /// so they are read by the same function (`super::entity::types_scopes_stamps`)
215 : /// rather than a second copy of it.
216 : use super::entity::types_scopes_stamps as extract;
217 :
218 : /// Entity-doc members that are NOT temporal attributes. `scope` may itself be
219 : /// instance-shaped (deletion instances, 020_19/20) — it lives in `meta`
220 : /// verbatim either way, exactly as the bridge kept it unpruned.
221 : const DOC_META: &[&str] = antares_model::ENTITY_META_KEYS;
222 :
223 : /// The meta an append leaves on a `temporal_entities` row that exists: the
224 : /// shell's (`EXCLUDED.meta`), but a stored `scope` that holds instances —
225 : /// its recorded history, which `push_scope` keeps in `meta` — stays.
226 : const KEPT_META: &str = "CASE WHEN jsonb_typeof(temporal_entities.meta->'scope'->0) = 'object'
227 : THEN EXCLUDED.meta || jsonb_build_object('scope', temporal_entities.meta->'scope')
228 : ELSE EXCLUDED.meta END";
229 :
230 : /// The meta-only document stored in `temporal_entities.meta`.
231 3040 : fn meta_of(doc: &Value) -> Value {
232 3040 : let mut m = serde_json::Map::new();
233 3040 : if let Some(obj) = doc.as_object() {
234 24320 : for k in DOC_META {
235 24320 : if let Some(v) = obj.get(*k) {
236 11617 : m.insert((*k).to_owned(), v.clone());
237 12703 : }
238 : }
239 0 : }
240 3040 : Value::Object(m)
241 3040 : }
242 :
243 : /// Decompose a doc's attribute arrays into row JSON for the multi-row
244 : /// INSERT. The `observed_at` string is derived ONLY from the instance
245 : /// document, so decomposing a reconstructed doc reproduces byte-identical
246 : /// keys (what the mutate diff relies on).
247 3332 : fn decompose(doc: &Value) -> Vec<Value> {
248 3332 : let mut rows: Vec<Value> = Vec::new();
249 3332 : if let Some(obj) = doc.as_object() {
250 11766 : for (attr, instances) in obj {
251 11766 : if DOC_META.contains(&attr.as_str()) {
252 4808 : continue;
253 6958 : }
254 6958 : let Some(arr) = instances.as_array() else {
255 38 : continue;
256 : };
257 17151 : for i in arr {
258 126271 : let s = |k: &str| i.get(k).and_then(Value::as_str);
259 : // 4.6.3: a comma seconds-fraction is legal in a request and
260 : // these stamps go straight into `::timestamptz` casts, which
261 : // refuse it — and the raise lands in the temporal drain,
262 : // which absorbs it, so the whole request's history would be
263 : // lost with a 2xx already returned. Timestamps only:
264 : // `datasetId` is a URI, where a comma is an ordinary
265 : // character and rewriting it would corrupt the id.
266 91969 : let ts = |k: &str| s(k).map(antares_store::filter::canonical_datetime);
267 17151 : let Some(instance_id) = s("instanceId") else {
268 0 : continue; // stamped by the API layer; belt only
269 : };
270 : // observed_at falls back through the instance's own
271 : // timestamps — deletion instances carry only deletedAt and
272 : // must NOT collapse onto the epoch (retention would reap them)
273 17151 : let observed = ts("observedAt")
274 17151 : .or_else(|| ts("modifiedAt"))
275 17151 : .or_else(|| ts("deletedAt"))
276 17151 : .or_else(|| ts("createdAt"))
277 17151 : .unwrap_or(std::borrow::Cow::Borrowed("1970-01-01T00:00:00Z"));
278 : // Provenance is a column, never data.
279 17151 : let mut data = i.clone();
280 17151 : let provenance = antares_store::take_provenance(&mut data);
281 : // the `instance_id` column holds it: ~70 B a row
282 17151 : if let Some(o) = data.as_object_mut() {
283 17151 : o.remove("instanceId");
284 17151 : }
285 17151 : rows.push(serde_json::json!({
286 17151 : "attr_id": attr,
287 17151 : "instance_id": instance_id,
288 17151 : "dataset_id": s("datasetId"),
289 17151 : "observed_at": observed,
290 17151 : "created_at": ts("createdAt").unwrap_or_else(|| observed.clone()),
291 17151 : "modified_at": ts("modifiedAt").unwrap_or(observed),
292 17151 : "deleted_at": ts("deletedAt"),
293 : // 4.22: the reap's index key (migration 0030); a stamp
294 : // chrono cannot read stays NULL there, and the daily
295 : // scan of `data` reaps it
296 17151 : "expires_at": ts("expiresAt")
297 17151 : .and_then(|t| chrono::DateTime::parse_from_rfc3339(&t).ok())
298 17151 : .map(|t| t.to_rfc3339()),
299 17151 : "data": data,
300 17151 : "provenance": provenance,
301 : }));
302 : }
303 : }
304 0 : }
305 3332 : rows
306 3332 : }
307 :
308 : /// One attribute's instances of history rows `alias`, as the reads group
309 : /// them: a row stores its instanceId in the `instance_id` column
310 : /// only, so an attribute comes back as `[instances, instance ids]` in one
311 : /// order, which [`unpack`] zips into the instances. A row written before
312 : /// the compact layout still holds `instanceId` in `data`, equal to the column. Building
313 : /// each row in SQL instead (`jsonb_build_array` or `||` per row) cost
314 : /// Postgres 10-35 % more than reading the old documents (240k rows).
315 1666 : fn insts_and_ids(alias: &str) -> String {
316 1666 : let key = format!("{alias}.created_at, {alias}.observed_at, {alias}.instance_id");
317 1666 : format!(
318 : "jsonb_build_array(jsonb_agg({alias}.data ORDER BY {key}), \
319 : to_jsonb(array_agg({alias}.instance_id ORDER BY {key})))"
320 : )
321 1666 : }
322 :
323 : /// The whole instance document of row `alias`, built in SQL, where a row is
324 : /// copied (attr_versions keeps whole documents) or read rarely. The column
325 : /// equals `data`'s own instanceId on a row written.
326 2945 : fn restamp(alias: &str) -> String {
327 2945 : format!("({alias}.data || jsonb_build_object('instanceId', {alias}.instance_id))")
328 2945 : }
329 :
330 : /// The history rows of one Entity over both parents (docs/src/storage.md
331 : /// "Integer history keys"): `attr_instances` and, in its row type, `attr_rows`
332 : /// (`attr_rows_of`, migration 0020), which stays empty until the cut-over. The
333 : /// caller's predicates reach each arm's btree; an ordered read with a LIMIT
334 : /// writes its arms out instead, each limited itself.
335 1640 : fn history(tenant: &str, entity: &str) -> String {
336 1640 : format!(
337 : "(SELECT * FROM attr_instances WHERE tenant_id = {tenant} AND entity_id = {entity} \
338 : UNION ALL SELECT * FROM attr_rows_of({tenant}, {entity}))"
339 : )
340 1640 : }
341 :
342 : /// The Entity `m` has a history row passing `preds` (on alias `ai`): the
343 : /// page qualification. Each parent is probed on its own along its btree,
344 : /// `dir` from the end the window opens towards, so the first row read is
345 : /// most likely in the window. An EXISTS over `history` read each Entity's
346 : /// rows oldest first from the widened column bound and fetched every row the
347 : /// text predicate then refused: ~115 rows an Entity on the bike-share
348 : /// history, past the 30 s statement timeout for 577 Entities.
349 : /// `attrs`, the bind of the Attribute IRIs a policy shows, keeps the probe to
350 : /// their rows.
351 251 : pub fn has_history(preds: &str, dir: &str, attrs: Option<usize>) -> String {
352 251 : let (x, r) = attrs.map_or_else(Default::default, |n| {
353 9 : (
354 9 : format!(" AND ai.attr_id = ANY(${n})"),
355 9 : format!(
356 9 : " AND ai.attr_no IN (SELECT i.id FROM iri_ids i \
357 9 : WHERE i.tenant_id = m.tenant_id AND i.iri = ANY(${n}))"
358 9 : ),
359 9 : )
360 9 : });
361 251 : format!(
362 : "((SELECT true FROM attr_instances ai \
363 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_id = m.id{x}{preds} \
364 : ORDER BY ai.attr_id {dir}, ai.observed_at {dir} LIMIT 1) IS NOT NULL \
365 : OR (SELECT true FROM attr_rows ai \
366 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_no = iri_no(m.tenant_id, m.id){r}{preds} \
367 : ORDER BY ai.attr_no {dir}, ai.observed_at {dir} LIMIT 1) IS NOT NULL)"
368 : )
369 251 : }
370 :
371 : /// Turns each attribute's `[instances, instance ids]` that
372 : /// [`insts_and_ids`] returned back into its instances.
373 1809 : fn unpack(doc: &mut Value) {
374 1809 : let Some(o) = doc.as_object_mut() else {
375 2 : return;
376 : };
377 8742 : for attr in o.values_mut() {
378 8742 : let Some([insts, ids]) = attr.as_array_mut().map(Vec::as_mut_slice) else {
379 5660 : continue;
380 : };
381 3082 : let (Some(list), Some(ids)) = (insts.as_array_mut(), ids.as_array_mut()) else {
382 29 : continue;
383 : };
384 3053 : if list.len() != ids.len() || !ids.iter().all(Value::is_string) {
385 4 : continue;
386 3049 : }
387 26022 : for (inst, id) in list.iter_mut().zip(ids.iter_mut()) {
388 26022 : if let Some(i) = inst.as_object_mut() {
389 26022 : i.insert("instanceId".to_owned(), id.take());
390 26022 : }
391 : }
392 3049 : let list = insts.take();
393 3049 : *attr = list;
394 : }
395 1809 : }
396 :
397 : /// `ANTARES_TEMPORAL_DEDUP=off` records a row per write again; anything else,
398 : /// or unset, records a new instance only (see [`unseen`]).
399 2118 : fn temporal_dedup() -> bool {
400 : static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
401 2118 : *ON.get_or_init(|| std::env::var("ANTARES_TEMPORAL_DEDUP").as_deref() != Ok("off"))
402 2118 : }
403 :
404 : /// An instance's stamps: two instances that differ only in these are the same
405 : /// observation (CIM 009 4.5: a static Property is one instance).
406 : const STAMPS: &str = "ARRAY['instanceId','createdAt','modifiedAt']";
407 :
408 : /// The provenance of a row `e` that corrects the stored one.
409 : const CORRECTED: &str = "jsonb_set(COALESCE(NULLIF(e->'provenance', 'null'::jsonb), '{}'::jsonb), '{operation}', '\"correct\"')";
410 :
411 : /// The geometry column of an instance row `e` (see `insert_rows`).
412 : const GEO_VALUE: &str = "CASE WHEN jsonb_typeof(e->'data'->'value') = 'object'
413 : AND e->'data'->'value'->>'type' IN
414 : ('Point','MultiPoint','LineString','MultiLineString',
415 : 'Polygon','MultiPolygon')
416 : THEN try_geomfromgeojson((e->'data'->'value')::text) END";
417 :
418 : /// Whether the newest stored instance of row `e`'s (attribute, datasetId) is
419 : /// another instance with the same value: one row read by the index
420 : /// (tenant_id, entity_id, attr_id, observed_at, …) read backwards, false when none is stored.
421 : /// The current and the last week first: unbounded, the newest row is
422 : /// a merge over every weekly partition, one cold index descent each. A row
423 : /// found there is the newest of all; only an attribute silent for longer
424 : /// reads the whole history, and COALESCE runs that probe for it alone.
425 : /// Each parent is its own ordered arm with its own LIMIT: Postgres
426 : /// merges UNION ALL arms in index order only when each arm is limited itself.
427 : const NEWEST_REPEATS: &str = "COALESCE((
428 : SELECT x.instance_id <> e->>'instance_id'
429 : AND x.data - ARRAY['instanceId','createdAt','modifiedAt']
430 : = (e->'data') - ARRAY['instanceId','createdAt','modifiedAt']
431 : FROM ((SELECT x.instance_id, x.data, x.observed_at FROM attr_instances x
432 : WHERE x.tenant_id = $1 AND x.entity_id = $2 AND x.attr_id = e->>'attr_id'
433 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
434 : AND x.observed_at >= date_trunc('week', now()) - interval '1 week'
435 : ORDER BY x.observed_at DESC LIMIT 1)
436 : UNION ALL
437 : (SELECT x.instance_id, x.data, x.observed_at FROM attr_rows_at($1, $2, e->>'attr_id') x
438 : WHERE x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
439 : AND x.observed_at >= date_trunc('week', now()) - interval '1 week'
440 : ORDER BY x.observed_at DESC LIMIT 1)) x
441 : ORDER BY x.observed_at DESC LIMIT 1), (
442 : SELECT x.instance_id <> e->>'instance_id'
443 : AND x.data - ARRAY['instanceId','createdAt','modifiedAt']
444 : = (e->'data') - ARRAY['instanceId','createdAt','modifiedAt']
445 : FROM ((SELECT x.instance_id, x.data, x.observed_at FROM attr_instances x
446 : WHERE x.tenant_id = $1 AND x.entity_id = $2 AND x.attr_id = e->>'attr_id'
447 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
448 : ORDER BY x.observed_at DESC LIMIT 1)
449 : UNION ALL
450 : (SELECT x.instance_id, x.data, x.observed_at FROM attr_rows_at($1, $2, e->>'attr_id') x
451 : WHERE x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
452 : ORDER BY x.observed_at DESC LIMIT 1)) x
453 : ORDER BY x.observed_at DESC LIMIT 1), false)";
454 :
455 : /// The earliest and latest `observed_at` of `rows`, bound beside every probe
456 : /// that matches a row by its `observed_at`. The equality reaches
457 : /// the partition key only through a join, so a generic plan reads every
458 : /// weekly partition of the Entity; a bound parameter prunes them to the weeks
459 : /// the batch writes before the statement runs. A stamp chrono cannot read
460 : /// leaves the range open, as the cast in the statement judges it.
461 2925 : fn observed_range(rows: &[Value]) -> (String, String) {
462 2925 : let mut range: Option<(
463 2925 : chrono::DateTime<chrono::FixedOffset>,
464 2925 : chrono::DateTime<chrono::FixedOffset>,
465 2925 : )> = None;
466 14227 : for r in rows {
467 14227 : let Some(t) = r["observed_at"]
468 14227 : .as_str()
469 14227 : .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
470 : else {
471 0 : return ("-infinity".into(), "infinity".into());
472 : };
473 14227 : range = Some(match range {
474 2925 : None => (t, t),
475 11302 : Some((lo, hi)) => (lo.min(t), hi.max(t)),
476 : });
477 : }
478 2925 : match range {
479 2925 : Some((lo, hi)) => (lo.to_rfc3339(), hi.to_rfc3339()),
480 0 : None => ("-infinity".into(), "infinity".into()),
481 : }
482 2925 : }
483 :
484 : /// The rows of one append that are new observations: of rows with an
485 : /// `observedAt`, one per (attribute, datasetId, observedAt), and of rows
486 : /// without, each one that differs but for its stamps from the row kept
487 : /// before it in its (attribute, datasetId) — the rule NEWEST_REPEATS applies
488 : /// against the table, so a value that comes back after another is recorded.
489 : /// A row kept after another of its series is marked `follows`: its
490 : /// predecessor is that row, never the table's newest, and `insert_rows`
491 : /// holds only a series' first row against what the table already holds.
492 2122 : fn unseen(rows: Vec<Value>) -> Vec<Value> {
493 2122 : let mut observed = std::collections::HashSet::new();
494 2122 : let mut last: std::collections::HashMap<(String, String), Value> =
495 2122 : std::collections::HashMap::new();
496 2122 : rows.into_iter()
497 4759 : .filter_map(|mut r| {
498 19036 : let field = |k: &str| r.get(k).cloned().unwrap_or(Value::Null);
499 4759 : let series = (
500 4759 : field("attr_id").to_string(),
501 4759 : field("dataset_id").to_string(),
502 4759 : );
503 4759 : let at = field("observed_at").to_string();
504 4759 : let mut d = field("data");
505 4759 : let has_observed_at = d.get("observedAt").is_some();
506 4759 : if let Some(o) = d.as_object_mut() {
507 14277 : for k in ["instanceId", "createdAt", "modifiedAt"] {
508 14277 : o.remove(k);
509 14277 : }
510 0 : }
511 4759 : if has_observed_at {
512 963 : if !observed.insert((series.0.clone(), series.1.clone(), at)) {
513 2 : return None;
514 961 : }
515 3796 : } else if last.get(&series) == Some(&d) {
516 2 : return None;
517 3794 : }
518 4755 : if last.insert(series, d).is_some() {
519 19 : if let Some(o) = r.as_object_mut() {
520 19 : o.insert("follows".into(), Value::Bool(true));
521 19 : }
522 4736 : }
523 4755 : Some(r)
524 4759 : })
525 2122 : .collect()
526 2122 : }
527 :
528 : /// How [`insert_rows`] writes one batch: `dedup` records new observations
529 : /// only, `short` writes every row to the unlogged child.
530 : #[derive(Clone, Copy)]
531 : struct Route {
532 : dedup: bool,
533 : short: bool,
534 : }
535 :
536 : /// Multi-row instance upsert inside the caller's transaction. With `dedup`
537 : /// (the auto-recording append) a row is written only when it is a new
538 : /// observation: an instance with an `observedAt` already stored for
539 : /// its (attribute, datasetId) is updated in place, keeping its instanceId,
540 : /// and one without is skipped when the newest stored instance holds the same
541 : /// value. tenant_id is part of every comparison.
542 : ///
543 : /// A row observed from the cut-over on goes to `attr_rows`, any
544 : /// earlier one to `attr_instances`, a late write included: one observation
545 : /// always meets its stored instance in the same parent. Every row of a
546 : /// `short` write goes to `attr_rows` as `short`, its unlogged child.
547 3028 : async fn insert_rows(
548 3028 : tx: &mut sqlx::PgConnection,
549 3028 : keys: &Keys,
550 3028 : ensured: &mut std::collections::HashSet<(String, String)>,
551 3028 : tenant: &TenantId,
552 3028 : entity_id: &str,
553 3028 : rows: Vec<Value>,
554 3028 : Route { dedup, short }: Route,
555 3028 : ) -> Result<(), sqlx::Error> {
556 3028 : let dedup = dedup && temporal_dedup();
557 3028 : let rows = if dedup { unseen(rows) } else { rows };
558 3028 : if rows.is_empty() {
559 104 : return Ok(());
560 2924 : }
561 : // The Attributes given an instance in a dataset (migration 0009): lastN
562 : // keeps every row of these, so the read needs to know them.
563 2924 : let mut in_dataset: Vec<String> = rows
564 2924 : .iter()
565 14227 : .filter(|r| r["dataset_id"].is_string())
566 2924 : .filter_map(|r| r["attr_id"].as_str().map(str::to_owned))
567 2924 : .collect();
568 2924 : in_dataset.sort_unstable();
569 2924 : in_dataset.dedup();
570 : // The newest instance whose observedAt the rank may order unlike
571 : // observed_at (migration 0028): lastN reads past its N newest only there.
572 2924 : let offset_until = rows
573 2924 : .iter()
574 14227 : .filter(|r| !utc_observed(&r["data"]))
575 4020 : .filter_map(|r| r["observed_at"].as_str())
576 4020 : .filter_map(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
577 2924 : .max();
578 2924 : let (keyed, text): (Vec<Value>, Vec<Value>) = if short {
579 27 : (rows, Vec::new())
580 : } else {
581 2897 : let cutover = keys.cutover(tx).await?;
582 14140 : rows.into_iter().partition(|r| is_keyed(r, cutover))
583 : };
584 2924 : if !text.is_empty() {
585 2885 : insert_text(tx, tenant, entity_id, text, dedup).await?;
586 39 : }
587 2924 : if !keyed.is_empty() {
588 40 : let mut iris = vec![entity_id.to_owned()];
589 40 : iris.extend(
590 40 : keyed
591 40 : .iter()
592 136 : .filter_map(|r| r["attr_id"].as_str().map(str::to_owned)),
593 : );
594 40 : keys.ensure(tx, tenant, iris, ensured).await?;
595 40 : if let Err(e) = insert_keyed(tx, tenant, entity_id, keyed, dedup, short).await {
596 : // an IRI the known set holds is gone: another pod purged the Tenant
597 0 : if matches!(&e, sqlx::Error::Database(d) if d.code().as_deref() == Some("23502")) {
598 0 : keys.forget(tenant);
599 0 : }
600 0 : return Err(e);
601 40 : }
602 2884 : }
603 2924 : if let Some(t) = offset_until {
604 : // Every caller writes the Entity's temporal row before its instances.
605 2107 : sqlx::query(
606 2107 : "UPDATE temporal_entities SET offset_until = $3
607 2107 : WHERE tenant_id = $1 AND id = $2 AND offset_until < $3",
608 2107 : )
609 2107 : .bind(tenant.as_str())
610 2107 : .bind(entity_id)
611 2107 : .bind(t)
612 2107 : .execute(&mut *tx)
613 2107 : .await?;
614 817 : }
615 2924 : if !in_dataset.is_empty() {
616 : // Every caller writes the Entity's temporal row before its instances.
617 394 : sqlx::query(
618 394 : "UPDATE temporal_entities
619 394 : SET dataset_attrs = ARRAY(SELECT DISTINCT x FROM unnest(dataset_attrs || $3) AS x ORDER BY x)
620 394 : WHERE tenant_id = $1 AND id = $2 AND NOT dataset_attrs @> $3",
621 394 : )
622 394 : .bind(tenant.as_str())
623 394 : .bind(entity_id)
624 394 : .bind(&in_dataset)
625 394 : .execute(&mut *tx)
626 394 : .await?;
627 2530 : }
628 2924 : Ok(())
629 3028 : }
630 :
631 : /// Whether an instance's `observedAt` is a UTC RFC 3339 stamp, whose text
632 : /// key (`dt_key`) orders exactly like its instant. Any other instance, one
633 : /// without observedAt included, may rank unlike `observed_at` in lastN.
634 14245 : fn utc_observed(data: &Value) -> bool {
635 14245 : data.get("observedAt")
636 14245 : .and_then(Value::as_str)
637 : // `dt_key` reads the first 19 bytes as the wall time: only the `T`
638 : // spelling of the separator orders with the rest
639 14245 : .filter(|t| t.ends_with('Z') && t.as_bytes().get(10) == Some(&b'T'))
640 14245 : .is_some_and(|t| {
641 10213 : chrono::DateTime::parse_from_rfc3339(&antares_store::filter::canonical_datetime(t))
642 10213 : .is_ok()
643 10213 : })
644 14245 : }
645 :
646 : /// Whether row `r` is observed at or after `cutover`. A stamp chrono cannot
647 : /// read stays in the text shape, where its cast judges it as before.
648 14150 : fn is_keyed(r: &Value, cutover: Option<chrono::DateTime<chrono::Utc>>) -> bool {
649 14150 : let Some(cut) = cutover else {
650 14062 : return false;
651 : };
652 88 : r["observed_at"]
653 88 : .as_str()
654 88 : .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
655 88 : .is_some_and(|t| t >= cut)
656 14150 : }
657 :
658 : /// The `attr_rows` key of an instance id and whether it is the broker's own:
659 : /// the uuid of `urn:ngsi-ld:Instance:<uuid>` in its canonical lowercase form,
660 : /// which the read side prints back as is; any other id keys by its uuid v5
661 : /// and stays in `data` as `instanceId` (docs/src/storage.md).
662 152 : fn instance_no(id: &str) -> (uuid::Uuid, bool) {
663 152 : let own = id.strip_prefix("urn:ngsi-ld:Instance:").and_then(|s| {
664 142 : uuid::Uuid::try_parse(s)
665 142 : .ok()
666 142 : .filter(|u| u.hyphenated().to_string() == s)
667 142 : });
668 152 : match own {
669 122 : Some(u) => (u, true),
670 30 : None => (
671 30 : uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, id.as_bytes()),
672 30 : false,
673 30 : ),
674 : }
675 152 : }
676 :
677 : /// [`insert_rows`] into `attr_instances`, the text shape.
678 2885 : async fn insert_text(
679 2885 : tx: &mut sqlx::PgConnection,
680 2885 : tenant: &TenantId,
681 2885 : entity_id: &str,
682 2885 : rows: Vec<Value>,
683 2885 : dedup: bool,
684 2885 : ) -> Result<(), sqlx::Error> {
685 2885 : let (lo, hi) = observed_range(&rows);
686 : // geo_value: extracted per instance when the value LOOKS like a GeoJSON
687 : // geometry; try_geomfromgeojson (0001_init.sql) maps anything PostGIS rejects to
688 : // NULL, which the S3 prefilter treats as "reaches the evaluator".
689 2885 : let rows = Value::Array(rows);
690 : // The stored instance of the same (attribute, datasetId) a row of `e` repeats.
691 : // Its tenant and entity predicates stay written in each statement, where the tenant check
692 : // (tests/attr_instances_tenant_predicate.rs) reads them.
693 2885 : let same = "x.attr_id = e->>'attr_id'
694 2885 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
695 2885 : AND x.instance_id <> e->>'instance_id'";
696 : // A row a write replaces with another value is a correction: the stored
697 : // version is kept in attr_versions before the row below takes the next
698 : // version number in place. Its two targets
699 : // are the two below: the same instanceId (the upsert's conflict), and with
700 : // `dedup` the same (attribute, datasetId, observedAt).
701 2885 : sqlx::query(sqlx::AssertSqlSafe(format!(
702 2885 : "INSERT INTO attr_versions
703 2885 : (tenant_id, entity_id, attr_id, instance_id, dataset_id, observed_at, version, data, provenance,
704 2885 : expires_at)
705 2885 : SELECT x.tenant_id, x.entity_id, x.attr_id, x.instance_id, x.dataset_id, x.observed_at,
706 2885 : x.version, {x_data}, x.provenance, x.expires_at
707 2885 : FROM jsonb_array_elements($3::jsonb) AS e
708 2885 : JOIN attr_instances x ON x.tenant_id = $1 AND x.entity_id = $2
709 2885 : AND x.attr_id = e->>'attr_id' AND x.observed_at = (e->>'observed_at')::timestamptz
710 2885 : AND x.observed_at BETWEEN $5::timestamptz AND $6::timestamptz
711 2885 : WHERE (x.instance_id = e->>'instance_id'
712 2885 : OR ($4 AND e->'data' ? 'observedAt'
713 2885 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'))
714 2885 : AND x.data - {STAMPS} IS DISTINCT FROM (e->'data') - {STAMPS}
715 2885 : ON CONFLICT DO NOTHING",
716 2885 : x_data = restamp("x")
717 2885 : )))
718 2885 : .bind(tenant.as_str())
719 2885 : .bind(entity_id)
720 2885 : .bind(&rows)
721 2885 : .bind(dedup)
722 2885 : .bind(&lo)
723 2885 : .bind(&hi)
724 2885 : .execute(&mut *tx)
725 2885 : .await?;
726 2885 : if dedup {
727 : // Same observedAt, another value: that observation was corrected.
728 2074 : sqlx::query(sqlx::AssertSqlSafe(format!(
729 2074 : "UPDATE attr_instances x
730 2074 : SET data = (e->'data') - 'instanceId',
731 2074 : version = x.version + 1, provenance = {CORRECTED},
732 2074 : modified_at = (e->>'modified_at')::timestamptz,
733 2074 : deleted_at = (e->>'deleted_at')::timestamptz, geo_value = {GEO_VALUE},
734 2074 : expires_at = (e->>'expires_at')::timestamptz
735 2074 : FROM jsonb_array_elements($3::jsonb) AS e
736 2074 : WHERE x.tenant_id = $1 AND x.entity_id = $2 AND e->'data' ? 'observedAt' AND {same}
737 2074 : AND x.observed_at = (e->>'observed_at')::timestamptz
738 2074 : AND x.observed_at BETWEEN $4::timestamptz AND $5::timestamptz
739 2074 : AND x.data - {STAMPS} IS DISTINCT FROM (e->'data') - {STAMPS}"
740 2074 : )))
741 2074 : .bind(tenant.as_str())
742 2074 : .bind(entity_id)
743 2074 : .bind(&rows)
744 2074 : .bind(&lo)
745 2074 : .bind(&hi)
746 2074 : .execute(&mut *tx)
747 2074 : .await?;
748 811 : }
749 : // `$4` off: today's row per write. On: a row with an observedAt already
750 : // stored, or without one equal to the newest stored instance, is no new
751 : // observation; a row that `follows` another of the batch was compared
752 : // with that one by `unseen`. Both probes ride
753 : // (tenant_id, entity_id, attr_id, observed_at, …)
754 : // and read one row: the second is NEWEST_REPEATS, never a scan of the
755 : // attribute's whole history (that scan cost ~90 ms a station at a
756 : // month of history, and a minute's batch of 2,063 stations recorded nothing).
757 2885 : sqlx::query(sqlx::AssertSqlSafe(format!(
758 2885 : "INSERT INTO attr_instances
759 2885 : (tenant_id, entity_id, attr_id, instance_id, dataset_id, observed_at,
760 2885 : created_at, modified_at, deleted_at, data, geo_value, provenance, expires_at)
761 2885 : SELECT $1, $2, e->>'attr_id', e->>'instance_id', e->>'dataset_id',
762 2885 : (e->>'observed_at')::timestamptz, (e->>'created_at')::timestamptz,
763 2885 : (e->>'modified_at')::timestamptz, (e->>'deleted_at')::timestamptz,
764 2885 : e->'data', {GEO_VALUE}, NULLIF(e->'provenance', 'null'::jsonb),
765 2885 : (e->>'expires_at')::timestamptz
766 2885 : FROM jsonb_array_elements($3::jsonb) AS e
767 2885 : WHERE NOT $4 OR NOT CASE
768 2885 : WHEN e->'data' ? 'observedAt' THEN EXISTS (
769 2885 : SELECT 1 FROM attr_instances x
770 2885 : WHERE x.tenant_id = $1 AND x.entity_id = $2 AND {same}
771 2885 : AND x.observed_at = (e->>'observed_at')::timestamptz
772 2885 : AND x.observed_at BETWEEN $5::timestamptz AND $6::timestamptz)
773 2885 : ELSE NOT (e ? 'follows') AND {NEWEST_REPEATS}
774 2885 : END
775 2885 : ON CONFLICT (tenant_id, entity_id, attr_id, instance_id, observed_at)
776 2885 : DO UPDATE SET data = EXCLUDED.data, modified_at = EXCLUDED.modified_at,
777 2885 : dataset_id = EXCLUDED.dataset_id,
778 2885 : deleted_at = EXCLUDED.deleted_at,
779 2885 : geo_value = EXCLUDED.geo_value,
780 2885 : expires_at = EXCLUDED.expires_at,
781 2885 : version = attr_instances.version + 1,
782 2885 : provenance = jsonb_set(COALESCE(EXCLUDED.provenance, '{{}}'::jsonb),
783 2885 : '{{operation}}', '\"correct\"')
784 2885 : WHERE attr_instances.data - {STAMPS} IS DISTINCT FROM EXCLUDED.data - {STAMPS}"
785 2885 : )))
786 2885 : .bind(tenant.as_str())
787 2885 : .bind(entity_id)
788 2885 : .bind(&rows)
789 2885 : .bind(dedup)
790 2885 : .bind(&lo)
791 2885 : .bind(&hi)
792 2885 : .execute(&mut *tx)
793 2885 : .await?;
794 2885 : Ok(())
795 2885 : }
796 :
797 : /// [`insert_rows`] into `attr_rows`: [`insert_text`]'s three statements over
798 : /// the integer keys. `iri_no` binds the dictionary ids [`Keys::ensure`] wrote;
799 : /// `instance_no` keys the instance, and a client's own id rides in `data`.
800 40 : async fn insert_keyed(
801 40 : tx: &mut sqlx::PgConnection,
802 40 : tenant: &TenantId,
803 40 : entity_id: &str,
804 40 : mut rows: Vec<Value>,
805 40 : dedup: bool,
806 40 : short: bool,
807 40 : ) -> Result<(), sqlx::Error> {
808 136 : for r in &mut rows {
809 136 : let id = r["instance_id"].as_str().unwrap_or_default().to_owned();
810 136 : let (no, own) = instance_no(&id);
811 136 : r["instance_no"] = Value::String(no.to_string());
812 136 : if !own {
813 16 : if let Some(d) = r["data"].as_object_mut() {
814 16 : d.insert("instanceId".to_owned(), Value::String(id));
815 16 : }
816 120 : }
817 : }
818 40 : let (lo, hi) = observed_range(&rows);
819 40 : let rows = Value::Array(rows);
820 : // the stored instance of the same (attribute, datasetId) a row of `e` repeats
821 40 : let same = "x.attr_no = iri_no($1, e->>'attr_id')
822 40 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'
823 40 : AND x.instance_no <> (e->>'instance_no')::uuid";
824 : // attr_versions keeps whole documents in the text shape
825 40 : sqlx::query(sqlx::AssertSqlSafe(format!(
826 40 : "INSERT INTO attr_versions
827 40 : (tenant_id, entity_id, attr_id, instance_id, dataset_id, observed_at, version, data, provenance,
828 40 : expires_at)
829 40 : SELECT x.tenant_id, x.entity_id, x.attr_id, x.instance_id, x.dataset_id, x.observed_at,
830 40 : x.version, {x_data}, x.provenance, x.expires_at
831 40 : FROM jsonb_array_elements($3::jsonb) AS e
832 40 : CROSS JOIN LATERAL attr_rows_at($1, $2, e->>'attr_id') x
833 40 : WHERE x.observed_at = (e->>'observed_at')::timestamptz
834 40 : AND x.observed_at BETWEEN $5::timestamptz AND $6::timestamptz
835 40 : AND (x.instance_id = e->>'instance_id'
836 40 : OR ($4 AND e->'data' ? 'observedAt'
837 40 : AND x.dataset_id IS NOT DISTINCT FROM e->>'dataset_id'))
838 40 : AND x.data - {STAMPS} IS DISTINCT FROM (e->'data') - {STAMPS}
839 40 : ON CONFLICT DO NOTHING",
840 40 : x_data = restamp("x")
841 40 : )))
842 40 : .bind(tenant.as_str())
843 40 : .bind(entity_id)
844 40 : .bind(&rows)
845 40 : .bind(dedup)
846 40 : .bind(&lo)
847 40 : .bind(&hi)
848 40 : .execute(&mut *tx)
849 40 : .await?;
850 40 : if dedup {
851 : // Same observedAt, another value: corrected in place, keeping the
852 : // stored instance's id (a client's one in `data`).
853 32 : sqlx::query(sqlx::AssertSqlSafe(format!(
854 32 : "UPDATE attr_rows x
855 32 : SET data = ((e->'data') - 'instanceId')
856 32 : || CASE WHEN x.data ? 'instanceId'
857 32 : THEN jsonb_build_object('instanceId', x.data->'instanceId')
858 32 : ELSE '{{}}'::jsonb END,
859 32 : version = x.version + 1, provenance = {CORRECTED},
860 32 : modified_at = (e->>'modified_at')::timestamptz,
861 32 : deleted_at = (e->>'deleted_at')::timestamptz, geo_value = {GEO_VALUE},
862 32 : expires_at = (e->>'expires_at')::timestamptz
863 32 : FROM jsonb_array_elements($3::jsonb) AS e
864 32 : WHERE x.tenant_id = $1 AND x.entity_no = iri_no($1, $2)
865 32 : AND e->'data' ? 'observedAt' AND {same}
866 32 : AND x.observed_at = (e->>'observed_at')::timestamptz
867 32 : AND x.observed_at BETWEEN $4::timestamptz AND $5::timestamptz
868 32 : AND x.data - {STAMPS} IS DISTINCT FROM (e->'data') - {STAMPS}"
869 32 : )))
870 32 : .bind(tenant.as_str())
871 32 : .bind(entity_id)
872 32 : .bind(&rows)
873 32 : .bind(&lo)
874 32 : .bind(&hi)
875 32 : .execute(&mut *tx)
876 32 : .await?;
877 8 : }
878 40 : sqlx::query(sqlx::AssertSqlSafe(format!(
879 40 : "INSERT INTO attr_rows
880 40 : (tenant_id, entity_no, attr_no, instance_no, dataset_id, observed_at,
881 40 : created_at, modified_at, deleted_at, data, geo_value, provenance, short, expires_at)
882 40 : SELECT $1, iri_no($1, $2), iri_no($1, e->>'attr_id'), (e->>'instance_no')::uuid,
883 40 : e->>'dataset_id',
884 40 : (e->>'observed_at')::timestamptz, (e->>'created_at')::timestamptz,
885 40 : (e->>'modified_at')::timestamptz, (e->>'deleted_at')::timestamptz,
886 40 : e->'data', {GEO_VALUE}, NULLIF(e->'provenance', 'null'::jsonb), $7,
887 40 : (e->>'expires_at')::timestamptz
888 40 : FROM jsonb_array_elements($3::jsonb) AS e
889 40 : WHERE NOT $4 OR NOT CASE
890 40 : WHEN e->'data' ? 'observedAt' THEN EXISTS (
891 40 : SELECT 1 FROM attr_rows x
892 40 : WHERE x.tenant_id = $1 AND x.entity_no = iri_no($1, $2) AND {same}
893 40 : AND x.observed_at = (e->>'observed_at')::timestamptz
894 40 : AND x.observed_at BETWEEN $5::timestamptz AND $6::timestamptz)
895 40 : ELSE NOT (e ? 'follows') AND {NEWEST_REPEATS}
896 40 : END
897 40 : ON CONFLICT (tenant_id, entity_no, attr_no, observed_at, instance_no, short)
898 40 : DO UPDATE SET data = EXCLUDED.data, modified_at = EXCLUDED.modified_at,
899 40 : dataset_id = EXCLUDED.dataset_id,
900 40 : deleted_at = EXCLUDED.deleted_at,
901 40 : geo_value = EXCLUDED.geo_value,
902 40 : expires_at = EXCLUDED.expires_at,
903 40 : version = attr_rows.version + 1,
904 40 : provenance = jsonb_set(COALESCE(EXCLUDED.provenance, '{{}}'::jsonb),
905 40 : '{{operation}}', '\"correct\"')
906 40 : WHERE attr_rows.data - {STAMPS} IS DISTINCT FROM EXCLUDED.data - {STAMPS}"
907 40 : )))
908 40 : .bind(tenant.as_str())
909 40 : .bind(entity_id)
910 40 : .bind(&rows)
911 40 : .bind(dedup)
912 40 : .bind(&lo)
913 40 : .bind(&hi)
914 40 : .bind(short)
915 40 : .execute(&mut *tx)
916 40 : .await?;
917 40 : Ok(())
918 40 : }
919 :
920 : /// Removes every instance and every kept version of one Entity: what 5.6.16
921 : /// and an expired Entity's reap take, versions included.
922 1891 : async fn wipe_instances(
923 1891 : tx: &mut sqlx::PgConnection,
924 1891 : tenant: &TenantId,
925 1891 : id: &str,
926 1891 : ) -> Result<(), sqlx::Error> {
927 1891 : sqlx::query("DELETE FROM attr_instances WHERE tenant_id = $1 AND entity_id = $2")
928 1891 : .bind(tenant.as_str())
929 1891 : .bind(id)
930 1891 : .execute(&mut *tx)
931 1891 : .await?;
932 1891 : sqlx::query("DELETE FROM attr_rows WHERE tenant_id = $1 AND entity_no = iri_no($1, $2)")
933 1891 : .bind(tenant.as_str())
934 1891 : .bind(id)
935 1891 : .execute(&mut *tx)
936 1891 : .await?;
937 1891 : sqlx::query("DELETE FROM attr_versions WHERE tenant_id = $1 AND entity_id = $2")
938 1891 : .bind(tenant.as_str())
939 1891 : .bind(id)
940 1891 : .execute(&mut *tx)
941 1891 : .await?;
942 1891 : Ok(())
943 1891 : }
944 :
945 : // TemporalFilter lives in `store::filter`; re-exported for path compat.
946 : pub use crate::store::filter::{TemporalFilter, TemporalOutcome};
947 :
948 : /// The correlated subquery reconstructing the attribute object for the meta
949 : /// row aliased `m`, with the 4.11 range and the lastN RANK() cap applied over
950 : /// the rows (byte-exact against the API window: predicates and ordering run
951 : /// on the instance JSON with COLLATE "C", never on the partition column).
952 : /// Returns the SQL fragment + its text binds, numbered from `first_bind`;
953 : /// `None` when a range is present but outside the compiler's exact subset —
954 : /// the caller then reconstructs unpruned and the window stays the arbiter.
955 1378 : fn attr_object_expr(f: &TemporalFilter<'_>, first_bind: usize) -> Option<(String, Vec<String>)> {
956 1378 : let (range_and, mut binds) = window_sql(f, first_bind)?;
957 1376 : let tp = first_bind;
958 1376 : let expr = match f.last_n {
959 491 : Some(n) => {
960 491 : let n_bind = first_bind.saturating_add(binds.len());
961 491 : binds.push(n.to_string());
962 : // 4.11 lastN keeps the N most recent INSTANTS, so the rank orders
963 : // on the same canonical key the window compares on — raw bytes
964 : // put "…00.000Z" after "…00Z" and kept the wrong N.
965 491 : let order_key = crate::compile::temporal::dt_key_sql(&format!("(ai.data ->> ${tp})"));
966 : // firstN keeps the N oldest (TS 104 176 6.4.7.3), the exact
967 : // reverse: the API window sorts an instance without the
968 : // timeproperty before every instant, so it ranks first ascending
969 491 : let dir = if f.oldest_first {
970 378 : "ASC NULLS FIRST"
971 : } else {
972 113 : "DESC NULLS LAST"
973 : };
974 491 : let rows = newest_rows(f, first_bind, n_bind)
975 491 : .unwrap_or_else(|| format!("{} ai WHERE true", history("m.tenant_id", "m.id")));
976 491 : format!(
977 : "COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (\
978 : SELECT s.attr_id, {s_insts} AS insts \
979 : FROM (SELECT ai.*, rank() OVER (PARTITION BY ai.attr_id, ai.data ->> 'datasetId' \
980 : ORDER BY {order_key} {dir}) AS rk \
981 : FROM {rows}{range_and}) s \
982 : WHERE s.rk <= ${n_bind}::bigint GROUP BY s.attr_id) g), '{{}}'::jsonb)",
983 491 : s_insts = insts_and_ids("s")
984 : )
985 : }
986 : None => {
987 : // No range and no lastN: nothing names the timeproperty, so the
988 : // bind `window_sql` reserved for it is not a parameter of the
989 : // statement the caller assembles. Handing it back anyway would
990 : // make every plain Retrieve Temporal Evolution bind one argument
991 : // more than its SQL declares.
992 885 : if f.range.is_none() {
993 625 : binds.clear();
994 625 : }
995 885 : format!(
996 : "COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (\
997 : SELECT ai.attr_id, {ai_insts} AS insts \
998 : FROM {rows} ai \
999 : WHERE true{range_and} \
1000 : GROUP BY ai.attr_id) g), '{{}}'::jsonb)",
1001 885 : ai_insts = insts_and_ids("ai"),
1002 885 : rows = history("m.tenant_id", "m.id")
1003 : )
1004 : }
1005 : };
1006 1376 : Some((expr, binds))
1007 1378 : }
1008 :
1009 : /// [`newest_rows`]' statement: `{window}` is its range bound on `t`
1010 : /// (`{xwindow}`, `{rwindow}` the same on `x` and `r`), `${n}` the bind of N.
1011 : ///
1012 : /// Distinct Attributes by a skip scan of the same btree, in each parent,
1013 : /// inside the window's column bound: an Attribute with no row
1014 : /// there contributes no row anyway, and the bound keeps the scan to the
1015 : /// partitions the window reaches instead of every week of the retention;
1016 : /// `OFFSET 0` keeps the instance read a parameterised index
1017 : /// range per Attribute, which a hash join over the whole Entity would
1018 : /// otherwise replace. The N-th newest takes the newest N of each parent,
1019 : /// each arm limited itself so both read their btree backwards, over the rows
1020 : /// that carry observedAt only: the rank puts a row without it after every
1021 : /// one with it (`NULLS LAST`), whatever its `observed_at` column says.
1022 : const NEWEST_ROWS: &str = "(WITH RECURSIVE a(attr) AS (\
1023 : SELECT min(x.attr_id) FROM attr_instances x \
1024 : WHERE x.tenant_id = m.tenant_id AND x.entity_id = m.id{xwindow} \
1025 : UNION ALL \
1026 : SELECT (SELECT min(x.attr_id) FROM attr_instances x \
1027 : WHERE x.tenant_id = m.tenant_id AND x.entity_id = m.id AND x.attr_id > a.attr{xwindow}) \
1028 : FROM a WHERE a.attr IS NOT NULL), \
1029 : b(no) AS (\
1030 : SELECT min(r.attr_no) FROM attr_rows r \
1031 : WHERE r.tenant_id = m.tenant_id AND r.entity_no = iri_no(m.tenant_id, m.id){rwindow} \
1032 : UNION ALL \
1033 : SELECT (SELECT min(r.attr_no) FROM attr_rows r \
1034 : WHERE r.tenant_id = m.tenant_id AND r.entity_no = iri_no(m.tenant_id, m.id) \
1035 : AND r.attr_no > b.no{rwindow}) \
1036 : FROM b WHERE b.no IS NOT NULL), \
1037 : n(attr) AS (\
1038 : SELECT a.attr FROM a WHERE a.attr IS NOT NULL \
1039 : UNION \
1040 : SELECT i.iri FROM b JOIN iri_ids i ON i.tenant_id = m.tenant_id AND i.id = b.no) \
1041 : SELECT ai.* FROM n a \
1042 : CROSS JOIN LATERAL (SELECT CASE WHEN a.attr = ANY(m.dataset_attrs) THEN '-infinity'::timestamptz \
1043 : ELSE (SELECT CASE WHEN k.nth - interval '28 hours' > m.offset_until THEN k.nth \
1044 : ELSE k.nth - interval '28 hours' END FROM (SELECT COALESCE((SELECT t.observed_at FROM (\
1045 : (SELECT t.observed_at FROM attr_instances t \
1046 : WHERE t.tenant_id = m.tenant_id AND t.entity_id = m.id AND t.attr_id = a.attr{window} \
1047 : AND t.data ->> 'observedAt' IS NOT NULL \
1048 : ORDER BY t.observed_at DESC LIMIT greatest(${n}::bigint, 1)) \
1049 : UNION ALL \
1050 : (SELECT t.observed_at FROM attr_rows_at(m.tenant_id, m.id, a.attr) t \
1051 : WHERE t.data ->> 'observedAt' IS NOT NULL{window} \
1052 : ORDER BY t.observed_at DESC LIMIT greatest(${n}::bigint, 1))) t \
1053 : ORDER BY t.observed_at DESC OFFSET greatest(${n}::bigint - 1, 0) LIMIT 1), \
1054 : '-infinity'::timestamptz) AS nth) k) END AS th) c \
1055 : CROSS JOIN LATERAL (SELECT * FROM attr_instances ai \
1056 : WHERE ai.tenant_id = m.tenant_id AND ai.entity_id = m.id AND ai.attr_id = a.attr \
1057 : AND ai.observed_at >= c.th \
1058 : UNION ALL SELECT * FROM attr_rows_at(m.tenant_id, m.id, a.attr) ai \
1059 : WHERE ai.observed_at >= c.th OFFSET 0) ai) ai WHERE true";
1060 :
1061 : /// lastN's candidate rows, read through the (tenant, entity, attr,
1062 : /// observed_at DESC) btree instead of ranking every instance of the window:
1063 : /// per Attribute, the rows at most 28 hours older than its N-th newest
1064 : /// `observed_at`. A SUPERSET of what the rank keeps: the parsed column and
1065 : /// the text key the rank orders on diverge by at most the two RFC 3339
1066 : /// offsets (±14 h each), so a row more than 28 h older than the N-th newest
1067 : /// has N rows strictly above it under either order. KPI histories of 22k
1068 : /// instances per Entity passed the gateway's 30 s ranking all of them;
1069 : /// this reads about N plus 28 hours of rows per Attribute.
1070 : /// Where the Entity's `offset_until` (migration 0028) lies more than 28 h
1071 : /// before the N-th newest, every row within those 28 h is a UTC stamp,
1072 : /// whose text key orders like `observed_at`: the N newest and their ties are
1073 : /// then exactly what the rank keeps, and the read stops there (the
1074 : /// bike-share stations write ~1200 rows a day each).
1075 : ///
1076 : /// `None` (rank the whole window, as before) where the superset argument
1077 : /// does not hold or no index serves it:
1078 : /// - a timeproperty other than observedAt (no btree on its column);
1079 : /// - firstN, and a window closed towards now (`before`, `between`): rows the
1080 : /// window drops could raise the N-th newest above what it keeps;
1081 : /// - an Attribute with any instance in a dataset: the rank partitions by
1082 : /// datasetId, and a sparse dataset's newest may sit below the Attribute's
1083 : /// N-th; such an Attribute keeps every row (`-infinity`), found through the
1084 : /// Entity's `dataset_attrs` (migration 0009), which the write path keeps.
1085 497 : fn newest_rows(f: &TemporalFilter<'_>, first_bind: usize, n_bind: usize) -> Option<String> {
1086 497 : if f.oldest_first || f.timeproperty != "observedAt" {
1087 379 : return None;
1088 118 : }
1089 121 : let bound = |alias| match &f.range {
1090 168 : None => Some(String::new()),
1091 118 : Some(r) if r.timerel == "any" => Some(String::new()),
1092 118 : Some(r) if r.timerel == "after" => {
1093 84 : crate::compile::temporal::column_range_bound(r, alias, first_bind.saturating_add(1))
1094 84 : .map(|b| format!(" AND {b}"))
1095 : }
1096 34 : Some(_) => None,
1097 286 : };
1098 : Some(
1099 118 : NEWEST_ROWS
1100 118 : .replace("{window}", &bound("t")?)
1101 84 : .replace("{xwindow}", &bound("x")?)
1102 84 : .replace("{rwindow}", &bound("r")?)
1103 84 : .replace("{n}", &n_bind.to_string()),
1104 : )
1105 497 : }
1106 :
1107 : /// The 4.11 window over the instance rows `ai` of the meta row `m`: the
1108 : /// ` AND …` fragment plus its text binds — $first_bind is always the
1109 : /// timeproperty (predicate member / order key), the range binds follow.
1110 : /// `None` when the range is outside the compiler's exact subset.
1111 1410 : fn window_sql(f: &TemporalFilter<'_>, first_bind: usize) -> Option<(String, Vec<String>)> {
1112 1410 : let mut binds = vec![f.timeproperty.to_owned()];
1113 1410 : let mut range_and = String::new();
1114 1410 : if let Some(r) = &f.range {
1115 575 : let c = crate::compile::temporal::compile_instance_range(r, "ai.data", first_bind)?;
1116 573 : range_and = format!(" AND {}", c.sql);
1117 573 : debug_assert_eq!(c.binds.first().map(String::as_str), Some(f.timeproperty));
1118 573 : binds.extend(c.binds.into_iter().skip(1));
1119 : // widened COLUMN bound on the SAME binds ($first_bind+1 = timeAt):
1120 : // lets the (tenant, entity, attr, observed_at) btree serve the range;
1121 : // the byte-exact text predicate above still decides membership
1122 557 : if let Some(cb) =
1123 573 : crate::compile::temporal::column_range_bound(r, "ai", first_bind.saturating_add(1))
1124 557 : {
1125 557 : range_and.push_str(&format!(" AND {cb}"));
1126 557 : }
1127 835 : }
1128 1408 : Some((range_and, binds))
1129 1410 : }
1130 :
1131 : /// Timestamp text in the shape the API's aggregated rows carry.
1132 : const BUCKET_TS: &str = r#"to_char({} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"')"#;
1133 :
1134 : /// 4.5.19 / 5.7.4.4 aggregated temporal representation computed in SQL for
1135 : /// the meta row `m`: per attribute the bucket matrix of every requested
1136 : /// method as `[value, start, end]` rows, in one object
1137 : /// `{"bad": <any windowed value non-numeric>, "attrs": {iri: {"type":
1138 : /// "Property", method: rows…}}}`. Only the numeric class (numbers and
1139 : /// booleans, Table 4.5.19.1-1) is computed here; `bad` tells the caller to
1140 : /// reconstruct instead so the API keeps every other class and the
1141 : /// eligibility errors. Buckets: `period_secs` wide from the anchor (the
1142 : /// request's timeAt, else the attribute's first instant); no period = one
1143 : /// bucket spanning the query's whole time range (4.5.19.1 PT0S), with the
1144 : /// edge 4.11 leaves open closed by the attribute's own first or last
1145 : /// instant — the API's own rule.
1146 32 : fn aggregate_expr(
1147 32 : f: &TemporalFilter<'_>,
1148 32 : agg: &crate::store::filter::Aggregate<'_>,
1149 32 : first_bind: usize,
1150 32 : ) -> Option<(String, Vec<String>)> {
1151 32 : let (range_and, mut binds) = window_sql(f, first_bind)?;
1152 32 : let tp = first_bind;
1153 32 : let col = match f.timeproperty {
1154 32 : "observedAt" => "observed_at",
1155 0 : "createdAt" => "created_at",
1156 0 : "modifiedAt" => "modified_at",
1157 0 : _ => return None,
1158 : };
1159 32 : let anchor = match agg.anchor {
1160 31 : Some(a) => {
1161 31 : binds.push(antares_store::filter::canonical_datetime(a).into_owned());
1162 31 : format!(
1163 : "${}::timestamptz",
1164 31 : first_bind.saturating_add(binds.len()).saturating_sub(1)
1165 : )
1166 : }
1167 1 : None => format!("min(ai.{col}) OVER (PARTITION BY ai.attr_id)"),
1168 : };
1169 32 : let (bs, be) = match agg.period_secs {
1170 : None => {
1171 : // 4.5.19.1: a zero duration "is interpreted as a duration
1172 : // spanning the whole time range specified by the temporal
1173 : // query". `before` names only the range's end and `after` only
1174 : // its start (4.11), so the data closes the other edge.
1175 15 : let (qs, qe) = match &f.range {
1176 14 : Some(r) if r.timerel == "before" => (None, Some(r.time_at)),
1177 13 : Some(r) if r.timerel == "between" => (Some(r.time_at), r.end_time_at),
1178 12 : Some(r) if r.timerel == "after" => (Some(r.time_at), None),
1179 1 : _ => (None, None),
1180 : };
1181 15 : let start = match qs {
1182 13 : Some(v) => {
1183 13 : binds.push(antares_store::filter::canonical_datetime(v).into_owned());
1184 13 : format!(
1185 : "${}::timestamptz",
1186 13 : first_bind.saturating_add(binds.len()).saturating_sub(1)
1187 : )
1188 : }
1189 2 : None => "s0.first_ts".to_owned(),
1190 : };
1191 15 : let end = match qe {
1192 2 : Some(v) => {
1193 2 : binds.push(antares_store::filter::canonical_datetime(v).into_owned());
1194 2 : format!(
1195 : "${}::timestamptz",
1196 2 : first_bind.saturating_add(binds.len()).saturating_sub(1)
1197 : )
1198 : }
1199 13 : None => "s0.last + interval '1 second'".to_owned(),
1200 : };
1201 15 : (start, end)
1202 : }
1203 17 : Some(sc) => {
1204 17 : binds.push(sc.to_string());
1205 17 : let n = first_bind.saturating_add(binds.len()).saturating_sub(1);
1206 17 : let start = format!("date_bin(make_interval(secs => ${n}::bigint), s0.ts, s0.anchor)");
1207 17 : (
1208 17 : start.clone(),
1209 17 : format!("{start} + make_interval(secs => ${n}::bigint)"),
1210 17 : )
1211 : }
1212 : };
1213 32 : let mut aggs = String::new();
1214 32 : let mut rows = String::new();
1215 32 : let mut pairs = String::new();
1216 67 : for (i, m) in agg.methods.iter().enumerate() {
1217 : // The SQL and the member name both come out of this table, so the
1218 : // request's own string never reaches the statement.
1219 67 : let (name, sql) = match m.as_str() {
1220 67 : "totalCount" => ("totalCount", "count(*)"),
1221 53 : "distinctCount" => ("distinctCount", "count(DISTINCT s.v)"),
1222 53 : "min" => ("min", "min(s.v)"),
1223 49 : "max" => ("max", "max(s.v)"),
1224 30 : "sum" => ("sum", "sum(s.v)"),
1225 30 : "avg" => ("avg", "avg(s.v)"),
1226 0 : "stddev" => ("stddev", "stddev_pop(s.v)"),
1227 0 : "sumsq" => ("sumsq", "sum(s.v * s.v)"),
1228 0 : _ => return None,
1229 : };
1230 67 : aggs.push_str(&format!(", {sql} AS m{i}"));
1231 67 : rows.push_str(&format!(
1232 67 : ", jsonb_agg(jsonb_build_array(g.m{i}, {}, {}) ORDER BY g.bs) AS r{i}",
1233 67 : BUCKET_TS.replace("{}", "g.bs"),
1234 67 : BUCKET_TS.replace("{}", "g.be")
1235 67 : ));
1236 67 : pairs.push_str(&format!(", '{name}', b.r{i}"));
1237 : }
1238 32 : let hist = history("m.tenant_id", "m.id");
1239 32 : let expr = format!(
1240 : "(SELECT jsonb_build_object('bad', bool_or(b.bad), 'attrs', \
1241 : jsonb_object_agg(b.attr_id, jsonb_build_object('type', 'Property'{pairs}))) \
1242 : FROM (SELECT g.attr_id, bool_or(g.bad) AS bad{rows} \
1243 : FROM (SELECT s.attr_id, s.bs, s.be, bool_or(s.bad) AS bad{aggs} \
1244 : FROM (SELECT s0.attr_id, s0.v, s0.bad, {bs} AS bs, {be} AS be \
1245 : FROM (SELECT ai.attr_id, \
1246 : CASE WHEN jsonb_typeof(ai.data -> 'value') = 'boolean' \
1247 : THEN (CASE WHEN (ai.data ->> 'value')::boolean THEN 1 ELSE 0 END)::float8 \
1248 : WHEN jsonb_typeof(ai.data -> 'value') = 'number' \
1249 : THEN (ai.data ->> 'value')::float8 END AS v, \
1250 : jsonb_typeof(ai.data -> 'value') IS DISTINCT FROM 'number' \
1251 : AND jsonb_typeof(ai.data -> 'value') IS DISTINCT FROM 'boolean' AS bad, \
1252 : ai.{col} AS ts, {anchor} AS anchor, \
1253 : min(ai.{col}) OVER (PARTITION BY ai.attr_id) AS first_ts, \
1254 : max(ai.{col}) OVER (PARTITION BY ai.attr_id) AS last \
1255 : FROM {hist} ai \
1256 : WHERE jsonb_typeof(ai.data -> ${tp}) = 'string'{range_and}) s0) s \
1257 : GROUP BY s.attr_id, s.bs, s.be) g \
1258 : GROUP BY g.attr_id) b)"
1259 : );
1260 32 : Some((expr, binds))
1261 32 : }
1262 :
1263 : impl PgTemporalStore {
1264 153 : pub fn new(pool: PgPool) -> Self {
1265 153 : Self {
1266 153 : pool,
1267 153 : temporal_only: false,
1268 153 : keys: Keys::default(),
1269 153 : }
1270 153 : }
1271 :
1272 : /// A store that routes by `cutover` instead of `history_cutover`, so a
1273 : /// test sets its own cut-over without moving every other test's rows.
1274 : #[cfg(any(test, feature = "test-kit"))]
1275 10 : pub fn with_cutover(pool: PgPool, cutover: Option<chrono::DateTime<chrono::Utc>>) -> Self {
1276 10 : let mut s = Self::new(pool);
1277 10 : s.keys.cutover = tokio::sync::OnceCell::new_with(Some(cutover));
1278 10 : s
1279 10 : }
1280 :
1281 : /// Replaces which writes keep short history with `short`.
1282 4 : pub fn set_short_history(&self, short: antares_store::ShortHistory) {
1283 4 : *self
1284 4 : .keys
1285 4 : .short
1286 4 : .write()
1287 4 : .unwrap_or_else(std::sync::PoisonError::into_inner) = std::sync::Arc::new(short);
1288 4 : }
1289 :
1290 : /// Forgets the dictionary entries of `tenant` after its purge.
1291 26 : pub fn forget_tenant(&self, tenant: &TenantId) {
1292 26 : self.keys.forget(tenant);
1293 26 : }
1294 :
1295 : /// `false` when the id already exists (create semantics, like the memory
1296 : /// store's `create`).
1297 : ///
1298 : /// 4.22: an expired one does NOT exist, so it is dropped here — history
1299 : /// included — and the insert below is then an ordinary create. Leaving it
1300 : /// in place would make `create` report a conflict for an id every read
1301 : /// calls absent, and `AnyStore::upsert` reads that `false` as "already
1302 : /// there" and falls through to `mutate`, which refuses an expired entity
1303 : /// too: the upsert would write nothing and report success.
1304 629 : pub async fn create(
1305 629 : &self,
1306 629 : tenant: &TenantId,
1307 629 : id: &str,
1308 629 : doc: &Value,
1309 629 : ) -> Result<bool, sqlx::Error> {
1310 629 : let (types, scopes, created, modified) = extract(doc);
1311 629 : let meta = meta_of(doc);
1312 629 : let rows = decompose(doc);
1313 629 : let mut tx = super::begin(&self.pool).await?;
1314 629 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1315 629 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
1316 629 : let reaped = sqlx::query(sqlx::AssertSqlSafe(format!(
1317 629 : "DELETE FROM temporal_entities m \
1318 629 : WHERE m.tenant_id = $1 AND m.id = $2 AND NOT {NOT_EXPIRED}"
1319 629 : )))
1320 629 : .bind(tenant.as_str())
1321 629 : .bind(id)
1322 629 : .execute(&mut *tx)
1323 629 : .await?
1324 629 : .rows_affected();
1325 629 : if reaped == 1 {
1326 1 : wipe_instances(&mut tx, tenant, id).await?;
1327 628 : }
1328 629 : let n = sqlx::query(
1329 629 : "INSERT INTO temporal_entities
1330 629 : (tenant_id, id, types, scopes, meta, created_at, modified_at)
1331 629 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz)
1332 629 : ON CONFLICT (tenant_id, id) DO NOTHING",
1333 629 : )
1334 629 : .bind(tenant.as_str())
1335 629 : .bind(id)
1336 629 : .bind(&types)
1337 629 : .bind(&scopes)
1338 629 : .bind(&meta)
1339 629 : .bind(&created)
1340 629 : .bind(&modified)
1341 629 : .execute(&mut *tx)
1342 629 : .await?
1343 629 : .rows_affected();
1344 629 : if n == 1 {
1345 622 : let mut ensured = std::collections::HashSet::new();
1346 622 : let short = self.keys.is_short(tenant, &types, doc);
1347 622 : insert_rows(
1348 622 : &mut tx,
1349 622 : &self.keys,
1350 622 : &mut ensured,
1351 622 : tenant,
1352 622 : id,
1353 622 : rows,
1354 622 : Route {
1355 622 : dedup: false,
1356 622 : short,
1357 622 : },
1358 622 : )
1359 622 : .await?;
1360 7 : }
1361 629 : tx.commit().await?;
1362 629 : Ok(n == 1)
1363 629 : }
1364 :
1365 : /// Append-only fast path (auto-recording, 5.6.12 adds): NO reconstruction,
1366 : /// no history read — a shell meta insert (first touch) plus one multi-row
1367 : /// instance upsert. This is the write the old full-resync made O(history).
1368 : ///
1369 : /// Conditional on the entity still existing. 5.6.6 Delete Entity removes
1370 : /// the entity and the temporal evolution recorded for it, in that order and
1371 : /// in two transactions; an auto-recording append that overlaps the delete
1372 : /// would otherwise commit history for an entity that is gone, and no later
1373 : /// delete would ever clean it. The `FOR KEY SHARE` lock is what makes the
1374 : /// check hold: it lets concurrent updates of the same entity through and
1375 : /// makes a concurrent DELETE wait until this append has committed.
1376 136 : pub async fn append(
1377 136 : &self,
1378 136 : tenant: &TenantId,
1379 136 : id: &str,
1380 136 : shell: &Value,
1381 136 : additions: &Value,
1382 136 : ) -> Result<(), sqlx::Error> {
1383 136 : let mut tx = super::begin(&self.pool).await?;
1384 136 : let mut ensured = std::collections::HashSet::new();
1385 136 : self.append_in(&mut tx, &mut ensured, tenant, id, shell, additions)
1386 136 : .await?;
1387 136 : tx.commit().await?;
1388 136 : Ok(())
1389 136 : }
1390 :
1391 : /// [`Self::append`] of a whole drained request in one transaction:
1392 : /// a batch of N Entities costs one BEGIN, not N. Each append
1393 : /// keeps its own liveness check and tenant.
1394 1917 : pub async fn append_many(
1395 1917 : &self,
1396 1917 : appends: &[(&TenantId, &str, &Value, Value)],
1397 1917 : ) -> Result<(), sqlx::Error> {
1398 1917 : if appends.is_empty() {
1399 0 : return Ok(());
1400 1917 : }
1401 1917 : let mut tx = super::begin(&self.pool).await?;
1402 1917 : let mut ensured = std::collections::HashSet::new();
1403 : // After the cut-over or for a short write, the batch's first sights go
1404 : // in first, in one order: an Entity at a time, two batches could each
1405 : // wait on an IRI the other wrote first. Only live Entities count:
1406 : // append_in skips the others, and their IRIs would stay in iri_ids
1407 : // naming nothing. The FOR KEY SHARE taken here holds until commit, so
1408 : // append_in's own check finds the same Entities.
1409 1917 : let short = appends
1410 1917 : .iter()
1411 1985 : .any(|(tenant, _, shell, _)| self.keys.is_short(tenant, &extract(shell).0, shell));
1412 1917 : if short || self.keys.cutover(&mut tx).await?.is_some() {
1413 : // the batch's Entities per Tenant, as (id, additions)
1414 : type Docs<'a> = Vec<(&'a str, &'a Value)>;
1415 1 : let mut by_tenant: std::collections::BTreeMap<&str, (&TenantId, Docs)> =
1416 1 : std::collections::BTreeMap::new();
1417 1 : for (tenant, id, _, additions) in appends {
1418 1 : by_tenant
1419 1 : .entry(tenant.as_str())
1420 1 : .or_insert_with(|| (tenant, Vec::new()))
1421 : .1
1422 1 : .push((id, additions));
1423 : }
1424 1 : for (tenant, docs) in by_tenant.into_values() {
1425 1 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1426 1 : let live = self.live_of(&mut tx, tenant, &docs).await?;
1427 1 : let iris = docs
1428 1 : .into_iter()
1429 1 : .filter(|(id, _)| live.contains(*id))
1430 1 : .flat_map(|(id, additions)| iris_of(id, additions))
1431 1 : .collect();
1432 1 : self.keys
1433 1 : .ensure(&mut tx, tenant, iris, &mut ensured)
1434 1 : .await?;
1435 : }
1436 1916 : }
1437 1985 : for (tenant, id, shell, additions) in appends {
1438 1985 : self.append_in(&mut tx, &mut ensured, tenant, id, shell, additions)
1439 1985 : .await?;
1440 : }
1441 1917 : tx.commit().await?;
1442 1917 : Ok(())
1443 1917 : }
1444 :
1445 : /// The ids of `docs` stored in `entities` (all of them when this store
1446 : /// serves only the temporal seam), each locked FOR KEY SHARE until the
1447 : /// caller's commit.
1448 1 : async fn live_of(
1449 1 : &self,
1450 1 : tx: &mut sqlx::Transaction<'static, sqlx::Postgres>,
1451 1 : tenant: &TenantId,
1452 1 : docs: &[(&str, &Value)],
1453 1 : ) -> Result<std::collections::HashSet<String>, sqlx::Error> {
1454 1 : let mut ids: Vec<&str> = docs.iter().map(|(id, _)| *id).collect();
1455 1 : if self.temporal_only {
1456 0 : return Ok(ids.into_iter().map(str::to_owned).collect());
1457 1 : }
1458 1 : ids.sort_unstable();
1459 1 : ids.dedup();
1460 1 : let live = sqlx::query_scalar(
1461 1 : "SELECT id FROM entities WHERE tenant_id = $1 AND id = ANY($2)
1462 1 : ORDER BY id FOR KEY SHARE",
1463 1 : )
1464 1 : .bind(tenant.as_str())
1465 1 : .bind(&ids)
1466 1 : .fetch_all(&mut **tx)
1467 1 : .await?;
1468 1 : Ok(live.into_iter().collect())
1469 1 : }
1470 :
1471 2121 : async fn append_in(
1472 2121 : &self,
1473 2121 : tx: &mut sqlx::Transaction<'static, sqlx::Postgres>,
1474 2121 : ensured: &mut std::collections::HashSet<(String, String)>,
1475 2121 : tenant: &TenantId,
1476 2121 : id: &str,
1477 2121 : shell: &Value,
1478 2121 : additions: &Value,
1479 2121 : ) -> Result<(), sqlx::Error> {
1480 2121 : let (types, scopes, created, modified) = extract(shell);
1481 2121 : let meta = meta_of(shell);
1482 2121 : let rows = decompose(additions);
1483 2121 : crate::store::pg::set_tenant(tx, tenant).await?;
1484 2121 : if !self.temporal_only {
1485 2083 : let live = sqlx::query(
1486 2083 : "SELECT 1 FROM entities WHERE tenant_id = $1 AND id = $2 FOR KEY SHARE",
1487 2083 : )
1488 2083 : .bind(tenant.as_str())
1489 2083 : .bind(id)
1490 2083 : .fetch_optional(&mut **tx)
1491 2083 : .await?;
1492 2083 : if live.is_none() {
1493 3 : return Ok(());
1494 2080 : }
1495 38 : }
1496 : // DO NOTHING froze types/scopes at first touch — an
1497 : // entity gaining a type stayed invisible to type-filtered
1498 : // temporal queries forever. The shell carries the CURRENT
1499 : // entity, so refresh on change; the IS DISTINCT FROM guard keeps
1500 : // the common no-change append from churning the meta row.
1501 : // The one meta member the shell cannot refresh is a `scope` that
1502 : // holds instances (4.5.6 scope history, its deletion instances):
1503 : // the shell's is the current string array, and writing it over the
1504 : // history erased every scope change recorded so far.
1505 2118 : sqlx::query(sqlx::AssertSqlSafe(format!(
1506 2118 : "INSERT INTO temporal_entities
1507 2118 : (tenant_id, id, types, scopes, meta, created_at, modified_at)
1508 2118 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz)
1509 2118 : ON CONFLICT (tenant_id, id) DO UPDATE SET
1510 2118 : types = EXCLUDED.types,
1511 2118 : scopes = EXCLUDED.scopes,
1512 2118 : meta = {KEPT_META},
1513 2118 : modified_at = EXCLUDED.modified_at
1514 2118 : WHERE temporal_entities.types IS DISTINCT FROM EXCLUDED.types
1515 2118 : OR temporal_entities.scopes IS DISTINCT FROM EXCLUDED.scopes
1516 2118 : OR temporal_entities.meta IS DISTINCT FROM {KEPT_META}"
1517 2118 : )))
1518 2118 : .bind(tenant.as_str())
1519 2118 : .bind(id)
1520 2118 : .bind(&types)
1521 2118 : .bind(&scopes)
1522 2118 : .bind(&meta)
1523 2118 : .bind(&created)
1524 2118 : .bind(&modified)
1525 2118 : .execute(&mut **tx)
1526 2118 : .await?;
1527 2118 : let short = self.keys.is_short(tenant, &types, shell);
1528 2118 : let route = Route { dedup: true, short };
1529 2118 : insert_rows(tx, &self.keys, ensured, tenant, id, rows, route).await
1530 2121 : }
1531 :
1532 563 : pub async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
1533 563 : self.get_range(tenant, id, &TemporalFilter::default()).await
1534 563 : }
1535 :
1536 : /// 5.6.16 Delete Temporal Evolution: `true` when it was there, `false`
1537 : /// for "no existing Entity whose id (URI) is equivalent held locally",
1538 : /// which the caller answers as ResourceNotFound. 4.22 decides what "held
1539 : /// locally" means, the same way `get_range` and `query` decide it.
1540 4805 : pub async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, sqlx::Error> {
1541 4805 : let mut tx = super::begin(&self.pool).await?;
1542 4805 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1543 4805 : let n = sqlx::query(sqlx::AssertSqlSafe(format!(
1544 4805 : "DELETE FROM temporal_entities m \
1545 4805 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED}"
1546 4805 : )))
1547 4805 : .bind(tenant.as_str())
1548 4805 : .bind(id)
1549 4805 : .execute(&mut *tx)
1550 4805 : .await?
1551 4805 : .rows_affected();
1552 : // no FK: a partitioned table cannot be the referencing side of a
1553 : // cascade from temporal_entities — clean the instances explicitly,
1554 : // and only for the row this call actually removed. An expired
1555 : // entity is refused above and keeps its history until the 4.22
1556 : // reap or a create replaces it; wiping it here would destroy the
1557 : // history behind a 404.
1558 4805 : if n == 1 {
1559 1890 : wipe_instances(&mut tx, tenant, id).await?;
1560 2915 : }
1561 4805 : tx.commit().await?;
1562 4805 : Ok(n == 1)
1563 4805 : }
1564 :
1565 : /// Erasure: the rows of `attrs` of one Entity,
1566 : /// current and superseded, observed at or before `before` (`None`: all);
1567 : /// `attrs` `None` takes every Attribute, and the Entity's history itself
1568 : /// once it holds no row. How many rows went.
1569 1 : pub async fn erase(
1570 1 : &self,
1571 1 : tenant: &TenantId,
1572 1 : id: &str,
1573 1 : attrs: Option<&[String]>,
1574 1 : before: Option<&str>,
1575 1 : ) -> Result<u64, sqlx::Error> {
1576 1 : let mut tx = super::begin(&self.pool).await?;
1577 1 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1578 1 : let names = attrs.map(<[String]>::to_vec);
1579 1 : let mut n = 0u64;
1580 3 : for sql in [
1581 1 : "DELETE FROM attr_instances WHERE tenant_id = $1 AND entity_id = $2 \
1582 1 : AND ($3::text[] IS NULL OR attr_id = ANY($3)) \
1583 1 : AND ($4::timestamptz IS NULL OR observed_at <= $4::timestamptz)",
1584 1 : "DELETE FROM attr_rows WHERE tenant_id = $1 AND entity_no = iri_no($1, $2) \
1585 1 : AND ($3::text[] IS NULL OR attr_no IN \
1586 1 : (SELECT i.id FROM iri_ids i WHERE i.tenant_id = $1 AND i.iri = ANY($3))) \
1587 1 : AND ($4::timestamptz IS NULL OR observed_at <= $4::timestamptz)",
1588 1 : "DELETE FROM attr_versions WHERE tenant_id = $1 AND entity_id = $2 \
1589 1 : AND ($3::text[] IS NULL OR attr_id = ANY($3)) \
1590 1 : AND ($4::timestamptz IS NULL OR observed_at <= $4::timestamptz)",
1591 1 : ] {
1592 3 : let gone = sqlx::query(sql)
1593 3 : .bind(tenant.as_str())
1594 3 : .bind(id)
1595 3 : .bind(&names)
1596 3 : .bind(before)
1597 3 : .execute(&mut *tx)
1598 3 : .await?
1599 3 : .rows_affected();
1600 3 : n = n.saturating_add(gone);
1601 : }
1602 1 : if attrs.is_none() {
1603 0 : sqlx::query(
1604 0 : "DELETE FROM temporal_entities t WHERE t.tenant_id = $1 AND t.id = $2 \
1605 0 : AND NOT EXISTS (SELECT 1 FROM attr_instances a \
1606 0 : WHERE a.tenant_id = t.tenant_id AND a.entity_id = t.id) \
1607 0 : AND NOT EXISTS (SELECT 1 FROM attr_rows r WHERE r.tenant_id = t.tenant_id \
1608 0 : AND r.entity_no = iri_no(t.tenant_id, t.id))",
1609 0 : )
1610 0 : .bind(tenant.as_str())
1611 0 : .bind(id)
1612 0 : .execute(&mut *tx)
1613 0 : .await?;
1614 1 : }
1615 1 : tx.commit().await?;
1616 1 : Ok(n)
1617 1 : }
1618 :
1619 : /// Every version of the rows of one Entity's history by Attribute: each instance with its `version` and `provenance`,
1620 : /// a superseded one with `supersededAt`, oldest first per measurement.
1621 7 : pub async fn versions(
1622 7 : &self,
1623 7 : tenant: &TenantId,
1624 7 : id: &str,
1625 7 : ) -> Result<serde_json::Map<String, Value>, sqlx::Error> {
1626 7 : let mut tx = super::begin(&self.pool).await?;
1627 7 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1628 7 : let rows: Vec<(String, Value)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
1629 7 : "SELECT attr_id, data || jsonb_build_object('version', version) || jsonb_strip_nulls( \
1630 7 : jsonb_build_object('provenance', provenance, 'supersededAt', \
1631 7 : to_char(superseded_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"'))) \
1632 7 : FROM (SELECT attr_id, observed_at, dataset_id, version, data, provenance, superseded_at \
1633 7 : FROM attr_versions WHERE tenant_id = $1 AND entity_id = $2 \
1634 7 : UNION ALL \
1635 7 : SELECT ai.attr_id, ai.observed_at, ai.dataset_id, ai.version, {}, ai.provenance, NULL \
1636 7 : FROM {} ai) v \
1637 7 : ORDER BY attr_id, observed_at, dataset_id NULLS FIRST, version",
1638 7 : restamp("ai"),
1639 7 : history("$1", "$2")
1640 7 : )))
1641 7 : .bind(tenant.as_str())
1642 7 : .bind(id)
1643 7 : .fetch_all(&mut *tx)
1644 7 : .await?;
1645 7 : tx.commit().await?;
1646 7 : let mut out = serde_json::Map::new();
1647 387 : for (attr, instance) in rows {
1648 387 : if let Some(list) = out
1649 387 : .entry(attr)
1650 387 : .or_insert_with(|| Value::Array(Vec::new()))
1651 387 : .as_array_mut()
1652 387 : {
1653 387 : list.push(instance);
1654 387 : }
1655 : }
1656 7 : Ok(out)
1657 7 : }
1658 :
1659 : /// Temporal query: entity narrowing (ids/types/attrs) in the WHERE,
1660 : /// instance pruning (range + lastN cap) in the reconstruction, and —
1661 : /// when the caller passes a page — entity qualification + LIMIT/OFFSET
1662 : /// in SQL, so a temporal query no longer materializes the whole tenant.
1663 504 : pub async fn query(
1664 504 : &self,
1665 504 : tenant: &TenantId,
1666 504 : f: &TemporalFilter<'_>,
1667 504 : ) -> Result<TemporalOutcome, sqlx::Error> {
1668 504 : self.query_inner(tenant, f, f.aggregate.is_some()).await
1669 504 : }
1670 :
1671 : /// `push_agg`: compute the filter's 4.5.19 aggregation in SQL; a row
1672 : /// whose windowed values are not all numeric makes the whole query fall
1673 : /// back to instance reconstruction (one extra round trip, only then).
1674 505 : async fn query_inner(
1675 505 : &self,
1676 505 : tenant: &TenantId,
1677 505 : f: &TemporalFilter<'_>,
1678 505 : push_agg: bool,
1679 505 : ) -> Result<TemporalOutcome, sqlx::Error> {
1680 : enum B {
1681 : Text(String),
1682 : Arr(Vec<String>),
1683 : Num(i64),
1684 : Float(f64),
1685 : }
1686 505 : let mut binds: Vec<B> = vec![B::Text(tenant.as_str().to_owned())];
1687 : // 4.22: an expired ENTITY is invalid on temporal reads too — filter it
1688 : // in SQL (no bind) so paging/totals stay exact. Expired instances are
1689 : // stripped at the read boundary (any.rs). Literal, applies to the
1690 : // fallback count query too.
1691 505 : let mut wheres = vec!["m.tenant_id = $1".to_owned(), NOT_EXPIRED.to_owned()];
1692 505 : if let Some(ids) = f.ids {
1693 127 : binds.push(B::Arr(super::storable(ids)));
1694 127 : wheres.push(format!("m.id = ANY(${})", binds.len()));
1695 378 : }
1696 505 : if let Some(types) = f.types {
1697 318 : // overlap: entity has ANY of the wanted types (flat OR list).
1698 318 : // `&&` alone is no index condition under RLS (not leakproof):
1699 318 : // the first-type arm and the multi-type arm are (migration 0015),
1700 318 : // and they BitmapOr instead of reading the Tenant.
1701 318 : binds.push(B::Arr(super::storable(types)));
1702 318 : let n = binds.len();
1703 318 : wheres.push(format!(
1704 318 : "(m.types[1] = ANY(${n}) OR (cardinality(m.types) > 1 AND m.types && ${n}))"
1705 318 : ));
1706 318 : }
1707 505 : if let Some(attrs) = f.attrs {
1708 32 : binds.push(B::Arr(super::storable(attrs)));
1709 32 : wheres.push(format!(
1710 32 : "(EXISTS (SELECT 1 FROM attr_instances x WHERE x.tenant_id = m.tenant_id \
1711 32 : AND x.entity_id = m.id AND x.attr_id = ANY(${n})) \
1712 32 : OR EXISTS (SELECT 1 FROM attr_rows r WHERE r.tenant_id = m.tenant_id \
1713 32 : AND r.entity_no = iri_no(m.tenant_id, m.id) AND r.attr_no IN \
1714 32 : (SELECT i.id FROM iri_ids i WHERE i.tenant_id = m.tenant_id AND i.iri = ANY(${n}))))",
1715 32 : n = binds.len()
1716 32 : ));
1717 473 : }
1718 : // page pushdown: only when the caller passed one AND the range (if
1719 : // any) compiles — SQL then also applies the evaluator's entity-
1720 : // qualification rule (≥1 instance, in-window when ranged). WHERE
1721 : // binds come FIRST so the fallback count query (offset past the end)
1722 : // can reuse them with identical numbering.
1723 : // None: no range to compile. Some(None): a range that does not
1724 : // compile, which is the case page pushdown must not take — the SQL
1725 : // would qualify entities on a window it cannot express.
1726 505 : let compiled_range = f.range.as_ref().map(|r| {
1727 446 : crate::compile::temporal::compile_instance_range(
1728 446 : r,
1729 : "ai.data",
1730 446 : binds.len().saturating_add(1),
1731 : )
1732 446 : });
1733 505 : let mut paged = false;
1734 505 : if f.page.is_some() && !matches!(compiled_range, Some(None)) {
1735 246 : let mut preds = String::new();
1736 246 : if let (Some(r), Some(Some(c))) = (&f.range, compiled_range) {
1737 244 : let n = binds.len().saturating_add(1);
1738 566 : for b in c.binds {
1739 566 : binds.push(B::Text(b));
1740 566 : }
1741 244 : preds.push_str(&format!(" AND {}", c.sql));
1742 : // index-serving widened bound on the same binds ($n+1 = timeAt)
1743 244 : if let Some(cb) =
1744 244 : crate::compile::temporal::column_range_bound(r, "ai", n.saturating_add(1))
1745 244 : {
1746 244 : preds.push_str(&format!(" AND {cb}"));
1747 244 : }
1748 2 : }
1749 246 : let dir = match &f.range {
1750 244 : Some(r) if r.timerel == "before" => "ASC",
1751 208 : _ => "DESC",
1752 : };
1753 246 : let shown = f.page_attrs.map(|attrs| {
1754 7 : binds.push(B::Arr(super::storable(attrs)));
1755 7 : binds.len()
1756 7 : });
1757 246 : wheres.push(has_history(&preds, dir, shown));
1758 246 : paged = true;
1759 259 : }
1760 : // 5.7.4.4 S2 superset prefilter: entities with no windowed instance
1761 : // satisfying the compilable part of q= are never reconstructed. The
1762 : // API arbiter re-evaluates q on every row that comes back, so this
1763 : // can only narrow (compile::qprefilter invariant), never decide.
1764 505 : if let Some(qn) = f.q {
1765 146 : if let Some(c) = crate::compile::qprefilter::compile_prefilter(
1766 146 : qn,
1767 146 : f.range.as_ref(),
1768 146 : "m",
1769 146 : binds.len().saturating_add(1),
1770 146 : f.expand,
1771 146 : ) {
1772 1754 : for b in c.binds {
1773 1754 : binds.push(B::Text(b));
1774 1754 : }
1775 124 : wheres.push(c.sql);
1776 22 : }
1777 359 : }
1778 : // 5.7.4.4 S3 superset prefilter: entities with no windowed instance
1779 : // of the geoproperty possibly satisfying the geoquery are never
1780 : // reconstructed. NULL geo_value (rows with no extracted geometry) always
1781 : // survives; GeoQuery::matches stays the arbiter.
1782 505 : for (spec, iri) in f.geo {
1783 29 : let attr_bind = binds.len().saturating_add(1);
1784 29 : let mut win_binds: Vec<String> = Vec::new();
1785 29 : let mut window = String::new();
1786 29 : if let Some(r) = &f.range {
1787 29 : if let Some(cb) = crate::compile::temporal::column_range_bound(
1788 29 : r,
1789 29 : "gi",
1790 29 : attr_bind.saturating_add(1),
1791 29 : ) {
1792 : // 4.6.3 lets a request spell the seconds fraction with a
1793 : // comma; the bound lands in a `::timestamptz` cast, which
1794 : // takes only the point form. Every other range bind is
1795 : // canonicalized for exactly this reason.
1796 29 : win_binds
1797 29 : .push(antares_store::filter::canonical_datetime(r.time_at).into_owned());
1798 29 : if r.timerel == "between" {
1799 22 : if let Some(e) = r.end_time_at {
1800 22 : win_binds
1801 22 : .push(antares_store::filter::canonical_datetime(e).into_owned());
1802 22 : }
1803 7 : }
1804 29 : window = format!(" AND {cb}");
1805 0 : }
1806 0 : }
1807 29 : if let Some(c) = crate::compile::geo::compile_geo_instance(
1808 29 : spec,
1809 29 : "gi.geo_value",
1810 29 : attr_bind.saturating_add(1).saturating_add(win_binds.len()),
1811 29 : ) {
1812 29 : binds.push(B::Text(iri.to_owned()));
1813 51 : for b in win_binds {
1814 51 : binds.push(B::Text(b));
1815 51 : }
1816 29 : for b in c.geo_binds {
1817 29 : binds.push(B::Text(b));
1818 29 : }
1819 29 : for n in c.num_binds {
1820 17 : binds.push(B::Float(n));
1821 17 : }
1822 29 : wheres.push(format!(
1823 : "EXISTS (SELECT 1 FROM (SELECT * FROM attr_instances \
1824 : WHERE tenant_id = m.tenant_id AND entity_id = m.id AND attr_id = ${attr_bind} \
1825 : UNION ALL SELECT * FROM attr_rows_at(m.tenant_id, m.id, ${attr_bind})) gi \
1826 : WHERE true{window} AND {})",
1827 : c.sql
1828 : ));
1829 0 : }
1830 : }
1831 505 : let n_where = binds.len();
1832 505 : let where_sql = wheres.join(" AND ");
1833 505 : let agg = if push_agg {
1834 32 : f.aggregate
1835 32 : .as_ref()
1836 32 : .and_then(|a| aggregate_expr(f, a, n_where.saturating_add(1)))
1837 : } else {
1838 473 : None
1839 : };
1840 505 : let aggregated = agg.is_some();
1841 505 : let (attr_expr, extra) = match agg {
1842 32 : Some((e, b)) => (format!("jsonb_build_object('$agg', {e})"), b),
1843 473 : None => match attr_object_expr(f, n_where.saturating_add(1)) {
1844 473 : Some(v) => v,
1845 : // refused range shape: reconstruct unpruned, window arbitrates.
1846 : // The default filter has neither range nor lastN, the two
1847 : // shapes the compiler can refuse: pinned by
1848 : // `the_unpruned_fallback_always_compiles`. A request reaches
1849 : // this line: a refusal is an error, never a panic.
1850 0 : None => attr_object_expr(&TemporalFilter::default(), n_where.saturating_add(1))
1851 0 : .ok_or_else(|| {
1852 0 : sqlx::Error::Protocol("the unpruned temporal shape did not compile".into())
1853 0 : })?,
1854 : },
1855 : };
1856 505 : binds.extend(extra.into_iter().map(B::Text));
1857 505 : let mut select_total = String::new();
1858 505 : let mut tail = " ORDER BY m.id".to_owned();
1859 : // `paged` is set only inside `if f.page.is_some()`, so binding the
1860 : // two together says that in the type system instead of in a comment.
1861 505 : if let (true, Some(page)) = (paged, f.page.as_ref()) {
1862 246 : select_total = ", count(*) OVER () AS total".into();
1863 246 : binds.push(B::Num(page.limit));
1864 246 : tail.push_str(&format!(" LIMIT ${}", binds.len()));
1865 246 : binds.push(B::Num(page.offset));
1866 246 : tail.push_str(&format!(" OFFSET ${}", binds.len()));
1867 289 : } else {
1868 259 : // No page pushed down: the caller still has to filter, so the only
1869 259 : // bound on this statement is the safety ceiling — without it a
1870 259 : // bare `?timerel=…` reconstructs every temporal entity of the
1871 259 : // tenant into one Vec.
1872 259 : binds.push(B::Num(super::entity::MAX_UNDECIDED_ROWS));
1873 259 : tail.push_str(&format!(" LIMIT ${}", binds.len()));
1874 259 : }
1875 505 : let sql = format!(
1876 : "SELECT m.meta || {attr_expr}{select_total} FROM temporal_entities m \
1877 : WHERE {where_sql}{tail}"
1878 : );
1879 505 : let first: Result<Option<TemporalOutcome>, sqlx::Error> = async {
1880 505 : let mut tx = super::begin(&self.pool).await?;
1881 505 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1882 : // `sql` is compiler literals + $n placeholders only.
1883 505 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone()));
1884 5603 : for b in &binds {
1885 5603 : qy = match b {
1886 4351 : B::Text(s) => qy.bind(s),
1887 484 : B::Arr(v) => qy.bind(v),
1888 751 : B::Num(n) => qy.bind(n),
1889 17 : B::Float(x) => qy.bind(x),
1890 : };
1891 : }
1892 505 : let rows = qy.fetch_all(&mut *tx).await?;
1893 505 : let mut total = if paged {
1894 246 : rows.first().map(|r| r.get::<i64, _>(1))
1895 : } else {
1896 259 : None
1897 : };
1898 505 : if paged && total.is_none() {
1899 : // offset past the end: the window function came back with the
1900 : // page, which is empty — count separately with the WHERE binds
1901 32 : let count_sql =
1902 32 : format!("SELECT count(*) FROM temporal_entities m WHERE {where_sql}");
1903 32 : let mut cq = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql));
1904 270 : for b in binds.iter().take(n_where) {
1905 270 : cq = match b {
1906 256 : B::Text(s) => cq.bind(s),
1907 14 : B::Arr(v) => cq.bind(v),
1908 0 : B::Num(n) => cq.bind(n),
1909 0 : B::Float(x) => cq.bind(x),
1910 : };
1911 : }
1912 32 : total = Some(cq.fetch_one(&mut *tx).await?);
1913 473 : }
1914 505 : tx.commit().await?;
1915 505 : super::entity::check_ceiling(paged, rows.len(), super::entity::MAX_UNDECIDED_ROWS)?;
1916 1102 : let mut docs: Vec<Value> = rows.into_iter().map(|r| r.get::<Value, _>(0)).collect();
1917 505 : docs.iter_mut().for_each(unpack);
1918 505 : if aggregated {
1919 34 : for d in &mut docs {
1920 34 : let Some(o) = d.as_object_mut() else { continue };
1921 34 : let Some(agg) = o.remove("$agg") else {
1922 0 : continue;
1923 : };
1924 34 : if agg.get("bad").and_then(Value::as_bool) == Some(true) {
1925 1 : return Ok(None);
1926 33 : }
1927 33 : if let Some(attrs) = agg.get("attrs").and_then(Value::as_object) {
1928 61 : for (k, v) in attrs {
1929 61 : o.insert(k.clone(), v.clone());
1930 61 : }
1931 0 : }
1932 : }
1933 473 : }
1934 504 : Ok(Some(TemporalOutcome {
1935 504 : rows: docs,
1936 504 : paged,
1937 504 : total,
1938 504 : aggregated,
1939 504 : }))
1940 505 : }
1941 505 : .await;
1942 505 : match first? {
1943 504 : Some(out) => Ok(out),
1944 : // the fallback re-runs the same statement without the pushed
1945 : // aggregation; boxed because an async fn cannot recurse inline
1946 1 : None => Box::pin(self.query_inner(tenant, f, false)).await,
1947 : }
1948 505 : }
1949 :
1950 : /// Single-entity fetch with the same instance pruning (Retrieve
1951 : /// Temporal Evolution, 5.7.3). `None` = entity absent.
1952 889 : pub async fn get_range(
1953 889 : &self,
1954 889 : tenant: &TenantId,
1955 889 : id: &str,
1956 889 : f: &TemporalFilter<'_>,
1957 889 : ) -> Result<Option<Value>, sqlx::Error> {
1958 : // Same unpruned fallback as `query_inner`, and compiling for the
1959 : // same reason: no range and no lastN is the shape that never refuses.
1960 : // A request reaches this line: a refusal here is an error, never a
1961 : // panic.
1962 889 : let (attr_expr, binds) = match attr_object_expr(f, 3) {
1963 889 : Some(v) => v,
1964 0 : None => attr_object_expr(&TemporalFilter::default(), 3).ok_or_else(|| {
1965 0 : sqlx::Error::Protocol("the unpruned temporal shape did not compile".into())
1966 0 : })?,
1967 : };
1968 : // 4.22: an expired entity is invalid → None (404), same as get().
1969 889 : let sql = format!(
1970 : "SELECT m.meta || {attr_expr} FROM temporal_entities m \
1971 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED}"
1972 : );
1973 889 : let mut tx = super::begin(&self.pool).await?;
1974 889 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1975 889 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
1976 889 : .bind(tenant.as_str())
1977 889 : .bind(id);
1978 889 : for b in &binds {
1979 694 : qy = qy.bind(b);
1980 694 : }
1981 889 : let row = qy.fetch_optional(&mut *tx).await?;
1982 889 : tx.commit().await?;
1983 889 : Ok(row.map(|r| {
1984 413 : let mut doc = r.get::<Value, _>(0);
1985 413 : unpack(&mut doc);
1986 413 : doc
1987 413 : }))
1988 889 : }
1989 :
1990 14 : pub async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, sqlx::Error> {
1991 14 : Ok(self.query(tenant, &TemporalFilter::default()).await?.rows)
1992 14 : }
1993 :
1994 : /// Row-locked read-modify-write over the RECONSTRUCTED doc, written back
1995 : /// as a DELTA: rows are diffed by (attr, instanceId) and only moved,
1996 : /// changed or removed instances touch the table — never a full-history
1997 : /// rewrite (the old resync's O(history) write amplification, and the
1998 : /// thing that fought hypertable compression).
1999 317 : pub async fn mutate<T, E>(
2000 317 : &self,
2001 317 : tenant: &TenantId,
2002 317 : id: &str,
2003 317 : f: impl FnOnce(&mut Value) -> Result<T, E>,
2004 317 : ) -> Result<Option<Result<T, E>>, sqlx::Error> {
2005 317 : let mut tx = super::begin(&self.pool).await?;
2006 317 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
2007 : // the meta row is the serialization point (FOR UPDATE). 4.22
2008 : // qualifies it: 5.6.12 to 5.6.15 reach the history through here,
2009 : // and none of them may modify an entity every read calls absent.
2010 317 : let row = sqlx::query(sqlx::AssertSqlSafe(format!(
2011 317 : "SELECT m.meta FROM temporal_entities m \
2012 317 : WHERE m.tenant_id = $1 AND m.id = $2 AND {NOT_EXPIRED} FOR UPDATE"
2013 317 : )))
2014 317 : .bind(tenant.as_str())
2015 317 : .bind(id)
2016 317 : .fetch_optional(&mut *tx)
2017 317 : .await?;
2018 317 : let Some(row) = row else {
2019 27 : tx.commit().await?;
2020 27 : return Ok(None);
2021 : };
2022 290 : let meta: Value = row.get(0);
2023 : // reconstruct the full doc inside the same transaction
2024 290 : let attrs_row = sqlx::query(sqlx::AssertSqlSafe(format!(
2025 290 : "SELECT COALESCE((SELECT jsonb_object_agg(g.attr_id, g.insts) FROM (
2026 290 : SELECT ai.attr_id,
2027 290 : {} AS insts
2028 290 : FROM {} ai
2029 290 : GROUP BY ai.attr_id) g), '{{}}'::jsonb)",
2030 290 : insts_and_ids("ai"),
2031 290 : history("$1", "$2")
2032 290 : )))
2033 290 : .bind(tenant.as_str())
2034 290 : .bind(id)
2035 290 : .fetch_one(&mut *tx)
2036 290 : .await?;
2037 290 : let mut attrs: Value = attrs_row.get(0);
2038 290 : unpack(&mut attrs);
2039 290 : let mut doc = meta;
2040 290 : if let (Some(d), Some(a)) = (doc.as_object_mut(), attrs.as_object()) {
2041 770 : for (k, v) in a {
2042 770 : d.insert(k.clone(), v.clone());
2043 770 : }
2044 0 : }
2045 290 : let before_rows = decompose(&doc);
2046 290 : match f(&mut doc) {
2047 288 : Ok(t) => {
2048 288 : let (types, scopes, _created, modified) = extract(&doc);
2049 288 : let after_rows = decompose(&doc);
2050 : // diff by logical identity (attr, instanceId); a changed
2051 : // observed_at moves the physical row → delete + insert
2052 4972 : let key = |r: &Value| -> (String, String) {
2053 4972 : (
2054 4972 : r["attr_id"].as_str().unwrap_or("").to_owned(),
2055 4972 : r["instance_id"].as_str().unwrap_or("").to_owned(),
2056 4972 : )
2057 4972 : };
2058 288 : let old: std::collections::HashMap<(String, String), &Value> =
2059 1470 : before_rows.iter().map(|r| (key(r), r)).collect();
2060 288 : let new: std::collections::HashMap<(String, String), &Value> =
2061 1751 : after_rows.iter().map(|r| (key(r), r)).collect();
2062 288 : let mut deletes: Vec<Value> = Vec::new();
2063 1470 : for (k, o) in &old {
2064 1470 : match new.get(k) {
2065 19 : None => deletes.push(serde_json::json!({"a": k.0, "i": k.1})),
2066 1451 : Some(n) if n["observed_at"] != o["observed_at"] => {
2067 0 : deletes.push(serde_json::json!({"a": k.0, "i": k.1}))
2068 : }
2069 1451 : _ => {}
2070 : }
2071 : }
2072 288 : let upserts: Vec<Value> = after_rows
2073 288 : .iter()
2074 1751 : .filter(|r| old.get(&key(r)).is_none_or(|o| *o != *r))
2075 288 : .cloned()
2076 288 : .collect();
2077 288 : if !deletes.is_empty() {
2078 : // 5.6.15 deletes an instance: its last version is kept,
2079 : // so a deletion is never without a record.
2080 13 : sqlx::query(sqlx::AssertSqlSafe(format!(
2081 13 : "INSERT INTO attr_versions
2082 13 : (tenant_id, entity_id, attr_id, instance_id, dataset_id, observed_at,
2083 13 : version, data, provenance, expires_at)
2084 13 : SELECT t.tenant_id, t.entity_id, t.attr_id, t.instance_id, t.dataset_id,
2085 13 : t.observed_at, t.version, {}, t.provenance, t.expires_at
2086 13 : FROM {} t JOIN jsonb_array_elements($3::jsonb) AS e
2087 13 : ON t.attr_id = e->>'a' AND t.instance_id = e->>'i'
2088 13 : ON CONFLICT DO NOTHING",
2089 13 : restamp("t"),
2090 13 : history("$1", "$2")
2091 13 : )))
2092 13 : .bind(tenant.as_str())
2093 13 : .bind(id)
2094 13 : .bind(Value::Array(deletes.clone()))
2095 13 : .execute(&mut *tx)
2096 13 : .await?;
2097 26 : for sql in [
2098 13 : "DELETE FROM attr_instances t
2099 13 : USING jsonb_array_elements($3::jsonb) AS e
2100 13 : WHERE t.tenant_id = $1 AND t.entity_id = $2
2101 13 : AND t.attr_id = e->>'a' AND t.instance_id = e->>'i'",
2102 13 : "DELETE FROM attr_rows t
2103 13 : USING jsonb_array_elements($3::jsonb) AS e
2104 13 : WHERE t.tenant_id = $1 AND t.entity_no = iri_no($1, $2)
2105 13 : AND t.attr_no = iri_no($1, e->>'a')
2106 13 : AND COALESCE(t.data->>'instanceId',
2107 13 : 'urn:ngsi-ld:Instance:' || t.instance_no::text) = e->>'i'",
2108 13 : ] {
2109 26 : sqlx::query(sql)
2110 26 : .bind(tenant.as_str())
2111 26 : .bind(id)
2112 26 : .bind(Value::Array(deletes.clone()))
2113 26 : .execute(&mut *tx)
2114 26 : .await?;
2115 : }
2116 275 : }
2117 288 : let mut ensured = std::collections::HashSet::new();
2118 288 : insert_rows(
2119 288 : &mut tx,
2120 288 : &self.keys,
2121 288 : &mut ensured,
2122 288 : tenant,
2123 288 : id,
2124 288 : upserts,
2125 288 : Route {
2126 288 : dedup: false,
2127 288 : short: self.keys.is_short(tenant, &types, &doc),
2128 288 : },
2129 288 : )
2130 288 : .await?;
2131 288 : sqlx::query(
2132 288 : "UPDATE temporal_entities SET meta = $3, types = $4, scopes = $5,
2133 288 : modified_at = $6::timestamptz
2134 288 : WHERE tenant_id = $1 AND id = $2",
2135 288 : )
2136 288 : .bind(tenant.as_str())
2137 288 : .bind(id)
2138 288 : .bind(meta_of(&doc))
2139 288 : .bind(&types)
2140 288 : .bind(&scopes)
2141 288 : .bind(&modified)
2142 288 : .execute(&mut *tx)
2143 288 : .await?;
2144 288 : tx.commit().await?;
2145 288 : Ok(Some(Ok(t)))
2146 : }
2147 2 : Err(e) => {
2148 2 : tx.rollback().await?;
2149 2 : Ok(Some(Err(e)))
2150 : }
2151 : }
2152 317 : }
2153 : }
2154 :
2155 : #[cfg(test)]
2156 : mod tests {
2157 : use super::*;
2158 : use crate::compile::temporal::InstanceRange;
2159 :
2160 400012 : fn pair(n: usize) -> (String, String) {
2161 400012 : ("t".to_owned(), format!("urn:ngsi-ld:E:{n}"))
2162 400012 : }
2163 :
2164 : /// the known set holds at most two generations; a pair read
2165 : /// while old survives the next ageing, one never read again is dropped.
2166 : #[test]
2167 2 : fn known_ages_by_generation_and_keeps_what_it_reads() {
2168 2 : let mut k = Known::default();
2169 200000 : for n in 0..KNOWN_GEN {
2170 200000 : k.insert(pair(n));
2171 200000 : }
2172 2 : k.insert(pair(KNOWN_GEN));
2173 2 : assert_eq!(k.old.len(), KNOWN_GEN);
2174 2 : assert_eq!(k.new.len(), 1);
2175 2 : assert!(k.contains(&pair(0)));
2176 2 : assert!(!k.old.contains(&pair(0)));
2177 199998 : for n in 1..KNOWN_GEN {
2178 199998 : k.insert(pair(KNOWN_GEN.saturating_add(n)));
2179 199998 : }
2180 2 : k.insert(pair(KNOWN_GEN.saturating_mul(3)));
2181 2 : assert!(k.contains(&pair(0)));
2182 2 : assert!(!k.contains(&pair(1)));
2183 2 : assert!(!k.contains(&pair(KNOWN_GEN.saturating_mul(4))));
2184 200004 : k.retain(|(t, _)| t != "t");
2185 2 : assert!(k.new.is_empty());
2186 2 : assert!(k.old.is_empty());
2187 2 : }
2188 :
2189 : /// the broker's own canonical id keys by its uuid; any other id,
2190 : /// an upper-case uuid included, keys by its v5 and stays in data.
2191 : #[test]
2192 2 : fn instance_no_keys_own_ids_by_their_uuid_and_others_by_v5() {
2193 2 : let own = "urn:ngsi-ld:Instance:0f8fad5b-d9cb-469f-a165-70867728950e";
2194 2 : let (no, is_own) = instance_no(own);
2195 2 : assert_eq!(
2196 2 : (no.to_string(), is_own),
2197 2 : ("0f8fad5b-d9cb-469f-a165-70867728950e".to_owned(), true)
2198 : );
2199 2 : let upper = "urn:ngsi-ld:Instance:0F8FAD5B-D9CB-469F-A165-70867728950E";
2200 6 : for other in [upper, "urn:client:speed:thirty", "urn:ngsi-ld:Instance:x"] {
2201 6 : let (no, is_own) = instance_no(other);
2202 6 : assert_eq!((no.get_version_num(), is_own), (5, false), "{other}");
2203 6 : assert_eq!(no, instance_no(other).0, "{other}: stable");
2204 : }
2205 2 : assert_ne!(
2206 2 : instance_no(upper).0,
2207 : no,
2208 : "a v5 never equals the uuid it spells"
2209 : );
2210 2 : }
2211 :
2212 : /// a row from the cut-over on is keyed; an earlier one, an
2213 : /// unreadable stamp and no cut-over stay in the text shape.
2214 : #[test]
2215 2 : fn is_keyed_routes_by_observed_at_against_the_cut_over() {
2216 2 : let cut = chrono::DateTime::parse_from_rfc3339("2026-10-26T00:00:00Z")
2217 2 : .ok()
2218 2 : .map(|t| t.to_utc());
2219 10 : let row = |at: &str| serde_json::json!({"observed_at": at});
2220 2 : let got = [
2221 2 : is_keyed(&row("2026-10-26T00:00:00Z"), cut),
2222 2 : is_keyed(&row("2026-10-26T01:00:00+02:00"), cut),
2223 2 : is_keyed(&row("2026-10-25T23:59:59Z"), cut),
2224 2 : is_keyed(&row("not a stamp"), cut),
2225 2 : is_keyed(&row("2027-01-01T00:00:00Z"), None),
2226 2 : ];
2227 2 : assert_eq!(got, [true, false, false, false, false]);
2228 2 : }
2229 :
2230 : /// the IRIs a batch append writes first are the Entity id and its
2231 : /// Attribute names, never a meta member.
2232 : #[test]
2233 2 : fn iris_of_names_the_entity_and_its_attributes() {
2234 2 : let doc = serde_json::json!({"id": "x", "type": "T", "createdAt": "2026-01-01T00:00:00Z",
2235 2 : "https://a": [{}], "https://b": {"not": "instances"}});
2236 2 : assert_eq!(
2237 2 : iris_of("urn:e", &doc),
2238 2 : vec!["urn:e".to_owned(), "https://a".to_owned()]
2239 : );
2240 2 : }
2241 :
2242 : /// a row's instanceId lives in its column; `unpack` puts it back
2243 : /// and leaves stored instances and meta members alone.
2244 : #[test]
2245 2 : fn instance_id_leaves_data_and_unpack_puts_it_back() {
2246 2 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed";
2247 2 : let inst = serde_json::json!({"type": "Property", "value": 1,
2248 2 : "instanceId": "urn:ngsi-ld:Instance:1"});
2249 2 : let rows = decompose(&serde_json::json!({"id": "urn:ngsi-ld:V:1", speed: [inst.clone()]}));
2250 2 : assert_eq!(rows.len(), 1);
2251 2 : assert_eq!(
2252 2 : rows[0]["data"],
2253 2 : serde_json::json!({"type": "Property", "value": 1})
2254 : );
2255 2 : assert_eq!(rows[0]["instance_id"], "urn:ngsi-ld:Instance:1");
2256 : // a row written keeps its instanceId in data
2257 2 : let stored = serde_json::json!({"type": "Property", "value": 2, "instanceId": "urn:ngsi-ld:Instance:2"});
2258 2 : let mut doc = serde_json::json!({
2259 2 : "id": "urn:ngsi-ld:V:1", "type": "V", "scope": ["/a", "/b"],
2260 2 : speed: [[rows[0]["data"].clone(), stored.clone()],
2261 : ["urn:ngsi-ld:Instance:1", "urn:ngsi-ld:Instance:2"]],
2262 2 : "short": [[{"value": 1}], ["x", "y"]],
2263 2 : "numbers": [[{"value": 1}], [1]]
2264 : });
2265 2 : unpack(&mut doc);
2266 2 : assert_eq!(
2267 : doc,
2268 2 : serde_json::json!({"id": "urn:ngsi-ld:V:1", "type": "V", "scope": ["/a", "/b"],
2269 2 : speed: [inst, stored],
2270 2 : "short": [[{"value": 1}], ["x", "y"]],
2271 2 : "numbers": [[{"value": 1}], [1]]})
2272 : );
2273 2 : let mut scalar = serde_json::json!(1);
2274 2 : unpack(&mut scalar);
2275 2 : assert_eq!(scalar, serde_json::json!(1));
2276 2 : }
2277 :
2278 : /// the dedup of an instance without observedAt reads the newest
2279 : /// stored row through the index and nothing older. A correlated max()
2280 : /// compared row by row read the attribute's whole history per write
2281 : /// (~90 ms a station at a month deep).
2282 : #[test]
2283 2 : fn the_newest_row_probe_reads_one_row() {
2284 2 : assert!(NEWEST_REPEATS.contains("ORDER BY x.observed_at DESC LIMIT 1"));
2285 2 : assert!(!NEWEST_REPEATS.to_lowercase().contains("max("));
2286 : // Its stamps are the ones every other comparison here ignores.
2287 2 : assert!(NEWEST_REPEATS.contains(&STAMPS["ARRAY".len()..]));
2288 2 : assert!(NEWEST_REPEATS.contains("x.tenant_id = $1 AND x.entity_id = $2"));
2289 2 : }
2290 :
2291 : /// Two callers fall back to `attr_object_expr(&TemporalFilter::default())`
2292 : /// when a range is outside the compiler's exact subset, and unwrap it:
2293 : /// the default filter has no range and no `lastN`, and `None` is returned
2294 : /// only for a range the compiler refuses. That is the reconstruct-unpruned
2295 : /// path, so if it ever stopped compiling the broker would panic on
2296 : /// precisely the queries the fallback exists to serve. Pinned here rather
2297 : /// than argued in a comment.
2298 : #[test]
2299 2 : fn the_unpruned_fallback_always_compiles() {
2300 2 : let f = TemporalFilter::default();
2301 2 : assert!(f.range.is_none(), "the default filter carries no range");
2302 2 : assert!(f.last_n.is_none(), "the default filter carries no lastN");
2303 8 : for first_bind in [1, 3, 4, 17] {
2304 8 : let built = attr_object_expr(&f, first_bind);
2305 8 : let (sql, binds) = built.unwrap_or_else(|| {
2306 0 : panic!("the unpruned fallback failed to compile at bind {first_bind}")
2307 : });
2308 8 : assert!(!sql.is_empty(), "empty SQL at bind {first_bind}");
2309 8 : assert!(
2310 8 : !sql.contains('$'),
2311 : "the unpruned fragment names no placeholder: {sql}"
2312 : );
2313 8 : assert!(
2314 8 : binds.is_empty(),
2315 : "a fragment that names no placeholder must hand back no bind: {binds:?}"
2316 : );
2317 : }
2318 2 : }
2319 :
2320 2 : fn between<'a>() -> TemporalFilter<'a> {
2321 2 : TemporalFilter {
2322 2 : range: Some(InstanceRange {
2323 2 : timerel: "between",
2324 2 : time_at: "2026-01-01T00:00:00Z",
2325 2 : end_time_at: Some("2026-02-01T00:00:00Z"),
2326 2 : timeproperty: "observedAt",
2327 2 : }),
2328 2 : ..Default::default()
2329 2 : }
2330 2 : }
2331 :
2332 : /// The temporal tables extract their four columns with the same function
2333 : /// the entity table uses, so what holds there holds here: `type` and
2334 : /// `scope` read as arrays whichever shape the document spells them in,
2335 : /// and 4.6.3's comma seconds-fraction is canonicalised. A stamp that
2336 : /// reaches a `::timestamptz` bind with its comma intact is rejected by
2337 : /// Postgres, and the append fails on a document the API accepted.
2338 : #[test]
2339 2 : fn extract_canonicalises_the_stamps_and_takes_type_in_either_shape() {
2340 2 : let (types, scopes, created, modified) = extract(&serde_json::json!({
2341 2 : "id": "urn:x", "type": "T", "scope": ["/a", "/b"],
2342 2 : "createdAt": "2026-01-01T00:00:00,500Z",
2343 2 : "modifiedAt": "2026-01-01T00:00:01.500Z"}));
2344 2 : assert_eq!(types, ["T"]);
2345 2 : assert_eq!(scopes.expect("scopes"), ["/a", "/b"]);
2346 2 : assert_eq!(
2347 : created, "2026-01-01T00:00:00.500Z",
2348 : "the comma is the fraction"
2349 : );
2350 2 : assert_eq!(modified, "2026-01-01T00:00:01.500Z");
2351 2 : let (types, scopes, created, _) = extract(&serde_json::json!({"id": "urn:x"}));
2352 2 : assert!(types.is_empty(), "no type is no types, never a panic");
2353 2 : assert!(
2354 2 : scopes.is_none(),
2355 : "an absent scope stays absent, not an empty list"
2356 : );
2357 2 : assert_eq!(
2358 : created, "1970-01-01T00:00:00Z",
2359 : "the belt stamp is usable as a bind"
2360 : );
2361 2 : }
2362 :
2363 : #[test]
2364 2 : fn meta_of_keeps_only_meta_members() {
2365 2 : let doc = serde_json::json!({
2366 2 : "id": "urn:x", "type": ["T"], "createdAt": "c", "modifiedAt": "m",
2367 2 : "https://a/attr": [{"instanceId": "i1"}]
2368 : });
2369 2 : let m = meta_of(&doc);
2370 2 : assert!(m.get("id").is_some() && m.get("https://a/attr").is_none());
2371 2 : }
2372 :
2373 : #[test]
2374 2 : fn range_binds_are_numbered_from_first_bind() {
2375 2 : let (expr, binds) = attr_object_expr(&between(), 2).expect("compiles");
2376 2 : assert_eq!(
2377 : binds,
2378 2 : vec!["observedAt", "2026-01-01T00:00:00Z", "2026-02-01T00:00:00Z"]
2379 : );
2380 2 : assert!(
2381 2 : expr.contains("$2") && expr.contains("$3") && expr.contains("$4"),
2382 : "{expr}"
2383 : );
2384 : // predicates run on the instance JSON, never the partition column
2385 2 : assert!(expr.contains("ai.data ->>"), "{expr}");
2386 2 : }
2387 :
2388 : #[test]
2389 2 : fn last_n_caps_with_rank_per_attr_and_dataset_never_row_number() {
2390 2 : let f = TemporalFilter {
2391 2 : last_n: Some(5),
2392 2 : ..Default::default()
2393 2 : };
2394 2 : let (expr, binds) = attr_object_expr(&f, 1).expect("compiles");
2395 : // RANK keeps timestamp ties; ROW_NUMBER would cut an instance the
2396 : // API-side per-attr lastN still wants (the tie-break divergence bug)
2397 2 : assert!(expr.contains("rank() OVER"), "{expr}");
2398 2 : assert!(!expr.contains("row_number"), "{expr}");
2399 2 : assert!(
2400 2 : expr.contains("PARTITION BY ai.attr_id, ai.data ->> 'datasetId'"),
2401 : "{expr}"
2402 : );
2403 2 : assert!(expr.contains("COLLATE \"C\" DESC NULLS LAST"), "{expr}");
2404 2 : assert_eq!(binds, vec!["observedAt", "5"]);
2405 2 : }
2406 :
2407 : /// lastN's distinct-Attribute skip scan keeps to the window's
2408 : /// column bound in both parents, so it reads the weeks the window
2409 : /// reaches and not every week of the retention; no window, no bound.
2410 : #[test]
2411 2 : fn the_attribute_skip_scan_keeps_to_the_window() {
2412 2 : let after = TemporalFilter {
2413 2 : last_n: Some(10),
2414 2 : range: Some(InstanceRange {
2415 2 : timerel: "after",
2416 2 : time_at: "2026-10-24T00:00:00Z",
2417 2 : end_time_at: None,
2418 2 : timeproperty: "observedAt",
2419 2 : }),
2420 2 : ..Default::default()
2421 2 : };
2422 2 : let sql = newest_rows(&after, 3, 7).expect("after reads the newest rows");
2423 2 : assert!(!sql.contains("window}"), "{sql}");
2424 6 : for alias in ["x", "r", "t"] {
2425 6 : let bound = format!("{alias}.observed_at >= $4::timestamptz - interval '48 hours'");
2426 6 : assert!(sql.contains(&bound), "{bound} missing: {sql}");
2427 : }
2428 2 : assert_eq!(sql.matches("x.observed_at >= $4").count(), 2, "{sql}");
2429 2 : assert_eq!(sql.matches("r.observed_at >= $4").count(), 2, "{sql}");
2430 2 : let unbounded = TemporalFilter {
2431 2 : last_n: Some(10),
2432 2 : ..Default::default()
2433 2 : };
2434 2 : let sql = newest_rows(&unbounded, 3, 4).expect("no range reads the newest rows");
2435 2 : assert!(!sql.contains("window}"), "{sql}");
2436 2 : assert!(!sql.contains("observed_at >= $"), "{sql}");
2437 2 : }
2438 :
2439 : /// the page qualification probes each parent newest first (oldest
2440 : /// first for `before`), one row along its btree, never an EXISTS over the
2441 : /// union that reads an Entity's rows in the btree's own order.
2442 : #[test]
2443 2 : fn the_page_qualification_probes_one_row_per_parent() {
2444 2 : let sql = has_history(" AND ai.x", "DESC", None);
2445 2 : assert!(
2446 2 : sql.contains("ORDER BY ai.attr_id DESC, ai.observed_at DESC LIMIT 1"),
2447 : "{sql}"
2448 : );
2449 2 : assert!(
2450 2 : sql.contains("ORDER BY ai.attr_no DESC, ai.observed_at DESC LIMIT 1"),
2451 : "{sql}"
2452 : );
2453 2 : assert_eq!(sql.matches(" AND ai.x").count(), 2, "{sql}");
2454 2 : assert!(!sql.contains("EXISTS"), "{sql}");
2455 2 : assert!(!sql.contains("ANY("), "{sql}");
2456 2 : }
2457 :
2458 : /// under a policy the page qualifies an Entity only through the
2459 : /// Attributes its rules show, in both parents, on the one bind.
2460 : #[test]
2461 2 : fn the_page_qualification_keeps_to_the_shown_attributes() {
2462 2 : let sql = has_history(" AND ai.x", "DESC", Some(9));
2463 2 : assert!(sql.contains("ai.attr_id = ANY($9)"), "{sql}");
2464 2 : assert!(
2465 2 : sql.contains("ai.attr_no IN (SELECT i.id FROM iri_ids i WHERE i.tenant_id = m.tenant_id AND i.iri = ANY($9))"),
2466 : "{sql}"
2467 : );
2468 2 : }
2469 :
2470 : /// the read takes the N newest alone only where the Entity's
2471 : /// last non-UTC stamp lies more than 28 h before the N-th newest.
2472 : #[test]
2473 2 : fn the_newest_rows_widen_only_near_an_offset_stamp() {
2474 2 : let f = TemporalFilter {
2475 2 : last_n: Some(10),
2476 2 : ..Default::default()
2477 2 : };
2478 2 : let sql = newest_rows(&f, 3, 4).expect("reads the newest rows");
2479 2 : assert!(
2480 2 : sql.contains(
2481 2 : "CASE WHEN k.nth - interval '28 hours' > m.offset_until THEN k.nth \
2482 2 : ELSE k.nth - interval '28 hours' END"
2483 2 : ),
2484 : "{sql}"
2485 : );
2486 2 : assert!(sql.contains("ai.observed_at >= c.th"), "{sql}");
2487 2 : }
2488 :
2489 : /// a UTC RFC 3339 stamp keys like its instant; an offset, a space
2490 : /// separator, a non-stamp or no observedAt does not.
2491 : #[test]
2492 2 : fn only_a_utc_t_stamp_orders_like_its_instant() {
2493 18 : for (data, utc) in [
2494 2 : (
2495 2 : serde_json::json!({"observedAt": "2026-10-26T12:00:00Z"}),
2496 2 : true,
2497 2 : ),
2498 2 : (
2499 2 : serde_json::json!({"observedAt": "2026-10-26T12:00:00,5Z"}),
2500 2 : true,
2501 2 : ),
2502 2 : (
2503 2 : serde_json::json!({"observedAt": "2026-10-26T12:00:00.123456789Z"}),
2504 2 : true,
2505 2 : ),
2506 2 : (
2507 2 : serde_json::json!({"observedAt": "2026-10-26T12:00:00+00:00"}),
2508 2 : false,
2509 2 : ),
2510 2 : (
2511 2 : serde_json::json!({"observedAt": "2026-10-26T14:00:00+02:00"}),
2512 2 : false,
2513 2 : ),
2514 2 : (
2515 2 : serde_json::json!({"observedAt": "2026-10-26 12:00:00Z"}),
2516 2 : false,
2517 2 : ),
2518 2 : (serde_json::json!({"observedAt": "yesterdayZ"}), false),
2519 2 : (serde_json::json!({"observedAt": 5}), false),
2520 2 : (serde_json::json!({"value": 1}), false),
2521 2 : ] {
2522 18 : assert_eq!(utc_observed(&data), utc, "{data}");
2523 : }
2524 2 : }
2525 :
2526 : #[test]
2527 2 : fn first_n_ranks_the_oldest_first() {
2528 2 : let f = TemporalFilter {
2529 2 : last_n: Some(5),
2530 2 : oldest_first: true,
2531 2 : ..Default::default()
2532 2 : };
2533 2 : let (expr, binds) = attr_object_expr(&f, 1).expect("compiles");
2534 2 : assert!(expr.contains("COLLATE \"C\" ASC NULLS FIRST"), "{expr}");
2535 2 : assert!(!expr.contains("DESC"), "{expr}");
2536 2 : assert_eq!(binds, vec!["observedAt", "5"]);
2537 2 : }
2538 :
2539 : #[test]
2540 2 : fn refused_range_shape_refuses_the_pruning() {
2541 2 : let f = TemporalFilter {
2542 2 : range: Some(InstanceRange {
2543 2 : timerel: "since", // not a 4.11 relation
2544 2 : time_at: "t",
2545 2 : end_time_at: None,
2546 2 : timeproperty: "observedAt",
2547 2 : }),
2548 2 : last_n: Some(3),
2549 2 : ..Default::default()
2550 2 : };
2551 : // half-pruning would silently skip the range: refuse instead
2552 2 : assert!(attr_object_expr(&f, 1).is_none());
2553 2 : }
2554 :
2555 : /// S3-06: without observedAt a row is dropped only when it repeats the
2556 : /// row kept before it in its (attribute, datasetId), as NEWEST_REPEATS
2557 : /// judges against the table: a value that comes back after another is
2558 : /// a new observation, and only a batch's first row of a series is held
2559 : /// against the table.
2560 : #[test]
2561 2 : fn unseen_keeps_a_value_that_returns_after_another() {
2562 12 : let row = |v: i64, i: &str| {
2563 12 : serde_json::json!({"attr_id": "a", "dataset_id": null,
2564 12 : "instance_id": i, "observed_at": "2026-01-01T00:00:00Z",
2565 12 : "data": {"type": "Property", "value": v}})
2566 12 : };
2567 2 : let kept = unseen(vec![row(1, "i1"), row(2, "i2"), row(1, "i3")]);
2568 2 : assert_eq!(kept.len(), 3, "{kept:?}");
2569 2 : let kept = unseen(vec![row(1, "i1"), row(1, "i2"), row(2, "i3")]);
2570 4 : let ids: Vec<_> = kept.iter().map(|r| r["instance_id"].clone()).collect();
2571 2 : assert_eq!(ids, ["i1", "i3"], "a repeat in a row is still one");
2572 2 : }
2573 :
2574 : #[test]
2575 2 : fn decompose_never_parks_deletion_instances_on_the_epoch() {
2576 2 : let doc = serde_json::json!({
2577 2 : "id": "urn:x",
2578 2 : "https://a/attr": [{
2579 2 : "type": "Property", "value": "urn:ngsi-ld:null",
2580 2 : "instanceId": "urn:ngsi-ld:Instance:d1",
2581 2 : "deletedAt": "2026-08-08T10:00:00Z"
2582 : }]
2583 : });
2584 2 : let rows = decompose(&doc);
2585 2 : assert_eq!(rows.len(), 1);
2586 2 : assert_eq!(rows[0]["observed_at"], "2026-08-08T10:00:00Z");
2587 2 : }
2588 : }
|
