diff --git a/crates/datawal-cli/src/commands.rs b/crates/datawal-cli/src/commands.rs
index 62fa4b1..6846be8 100644
--- a/crates/datawal-cli/src/commands.rs
+++ b/crates/datawal-cli/src/commands.rs
@@ -126,7 +126,7 @@ fn cmd_get(args: GetArgs, json: bool) -> Result<ExitCode> {
     let mode: BytesMode = args.bytes.into();
     let truncate = truncate_for(args.no_truncate);
 
-    let store = DataWal::open(&args.store)
+    let mut store = DataWal::open(&args.store)
         .with_context(|| format!("open store {}", args.store.display()))?;
     match store.get(&key)? {
         Some(value) => {
diff --git a/crates/datawal-core/benches/compaction.rs b/crates/datawal-core/benches/compaction.rs
index fe54c46..d86b666 100644
--- a/crates/datawal-core/benches/compaction.rs
+++ b/crates/datawal-core/benches/compaction.rs
@@ -83,7 +83,7 @@ fn bench_compact_to_delete_heavy(c: &mut Criterion) {
     let mut group = c.benchmark_group("datawal_compact_to_delete_heavy");
 
     for &(label, ratio) in LIVE_RATIOS {
-        let (_src, kv) = populated_with_deletes(ratio);
+        let (_src, mut kv) = populated_with_deletes(ratio);
 
         group.bench_function(BenchmarkId::from_parameter(label), |b| {
             b.iter_with_setup(bench_tempdir, |out_dir| {
@@ -101,7 +101,7 @@ fn bench_compact_to_overwrite_heavy(c: &mut Criterion) {
     let mut group = c.benchmark_group("datawal_compact_to_overwrite_heavy");
 
     for &(label, ratio) in LIVE_RATIOS {
-        let (_src, kv) = populated_with_overwrites(ratio);
+        let (_src, mut kv) = populated_with_overwrites(ratio);
 
         group.bench_function(BenchmarkId::from_parameter(label), |b| {
             b.iter_with_setup(bench_tempdir, |out_dir| {
@@ -118,7 +118,7 @@ fn bench_export_jsonl(c: &mut Criterion) {
     let mut group = c.benchmark_group("datawal_export_jsonl");
 
     for &(label, ratio) in LIVE_RATIOS {
-        let (_src, kv) = populated_with_deletes(ratio);
+        let (_src, mut kv) = populated_with_deletes(ratio);
 
         group.bench_function(BenchmarkId::from_parameter(label), |b| {
             b.iter_with_setup(
diff --git a/crates/datawal-core/benches/datawal_kv.rs b/crates/datawal-core/benches/datawal_kv.rs
index f47534e..cf035f7 100644
--- a/crates/datawal-core/benches/datawal_kv.rs
+++ b/crates/datawal-core/benches/datawal_kv.rs
@@ -72,7 +72,7 @@ fn bench_get(c: &mut Criterion) {
 
     for &n in KEYDIR_SIZES {
         group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
-            let (_dir, kv) = populated_kv(n);
+            let (_dir, mut kv) = populated_kv(n);
             // Round-robin over the populated key range.
             let mut i = 0usize;
             b.iter(|| {
diff --git a/crates/datawal-core/examples/datawal_kv_demo.rs b/crates/datawal-core/examples/datawal_kv_demo.rs
index f10b383..4f96384 100644
--- a/crates/datawal-core/examples/datawal_kv_demo.rs
+++ b/crates/datawal-core/examples/datawal_kv_demo.rs
@@ -57,7 +57,7 @@ fn main() -> Result<()> {
     );
 
     // Reopen the compacted directory and verify state.
-    let kv2 = DataWal::open(&compact_dir)?;
+    let mut kv2 = DataWal::open(&compact_dir)?;
     println!("after compact reopen: len={}", kv2.len());
     println!("  a = {:?}", kv2.get(b"a")?);
     println!("  b contains_key = {}", kv2.contains_key(b"b"));
diff --git a/crates/datawal-core/examples/soak.rs b/crates/datawal-core/examples/soak.rs
index 73beb9f..e573248 100644
--- a/crates/datawal-core/examples/soak.rs
+++ b/crates/datawal-core/examples/soak.rs
@@ -452,7 +452,7 @@ fn run(cli: Cli) -> Result<()> {
     drop(log);
 
     // Final-state consistency: reopen and compare keydir to `expected`.
-    let reopen = DataWal::open(&cli.work_dir).context("final reopen")?;
+    let mut reopen = DataWal::open(&cli.work_dir).context("final reopen")?;
     let observed_keys: std::collections::HashSet<Vec<u8>> = reopen.keys().into_iter().collect();
     let expected_keys: std::collections::HashSet<Vec<u8>> = expected.keys().cloned().collect();
 
diff --git a/crates/datawal-core/src/datawal.rs b/crates/datawal-core/src/datawal.rs
index ed3bbb0..e20aa2c 100644
--- a/crates/datawal-core/src/datawal.rs
+++ b/crates/datawal-core/src/datawal.rs
@@ -1,22 +1,40 @@
 //! Last-write-wins bytes-based KV projection over [`RecordLog`].
 //!
-//! v0.1-pre semantics:
+//! v0.1.4 semantics:
 //! - `put(key, value)` appends a `Put` record and updates the in-memory keydir.
 //! - `delete(key)` appends a `Delete` (tombstone) record and removes the key
 //!   from the keydir.
-//! - `get(key)` reads from the in-memory keydir (payload is held in memory).
-//! - `open(dir)` rebuilds the keydir by scanning the log from segment 1 in
-//!   order; the on-disk order is authoritative and **last-write-wins**.
+//! - `get(key)` looks up the keydir, then performs a positional read against
+//!   the segment file (`pread` on Unix), validates the per-record CRC32C, and
+//!   returns the payload bytes. Takes `&mut self` because it maintains an
+//!   internal LRU of file descriptors (see [`fd_pool`](crate::fd_pool)).
+//! - `open(dir)` rebuilds the keydir by scanning the log lazily via
+//!   [`RecordLog::scan_iter`]; payloads are **not** materialised in memory.
 //! - `compact_to(out_dir)` writes a new datawal log into `out_dir` containing
-//!   exactly one `Put` per live key. The original log is untouched.
+//!   exactly one `Put` per live key. Payloads are read from the source via
+//!   the same `get` path used by callers (CRC re-validated). The original
+//!   log is untouched.
 //! - `export_jsonl(out)` writes one JSON object per live key to `out` using
 //!   `safeatomic_rs::write_atomic` (base64-encoded bytes).
 //!
-//! Out of scope for v0.1-pre:
+//! ## Why `get` takes `&mut self`
+//!
+//! Pre-0.1.4 the keydir stored the full payload alongside the key, so
+//! `get` was a pure HashMap lookup that fit `&self`. In 0.1.4 the keydir
+//! stores a [`RecordRef`] (segment + offset + length) instead, and `get`
+//! performs disk I/O plus CRC validation. To avoid re-opening the same
+//! segment file on every call, `DataWal` keeps a small LRU of read-only
+//! file descriptors keyed by segment id; touching that cache requires
+//! `&mut`. The trade-off is a smaller `DataWal` footprint in RAM (only
+//! key bytes and 16-byte refs are kept) at the cost of one read syscall
+//! and one CRC check per `get`.
+//!
+//! Out of scope for v0.1.4:
 //! - In-place `compact()` that swaps segments atomically.
 //! - Transactions / atomic multi-record commits.
 //! - Concurrent writers.
 //! - Iteration over historical record versions.
+//! - Asynchronous I/O.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
@@ -26,38 +44,47 @@ use base64::engine::general_purpose::STANDARD as B64;
 use base64::Engine as _;
 use serde::Serialize;
 
-use crate::format::RecordType;
-use crate::record_log::{Record, RecordLog};
+use crate::fd_pool::{FdPool, DEFAULT_CAPACITY};
+use crate::format::{decode_next, DecodeOutcome, RecordType, HEADER_LEN};
+use crate::record_log::{RecordLog, RecordRef};
 
 /// Stats reported by a successful compaction.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct CompactionStats {
     /// Number of live keys written to the output log.
     pub live_keys: u64,
-    /// Number of records written. Equal to `live_keys` in v0.1-pre.
+    /// Number of records written. Equal to `live_keys` in v0.1.4.
     pub records_written: u64,
     /// Total framed bytes written. Useful for sanity checks.
     pub bytes_written: u64,
 }
 
 /// Bytes-based key/value store backed by a [`RecordLog`].
+///
+/// See the [module docs](crate) for the semantic contract and the rationale
+/// for the `&mut self` signatures on read paths.
 #[derive(Debug)]
 pub struct DataWal {
     log: RecordLog,
-    /// In-memory live keydir. v0.1-pre stores the full payload alongside
-    /// the key for simplicity; a future variant will switch to
-    /// `HashMap<Vec<u8>, RecordRef>` and read by offset.
-    map: HashMap<Vec<u8>, Vec<u8>>,
+    /// In-memory live keydir. Stores only on-disk references; payloads
+    /// are read on demand through `fd_pool`.
+    map: HashMap<Vec<u8>, RecordRef>,
+    /// Bounded LRU of read-only segment file descriptors.
+    fd_pool: FdPool,
 }
 
 impl DataWal {
     /// Open (or create) a `DataWal` rooted at `dir`. Rebuilds the keydir
-    /// by scanning the underlying record log in physical order.
+    /// by scanning the underlying record log lazily; payloads are not
+    /// materialised in memory.
     pub fn open(dir: &Path) -> Result<Self> {
-        let mut log = RecordLog::open(dir)?;
-        let records = log.scan()?;
-        let map = rebuild_keydir(&records);
-        Ok(Self { log, map })
+        let log = RecordLog::open(dir)?;
+        let map = rebuild_keydir_lazy(&log)?;
+        Ok(Self {
+            log,
+            map,
+            fd_pool: FdPool::with_capacity(DEFAULT_CAPACITY),
+        })
     }
 
     /// Path of the directory backing this store.
@@ -75,33 +102,63 @@ impl DataWal {
         self.map.is_empty()
     }
 
-    /// Returns `true` if `key` is currently live.
+    /// Returns `true` if `key` is currently live. Does not read from disk.
     pub fn contains_key(&self, key: &[u8]) -> bool {
         self.map.contains_key(key)
     }
 
-    /// Returns a copy of the value for `key`, if live.
-    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
-        Ok(self.map.get(key).cloned())
+    /// Returns the on-disk reference for `key`, if live. Does not read
+    /// the payload from disk.
+    pub fn ref_of(&self, key: &[u8]) -> Option<RecordRef> {
+        self.map.get(key).copied()
+    }
+
+    /// Reads the live value for `key`, validating the per-record CRC32C.
+    ///
+    /// Performs a positional read (`pread` on Unix) against the segment
+    /// holding the most recent `Put` for `key`. Takes `&mut self` because
+    /// it maintains an internal LRU of read-only segment file descriptors;
+    /// see the [module docs](crate) for the rationale.
+    ///
+    /// Returns `Err` if the on-disk frame is structurally invalid, the
+    /// CRC does not match, or the segment file cannot be opened.
+    pub fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>> {
+        let rref = match self.map.get(key).copied() {
+            Some(r) => r,
+            None => return Ok(None),
+        };
+        let bytes =
+            self.fd_pool
+                .read_at(self.log.dir(), rref.segment, rref.offset, rref.len as usize)?;
+        decode_payload_for_get(&bytes, rref).map(Some)
     }
 
-    /// All live keys, unordered (HashMap iteration order).
+    /// All live keys, unordered (HashMap iteration order). Does not read
+    /// from disk.
     pub fn keys(&self) -> Vec<Vec<u8>> {
         self.map.keys().cloned().collect()
     }
 
-    /// All live `(key, value)` pairs, unordered.
-    pub fn items(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
-        self.map
-            .iter()
-            .map(|(k, v)| (k.clone(), v.clone()))
-            .collect()
+    /// All live `(key, value)` pairs, unordered. Performs one disk read
+    /// per key (CRC validated) and takes `&mut self` for the same reason
+    /// `get` does. Returns the first I/O or CRC error if any.
+    pub fn items(&mut self) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+        let keys: Vec<Vec<u8>> = self.map.keys().cloned().collect();
+        let mut out = Vec::with_capacity(keys.len());
+        for k in keys {
+            let v = self.get(&k)?.ok_or_else(|| {
+                anyhow::anyhow!("datawal: keydir desync for key (len={})", k.len())
+            })?;
+            out.push((k, v));
+        }
+        Ok(out)
     }
 
-    /// Append a `Put` record and update the keydir.
+    /// Append a `Put` record and update the keydir with the returned
+    /// on-disk reference.
     pub fn put(&mut self, key: &[u8], value: &[u8]) -> Result<()> {
-        self.log.append_record(RecordType::Put, key, value)?;
-        self.map.insert(key.to_vec(), value.to_vec());
+        let rref = self.log.append_record(RecordType::Put, key, value)?;
+        self.map.insert(key.to_vec(), rref);
         Ok(())
     }
 
@@ -131,9 +188,14 @@ impl DataWal {
     /// `out_dir` must be empty (or non-existent). Tombstones are not
     /// carried over: deleted keys simply disappear. The original
     /// `self` is **not** modified.
-    pub fn compact_to(&self, out_dir: &Path) -> Result<CompactionStats> {
+    ///
+    /// Each live value is read from the source via the same disk-read /
+    /// CRC-validate path used by [`Self::get`]; this re-validates every
+    /// record copied to the compacted target, at the cost of one read
+    /// syscall per key.
+    pub fn compact_to(&mut self, out_dir: &Path) -> Result<CompactionStats> {
         // Refuse to clobber an existing non-empty directory or an existing
-        // datawal log directory. v0.1-pre keeps this strict on purpose.
+        // datawal log directory. v0.1.4 keeps this strict on purpose.
         if out_dir.exists() {
             let is_empty = std::fs::read_dir(out_dir)
                 .with_context(|| {
@@ -152,13 +214,17 @@ impl DataWal {
             }
         }
 
+        // Sort keys for deterministic output. Useful for diffing snapshots.
+        let mut sorted_keys: Vec<Vec<u8>> = self.map.keys().cloned().collect();
+        sorted_keys.sort();
+
         let mut out_log = RecordLog::open(out_dir)?;
         let mut stats = CompactionStats::default();
-        // Sort keys for deterministic output. Useful for diffing snapshots.
-        let mut sorted: Vec<(&Vec<u8>, &Vec<u8>)> = self.map.iter().collect();
-        sorted.sort_by(|a, b| a.0.cmp(b.0));
-        for (k, v) in sorted {
-            let r = out_log.append_record(RecordType::Put, k, v)?;
+        for k in sorted_keys {
+            let v = self
+                .get(&k)?
+                .ok_or_else(|| anyhow::anyhow!("datawal: keydir desync during compact_to"))?;
+            let r = out_log.append_record(RecordType::Put, &k, &v)?;
             stats.records_written += 1;
             stats.bytes_written += r.len as u64;
         }
@@ -178,20 +244,26 @@ impl DataWal {
     /// Keys are emitted in sorted byte order for deterministic output.
     /// This is **not** an analytics format: keys and values are opaque
     /// bytes; consumers must base64-decode them.
-    pub fn export_jsonl(&self, out_path: &Path) -> Result<()> {
+    ///
+    /// Performs one disk read per live key (CRC validated) and therefore
+    /// takes `&mut self`.
+    pub fn export_jsonl(&mut self, out_path: &Path) -> Result<()> {
         #[derive(Serialize)]
         struct Row<'a> {
             key_b64: &'a str,
             value_b64: &'a str,
         }
 
-        let mut sorted: Vec<(&Vec<u8>, &Vec<u8>)> = self.map.iter().collect();
-        sorted.sort_by(|a, b| a.0.cmp(b.0));
+        let mut sorted_keys: Vec<Vec<u8>> = self.map.keys().cloned().collect();
+        sorted_keys.sort();
 
         let mut buf = String::new();
-        for (k, v) in sorted {
-            let k_b64 = B64.encode(k);
-            let v_b64 = B64.encode(v);
+        for k in sorted_keys {
+            let v = self
+                .get(&k)?
+                .ok_or_else(|| anyhow::anyhow!("datawal: keydir desync during export_jsonl"))?;
+            let k_b64 = B64.encode(&k);
+            let v_b64 = B64.encode(&v);
             let line = serde_json::to_string(&Row {
                 key_b64: &k_b64,
                 value_b64: &v_b64,
@@ -219,19 +291,30 @@ impl DataWal {
     }
 }
 
-/// Replay a scanned record list into a live keydir.
+/// Replay the log lazily through [`RecordLog::scan_iter`] to build a
+/// `key -> RecordRef` map. Last-write-wins by physical order.
 ///
-/// Last-write-wins by physical order: later records (later segment, later
-/// offset within the same segment) overwrite earlier ones.
-fn rebuild_keydir(records: &[Record]) -> HashMap<Vec<u8>, Vec<u8>> {
-    let mut map: HashMap<Vec<u8>, Vec<u8>> = HashMap::new();
-    for r in records {
-        match r.record_type {
+/// Payloads are not materialised: only `RecordRef`s are inserted into
+/// the map. Tombstones (`Delete`) remove the prior entry, mirroring the
+/// pre-0.1.4 behaviour.
+fn rebuild_keydir_lazy(log: &RecordLog) -> Result<HashMap<Vec<u8>, RecordRef>> {
+    let mut map: HashMap<Vec<u8>, RecordRef> = HashMap::new();
+    let iter = log.scan_iter()?;
+    for rec in iter {
+        let rec = rec?;
+        match rec.record_type {
             RecordType::Put => {
-                map.insert(r.key.clone(), r.payload.clone());
+                map.insert(
+                    rec.key,
+                    RecordRef {
+                        segment: rec.segment,
+                        offset: rec.offset,
+                        len: rec.len,
+                    },
+                );
             }
             RecordType::Delete => {
-                map.remove(&r.key);
+                map.remove(&rec.key);
             }
             RecordType::Raw => {
                 // Raw records are not part of the KV projection. They are
@@ -239,5 +322,73 @@ fn rebuild_keydir(records: &[Record]) -> HashMap<Vec<u8>, Vec<u8>> {
             }
         }
     }
-    map
+    Ok(map)
+}
+
+/// Decode the payload at a positionally-read frame, sanity-checking the
+/// frame against the `RecordRef` that pointed at it.
+///
+/// Returns the payload bytes on success, or an `anyhow` error otherwise.
+fn decode_payload_for_get(bytes: &[u8], rref: RecordRef) -> Result<Vec<u8>> {
+    if bytes.len() < HEADER_LEN {
+        anyhow::bail!(
+            "datawal: short read for segment {:08} offset {} (got {}, need at least {})",
+            rref.segment,
+            rref.offset,
+            bytes.len(),
+            HEADER_LEN
+        );
+    }
+    match decode_next(bytes, 0) {
+        Ok(DecodeOutcome::Ok {
+            record_type,
+            key: _key,
+            payload,
+            bytes_consumed,
+            ..
+        }) => {
+            if bytes_consumed != rref.len {
+                anyhow::bail!(
+                    "datawal: frame length mismatch at segment {:08} offset {} (decoded {} vs ref {})",
+                    rref.segment,
+                    rref.offset,
+                    bytes_consumed,
+                    rref.len
+                );
+            }
+            // The keydir only points at Put records, so anything else here
+            // is corruption — surface it explicitly instead of silently
+            // returning bytes.
+            if !matches!(record_type, RecordType::Put) {
+                anyhow::bail!(
+                    "datawal: record at segment {:08} offset {} is not a Put (found {:?})",
+                    rref.segment,
+                    rref.offset,
+                    record_type
+                );
+            }
+            Ok(payload)
+        }
+        Ok(DecodeOutcome::Truncated { available, needed }) => {
+            anyhow::bail!(
+                "datawal: truncated frame at segment {:08} offset {} (available {}, needed {})",
+                rref.segment,
+                rref.offset,
+                available,
+                needed
+            );
+        }
+        Ok(DecodeOutcome::CrcMismatch { bytes_consumed }) => {
+            anyhow::bail!(
+                "datawal: CRC mismatch at segment {:08} offset {} (size {})",
+                rref.segment,
+                rref.offset,
+                bytes_consumed
+            );
+        }
+        Err(e) => Err(anyhow::Error::new(e).context(format!(
+            "datawal: structural decode error at segment {:08} offset {}",
+            rref.segment, rref.offset
+        ))),
+    }
 }
diff --git a/crates/datawal-core/src/fd_pool.rs b/crates/datawal-core/src/fd_pool.rs
new file mode 100644
index 0000000..e3b5f7b
--- /dev/null
+++ b/crates/datawal-core/src/fd_pool.rs
@@ -0,0 +1,219 @@
+//! Small bounded LRU cache of read-only segment file descriptors.
+//!
+//! Used by [`crate::DataWal`] (v0.1.4+) to avoid reopening the same
+//! `[0-9]{8}.dwal` segment on every `get`.
+//!
+//! Design notes:
+//!
+//! - Hard cap on entries (`capacity`). Once full, the least-recently-used
+//!   segment id is evicted.
+//! - All file descriptors are opened **read-only**. They never compete
+//!   with the writer's exclusive append-mode handle for that same segment.
+//!   On Unix this is well-defined; on macOS/Linux a read-only `pread`
+//!   against a file being appended to by the same process is safe and
+//!   returns bytes that were already written (length checked by caller).
+//! - The pool is intentionally not `Sync`. `DataWal::get` takes `&mut
+//!   self`, so a single-threaded borrow is enough.
+//! - Eviction never panics. If `open` fails the pool is left unchanged
+//!   and the error is propagated.
+//! - A capacity of 0 means: open on every call, never cache. Useful for
+//!   tests.
+
+use std::collections::VecDeque;
+use std::fs::File;
+#[cfg(not(unix))]
+use std::io::Read;
+#[cfg(unix)]
+use std::os::unix::fs::FileExt;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+
+use crate::segment::segment_path;
+
+/// Default cap. Tuned for the soak driver: a 4-segment hot working set
+/// covers the common case (active + compaction target + a recent rotated
+/// pair). Adjust via [`FdPool::with_capacity`] for benchmarks.
+pub const DEFAULT_CAPACITY: usize = 16;
+
+/// Tiny LRU of read-only segment file handles, keyed by segment id.
+#[derive(Debug)]
+pub(crate) struct FdPool {
+    capacity: usize,
+    /// `(segment_id, File)` pairs. Most recently used at the back.
+    entries: VecDeque<(u32, File)>,
+}
+
+impl FdPool {
+    /// New pool with the given hard cap.
+    pub fn with_capacity(capacity: usize) -> Self {
+        Self {
+            capacity,
+            entries: VecDeque::with_capacity(capacity.max(1)),
+        }
+    }
+
+    /// Pre-existing entries.
+    #[cfg(test)]
+    pub fn len(&self) -> usize {
+        self.entries.len()
+    }
+
+    /// Look the cached fd for `segment_id`, opening it under `dir` if not
+    /// already cached. Touches LRU order on hit.
+    fn touch_or_open(&mut self, dir: &Path, segment_id: u32) -> Result<&mut File> {
+        // Hit: move to the back (most-recently-used) and return.
+        if let Some(pos) = self.entries.iter().position(|(id, _)| *id == segment_id) {
+            let entry = self.entries.remove(pos).expect("position just found");
+            self.entries.push_back(entry);
+            return Ok(&mut self.entries.back_mut().expect("just pushed").1);
+        }
+
+        // Miss: open read-only, evict if needed, push back.
+        let path = segment_path(dir, segment_id);
+        let file = File::open(&path)
+            .with_context(|| format!("datawal: open segment {}", path.display()))?;
+
+        if self.capacity == 0 {
+            // Caller wants no caching. Return a temporary handle that
+            // will be dropped after this call by overwriting `entries`.
+            self.entries.clear();
+            self.entries.push_back((segment_id, file));
+            return Ok(&mut self.entries.back_mut().expect("just pushed").1);
+        }
+
+        if self.entries.len() >= self.capacity {
+            self.entries.pop_front();
+        }
+        self.entries.push_back((segment_id, file));
+        Ok(&mut self.entries.back_mut().expect("just pushed").1)
+    }
+
+    /// Read exactly `len` bytes starting at `offset` from segment
+    /// `segment_id` under `dir`.
+    ///
+    /// Returns an owned `Vec<u8>` with the bytes. Uses positional reads
+    /// (`pread` on Unix) so the underlying file descriptor offset is not
+    /// disturbed. On non-Unix platforms falls back to `seek + read`.
+    pub fn read_at(
+        &mut self,
+        dir: &Path,
+        segment_id: u32,
+        offset: u64,
+        len: usize,
+    ) -> Result<Vec<u8>> {
+        let file = self.touch_or_open(dir, segment_id)?;
+        let mut buf = vec![0u8; len];
+
+        #[cfg(unix)]
+        {
+            file.read_exact_at(&mut buf, offset).with_context(|| {
+                format!(
+                    "datawal: pread {} bytes at offset {} of segment {:08}.dwal",
+                    len, offset, segment_id
+                )
+            })?;
+        }
+        #[cfg(not(unix))]
+        {
+            use std::io::{Seek, SeekFrom};
+            file.seek(SeekFrom::Start(offset)).with_context(|| {
+                format!(
+                    "datawal: seek to offset {} of segment {:08}.dwal",
+                    offset, segment_id
+                )
+            })?;
+            file.read_exact(&mut buf).with_context(|| {
+                format!(
+                    "datawal: read {} bytes at offset {} of segment {:08}.dwal",
+                    len, offset, segment_id
+                )
+            })?;
+        }
+
+        Ok(buf)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use tempfile::tempdir;
+
+    fn seed_segment(dir: &Path, id: u32, bytes: &[u8]) {
+        let p = segment_path(dir, id);
+        let mut f = File::create(&p).unwrap();
+        f.write_all(bytes).unwrap();
+        f.sync_all().unwrap();
+    }
+
+    #[test]
+    fn read_at_returns_requested_bytes() {
+        let dir = tempdir().unwrap();
+        seed_segment(dir.path(), 1, b"hello world");
+        let mut pool = FdPool::with_capacity(4);
+        let got = pool.read_at(dir.path(), 1, 6, 5).unwrap();
+        assert_eq!(got, b"world");
+    }
+
+    #[test]
+    fn second_read_reuses_cached_fd() {
+        let dir = tempdir().unwrap();
+        seed_segment(dir.path(), 1, b"abcdefgh");
+        let mut pool = FdPool::with_capacity(2);
+        pool.read_at(dir.path(), 1, 0, 4).unwrap();
+        pool.read_at(dir.path(), 1, 4, 4).unwrap();
+        assert_eq!(pool.len(), 1);
+    }
+
+    #[test]
+    fn lru_evicts_oldest_at_capacity() {
+        let dir = tempdir().unwrap();
+        for id in 1..=3u32 {
+            seed_segment(dir.path(), id, b"xxxx");
+        }
+        let mut pool = FdPool::with_capacity(2);
+        pool.read_at(dir.path(), 1, 0, 1).unwrap();
+        pool.read_at(dir.path(), 2, 0, 1).unwrap();
+        assert_eq!(pool.len(), 2);
+        pool.read_at(dir.path(), 3, 0, 1).unwrap();
+        // 1 was the oldest -> evicted.
+        assert_eq!(pool.len(), 2);
+        // Verify by re-reading: would re-open if evicted; still works regardless.
+        pool.read_at(dir.path(), 1, 0, 1).unwrap();
+        assert_eq!(pool.len(), 2);
+    }
+
+    #[test]
+    fn capacity_zero_disables_cache() {
+        let dir = tempdir().unwrap();
+        seed_segment(dir.path(), 1, b"abcd");
+        let mut pool = FdPool::with_capacity(0);
+        pool.read_at(dir.path(), 1, 0, 4).unwrap();
+        pool.read_at(dir.path(), 1, 0, 4).unwrap();
+        // With cap 0 the entry is overwritten each call (single slot).
+        assert!(pool.len() <= 1);
+    }
+
+    #[test]
+    fn missing_segment_returns_error() {
+        let dir = tempdir().unwrap();
+        let mut pool = FdPool::with_capacity(2);
+        let err = pool.read_at(dir.path(), 7, 0, 1).unwrap_err();
+        assert!(format!("{err:#}").contains("open segment"));
+    }
+
+    #[test]
+    fn out_of_bounds_read_errors() {
+        let dir = tempdir().unwrap();
+        seed_segment(dir.path(), 1, b"abc");
+        let mut pool = FdPool::with_capacity(2);
+        let err = pool.read_at(dir.path(), 1, 0, 16).unwrap_err();
+        assert!(
+            format!("{err:#}").to_lowercase().contains("eof")
+                || format!("{err:#}").to_lowercase().contains("pread")
+                || format!("{err:#}").to_lowercase().contains("read")
+        );
+    }
+}
diff --git a/crates/datawal-core/src/lib.rs b/crates/datawal-core/src/lib.rs
index 3544a94..19c676a 100644
--- a/crates/datawal-core/src/lib.rs
+++ b/crates/datawal-core/src/lib.rs
@@ -56,6 +56,7 @@ pub mod lock;
 pub mod segment;
 
 mod datawal;
+mod fd_pool;
 mod record_log;
 mod record_log_reader;
 
diff --git a/crates/datawal-core/tests/corpus_fixtures.rs b/crates/datawal-core/tests/corpus_fixtures.rs
index 332c115..0199c8d 100644
--- a/crates/datawal-core/tests/corpus_fixtures.rs
+++ b/crates/datawal-core/tests/corpus_fixtures.rs
@@ -167,7 +167,7 @@ fn unknown_version_is_hard_error() {
 #[test]
 fn delete_tombstone_projects_to_lww_keydir() {
     let (_tmp, dir) = copy_fixture("delete_tombstone");
-    let kv = DataWal::open(&dir).expect("open");
+    let mut kv = DataWal::open(&dir).expect("open");
     // Source wrote: put alpha=1, put beta=2, put alpha=3, del beta.
     assert_eq!(kv.len(), 1);
     assert_eq!(kv.get(b"alpha").expect("get"), Some(b"3".to_vec()));
@@ -200,7 +200,7 @@ fn delete_tombstone_underlying_log_replays_all_records() {
 #[test]
 fn compact_to_output_contains_only_live_keys_as_puts() {
     let (_tmp, dir) = copy_fixture("compact_to_output");
-    let kv = DataWal::open(&dir).expect("open");
+    let mut kv = DataWal::open(&dir).expect("open");
     assert_eq!(kv.len(), 2);
     assert_eq!(kv.get(b"keep").expect("keep"), Some(b"final".to_vec()));
     assert_eq!(kv.get(b"other").expect("other"), Some(b"value".to_vec()));
diff --git a/crates/datawal-core/tests/crash_injection.rs b/crates/datawal-core/tests/crash_injection.rs
index fcca6b3..f80ddfc 100644
--- a/crates/datawal-core/tests/crash_injection.rs
+++ b/crates/datawal-core/tests/crash_injection.rs
@@ -233,7 +233,7 @@ fn child_compact_to(dir: &Path) -> ! {
         // Each iteration: open, compact, drop, remove. Open takes a
         // fresh fs2 lock; SIGKILL during the loop releases the lock
         // automatically.
-        let kv = DataWal::open(&source).expect("child: open source");
+        let mut kv = DataWal::open(&source).expect("child: open source");
         kv.compact_to(&out_dir).expect("child: compact_to");
         drop(kv);
         let _ = writeln!(stdout, "compacted {i}");
@@ -259,7 +259,7 @@ fn child_export_jsonl(dir: &Path) -> ! {
     let mut stdout = std::io::stdout().lock();
     for i in 0..MAX_RECORDS {
         let out = dir.join(format!("export_{i}.jsonl"));
-        let kv = DataWal::open(&source).expect("child: open source");
+        let mut kv = DataWal::open(&source).expect("child: open source");
         kv.export_jsonl(&out).expect("child: export_jsonl");
         drop(kv);
         let _ = writeln!(stdout, "exported {i}");
@@ -627,7 +627,7 @@ fn crash_compact_to() {
         // that segment_digest might miss (e.g. a lock-file artefact
         // we don't account for would be caught by the digest, but
         // we also want to verify the projection is correct).
-        let kv = DataWal::open(&source).expect("parent: reopen source");
+        let mut kv = DataWal::open(&source).expect("parent: reopen source");
         let mut actual: BTreeMap<Vec<u8>, Vec<u8>> = BTreeMap::new();
         for k in expected.keys() {
             let v = kv
@@ -720,7 +720,7 @@ fn crash_export_jsonl() {
 
         // Invariant 2: source still projects to the original key
         // set.
-        let kv = DataWal::open(&source).expect("parent: reopen source");
+        let mut kv = DataWal::open(&source).expect("parent: reopen source");
         for (k, v) in &expected {
             let got = kv
                 .get(k)
diff --git a/crates/datawal-core/tests/datawal.rs b/crates/datawal-core/tests/datawal.rs
index e82ac6b..66200f4 100644
--- a/crates/datawal-core/tests/datawal.rs
+++ b/crates/datawal-core/tests/datawal.rs
@@ -97,7 +97,7 @@ fn open_rebuilds_keydir_from_log() {
         kv.delete(b"b").unwrap();
         kv.fsync().unwrap();
     }
-    let kv = DataWal::open(dir.path()).unwrap();
+    let mut kv = DataWal::open(dir.path()).unwrap();
     assert_eq!(kv.len(), 1);
     assert_eq!(kv.get(b"a").unwrap().as_deref(), Some(&b"3"[..]));
     assert_eq!(kv.get(b"b").unwrap(), None);
@@ -125,7 +125,7 @@ fn compact_to_preserves_live_state() {
     assert_eq!(stats.records_written, 3);
     assert!(stats.bytes_written > 0);
 
-    let kv2 = DataWal::open(&dst_path).unwrap();
+    let mut kv2 = DataWal::open(&dst_path).unwrap();
     assert_eq!(kv2.len(), 3);
     assert_eq!(kv2.get(b"a").unwrap().as_deref(), Some(&b"3"[..]));
     assert_eq!(kv2.get(b"b").unwrap().as_deref(), Some(&b"2"[..]));
diff --git a/crates/datawal-core/tests/integration.rs b/crates/datawal-core/tests/integration.rs
index a5d1bb0..395f3b7 100644
--- a/crates/datawal-core/tests/integration.rs
+++ b/crates/datawal-core/tests/integration.rs
@@ -52,7 +52,7 @@ fn datawal_reopen_after_put_delete() {
         kv.put(b"k01", b"v1-bis").unwrap();
         kv.fsync().unwrap();
     }
-    let kv = DataWal::open(dir.path()).unwrap();
+    let mut kv = DataWal::open(dir.path()).unwrap();
     assert_eq!(kv.len(), 10, "10 odd keys should survive");
     for i in (0..20u32).step_by(2) {
         let key = format!("k{i:02}");
diff --git a/crates/datawal-core/tests/keydir_by_offset.rs b/crates/datawal-core/tests/keydir_by_offset.rs
new file mode 100644
index 0000000..76a6a5f
--- /dev/null
+++ b/crates/datawal-core/tests/keydir_by_offset.rs
@@ -0,0 +1,279 @@
+//! Tests pinning the v0.1.4 keydir-by-offset behaviour.
+//!
+//! Before this PR, `DataWal::get` returned a `Vec<u8>` cloned from an
+//! in-memory `HashMap<Vec<u8>, Vec<u8>>`. After this PR, the keydir holds
+//! `RecordRef`s and every `get` re-reads the segment file and revalidates
+//! the CRC against the stored payload. These tests pin that contract.
+
+use std::fs::OpenOptions;
+use std::io::{Read, Seek, SeekFrom, Write};
+use std::path::{Path, PathBuf};
+
+use datawal::DataWal;
+use tempfile::tempdir;
+
+fn seg_path(dir: &Path, id: u32) -> PathBuf {
+    dir.join(format!("{:08}.dwal", id))
+}
+
+#[test]
+fn get_reads_payload_lazily_after_drop_and_reopen() {
+    // open, put, drop -> reopen, the keydir should hold a RecordRef
+    // pointing at the on-disk record. get must read it through pread+CRC.
+    let dir = tempdir().unwrap();
+    {
+        let mut kv = DataWal::open(dir.path()).expect("open");
+        kv.put(b"alpha", b"one").expect("put alpha");
+        kv.put(b"beta", b"two").expect("put beta");
+        kv.fsync().expect("fsync");
+    }
+
+    let mut kv = DataWal::open(dir.path()).expect("reopen");
+    assert_eq!(kv.get(b"alpha").unwrap().as_deref(), Some(&b"one"[..]));
+    assert_eq!(kv.get(b"beta").unwrap().as_deref(), Some(&b"two"[..]));
+    assert!(kv.get(b"missing").unwrap().is_none());
+}
+
+#[test]
+fn get_revalidates_crc_and_errors_on_corruption() {
+    // We need the keydir to point at the record BEFORE the bytes are
+    // corrupted; otherwise reopen's longest-valid-prefix recovery would
+    // treat the corrupted record as a tail truncate and the key would
+    // simply be missing. The point of this test is to prove that `get`
+    // itself revalidates the CRC, not that recovery rejects it.
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    kv.put(b"k", b"some-payload-that-will-be-corrupted-on-disk")
+        .expect("put");
+    kv.fsync().expect("fsync");
+
+    // Confirm the record is in the keydir before corruption.
+    assert!(kv.ref_of(b"k").is_some(), "record must be in keydir");
+
+    // Flip a single byte inside the payload region of the active segment.
+    // The frame CRC covers the payload, so any flip yields a CRC mismatch
+    // on the read path.
+    let seg = seg_path(dir.path(), 1);
+    let len = std::fs::metadata(&seg).unwrap().len();
+    let target_offset = len - 5;
+    let mut f = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(&seg)
+        .unwrap();
+    f.seek(SeekFrom::Start(target_offset)).unwrap();
+    let mut byte = [0u8; 1];
+    f.read_exact(&mut byte).unwrap();
+    byte[0] ^= 0xFF;
+    f.seek(SeekFrom::Start(target_offset)).unwrap();
+    f.write_all(&byte).unwrap();
+    f.sync_all().unwrap();
+    drop(f);
+
+    let err = kv
+        .get(b"k")
+        .expect_err("get should fail on corrupt payload");
+    let msg = format!("{err:#}").to_lowercase();
+    assert!(
+        msg.contains("crc") || msg.contains("decode"),
+        "expected CRC/decode error, got: {msg}"
+    );
+}
+
+#[test]
+fn ref_of_tracks_live_keys_and_returns_none_after_delete() {
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+
+    assert!(kv.ref_of(b"x").is_none());
+
+    kv.put(b"x", b"v1").expect("put");
+    kv.fsync().expect("fsync");
+    let r1 = kv.ref_of(b"x").expect("x should be live");
+    assert_eq!(r1.segment, 1);
+    assert!(r1.len > 0);
+
+    kv.delete(b"x").expect("delete");
+    kv.fsync().expect("fsync");
+    assert!(
+        kv.ref_of(b"x").is_none(),
+        "ref_of should return None after delete tombstone"
+    );
+    assert_eq!(
+        kv.get(b"x").unwrap(),
+        None,
+        "get should agree with ref_of after delete"
+    );
+}
+
+#[test]
+fn ref_of_advances_to_latest_overwrite() {
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    kv.put(b"k", b"first").expect("put first");
+    let r1 = kv.ref_of(b"k").expect("first ref");
+    kv.put(b"k", b"second-longer").expect("put second");
+    let r2 = kv.ref_of(b"k").expect("second ref");
+
+    // The second ref must point past the first one in the same segment
+    // (overwrites are appends).
+    assert_eq!(r1.segment, r2.segment, "same segment for both writes");
+    assert!(
+        r2.offset > r1.offset,
+        "second ref must be past the first: r1={r1:?} r2={r2:?}"
+    );
+
+    kv.fsync().expect("fsync");
+    assert_eq!(
+        kv.get(b"k").unwrap().as_deref(),
+        Some(&b"second-longer"[..])
+    );
+}
+
+#[test]
+fn get_works_for_records_in_sealed_segment_after_rotation() {
+    // Put records into segment 1, force rotation by directly accessing the
+    // RecordLog handle, then keep writing into segment 2. The first key's
+    // RecordRef must still resolve via the fd pool.
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    kv.put(b"sealed-key", b"sealed-value")
+        .expect("put pre-rotate");
+    kv.fsync().expect("fsync");
+
+    // RecordLog::rotate is not exposed via DataWal, but compact_to into a
+    // new dir + reopen gives us a comparable structure: the live state
+    // survives a snapshot-style rebuild.
+    let dst = tempdir().unwrap();
+    let _ = kv.compact_to(dst.path()).expect("compact_to");
+
+    let mut kv2 = DataWal::open(dst.path()).expect("open compacted");
+    kv2.put(b"new-key", b"new-value").expect("put post-compact");
+    kv2.fsync().expect("fsync");
+
+    assert_eq!(
+        kv2.get(b"sealed-key").unwrap().as_deref(),
+        Some(&b"sealed-value"[..])
+    );
+    assert_eq!(
+        kv2.get(b"new-key").unwrap().as_deref(),
+        Some(&b"new-value"[..])
+    );
+}
+
+#[test]
+fn fd_pool_reuses_fd_across_repeated_gets() {
+    // Indirect check: do many gets and confirm the working set doesn't
+    // explode in number of file descriptors. We can't see the fd pool
+    // directly (private), but a process-level fd count check on /proc
+    // is a good smoke test on Linux. On non-Linux this just exercises
+    // the happy path.
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    for i in 0u32..100 {
+        let k = format!("k{i:03}").into_bytes();
+        let v = format!("v{i:03}").into_bytes();
+        kv.put(&k, &v).expect("put");
+    }
+    kv.fsync().expect("fsync");
+
+    #[cfg(target_os = "linux")]
+    let before = std::fs::read_dir("/proc/self/fd").map(|d| d.count()).ok();
+
+    for _ in 0..5 {
+        for i in 0u32..100 {
+            let k = format!("k{i:03}").into_bytes();
+            let v = format!("v{i:03}").into_bytes();
+            assert_eq!(kv.get(&k).unwrap().as_deref(), Some(&v[..]));
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    {
+        let after = std::fs::read_dir("/proc/self/fd").map(|d| d.count()).ok();
+        if let (Some(b), Some(a)) = (before, after) {
+            // The pool has a small fixed capacity (DEFAULT_CAPACITY = 16).
+            // After 500 gets across the same segment, the fd count should
+            // not grow by more than the pool capacity plus slack for
+            // harness/runner-dependent fds (CI runners can hold several
+            // extra fds open compared to local containers).
+            assert!(
+                a <= b + 32,
+                "fd count grew from {b} to {a} -- fd pool not reusing"
+            );
+        }
+    }
+}
+
+#[test]
+fn items_returns_all_live_pairs_and_revalidates_crc() {
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    kv.put(b"a", b"1").expect("put a");
+    kv.put(b"b", b"2").expect("put b");
+    kv.put(b"c", b"3").expect("put c");
+    kv.delete(b"b").expect("delete b");
+    kv.fsync().expect("fsync");
+
+    let mut items = kv.items().expect("items");
+    items.sort();
+    assert_eq!(
+        items,
+        vec![
+            (b"a".to_vec(), b"1".to_vec()),
+            (b"c".to_vec(), b"3".to_vec())
+        ]
+    );
+}
+
+#[test]
+fn compact_to_preserves_live_state_via_offset_keydir() {
+    let dir = tempdir().unwrap();
+    let mut kv = DataWal::open(dir.path()).expect("open");
+    for i in 0u32..50 {
+        let k = format!("k{i:02}").into_bytes();
+        let v = format!("v{i:02}-payload").into_bytes();
+        kv.put(&k, &v).expect("put");
+    }
+    // Overwrite half of them.
+    for i in 0u32..25 {
+        let k = format!("k{i:02}").into_bytes();
+        let v = format!("OVERWRITE-{i:02}").into_bytes();
+        kv.put(&k, &v).expect("overwrite");
+    }
+    // Delete a few.
+    for i in 40u32..45 {
+        let k = format!("k{i:02}").into_bytes();
+        kv.delete(&k).expect("delete");
+    }
+    kv.fsync().expect("fsync");
+
+    let dst = tempdir().unwrap();
+    let stats = kv.compact_to(dst.path()).expect("compact_to");
+    assert_eq!(stats.live_keys, 50 - 5);
+
+    let mut compacted = DataWal::open(dst.path()).expect("open compacted");
+    // First 25 carry OVERWRITE-* values.
+    for i in 0u32..25 {
+        let k = format!("k{i:02}").into_bytes();
+        let v = format!("OVERWRITE-{i:02}").into_bytes();
+        assert_eq!(compacted.get(&k).unwrap().as_deref(), Some(&v[..]));
+    }
+    // Next 15 (25..40) carry the original payload.
+    for i in 25u32..40 {
+        let k = format!("k{i:02}").into_bytes();
+        let v = format!("v{i:02}-payload").into_bytes();
+        assert_eq!(compacted.get(&k).unwrap().as_deref(), Some(&v[..]));
+    }
+    // Deleted ones are gone.
+    for i in 40u32..45 {
+        let k = format!("k{i:02}").into_bytes();
+        assert!(compacted.get(&k).unwrap().is_none());
+    }
+    // Last 5 still present.
+    for i in 45u32..50 {
+        let k = format!("k{i:02}").into_bytes();
+        let v = format!("v{i:02}-payload").into_bytes();
+        assert_eq!(compacted.get(&k).unwrap().as_deref(), Some(&v[..]));
+    }
+}
diff --git a/crates/datawal-core/tests/proptest_recovery.rs b/crates/datawal-core/tests/proptest_recovery.rs
index 02fcc4d..b674220 100644
--- a/crates/datawal-core/tests/proptest_recovery.rs
+++ b/crates/datawal-core/tests/proptest_recovery.rs
@@ -131,7 +131,7 @@ proptest! {
         {
             let _ = apply_seq_to_datawal(tmp.path(), &seq);
         }
-        let wal = DataWal::open(tmp.path()).expect("reopen at end");
+        let mut wal = DataWal::open(tmp.path()).expect("reopen at end");
 
         let expected = replay_expected(&seq);
         let mut got: BTreeMap<Vec<u8>, Vec<u8>> = BTreeMap::new();
@@ -158,7 +158,7 @@ proptest! {
             wal.delete(&[k]).unwrap();
             wal.put(&[k], &v).unwrap();
         }
-        let wal = DataWal::open(tmp.path()).unwrap();
+        let mut wal = DataWal::open(tmp.path()).unwrap();
         let got = wal.get(&[k]).unwrap();
         prop_assert_eq!(got, Some(v));
     }
@@ -179,18 +179,18 @@ proptest! {
 
         let stats_keys = {
             let _ = apply_seq_to_datawal(src.path(), &seq);
-            let src_wal = DataWal::open(src.path()).expect("reopen src");
+            let mut src_wal = DataWal::open(src.path()).expect("reopen src");
             src_wal.compact_to(&out_dir).expect("compact_to");
             src_wal.keys()
         };
 
-        let compacted = DataWal::open(&out_dir).expect("open compacted");
+        let mut compacted = DataWal::open(&out_dir).expect("open compacted");
         let mut got: BTreeMap<Vec<u8>, Vec<u8>> = BTreeMap::new();
         for k in compacted.keys() {
             got.insert(k.clone(), compacted.get(&k).unwrap().unwrap());
         }
         let mut expected: BTreeMap<Vec<u8>, Vec<u8>> = BTreeMap::new();
-        let src_again = DataWal::open(src.path()).expect("reopen src again");
+        let mut src_again = DataWal::open(src.path()).expect("reopen src again");
         for k in stats_keys {
             expected.insert(k.clone(), src_again.get(&k).unwrap().unwrap());
         }
@@ -273,7 +273,7 @@ proptest! {
                 }
             }
         }
-        let wal = DataWal::open(tmp.path()).unwrap();
+        let mut wal = DataWal::open(tmp.path()).unwrap();
         let mut got: BTreeMap<Vec<u8>, Vec<u8>> = BTreeMap::new();
         for k in wal.keys() {
             got.insert(k.clone(), wal.get(&k).unwrap().unwrap());