engramdb_core/
store.rs

1//! Store-I 的批量 gather：分片文件 + 行号 → 组装 `[n, width]` 缓冲区。
2//!
3//! 最小实现：每个 shard 一个文件，`read_at` 按 badge 读取并拷贝行。
4//! 大数据量生产版：io_uring / preadv 多队列、页缓存命中判定、双缓冲（M1 实现）。
5//! 本文件为 M0-P1 探针服务：真实调用路径（行号寻址）不变。
6
7use std::fs::File;
8use std::os::unix::fs::FileExt;
9use std::path::Path;
10
11use crate::layout::Layout;
12
13pub struct ShardedStore {
14    pub layout: Layout,
15    files: Vec<File>,
16}
17
18impl ShardedStore {
19    pub fn open(dir: &Path, layout: Layout) -> std::io::Result<Self> {
20        let mut files = Vec::with_capacity(layout.shards as usize);
21        for i in 0..layout.shards {
22            let p = dir.join(format!("shard_{:03}.bin", i));
23            files.push(File::open(p)?);
24        }
25        Ok(Self { layout, files })
26    }
27
28    /// 批取 `keys` 行到 `out[n*width]`（按 rowid 顺序；顺序无关，因为 gather 是按 key 的）。
29    pub fn gather(&self, keys: &[u64], out: &mut [u8]) -> std::io::Result<()> {
30        let w = self.layout.width as usize;
31        let rb = self.layout.row_bytes as usize;
32        let mut badge_buf = vec![0u8; self.layout.badge_bytes() as usize];
33        let mut current_badge = u64::MAX;
34        for (i, &k) in keys.iter().enumerate() {
35            let (shard, badge, in_badge) = self.layout.locate(k);
36            let off = badge * self.layout.badge_bytes();
37            if (shard, badge) != (current_badge, current_badge >> 32) {
38                // 一次读取 badge 块（4KB 对齐粒度）
39                self.files[shard as usize].read_exact_at(&mut badge_buf, off)?;
40                current_badge = shard << 32 | badge;
41            }
42            let src = in_badge as usize * rb;
43            let dst = i * w;
44            out[dst..dst + w].copy_from_slice(&badge_buf[src..src + w]);
45        }
46        Ok(())
47    }
48}
engramdb_core/store.rs

engramdb_core/
store.rs
