diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs index fc1b39374a2..e1e079b2c91 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment.rs @@ -15,6 +15,7 @@ use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use rusqlite::types::Value; use rusqlite::{OptionalExtension, Transaction, params}; use tracedecay_code_index::production::{ CodeIndexExecutionControlV1, SealedLexicalFileReplacementV1, VerifiedSealedLexicalPageV1, @@ -238,19 +239,6 @@ impl ShiftV1 { }) } - /// An SQL expression giving `column`'s successor value for a carried - /// row; it reads only the parent value, so one `UPDATE` applies it. - fn sql_image(&self, column: &str) -> String { - let mut expression = String::from("CASE"); - for ((_, end), growth) in self.replaced.iter().zip(&self.growth).rev() { - expression.push_str(&format!( - " WHEN {column} >= {end} THEN {column} + ({growth})" - )); - } - expression.push_str(&format!(" ELSE {column} END")); - expression - } - fn sql_replaced(&self, column: &str) -> String { if self.replaced.is_empty() { return "0".to_owned(); @@ -636,25 +624,67 @@ fn shift_rowid_keys( table: &str, column: &str, shift: &ShiftV1, + control: &dyn CodeIndexExecutionControlV1, ) -> Result<(), CodeLexicalArtifactErrorV1> { let Some(first) = shift.first_affected().filter(|_| shift.shifts()) else { return Ok(()); }; - transaction - .execute( - &format!( - "UPDATE {table} SET {column} = -({image}) - 1 WHERE {column} >= {first}", - image = shift.sql_image(column) - ), - [], - ) - .map_err(sqlite_error)?; - transaction - .execute( - &format!("UPDATE {table} SET {column} = -{column} - 1 WHERE {column} < 0"), - [], - ) + // A bulk UPDATE of INTEGER PRIMARY KEY values builds a statement-local + // ephemeral key index that can spill despite an empty TEMP pager afterward. + // Use bounded key-ordered reads and exact updates in both staging passes. + let mut update = transaction + .prepare_cached(&format!( + "UPDATE {table} SET {column} = ?2 WHERE {column} = ?1" + )) .map_err(sqlite_error)?; + for staging in [true, false] { + let predicate = if staging { + format!("{column} >= {first}") + } else { + format!("{column} < 0") + }; + let first_sql = + format!("SELECT {column} FROM {table} WHERE {predicate} ORDER BY {column} LIMIT ?1"); + let after_sql = format!( + "SELECT {column} FROM {table} WHERE {predicate} AND {column} > ?1 ORDER BY {column} LIMIT ?2" + ); + let mut last: Option = None; + loop { + checkpoint(control)?; + let window: Vec = postings::scan_window( + transaction, + &first_sql, + &after_sql, + &last, + |key| vec![Value::Integer(*key)], + |row| row.get(0), + )?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(*final_key); + for key in window { + checkpoint(control)?; + let target = if staging { + let mapped = shift + .map(u64::try_from(key).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "replaced lexical row survived deletion".to_owned(), + ) + })?; + -i64::try_from(mapped).map_err(contract_number)? - 1 + } else { + -(key + 1) + }; + if update.execute(params![key, target]).map_err(sqlite_error)? != 1 { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried lexical row changed during key shift".to_owned(), + )); + } + } + } + } Ok(()) } @@ -673,7 +703,13 @@ fn carry_row_blocks( [], ) .map_err(sqlite_error)?; - shift_rowid_keys(transaction, "row_blocks", "first_document", documents)?; + shift_rowid_keys( + transaction, + "row_blocks", + "first_document", + documents, + control, + )?; let mut insert = transaction .prepare_cached("INSERT INTO row_blocks(first_document, payload) VALUES (?1, ?2)") .map_err(sqlite_error)?; @@ -704,15 +740,59 @@ fn carry_row_chunks( ) .map_err(sqlite_error)?; if let Some(first) = documents.first_affected().filter(|_| documents.shifts()) { - transaction - .execute( - &format!( - "UPDATE row_chunks SET document_id = {} WHERE document_id >= {first}", - documents.sql_image("document_id") - ), - [], + // The chunk ID is the stable primary key; the document ID is only a + // value. Seek primary-key windows so a large carry can observe + // cancellation between updates without creating a sorted TEMP copy. + let mut update = transaction + .prepare_cached( + "UPDATE row_chunks SET document_id = ?3 WHERE chunk_id = ?1 AND document_id = ?2", ) .map_err(sqlite_error)?; + let first_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} ORDER BY chunk_id LIMIT ?1" + ); + let after_sql = format!( + "SELECT chunk_id, document_id FROM row_chunks WHERE document_id >= {first} AND chunk_id > ?1 ORDER BY chunk_id LIMIT ?2" + ); + let mut last: Option = None; + loop { + checkpoint(control)?; + let window: Vec<(Value, i64)> = postings::scan_window( + transaction, + &first_sql, + &after_sql, + &last, + |chunk_id| vec![chunk_id.clone()], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + let Some((final_chunk, _)) = window.last() else { + break; + }; + last = Some(final_chunk.clone()); + for (chunk_id, old_document) in window { + checkpoint(control)?; + let mapped = documents + .map(u64::try_from(old_document).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "replaced lexical chunk survived deletion".to_owned(), + ) + })?; + let mapped = i64::try_from(mapped).map_err(contract_number)?; + if mapped == old_document { + continue; + } + if update + .execute(params![chunk_id, old_document, mapped]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried lexical chunk changed during remap".to_owned(), + )); + } + } + } } let mut insert = transaction .prepare_cached("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") @@ -819,3 +899,6 @@ fn write_page_rows( } Ok(rows.len() as u64) } + +#[cfg(test)] +mod tests; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs index 9c229319776..1f66e77c207 100644 --- a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones.rs @@ -4,6 +4,7 @@ use std::collections::BTreeMap; +use rusqlite::types::Value; use rusqlite::{Transaction, params}; use tracedecay_code_index::production::CodeIndexExecutionControlV1; @@ -13,6 +14,7 @@ use super::super::format::contract_number; use super::super::prepared::PreparedCloneBodyV1; use super::super::row_codec::stored_symbol_key; use super::super::{CodeLexicalArtifactErrorV1, checkpoint, sqlite_error}; +use super::postings::scan_window; use super::{ReplacedFilesV1, ShiftV1, shift_rowid_keys, shifted}; /// One successor clone body with the occurrence ordinal it takes. @@ -52,19 +54,12 @@ pub(super) fn carry_clone_rows( [], ) .map_err(sqlite_error)?; - if let Some(first) = clones.first_affected().filter(|_| clones.shifts()) { - // The posting's ordinal is part of its key, so shifted rows are - // rewritten through a temporary copy. - transaction - .execute_batch(&format!( - "CREATE TEMP TABLE carried_clone_exact AS SELECT class, normalization_revision, digest, {image} AS occurrence_ordinal FROM clone_exact_postings WHERE occurrence_ordinal >= {first}; - DELETE FROM clone_exact_postings WHERE occurrence_ordinal >= {first}; - INSERT INTO clone_exact_postings(class, normalization_revision, digest, occurrence_ordinal) SELECT class, normalization_revision, digest, occurrence_ordinal FROM temp.carried_clone_exact; - DROP TABLE temp.carried_clone_exact;", - image = clones.sql_image("occurrence_ordinal") - )) - .map_err(sqlite_error)?; - } + // A TEMP copy of every surviving exact posting retains unlinked pager pages + // after DROP until this long-lived builder resets its TEMP schema. Move the + // composite primary keys through bounded, ordered windows and a negative + // ordinal namespace. A single bulk UPDATE can itself spill an unlinked + // statement-local key index, even when the TEMP pager has zero pages. + shift_clone_posting_keys(transaction, clones, control)?; { let mut insert = transaction .prepare_cached( @@ -94,7 +89,7 @@ pub(super) fn carry_clone_rows( [], ) .map_err(sqlite_error)?; - shift_rowid_keys(transaction, "clone_occurrences", "ordinal", clones)?; + shift_rowid_keys(transaction, "clone_occurrences", "ordinal", clones, control)?; let mut parent_payloads = BTreeMap::>::new(); { let mut statement = transaction @@ -140,6 +135,120 @@ pub(super) fn carry_clone_rows( renumber_clone_payloads(transaction, &parent_payloads, &successor_payloads, control) } +/// A key-ordered window, so SQLite can seek the WITHOUT ROWID primary key +/// without sorting the entire posting index into an external TEMP file. +type ClonePostingKey = (i64, i64, Vec, i64); + +fn clone_posting_window( + transaction: &Transaction<'_>, + predicate: &str, + last: &Option, +) -> Result, CodeLexicalArtifactErrorV1> { + let columns = "class, normalization_revision, digest, occurrence_ordinal"; + scan_window( + transaction, + &format!( + "SELECT {columns} FROM clone_exact_postings WHERE {predicate} ORDER BY {columns} LIMIT ?1" + ), + &format!( + "SELECT {columns} FROM clone_exact_postings WHERE {predicate} AND ({columns}) > (?1, ?2, ?3, ?4) ORDER BY {columns} LIMIT ?5" + ), + last, + |(class, revision, digest, ordinal)| { + vec![ + Value::Integer(*class), + Value::Integer(*revision), + Value::Blob(digest.clone()), + Value::Integer(*ordinal), + ] + }, + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) +} + +/// Shift the occurrence component of the WITHOUT ROWID posting primary key. +/// Both passes seek bounded primary-key windows and update one keyed row at a +/// time: SQLite's single bulk UPDATE can allocate an unlinked statement-local +/// key index independently of the TEMP b-tree pager. The negative namespace +/// prevents collisions with positive keys while a window advances. +fn shift_clone_posting_keys( + transaction: &Transaction<'_>, + clones: &ShiftV1, + control: &dyn CodeIndexExecutionControlV1, +) -> Result<(), CodeLexicalArtifactErrorV1> { + let Some(first) = clones.first_affected().filter(|_| clones.shifts()) else { + return Ok(()); + }; + let mut update = transaction + .prepare_cached( + "UPDATE clone_exact_postings SET occurrence_ordinal = ?5 + WHERE class = ?1 AND normalization_revision = ?2 + AND digest = ?3 AND occurrence_ordinal = ?4", + ) + .map_err(sqlite_error)?; + let mut last = None; + loop { + checkpoint(control)?; + let window = clone_posting_window( + transaction, + &format!("occurrence_ordinal >= {first}"), + &last, + )?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(final_key.clone()); + for (class, revision, digest, ordinal) in window { + checkpoint(control)?; + let mapped = clones + .map(u64::try_from(ordinal).map_err(contract_number)?)? + .ok_or_else(|| { + CodeLexicalArtifactErrorV1::Corrupt( + "carried clone posting belongs to a replaced range".to_owned(), + ) + })?; + let mapped = i64::try_from(mapped).map_err(contract_number)?; + // -mapped - 1 fits even when mapped == i64::MAX. + let staged = -mapped - 1; + if update + .execute(params![class, revision, digest, ordinal, staged]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "carried clone posting changed during key shift".to_owned(), + )); + } + } + } + // All moved keys are now negative and disjoint from every unchanged key. + // Keyset pagination remains stable as restored positive keys leave the + // scan; compute -(ordinal + 1) to handle an i64::MIN staged key. + let mut last = None; + loop { + checkpoint(control)?; + let window = clone_posting_window(transaction, "occurrence_ordinal < 0", &last)?; + let Some(final_key) = window.last() else { + break; + }; + last = Some(final_key.clone()); + for (class, revision, digest, ordinal) in window { + checkpoint(control)?; + let restored = -(ordinal + 1); + if update + .execute(params![class, revision, digest, ordinal, restored]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "staged clone posting changed during key restoration".to_owned(), + )); + } + } + } + Ok(()) +} + /// Where a run's first successor occurrence lands: its parent start moved /// by the growth of the runs before it. `index` pairs the run with its /// replaced range, since two runs without clone bodies can replace the same @@ -161,6 +270,49 @@ fn clone_range_image( shifted(start, growth_before) } +/// Restore staged negative payload ordinals without the statement-local +/// key index that a bulk INTEGER PRIMARY KEY UPDATE can create. Restored +/// positive keys cannot re-enter the negative-key scan. +fn restore_clone_payload_ordinals( + transaction: &Transaction<'_>, + control: &dyn CodeIndexExecutionControlV1, +) -> Result<(), CodeLexicalArtifactErrorV1> { + let mut update = transaction + .prepare_cached("UPDATE clone_body_payloads SET ordinal = ?2 WHERE ordinal = ?1") + .map_err(sqlite_error)?; + let mut last: Option = None; + loop { + checkpoint(control)?; + let window: Vec = scan_window( + transaction, + "SELECT ordinal FROM clone_body_payloads WHERE ordinal < 0 ORDER BY ordinal LIMIT ?1", + "SELECT ordinal FROM clone_body_payloads WHERE ordinal < 0 AND ordinal > ?1 ORDER BY ordinal LIMIT ?2", + &last, + |ordinal| vec![Value::Integer(*ordinal)], + |row| row.get(0), + )?; + let Some(final_ordinal) = window.last() else { + break; + }; + last = Some(*final_ordinal); + for ordinal in window { + checkpoint(control)?; + // Reversing -number-1 as -(ordinal+1) also handles i64::MIN. + let restored = -(ordinal + 1); + if update + .execute(params![ordinal, restored]) + .map_err(sqlite_error)? + != 1 + { + return Err(CodeLexicalArtifactErrorV1::Corrupt( + "staged clone payload changed during key restoration".to_owned(), + )); + } + } + } + Ok(()) +} + /// Number payloads by first use over the occurrence sequence, retire the /// ones no occurrence uses, and point every occurrence at its number. fn renumber_clone_payloads( @@ -230,6 +382,7 @@ fn renumber_clone_payloads( } } for (ordinal, digest) in parent_payloads { + checkpoint(control)?; match numbered.get(digest) { None => { transaction @@ -250,12 +403,7 @@ fn renumber_clone_payloads( Some(_) => {} } } - transaction - .execute( - "UPDATE clone_body_payloads SET ordinal = -ordinal - 1 WHERE ordinal < 0", - [], - ) - .map_err(sqlite_error)?; + restore_clone_payload_ordinals(transaction, control)?; { let mut insert = transaction .prepare_cached( @@ -278,3 +426,6 @@ fn renumber_clone_payloads( } Ok(()) } + +#[cfg(test)] +mod tests; diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs new file mode 100644 index 00000000000..2c2b2a08ea7 --- /dev/null +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/clones/tests.rs @@ -0,0 +1,665 @@ +use std::collections::BTreeMap; +use std::path::Path; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use rusqlite::{Connection, ErrorCode, params}; +use tempfile::{TempDir, tempdir}; +use tracedecay_code_index::production::CodeIndexExecutionControlV1; + +use super::super::CARRY_SCAN_ROWS; +use super::{ + ClonePostingKey, CodeLexicalArtifactErrorV1, ShiftV1, renumber_clone_payloads, + shift_clone_posting_keys, +}; + +struct ActiveControl; + +impl CodeIndexExecutionControlV1 for ActiveControl { + fn is_cancelled(&self) -> bool { + false + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +struct CancelAfter { + checks: AtomicUsize, + limit: usize, +} + +impl CodeIndexExecutionControlV1 for CancelAfter { + fn is_cancelled(&self) -> bool { + self.checks.fetch_add(1, Ordering::Relaxed) >= self.limit + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +fn open_fixture(root: &Path) -> Connection { + let connection = Connection::open(root.join("clones.sqlite")).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store = FILE; + PRAGMA cache_size = 8; + PRAGMA temp.cache_size = 8;", + ) + .unwrap(); + connection +} + +fn fixture() -> (TempDir, Connection) { + let root = tempdir().unwrap(); + let mut connection = open_fixture(root.path()); + connection + .execute_batch( + "CREATE TABLE clone_exact_postings ( + class INTEGER NOT NULL, + normalization_revision INTEGER NOT NULL, + digest BLOB NOT NULL, + occurrence_ordinal INTEGER NOT NULL, + PRIMARY KEY(class, normalization_revision, digest, occurrence_ordinal) + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('survivor');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + { + let mut insert = transaction + .prepare( + "INSERT INTO clone_exact_postings + (class, normalization_revision, digest, occurrence_ordinal) + VALUES (?1, ?2, ?3, ?4)", + ) + .unwrap(); + // Consecutive ordinals share a key prefix. Expansion really collides + // without staging, and each family crosses multiple scan windows. + let digest = [7_u8; 32]; + for ordinal in 1_i64..=16_000 { + for class in 0_i64..3 { + insert + .execute(params![class, class + 1, digest.as_slice(), ordinal]) + .unwrap(); + } + } + } + transaction.commit().unwrap(); + (root, connection) +} + +fn postings(connection: &Connection) -> Vec { + let mut statement = connection + .prepare( + "SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings + ORDER BY class, normalization_revision, digest, occurrence_ordinal", + ) + .unwrap(); + statement + .query_map([], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) + }) + .unwrap() + .collect::>>() + .unwrap() +} + +fn integrity(connection: &Connection, schema: &str) { + let result: String = connection + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(result, "ok", "{schema} page integrity"); +} + +fn expected_postings(original: &[ClonePostingKey], shift: &ShiftV1) -> Vec { + let mut expected: Vec<_> = original + .iter() + .filter_map(|(class, revision, digest, ordinal)| { + shift + .map(u64::try_from(*ordinal).unwrap()) + .unwrap() + .map(|mapped| { + ( + *class, + *revision, + digest.clone(), + i64::try_from(mapped).unwrap(), + ) + }) + }) + .collect(); + expected.sort(); + expected +} + +#[test] +fn carried_clone_keys_shift_without_retained_file_backed_temp_pages() { + let (root, mut connection) = fixture(); + let original = postings(&connection); + assert_eq!(original.len(), 48_000); + assert_eq!( + connection + .pragma_query_value(None, "temp_store", |row| row.get::<_, i64>(0)) + .unwrap(), + 1 + ); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "auto_vacuum", |row| row.get::<_, i64>(0)) + .unwrap(), + 0 + ); + for (index, shift) in [ + ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), + ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), + ] + .iter() + .enumerate() + { + let expected = expected_postings(&original, shift); + assert_eq!(expected.len(), 47_964); + assert_ne!(expected, original); + let transaction = connection.transaction().unwrap(); + transaction + .execute( + &format!( + "DELETE FROM clone_exact_postings WHERE {}", + shift.sql_replaced("occurrence_ordinal") + ), + [], + ) + .unwrap(); + shift_clone_posting_keys(&transaction, shift, &ActiveControl).unwrap(); + assert_eq!(postings(&transaction), expected); + assert_eq!( + transaction + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "the remap must not retain TEMP pager pages" + ); + integrity(&transaction, "main"); + integrity(&transaction, "temp"); + if index == 0 { + transaction.rollback().unwrap(); + } else { + transaction.commit().unwrap(); + } + drop(connection); + connection = open_fixture(root.path()); + let committed = if index == 0 { &original } else { &expected }; + assert_eq!( + &postings(&connection), + committed, + "a fresh on-disk connection must see exactly the committed keys" + ); + let survivor: String = connection + .query_row("SELECT value FROM independent", [], |row| row.get(0)) + .unwrap(); + assert_eq!(survivor, "survivor"); + integrity(&connection, "main"); + } + connection + .execute_batch( + "CREATE TEMP TABLE old_style_clone_copy AS + SELECT class, normalization_revision, digest, occurrence_ordinal + FROM clone_exact_postings; + DROP TABLE temp.old_style_clone_copy;", + ) + .unwrap(); + let retained_pages: i64 = connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get(0)) + .unwrap(); + assert!( + retained_pages > 32, + "positive TEMP-copy control: {retained_pages}" + ); + integrity(&connection, "main"); + integrity(&connection, "temp"); +} + +#[test] +fn clone_remap_stages_colliding_keys_and_rolls_back_interrupted_windows() { + let (root, mut connection) = fixture(); + let original = postings(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + { + let transaction = connection.transaction().unwrap(); + let error = transaction + .execute( + "UPDATE clone_exact_postings SET occurrence_ordinal = occurrence_ordinal + 1", + [], + ) + .unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); + transaction.rollback().unwrap(); + } + // Interrupt after at least one complete window in each staging direction. + for limit in [CARRY_SCAN_ROWS + 2, original.len() + CARRY_SCAN_ROWS + 32] { + let control = CancelAfter { + checks: AtomicUsize::new(0), + limit, + }; + { + let transaction = connection.transaction().unwrap(); + let error = shift_clone_posting_keys(&transaction, &shift, &control).unwrap_err(); + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + let negative: i64 = transaction + .query_row( + "SELECT COUNT(*) FROM clone_exact_postings WHERE occurrence_ordinal < 0", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(negative > 0 && negative < 48_000); + integrity(&transaction, "main"); + // Drop must roll back partially staged/restored keys, not publish them. + } + drop(connection); + connection = open_fixture(root.path()); + assert_eq!(postings(&connection), original); + integrity(&connection, "main"); + } + let transaction = connection.transaction().unwrap(); + shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); + assert_eq!(postings(&transaction), expected_postings(&original, &shift)); + transaction.commit().unwrap(); +} + +fn clone_payload_fixture() -> (TempDir, Connection, BTreeMap>) { + let root = tempdir().unwrap(); + let mut connection = open_fixture(root.path()); + connection + .execute_batch( + "CREATE TABLE clone_body_payloads( + ordinal INTEGER PRIMARY KEY, + payload_digest BLOB NOT NULL, + payload BLOB NOT NULL + ); + CREATE TABLE clone_occurrences( + ordinal INTEGER PRIMARY KEY, + payload_ordinal INTEGER NOT NULL + ); + CREATE TABLE independent_payload(value TEXT NOT NULL); + INSERT INTO independent_payload VALUES ('survivor');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + let mut parent = BTreeMap::new(); + { + let mut payload_insert = transaction + .prepare( + "INSERT INTO clone_body_payloads + (ordinal, payload_digest, payload) VALUES (?1, ?2, ?3)", + ) + .unwrap(); + let mut occurrence_insert = transaction + .prepare("INSERT INTO clone_occurrences(ordinal, payload_ordinal) VALUES (?1, ?2)") + .unwrap(); + for ordinal in 1_i64..=16_000 { + let digest = ordinal.to_be_bytes().repeat(4); + let payload = ordinal.to_be_bytes().repeat(32); + payload_insert + .execute(params![ordinal, digest.as_slice(), payload.as_slice()]) + .unwrap(); + // First use is deliberately reverse-ordered relative to the + // existing payload keys, so every old ordinal must be rekeyed. + occurrence_insert + .execute(params![ordinal, 16_001 - ordinal]) + .unwrap(); + parent.insert(ordinal, digest); + } + } + transaction.commit().unwrap(); + (root, connection, parent) +} + +fn payload_rows(connection: &Connection) -> Vec<(i64, Vec, Vec)> { + let mut query = connection + .prepare( + "SELECT ordinal, payload_digest, payload + FROM clone_body_payloads ORDER BY ordinal", + ) + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))) + .unwrap() + .collect::>>() + .unwrap() +} + +fn payload_occurrences(connection: &Connection) -> Vec<(i64, i64)> { + let mut query = connection + .prepare("SELECT ordinal, payload_ordinal FROM clone_occurrences ORDER BY ordinal") + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) + .unwrap() + .collect::>>() + .unwrap() +} + +#[test] +fn clone_payload_renumber_preserves_durable_rows_and_interrupts_restoration() { + let (root, mut connection, parent) = clone_payload_fixture(); + let original = payload_rows(&connection); + let original_occurrences = payload_occurrences(&connection); + assert_eq!(original.len(), 16_000); + assert_eq!(original_occurrences.len(), 16_000); + let mut expected: Vec<_> = original + .iter() + .map(|(ordinal, digest, payload)| (16_001 - ordinal, digest.clone(), payload.clone())) + .collect(); + expected.sort_by_key(|row| row.0); + assert_ne!(expected, original); + let empty = BTreeMap::::new(); + + // A direct positive-key renumber would collide with an occupied key. + { + let transaction = connection.transaction().unwrap(); + let error = transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = 16001 - ordinal", + [], + ) + .unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); + transaction.rollback().unwrap(); + } + + // The operation checkpoints during both the scan and each old-key write. + // Stop after staging all 16,000 keys and restoring one bounded window. + { + let transaction = connection.transaction().unwrap(); + let stop = CancelAfter { + checks: AtomicUsize::new(0), + limit: 16_000 * 2 + CARRY_SCAN_ROWS + 8, + }; + let error = renumber_clone_payloads(&transaction, &parent, &empty, &stop).unwrap_err(); + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + let negatives: i64 = transaction + .query_row( + "SELECT COUNT(*) FROM clone_body_payloads WHERE ordinal < 0", + [], + |row| row.get(0), + ) + .unwrap(); + assert!( + negatives > 0 && negatives < 16_000, + "interruption must happen after partial restoration: {negatives}" + ); + integrity(&transaction, "main"); + transaction.rollback().unwrap(); + } + drop(connection); + connection = open_fixture(root.path()); + assert_eq!(payload_rows(&connection), original); + assert_eq!(payload_occurrences(&connection), original_occurrences); + integrity(&connection, "main"); + + let transaction = connection.transaction().unwrap(); + renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); + assert_eq!(payload_rows(&transaction), expected); + assert_eq!( + payload_occurrences(&transaction), + (1_i64..=16_000) + .map(|ordinal| (ordinal, ordinal)) + .collect::>() + ); + integrity(&transaction, "main"); + transaction.commit().unwrap(); + drop(connection); + let reopened = open_fixture(root.path()); + assert_eq!(payload_rows(&reopened), expected); + assert_eq!( + payload_occurrences(&reopened), + (1_i64..=16_000) + .map(|ordinal| (ordinal, ordinal)) + .collect::>() + ); + let survivor: String = reopened + .query_row("SELECT value FROM independent_payload", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(survivor, "survivor"); + integrity(&reopened, "main"); + integrity(&reopened, "temp"); +} + +#[cfg(target_os = "linux")] +mod physical_spill { + use std::io; + use std::process::Command; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + + use rusqlite::params; + + use super::{ + ActiveControl, ShiftV1, clone_payload_fixture, expected_postings, fixture, payload_rows, + postings, renumber_clone_payloads, shift_clone_posting_keys, + }; + + fn spill_bytes() -> io::Result { + let mut bytes = 0; + for entry in std::fs::read_dir("/proc/self/fd")? { + let path = entry?.path(); + let target = match std::fs::read_link(&path) { + Ok(target) => target, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if target + .file_name() + .is_some_and(|name| name.to_string_lossy().starts_with("etilqs_")) + { + bytes += std::fs::metadata(path)?.len(); + } + } + Ok(bytes) + } + + #[test] + fn clone_payload_renumber_has_no_unlinked_statement_spill() { + const CHILD: &str = "TRACEDECAY_PAYLOAD_SPILL_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .arg("clone_payload_renumber_has_no_unlinked_statement_spill") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!( + stdout.contains("1 passed"), + "child test was vacuous: {stdout}" + ); + return; + } + + let (_root, mut connection, parent) = clone_payload_fixture(); + let original = payload_rows(&connection); + let peak = Arc::new(AtomicU64::new(0)); + let unavailable = Arc::new(AtomicBool::new(false)); + let observed = Arc::clone(&peak); + let unavailable_probe = Arc::clone(&unavailable); + connection + .progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + unavailable_probe.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); + + { + let transaction = connection.transaction().unwrap(); + for ordinal in 1_i64..=16_000 { + transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = ?2 WHERE ordinal = ?1", + params![ordinal, -(16_001 - ordinal) - 1], + ) + .unwrap(); + } + // Old bulk restoration, with the actual production key pattern. + transaction + .execute( + "UPDATE clone_body_payloads SET ordinal = -ordinal - 1 WHERE ordinal < 0", + [], + ) + .unwrap(); + transaction.rollback().unwrap(); + } + assert!(!unavailable.load(Ordering::Relaxed)); + assert!( + peak.load(Ordering::Relaxed) > 0, + "bulk payload restoration must produce an observable spill" + ); + assert_eq!(spill_bytes().unwrap(), 0); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "TEMP pager counts miss the transient statement-local spill" + ); + assert_eq!(payload_rows(&connection), original); + + peak.store(0, Ordering::Relaxed); + let empty = std::collections::BTreeMap::::new(); + { + let transaction = connection.transaction().unwrap(); + renumber_clone_payloads(&transaction, &parent, &empty, &ActiveControl).unwrap(); + transaction.commit().unwrap(); + } + connection + .progress_handler(0, None:: bool>) + .unwrap(); + assert!(!unavailable.load(Ordering::Relaxed)); + assert_eq!( + peak.load(Ordering::Relaxed), + 0, + "bounded payload key restoration must not create a file spill" + ); + assert_eq!(spill_bytes().unwrap(), 0); + assert_ne!(payload_rows(&connection), original); + } + + #[test] + fn clone_posting_remap_has_no_implicit_file_spill() { + const CHILD: &str = "TRACEDECAY_CLONE_SPILL_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + // Other unit tests may spill concurrently; only measure our child. + let output = Command::new(std::env::current_exe().unwrap()) + .arg("clone_posting_remap_has_no_implicit_file_spill") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!( + stdout.contains("1 passed"), + "child did not execute: {stdout}" + ); + return; + } + let (_root, mut connection) = fixture(); + let original = postings(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + let peak = Arc::new(AtomicU64::new(0)); + let unavailable = Arc::new(AtomicBool::new(false)); + let observed_peak = Arc::clone(&peak); + let failed_probe = Arc::clone(&unavailable); + connection + .progress_handler( + 4096, + Some(move || match spill_bytes() { + Ok(bytes) => { + observed_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + failed_probe.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); + { + let transaction = connection.transaction().unwrap(); + // Bulk key updates are the positive control for statement-local + // spills that bounded production key updates avoid. + transaction + .execute( + "UPDATE clone_exact_postings + SET occurrence_ordinal = -occurrence_ordinal - 2 + WHERE occurrence_ordinal > 0", + [], + ) + .unwrap(); + transaction + .execute( + "UPDATE clone_exact_postings + SET occurrence_ordinal = -occurrence_ordinal - 1 + WHERE occurrence_ordinal < 0", + [], + ) + .unwrap(); + transaction.rollback().unwrap(); + } + assert!(!unavailable.load(Ordering::Relaxed)); + assert!( + peak.load(Ordering::Relaxed) > 0, + "bulk UPDATE must really spill" + ); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "page_count", |row| row.get::<_, i64>(0)) + .unwrap(), + 0, + "TEMP page_count alone misses the just-freed statement spill" + ); + assert_eq!(postings(&connection), original); + assert_eq!(spill_bytes().unwrap(), 0); + peak.store(0, Ordering::Relaxed); + { + let transaction = connection.transaction().unwrap(); + shift_clone_posting_keys(&transaction, &shift, &ActiveControl).unwrap(); + transaction.commit().unwrap(); + } + connection + .progress_handler(0, None:: bool>) + .unwrap(); + assert!(!unavailable.load(Ordering::Relaxed)); + assert_eq!(peak.load(Ordering::Relaxed), 0, "no implicit remap spill"); + assert_eq!(spill_bytes().unwrap(), 0); + assert_eq!(postings(&connection), expected_postings(&original, &shift)); + } +} diff --git a/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs new file mode 100644 index 00000000000..af1b5d10462 --- /dev/null +++ b/crates/tracedecay-query/src/retrieval/lexical/projection/artifact/increment/tests.rs @@ -0,0 +1,492 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; + +use rusqlite::types::Value; +use rusqlite::{Connection, ErrorCode, params}; +use tracedecay_code_index::production::CodeIndexExecutionControlV1; + +use super::{CodeLexicalArtifactErrorV1, ShiftV1, carry_row_chunks, shift_rowid_keys}; + +struct Active; +impl CodeIndexExecutionControlV1 for Active { + fn is_cancelled(&self) -> bool { + false + } + + fn is_deadline_exceeded(&self) -> bool { + false + } +} + +fn fixture() -> (tempfile::TempDir, Connection) { + let directory = tempfile::tempdir().unwrap(); + let mut connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store=FILE; + PRAGMA cache_size=8; + PRAGMA temp.cache_size=8; + CREATE TABLE row_blocks( + first_document INTEGER PRIMARY KEY, payload BLOB NOT NULL + ); + CREATE TABLE clone_occurrences( + ordinal INTEGER PRIMARY KEY, + symbol_key BLOB NOT NULL UNIQUE, + payload_ordinal INTEGER NOT NULL + ); + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('live');", + ) + .unwrap(); + let transaction = connection.transaction().unwrap(); + { + let mut block = transaction + .prepare("INSERT INTO row_blocks(first_document,payload) VALUES(?1,?2)") + .unwrap(); + let mut occurrence = transaction + .prepare( + "INSERT INTO clone_occurrences(ordinal,symbol_key,payload_ordinal) + VALUES(?1,?2,?3)", + ) + .unwrap(); + for n in 1_i64..=16_000 { + block.execute(params![n, [7_u8; 32].as_slice()]).unwrap(); + occurrence + .execute(params![n, n.to_be_bytes().as_slice(), n % 31]) + .unwrap(); + } + } + transaction.commit().unwrap(); + (directory, connection) +} + +fn blocks(c: &Connection) -> Vec<(i64, Vec)> { + let mut query = c + .prepare("SELECT first_document,payload FROM row_blocks ORDER BY first_document") + .unwrap(); + query + .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .unwrap() + .collect::>() + .unwrap() +} + +fn occurrences(c: &Connection) -> Vec<(i64, Vec, i64)> { + let mut query = c + .prepare( + "SELECT ordinal,symbol_key,payload_ordinal + FROM clone_occurrences ORDER BY ordinal", + ) + .unwrap(); + query + .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) + .unwrap() + .collect::>() + .unwrap() +} + +fn integrity(c: &Connection) { + for schema in ["main", "temp"] { + let actual: String = c + .query_row(&format!("PRAGMA {schema}.integrity_check"), [], |r| { + r.get(0) + }) + .unwrap(); + assert_eq!(actual, "ok", "{schema} remains intact"); + } +} + +#[test] +fn lexical_rowid_carry_preserves_disk_rows_across_replacements() { + let (directory, mut connection) = fixture(); + let original_blocks = blocks(&connection); + let original_occurrences = occurrences(&connection); + assert_eq!(original_blocks.len(), 16_000); + assert_eq!(original_occurrences.len(), 16_000); + let shifts = [ + ShiftV1::new([((11, 20), 5), ((36, 39), 6)]).unwrap(), + ShiftV1::new([((11, 20), 12), ((36, 39), 1)]).unwrap(), + ]; + // Positive collision control: old in-place positive keys cannot just move. + { + let tx = connection.transaction().unwrap(); + let error = tx + .execute("UPDATE row_blocks SET first_document=first_document+1", []) + .unwrap_err(); + assert_eq!( + error.sqlite_error_code(), + Some(ErrorCode::ConstraintViolation) + ); + tx.rollback().unwrap(); + } + for (index, shift) in shifts.iter().enumerate() { + let expected_blocks: Vec<_> = original_blocks + .iter() + .filter_map(|(key, blob)| { + shift + .map(u64::try_from(*key).unwrap()) + .unwrap() + .map(|mapped| (i64::try_from(mapped).unwrap(), blob.clone())) + }) + .collect(); + let expected_occurrences: Vec<_> = original_occurrences + .iter() + .filter_map(|(key, symbol, payload)| { + shift + .map(u64::try_from(*key).unwrap()) + .unwrap() + .map(|mapped| (i64::try_from(mapped).unwrap(), symbol.clone(), *payload)) + }) + .collect(); + let tx = connection.transaction().unwrap(); + for (table, column) in [ + ("row_blocks", "first_document"), + ("clone_occurrences", "ordinal"), + ] { + tx.execute( + &format!("DELETE FROM {table} WHERE {}", shift.sql_replaced(column)), + [], + ) + .unwrap(); + shift_rowid_keys(&tx, table, column, shift, &Active).unwrap(); + } + assert_eq!(blocks(&tx), expected_blocks); + assert_eq!(occurrences(&tx), expected_occurrences); + assert_eq!( + tx.pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)) + .unwrap(), + 0 + ); + integrity(&tx); + if index == 0 { + tx.rollback().unwrap(); + } else { + tx.commit().unwrap(); + } + drop(connection); + connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + assert_eq!( + blocks(&connection), + if index == 0 { + original_blocks.clone() + } else { + expected_blocks + } + ); + assert_eq!( + occurrences(&connection), + if index == 0 { + original_occurrences.clone() + } else { + expected_occurrences + } + ); + let live: String = connection + .query_row("SELECT value FROM independent", [], |r| r.get(0)) + .unwrap(); + assert_eq!(live, "live"); + integrity(&connection); + } +} + +#[test] +fn interrupted_rowid_carry_rolls_back_partial_key_staging() { + let (directory, mut connection) = fixture(); + let original_blocks = blocks(&connection); + let original_occurrences = occurrences(&connection); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + struct CancelAfter { + probes: std::sync::atomic::AtomicUsize, + limit: usize, + } + impl CodeIndexExecutionControlV1 for CancelAfter { + fn is_cancelled(&self) -> bool { + self.probes + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + >= self.limit + } + + fn is_deadline_exceeded(&self) -> bool { + false + } + } + for occurrences_family in [false, true] { + let tx = connection.transaction().unwrap(); + let cancel = CancelAfter { + probes: std::sync::atomic::AtomicUsize::new(0), + limit: 4_100, + }; + let (table, column) = if occurrences_family { + ("clone_occurrences", "ordinal") + } else { + ("row_blocks", "first_document") + }; + let error = shift_rowid_keys(&tx, table, column, &shift, &cancel).unwrap_err(); + assert!(matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_))); + if occurrences_family { + assert_ne!(occurrences(&tx), original_occurrences); + } else { + assert_ne!(blocks(&tx), original_blocks); + } + integrity(&tx); + drop(tx); + drop(connection); + connection = Connection::open(directory.path().join("carry.sqlite")).unwrap(); + assert_eq!(blocks(&connection), original_blocks); + assert_eq!(occurrences(&connection), original_occurrences); + integrity(&connection); + } +} + +#[cfg(target_os = "linux")] +#[test] +fn lexical_rowid_shift_does_not_spill_unlinked_sqlite_files() { + use std::io; + use std::process::Command; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + + fn spill_bytes() -> io::Result { + let mut total = 0; + for entry in std::fs::read_dir("/proc/self/fd")? { + let path = entry?.path(); + let target = match std::fs::read_link(&path) { + Ok(target) => target, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if target + .file_name() + .is_some_and(|name| name.to_string_lossy().starts_with("etilqs_")) + { + total += std::fs::metadata(path)?.len(); + } + } + Ok(total) + } + + const CHILD: &str = "TRACEDECAY_ROWID_CARRY_SPILL_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .arg("lexical_rowid_shift_does_not_spill_unlinked_sqlite_files") + .arg("--test-threads=1") + .arg("--nocapture") + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{stdout}\n{stderr}"); + assert!( + stdout.contains("1 passed"), + "child test was vacuous: {stdout}" + ); + return; + } + let (_directory, mut connection) = fixture(); + let shift = ShiftV1::new([((1, 1), 1)]).unwrap(); + let original = blocks(&connection); + let peak = Arc::new(AtomicU64::new(0)); + let unreadable = Arc::new(AtomicBool::new(false)); + let observer_peak = Arc::clone(&peak); + let observer_failed = Arc::clone(&unreadable); + connection + .progress_handler( + 1_024, + Some(move || match spill_bytes() { + Ok(bytes) => { + observer_peak.fetch_max(bytes, Ordering::Relaxed); + false + } + Err(_) => { + observer_failed.store(true, Ordering::Relaxed); + true + } + }), + ) + .unwrap(); + { + let tx = connection.transaction().unwrap(); + tx.execute( + "UPDATE row_blocks SET first_document=-first_document-1 WHERE first_document>0", + [], + ) + .unwrap(); + // The second bulk UPDATE materializes the transient rowid set. + tx.execute( + "UPDATE row_blocks SET first_document=-first_document-1 WHERE first_document<0", + [], + ) + .unwrap(); + tx.rollback().unwrap(); + } + assert!(!unreadable.load(Ordering::Relaxed)); + assert!( + peak.load(Ordering::Relaxed) > 0, + "old bulk rowid rewrite must spill" + ); + assert_eq!( + connection + .pragma_query_value(Some("temp"), "page_count", |r| r.get::<_, i64>(0)) + .unwrap(), + 0, + "TEMP pager counts cannot observe the transient statement spill" + ); + assert_eq!(blocks(&connection), original); + assert_eq!(spill_bytes().unwrap(), 0); + peak.store(0, Ordering::Relaxed); + { + let tx = connection.transaction().unwrap(); + shift_rowid_keys(&tx, "row_blocks", "first_document", &shift, &Active).unwrap(); + tx.commit().unwrap(); + } + connection + .progress_handler(0, None:: bool>) + .unwrap(); + assert!(!unreadable.load(Ordering::Relaxed)); + assert_eq!( + peak.load(Ordering::Relaxed), + 0, + "bounded rowid shift must not spill" + ); + assert_eq!(spill_bytes().unwrap(), 0); + assert_ne!(blocks(&connection), original); + integrity(&connection); +} + +#[test] +fn carried_row_chunks_check_cancellation_and_preserve_disk_rows() { + struct StopAfter { + checks: AtomicUsize, + deadline: bool, + } + impl CodeIndexExecutionControlV1 for StopAfter { + fn is_cancelled(&self) -> bool { + !self.deadline && self.checks.fetch_add(1, Ordering::Relaxed) >= 4_100 + } + + fn is_deadline_exceeded(&self) -> bool { + self.deadline && self.checks.fetch_add(1, Ordering::Relaxed) >= 4_100 + } + } + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("row-chunks.sqlite"); + let mut connection = Connection::open(&path).unwrap(); + connection + .execute_batch( + "PRAGMA temp_store=FILE; + PRAGMA cache_size=8; + CREATE TABLE row_chunks( + chunk_id BLOB NOT NULL PRIMARY KEY, + document_id INTEGER NOT NULL + ) WITHOUT ROWID; + CREATE TABLE independent(value TEXT NOT NULL); + INSERT INTO independent VALUES ('live');", + ) + .unwrap(); + { + let tx = connection.transaction().unwrap(); + let mut insert = tx + .prepare("INSERT INTO row_chunks(chunk_id, document_id) VALUES (?1, ?2)") + .unwrap(); + for document in 1_i64..=12_000 { + // Canonical chunk digests are BLOB keys; other valid IDs retain + // TEXT storage. Both must survive the same mixed-key scan. + let chunk_id = if document % 2 == 0 { + Value::Text(format!("literal-chunk:{document:05}")) + } else { + Value::Blob(document.to_be_bytes().repeat(4)) + }; + insert.execute(params![chunk_id, document]).unwrap(); + } + drop(insert); + tx.commit().unwrap(); + } + + let chunks = |db: &Connection| -> Vec<(Value, i64)> { + let mut query = db + .prepare("SELECT chunk_id, document_id FROM row_chunks ORDER BY chunk_id") + .unwrap(); + query + .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) + .unwrap() + .collect::>() + .unwrap() + }; + let original = chunks(&connection); + assert_eq!(original.len(), 12_000); + let (blob_keys, text_keys): (i64, i64) = connection + .query_row( + "SELECT SUM(typeof(chunk_id) = 'blob'), SUM(typeof(chunk_id) = 'text') + FROM row_chunks", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!((blob_keys, text_keys), (6_000, 6_000)); + let shift = ShiftV1::new([((8, 14), 4), ((45, 52), 10)]).unwrap(); + // Reference mapping is deliberately independent of ShiftV1::map: + // two replaced ranges remove 13 old chunks, shifting the intervening + // documents down by two and the trailing suffix up by one. + let expected: Vec<_> = original + .iter() + .filter_map(|(chunk, document)| { + let mapped = match *document { + 8..=13 | 45..=51 => None, + 14..=44 => Some(*document - 2), + 52.. => Some(*document + 1), + _ => Some(*document), + }; + mapped.map(|mapped| (chunk.clone(), mapped)) + }) + .collect(); + assert_eq!(expected.len(), 11_987); + assert_ne!(expected, original); + + for deadline in [false, true] { + let tx = connection.transaction().unwrap(); + let control = StopAfter { + checks: AtomicUsize::new(0), + deadline, + }; + let error = carry_row_chunks(&tx, &shift, &[], &control).unwrap_err(); + assert!( + matches!(error, CodeLexicalArtifactErrorV1::Interrupted(_)), + "both cancellation and deadline must interrupt the real remap: {error:?}" + ); + let partial = chunks(&tx); + assert_ne!(partial, original, "the interrupted carry did actual writes"); + assert_ne!(partial, expected, "it stopped before completing the remap"); + integrity(&tx); + tx.rollback().unwrap(); + drop(connection); + connection = Connection::open(&path).unwrap(); + assert_eq!( + chunks(&connection), + original, + "rollback must restore every chunk" + ); + integrity(&connection); + } + + let tx = connection.transaction().unwrap(); + carry_row_chunks(&tx, &shift, &[], &Active).unwrap(); + assert_eq!(chunks(&tx), expected); + integrity(&tx); + tx.commit().unwrap(); + drop(connection); + let reopened = Connection::open(&path).unwrap(); + assert_eq!( + chunks(&reopened), + expected, + "commit must survive a fresh connection" + ); + assert_eq!( + reopened + .query_row("SELECT value FROM independent", [], |row| row + .get::<_, String>(0)) + .unwrap(), + "live" + ); + integrity(&reopened); +}