diff --git a/Cargo.lock b/Cargo.lock index ed705d32..47f3ce03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6474,6 +6474,7 @@ dependencies = [ "anyhow", "common", "prost", + "rust_decimal", "serde", "serde_json", "sqlx", diff --git a/common/proto/search.proto b/common/proto/search.proto index edf8eb7d..8e4f5448 100644 --- a/common/proto/search.proto +++ b/common/proto/search.proto @@ -4,6 +4,11 @@ package komp_ac.search; service Searcher { rpc Search(SearchRequest) returns (SearchResponse); + rpc Count(SearchRequest) returns (SearchCountResponse); +} + +message SearchCountResponse { + uint64 count = 1; } enum MatchMode { @@ -26,6 +31,16 @@ message SearchRequest { optional uint32 limit = 5; optional uint32 offset = 6; optional SearchOrder order = 7; + // Current rows are the default so pickers and relationship counts never + // select historical records. Callers may explicitly search immutable row + // archives as well. + SearchVersionScope version_scope = 8; +} + +enum SearchVersionScope { + SEARCH_VERSION_SCOPE_CURRENT = 0; + SEARCH_VERSION_SCOPE_ARCHIVED = 1; + SEARCH_VERSION_SCOPE_ALL = 2; } message SearchOrder { @@ -48,6 +63,10 @@ message SearchResponse { repeated string row_display_values = 5; repeated string row_display_columns = 6; optional uint64 position = 7; + // Version 0 means the source predates managed row versioning. New managed + // rows always return a positive version. + int64 version_number = 8; + bool archived = 9; } repeated Hit hits = 1; } diff --git a/common/src/proto/descriptor.bin b/common/src/proto/descriptor.bin index 47496694..aaf51e5f 100644 Binary files a/common/src/proto/descriptor.bin and b/common/src/proto/descriptor.bin differ diff --git a/common/src/proto/komp_ac.search.rs b/common/src/proto/komp_ac.search.rs index 1a7b0736..e0f5bcc9 100644 --- a/common/src/proto/komp_ac.search.rs +++ b/common/src/proto/komp_ac.search.rs @@ -1,4 +1,9 @@ // This file is @generated by prost-build. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SearchCountResponse { + #[prost(uint64, tag = "1")] + pub count: u64, +} #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ColumnConstraint { #[prost(string, tag = "1")] @@ -24,6 +29,11 @@ pub struct SearchRequest { pub offset: ::core::option::Option, #[prost(message, optional, tag = "7")] pub order: ::core::option::Option, + /// Current rows are the default so pickers and relationship counts never + /// select historical records. Callers may explicitly search immutable row + /// archives as well. + #[prost(enumeration = "SearchVersionScope", tag = "8")] + pub version_scope: i32, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct SearchOrder { @@ -59,6 +69,12 @@ pub mod search_response { >, #[prost(uint64, optional, tag = "7")] pub position: ::core::option::Option, + /// Version 0 means the source predates managed row versioning. New managed + /// rows always return a positive version. + #[prost(int64, tag = "8")] + pub version_number: i64, + #[prost(bool, tag = "9")] + pub archived: bool, } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] @@ -92,6 +108,35 @@ impl MatchMode { } #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] +pub enum SearchVersionScope { + Current = 0, + Archived = 1, + All = 2, +} +impl SearchVersionScope { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Current => "SEARCH_VERSION_SCOPE_CURRENT", + Self::Archived => "SEARCH_VERSION_SCOPE_ARCHIVED", + Self::All => "SEARCH_VERSION_SCOPE_ALL", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "SEARCH_VERSION_SCOPE_CURRENT" => Some(Self::Current), + "SEARCH_VERSION_SCOPE_ARCHIVED" => Some(Self::Archived), + "SEARCH_VERSION_SCOPE_ALL" => Some(Self::All), + _ => None, + } + } +} +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] pub enum SearchOrderDirection { Unspecified = 0, Asc = 1, @@ -231,6 +276,30 @@ pub mod searcher_client { .insert(GrpcMethod::new("komp_ac.search.Searcher", "Search")); self.inner.unary(req, path, codec).await } + pub async fn count( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/komp_ac.search.Searcher/Count", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("komp_ac.search.Searcher", "Count")); + self.inner.unary(req, path, codec).await + } } } /// Generated server implementations. @@ -250,6 +319,13 @@ pub mod searcher_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn count( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } #[derive(Debug)] pub struct SearcherServer { @@ -370,6 +446,49 @@ pub mod searcher_server { }; Box::pin(fut) } + "/komp_ac.search.Searcher/Count" => { + #[allow(non_camel_case_types)] + struct CountSvc(pub Arc); + impl tonic::server::UnaryService + for CountSvc { + type Response = super::SearchCountResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::count(&inner, request).await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = CountSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } _ => { Box::pin(async move { let mut response = http::Response::new( diff --git a/common/src/search.rs b/common/src/search.rs index 255ec3ae..8eecd1be 100644 --- a/common/src/search.rs +++ b/common/src/search.rs @@ -18,6 +18,17 @@ pub const F_DATA_WORD: &str = "data_word"; 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"; + +/// Root for the current on-disk index format. The format component prevents +/// an executable with new projection semantics from opening old documents. +pub fn search_index_root() -> PathBuf { + std::env::var_os("TANTIVY_INDEX_DIR") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("./tantivy_indexes")) + .join(SEARCH_INDEX_FORMAT_DIRECTORY) +} pub const TOK_WORD: &str = "kw_word"; pub const TOK_NGRAM: &str = "kw_ngram"; @@ -33,6 +44,30 @@ pub fn search_row_key(table_name: &str, row_id: i64) -> String { format!("{}:{}", table_name, row_id) } +/// Returns the unique index key for an immutable archived row version. +pub fn archived_search_row_key( + table_definition_id: i64, + row_id: i64, + version_number: i64, +) -> String { + format!( + "{ARCHIVED_ROW_KEY_PREFIX}{table_definition_id}:{row_id}:{version_number}" + ) +} + +/// Decodes an archived index key into (table definition, row, version). +pub fn parse_archived_search_row_key(row_key: &str) -> Option<(i64, i64, i64)> { + let encoded = row_key.strip_prefix(ARCHIVED_ROW_KEY_PREFIX)?; + let mut parts = encoded.split(':'); + let table_definition_id = parts.next()?.parse().ok()?; + let row_id = parts.next()?.parse().ok()?; + let version_number = parts.next()?.parse().ok()?; + if parts.next().is_some() { + return None; + } + Some((table_definition_id, row_id, version_number)) +} + /// Normalizes user-entered values for exact-mode terms. pub fn normalize_exact(input: &str) -> String { let trimmed = input.trim(); @@ -184,6 +219,19 @@ impl SchemaFields { } } +#[cfg(test)] +mod tests { + use super::{archived_search_row_key, parse_archived_search_row_key}; + + #[test] + fn archived_row_key_round_trips_stable_identity() { + let key = archived_search_row_key(42, 7, 3); + assert_eq!(parse_archived_search_row_key(&key), Some((42, 7, 3))); + assert_eq!(parse_archived_search_row_key("customers:7"), None); + assert_eq!(parse_archived_search_row_key("__archive__:42:7"), None); + } +} + fn get_field(schema: &Schema, name: &str) -> tantivy::Result { schema.get_field(name).map_err(|e| { tantivy::TantivyError::SchemaError(format!("schema is missing field '{name}': {e}")) diff --git a/search/Cargo.toml b/search/Cargo.toml index 6de946fd..a7b90850 100644 --- a/search/Cargo.toml +++ b/search/Cargo.toml @@ -9,6 +9,7 @@ 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 fb772ee9..23585e2d 100644 --- a/search/src/lib.rs +++ b/search/src/lib.rs @@ -7,21 +7,25 @@ use std::sync::{Arc, Mutex}; use common::proto::komp_ac::search::searcher_server::Searcher; pub use common::proto::komp_ac::search::searcher_server::SearcherServer; use common::proto::komp_ac::search::{ - SearchOrderDirection, SearchRequest, SearchResponse, search_response::Hit, + SearchCountResponse, SearchOrderDirection, SearchRequest, SearchResponse, SearchVersionScope, + search_response::Hit, }; -use common::search::{SchemaFields, register_tokenizers, search_index_path}; -use common::system_column::is_system_column; +use common::search::{ + SchemaFields, 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::{ ConstraintMode, SearchConstraint, SearchConstraintTarget, build_master_query, }; use sqlx::{AssertSqlSafe, PgPool, Row}; use tantivy::collector::TopDocs; +use tantivy::query::Query; use tantivy::schema::Value; use tantivy::{Index, IndexReader, ReloadPolicy, TantivyDocument}; use tonic::{Request, Response, Status}; use tracing::info; -const INDEX_ROOT: &str = "./tantivy_indexes"; const DEFAULT_RESULT_LIMIT: usize = 60; const HARD_RESULT_LIMIT: usize = 200; const DEFAULT_LIST_LIMIT: usize = 5; @@ -65,6 +69,11 @@ impl SearcherService { } if !normalized.has_input() { + if normalized.version_scope != SearchVersionScope::Current { + return Err(Status::invalid_argument( + "archived searches require text or a column constraint", + )); + } let Some(table_name) = normalized.table_name.as_deref() else { return Err(Status::invalid_argument( "table_name is required when query is empty", @@ -83,7 +92,10 @@ impl SearcherService { return Ok(Response::new(SearchResponse { hits })); } - let index_path = search_index_path(Path::new(INDEX_ROOT), &normalized.profile_name); + let index_path = search_index_path( + &common::search::search_index_root(), + &normalized.profile_name, + ); if !index_path.exists() { return Err(Status::not_found(format!( "No search index found for profile '{}'", @@ -100,6 +112,10 @@ 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, @@ -110,6 +126,7 @@ impl SearcherService { normalized.limit.unwrap_or(DEFAULT_RESULT_LIMIT), normalized.offset, normalized.order.as_ref(), + normalized.version_scope, ) .await?; @@ -133,6 +150,163 @@ impl SearcherService { Ok(Response::new(SearchResponse { hits })) } + + async fn run_count_rpc( + &self, + request: Request, + ) -> Result, Status> { + let normalized = normalize_request(request.into_inner())?; + let table_name = normalized.table_name.as_deref().ok_or_else(|| { + Status::invalid_argument("table_name is required when counting search results") + })?; + if !normalized.has_input() { + return Err(Status::invalid_argument( + "counting search results requires text or a column constraint", + )); + } + if !profile_exists(&self.pool, &normalized.profile_name).await? + || !table_exists(&self.pool, &normalized.profile_name, table_name).await? + { + return Err(Status::not_found("Search table was not found")); + } + let index_path = search_index_path( + &common::search::search_index_root(), + &normalized.profile_name, + ); + if !index_path.exists() { + return Err(Status::not_found(format!( + "No search index found for profile '{}'", + normalized.profile_name + ))); + } + let constraints = resolve_constraints( + &self.pool, + &normalized.profile_name, + Some(table_name), + &normalized.must, + ) + .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, + &normalized.free_query, + &constraints, + Some(table_name), + normalized.version_scope, + )?; + let count = count_authoritative_matches( + &self.pool, + &profile, + &normalized.profile_name, + table_name, + &*query, + ) + .await?; + Ok(Response::new(SearchCountResponse { count })) + } +} + +async fn count_authoritative_matches( + pool: &PgPool, + profile: &ProfileIndex, + profile_name: &str, + table_name: &str, + 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 best_score = documents + .first() + .map(|(score, _)| *score) + .unwrap_or_default(); + let score_floor = if best_score > 0.0 { + best_score * SEARCH_SCORE_RELATIVE_FLOOR + } else { + 0.0 + }; + let mut current_ids = 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; + }; + 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); + } + } + current_ids.sort_unstable(); + current_ids.dedup(); + archives.sort_unstable(); + archives.dedup(); + + let current_count = if current_ids.is_empty() { + 0 + } else { + let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; + sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( + "SELECT COUNT(*) FROM {qualified_table} WHERE deleted = FALSE AND id = ANY($1)" + ))) + .bind(¤t_ids) + .fetch_one(pool) + .await + .map_err(|error| Status::internal(format!("Current search count validation failed: {error}")))? + }; + let archive_count = if archives.is_empty() { + 0 + } else { + let definition_ids = archives.iter().map(|item| item.0).collect::>(); + let row_ids = archives.iter().map(|item| item.1).collect::>(); + let versions = archives.iter().map(|item| item.2).collect::>(); + sqlx::query_scalar::<_, i64>( + r#"SELECT COUNT(*) + FROM UNNEST($1::BIGINT[], $2::BIGINT[], $3::BIGINT[]) + AS requested(table_definition_id, source_row_id, version_number) + JOIN table_row_archives archive + ON archive.table_definition_id = requested.table_definition_id + AND archive.source_row_id = requested.source_row_id + AND archive.version_number = requested.version_number + JOIN table_definitions definition ON definition.id = archive.table_definition_id + JOIN schemas owner ON owner.id = definition.schema_id + WHERE definition.table_name = $4 + AND definition.deleted = FALSE + AND (owner.name = $5 OR definition.is_global = TRUE)"#, + ) + .bind(&definition_ids) + .bind(&row_ids) + .bind(&versions) + .bind(table_name) + .bind(profile_name) + .fetch_one(pool) + .await + .map_err(|error| Status::internal(format!("Archived search count validation failed: {error}")))? + }; + u64::try_from(current_count.saturating_add(archive_count)) + .map_err(|_| Status::internal("Search count was negative")) } struct ProfileIndex { @@ -176,6 +350,7 @@ struct NormalizedSearchRequest { limit: Option, offset: usize, order: Option, + version_scope: SearchVersionScope, } #[derive(Debug)] @@ -191,6 +366,45 @@ struct NormalizedColumnConstraint { mode: ConstraintMode, } +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()) +} + +fn public_search_system_type(column: &str) -> Option<&'static str> { + match column { + "id" | "row_revision" => Some("bigint"), + "created_at" => Some("instant"), + _ => None, + } +} + +#[derive(Clone, Debug)] +struct SearchCandidate { + score: f32, + row_id: i64, + table_name: String, + row_key: String, +} + impl NormalizedSearchRequest { fn has_input(&self) -> bool { !self.free_query.is_empty() || !self.must.is_empty() @@ -278,7 +492,9 @@ async fn table_exists(pool: &PgPool, profile_name: &str, table_name: &str) -> Re SELECT 1 FROM table_definitions td JOIN schemas s ON td.schema_id = s.id - WHERE s.name = $1 AND td.table_name = $2 + WHERE td.table_name = $2 + AND td.deleted = FALSE + AND (s.name = $1 OR td.is_global = TRUE) ) "#, ) @@ -290,6 +506,30 @@ async fn table_exists(pool: &PgPool, profile_name: &str, table_name: &str) -> Re Ok(exists) } +async fn qualified_visible_table( + pool: &PgPool, + profile_name: &str, + table_name: &str, +) -> Result { + let storage_schema = sqlx::query_scalar::<_, String>( + r#"SELECT owner.name + FROM table_definitions definition + JOIN schemas owner ON owner.id = definition.schema_id + WHERE definition.table_name = $2 + AND definition.deleted = FALSE + AND (owner.name = $1 OR definition.is_global = TRUE) + ORDER BY definition.is_global ASC + LIMIT 1"#, + ) + .bind(profile_name) + .bind(table_name) + .fetch_optional(pool) + .await + .map_err(|error| Status::internal(format!("Table storage lookup failed: {error}")))? + .ok_or_else(|| Status::not_found(format!("Table '{table_name}' was not found")))?; + Ok(qualify_profile_table(&storage_schema, table_name)) +} + fn normalize_request(req: SearchRequest) -> Result { let profile_name = req.profile_name.trim(); if profile_name.is_empty() { @@ -353,11 +593,19 @@ fn normalize_request(req: SearchRequest) -> Result Result Result, Status> { let mut resolved = Vec::with_capacity(must.len()); for constraint in must { - let targets = resolve_constraint_targets(pool, profile_name, table_filter, &constraint.column) - .await?; + let targets = resolve_constraint_targets( + pool, + profile_name, + table_filter, + &constraint.column, + &constraint.query, + constraint.mode, + ) + .await?; resolved.push(SearchConstraint { targets, - query: constraint.query.clone(), mode: constraint.mode, }); } @@ -402,16 +657,26 @@ async fn resolve_constraint_targets( profile_name: &str, table_filter: Option<&str>, column: &str, + query: &str, + mode: ConstraintMode, ) -> Result, Status> { let rows = if let Some(table_name) = table_filter { sqlx::query( r#" - SELECT td.table_name, tdc.physical_name + SELECT td.table_name, tdc.physical_name, tdc.field_type FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id - WHERE s.name = $1 - AND td.table_name = $2 + WHERE td.id = ( + SELECT visible.id + FROM table_definitions visible + JOIN schemas owner ON owner.id = visible.schema_id + WHERE visible.table_name = $2 + AND visible.deleted = FALSE + AND (owner.name = $1 OR visible.is_global = TRUE) + ORDER BY visible.is_global ASC + LIMIT 1 + ) AND tdc.display_name = $3 "#, ) @@ -423,11 +688,12 @@ async fn resolve_constraint_targets( } else { sqlx::query( r#" - SELECT td.table_name, tdc.physical_name + SELECT td.table_name, tdc.physical_name, tdc.field_type FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id - WHERE s.name = $1 + WHERE (s.name = $1 OR td.is_global = TRUE) + AND td.deleted = FALSE AND tdc.display_name = $2 "#, ) @@ -438,15 +704,49 @@ async fn resolve_constraint_targets( } .map_err(|e| Status::internal(format!("Column mapping lookup failed: {}", e)))?; - if rows.is_empty() { - return Err(Status::invalid_argument(format!( - "Column alias '{}' was not found{}", - column, - table_filter - .map(|table_name| format!(" in table '{}'", table_name)) - .unwrap_or_default() - ))); - } + let rows = if rows.is_empty() { + let Some(field_type) = public_search_system_type(column) else { + return Err(Status::invalid_argument(format!( + "Column alias '{}' was not found{}", + column, + table_filter + .map(|table_name| format!(" in table '{}'", table_name)) + .unwrap_or_default() + ))); + }; + let table_names = if let Some(table_name) = table_filter { + vec![table_name.to_string()] + } else { + sqlx::query_scalar::<_, String>( + r#"SELECT DISTINCT definition.table_name + FROM table_definitions definition + JOIN schemas owner ON owner.id = definition.schema_id + WHERE definition.deleted = FALSE + AND (owner.name = $1 OR definition.is_global = TRUE)"#, + ) + .bind(profile_name) + .fetch_all(pool) + .await + .map_err(|error| Status::internal(format!( + "System-column search table lookup failed: {error}" + )))? + }; + let query = if mode == ConstraintMode::Exact { + normalize_constraint_value(query, field_type)? + } else { + query.to_string() + }; + return Ok(table_names + .into_iter() + .map(|table_name| SearchConstraintTarget { + table_name: table_filter.is_none().then_some(table_name), + column: column.to_string(), + query: query.clone(), + }) + .collect()); + } else { + rows + }; let mut seen = HashSet::new(); let mut targets = Vec::new(); @@ -457,6 +757,9 @@ async fn resolve_constraint_targets( let physical_name: String = row .try_get("physical_name") .map_err(|e| Status::internal(format!("Column mapping physical read failed: {}", e)))?; + let field_type: String = row + .try_get("field_type") + .map_err(|e| Status::internal(format!("Column mapping type read failed: {e}")))?; let key = (table_name.clone(), physical_name.clone()); if seen.insert(key) { targets.push(SearchConstraintTarget { @@ -466,6 +769,11 @@ async fn resolve_constraint_targets( Some(table_name) }, column: physical_name, + query: if mode == ConstraintMode::Exact { + normalize_constraint_value(query, &field_type)? + } else { + query.to_string() + }, }); } } @@ -484,8 +792,16 @@ async fn table_physical_to_display_map( FROM schemas s JOIN table_definitions td ON td.schema_id = s.id JOIN table_definition_columns tdc ON tdc.table_definition_id = td.id - WHERE s.name = $1 - AND td.table_name = $2 + WHERE td.id = ( + SELECT visible.id + FROM table_definitions visible + JOIN schemas owner ON owner.id = visible.schema_id + WHERE visible.table_name = $2 + AND visible.deleted = FALSE + AND (owner.name = $1 OR visible.is_global = TRUE) + ORDER BY visible.is_global ASC + LIMIT 1 + ) "#, ) .bind(profile_name) @@ -507,6 +823,43 @@ async fn table_physical_to_display_map( Ok(mapping) } +async fn table_internal_column_names( + pool: &PgPool, + profile_name: &str, + table_name: &str, +) -> Result, Status> { + let rows = sqlx::query_scalar::<_, String>( + r#" + SELECT link.version_column_name + FROM schemas s + JOIN table_definitions td ON td.schema_id = s.id + LEFT JOIN table_definition_links link ON link.source_table_id = td.id + WHERE td.id = ( + SELECT visible.id + FROM table_definitions visible + JOIN schemas owner ON owner.id = visible.schema_id + WHERE visible.table_name = $2 + AND visible.deleted = FALSE + AND (owner.name = $1 OR visible.is_global = TRUE) + ORDER BY visible.is_global ASC + LIMIT 1 + ) + AND link.version_column_name IS NOT NULL + "#, + ) + .bind(profile_name) + .bind(table_name) + .fetch_all(pool) + .await + .map_err(|e| Status::internal(format!("Internal column lookup failed: {e}")))?; + + let mut internal = internal_column_names() + .map(str::to_string) + .collect::>(); + internal.extend(rows); + Ok(internal) +} + async fn table_row_display_columns( pool: &PgPool, profile_name: &str, @@ -517,7 +870,16 @@ async fn table_row_display_columns( SELECT td.row_display_columns FROM schemas s JOIN table_definitions td ON td.schema_id = s.id - WHERE s.name = $1 AND td.table_name = $2 + WHERE td.id = ( + SELECT visible.id + FROM table_definitions visible + JOIN schemas owner ON owner.id = visible.schema_id + WHERE visible.table_name = $2 + AND visible.deleted = FALSE + AND (owner.name = $1 OR visible.is_global = TRUE) + ORDER BY visible.is_global ASC + LIMIT 1 + ) "#, ) .bind(profile_name) @@ -530,11 +892,15 @@ async fn table_row_display_columns( fn remap_json_to_display_names( value: serde_json::Value, physical_to_display: &HashMap, + internal_columns: &HashSet, ) -> Result { match value { serde_json::Value::Object(object) => { let mut remapped = serde_json::Map::with_capacity(object.len()); for (key, value) in object { + if internal_columns.contains(&key) { + continue; + } let final_key = match physical_to_display.get(&key) { Some(display_name) => display_name.clone(), None if is_system_column(&key) => key, @@ -607,7 +973,10 @@ async fn resolve_order_column( .find(|(_, display)| display.eq_ignore_ascii_case(&requested_column)) .map(|(physical, _)| physical.clone()) .or_else(|| { - is_system_column(&requested_column).then(|| requested_column.clone()) + (is_system_column(&requested_column) + && !is_internal_column(&requested_column) + && !physical_to_display.contains_key(&requested_column)) + .then(|| requested_column.clone()) }) .ok_or_else(|| { Status::invalid_argument(format!( @@ -620,7 +989,18 @@ async fn resolve_order_column( r#" SELECT column_name FROM information_schema.columns - WHERE table_schema = $1 AND table_name = $2 AND LOWER(column_name) = LOWER($3) + WHERE table_schema = ( + SELECT owner.name + FROM table_definitions definition + JOIN schemas owner ON owner.id = definition.schema_id + WHERE definition.table_name = $2 + AND definition.deleted = FALSE + AND (owner.name = $1 OR definition.is_global = TRUE) + ORDER BY definition.is_global ASC + LIMIT 1 + ) + AND table_name = $2 + AND LOWER(column_name) = LOWER($3) "#, ) .bind(profile_name) @@ -688,6 +1068,7 @@ async fn fetch_ordered_rows( offset: usize, ) -> Result, Status> { let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?; + let internal_columns = table_internal_column_names(pool, profile_name, table_name).await?; let display_columns = table_row_display_columns(pool, profile_name, table_name).await?; let (resolved_order, direction) = match order { Some(order) => ( @@ -696,6 +1077,7 @@ async fn fetch_ordered_rows( ), None => (ResolvedOrderColumn::Position, SearchOrderDirection::Desc), }; + let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let sql = format!( "WITH positioned AS (\ SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \ @@ -703,7 +1085,7 @@ async fn fetch_ordered_rows( ) \ SELECT id, to_jsonb(positioned) - 'picker_position' AS data, picker_position \ FROM positioned ORDER BY {} LIMIT $1 OFFSET $2", - qualify_profile_table(profile_name, table_name), + qualified_table, order_clause(&resolved_order, direction), ); @@ -726,7 +1108,15 @@ async fn fetch_ordered_rows( let position: i64 = row.try_get("picker_position").map_err(|error| { Status::internal(format!("Search position read failed: {}", error)) })?; - let json_data = remap_json_to_display_names(json_data, &physical_to_display)?; + let version_number = json_data + .get("version") + .and_then(|value| value.as_i64()) + .unwrap_or(0); + let json_data = remap_json_to_display_names( + json_data, + &physical_to_display, + &internal_columns, + )?; let row_display_values = row_display_values(&json_data, &display_columns); Ok(Hit { id, @@ -736,6 +1126,8 @@ async fn fetch_ordered_rows( row_display_values, row_display_columns: display_columns.clone(), position: u64::try_from(position).ok(), + version_number, + archived: false, }) }) .collect() @@ -751,6 +1143,7 @@ async fn run_search( limit: usize, offset: usize, order: Option<&NormalizedSearchOrder>, + version_scope: SearchVersionScope, ) -> Result, Status> { let master_query = build_master_query( &profile.index, @@ -758,6 +1151,7 @@ async fn run_search( free_query, must, table_filter, + version_scope, )?; let searcher = profile.reader.searcher(); @@ -797,7 +1191,7 @@ async fn run_search( .collect::>() }; - let mut candidates: Vec<(f32, i64, String)> = Vec::with_capacity(page_docs.len()); + let mut candidates: Vec = Vec::with_capacity(page_docs.len()); for (score, doc_address) in page_docs { let doc: TantivyDocument = searcher .doc(doc_address) @@ -814,7 +1208,18 @@ async fn run_search( else { continue; }; - candidates.push((score, pg_id as i64, table_name.to_string())); + let Some(row_key) = doc + .get_first(profile.fields.row_key) + .and_then(|value| value.as_str()) + else { + continue; + }; + candidates.push(SearchCandidate { + score, + row_id: pg_id as i64, + table_name: table_name.to_string(), + row_key: row_key.to_string(), + }); } if candidates.is_empty() { @@ -836,23 +1241,29 @@ async fn run_search( } let mut ids_by_table: HashMap> = HashMap::new(); - for (_, pg_id, table_name) in &candidates { + for candidate in &candidates { + if parse_archived_search_row_key(&candidate.row_key).is_some() { + continue; + } ids_by_table - .entry(table_name.clone()) + .entry(candidate.table_name.clone()) .or_default() - .push(*pg_id); + .push(candidate.row_id); } - let mut content_map: HashMap<(String, i64), (String, Vec, Vec)> = + let mut content_map: HashMap, Vec, i64, bool)> = HashMap::new(); for (table_name, pg_ids) in ids_by_table { validate_identifier(&table_name, "table_name")?; let physical_to_display = table_physical_to_display_map(pool, profile_name, &table_name).await?; + let internal_columns = + 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 sql = format!( "SELECT id, to_jsonb(t) AS data FROM {} t WHERE deleted = FALSE AND id = ANY($1)", - qualify_profile_table(profile_name, &table_name) + qualified_table ); let rows = sqlx::query(AssertSqlSafe(sql)) .bind(&pg_ids) @@ -863,28 +1274,118 @@ 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 json_data = remap_json_to_display_names(json_data, &physical_to_display)?; + 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, + &physical_to_display, + &internal_columns, + )?; let display_values = row_display_values(&json_data, &display_columns); - content_map.insert( - (table_name.clone(), id), - (json_data.to_string(), display_values, display_columns.clone()), + content_map.insert(search_row_key(&table_name, id), ( + json_data.to_string(), + display_values, + display_columns.clone(), + version, + false, + )); + } + } + + let mut archives_by_table: HashMap> = HashMap::new(); + for candidate in &candidates { + if let Some((table_definition_id, row_id, version)) = + parse_archived_search_row_key(&candidate.row_key) + { + archives_by_table + .entry(candidate.table_name.clone()) + .or_default() + .push((table_definition_id, row_id, version)); + } + } + for (table_name, archives) in archives_by_table { + let physical_to_display = + table_physical_to_display_map(pool, profile_name, &table_name).await?; + let internal_columns = + table_internal_column_names(pool, profile_name, &table_name).await?; + let display_columns = table_row_display_columns(pool, profile_name, &table_name).await?; + let table_definition_ids = archives.iter().map(|item| item.0).collect::>(); + let row_ids = archives.iter().map(|item| item.1).collect::>(); + let versions = archives.iter().map(|item| item.2).collect::>(); + let rows = sqlx::query( + r#"SELECT archive.table_definition_id, archive.source_row_id, + archive.version_number, archive.snapshot + FROM UNNEST($1::BIGINT[], $2::BIGINT[], $3::BIGINT[]) + AS requested(table_definition_id, source_row_id, version_number) + JOIN table_row_archives archive + ON archive.table_definition_id = requested.table_definition_id + AND archive.source_row_id = requested.source_row_id + AND archive.version_number = requested.version_number + JOIN table_definitions definition + ON definition.id = archive.table_definition_id + JOIN schemas owner ON owner.id = definition.schema_id + WHERE definition.table_name = $4 + AND definition.deleted = FALSE + AND (owner.name = $5 OR definition.is_global = TRUE)"#, + ) + .bind(&table_definition_ids) + .bind(&row_ids) + .bind(&versions) + .bind(&table_name) + .bind(profile_name) + .fetch_all(pool) + .await + .map_err(|error| Status::internal(format!( + "Archived search row lookup failed: {error}" + )))?; + for row in rows { + let table_definition_id: i64 = row.try_get("table_definition_id").map_err(|error| { + Status::internal(format!("Archived table id read failed: {error}")) + })?; + let row_id: i64 = row.try_get("source_row_id").map_err(|error| { + Status::internal(format!("Archived row id read failed: {error}")) + })?; + let version: i64 = row.try_get("version_number").map_err(|error| { + Status::internal(format!("Archived version read failed: {error}")) + })?; + let snapshot: serde_json::Value = row.try_get("snapshot").map_err(|error| { + Status::internal(format!("Archived snapshot read failed: {error}")) + })?; + let row_key = common::search::archived_search_row_key( + table_definition_id, + row_id, + version, ); + let json_data = remap_json_to_display_names( + snapshot, + &physical_to_display, + &internal_columns, + )?; + let display_values = row_display_values(&json_data, &display_columns); + content_map.insert(row_key, ( + json_data.to_string(), + display_values, + display_columns.clone(), + version, + true, + )); } } Ok(candidates .into_iter() - .filter_map(|(score, pg_id, table_name)| { + .filter_map(|candidate| { content_map - .get(&(table_name.clone(), pg_id)) - .map(|(content_json, row_display_values, row_display_columns)| Hit { - id: pg_id, - score, + .get(&candidate.row_key) + .map(|(content_json, row_display_values, row_display_columns, version_number, archived)| Hit { + id: candidate.row_id, + score: candidate.score, content_json: content_json.clone(), - table_name, + table_name: candidate.table_name, row_display_values: row_display_values.clone(), row_display_columns: row_display_columns.clone(), position: None, + version_number: *version_number, + archived: *archived, }) }) .collect()) @@ -894,15 +1395,17 @@ async fn fetch_ordered_candidate_rows( pool: &PgPool, profile_name: &str, table_name: &str, - candidates: &[(f32, i64, String)], + candidates: &[SearchCandidate], order: &NormalizedSearchOrder, limit: usize, offset: usize, ) -> Result, Status> { let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?; + let internal_columns = table_internal_column_names(pool, profile_name, table_name).await?; let display_columns = table_row_display_columns(pool, profile_name, table_name).await?; let resolved_order = resolve_order_column(pool, profile_name, table_name, &order.column).await?; + let qualified_table = qualified_visible_table(pool, profile_name, table_name).await?; let sql = format!( "WITH positioned AS (\ SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \ @@ -915,16 +1418,16 @@ async fn fetch_ordered_candidate_rows( AS candidate(candidate_id, candidate_score, candidate_group) \ ON candidate_id = positioned.id \ ORDER BY {} LIMIT $4 OFFSET $5", - qualify_profile_table(profile_name, table_name), + qualified_table, ranked_order_clause(&resolved_order, order.direction), ); let ids = candidates .iter() - .map(|(_, id, _)| *id) + .map(|candidate| candidate.row_id) .collect::>(); let scores = candidates .iter() - .map(|(score, _, _)| *score) + .map(|candidate| candidate.score) .collect::>(); let best_score = scores .iter() @@ -960,7 +1463,15 @@ async fn fetch_ordered_candidate_rows( let score: f32 = row.try_get("candidate_score").map_err(|error| { Status::internal(format!("Search score read failed: {}", error)) })?; - let json_data = remap_json_to_display_names(json_data, &physical_to_display)?; + let version_number = json_data + .get("version") + .and_then(|value| value.as_i64()) + .unwrap_or(0); + let json_data = remap_json_to_display_names( + json_data, + &physical_to_display, + &internal_columns, + )?; let display_values = row_display_values(&json_data, &display_columns); Ok(Hit { id, @@ -970,6 +1481,8 @@ async fn fetch_ordered_candidate_rows( row_display_values: display_values, row_display_columns: display_columns.clone(), position: u64::try_from(position).ok(), + version_number, + archived: false, }) }) .collect() @@ -983,6 +1496,14 @@ impl Searcher for SearcherService { ) -> Result, Status> { self.run_rpc(request).await } + + + async fn count( + &self, + request: Request, + ) -> Result, Status> { + self.run_count_rpc(request).await + } } #[cfg(test)] @@ -996,6 +1517,7 @@ mod tests { let mapped = remap_json_to_display_names( serde_json::json!({"1": "Acme", "id": 4}), &mappings, + &HashSet::new(), ) .unwrap(); @@ -1007,6 +1529,7 @@ mod tests { let error = remap_json_to_display_names( serde_json::json!({"1": "Acme", "2": "hidden"}), &HashMap::from([("1".to_string(), "customer".to_string())]), + &HashSet::new(), ) .unwrap_err(); @@ -1014,6 +1537,35 @@ mod tests { assert!(!error.message().contains("'2'")); } + #[test] + fn search_response_mapping_omits_registered_internal_columns() { + let mapped = remap_json_to_display_names( + serde_json::json!({ + "1": 7, + "1_version": 3, + "account_id": 11, + "account_id_version": 4, + "version": 5, + "id": 9 + }), + &HashMap::from([ + ("1".to_string(), "adresar".to_string()), + ("account_id".to_string(), "account".to_string()), + ]), + &HashSet::from([ + "1_version".to_string(), + "account_id_version".to_string(), + "version".to_string(), + ]), + ) + .unwrap(); + + assert_eq!( + mapped, + serde_json::json!({"adresar": 7, "account": 11, "id": 9}) + ); + } + fn request(order: Option, table_name: Option<&str>) -> SearchRequest { SearchRequest { profile_name: "finance".to_string(), @@ -1023,6 +1575,7 @@ mod tests { limit: Some(20), offset: Some(40), order, + version_scope: 0, } } diff --git a/search/src/query_builder.rs b/search/src/query_builder.rs index ca4bfd6e..4324a332 100644 --- a/search/src/query_builder.rs +++ b/search/src/query_builder.rs @@ -1,10 +1,11 @@ use common::search::{ - json_path_term, normalize_column_name, normalize_exact, tokenize_ngram, tokenize_word, - SchemaFields, + ARCHIVED_ROW_KEY_PREFIX, 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, - TermQuery, + RegexQuery, TermQuery, }; use tantivy::schema::{IndexRecordOption, Term}; use tantivy::Index; @@ -19,7 +20,6 @@ pub enum ConstraintMode { #[derive(Clone, Debug)] pub struct SearchConstraint { pub targets: Vec, - pub query: String, pub mode: ConstraintMode, } @@ -27,6 +27,7 @@ pub struct SearchConstraint { pub struct SearchConstraintTarget { pub table_name: Option, pub column: String, + pub query: String, } pub fn build_master_query( @@ -35,6 +36,7 @@ pub fn build_master_query( free_query: &str, must: &[SearchConstraint], table_filter: Option<&str>, + version_scope: SearchVersionScope, ) -> Result, Status> { let mut clauses: Vec<(Occur, Box)> = Vec::new(); let mut has_search_clause = false; @@ -60,6 +62,17 @@ 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}")))?; + match version_scope { + SearchVersionScope::Current => clauses.push((Occur::MustNot, Box::new(archived_rows))), + SearchVersionScope::Archived => clauses.push((Occur::Must, Box::new(archived_rows))), + SearchVersionScope::All => {} + } + if !has_search_clause { return Ok(Box::new(EmptyQuery)); } @@ -75,9 +88,9 @@ fn constraint_predicate( for target in &constraint.targets { let column_predicate = match constraint.mode { - ConstraintMode::Exact => exact_predicate(fields, &target.column, &constraint.query)?, + ConstraintMode::Exact => exact_predicate(fields, &target.column, &target.query)?, ConstraintMode::Fuzzy => { - fuzzy_predicate_scoped(fields, &target.column, &constraint.query)? + fuzzy_predicate_scoped(fields, &target.column, &target.query)? } }; @@ -289,3 +302,260 @@ fn fuzzy_distance(word_len: usize) -> Option { _ => Some(2), } } + +#[cfg(test)] +mod tests { + use super::*; + use common::search::{archived_search_row_key, create_search_schema, register_tokenizers}; + use tantivy::collector::Count; + use tantivy::schema::OwnedValue; + use tantivy::TantivyDocument; + + struct TestDocument<'a> { + row_key: String, + row_id: u64, + table_name: &'a str, + values: &'a [(&'a str, &'a str)], + } + + fn index_documents(documents: &[TestDocument<'_>]) -> (Index, SchemaFields) { + let schema = create_search_schema(); + let index = Index::create_in_ram(schema.clone()); + register_tokenizers(&index).expect("tokenizers should register"); + let fields = SchemaFields::from(&schema).expect("schema should match"); + let mut writer = index.writer(50_000_000).expect("writer should open"); + + for source in documents { + let mut document = TantivyDocument::default(); + document.add_u64(fields.pg_id, source.row_id); + 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(); + for (column, value) in source.values { + document.add_text(fields.all_text, value); + object.insert((*column).to_string(), OwnedValue::from(*value)); + } + document.add_object(fields.data_word, object.clone()); + document.add_object(fields.data_ngram, object.clone()); + document.add_object(fields.data_exact, object); + writer.add_document(document).expect("document should index"); + } + writer.commit().expect("documents should commit"); + (index, fields) + } + + fn index_versions() -> (Index, SchemaFields) { + index_documents(&[ + TestDocument { + row_key: "customers:7".to_string(), + row_id: 7, + table_name: "customers", + values: &[("1", "new")], + }, + TestDocument { + row_key: archived_search_row_key(40, 7, 1), + row_id: 7, + table_name: "customers", + values: &[("1", "old")], + }, + ]) + } + + fn count(index: &Index, query: &dyn Query) -> usize { + index + .reader() + .expect("reader should open") + .searcher() + .search(query, &Count) + .expect("query should run") + } + + #[test] + fn version_scope_separates_current_and_archived_documents() { + let (index, fields) = index_versions(); + let constraint = SearchConstraint { + targets: vec![SearchConstraintTarget { + table_name: None, + column: "1".to_string(), + query: "old".to_string(), + }], + mode: ConstraintMode::Exact, + }; + + let current = build_master_query( + &index, + &fields, + "", + std::slice::from_ref(&constraint), + Some("customers"), + SearchVersionScope::Current, + ) + .expect("current query should build"); + let archived = build_master_query( + &index, + &fields, + "", + &[constraint], + Some("customers"), + SearchVersionScope::Archived, + ) + .expect("archive query should build"); + + assert_eq!(count(&index, &*current), 0); + assert_eq!(count(&index, &*archived), 1); + } + + #[test] + fn physical_constraint_survives_any_display_alias_rename() { + let (index, fields) = index_versions(); + for public_alias in ["adresar", "customer", "renamed_again"] { + let resolved_physical_column = if !public_alias.is_empty() { "1" } else { unreachable!() }; + let query = build_master_query( + &index, + &fields, + "", + &[SearchConstraint { + targets: vec![SearchConstraintTarget { + table_name: None, + column: resolved_physical_column.to_string(), + query: "new".to_string(), + }], + mode: ConstraintMode::Exact, + }], + Some("customers"), + SearchVersionScope::Current, + ) + .expect("renamed alias should resolve to the same physical query"); + assert_eq!(count(&index, &*query), 1, "alias {public_alias}"); + } + } + + #[test] + fn exact_constraints_are_anded_and_table_scoped() { + let (index, fields) = index_documents(&[ + TestDocument { + row_key: "customers:1".to_string(), + row_id: 1, + table_name: "customers", + values: &[("1", "Alice Example"), ("2", "Bratislava")], + }, + TestDocument { + row_key: "customers:2".to_string(), + row_id: 2, + table_name: "customers", + values: &[("1", "Alice Example"), ("2", "Kosice")], + }, + TestDocument { + row_key: "suppliers:3".to_string(), + row_id: 3, + table_name: "suppliers", + values: &[("1", "Alice Example"), ("2", "Bratislava")], + }, + ]); + let constraints = [ + SearchConstraint { + targets: vec![SearchConstraintTarget { + table_name: None, + column: "1".to_string(), + query: "Alice Example".to_string(), + }], + mode: ConstraintMode::Exact, + }, + SearchConstraint { + targets: vec![SearchConstraintTarget { + table_name: None, + column: "2".to_string(), + query: "Bratislava".to_string(), + }], + mode: ConstraintMode::Exact, + }, + ]; + let query = build_master_query( + &index, + &fields, + "", + &constraints, + Some("customers"), + SearchVersionScope::Current, + ) + .expect("exact query should build"); + + assert_eq!(count(&index, &*query), 1); + } + + #[test] + fn fuzzy_constraints_stay_within_the_resolved_physical_column() { + let (index, fields) = index_documents(&[ + TestDocument { + row_key: "customers:1".to_string(), + row_id: 1, + table_name: "customers", + values: &[("1", "Alice"), ("2", "Bratislava")], + }, + TestDocument { + row_key: "customers:2".to_string(), + row_id: 2, + table_name: "customers", + values: &[("1", "Bratislava"), ("2", "Kosice")], + }, + ]); + let constraint = |column: &str| SearchConstraint { + targets: vec![SearchConstraintTarget { + table_name: None, + column: column.to_string(), + query: "Bratislva".to_string(), + }], + mode: ConstraintMode::Fuzzy, + }; + let address_query = build_master_query( + &index, + &fields, + "", + &[constraint("2")], + Some("customers"), + SearchVersionScope::Current, + ) + .expect("fuzzy address query should build"); + let name_query = build_master_query( + &index, + &fields, + "", + &[constraint("1")], + Some("customers"), + SearchVersionScope::Current, + ) + .expect("fuzzy name query should build"); + + assert_eq!(count(&index, &*address_query), 1); + assert_eq!(count(&index, &*name_query), 1); + } + + #[test] + fn free_text_searches_all_public_values_but_respects_table_filter() { + let (index, fields) = index_documents(&[ + TestDocument { + row_key: "customers:1".to_string(), + row_id: 1, + table_name: "customers", + values: &[("1", "Alice Example")], + }, + TestDocument { + row_key: "suppliers:2".to_string(), + row_id: 2, + table_name: "suppliers", + values: &[("8", "Alice Example")], + }, + ]); + let query = build_master_query( + &index, + &fields, + "Alice Example", + &[], + Some("customers"), + SearchVersionScope::Current, + ) + .expect("free-text query should build"); + + assert_eq!(count(&index, &*query), 1); + } +}