sui_rpc_store/indexer/
transaction_bitmap.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Sequential pipeline that populates the
5//! [`schema::transaction_bitmap`](crate::schema::transaction_bitmap)
6//! CF.
7//!
8//! For every transaction in the checkpoint the pipeline:
9//!
10//! 1. Visits every dimension candidate via
11//!    [`sui_inverted_index::for_each_transaction_dimension`].
12//! 2. Encodes each `(dimension, value)` into a `dimension_key` via
13//!    [`sui_inverted_index::encode_dimension_key`] and dedupes
14//!    per-tx, so a transaction matching the same dimension
15//!    multiple times contributes a single bit per
16//!    `(dim_key, bucket)`.
17//! 3. Groups `tx_seq` bits by `(dim_key, tx_seq / TX_BUCKET_SIZE)`,
18//!    folding any number of transactions in the checkpoint into
19//!    one `RoaringBitmap` per group.
20//!
21//! Multiple checkpoints landing in the same commit batch are
22//! folded into the same `RoaringBitmap` per group via the
23//! handler's `batch` callback, so the commit path emits one
24//! merge operand per `(dim_key, bucket)` regardless of how many
25//! checkpoints contributed.
26
27use std::collections::HashMap;
28use std::collections::HashSet;
29use std::sync::Arc;
30
31use async_trait::async_trait;
32use roaring::RoaringBitmap;
33use sui_indexer_alt_framework::pipeline::Processor;
34use sui_indexer_alt_framework::pipeline::sequential;
35use sui_inverted_index::encode_dimension_key;
36use sui_inverted_index::for_each_transaction_dimension;
37use sui_types::full_checkpoint_content::Checkpoint;
38
39use crate::indexer::Schema;
40use crate::indexer::Store;
41use crate::indexer::tx_seq_at;
42use crate::schema::transaction_bitmap;
43use crate::schema::transaction_bitmap::TX_BUCKET_SIZE;
44use crate::schema::transaction_bitmap::bit_of;
45use crate::schema::transaction_bitmap::bucket_of;
46
47/// Pipeline marker for `transaction_bitmap`.
48pub struct TransactionBitmap;
49
50/// One pre-built bitmap for a single `(dimension_key, bucket)`
51/// pair, ready to be staged as a merge operand against the CF.
52pub struct Row {
53    pub dimension_key: Vec<u8>,
54    pub bucket: u64,
55    pub bitmap: RoaringBitmap,
56}
57
58#[async_trait]
59impl Processor for TransactionBitmap {
60    const NAME: &'static str = "transaction_bitmap";
61    type Value = Row;
62
63    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Row>> {
64        let mut groups: HashMap<(Vec<u8>, u64), RoaringBitmap> = HashMap::new();
65        let mut dim_keys: HashSet<Vec<u8>> = HashSet::new();
66
67        for (i, tx) in checkpoint.transactions.iter().enumerate() {
68            let tx_seq = tx_seq_at(checkpoint, i);
69            let bucket = bucket_of(tx_seq);
70            let bit = bit_of(tx_seq);
71
72            // Dedupe `(dim, value)` pairs that occur multiple
73            // times in the same tx (e.g. AffectedAddress when an
74            // address shows up in several object changes).
75            // Without this we'd add the same bit to the bitmap
76            // repeatedly — not incorrect, but redundant work.
77            dim_keys.clear();
78            for_each_transaction_dimension(
79                &tx.transaction,
80                &tx.effects,
81                tx.events.as_ref(),
82                &checkpoint.object_set,
83                |dim, value| {
84                    dim_keys.insert(encode_dimension_key(dim, value));
85                },
86            );
87
88            for dim_key in dim_keys.drain() {
89                groups.entry((dim_key, bucket)).or_default().insert(bit);
90            }
91        }
92
93        Ok(groups
94            .into_iter()
95            .map(|((dim_key, bucket), bitmap)| Row {
96                dimension_key: dim_key,
97                bucket,
98                bitmap,
99            })
100            .collect())
101    }
102}
103
104#[async_trait]
105impl sequential::Handler for TransactionBitmap {
106    type Store = Store;
107    /// Fold operands from multiple checkpoints together so we
108    /// emit at most one merge operand per `(dim_key, bucket)`
109    /// per commit — even if many checkpoints land in the same
110    /// batch.
111    type Batch = HashMap<(Vec<u8>, u64), RoaringBitmap>;
112
113    fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Row>) {
114        for row in values {
115            let entry = batch.entry((row.dimension_key, row.bucket)).or_default();
116            *entry |= row.bitmap;
117        }
118    }
119
120    async fn commit<'a>(
121        &self,
122        batch: &Self::Batch,
123        conn: &mut sui_consistent_store::Connection<'a, Schema>,
124    ) -> anyhow::Result<usize> {
125        let cf = &conn.store.schema().transaction_bitmap;
126        for ((dim_key, bucket), bitmap) in batch {
127            let (k, v) = transaction_bitmap::store_bitmap(dim_key.clone(), *bucket, bitmap.clone());
128            conn.batch.merge(cf, &k, &v)?;
129        }
130        Ok(batch.len())
131    }
132}
133
134// Re-export for documentation cross-referencing — silence the
135// "unused import" lint without an `#[allow]`.
136#[allow(dead_code)]
137const _BUCKET_SIZE_DOC: u64 = TX_BUCKET_SIZE;
138
139#[cfg(test)]
140mod tests {
141    use std::sync::Arc;
142
143    use sui_types::test_checkpoint_data_builder::TestCheckpointBuilder;
144
145    use super::*;
146
147    #[tokio::test]
148    async fn process_runs_against_synthetic_checkpoint() {
149        let checkpoint = Arc::new(TestCheckpointBuilder::new(1).build_checkpoint());
150        let _ = TransactionBitmap.process(&checkpoint).await.unwrap();
151    }
152}
sui_rpc_store/indexer/transaction_bitmap.rs

sui_rpc_store/indexer/
transaction_bitmap.rs
