From a865cb918456d8b458a19c06ceac993ee0b286ad Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:04:24 -0800 Subject: [PATCH 1/7] lore: prune old versions of the lore tables after each lore run compact_lore_tables() skipped the lore table entirely and never touched lore_vectors, so neither ever dropped an old version. Every hourly refresh rewrites the five lore FTS indices in full (the body index alone is over 800 MB) and appends dozens of fragments, and all of it stayed on disk: on one host lore grew from 3.9 GB to 20 GB in two days, 17 GB of it superseded FTS indices. It also wedged FTS updates. optimize_lore_fts_indices() refuses to run once lore.lance/_indices holds more than 100 entries, and with nothing pruned the count only grew, by five per run. After about 14 hourly runs the indices stopped covering new mail for good. Prune old versions of lore from compact_lore_tables(), which only the lore run calls. Its Compact and index optimization phases, which the existing comment blames for running out of memory, stay skipped; prune reads only each manifest's index metadata. optimize_database() still skips lore: with delete_unverified set, a prune from another process could delete files a running lore refresh has not committed. Add lore_vectors to the tables compact_lore_tables() handles. Put the prune in one helper, prune_old_versions(), used by compact_lore_tables(), optimize_database() and create_lore_fts_indices(), which carried its own copy of the same call to clean up after rebuilding the FTS indices. The prune runs before the run's FTS update, so one superseded set of FTS indices remains on disk until the next run. It deletes their files but leaves their directories behind, so count only non-empty entries in the _indices guard. --- src/database/schema.rs | 102 ++++++++++++++++++++++++++++------------- 1 file changed, 71 insertions(+), 31 deletions(-) diff --git a/src/database/schema.rs b/src/database/schema.rs index 56f0a58..b85edf1 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -1721,18 +1721,7 @@ impl SchemaManager { // Prune orphaned index data left behind by drop_index(). if dropped { tracing::info!("Pruning orphaned index data from lore table..."); - if let Err(e) = table - .optimize(OptimizeAction::Prune { - older_than: Some( - lancedb::table::Duration::try_seconds(0).expect("valid duration"), - ), - delete_unverified: Some(true), - error_if_tagged_old_versions: Some(false), - }) - .await - { - tracing::warn!("Failed to prune lore table after FTS rebuild: {}", e); - } + Self::prune_old_versions(&table, "lore").await; } Ok(()) @@ -1796,13 +1785,25 @@ impl SchemaManager { // drives semcode-index into swap and gets it OOM-killed. // Query correctness is preserved regardless: unindexed rows // still fall back to a brute-force scan. + // + // Count only non-empty entries: pruning old versions deletes a + // superseded index's files but leaves its directory behind. + // Those empty directories accumulate, five per lore run; they + // cost a directory entry each and nothing else. const MAX_LORE_INDEX_FRAGMENTS: usize = 100; let uri = self.connection.uri(); let indices_dir = std::path::Path::new(uri) .join("lore.lance") .join("_indices"); if let Ok(rd) = std::fs::read_dir(&indices_dir) { - let count = rd.count(); + let count = rd + .filter_map(|entry| entry.ok()) + .filter(|entry| { + std::fs::read_dir(entry.path()) + .map(|mut files| files.next().is_some()) + .unwrap_or(true) + }) + .count(); if count > MAX_LORE_INDEX_FRAGMENTS { tracing::warn!( "Skipping lore FTS index optimization: \ @@ -2050,19 +2051,24 @@ impl SchemaManager { /// the database, including code-index tables and content shards /// that a lore run never touches. On memory-constrained systems /// the combined working set of those compactions triggers the OOM - /// killer. This method limits work to the two lore tables and + /// killer. This method limits work to the lore tables and /// processes them sequentially to keep peak memory low. pub async fn compact_lore_tables(&self) -> Result<()> { tracing::info!("Running compaction for lore tables..."); let table_names = self.connection.table_names().execute().await?; - let lore_tables = ["lore", "lore_indexed_commits"]; + let lore_tables = ["lore", "lore_indexed_commits", "lore_vectors"]; for name in &lore_tables { if !table_names.iter().any(|n| n == name) { continue; } - match Self::optimize_single_table(&self.connection, name).await { + let outcome = if *name == "lore" { + Self::prune_lore_table(&self.connection).await + } else { + Self::optimize_single_table(&self.connection, name).await + }; + match outcome { Ok(OptimizeOutcome::Optimized) => { tracing::info!("Compacted table {}", name); } @@ -2081,6 +2087,51 @@ impl SchemaManager { Ok(()) } + /// Delete every version of `table` but the latest, along with the + /// data and index files only those versions referenced. Returns + /// false if the prune failed. + async fn prune_old_versions(table: &lancedb::table::Table, table_name: &str) -> bool { + match table + .optimize(OptimizeAction::Prune { + older_than: Some(lancedb::table::Duration::try_seconds(0).expect("valid duration")), + delete_unverified: Some(true), + error_if_tagged_old_versions: Some(false), + }) + .await + { + Ok(_) => true, + Err(e) => { + tracing::warn!( + "Failed to prune old versions from table {}: {}", + table_name, + e + ); + false + } + } + } + + /// Prune old versions of the lore table, which optimize_single_table() + /// skips entirely. Every lore run rewrites the FTS indices, and + /// unless old versions are pruned the superseded copies stay under + /// _indices/ until the count trips the guard in + /// optimize_lore_fts_indices() and FTS updates stop for good. Prune + /// reads only each manifest's index metadata, not the index files, + /// so it avoids the memory cost of Compact. + /// + /// Only the lore run calls this, not optimize_database(): with + /// delete_unverified set, a prune can delete files that another + /// process writing to lore has not committed yet. That also relies + /// on lore runs not overlapping, e.g. as a systemd oneshot unit ensures. + async fn prune_lore_table(connection: &Connection) -> Result { + let table = connection.open_table("lore").execute().await?; + Ok(if Self::prune_old_versions(&table, "lore").await { + OptimizeOutcome::Optimized + } else { + OptimizeOutcome::PartialFailure + }) + } + /// Optimize a single table - runs compact, prune, and index operations /// /// Tables with fewer than 1000 rows are skipped since the overhead of @@ -2098,11 +2149,12 @@ impl SchemaManager { // the operation completes. On memory-constrained systems the // resident set grows into swap and the OOM killer terminates // semcode-index before compaction finishes, leaving fresh - // delta fragments behind each time. Skip the table entirely. + // delta fragments behind each time. Skip the table entirely; + // compact_lore_tables() prunes its old versions. if table_name == "lore" { tracing::info!( "Skipping generic optimization for lore table \ - (handled by optimize_lore_fts_indices)" + (pruned by compact_lore_tables)" ); return Ok(OptimizeOutcome::Skipped); } @@ -2165,19 +2217,7 @@ impl SchemaManager { } // 2. Prune ALL old versions - if let Err(e) = table - .optimize(OptimizeAction::Prune { - older_than: Some(lancedb::table::Duration::try_seconds(0).expect("valid duration")), - delete_unverified: Some(true), - error_if_tagged_old_versions: Some(false), - }) - .await - { - tracing::warn!( - "Failed to prune old versions from table {}: {}", - table_name, - e - ); + if !Self::prune_old_versions(&table, table_name).await { success = false; } From 79d705e41321466a602bd4d00981d16f027adb99 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:04:24 -0800 Subject: [PATCH 2/7] lore: merge partial batches before inserting them Each lore parser thread sends a batch once it holds batch_size emails, and flushes whatever it has left when the commit list runs out. On an hourly refresh that adds a few hundred emails per archive, every thread ends up with a short batch, and each of them became its own append to lore and lore_indexed_commits: one host's refreshes added 40 to 80 fragments to each table per run, and the lore_vectors step then made one merge_insert per fragment. A parser only sends a short batch when its input has run out, so when an inserter receives one, keep receiving until it holds at least batch_size emails or every parser has finished. Full batches received on their own are inserted as before, and a merged batch stays under twice batch_size. In a simulated hourly refresh of 50 new emails with 32 parser threads, lore gains one fragment per run instead of 32. --- src/indexer.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/src/indexer.rs b/src/indexer.rs index 2752ea1..1573023 100644 --- a/src/indexer.rs +++ b/src/indexer.rs @@ -957,10 +957,24 @@ pub async fn process_lore_commits_pipeline( let handle = tokio::spawn(async move { loop { - // Get next batch from shared receiver + // Get next batch from shared receiver. A parser only + // sends a short batch when its input has run out, so + // keep receiving until the batch holds at least + // batch_size emails or every parser has finished. + // Otherwise each parser's last partial batch becomes + // its own small append, dozens of tiny fragments per + // archive on a refresh. let batch = { match result_rx_clone.lock() { - Ok(rx) => rx.recv(), + Ok(rx) => rx.recv().map(|mut emails| { + while emails.len() < batch_size { + match rx.recv() { + Ok(more) => emails.extend(more), + Err(_) => break, + } + } + emails + }), Err(e) => { error!("Inserter {} failed to lock receiver: {}", inserter_id, e); break; From 338aa4731fab44e2bbda09b065d6a4ae440baa10 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:04:25 -0800 Subject: [PATCH 3/7] lore: buffer vectors before merging them into lore_vectors update_lore_vectors() workers send one Vec per lore RecordBatch that still has emails without a vector, and the inserter ran a merge_insert into lore_vectors for each of them. On an incremental run most of those hold only a few vectors, while every merge_insert joins against the whole table: one host's day of new mail (6,228 emails) took about 1,600 merge_inserts and 56 minutes, nearly all of it in those joins, and each hourly run added 40 to 80 fragments to lore_vectors. Buffer vectors in the inserter and write once at least 8192 have accumulated, flushing the remainder when the workers finish. The merge semantics are unchanged. merge_insert inserts every unmatched source row, so keep one entry per message_id in a buffer, as separate merges used to. An hourly-sized run now adds one fragment. The progress bar still advances on every Vec the inserter receives, not at each write, so on an incremental run it does not sit at zero until the final write and then jump to the total. Its count can now run ahead of what has been written, so its message reads "Vectorized N emails" rather than "Inserted N vectors". --- src/database/search.rs | 39 ++++++++++++++++++++++++++++++++------- 1 file changed, 32 insertions(+), 7 deletions(-) diff --git a/src/database/search.rs b/src/database/search.rs index 56458cf..b958ee8 100644 --- a/src/database/search.rs +++ b/src/database/search.rs @@ -2950,20 +2950,45 @@ impl VectorSearchManager { .execute() .await?; + // Workers send one Vec per lore RecordBatch that still has + // emails without a vector. On an incremental run most of + // those hold only a few, and every merge_insert joins + // against the whole table, so write once at least + // LORE_VECTOR_FLUSH_ROWS have accumulated. + const LORE_VECTOR_FLUSH_ROWS: usize = 8192; + let mut pending: Vec = Vec::new(); let mut total_inserted = 0; - while let Some(batch_vectors) = result_rx.recv().await { + let mut open = true; + while open { + match result_rx.recv().await { + Some(batch_vectors) => { + let received = batch_vectors.len(); + let count = + processed_clone.fetch_add(received, Ordering::Relaxed) + received; + pb_clone.set_position(count as u64); + pb_clone.set_message(format!("Vectorized {} emails", count)); + pending.extend(batch_vectors); + } + None => open = false, + } + if pending.is_empty() || (open && pending.len() < LORE_VECTOR_FLUSH_ROWS) { + continue; + } + + // merge_insert inserts every unmatched source row, so two + // copies of a message_id in one buffer would become two + // rows. Keep one. + let batch = crate::database::one_row_per_key(std::mem::take(&mut pending), |e| { + e.content_hash.clone() + }); if let Err(e) = - insert_lore_vectors_batch_with_table(&lore_vectors_table, &batch_vectors).await + insert_lore_vectors_batch_with_table(&lore_vectors_table, &batch).await { tracing::error!("Failed to insert lore vectors batch: {}", e); return Err(e); } - total_inserted += batch_vectors.len(); - let count = processed_clone.fetch_add(batch_vectors.len(), Ordering::Relaxed) - + batch_vectors.len(); - pb_clone.set_position(count as u64); - pb_clone.set_message(format!("Inserted {} vectors", count)); + total_inserted += batch.len(); } Ok::<_, anyhow::Error>(total_inserted) }); From b0cdd86b29f100efab5359fdec98a1059e7017e8 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:48:39 -0800 Subject: [PATCH 4/7] lore: compact lore_indexed_commits whatever its fragment count A cold lore build appends one lore_indexed_commits fragment per inserted batch: 1,013 for a production-sized set of archives, 3,660 for an upstream build. optimize_single_table() skips compaction above 500 fragments and nothing else brings the count down, so the table stays over the limit for good. Rebuilding with --clear, as the skip message advises, recreates the table over the limit. Exempt this table from the limit and keep the limit for every other table. Databases already over it compact on their next lore run that adds mail. The limit guards against a lance compaction CPU spin seen with thousands of small fragments (61e5493). lore_indexed_commits holds one 40-char SHA per row and no index. On lance 4.0.0 a table of that shape compacts in under 5 s at both 1,013 and 3,660 fragments, with no spin, whether compacted alone, as compact_lore_tables() does, or eight tables at a time, as optimize_database() does. The spin does not reproduce on lance 1.0.1 either, the version 61e5493 saw it on, so its cause is still unknown. --- Cargo.lock | 1 + Cargo.toml | 5 +++ src/database/schema.rs | 93 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 97 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bfd6219..1087441 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6263,6 +6263,7 @@ dependencies = [ "gix-pack", "hex", "indicatif 0.18.6", + "lance", "lancedb", "model2vec-rs", "ndarray 0.17.2", diff --git a/Cargo.toml b/Cargo.toml index 32ed797..5346340 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -66,6 +66,11 @@ chrono = "0.4.44" chrono-english = "0.1" # Natural language date parsing base64 = "0.22" # Base64 decoding for MIME-encoded email bodies +[dev-dependencies] +# Tests reach lance's WriteParams, which lancedb does not re-export. Pin +# the exact version lancedb resolves to, so the two cannot drift apart. +lance = { version = "=4.0.0", default-features = false } + [[bin]] name = "semcode-index" path = "src/bin/index.rs" diff --git a/src/database/schema.rs b/src/database/schema.rs index b85edf1..0c63861 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -2181,11 +2181,22 @@ impl SchemaManager { let mut success = true; - // 1. Compact files + // 1. Compact files. Tables above MAX_COMPACT_FRAGMENTS are left + // alone, as a guard against lance compaction spinning on thousands + // of small fragments. lore_indexed_commits is exempt: it holds one + // 40-char SHA per row and no index, and a table of that shape + // compacts in seconds at 3,660 fragments on lance 4.0.0. A cold + // lore build leaves it with one fragment per inserted batch, over a + // thousand for a full archive set, so the guard would keep it + // uncompacted. const MAX_COMPACT_FRAGMENTS: usize = 500; + let fragment_limit_applies = table_name != "lore_indexed_commits"; let should_compact = match table.stats().await { - Ok(stats) if stats.fragment_stats.num_fragments > MAX_COMPACT_FRAGMENTS => { + Ok(stats) + if fragment_limit_applies + && stats.fragment_stats.num_fragments > MAX_COMPACT_FRAGMENTS => + { tracing::warn!( "Skipping compaction for table {} \ ({} fragments exceeds {} limit -- \ @@ -2521,3 +2532,81 @@ impl SchemaManager { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::StringArray; + use lancedb::connection::LanceFileVersion; + use lancedb::database::listing::ListingDatabaseOptions; + + async fn fresh_schema_manager() -> (tempfile::TempDir, SchemaManager) { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("schema.lancedb"); + let options = ListingDatabaseOptions::builder() + .data_storage_version(LanceFileVersion::V2_2) + .build(); + let connection = lancedb::connect(db_path.to_str().unwrap()) + .database_options(&options) + .execute() + .await + .unwrap(); + (dir, SchemaManager::new(connection)) + } + + /// A cold lore build appends one lore_indexed_commits fragment per + /// batch. 600 two-row fragments put the table above + /// MAX_COMPACT_FRAGMENTS and above MIN_ROWS_FOR_OPTIMIZATION, so the + /// lore compaction has to compact it rather than skip it. One write + /// capped at two rows per file builds them all in a single commit. + #[tokio::test] + async fn lore_indexed_commits_compacts_above_the_fragment_limit() { + const FRAGMENTS: usize = 600; + const ROWS_PER_FRAGMENT: usize = 2; + + let (_dir, schema_manager) = fresh_schema_manager().await; + schema_manager + .create_lore_indexed_commits_table() + .await + .unwrap(); + let table = schema_manager + .connection + .open_table("lore_indexed_commits") + .execute() + .await + .unwrap(); + let shas: Vec = (0..FRAGMENTS * ROWS_PER_FRAGMENT) + .map(|row| format!("{row:040x}")) + .collect(); + let batch = RecordBatch::try_new( + table.schema().await.unwrap(), + vec![Arc::new(StringArray::from(shas))], + ) + .unwrap(); + table + .add(vec![batch]) + .write_options(lancedb::table::WriteOptions { + lance_write_params: Some(lance::dataset::WriteParams { + max_rows_per_file: ROWS_PER_FRAGMENT, + max_rows_per_group: ROWS_PER_FRAGMENT, + mode: lance::dataset::WriteMode::Append, + ..Default::default() + }), + }) + .execute() + .await + .unwrap(); + let before = table.stats().await.unwrap().fragment_stats.num_fragments; + assert_eq!(before, FRAGMENTS); + + schema_manager.compact_lore_tables().await.unwrap(); + + table.checkout_latest().await.unwrap(); + let after = table.stats().await.unwrap().fragment_stats.num_fragments; + assert!(after < 10, "{after} fragments left after compaction"); + assert_eq!( + table.count_rows(None).await.unwrap(), + FRAGMENTS * ROWS_PER_FRAGMENT + ); + } +} From 4d66a4526f91337eafb22008935686fa997ca279 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:04:25 -0800 Subject: [PATCH 5/7] lore: report skipped FTS updates and compaction problems semcode-index printed "Lore table compaction completed successfully" and "FTS indices updated successfully" whatever had happened. The reasons for skipping or failing went to tracing::warn!, which the default SEMCODE_DEBUG level filters out, so the lore FTS indices stopped covering new mail for two days on one host without a line in the hourly unit's log. Print the FTS skip reasons and lore compaction problems to stderr. optimize_lore_fts_indices() and compact_lore_tables() now return whether they did their work, and the success lines are printed only when they did. optimize_single_table() also reports a table it did not compact because it has more than 500 fragments as a partial failure rather than success, since its fragment count only grows from there. --- src/bin/index.rs | 12 ++++++---- src/database/connection.rs | 6 +++-- src/database/schema.rs | 45 ++++++++++++++++++++++++++------------ 3 files changed, 43 insertions(+), 20 deletions(-) diff --git a/src/bin/index.rs b/src/bin/index.rs index aaa1fa3..efc33bb 100644 --- a/src/bin/index.rs +++ b/src/bin/index.rs @@ -1202,7 +1202,8 @@ async fn main() -> Result<()> { // combined working set triggers the OOM killer. println!("\nCompacting lore tables..."); match db_manager.compact_lore_tables().await { - Ok(_) => println!("Lore table compaction completed successfully"), + Ok(true) => println!("Lore table compaction completed successfully"), + Ok(false) => {} Err(e) => error!("Failed to compact lore tables: {}", e), } @@ -1216,7 +1217,8 @@ async fn main() -> Result<()> { Err(e) => eprintln!("Warning: Failed to ensure FTS indices: {}", e), } match db_manager.optimize_lore_fts_indices().await { - Ok(_) => println!("FTS indices updated successfully"), + Ok(true) => println!("FTS indices updated successfully"), + Ok(false) => {} Err(e) => eprintln!("Warning: Failed to optimize FTS indices: {}", e), } } @@ -1376,7 +1378,8 @@ async fn main() -> Result<()> { if total_new_emails > 0 { println!("\nCompacting lore tables..."); match db_manager.compact_lore_tables().await { - Ok(_) => println!("Lore table compaction completed successfully"), + Ok(true) => println!("Lore table compaction completed successfully"), + Ok(false) => {} Err(e) => error!("Failed to compact lore tables: {}", e), } @@ -1386,7 +1389,8 @@ async fn main() -> Result<()> { Err(e) => eprintln!("Warning: Failed to ensure FTS indices: {}", e), } match db_manager.optimize_lore_fts_indices().await { - Ok(_) => println!("FTS indices updated successfully"), + Ok(true) => println!("FTS indices updated successfully"), + Ok(false) => {} Err(e) => eprintln!("Warning: Failed to optimize FTS indices: {}", e), } } diff --git a/src/database/connection.rs b/src/database/connection.rs index 22d8d48..47e51c7 100644 --- a/src/database/connection.rs +++ b/src/database/connection.rs @@ -889,7 +889,8 @@ impl DatabaseManager { self.schema_manager.compact_and_cleanup().await } - pub async fn compact_lore_tables(&self) -> Result<()> { + /// Returns false, after printing the problem, if any table failed. + pub async fn compact_lore_tables(&self) -> Result { self.schema_manager.compact_lore_tables().await } @@ -909,7 +910,8 @@ impl DatabaseManager { } /// Merge newly-inserted rows into existing lore FTS indices. - pub async fn optimize_lore_fts_indices(&self) -> Result<()> { + /// Returns false, after printing why, if the merge was skipped. + pub async fn optimize_lore_fts_indices(&self) -> Result { self.schema_manager.optimize_lore_fts_indices().await } diff --git a/src/database/schema.rs b/src/database/schema.rs index 0c63861..6bfc32e 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -1776,7 +1776,9 @@ impl SchemaManager { /// brute-force fallback at query time, so queries remain correct /// even before this call. Running optimize merges those rows /// into the inverted index structure, eliminating the scan cost. - pub async fn optimize_lore_fts_indices(&self) -> Result<()> { + /// + /// Returns false, after printing why, if it skipped the merge. + pub async fn optimize_lore_fts_indices(&self) -> Result { // Guard against running on a table with a large _indices/ // backlog. lance/index/append.rs opens every delta fragment // for a column before merging any, so peak memory scales @@ -1805,15 +1807,14 @@ impl SchemaManager { }) .count(); if count > MAX_LORE_INDEX_FRAGMENTS { - tracing::warn!( - "Skipping lore FTS index optimization: \ + eprintln!( + "Warning: Skipping lore FTS index optimization: \ {} _indices/ fragments exceeds {} threshold. \ Queries remain correct via brute-force fallback. \ Rebuild the lore table on a larger host to recover.", - count, - MAX_LORE_INDEX_FRAGMENTS + count, MAX_LORE_INDEX_FRAGMENTS ); - return Ok(()); + return Ok(false); } } @@ -1829,8 +1830,8 @@ impl SchemaManager { .collect(); if fts_index_names.is_empty() { - tracing::warn!("Skipping lore FTS optimization: no FTS indices found"); - return Ok(()); + eprintln!("Warning: Skipping lore FTS index optimization: no FTS indices found"); + return Ok(false); } let start_time = std::time::Instant::now(); @@ -1850,7 +1851,7 @@ impl SchemaManager { "Lore FTS index optimization completed in {:.1}s", elapsed.as_secs_f64() ); - Ok(()) + Ok(true) } /// Shared helper: create all 5 FTS indices on an already-opened table. @@ -2053,12 +2054,19 @@ impl SchemaManager { /// the combined working set of those compactions triggers the OOM /// killer. This method limits work to the lore tables and /// processes them sequentially to keep peak memory low. - pub async fn compact_lore_tables(&self) -> Result<()> { + /// + /// Returns false, after printing the problem, if any table failed. + pub async fn compact_lore_tables(&self) -> Result { tracing::info!("Running compaction for lore tables..."); let table_names = self.connection.table_names().execute().await?; let lore_tables = ["lore", "lore_indexed_commits", "lore_vectors"]; + // Print problems to stderr: the tracing warnings carrying the + // details are filtered out below SEMCODE_DEBUG=warn, and a prune + // that fails silently lets _indices/ grow until lore FTS updates + // stop, as they did when lore was never pruned. + let mut all_ok = true; for name in &lore_tables { if !table_names.iter().any(|n| n == name) { continue; @@ -2076,15 +2084,21 @@ impl SchemaManager { tracing::info!("Skipped table {} (too small)", name); } Ok(OptimizeOutcome::PartialFailure) => { - tracing::warn!("Partial failure compacting table {}", name); + eprintln!( + "Warning: Partial failure compacting table {} \ + (set SEMCODE_DEBUG=warn for details)", + name + ); + all_ok = false; } Err(e) => { - tracing::warn!("Failed to compact table {}: {}", name, e); + eprintln!("Warning: Failed to compact table {}: {}", name, e); + all_ok = false; } } } - Ok(()) + Ok(all_ok) } /// Delete every version of `table` but the latest, along with the @@ -2205,6 +2219,9 @@ impl SchemaManager { stats.fragment_stats.num_fragments, MAX_COMPACT_FRAGMENTS ); + // Not compacting is not success: the fragment count only + // grows from here, and callers report the outcome. + success = false; false } Ok(_) => true, @@ -2599,7 +2616,7 @@ mod tests { let before = table.stats().await.unwrap().fragment_stats.num_fragments; assert_eq!(before, FRAGMENTS); - schema_manager.compact_lore_tables().await.unwrap(); + assert!(schema_manager.compact_lore_tables().await.unwrap()); table.checkout_latest().await.unwrap(); let after = table.stats().await.unwrap().fragment_stats.num_fragments; From f5a49a516235defa08662e297ea11b21f67fc0f0 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 08:33:27 -0800 Subject: [PATCH 6/7] lore: rebuild the non-nullable lore BTree indices each run The lore BTree indices are created at schema setup, usually on an empty table, and nothing updates them afterwards: `optimize_single_table()` skips lore, and `optimize_lore_fts_indices()` merges only the FTS indices. In a database built from scratch they cover no rows, so every lookup by `message_id` scans the whole table, and so does every `in_reply_to` lookup a thread walk makes. A long-lived database had about 106,000 of its 998,000 emails outside them, a number every refresh adds to. `compact_lore_tables()` now rebuilds, after pruning old versions, the BTree indices on non-nullable columns: `message_id`, `git_commit_sha` and `date`, plus `from` and `subject` until the first FTS build replaces those with FTS indices. It builds them from the whole table each run rather than merging only the new rows into them, so no earlier failed run, and no rewrite of the table's fragments, can leave rows out. The BTrees on the nullable `in_reply_to` and `references` columns stay as they are, because lance 4.0.0 writes a BTree over a nullable string column of that size into pages it cannot decode: reading the index back panics, and so does any lookup that lands on a bad page. `in_reply_to` lookups therefore still scan the rows its index does not cover. A comment at the filter names the defect, so a lance upgrade can lift it. A failed rebuild is reported as a partial failure, like a failed prune. The rebuild runs as its own task, so a panic inside lance is reported the same way instead of ending semcode-index before the FTS update. `prune_lore_table()` is renamed `maintain_lore_table()` to match. On a copy of a production database (about 998,000 rows, load average 100 to 170 on 316 CPUs), rebuilding the three BTrees took 1.6 to 1.8 s every run, about twice as long as merging new rows, and peaked at 265 MB on its own. A run that rebuilds them peaks at 1.7 GB, most of it from listing the table's indices, which opens the FTS indices that the FTS update opens anyway. The index copies a rebuild replaces stay under `_indices/` until the next run prunes them, adding one BTree set to the non-empty entries `optimize_lore_fts_indices()` counts against its limit of 100. --- src/database/schema.rs | 274 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 260 insertions(+), 14 deletions(-) diff --git a/src/database/schema.rs b/src/database/schema.rs index 6bfc32e..b3e4792 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -1790,8 +1790,11 @@ impl SchemaManager { // // Count only non-empty entries: pruning old versions deletes a // superseded index's files but leaves its directory behind. - // Those empty directories accumulate, five per lore run; they - // cost a directory entry each and nothing else. + // Those empty directories accumulate, one per FTS and BTree + // index each lore run; they cost a directory entry each and + // nothing else. What stays non-empty is the current set of + // indices plus the BTree indices that compact_lore_tables() + // superseded earlier in this run, which the next run prunes. const MAX_LORE_INDEX_FRAGMENTS: usize = 100; let uri = self.connection.uri(); let indices_dir = std::path::Path::new(uri) @@ -2072,7 +2075,7 @@ impl SchemaManager { continue; } let outcome = if *name == "lore" { - Self::prune_lore_table(&self.connection).await + Self::maintain_lore_table(&self.connection).await } else { Self::optimize_single_table(&self.connection, name).await }; @@ -2125,27 +2128,130 @@ impl SchemaManager { } } - /// Prune old versions of the lore table, which optimize_single_table() - /// skips entirely. Every lore run rewrites the FTS indices, and - /// unless old versions are pruned the superseded copies stay under - /// _indices/ until the count trips the guard in - /// optimize_lore_fts_indices() and FTS updates stop for good. Prune - /// reads only each manifest's index metadata, not the index files, - /// so it avoids the memory cost of Compact. + /// Prune old versions of the lore table and rebuild the BTree indices + /// on its non-nullable columns: optimize_single_table() skips lore + /// entirely. + /// + /// Every lore run rewrites the FTS indices, and unless old versions + /// are pruned the superseded copies stay under _indices/ until the + /// count trips the guard in optimize_lore_fts_indices() and FTS + /// updates stop for good. Prune reads only each manifest's index + /// metadata, not the index files, so it avoids the memory cost of + /// Compact. + /// + /// The BTree rebuild runs after the prune, so the BTree files it + /// supersedes stay until the next run prunes them. /// /// Only the lore run calls this, not optimize_database(): with /// delete_unverified set, a prune can delete files that another /// process writing to lore has not committed yet. That also relies /// on lore runs not overlapping, e.g. as a systemd oneshot unit ensures. - async fn prune_lore_table(connection: &Connection) -> Result { + async fn maintain_lore_table(connection: &Connection) -> Result { let table = connection.open_table("lore").execute().await?; - Ok(if Self::prune_old_versions(&table, "lore").await { + let pruned = Self::prune_old_versions(&table, "lore").await; + let btrees_current = Self::rebuild_lore_btree_indices(&table).await; + Ok(if pruned && btrees_current { OptimizeOutcome::Optimized } else { OptimizeOutcome::PartialFailure }) } + /// Run one maintenance step on the lore table and report whether it + /// succeeded, logging why not. The step runs as its own task: a lance + /// panic would otherwise end semcode-index before the FTS update, on + /// every run until the table is repaired by hand, and as a task it + /// becomes a JoinError. The next run's prune deletes the files an + /// aborted step wrote. + async fn run_lore_task(what: &str, step: F) -> bool + where + T: Send + 'static, + F: std::future::Future> + Send + 'static, + { + let start_time = std::time::Instant::now(); + match tokio::spawn(step).await { + Ok(Ok(_)) => { + tracing::info!( + "Lore {} completed in {:.1}s", + what, + start_time.elapsed().as_secs_f64() + ); + true + } + Ok(Err(e)) => { + tracing::warn!("Lore {} failed: {}", what, e); + false + } + Err(e) => { + tracing::warn!("Lore {} did not finish: {}", what, e); + false + } + } + } + + /// Build the lore table's BTree indices on non-nullable columns again + /// from the whole table, under their existing names. The BTrees are + /// created at schema setup, usually on an empty table, and a lookup by + /// message_id scans every row they do not cover. Returns false, after + /// logging why, on failure; a table without such indices has nothing + /// to rebuild. + /// + /// A rebuild rather than lance's in-place update, which merges only + /// new rows: a rebuild depends on nothing but the table as it is now, + /// so no earlier failed run or rewrite of the table's fragments can + /// leave rows out of the result. + /// + /// BTrees on nullable columns (in_reply_to, references) are left as + /// they are: lance 4.0.0 writes a BTree over a nullable string column + /// with enough rows into pages it cannot decode. Reading the index + /// back panics in lance-encoding, and so does any lookup that lands on + /// a bad page. Rebuilding them would replace a stale but readable + /// index with an unreadable one. Lift this once semcode moves to a + /// lance whose writer reads back its own output. + async fn rebuild_lore_btree_indices(table: &lancedb::table::Table) -> bool { + use lancedb::index::IndexType; + + let (indices, schema) = match tokio::try_join!(table.list_indices(), table.schema()) { + Ok(found) => found, + Err(e) => { + tracing::warn!("Failed to read indices of table lore: {}", e); + return false; + } + }; + let non_nullable = |column: &String| { + schema + .field_with_name(column) + .is_ok_and(|field| !field.is_nullable()) + }; + let btrees: Vec = indices + .into_iter() + .filter(|index| { + index.index_type == IndexType::BTree && index.columns.iter().all(non_nullable) + }) + .collect(); + if btrees.is_empty() { + return true; + } + + tracing::info!("Rebuilding {} lore BTree indices...", btrees.len()); + let table = table.clone(); + Self::run_lore_task("BTree index rebuild", async move { + for index in btrees { + table + .create_index( + &index.columns, + LanceIndex::BTree(BTreeIndexBuilder::default()), + ) + .name(index.name) + .replace(true) + .execute() + .await?; + } + Ok(()) + }) + .await + } + /// Optimize a single table - runs compact, prune, and index operations /// /// Tables with fewer than 1000 rows are skipped since the overhead of @@ -2164,11 +2270,12 @@ impl SchemaManager { // resident set grows into swap and the OOM killer terminates // semcode-index before compaction finishes, leaving fresh // delta fragments behind each time. Skip the table entirely; - // compact_lore_tables() prunes its old versions. + // compact_lore_tables() prunes its old versions and rebuilds the + // BTree indices on its non-nullable columns. if table_name == "lore" { tracing::info!( "Skipping generic optimization for lore table \ - (pruned by compact_lore_tables)" + (maintained by compact_lore_tables)" ); return Ok(OptimizeOutcome::Skipped); } @@ -2553,9 +2660,12 @@ impl SchemaManager { #[cfg(test)] mod tests { use super::*; + use crate::database::DatabaseManager; + use crate::types::LoreEmailInfo; use arrow::array::StringArray; use lancedb::connection::LanceFileVersion; use lancedb::database::listing::ListingDatabaseOptions; + use lancedb::index::IndexType; async fn fresh_schema_manager() -> (tempfile::TempDir, SchemaManager) { let dir = tempfile::tempdir().unwrap(); @@ -2626,4 +2736,140 @@ mod tests { FRAGMENTS * ROWS_PER_FRAGMENT ); } + + /// BTrees on non-nullable columns that every lore database carries: + /// each lore run brings them up to date. + const UPDATED_BTREES: [&str; 3] = ["message_id_idx", "git_commit_sha_idx", "date_idx"]; + + /// BTrees on nullable columns, which lance 4.0.0 cannot rewrite + /// readably: lore runs leave them as they were created, on the empty + /// table. + const NULLABLE_BTREES: [&str; 2] = ["in_reply_to_idx", "references_idx"]; + + async fn lore_database() -> (tempfile::TempDir, DatabaseManager) { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("lore-btree.db"); + let db = DatabaseManager::new(db_path.to_str().unwrap(), String::new()) + .await + .unwrap(); + db.create_tables().await.unwrap(); + (dir, db) + } + + /// Emails `first..first + count`, every second one a reply. + fn emails(first: usize, count: usize) -> Vec { + (first..first + count) + .map(|i| LoreEmailInfo { + git_commit_sha: format!("{i:040x}"), + from: format!("Sender {i} "), + date: "Tue, 10 Nov 2026 12:00:00 +0000".to_string(), + date_timestamp: 1_794_312_000 + i as i64, + message_id: format!(""), + in_reply_to: (i % 2 == 1).then(|| format!("", i - 1)), + subject: format!("[PATCH] change {i}"), + references: (i % 2 == 1).then(|| format!("", i - 1)), + recipients: "list@example.org".to_string(), + body: format!("body {i}"), + symbols: Vec::new(), + }) + .collect() + } + + /// Every BTree index on lore as (name, indexed rows, unindexed rows). + async fn btree_coverage(db: &DatabaseManager) -> Vec<(String, usize, usize)> { + let table = db.connection().open_table("lore").execute().await.unwrap(); + let mut coverage = Vec::new(); + for index in table.list_indices().await.unwrap() { + if index.index_type != IndexType::BTree { + continue; + } + let stats = table.index_stats(&index.name).await.unwrap().unwrap(); + coverage.push((index.name, stats.num_indexed_rows, stats.num_unindexed_rows)); + } + coverage.sort(); + coverage + } + + /// Asserts that every BTree on a non-nullable column covers all `rows` + /// and that the nullable-column BTrees still cover none. + fn assert_covers(coverage: &[(String, usize, usize)], rows: usize) { + for name in UPDATED_BTREES.iter().chain(&NULLABLE_BTREES) { + assert!( + coverage.iter().any(|(n, _, _)| n == name), + "{name} missing from {coverage:?}" + ); + } + for (name, indexed, unindexed) in coverage { + let expected = if NULLABLE_BTREES.contains(&name.as_str()) { + (0, rows) + } else { + (rows, 0) + }; + assert_eq!( + (*indexed, *unindexed), + expected, + "{name}: indexed and unindexed rows" + ); + } + } + + #[tokio::test] + async fn lore_btree_indices_cover_new_rows() { + let (_dir, db) = lore_database().await; + assert!(db + .insert_lore_emails(&emails(0, 50)) + .await + .unwrap() + .is_empty()); + + assert!(db.compact_lore_tables().await.unwrap()); + + assert_covers(&btree_coverage(&db).await, 50); + } + + /// A later run, with the FTS indices in place as every run after the + /// first has them: the BTrees take in the new rows and the FTS indices + /// are left as they were. + #[tokio::test] + async fn lore_btree_indices_stay_current_across_runs() { + let (_dir, db) = lore_database().await; + assert!(db + .insert_lore_emails(&emails(0, 50)) + .await + .unwrap() + .is_empty()); + assert!(db.compact_lore_tables().await.unwrap()); + db.ensure_lore_fts_indices().await.unwrap(); + assert!(db.optimize_lore_fts_indices().await.unwrap()); + + assert!(db + .insert_lore_emails(&emails(50, 30)) + .await + .unwrap() + .is_empty()); + assert!(db.compact_lore_tables().await.unwrap()); + + assert_covers(&btree_coverage(&db).await, 80); + let table = db.connection().open_table("lore").execute().await.unwrap(); + let mut fts: Vec<(String, usize)> = Vec::new(); + for index in table.list_indices().await.unwrap() { + if index.index_type == IndexType::FTS { + let stats = table.index_stats(&index.name).await.unwrap().unwrap(); + fts.push((index.name, stats.num_indexed_rows)); + } + } + fts.sort(); + assert_eq!( + fts, + [ + "body_idx", + "from_idx", + "recipients_idx", + "subject_idx", + "symbols_idx" + ] + .map(|name| (name.to_string(), 50)) + .to_vec() + ); + } } From cbdffc9ba9f7aadb4265d161335ba2709ea9e3e8 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 09:13:49 -0800 Subject: [PATCH 7/7] lore: compact the lore table once it has many small fragments Every lore run appends a few fragments to `lore`, and nothing merges them. A production database had reached 970 fragments, most of them a few rows, and queries that scan lore pay for each one: on a compacted copy, the thread lookup's `references` scan used 27 to 40 s of CPU instead of 51 to 56 s, and a date-bounded subject search 1.0 to 1.3 s instead of 2.0 to 2.8 s. `optimize_single_table()` skips the table, because a compaction that remaps every index on it peaked at 8.9 GB on a production-sized table, and panics on the `references` BTree, which lance 4.0.0 writes in a form it cannot read back. `compact_lore_tables()` now prunes, then compacts `lore` once it has more than 256 fragments, then rebuilds the BTrees as before. Fragments of 150,000 rows or more are not rewritten. Index remapping is deferred: lance records the old-to-new fragment mapping in its fragment-reuse index and leaves the indices as they are, which keeps memory low. The compaction runs as its own task, like the rebuild, so a panic in it is reported as a partial failure. Reads and the FTS update apply the fragment-reuse mapping, but lance 4.0.0's BTree update does not: after such a compaction it drops every row the index held in a rewritten fragment, while the index statistics still report full coverage. That is why the BTrees are rebuilt from the whole table each run rather than updated, and a new test checks that lookups find every row after a compaction of already-indexed fragments. #122, which encodes lore vectors in 64-email chunks, should land before or with this commit: larger fragments let `update_lore_vectors()` hand its workers full 1,024-row batches, which raises the peak of a cold `--vectors` run. Costs and caveats: - On a copy of the production database (998,000 rows, host load 130 to 200 on 316 CPUs), the first compaction took 970 fragments to 6 in 4.0 s, and the BTree rebuild after it 1.6 s, with a process peak of 3.3 GB. - A table built from scratch by the #112/#113 lore pipeline holds only small fragments (997,703 rows in 1,043 fragments), so its first compaction rewrites all of it: 30 to 38 s and a 7.6 GB peak with the rewrite tasks run one at a time (8.4 GB at two, 9.0 GB at four), measured in a process of its own. About 3 GB of that is glibc arena fragmentation: with MALLOC_ARENA_MAX=2 it peaks at 4.2 GB. - A cold build now ends with this compaction, in the process whose indexing peaked at 14.5 GB; that whole run's peak was not measured. Neither was an upstream-built table, which holds more, smaller fragments. - Replaced fragments stay on disk until the next run's prune, so a query still reading them is not cut off. That is about 0.25 GB for an hour on production, and the whole table's data once on a cold database (2.3 GB more here). - lance 4.0.0 never trims the fragment-reuse index, so each compaction adds a version to it, a few KB after the first. At 256 fragments a table compacts about every 25 hourly runs, so about 14 versions a year. --- src/database/schema.rs | 307 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 295 insertions(+), 12 deletions(-) diff --git a/src/database/schema.rs b/src/database/schema.rs index b3e4792..60a0479 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -7,7 +7,7 @@ use futures::{stream, StreamExt, TryStreamExt}; use lancedb::connection::Connection; use lancedb::index::{scalar::BTreeIndexBuilder, scalar::FtsIndexBuilder, Index as LanceIndex}; use lancedb::query::{ExecutableQuery, QueryBase}; -use lancedb::table::{OptimizeAction, OptimizeOptions}; +use lancedb::table::{CompactionOptions, OptimizeAction, OptimizeOptions}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -2128,19 +2128,19 @@ impl SchemaManager { } } - /// Prune old versions of the lore table and rebuild the BTree indices - /// on its non-nullable columns: optimize_single_table() skips lore - /// entirely. + /// Prune the lore table's old versions, merge its small fragments and + /// rebuild the BTree indices on its non-nullable columns: + /// optimize_single_table() skips lore entirely. /// /// Every lore run rewrites the FTS indices, and unless old versions /// are pruned the superseded copies stay under _indices/ until the /// count trips the guard in optimize_lore_fts_indices() and FTS /// updates stop for good. Prune reads only each manifest's index - /// metadata, not the index files, so it avoids the memory cost of - /// Compact. + /// metadata, not the index files. /// - /// The BTree rebuild runs after the prune, so the BTree files it - /// supersedes stay until the next run prunes them. + /// The prune runs first, so the fragments a compaction replaces and the + /// BTree files the rebuild supersedes stay until the next run prunes + /// them, and a query still reading them is not cut off. /// /// Only the lore run calls this, not optimize_database(): with /// delete_unverified set, a prune can delete files that another @@ -2149,14 +2149,79 @@ impl SchemaManager { async fn maintain_lore_table(connection: &Connection) -> Result { let table = connection.open_table("lore").execute().await?; let pruned = Self::prune_old_versions(&table, "lore").await; + let compacted = Self::compact_lore_fragments(&table).await; let btrees_current = Self::rebuild_lore_btree_indices(&table).await; - Ok(if pruned && btrees_current { + Ok(if pruned && compacted && btrees_current { OptimizeOutcome::Optimized } else { OptimizeOutcome::PartialFailure }) } + /// Rewrite the lore table's small fragments into fragments of up to + /// LORE_TARGET_ROWS_PER_FRAGMENT rows once it has more than + /// LORE_COMPACT_FRAGMENTS. Each lore run appends a few fragments, and + /// queries that scan lore pay per fragment. Returns false, after + /// logging why, on failure. + /// + /// Fragments at or above the target are not candidates, and lance + /// 4.0.0 merges only fragments covered by the same set of indices, so + /// a compaction rewrites the fragments appended since the last one and + /// leaves an earlier compaction's output alone even when it is under + /// the target: one leftover fragment per compaction. + /// + /// Index remapping is deferred: lance records the old-to-new fragment + /// mapping in its fragment-reuse index and applies it when an index is + /// read, instead of rewriting every index now, which is what costs + /// memory on a table with large FTS indices. The FTS update applies + /// the mapping too, but lance 4.0.0's BTree update does not: it drops + /// every row the index held in a rewritten fragment, while the index + /// statistics still report full coverage. The BTrees are rebuilt + /// from the table as it is instead (rebuild_lore_btree_indices()), so + /// the mapping does not concern them. lance 4.0.0 never trims the + /// fragment-reuse index, which gains a version of a few KB per + /// compaction; the fragment threshold keeps that to one compaction + /// every few hundred appended fragments. + /// + /// A table of nothing but small fragments, such as one built from + /// scratch, is rewritten whole on its first compaction, one task per + /// LORE_TARGET_ROWS_PER_FRAGMENT rows. The tasks run one at a time: + /// each one in flight adds to peak memory, and a compaction this + /// large happens once per database. + async fn compact_lore_fragments(table: &lancedb::table::Table) -> bool { + const LORE_COMPACT_FRAGMENTS: usize = 256; + const LORE_TARGET_ROWS_PER_FRAGMENT: usize = 150_000; + const LORE_COMPACT_THREADS: usize = 1; + + let fragments = match table.stats().await { + Ok(stats) => stats.fragment_stats.num_fragments, + Err(e) => { + tracing::warn!("Failed to read stats for table lore: {}", e); + return false; + } + }; + if fragments <= LORE_COMPACT_FRAGMENTS { + return true; + } + + tracing::info!("Compacting lore table ({} fragments)...", fragments); + let table = table.clone(); + Self::run_lore_task("compaction", async move { + table + .optimize(OptimizeAction::Compact { + options: CompactionOptions { + target_rows_per_fragment: LORE_TARGET_ROWS_PER_FRAGMENT, + defer_index_remap: true, + num_threads: Some(LORE_COMPACT_THREADS), + ..Default::default() + }, + remap_options: None, + }) + .await + }) + .await + } + /// Run one maintenance step on the lore table and report whether it /// succeeded, logging why not. The step runs as its own task: a lance /// panic would otherwise end semcode-index before the FTS update, on @@ -2270,8 +2335,9 @@ impl SchemaManager { // resident set grows into swap and the OOM killer terminates // semcode-index before compaction finishes, leaving fresh // delta fragments behind each time. Skip the table entirely; - // compact_lore_tables() prunes its old versions and rebuilds the - // BTree indices on its non-nullable columns. + // compact_lore_tables() prunes its old versions, compacts it + // without remapping its indices and rebuilds the BTree indices on + // its non-nullable columns. if table_name == "lore" { tracing::info!( "Skipping generic optimization for lore table \ @@ -2665,7 +2731,9 @@ mod tests { use arrow::array::StringArray; use lancedb::connection::LanceFileVersion; use lancedb::database::listing::ListingDatabaseOptions; + use lancedb::index::scalar::FullTextSearchQuery; use lancedb::index::IndexType; + use lancedb::query::Select; async fn fresh_schema_manager() -> (tempfile::TempDir, SchemaManager) { let dir = tempfile::tempdir().unwrap(); @@ -2769,7 +2837,7 @@ mod tests { subject: format!("[PATCH] change {i}"), references: (i % 2 == 1).then(|| format!("", i - 1)), recipients: "list@example.org".to_string(), - body: format!("body {i}"), + body: format!("body {i} {}", if i % 3 == 0 { "needle" } else { "hay" }), symbols: Vec::new(), }) .collect() @@ -2872,4 +2940,219 @@ mod tests { .to_vec() ); } + + async fn lore_fragments(db: &DatabaseManager) -> usize { + let table = db.connection().open_table("lore").execute().await.unwrap(); + table.stats().await.unwrap().fragment_stats.num_fragments + } + + /// Adds emails `first..first + 2 * fragments` to lore as `fragments` + /// fragments of two rows, the layout many small appends leave, in one + /// commit, since an append per fragment makes the test slow. The rows + /// are built by insert_lore_emails() in a scratch database and copied + /// over. + async fn append_lore_fragments(db: &DatabaseManager, first: usize, fragments: usize) { + let (_scratch_dir, scratch) = lore_database().await; + let failed = scratch + .insert_lore_emails(&emails(first, 2 * fragments)) + .await + .unwrap(); + assert!(failed.is_empty()); + let rows: Vec = scratch + .connection() + .open_table("lore") + .execute() + .await + .unwrap() + .query() + .execute() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + db.connection() + .open_table("lore") + .execute() + .await + .unwrap() + .add(rows) + .write_options(lancedb::table::WriteOptions { + lance_write_params: Some(lance::dataset::WriteParams { + max_rows_per_file: 2, + max_rows_per_group: 2, + mode: lance::dataset::WriteMode::Append, + ..Default::default() + }), + }) + .execute() + .await + .unwrap(); + } + + /// Sorted message_ids of the emails whose body FTS matches `needle`. + async fn needle_message_ids(db: &DatabaseManager) -> Vec { + let table = db.connection().open_table("lore").execute().await.unwrap(); + let query = FullTextSearchQuery::new("needle".to_string()) + .with_column("body".to_string()) + .unwrap(); + let batches: Vec = table + .query() + .full_text_search(query) + .select(Select::columns(&["message_id"])) + .limit(100_000) + .execute() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + let mut ids: Vec = batches + .iter() + .flat_map(|batch| { + let column = batch + .column_by_name("message_id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + (0..column.len()).map(|row| column.value(row).to_string()) + }) + .collect(); + ids.sort(); + ids + } + + /// The message_ids `needle_message_ids()` must return once emails + /// `0..count` are stored. + fn expected_needles(count: usize) -> Vec { + let mut ids: Vec = (0..count) + .filter(|i| i % 3 == 0) + .map(|i| format!("")) + .collect(); + ids.sort(); + ids + } + + /// Above the fragment threshold the lore table is compacted, queries + /// return the same rows through the indices built before it, and both + /// kinds of index keep taking in new rows afterwards. + #[tokio::test] + async fn compact_lore_merges_small_fragments() { + const FRAGMENTS: usize = 257; + let (_dir, db) = lore_database().await; + append_lore_fragments(&db, 0, FRAGMENTS).await; + db.ensure_lore_fts_indices().await.unwrap(); + assert_eq!(lore_fragments(&db).await, FRAGMENTS); + assert_eq!( + needle_message_ids(&db).await, + expected_needles(2 * FRAGMENTS) + ); + + assert!(db.compact_lore_tables().await.unwrap()); + + assert!(lore_fragments(&db).await < 10); + let table = db.connection().open_table("lore").execute().await.unwrap(); + assert_eq!(table.count_rows(None).await.unwrap(), 2 * FRAGMENTS); + assert_eq!( + needle_message_ids(&db).await, + expected_needles(2 * FRAGMENTS) + ); + assert_covers(&btree_coverage(&db).await, 2 * FRAGMENTS); + let found = db + .get_lore_email_by_message_id("") + .await + .unwrap() + .unwrap(); + assert_eq!(found.subject, "[PATCH] change 7"); + + let failed = db + .insert_lore_emails(&emails(2 * FRAGMENTS, 10)) + .await + .unwrap(); + assert!(failed.is_empty()); + assert!(db.optimize_lore_fts_indices().await.unwrap()); + assert!(db.compact_lore_tables().await.unwrap()); + assert_eq!( + needle_message_ids(&db).await, + expected_needles(2 * FRAGMENTS + 10) + ); + assert_covers(&btree_coverage(&db).await, 2 * FRAGMENTS + 10); + } + + /// At or below the fragment threshold the lore table is not compacted, + /// so a run adds no version to lance's fragment-reuse index. + #[tokio::test] + async fn compact_lore_leaves_few_fragments_alone() { + let (_dir, db) = lore_database().await; + append_lore_fragments(&db, 0, 256).await; + assert_eq!(lore_fragments(&db).await, 256); + + assert!(db.compact_lore_tables().await.unwrap()); + + assert_eq!(lore_fragments(&db).await, 256); + } + + /// How many rows `column IN (values)` finds. The column's BTree + /// serves the lookup, so a row the index lost is not found. + async fn rows_found(db: &DatabaseManager, column: &str, values: &[String]) -> usize { + let list = values + .iter() + .map(|value| format!("'{value}'")) + .collect::>() + .join(", "); + let table = db.connection().open_table("lore").execute().await.unwrap(); + let batches: Vec = table + .query() + .only_if(format!("{column} IN ({list})")) + .select(Select::columns(&["message_id"])) + .execute() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + batches.iter().map(|batch| batch.num_rows()).sum() + } + + /// Asserts that `message_id` and `git_commit_sha` lookups find each of + /// emails `0..count` exactly once. + async fn assert_lookups_find(db: &DatabaseManager, count: usize) { + let ids: Vec = (0..count) + .map(|i| format!("")) + .collect(); + let shas: Vec = (0..count).map(|i| format!("{i:040x}")).collect(); + assert_eq!(rows_found(db, "message_id", &ids).await, count); + assert_eq!(rows_found(db, "git_commit_sha", &shas).await, count); + } + + /// A compaction rewrites fragments the BTrees already index, as every + /// compaction after a database's first does. Lookups through the + /// BTrees must still find those rows, in that run and the next. + #[tokio::test] + async fn compact_lore_keeps_indexed_rows_findable() { + let (_dir, db) = lore_database().await; + append_lore_fragments(&db, 0, 200).await; + assert!(db.compact_lore_tables().await.unwrap()); + db.ensure_lore_fts_indices().await.unwrap(); + assert_covers(&btree_coverage(&db).await, 400); + append_lore_fragments(&db, 400, 57).await; + assert_eq!(lore_fragments(&db).await, 257); + + assert!(db.compact_lore_tables().await.unwrap()); + + assert!(lore_fragments(&db).await < 10); + assert_covers(&btree_coverage(&db).await, 514); + assert_lookups_find(&db, 514).await; + + let failed = db.insert_lore_emails(&emails(514, 4)).await.unwrap(); + assert!(failed.is_empty()); + assert!(db.optimize_lore_fts_indices().await.unwrap()); + assert!(db.compact_lore_tables().await.unwrap()); + + assert_covers(&btree_coverage(&db).await, 518); + assert_lookups_find(&db, 518).await; + assert_eq!(needle_message_ids(&db).await, expected_needles(518)); + } }