diff --git a/Cargo.lock b/Cargo.lock index 47f3ce03..ed705d32 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6474,7 +6474,6 @@ dependencies = [ "anyhow", "common", "prost", - "rust_decimal", "serde", "serde_json", "sqlx", diff --git a/client b/client index cbe1f9ad..f8bc80a4 160000 --- a/client +++ b/client @@ -1 +1 @@ -Subproject commit cbe1f9adc687c46b168daca43f6d96a90e34f839 +Subproject commit f8bc80a415138e4eb5cb9b70523a9556250d21d4 diff --git a/common/src/search.rs b/common/src/search.rs index 8eecd1be..935d361e 100644 --- a/common/src/search.rs +++ b/common/src/search.rs @@ -2,7 +2,7 @@ use std::path::{Path, PathBuf}; use tantivy::schema::{ Field, IndexRecordOption, JsonObjectOptions, Schema, Term, TextFieldIndexing, TextOptions, - INDEXED, STORED, STRING, + FAST, INDEXED, STORED, STRING, }; use tantivy::tokenizer::{ AsciiFoldingFilter, LowerCaser, NgramTokenizer, RawTokenizer, RemoveLongFilter, @@ -11,6 +11,10 @@ use tantivy::tokenizer::{ use tantivy::Index; pub const F_PG_ID: &str = "pg_id"; +pub const F_ROW_REVISION: &str = "row_revision"; +pub const F_IS_ARCHIVED: &str = "is_archived"; +pub const F_TABLE_DEFINITION_ID: &str = "table_definition_id"; +pub const F_VERSION_NUMBER: &str = "version_number"; pub const F_TABLE_NAME: &str = "table_name"; pub const F_ROW_KEY: &str = "row_key"; pub const F_ALL_TEXT: &str = "all_text"; @@ -19,7 +23,7 @@ pub const F_DATA_NGRAM: &str = "data_ngram"; pub const F_DATA_EXACT: &str = "data_exact"; pub const JOURNAL_TABLE_NAME: &str = "general_ledger"; pub const ARCHIVED_ROW_KEY_PREFIX: &str = "__archive__:"; -pub const SEARCH_INDEX_FORMAT_DIRECTORY: &str = "v2"; +pub const SEARCH_INDEX_FORMAT_DIRECTORY: &str = "v3"; /// Root for the current on-disk index format. The format component prevents /// an executable with new projection semantics from opening old documents. @@ -84,6 +88,38 @@ pub fn normalize_exact(input: &str) -> String { out } +/// Canonicalizes an exact constraint with the same scalar spelling used by +/// the indexer. Catalog field types are public type names, not PostgreSQL type +/// names, so this deliberately matches the catalog vocabulary. +pub fn canonical_exact_search_value(input: &str, field_type: &str) -> Result { + let normalized_type = field_type.trim().to_ascii_lowercase(); + if matches!(normalized_type.as_str(), "numeric" | "money") + || normalized_type.starts_with("decimal(") + { + return input + .parse::() + .map(|value| value.normalize().to_string()) + .map_err(|error| format!("Exact numeric search value is invalid: {error}")); + } + if matches!(normalized_type.as_str(), "int" | "bigint") + || normalized_type.starts_with("link(") + { + return input + .parse::() + .map(|value| value.to_string()) + .map_err(|error| format!("Exact integer search value is invalid: {error}")); + } + Ok(input.to_string()) +} + +/// Canonical spelling for a JSON number stored in a Tantivy text term. +pub fn canonical_search_number(input: &str) -> String { + input + .parse::() + .map(|value| value.normalize().to_string()) + .unwrap_or_else(|_| input.to_string()) +} + /// Normalizes a column name to the JSON-key form used at index time. pub fn normalize_column_name(column: &str) -> String { column.to_ascii_lowercase() @@ -93,7 +129,11 @@ pub fn normalize_column_name(column: &str) -> String { pub fn create_search_schema() -> Schema { let mut schema_builder = Schema::builder(); - schema_builder.add_u64_field(F_PG_ID, INDEXED | STORED); + schema_builder.add_u64_field(F_PG_ID, INDEXED | STORED | FAST); + schema_builder.add_u64_field(F_ROW_REVISION, STORED | FAST); + schema_builder.add_u64_field(F_IS_ARCHIVED, INDEXED | STORED | FAST); + schema_builder.add_u64_field(F_TABLE_DEFINITION_ID, STORED | FAST); + schema_builder.add_u64_field(F_VERSION_NUMBER, STORED | FAST); schema_builder.add_text_field(F_TABLE_NAME, STRING | STORED); schema_builder.add_text_field(F_ROW_KEY, STRING | STORED); schema_builder.add_text_field(F_ALL_TEXT, text_options(TOK_WORD)); @@ -197,6 +237,10 @@ pub fn json_path_term(field: Field, column: &str, text: &str) -> Term { /// Returns all required schema fields or fails loudly on mismatch. pub struct SchemaFields { pub pg_id: Field, + pub row_revision: Field, + pub is_archived: Field, + pub table_definition_id: Field, + pub version_number: Field, pub table_name: Field, pub row_key: Field, pub all_text: Field, @@ -209,6 +253,10 @@ impl SchemaFields { pub fn from(schema: &Schema) -> tantivy::Result { Ok(Self { pg_id: get_field(schema, F_PG_ID)?, + row_revision: get_field(schema, F_ROW_REVISION)?, + is_archived: get_field(schema, F_IS_ARCHIVED)?, + table_definition_id: get_field(schema, F_TABLE_DEFINITION_ID)?, + version_number: get_field(schema, F_VERSION_NUMBER)?, table_name: get_field(schema, F_TABLE_NAME)?, row_key: get_field(schema, F_ROW_KEY)?, all_text: get_field(schema, F_ALL_TEXT)?, @@ -221,7 +269,9 @@ impl SchemaFields { #[cfg(test)] mod tests { - use super::{archived_search_row_key, parse_archived_search_row_key}; + use super::{ + archived_search_row_key, canonical_exact_search_value, parse_archived_search_row_key, + }; #[test] fn archived_row_key_round_trips_stable_identity() { @@ -230,6 +280,19 @@ mod tests { assert_eq!(parse_archived_search_row_key("customers:7"), None); assert_eq!(parse_archived_search_row_key("__archive__:42:7"), None); } + + #[test] + fn exact_scalar_values_follow_catalog_field_types() { + assert_eq!(canonical_exact_search_value("001", "int").unwrap(), "1"); + assert_eq!(canonical_exact_search_value("001", "bigint").unwrap(), "1"); + assert_eq!(canonical_exact_search_value("001", "link(adresar)").unwrap(), "1"); + assert_eq!(canonical_exact_search_value("10.50", "numeric").unwrap(), "10.5"); + assert_eq!(canonical_exact_search_value("10.50", "money").unwrap(), "10.5"); + assert_eq!( + canonical_exact_search_value("10.50", "decimal(12, 2)").unwrap(), + "10.5" + ); + } } fn get_field(schema: &Schema, name: &str) -> tantivy::Result { diff --git a/search/Cargo.toml b/search/Cargo.toml index a7b90850..6de946fd 100644 --- a/search/Cargo.toml +++ b/search/Cargo.toml @@ -9,7 +9,6 @@ anyhow = { workspace = true } prost = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -rust_decimal = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tracing = { workspace = true } diff --git a/search/src/lib.rs b/search/src/lib.rs index 23585e2d..c9dbd919 100644 --- a/search/src/lib.rs +++ b/search/src/lib.rs @@ -11,8 +11,9 @@ use common::proto::komp_ac::search::{ search_response::Hit, }; use common::search::{ - SchemaFields, parse_archived_search_row_key, register_tokenizers, search_index_path, - search_row_key, + F_IS_ARCHIVED, F_PG_ID, F_ROW_REVISION, F_TABLE_DEFINITION_ID, F_VERSION_NUMBER, + SchemaFields, canonical_exact_search_value, parse_archived_search_row_key, + register_tokenizers, search_index_path, search_row_key, }; use common::system_column::{internal_column_names, is_internal_column, is_system_column}; use query_builder::{ @@ -31,6 +32,7 @@ const HARD_RESULT_LIMIT: usize = 200; const DEFAULT_LIST_LIMIT: usize = 5; const SEARCH_SCORE_RELATIVE_FLOOR: f32 = 0.25; const SEARCH_SCORE_GROUP_WIDTH: f32 = 0.25; +const SEARCH_VALIDATION_OVERFETCH_FACTOR: usize = 4; pub struct SearcherService { pub pool: PgPool, @@ -112,10 +114,6 @@ impl SearcherService { .await?; let profile = profile_index(&self.profiles, &normalized.profile_name, &index_path)?; - profile - .reader - .reload() - .map_err(|error| Status::internal(format!("Search index reload failed: {error}")))?; let mut hits = run_search( &self.pool, &profile, @@ -127,6 +125,7 @@ impl SearcherService { normalized.offset, normalized.order.as_ref(), normalized.version_scope, + false, ) .await?; @@ -187,10 +186,6 @@ impl SearcherService { ) .await?; let profile = profile_index(&self.profiles, &normalized.profile_name, &index_path)?; - profile - .reader - .reload() - .map_err(|error| Status::internal(format!("Search index reload failed: {error}")))?; let query = build_master_query( &profile.index, &profile.fields, @@ -219,12 +214,7 @@ async fn count_authoritative_matches( query: &dyn Query, ) -> Result { let searcher = profile.reader.searcher(); - let documents = searcher - .search( - query, - &TopDocs::with_limit(searcher.num_docs() as usize).order_by_score(), - ) - .map_err(|error| Status::internal(format!("Search count failed: {error}")))?; + let documents = top_documents_for_count(&searcher, query)?; let best_score = documents .first() .map(|(score, _)| *score) @@ -234,44 +224,92 @@ async fn count_authoritative_matches( } else { 0.0 }; - let mut current_ids = Vec::new(); + let mut current_rows = Vec::new(); let mut archives = Vec::new(); for (score, address) in documents { if score < score_floor { continue; } - let document: TantivyDocument = searcher - .doc(address) - .map_err(|error| Status::internal(format!("Search count document read failed: {error}")))?; - let Some(row_key) = document - .get_first(profile.fields.row_key) - .and_then(|value| value.as_str()) - else { - continue; + let segment = searcher.segment_reader(address.segment_ord); + let fast_u64 = |field_name: &str| -> Result { + segment + .fast_fields() + .u64(field_name) + .map_err(|error| Status::internal(format!( + "Search count fast field '{field_name}' failed: {error}" + )))? + .first(address.doc_id) + .and_then(|value| i64::try_from(value).ok()) + .ok_or_else(|| Status::internal(format!( + "Search count document has no valid '{field_name}'" + ))) }; - if let Some(archive) = parse_archived_search_row_key(row_key) { - archives.push(archive); - } else if let Some(row_id) = document - .get_first(profile.fields.pg_id) - .and_then(|value| value.as_u64()) - .and_then(|value| i64::try_from(value).ok()) - { - current_ids.push(row_id); + let row_id = fast_u64(F_PG_ID)?; + if fast_u64(F_IS_ARCHIVED)? != 0 { + archives.push(( + fast_u64(F_TABLE_DEFINITION_ID)?, + row_id, + fast_u64(F_VERSION_NUMBER)?, + )); + } else { + current_rows.push((row_id, fast_u64(F_ROW_REVISION)?)); } } - current_ids.sort_unstable(); - current_ids.dedup(); + current_rows.sort_unstable(); + current_rows.dedup(); archives.sort_unstable(); archives.dedup(); - let current_count = if current_ids.is_empty() { + count_validated_candidates( + pool, + profile_name, + table_name, + ¤t_rows, + &archives, + ) + .await +} + +fn top_documents_for_count( + searcher: &tantivy::Searcher, + query: &dyn Query, +) -> Result, Status> { + if searcher.num_docs() == 0 { + return Ok(Vec::new()); + } + // DocAddress is local to this exact Searcher snapshot. The caller must use + // the same searcher to load every address returned here. + searcher + .search( + query, + &TopDocs::with_limit(searcher.num_docs() as usize).order_by_score(), + ) + .map_err(|error| Status::internal(format!("Search count failed: {error}"))) +} + +async fn count_validated_candidates( + pool: &PgPool, + profile_name: &str, + table_name: &str, + current_rows: &[(i64, i64)], + archives: &[(i64, i64, i64)], +) -> Result { + let current_count = if current_rows.is_empty() { 0 } else { let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; + let current_ids = current_rows.iter().map(|item| item.0).collect::>(); + let current_revisions = current_rows.iter().map(|item| item.1).collect::>(); sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( - "SELECT COUNT(*) FROM {qualified_table} WHERE deleted = FALSE AND id = ANY($1)" + "SELECT COUNT(*) FROM {qualified_table} current_row \ + JOIN UNNEST($1::BIGINT[], $2::BIGINT[]) \ + AS candidate(id, row_revision) \ + ON candidate.id = current_row.id \ + AND candidate.row_revision = current_row.row_revision \ + WHERE current_row.deleted = FALSE" ))) .bind(¤t_ids) + .bind(¤t_revisions) .fetch_one(pool) .await .map_err(|error| Status::internal(format!("Current search count validation failed: {error}")))? @@ -367,26 +405,7 @@ struct NormalizedColumnConstraint { } fn normalize_constraint_value(query: &str, field_type: &str) -> Result { - let normalized_type = field_type.trim().to_ascii_lowercase(); - if normalized_type == "money" || normalized_type.starts_with("decimal(") { - return query - .parse::() - .map(|value| value.normalize().to_string()) - .map_err(|error| Status::invalid_argument(format!( - "Exact numeric search value is invalid: {error}" - ))); - } - if matches!(normalized_type.as_str(), "integer" | "bigint") - || normalized_type.starts_with("link(") - { - return query - .parse::() - .map(|value| value.to_string()) - .map_err(|error| Status::invalid_argument(format!( - "Exact integer search value is invalid: {error}" - ))); - } - Ok(query.to_string()) + canonical_exact_search_value(query, field_type).map_err(Status::invalid_argument) } fn public_search_system_type(column: &str) -> Option<&'static str> { @@ -401,6 +420,7 @@ fn public_search_system_type(column: &str) -> Option<&'static str> { struct SearchCandidate { score: f32, row_id: i64, + row_revision: i64, table_name: String, row_key: String, } @@ -1144,6 +1164,7 @@ async fn run_search( offset: usize, order: Option<&NormalizedSearchOrder>, version_scope: SearchVersionScope, + exhaustive_validation: bool, ) -> Result, Status> { let master_query = build_master_query( &profile.index, @@ -1155,10 +1176,13 @@ async fn run_search( )?; let searcher = profile.reader.searcher(); - let window_limit = if order.is_some() { + let window_limit = if order.is_some() || exhaustive_validation { searcher.num_docs() as usize } else { - offset.saturating_add(limit) + offset + .saturating_add(limit) + .saturating_mul(SEARCH_VALIDATION_OVERFETCH_FACTOR) + .min(searcher.num_docs() as usize) }; if window_limit == 0 { return Ok(Vec::new()); @@ -1181,15 +1205,7 @@ async fn run_search( .into_iter() .filter(|(score, _)| *score >= score_floor) .collect::>(); - let page_docs = if order.is_some() { - eligible_docs - } else { - eligible_docs - .into_iter() - .skip(offset) - .take(limit) - .collect::>() - }; + let page_docs = eligible_docs; let mut candidates: Vec = Vec::with_capacity(page_docs.len()); for (score, doc_address) in page_docs { @@ -1214,9 +1230,17 @@ async fn run_search( else { continue; }; + let Some(row_revision) = doc + .get_first(profile.fields.row_revision) + .and_then(|value| value.as_u64()) + .and_then(|value| i64::try_from(value).ok()) + else { + continue; + }; candidates.push(SearchCandidate { score, row_id: pg_id as i64, + row_revision, table_name: table_name.to_string(), row_key: row_key.to_string(), }); @@ -1240,20 +1264,20 @@ async fn run_search( .await; } - let mut ids_by_table: HashMap> = HashMap::new(); + let mut rows_by_table: HashMap> = HashMap::new(); for candidate in &candidates { if parse_archived_search_row_key(&candidate.row_key).is_some() { continue; } - ids_by_table + rows_by_table .entry(candidate.table_name.clone()) .or_default() - .push(candidate.row_id); + .push((candidate.row_id, candidate.row_revision)); } let mut content_map: HashMap, Vec, i64, bool)> = HashMap::new(); - for (table_name, pg_ids) in ids_by_table { + for (table_name, candidate_rows) in rows_by_table { validate_identifier(&table_name, "table_name")?; let physical_to_display = table_physical_to_display_map(pool, profile_name, &table_name).await?; @@ -1261,6 +1285,8 @@ async fn run_search( table_internal_column_names(pool, profile_name, &table_name).await?; let display_columns = table_row_display_columns(pool, profile_name, &table_name).await?; let qualified_table = qualified_visible_table(pool, profile_name, &table_name).await?; + let pg_ids = candidate_rows.iter().map(|item| item.0).collect::>(); + let revisions_by_id = candidate_rows.into_iter().collect::>(); let sql = format!( "SELECT id, to_jsonb(t) AS data FROM {} t WHERE deleted = FALSE AND id = ANY($1)", qualified_table @@ -1274,6 +1300,12 @@ async fn run_search( for row in rows { let id: i64 = row.try_get("id").unwrap_or_default(); let json_data: serde_json::Value = row.try_get("data").unwrap_or_default(); + let row_revision = json_data + .get("row_revision") + .and_then(|value| value.as_i64()); + if revisions_by_id.get(&id).copied() != row_revision { + continue; + } let version = json_data.get("version").and_then(|value| value.as_i64()).unwrap_or(0); let json_data = remap_json_to_display_names( json_data, @@ -1371,7 +1403,7 @@ async fn run_search( } } - Ok(candidates + let hits = candidates .into_iter() .filter_map(|candidate| { content_map @@ -1388,7 +1420,32 @@ async fn run_search( archived: *archived, }) }) - .collect()) + .collect::>(); + if order.is_some() { + Ok(hits) + } else { + let page = hits.into_iter().skip(offset).take(limit).collect::>(); + if !exhaustive_validation + && page.len() < limit + && window_limit < searcher.num_docs() as usize + { + return Box::pin(run_search( + pool, + profile, + profile_name, + table_filter, + free_query, + must, + limit, + offset, + order, + version_scope, + true, + )) + .await; + } + Ok(page) + } } async fn fetch_ordered_candidate_rows( @@ -1414,10 +1471,11 @@ async fn fetch_ordered_candidate_rows( SELECT positioned.id, to_jsonb(positioned) - 'picker_position' AS data, \ picker_position, candidate_score \ FROM positioned \ - JOIN UNNEST($1::BIGINT[], $2::REAL[], $3::INTEGER[]) \ - AS candidate(candidate_id, candidate_score, candidate_group) \ + JOIN UNNEST($1::BIGINT[], $2::BIGINT[], $3::REAL[], $4::INTEGER[]) \ + AS candidate(candidate_id, candidate_revision, candidate_score, candidate_group) \ ON candidate_id = positioned.id \ - ORDER BY {} LIMIT $4 OFFSET $5", + AND candidate_revision = positioned.row_revision \ + ORDER BY {} LIMIT $5 OFFSET $6", qualified_table, ranked_order_clause(&resolved_order, order.direction), ); @@ -1429,6 +1487,10 @@ async fn fetch_ordered_candidate_rows( .iter() .map(|candidate| candidate.score) .collect::>(); + let revisions = candidates + .iter() + .map(|candidate| candidate.row_revision) + .collect::>(); let best_score = scores .iter() .copied() @@ -1440,6 +1502,7 @@ async fn fetch_ordered_candidate_rows( .collect::>(); let rows = sqlx::query(AssertSqlSafe(sql)) .bind(&ids) + .bind(&revisions) .bind(&scores) .bind(&groups) .bind(limit as i64) @@ -1510,6 +1573,8 @@ impl Searcher for SearcherService { mod tests { use super::*; use common::proto::komp_ac::search::{SearchOrder, SearchOrderDirection}; + use common::search::create_search_schema; + use tantivy::query::AllQuery; #[test] fn search_response_mapping_exposes_aliases_only() { @@ -1524,6 +1589,24 @@ mod tests { assert_eq!(mapped, serde_json::json!({"customer": "Acme", "id": 4})); } + #[test] + fn count_collector_returns_zero_documents_for_an_empty_index() { + let schema = create_search_schema(); + let index = Index::create_in_ram(schema.clone()); + let profile = ProfileIndex { + reader: index.reader().expect("empty reader should open"), + fields: SchemaFields::from(&schema).expect("search schema should match"), + index, + }; + + assert_eq!( + top_documents_for_count(&profile.reader.searcher(), &AllQuery) + .expect("empty count should not construct a zero-limit collector") + .len(), + 0, + ); + } + #[test] fn search_response_mapping_fails_instead_of_leaking_a_real_name() { let error = remap_json_to_display_names( diff --git a/search/src/query_builder.rs b/search/src/query_builder.rs index 4324a332..71980bcb 100644 --- a/search/src/query_builder.rs +++ b/search/src/query_builder.rs @@ -1,11 +1,11 @@ use common::search::{ - ARCHIVED_ROW_KEY_PREFIX, SchemaFields, json_path_term, normalize_column_name, - normalize_exact, tokenize_ngram, tokenize_word, + SchemaFields, json_path_term, normalize_column_name, normalize_exact, tokenize_ngram, + tokenize_word, }; use common::proto::komp_ac::search::SearchVersionScope; use tantivy::query::{ BooleanQuery, BoostQuery, EmptyQuery, FuzzyTermQuery, Occur, PhraseQuery, Query, QueryParser, - RegexQuery, TermQuery, + TermQuery, }; use tantivy::schema::{IndexRecordOption, Term}; use tantivy::Index; @@ -62,14 +62,15 @@ pub fn build_master_query( )); } - let archived_rows = RegexQuery::from_pattern( - &format!("{ARCHIVED_ROW_KEY_PREFIX}.*"), - fields.row_key, - ) - .map_err(|error| Status::internal(format!("Archived-row query build failed: {error}")))?; + let archived_rows = |archived| { + Box::new(TermQuery::new( + Term::from_field_u64(fields.is_archived, u64::from(archived)), + IndexRecordOption::Basic, + )) as Box + }; match version_scope { - SearchVersionScope::Current => clauses.push((Occur::MustNot, Box::new(archived_rows))), - SearchVersionScope::Archived => clauses.push((Occur::Must, Box::new(archived_rows))), + SearchVersionScope::Current => clauses.push((Occur::Must, archived_rows(false))), + SearchVersionScope::Archived => clauses.push((Occur::Must, archived_rows(true))), SearchVersionScope::All => {} } @@ -328,6 +329,17 @@ mod tests { for source in documents { let mut document = TantivyDocument::default(); document.add_u64(fields.pg_id, source.row_id); + document.add_u64(fields.row_revision, 1); + let archive = common::search::parse_archived_search_row_key(&source.row_key); + document.add_u64(fields.is_archived, u64::from(archive.is_some())); + document.add_u64( + fields.table_definition_id, + archive.map(|item| item.0 as u64).unwrap_or_default(), + ); + document.add_u64( + fields.version_number, + archive.map(|item| item.2 as u64).unwrap_or_default(), + ); document.add_text(fields.table_name, source.table_name); document.add_text(fields.row_key, &source.row_key); let mut object = std::collections::BTreeMap::new(); diff --git a/server b/server index 3b7914b5..714104f7 160000 --- a/server +++ b/server @@ -1 +1 @@ -Subproject commit 3b7914b5e69dd56b9246ba8b1d390fd1668dc0da +Subproject commit 714104f7605806bf407337fa639418f350d15e60