filipriec/komp_ac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

search fix to new architecture

Browse Source
This commit is contained in:
Priec
2026-08-13 21:23:49 +02:00
parent 8662b00ccf
commit 810c8ffd08
8 changed files with 1069 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -6474,6 +6474,7 @@ dependencies = [
"anyhow",
"common",
"prost",
"rust_decimal",
"serde",
"serde_json",
"sqlx",

19
common/proto/search.proto
View File

@@ -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;
}

BIN
common/src/proto/descriptor.bin
View File

Binary file not shown.

119
common/src/proto/komp_ac.search.rs
View File

@@ -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<u32>,
#[prost(message, optional, tag = "7")]
pub order: ::core::option::Option<SearchOrder>,
/// 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<u64>,
/// 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<Self> {
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<super::SearchRequest>,
) -> std::result::Result<
tonic::Response<super::SearchCountResponse>,
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<super::SearchRequest>,
) -> std::result::Result<tonic::Response<super::SearchResponse>, tonic::Status>;
async fn count(
&self,
request: tonic::Request<super::SearchRequest>,
) -> std::result::Result<
tonic::Response<super::SearchCountResponse>,
tonic::Status,
>;
}
#[derive(Debug)]
pub struct SearcherServer<T> {
@@ -370,6 +446,49 @@ pub mod searcher_server {
};
Box::pin(fut)
}
"/komp_ac.search.Searcher/Count" => {
#[allow(non_camel_case_types)]
struct CountSvc<T: Searcher>(pub Arc<T>);
impl<T: Searcher> tonic::server::UnaryService<super::SearchRequest>
for CountSvc<T> {
type Response = super::SearchCountResponse;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::SearchRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Searcher>::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(

48
common/src/search.rs
View File

@@ -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<Field> {
schema.get_field(name).map_err(|e| {
tantivy::TantivyError::SchemaError(format!("schema is missing field '{name}': {e}"))

1
search/Cargo.toml
View File

@@ -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 }

657
search/src/lib.rs
View File

@@ -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<SearchRequest>,
) -> Result<Response<SearchCountResponse>, 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<u64, Status> {
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(&current_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::<Vec<_>>();
let row_ids = archives.iter().map(|item| item.1).collect::<Vec<_>>();
let versions = archives.iter().map(|item| item.2).collect::<Vec<_>>();
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<usize>,
offset: usize,
order: Option<NormalizedSearchOrder>,
version_scope: SearchVersionScope,
}
#[derive(Debug)]
@@ -191,6 +366,45 @@ struct NormalizedColumnConstraint {
mode: ConstraintMode,
}
fn normalize_constraint_value(query: &str, field_type: &str) -> Result<String, Status> {
let normalized_type = field_type.trim().to_ascii_lowercase();
if normalized_type == "money" || normalized_type.starts_with("decimal(") {
return query
.parse::<rust_decimal::Decimal>()
.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::<i64>()
.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<String, Status> {
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<NormalizedSearchRequest, Status> {
let profile_name = req.profile_name.trim();
if profile_name.is_empty() {
@@ -353,11 +593,19 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
})
.transpose()?;
let version_scope = SearchVersionScope::try_from(req.version_scope)
.map_err(|_| Status::invalid_argument("version_scope is invalid"))?;
if order.is_some() && table_name.is_none() {
return Err(Status::invalid_argument(
"table_name is required when order is specified",
));
}
if order.is_some() && version_scope != SearchVersionScope::Current {
return Err(Status::invalid_argument(
"ordering archived search results is not supported",
));
}
Ok(NormalizedSearchRequest {
profile_name: profile_name.to_string(),
@@ -367,6 +615,7 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
limit,
offset,
order,
version_scope,
})
}
@@ -385,12 +634,18 @@ async fn resolve_constraints(
) -> Result<Vec<SearchConstraint>, 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<Vec<SearchConstraintTarget>, 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<HashSet<String>, 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::<HashSet<_>>();
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<String, String>,
internal_columns: &HashSet<String>,
) -> Result<serde_json::Value, Status> {
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<Vec<Hit>, 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<Vec<Hit>, 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::<Vec<_>>()
};
let mut candidates: Vec<(f32, i64, String)> = Vec::with_capacity(page_docs.len());
let mut candidates: Vec<SearchCandidate> = 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<String, Vec<i64>> = 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<String>, Vec<String>)> =
let mut content_map: HashMap<String, (String, Vec<String>, Vec<String>, 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<String, Vec<(i64, i64, i64)>> = 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::<Vec<_>>();
let row_ids = archives.iter().map(|item| item.1).collect::<Vec<_>>();
let versions = archives.iter().map(|item| item.2).collect::<Vec<_>>();
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<Vec<Hit>, 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::<Vec<_>>();
let scores = candidates
.iter()
.map(|(score, _, _)| *score)
.map(|candidate| candidate.score)
.collect::<Vec<_>>();
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<Response<SearchResponse>, Status> {
self.run_rpc(request).await
}
async fn count(
&self,
request: Request<SearchRequest>,
) -> Result<Response<SearchCountResponse>, 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<SearchOrder>, 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,
}
}

282
search/src/query_builder.rs
View File

@@ -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<SearchConstraintTarget>,
pub query: String,
pub mode: ConstraintMode,
}
@@ -27,6 +27,7 @@ pub struct SearchConstraint {
pub struct SearchConstraintTarget {
pub table_name: Option<String>,
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<Box<dyn Query>, Status> {
let mut clauses: Vec<(Occur, Box<dyn Query>)> = 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<u8> {
_ => 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);
}
}