use rustc_hash::FxHashMap;
use std::fs;
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
const VIEW_FILE: &str = crate::data_dir::VIEW_REGISTRY_FILE;
const VIEW_MAGIC: &[u8; 4] = b"BVIW";
const VIEW_VERSION: u16 = 1;
#[cfg(test)]
thread_local! {
static VIEW_PERSIST_FAIL_BEFORE_RENAME: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
static VIEW_PERSIST_FAIL_AFTER_RENAME: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
}
#[cfg(test)]
fn take_view_persist_fail_before_rename() -> bool {
VIEW_PERSIST_FAIL_BEFORE_RENAME.with(|failpoint| {
let fail = failpoint.get();
failpoint.set(false);
fail
})
}
#[cfg(test)]
fn take_view_persist_fail_after_rename() -> bool {
VIEW_PERSIST_FAIL_AFTER_RENAME.with(|failpoint| {
let fail = failpoint.get();
failpoint.set(false);
fail
})
}
struct ViewPersistError {
source: io::Error,
published: bool,
}
impl ViewPersistError {
fn before_publication(source: io::Error) -> Self {
Self {
source,
published: false,
}
}
fn after_publication(source: io::Error) -> Self {
Self {
source,
published: true,
}
}
fn into_io(self) -> io::Error {
self.source
}
}
#[derive(Debug, Clone)]
pub struct ViewDef {
pub name: String,
pub query: String,
pub depends_on: Vec<String>,
pub dirty: bool,
}
pub struct ViewRegistry {
views: FxHashMap<String, ViewDef>,
deps: FxHashMap<String, Vec<String>>,
data_dir: PathBuf,
}
impl ViewRegistry {
pub fn new(data_dir: &Path) -> Self {
ViewRegistry {
views: FxHashMap::default(),
deps: FxHashMap::default(),
data_dir: data_dir.to_path_buf(),
}
}
pub fn open(data_dir: &Path) -> io::Result<Self> {
let path = data_dir.join(VIEW_FILE);
if !path.exists() {
return Ok(Self::new(data_dir));
}
let defs = read_view_file(&path)?;
let mut reg = Self::new(data_dir);
for def in defs {
reg.insert_def(def);
}
Ok(reg)
}
pub fn register(&mut self, def: ViewDef) -> io::Result<()> {
if self.views.contains_key(&def.name) {
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
format!("view '{}' already exists", def.name),
));
}
let name = def.name.clone();
self.insert_def(def);
if let Err(e) = self.persist_for_lifecycle() {
if !e.published {
self.remove_def(&name);
}
return Err(e.into_io());
}
Ok(())
}
pub fn unregister(&mut self, name: &str) -> io::Result<()> {
let def = self.remove_def(name).ok_or_else(|| {
io::Error::new(io::ErrorKind::NotFound, format!("view '{name}' not found"))
})?;
if let Err(e) = self.persist_for_lifecycle() {
if !e.published {
self.insert_def(def);
}
return Err(e.into_io());
}
Ok(())
}
pub fn get(&self, name: &str) -> Option<&ViewDef> {
self.views.get(name)
}
#[inline]
pub fn is_view(&self, name: &str) -> bool {
self.views.contains_key(name)
}
pub fn mark_dirty(&mut self, view_name: &str) -> io::Result<()> {
let changed = match self.views.get_mut(view_name) {
Some(def) if !def.dirty => {
def.dirty = true;
true
}
_ => false,
};
if changed {
self.persist()
} else {
Ok(())
}
}
pub fn mark_dirty_in_memory(&mut self, view_name: &str) {
if let Some(def) = self.views.get_mut(view_name) {
def.dirty = true;
}
}
pub fn mark_clean(&mut self, view_name: &str) -> io::Result<()> {
let changed = match self.views.get_mut(view_name) {
Some(def) if def.dirty => {
def.dirty = false;
true
}
_ => false,
};
if !changed {
return Ok(());
}
if let Err(e) = self.persist() {
if let Some(def) = self.views.get_mut(view_name) {
def.dirty = true;
}
return Err(e);
}
Ok(())
}
pub fn mark_clean_in_memory(&mut self, view_name: &str) {
if let Some(def) = self.views.get_mut(view_name) {
def.dirty = false;
}
}
#[inline]
pub fn is_dirty(&self, view_name: &str) -> bool {
self.views.get(view_name).is_some_and(|d| d.dirty)
}
#[inline]
pub fn mark_dependents_dirty(&mut self, table: &str) -> io::Result<()> {
let any_clean = self.deps.get(table).is_some_and(|names| {
names
.iter()
.any(|n| self.views.get(n.as_str()).is_some_and(|d| !d.dirty))
});
if !any_clean {
return Ok(());
}
let mut queue: Vec<String> = vec![table.to_string()];
while let Some(source) = queue.pop() {
let names: Vec<String> = self.deps.get(source.as_str()).cloned().unwrap_or_default();
for name in names {
match self.views.get_mut(name.as_str()) {
Some(def) if !def.dirty => {
def.dirty = true;
queue.push(name);
}
_ => {}
}
}
}
self.persist()
}
pub fn dependents_of(&self, source: &str) -> Vec<String> {
let mut found: Vec<String> = Vec::new();
let mut queue: Vec<String> = vec![source.to_string()];
while let Some(current) = queue.pop() {
for name in self.deps.get(current.as_str()).into_iter().flatten() {
if !found.iter().any(|seen| seen == name) {
found.push(name.clone());
queue.push(name.clone());
}
}
}
found.sort();
found
}
pub fn list_views(&self) -> Vec<&str> {
self.views.keys().map(|k| k.as_str()).collect()
}
pub fn validate_backing_tables(&self, catalog: &crate::catalog::Catalog) -> io::Result<()> {
let mut missing: Vec<&str> = self
.views
.keys()
.map(|name| name.as_str())
.filter(|name| catalog.schema(name).is_none())
.collect();
if missing.is_empty() {
return Ok(());
}
missing.sort_unstable();
Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"{} registers materialized view{} {} but the catalog has no backing table{}; \
restore the missing table file/catalog entry from backup, or repair by \
removing/recreating the view after taking an operator-approved copy of the \
data directory",
self.data_dir.join(VIEW_FILE).display(),
if missing.len() == 1 { "" } else { "s" },
missing.join(", "),
if missing.len() == 1 { "" } else { "s" }
),
))
}
fn insert_def(&mut self, def: ViewDef) {
let name = def.name.clone();
for table in &def.depends_on {
self.deps
.entry(table.clone())
.or_default()
.push(name.clone());
}
self.views.insert(name, def);
}
fn remove_def(&mut self, name: &str) -> Option<ViewDef> {
let def = self.views.remove(name)?;
for table in &def.depends_on {
if let Some(list) = self.deps.get_mut(table) {
list.retain(|v| v != name);
if list.is_empty() {
self.deps.remove(table);
}
}
}
Some(def)
}
fn persist(&self) -> io::Result<()> {
self.persist_for_lifecycle()
.map_err(ViewPersistError::into_io)
}
fn persist_for_lifecycle(&self) -> Result<(), ViewPersistError> {
let path = self.data_dir.join(VIEW_FILE);
let tmp = self.data_dir.join(format!("{VIEW_FILE}.tmp"));
let defs: Vec<&ViewDef> = self.views.values().collect();
write_view_file(&tmp, &defs).map_err(ViewPersistError::before_publication)?;
#[cfg(test)]
if take_view_persist_fail_before_rename() {
return Err(ViewPersistError::before_publication(io::Error::other(
"injected view registry failure before rename",
)));
}
fs::rename(&tmp, &path).map_err(ViewPersistError::before_publication)?;
#[cfg(test)]
if take_view_persist_fail_after_rename() {
return Err(ViewPersistError::after_publication(io::Error::other(
"injected view registry failure after rename",
)));
}
crate::catalog::sync_directory(&self.data_dir)
.map_err(ViewPersistError::after_publication)?;
Ok(())
}
}
fn write_view_file(path: &Path, defs: &[&ViewDef]) -> io::Result<()> {
let mut buf: Vec<u8> = Vec::with_capacity(256);
buf.extend_from_slice(VIEW_MAGIC);
buf.extend_from_slice(&VIEW_VERSION.to_le_bytes());
buf.extend_from_slice(&(defs.len() as u32).to_le_bytes());
for def in defs {
let name = def.name.as_bytes();
buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
buf.extend_from_slice(name);
let query = def.query.as_bytes();
buf.extend_from_slice(&(query.len() as u32).to_le_bytes());
buf.extend_from_slice(query);
buf.extend_from_slice(&(def.depends_on.len() as u16).to_le_bytes());
for dep in &def.depends_on {
let d = dep.as_bytes();
buf.extend_from_slice(&(d.len() as u32).to_le_bytes());
buf.extend_from_slice(d);
}
buf.push(if def.dirty { 1 } else { 0 });
}
let mut f = fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(path)?;
f.write_all(&buf)?;
f.sync_data()?;
Ok(())
}
fn read_view_file(path: &Path) -> io::Result<Vec<ViewDef>> {
let mut f = fs::File::open(path)?;
let mut buf = Vec::new();
f.read_to_end(&mut buf)?;
let mut pos = 0usize;
if buf.len() < 10 || &buf[0..4] != VIEW_MAGIC {
return Err(io::Error::new(io::ErrorKind::InvalidData, "bad view magic"));
}
pos += 4;
let version = u16::from_le_bytes(
buf[pos..pos + 2]
.try_into()
.expect("invariant: buf.len() >= 10 checked above covers the 2-byte version field"),
);
pos += 2;
if version != VIEW_VERSION {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("unsupported view version: {version}"),
));
}
let n_views = u32::from_le_bytes(
buf[pos..pos + 4]
.try_into()
.expect("invariant: buf.len() >= 10 checked above covers the 4-byte n_views field"),
) as usize;
pos += 4;
let mut defs = Vec::with_capacity(n_views);
for _ in 0..n_views {
let name = read_str(&buf, &mut pos)?;
let query = read_str(&buf, &mut pos)?;
let n_deps = read_u16(&buf, &mut pos)? as usize;
let mut depends_on = Vec::with_capacity(n_deps);
for _ in 0..n_deps {
depends_on.push(read_str(&buf, &mut pos)?);
}
let dirty = read_u8(&buf, &mut pos)? != 0;
defs.push(ViewDef {
name,
query,
depends_on,
dirty,
});
}
Ok(defs)
}
fn read_u8(buf: &[u8], pos: &mut usize) -> io::Result<u8> {
if *pos >= buf.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"truncated view file",
));
}
let v = buf[*pos];
*pos += 1;
Ok(v)
}
fn read_u16(buf: &[u8], pos: &mut usize) -> io::Result<u16> {
if *pos + 2 > buf.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"truncated view file",
));
}
let v = u16::from_le_bytes(
buf[*pos..*pos + 2]
.try_into()
.expect("invariant: bounds checked immediately above (*pos + 2 <= buf.len())"),
);
*pos += 2;
Ok(v)
}
fn read_str(buf: &[u8], pos: &mut usize) -> io::Result<String> {
if *pos + 4 > buf.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"truncated view file",
));
}
let len = u32::from_le_bytes(
buf[*pos..*pos + 4]
.try_into()
.expect("invariant: bounds checked immediately above (*pos + 4 <= buf.len())"),
) as usize;
*pos += 4;
if *pos + len > buf.len() {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"truncated view file",
));
}
let s = std::str::from_utf8(&buf[*pos..*pos + len])
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "non-utf8 in view file"))?
.to_string();
*pos += len;
Ok(s)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::catalog::Catalog;
use crate::types::{ColumnDef, Schema, TypeId};
fn fail_next_view_persist_before_rename() {
VIEW_PERSIST_FAIL_BEFORE_RENAME.with(|failpoint| failpoint.set(true));
}
fn fail_next_view_persist_after_rename() {
VIEW_PERSIST_FAIL_AFTER_RENAME.with(|failpoint| failpoint.set(true));
}
fn temp_registry(name: &str) -> ViewRegistry {
let dir = std::env::temp_dir().join(format!("powdb_view_{name}_{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
ViewRegistry::new(&dir)
}
fn one_int_schema(name: &str) -> Schema {
Schema {
table_name: name.into(),
columns: vec![ColumnDef {
name: "id".into(),
type_id: TypeId::Int,
required: true,
position: 0,
}],
}
}
#[test]
fn test_read_view_file_garbage_errors_not_panic() {
let dir = std::env::temp_dir().join(format!("powdb_view_garbage_{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let cases: Vec<Vec<u8>> = vec![
vec![], b"BVIW".to_vec(), b"XXXX\x01\x00\x00\x00\x00\x00".to_vec(), {
let mut v = b"BVIW".to_vec();
v.extend_from_slice(&VIEW_VERSION.to_le_bytes());
v.extend_from_slice(&1u32.to_le_bytes());
v
},
{
let mut v = b"BVIW".to_vec();
v.extend_from_slice(&VIEW_VERSION.to_le_bytes());
v.extend_from_slice(&1u32.to_le_bytes());
v.extend_from_slice(&0xFFFF_FFFFu32.to_le_bytes());
v
},
];
for (i, bytes) in cases.iter().enumerate() {
let p = dir.join(format!("v{i}.bin"));
std::fs::write(&p, bytes).unwrap();
let result = read_view_file(&p);
assert!(
result.is_err(),
"expected Err for garbage view file case {i} ({bytes:?}), got Ok"
);
}
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn test_register_and_lookup() {
let mut reg = temp_registry("basic");
reg.register(ViewDef {
name: "ActiveUsers".into(),
query: "User filter .active = true".into(),
depends_on: vec!["User".into()],
dirty: false,
})
.unwrap();
assert!(reg.is_view("ActiveUsers"));
assert!(!reg.is_view("User"));
let def = reg.get("ActiveUsers").unwrap();
assert_eq!(def.query, "User filter .active = true");
}
#[test]
fn test_dirty_tracking() {
let mut reg = temp_registry("dirty");
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
assert!(!reg.is_dirty("V1"));
reg.mark_dependents_dirty("T1").unwrap();
assert!(reg.is_dirty("V1"));
reg.mark_clean("V1").unwrap();
assert!(!reg.is_dirty("V1"));
}
#[test]
fn test_dirty_flag_survives_reopen() {
let dir = std::env::temp_dir().join(format!("powdb_view_dirty_rt_{}", std::process::id()));
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).unwrap();
{
let mut reg = ViewRegistry::new(&dir);
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
reg.mark_dependents_dirty("T1").unwrap();
}
{
let mut reg = ViewRegistry::open(&dir).unwrap();
assert!(reg.is_dirty("V1"), "dirty flag must survive process exit");
reg.mark_clean("V1").unwrap();
}
let reg = ViewRegistry::open(&dir).unwrap();
assert!(!reg.is_dirty("V1"), "clean flag must survive process exit");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn test_mark_dependents_dirty_is_idempotent_after_first_transition() {
let dir =
std::env::temp_dir().join(format!("powdb_view_transition_{}", std::process::id()));
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).unwrap();
let mut reg = ViewRegistry::new(&dir);
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
reg.mark_dependents_dirty("T1").unwrap();
let path = dir.join(VIEW_FILE);
assert!(path.exists(), "the transition itself must write the file");
std::fs::remove_file(&path).unwrap();
for _ in 0..100 {
reg.mark_dependents_dirty("T1").unwrap();
}
assert!(
!path.exists(),
"re-marking an already dirty view must not rewrite views.bin"
);
reg.mark_clean("V1").unwrap();
assert!(path.exists());
std::fs::remove_file(&path).unwrap();
for _ in 0..100 {
reg.mark_clean("V1").unwrap();
}
assert!(!path.exists());
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn test_multi_dependency() {
let mut reg = temp_registry("multi");
reg.register(ViewDef {
name: "V1".into(),
query: "T1 inner join T2 on .id = .fk".into(),
depends_on: vec!["T1".into(), "T2".into()],
dirty: false,
})
.unwrap();
reg.mark_dependents_dirty("T2").unwrap();
assert!(reg.is_dirty("V1"));
}
#[test]
fn test_unregister() {
let mut reg = temp_registry("unreg");
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
reg.unregister("V1").unwrap();
assert!(!reg.is_view("V1"));
reg.mark_dependents_dirty("T1").unwrap();
}
#[test]
fn register_rolls_back_memory_when_registry_persist_fails_before_rename() {
let mut reg = temp_registry("register_fail");
fail_next_view_persist_before_rename();
let error = reg
.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap_err();
assert!(
error
.to_string()
.contains("injected view registry failure before rename"),
"expected injected registry failure, got: {error}"
);
assert!(
!reg.is_view("V1"),
"a failed register must not leave an in-memory ghost view"
);
assert!(
!reg.data_dir.join(VIEW_FILE).exists(),
"a failed register before rename must not create views.bin"
);
}
#[test]
fn unregister_rolls_back_memory_when_registry_persist_fails_before_rename() {
let mut reg = temp_registry("unregister_fail");
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
fail_next_view_persist_before_rename();
let error = reg.unregister("V1").unwrap_err();
assert!(
error
.to_string()
.contains("injected view registry failure before rename"),
"expected injected registry failure, got: {error}"
);
assert!(
reg.is_view("V1"),
"a failed unregister must keep the in-memory registry aligned with disk"
);
reg.mark_dependents_dirty("T1").unwrap();
assert!(
reg.is_dirty("V1"),
"rollback must also restore dependency edges"
);
let reopened = ViewRegistry::open(®.data_dir).unwrap();
assert!(
reopened.is_view("V1"),
"the view must still be registered after reopening"
);
}
#[test]
fn register_keeps_published_memory_when_registry_sync_fails_after_rename() {
let mut reg = temp_registry("register_after_rename_fail");
fail_next_view_persist_after_rename();
let error = reg
.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap_err();
assert!(
error
.to_string()
.contains("injected view registry failure after rename"),
"expected injected post-rename registry failure, got: {error}"
);
assert!(
reg.is_view("V1"),
"after rename, memory must keep the potentially published register state"
);
let reopened = ViewRegistry::open(®.data_dir).unwrap();
assert!(
reopened.is_view("V1"),
"the post-rename failure fixture publishes views.bin before returning the error"
);
}
#[test]
fn unregister_keeps_published_memory_when_registry_sync_fails_after_rename() {
let mut reg = temp_registry("unregister_after_rename_fail");
reg.register(ViewDef {
name: "V1".into(),
query: "T1".into(),
depends_on: vec!["T1".into()],
dirty: false,
})
.unwrap();
fail_next_view_persist_after_rename();
let error = reg.unregister("V1").unwrap_err();
assert!(
error
.to_string()
.contains("injected view registry failure after rename"),
"expected injected post-rename registry failure, got: {error}"
);
assert!(
!reg.is_view("V1"),
"after rename, memory must keep the potentially published unregister state"
);
let reopened = ViewRegistry::open(®.data_dir).unwrap();
assert!(
!reopened.is_view("V1"),
"the post-rename failure fixture publishes views.bin before returning the error"
);
}
#[test]
fn validate_backing_tables_accepts_dirty_views_and_normal_tables() {
let dir =
std::env::temp_dir().join(format!("powdb_view_validate_ok_{}", std::process::id()));
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).unwrap();
let mut catalog = Catalog::create(&dir).unwrap();
catalog.create_table(one_int_schema("Base")).unwrap();
catalog.create_table(one_int_schema("V")).unwrap();
catalog.create_table(one_int_schema("Ordinary")).unwrap();
let mut reg = ViewRegistry::new(&dir);
reg.register(ViewDef {
name: "V".into(),
query: "MissingSource { .id }".into(),
depends_on: vec!["MissingSource".into()],
dirty: true,
})
.unwrap();
reg.validate_backing_tables(&catalog)
.expect("dirty views and ordinary non-view tables are not backing-table damage");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn validate_backing_tables_rejects_registered_view_without_backing_table() {
let dir = std::env::temp_dir().join(format!(
"powdb_view_validate_missing_{}",
std::process::id()
));
std::fs::remove_dir_all(&dir).ok();
std::fs::create_dir_all(&dir).unwrap();
let mut catalog = Catalog::create(&dir).unwrap();
catalog.create_table(one_int_schema("Base")).unwrap();
let mut reg = ViewRegistry::new(&dir);
reg.register(ViewDef {
name: "V".into(),
query: "Base { .id }".into(),
depends_on: vec!["Base".into()],
dirty: false,
})
.unwrap();
let error = reg.validate_backing_tables(&catalog).unwrap_err();
let message = error.to_string();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
assert!(
message.contains("views.bin")
&& message.contains("materialized view V")
&& message.contains("no backing table")
&& message.contains("restore")
&& message.contains("recreating the view"),
"missing backing table error must be actionable, got: {message}"
);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn test_persist_and_reopen() {
let dir = std::env::temp_dir().join(format!("powdb_view_persist_{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
{
let mut reg = ViewRegistry::new(&dir);
reg.register(ViewDef {
name: "V1".into(),
query: "User filter .active = true".into(),
depends_on: vec!["User".into()],
dirty: true,
})
.unwrap();
}
let reg = ViewRegistry::open(&dir).unwrap();
assert!(reg.is_view("V1"));
let def = reg.get("V1").unwrap();
assert_eq!(def.query, "User filter .active = true");
assert!(def.dirty);
assert_eq!(def.depends_on, vec!["User".to_string()]);
}
}