store: Asyncify catalog::histogram_bounds

Author: lutter

store/postgres/src/catalog.rs

@@ -1061,7 +1061,7 @@ fn pg_stats_has_range_bounds_histogram(conn: &mut PgConnection) -> Result<bool,
         .map_err(StoreError::from)
 }
 
-pub(crate) fn histogram_bounds(
+pub(crate) async fn histogram_bounds(
     conn: &mut PgConnection,
     namespace: &Namespace,
     table: &SqlName,

store/postgres/src/copy.rs

@@ -121,7 +121,7 @@ struct CopyState {
 }
 
 impl CopyState {
-    fn new(
+    async fn new(
         conn: &mut PgConnection,
         primary: Primary,
         src: Arc<Layout>,
@@ -162,22 +162,22 @@ impl CopyState {
                         src.site.id
                     ));
                 }
-                Self::load(conn, primary, src, dst, target_block)
+                Self::load(conn, primary, src, dst, target_block).await
             }
-            None => Self::create(conn, primary.cheap_clone(), src, dst, target_block),
+            None => Self::create(conn, primary.cheap_clone(), src, dst, target_block).await,
         }?;
 
         Ok(state)
     }
 
-    fn load(
+    async fn load(
         conn: &mut PgConnection,
         primary: Primary,
         src: Arc<Layout>,
         dst: Arc<Layout>,
         target_block: BlockPtr,
     ) -> Result<CopyState, StoreError> {
-        let tables = TableState::load(conn, primary, src.as_ref(), dst.as_ref())?;
+        let tables = TableState::load(conn, primary, src.as_ref(), dst.as_ref()).await?;
         let (finished, mut unfinished): (Vec<_>, Vec<_>) =
             tables.into_iter().partition(|table| table.finished());
         unfinished.sort_by_key(|table| table.dst.object.to_string());
@@ -190,7 +190,7 @@ impl CopyState {
         })
     }
 
-    fn create(
+    async fn create(
         conn: &mut PgConnection,
         primary: Primary,
         src: Arc<Layout>,
@@ -212,15 +212,18 @@ impl CopyState {
         let mut unfinished = Vec::new();
         for dst_table in dst.tables.values() {
             if let Some(src_table) = src.table_for_entity(&dst_table.object).ok() {
-                unfinished.push(TableState::init(
-                    conn,
-                    primary.cheap_clone(),
-                    dst.site.clone(),
-                    &src,
-                    src_table.clone(),
-                    dst_table.clone(),
-                    &target_block,
-                )?);
+                unfinished.push(
+                    TableState::init(
+                        conn,
+                        primary.cheap_clone(),
+                        dst.site.clone(),
+                        &src,
+                        src_table.clone(),
+                        dst_table.clone(),
+                        &target_block,
+                    )
+                    .await?,
+                );
             }
         }
         unfinished.sort_by_key(|table| table.dst.object.to_string());
@@ -324,7 +327,7 @@ struct TableState {
 }
 
 impl TableState {
-    fn init(
+    async fn init(
         conn: &mut PgConnection,
         primary: Primary,
         dst_site: Arc<Site>,
@@ -334,7 +337,8 @@ impl TableState {
         target_block: &BlockPtr,
     ) -> Result<Self, StoreError> {
         let vid_range = VidRange::for_copy(conn, &src, target_block)?;
-        let batcher = VidBatcher::load(conn, &src_layout.site.namespace, src.as_ref(), vid_range)?;
+        let batcher =
+            VidBatcher::load(conn, &src_layout.site.namespace, src.as_ref(), vid_range).await?;
         Ok(Self {
             primary,
             src,
@@ -349,7 +353,7 @@ impl TableState {
         self.batcher.finished()
     }
 
-    fn load(
+    async fn load(
         conn: &mut PgConnection,
         primary: Primary,
         src_layout: &Layout,
@@ -408,7 +412,8 @@ impl TableState {
                 &src_layout.site.namespace,
                 &src,
                 VidRange::new(current_vid, target_vid),
-            )?
+            )
+            .await?
             .with_batch_size(size as usize);
 
             let state = TableState {
@@ -1128,7 +1133,7 @@ impl Connection {
         let primary = self.primary.cheap_clone();
         let mut state = self
             .transaction(|conn| {
-                async { CopyState::new(conn, primary, src, dst, target_block) }.scope_boxed()
+                CopyState::new(conn, primary, src, dst, target_block).scope_boxed()
             })?
             .await?;
 

store/postgres/src/relational/prune.rs

@@ -106,7 +106,7 @@ impl TablePair {
 
         // Determine the last vid that we need to copy
         let range = VidRange::for_prune(conn, &self.src, earliest_block, final_block)?;
-        let mut batcher = VidBatcher::load(conn, &self.src_nsp, &self.src, range)?;
+        let mut batcher = VidBatcher::load(conn, &self.src_nsp, &self.src, range).await?;
         tracker.start_copy_final(conn, &self.src, range).await?;
 
         while !batcher.finished() {
@@ -174,7 +174,7 @@ impl TablePair {
 
         // Determine the last vid that we need to copy
         let range = VidRange::for_prune(conn, &self.src, final_block + 1, BLOCK_NUMBER_MAX)?;
-        let mut batcher = VidBatcher::load(conn, &self.src.nsp, &self.src, range)?;
+        let mut batcher = VidBatcher::load(conn, &self.src.nsp, &self.src, range).await?;
         tracker.start_copy_nonfinal(conn, &self.src, range).await?;
 
         while !batcher.finished() {
@@ -479,7 +479,8 @@ impl Layout {
                     // Delete all entity versions whose range was closed
                     // before `req.earliest_block`
                     let range = VidRange::for_prune(conn, &table, 0, req.earliest_block)?;
-                    let mut batcher = VidBatcher::load(conn, &self.site.namespace, &table, range)?;
+                    let mut batcher =
+                        VidBatcher::load(conn, &self.site.namespace, &table, range).await?;
 
                     tracker.start_delete(conn, table, range, &batcher).await?;
                     while !batcher.finished() {

store/postgres/src/vid_batcher.rs

@@ -118,13 +118,13 @@ impl VidBatcher {
     /// The `vid_range` is inclusive, i.e., the batcher will iterate over
     /// all vids `vid_range.0 <= vid <= vid_range.1`; for an empty table,
     /// the `vid_range` must be set to `(-1, 0)`
-    pub fn load(
+    pub async fn load(
         conn: &mut PgConnection,
         nsp: &Namespace,
         table: &Table,
         vid_range: VidRange,
     ) -> Result<Self, StoreError> {
-        let bounds = catalog::histogram_bounds(conn, nsp, &table.name, VID_COLUMN)?;
+        let bounds = catalog::histogram_bounds(conn, nsp, &table.name, VID_COLUMN).await?;
         let batch_size = AdaptiveBatchSize::new(table);
         Self::new(bounds, vid_range, batch_size)
     }