Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Temporal maintenance: the broker's own scheduled
3 : //! job replaces TimescaleDB background workers in plain mode, and drives the
4 : //! retention knob in both modes.
5 : //!
6 : //! Single-winner rule: the run is claimed via
7 : //! `SELECT … FOR UPDATE SKIP LOCKED` on the `maintenance_jobs` row — N
8 : //! instances race, one wins, the rest skip. No coordinator.
9 : //!
10 : //! Plain-mode partitioning: weekly partitions are pre-created for a window
11 : //! around now; everything else (historic backfill) lands in the DEFAULT
12 : //! partition. Creating a partition whose range already has rows sitting in
13 : //! the DEFAULT partition fails in PostgreSQL, so a single row written with an
14 : //! `observedAt` past the window permanently blocks that week's partition and
15 : //! sends all of its later traffic to DEFAULT as well. Such a range is
16 : //! recovered rather than skipped: the rows are moved out of DEFAULT into a
17 : //! standalone table, which is then ATTACHed as the partition.
18 : //!
19 : //! The recovery belongs here and NOT at ingest, for two reasons. The first is
20 : //! internal: clamping an `observedAt` outside a horizon would let the
21 : //! `observed_at` column disagree with the raw timestamp string in `data`, and
22 : //! `compile::temporal::column_range_bound` may prune on that column only
23 : //! because it is a superset of the byte-exact text window (4.11). The second
24 : //! is normative: 4.8 defines `observedAt` as "the temporal Property at which a
25 : //! certain Property or Relationship became valid or was observed" and requires
26 : //! only that it be a 4.6.3 DateTime — there is no horizon and no error type
27 : //! for one, and a forecast Property legitimately becomes valid in the future.
28 : //! A well-formed future `observedAt` is therefore valid input the broker
29 : //! stores, not input it may refuse.
30 : //!
31 : //! Residual, deliberate: rows whose `observed_at` lies beyond the pre-created
32 : //! window stay in DEFAULT until their week enters it, and rows dated far
33 : //! enough ahead stay there indefinitely — retention only purges DEFAULT rows
34 : //! that are already OLD. Auto-adopting arbitrary future weeks is not the fix
35 : //! either: one row per week over a century would trade an oversized DEFAULT
36 : //! for thousands of partitions. The condition is reported instead (see
37 : //! `default_partition_load`), so an operator sees it rather than discovering
38 : //! it as a query slowdown.
39 :
40 : use sqlx::postgres::PgPool;
41 : use sqlx::{Acquire, Row};
42 :
43 : /// True when the timescaledb extension is CREATED in this database
44 : /// (per-database `pg_extension`, not "installed on the server").
45 2 : pub async fn timescale_present(pool: &PgPool) -> Result<bool, sqlx::Error> {
46 2 : let row = sqlx::query("SELECT 1 FROM pg_extension WHERE extname = 'timescaledb'")
47 2 : .fetch_optional(pool)
48 2 : .await?;
49 2 : Ok(row.is_some())
50 2 : }
51 :
52 : /// What the migrations actually built `attr_instances` as. Detected ONCE at
53 : /// startup from the catalog and pinned — never re-probed per tick, so the
54 : /// maintenance branch can never disagree with the DDL on disk (the
55 : /// "extension installed after first boot" trap).
56 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
57 : pub enum TemporalBackend {
58 : /// timescale hypertable (relkind 'r' + timescaledb catalog entry)
59 : Hypertable,
60 : /// native PARTITION BY RANGE (relkind 'p')
61 : Partitioned,
62 : }
63 :
64 : /// Inspect the catalog. Errors when `attr_instances` is neither a hypertable
65 : /// nor a partitioned table — that means the database was migrated under one
66 : /// extension state and is now running under another; refusing beats running
67 : /// the wrong maintenance jobs against mismatched DDL.
68 12 : pub async fn detect_temporal_backend(pool: &PgPool) -> Result<TemporalBackend, String> {
69 12 : let relkind: Option<String> = sqlx::query_scalar(
70 12 : "SELECT c.relkind::text FROM pg_class c
71 12 : WHERE c.relname = 'attr_instances' AND c.relnamespace = 'public'::regnamespace",
72 12 : )
73 12 : .fetch_optional(pool)
74 12 : .await
75 12 : .map_err(|e| e.to_string())?;
76 12 : match relkind.as_deref() {
77 12 : Some("p") => Ok(TemporalBackend::Partitioned),
78 4 : Some("r") => {
79 4 : let hyper = sqlx::query(
80 4 : "SELECT 1 FROM timescaledb_information.hypertables
81 4 : WHERE hypertable_name = 'attr_instances'",
82 4 : )
83 4 : .fetch_optional(pool)
84 4 : .await
85 4 : .map_err(|e| e.to_string())?;
86 4 : if hyper.is_some() {
87 4 : Ok(TemporalBackend::Hypertable)
88 : } else {
89 0 : Err(
90 0 : "attr_instances is a plain table — the database was migrated as a \
91 0 : hypertable and the timescaledb extension has since been removed, or \
92 0 : the catalog is damaged; refusing to run temporal maintenance"
93 0 : .into(),
94 0 : )
95 : }
96 : }
97 0 : other => Err(format!(
98 0 : "attr_instances has unexpected relkind {other:?} — migrations did not run?"
99 0 : )),
100 : }
101 12 : }
102 :
103 : /// How long a published claim-check row outlives its message. The changes
104 : /// stream keeps a message until every durable has acked it, so no bound is
105 : /// derivable from the bus — this is the operator-visible ceiling on how long
106 : /// a matcher may lag before an oversized change stops resolving.
107 : // ponytail: fixed window, an env knob if a deployment's matcher lag exceeds it
108 : const CLAIM_CHECK_HOURS: i64 = 24;
109 :
110 : /// One maintenance pass. Returns a short human-readable summary ("skipped"
111 : /// when another instance holds the claim). `retention_days = None` keeps
112 : /// history forever — retention is a deliberate deployment knob, never a
113 : /// default.
114 397 : pub async fn temporal_maintenance(
115 397 : pool: &PgPool,
116 397 : backend: TemporalBackend,
117 397 : retention_days: Option<i64>,
118 397 : ) -> Result<String, sqlx::Error> {
119 397 : let mut tx = pool.begin().await?;
120 397 : let claimed = sqlx::query(
121 397 : "SELECT name FROM maintenance_jobs WHERE name = 'temporal_partitions'
122 397 : FOR UPDATE SKIP LOCKED",
123 397 : )
124 397 : .fetch_optional(&mut *tx)
125 397 : .await?;
126 397 : if claimed.is_none() {
127 0 : return Ok("skipped: another instance holds the claim".into());
128 397 : }
129 : // retention DML is cross-tenant service work (the `antares.service`
130 : // escape in 0001_init.sql)
131 397 : crate::store::pg::set_service(&mut tx).await?;
132 397 : let mut done: Vec<String> = Vec::new();
133 : // The 4.22 reaps run in their own transactions AFTER this one commits: both
134 : // DELETEs grow with stored volume and can exceed the connection's
135 : // statement_timeout, and a reap that times out must not abort the partition
136 : // pre-creation that keeps ingest writable.
137 397 : if backend == TemporalBackend::Hypertable {
138 188 : if let Some(days) = retention_days {
139 0 : sqlx::query("SELECT public.drop_chunks('attr_instances', older_than => make_interval(days => $1::int))")
140 0 : .bind(days)
141 0 : .execute(&mut *tx)
142 0 : .await?;
143 0 : done.push(format!("timescale drop_chunks older than {days}d"));
144 188 : }
145 : } else {
146 : // weekly partitions for [now-1w, now+4w)
147 1045 : for off in -1i64..4 {
148 1045 : let row = sqlx::query(
149 1045 : "SELECT to_char(date_trunc('week', now()) + make_interval(weeks => $1::int), 'IYYY\"w\"IW') AS suffix,
150 1045 : (date_trunc('week', now()) + make_interval(weeks => $1::int))::text AS lo,
151 1045 : (date_trunc('week', now()) + make_interval(weeks => ($1::int) + 1))::text AS hi",
152 1045 : )
153 1045 : .bind(off)
154 1045 : .fetch_one(&mut *tx)
155 1045 : .await?;
156 1045 : let (suffix, lo, hi): (String, String, String) =
157 1045 : (row.get("suffix"), row.get("lo"), row.get("hi"));
158 : // The failure below is EXPECTED (see module docs), and in
159 : // PostgreSQL a failed statement aborts the whole transaction —
160 : // every later one then returns 25P02. Tolerating an error means
161 : // owning a savepoint to roll back to; without it the first
162 : // already-occupied range poisons the entire maintenance pass.
163 1045 : let mut sp = tx.begin().await?;
164 1045 : match sqlx::query(sqlx::AssertSqlSafe(create_partition_sql(&suffix, &lo, &hi)))
165 1045 : .execute(&mut *sp)
166 1045 : .await
167 : {
168 : Ok(_) => {
169 1044 : sp.commit().await?;
170 1044 : done.push(format!("partition {suffix}: ok"));
171 1044 : continue;
172 : }
173 1 : Err(_) => sp.rollback().await?,
174 : }
175 : // Rows for this range already sit in DEFAULT. Adopt them: the move
176 : // empties the range, so the ATTACH's revalidation of DEFAULT
177 : // passes. Its ACCESS EXCLUSIVE lock is bounded by the connection's
178 : // lock_timeout, and a loser simply retries on the next tick.
179 1 : let mut sp = tx.begin().await?;
180 1 : let mut adopted = Ok(());
181 3 : for stmt in adopt_default_rows_sql(&suffix, &lo, &hi) {
182 3 : adopted = sqlx::query(sqlx::AssertSqlSafe(stmt))
183 3 : .execute(&mut *sp)
184 3 : .await
185 3 : .map(|_| ());
186 3 : if adopted.is_err() {
187 0 : break;
188 3 : }
189 : }
190 1 : match adopted {
191 : Ok(()) => {
192 1 : sp.commit().await?;
193 1 : done.push(format!("partition {suffix}: adopted from default"));
194 : }
195 0 : Err(e) => {
196 : // warn, not debug: a PERMANENTLY failing create
197 : // (permissions) must be visible at default log level, or it
198 : // silently degrades to "everything lands in DEFAULT".
199 0 : sp.rollback().await?;
200 0 : tracing::warn!("partition attr_instances_{suffix} not created: {e}");
201 0 : done.push(format!("partition {suffix}: left in default"));
202 : }
203 : }
204 : }
205 : }
206 397 : sqlx::query("UPDATE maintenance_jobs SET last_run = now() WHERE name = 'temporal_partitions'")
207 397 : .execute(&mut *tx)
208 397 : .await?;
209 397 : tx.commit().await?;
210 : // Plain-mode retention runs AFTER the claim transaction commits, for the
211 : // same reason as the 4.22 reaps: DROP TABLE needs ACCESS EXCLUSIVE on the
212 : // parent and the DEFAULT purge grows with stored volume, so one lock
213 : // contention or one statement timeout would otherwise roll back the
214 : // partition pre-creation that keeps ingest writable.
215 397 : if backend == TemporalBackend::Partitioned {
216 209 : if let Some(days) = retention_days {
217 0 : match plain_retention(pool, days).await {
218 0 : Ok(lines) => done.extend(lines),
219 0 : Err(e) => done.push(format!("retention skipped ({e})")),
220 : }
221 209 : }
222 209 : if let Ok(Some(line)) = default_partition_load(pool).await {
223 0 : done.push(line);
224 209 : }
225 188 : }
226 : // 4.22 garbage collection, on both backends: reads already refuse expired
227 : // entities and instances, so these reaps only bound storage — the clause
228 : // itself sanctions deletion lagging expiresAt.
229 397 : match reap_expired_entities(pool).await {
230 393 : Ok(0) => {}
231 4 : Ok(n) => done.push(format!("reaped {n} expired transient entities (4.22)")),
232 0 : Err(e) => done.push(format!("entity reap skipped ({e})")),
233 : }
234 : // 4.22 also names Properties/Relationships: an attribute instance whose
235 : // expiresAt has passed "should be deleted from an NGSI-LD system". This
236 : // DELETE additionally contends with concurrent ingest on the same rows, and
237 : // a deadlock must cost only the reap (observed under ~1.2k msg/s ingest).
238 397 : match reap_expired_instances(pool).await {
239 394 : Ok(0) => {}
240 3 : Ok(n) => done.push(format!("reaped {n} expired attribute instances (4.22)")),
241 0 : Err(e) => done.push(format!("instance reap skipped ({e})")),
242 : }
243 : // Claim-check rows the drain kept because the bus could not carry their
244 : // bodies. The consumer resolves them off the message, so the window has to
245 : // outlive the message: reaping one early costs the notification it was
246 : // carrying, keeping one costs a row.
247 397 : match crate::store::pg::outbox::reap_published(pool, CLAIM_CHECK_HOURS).await {
248 397 : Ok(0) => {}
249 0 : Ok(n) => done.push(format!("reaped {n} published claim-check rows")),
250 0 : Err(e) => done.push(format!("claim-check reap skipped ({e})")),
251 : }
252 397 : if done.is_empty() {
253 185 : done.push("nothing to do".into());
254 394 : }
255 397 : Ok(done.join("; "))
256 397 : }
257 :
258 : /// Plain-mode retention, in its own transaction: drop every weekly partition
259 : /// whose whole range is older than the horizon, then purge the DEFAULT
260 : /// partition (which has no upper bound and is therefore never dropped) of the
261 : /// historic-backfill rows that landed there before their week existed.
262 : ///
263 : /// Both statements are idempotent, so a run that loses a lock simply repeats
264 : /// on the next tick. Service role: retention is cross-tenant work.
265 0 : async fn plain_retention(pool: &PgPool, days: i64) -> Result<Vec<String>, sqlx::Error> {
266 0 : let mut done: Vec<String> = Vec::new();
267 0 : let mut tx = pool.begin().await?;
268 0 : crate::store::pg::set_service(&mut tx).await?;
269 0 : let parts = sqlx::query(
270 0 : // The parent is pinned to the schema the migrations built. Without
271 0 : // it a same-named table in another schema contributes its children,
272 0 : // and the DROP below would name them by bare relname.
273 0 : "SELECT c.relname,
274 0 : pg_get_expr(c.relpartbound, c.oid) AS bound
275 0 : FROM pg_inherits i
276 0 : JOIN pg_class c ON c.oid = i.inhrelid
277 0 : JOIN pg_class p ON p.oid = i.inhparent
278 0 : WHERE p.relname = 'attr_instances'
279 0 : AND p.relnamespace = 'public'::regnamespace",
280 0 : )
281 0 : .fetch_all(&mut *tx)
282 0 : .await?;
283 0 : for r in parts {
284 0 : let name: String = r.get("relname");
285 0 : let bound: String = r.get::<Option<String>, _>("bound").unwrap_or_default();
286 : // bound looks like: FOR VALUES FROM ('<lo>') TO ('<hi>')
287 0 : let Some(hi) = bound
288 0 : .split("TO ('")
289 0 : .nth(1)
290 0 : .and_then(|s| s.split('\'').next())
291 : else {
292 0 : continue; // DEFAULT partition — never dropped
293 : };
294 0 : let expired: bool =
295 0 : sqlx::query_scalar("SELECT $1::timestamptz < now() - make_interval(days => $2::int)")
296 0 : .bind(hi)
297 0 : .bind(days)
298 0 : .fetch_one(&mut *tx)
299 0 : .await?;
300 0 : if expired {
301 0 : sqlx::query(sqlx::AssertSqlSafe(drop_partition_sql(&name)))
302 0 : .execute(&mut *tx)
303 0 : .await?;
304 0 : done.push(format!("dropped expired partition {name}"));
305 0 : }
306 : }
307 0 : let purged = sqlx::query(
308 0 : "DELETE FROM attr_instances_default
309 0 : WHERE observed_at < now() - make_interval(days => $1::int)",
310 0 : )
311 0 : .bind(days)
312 0 : .execute(&mut *tx)
313 0 : .await?
314 0 : .rows_affected();
315 0 : tx.commit().await?;
316 0 : if purged > 0 {
317 0 : done.push(format!(
318 0 : "purged {purged} expired rows from DEFAULT partition"
319 0 : ));
320 0 : }
321 0 : Ok(done)
322 0 : }
323 :
324 : /// Rows the DEFAULT partition holds, and a warning once it stops being
325 : /// incidental. Every row there is one no weekly partition covers — historic
326 : /// backfill, or an `observedAt` dated beyond the pre-created window — so the
327 : /// count is what makes an unpartitioned pile visible before it shows up as a
328 : /// query that stopped pruning. `reltuples` is the planner's estimate, so this
329 : /// costs a catalog lookup rather than a scan of the pile it is measuring;
330 : /// `-1` means "never analysed", which is reported as nothing.
331 : const DEFAULT_PARTITION_WARN_ROWS: i64 = 100_000;
332 :
333 209 : async fn default_partition_load(pool: &PgPool) -> Result<Option<String>, sqlx::Error> {
334 209 : let rows: Option<f32> = sqlx::query_scalar(
335 209 : "SELECT reltuples FROM pg_class
336 209 : WHERE relname = 'attr_instances_default' AND relnamespace = 'public'::regnamespace",
337 209 : )
338 209 : .fetch_optional(pool)
339 209 : .await?;
340 209 : let est = rows.unwrap_or(-1.0) as i64;
341 209 : if est < DEFAULT_PARTITION_WARN_ROWS {
342 209 : return Ok(None);
343 0 : }
344 0 : tracing::warn!(
345 : "attr_instances_default holds ~{est} rows: instances are being written outside the \
346 : maintained partition window (historic backfill, or an observedAt far in the future); \
347 : queries over those ranges cannot prune"
348 : );
349 0 : Ok(Some(format!("default partition ~{est} rows")))
350 209 : }
351 :
352 : /// The 4.22 expired-entity DELETE, isolated so a reap that outruns
353 : /// `statement_timeout` costs only itself. Served by the partial index on
354 : /// `expires_at` (0001_init.sql) — without it this is a sequential scan of
355 : /// every entity in the deployment. Service role: the reap is cross-tenant work
356 : /// (RLS would hide other tenants' rows).
357 397 : async fn reap_expired_entities(pool: &PgPool) -> Result<u64, sqlx::Error> {
358 397 : let mut tx = pool.begin().await?;
359 397 : crate::store::pg::set_service(&mut tx).await?;
360 397 : let n = sqlx::query("DELETE FROM entities WHERE expires_at IS NOT NULL AND expires_at < now()")
361 397 : .execute(&mut *tx)
362 397 : .await?
363 397 : .rows_affected();
364 397 : tx.commit().await?;
365 397 : Ok(n)
366 397 : }
367 :
368 : /// Drop one partition, named the way the catalog row spells it. The name
369 : /// comes back from `pg_class.relname` as the identifier itself, not as SQL:
370 : /// unquoted it would be folded to lower case and would need to be a bare
371 : /// identifier to parse at all, so a partition whose name is neither — one
372 : /// created by hand, or by a migration that quoted it — is a partition
373 : /// retention fails on, every tick, for as long as it exists. Schema-
374 : /// qualified for the same reason the catalog query is: the row names a table
375 : /// in `public`, and `search_path` is not what should decide which one.
376 6 : fn drop_partition_sql(name: &str) -> String {
377 6 : format!("DROP TABLE public.\"{}\"", name.replace('"', "\"\""))
378 6 : }
379 :
380 : /// One weekly partition, created directly under the parent. Fails while the
381 : /// DEFAULT partition still holds a row in `[lo, hi)`.
382 1047 : fn create_partition_sql(suffix: &str, lo: &str, hi: &str) -> String {
383 1047 : format!(
384 : "CREATE TABLE IF NOT EXISTS attr_instances_{suffix} PARTITION OF attr_instances \
385 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
386 : )
387 1047 : }
388 :
389 : /// Recovery for a range DEFAULT already holds rows for, in execution order:
390 : /// build the week's table STANDALONE (a `PARTITION OF` would fail again), move
391 : /// exactly `[lo, hi)` out of DEFAULT into it, then ATTACH. The move is what
392 : /// makes the ATTACH legal, and its bounds are what keep every other row in
393 : /// DEFAULT. Run as one unit — a partial application would strand rows in an
394 : /// unattached table.
395 5 : fn adopt_default_rows_sql(suffix: &str, lo: &str, hi: &str) -> [String; 3] {
396 5 : [
397 5 : format!(
398 5 : "CREATE TABLE IF NOT EXISTS attr_instances_{suffix} \
399 5 : (LIKE attr_instances INCLUDING DEFAULTS INCLUDING CONSTRAINTS)"
400 5 : ),
401 5 : format!(
402 5 : "WITH moved AS (DELETE FROM attr_instances_default \
403 5 : WHERE observed_at >= '{lo}' AND observed_at < '{hi}' RETURNING *) \
404 5 : INSERT INTO attr_instances_{suffix} SELECT * FROM moved"
405 5 : ),
406 5 : format!(
407 5 : "ALTER TABLE attr_instances ATTACH PARTITION attr_instances_{suffix} \
408 5 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
409 5 : ),
410 5 : ]
411 5 : }
412 :
413 : /// The 4.22 expired-instance DELETE, isolated so a deadlock with concurrent
414 : /// ingest never poisons the main maintenance transaction. Service role: the
415 : /// reap is cross-tenant work (RLS would hide other tenants' rows).
416 397 : async fn reap_expired_instances(pool: &PgPool) -> Result<u64, sqlx::Error> {
417 397 : let mut tx = pool.begin().await?;
418 397 : crate::store::pg::set_service(&mut tx).await?;
419 397 : let n = sqlx::query(
420 397 : // try_timestamptz (0001_init.sql), not a bare cast: `expiresAt` is
421 397 : // jsonb TEXT and a stamp PostgreSQL cannot parse would abort this
422 397 : // DELETE for the whole deployment, every tick, forever.
423 397 : "DELETE FROM attr_instances
424 397 : WHERE try_timestamptz(data->>'expiresAt') < now()",
425 397 : )
426 397 : .execute(&mut *tx)
427 397 : .await?
428 397 : .rows_affected();
429 397 : tx.commit().await?;
430 397 : Ok(n)
431 397 : }
432 :
433 : #[cfg(test)]
434 : mod tests {
435 : use super::*;
436 :
437 : const LO: &str = "2026-08-17T00:00:00+00";
438 : const HI: &str = "2026-08-24T00:00:00+00";
439 :
440 : /// The partition name is an identifier the catalog hands back, not SQL.
441 : /// A bare interpolation drops any partition whose name needs quoting and
442 : /// resolves through `search_path` instead of the schema the row is in.
443 : #[test]
444 2 : fn a_partition_is_dropped_by_the_name_the_catalog_spells() {
445 2 : assert_eq!(
446 2 : drop_partition_sql("attr_instances_2026w34"),
447 : "DROP TABLE public.\"attr_instances_2026w34\""
448 : );
449 2 : assert_eq!(
450 2 : drop_partition_sql("Weird Name"),
451 : "DROP TABLE public.\"Weird Name\"",
452 : "a name that is not a bare identifier still names one table"
453 : );
454 2 : assert_eq!(
455 2 : drop_partition_sql("a\"b"),
456 : "DROP TABLE public.\"a\"\"b\"",
457 : "a quote in the name closes nothing"
458 : );
459 2 : }
460 :
461 : #[test]
462 2 : fn create_partition_covers_exactly_the_week() {
463 2 : let sql = create_partition_sql("2026w34", LO, HI);
464 2 : assert!(sql.contains("attr_instances_2026w34 PARTITION OF attr_instances"));
465 2 : assert!(sql.contains(&format!("FROM ('{LO}') TO ('{HI}')")));
466 2 : }
467 :
468 : /// The week's table must be built standalone: `CREATE ... PARTITION OF`
469 : /// is the statement that just failed, so repeating it cannot recover the
470 : /// range.
471 : #[test]
472 2 : fn adopt_builds_the_table_standalone_then_attaches_it() {
473 2 : let [create, _move, attach] = adopt_default_rows_sql("2026w34", LO, HI);
474 2 : assert!(create.contains("LIKE attr_instances"), "{create}");
475 2 : assert!(!create.contains("PARTITION OF"), "{create}");
476 2 : assert!(
477 2 : attach
478 2 : .starts_with("ALTER TABLE attr_instances ATTACH PARTITION attr_instances_2026w34"),
479 : "{attach}"
480 : );
481 2 : assert!(
482 2 : attach.contains(&format!("FROM ('{LO}') TO ('{HI}')")),
483 : "{attach}"
484 : );
485 2 : }
486 :
487 : /// The move is bounded by the partition range on BOTH sides. Unbounded, it
488 : /// would empty the DEFAULT partition of every historic-backfill row in the
489 : /// deployment and stuff them into one week.
490 : #[test]
491 2 : fn adopt_moves_only_the_partition_range_out_of_default() {
492 2 : let [_create, mv, _attach] = adopt_default_rows_sql("2026w34", LO, HI);
493 2 : assert!(mv.contains("DELETE FROM attr_instances_default"), "{mv}");
494 2 : assert!(mv.contains(&format!("observed_at >= '{LO}'")), "{mv}");
495 2 : assert!(mv.contains(&format!("observed_at < '{HI}'")), "{mv}");
496 2 : assert!(
497 2 : mv.contains("INSERT INTO attr_instances_2026w34"),
498 : "moved rows must land in the week's table: {mv}"
499 : );
500 2 : }
501 : }
|
