Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fde43774b4 |
@@ -265,12 +265,21 @@ async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
|
||||
for field in T::all_fields() {
|
||||
if field.is_text() || field.is_json() {
|
||||
let column_name = field.column();
|
||||
// inbuxa: with GIN's default fastupdate=on, new entries wait in
|
||||
// an unindexed pending list that every search scans in full
|
||||
// until a VACUUM (or 4 MB of backlog) merges it. On a mailbox
|
||||
// taking steady mail that list never drains and searches slow
|
||||
// from milliseconds to hundreds of them. Pay the index update
|
||||
// at insert time instead.
|
||||
let index_name = format!("gin_{table_name}_{column_name}");
|
||||
let create_index_query = format!(
|
||||
"CREATE INDEX IF NOT EXISTS gin_{table_name}_{column_name} ON {table_name} USING GIN({column_name})",
|
||||
"CREATE INDEX IF NOT EXISTS {index_name} ON {table_name} USING GIN({column_name}) WITH (fastupdate = off)",
|
||||
);
|
||||
conn.execute(&create_index_query, &[])
|
||||
.await
|
||||
.map_err(into_error)?;
|
||||
// Indexes made before this change keep fastupdate=on
|
||||
disable_gin_fastupdate(conn, &index_name).await;
|
||||
}
|
||||
|
||||
if field.is_indexed() {
|
||||
@@ -287,6 +296,69 @@ async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// inbuxa: turns fastupdate off on a GIN index made with the default and
|
||||
/// merges the pending list it has built up. Idempotent: an index that already
|
||||
/// has the option is left alone, so this costs one catalog read per index at
|
||||
/// startup. A failure is logged and startup goes on, since search still works,
|
||||
/// only slower.
|
||||
async fn disable_gin_fastupdate(conn: &Object, index_name: &str) {
|
||||
if let Err(err) = try_disable_gin_fastupdate(conn, index_name).await {
|
||||
trc::event!(
|
||||
Store(trc::StoreEvent::PostgresqlError),
|
||||
Details = format!("Failed to turn off fastupdate on search index {index_name}"),
|
||||
Reason = err.to_string(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn try_disable_gin_fastupdate(conn: &Object, index_name: &str) -> trc::Result<()> {
|
||||
let options = conn
|
||||
.query_opt(
|
||||
"SELECT COALESCE(reloptions, '{}')::text[] FROM pg_class WHERE oid = to_regclass($1)",
|
||||
&[&index_name],
|
||||
)
|
||||
.await
|
||||
.map_err(into_error)?
|
||||
.map(|row| row.try_get::<_, Vec<String>>(0))
|
||||
.transpose()
|
||||
.map_err(into_error)?;
|
||||
let Some(options) = options else {
|
||||
return Ok(());
|
||||
};
|
||||
if gin_fastupdate_is_off(&options) {
|
||||
return Ok(());
|
||||
}
|
||||
// SET (fastupdate) takes a SHARE UPDATE EXCLUSIVE lock, which doesn't
|
||||
// block reads or writes. Turning it off stops new entries going to the
|
||||
// pending list but doesn't flush the entries already there.
|
||||
conn.execute(
|
||||
&format!("ALTER INDEX {index_name} SET (fastupdate = off)"),
|
||||
&[],
|
||||
)
|
||||
.await
|
||||
.map_err(into_error)?;
|
||||
conn.query_one(
|
||||
"SELECT gin_clean_pending_list($1::text::regclass)",
|
||||
&[&index_name],
|
||||
)
|
||||
.await
|
||||
.map_err(into_error)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Whether a relation's reloptions turn GIN's fastupdate off.
|
||||
fn gin_fastupdate_is_off(options: &[String]) -> bool {
|
||||
options.iter().any(|option| {
|
||||
option.split_once('=').is_some_and(|(name, value)| {
|
||||
name.trim().eq_ignore_ascii_case("fastupdate")
|
||||
&& matches!(
|
||||
value.trim().to_ascii_lowercase().as_str(),
|
||||
"off" | "false" | "no" | "0" | "f" | "n"
|
||||
)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
async fn discover_ts_configs(pool: &Pool) -> AHashSet<&'static str> {
|
||||
let mut ts_configs = AHashSet::from_iter([PG_FALLBACK_LANG, PG_UNSTEMMED_LANG]);
|
||||
|
||||
|
||||
@@ -21,6 +21,8 @@ pub mod replica_mysql; // inbuxa: read replicas on MySQL
|
||||
#[cfg(all(feature = "postgres", feature = "redis"))]
|
||||
pub mod replica_cluster; // inbuxa: read replicas across nodes
|
||||
pub mod scaleout; // inbuxa: scale-out storage
|
||||
#[cfg(feature = "postgres")]
|
||||
pub mod search_gin; // inbuxa: GIN indexes without a pending list
|
||||
#[cfg(any(feature = "postgres", feature = "mysql"))]
|
||||
pub mod sql_timeout;
|
||||
pub mod task_locks; // inbuxa: task locks across nodes
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! PostgreSQL full-text GIN indexes are built with fastupdate off, and an
|
||||
//! index made earlier with the default is switched over at startup. With
|
||||
//! fastupdate on, new entries wait in a pending list that every search scans
|
||||
//! in full until VACUUM merges it.
|
||||
|
||||
use crate::utils::storage::build_data_store;
|
||||
use registry::schema::structs::DataStore;
|
||||
use store::{Rows, SearchStore, Store};
|
||||
|
||||
const SCHEMA: &str = "gin_fastupdate_test";
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
pub async fn postgres_gin_fastupdate() {
|
||||
println!("Running PostgreSQL GIN fastupdate test...");
|
||||
|
||||
// Work in a schema of our own so the shared search tables are untouched
|
||||
let admin = Store::build(build_data_store("PostgreSql", "").await)
|
||||
.await
|
||||
.expect("Failed to connect to PostgreSQL");
|
||||
for query in [
|
||||
format!("DROP SCHEMA IF EXISTS {SCHEMA} CASCADE"),
|
||||
format!("CREATE SCHEMA {SCHEMA}"),
|
||||
] {
|
||||
admin.sql_query::<usize>(&query, vec![]).await.unwrap();
|
||||
}
|
||||
let DataStore::PostgreSql(mut config) = build_data_store("PostgreSql", "").await else {
|
||||
unreachable!()
|
||||
};
|
||||
config.options = Some(format!("-c search_path={SCHEMA}"));
|
||||
let store = Store::build(DataStore::PostgreSql(config))
|
||||
.await
|
||||
.expect("Failed to connect to PostgreSQL");
|
||||
let search = SearchStore::Store(store.clone());
|
||||
|
||||
// A fresh schema
|
||||
search.create_indexes().await.unwrap();
|
||||
let indexes = gin_indexes(&admin).await;
|
||||
assert!(
|
||||
indexes.len() >= 4,
|
||||
"expected the search GIN indexes, found {indexes:?}"
|
||||
);
|
||||
for (name, options) in &indexes {
|
||||
assert!(
|
||||
options.contains("fastupdate=off"),
|
||||
"fresh index {name} has options {options:?}"
|
||||
);
|
||||
}
|
||||
|
||||
// A schema from before the change: the same indexes, made with the
|
||||
// default fastupdate=on, and a pending list with something in it
|
||||
for (name, _) in &indexes {
|
||||
admin
|
||||
.sql_query::<usize>(
|
||||
&format!("ALTER INDEX {SCHEMA}.{name} RESET (fastupdate)"),
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
for (name, options) in gin_indexes(&admin).await {
|
||||
assert!(
|
||||
!options.contains("fastupdate"),
|
||||
"index {name} still has options {options:?}"
|
||||
);
|
||||
}
|
||||
admin
|
||||
.sql_query::<usize>(
|
||||
&format!(
|
||||
"INSERT INTO {SCHEMA}.s_email (accid, docid, subj, body) \
|
||||
SELECT 1, n, to_tsvector('simple', 'pending subject ' || n), \
|
||||
to_tsvector('simple', 'pending body text ' || n) \
|
||||
FROM generate_series(1, 500) n"
|
||||
),
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
pending_tuples(&admin, "gin_s_email_body").await > 0,
|
||||
"no pending list to merge"
|
||||
);
|
||||
|
||||
// Startup on the existing schema switches every index over and merges
|
||||
// what was pending
|
||||
search.create_indexes().await.unwrap();
|
||||
for (name, options) in gin_indexes(&admin).await {
|
||||
assert!(
|
||||
options.contains("fastupdate=off"),
|
||||
"existing index {name} has options {options:?} after startup"
|
||||
);
|
||||
}
|
||||
assert_eq!(pending_tuples(&admin, "gin_s_email_body").await, 0);
|
||||
|
||||
// And a second startup changes nothing
|
||||
search.create_indexes().await.unwrap();
|
||||
for (name, options) in gin_indexes(&admin).await {
|
||||
assert!(options.contains("fastupdate=off"), "{name}: {options:?}");
|
||||
}
|
||||
|
||||
admin
|
||||
.sql_query::<usize>(&format!("DROP SCHEMA {SCHEMA} CASCADE"), vec![])
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// The GIN indexes in the test schema with their reloptions.
|
||||
async fn gin_indexes(admin: &Store) -> Vec<(String, String)> {
|
||||
admin
|
||||
.sql_query::<Rows>(
|
||||
&format!(
|
||||
"SELECT c.relname::text, COALESCE(array_to_string(c.reloptions, ','), '') \
|
||||
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \
|
||||
JOIN pg_am a ON a.oid = c.relam \
|
||||
WHERE n.nspname = '{SCHEMA}' AND c.relkind = 'i' AND a.amname = 'gin' \
|
||||
ORDER BY 1"
|
||||
),
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.rows
|
||||
.into_iter()
|
||||
.map(|row| {
|
||||
let mut values = row.values.into_iter();
|
||||
(
|
||||
values.next().unwrap().to_str().into_owned(),
|
||||
values.next().unwrap().to_str().into_owned(),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Tuples waiting in a GIN index's pending list (pgstattuple is a contrib
|
||||
/// extension the test database has).
|
||||
async fn pending_tuples(admin: &Store, index: &str) -> i64 {
|
||||
admin
|
||||
.sql_query::<usize>("CREATE EXTENSION IF NOT EXISTS pgstattuple", vec![])
|
||||
.await
|
||||
.unwrap();
|
||||
admin
|
||||
.sql_query::<Rows>(
|
||||
&format!("SELECT pending_tuples FROM pgstatginindex('{SCHEMA}.{index}'::regclass)"),
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.rows
|
||||
.into_iter()
|
||||
.next()
|
||||
.and_then(|row| row.values.into_iter().next())
|
||||
.map(|value| value.to_str().parse::<i64>().unwrap())
|
||||
.unwrap()
|
||||
}
|
||||
Reference in New Issue
Block a user