siftdb_core/

storage.rs

use crate::types::{Frame, FileHeader, HandleMetadata};
use anyhow::Result;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

/// Segment writer for append-only storage
pub struct SegmentWriter {
    file: File,
    #[allow(dead_code)]
    path: PathBuf,
    current_offset: u64,
    seg_id: u32,
}

impl SegmentWriter {
    pub fn new(store_path: &Path, seg_id: u32) -> Result<Self> {
        let path = store_path.join(format!("seg-{:06}.sift", seg_id));
        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)?;
        
        // Get current file size to track offset
        let current_offset = file.metadata()?.len();
        
        Ok(Self {
            file,
            path,
            current_offset,
            seg_id,
        })
    }
    
    pub fn write_frame(&mut self, frame: &Frame) -> Result<HandleMetadata> {
        let start_offset = self.current_offset;
        
        // Calculate frame length (excluding the frame_len field itself)
        let frame_content_len = frame.frame_size() - 4;
        
        // Write frame_len
        self.file.write_all(&(frame_content_len as u32).to_le_bytes())?;
        
        // Write header_crc32
        let header_bytes = frame.header.to_bytes();
        let header_crc = crc32fast::hash(&header_bytes);
        self.file.write_all(&header_crc.to_le_bytes())?;
        
        // Write FileHeader (64 bytes)
        self.file.write_all(&header_bytes)?;
        
        // Write content_crc32
        let content_crc = crc32fast::hash(&frame.content);
        self.file.write_all(&content_crc.to_le_bytes())?;
        
        // Write content
        self.file.write_all(&frame.content)?;
        
        // Write line table
        let line_tbl_offset = (self.current_offset + 4 + 4 + 64 + 4 + frame.content.len() as u64) as u32;
        self.file.write_all(&frame.line_table)?;
        
        // Write frame_crc32 (CRC of everything except frame_len and this CRC itself)
        let mut frame_data = Vec::new();
        frame_data.extend_from_slice(&header_crc.to_le_bytes());
        frame_data.extend_from_slice(&header_bytes);
        frame_data.extend_from_slice(&content_crc.to_le_bytes());
        frame_data.extend_from_slice(&frame.content);
        frame_data.extend_from_slice(&frame.line_table);
        let frame_crc = crc32fast::hash(&frame_data);
        self.file.write_all(&frame_crc.to_le_bytes())?;
        
        // Pad to 4KiB alignment
        let total_written = frame.frame_size();
        let padded_size = frame.padded_frame_size();
        let padding_needed = padded_size - total_written;
        if padding_needed > 0 {
            let padding = vec![0u8; padding_needed];
            self.file.write_all(&padding)?;
        }
        
        self.file.flush()?;
        
        let metadata = HandleMetadata {
            seg_id: self.seg_id,
            offset: start_offset,
            frame_len: padded_size as u32,
            line_tbl_offset,
        };
        
        self.current_offset += padded_size as u64;
        
        Ok(metadata)
    }
    
    pub fn current_offset(&self) -> u64 {
        self.current_offset
    }
    
    pub fn seg_id(&self) -> u32 {
        self.seg_id
    }
}

/// Segment reader for reading frames
pub struct SegmentReader {
    file: File,
    #[allow(dead_code)]
    seg_id: u32,
}

impl SegmentReader {
    pub fn new(store_path: &Path, seg_id: u32) -> Result<Self> {
        let path = store_path.join(format!("seg-{:06}.sift", seg_id));
        let file = File::open(path)?;
        
        Ok(Self { file, seg_id })
    }
    
