BM25

core/src/cache.rs

@@ -103,7 +103,7 @@ pub async fn refresh_job_cache(
     job_cache: &Arc<RwLock<HashMap<String, VectorizeJob>>>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let all_jobs: Vec<VectorizeJob> = sqlx::query_as(
-        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model FROM vectorize.job",
+        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model, bm25_enabled FROM vectorize.job",
     )
     .fetch_all(db_pool)
     .await?;
@@ -129,7 +129,7 @@ pub async fn load_initial_job_cache(
     pool: &sqlx::PgPool,
 ) -> Result<HashMap<String, VectorizeJob>, sqlx::Error> {
     let all_jobs: Vec<VectorizeJob> = sqlx::query_as(
-        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model FROM vectorize.job",
+        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model, bm25_enabled FROM vectorize.job",
     )
     .fetch_all(pool)
     .await?;

core/src/db.rs

@@ -7,8 +7,8 @@ pub async fn get_vectorize_job(
 ) -> Result<VectorizeJob, VectorizeError> {
     // Changed return type
     let row = sqlx::query(
-        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model 
-         FROM vectorize.job 
+        "SELECT job_name, src_table, src_schema, src_columns, primary_key, update_time_col, model, bm25_enabled
+         FROM vectorize.job
          WHERE job_name = $1",
     )
     .bind(job_name)

core/src/init.rs

@@ -77,9 +77,10 @@ pub async fn init_vectorize(pool: &PgPool) -> Result<(), VectorizeError> {
         return Ok(());
     } else {
         // these statements are critical, so we fail if they error
+        // Note: the vectorize.job table itself is created/evolved via the
+        // vectorize-server crate's sqlx migrations (server/migrations), not here.
         let statements_nofail = vec![
             "CREATE SCHEMA IF NOT EXISTS vectorize;".to_string(),
-            query::create_vectorize_table(),
             query::handle_table_update(),
             query::create_batch_texts_fn(),
         ];
@@ -126,15 +127,16 @@ pub async fn initialize_job(
     // create the job record
     let mut tx = pool.begin().await?;
     let job_id: Uuid = sqlx::query_scalar("
-        INSERT INTO vectorize.job (job_name, src_schema, src_table, src_columns, primary_key, update_time_col, model)
-        VALUES ($1, $2, $3, $4, $5, $6, $7)
+        INSERT INTO vectorize.job (job_name, src_schema, src_table, src_columns, primary_key, update_time_col, model, bm25_enabled)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
         ON CONFLICT (job_name) DO UPDATE SET
             src_schema = EXCLUDED.src_schema,
             src_table = EXCLUDED.src_table,
             src_columns = EXCLUDED.src_columns,
             primary_key = EXCLUDED.primary_key,
             update_time_col = EXCLUDED.update_time_col,
-            model = EXCLUDED.model
+            model = EXCLUDED.model,
+            bm25_enabled = EXCLUDED.bm25_enabled
         RETURNING id")
         .bind(job_request.job_name.clone())
         .bind(job_request.src_schema.clone())
@@ -143,6 +145,7 @@ pub async fn initialize_job(
         .bind(job_request.primary_key.clone())
         .bind(job_request.update_time_col.clone())
         .bind(job_request.model.to_string())
+        .bind(job_request.bm25_enabled)
         .fetch_one(&mut *tx)
         .await?;
 

core/src/query.rs

@@ -202,23 +202,6 @@ pub fn check_input(input: &str) -> Result<()> {
     }
 }
 
-pub fn create_vectorize_table() -> String {
-    "CREATE TABLE IF NOT EXISTS vectorize.job
-        (
-            id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
-            job_name TEXT NOT NULL UNIQUE,
-            src_schema TEXT NOT NULL,
-            src_table TEXT NOT NULL,
-            src_columns TEXT[] NOT NULL,
-            primary_key TEXT NOT NULL,
-            update_time_col TEXT NOT NULL,
-            model TEXT NOT NULL,
-            params JSONB
-        );
-        "
-    .to_string()
-}
-
 pub fn init_index_query(job_name: &str, idx_type: &str, job_params: &JobParams) -> String {
     check_input(job_name).expect("invalid job name");
     let src_schema = job_params.schema.clone();
@@ -699,16 +682,20 @@ pub fn join_table_cosine_similarity(
     )
 }
 
-fn build_where_filter(filters: &BTreeMap<String, FilterValue>) -> String {
+fn build_where_filter_with_start(filters: &BTreeMap<String, FilterValue>, start: i16) -> String {
     let mut where_filter = "WHERE 1=1".to_string();
-    for (bind_value_counter, (column, filter_value)) in (3_i16..).zip(filters.iter()) {
+    for (bind_value_counter, (column, filter_value)) in (start..).zip(filters.iter()) {
         let operator = filter_value.operator.to_sql();
         let filt = format!(" AND t0.\"{column}\" {operator} ${bind_value_counter}");
         where_filter.push_str(&filt);
     }
     where_filter
 }
 
+fn build_where_filter(filters: &BTreeMap<String, FilterValue>) -> String {
+    build_where_filter_with_start(filters, 3)
+}
+
 /// Generates the core hybrid search SELECT that returns raw table rows.
 /// `$1::vector` and `$2` are sqlx bind parameter placeholders for the embedding and query text.
 #[allow(clippy::too_many_arguments)]
@@ -855,6 +842,107 @@ pub fn hybrid_search_query_rows(
         &where_filter,
     )
 }
+/// Generates the three-arm hybrid search SQL (semantic + FTS + BM25).
+///
+/// Bind parameters:
+///   $1 - embedding vector
+///   $2 - FTS query text
+///   $3 - BM25 ranked PKs as text[] (ordered by Tantivy BM25 score)
+///   $4+ - filter values
+///
+/// The BM25 CTE joins the `$3` array back to the source table so the PK
+/// type is resolved correctly regardless of whether it is int, uuid, or text.
+#[allow(clippy::too_many_arguments)]
+pub fn hybrid_search_query_with_bm25(
+    job_name: &str,
+    src_schema: &str,
+    src_table: &str,
+    join_key: &str,
+    return_columns: &[String],
+    window_size: i32,
+    limit: i32,
+    rrf_k: f32,
+    semantic_weight: f32,
+    fts_weight: f32,
+    bm25_weight: f32,
+    filters: &BTreeMap<String, FilterValue>,
+) -> String {
+    let cols = return_columns
+        .iter()
+        .map(|s| format!("t0.{s}"))
+        .collect::<Vec<_>>()
+        .join(",");
+    // filters start at $4 because $3 is the BM25 PKs array
+    let where_filter = build_where_filter_with_start(filters, 4);
+
+    let inner = format!(
+        "
+        WITH bm25_cte AS (
+            SELECT t0.{join_key}, b.bm25_rank
+            FROM (
+                SELECT val AS pk_text, ord::int AS bm25_rank
+                FROM unnest($3::text[]) WITH ORDINALITY AS t(val, ord)
+            ) b
+            JOIN {src_schema}.{src_table} t0 ON t0.{join_key}::text = b.pk_text
+        )
+        SELECT {cols}, t.rrf_score, t.semantic_rank, t.fts_rank, t.bm25_rank, t.similarity_score
+        FROM (
+            SELECT
+                COALESCE(s.{join_key}, f.{join_key}, bm25.{join_key}) AS {join_key},
+                s.semantic_rank,
+                s.similarity_score,
+                f.fts_rank,
+                bm25.bm25_rank,
+                (
+                    COALESCE({semantic_weight}::float / ({rrf_k} + s.semantic_rank), 0) +
+                    COALESCE({fts_weight}::float / ({rrf_k} + f.fts_rank), 0) +
+                    COALESCE({bm25_weight}::float / ({rrf_k} + bm25.bm25_rank), 0)
+                ) AS rrf_score
+            FROM (
+                SELECT
+                    {join_key},
+                    distance,
+                    ROW_NUMBER() OVER (ORDER BY distance) AS semantic_rank,
+                    1 - distance AS similarity_score
+                FROM (
+                    SELECT
+                        {join_key},
+                        embeddings <=> $1::vector AS distance
+                    FROM vectorize._embeddings_{job_name}
+                ) sub
+                ORDER BY distance
+                LIMIT {window_size}
+            ) s
+            FULL OUTER JOIN (
+                SELECT
+                    {join_key},
+                    ROW_NUMBER() OVER (ORDER BY ts_rank_cd(search_tokens, query) DESC) AS fts_rank
+                FROM vectorize._search_tokens_{job_name},
+                     to_tsquery('english',
+                         NULLIF(
+                             replace(plainto_tsquery('english', $2)::text, ' & ', ' | '),
+                             ''
+                         )
+                     ) AS query
+                WHERE search_tokens @@ query
+                ORDER BY ts_rank_cd(search_tokens, query) DESC
+                LIMIT {window_size}
+            ) f ON s.{join_key} = f.{join_key}
+            FULL OUTER JOIN bm25_cte bm25
+                ON COALESCE(s.{join_key}, f.{join_key}) = bm25.{join_key}
+        ) t
+        INNER JOIN {src_schema}.{src_table} t0 ON t0.{join_key} = t.{join_key}
+        {where_filter}
+        ORDER BY t.rrf_score DESC
+        LIMIT {limit}"
+    );
+
+    format!(
+        "SELECT to_jsonb(t) as results FROM ({inner}
+    ) t"
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

core/src/types.rs

@@ -21,6 +21,10 @@ pub struct VectorizeJob {
         serialize_with = "model_to_string"
     )]
     pub model: Model,
+    /// Opt-in flag to build and maintain an in-memory BM25 (Tantivy) index for this job.
+    /// Defaults to false so BM25 indexing never runs unless explicitly requested.
+    #[serde(default)]
+    pub bm25_enabled: bool,
 }
 
 #[allow(non_camel_case_types)]

server/Cargo.toml

@@ -46,7 +46,7 @@ reqwest = { version = "0.12.16", features = ["json"] }
 serde = "1.0.219"
 serde_json = "1.0"
 sqlparser = "0.51"
-sqlx = { workspace = true}
+sqlx = { workspace = true, features = ["migrate"] }
 thiserror = "2.0.12"
 tiktoken-rs = "0.7.0"
 tokio = { version = "1.0", features = ["full"] }
@@ -55,6 +55,7 @@ url = "2.2"
 utoipa = { version = "4", features = ["actix_extras", "chrono", "uuid"] }
 utoipa-swagger-ui  = { version = "7", features = ["actix-web"] }
 uuid = { version = "1.16.0", features = ["v4", "fast-rng", "macro-diagnostics", "serde"] }
+tantivy = "0.22"
 
 [dev-dependencies]
 pgvector = { version = "0.4.1", features = ["postgres", "sqlx"] }

server/migrations/0001_init_job_table.sql

@@ -0,0 +1,15 @@
+-- Baseline: reflects the vectorize.job table as already deployed prior to
+-- adopting sqlx migrations. Uses IF NOT EXISTS so this is a no-op against
+-- any database where vectorize-core already created this table by hand.
+CREATE TABLE IF NOT EXISTS vectorize.job
+(
+    id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
+    job_name TEXT NOT NULL UNIQUE,
+    src_schema TEXT NOT NULL,
+    src_table TEXT NOT NULL,
+    src_columns TEXT[] NOT NULL,
+    primary_key TEXT NOT NULL,
+    update_time_col TEXT NOT NULL,
+    model TEXT NOT NULL,
+    params JSONB
+);

server/migrations/0002_add_bm25_enabled.sql

@@ -0,0 +1,4 @@
+-- Opt-in flag for building/maintaining an in-memory BM25 (Tantivy) index per job.
+-- Defaults to false so upgraded deployments never get BM25 indexing turned on
+-- for existing jobs without an explicit opt-in via POST /table.
+ALTER TABLE vectorize.job ADD COLUMN IF NOT EXISTS bm25_enabled BOOLEAN NOT NULL DEFAULT false;

server/src/app_state.rs

@@ -1,11 +1,12 @@
 use std::collections::HashMap;
 use std::sync::Arc;
-use tokio::sync::RwLock;
+use tokio::sync::{Mutex, RwLock};
 use tracing::error;
 use vectorize_core::config::Config;
 use vectorize_core::types::VectorizeJob;
 use vectorize_worker::WorkerHealth;
 
+use crate::bm25::BM25Index;
 use crate::cache;
 
 #[derive(Debug, thiserror::Error)]
@@ -27,6 +28,8 @@ pub struct AppState {
     pub job_cache: Arc<RwLock<HashMap<String, VectorizeJob>>>,
     /// worker health monitoring data
     pub worker_health: Arc<RwLock<WorkerHealth>>,
+    /// in-memory BM25 indexes keyed by job_name; rebuilt from source on startup
+    pub bm25_indexes: Arc<RwLock<HashMap<String, Arc<Mutex<BM25Index>>>>>,
 }
 
 impl AppState {
@@ -45,6 +48,10 @@ impl AppState {
             .await
             .map_err(|e| format!("Failed to initialize project: {e}"))?;
 
+        crate::db::run_migrations(&db_pool)
+            .await
+            .map_err(|e| format!("Failed to run migrations: {e}"))?;
+
         // load initial job cache
         let job_cache = cache::load_initial_job_cache(&db_pool)
             .await
@@ -66,12 +73,61 @@ impl AppState {
             last_error: None,
         }));
 
+        // Create empty BM25 index map, then kick off background population for
+        // every job that has opted in via `bm25_enabled`, so the server stays
+        // non-blocking at startup and jobs that never asked for BM25 incur no cost.
+        let bm25_indexes: Arc<RwLock<HashMap<String, Arc<Mutex<BM25Index>>>>> =
+            Arc::new(RwLock::new(HashMap::new()));
+
+        {
+            let jobs: Vec<VectorizeJob> = {
+                let cache = job_cache.read().await;
+                cache
+                    .values()
+                    .filter(|job| job.bm25_enabled)
+                    .cloned()
+                    .collect()
+            };
+            for job in jobs {
+                if job.job_name.is_empty() {
+                    continue;
+                }
+                match BM25Index::new() {
+                    Ok(idx) => {
+                        let idx = Arc::new(Mutex::new(idx));
+                        bm25_indexes
+                            .write()
+                            .await
+                            .insert(job.job_name.clone(), idx.clone());
+                        let pool = db_pool.clone();
+                        tokio::spawn(async move {
+                            crate::bm25::populate_bm25_index(&pool, &job, idx).await;
+                        });
+                    }
+                    Err(e) => {
+                        tracing::warn!("Failed to create BM25 index for job {}: {e}", job.job_name);
+                    }
+                }
+            }
+        }
+
+        // Background sync: pick up changed documents every 30 s.
+        {
+            let pool = db_pool.clone();
+            let indexes = bm25_indexes.clone();
+            let cache = job_cache.clone();
+            tokio::spawn(async move {
+                crate::bm25::start_bm25_sync_task(pool, indexes, cache).await;
+            });
+        }
+
         Ok(AppState {
             config,
             db_pool,
             cache_pool,
             job_cache,
             worker_health,
+            bm25_indexes,
         })
     }
 

server/src/bin/worker.rs

@@ -21,6 +21,10 @@ async fn main() {
         .await
         .expect("Failed to initialize project");
 
+    vectorize_server::db::run_migrations(&pool)
+        .await
+        .expect("Failed to run migrations");
+
     let queue = pgmq::PGMQueueExt::new_with_pool(pool.clone()).await;
 
     loop {