1use crate::footer::Precision;
3use ailake_core::{AilakeError, AilakeResult, Centroid, VectorMetric};
4use ailake_index::{AnyIndex, HnswIndex, IvfPqSerializer, MmapLoader};
5use ailake_parquet::ParquetVectorReader;
6use arrow_array::RecordBatch;
7use bytes::Bytes;
8
9use crate::footer::{
10 parquet_footer_start, AilakeHeader, AilakeTrailer, DistanceMetric, AILK_FTS_HEADER_SIZE,
11 AILK_FTS_MAGIC, FLAG_INDEX_IVF_PQ, HEADER_SIZE, KV_FTS_OFFSET, TRAILER_SIZE,
12};
13
14pub struct AilakeFileReader {
15 bytes: Bytes,
16 vector_column: String,
17 #[allow(dead_code)]
18 dim: u32,
19}
20
21impl AilakeFileReader {
22 pub fn new(bytes: Bytes, vector_column: &str, dim: u32) -> Self {
23 Self {
24 bytes,
25 vector_column: vector_column.to_string(),
26 dim,
27 }
28 }
29
30 pub fn ailk_offset(&self) -> AilakeResult<u64> {
36 let reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
37 if let Some(val) = reader.kv_metadata("ailake.footer_offset")? {
38 return val.parse::<u64>().map_err(|_| AilakeError::NotAnAilakeFile);
39 }
40 self.ailk_offset_from_trailer()
41 }
42
43 pub fn ailk_offset_for_column(&self, column: &str) -> AilakeResult<u64> {
50 let reader = ParquetVectorReader::new(self.bytes.clone(), column);
51 let col_key = format!("ailake.{column}.footer_offset");
52 if let Some(val) = reader.kv_metadata(&col_key)? {
53 return val.parse::<u64>().map_err(|_| AilakeError::NotAnAilakeFile);
54 }
55 if let Some(val) = reader.kv_metadata("ailake.footer_offset")? {
56 return val.parse::<u64>().map_err(|_| AilakeError::NotAnAilakeFile);
57 }
58 self.ailk_offset_from_trailer()
59 }
60
61 pub fn has_column_footer(&self, column: &str) -> bool {
71 let reader = ParquetVectorReader::new(self.bytes.clone(), column);
72 let col_key = format!("ailake.{column}.footer_offset");
73 matches!(reader.kv_metadata(&col_key), Ok(Some(_)))
74 }
75
76 fn ailk_offset_from_trailer(&self) -> AilakeResult<u64> {
83 let buf = self.bytes.as_ref();
84 let footer_start = parquet_footer_start(buf)?;
85 if footer_start < TRAILER_SIZE {
86 return Err(AilakeError::NotAnAilakeFile);
87 }
88 let trailer_start = footer_start - TRAILER_SIZE;
89 let trailer_bytes: &[u8; TRAILER_SIZE] = buf[trailer_start..footer_start]
90 .try_into()
91 .map_err(|_| AilakeError::NotAnAilakeFile)?;
92 let trailer = AilakeTrailer::from_bytes(trailer_bytes)?;
93 Ok(trailer.footer_offset)
94 }
95
96 pub fn is_ailake_file(&self) -> bool {
98 self.ailk_offset().is_ok()
99 }
100
101 pub fn read_header(&self) -> AilakeResult<AilakeHeader> {
103 self.read_header_at_offset(self.ailk_offset()?)
104 }
105
106 pub fn read_header_for_column(&self, column: &str) -> AilakeResult<AilakeHeader> {
111 self.read_header_at_offset(self.ailk_offset_for_column(column)?)
112 }
113
114 fn read_header_at_offset(&self, offset: u64) -> AilakeResult<AilakeHeader> {
115 let offset = offset as usize;
116 let header_end = offset
117 .checked_add(HEADER_SIZE)
118 .ok_or(AilakeError::NotAnAilakeFile)?;
119 if header_end > self.bytes.len() {
120 return Err(AilakeError::NotAnAilakeFile);
121 }
122 let header_bytes: &[u8; HEADER_SIZE] = self.bytes[offset..header_end]
123 .try_into()
124 .map_err(|_| AilakeError::NotAnAilakeFile)?;
125 AilakeHeader::from_bytes(header_bytes)
126 }
127
128 pub fn get_centroid(&self) -> AilakeResult<Centroid> {
130 let ailk_start = self.ailk_offset()? as usize;
131 let header = self.read_header()?;
132 let centroid_start = ailk_start
133 .checked_add(header.centroid_offset as usize)
134 .ok_or(AilakeError::NotAnAilakeFile)?;
135 let centroid_end = centroid_start
136 .checked_add(header.centroid_len as usize)
137 .ok_or(AilakeError::NotAnAilakeFile)?;
138
139 if centroid_end > self.bytes.len() {
140 return Err(AilakeError::NotAnAilakeFile);
141 }
142
143 let centroid_data = &self.bytes[centroid_start..centroid_end];
144 let dim = header.dim as usize;
145 let expected_len = dim.checked_mul(4).and_then(|v| v.checked_add(4)).ok_or(
146 AilakeError::InvalidCentroidLength {
147 expected_dim: header.dim,
148 actual: centroid_data.len(),
149 },
150 )?;
151 if centroid_data.len() != expected_len {
152 return Err(AilakeError::InvalidCentroidLength {
153 expected_dim: header.dim,
154 actual: centroid_data.len(),
155 });
156 }
157
158 let values: Vec<f32> = centroid_data[..dim * 4]
159 .chunks_exact(4)
160 .map(|b| {
161 f32::from_le_bytes(
162 b.try_into()
163 .expect("chunks_exact(4) guarantees 4-byte slices"),
164 )
165 })
166 .collect();
167 let radius = f32::from_le_bytes(
168 centroid_data[dim * 4..]
169 .try_into()
170 .expect("invariant: validated len == dim*4 + 4 above"),
171 );
172 let metric = distance_metric_to_vector_metric(header.distance_metric);
173
174 Ok(Centroid {
175 values,
176 radius,
177 metric,
178 })
179 }
180
181 pub fn load_index(&self) -> AilakeResult<HnswIndex> {
183 self.load_index_for_column(&self.vector_column.clone())
184 }
185
186 pub fn load_index_for_column(&self, column: &str) -> AilakeResult<HnswIndex> {
191 let ailk_start = self.ailk_offset_for_column(column)? as usize;
192 let header = self.read_header_at_offset(ailk_start as u64)?;
193
194 let hnsw_start = ailk_start
195 .checked_add(header.hnsw_offset as usize)
196 .ok_or(AilakeError::NotAnAilakeFile)?;
197 let hnsw_end = hnsw_start
198 .checked_add(header.hnsw_len as usize)
199 .ok_or(AilakeError::NotAnAilakeFile)?;
200
201 if hnsw_end > self.bytes.len() {
202 return Err(AilakeError::NotAnAilakeFile);
203 }
204 let mut idx = MmapLoader::from_bytes(&self.bytes[hnsw_start..hnsw_end])?;
205 if header.precision == Precision::F16 {
206 idx.quantize_to_f16();
207 }
208 Ok(idx)
209 }
210
211 pub fn load_any_index(&self) -> AilakeResult<AnyIndex> {
213 self.load_any_index_for_column(&self.vector_column.clone())
214 }
215
216 pub fn load_any_index_for_column(&self, column: &str) -> AilakeResult<AnyIndex> {
218 let ailk_start = self.ailk_offset_for_column(column)? as usize;
219 let header = self.read_header_at_offset(ailk_start as u64)?;
220
221 let index_start = ailk_start
222 .checked_add(header.hnsw_offset as usize)
223 .ok_or(AilakeError::NotAnAilakeFile)?;
224 let index_end = index_start
225 .checked_add(header.hnsw_len as usize)
226 .ok_or(AilakeError::NotAnAilakeFile)?;
227
228 if index_end > self.bytes.len() {
229 return Err(AilakeError::NotAnAilakeFile);
230 }
231 let index_bytes = &self.bytes[index_start..index_end];
232
233 if header.flags & FLAG_INDEX_IVF_PQ != 0 {
234 let idx = IvfPqSerializer::from_bytes(index_bytes)?;
235 Ok(AnyIndex::IvfPq(idx))
236 } else {
237 let mut idx = MmapLoader::from_bytes(index_bytes)?;
238 if header.precision == Precision::F16 {
239 idx.quantize_to_f16();
240 }
241 Ok(AnyIndex::Hnsw(idx))
242 }
243 }
244
245 pub fn read_parquet(&self) -> AilakeResult<(RecordBatch, Vec<Vec<f32>>)> {
248 let reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
249 reader.read_all()
250 }
251
252 pub fn read_parquet_filtered(
256 &self,
257 filter: &ailake_core::ColumnFilter,
258 ) -> AilakeResult<(RecordBatch, Vec<Vec<f32>>)> {
259 let reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
260 reader.read_all_filtered(filter)
261 }
262
263 pub fn matching_row_ids(
269 &self,
270 filter: &ailake_core::ColumnFilter,
271 ) -> AilakeResult<std::collections::HashSet<u64>> {
272 let reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
273 reader.matching_row_ids(filter)
274 }
275
276 pub fn load_fts_blob(&self) -> AilakeResult<Option<Bytes>> {
281 let pq_reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
282 let fts_offset_str = match pq_reader.kv_metadata(KV_FTS_OFFSET)? {
283 Some(s) => s,
284 None => return Ok(None),
285 };
286 let fts_abs: usize = fts_offset_str
287 .parse::<u64>()
288 .map_err(|e| {
289 AilakeError::Fts(format!(
290 "invalid FTS section offset '{fts_offset_str}': {e}"
291 ))
292 })?
293 .try_into()
294 .map_err(|_| AilakeError::Fts("FTS section offset exceeds address space".into()))?;
295
296 if fts_abs + AILK_FTS_HEADER_SIZE > self.bytes.len() {
298 return Err(AilakeError::Fts("AILK_FTS header out of bounds".into()));
299 }
300 let hdr = &self.bytes[fts_abs..fts_abs + AILK_FTS_HEADER_SIZE];
301 if hdr[0..4] != AILK_FTS_MAGIC {
302 return Err(AilakeError::Fts(format!(
303 "bad AILK_FTS magic: {:?}",
304 &hdr[0..4]
305 )));
306 }
307 let blob_len = u64::from_le_bytes(hdr[8..16].try_into().unwrap()) as usize;
308 let blob_start = fts_abs + AILK_FTS_HEADER_SIZE;
309 let blob_end = blob_start + blob_len;
310 if blob_end > self.bytes.len() {
311 return Err(AilakeError::Fts("AILK_FTS blob out of bounds".into()));
312 }
313 Ok(Some(self.bytes.slice(blob_start..blob_end)))
314 }
315
316 pub fn verify_integrity(&self) -> AilakeResult<()> {
321 let header = self.read_header()?;
322 let index = self.load_any_index()?;
323 let reader = ParquetVectorReader::new(self.bytes.clone(), &self.vector_column);
324 let parquet_count = reader.record_count()?;
325
326 if parquet_count != index.node_count() {
327 return Err(AilakeError::RowCountMismatch {
328 parquet: parquet_count,
329 hnsw: index.node_count(),
330 });
331 }
332 if parquet_count != header.record_count {
333 return Err(AilakeError::RowCountMismatch {
334 parquet: parquet_count,
335 hnsw: header.record_count,
336 });
337 }
338 Ok(())
339 }
340}
341
342fn distance_metric_to_vector_metric(dm: DistanceMetric) -> VectorMetric {
343 match dm {
344 DistanceMetric::Cosine => VectorMetric::Cosine,
345 DistanceMetric::Euclidean => VectorMetric::Euclidean,
346 DistanceMetric::DotProduct => VectorMetric::DotProduct,
347 DistanceMetric::NormalizedCosine => VectorMetric::NormalizedCosine,
348 }
349}
350
351#[cfg(test)]
352mod tests {
353 use super::*;
354 use crate::writer::AilakeFileWriter;
355 use ailake_core::{VectorMetric, VectorPrecision, VectorStoragePolicy};
356 use arrow_array::{Int32Array, RecordBatch};
357 use arrow_schema::{DataType, Field, Schema};
358 use std::sync::Arc;
359
360 fn make_policy(dim: u32) -> VectorStoragePolicy {
361 VectorStoragePolicy {
362 column_name: "embedding".to_string(),
363 dim,
364 metric: VectorMetric::Cosine,
365 precision: VectorPrecision::F16,
366 pq: None,
367 keep_raw_for_reranking: true,
368 pre_normalize: false,
369 hnsw_m: None,
370 hnsw_ef_construction: None,
371 ivf_residual: false,
372 embedding_model: None,
373 modality: None,
374 partition_by: None,
375 partition_value: None,
376 partition_column_type: None,
377 partition_fields: vec![],
378 }
379 }
380
381 fn write_file(rows: usize, dim: u32) -> Bytes {
382 let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
383 let ids: Vec<i32> = (0..rows as i32).collect();
384 let batch = RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(ids))]).unwrap();
385 let embs: Vec<Vec<f32>> = (0..rows)
386 .map(|i| {
387 let mut v = vec![0.0f32; dim as usize];
388 v[i % dim as usize] = 1.0;
389 v
390 })
391 .collect();
392 AilakeFileWriter::new(make_policy(dim))
393 .write(&batch, &embs)
394 .unwrap()
395 }
396
397 #[test]
398 fn is_ailake_file() {
399 let file = write_file(3, 4);
400 let reader = AilakeFileReader::new(file, "embedding", 4);
401 assert!(reader.is_ailake_file());
402 }
403
404 #[test]
405 fn integrity_check_passes() {
406 let file = write_file(10, 8);
407 let reader = AilakeFileReader::new(file, "embedding", 8);
408 reader.verify_integrity().unwrap();
409 }
410
411 #[test]
412 fn centroid_has_correct_dim() {
413 let file = write_file(5, 4);
414 let reader = AilakeFileReader::new(file, "embedding", 4);
415 let centroid = reader.get_centroid().unwrap();
416 assert_eq!(centroid.values.len(), 4);
417 }
418
419 #[test]
420 fn search_finds_nearest() {
421 let dim = 4u32;
422 let file = write_file(4, dim);
423 let reader = AilakeFileReader::new(file, "embedding", dim);
424 let index = reader.load_index().unwrap();
425 let query = vec![1.0f32, 0.0, 0.0, 0.0];
426 let results = index.search(&query, 1, 50);
427 assert_eq!(results.len(), 1);
428 assert_eq!(results[0].0, ailake_core::RowId::new(0));
429 }
430
431 #[test]
432 fn parquet_read_returns_tabular_data() {
433 let file = write_file(3, 4);
434 let reader = AilakeFileReader::new(file, "embedding", 4);
435 let (batch, embs) = reader.read_parquet().unwrap();
436 assert_eq!(batch.num_rows(), 3);
437 assert_eq!(embs.len(), 3);
438 }
439
440 #[test]
447 fn corrupted_hnsw_offset_errors_instead_of_panicking() {
448 let file = write_file(3, 4);
449 let reader = AilakeFileReader::new(file.clone(), "embedding", 4);
450 let ailk_start = reader.ailk_offset().unwrap() as usize;
451
452 let mut corrupted = file.to_vec();
454 let field_start = ailk_start + 40;
455 corrupted[field_start..field_start + 8].copy_from_slice(&u64::MAX.to_le_bytes());
456 let corrupted = Bytes::from(corrupted);
457
458 let r1 = AilakeFileReader::new(corrupted.clone(), "embedding", 4);
459 assert!(r1.load_index_for_column("embedding").is_err());
460 let r2 = AilakeFileReader::new(corrupted, "embedding", 4);
461 assert!(r2.load_any_index_for_column("embedding").is_err());
462 }
463
464 #[test]
471 fn has_column_footer_does_not_false_positive_on_primary_fallback() {
472 let file = write_file(3, 4);
473 let reader = AilakeFileReader::new(file, "embedding", 4);
474
475 assert!(!reader.has_column_footer("embedding_v2"));
479 assert!(
480 reader.ailk_offset_for_column("embedding_v2").is_ok(),
481 "sanity: ailk_offset_for_column's primary-fallback behavior is what \
482 has_column_footer exists to avoid — if this assert ever fails, the \
483 fallback was removed and has_column_footer may be redundant"
484 );
485 }
486
487 #[test]
491 fn has_column_footer_detects_genuine_extra_column() {
492 use crate::writer::VectorColumnBatch;
493
494 let dim = 4u32;
495 let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
496 let batch =
497 RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![0i32, 1, 2]))])
498 .unwrap();
499 let primary_embs: Vec<Vec<f32>> = vec![vec![1.0, 0.0, 0.0, 0.0]; 3];
500 let extra_embs: Vec<Vec<f32>> = vec![vec![0.0, 1.0, 0.0, 0.0]; 3];
501 let extra_policy = make_policy(dim);
502 let mut extra_policy = extra_policy.clone();
503 extra_policy.column_name = "embedding_v2".to_string();
504
505 let primary_policy = make_policy(dim);
506 let file_bytes = AilakeFileWriter::new(primary_policy.clone())
507 .write_multi(
508 &batch,
509 &[
510 VectorColumnBatch {
511 policy: &primary_policy,
512 embeddings: &primary_embs,
513 },
514 VectorColumnBatch {
515 policy: &extra_policy,
516 embeddings: &extra_embs,
517 },
518 ],
519 )
520 .unwrap();
521
522 let reader = AilakeFileReader::new(file_bytes, "embedding", dim);
523 assert!(reader.has_column_footer("embedding_v2"));
524 assert!(!reader.has_column_footer("embedding_v3"));
526 }
527
528 #[test]
529 fn corrupt_ailk_magic_returns_err_no_panic() {
530 let file = write_file(3, 4);
531 let ailk_start = {
532 let reader = AilakeFileReader::new(file.clone(), "embedding", 4);
533 reader.ailk_offset().unwrap() as usize
534 };
535 let mut corrupted = file.to_vec();
536 corrupted[ailk_start..ailk_start + 4].copy_from_slice(b"BADC");
537 let corrupted = Bytes::from(corrupted);
538 let reader = AilakeFileReader::new(corrupted, "embedding", 4);
539 assert!(reader.load_index_for_column("embedding").is_err());
540 }
541
542 #[test]
543 fn corrupt_hnsw_bincode_returns_err_no_panic() {
544 let file = write_file(3, 4);
545 let ailk_start = {
546 let reader = AilakeFileReader::new(file.clone(), "embedding", 4);
547 reader.ailk_offset().unwrap() as usize
548 };
549 let mut corrupted = file.to_vec();
552 let header_bytes = &corrupted[ailk_start..];
555 let hnsw_offset = u64::from_le_bytes(header_bytes[48..56].try_into().unwrap()) as usize;
556 let hnsw_len = u64::from_le_bytes(header_bytes[40..48].try_into().unwrap()) as usize;
557 let blob_start = ailk_start + hnsw_offset;
558 let garbage = vec![0xABu8; hnsw_len];
559 corrupted[blob_start..blob_start + hnsw_len].copy_from_slice(&garbage);
560 let corrupted = Bytes::from(corrupted);
561
562 let reader = AilakeFileReader::new(corrupted, "embedding", 4);
563 let result = reader.load_index_for_column("embedding");
564 assert!(
565 result.is_err(),
566 "corrupted HNSW bincode must return Err, got Ok"
567 );
568 }
569
570 #[test]
571 fn truncated_parquet_fails_gracefully() {
572 let file = write_file(3, 4);
573 let half = file.len() / 2;
574 let truncated = file.slice(..half);
575 let reader = AilakeFileReader::new(truncated, "embedding", 4);
576 assert!(!reader.is_ailake_file());
577 let result = reader.load_index_for_column("embedding");
578 assert!(result.is_err());
579 let result = reader.load_any_index_for_column("embedding");
580 assert!(result.is_err());
581 }
582
583 #[test]
584 fn empty_file_fails_gracefully() {
585 let empty = Bytes::new();
586 let reader = AilakeFileReader::new(empty, "embedding", 4);
587 let result = reader.load_index_for_column("embedding");
588 assert!(result.is_err());
589 }
590
591 #[test]
592 fn missing_parquet_footer_fails_gracefully() {
593 let file = write_file(3, 4);
594 let parquet_footer = "PAR1";
595 if let Some(pos) = file
596 .windows(4)
597 .rposition(|w| w == parquet_footer.as_bytes())
598 {
599 let truncated = file.slice(..pos);
600 let reader = AilakeFileReader::new(truncated, "embedding", 4);
601 assert!(!reader.is_ailake_file());
602 let result = reader.load_index_for_column("embedding");
603 assert!(result.is_err());
604 }
605 }
606
607 #[test]
608 fn zero_vector_query_returns_sane_distances() {
609 let dim = 4u32;
610 let file = write_file(3, dim);
611 let reader = AilakeFileReader::new(file, "embedding", dim);
612 let query = vec![0.0f32; dim as usize];
613 let index = reader.load_index().unwrap();
614 let results = index.search(&query, 3, 50);
615 for (_, d) in &results {
616 assert!(d.is_finite(), "distance must be finite");
617 assert!(*d <= 2.0, "cosine distance max is 2.0");
618 }
619 }
620}