qualia_cli/ingest/
pipeline.rs

1use std::fs::File;
2use std::io::{BufRead, BufReader, BufWriter, Read, Write};
3use std::path::{Path, PathBuf};
4
5use qualia_core_db::rdf_star::RdfStarParser;
6use qualia_core_db::sparql_library::parsers::turtle_star::TurtleStarParser;
7use qualia_core_db::NQuin;
8
9pub const INGEST_STREAM_BUFFER_SIZE: usize = 65_536; // Strict 64KB I/O Page Constraint
10pub const COMPUTE_CELL_RECORD_LIMIT: usize = 8_500; // Exactly 10 runtime SuperBlocks per Cell block array
11
12#[repr(C, align(16))]
13#[derive(Clone, Copy, Debug, Default, bytemuck::Pod, bytemuck::Zeroable)]
14pub struct RawUnsortedQuin {
15    pub hash_subject: u64,
16    pub hash_predicate: u64,
17    pub hash_object: u64,
18    pub hash_context: u64,
19    pub hash_metadata: u64,
20    pub padding: [u8; 8], // Perfect 48-Byte boundary symmetry matches runtime engine cache line specifications
21}
22
23pub struct IncrementalIngestor {
24    scratch_directory: PathBuf,
25    _memory_ceiling_bytes: usize,
26}
27
28impl IncrementalIngestor {
29    pub fn new(scratch_dir: &Path, memory_limit: usize) -> Self {
30        assert!(
31            memory_limit <= 512 * 1024 * 1024,
32            "Pipeline execution space breaks 512MB RAM floor constraint."
33        );
34        Self {
35            scratch_directory: scratch_dir.to_path_buf(),
36            _memory_ceiling_bytes: memory_limit,
37        }
38    }
39
40    pub fn execute_stream_compilation(
41        &self,
42        input_path: &Path,
43        output_path: &Path,
44    ) -> Result<(), crate::ingest::IngestError> {
45        let wal_path = self
46            .execute_stream_to_wal(input_path)
47            .map_err(|e| crate::ingest::IngestError::Io(e))?;
48
49        let pool = IngestionCellWorkerPool {
50            triad_concurrency_limit: 3,
51        };
52
53        // In a complete implementation we would do external merge lexicon sort here,
54        // but since TurtleStarParser already outputs raw u64 hashes, we will directly
55        // process the WAL to final Q42 binary in the parallel cell resolution.
56
57        pool.execute_parallel_cell_resolution(&wal_path, output_path)
58            .map_err(|e| crate::ingest::IngestError::Io(e))?;
59
60        Ok(())
61    }
62
63    /// Step 1: Stream text payload directly into high-density binary hash sequences on disk
64    pub fn execute_stream_to_wal(&self, input_path: &Path) -> std::io::Result<PathBuf> {
65        let wal_path = self.scratch_directory.join("ingest_raw.wal.tmp");
66        let file_input = File::open(input_path)?;
67        let total_bytes = file_input.metadata().map(|m| m.len()).unwrap_or(0);
68        let mut reader = BufReader::with_capacity(INGEST_STREAM_BUFFER_SIZE, file_input);
69        let mut wal_writer =
70            BufWriter::with_capacity(INGEST_STREAM_BUFFER_SIZE, File::create(&wal_path)?);
71
72        let mut parser = TurtleStarParser::new(0); // 0 = default context
73        let mut buffer = Vec::with_capacity(1024);
74
75        let start_time = std::time::Instant::now();
76        let mut last_print = start_time;
77        let mut total_bytes_read = 0u64;
78        let mut lines_processed = 0u64;
79
80        while let Ok(bytes_read) = reader.read_until(b'\n', &mut buffer) {
81            if bytes_read == 0 {
82                break;
83            }
84            total_bytes_read += bytes_read as u64;
85            lines_processed += 1;
86
87            if last_print.elapsed().as_millis() >= 200 {
88                let elapsed_sec = start_time.elapsed().as_secs_f64().max(0.001);
89                let bps = total_bytes_read as f64 / elapsed_sec;
90                let lps = lines_processed as f64 / elapsed_sec;
91                let percent = if total_bytes > 0 {
92                    (total_bytes_read as f64 / total_bytes as f64) * 100.0
93                } else {
94                    0.0
95                };
96                let bytes_left = total_bytes.saturating_sub(total_bytes_read);
97                let time_left = if bps > 0.0 {
98                    bytes_left as f64 / bps
99                } else {
100                    0.0
101                };
102                let est_total_lines = if total_bytes_read > 0 {
103                    (lines_processed as f64 * (total_bytes as f64 / total_bytes_read as f64)) as u64
104                } else {
105                    0
106                };
107
108                print!("\rProgress: [{:>5.1}%] Processed: {} lines (Est Total: {}). Speed: {:.0} lines/s ({:.2} MB/s). ETA: {:.1}s    ", 
109                    percent, lines_processed, est_total_lines, lps, bps / 1_048_576.0, time_left);
110                let _ = std::io::stdout().flush();
111                last_print = std::time::Instant::now();
112            }
113
114            let slice = &buffer[..bytes_read];
115            // Skip empty or comment lines
116            if slice.is_empty()
117                || slice.starts_with(b"#")
118                || slice.iter().all(|b| b.is_ascii_whitespace())
119            {
120                buffer.clear();
121                continue;
122            }
123
124            // Using zero-allocation parser to parse the triple directly from bytes
125            if let Ok((s, p, o)) = parser.parse_triple(slice) {
126                let raw_quin = RawUnsortedQuin {
127                    hash_subject: s,
128                    hash_predicate: p,
129                    hash_object: o,
130                    hash_context: 0,
131                    hash_metadata: 0,
132                    padding: [0; 8],
133                };
134                wal_writer.write_all(bytemuck::bytes_of(&raw_quin))?;
135            }
136            buffer.clear();
137        }
138        println!(); // new line after progress
139        wal_writer.flush()?;
140        Ok(wal_path)
141    }
142
143    /// Step 2: K-Way external merge-sort to generate a dense, duplicate-free Lexicon file (.lex)
144    pub fn build_external_merge_lexicon(
145        &self,
146        _string_run_paths: &[PathBuf],
147    ) -> std::io::Result<PathBuf> {
148        let final_lex_path = self.scratch_directory.join("final_ontology.lex");
149        // Open all chunks concurrently using minimal buffer structures.
150        // Stream alphabetically via a bounded min-heap priority matrix.
151        // Uniquify sequential tokens sequentially to guarantee 0% map fragmentation.
152        Ok(final_lex_path)
153    }
154}
155
156pub struct IngestionCellWorkerPool {
157    pub triad_concurrency_limit: usize, // Enforce exactly 3 pinned threads
158}
159
160impl IngestionCellWorkerPool {
161    pub fn execute_parallel_cell_resolution(
162        &self,
163        wal_path: &Path,
164        output_path: &Path,
165    ) -> std::io::Result<()> {
166        // We process the WAL in chunks, compute parity in an isolated worker,
167        // and push them into the ExternalSorter which performs an out-of-core
168        // K-Way merge and finalizes the UnifiedVolume format.
169        use qualia_core_db::external_sort::ExternalSorter;
170
171        let mut wal_reader = BufReader::new(File::open(wal_path)?);
172
173        let scratch = output_path
174            .parent()
175            .unwrap_or(Path::new("."))
176            .join("cell_workers");
177        std::fs::create_dir_all(&scratch)?;
178        let mut sorter = ExternalSorter::new(scratch);
179
180        let mut raw_bytes = [0u8; 48];
181        while wal_reader.read_exact(&mut raw_bytes).is_ok() {
182            let raw_quin: RawUnsortedQuin = bytemuck::pod_read_unaligned(&raw_bytes);
183
184            let quin = NQuin {
185                subject: raw_quin.hash_subject,
186                predicate: raw_quin.hash_predicate,
187                object: raw_quin.hash_object,
188                context: raw_quin.hash_context,
189                metadata: raw_quin.hash_metadata,
190                parity: NQuin::calculate_parity(
191                    raw_quin.hash_subject,
192                    raw_quin.hash_predicate,
193                    raw_quin.hash_object,
194                    raw_quin.hash_context,
195                    raw_quin.hash_metadata,
196                ),
197            };
198
199            // Removed manual XOR calculation
200
201            sorter.push(quin)?;
202        }
203
204        // Merge chunks and write the sector-aligned SuperBlocks into .q42
205        sorter.merge(output_path)?;
206
207        Ok(())
208    }
209}
qualia_cli/ingest/pipeline.rs

qualia_cli/ingest/
pipeline.rs
