perf(catalog): reuse loaded catalog state in pending_sync_rows

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<pgrx::pg_sys::Oid>,
+) -> 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())
     })
 }

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<u32> = 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));

graph/src/sql_sync.rs

@@ -150,17 +150,25 @@ pub(crate) fn disabled_graph_trigger_count() -> safety::GraphResult<i32> {
     .map_err(|e| safety::GraphError::Internal(format!("trigger status check failed: {}", e)))
 }
 
-pub(crate) fn pending_sync_rows(applied_sync_id: i64) -> safety::GraphResult<i64> {
+pub(crate) fn pending_sync_rows(
+    applied_sync_id: i64,
+    applicable_table_oids: &[u32],
+) -> safety::GraphResult<i64> {
+    if applicable_table_oids.is_empty() {
+        return Ok(0);
+    }
     Spi::get_one_with_args::<i64>(
-        "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<i64> {
-    let applicable_table_oids = SyncReplayContext::load()?.applicable_table_oids();
+pub(crate) fn pending_sync_rows_direct(
+    applied_sync_id: i64,
+    applicable_table_oids: Vec<u32>,
+) -> safety::GraphResult<i64> {
     if applicable_table_oids.is_empty() {
         return Ok(0);
     }