diff --git a/crates/persistence/README.md b/crates/persistence/README.md index b872330a30..069d0f6d81 100644 --- a/crates/persistence/README.md +++ b/crates/persistence/README.md @@ -563,6 +563,11 @@ cargo build --bin hfs --release SQLite handles all CRUD operations, versioning, history, and search using its built-in FTS5 full-text search engine. Data is stored in `fhir.db` by default. +For a first-page search on a single parameter, SQLite reads at most 2,001 index rows of that parameter to decide +whether its key list is short enough to drive the query from, instead of letting the planner walk `resources` in +`last_updated` order. Set `HFS_SQLITE_SEARCH_PROBE=0` (`false`, `off` and `no` also work) to turn the probe off and +always use the planner's plan; it is read once per process and defaults to on. Results are the same either way. + ### SQLite + Elasticsearch SQLite handles CRUD, versioning, history, and transactions. Elasticsearch handles all search operations with: diff --git a/crates/persistence/src/backends/sqlite/schema.rs b/crates/persistence/src/backends/sqlite/schema.rs index f9be8adc7d..b87a871d68 100644 --- a/crates/persistence/src/backends/sqlite/schema.rs +++ b/crates/persistence/src/backends/sqlite/schema.rs @@ -10,9 +10,17 @@ use crate::core::bulk_submit_legacy::{ use crate::error::StorageResult; /// Current schema version. -pub const SCHEMA_VERSION: i32 = 41; - -/// The `search_index` value indexes. Excludes `idx_search_composite`, which the +pub const SCHEMA_VERSION: i32 = 46; + +/// The `search_index` value indexes. `idx_search_token`, `idx_search_quantity` +/// and `idx_search_composite_cov` end in (or carry) `resource_key`, so the +/// `rowid IN (SELECT resource_key FROM search_index ...)` membership +/// subqueries are answered from the index alone (v42). `idx_search_token` is +/// keyed code-first (v46: `value_token_code, resource_key, value_token_system, +/// composite_group, is_contained`), so a token search with no system +/// (`code=8302-2`) seeks it, one with a system and a code seeks it by the code +/// and tests the system in the index, and the token half of a token+quantity +/// composite (`code-value-quantity`) is paired from it too. Excludes `idx_search_composite`, which the /// delete-by-resource path needs at all times, and `idx_search_token_display`, /// dropped in v29 for write volume (#945): `value_token_display` is populated on /// most Coding rows but its only reader is the uncommon token `:text` / @@ -24,14 +32,14 @@ pub const SCHEMA_VERSION: i32 = 41; /// /// Keep each entry's SQL byte-for-byte what the migration ladder creates, /// normalised to one line, so the self-heal on startup and the ladder agree. -pub(crate) const SEARCH_VALUE_INDEXES: [(&str, &str); 16] = [ +pub(crate) const SEARCH_VALUE_INDEXES: [(&str, &str); 17] = [ ( "idx_search_string", "CREATE INDEX IF NOT EXISTS idx_search_string ON search_index(tenant_id, resource_type, param_name, value_string) WHERE value_string IS NOT NULL", ), ( "idx_search_token", - "CREATE INDEX IF NOT EXISTS idx_search_token ON search_index(tenant_id, resource_type, param_name, value_token_system, value_token_code) WHERE value_token_system IS NOT NULL OR value_token_code IS NOT NULL", + "CREATE INDEX IF NOT EXISTS idx_search_token ON search_index(tenant_id, resource_type, param_name, value_token_code, resource_key, value_token_system, composite_group, is_contained) WHERE value_token_system IS NOT NULL OR value_token_code IS NOT NULL", ), ( "idx_search_date", @@ -51,7 +59,7 @@ pub(crate) const SEARCH_VALUE_INDEXES: [(&str, &str); 16] = [ ), ( "idx_search_quantity", - "CREATE INDEX IF NOT EXISTS idx_search_quantity ON search_index(tenant_id, resource_type, param_name, value_quantity_value, value_quantity_unit) WHERE value_quantity_value IS NOT NULL", + "CREATE INDEX IF NOT EXISTS idx_search_quantity ON search_index(tenant_id, resource_type, param_name, value_quantity_value, value_quantity_unit, resource_key) WHERE value_quantity_value IS NOT NULL", ), ( "idx_search_reference", @@ -89,6 +97,10 @@ pub(crate) const SEARCH_VALUE_INDEXES: [(&str, &str); 16] = [ "idx_search_contained_composite_legacy", "CREATE INDEX IF NOT EXISTS idx_search_contained_composite_legacy ON search_index(tenant_id, contained_type, param_name) WHERE is_contained = 1 AND composite_group IS NOT NULL AND composite_slot IS NULL", ), + ( + "idx_search_composite_cov", + "CREATE INDEX IF NOT EXISTS idx_search_composite_cov ON search_index(tenant_id, resource_type, param_name, resource_key, composite_group, composite_slot, value_token_system, value_token_code, value_quantity_value, value_quantity_unit, value_quantity_system) WHERE composite_group IS NOT NULL AND is_contained = 0", + ), ]; /// Creates every missing [`SEARCH_VALUE_INDEXES`] entry and returns how many @@ -99,22 +111,65 @@ pub(crate) const SEARCH_VALUE_INDEXES: [(&str, &str); 16] = [ pub(crate) fn ensure_search_value_indexes(conn: &Connection) -> StorageResult { let mut created = 0; for (name, sql) in SEARCH_VALUE_INDEXES { - let exists: bool = conn + let existing: Option = conn .query_row( - "SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = ?1", + "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?1", [name], - |_| Ok(()), + |row| row.get(0), ) - .is_ok(); - if !exists { - conn.execute(sql, []) - .map_err(|e| migration_err(format!("recreate {name}: {e}")))?; - created += 1; + .ok(); + match existing { + None => { + conn.execute(sql, []) + .map_err(|e| migration_err(format!("recreate {name}: {e}")))?; + created += 1; + } + // The covering indexes (v42) keep their name across shapes, so an + // older binary after a rollback, or an out-of-band step, can leave + // the pre-v42 non-covering form behind at v42 forever. Results stay + // correct but the covering speedup silently vanishes, so compare + // the shape and rebuild on a mismatch. One query per index when + // everything matches. + Some(current) + if COVERING_SEARCH_INDEXES.contains(&name) + && normalise_index_sql(¤t) != normalise_index_sql(sql) => + { + let started = std::time::Instant::now(); + conn.execute(&format!("DROP INDEX IF EXISTS {name}"), []) + .map_err(|e| migration_err(format!("drop stale {name}: {e}")))?; + conn.execute(sql, []) + .map_err(|e| migration_err(format!("rebuild {name}: {e}")))?; + tracing::warn!( + index = name, + elapsed_ms = started.elapsed().as_millis() as u64, + "search_index value index had a stale (non-covering) shape and has been rebuilt" + ); + created += 1; + } + Some(_) => {} } } Ok(created) } +/// The v42 covering indexes whose shape [`ensure_search_value_indexes`] checks. +const COVERING_SEARCH_INDEXES: [&str; 3] = [ + "idx_search_quantity", + "idx_search_token", + "idx_search_composite_cov", +]; + +/// Lower-cases, drops `IF NOT EXISTS` (SQLite stores the statement without it) +/// and collapses whitespace, so a stored definition compares with the canonical +/// one regardless of formatting. +fn normalise_index_sql(sql: &str) -> String { + sql.to_lowercase() + .replace("if not exists ", "") + .split_whitespace() + .collect::>() + .join(" ") +} + /// Drops every [`SEARCH_VALUE_INDEXES`] entry. The bulk index rebuild's first /// half: with the value indexes gone, a rebuild's rows land in the table /// (an append) and `idx_search_composite` alone, and the indexes are then @@ -172,7 +227,7 @@ pub fn initialize_schema(conn: &Connection) -> StorageResult<()> { if recreated > 0 { tracing::warn!( recreated, - "search_index value indexes were missing at startup and have been rebuilt" + "search_index value indexes were missing or stale at startup and have been rebuilt" ); } @@ -344,7 +399,6 @@ fn create_schema_v1(conn: &Connection) -> StorageResult<()> { fn create_indexes(conn: &Connection) -> StorageResult<()> { let indexes = [ // Resources table indexes - "CREATE INDEX IF NOT EXISTS idx_resources_type ON resources(tenant_id, resource_type)", "CREATE INDEX IF NOT EXISTS idx_resources_updated ON resources(tenant_id, last_updated)", "CREATE INDEX IF NOT EXISTS idx_resources_reindex ON resources(tenant_id, resource_type, last_updated, id)", // Covering index for the live-resource counts the dashboard runs @@ -353,13 +407,13 @@ fn create_indexes(conn: &Connection) -> StorageResult<()> { // prefix, and `resource_type` then arrives already sorted *within* // that prefix — the `GROUP BY` needs no sort step and every column the // query mentions lives in the index, so the table is never touched. - // `idx_resources_type` cannot serve these: it lacks `is_deleted`, so + // A `(tenant_id, resource_type)` index (the former `idx_resources_type`, + // dropped in v45) cannot serve these: it lacks `is_deleted`, so // SQLite can walk it for the tenant but must then fetch each matching // row from the table just to test `is_deleted = 0` — millions of row // visits on a large corpus. "CREATE INDEX IF NOT EXISTS idx_resources_live_type ON resources(tenant_id, is_deleted, resource_type)", // History table indexes - "CREATE INDEX IF NOT EXISTS idx_history_resource ON resource_history(tenant_id, resource_type, id)", "CREATE INDEX IF NOT EXISTS idx_history_updated ON resource_history(tenant_id, last_updated)", // Search index indexes "CREATE INDEX IF NOT EXISTS idx_search_string ON search_index(tenant_id, resource_type, param_name, value_string) WHERE value_string IS NOT NULL", @@ -483,6 +537,11 @@ fn migrate_schema(conn: &Connection, from_version: i32, log_steps: bool) -> Stor 38 => migrate_v38_to_v39(conn)?, 39 => migrate_v39_to_v40(conn)?, 40 => migrate_v40_to_v41(conn)?, + 41 => migrate_v41_to_v42(conn)?, + 42 => migrate_v42_to_v43(conn)?, + 43 => migrate_v43_to_v44(conn)?, + 44 => migrate_v44_to_v45(conn)?, + 45 => migrate_v45_to_v46(conn)?, _ => { return Err(crate::error::StorageError::Backend( crate::error::BackendError::Internal { @@ -2176,6 +2235,180 @@ fn migrate_v40_to_v41(conn: &Connection) -> StorageResult<()> { .map_err(|e| migration_err(format!("v41 terminology tables: {e}"))) } +/// Migrate from schema version 41 to version 42. +/// +/// Makes the value indexes behind the hottest search shapes covering by +/// appending `resource_key`: `idx_search_quantity` and `idx_search_token` are +/// extended in place (same names, same partial predicates), and +/// `idx_search_composite_cov` is added beside `idx_search_composite`, which the +/// delete-by-resource path still needs. Without `resource_key` every index hit +/// costs a random read of its `search_index` row just to fetch the key for the +/// membership subquery. Measured on a 30% corpus copy: value-quantity `lt` +/// 3.3 s -> 0.04 s, category 0.54 s -> 0.025 s, composite 0.9 s -> 0.055 s, for +/// about +490 MB of file. Each build scans `search_index` and can take minutes +/// on a large database, so progress is logged per index. An index already of +/// the new shape is left alone, so a replay costs nothing. +/// +/// Operational cost: `CREATE INDEX` needs temporary space about the size of the +/// index being built and holds the write lock while it runs, so writes block +/// for the duration (about 16 s at 6 GB). Plan the upgrade window accordingly. +/// The startup self-repair ([`ensure_search_value_indexes`]) re-applies the +/// same rebuild if these indexes are ever found in the old shape again. +fn migrate_v41_to_v42(conn: &Connection) -> StorageResult<()> { + for (name, current_marker) in [ + ( + "idx_search_quantity", + "%value_quantity_unit, resource_key)%", + ), + ("idx_search_token", "%value_token_code, resource_key%"), + ("idx_search_composite_cov", "%"), + ] { + let started = std::time::Instant::now(); + let is_current: bool = conn + .query_row( + "SELECT sql LIKE ?2 FROM sqlite_master WHERE type = 'index' AND name = ?1", + rusqlite::params![name, current_marker], + |row| row.get(0), + ) + .unwrap_or(false); + if !is_current { + conn.execute(&format!("DROP INDEX IF EXISTS {name}"), []) + .map_err(|e| migration_err(format!("v42 drop {name}: {e}")))?; + let sql = SEARCH_VALUE_INDEXES + .iter() + .find(|(n, _)| *n == name) + .map(|(_, sql)| *sql) + .expect("v42 index is in SEARCH_VALUE_INDEXES"); + conn.execute(sql, []) + .map_err(|e| migration_err(format!("v42 create {name}: {e}")))?; + tracing::info!( + index = name, + elapsed_ms = started.elapsed().as_millis() as u64, + "Built covering search index" + ); + } + } + Ok(()) +} + +/// Migrate from schema version 42 to version 43. +/// +/// A no-op that keeps the version number. An earlier build of this change built +/// `idx_search_composite_code` here; v46 supersedes it (the token half of a composite is read from the +/// reshaped `idx_search_token`), so a database coming from v42 skips the build +/// instead of paying for an index that v46 would drop. A database that already +/// carries the index (it ran the original v43) has it dropped by v46. +fn migrate_v42_to_v43(_conn: &Connection) -> StorageResult<()> { + Ok(()) +} + +/// Migrate from schema version 43 to version 44. +/// +/// A no-op, for the same reason as v43: an earlier build built +/// `idx_search_token_code`, a second code-first copy of every token row. It +/// cost about a third more WAL per create (one more b-tree page per token row) +/// and v46 gets the same code seek by reshaping `idx_search_token` itself, so +/// a database coming from v43 skips the build, and one that has the index has +/// it dropped by v46. +fn migrate_v43_to_v44(_conn: &Connection) -> StorageResult<()> { + Ok(()) +} + +/// Migrate from schema version 44 to version 45. +/// +/// Drops the two indexes that are exact prefixes of the key that already +/// exists, so they only cost a b-tree insertion per write and a few MB each: +/// `idx_resources_type (tenant_id, resource_type)` is a prefix of the primary +/// key `(tenant_id, resource_type, id)` (and of `idx_resources_reindex`), and +/// `idx_history_resource (tenant_id, resource_type, id)` is a prefix of +/// `resource_history`'s primary key `(tenant_id, resource_type, id, +/// version_id)`. Every lookup they served seeks the longer key on the same +/// equality prefix. Dropped only when the stored definition is exactly the +/// historical one: an index of that name with another shape is left alone. +fn migrate_v44_to_v45(conn: &Connection) -> StorageResult<()> { + for (name, definition) in [ + ( + "idx_resources_type", + "CREATE INDEX idx_resources_type ON resources(tenant_id, resource_type)", + ), + ( + "idx_history_resource", + "CREATE INDEX idx_history_resource ON resource_history(tenant_id, resource_type, id)", + ), + ] { + let stored: Option = conn + .query_row( + "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?1", + [name], + |row| row.get(0), + ) + .ok(); + if stored.is_some_and(|sql| normalise_index_sql(&sql) == normalise_index_sql(definition)) { + conn.execute(&format!("DROP INDEX IF EXISTS {name}"), []) + .map_err(|e| migration_err(format!("v45 drop {name}: {e}")))?; + } + } + Ok(()) +} + +/// Migrate from schema version 45 to version 46. +/// +/// Reshapes `idx_search_token` code-first, `(tenant_id, resource_type, +/// param_name, value_token_code, resource_key, value_token_system, +/// composite_group, is_contained)` with the same partial predicate, and drops +/// the two extra indexes the earlier v43 and v44 added +/// (`idx_search_composite_code`, `idx_search_token_code`). The old shape began +/// its values with the system, so a token search without one (`code=8302-2`, +/// `category=laboratory`, `status=finished`) could not seek it and read every +/// row of the parameter. The new shape seeks by code (a search with a system +/// and a code seeks the code and tests the system in the index), is covering +/// for the membership subquery, carries `composite_group` and `is_contained` +/// so the token half of a token+quantity composite is paired from the index +/// alone, and has `resource_key` right after the code so the `EXISTS` forms +/// seek `(code, resource_key)`. It replaces the index instead of adding a +/// twin: a second token index cost about a third more WAL per create (a +/// b-tree page per token row), the same single index costs what the old one +/// did. The one search that loses its seek is a bare `system|` token (no +/// code), which now reads the parameter's index range instead. +/// +/// The rebuild scans `search_index` and holds the write lock while it runs +/// (about 10 s per GB of index at 6 GB); the startup self-repair +/// ([`ensure_search_value_indexes`]) compares the shape of +/// `idx_search_token` and would rebuild it anyway if it were ever found in the +/// old one. +fn migrate_v45_to_v46(conn: &Connection) -> StorageResult<()> { + for legacy in ["idx_search_composite_code", "idx_search_token_code"] { + conn.execute(&format!("DROP INDEX IF EXISTS {legacy}"), []) + .map_err(|e| migration_err(format!("v46 drop {legacy}: {e}")))?; + } + let (name, sql) = SEARCH_VALUE_INDEXES + .iter() + .find(|(n, _)| *n == "idx_search_token") + .copied() + .expect("idx_search_token is in SEARCH_VALUE_INDEXES"); + let stored: Option = conn + .query_row( + "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?1", + [name], + |row| row.get(0), + ) + .ok(); + if stored.is_some_and(|stored| normalise_index_sql(&stored) == normalise_index_sql(sql)) { + return Ok(()); + } + let started = std::time::Instant::now(); + conn.execute(&format!("DROP INDEX IF EXISTS {name}"), []) + .map_err(|e| migration_err(format!("v46 drop {name}: {e}")))?; + conn.execute(sql, []) + .map_err(|e| migration_err(format!("v46 create {name}: {e}")))?; + tracing::info!( + index = name, + elapsed_ms = started.elapsed().as_millis() as u64, + "Rebuilt the token search index code-first" + ); + Ok(()) +} + fn migration_err(message: String) -> crate::error::StorageError { crate::error::StorageError::Backend(crate::error::BackendError::Internal { backend_name: "sqlite".to_string(), @@ -3266,6 +3499,300 @@ mod tests { assert_eq!(version, SCHEMA_VERSION); } + /// v41 -> v42: the old-shape quantity/token indexes are rebuilt with + /// `resource_key`, the covering composite index is added, and a replay is a + /// no-op. A subquery over the new indexes must be answered covering. + #[test] + fn migrate_v41_to_v42_makes_value_indexes_covering() { + let conn = Connection::open_in_memory().unwrap(); + initialize_schema(&conn).unwrap(); + conn.execute_batch( + "DROP INDEX idx_search_quantity; + DROP INDEX idx_search_token; + DROP INDEX idx_search_composite_cov; + CREATE INDEX idx_search_quantity ON search_index(tenant_id, resource_type, param_name, value_quantity_value, value_quantity_unit) WHERE value_quantity_value IS NOT NULL; + CREATE INDEX idx_search_token ON search_index(tenant_id, resource_type, param_name, value_token_system, value_token_code) WHERE value_token_system IS NOT NULL OR value_token_code IS NOT NULL;", + ) + .unwrap(); + set_schema_version(&conn, 41).unwrap(); + initialize_schema(&conn).unwrap(); + assert_eq!(get_schema_version(&conn).unwrap(), SCHEMA_VERSION); + let sql_of = |name: &str| -> String { + conn.query_row( + "SELECT sql FROM sqlite_master WHERE type='index' AND name=?1", + [name], + |r| r.get(0), + ) + .unwrap() + }; + assert!(sql_of("idx_search_quantity").contains("value_quantity_unit, resource_key)")); + assert!( + sql_of("idx_search_token") + .contains("value_token_code, resource_key, value_token_system") + ); + assert!( + sql_of("idx_search_composite_cov") + .contains("param_name, resource_key, composite_group") + ); + // The quantity subquery is covering. + let plan: Vec = conn + .prepare( + "EXPLAIN QUERY PLAN SELECT resource_key FROM search_index WHERE tenant_id = ?1 \ + AND resource_type = ?2 AND param_name = 'value-quantity' AND value_quantity_value > 100", + ) + .unwrap() + .query_map(["t", "Observation"], |r| r.get::<_, String>(3)) + .unwrap() + .map(|r| r.unwrap()) + .collect(); + assert!( + plan.iter() + .any(|l| l.contains("COVERING INDEX idx_search_quantity")), + "plan: {plan:?}" + ); + // Replay is a no-op. + migrate_v41_to_v42(&conn).unwrap(); + } + + /// v45 -> v46 drops the two extra code-first indexes of the earlier + /// v43/v44, rebuilds `idx_search_token` code-first, and a replay is a + /// no-op. The searches the reshape serves seek it: a bare code, a + /// `system|code`, and the token half of a token+quantity composite paired + /// from the index alone with the `EXISTS` form seeking `(code, + /// resource_key)`; a bare `system|` still answers from the index range. + #[test] + fn migrate_v45_to_v46_reshapes_the_token_index_code_first() { + let conn = Connection::open_in_memory().unwrap(); + initialize_schema(&conn).unwrap(); + // A v45 database as the earlier branch left it. + conn.execute_batch( + "DROP INDEX idx_search_token; + CREATE INDEX idx_search_token ON search_index(tenant_id, resource_type, param_name, value_token_system, value_token_code, resource_key) WHERE value_token_system IS NOT NULL OR value_token_code IS NOT NULL; + CREATE INDEX idx_search_composite_code ON search_index(tenant_id, resource_type, param_name, value_token_code, resource_key, composite_group, value_token_system) WHERE composite_group IS NOT NULL AND is_contained = 0 AND value_token_code IS NOT NULL; + CREATE INDEX idx_search_token_code ON search_index(tenant_id, resource_type, param_name, value_token_code, value_token_system, resource_key) WHERE value_token_code IS NOT NULL;", + ) + .unwrap(); + set_schema_version(&conn, 45).unwrap(); + initialize_schema(&conn).unwrap(); + assert_eq!(get_schema_version(&conn).unwrap(), SCHEMA_VERSION); + let exists = |name: &str| -> bool { + conn.query_row( + "SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1", + [name], + |r| r.get::<_, i64>(0), + ) + .unwrap() + == 1 + }; + assert!(!exists("idx_search_composite_code") && !exists("idx_search_token_code")); + let (sql, rowid) = index_sql_and_rowid(&conn, "idx_search_token"); + assert!( + sql.contains("param_name, value_token_code, resource_key, value_token_system, composite_group, is_contained)"), + "{sql}" + ); + // A replay leaves the rebuilt index alone. + migrate_v45_to_v46(&conn).unwrap(); + assert_eq!(index_sql_and_rowid(&conn, "idx_search_token").1, rowid); + + let plan_of = |sql: &str| -> String { + conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .unwrap() + .query_map([], |r| r.get::<_, String>(3)) + .unwrap() + .map(|r| r.unwrap()) + .collect::>() + .join(" | ") + }; + let membership = |predicate: &str| { + plan_of(&format!( + "SELECT resource_key FROM search_index WHERE tenant_id = 't' \ + AND resource_type = 'Observation' AND param_name = 'code' AND {predicate}" + )) + }; + for predicate in [ + "value_token_code = '8302-2'", + "value_token_system IN ('http://loinc.org', 'x') AND value_token_code = '8302-2'", + "(value_token_system IS NULL OR value_token_system IN ('', 'x')) AND value_token_code = '8302-2'", + ] { + let plan = membership(predicate); + assert!( + plan.contains("COVERING INDEX idx_search_token (tenant_id=? AND resource_type=? AND param_name=? AND value_token_code=?"), + "{predicate}: {plan}" + ); + } + // A bare `system|` has no seek on the system any more, but the index + // still answers it, covering. + let system_only = membership("value_token_system = 'http://loinc.org'"); + assert!( + system_only.contains("COVERING INDEX idx_search_token"), + "{system_only}" + ); + // The code-first pairing of a token+quantity composite: the token + // rows seek the index by code, each pairs with its quantity row of the + // same group through the covering composite index, and the + // `EXISTS` form probes `(code, resource_key)`. + let pairing = "c.tenant_id = 't' AND c.resource_type = 'Observation' AND c.param_name = 'code-value-quantity' \ + AND c.is_contained = 0 AND c.composite_group IS NOT NULL AND c.value_token_code IS NOT NULL \ + AND (value_token_code = '8302-2') AND EXISTS (SELECT 1 FROM search_index AS q \ + INDEXED BY idx_search_composite_cov WHERE q.tenant_id = 't' AND q.resource_type = 'Observation' \ + AND q.param_name = 'code-value-quantity' AND q.resource_key = c.resource_key \ + AND q.composite_group = c.composite_group AND q.composite_group IS NOT NULL \ + AND q.is_contained = 0 AND (value_quantity_value > 5))"; + let rows = plan_of(&format!( + "SELECT c.resource_key FROM search_index AS c INDEXED BY idx_search_token WHERE {pairing}" + )); + assert!( + rows.contains("SEARCH c USING COVERING INDEX idx_search_token (tenant_id=? AND resource_type=? AND param_name=? AND value_token_code=?)"), + "{rows}" + ); + let window = plan_of(&format!( + "SELECT 1 FROM resources AS w WHERE EXISTS (SELECT 1 FROM search_index AS c \ + INDEXED BY idx_search_token WHERE c.resource_key = w.rowid AND {pairing})" + )); + assert!( + window.contains("USING COVERING INDEX idx_search_token (tenant_id=? AND resource_type=? AND param_name=? AND value_token_code=? AND resource_key=?)"), + "{window}" + ); + } + + /// v42 -> v43 and v43 -> v44 build nothing any more: the indexes they + /// used to build are superseded by v46, which drops them when present. + #[test] + fn migrate_v43_and_v44_build_no_index() { + let conn = Connection::open_in_memory().unwrap(); + initialize_schema(&conn).unwrap(); + let count = |conn: &Connection| -> i64 { + conn.query_row( + "SELECT count(*) FROM sqlite_master WHERE type='index'", + [], + |r| r.get(0), + ) + .unwrap() + }; + let before = count(&conn); + migrate_v42_to_v43(&conn).unwrap(); + migrate_v43_to_v44(&conn).unwrap(); + assert_eq!(count(&conn), before); + } + + /// v44 -> v45 drops the two exact-prefix indexes and nothing else; the + /// lookups they served seek the primary keys. A same-named index of + /// another shape is left alone, and a replay is a no-op. + #[test] + fn migrate_v44_to_v45_drops_the_prefix_indexes() { + let conn = Connection::open_in_memory().unwrap(); + initialize_schema(&conn).unwrap(); + let exists = |name: &str| -> bool { + conn.query_row( + "SELECT count(*) FROM sqlite_master WHERE type='index' AND name=?1", + [name], + |r| r.get::<_, i64>(0), + ) + .unwrap() + == 1 + }; + assert!(!exists("idx_resources_type") && !exists("idx_history_resource")); + // An older database has them; the ladder drops them. + conn.execute_batch( + "CREATE INDEX idx_resources_type ON resources(tenant_id, resource_type); + CREATE INDEX idx_history_resource ON resource_history(tenant_id, resource_type, id);", + ) + .unwrap(); + set_schema_version(&conn, 44).unwrap(); + initialize_schema(&conn).unwrap(); + assert!(!exists("idx_resources_type") && !exists("idx_history_resource")); + // Other shapes under the same names survive. + conn.execute_batch( + "CREATE INDEX idx_resources_type ON resources(tenant_id, resource_type, id);", + ) + .unwrap(); + migrate_v44_to_v45(&conn).unwrap(); + assert!(exists("idx_resources_type")); + conn.execute_batch("DROP INDEX idx_resources_type").unwrap(); + migrate_v44_to_v45(&conn).unwrap(); + // The lookups they served use the primary keys. + let plan_of = |sql: &str| -> String { + conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .unwrap() + .query_map([], |r| r.get::<_, String>(3)) + .unwrap() + .map(|r| r.unwrap()) + .collect::>() + .join(" | ") + }; + let resources = + plan_of("SELECT id FROM resources WHERE tenant_id = 't' AND resource_type = 'Patient'"); + assert!( + resources.contains("sqlite_autoindex_resources_1") + || resources.contains("idx_resources_reindex"), + "{resources}" + ); + let history = plan_of( + "SELECT version_id FROM resource_history WHERE tenant_id = 't' AND resource_type = 'Patient' AND id = 'x'", + ); + assert!( + history.contains("sqlite_autoindex_resource_history_1"), + "{history}" + ); + } + + fn index_sql_and_rowid(conn: &Connection, name: &str) -> (String, i64) { + conn.query_row( + "SELECT sql, rowid FROM sqlite_master WHERE type='index' AND name=?1", + [name], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap() + } + + /// An old-shape index left at v42 (rollback to a v41 binary, out-of-band + /// step) is detected and rebuilt covering by the startup self-repair; a + /// second pass changes nothing. + #[test] + #[tracing_test::traced_test] + fn startup_repair_rebuilds_stale_non_covering_index() { + let conn = Connection::open_in_memory().unwrap(); + initialize_schema(&conn).unwrap(); + conn.execute_batch( + "DROP INDEX idx_search_quantity; + CREATE INDEX idx_search_quantity ON search_index(tenant_id, resource_type, param_name, value_quantity_value, value_quantity_unit) WHERE value_quantity_value IS NOT NULL;", + ) + .unwrap(); + assert!( + !index_sql_and_rowid(&conn, "idx_search_quantity") + .0 + .contains("resource_key") + ); + + initialize_schema(&conn).unwrap(); + let (sql, rowid) = index_sql_and_rowid(&conn, "idx_search_quantity"); + assert!(sql.contains("resource_key"), "sql: {sql}"); + assert!(logs_contain("stale (non-covering) shape")); + let plan: Vec = conn + .prepare( + "EXPLAIN QUERY PLAN SELECT resource_key FROM search_index WHERE tenant_id = ?1 \ + AND resource_type = ?2 AND param_name = 'value-quantity' AND value_quantity_value > 100", + ) + .unwrap() + .query_map(["t", "Observation"], |r| r.get::<_, String>(3)) + .unwrap() + .map(|r| r.unwrap()) + .collect(); + assert!( + plan.iter() + .any(|l| l.contains("COVERING INDEX idx_search_quantity")), + "plan: {plan:?}" + ); + + // Idempotent: nothing rebuilt, the sqlite_master entry is untouched. + assert_eq!(ensure_search_value_indexes(&conn).unwrap(), 0); + initialize_schema(&conn).unwrap(); + assert_eq!( + index_sql_and_rowid(&conn, "idx_search_quantity"), + (sql, rowid) + ); + } + #[test] fn composite_slot_migration_preserves_legacy_rows_and_replays() { let conn = Connection::open_in_memory().unwrap(); @@ -3278,6 +3805,7 @@ mod tests { conn.execute_batch( "DROP INDEX idx_search_composite_legacy; DROP INDEX idx_search_contained_composite_legacy; + DROP INDEX idx_search_composite_cov; ALTER TABLE search_index DROP COLUMN composite_slot; PRAGMA foreign_keys = OFF; INSERT INTO search_index (tenant_id, resource_type, resource_id, param_name, diff --git a/crates/persistence/src/backends/sqlite/search/mod.rs b/crates/persistence/src/backends/sqlite/search/mod.rs index 6668a2b3eb..2fbd628cad 100644 --- a/crates/persistence/src/backends/sqlite/search/mod.rs +++ b/crates/persistence/src/backends/sqlite/search/mod.rs @@ -20,5 +20,6 @@ pub mod writer; pub use filter_parser::{FilterExpr, FilterOp, FilterParseError, FilterParser, FilterSqlGenerator}; pub use parameter_handlers::CompositeComponentDef; pub use query_builder::{KeysetKey, QueryBuilder, SortValueKind, SqlFragment, SqlParam}; +pub(crate) use query_builder::{is_parameter_rows, membership_exists_predicate}; pub use strategy::{SearchStrategyCapability, SqliteSearchStrategy}; pub use writer::{SqlValue, SqliteSearchIndexWriter}; diff --git a/crates/persistence/src/backends/sqlite/search/query_builder.rs b/crates/persistence/src/backends/sqlite/search/query_builder.rs index 208f210d57..9aa3e80f13 100644 --- a/crates/persistence/src/backends/sqlite/search/query_builder.rs +++ b/crates/persistence/src/backends/sqlite/search/query_builder.rs @@ -8,8 +8,8 @@ use std::collections::HashSet; use crate::error::SearchError; use crate::types::{ - CompartmentMembership, ContainedMode, SearchModifier, SearchParamType, SearchParameter, - SearchQuery, SearchValue, strip_reference_version, + CompartmentMembership, CompositeSearchComponent, ContainedMode, SearchModifier, + SearchParamType, SearchParameter, SearchQuery, SearchValue, strip_reference_version, }; use super::modifier_handlers::{build_missing_condition, get_missing_value, is_missing_modifier}; @@ -234,11 +234,128 @@ impl ParameterCondition { /// the searched tenant (`?1`) and type (`?2`), that satisfy `condition`. fn parameter_rows(name: &str, condition: &str) -> String { format!( - "SELECT resource_key FROM search_index WHERE tenant_id = ?1 AND resource_type = ?2 AND param_name = {} AND ({condition})", + "{PARAMETER_ROWS_HEAD}param_name = {} AND ({condition})", sql_string_literal(name) ) } +/// What every [`parameter_rows`] subquery starts with, up to its +/// `param_name` test. +const PARAMETER_ROWS_HEAD: &str = + "SELECT resource_key FROM search_index WHERE tenant_id = ?1 AND resource_type = ?2 AND "; + +/// Whether `sql` is exactly one [`parameter_rows`] select: a single +/// parameter's membership, nothing intersected with it. +pub(crate) fn is_parameter_rows(sql: &str) -> bool { + (sql.starts_with(PARAMETER_ROWS_HEAD) || sql.starts_with(COMPOSITE_PAIR_HEAD)) + && !sql.contains(" INTERSECT ") +} + +/// The same membership as a correlated `EXISTS` on the outer `resources` row, +/// for a subquery built by [`parameter_rows`]: `rowid IN ()` becomes +/// `EXISTS ( for outer_key)`, which lets the outer query walk +/// `resources` in page order and stop at the first full page instead of +/// materializing every matching key first. Reuses the subquery's numbered +/// placeholders, so the same parameters bind. `None` for anything that is not +/// one plain `parameter_rows` select or a `UNION ALL` of them (a date's +/// ranges, tested as `EXISTS (…) OR EXISTS (…)`); an `INTERSECT` or a nested +/// subquery keeps `rowid IN (…)`. +pub(crate) fn membership_exists_predicate(rows_sql: &str, outer_key: &str) -> Option { + // The code-first composite pairing: its one nested `EXISTS` stays inside, + // and `c.resource_key = outer_key` joins the seek on `(…, code, + // resource_key)`. + if let Some(rest) = rows_sql.strip_prefix(COMPOSITE_PAIR_HEAD) { + if [" UNION ", " INTERSECT ", " EXCEPT ", " GROUP BY "] + .iter() + .any(|keyword| rest.contains(keyword)) + || rest.matches("SELECT").count() != 1 + { + return None; + } + return Some(format!( + "EXISTS (SELECT 1 FROM search_index AS c INDEXED BY idx_search_token \ + WHERE c.tenant_id = ?1 AND c.resource_type = ?2 AND c.resource_key = {outer_key} AND {rest})" + )); + } + // A `UNION ALL` of parameter selects (a date's `ne`/`le`/`ge` ranges): a + // resource is a member when any one range holds for it, so the branches + // are `EXISTS`-es joined by `OR`. + if rows_sql.contains(" UNION ALL ") { + let branches = rows_sql + .split(" UNION ALL ") + .map(|branch| parameter_exists_predicate(branch, outer_key)) + .collect::>>()?; + return Some(format!("({})", branches.join(" OR "))); + } + parameter_exists_predicate(rows_sql, outer_key) +} + +/// [`membership_exists_predicate`] for one plain [`parameter_rows`] select. +fn parameter_exists_predicate(rows_sql: &str, outer_key: &str) -> Option { + let rest = rows_sql.strip_prefix(PARAMETER_ROWS_HEAD)?; + if [" UNION ", " INTERSECT ", " EXCEPT ", "SELECT"] + .iter() + .any(|keyword| rest.contains(keyword)) + { + return None; + } + Some(format!( + "EXISTS (SELECT 1 FROM search_index INDEXED BY idx_search_composite \ + WHERE tenant_id = ?1 AND resource_type = ?2 AND resource_key = {outer_key} AND {rest})" + )) +} + +/// What a [`composite_pair_rows`] subquery starts with, up to its +/// `param_name` test. +const COMPOSITE_PAIR_HEAD: &str = "SELECT c.resource_key FROM search_index AS c INDEXED BY idx_search_token WHERE c.tenant_id = ?1 AND c.resource_type = ?2 AND "; + +/// The keys of the resources with a composite group in which a token row +/// matches the token component and a row of the same group matches the +/// quantity component: the rows of the whole-parameter +/// `GROUP BY resource_key, composite_group HAVING MAX(CASE token) AND +/// MAX(CASE quantity)` for a two-component token+quantity composite, without +/// reading every row of the parameter. The code-first token index +/// (`idx_search_token`, which carries `composite_group` and `is_contained`) +/// finds the token rows by code; each one is +/// then paired by a probe of `idx_search_composite_cov` on `(resource_key, +/// composite_group)`. A key can be listed once per matching token row, which +/// the `rowid IN (…)` callers ignore. +/// +/// `None` (keep the `GROUP BY`) unless the composite is exactly one token and +/// one quantity component and the token part tests `value_token_code = ?`: a +/// `system|` token has no code to seek. The fragments' columns are unqualified; the token part sits in the +/// outer select, so it binds to `c`, and the quantity part in the `EXISTS`, +/// so it binds to `q`. Both `INDEXED BY` hints need the partial-index terms +/// spelled out in the `WHERE`. +fn composite_pair_rows( + name: &str, + components: &[CompositeSearchComponent], + fragments: &[SqlFragment], +) -> Option { + if components.len() != 2 || fragments.len() != 2 { + return None; + } + let (token, quantity) = match (components[0].param_type, components[1].param_type) { + (SearchParamType::Token, SearchParamType::Quantity) => (&fragments[0], &fragments[1]), + (SearchParamType::Quantity, SearchParamType::Token) => (&fragments[1], &fragments[0]), + _ => return None, + }; + if !token.sql.contains("value_token_code = ?") { + return None; + } + let name = sql_string_literal(name); + Some(format!( + "{COMPOSITE_PAIR_HEAD}c.param_name = {name} AND c.is_contained = 0 \ + AND c.composite_group IS NOT NULL AND c.value_token_code IS NOT NULL AND ({token_sql}) \ + AND EXISTS (SELECT 1 FROM search_index AS q INDEXED BY idx_search_composite_cov \ + WHERE q.tenant_id = ?1 AND q.resource_type = ?2 AND q.param_name = {name} \ + AND q.resource_key = c.resource_key AND q.composite_group = c.composite_group \ + AND q.composite_group IS NOT NULL AND q.is_contained = 0 AND ({quantity_sql}))", + token_sql = token.sql, + quantity_sql = quantity.sql, + )) +} + /// Builds SQL queries from FHIR search parameters. pub struct QueryBuilder { /// The tenant ID for the query. @@ -252,6 +369,12 @@ pub struct QueryBuilder { param_offset: usize, /// Whether to skip tenant/resource type params (they're shared with outer query). skip_base_params: bool, + /// Whether a token+quantity composite is built as the code-first pairing + /// ([`composite_pair_rows`], hinted to `idx_search_token` and + /// `idx_search_composite_cov`) instead of a whole-parameter `GROUP BY`. + /// Off unless the caller has seen both indexes exist: `INDEXED BY` fails + /// to prepare against a missing one (a bulk rebuild drops them). + composite_rewrite: bool, } impl QueryBuilder { @@ -262,9 +385,18 @@ impl QueryBuilder { resource_type: resource_type.into(), param_offset: 0, skip_base_params: false, + composite_rewrite: false, } } + /// Builds a token+quantity composite as the code-first pairing, which + /// needs `idx_search_token` (in its code-first shape) and + /// `idx_search_composite_cov` to exist (see [`composite_pair_rows`]). The rows are the same set. + pub fn with_composite_rewrite(mut self, enabled: bool) -> Self { + self.composite_rewrite = enabled; + self + } + /// Sets the parameter offset for embedded subqueries. /// /// When the generated SQL will be embedded in an outer query that already @@ -977,6 +1109,22 @@ impl QueryBuilder { param_offset + total_params, ) { Some(fragments) if !fragments.is_empty() => { + // One value only: an OR of values is wrapped by the caller + // in a scan of the type's rows (not a positive membership), + // and there the pairing measured slower than the GROUP BY + // (0.30 s -> 0.59 s for two wide codes). + if self.composite_rewrite + && param.values.len() == 1 + && let Some(rows) = + composite_pair_rows(¶m.name, ¶m.components, &fragments) + { + for f in fragments { + total_params += f.params.len(); + params.extend(f.params); + } + or_conditions.push(format!("resource_key IN ({rows})")); + continue; + } let havings: Vec = fragments .iter() .map(|f| format!("MAX(CASE WHEN {} THEN 1 ELSE 0 END) = 1", f.sql)) @@ -1476,6 +1624,36 @@ fn positive_membership_subquery(condition: &str) -> Option<&str> { mod tests { use super::*; + #[test] + fn exists_form_only_for_one_plain_parameter_select() { + let rows = parameter_rows("code", "value_token_code = ?3"); + let exists = membership_exists_predicate(&rows, "w.rk").unwrap(); + assert!( + exists + .starts_with("EXISTS (SELECT 1 FROM search_index INDEXED BY idx_search_composite") + ); + assert!( + exists.contains( + "resource_key = w.rk AND param_name = 'code' AND (value_token_code = ?3)" + ) + ); + assert!(is_parameter_rows(&rows)); + let union = format!("{rows} UNION ALL {rows}"); + let exists_union = membership_exists_predicate(&union, "w.rk").unwrap(); + assert!(exists_union.starts_with("(EXISTS (SELECT 1 FROM search_index INDEXED BY")); + assert_eq!(exists_union.matches(") OR EXISTS (").count(), 1); + let nested = format!("{rows} UNION ALL SELECT resource_key FROM search_index"); + assert!(membership_exists_predicate(&nested, "w.rk").is_none()); + let union_all_then_union = format!("{rows} UNION {rows}"); + assert!(membership_exists_predicate(&union_all_then_union, "w.rk").is_none()); + let intersect = format!("{rows} INTERSECT {rows}"); + assert!(!is_parameter_rows(&intersect)); + assert!( + membership_exists_predicate("SELECT DISTINCT resource_key FROM search_index", "w.rk") + .is_none() + ); + } + #[test] fn positive_membership_subquery_takes_exactly_one_membership() { assert_eq!( diff --git a/crates/persistence/src/backends/sqlite/search/writer.rs b/crates/persistence/src/backends/sqlite/search/writer.rs index 3660290d6a..059459386c 100644 --- a/crates/persistence/src/backends/sqlite/search/writer.rs +++ b/crates/persistence/src/backends/sqlite/search/writer.rs @@ -193,6 +193,32 @@ impl SqliteSearchIndexWriter { }) } + /// Most rows [`Self::insert_sql_rows`] carries: 512 x 28 = 14,336 + /// parameters, well under SQLite's variable limit. + pub const MAX_ROWS_PER_INSERT: usize = 512; + + /// One INSERT carrying `rows` rows (1..=[`Self::MAX_ROWS_PER_INSERT`], each + /// [`Self::COLUMNS`] positional parameters): a resource's whole index in + /// one statement instead of one per row (or one per eight). The text is + /// built once per row count and kept, so the connection's prepared + /// statement cache sees one stable string per count. + pub fn insert_sql_rows(rows: usize) -> &'static str { + use std::collections::HashMap; + use std::sync::{Mutex, OnceLock}; + assert!((1..=Self::MAX_ROWS_PER_INSERT).contains(&rows)); + static CACHE: OnceLock>> = OnceLock::new(); + let mut cache = CACHE + .get_or_init(Default::default) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + cache.entry(rows).or_insert_with(|| { + let base = Self::insert_sql(); + let cols = &base[..base.find("VALUES").expect("insert_sql has VALUES")]; + let group = format!("({})", ["?"; Self::COLUMNS].join(", ")); + Box::leak(format!("{cols}VALUES {}", vec![group; rows].join(", ")).into_boxed_str()) + }) + } + /// Number of base columns [`Self::insert_sql`] writes, and so the length of /// every vector [`Self::to_sql_params`] returns. pub const COLUMNS: usize = 28; @@ -465,6 +491,31 @@ mod tests { use crate::search::DateEnd; use crate::types::{DatePrecision, SearchParamType}; + #[test] + fn multi_row_insert_has_one_group_per_row_and_is_stable() { + for rows in [ + 1usize, + 2, + 8, + 19, + SqliteSearchIndexWriter::MAX_ROWS_PER_INSERT, + ] { + let sql = SqliteSearchIndexWriter::insert_sql_rows(rows); + assert_eq!( + sql.matches('?').count(), + rows * SqliteSearchIndexWriter::COLUMNS + ); + assert!(std::ptr::eq( + sql, + SqliteSearchIndexWriter::insert_sql_rows(rows) + )); + } + assert_eq!( + SqliteSearchIndexWriter::insert_sql_rows(8), + SqliteSearchIndexWriter::insert_sql_rows8() + ); + } + #[test] fn composite_slot_survives_single_batch_and_contained_inserts() { let conn = rusqlite::Connection::open_in_memory().unwrap(); diff --git a/crates/persistence/src/backends/sqlite/search_impl.rs b/crates/persistence/src/backends/sqlite/search_impl.rs index f6c5921c57..af42d9f8a1 100644 --- a/crates/persistence/src/backends/sqlite/search_impl.rs +++ b/crates/persistence/src/backends/sqlite/search_impl.rs @@ -27,7 +27,10 @@ use crate::types::{ }; use super::SqliteBackend; -use super::search::{KeysetKey, QueryBuilder, SortValueKind, SqlParam}; +use super::search::{ + KeysetKey, QueryBuilder, SortValueKind, SqlParam, is_parameter_rows, + membership_exists_predicate, +}; fn internal_error(message: String) -> StorageError { StorageError::Backend(BackendError::Internal { @@ -382,6 +385,37 @@ impl SqliteBackend { tenant: &TenantContext, query: &SearchQuery, total: Option, + ) -> StorageResult { + self.search_with_connection_probe(conn, tenant, query, total, probe_enabled()) + } + + /// [`Self::search_with_connection`] with the bounded probe forced on or + /// off (`HFS_SQLITE_SEARCH_PROBE` decides in production). + #[allow(clippy::type_complexity)] + pub(crate) fn search_with_connection_probe( + &self, + conn: &rusqlite::Connection, + tenant: &TenantContext, + query: &SearchQuery, + total: Option, + probe: bool, + ) -> StorageResult { + with_composite_rewrite(conn, query, |rewrite| { + self.search_with_connection_plan(conn, tenant, query, total, probe, rewrite) + }) + } + + /// [`Self::search_with_connection_probe`] with the code-first composite + /// pairing forced on or off. + #[allow(clippy::type_complexity)] + pub(crate) fn search_with_connection_plan( + &self, + conn: &rusqlite::Connection, + tenant: &TenantContext, + query: &SearchQuery, + total: Option, + probe: bool, + composite_rewrite: bool, ) -> StorageResult { reject_unsupported_metadata_modifier(query)?; reject_legacy_composites(conn, tenant, query)?; @@ -412,8 +446,9 @@ impl SqliteBackend { let param_offset = if cursor.is_some() { 4 } else { 2 }; let search_filter = if !query.parameters.is_empty() || query.compartment.is_some() { - let builder = - QueryBuilder::new(tenant_id, resource_type).with_param_offset(param_offset); + let builder = QueryBuilder::new(tenant_id, resource_type) + .with_param_offset(param_offset) + .with_composite_rewrite(composite_rewrite); let fragment = builder.build(query); if !fragment.sql.is_empty() { Some(fragment) @@ -423,12 +458,43 @@ impl SqliteBackend { } else { None }; - let filter_clause = search_filter + let mut filter_clause = search_filter .as_ref() .map(|f| format!(" AND rowid IN ({})", f.sql)) .unwrap_or_default(); let first_page_source = first_page_source(search_filter.is_some(), query, count); - let search_params = search_filter.map(|f| f.params).unwrap_or_default(); + // First page of a one-parameter membership search: the probe may pick + // a cheaper plan for it (same rows, same order). + let mut probed_keys: Option> = None; + let first_page_plan = match &search_filter { + Some(f) + if probe + && cursor.is_none() + && query.offset.is_none() + && query.sort.is_empty() + && query.compartment.is_none() + && query.parameters.len() == 1 + && count > 1 => + { + let (plan, keys) = + probe_first_page(conn, tenant_id, resource_type, &f.sql, &f.params); + if plan == FirstPagePlan::ListDriven { + probed_keys = Some(keys); + } + plan + } + _ => FirstPagePlan::Planner, + }; + // A short key list is already in hand: the page reads those rows by + // rowid instead of evaluating the filter a second time (a pairing of + // token and quantity rows costs as much as the probe did). + if let Some(keys) = &probed_keys { + let placeholders = (0..keys.len()) + .map(|i| format!("?{}", param_offset + 1 + i)) + .collect::>() + .join(", "); + filter_clause = format!(" AND rowid IN ({placeholders})"); + } // SELECT the sort key alongside the row so the next cursor can be built. let select_cols = match &keyset { @@ -446,6 +512,39 @@ impl SqliteBackend { QueryBuilder::new(tenant_id, resource_type).build_order_by(query) }; + // The bounded early-exit attempts, tried before the page statement and + // kept only when one fills the page: the newest [`WINDOWS`] resources + // in turn. A short window (the filter matches few of the newest + // resources) widens before the page statement runs, since a miss only + // costs the window. + let window_sqls: Vec = match (first_page_plan, &search_filter) { + (FirstPagePlan::ExistsDriven, Some(f)) => membership_exists_predicate(&f.sql, "w.rk") + .map(|exists| { + WINDOWS + .iter() + .map(|window| { + format!( + "SELECT {select_cols} FROM \ + (SELECT rowid AS rk, last_updated AS lu, id AS rid \ + FROM resources INDEXED BY idx_resources_reindex \ + WHERE tenant_id = ?1 AND resource_type = ?2 AND is_deleted = 0 \ + ORDER BY last_updated DESC, id ASC LIMIT {window}) AS w \ + CROSS JOIN resources ON resources.rowid = w.rk \ + WHERE {exists} \ + ORDER BY w.lu DESC, w.rid ASC LIMIT {lim}", + lim = count + 1, + ) + }) + .collect() + }) + .unwrap_or_default(), + _ => Vec::new(), + }; + let search_params = match probed_keys { + Some(keys) => keys.into_iter().map(SqlParam::Integer).collect(), + None => search_filter.map(|f| f.params).unwrap_or_default(), + }; + // Build query based on pagination mode. let (sql, has_previous) = if let (Some(cursor), Some(k)) = (&cursor, &keyset) { let backward = cursor.direction() == CursorDirection::Previous; @@ -474,22 +573,23 @@ impl SqliteBackend { ); (sql, offset > 0) } else { - let sql = format!( - "SELECT {cols} FROM {first_page_source} \ - WHERE tenant_id = ?1 AND resource_type = ?2 AND is_deleted = 0{filter} \ - {order} LIMIT {lim}", - cols = select_cols, - filter = filter_clause, - order = order_by, - lim = count + 1, - ); + let page = |source: &str, filter: &str| { + format!( + "SELECT {cols} FROM {source} \ + WHERE tenant_id = ?1 AND resource_type = ?2 AND is_deleted = 0{filter} \ + {order} LIMIT {lim}", + cols = select_cols, + order = order_by, + lim = count + 1, + ) + }; + let sql = match first_page_plan { + FirstPagePlan::ListDriven => page("resources NOT INDEXED", &filter_clause), + _ => page(first_page_source, &filter_clause), + }; (sql, false) }; - let mut stmt = conn - .prepare(&sql) - .or_query_error("Failed to prepare search query")?; - // Bind params: tenant, type, then (cursor) sort value + id, then filter params. let mut all_params: Vec> = vec![ Box::new(tenant_id.to_string()), @@ -511,8 +611,11 @@ impl SqliteBackend { // The sort key (column 5) is selected only when keyset paging is active. let sort_kind = keyset.as_ref().map(|k| k.kind); - let raw_rows: Vec<(String, String, Vec, String, String, Option)> = stmt - .query_map(param_refs.as_slice(), |row| { + let run = |sql: &str| -> StorageResult> { + let mut stmt = conn + .prepare(sql) + .or_query_error("Failed to prepare search query")?; + stmt.query_map(param_refs.as_slice(), |row| { let id: String = row.get(0)?; let version_id: String = row.get(1)?; let data: Vec = row.get(2)?; @@ -530,7 +633,18 @@ impl SqliteBackend { }) .or_query_error("Failed to execute search")? .collect::, _>>() - .or_query_error("Failed to read row")?; + .or_query_error("Failed to read row") + }; + // A window attempt may fail to prepare (an index dropped for a bulk + // load) or fall short of a full page: either way the next, wider + // window and finally the page statement run. + let windowed = window_sqls + .iter() + .find_map(|window| run(window).ok().filter(|rows| rows.len() > count)); + let raw_rows = match windowed { + Some(rows) => rows, + None => run(&sql)?, + }; // Parse rows, carrying the sort key for cursor construction. let mut parsed: Vec<(StoredResource, Option)> = Vec::new(); @@ -624,6 +738,109 @@ impl SqliteBackend { } } +/// One `resources` row as the page statements return it, with its sort key. +type RawRow = (String, String, Vec, String, String, Option); + +/// Newest `resources` rows the early-exit attempts look at, narrowest first. +/// A page the 512 newest cannot fill (a date range that the newest resources +/// mostly miss) is retried on 4,096 before the planner's plan runs. +const WINDOWS: [usize; 2] = [512, 4096]; + +/// Rows of a single parameter's index the probe reads at most. A filter with +/// fewer rows is "selective": its whole key list is cheap to build. +const PROBE_CAP: usize = 2001; + +/// Whether the bounded probe is on: `HFS_SQLITE_SEARCH_PROBE`, read once. +/// Unset or empty means on; `0`, `false`, `off` or `no` turn it off. +fn probe_enabled() -> bool { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| match std::env::var("HFS_SQLITE_SEARCH_PROBE") { + Ok(value) => !matches!( + value.trim().to_ascii_lowercase().as_str(), + "0" | "false" | "off" | "no" + ), + Err(_) => true, + }) +} + +/// How the first page of a one-parameter membership search reaches its rows. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum FirstPagePlan { + /// The planner's choice: `rowid IN ()` with `resources` walked in + /// `last_updated` order. + Planner, + /// The key list is short (under [`PROBE_CAP`] index rows): drive from it + /// and read `resources` by rowid, sorting the few rows. The probe's own + /// keys are bound to the page statement, so the filter runs once. + ListDriven, + /// The key list is long: try the newest [`WINDOWS`] resources in page + /// order, each tested against the index with a correlated `EXISTS`; the + /// page is final when the window fills it, else the planner's plan runs. + ExistsDriven, +} + +/// Picks the [`FirstPagePlan`] with one bounded probe over the filter's own +/// index rows (`SELECT … FROM ( LIMIT 2001)`, a seek that stops early). +/// Any shape other than one plain parameter membership, and any probe error, +/// is [`FirstPagePlan::Planner`]: the result of every plan is the same set in +/// the same `ORDER BY last_updated DESC, id` order, so this only chooses cost. +/// +/// Only the tests call it; the production path reaches the same plans through +/// the probe's own keys. +#[cfg(test)] +fn choose_first_page_plan( + conn: &rusqlite::Connection, + tenant_id: &str, + resource_type: &str, + rows_sql: &str, + filter_params: &[SqlParam], +) -> FirstPagePlan { + probe_first_page(conn, tenant_id, resource_type, rows_sql, filter_params).0 +} + +/// [`choose_first_page_plan`], and for [`FirstPagePlan::ListDriven`] the +/// distinct `resource_key`s the probe read (under [`PROBE_CAP`] of them), so +/// the page can be read by rowid without evaluating the filter again. +fn probe_first_page( + conn: &rusqlite::Connection, + tenant_id: &str, + resource_type: &str, + rows_sql: &str, + filter_params: &[SqlParam], +) -> (FirstPagePlan, Vec) { + if !is_parameter_rows(rows_sql) { + return (FirstPagePlan::Planner, Vec::new()); + } + let probe = format!("SELECT resource_key FROM ({rows_sql} LIMIT {PROBE_CAP})"); + let mut bound: Vec> = vec![ + Box::new(tenant_id.to_string()), + Box::new(resource_type.to_string()), + ]; + for param in filter_params { + bound.push(match param { + SqlParam::String(s) => Box::new(s.clone()), + SqlParam::Integer(i) => Box::new(*i), + SqlParam::Float(f) => Box::new(*f), + SqlParam::Null => Box::new(Option::::None), + }); + } + let refs: Vec<&dyn rusqlite::ToSql> = bound.iter().map(|p| p.as_ref()).collect(); + let keys = conn.prepare(&probe).and_then(|mut stmt| { + stmt.query_map(refs.as_slice(), |row| row.get::<_, Option>(0))? + .collect::, _>>() + }); + match keys { + Ok(keys) if keys.len() < PROBE_CAP => { + let mut keys: Vec = keys.into_iter().flatten().collect(); + keys.sort_unstable(); + keys.dedup(); + (FirstPagePlan::ListDriven, keys) + } + Ok(_) => (FirstPagePlan::ExistsDriven, Vec::new()), + Err(_) => (FirstPagePlan::Planner, Vec::new()), + } +} + /// The table expression of a first page (no cursor, no offset) of `count` /// resources: `resources`, or `resources NOT INDEXED` for a filtered one-row /// page on the default order. @@ -644,6 +861,50 @@ fn first_page_source(filtered: bool, query: &SearchQuery, count: usize) -> &'sta } } +/// Whether `query` can use the code-first composite pairing: it has a +/// composite parameter and both indexes the pairing is hinted to exist, the +/// token index in its code-first shape (a database still carrying the earlier, +/// system-first `idx_search_token` would answer the hinted statement with a +/// table lookup per token row). `INDEXED BY` makes a statement fail to prepare +/// against a missing index, and a bulk rebuild drops the value indexes for its +/// duration. +fn composite_rewrite_available(conn: &rusqlite::Connection, query: &SearchQuery) -> bool { + use crate::types::SearchParamType; + if !query + .parameters + .iter() + .any(|p| p.param_type == SearchParamType::Composite) + { + return false; + } + conn.query_row( + "SELECT count(*) FROM sqlite_master WHERE type = 'index' AND ( \ + name = 'idx_search_composite_cov' \ + OR (name = 'idx_search_token' AND sql LIKE '%value_token_code, resource_key, value_token_system, composite_group%'))", + [], + |row| row.get::<_, i64>(0), + ) + .map(|n| n == 2) + .unwrap_or(false) +} + +/// Runs `run(rewrite)` with the composite pairing on when +/// [`composite_rewrite_available`], and again with it off if an index was +/// dropped between that check and the statement's prepare. +fn with_composite_rewrite( + conn: &rusqlite::Connection, + query: &SearchQuery, + run: impl Fn(bool) -> StorageResult, +) -> StorageResult { + if !composite_rewrite_available(conn, query) { + return run(false); + } + match run(true) { + Err(error) if error.to_string().contains("no such index") => run(false), + result => result, + } +} + /// The id-only page of [`SearchProvider::search_ids`], run on a blocking /// thread by the caller. fn search_ids_on( @@ -651,6 +912,18 @@ fn search_ids_on( tenant: &TenantContext, query: &SearchQuery, cursor: Option, +) -> StorageResult> { + with_composite_rewrite(conn, query, |rewrite| { + search_ids_on_plan(conn, tenant, query, cursor.clone(), rewrite) + }) +} + +fn search_ids_on_plan( + conn: &rusqlite::Connection, + tenant: &TenantContext, + query: &SearchQuery, + cursor: Option, + composite_rewrite: bool, ) -> StorageResult> { let tenant_id = tenant.tenant_id().as_str(); let resource_type = &query.resource_type; @@ -658,6 +931,7 @@ fn search_ids_on( let search_filter = if !query.parameters.is_empty() { let fragment = QueryBuilder::new(tenant_id, resource_type) .with_param_offset(param_offset) + .with_composite_rewrite(composite_rewrite) .build(query); if fragment.sql.is_empty() { None @@ -826,6 +1100,17 @@ fn search_count_on( conn: &rusqlite::Connection, tenant: &TenantContext, query: &SearchQuery, +) -> StorageResult { + with_composite_rewrite(conn, query, |rewrite| { + search_count_on_plan(conn, tenant, query, rewrite) + }) +} + +fn search_count_on_plan( + conn: &rusqlite::Connection, + tenant: &TenantContext, + query: &SearchQuery, + composite_rewrite: bool, ) -> StorageResult { reject_legacy_composites(conn, tenant, query)?; let tenant_id = tenant.tenant_id().as_str(); @@ -835,7 +1120,9 @@ fn search_count_on( let (sql, all_params): (String, Vec>) = if !query.parameters.is_empty() || query.compartment.is_some() { - let builder = QueryBuilder::new(tenant_id, resource_type).with_param_offset(2); + let builder = QueryBuilder::new(tenant_id, resource_type) + .with_param_offset(2) + .with_composite_rewrite(composite_rewrite); let fragment = builder.build(query); let mut params: Vec> = vec![ @@ -3118,7 +3405,7 @@ mod tests { ] { TRACED_SQL.with(|traced| traced.borrow_mut().clear()); let result = backend - .search_with_connection(&conn, &tenant, &query(count), None) + .search_with_connection_probe(&conn, &tenant, &query(count), None, false) .unwrap(); let ids: Vec<&str> = result.resources.items.iter().map(|r| r.id()).collect(); assert_eq!(ids.len(), count.min(2) as usize, "{ids:?}"); @@ -3787,4 +4074,554 @@ mod tests { let result = backend.search(&tenant, &q).await.unwrap(); assert!(result.resources.items.is_empty()); } + + // ---- bounded probe + first-page plan selection ---- + + use crate::types::{CompositeSearchComponent, SearchPrefix}; + + fn obs_param( + name: &str, + param_type: SearchParamType, + prefix: SearchPrefix, + value: &str, + ) -> SearchParameter { + SearchParameter { + name: name.to_string(), + param_type, + modifier: None, + values: vec![SearchValue::new(prefix, value)], + chain: vec![], + components: vec![], + } + } + + /// 2,300 Observations: `B2001` is indexed on exactly 2,001 of them and + /// `B2000` on exactly 2,000, so a probe capped at 2,001 rows sees the + /// boundary on either side. + async fn probe_fixture() -> (SqliteBackend, TenantContext) { + let backend = create_test_backend(); + let tenant = create_test_tenant(); + for i in 0..2300usize { + let mut codings = + vec![json!({"system": "http://loinc.org", "code": format!("C{}", i % 7)})]; + if i < 2001 { + codings.push(json!({"system": "http://x.org", "code": "B2001"})); + } + if i < 2000 { + codings.push(json!({"system": "http://x.org", "code": "B2000"})); + } + backend + .create( + &tenant, + "Observation", + json!({ + "resourceType": "Observation", + "id": format!("o{i:04}"), + "status": "final", + "category": [{"coding": [{ + "system": "http://terminology.hl7.org/CodeSystem/observation-category", + "code": if i % 10 == 0 { "vital-signs" } else { "laboratory" }}]}], + "code": {"coding": codings}, + "subject": {"reference": format!("Patient/p{}", i % 50)}, + "effectiveDateTime": format!("{}-01-15T10:00:00Z", 2000 + i % 20), + "valueQuantity": {"value": (i % 500) as f64, "unit": "mg", + "system": "http://unitsofmeasure.org", "code": "mg"} + }), + FhirVersion::default(), + ) + .await + .unwrap(); + } + (backend, tenant) + } + + fn probe_plan_of( + backend: &SqliteBackend, + tenant: &TenantContext, + query: &SearchQuery, + ) -> FirstPagePlan { + let conn = backend.get_connection().unwrap(); + let fragment = QueryBuilder::new(tenant.tenant_id().as_str(), &query.resource_type) + .with_param_offset(2) + .build(query); + choose_first_page_plan( + &conn, + tenant.tenant_id().as_str(), + &query.resource_type, + &fragment.sql, + &fragment.params, + ) + } + + /// Every page of `query` through the planner path (probe off) and the + /// probe path, compared id by id. + fn assert_same_pages(backend: &SqliteBackend, tenant: &TenantContext, query: &SearchQuery) { + let conn = backend.get_connection().unwrap(); + let mut cursor_old: Option = None; + let mut cursor_new: Option = None; + let mut seen = 0usize; + for page_no in 0..6 { + let mut q_old = query.clone(); + q_old.cursor = cursor_old.clone(); + let mut q_new = query.clone(); + q_new.cursor = cursor_new.clone(); + let old = backend + .search_with_connection_probe(&conn, tenant, &q_old, None, false) + .unwrap(); + let new = backend + .search_with_connection_probe(&conn, tenant, &q_new, None, true) + .unwrap(); + let ids = |r: &SearchResult| -> Vec { + r.resources + .items + .iter() + .map(|x| x.id().to_string()) + .collect() + }; + assert_eq!( + ids(&old), + ids(&new), + "page {page_no} of {:?}", + query.parameters + ); + assert_eq!( + old.resources.page_info.has_next, + new.resources.page_info.has_next + ); + seen += ids(&new).len(); + cursor_old = old.resources.page_info.next_cursor.clone(); + cursor_new = new.resources.page_info.next_cursor.clone(); + if cursor_old.is_none() { + break; + } + } + assert!( + seen > 0 || query.parameters.is_empty(), + "{:?} matched nothing", + query.parameters + ); + } + + #[tokio::test] + async fn probe_plan_gives_the_same_pages_as_the_planner() { + let (backend, tenant) = probe_fixture().await; + let mut shapes: Vec = vec![ + // token: selective, dense, boundary, rare + obs_param("code", SearchParamType::Token, SearchPrefix::Eq, "C3"), + obs_param( + "code", + SearchParamType::Token, + SearchPrefix::Eq, + "http://loinc.org|C5", + ), + obs_param("code", SearchParamType::Token, SearchPrefix::Eq, "B2000"), + obs_param("code", SearchParamType::Token, SearchPrefix::Eq, "B2001"), + obs_param( + "category", + SearchParamType::Token, + SearchPrefix::Eq, + "laboratory", + ), + obs_param( + "category", + SearchParamType::Token, + SearchPrefix::Eq, + "vital-signs", + ), + // quantity: dense, selective, empty + obs_param( + "value-quantity", + SearchParamType::Quantity, + SearchPrefix::Lt, + "460", + ), + obs_param( + "value-quantity", + SearchParamType::Quantity, + SearchPrefix::Gt, + "450", + ), + obs_param( + "value-quantity", + SearchParamType::Quantity, + SearchPrefix::Le, + "20", + ), + // date: dense, selective, union (le/ge), empty + obs_param( + "date", + SearchParamType::Date, + SearchPrefix::Gt, + "2005-01-01", + ), + obs_param( + "date", + SearchParamType::Date, + SearchPrefix::Gt, + "2018-01-01", + ), + obs_param("date", SearchParamType::Date, SearchPrefix::Le, "2003"), + obs_param("date", SearchParamType::Date, SearchPrefix::Ge, "2010"), + // reference + obs_param( + "subject", + SearchParamType::Reference, + SearchPrefix::Eq, + "Patient/p7", + ), + ]; + // composite stays on the planner path but must still agree + shapes.push(SearchParameter { + name: "code-value-quantity".into(), + param_type: SearchParamType::Composite, + values: vec![SearchValue::eq("C3$gt100")], + components: vec![ + CompositeSearchComponent { + param_type: SearchParamType::Token, + param_name: "code".into(), + }, + CompositeSearchComponent { + param_type: SearchParamType::Quantity, + param_name: "value-quantity".into(), + }, + ], + ..Default::default() + }); + for param in shapes { + for count in [2u32, 7, 50, 250] { + let mut query = SearchQuery::new("Observation").with_parameter(param.clone()); + query.count = Some(count); + assert_same_pages(&backend, &tenant, &query); + } + } + // a value no row has: both plans return an empty page + let mut empty = SearchQuery::new("Observation").with_parameter(obs_param( + "code", + SearchParamType::Token, + SearchPrefix::Eq, + "no-such-code", + )); + empty.count = Some(20); + let conn = backend.get_connection().unwrap(); + for probe in [false, true] { + let r = backend + .search_with_connection_probe(&conn, &tenant, &empty, None, probe) + .unwrap(); + assert!(r.resources.items.is_empty()); + } + } + + fn cvq(values: &[&str]) -> SearchParameter { + SearchParameter { + name: "code-value-quantity".into(), + param_type: SearchParamType::Composite, + values: values.iter().map(|v| SearchValue::eq(*v)).collect(), + components: vec![ + CompositeSearchComponent { + param_type: SearchParamType::Token, + param_name: "code".into(), + }, + CompositeSearchComponent { + param_type: SearchParamType::Quantity, + param_name: "value-quantity".into(), + }, + ], + ..Default::default() + } + } + + /// The code-first composite pairing returns the pages, the page flags and + /// the total of the whole-parameter `GROUP BY`, with the probe on and off, + /// for a selective code, a code past the probe cap (the `EXISTS`-driven + /// window), an empty one, an OR of values, and the shapes that keep the + /// `GROUP BY` (`system|` without a code). + #[tokio::test] + async fn composite_pairing_gives_the_same_pages_as_the_group_by() { + let (backend, tenant) = probe_fixture().await; + let conn = backend.get_connection().unwrap(); + let ids = |r: &SearchResult| -> Vec { + r.resources + .items + .iter() + .map(|x| x.id().to_string()) + .collect() + }; + let mut paired_somewhere = false; + for values in [ + vec!["C3$gt100"], + vec!["http://loinc.org|C5$lt20"], + vec!["B2001$ge0"], + vec!["B2001$gt400"], + vec!["B2000$lt3"], + vec!["C1$gt1000"], + vec!["no-such$gt1"], + vec!["C3$gt480", "C5$lt3"], + vec!["http://loinc.org|$gt490"], + vec!["|C2$gt490"], + ] { + for count in [2u32, 7, 50] { + for probe in [false, true] { + let mut query = SearchQuery::new("Observation").with_parameter(cvq(&values)); + query.count = Some(count); + let mut cursor: Option = None; + for page_no in 0..4 { + query.cursor = cursor.clone(); + let old = backend + .search_with_connection_plan(&conn, &tenant, &query, None, probe, false) + .unwrap(); + let new = backend + .search_with_connection_plan(&conn, &tenant, &query, None, probe, true) + .unwrap(); + assert_eq!( + ids(&old), + ids(&new), + "{values:?} count {count} probe {probe} page {page_no}" + ); + assert_eq!( + old.resources.page_info.has_next, + new.resources.page_info.has_next + ); + paired_somewhere |= !ids(&new).is_empty(); + cursor = old.resources.page_info.next_cursor.clone(); + if cursor.is_none() { + break; + } + } + } + } + let query = SearchQuery::new("Observation").with_parameter(cvq(&values)); + assert_eq!( + search_count_on_plan(&conn, &tenant, &query, false).unwrap(), + search_count_on_plan(&conn, &tenant, &query, true).unwrap(), + "{values:?} total" + ); + } + assert!(paired_somewhere); + } + + /// The pairing is chosen only for a token(code)+quantity composite; the + /// wide one past the probe cap is `EXISTS`-driven, and a missing hinted + /// index falls back to the `GROUP BY` instead of failing the search. + #[tokio::test] + async fn composite_pairing_plan_and_missing_index_fallback() { + let (backend, tenant) = probe_fixture().await; + let conn = backend.get_connection().unwrap(); + let build = |values: &[&str], rewrite: bool| { + QueryBuilder::new(tenant.tenant_id().as_str(), "Observation") + .with_param_offset(2) + .with_composite_rewrite(rewrite) + .build(&SearchQuery::new("Observation").with_parameter(cvq(values))) + }; + assert!( + build(&["C3$gt1"], true) + .sql + .contains("INDEXED BY idx_search_token") + ); + assert!(!build(&["C3$gt1"], false).sql.contains("INDEXED BY")); + assert!( + !build(&["http://loinc.org|$gt1"], true) + .sql + .contains("INDEXED BY") + ); + // An OR of values keeps the GROUP BY (the pairing measured slower there). + assert!( + !build(&["C3$gt1", "C5$gt1"], true) + .sql + .contains("INDEXED BY") + ); + let wide = build(&["B2001$ge0"], true); + assert_eq!( + choose_first_page_plan(&conn, "test-tenant", "Observation", &wide.sql, &wide.params), + FirstPagePlan::ExistsDriven + ); + assert!( + membership_exists_predicate(&wide.sql, "w.rk") + .unwrap() + .contains("c.resource_key = w.rk") + ); + let narrow = build(&["C3$gt400"], true); + assert_eq!( + choose_first_page_plan( + &conn, + "test-tenant", + "Observation", + &narrow.sql, + &narrow.params + ), + FirstPagePlan::ListDriven + ); + + let mut query = SearchQuery::new("Observation").with_parameter(cvq(&["C3$gt100"])); + query.count = Some(5); + let with_index = backend + .search_with_connection(&conn, &tenant, &query, None) + .unwrap(); + assert!(!with_index.resources.items.is_empty()); + conn.execute("DROP INDEX idx_search_token", []).unwrap(); + let without = backend + .search_with_connection(&conn, &tenant, &query, None) + .unwrap(); + let ids = |r: &SearchResult| -> Vec { + r.resources + .items + .iter() + .map(|x| x.id().to_string()) + .collect() + }; + assert_eq!(ids(&with_index), ids(&without)); + // Forcing the pairing against the missing index is the error the + // fallback absorbs. + let forced = backend.search_with_connection_plan(&conn, &tenant, &query, None, false, true); + assert!(forced.unwrap_err().to_string().contains("no such index")); + assert_eq!( + search_count_on(&conn, &tenant, &query).unwrap(), + search_count_on_plan(&conn, &tenant, &query, false).unwrap() + ); + } + + /// A broad date search (`le`: a `UNION ALL` of two ranges) whose matches + /// are all older than the newest 512 resources: the first window comes up + /// empty, the 4,096 window fills the page, and the page is the planner's. + #[tokio::test] + async fn union_date_search_widens_the_recency_window() { + let backend = create_test_backend(); + let tenant = create_test_tenant(); + for i in 0..2600usize { + backend + .create( + &tenant, + "Observation", + json!({ + "resourceType": "Observation", + "id": format!("d{i:04}"), + "status": "final", + "code": {"coding": [{"system": "http://loinc.org", "code": "X"}]}, + // the 2,050 oldest are the only matches of `le2000` + "effectiveDateTime": if i < 2050 { "1990-01-15T10:00:00Z" } else { "2020-01-15T10:00:00Z" }, + }), + FhirVersion::default(), + ) + .await + .unwrap(); + } + let query = SearchQuery::new("Observation").with_parameter(obs_param( + "date", + SearchParamType::Date, + SearchPrefix::Le, + "2000", + )); + assert_eq!( + probe_plan_of(&backend, &tenant, &query), + FirstPagePlan::ExistsDriven + ); + let conn = backend.get_connection().unwrap(); + conn.trace_v2( + rusqlite::trace::TraceEventCodes::SQLITE_TRACE_STMT, + Some(record_traced_sql), + ); + TRACED_SQL.with(|t| t.borrow_mut().clear()); + let probed = backend + .search_with_connection_probe(&conn, &tenant, &query, None, true) + .unwrap(); + let windows: Vec = TRACED_SQL.with(|t| { + t.borrow() + .iter() + .filter(|s| s.contains("INDEXED BY idx_resources_reindex")) + .cloned() + .collect() + }); + assert_eq!(windows.len(), 2, "{windows:?}"); + assert!(windows[0].contains("LIMIT 512)") && windows[1].contains("LIMIT 4096)")); + assert!(windows[1].contains(") OR EXISTS (")); + let planned = backend + .search_with_connection_probe(&conn, &tenant, &query, None, false) + .unwrap(); + let ids = |r: &SearchResult| -> Vec { + r.resources + .items + .iter() + .map(|x| x.id().to_string()) + .collect() + }; + assert_eq!(ids(&probed), ids(&planned)); + assert_eq!(ids(&probed).len(), 100); + assert!(probed.resources.page_info.has_next); + } + + #[tokio::test] + async fn probe_boundary_is_2001_index_rows() { + let (backend, tenant) = probe_fixture().await; + let plan = |value: &str| { + let mut q = SearchQuery::new("Observation").with_parameter(obs_param( + "code", + SearchParamType::Token, + SearchPrefix::Eq, + value, + )); + q.count = Some(10); + probe_plan_of(&backend, &tenant, &q) + }; + assert_eq!(plan("B2000"), FirstPagePlan::ListDriven, "2000 rows"); + assert_eq!(plan("B2001"), FirstPagePlan::ExistsDriven, "2001 rows"); + assert_eq!(plan("C3"), FirstPagePlan::ListDriven); + assert_eq!(plan("no-such-code"), FirstPagePlan::ListDriven); + // composites, `:not` and two parameters never take the probe + let mut two = SearchQuery::new("Observation") + .with_parameter(obs_param( + "code", + SearchParamType::Token, + SearchPrefix::Eq, + "B2001", + )) + .with_parameter(obs_param( + "category", + SearchParamType::Token, + SearchPrefix::Eq, + "laboratory", + )); + two.count = Some(10); + assert_eq!( + probe_plan_of(&backend, &tenant, &two), + FirstPagePlan::Planner + ); + } + + #[tokio::test] + async fn probe_picks_the_measured_plans_for_the_page_statement() { + let (backend, tenant) = probe_fixture().await; + let conn = backend.get_connection().unwrap(); + conn.trace_v2( + rusqlite::trace::TraceEventCodes::SQLITE_TRACE_STMT, + Some(record_traced_sql), + ); + for (value, needle) in [ + ("B2000", "resources NOT INDEXED"), + ( + "B2001", + "EXISTS (SELECT 1 FROM search_index INDEXED BY idx_search_composite", + ), + ] { + TRACED_SQL.with(|t| t.borrow_mut().clear()); + let mut q = SearchQuery::new("Observation").with_parameter(obs_param( + "code", + SearchParamType::Token, + SearchPrefix::Eq, + value, + )); + q.count = Some(10); + backend + .search_with_connection_probe(&conn, &tenant, &q, None, true) + .unwrap(); + let page = TRACED_SQL + .with(|t| { + t.borrow() + .iter() + .find(|s| s.starts_with("SELECT id, version_id, data")) + .cloned() + }) + .expect("the page statement"); + assert!(page.contains(needle), "{value}: {page}"); + } + conn.trace_v2(rusqlite::trace::TraceEventCodes::empty(), None); + } } diff --git a/crates/persistence/src/backends/sqlite/storage.rs b/crates/persistence/src/backends/sqlite/storage.rs index 78b50014f1..4f15eabe25 100644 --- a/crates/persistence/src/backends/sqlite/storage.rs +++ b/crates/persistence/src/backends/sqlite/storage.rs @@ -151,7 +151,8 @@ const RECENT_WRITES_SQL: &str = "SELECT COUNT(*) FROM (SELECT 1 FROM resource_hi /// One type's bucketed history deltas (`count_deltas_by_bucket`): `?1` /// tenant, `?2` type, `?3` window bound, `?4` bucket width. The unary `+` keeps -/// the planner off `idx_history_resource`, whose `(tenant_id, resource_type)` +/// the planner off `idx_history_resource` (present on databases created before +/// schema v45, which drops it), whose `(tenant_id, resource_type)` /// equality looks better unanalysed but visits every version of the type with /// a table read each; `idx_history_updated` covers the query (schema v40, /// #2107), so it reads only the window, from the index alone. @@ -304,7 +305,7 @@ impl ResourceStorage for SqliteBackend { ) -> StorageResult { tenant.check_permission(Operation::Create, resource_type)?; - let conn = self.get_connection()?; + let mut conn = self.get_connection()?; let tenant_id = tenant.tenant_id().as_str(); // Extract or generate ID @@ -314,8 +315,42 @@ impl ResourceStorage for SqliteBackend { .map(String::from) .unwrap_or_else(crate::types::new_resource_id); + // Ensure the resource has correct type and id + let mut resource = resource; + if let Some(obj) = resource.as_object_mut() { + obj.insert( + "resourceType".to_string(), + Value::String(resource_type.to_string()), + ); + obj.insert("id".to_string(), Value::String(id.clone())); + } + + // Serialize the resource data + let data = serde_json::to_vec(&resource) + .map_err(|e| serialization_error(format!("Failed to serialize resource: {}", e)))?; + + // Extract the search values before taking the write lock: it is pure + // CPU, and SQLite has one writer. + let prepared = (!self.is_search_offloaded()) + .then(|| self.prepare_index(tenant_id, resource_type, &id, &resource)); + + let version_id = "1"; + let fhir_version_str = fhir_version.as_mime_param(); + + // The resource row, its history row and its search index (with the + // full-text rows) are ONE transaction, as update and delete already + // are. This used to be ~17 auto-committed statements: each its own + // WAL commit (+34% WAL per create, and one fsync per statement under + // `synchronous=FULL`: 46 ms against 6 ms), and a failure part-way + // left a resource with no, or half, of its search index. + // IMMEDIATE takes the write lock up front, where the busy handler + // applies, so the existence check below cannot race another create. + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|e| internal_error(format!("Failed to begin create: {}", e)))?; + // Check if resource already exists - let exists: bool = conn + let exists: bool = tx .query_row( "SELECT 1 FROM resources WHERE tenant_id = ?1 AND resource_type = ?2 AND id = ?3", params![tenant_id, resource_type, id], @@ -330,27 +365,14 @@ impl ResourceStorage for SqliteBackend { })); } - // Ensure the resource has correct type and id - let mut resource = resource; - if let Some(obj) = resource.as_object_mut() { - obj.insert( - "resourceType".to_string(), - Value::String(resource_type.to_string()), - ); - obj.insert("id".to_string(), Value::String(id.clone())); - } - - // Serialize the resource data - let data = serde_json::to_vec(&resource) - .map_err(|e| serialization_error(format!("Failed to serialize resource: {}", e)))?; - + // Stamped with the write lock held, so a create queued behind another + // writer cannot commit with an earlier `last_updated` than one that + // committed before it (`_since`, `_lastUpdated`, bulk export). let now = Utc::now(); let last_updated = now.to_rfc3339(); - let version_id = "1"; - let fhir_version_str = fhir_version.as_mime_param(); // Insert the resource - conn.execute( + tx.execute( "INSERT INTO resources (tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7)", params![tenant_id, resource_type, id, version_id, data, last_updated, fhir_version_str], @@ -358,7 +380,7 @@ impl ResourceStorage for SqliteBackend { .map_err(|e| internal_error(format!("Failed to insert resource: {}", e)))?; // Insert into history - conn.execute( + tx.execute( "INSERT INTO resource_history (tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7)", params![tenant_id, resource_type, id, version_id, data, last_updated, fhir_version_str], @@ -366,7 +388,12 @@ impl ResourceStorage for SqliteBackend { .map_err(|e| internal_error(format!("Failed to insert history: {}", e)))?; // Index the resource for search - self.index_resource(&conn, tenant_id, resource_type, &id, &resource)?; + if let Some(prepared) = prepared { + self.write_prepared_index(&tx, tenant_id, resource_type, &id, &resource, prepared)?; + } + + tx.commit() + .map_err(|e| internal_error(format!("Failed to commit create: {}", e)))?; // An overlay-affecting SearchParameter write changes this tenant's // registry — drop the cached copy so the next access rebuilds from @@ -1505,10 +1532,17 @@ impl SqliteBackend { ) -> StorageResult { tenant.check_permission(Operation::Update, resource_type)?; - let conn = self.get_connection()?; + let mut conn = self.get_connection()?; let tenant_id = tenant.tenant_id().as_str(); - let row: Result<(String, String), _> = conn.query_row( + // One IMMEDIATE transaction, as update and delete: the restored row, + // its history entry and its rebuilt search index land together or not + // at all. + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|e| internal_error(format!("Failed to begin restore: {}", e)))?; + + let row: Result<(String, String), _> = tx.query_row( "SELECT version_id, fhir_version FROM resources WHERE tenant_id = ?1 AND resource_type = ?2 AND id = ?3 AND is_deleted = 1", params![tenant_id, resource_type, id], @@ -1550,7 +1584,7 @@ impl SqliteBackend { let now = Utc::now(); let last_updated = now.to_rfc3339(); - conn.execute( + tx.execute( "UPDATE resources SET version_id = ?1, data = ?2, last_updated = ?3, is_deleted = 0, deleted_at = NULL WHERE tenant_id = ?4 AND resource_type = ?5 AND id = ?6", @@ -1565,7 +1599,7 @@ impl SqliteBackend { ) .map_err(|e| internal_error(format!("Failed to restore resource: {}", e)))?; - conn.execute( + tx.execute( "INSERT INTO resource_history (tenant_id, resource_type, id, version_id, data, last_updated, is_deleted, fhir_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7)", params![tenant_id, resource_type, id, new_version_str, data, last_updated, fhir_version_str], @@ -1574,8 +1608,11 @@ impl SqliteBackend { // The delete dropped the search index entries; rebuild them for the // resource that is live again. - self.delete_search_index(&conn, tenant_id, resource_type, id)?; - self.index_resource(&conn, tenant_id, resource_type, id, &resource)?; + self.delete_search_index(&tx, tenant_id, resource_type, id)?; + self.index_resource(&tx, tenant_id, resource_type, id, &resource)?; + + tx.commit() + .map_err(|e| internal_error(format!("Failed to commit restore: {}", e)))?; // A restored overlay-affecting SearchParameter re-enters this tenant's // overlay. @@ -1816,33 +1853,24 @@ impl SqliteBackend { } match prepared.rows { Ok(rows) => { - // Rows are written eight at a time: a single-row INSERT stepped - // once per row spends a measurable share of its time entering - // and leaving the statement, and every resource writes ~10-30 - // rows. The B-tree work is unchanged; only the per-statement - // overhead is amortized (measured +9% bulk-ingest throughput - // on the real 31 GB manifest). + // One multi-row INSERT carries the resource's whole index: a + // single-row INSERT stepped once per row spends a measurable + // share of its time entering and leaving the statement, and + // every resource writes ~10-30 rows (eight at a time left up + // to seven single-row steps behind). The rows, their order + // and the B-tree work are unchanged; only the per-statement + // overhead is amortized. let _span = crate::perf::span(crate::perf::Phase::IndexInsert); - let mut i = 0; - while i + 8 <= rows.len() { - let refs: Vec<&dyn ToSql> = rows[i..i + 8] + for chunk in rows.chunks(SqliteSearchIndexWriter::MAX_ROWS_PER_INSERT) { + let refs: Vec<&dyn ToSql> = chunk .iter() .flatten() .map(|p| self.sql_value_to_ref(p)) .collect(); - conn.prepare_cached(SqliteSearchIndexWriter::insert_sql_rows8()) + conn.prepare_cached(SqliteSearchIndexWriter::insert_sql_rows(chunk.len())) .and_then(|mut s| s.execute(refs.as_slice())) .map_err(|e| internal_error(format!("multi-row index insert: {e}")))?; - i += 8; - count += 8; - } - for row in &rows[i..] { - let refs: Vec<&dyn ToSql> = - row.iter().map(|p| self.sql_value_to_ref(p)).collect(); - conn.prepare_cached(SqliteSearchIndexWriter::insert_sql()) - .and_then(|mut s| s.execute(refs.as_slice())) - .map_err(|e| internal_error(format!("index insert: {e}")))?; - count += 1; + count += chunk.len(); } for row in &prepared.contained_rows { let refs: Vec<&dyn ToSql> = @@ -4982,6 +5010,138 @@ mod tests { ) } + /// A create is one transaction: when the search index cannot be written, + /// neither the resource nor its history row survives (a create used to + /// auto-commit each statement and left a resource with no index). The same + /// holds for the restore of a deleted resource. + #[tokio::test] + async fn create_and_restore_roll_back_when_indexing_fails() { + let backend = create_test_backend(); + let tenant = create_test_tenant(); + let patient = + |id: &str| json!({"resourceType": "Patient", "id": id, "name": [{"family": "Smith"}]}); + + backend + .create(&tenant, "Patient", patient("gone"), FhirVersion::default()) + .await + .unwrap(); + backend.delete(&tenant, "Patient", "gone").await.unwrap(); + let counts = |conn: &rusqlite::Connection, id: &str| -> (i64, i64, i64, String) { + conn.query_row( + "SELECT (SELECT count(*) FROM resources WHERE id = ?1), \ + (SELECT count(*) FROM resource_history WHERE id = ?1), \ + (SELECT count(*) FROM search_index WHERE resource_id = ?1), \ + COALESCE((SELECT version_id || '/' || is_deleted FROM resources WHERE id = ?1), '-')", + [id], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)), + ) + .unwrap() + }; + { + let conn = backend.get_connection().unwrap(); + conn.execute_batch( + "CREATE TRIGGER fail_index BEFORE INSERT ON search_index \ + BEGIN SELECT RAISE(ABORT, 'index write failed'); END;", + ) + .unwrap(); + } + let created = backend + .create(&tenant, "Patient", patient("new"), FhirVersion::default()) + .await; + assert!(created.is_err()); + let restored = backend + .create_or_update( + &tenant, + "Patient", + "gone", + patient("gone"), + FhirVersion::default(), + ) + .await; + assert!(restored.is_err()); + let conn = backend.get_connection().unwrap(); + assert_eq!(counts(&conn, "new"), (0, 0, 0, "-".to_string())); + // The deleted resource is still deleted, at version 2, with its history. + assert_eq!(counts(&conn, "gone"), (1, 2, 0, "2/1".to_string())); + + conn.execute_batch("DROP TRIGGER fail_index").unwrap(); + drop(conn); + backend + .create(&tenant, "Patient", patient("new"), FhirVersion::default()) + .await + .unwrap(); + backend + .create_or_update( + &tenant, + "Patient", + "gone", + patient("gone"), + FhirVersion::default(), + ) + .await + .unwrap(); + let conn = backend.get_connection().unwrap(); + assert_eq!(counts(&conn, "new").0, 1); + assert!(counts(&conn, "new").2 > 0); + assert_eq!(counts(&conn, "gone").3, "3/0"); + assert!(counts(&conn, "gone").2 > 0); + } + + /// A create commits once: the statements of one create are a single + /// `BEGIN IMMEDIATE … COMMIT`, not one commit per statement. + #[tokio::test] + async fn create_commits_once() { + thread_local! { + static STMTS: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; + } + fn record(event: rusqlite::trace::TraceEvent<'_>) { + if let rusqlite::trace::TraceEvent::Stmt(statement, _) = event + && let Some(sql) = statement.expanded_sql() + { + STMTS.with(|s| s.borrow_mut().push(sql)); + } + } + let backend = create_test_backend(); + let tenant = create_test_tenant(); + let conn = backend.get_connection().unwrap(); + conn.trace_v2( + rusqlite::trace::TraceEventCodes::SQLITE_TRACE_STMT, + Some(record), + ); + drop(conn); + // The in-memory pool may hand the same connection back; if not, the + // trace is on another and the assertion below says so. + backend + .create( + &tenant, + "Patient", + json!({"resourceType": "Patient", "id": "one", "name": [{"family": "Smith"}]}), + FhirVersion::default(), + ) + .await + .unwrap(); + // FTS5 reports its own internal statements (`-- …`), some of them while + // the COMMIT runs; only the application's statements count here. + let log: Vec = STMTS + .with(|s| s.borrow().clone()) + .into_iter() + .filter(|s| !s.starts_with("--")) + .collect(); + let begins = log.iter().filter(|s| s.starts_with("BEGIN")).count(); + let commits = log.iter().filter(|s| s.starts_with("COMMIT")).count(); + assert!(!log.is_empty(), "no statements traced"); + assert_eq!((begins, commits), (1, 1), "{log:?}"); + assert!( + log.first() + .is_some_and(|s| s.starts_with("BEGIN IMMEDIATE")), + "{log:?}" + ); + assert!( + log.last().is_some_and(|s| s.starts_with("COMMIT")), + "{log:?}" + ); + } + /// #1078: SQLite's `count_all_types` / `count_deltas_by_bucket` are real /// aggregates, so it opts in to `supports_type_counts`. #[test] @@ -5219,7 +5379,8 @@ mod tests { /// #2107: the dashboard's bucketed history counts read only the window, /// from `idx_history_updated` alone — never a table row per version, and - /// never every version of a type through `idx_history_resource`. + /// never every version of a type through `idx_history_resource` (databases + /// below schema v45 still have it). #[test] fn test_count_deltas_query_plans_are_covered_by_the_history_index() { let backend = create_test_backend();