1mod cover;
26mod fastcover;
27mod frequency;
28mod reservoir;
29
30use crate::bit_io::BitWriter;
31use crate::blocks::sequence_section::{
32 MAX_LITERAL_LENGTH_CODE, MAX_MATCH_LENGTH_CODE, MAX_OFFSET_CODE,
33};
34use crate::decoding::dictionary::MAGIC_NUM as DICT_MAGIC_NUM;
35use crate::decoding::sequence_section_decoder::{LL_MAX_LOG, ML_MAX_LOG, OF_MAX_LOG};
36use crate::dictionary::reservoir::create_sample;
37use crate::fse::fse_encoder::{self, build_table_from_symbol_counts};
38use crate::huff0::HuffmanTable as HuffmanDecoderTable;
39use crate::huff0::huff0_encoder::{HuffmanEncoder, HuffmanTable as HuffmanEncoderTable};
40use core::cmp::Reverse;
41use cover::*;
42pub use fastcover::{
43 DEFAULT_D_CANDIDATES, DEFAULT_F_CANDIDATES, DEFAULT_K_CANDIDATES, FastCoverParams,
44 FastCoverTuned,
45};
46use std::{
47 boxed::Box,
48 collections::{BinaryHeap, HashMap},
49 format,
50 fs::{self, File},
51 io::{self, Read},
52 path::{Path, PathBuf},
53 vec,
58 vec::Vec,
59};
60
61const MAX_TRAINING_PREALLOC_BYTES: usize = 8 * 1024 * 1024;
62const MAX_HUFFMAN_STATS_BYTES: usize = 64 * 1024;
63
64pub const MIN_TRAINED_DICT_SIZE: usize = DICT_MAGIC_NUM.len() + 4 + 12 + 8;
72
73#[derive(Debug, Clone)]
75pub struct FastCoverOptions {
76 pub optimize: bool,
77 pub split_point: f64,
78 pub accel: usize,
79 pub k: usize,
80 pub d: usize,
81 pub f: u32,
84 pub k_candidates: Vec<usize>,
85 pub d_candidates: Vec<usize>,
86 pub f_candidates: Vec<u32>,
87}
88
89impl Default for FastCoverOptions {
90 fn default() -> Self {
91 Self {
92 optimize: true,
93 split_point: 0.75,
94 accel: 1,
95 k: 256,
96 d: 8,
97 f: 20,
98 k_candidates: DEFAULT_K_CANDIDATES.to_vec(),
99 d_candidates: DEFAULT_D_CANDIDATES.to_vec(),
100 f_candidates: DEFAULT_F_CANDIDATES.to_vec(),
101 }
102 }
103}
104
105#[derive(Debug, Clone, Copy, Default)]
106pub struct FinalizeOptions {
107 pub dict_id: Option<u32>,
108}
109
110pub(super) struct DictParams {
115 pub segment_size: u32,
125}
126
127pub fn create_raw_dict_from_dir<P: AsRef<Path>, W: io::Write>(
147 path: P,
148 output: &mut W,
149 dict_size: usize,
150) -> Result<(), io::Error> {
151 let mut file_paths: Vec<PathBuf> = Vec::new();
153 let dir: fs::ReadDir = fs::read_dir(path)?;
154 fn recurse_read(dir: fs::ReadDir, file_paths: &mut Vec<PathBuf>) -> Result<(), io::Error> {
155 for entry in dir {
156 let entry = entry?;
157 if entry.file_type()?.is_dir() {
158 recurse_read(fs::read_dir(entry.path())?, file_paths)?;
159 } else {
160 file_paths.push(entry.path());
161 }
162 }
163 Ok(())
164 }
165 recurse_read(dir, &mut file_paths)?;
166
167 let mut total_file_len: u64 = 0;
169 let mut file_handles: Vec<fs::File> = Vec::new();
170 for path in file_paths {
171 let handle = File::open(path)?;
172 total_file_len += handle.metadata()?.len();
173 file_handles.push(handle);
174 }
175 let empty_reader: Box<dyn Read> = Box::new(io::empty());
176 let chained_files = file_handles
177 .iter()
178 .fold(empty_reader, |acc, reader| Box::new(acc.chain(reader)));
179
180 create_raw_dict_from_source(chained_files, total_file_len as usize, output, dict_size)?;
182 Ok(())
183}
184
185pub fn create_raw_dict_from_source<R: io::Read, W: io::Write>(
205 mut source: R,
206 source_size: usize,
207 output: &mut W,
208 dict_size: usize,
209) -> io::Result<()> {
210 if dict_size == 0 {
211 return Ok(());
212 }
213 let prealloc = source_size.min(MAX_TRAINING_PREALLOC_BYTES);
214 let mut all = Vec::with_capacity(prealloc);
215 source.read_to_end(&mut all)?;
216 create_raw_dict_from_slice(&all, output, dict_size)
217}
218
219pub fn create_raw_dict_from_slice<W: io::Write>(
241 all: &[u8],
242 output: &mut W,
243 dict_size: usize,
244) -> io::Result<()> {
245 if dict_size == 0 || all.is_empty() {
246 return Ok(());
247 }
248
249 if all.len() < K {
250 let keep = usize::min(all.len(), dict_size);
251 output.write_all(&all[all.len() - keep..])?;
252 return Ok(());
253 }
254
255 let source_size = all.len();
256 vprintln!("create_dict: creating {dict_size} byte dict from {source_size} byte source");
257
258 let params = DictParams { segment_size: 2048 };
259 let num_segments = usize::max(1, source_size / params.segment_size as usize);
260 let denom = usize::max(1, source_size / (2 * num_segments));
264 let sample_scale = usize::max(1, usize::min(denom, 256));
265 let mut sample_size = source_size / sample_scale;
266 sample_size = usize::max(sample_size, usize::min(source_size, 16));
267 vprintln!("create_dict: creating {sample_size} byte sample of collection");
268 let mut sample_reader = all;
269 let collection_sample = create_sample(&mut sample_reader, sample_size);
270
271 let mut pool: BinaryHeap<Reverse<Segment>> = BinaryHeap::new();
277 let (num_epochs, epoch_size_kmers) = compute_epoch_info(¶ms, dict_size, source_size / K);
278 let epoch_size = usize::max(K, epoch_size_kmers * K);
283 vprintln!("create_dict: computed epoch info, using {num_epochs} epochs of {epoch_size} bytes");
284 let mut epoch_counter = 0;
285 let mut ctx = Context {
286 frequencies: HashMap::with_capacity(epoch_size / K),
287 };
288 for epoch_idx in 0..num_epochs {
292 let start = epoch_idx * epoch_size;
293 if start >= all.len() {
294 break;
295 }
296 let end = if epoch_idx + 1 == num_epochs {
297 all.len()
298 } else {
299 usize::min(start + epoch_size, all.len())
300 };
301 let epoch = &all[start..end];
302 epoch_counter += 1;
303 let best_segment = pick_best_segment(¶ms, &mut ctx, epoch, &collection_sample);
304 vprintln!(
305 "\tcreate_dict: epoch {epoch_counter}/{num_epochs} has best segment score {}",
306 best_segment.score
307 );
308 pool.push(Reverse(best_segment));
309 ctx.frequencies.clear();
311 }
312 vprintln!(
313 "create_dict: {epoch_counter} epochs written, writing {} segments",
314 pool.len()
315 );
316 while let Some(segment) = pool.pop() {
319 output.write_all(&segment.0.raw)?;
320 }
321 Ok(())
322}
323
324fn strided_index(i: usize, len: usize) -> usize {
332 ((i as u64 * len as u64) / MAX_HUFFMAN_STATS_BYTES as u64) as usize
333}
334
335fn serialize_huffman_table(sample_data: &[u8], raw_content: &[u8]) -> io::Result<Vec<u8>> {
336 fn bounded_huffman_stats(data: &[u8]) -> Vec<u8> {
337 if data.len() <= MAX_HUFFMAN_STATS_BYTES {
338 return data.to_vec();
339 }
340
341 let mut stats = Vec::with_capacity(MAX_HUFFMAN_STATS_BYTES);
342 for i in 0..MAX_HUFFMAN_STATS_BYTES {
343 stats.push(data[strided_index(i, data.len())]);
344 }
345 stats
346 }
347
348 let source = if sample_data.len() >= 2 {
349 sample_data
350 } else {
351 raw_content
352 };
353 let mut stats = bounded_huffman_stats(source);
354 if stats.len() < 2 || stats.iter().all(|b| *b == stats[0]) {
355 stats = (0u8..128).collect();
361 }
362
363 let mut table = HuffmanEncoderTable::build_from_data(stats.as_slice());
364 if table.writeable_table_description_size().is_none() {
365 stats = (0u8..128).collect();
369 table = HuffmanEncoderTable::build_from_data(stats.as_slice());
370 }
371 let mut writer = BitWriter::new();
372 let mut encoder = HuffmanEncoder::new(&table, &mut writer);
373 encoder.encode(&[stats[0]], true);
374 let encoded = writer.dump();
375
376 let mut decoder = HuffmanDecoderTable::new();
377 let table_size = decoder
378 .build_decoder(encoded.as_slice())
379 .map_err(|e| io::Error::other(format!("failed to decode generated huffman table: {e}")))?;
380 Ok(encoded[..table_size as usize].to_vec())
381}
382
383fn serialize_fse_table(table: &fse_encoder::FSETable) -> Vec<u8> {
384 let mut writer = BitWriter::new();
385 table.write_table(&mut writer);
386 writer.dump()
387}
388
389fn bounded_fse_symbols(data: &[u8], max_symbol: u8) -> Vec<u8> {
390 let modulo = u16::from(max_symbol) + 1;
391 if data.is_empty() {
392 return Vec::from([0u8]);
393 }
394 if data.len() <= MAX_HUFFMAN_STATS_BYTES {
395 return data
396 .iter()
397 .map(|b| (u16::from(*b) % modulo) as u8)
398 .collect();
399 }
400
401 let mut out = Vec::with_capacity(MAX_HUFFMAN_STATS_BYTES);
402 for i in 0..MAX_HUFFMAN_STATS_BYTES {
403 let idx = strided_index(i, data.len());
404 out.push((u16::from(data[idx]) % modulo) as u8);
405 }
406 out
407}
408
409fn serialize_fse_table_from_corpus(
410 sample_data: &[u8],
411 raw_content: &[u8],
412 max_symbol: u8,
413 max_log: u8,
414) -> io::Result<Vec<u8>> {
415 fn counts_total_for_source(source: &[u8], max_symbol: u8, counts: &mut [usize]) -> usize {
416 counts.fill(0);
417 for symbol in bounded_fse_symbols(source, max_symbol) {
418 counts[usize::from(symbol)] += 1;
419 }
420 counts.iter().sum::<usize>()
421 }
422
423 let mut counts = vec![0usize; usize::from(max_symbol) + 1];
424 let using_sample = !sample_data.is_empty();
425 let mut total = counts_total_for_source(
426 if using_sample {
427 sample_data
428 } else {
429 raw_content
430 },
431 max_symbol,
432 &mut counts,
433 );
434 if total <= 1 && using_sample && !raw_content.is_empty() {
435 total = counts_total_for_source(raw_content, max_symbol, &mut counts);
436 }
437 if total <= 1 {
438 return Err(io::Error::new(
439 io::ErrorKind::InvalidInput,
440 "insufficient symbol statistics for FSE table",
441 ));
442 }
443 let table = build_table_from_symbol_counts(&counts, max_log, false);
444 Ok(serialize_fse_table(&table))
445}
446
447fn finalized_content_budget(
448 sample_data: &[u8],
449 raw_fallback: &[u8],
450 dict_size: usize,
451) -> io::Result<usize> {
452 let min_content_size = 8usize;
453 let huf_len = serialize_huffman_table(sample_data, raw_fallback)?.len();
454 let of_len =
455 serialize_fse_table_from_corpus(sample_data, raw_fallback, MAX_OFFSET_CODE, OF_MAX_LOG)?
456 .len();
457 let ml_len = serialize_fse_table_from_corpus(
458 sample_data,
459 raw_fallback,
460 MAX_MATCH_LENGTH_CODE,
461 ML_MAX_LOG,
462 )?
463 .len();
464 let ll_len = serialize_fse_table_from_corpus(
465 sample_data,
466 raw_fallback,
467 MAX_LITERAL_LENGTH_CODE,
468 LL_MAX_LOG,
469 )?
470 .len();
471
472 let header_len = DICT_MAGIC_NUM.len() + 4 + huf_len + of_len + ml_len + ll_len + 12;
473 let max_content_budget = dict_size.saturating_sub(header_len);
474 if max_content_budget < min_content_size {
475 return Err(io::Error::new(
476 io::ErrorKind::InvalidInput,
477 "dictionary size too small to fit header and offset history",
478 ));
479 }
480 Ok(max_content_budget)
481}
482
483fn derive_dict_id(raw_content: &[u8]) -> u32 {
484 let mut h = 0xcbf29ce484222325u64;
485 for &b in raw_content {
486 h ^= u64::from(b);
487 h = h.wrapping_mul(0x100000001b3);
488 }
489 let compliant = (h % ((1u64 << 31) - 32768)) + 32768;
490 compliant as u32
491}
492
493pub fn finalize_raw_dict(
496 raw_content: &[u8],
497 sample_data: &[u8],
498 dict_size: usize,
499 options: FinalizeOptions,
500) -> io::Result<Vec<u8>> {
501 if raw_content.is_empty() {
502 return Err(io::Error::new(
503 io::ErrorKind::InvalidInput,
504 "raw dictionary content must not be empty",
505 ));
506 }
507 let mut out = Vec::with_capacity(dict_size.max(256));
508 out.extend_from_slice(&DICT_MAGIC_NUM);
509 let dict_id = options
510 .dict_id
511 .unwrap_or_else(|| derive_dict_id(raw_content));
512 if dict_id == 0 {
513 return Err(io::Error::new(
514 io::ErrorKind::InvalidInput,
515 "dictionary id must be non-zero",
516 ));
517 }
518 out.extend_from_slice(&dict_id.to_le_bytes());
519 out.extend_from_slice(serialize_huffman_table(sample_data, raw_content)?.as_slice());
520 out.extend_from_slice(
521 serialize_fse_table_from_corpus(sample_data, raw_content, MAX_OFFSET_CODE, OF_MAX_LOG)?
522 .as_slice(),
523 );
524 out.extend_from_slice(
525 serialize_fse_table_from_corpus(
526 sample_data,
527 raw_content,
528 MAX_MATCH_LENGTH_CODE,
529 ML_MAX_LOG,
530 )?
531 .as_slice(),
532 );
533 out.extend_from_slice(
534 serialize_fse_table_from_corpus(
535 sample_data,
536 raw_content,
537 MAX_LITERAL_LENGTH_CODE,
538 LL_MAX_LOG,
539 )?
540 .as_slice(),
541 );
542
543 out.extend_from_slice(&1u32.to_le_bytes());
545 out.extend_from_slice(&4u32.to_le_bytes());
546 out.extend_from_slice(&8u32.to_le_bytes());
547
548 let min_content_size = 8usize;
549 let max_content_budget = dict_size.saturating_sub(out.len());
550 if max_content_budget < min_content_size {
551 return Err(io::Error::new(
552 io::ErrorKind::InvalidInput,
553 "dictionary size too small to fit header and offset history",
554 ));
555 }
556
557 let content = if raw_content.len() > max_content_budget {
558 &raw_content[raw_content.len() - max_content_budget..]
559 } else {
560 raw_content
561 };
562 if content.len() < min_content_size {
563 out.resize(out.len() + (min_content_size - content.len()), 0);
564 }
565 out.extend_from_slice(content);
566 Ok(out)
567}
568
569fn train_fastcover_internal(
572 sample: &[u8],
573 dict_size: usize,
574 options: &FastCoverOptions,
575) -> io::Result<(Vec<u8>, FastCoverTuned)> {
576 let trained = if options.optimize {
577 fastcover::optimize_fastcover_raw(
578 sample,
579 dict_size,
580 options.split_point,
581 options.accel,
582 options.d_candidates.as_slice(),
583 options.f_candidates.as_slice(),
584 options.k_candidates.as_slice(),
585 )
586 } else {
587 let params = fastcover::normalize_fastcover_params(FastCoverParams {
588 k: options.k,
589 d: options.d,
590 f: options.f,
591 accel: options.accel,
592 });
593 fastcover::train_fastcover_raw(sample, dict_size, params).map(|dict| {
594 (
595 dict,
596 FastCoverTuned {
597 k: params.k,
598 d: params.d,
599 f: params.f,
600 accel: params.accel,
601 score: 0,
602 },
603 )
604 })
605 };
606 trained.map_err(io::Error::from)
607}
608
609impl From<fastcover::TableTooLarge> for io::Error {
610 fn from(table: fastcover::TableTooLarge) -> Self {
611 io::Error::new(
612 io::ErrorKind::OutOfMemory,
613 format!(
614 "a FastCOVER table of {} entries does not fit in memory; use a smaller f",
615 table.entries
616 ),
617 )
618 }
619}
620
621pub fn train_fastcover_raw_from_slice(
623 sample: &[u8],
624 dict_size: usize,
625 options: &FastCoverOptions,
626) -> io::Result<(Vec<u8>, FastCoverTuned)> {
627 if sample.is_empty() {
628 return Err(io::Error::new(
629 io::ErrorKind::InvalidInput,
630 "source stream is empty",
631 ));
632 }
633 let (dict, tuned) = train_fastcover_internal(sample, dict_size, options)?;
634 if dict.is_empty() && dict_size > 0 {
635 return Err(io::Error::new(
636 io::ErrorKind::InvalidInput,
637 "training sample is too small for FastCOVER",
638 ));
639 }
640 Ok((dict, tuned))
641}
642
643pub fn create_fastcover_raw_dict_from_source<R: io::Read, W: io::Write>(
648 mut source: R,
649 output: &mut W,
650 dict_size: usize,
651 options: &FastCoverOptions,
652) -> io::Result<FastCoverTuned> {
653 let mut sample = Vec::new();
654 source.read_to_end(&mut sample)?;
655 let (dict, tuned) = train_fastcover_raw_from_slice(sample.as_slice(), dict_size, options)?;
656 output.write_all(dict.as_slice())?;
657 Ok(tuned)
658}
659
660pub fn create_fastcover_dict_from_source<R: io::Read, W: io::Write>(
665 mut source: R,
666 output: &mut W,
667 dict_size: usize,
668 fastcover: &FastCoverOptions,
669 finalize: FinalizeOptions,
670) -> io::Result<FastCoverTuned> {
671 let mut sample = Vec::new();
672 source.read_to_end(&mut sample)?;
673 create_fastcover_dict_from_slice(sample.as_slice(), output, dict_size, fastcover, finalize)
674}
675
676pub fn create_fastcover_dict_from_slice<W: io::Write>(
683 sample: &[u8],
684 output: &mut W,
685 dict_size: usize,
686 fastcover: &FastCoverOptions,
687 finalize: FinalizeOptions,
688) -> io::Result<FastCoverTuned> {
689 if sample.is_empty() {
690 return Err(io::Error::new(
691 io::ErrorKind::InvalidInput,
692 "source stream is empty",
693 ));
694 }
695 let content_budget = finalized_content_budget(sample, sample, dict_size)?;
696 let (raw_dict, tuned) = train_fastcover_raw_from_slice(sample, content_budget, fastcover)?;
697
698 let finalized = finalize_raw_dict(raw_dict.as_slice(), sample, dict_size, finalize)?;
699 output.write_all(finalized.as_slice())?;
700 Ok(tuned)
701}
702
703#[cfg(feature = "bench-internals")]
711pub(crate) fn dict_roundtrip_fixture() -> (
712 alloc::vec::Vec<u8>,
713 alloc::vec::Vec<u8>,
714 alloc::vec::Vec<u8>,
715) {
716 use crate::decoding::Dictionary;
717 use crate::encoding::{CompressionLevel, FrameCompressor};
718
719 let mut sample = alloc::vec::Vec::new();
720 for i in 0..512u32 {
721 sample.extend_from_slice(
722 alloc::format!(
723 "tenant=demo table=orders key={i} region=eu payload=aaaaabbbbbcccccdddddeeeee\n"
724 )
725 .as_bytes(),
726 );
727 }
728
729 let dict_size = 4096usize;
730 let content_budget = finalized_content_budget(sample.as_slice(), sample.as_slice(), dict_size)
731 .expect("content budget should be computable");
732 let raw = fastcover::train_fastcover_raw(
733 sample.as_slice(),
734 content_budget,
735 fastcover::FastCoverParams {
736 k: 256,
737 d: 8,
738 f: 20,
739 accel: 1,
740 },
741 )
742 .expect("a 2^20 table fits");
743 let finalized = finalize_raw_dict(
744 raw.as_slice(),
745 sample.as_slice(),
746 dict_size,
747 FinalizeOptions::default(),
748 )
749 .expect("finalization should succeed");
750 let parsed =
751 Dictionary::decode_dict(finalized.as_slice()).expect("finalized dictionary should parse");
752 assert!(!parsed.dict_content.is_empty());
753
754 let mut payload = alloc::vec::Vec::new();
755 for idx in 0..96u32 {
756 payload.extend_from_slice(
757 alloc::format!("tenant=demo op=put key={idx} value=aaaaabbbbbcccccdddddeeeee\n")
758 .as_bytes(),
759 );
760 }
761
762 let mut compressed = alloc::vec::Vec::new();
763 let mut compressor = FrameCompressor::new(CompressionLevel::Fastest);
764 compressor
765 .set_dictionary(parsed)
766 .expect("dictionary should attach");
767 compressor.set_source(payload.as_slice());
768 compressor.set_drain(&mut compressed);
769 compressor.compress();
770
771 (finalized, compressed, payload)
772}
773
774#[cfg(test)]
775mod tests;