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

picker order by

Browse Source
This commit is contained in:
Priec
2026-07-16 14:38:12 +02:00
parent f2eb4d46bc
commit f13db931a4
6 changed files with 363 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

307
search/src/lib.rs
View File

@@ -6,7 +6,9 @@ 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::{SearchRequest, SearchResponse, search_response::Hit};
use common::proto::komp_ac::search::{
SearchOrderDirection, SearchRequest, SearchResponse, search_response::Hit,
};
use common::search::{SchemaFields, register_tokenizers, search_index_path};
use query_builder::{
ConstraintMode, SearchConstraint, SearchConstraintTarget, build_master_query,
@@ -67,10 +69,11 @@ impl SearcherService {
));
};
let hits = fetch_latest_rows(
let hits = fetch_ordered_rows(
&self.pool,
&normalized.profile_name,
table_name,
normalized.order.as_ref(),
normalized.limit.unwrap_or(DEFAULT_LIST_LIMIT),
normalized.offset,
)
@@ -104,13 +107,16 @@ impl SearcherService {
&resolved_must,
normalized.limit.unwrap_or(DEFAULT_RESULT_LIMIT),
normalized.offset,
normalized.order.as_ref(),
)
.await?;
hits.sort_by(|left, right| right.score.total_cmp(&left.score));
if let Some(limit) = normalized.limit {
if hits.len() > limit {
hits.truncate(limit);
if normalized.order.is_none() {
hits.sort_by(|left, right| right.score.total_cmp(&left.score));
if let Some(limit) = normalized.limit {
if hits.len() > limit {
hits.truncate(limit);
}
}
}
@@ -167,6 +173,13 @@ struct NormalizedSearchRequest {
must: Vec<NormalizedColumnConstraint>,
limit: Option<usize>,
offset: usize,
order: Option<NormalizedSearchOrder>,
}
#[derive(Debug)]
struct NormalizedSearchOrder {
column: String,
direction: SearchOrderDirection,
}
#[derive(Debug)]
@@ -315,6 +328,34 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
.limit
.map(|value| (value as usize).min(HARD_RESULT_LIMIT));
let offset = req.offset.unwrap_or_default() as usize;
let order = req
.order
.map(|order| {
let column = order.column.trim();
if column.is_empty() {
return Err(Status::invalid_argument("order.column must not be empty"));
}
validate_search_column(column)?;
let direction = SearchOrderDirection::try_from(order.direction).map_err(|_| {
Status::invalid_argument("order.direction must be ASC or DESC")
})?;
if direction == SearchOrderDirection::Unspecified {
return Err(Status::invalid_argument(
"order.direction must be ASC or DESC",
));
}
Ok(NormalizedSearchOrder {
column: column.to_string(),
direction,
})
})
.transpose()?;
if order.is_some() && table_name.is_none() {
return Err(Status::invalid_argument(
"table_name is required when order is specified",
));
}
Ok(NormalizedSearchRequest {
profile_name: profile_name.to_string(),
@@ -323,6 +364,7 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
must,
limit,
offset,
order,
})
}
@@ -507,18 +549,107 @@ fn row_display_value(value: &serde_json::Value, column: &str) -> String {
}
}
async fn fetch_latest_rows(
enum ResolvedOrderColumn {
Position,
Column(String),
}
fn order_direction_sql(direction: SearchOrderDirection) -> &'static str {
match direction {
SearchOrderDirection::Asc => "ASC",
SearchOrderDirection::Desc => "DESC",
SearchOrderDirection::Unspecified => unreachable!("order direction is normalized"),
}
}
async fn resolve_order_column(
pool: &PgPool,
profile_name: &str,
table_name: &str,
requested_column: &str,
) -> Result<ResolvedOrderColumn, Status> {
if requested_column.eq_ignore_ascii_case("position") {
return Ok(ResolvedOrderColumn::Position);
}
let requested_column = if requested_column.eq_ignore_ascii_case("row_display_column") {
table_row_display_column(pool, profile_name, table_name).await?
} else {
requested_column.to_string()
};
let physical_to_display =
table_physical_to_display_map(pool, profile_name, table_name).await?;
let physical_column = physical_to_display
.iter()
.find(|(physical, display)| {
physical.eq_ignore_ascii_case(&requested_column)
|| display.eq_ignore_ascii_case(&requested_column)
})
.map(|(physical, _)| physical.clone())
.unwrap_or_else(|| requested_column.clone());
let physical_column = sqlx::query_scalar::<_, String>(
r#"
SELECT column_name
FROM information_schema.columns
WHERE table_schema = $1 AND table_name = $2 AND LOWER(column_name) = LOWER($3)
"#,
)
.bind(profile_name)
.bind(table_name)
.bind(&physical_column)
.fetch_optional(pool)
.await
.map_err(|e| Status::internal(format!("Order column lookup failed: {}", e)))?;
let Some(physical_column) = physical_column else {
return Err(Status::invalid_argument(format!(
"Order column '{}' was not found in table '{}.{}'",
requested_column, profile_name, table_name
)));
};
Ok(ResolvedOrderColumn::Column(physical_column))
}
fn order_clause(column: &ResolvedOrderColumn, direction: SearchOrderDirection) -> String {
let direction = order_direction_sql(direction);
match column {
ResolvedOrderColumn::Position => {
format!("picker_position {}", direction)
}
ResolvedOrderColumn::Column(column) => {
let quoted_column = format!("\"{}\"", column.replace('"', "\"\""));
format!("{} {} NULLS LAST, id ASC", quoted_column, direction)
}
}
}
async fn fetch_ordered_rows(
pool: &PgPool,
profile_name: &str,
table_name: &str,
order: Option<&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 display_column = table_row_display_column(pool, profile_name, table_name).await?;
let (resolved_order, direction) = match order {
Some(order) => (
resolve_order_column(pool, profile_name, table_name, &order.column).await?,
order.direction,
),
None => (ResolvedOrderColumn::Position, SearchOrderDirection::Desc),
};
let sql = format!(
"SELECT id, to_jsonb(t) AS data FROM {} t WHERE deleted = FALSE ORDER BY id DESC LIMIT $1 OFFSET $2",
qualify_profile_table(profile_name, table_name)
"WITH positioned AS (\
SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \
FROM {} t WHERE deleted = FALSE\
) \
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),
order_clause(&resolved_order, direction),
);
let rows = sqlx::query(AssertSqlSafe(sql))
@@ -533,6 +664,7 @@ async fn fetch_latest_rows(
.map(|row| {
let id: i64 = row.try_get("id").unwrap_or_default();
let json_data: serde_json::Value = row.try_get("data").unwrap_or_default();
let position: i64 = row.try_get("picker_position").unwrap_or_default();
let json_data = remap_json_to_display_names(json_data, &physical_to_display);
let row_display_value = row_display_value(&json_data, &display_column);
Hit {
@@ -542,6 +674,7 @@ async fn fetch_latest_rows(
table_name: table_name.to_string(),
row_display_value,
row_display_column: display_column.clone(),
position: u64::try_from(position).ok(),
}
})
.collect())
@@ -556,6 +689,7 @@ async fn run_search(
must: &[SearchConstraint],
limit: usize,
offset: usize,
order: Option<&NormalizedSearchOrder>,
) -> Result<Vec<Hit>, Status> {
let master_query = build_master_query(
&profile.index,
@@ -566,7 +700,14 @@ async fn run_search(
)?;
let searcher = profile.reader.searcher();
let window_limit = offset.saturating_add(limit);
let window_limit = if order.is_some() {
searcher.num_docs() as usize
} else {
offset.saturating_add(limit)
};
if window_limit == 0 {
return Ok(Vec::new());
}
let top_docs = searcher
.search(&*master_query, &TopDocs::with_limit(window_limit).order_by_score())
.map_err(|e| Status::internal(format!("Search failed: {}", e)))?;
@@ -581,12 +722,19 @@ async fn run_search(
} else {
0.0
};
let page_docs = top_docs
let eligible_docs = top_docs
.into_iter()
.skip(offset)
.filter(|(score, _)| *score >= score_floor)
.take(limit)
.collect::<Vec<_>>();
let page_docs = if order.is_some() {
eligible_docs
} else {
eligible_docs
.into_iter()
.skip(offset)
.take(limit)
.collect::<Vec<_>>()
};
let mut candidates: Vec<(f32, i64, String)> = Vec::with_capacity(page_docs.len());
for (score, doc_address) in page_docs {
@@ -612,6 +760,20 @@ async fn run_search(
return Ok(vec![]);
}
if let Some(order) = order {
let table_name = table_filter.expect("ordered searches require a normalized table filter");
return fetch_ordered_candidate_rows(
pool,
profile_name,
table_name,
&candidates,
order,
limit,
offset,
)
.await;
}
let mut ids_by_table: HashMap<String, Vec<i64>> = HashMap::new();
for (_, pg_id, table_name) in &candidates {
ids_by_table
@@ -661,11 +823,72 @@ async fn run_search(
table_name,
row_display_value: row_display_value.clone(),
row_display_column: row_display_column.clone(),
position: None,
})
})
.collect())
}
async fn fetch_ordered_candidate_rows(
pool: &PgPool,
profile_name: &str,
table_name: &str,
candidates: &[(f32, i64, String)],
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 display_column = table_row_display_column(pool, profile_name, table_name).await?;
let resolved_order =
resolve_order_column(pool, profile_name, table_name, &order.column).await?;
let sql = format!(
"WITH positioned AS (\
SELECT t.*, ROW_NUMBER() OVER (ORDER BY id ASC) AS picker_position \
FROM {} t WHERE deleted = FALSE\
) \
SELECT id, to_jsonb(positioned) - 'picker_position' AS data, picker_position \
FROM positioned WHERE id = ANY($1) ORDER BY {} LIMIT $2 OFFSET $3",
qualify_profile_table(profile_name, table_name),
order_clause(&resolved_order, order.direction),
);
let ids = candidates
.iter()
.map(|(_, id, _)| *id)
.collect::<Vec<_>>();
let scores = candidates
.iter()
.map(|(score, id, _)| (*id, *score))
.collect::<HashMap<_, _>>();
let rows = sqlx::query(AssertSqlSafe(sql))
.bind(&ids)
.bind(limit as i64)
.bind(offset as i64)
.fetch_all(pool)
.await
.map_err(|e| Status::internal(format!("Ordered search query failed: {}", e)))?;
Ok(rows
.into_iter()
.map(|row| {
let id: i64 = row.try_get("id").unwrap_or_default();
let json_data: serde_json::Value = row.try_get("data").unwrap_or_default();
let position: i64 = row.try_get("picker_position").unwrap_or_default();
let json_data = remap_json_to_display_names(json_data, &physical_to_display);
let display_value = row_display_value(&json_data, &display_column);
Hit {
id,
score: scores.get(&id).copied().unwrap_or_default(),
content_json: json_data.to_string(),
table_name: table_name.to_string(),
row_display_value: display_value,
row_display_column: display_column.clone(),
position: u64::try_from(position).ok(),
}
})
.collect())
}
#[tonic::async_trait]
impl Searcher for SearcherService {
async fn search(
@@ -675,3 +898,61 @@ impl Searcher for SearcherService {
self.run_rpc(request).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use common::proto::komp_ac::search::{SearchOrder, SearchOrderDirection};
fn request(order: Option<SearchOrder>, table_name: Option<&str>) -> SearchRequest {
SearchRequest {
profile_name: "finance".to_string(),
table_name: table_name.map(str::to_string),
free_query: String::new(),
must: Vec::new(),
limit: Some(20),
offset: Some(40),
order,
}
}
#[test]
fn normalizes_typed_column_order() {
let normalized = normalize_request(request(
Some(SearchOrder {
column: "created_at".to_string(),
direction: SearchOrderDirection::Desc as i32,
}),
Some("invoice"),
))
.unwrap();
let order = normalized.order.unwrap();
assert_eq!(order.column, "created_at");
assert_eq!(order.direction, SearchOrderDirection::Desc);
}
#[test]
fn ordered_request_requires_a_table() {
let result = normalize_request(request(
Some(SearchOrder {
column: "position".to_string(),
direction: SearchOrderDirection::Desc as i32,
}),
None,
));
assert!(result.is_err());
}
#[test]
fn column_order_has_stable_id_tie_breaker() {
assert_eq!(
order_clause(
&ResolvedOrderColumn::Column("1".to_string()),
SearchOrderDirection::Asc,
),
"\"1\" ASC NULLS LAST, id ASC"
);
}
}