linkrs-storage 0.1.0

Storage engine with CSR and memory-mapped containers
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

//! Engine-level column statistics snapshot resolution.
//!
//! Resolves `(space, tag/edge_type, column)` to physical storage and
//! aggregates optimizer-facing snapshots:
//! - vertex side: per-shard zone-map bounds merged across all shards of the
//!   tag's [`ShardedVertexTable`];
//! - edge side: zone-map aggregated stats merged across all partitions of
//!   the edge type.
//!
//! All reads take short-lived table locks and touch only chunk metadata —
//! no rows are materialized (the property-table path falls back to a column
//! scan only when no zone maps have been built yet).

use std::sync::Arc;

use super::context::GraphStorageContext;
use super::ops::{edge_label_id, tag_label_id};
use crate::stats_reader::ColumnStatsSnapshot;
use crate::stats_reader::TableCardinalitySnapshot;

pub(crate) fn vertex_column_stats(
    ctx: &GraphStorageContext,
    space: &str,
    tag: &str,
    column: &str,
) -> Option<Arc<ColumnStatsSnapshot>> {
    let label = tag_label_id(ctx, space, tag).ok()??;
    let snapshot = ctx.data_store().with_vertex_tables(|tables| {
        tables
            .get(&label)
            .and_then(|table| table.column_stats_snapshot_at(column, ctx.get_read_timestamp()))
    })?;
    Some(Arc::new(snapshot))
}

pub(crate) fn vertex_table_stats(
    ctx: &GraphStorageContext,
    space: &str,
    tag: &str,
) -> Option<Arc<TableCardinalitySnapshot>> {
    let label = tag_label_id(ctx, space, tag).ok()??;
    let snapshot = ctx.data_store().with_vertex_tables(|tables| {
        tables
            .get(&label)
            .map(|table| table.table_cardinality_at(ctx.get_read_timestamp()))
    })?;
    Some(Arc::new(snapshot))
}

pub(crate) fn edge_table_stats(
    ctx: &GraphStorageContext,
    space: &str,
    edge_type: &str,
) -> Option<Arc<TableCardinalitySnapshot>> {
    let label = edge_label_id(ctx, space, edge_type).ok()??;
    let keys = ctx.data_store().edge_partition_keys(label).ok()?;
    if keys.is_empty() {
        return None;
    }

    // Snapshot-consistent sizing: same read timestamp and pending gate as
    // scans, merged across partitions. Per-partition snapshots carry
    // shard_count 1 so the merged count equals the partition count.
    let ts = ctx.get_read_timestamp();
    let gate = ctx.pending_gate();
    let mut acc = TableCardinalitySnapshot::default();
    for key in &keys {
        let snapshot = ctx
            .data_store()
            .with_single_edge_table(key, |store| Ok(store.table_cardinality_at(ts, &gate)))
            .ok()?;
        acc.absorb(&snapshot);
    }
    if acc.is_empty() {
        return None;
    }

    Some(Arc::new(acc))
}

pub(crate) fn edge_column_stats(
    ctx: &GraphStorageContext,
    space: &str,
    edge_type: &str,
    column: &str,
) -> Option<Arc<ColumnStatsSnapshot>> {
    let label = edge_label_id(ctx, space, edge_type).ok()??;
    let keys = ctx.data_store().edge_partition_keys(label).ok()?;
    if keys.is_empty() {
        return None;
    }

    let mut acc = ColumnStatsSnapshot::default();
    for key in &keys {
        let snapshot = ctx
            .data_store()
            .with_single_edge_table(key, |store| Ok(store.column_stats_snapshot(column)))
            .ok()??;
        acc.absorb(&snapshot);
    }
    if !acc.has_envelope() && acc.row_count == 0 {
        return None;
    }

    Some(Arc::new(acc))
}