engramdb_core/
store.rs

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

engramdb_core/
store.rs
