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

search properly aliasing now

Browse Source
This commit is contained in:
Priec
2026-07-09 13:14:06 +02:00
parent 98cf2370ec
commit 6711165c75
4 changed files with 224 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
search/src/lib.rs
View File

@@ -1,6 +1,6 @@
mod query_builder;
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::sync::{Arc, Mutex};
@@ -8,7 +8,9 @@ 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::search::{SchemaFields, register_tokenizers, search_index_path};
use query_builder::{ConstraintMode, SearchConstraint, build_master_query};
use query_builder::{
ConstraintMode, SearchConstraint, SearchConstraintTarget, build_master_query,
};
use sqlx::{AssertSqlSafe, PgPool, Row};
use tantivy::collector::TopDocs;
use tantivy::schema::Value;
@@ -82,6 +84,14 @@ impl SearcherService {
)));
}
let resolved_must = resolve_constraints(
&self.pool,
&normalized.profile_name,
normalized.table_name.as_deref(),
&normalized.must,
)
.await?;
let profile = profile_index(&self.profiles, &normalized.profile_name, &index_path)?;
let mut hits = run_search(
&self.pool,
@@ -89,7 +99,7 @@ impl SearcherService {
&normalized.profile_name,
normalized.table_name.as_deref(),
&normalized.free_query,
&normalized.must,
&resolved_must,
normalized.limit.unwrap_or(DEFAULT_RESULT_LIMIT),
)
.await?;
@@ -106,7 +116,7 @@ impl SearcherService {
normalized.profile_name,
normalized.table_name,
normalized.free_query,
normalized.must.len(),
resolved_must.len(),
hits.len()
);
@@ -151,10 +161,17 @@ struct NormalizedSearchRequest {
profile_name: String,
table_name: Option<String>,
free_query: String,
must: Vec<SearchConstraint>,
must: Vec<NormalizedColumnConstraint>,
limit: Option<usize>,
}
#[derive(Debug)]
struct NormalizedColumnConstraint {
column: String,
query: String,
mode: ConstraintMode,
}
impl NormalizedSearchRequest {
fn has_input(&self) -> bool {
!self.free_query.is_empty() || !self.must.is_empty()
@@ -283,7 +300,7 @@ fn normalize_request(req: SearchRequest) -> Result<NormalizedSearchRequest, Stat
));
}
must.push(SearchConstraint {
must.push(NormalizedColumnConstraint {
column: column.to_string(),
query: query.to_string(),
mode: constraint_mode_from_proto(constraint.mode),
@@ -310,12 +327,158 @@ fn constraint_mode_from_proto(raw_mode: i32) -> ConstraintMode {
}
}
async fn resolve_constraints(
pool: &PgPool,
profile_name: &str,
table_filter: Option<&str>,
must: &[NormalizedColumnConstraint],
) -> 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?;
resolved.push(SearchConstraint {
targets,
query: constraint.query.clone(),
mode: constraint.mode,
});
}
Ok(resolved)
}
async fn resolve_constraint_targets(
pool: &PgPool,
profile_name: &str,
table_filter: Option<&str>,
column: &str,
) -> Result<Vec<SearchConstraintTarget>, Status> {
let rows = if let Some(table_name) = table_filter {
sqlx::query(
r#"
SELECT td.table_name, tdc.physical_name
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
AND tdc.display_name = $3
"#,
)
.bind(profile_name)
.bind(table_name)
.bind(column)
.fetch_all(pool)
.await
} else {
sqlx::query(
r#"
SELECT td.table_name, tdc.physical_name
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 tdc.display_name = $2
"#,
)
.bind(profile_name)
.bind(column)
.fetch_all(pool)
.await
}
.map_err(|e| Status::internal(format!("Column mapping lookup failed: {}", e)))?;
if rows.is_empty() {
let target = SearchConstraintTarget {
table_name: table_filter.map(str::to_string),
column: column.to_string(),
};
return Ok(vec![target]);
}
let mut seen = HashSet::new();
let mut targets = Vec::new();
for row in rows {
let table_name: String = row
.try_get("table_name")
.map_err(|e| Status::internal(format!("Column mapping table read failed: {}", e)))?;
let physical_name: String = row
.try_get("physical_name")
.map_err(|e| Status::internal(format!("Column mapping physical read failed: {}", e)))?;
let key = (table_name.clone(), physical_name.clone());
if seen.insert(key) {
targets.push(SearchConstraintTarget {
table_name: if table_filter.is_some() {
None
} else {
Some(table_name)
},
column: physical_name,
});
}
}
Ok(targets)
}
async fn table_physical_to_display_map(
pool: &PgPool,
profile_name: &str,
table_name: &str,
) -> Result<HashMap<String, String>, Status> {
let rows = sqlx::query(
r#"
SELECT tdc.physical_name, tdc.display_name
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
"#,
)
.bind(profile_name)
.bind(table_name)
.fetch_all(pool)
.await
.map_err(|e| Status::internal(format!("Column mapping lookup failed: {}", e)))?;
let mut mapping = HashMap::with_capacity(rows.len());
for row in rows {
let physical_name: String = row
.try_get("physical_name")
.map_err(|e| Status::internal(format!("Column mapping physical read failed: {}", e)))?;
let display_name: String = row
.try_get("display_name")
.map_err(|e| Status::internal(format!("Column mapping display read failed: {}", e)))?;
mapping.insert(physical_name, display_name);
}
Ok(mapping)
}
fn remap_json_to_display_names(
value: serde_json::Value,
physical_to_display: &HashMap<String, String>,
) -> serde_json::Value {
match value {
serde_json::Value::Object(object) => {
let mut remapped = serde_json::Map::with_capacity(object.len());
for (key, value) in object {
let final_key = physical_to_display.get(&key).cloned().unwrap_or(key);
remapped.insert(final_key, value);
}
serde_json::Value::Object(remapped)
}
other => other,
}
}
async fn fetch_latest_rows(
pool: &PgPool,
profile_name: &str,
table_name: &str,
limit: usize,
) -> Result<Vec<Hit>, Status> {
let physical_to_display = table_physical_to_display_map(pool, profile_name, table_name).await?;
let sql = format!(
"SELECT id, to_jsonb(t) AS data FROM {} t WHERE deleted = FALSE ORDER BY id DESC LIMIT $1",
qualify_profile_table(profile_name, table_name)
@@ -332,6 +495,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 json_data = remap_json_to_display_names(json_data, &physical_to_display);
Hit {
id,
score: 0.0,
@@ -403,6 +567,8 @@ async fn run_search(
let mut content_map: HashMap<(String, i64), String> = 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 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)
@@ -417,6 +583,7 @@ 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);
content_map.insert((table_name.clone(), id), json_data.to_string());
}
}