Skip to content

Commit e03a549

Browse files
Zoran Cvetkovincrypto32
Zoran Cvetkov
authored and
incrypto32
committed
Add casuality region to the SQL querry
1 parent 74093de commit e03a549

File tree

9 files changed

+38
-18
lines changed

9 files changed

+38
-18
lines changed

‎graph/src/blockchain/block_stream.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
use crate::blockchain::SubgraphFilter;
2-
use crate::data_source::subgraph;
2+
use crate::data_source::{subgraph, CausalityRegion};
33
use crate::substreams::Clock;
44
use crate::substreams_rpc::response::Message as SubstreamsMessage;
55
use crate::substreams_rpc::BlockScopedData;
@@ -453,7 +453,7 @@ async fn get_entities_for_range(
453453
let entity_type = schema.entity_type(entity_name)?;
454454
entity_types.push(entity_type);
455455
}
456-
Ok(store.get_range(entity_types, from..to)?)
456+
Ok(store.get_range(entity_types, CausalityRegion::ONCHAIN, from..to)?)
457457
}
458458

459459
impl<C: Blockchain> TriggersAdapterWrapper<C> {

‎graph/src/components/store/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1043,6 +1043,7 @@ impl ReadStore for EmptyStore {
10431043
fn get_range(
10441044
&self,
10451045
_entity_types: Vec<EntityType>,
1046+
_causality_region: CausalityRegion,
10461047
_block_range: Range<BlockNumber>,
10471048
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
10481049
Ok(BTreeMap::new())

‎graph/src/components/store/traits.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -232,6 +232,7 @@ pub trait ReadStore: Send + Sync + 'static {
232232
fn get_range(
233233
&self,
234234
entity_types: Vec<EntityType>,
235+
causality_region: CausalityRegion,
235236
block_range: Range<BlockNumber>,
236237
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError>;
237238

@@ -260,9 +261,10 @@ impl<T: ?Sized + ReadStore> ReadStore for Arc<T> {
260261
fn get_range(
261262
&self,
262263
entity_types: Vec<EntityType>,
264+
causality_region: CausalityRegion,
263265
block_range: Range<BlockNumber>,
264266
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
265-
(**self).get_range(entity_types, block_range)
267+
(**self).get_range(entity_types, causality_region, block_range)
266268
}
267269

268270
fn get_derived(

‎store/postgres/src/deployment_store.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1060,11 +1060,12 @@ impl DeploymentStore {
10601060
&self,
10611061
site: Arc<Site>,
10621062
entity_types: Vec<EntityType>,
1063+
causality_region: CausalityRegion,
10631064
block_range: Range<BlockNumber>,
10641065
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
10651066
let mut conn = self.get_conn()?;
10661067
let layout = self.layout(&mut conn, site)?;
1067-
layout.find_range(&mut conn, entity_types, block_range)
1068+
layout.find_range(&mut conn, entity_types, causality_region, block_range)
10681069
}
10691070

10701071
pub(crate) fn get_derived(

‎store/postgres/src/relational.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -520,6 +520,7 @@ impl Layout {
520520
&self,
521521
conn: &mut PgConnection,
522522
entity_types: Vec<EntityType>,
523+
causality_region: CausalityRegion,
523524
block_range: Range<BlockNumber>,
524525
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
525526
let mut tables = vec![];
@@ -529,11 +530,11 @@ impl Layout {
529530
et_map.insert(et.to_string(), Arc::new(et));
530531
}
531532
let mut entities: BTreeMap<BlockNumber, Vec<EntityWithType>> = BTreeMap::new();
532-
let lower_vec = FindRangeQuery::new(&tables, false, block_range.clone())
533+
let lower_vec = FindRangeQuery::new(&tables, causality_region, false, block_range.clone())
533534
.get_results::<EntityDataExt>(conn)
534535
.optional()?
535536
.unwrap_or_default();
536-
let upper_vec = FindRangeQuery::new(&tables, true, block_range)
537+
let upper_vec = FindRangeQuery::new(&tables, causality_region, true, block_range)
537538
.get_results::<EntityDataExt>(conn)
538539
.optional()?
539540
.unwrap_or_default();

‎store/postgres/src/relational_queries.rs‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2024,6 +2024,7 @@ impl<'a, Conn> RunQueryDsl<Conn> for FindQuery<'a> {}
20242024
#[derive(Debug, Clone)]
20252025
pub struct FindRangeQuery<'a> {
20262026
tables: &'a Vec<&'a Table>,
2027+
causality_region: CausalityRegion,
20272028
is_upper_range: bool,
20282029
imm_range: EntityBlockRange,
20292030
mut_range: EntityBlockRange,
@@ -2032,13 +2033,15 @@ pub struct FindRangeQuery<'a> {
20322033
impl<'a> FindRangeQuery<'a> {
20332034
pub fn new(
20342035
tables: &'a Vec<&Table>,
2036+
causality_region: CausalityRegion,
20352037
is_upper_range: bool,
20362038
block_range: Range<BlockNumber>,
20372039
) -> Self {
20382040
let imm_range = EntityBlockRange::new(true, block_range.clone(), false);
20392041
let mut_range = EntityBlockRange::new(false, block_range, is_upper_range);
20402042
Self {
20412043
tables,
2044+
causality_region,
20422045
is_upper_range,
20432046
imm_range,
20442047
mut_range,
@@ -2075,12 +2078,12 @@ impl<'a> QueryFragment<Pg> for FindRangeQuery<'a> {
20752078
out.push_sql(" from ");
20762079
out.push_sql(table.qualified_name.as_str());
20772080
out.push_sql(" e\n where");
2078-
// TODO: add casuality region to the query
2079-
// if self.table.has_causality_region {
2080-
// out.push_sql("causality_region = ");
2081-
// out.push_bind_param::<Integer, _>(&self.key.causality_region)?;
2082-
// out.push_sql(" and ");
2083-
// }
2081+
// add casuality region to the query
2082+
if table.has_causality_region {
2083+
out.push_sql("causality_region = ");
2084+
out.push_bind_param::<Integer, _>(&self.causality_region)?;
2085+
out.push_sql(" and ");
2086+
}
20842087
if table.immutable {
20852088
self.imm_range.contains(&mut out)?;
20862089
} else {

‎store/postgres/src/writable.rs‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -355,12 +355,14 @@ impl SyncStore {
355355
fn get_range(
356356
&self,
357357
entity_types: Vec<EntityType>,
358+
causality_region: CausalityRegion,
358359
block_range: Range<BlockNumber>,
359360
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
360361
retry::forever(&self.logger, "get_range", || {
361362
self.writable.get_range(
362363
self.site.cheap_clone(),
363364
entity_types.clone(),
365+
causality_region,
364366
block_range.clone(),
365367
)
366368
})
@@ -1234,9 +1236,11 @@ impl Queue {
12341236
fn get_range(
12351237
&self,
12361238
entity_types: Vec<EntityType>,
1239+
causality_region: CausalityRegion,
12371240
block_range: Range<BlockNumber>,
12381241
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
1239-
self.store.get_range(entity_types, block_range)
1242+
self.store
1243+
.get_range(entity_types, causality_region, block_range)
12401244
}
12411245

12421246
fn get_derived(
@@ -1454,11 +1458,14 @@ impl Writer {
14541458
fn get_range(
14551459
&self,
14561460
entity_types: Vec<EntityType>,
1461+
causality_region: CausalityRegion,
14571462
block_range: Range<BlockNumber>,
14581463
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
14591464
match self {
1460-
Writer::Sync(store) => store.get_range(entity_types, block_range),
1461-
Writer::Async { queue, .. } => queue.get_range(entity_types, block_range),
1465+
Writer::Sync(store) => store.get_range(entity_types, causality_region, block_range),
1466+
Writer::Async { queue, .. } => {
1467+
queue.get_range(entity_types, causality_region, block_range)
1468+
}
14621469
}
14631470
}
14641471

@@ -1593,9 +1600,11 @@ impl ReadStore for WritableStore {
15931600
fn get_range(
15941601
&self,
15951602
entity_types: Vec<EntityType>,
1603+
causality_region: CausalityRegion,
15961604
block_range: Range<BlockNumber>,
15971605
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
1598-
self.writer.get_range(entity_types, block_range)
1606+
self.writer
1607+
.get_range(entity_types, causality_region, block_range)
15991608
}
16001609

16011610
fn get_derived(

‎store/test-store/tests/graph/entity_cache.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@ impl ReadStore for MockStore {
7070
fn get_range(
7171
&self,
7272
_entity_types: Vec<EntityType>,
73+
_causality_region: CausalityRegion,
7374
_block_range: Range<BlockNumber>,
7475
) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
7576
todo!()

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -343,7 +343,7 @@ fn read_range_test() {
343343
let br: Range<BlockNumber> = 0..18;
344344
let entity_types = vec![COUNTER_TYPE.clone(), COUNTER2_TYPE.clone()];
345345
let e: BTreeMap<i32, Vec<EntityWithType>> = writable
346-
.get_range(entity_types.clone(), br.clone())
346+
.get_range(entity_types.clone(), CausalityRegion::ONCHAIN, br.clone())
347347
.unwrap();
348348
assert_eq!(e.len(), 5);
349349
for en in &e {
@@ -356,7 +356,9 @@ fn read_range_test() {
356356
}
357357
writable.flush().await.unwrap();
358358
writable.deployment_synced().unwrap();
359-
let e: BTreeMap<i32, Vec<EntityWithType>> = writable.get_range(entity_types, br).unwrap();
359+
let e: BTreeMap<i32, Vec<EntityWithType>> = writable
360+
.get_range(entity_types, CausalityRegion::ONCHAIN, br)
361+
.unwrap();
360362
assert_eq!(e.len(), 7);
361363
for en in &e {
362364
let index = *en.0 - 1;

0 commit comments

Comments
 (0)