store: Implement Parquet dump for subgraph data

Wire dump queries and Parquet writing into Layout::dump(). This is the
core of the dump feature, covering both entity tables and the special
data_sources$ table.

For entity tables: build a DynamicSelectClause that selects vid, block
columns (split into lower/upper for mutable tables), causality_region,
and data columns. Use VidBatcher for adaptive batching, convert rows
to Arrow RecordBatch, and write via ParquetChunkWriter.

For data_sources$: use a concrete QueryableByName struct with raw SQL
(fixed schema, no DynamicSelectClause needed). Check table existence
via catalog::table_exists before attempting dump.

Metadata includes version, network, block pointers, entity count,
graft info, health, indexes, and per-table chunk tracking. Written
atomically via tmp+rename so its presence signals a complete dump.

Add entity_count() helper in detail.rs. Wire dump() through
DeploymentStore and SubgraphStore.

Author: lutter

store/postgres/src/deployment_store.rs

@@ -30,6 +30,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::convert::Into;
 use std::ops::Bound;
 use std::ops::{Deref, Range};
+use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::{atomic::AtomicUsize, Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -896,6 +897,18 @@ impl DeploymentStore {
 
         Ok(relational::prune::Viewer::new(self.pool.clone(), layout))
     }
+
+    pub(crate) async fn dump(&self, site: Arc<Site>, dir: PathBuf) -> Result<(), StoreError> {
+        let mut conn = self.pool.get_permitted().await?;
+        let layout = self.layout(&mut conn, site.cheap_clone()).await?;
+        let entity_count = crate::detail::entity_count(&mut conn, &site).await?;
+        // Loading the IndexList should happen inside dump, but the
+        // interface does not allow it; should be changed
+        let index_list = IndexList::load(&mut conn, site.cheap_clone(), self.clone()).await?;
+        layout
+            .dump(&mut conn, index_list, dir, &site.network, entity_count)
+            .await
+    }
 }
 
 /// Methods that back the trait `WritableStore`, but have small variations in their signatures

store/postgres/src/detail.rs

@@ -657,6 +657,18 @@ pub async fn deployment_entity(
     StoredDeploymentEntity(detail, manifest).into_subgraph_deployment_entity(schema)
 }
 
+/// Load the entity count for a deployment from `subgraphs.head`.
+pub async fn entity_count(conn: &mut AsyncPgConnection, site: &Site) -> Result<usize, StoreError> {
+    use subgraph_head as h;
+
+    let count: i64 = h::table
+        .find(site.id)
+        .select(h::entity_count)
+        .first(conn)
+        .await?;
+    Ok(count as usize)
+}
+
 #[derive(Queryable, Identifiable, Insertable)]
 #[diesel(table_name = graph_node_versions)]
 pub struct GraphNodeVersion {

store/postgres/src/parquet/mod.rs

@@ -1,2 +1,3 @@
-#[allow(dead_code)]
+pub(crate) mod convert;
 pub(crate) mod schema;
+pub(crate) mod writer;

store/postgres/src/relational.rs

@@ -15,6 +15,7 @@ mod ddl_tests;
 mod query_tests;
 
 pub(crate) mod dsl;
+pub(crate) mod dump;
 pub(crate) mod index;
 pub(crate) mod prune;
 pub(crate) mod rollup;