    pub fn read_frame(&mut self, metadata: &HandleMetadata) -> Result<Frame> {
        // Seek to frame start
        self.file.seek(SeekFrom::Start(metadata.offset))?;
        
        // Read frame_len
        let mut buf = [0u8; 4];
        self.file.read_exact(&mut buf)?;
        let _frame_len = u32::from_le_bytes(buf);
        
        // Read header_crc32
        self.file.read_exact(&mut buf)?;
        let expected_header_crc = u32::from_le_bytes(buf);
        
        // Read FileHeader (64 bytes)
        let mut header_buf = [0u8; 64];
        self.file.read_exact(&mut header_buf)?;
        let actual_header_crc = crc32fast::hash(&header_buf);
        
        if actual_header_crc != expected_header_crc {
            anyhow::bail!("Header CRC mismatch");
        }
        
        let header = FileHeader::from_bytes(&header_buf)?;
        
        // Read content_crc32
        self.file.read_exact(&mut buf)?;
        let expected_content_crc = u32::from_le_bytes(buf);
        
        // Read content
        let mut content = vec![0u8; header.content_len as usize];
        self.file.read_exact(&mut content)?;
        let actual_content_crc = crc32fast::hash(&content);
        
        if actual_content_crc != expected_content_crc {
            anyhow::bail!("Content CRC mismatch");
        }
        
        // Read line table
        let mut line_table = vec![0u8; header.line_tbl_len as usize];
        self.file.read_exact(&mut line_table)?;
        
        // Verify frame CRC
        self.file.read_exact(&mut buf)?;
        let expected_frame_crc = u32::from_le_bytes(buf);
        
        let mut frame_data = Vec::new();
        frame_data.extend_from_slice(&expected_header_crc.to_le_bytes());
        frame_data.extend_from_slice(&header_buf);
        frame_data.extend_from_slice(&expected_content_crc.to_le_bytes());
        frame_data.extend_from_slice(&content);
        frame_data.extend_from_slice(&line_table);
        let actual_frame_crc = crc32fast::hash(&frame_data);
        
        if actual_frame_crc != expected_frame_crc {
            anyhow::bail!("Frame CRC mismatch");
        }
        
        Ok(Frame {
            header,
            content,
            line_table,
        })
    }
}

/// Generate line table from content (delta-encoded varint positions of newlines)
pub fn generate_line_table(content: &[u8]) -> Vec<u8> {
    let mut positions = Vec::new();
    
    // Find all newline positions
    for (i, &byte) in content.iter().enumerate() {
        if byte == b'\n' {
            positions.push(i as u32);
        }
    }
    
    // Delta-encode and write as varints
    let mut line_table = Vec::new();
    let mut last_pos = 0u32;
    
    for pos in positions {
        let delta = pos - last_pos;
        write_varint(&mut line_table, delta as u64);
        last_pos = pos;
    }
    
    line_table
}

/// Decode line table to get newline positions
pub fn decode_line_table(line_table: &[u8]) -> Result<Vec<u32>> {
    let mut positions = Vec::new();
    let mut cursor = 0;
    let mut current_pos = 0u32;
    
    while cursor < line_table.len() {
        let (delta, bytes_read) = read_varint(&line_table[cursor..])?;
        current_pos += delta as u32;
        positions.push(current_pos);
        cursor += bytes_read;
    }
    
    Ok(positions)
}

/// Convert byte offset to line number using line table
pub fn byte_to_line(byte_offset: usize, newline_positions: &[u32]) -> u32 {
    // Binary search to find the line number
    match newline_positions.binary_search(&(byte_offset as u32)) {
        Ok(index) => (index + 1) as u32, // Found exact newline position
        Err(index) => (index + 1) as u32, // Insert position gives us the line number
    }
}

// Varint encoding/decoding utilities
fn write_varint(buf: &mut Vec<u8>, mut value: u64) {
    while value >= 0x80 {
        buf.push((value as u8) | 0x80);
        value >>= 7;
    }
    buf.push(value as u8);
}

fn read_varint(data: &[u8]) -> Result<(u64, usize)> {
    let mut result = 0u64;
    let mut shift = 0;
    let mut bytes_read = 0;
    
    for &byte in data {
        bytes_read += 1;
        result |= ((byte & 0x7F) as u64) << shift;
        
        if (byte & 0x80) == 0 {
            return Ok((result, bytes_read));
        }
        
        shift += 7;
        if shift >= 64 {
            anyhow::bail!("Varint too large");
        }
    }
    
    anyhow::bail!("Incomplete varint");
}