graph, store: 'graphman info': Do not get confused by copies

Because the code in primary::queries::fill_assignments used the deployment
hash to reference a deployment, it would get confused by copies since for
those several deployments have the same hash

Author: lutter

graph/src/components/store/mod.rs

@@ -3,6 +3,11 @@ mod err;
 mod traits;
 pub mod write;
 
+use diesel::deserialize::FromSql;
+use diesel::pg::Pg;
+use diesel::serialize::{Output, ToSql};
+use diesel::sql_types::Integer;
+use diesel_derives::{AsExpression, FromSqlRow};
 pub use entity_cache::{EntityCache, EntityLfuCache, GetScope, ModificationsAndCache};
 use slog::Logger;
 
@@ -691,7 +696,20 @@ pub struct StoredDynamicDataSource {
 /// identifier only has meaning in the context of a specific instance of
 /// graph-node. Only store code should ever construct or consume it; all
 /// other code passes it around as an opaque token.
-#[derive(Copy, Clone, CheapClone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash)]
+#[derive(
+    Copy,
+    Clone,
+    CheapClone,
+    Debug,
+    Serialize,
+    Deserialize,
+    PartialEq,
+    Eq,
+    Hash,
+    AsExpression,
+    FromSqlRow,
+)]
+#[diesel(sql_type = Integer)]
 pub struct DeploymentId(pub i32);
 
 impl Display for DeploymentId {
@@ -706,6 +724,19 @@ impl DeploymentId {
     }
 }
 
+impl FromSql<Integer, Pg> for DeploymentId {
+    fn from_sql(bytes: diesel::pg::PgValue) -> diesel::deserialize::Result<Self> {
+        let id = <i32 as FromSql<Integer, Pg>>::from_sql(bytes)?;
+        Ok(DeploymentId(id))
+    }
+}
+
+impl ToSql<Integer, Pg> for DeploymentId {
+    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> diesel::serialize::Result {
+        <i32 as ToSql<Integer, Pg>>::to_sql(&self.0, out)
+    }
+}
+
 /// A unique identifier for a deployment that specifies both its external
 /// identifier (`hash`) and its unique internal identifier (`id`) which
 /// ensures we are talking about a unique location for the deployment's data

store/postgres/src/primary.rs

@@ -450,6 +450,7 @@ mod queries {
     use diesel::sql_types::Text;
     use graph::prelude::NodeId;
     use graph::{
+        components::store::DeploymentId as GraphDeploymentId,
         data::subgraph::status,
         internal_error,
         prelude::{DeploymentHash, StoreError, SubgraphName},
@@ -646,18 +647,18 @@ mod queries {
         conn: &mut PgConnection,
         infos: &mut [status::Info],
     ) -> Result<(), StoreError> {
-        let ids: Vec<_> = infos.iter().map(|info| &info.subgraph).collect();
+        let ids: Vec<_> = infos.iter().map(|info| &info.id).collect();
         let nodes: HashMap<_, _> = a::table
             .inner_join(ds::table.on(ds::id.eq(a::id)))
-            .filter(ds::subgraph.eq_any(ids))
-            .select((ds::subgraph, a::node_id, a::paused_at.is_not_null()))
-            .load::<(String, String, bool)>(conn)?
+            .filter(ds::id.eq_any(ids))
+            .select((ds::id, a::node_id, a::paused_at.is_not_null()))
+            .load::<(GraphDeploymentId, String, bool)>(conn)?
             .into_iter()
-            .map(|(subgraph, node, paused)| (subgraph, (node, paused)))
+            .map(|(id, node, paused)| (id, (node, paused)))
             .collect();
         for info in infos {
-            info.node = nodes.get(&info.subgraph).map(|(node, _)| node.clone());
-            info.paused = nodes.get(&info.subgraph).map(|(_, paused)| *paused);
+            info.node = nodes.get(&info.id).map(|(node, _)| node.clone());
+            info.paused = nodes.get(&info.id).map(|(_, paused)| *paused);
         }
         Ok(())
     }