From 3e68482f544c2cac934df1742227af56a78242b8 Mon Sep 17 00:00:00 2001 From: "Edgeshot@27" Date: Wed, 5 Aug 2026 20:45:19 +0530 Subject: [PATCH] perf(catalog): reuse loaded catalog state in pending_sync_rows --- graph/src/sql_facade/admin.rs | 8 ++++++-- graph/src/sql_facade/runtime.rs | 11 ++++++++++- graph/src/sql_sync.rs | 18 +++++++++++++----- 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/graph/src/sql_facade/admin.rs b/graph/src/sql_facade/admin.rs index 6ee661f4..aacdaca0 100644 --- a/graph/src/sql_facade/admin.rs +++ b/graph/src/sql_facade/admin.rs @@ -32,9 +32,13 @@ fn selected_graph_id_for_current_role() -> String { security_definer )] #[search_path(pg_catalog, pg_temp)] -fn pending_sync_rows_for_current_role(applied_sync_id: i64) -> i64 { +fn pending_sync_rows_for_current_role( + applied_sync_id: i64, + applicable_table_oids: Vec, +) -> i64 { with_panic_boundary("_pending_sync_rows_for_current_role()", || { - crate::sql_sync::pending_sync_rows_direct(applied_sync_id) + let oids = applicable_table_oids.into_iter().map(|o| o.as_u32()).collect(); + crate::sql_sync::pending_sync_rows_direct(applied_sync_id, oids) .unwrap_or_else(|err| err.report()) }) } diff --git a/graph/src/sql_facade/runtime.rs b/graph/src/sql_facade/runtime.rs index 1fad8c73..e485158c 100644 --- a/graph/src/sql_facade/runtime.rs +++ b/graph/src/sql_facade/runtime.rs @@ -582,7 +582,16 @@ pub(crate) fn ensure_current_graph() -> safety::GraphResult<()> { let disabled = disabled_graph_trigger_count()?; let catalog_state = current_catalog_state()?; let applied_sync_id = ENGINE.with(|e| e.borrow().applied_sync_id); - let pending = pending_sync_rows(applied_sync_id)?; + let mut table_oids = std::collections::HashSet::new(); + for table in &catalog_state.0 { + table_oids.insert(table.table_oid); + } + for edge in &catalog_state.1 { + table_oids.insert(edge.from_table_oid); + table_oids.insert(edge.to_table_oid); + } + let applicable_table_oids: Vec = table_oids.into_iter().collect(); + let pending = pending_sync_rows(applied_sync_id, &applicable_table_oids)?; ENGINE.with(|e| { let mut eng = e.borrow_mut(); eng.refresh_observed_state(disabled, pending, &Ok(catalog_state)); diff --git a/graph/src/sql_sync.rs b/graph/src/sql_sync.rs index 8becdfcd..bca0585c 100644 --- a/graph/src/sql_sync.rs +++ b/graph/src/sql_sync.rs @@ -150,17 +150,25 @@ pub(crate) fn disabled_graph_trigger_count() -> safety::GraphResult { .map_err(|e| safety::GraphError::Internal(format!("trigger status check failed: {}", e))) } -pub(crate) fn pending_sync_rows(applied_sync_id: i64) -> safety::GraphResult { +pub(crate) fn pending_sync_rows( + applied_sync_id: i64, + applicable_table_oids: &[u32], +) -> safety::GraphResult { + if applicable_table_oids.is_empty() { + return Ok(0); + } Spi::get_one_with_args::( - "SELECT graph._pending_sync_rows_for_current_role($1)", - &[applied_sync_id.into()], + "SELECT graph._pending_sync_rows_for_current_role($1, $2)", + &[applied_sync_id.into(), applicable_table_oids.into()], ) .map_err(|e| safety::GraphError::Internal(format!("sync status check failed: {}", e)))? .ok_or_else(|| safety::GraphError::Internal("pending sync row count was null".to_string())) } -pub(crate) fn pending_sync_rows_direct(applied_sync_id: i64) -> safety::GraphResult { - let applicable_table_oids = SyncReplayContext::load()?.applicable_table_oids(); +pub(crate) fn pending_sync_rows_direct( + applied_sync_id: i64, + applicable_table_oids: Vec, +) -> safety::GraphResult { if applicable_table_oids.is_empty() { return Ok(0); }