store: Fix ordering issues with different id types

Author: incrypto32

store/postgres/src/relational.rs

@@ -589,7 +589,7 @@ impl Layout {
         while lower_now.is_some() || upper_now.is_some() {
             let (ewt, block) = match (lower_now, upper_now) {
                 (Some(lower), Some(upper)) => {
-                    match lower.cmp(&upper) {
+                    match lower.compare_with(&upper, &IdType::String)? {
                         std::cmp::Ordering::Greater => {
                             // we have upper bound at this block, but no lower bounds at the same block so it's deletion
                             let (ewt, block) = transform(upper, EntityOperationKind::Delete)?;

store/postgres/src/relational_queries.rs

@@ -558,45 +558,78 @@ impl EntityData {
     }
 }
 
-#[derive(QueryableByName, Clone, Debug, Default, Eq)]
+#[derive(QueryableByName, Clone, Debug, Default)]
 pub struct EntityDataExt {
     #[diesel(sql_type = Text)]
     pub entity: String,
     #[diesel(sql_type = Jsonb)]
     pub data: serde_json::Value,
     #[diesel(sql_type = Integer)]
     pub block_number: i32,
-    #[diesel(sql_type = Text)]
-    pub id: String,
+    #[diesel(sql_type = Binary)]
+    pub id: Vec<u8>,
     #[diesel(sql_type = BigInt)]
     pub vid: i64,
 }
 
-impl Ord for EntityDataExt {
-    fn cmp(&self, other: &Self) -> Ordering {
-        let ord = self.block_number.cmp(&other.block_number);
-        if ord != Ordering::Equal {
-            ord
-        } else {
-            let ord = self.entity.cmp(&other.entity);
-            if ord != Ordering::Equal {
-                ord
-            } else {
-                self.id.cmp(&other.id)
-            }
+#[derive(Debug)]
+pub enum IdCompareError {
+    InvalidUtf8(std::string::FromUtf8Error),
+    InvalidInt8Length(usize),
+}
+
+impl From<std::string::FromUtf8Error> for IdCompareError {
+    fn from(e: std::string::FromUtf8Error) -> Self {
+        IdCompareError::InvalidUtf8(e)
+    }
+}
+
+impl From<IdCompareError> for anyhow::Error {
+    fn from(e: IdCompareError) -> Self {
+        match e {
+            IdCompareError::InvalidUtf8(e) => anyhow::anyhow!("Invalid UTF-8 in ID: {}", e),
+            IdCompareError::InvalidInt8Length(len) => anyhow::anyhow!("Invalid Int8 ID length: expected 8 bytes, got {}", len),
         }
     }
 }
 
-impl PartialOrd for EntityDataExt {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
+impl From<IdCompareError> for StoreError {
+    fn from(e: IdCompareError) -> Self {
+        StoreError::Unknown(anyhow::Error::from(e))
     }
 }
 
-impl PartialEq for EntityDataExt {
-    fn eq(&self, other: &Self) -> bool {
-        self.cmp(other) == Ordering::Equal
+impl EntityDataExt {
+    pub fn compare_with(&self, other: &Self, idtype: &IdType) -> Result<Ordering, IdCompareError> {
+        let ord = self.block_number.cmp(&other.block_number);
+        if ord != Ordering::Equal {
+            return Ok(ord);
+        }
+        let ord = self.entity.cmp(&other.entity);
+        if ord != Ordering::Equal {
+            return Ok(ord);
+        }
+
+        match idtype {
+            IdType::String => {
+                let self_str = String::from_utf8(self.id.clone()).map_err(IdCompareError::InvalidUtf8)?;
+                let other_str = String::from_utf8(other.id.clone()).map_err(IdCompareError::InvalidUtf8)?;
+                Ok(self_str.cmp(&other_str))
+            }
+            IdType::Bytes => Ok(self.id.cmp(&other.id)),
+            IdType::Int8 => {
+                if self.id.len() < 8 {
+                    return Err(IdCompareError::InvalidInt8Length(self.id.len()));
+                }
+                if other.id.len() < 8 {
+                    return Err(IdCompareError::InvalidInt8Length(other.id.len()));
+                }
+
+                let self_int = i64::from_be_bytes(self.id[..8].try_into().unwrap());
+                let other_int = i64::from_be_bytes(other.id[..8].try_into().unwrap());
+                Ok(self_int.cmp(&other_int))
+            }
+        }
     }
 }
 

store/test-store/tests/postgres/writable.rs

@@ -29,6 +29,20 @@ const SCHEMA_GQL: &str = "
         id: ID!,
         count: Int!,
     }
+    type PoolCreated @entity(immutable: true) {
+        id: Bytes!,
+        token0: Bytes!,
+        token1: Bytes!,
+        fee: Int!,
+        tickSpacing: Int!,
+        pool: Bytes!,
+        blockNumber: BigInt!,
+        blockTimestamp: BigInt!,
+        transactionHash: Bytes!,
+        transactionFrom: Bytes!,
+        transactionGasPrice: BigInt!,
+        logIndex: BigInt!
+    }
 ";
 
 const COUNTER: &str = "Counter";
@@ -406,3 +420,71 @@ fn read_immutable_only_range_test() {
         assert_eq!(e.len(), 4);
     })
 }
+
+#[test]
+fn read_range_pool_created_test() {
+    run_test(|store, writable, sourceable, deployment| async move {
+        let result_entities = vec![
+            format!("(1, [EntitySourceOperation {{ entity_op: Create, entity_type: EntityType(PoolCreated), entity: Entity {{ blockNumber: BigInt(12369621), blockTimestamp: BigInt(1620243254), fee: Int(500), id: Bytes(0xff80818283848586), logIndex: BigInt(0), pool: Bytes(0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8), tickSpacing: Int(10), token0: Bytes(0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48), token1: Bytes(0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2), transactionFrom: Bytes(0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48), transactionGasPrice: BigInt(100000000000), transactionHash: Bytes(0x12340000000000000000000000000000000000000000000000000000000000000000000000000000) }}, vid: 1 }}])"),
+            format!("(2, [EntitySourceOperation {{ entity_op: Create, entity_type: EntityType(PoolCreated), entity: Entity {{ blockNumber: BigInt(12369622), blockTimestamp: BigInt(1620243255), fee: Int(3000), id: Bytes(0xff90919293949596), logIndex: BigInt(1), pool: Bytes(0x4585fe77225b41b697c938b018e2ac67ac5a20c0), tickSpacing: Int(60), token0: Bytes(0x2260fac5e5542a773aa44fbcfedf7c193bc2c599), token1: Bytes(0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2), transactionFrom: Bytes(0x2260fac5e5542a773aa44fbcfedf7c193bc2c599), transactionGasPrice: BigInt(100000000000), transactionHash: Bytes(0x12340000000000000000000000000000000000000000000000000000000000000000000000000001) }}, vid: 2 }}])"),
+        ];
+        
+        // Rest of the test remains the same
+        let subgraph_store = store.subgraph_store();
+        writable.deployment_synced().unwrap();
+
+        let pool_created_type = TEST_SUBGRAPH_SCHEMA.entity_type("PoolCreated").unwrap();
+        let entity_types = vec![pool_created_type.clone()];
+
+        for count in (1..=2).map(|x| x as i64) {
+            let id = if count == 1 {
+                "0xff80818283848586"
+            } else {
+                "0xff90919293949596"
+            };
+
+            let data = entity! { TEST_SUBGRAPH_SCHEMA =>
+                id: id,
+                token0: if count == 1 { "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" } else { "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599" },
+                token1: "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
+                fee: if count == 1 { 500 } else { 3000 },
+                tickSpacing: if count == 1 { 10 } else { 60 },
+                pool: if count == 1 { "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8" } else { "0x4585fe77225b41b697c938b018e2ac67ac5a20c0" },
+                blockNumber: 12369621 + count - 1,
+                blockTimestamp: 1620243254 + count - 1,
+                transactionHash: format!("0x1234{:0>76}", if count == 1 { "0" } else { "1" }),
+                transactionFrom: if count == 1 { "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" } else { "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599" },
+                transactionGasPrice: 100000000000i64,
+                logIndex: count - 1
+            };
+
+            let key = pool_created_type.parse_key(id).unwrap();
+            let op = EntityOperation::Set {
+                key: key.clone(),
+                data: EntityV::new(data, count),
+            };
+
+            transact_entity_operations(
+                &subgraph_store,
+                &deployment,
+                block_pointer(count as u8),
+                vec![op],
+            )
+            .await
+            .unwrap();
+        }
+        writable.flush().await.unwrap();
+        writable.deployment_synced().unwrap();
+
+        let br: Range<BlockNumber> = 0..18;
+        let e: BTreeMap<i32, Vec<EntitySourceOperation>> = sourceable
+            .get_range(entity_types.clone(), CausalityRegion::ONCHAIN, br.clone())
+            .unwrap();
+        assert_eq!(e.len(), 2);
+        for en in &e {
+            let index = *en.0 - 1;
+            let a = result_entities[index as usize].clone();
+            assert_eq!(a, format!("{:?}", en));
+        }
+    })
+}