feat: migrate-engine improvements (content validation, schema, streaming, incremental)

clawhdf5-migrate:
- Real content validation: the post-migration check reads the written HDF5
  back (new hdf5_reader) and compares actual content — chunk text, embeddings,
  and every session/entity/relation field — to the source, not just row counts.
  A representative sample of chunk rows is verified by default; --validate-full
  checks every row. A count-preserving corruption no longer passes.
- Configurable schema: SQL is built from a SchemaConfig (table + ordered column
  names, defaulting to the ZeroClaw layout) instead of hardcoded queries, with
  --chunks-table / --sessions-table / --entities-table / --relations-table.
- Streaming count pass: --dry-run does a COUNT(*)-only pass per table instead
  of loading every row.
- Incremental migration: --incremental reads the existing output, reads only
  source chunks with id greater than the last migrated id, and appends them
  (metadata groups refreshed from source) rather than re-migrating everything.

clawhdf5-format:
- read_as_f32 / read_as_f64 now decode IEEE-754 half-precision (2-byte) floats
  via a no_std-safe bit conversion — needed to read float16-stored embeddings
  back (e.g. for migrate's content validation), previously a TypeMismatch.

Tests: f16 read unit test; migrate tests for content-corruption detection,
custom table names, and incremental append; CLI smoke-tested end-to-end and the
dense/incremental output verified with h5py.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
osobh
2026-06-04 02:19:31 +00:00
co-authored by Claude Opus 4.8
parent 0754afb7f2
commit 8534c7d204
6 changed files with 800 additions and 180 deletions
+18
View File
@@ -3,6 +3,24 @@
## Unreleased ## Unreleased
### New Features ### New Features
- `clawhdf5-migrate`: substantial engine improvements:
- **Real content validation** — the post-migration check now reads the written
HDF5 back and compares actual content (chunk text, embeddings, and every
session/entity/relation field) against the source, not just row counts. A
representative sample of chunk rows is verified by default; `--validate-full`
checks every row. A corrupt migration that preserves counts no longer passes.
- **Configurable schema** — table names are no longer hardcoded; queries are
built from a `SchemaConfig` (table + ordered column names, defaulting to the
ZeroClaw layout) with `--chunks-table` / `--sessions-table` /
`--entities-table` / `--relations-table` overrides.
- **Streaming count pass** — `--dry-run` now does a `COUNT(*)`-only pass per
table instead of loading every row into memory.
- **Incremental migration** — `--incremental` reads the existing output, reads
only source chunks newer than the last migrated id, and appends them
(refreshing the metadata groups), instead of re-migrating everything.
- `clawhdf5-format`: read **IEEE-754 half-precision (f16)** floats. `read_as_f32`
/ `read_as_f64` previously only handled 4- and 8-byte floats; 2-byte floats
(e.g. float16-stored embeddings) now decode via a no_std-safe bit conversion.
- `clawhdf5-format`: **write multi-block fractal heaps** (root indirect block). - `clawhdf5-format`: **write multi-block fractal heaps** (root indirect block).
Dense attribute and dense link storage previously capped at a single direct Dense attribute and dense link storage previously capped at a single direct
block (~64 KiB of heap data — a few thousand attributes/links). When the block (~64 KiB of heap data — a few thousand attributes/links). When the
+93
View File
@@ -895,6 +895,7 @@ fn convert_to_f64(
Ok(v as f64) Ok(v as f64)
} }
8 => Ok(read_f64_bytes(bytes, order)), 8 => Ok(read_f64_bytes(bytes, order)),
2 => Ok(read_f16_bytes(bytes, order) as f64),
_ => Err(FormatError::DataSizeMismatch { _ => Err(FormatError::DataSizeMismatch {
expected: 8, expected: 8,
actual: *size as usize, actual: *size as usize,
@@ -1039,6 +1040,9 @@ pub fn read_as_f32(raw: &[u8], datatype: &Datatype) -> Result<Vec<f32>, FormatEr
Datatype::FloatingPoint { size: 8, .. } => { Datatype::FloatingPoint { size: 8, .. } => {
result.push(read_f64_bytes(chunk, &order) as f32); result.push(read_f64_bytes(chunk, &order) as f32);
} }
Datatype::FloatingPoint { size: 2, .. } => {
result.push(read_f16_bytes(chunk, &order));
}
Datatype::FixedPoint { Datatype::FixedPoint {
signed: true, signed: true,
size, size,
@@ -1490,6 +1494,53 @@ fn read_f64_bytes(bytes: &[u8], order: &DatatypeByteOrder) -> f64 {
f64::from_le_bytes(buf) f64::from_le_bytes(buf)
} }
/// Decode an IEEE-754 half-precision (binary16) value to `f32`. Pure integer
/// bit manipulation (no_std-safe, no `powi`/`libm`).
fn read_f16_bytes(bytes: &[u8], order: &DatatypeByteOrder) -> f32 {
let mut buf = [0u8; 2];
let len = bytes.len().min(2);
match order {
DatatypeByteOrder::BigEndian => {
for i in 0..len {
buf[i] = bytes[len - 1 - i];
}
}
_ => buf[..len].copy_from_slice(&bytes[..len]),
}
f16_bits_to_f32(u16::from_le_bytes(buf))
}
/// Convert the bit pattern of an IEEE-754 half (binary16) to an `f32`.
fn f16_bits_to_f32(h: u16) -> f32 {
let h = h as u32;
let sign = (h & 0x8000) << 16;
let exp = (h >> 10) & 0x1f;
let mant = h & 0x3ff;
let bits = if exp == 0 {
if mant == 0 {
sign // signed zero
} else {
// Subnormal: normalize into an f32 normal.
let mut e: i32 = -1;
let mut m = mant;
loop {
e += 1;
m <<= 1;
if m & 0x400 != 0 {
break;
}
}
let m = m & 0x3ff;
sign | (((127 - 15 - e) as u32) << 23) | (m << 13)
}
} else if exp == 0x1f {
sign | 0x7f80_0000 | (mant << 13) // inf / NaN
} else {
sign | ((exp + (127 - 15)) << 23) | (mant << 13)
};
f32::from_bits(bits)
}
fn read_f32_bytes(bytes: &[u8], order: &DatatypeByteOrder) -> f32 { fn read_f32_bytes(bytes: &[u8], order: &DatatypeByteOrder) -> f32 {
let mut buf = [0u8; 4]; let mut buf = [0u8; 4];
let len = bytes.len().min(4); let len = bytes.len().min(4);
@@ -1666,6 +1717,48 @@ mod tests {
use crate::dataspace::{Dataspace, DataspaceType}; use crate::dataspace::{Dataspace, DataspaceType};
use crate::datatype::{CharacterSet, StringPadding}; use crate::datatype::{CharacterSet, StringPadding};
fn f16_datatype() -> Datatype {
Datatype::FloatingPoint {
size: 2,
byte_order: DatatypeByteOrder::LittleEndian,
bit_offset: 0,
bit_precision: 16,
exponent_location: 10,
exponent_size: 5,
mantissa_location: 0,
mantissa_size: 10,
exponent_bias: 15,
}
}
// IEEE-754 half bit patterns for known values.
fn f16_bits(v: f32) -> u16 {
// Encode a few exact values used by the test.
match v {
x if x == 0.0 => 0x0000,
x if x == 1.0 => 0x3c00,
x if x == -2.0 => 0xc000,
x if x == 0.5 => 0x3800,
x if x == 65504.0 => 0x7bff, // f16 max
_ => panic!("unsupported test value {v}"),
}
}
#[test]
fn read_f16_as_f32_and_f64() {
let values = [0.0f32, 1.0, -2.0, 0.5, 65504.0];
let raw: Vec<u8> = values
.iter()
.flat_map(|&v| f16_bits(v).to_le_bytes())
.collect();
let dt = f16_datatype();
let got32 = read_as_f32(&raw, &dt).unwrap();
assert_eq!(got32, values);
let got64 = read_as_f64(&raw, &dt).unwrap();
let expect64: Vec<f64> = values.iter().map(|&v| v as f64).collect();
assert_eq!(got64, expect64);
}
fn reduced_int(signed: bool, precision: u16) -> Datatype { fn reduced_int(signed: bool, precision: u16) -> Datatype {
Datatype::FixedPoint { Datatype::FixedPoint {
size: 4, size: 4,
+159
View File
@@ -0,0 +1,159 @@
//! Read a migration HDF5 file back into the in-memory data model.
//!
//! Used to verify migrated content (real validation) and to merge new rows into
//! an existing output (incremental migration). Mirrors the layout produced by
//! [`crate::hdf5_writer`].
use clawhdf5::reader::{File, Group};
use clawhdf5_format::type_builders::AttrValue;
use crate::sqlite_reader::{Entity, MemoryChunk, Relation, Session, SqliteData};
type BoxErr = Box<dyn std::error::Error>;
fn read_strings(group: &Group<'_>, name: &str) -> Result<Vec<String>, BoxErr> {
Ok(group.dataset(name)?.read_string()?)
}
fn read_i64s(group: &Group<'_>, name: &str) -> Result<Vec<i64>, BoxErr> {
Ok(group.dataset(name)?.read_i64()?)
}
fn read_f64s(group: &Group<'_>, name: &str) -> Result<Vec<f64>, BoxErr> {
Ok(group.dataset(name)?.read_f64()?)
}
/// Read the embeddings dataset as a flat `Vec<f32>` of `n * dim` values,
/// handling both f32 and (lossy) f16 storage.
fn read_embeddings_flat(group: &Group<'_>) -> Result<Vec<f32>, BoxErr> {
Ok(group.dataset("embeddings")?.read_f32()?)
}
/// Read a migration HDF5 file into a [`SqliteData`].
pub fn read_hdf5(path: &str) -> Result<SqliteData, BoxErr> {
let file = File::open(path)?;
let embedding_dim = match file.root().attrs()?.get("embedding_dim") {
Some(AttrValue::I64(d)) => *d as usize,
_ => 0,
};
let chunks = read_chunks(&file, embedding_dim)?;
let sessions = read_sessions(&file)?;
let entities = read_entities(&file)?;
let relations = read_relations(&file)?;
Ok(SqliteData {
chunks,
sessions,
entities,
relations,
embedding_dim,
})
}
fn read_chunks(file: &File, dim: usize) -> Result<Vec<MemoryChunk>, BoxErr> {
let g = file.group("chunks")?;
let count = group_count(&g)?;
if count == 0 {
return Ok(Vec::new());
}
let ids = read_i64s(&g, "id")?;
let texts = read_strings(&g, "text")?;
let channels = read_strings(&g, "source_channel")?;
let timestamps = read_f64s(&g, "timestamp")?;
let session_ids = read_strings(&g, "session_id")?;
let tags = read_strings(&g, "tags")?;
let deleted = g.dataset("deleted")?.read_i32()?;
let emb_flat = read_embeddings_flat(&g)?;
let dim = dim.max(1);
let mut chunks = Vec::with_capacity(ids.len());
for (i, &id) in ids.iter().enumerate() {
let embedding = emb_flat
.get(i * dim..(i + 1) * dim)
.map(|s| s.to_vec())
.unwrap_or_default();
chunks.push(MemoryChunk {
id,
chunk: texts.get(i).cloned().unwrap_or_default(),
embedding,
source_channel: channels.get(i).cloned().unwrap_or_default(),
timestamp: timestamps.get(i).copied().unwrap_or(0.0),
session_id: session_ids.get(i).cloned().unwrap_or_default(),
tags: tags.get(i).cloned().unwrap_or_default(),
deleted: deleted.get(i).copied().unwrap_or(0),
});
}
Ok(chunks)
}
fn read_sessions(file: &File) -> Result<Vec<Session>, BoxErr> {
let g = file.group("sessions")?;
if group_count(&g)? == 0 {
return Ok(Vec::new());
}
let ids = read_strings(&g, "id")?;
let starts = read_i64s(&g, "start_idx")?;
let ends = read_i64s(&g, "end_idx")?;
let channels = read_strings(&g, "channel")?;
let timestamps = read_f64s(&g, "timestamp")?;
let summaries = read_strings(&g, "summary")?;
Ok((0..ids.len())
.map(|i| Session {
id: ids[i].clone(),
start_idx: starts.get(i).copied().unwrap_or(0),
end_idx: ends.get(i).copied().unwrap_or(0),
channel: channels.get(i).cloned().unwrap_or_default(),
timestamp: timestamps.get(i).copied().unwrap_or(0.0),
summary: summaries.get(i).cloned().unwrap_or_default(),
})
.collect())
}
fn read_entities(file: &File) -> Result<Vec<Entity>, BoxErr> {
let g = file.group("entities")?;
if group_count(&g)? == 0 {
return Ok(Vec::new());
}
let ids = read_i64s(&g, "id")?;
let names = read_strings(&g, "name")?;
let types = read_strings(&g, "type")?;
let emb_idxs = read_i64s(&g, "embedding_idx")?;
Ok((0..ids.len())
.map(|i| Entity {
id: ids[i],
name: names.get(i).cloned().unwrap_or_default(),
entity_type: types.get(i).cloned().unwrap_or_default(),
embedding_idx: emb_idxs.get(i).copied().unwrap_or(-1),
})
.collect())
}
fn read_relations(file: &File) -> Result<Vec<Relation>, BoxErr> {
let g = file.group("relations")?;
if group_count(&g)? == 0 {
return Ok(Vec::new());
}
let srcs = read_i64s(&g, "src")?;
let tgts = read_i64s(&g, "tgt")?;
let rels = read_strings(&g, "relation")?;
let weights = read_f64s(&g, "weight")?;
let timestamps = read_f64s(&g, "timestamp")?;
Ok((0..srcs.len())
.map(|i| Relation {
src: srcs[i],
tgt: tgts.get(i).copied().unwrap_or(0),
relation: rels.get(i).cloned().unwrap_or_default(),
weight: weights.get(i).copied().unwrap_or(1.0),
timestamp: timestamps.get(i).copied().unwrap_or(0.0),
})
.collect())
}
fn group_count(group: &Group<'_>) -> Result<u64, BoxErr> {
match group.attrs()?.get("count") {
Some(AttrValue::I64(n)) => Ok(*n as u64),
_ => Ok(0),
}
}
+260 -70
View File
@@ -1,9 +1,12 @@
mod hdf5_reader;
mod hdf5_writer; mod hdf5_writer;
mod sqlite_reader; mod sqlite_reader;
mod validate; mod validate;
use clap::Parser; use clap::Parser;
use sqlite_reader::SchemaConfig;
/// Migrate ZeroClaw agent memory from SQLite to HDF5 format. /// Migrate ZeroClaw agent memory from SQLite to HDF5 format.
#[derive(Parser, Debug)] #[derive(Parser, Debug)]
#[command(name = "clawhdf5-migrate", version, about)] #[command(name = "clawhdf5-migrate", version, about)]
@@ -48,45 +51,126 @@ struct Cli {
#[arg(long)] #[arg(long)]
dry_run: bool, dry_run: bool,
/// Content-check every migrated row (default: a representative sample)
#[arg(long)]
validate_full: bool,
/// Append only rows newer than the existing output (by chunk id), merging
/// into the file at --hdf5 if it exists
#[arg(long)]
incremental: bool,
/// Override the SQLite table name for memory chunks
#[arg(long)]
chunks_table: Option<String>,
/// Override the SQLite table name for sessions
#[arg(long)]
sessions_table: Option<String>,
/// Override the SQLite table name for entities
#[arg(long)]
entities_table: Option<String>,
/// Override the SQLite table name for relations
#[arg(long)]
relations_table: Option<String>,
/// Print progress /// Print progress
#[arg(long)] #[arg(long)]
verbose: bool, verbose: bool,
} }
/// Build the schema config from CLI table-name overrides (defaults otherwise).
fn schema_from_cli(cli: &Cli) -> SchemaConfig {
let mut c = SchemaConfig::default();
if let Some(t) = &cli.chunks_table {
c.chunks.table = t.clone();
}
if let Some(t) = &cli.sessions_table {
c.sessions.table = t.clone();
}
if let Some(t) = &cli.entities_table {
c.entities.table = t.clone();
}
if let Some(t) = &cli.relations_table {
c.relations.table = t.clone();
}
c
}
fn main() -> Result<(), Box<dyn std::error::Error>> { fn main() -> Result<(), Box<dyn std::error::Error>> {
let cli = Cli::parse(); let cli = Cli::parse();
let schema = schema_from_cli(&cli);
// Dry run: a fast count-only pass that does not buffer the database.
if cli.dry_run {
let counts = sqlite_reader::read_counts(&cli.sqlite, cli.skip_deleted, &schema)?;
eprintln!("Dry run — no output file written.");
eprintln!(
"Would migrate: {} chunks, {} sessions, {} entities, {} relations",
counts.chunks, counts.sessions, counts.entities, counts.relations
);
return Ok(());
}
if cli.verbose { if cli.verbose {
eprintln!("Reading SQLite database: {}", cli.sqlite); eprintln!("Reading SQLite database: {}", cli.sqlite);
} }
let data = sqlite_reader::read_sqlite(&cli.sqlite, cli.skip_deleted, cli.embedding_dim)?; // Incremental: merge new rows into the existing output (if present).
let incremental_base = if cli.incremental && std::path::Path::new(&cli.hdf5).exists() {
Some(hdf5_reader::read_hdf5(&cli.hdf5)?)
} else {
None
};
let min_chunk_id = incremental_base
.as_ref()
.map(|d| d.chunks.iter().map(|c| c.id).max().unwrap_or(0))
.unwrap_or(0);
let dim_hint = cli
.embedding_dim
.or_else(|| incremental_base.as_ref().map(|d| d.embedding_dim));
let source = if min_chunk_id > 0 {
sqlite_reader::read_sqlite_filtered(
&cli.sqlite,
cli.skip_deleted,
dim_hint,
&schema,
min_chunk_id,
)?
} else {
sqlite_reader::read_sqlite(&cli.sqlite, cli.skip_deleted, dim_hint, &schema)?
};
// Build the dataset to write: either the source alone, or the existing
// output plus the newly-read rows (metadata groups refreshed from source).
let data = match incremental_base {
Some(mut base) => {
let added = source.chunks.len();
base.chunks.extend(source.chunks);
base.sessions = source.sessions;
base.entities = source.entities;
base.relations = source.relations;
base.embedding_dim = source.embedding_dim.max(base.embedding_dim);
if cli.verbose {
eprintln!("Incremental: appended {added} new chunks (id > {min_chunk_id})");
}
base
}
None => source,
};
if cli.verbose { if cli.verbose {
eprintln!( eprintln!(
"Read {} chunks, {} sessions, {} entities, {} relations", "Migrating {} chunks, {} sessions, {} entities, {} relations (dim={})",
data.chunks.len(),
data.sessions.len(),
data.entities.len(),
data.relations.len()
);
eprintln!("Embedding dimension: {}", data.embedding_dim);
}
if cli.dry_run {
eprintln!("Dry run — no output file written.");
eprintln!(
"Would migrate: {} chunks, {} sessions, {} entities, {} relations (dim={})",
data.chunks.len(), data.chunks.len(),
data.sessions.len(), data.sessions.len(),
data.entities.len(), data.entities.len(),
data.relations.len(), data.relations.len(),
data.embedding_dim data.embedding_dim
); );
return Ok(());
}
if cli.verbose {
eprintln!("Writing HDF5 file: {}", cli.hdf5); eprintln!("Writing HDF5 file: {}", cli.hdf5);
} }
@@ -101,25 +185,19 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
hdf5_writer::write_hdf5(&cli.hdf5, &data, &opts)?; hdf5_writer::write_hdf5(&cli.hdf5, &data, &opts)?;
if cli.verbose { if cli.verbose {
eprintln!("Validating output..."); eprintln!("Validating output (content check)...");
} }
let summary = validate::validate_hdf5( let summary = validate::validate_hdf5(&cli.hdf5, &data, cli.validate_full, cli.float16)?;
&cli.hdf5,
data.chunks.len(),
data.sessions.len(),
data.entities.len(),
data.relations.len(),
data.embedding_dim,
)?;
eprintln!( eprintln!(
"Migration complete: {} chunks, {} sessions, {} entities, {} relations (dim={})", "Migration complete: {} chunks, {} sessions, {} entities, {} relations (dim={}); {} rows content-verified",
summary.chunks, summary.chunks,
summary.sessions, summary.sessions,
summary.entities, summary.entities,
summary.relations, summary.relations,
summary.embedding_dim summary.embedding_dim,
summary.rows_checked,
); );
Ok(()) Ok(())
@@ -231,7 +309,7 @@ mod tests {
insert_relation(&conn, 1, 1, "self"); insert_relation(&conn, 1, 1, "self");
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "test-agent".into(), agent_id: "test-agent".into(),
embedder: "test-embed".into(), embedder: "test-embed".into(),
@@ -241,7 +319,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 2, 1, 1, 1, 8).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.chunks, 2); assert_eq!(summary.chunks, 2);
assert_eq!(summary.sessions, 1); assert_eq!(summary.sessions, 1);
assert_eq!(summary.entities, 1); assert_eq!(summary.entities, 1);
@@ -262,7 +340,7 @@ mod tests {
insert_chunk(&conn, 3, "also active", &make_embedding(4, 3.0), 0); insert_chunk(&conn, 3, "also active", &make_embedding(4, 3.0), 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, true, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, true, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 2); assert_eq!(data.chunks.len(), 2);
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
@@ -274,7 +352,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 2, 0, 0, 0, 4).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.chunks, 2); assert_eq!(summary.chunks, 2);
} }
@@ -289,7 +367,7 @@ mod tests {
insert_chunk(&conn, 2, "deleted", &make_embedding(4, 2.0), 1); insert_chunk(&conn, 2, "deleted", &make_embedding(4, 2.0), 1);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 2); assert_eq!(data.chunks.len(), 2);
} }
@@ -303,7 +381,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0); insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.embedding_dim, 16); assert_eq!(data.embedding_dim, 16);
} }
@@ -317,7 +395,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0); insert_chunk(&conn, 1, "test", &make_embedding(16, 0.5), 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, Some(8)).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, Some(8), &SchemaConfig::default()).unwrap();
assert_eq!(data.embedding_dim, 8); assert_eq!(data.embedding_dim, 8);
// Embedding truncated to dim 8 // Embedding truncated to dim 8
assert_eq!(data.chunks[0].embedding.len(), 8); assert_eq!(data.chunks[0].embedding.len(), 8);
@@ -335,7 +413,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &emb, 0); insert_chunk(&conn, 1, "test", &emb, 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(), agent_id: "t".into(),
embedder: "t".into(), embedder: "t".into(),
@@ -345,8 +423,8 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
// Verify file was created and is valid // Content-validate with the float16 tolerance enabled.
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 1, 0, 0, 0, 4).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, true, true).unwrap();
assert_eq!(summary.chunks, 1); assert_eq!(summary.chunks, 1);
// Verify float16 values are within tolerance // Verify float16 values are within tolerance
@@ -375,7 +453,7 @@ mod tests {
} }
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts_compressed = hdf5_writer::WriteOptions { let opts_compressed = hdf5_writer::WriteOptions {
agent_id: "t".into(), agent_id: "t".into(),
@@ -415,7 +493,7 @@ mod tests {
drop(conn); drop(conn);
// Simulate dry-run: read data but don't write // Simulate dry-run: read data but don't write
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 1); assert_eq!(data.chunks.len(), 1);
assert!(!h5_path.exists()); assert!(!h5_path.exists());
} }
@@ -427,7 +505,7 @@ mod tests {
let db_path = create_test_db(&dir); let db_path = create_test_db(&dir);
let h5_path = dir.path().join("out.h5"); let h5_path = dir.path().join("out.h5");
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 0); assert_eq!(data.chunks.len(), 0);
assert_eq!(data.sessions.len(), 0); assert_eq!(data.sessions.len(), 0);
assert_eq!(data.entities.len(), 0); assert_eq!(data.entities.len(), 0);
@@ -442,7 +520,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 0, 0, 0, 0, 0).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.chunks, 0); assert_eq!(summary.chunks, 0);
} }
@@ -465,7 +543,7 @@ mod tests {
} }
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 1000); assert_eq!(data.chunks.len(), 1000);
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
@@ -478,7 +556,7 @@ mod tests {
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = let summary =
validate::validate_hdf5(h5_path.to_str().unwrap(), 1000, 0, 0, 0, 64).unwrap(); validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.chunks, 1000); assert_eq!(summary.chunks, 1000);
} }
@@ -495,7 +573,7 @@ mod tests {
insert_session(&conn, "session-gamma", 21, 30); insert_session(&conn, "session-gamma", 21, 30);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.sessions.len(), 3); assert_eq!(data.sessions.len(), 3);
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
@@ -507,7 +585,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 0, 3, 0, 0, 0).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.sessions, 3); assert_eq!(summary.sessions, 3);
} }
@@ -527,7 +605,7 @@ mod tests {
insert_relation(&conn, 2, 3, "uses"); insert_relation(&conn, 2, 3, "uses");
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.entities.len(), 3); assert_eq!(data.entities.len(), 3);
assert_eq!(data.relations.len(), 3); assert_eq!(data.relations.len(), 3);
@@ -540,7 +618,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 0, 0, 3, 3, 0).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.entities, 3); assert_eq!(summary.entities, 3);
assert_eq!(summary.relations, 3); assert_eq!(summary.relations, 3);
} }
@@ -556,7 +634,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &make_embedding(4, 1.0), 0); insert_chunk(&conn, 1, "test", &make_embedding(4, 1.0), 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(), agent_id: "t".into(),
embedder: "t".into(), embedder: "t".into(),
@@ -566,15 +644,15 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
// Expect 5 chunks but only 1 was written // Validating against a source with an extra (unwritten) chunk must fail.
let result = validate::validate_hdf5(h5_path.to_str().unwrap(), 5, 0, 0, 0, 4); let mut bigger = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default())
.unwrap();
let mut extra = bigger.chunks[0].clone();
extra.id = 999;
bigger.chunks.push(extra);
let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &bigger, false, false);
assert!(result.is_err()); assert!(result.is_err());
assert!( assert!(result.unwrap_err().to_string().contains("count mismatch"));
result
.unwrap_err()
.to_string()
.contains("Chunk count mismatch")
);
} }
// ---------- Test 14: Metadata attributes are stored ---------- // ---------- Test 14: Metadata attributes are stored ----------
@@ -588,7 +666,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &make_embedding(8, 1.0), 0); insert_chunk(&conn, 1, "test", &make_embedding(8, 1.0), 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "my-agent-42".into(), agent_id: "my-agent-42".into(),
embedder: "openai-ada".into(), embedder: "openai-ada".into(),
@@ -634,7 +712,7 @@ mod tests {
insert_chunk(&conn, 1, "test", &emb, 0); insert_chunk(&conn, 1, "test", &emb, 0);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(), agent_id: "t".into(),
embedder: "t".into(), embedder: "t".into(),
@@ -680,7 +758,7 @@ mod tests {
drop(conn); drop(conn);
// Skip deleted // Skip deleted
let data = sqlite_reader::read_sqlite(&db_path, true, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, true, None, &SchemaConfig::default()).unwrap();
assert_eq!(data.chunks.len(), 4); // chunk 3 is deleted assert_eq!(data.chunks.len(), 4); // chunk 3 is deleted
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
@@ -692,7 +770,7 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), 4, 2, 2, 1, 16).unwrap(); let summary = validate::validate_hdf5(h5_path.to_str().unwrap(), &data, false, false).unwrap();
assert_eq!(summary.chunks, 4); assert_eq!(summary.chunks, 4);
assert_eq!(summary.sessions, 2); assert_eq!(summary.sessions, 2);
assert_eq!(summary.entities, 2); assert_eq!(summary.entities, 2);
@@ -711,7 +789,7 @@ mod tests {
insert_session(&conn, "s1", 0, 10); insert_session(&conn, "s1", 0, 10);
drop(conn); drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None).unwrap(); let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions { let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(), agent_id: "t".into(),
embedder: "t".into(), embedder: "t".into(),
@@ -721,13 +799,125 @@ mod tests {
}; };
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap(); hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
let result = validate::validate_hdf5(h5_path.to_str().unwrap(), 0, 99, 0, 0, 0); // Validating against a source whose session content differs must fail.
let mut tampered = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default())
.unwrap();
tampered.sessions[0].summary = "DIFFERENT".into();
let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &tampered, false, false);
assert!(result.is_err()); assert!(result.is_err());
assert!( assert!(result.unwrap_err().to_string().contains("session"));
result }
.unwrap_err()
.to_string() // ---------- Real content validation catches corrupt embeddings ----------
.contains("Session count mismatch") #[test]
); fn test_content_validation_catches_embedding_corruption() {
let dir = TempDir::new().unwrap();
let db_path = create_test_db(&dir);
let h5_path = dir.path().join("out.h5");
let conn = Connection::open(&db_path).unwrap();
insert_chunk(&conn, 1, "hello", &make_embedding(8, 1.0), 0);
drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default()).unwrap();
let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(),
embedder: "t".into(),
compression: false,
compression_level: 4,
float16: false,
};
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
// A source whose embedding differs (but counts match) must fail validation.
let mut tampered = sqlite_reader::read_sqlite(&db_path, false, None, &SchemaConfig::default())
.unwrap();
tampered.chunks[0].embedding[3] += 9.0;
let result = validate::validate_hdf5(h5_path.to_str().unwrap(), &tampered, true, false);
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("embedding"));
}
// ---------- Configurable schema: custom table names ----------
#[test]
fn test_configurable_table_names() {
let dir = TempDir::new().unwrap();
let db_path = dir.path().join("custom.db");
let path_str = db_path.to_str().unwrap().to_string();
let conn = Connection::open(&path_str).unwrap();
// Chunks live in a differently-named table; the others use defaults.
conn.execute_batch(
"CREATE TABLE my_chunks (
id INTEGER PRIMARY KEY, chunk TEXT, embedding BLOB,
source_channel TEXT, timestamp REAL, session_id TEXT, tags TEXT, deleted INTEGER
);
CREATE TABLE sessions (id TEXT, start_idx INTEGER, end_idx INTEGER, channel TEXT, timestamp REAL, summary TEXT);
CREATE TABLE entities (id INTEGER, name TEXT, type TEXT, embedding_idx INTEGER);
CREATE TABLE relations (src INTEGER, tgt INTEGER, relation TEXT, weight REAL, timestamp REAL);",
)
.unwrap();
let blob: Vec<u8> = make_embedding(4, 1.0).iter().flat_map(|v| v.to_le_bytes()).collect();
conn.execute(
"INSERT INTO my_chunks VALUES (1, 'hi', ?1, 'api', 1.0, 's', '', 0)",
rusqlite::params![blob],
)
.unwrap();
drop(conn);
let mut schema = SchemaConfig::default();
schema.chunks.table = "my_chunks".into();
let data = sqlite_reader::read_sqlite(&path_str, false, None, &schema).unwrap();
assert_eq!(data.chunks.len(), 1);
assert_eq!(data.chunks[0].chunk, "hi");
assert_eq!(data.embedding_dim, 4);
// Counts pass should also honor the custom table name.
let counts = sqlite_reader::read_counts(&path_str, false, &schema).unwrap();
assert_eq!(counts.chunks, 1);
}
// ---------- Incremental migration appends only new rows ----------
#[test]
fn test_incremental_migration() {
let dir = TempDir::new().unwrap();
let db_path = create_test_db(&dir);
let h5_path = dir.path().join("out.h5");
let cfg = SchemaConfig::default();
let opts = hdf5_writer::WriteOptions {
agent_id: "t".into(),
embedder: "t".into(),
compression: false,
compression_level: 4,
float16: false,
};
// First migration: 2 chunks.
let conn = Connection::open(&db_path).unwrap();
insert_chunk(&conn, 1, "one", &make_embedding(4, 1.0), 0);
insert_chunk(&conn, 2, "two", &make_embedding(4, 2.0), 0);
drop(conn);
let data = sqlite_reader::read_sqlite(&db_path, false, None, &cfg).unwrap();
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &data, &opts).unwrap();
// Add two more rows, then migrate incrementally.
let conn = Connection::open(&db_path).unwrap();
insert_chunk(&conn, 3, "three", &make_embedding(4, 3.0), 0);
insert_chunk(&conn, 4, "four", &make_embedding(4, 4.0), 0);
drop(conn);
let base = hdf5_reader::read_hdf5(h5_path.to_str().unwrap()).unwrap();
let max_id = base.chunks.iter().map(|c| c.id).max().unwrap_or(0);
assert_eq!(max_id, 2);
let new = sqlite_reader::read_sqlite_filtered(&db_path, false, Some(4), &cfg, max_id).unwrap();
assert_eq!(new.chunks.len(), 2); // only id 3 and 4
let mut merged = base;
merged.chunks.extend(new.chunks);
hdf5_writer::write_hdf5(h5_path.to_str().unwrap(), &merged, &opts).unwrap();
let final_data = hdf5_reader::read_hdf5(h5_path.to_str().unwrap()).unwrap();
assert_eq!(final_data.chunks.len(), 4);
let texts: Vec<&str> = final_data.chunks.iter().map(|c| c.chunk.as_str()).collect();
assert_eq!(texts, vec!["one", "two", "three", "four"]);
} }
} }
+148 -20
View File
@@ -53,9 +53,115 @@ pub struct SqliteData {
pub embedding_dim: usize, pub embedding_dim: usize,
} }
/// A table name plus the ordered column names the reader maps by position.
#[derive(Debug, Clone)]
pub struct TableSchema {
pub table: String,
pub columns: Vec<&'static str>,
}
/// Configurable mapping from a SQLite layout to the migration's data model.
///
/// Defaults to the ZeroClaw schema; the CLI can override the table names so the
/// tool can migrate databases whose tables are named differently. Column names
/// (and order) are part of the config too, so a library caller can remap them.
#[derive(Debug, Clone)]
pub struct SchemaConfig {
pub chunks: TableSchema,
pub sessions: TableSchema,
pub entities: TableSchema,
pub relations: TableSchema,
}
impl Default for SchemaConfig {
fn default() -> Self {
SchemaConfig {
chunks: TableSchema {
table: "memory_chunks".into(),
columns: vec![
"id",
"chunk",
"embedding",
"source_channel",
"timestamp",
"session_id",
"tags",
"deleted",
],
},
sessions: TableSchema {
table: "sessions".into(),
columns: vec!["id", "start_idx", "end_idx", "channel", "timestamp", "summary"],
},
entities: TableSchema {
table: "entities".into(),
columns: vec!["id", "name", "type", "embedding_idx"],
},
relations: TableSchema {
table: "relations".into(),
columns: vec!["src", "tgt", "relation", "weight", "timestamp"],
},
}
}
}
impl TableSchema {
fn select(&self, where_clause: &str) -> String {
format!(
"SELECT {} FROM {}{}",
self.columns.join(", "),
self.table,
where_clause
)
}
}
/// Row counts for each table — a fast pass that does not load row contents.
/// Used for `--dry-run` and progress without buffering the whole database.
#[derive(Debug, Default, Clone, Copy)]
pub struct RowCounts {
pub chunks: u64,
pub sessions: u64,
pub entities: u64,
pub relations: u64,
}
fn count_rows(conn: &Connection, table: &str, where_clause: &str) -> SqlResult<u64> {
conn.query_row(
&format!("SELECT COUNT(*) FROM {table}{where_clause}"),
[],
|r| r.get(0),
)
}
/// Count rows in each table without reading their contents.
pub fn read_counts(
path: &str,
skip_deleted: bool,
config: &SchemaConfig,
) -> Result<RowCounts, Box<dyn std::error::Error>> {
let conn = Connection::open(path)?;
let deleted_col = config.chunks.columns.get(7).copied().unwrap_or("deleted");
let chunk_where = if skip_deleted {
format!(" WHERE {deleted_col} = 0")
} else {
String::new()
};
Ok(RowCounts {
chunks: count_rows(&conn, &config.chunks.table, &chunk_where)?,
sessions: count_rows(&conn, &config.sessions.table, "")?,
entities: count_rows(&conn, &config.entities.table, "")?,
relations: count_rows(&conn, &config.relations.table, "")?,
})
}
/// Auto-detect embedding dimension from the first chunk's BLOB size. /// Auto-detect embedding dimension from the first chunk's BLOB size.
fn detect_embedding_dim(conn: &Connection) -> SqlResult<Option<usize>> { fn detect_embedding_dim(conn: &Connection, config: &SchemaConfig) -> SqlResult<Option<usize>> {
let mut stmt = conn.prepare("SELECT embedding FROM memory_chunks LIMIT 1")?; let emb_col = config.chunks.columns.get(2).copied().unwrap_or("embedding");
let mut stmt = conn.prepare(&format!(
"SELECT {emb_col} FROM {} LIMIT 1",
config.chunks.table
))?;
let mut rows = stmt.query([])?; let mut rows = stmt.query([])?;
if let Some(row) = rows.next()? { if let Some(row) = rows.next()? {
let blob: Vec<u8> = row.get(0)?; let blob: Vec<u8> = row.get(0)?;
@@ -80,18 +186,31 @@ pub fn read_sqlite(
path: &str, path: &str,
skip_deleted: bool, skip_deleted: bool,
embedding_dim: Option<usize>, embedding_dim: Option<usize>,
config: &SchemaConfig,
) -> Result<SqliteData, Box<dyn std::error::Error>> {
read_sqlite_filtered(path, skip_deleted, embedding_dim, config, 0)
}
/// Like [`read_sqlite`] but only reads chunks whose id is greater than
/// `min_chunk_id` (0 = all). Used for incremental migration.
pub fn read_sqlite_filtered(
path: &str,
skip_deleted: bool,
embedding_dim: Option<usize>,
config: &SchemaConfig,
min_chunk_id: i64,
) -> Result<SqliteData, Box<dyn std::error::Error>> { ) -> Result<SqliteData, Box<dyn std::error::Error>> {
let conn = Connection::open(path)?; let conn = Connection::open(path)?;
let dim = match embedding_dim { let dim = match embedding_dim {
Some(d) => d, Some(d) => d,
None => detect_embedding_dim(&conn)?.unwrap_or(0), None => detect_embedding_dim(&conn, config)?.unwrap_or(0),
}; };
let chunks = read_chunks(&conn, skip_deleted, dim)?; let chunks = read_chunks(&conn, skip_deleted, dim, config, min_chunk_id)?;
let sessions = read_sessions(&conn)?; let sessions = read_sessions(&conn, config)?;
let entities = read_entities(&conn)?; let entities = read_entities(&conn, config)?;
let relations = read_relations(&conn)?; let relations = read_relations(&conn, config)?;
Ok(SqliteData { Ok(SqliteData {
chunks, chunks,
@@ -106,16 +225,26 @@ fn read_chunks(
conn: &Connection, conn: &Connection,
skip_deleted: bool, skip_deleted: bool,
expected_dim: usize, expected_dim: usize,
config: &SchemaConfig,
min_chunk_id: i64,
) -> SqlResult<Vec<MemoryChunk>> { ) -> SqlResult<Vec<MemoryChunk>> {
let sql = if skip_deleted { let id_col = config.chunks.columns.first().copied().unwrap_or("id");
"SELECT id, chunk, embedding, source_channel, timestamp, session_id, tags, deleted \ let deleted_col = config.chunks.columns.get(7).copied().unwrap_or("deleted");
FROM memory_chunks WHERE deleted = 0" let mut conds = Vec::new();
if skip_deleted {
conds.push(format!("{deleted_col} = 0"));
}
if min_chunk_id > 0 {
conds.push(format!("{id_col} > {min_chunk_id}"));
}
let where_clause = if conds.is_empty() {
String::new()
} else { } else {
"SELECT id, chunk, embedding, source_channel, timestamp, session_id, tags, deleted \ format!(" WHERE {}", conds.join(" AND "))
FROM memory_chunks"
}; };
let sql = config.chunks.select(&where_clause);
let mut stmt = conn.prepare(sql)?; let mut stmt = conn.prepare(&sql)?;
let rows = stmt.query_map([], |row| { let rows = stmt.query_map([], |row| {
let blob: Vec<u8> = row.get(2)?; let blob: Vec<u8> = row.get(2)?;
let mut embedding = blob_to_f32(&blob); let mut embedding = blob_to_f32(&blob);
@@ -140,9 +269,8 @@ fn read_chunks(
rows.collect() rows.collect()
} }
fn read_sessions(conn: &Connection) -> SqlResult<Vec<Session>> { fn read_sessions(conn: &Connection, config: &SchemaConfig) -> SqlResult<Vec<Session>> {
let mut stmt = let mut stmt = conn.prepare(&config.sessions.select(""))?;
conn.prepare("SELECT id, start_idx, end_idx, channel, timestamp, summary FROM sessions")?;
let rows = stmt.query_map([], |row| { let rows = stmt.query_map([], |row| {
Ok(Session { Ok(Session {
id: row.get(0)?, id: row.get(0)?,
@@ -156,8 +284,8 @@ fn read_sessions(conn: &Connection) -> SqlResult<Vec<Session>> {
rows.collect() rows.collect()
} }
fn read_entities(conn: &Connection) -> SqlResult<Vec<Entity>> { fn read_entities(conn: &Connection, config: &SchemaConfig) -> SqlResult<Vec<Entity>> {
let mut stmt = conn.prepare("SELECT id, name, type, embedding_idx FROM entities")?; let mut stmt = conn.prepare(&config.entities.select(""))?;
let rows = stmt.query_map([], |row| { let rows = stmt.query_map([], |row| {
Ok(Entity { Ok(Entity {
id: row.get(0)?, id: row.get(0)?,
@@ -169,8 +297,8 @@ fn read_entities(conn: &Connection) -> SqlResult<Vec<Entity>> {
rows.collect() rows.collect()
} }
fn read_relations(conn: &Connection) -> SqlResult<Vec<Relation>> { fn read_relations(conn: &Connection, config: &SchemaConfig) -> SqlResult<Vec<Relation>> {
let mut stmt = conn.prepare("SELECT src, tgt, relation, weight, timestamp FROM relations")?; let mut stmt = conn.prepare(&config.relations.select(""))?;
let rows = stmt.query_map([], |row| { let rows = stmt.query_map([], |row| {
Ok(Relation { Ok(Relation {
src: row.get(0)?, src: row.get(0)?,
+122 -90
View File
@@ -1,5 +1,7 @@
use clawhdf5::reader::File; use crate::hdf5_reader::read_hdf5;
use clawhdf5_format::type_builders::AttrValue; use crate::sqlite_reader::SqliteData;
type BoxErr = Box<dyn std::error::Error>;
/// Summary of a migration validation. /// Summary of a migration validation.
#[derive(Debug)] #[derive(Debug)]
@@ -9,119 +11,149 @@ pub struct ValidationSummary {
pub entities: u64, pub entities: u64,
pub relations: u64, pub relations: u64,
pub embedding_dim: u64, pub embedding_dim: u64,
/// Number of rows whose full content was compared against the source.
pub rows_checked: u64,
} }
/// Validate an HDF5 file written by the migration tool. /// Validate a migrated HDF5 file against the source data.
/// ///
/// Checks that row counts and embedding dimensions match expectations. /// Reads the written file back and compares actual content — chunk text,
/// embeddings, and every session/entity/relation field — to the source, not
/// just the row counts. When `full` is false a representative sample of chunk
/// rows is content-checked (counts and all other groups are always checked in
/// full); when `full` is true every chunk row is compared too. `float16` widens
/// the embedding tolerance to allow for half-precision quantization.
pub fn validate_hdf5( pub fn validate_hdf5(
path: &str, path: &str,
expected_chunks: usize, source: &SqliteData,
expected_sessions: usize, full: bool,
expected_entities: usize, float16: bool,
expected_relations: usize, ) -> Result<ValidationSummary, BoxErr> {
expected_dim: usize, let got = read_hdf5(path)?;
) -> Result<ValidationSummary, Box<dyn std::error::Error>> {
let file = File::open(path)?;
let root = file.root();
// Read root attributes // ---- Counts ----
let attrs = root.attrs()?; check_count("chunk", got.chunks.len(), source.chunks.len())?;
let stored_dim = match attrs.get("embedding_dim") { check_count("session", got.sessions.len(), source.sessions.len())?;
Some(AttrValue::I64(d)) => *d as u64, check_count("entity", got.entities.len(), source.entities.len())?;
_ => 0, check_count("relation", got.relations.len(), source.relations.len())?;
}; if got.embedding_dim != source.embedding_dim {
// Validate chunks group
let chunks_group = file.group("chunks")?;
let chunk_attrs = chunks_group.attrs()?;
let chunk_count = match chunk_attrs.get("count") {
Some(AttrValue::I64(n)) => *n as u64,
_ => 0,
};
if chunk_count != expected_chunks as u64 {
return Err(format!( return Err(format!(
"Chunk count mismatch: HDF5 has {}, expected {}", "embedding_dim mismatch: HDF5 has {}, source has {}",
chunk_count, expected_chunks got.embedding_dim, source.embedding_dim
) )
.into()); .into());
} }
// Validate embedding dimensions if chunks exist // ---- Chunk content (sampled or full) ----
if chunk_count > 0 && expected_dim > 0 { let (emb_abs, emb_rel) = if float16 { (1e-2, 1e-2) } else { (1e-4, 0.0) };
let emb_ds = chunks_group.dataset("embeddings")?; let mut rows_checked = 0u64;
let shape = emb_ds.shape()?; for i in sample_indices(source.chunks.len(), full) {
if shape.len() == 2 && shape[1] != expected_dim as u64 { let (s, g) = (&source.chunks[i], &got.chunks[i]);
if s.id != g.id {
return Err(field_err("chunk", i, "id", s.id, g.id));
}
if s.chunk != g.chunk {
return Err(format!( return Err(format!(
"Embedding dim mismatch: HDF5 has {}, expected {}", "chunk[{i}].text mismatch: source {:?}, HDF5 {:?}",
shape[1], expected_dim truncate(&s.chunk),
truncate(&g.chunk)
) )
.into()); .into());
} }
if s.session_id != g.session_id || s.source_channel != g.source_channel || s.tags != g.tags
if stored_dim != expected_dim as u64 { {
return Err(format!("chunk[{i}] string field mismatch").into());
}
if s.deleted != g.deleted {
return Err(field_err("chunk", i, "deleted", s.deleted, g.deleted));
}
if s.embedding.len() != g.embedding.len() {
return Err(format!( return Err(format!(
"Embedding dim attr mismatch: HDF5 attr={}, expected {}", "chunk[{i}] embedding length mismatch: {} vs {}",
stored_dim, expected_dim s.embedding.len(),
g.embedding.len()
) )
.into()); .into());
} }
for (k, (&a, &b)) in s.embedding.iter().zip(g.embedding.iter()).enumerate() {
if (a - b).abs() > emb_abs + emb_rel * a.abs() {
return Err(format!(
"chunk[{i}].embedding[{k}] mismatch: source {a}, HDF5 {b}"
)
.into());
}
}
rows_checked += 1;
} }
// Validate sessions group // ---- Other groups (always full — they are small) ----
let sessions_group = file.group("sessions")?; for (i, (s, g)) in source.sessions.iter().zip(got.sessions.iter()).enumerate() {
let sess_attrs = sessions_group.attrs()?; if s.id != g.id
let session_count = match sess_attrs.get("count") { || s.start_idx != g.start_idx
Some(AttrValue::I64(n)) => *n as u64, || s.end_idx != g.end_idx
_ => 0, || s.channel != g.channel
}; || s.summary != g.summary
{
if session_count != expected_sessions as u64 { return Err(format!("session[{i}] mismatch").into());
return Err(format!( }
"Session count mismatch: HDF5 has {}, expected {}", rows_checked += 1;
session_count, expected_sessions
)
.into());
} }
for (i, (s, g)) in source.entities.iter().zip(got.entities.iter()).enumerate() {
// Validate entities group if s.id != g.id
let entities_group = file.group("entities")?; || s.name != g.name
let ent_attrs = entities_group.attrs()?; || s.entity_type != g.entity_type
let entity_count = match ent_attrs.get("count") { || s.embedding_idx != g.embedding_idx
Some(AttrValue::I64(n)) => *n as u64, {
_ => 0, return Err(format!("entity[{i}] mismatch").into());
}; }
rows_checked += 1;
if entity_count != expected_entities as u64 {
return Err(format!(
"Entity count mismatch: HDF5 has {}, expected {}",
entity_count, expected_entities
)
.into());
} }
for (i, (s, g)) in source.relations.iter().zip(got.relations.iter()).enumerate() {
// Validate relations group if s.src != g.src || s.tgt != g.tgt || s.relation != g.relation {
let relations_group = file.group("relations")?; return Err(format!("relation[{i}] mismatch").into());
let rel_attrs = relations_group.attrs()?; }
let relation_count = match rel_attrs.get("count") { rows_checked += 1;
Some(AttrValue::I64(n)) => *n as u64,
_ => 0,
};
if relation_count != expected_relations as u64 {
return Err(format!(
"Relation count mismatch: HDF5 has {}, expected {}",
relation_count, expected_relations
)
.into());
} }
Ok(ValidationSummary { Ok(ValidationSummary {
chunks: chunk_count, chunks: got.chunks.len() as u64,
sessions: session_count, sessions: got.sessions.len() as u64,
entities: entity_count, entities: got.entities.len() as u64,
relations: relation_count, relations: got.relations.len() as u64,
embedding_dim: stored_dim, embedding_dim: got.embedding_dim as u64,
rows_checked,
}) })
} }
fn check_count(kind: &str, got: usize, expected: usize) -> Result<(), BoxErr> {
if got != expected {
return Err(format!("{kind} count mismatch: HDF5 has {got}, source has {expected}").into());
}
Ok(())
}
fn field_err<T: std::fmt::Display>(kind: &str, i: usize, field: &str, s: T, g: T) -> BoxErr {
format!("{kind}[{i}].{field} mismatch: source {s}, HDF5 {g}").into()
}
fn truncate(s: &str) -> String {
if s.len() <= 40 {
s.to_string()
} else {
format!("{}…", &s[..40])
}
}
/// Indices of chunk rows to content-check. Full = all; otherwise a spread of
/// representative rows (first/last and evenly-spaced interior samples).
fn sample_indices(n: usize, full: bool) -> Vec<usize> {
if n == 0 {
return Vec::new();
}
if full || n <= 16 {
return (0..n).collect();
}
let mut idx: Vec<usize> = (0..16).map(|k| k * (n - 1) / 15).collect();
idx.dedup();
idx
}