use std::collections::{HashMap, HashSet};
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
use std::path::Path;
use std::sync::Arc;
use memmap2::MmapMut;
use serde_json;
use crate::graph::io::ntriples::{
ColMapEntry, ColumnTypeMeta, FixedColMeta, RegionMeta, StrColMeta,
};
use crate::graph::schema::StringInterner;
use crate::graph::storage::column_store::{ColumnStore, TypedColumn};
use crate::graph::storage::mapped::mmap_vec::{MmapBytes, MmapOrVec};
use rustc_hash::FxHashMap;
#[allow(dead_code)] pub struct WriteResult {
pub written: HashSet<String>,
pub unhandled: HashSet<String>,
}
pub fn write_unified_columns(
dir: &Path,
column_stores: &HashMap<String, Arc<ColumnStore>>,
_interner: &StringInterner,
) -> io::Result<WriteResult> {
let seg0 = dir.join("seg_000");
fs::create_dir_all(&seg0)?;
let bin_path = seg0.join("columns.bin");
let json_path = seg0.join("columns_meta.json");
struct PlannedType {
type_name: String,
meta: ColumnTypeMeta,
sources: Vec<(usize, Vec<u8>)>,
}
let mut planned: Vec<PlannedType> = Vec::with_capacity(column_stores.len());
let mut unhandled: HashSet<String> = HashSet::new();
let mut cursor: usize = 0;
let mut type_names: Vec<&String> = column_stores.keys().collect();
type_names.sort();
for type_name in type_names {
let store = &column_stores[type_name];
let has_mixed = store_has_mixed(store);
if has_mixed {
unhandled.insert(type_name.clone());
continue;
}
let row_count = store.row_count();
let (id_is_string, id_data_bytes, id_nulls_bytes, id_str_data_bytes, id_str_offsets_bytes) =
extract_id_column(store);
let mut sources: Vec<(usize, Vec<u8>)> = Vec::new();
let mut id_data = RegionMeta { offset: 0, len: 0 };
let mut id_nulls = RegionMeta { offset: 0, len: 0 };
let mut id_str_data = RegionMeta { offset: 0, len: 0 };
let mut id_str_offsets = RegionMeta { offset: 0, len: 0 };
if id_is_string {
(id_str_data, cursor) = plan_region(cursor, &id_str_data_bytes);
sources.push((id_str_data.offset, id_str_data_bytes));
(id_str_offsets, cursor) = plan_region(cursor, &id_str_offsets_bytes);
sources.push((id_str_offsets.offset, id_str_offsets_bytes));
(id_nulls, cursor) = plan_region(cursor, &id_nulls_bytes);
sources.push((id_nulls.offset, id_nulls_bytes));
} else if !id_data_bytes.is_empty() {
(id_data, cursor) = plan_region(cursor, &id_data_bytes);
sources.push((id_data.offset, id_data_bytes));
(id_nulls, cursor) = plan_region(cursor, &id_nulls_bytes);
sources.push((id_nulls.offset, id_nulls_bytes));
}
let (title_data_bytes, title_offsets_bytes, title_nulls_bytes) =
extract_title_column(store);
let (title_data, c) = plan_region(cursor, &title_data_bytes);
cursor = c;
sources.push((title_data.offset, title_data_bytes));
let (title_offsets, c) = plan_region(cursor, &title_offsets_bytes);
cursor = c;
sources.push((title_offsets.offset, title_offsets_bytes));
let (title_nulls, c) = plan_region(cursor, &title_nulls_bytes);
cursor = c;
sources.push((title_nulls.offset, title_nulls_bytes));
let mut col_map: Vec<ColMapEntry> = Vec::new();
let mut fixed_cols: Vec<FixedColMeta> = Vec::new();
let mut str_cols: Vec<StrColMeta> = Vec::new();
for (slot, ik) in store.schema().iter() {
let s = slot as usize;
let col = match store.column(s) {
Some(c) => c,
None => continue,
};
match col {
TypedColumn::Mixed { .. } => {
unreachable!("Mixed column slipped past store_has_mixed");
}
TypedColumn::Int64 { data, nulls } => {
let (data_r, c) = plan_region(cursor, data.as_raw_bytes());
cursor = c;
sources.push((data_r.offset, data.as_raw_bytes().to_vec()));
let (nulls_r, c) = plan_region(cursor, nulls.as_raw_bytes());
cursor = c;
sources.push((nulls_r.offset, nulls.as_raw_bytes().to_vec()));
let idx = fixed_cols.len();
fixed_cols.push(FixedColMeta {
col_type_str: "int64".into(),
data: data_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "int64".into(),
idx,
});
}
TypedColumn::Float64 { data, nulls } => {
let (data_r, c) = plan_region(cursor, data.as_raw_bytes());
cursor = c;
sources.push((data_r.offset, data.as_raw_bytes().to_vec()));
let (nulls_r, c) = plan_region(cursor, nulls.as_raw_bytes());
cursor = c;
sources.push((nulls_r.offset, nulls.as_raw_bytes().to_vec()));
let idx = fixed_cols.len();
fixed_cols.push(FixedColMeta {
col_type_str: "float64".into(),
data: data_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "float64".into(),
idx,
});
}
TypedColumn::UniqueId { data, nulls } => {
let (data_r, c) = plan_region(cursor, data.as_raw_bytes());
cursor = c;
sources.push((data_r.offset, data.as_raw_bytes().to_vec()));
let (nulls_r, c) = plan_region(cursor, nulls.as_raw_bytes());
cursor = c;
sources.push((nulls_r.offset, nulls.as_raw_bytes().to_vec()));
let idx = fixed_cols.len();
fixed_cols.push(FixedColMeta {
col_type_str: "uniqueid".into(),
data: data_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "uniqueid".into(),
idx,
});
}
TypedColumn::Bool { data, nulls } => {
let (data_r, c) = plan_region(cursor, data.as_raw_bytes());
cursor = c;
sources.push((data_r.offset, data.as_raw_bytes().to_vec()));
let (nulls_r, c) = plan_region(cursor, nulls.as_raw_bytes());
cursor = c;
sources.push((nulls_r.offset, nulls.as_raw_bytes().to_vec()));
let idx = fixed_cols.len();
fixed_cols.push(FixedColMeta {
col_type_str: "bool".into(),
data: data_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "bool".into(),
idx,
});
}
TypedColumn::Date { data, nulls } => {
let (data_r, c) = plan_region(cursor, data.as_raw_bytes());
cursor = c;
sources.push((data_r.offset, data.as_raw_bytes().to_vec()));
let (nulls_r, c) = plan_region(cursor, nulls.as_raw_bytes());
cursor = c;
sources.push((nulls_r.offset, nulls.as_raw_bytes().to_vec()));
let idx = fixed_cols.len();
fixed_cols.push(FixedColMeta {
col_type_str: "date".into(),
data: data_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "date".into(),
idx,
});
}
TypedColumn::Str {
offsets,
data,
nulls,
relocated,
} => {
let (data_bytes, offsets_bytes, nulls_bytes) =
pack_str_column(offsets, data, nulls, relocated);
let (data_r, c) = plan_region(cursor, &data_bytes);
cursor = c;
sources.push((data_r.offset, data_bytes));
let (offsets_r, c) = plan_region(cursor, &offsets_bytes);
cursor = c;
sources.push((offsets_r.offset, offsets_bytes));
let (nulls_r, c) = plan_region(cursor, &nulls_bytes);
cursor = c;
sources.push((nulls_r.offset, nulls_bytes));
let idx = str_cols.len();
str_cols.push(StrColMeta {
data: data_r,
offsets: offsets_r,
nulls: nulls_r,
});
col_map.push(ColMapEntry {
key_u64: ik.as_u64(),
col_type_str: "string".into(),
idx,
});
}
}
}
let (overflow_offsets, overflow_data, has_overflow) =
if let (Some(off_bytes), Some(data_bytes)) =
(store.overflow_offsets_bytes(), store.overflow_data_bytes())
{
let (off_r, c) = plan_region(cursor, &off_bytes);
cursor = c;
sources.push((off_r.offset, off_bytes));
let (data_r, c) = plan_region(cursor, &data_bytes);
cursor = c;
sources.push((data_r.offset, data_bytes));
(off_r, data_r, true)
} else {
(
RegionMeta { offset: 0, len: 0 },
RegionMeta { offset: 0, len: 0 },
false,
)
};
let meta = ColumnTypeMeta {
type_name: type_name.clone(),
row_count,
id_is_string,
id_data,
id_nulls,
id_str_data,
id_str_offsets,
title_data,
title_offsets,
title_nulls,
col_map,
fixed_cols,
str_cols,
overflow_offsets,
overflow_data,
has_overflow,
};
planned.push(PlannedType {
type_name: type_name.clone(),
meta,
sources,
});
}
let total_bytes = cursor;
if total_bytes == 0 {
let _ = fs::remove_file(&bin_path);
let _ = fs::remove_file(&json_path);
return Ok(WriteResult {
written: HashSet::new(),
unhandled,
});
}
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&bin_path)?;
file.set_len(total_bytes as u64)?;
let mut mmap = unsafe { MmapMut::map_mut(&file)? };
for pt in &planned {
for (off, bytes) in &pt.sources {
let dst = &mut mmap[*off..*off + bytes.len()];
dst.copy_from_slice(bytes);
}
}
mmap.flush()?;
let metas: Vec<ColumnTypeMeta> = planned.iter().map(|pt| pt.meta.clone()).collect();
let json = serde_json::to_string_pretty(&metas).map_err(io::Error::other)?;
let mut f = File::create(&json_path)?;
f.write_all(json.as_bytes())?;
f.sync_all()?;
let written: HashSet<String> = planned.into_iter().map(|pt| pt.type_name).collect();
Ok(WriteResult { written, unhandled })
}
#[inline]
fn plan_region(cursor: usize, bytes: &[u8]) -> (RegionMeta, usize) {
let region = RegionMeta {
offset: cursor,
len: bytes.len(),
};
(region, cursor + bytes.len())
}
fn pack_str_column(
offsets: &MmapOrVec<u64>,
data: &MmapBytes,
nulls: &MmapOrVec<u8>,
relocated: &FxHashMap<u32, String>,
) -> (Vec<u8>, Vec<u8>, Vec<u8>) {
let row_count = nulls.len();
let nulls_bytes = nulls.as_raw_bytes().to_vec();
let leading_zero = offsets.len() == row_count + 1;
if relocated.is_empty() {
let off_bytes = offsets.as_raw_bytes();
let off_slice = if leading_zero {
&off_bytes[8..]
} else {
off_bytes
};
return (
data.as_raw_bytes().to_vec(),
off_slice.to_vec(),
nulls_bytes,
);
}
let offsets = offsets.as_slice();
let source = data.as_raw_bytes();
let mut new_data: Vec<u8> = Vec::with_capacity(source.len());
let mut new_offsets: Vec<u8> = Vec::with_capacity(row_count * 8);
for row in 0..row_count {
if nulls.get(row) == 0 {
match relocated.get(&(row as u32)) {
Some(s) => new_data.extend_from_slice(s.as_bytes()),
None => {
let range = if leading_zero {
offsets.get(row).copied().zip(offsets.get(row + 1).copied())
} else if row == 0 {
offsets.first().copied().map(|end| (0, end))
} else {
offsets.get(row - 1).copied().zip(offsets.get(row).copied())
};
if let Some(bytes) =
range.and_then(|(start, end)| source.get(start as usize..end as usize))
{
new_data.extend_from_slice(bytes);
}
}
}
}
new_offsets.extend_from_slice(&(new_data.len() as u64).to_le_bytes());
}
(new_data, new_offsets, nulls_bytes)
}
fn store_has_mixed(store: &ColumnStore) -> bool {
if store
.columns_ref()
.any(|c| matches!(c, TypedColumn::Mixed { .. }))
{
return true;
}
if let Some(c) = store.id_column_ref() {
if matches!(c, TypedColumn::Mixed { .. }) {
return true;
}
}
if let Some(c) = store.title_column_ref() {
if matches!(c, TypedColumn::Mixed { .. }) {
return true;
}
}
false
}
fn extract_id_column(store: &ColumnStore) -> (bool, Vec<u8>, Vec<u8>, Vec<u8>, Vec<u8>) {
match store.id_column_ref() {
Some(TypedColumn::Str {
offsets,
data,
nulls,
relocated,
}) => {
let (data_bytes, offsets_bytes, nulls_bytes) =
pack_str_column(offsets, data, nulls, relocated);
(true, Vec::new(), nulls_bytes, data_bytes, offsets_bytes)
}
Some(TypedColumn::UniqueId { data, nulls }) => (
false,
data.as_raw_bytes().to_vec(),
nulls.as_raw_bytes().to_vec(),
Vec::new(),
Vec::new(),
),
Some(TypedColumn::Int64 { data, nulls }) => (
false,
data.as_raw_bytes().to_vec(),
nulls.as_raw_bytes().to_vec(),
Vec::new(),
Vec::new(),
),
_ => (false, Vec::new(), Vec::new(), Vec::new(), Vec::new()),
}
}
fn extract_title_column(store: &ColumnStore) -> (Vec<u8>, Vec<u8>, Vec<u8>) {
match store.title_column_ref() {
Some(TypedColumn::Str {
offsets,
data,
nulls,
relocated,
}) => pack_str_column(offsets, data, nulls, relocated),
_ => (Vec::new(), Vec::new(), Vec::new()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::datatypes::values::Value;
fn decode(data: &[u8], offsets: &[u8], nulls: &[u8]) -> Vec<Option<String>> {
let ends: Vec<u64> = offsets
.as_chunks::<8>()
.0
.iter()
.map(|c| u64::from_le_bytes(*c))
.collect();
assert_eq!(ends.len(), nulls.len(), "one end offset per row");
let mut out = Vec::with_capacity(nulls.len());
for (row, &is_null) in nulls.iter().enumerate() {
if is_null != 0 {
out.push(None);
continue;
}
let start = if row == 0 { 0 } else { ends[row - 1] } as usize;
let end = ends[row] as usize;
out.push(Some(String::from_utf8(data[start..end].to_vec()).unwrap()));
}
out
}
fn build(values: &[Option<&str>]) -> TypedColumn {
let mut col = TypedColumn::from_type_str("string");
for v in values {
match v {
Some(s) => col.push(&Value::String((*s).to_string())).unwrap(),
None => col.push_null(),
}
}
col
}
fn pack(col: &TypedColumn) -> Vec<Option<String>> {
let TypedColumn::Str {
offsets,
data,
nulls,
relocated,
} = col
else {
panic!("expected a Str column");
};
let (data_bytes, offsets_bytes, nulls_bytes) =
pack_str_column(offsets, data, nulls, relocated);
decode(&data_bytes, &offsets_bytes, &nulls_bytes)
}
#[test]
fn packs_a_column_with_no_overlay() {
let col = build(&[Some("a"), None, Some("ccc")]);
assert_eq!(
pack(&col),
vec![Some("a".into()), None, Some("ccc".into())],
"the leading zero must be stripped, not emitted as row 0's end"
);
}
#[test]
fn folds_a_differing_length_overwrite_back_into_the_layout() {
let mut col = build(&[Some("aa"), Some("bb"), Some("cc")]);
col.set(1, &Value::String("a-much-longer-value".into()))
.unwrap();
assert_eq!(
pack(&col),
vec![
Some("aa".into()),
Some("a-much-longer-value".into()),
Some("cc".into()),
],
"the overlaid row must carry the new value and its neighbours the old ones"
);
}
#[test]
fn folds_a_shorter_overwrite_and_renumbers_the_tail() {
let mut col = build(&[Some("aaaa"), Some("bbbb"), Some("cccc")]);
col.set(0, &Value::String("z".into())).unwrap();
assert_eq!(
pack(&col),
vec![Some("z".into()), Some("bbbb".into()), Some("cccc".into())],
"shrinking row 0 must shift every later row's offsets down"
);
}
#[test]
fn folds_an_overlay_onto_a_column_whose_rows_are_all_null() {
let mut col = build(&[None, None, None]);
col.set(2, &Value::String("xyz".into())).unwrap();
assert_eq!(pack(&col), vec![None, None, Some("xyz".into())]);
}
#[test]
fn a_null_row_advances_an_offset_without_contributing_bytes() {
let mut col = build(&[Some("aa"), None, Some("cc")]);
col.set(0, &Value::String("longer".into())).unwrap();
assert_eq!(
pack(&col),
vec![Some("longer".into()), None, Some("cc".into())]
);
}
#[test]
fn folds_an_overlay_onto_the_cumulative_ends_offset_form() {
let mut col = TypedColumn::Str {
offsets: MmapOrVec::from_vec(vec![2u64, 4, 6]),
data: {
let mut d = MmapBytes::new();
d.extend(b"aabbcc").unwrap();
d
},
nulls: MmapOrVec::from_vec(vec![0u8, 0, 0]),
relocated: FxHashMap::default(),
};
if let TypedColumn::Str { relocated, .. } = &mut col {
relocated.insert(1, "BB-longer".to_string());
}
assert_eq!(
pack(&col),
vec![
Some("aa".into()),
Some("BB-longer".into()),
Some("cc".into()),
]
);
}
}