banyan_utils/
tags.rs

1use crate::{tag_index::map_to_index_set, tag_index::TagIndex, tag_index::TagSet};
2use banyan::query::Query;
3use banyan::{index::*, TreeTypes};
4use libipld::{
5    cbor::DagCborCodec,
6    codec::{Decode, Encode},
7    Cid, DagCbor,
8};
9use multihash::{Code, MultihashDigest};
10use serde::{Deserialize, Serialize};
11use std::{
12    convert::{TryFrom, TryInto},
13    fmt,
14    io::{Read, Seek, Write},
15    iter::FromIterator,
16    str::FromStr,
17};
18use vec_collections::VecSet;
19
20#[derive(Debug, Clone)]
21pub struct TT {}
22
23#[derive(Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
24pub struct Sha256Digest([u8; 32]);
25
26impl Decode<DagCborCodec> for Sha256Digest {
27    fn decode<R: Read + Seek>(c: DagCborCodec, r: &mut R) -> anyhow::Result<Self> {
28        Self::try_from(Cid::decode(c, r)?)
29    }
30}
31impl Encode<DagCborCodec> for Sha256Digest {
32    fn encode<W: Write>(&self, c: DagCborCodec, w: &mut W) -> anyhow::Result<()> {
33        Cid::encode(&Cid::from(*self), c, w)
34    }
35}
36
37impl Sha256Digest {
38    pub fn new(data: &[u8]) -> Self {
39        let mh = multihash::Code::Sha2_256.digest(data);
40        Sha256Digest(mh.digest().try_into().unwrap())
41    }
42
43    pub fn digest(data: &[u8]) -> Self {
44        let mh = Code::Sha2_256.digest(data);
45        Sha256Digest(mh.digest().try_into().unwrap())
46    }
47}
48
49impl From<Sha256Digest> for Cid {
50    fn from(value: Sha256Digest) -> Self {
51        // https://github.com/multiformats/multicodec/blob/master/table.csv
52        let mh = multihash::Multihash::wrap(0x12, &value.0).unwrap();
53        Cid::new_v1(0x71, mh)
54    }
55}
56
57impl TryFrom<Cid> for Sha256Digest {
58    type Error = anyhow::Error;
59
60    fn try_from(value: Cid) -> Result<Self, Self::Error> {
61        anyhow::ensure!(value.codec() == 0x71, "Unexpected codec");
62        anyhow::ensure!(value.hash().code() == 0x12, "Unexpected hash algorithm");
63        let digest: [u8; 32] = value.hash().digest().try_into()?;
64        Ok(Self(digest))
65    }
66}
67
68impl FromStr for Sha256Digest {
69    type Err = anyhow::Error;
70    fn from_str(s: &str) -> Result<Self, Self::Err> {
71        let cid = Cid::from_str(s)?;
72        cid.try_into()
73    }
74}
75
76impl fmt::Display for Sha256Digest {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        write!(f, "{}", Cid::from(*self))
79    }
80}
81
82impl fmt::Debug for Sha256Digest {
83    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84        write!(f, "{}", Cid::from(*self))
85    }
86}
87
88impl TreeTypes for TT {
89    type Key = Key;
90    type KeySeq = KeySeq;
91    type Summary = Key;
92    type SummarySeq = KeySeq;
93    type Link = Sha256Digest;
94}
95
96#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
97pub struct Key {
98    time: TimeData,
99    tags: TagSet,
100}
101
102#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
103pub struct TimeData {
104    min_lamport: u64,
105    min_time: u64,
106    max_time: u64,
107}
108
109impl TimeData {
110    fn intersects(&self, that: &Self) -> bool {
111        if self.max_time < that.min_time {
112            return false;
113        }
114        if self.min_time > that.max_time {
115            return false;
116        }
117        true
118    }
119
120    fn contains(&self, that: &Self) -> bool {
121        if that.min_time < self.min_time {
122            return false;
123        }
124        if that.max_time > self.max_time {
125            return false;
126        }
127        true
128    }
129
130    fn combine(&mut self, b: &Self) {
131        self.min_lamport = self.min_lamport.min(b.min_lamport);
132        self.min_time = self.min_time.min(b.min_time);
133        self.max_time = self.max_time.max(b.max_time);
134    }
135}
136
137#[allow(dead_code)]
138impl Key {
139    pub fn single(lamport: u64, time: u64, tags: TagSet) -> Self {
140        Self {
141            time: TimeData {
142                min_lamport: lamport,
143                min_time: time,
144                max_time: time,
145            },
146            tags,
147        }
148    }
149
150    pub fn filter_tags(tags: TagSet) -> Self {
151        Self {
152            time: TimeData {
153                min_lamport: u64::MIN,
154                min_time: u64::MIN,
155                max_time: u64::MAX,
156            },
157            tags,
158        }
159    }
160
161    pub fn range(min_time: u64, max_time: u64, tags: TagSet) -> Self {
162        Self {
163            time: TimeData {
164                min_lamport: 0,
165                min_time,
166                max_time,
167            },
168            tags,
169        }
170    }
171
172    fn intersects(&self, that: &Key) -> bool {
173        self.time.intersects(&that.time) && !self.tags.is_disjoint(&that.tags)
174    }
175
176    fn contains(&self, that: &Key) -> bool {
177        self.time.contains(&that.time) && self.tags.is_superset(&that.tags)
178    }
179}
180
181#[allow(dead_code)]
182impl Key {
183    fn combine(&mut self, b: &Self) {
184        self.time.combine(&b.time);
185        self.tags.extend(b.tags.iter().cloned());
186    }
187}
188
189#[derive(Debug)]
190pub struct DnfQuery(pub Vec<Key>);
191
192#[allow(dead_code)]
193impl DnfQuery {
194    fn intersects(&self, v: &Key) -> bool {
195        self.0.iter().any(|x| x.intersects(v))
196    }
197    fn contains(&self, v: &Key) -> bool {
198        self.0.iter().any(|x| x.contains(v))
199    }
200    fn map_into<'a>(&self, keyseq: &'a KeySeq) -> TranslatedDnfQuery<'a> {
201        TranslatedDnfQuery {
202            query: self
203                .0
204                .iter()
205                .map(|key| {
206                    map_to_index_set(&keyseq.tags.tags, &key.tags).map(|index_set| TranslatedKey {
207                        index_set,
208                        time: key.time,
209                    })
210                })
211                .collect::<Option<Vec<_>>>()
212                .unwrap_or_default(),
213            seq: keyseq,
214        }
215    }
216}
217
218type IndexSet = VecSet<[u32; 4]>;
219
220struct TranslatedKey {
221    index_set: IndexSet,
222    time: TimeData,
223}
224
225struct TranslatedDnfQuery<'a> {
226    query: Vec<TranslatedKey>,
227    seq: &'a KeySeq,
228}
229
230impl<'a> TranslatedDnfQuery<'a> {
231    fn intersects(&self, i: usize) -> bool {
232        self.query.iter().any(|q| {
233            q.time.intersects(&self.seq.time(i).unwrap())
234                && !q.index_set.is_disjoint(&self.seq.tags.elements[i])
235        })
236    }
237
238    fn contains(&self, i: usize) -> bool {
239        self.query.iter().any(|q| {
240            q.time.contains(&self.seq.time(i).unwrap())
241                && q.index_set.is_superset(&self.seq.tags.elements[i])
242        })
243    }
244
245    fn intersecting(&self, matching: &mut [bool]) {
246        for i in 0..self.seq.len().min(matching.len()) {
247            if matching[i] {
248                matching[i] = self.intersects(i);
249            }
250        }
251    }
252
253    fn containing(&self, matching: &mut [bool]) {
254        for i in 0..self.seq.len().min(matching.len()) {
255            if matching[i] {
256                matching[i] = self.contains(i);
257            }
258        }
259    }
260}
261
262impl Query<TT> for DnfQuery {
263    fn intersecting(&self, _: u64, x: &BranchIndex<TT>, matching: &mut [bool]) {
264        self.map_into(&x.summaries).intersecting(matching);
265        // for i in 0..x.summaries.len().min(matching.len()) {
266        //     if matching[i] {
267        //         matching[i] = self.intersects(&x.summaries.get(i).unwrap());
268        //     }
269        // }
270    }
271    fn containing(&self, _: u64, x: &LeafIndex<TT>, matching: &mut [bool]) {
272        self.map_into(&x.keys).containing(matching);
273        // for i in 0..x.keys.len().min(matching.len()) {
274        //     if matching[i] {
275        //         matching[i] = self.contains(&x.keys.get(i).unwrap());
276        //     }
277        // }
278    }
279}
280
281#[derive(Clone, Debug, Serialize, Deserialize, DagCbor)]
282pub struct KeySeq {
283    min_lamport: Vec<u64>,
284    min_time: Vec<u64>,
285    max_time: Vec<u64>,
286    tags: TagIndex,
287}
288
289impl KeySeq {
290    fn time(&self, i: usize) -> Option<TimeData> {
291        if i < self.min_lamport.len() {
292            Some(TimeData {
293                min_time: self.min_time[i],
294                max_time: self.max_time[i],
295                min_lamport: self.min_lamport[i],
296            })
297        } else {
298            None
299        }
300    }
301}
302
303impl CompactSeq for KeySeq {
304    type Item = Key;
305
306    fn get(&self, index: usize) -> Option<Key> {
307        if let (Some(min_lamport), Some(min_time), Some(max_time), Some(tags)) = (
308            self.min_lamport.get(index),
309            self.min_time.get(index),
310            self.max_time.get(index),
311            self.tags.get(index),
312        ) {
313            Some(Key {
314                time: TimeData {
315                    min_lamport: *min_lamport,
316                    min_time: *min_time,
317                    max_time: *max_time,
318                },
319                tags,
320            })
321        } else {
322            None
323        }
324    }
325
326    fn len(&self) -> usize {
327        self.tags.elements.len()
328    }
329
330    fn estimated_size(&self) -> usize {
331        self.len() * 128
332    }
333}
334
335impl Summarizable<Key> for KeySeq {
336    fn summarize(&self) -> Key {
337        let max_time = *self.max_time.iter().max().unwrap();
338        let min_time = *self.min_time.iter().min().unwrap();
339        let min_lamport = *self.min_lamport.iter().min().unwrap();
340        let tags = self.tags.tags.clone();
341        Key {
342            time: TimeData {
343                min_lamport,
344                min_time,
345                max_time,
346            },
347            tags,
348        }
349        // let mut result = self.get(0).unwrap();
350        // for i in 1..self.tags.elements.len() {
351        //     result.combine(&self.get(i).unwrap());
352        // }
353        // result
354    }
355}
356
357impl FromIterator<Key> for KeySeq {
358    fn from_iter<T: IntoIterator<Item = Key>>(iter: T) -> Self {
359        let mut min_lamport = Vec::new();
360        let mut min_time = Vec::new();
361        let mut max_time = Vec::new();
362        let mut tag_index = Vec::new();
363        for value in iter.into_iter() {
364            min_lamport.push(value.time.min_lamport);
365            min_time.push(value.time.min_time);
366            max_time.push(value.time.max_time);
367            tag_index.push(value.tags);
368        }
369        let tag_index = TagIndex::from_elements(&tag_index);
370        Self {
371            min_lamport,
372            min_time,
373            max_time,
374            tags: tag_index,
375        }
376    }
377}
banyan_utils/tags.rs

banyan_utils/
tags.rs
