1use rudb_common::{Error, Result};
49
50use crate::fsst::SymbolTable;
51use crate::integer;
52use crate::reader::Reader;
53use crate::sketch::Sketch;
54use crate::string;
55
56const FLOOR_SAMPLE_BYTES: usize = 2 * 1024;
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum Strategy {
63 Independent = 0,
65 SharedTable = 1,
67 SharedDict = 2,
69}
70
71impl Strategy {
72 fn tag(self) -> u8 {
73 self as u8
74 }
75
76 fn from_tag(tag: u8) -> Result<Self> {
77 match tag {
78 0 => Ok(Self::Independent),
79 1 => Ok(Self::SharedTable),
80 2 => Ok(Self::SharedDict),
81 other => Err(Error::internal(format!("unknown column group tag {other}"))),
82 }
83 }
84
85 #[must_use]
87 pub fn name(self) -> &'static str {
88 match self {
89 Self::Independent => "INDEPENDENT",
90 Self::SharedTable => "SHARED_TABLE",
91 Self::SharedDict => "SHARED_DICT",
92 }
93 }
94}
95
96pub fn encode_group(columns: &[&[&[u8]]]) -> Result<Vec<u8>> {
107 let mut best: Option<Vec<u8>> = None;
108 for strategy in [Strategy::Independent, Strategy::SharedTable, Strategy::SharedDict] {
109 let Some(bytes) = encode_as(strategy, columns)? else {
110 continue;
111 };
112 if best.as_ref().is_none_or(|current| bytes.len() < current.len()) {
113 best = Some(bytes);
114 }
115 }
116 best.ok_or_else(|| Error::internal("no strategy applied to the column group"))
117}
118
119pub fn decode_group(bytes: &[u8]) -> Result<Vec<Vec<Vec<u8>>>> {
125 let mut reader = Reader::new(bytes);
126 let columns = decode_at(&mut reader)?;
127 if reader.remaining() != 0 {
128 return Err(Error::internal(format!(
129 "{} bytes left over after decoding a column group",
130 reader.remaining()
131 )));
132 }
133 Ok(columns)
134}
135
136pub fn strategy_sizes(columns: &[&[&[u8]]]) -> Result<Vec<(Strategy, usize)>> {
142 let mut sizes = Vec::new();
143 for strategy in [Strategy::Independent, Strategy::SharedTable, Strategy::SharedDict] {
144 if let Some(bytes) = encode_as(strategy, columns)? {
145 sizes.push((strategy, bytes.len()));
146 }
147 }
148 Ok(sizes)
149}
150
151pub fn describe(bytes: &[u8]) -> Result<String> {
157 let mut reader = Reader::new(bytes);
158 describe_at(&mut reader)
159}
160
161pub fn dictionary_groups(sketches: &[Sketch], threshold: f64) -> Result<Vec<Vec<usize>>> {
183 let mut parent: Vec<usize> = (0..sketches.len()).collect();
184 for left in 0..sketches.len() {
185 for right in (left + 1)..sketches.len() {
186 if sketches[left].jaccard(&sketches[right])? >= threshold {
187 let (a, b) = (find(&mut parent, left), find(&mut parent, right));
188 if a != b {
189 let (low, high) = if a < b { (a, b) } else { (b, a) };
192 parent[high] = low;
193 }
194 }
195 }
196 }
197 let mut groups: Vec<Vec<usize>> = Vec::new();
198 let mut roots: Vec<usize> = Vec::new();
199 for column in 0..sketches.len() {
200 let root = find(&mut parent, column);
201 match roots.iter().position(|seen| *seen == root) {
202 Some(at) => groups[at].push(column),
203 None => {
204 roots.push(root);
205 groups.push(vec![column]);
206 }
207 }
208 }
209 Ok(groups)
210}
211
212fn find(parent: &mut [usize], mut node: usize) -> usize {
213 while parent[node] != node {
214 parent[node] = parent[parent[node]];
215 node = parent[node];
216 }
217 node
218}
219
220fn encode_as(strategy: Strategy, columns: &[&[&[u8]]]) -> Result<Option<Vec<u8>>> {
221 let mut out = vec![strategy.tag()];
222 put_u32(&mut out, u32::try_from(columns.len()).map_err(|_| too_many(columns.len()))?);
223 match strategy {
224 Strategy::Independent => {
225 for column in columns {
226 out.extend_from_slice(&string::encode(column)?);
227 }
228 }
229 Strategy::SharedTable => {
230 if columns.len() < 2 {
231 return Ok(None);
232 }
233 let table = SymbolTable::train(&shared_sample(columns));
234 if table.is_empty() {
235 return Ok(None);
236 }
237 table.serialize(&mut out);
238 for column in columns {
239 put_u32(&mut out, u32::try_from(column.len()).map_err(|_| too_many(column.len()))?);
240 let mut compressed = Vec::new();
241 let mut lengths = Vec::with_capacity(column.len());
242 for value in *column {
243 let before = compressed.len();
244 table.compress(value, &mut compressed);
245 lengths.push((compressed.len() - before) as i64);
246 }
247 out.extend_from_slice(&integer::encode(&lengths)?);
248 out.extend_from_slice(&compressed);
249 }
250 }
251 Strategy::SharedDict => {
252 if columns.len() < 2 {
253 return Ok(None);
254 }
255 let dictionary = union_values(columns);
256 let total: usize = columns.iter().map(|column| column.len()).sum();
257 if dictionary.is_empty() || dictionary.len() >= total {
260 return Ok(None);
261 }
262 let entries: Vec<&[u8]> = dictionary.iter().map(Vec::as_slice).collect();
263 out.extend_from_slice(&string::encode(&entries)?);
264 for column in columns {
265 let codes = codes_over(column, &dictionary);
266 out.extend_from_slice(&integer::encode(&codes)?);
267 }
268 }
269 }
270 Ok(Some(out))
271}
272
273fn decode_at(reader: &mut Reader<'_>) -> Result<Vec<Vec<Vec<u8>>>> {
274 let strategy = Strategy::from_tag(reader.u8()?)?;
275 let count = reader.u32()? as usize;
276 let mut columns = Vec::with_capacity(count.min(1024));
277 match strategy {
278 Strategy::Independent => {
279 for _ in 0..count {
280 let (values, used) = string::decode_prefix(reader.rest())?;
281 reader.skip(used)?;
282 columns.push(values);
283 }
284 }
285 Strategy::SharedTable => {
286 let (table, used) = SymbolTable::deserialize(reader.rest())?;
287 reader.skip(used)?;
288 for _ in 0..count {
289 let rows = reader.u32()? as usize;
290 let (lengths, used) = integer::decode_prefix(reader.rest())?;
291 reader.skip(used)?;
292 if lengths.len() != rows {
293 return Err(Error::internal(format!(
294 "a column says it holds {rows} values and has {} lengths",
295 lengths.len()
296 )));
297 }
298 let mut values = Vec::with_capacity(rows);
299 for length in lengths {
300 let length = usize::try_from(length)
301 .map_err(|_| Error::internal("a negative compressed length"))?;
302 let compressed = reader.bytes(length)?;
303 let mut value = Vec::new();
304 table.decompress(compressed, &mut value)?;
305 values.push(value);
306 }
307 columns.push(values);
308 }
309 }
310 Strategy::SharedDict => {
311 let (dictionary, used) = string::decode_prefix(reader.rest())?;
312 reader.skip(used)?;
313 for _ in 0..count {
314 let (codes, used) = integer::decode_prefix(reader.rest())?;
315 reader.skip(used)?;
316 let mut values = Vec::with_capacity(codes.len());
317 for code in codes {
318 let entry = usize::try_from(code)
319 .ok()
320 .and_then(|index| dictionary.get(index))
321 .ok_or_else(|| {
322 Error::internal(format!("code {code} is not in the shared dictionary"))
323 })?;
324 values.push(entry.clone());
325 }
326 columns.push(values);
327 }
328 }
329 }
330 Ok(columns)
331}
332
333fn describe_at(reader: &mut Reader<'_>) -> Result<String> {
334 let strategy = Strategy::from_tag(reader.u8()?)?;
335 let count = reader.u32()? as usize;
336 let mut parts = Vec::with_capacity(count.min(1024));
337 let head = match strategy {
338 Strategy::Independent => {
339 for _ in 0..count {
340 let (text, used) = string::describe_prefix(reader.rest())?;
341 reader.skip(used)?;
342 parts.push(text);
343 }
344 "INDEPENDENT".to_string()
345 }
346 Strategy::SharedTable => {
347 let (table, used) = SymbolTable::deserialize(reader.rest())?;
348 reader.skip(used)?;
349 for _ in 0..count {
350 let rows = reader.u32()? as usize;
351 let (text, _) = integer::describe_prefix(reader.rest())?;
354 let (lengths, used) = integer::decode_prefix(reader.rest())?;
355 reader.skip(used)?;
356 if lengths.len() != rows {
357 return Err(Error::internal("a column group disagrees with itself"));
358 }
359 let bytes: i64 = lengths.iter().sum();
360 reader.skip(usize::try_from(bytes).map_err(|_| {
361 Error::internal("a column group has a negative compressed size")
362 })?)?;
363 parts.push(text);
364 }
365 format!("SHARED_TABLE[{}]", table.len())
366 }
367 Strategy::SharedDict => {
368 let (text, used) = string::describe_prefix(reader.rest())?;
369 reader.skip(used)?;
370 for _ in 0..count {
371 let (codes, used) = integer::describe_prefix(reader.rest())?;
372 reader.skip(used)?;
373 parts.push(codes);
374 }
375 format!("SHARED_DICT({text})")
376 }
377 };
378 Ok(format!("{head}({})", parts.join(", ")))
379}
380
381fn shared_sample<'a>(columns: &[&[&'a [u8]]]) -> Vec<&'a [u8]> {
396 let sizes: Vec<usize> =
397 columns.iter().map(|column| column.iter().map(|value| value.len()).sum()).collect();
398 let total: usize = sizes.iter().sum();
399 let floor = FLOOR_SAMPLE_BYTES;
400 let mut sample = Vec::new();
401 for (column, bytes) in columns.iter().zip(&sizes) {
402 let share = if total == 0 {
403 floor
404 } else {
405 (string::SAMPLE_BYTES as u128 * *bytes as u128 / total as u128) as usize
406 };
407 sample.extend(string::sample_bytes_of(column, share.max(floor)));
408 }
409 sample
410}
411
412fn union_values(columns: &[&[&[u8]]]) -> Vec<Vec<u8>> {
414 let mut values: Vec<Vec<u8>> =
415 columns.iter().flat_map(|column| column.iter().map(|value| value.to_vec())).collect();
416 values.sort_unstable();
417 values.dedup();
418 values
419}
420
421fn codes_over(values: &[&[u8]], dictionary: &[Vec<u8>]) -> Vec<i64> {
422 values
423 .iter()
424 .map(|value| {
425 dictionary
426 .binary_search_by(|entry| entry.as_slice().cmp(value))
427 .expect("the dictionary is the union of the columns in this group")
428 as i64
429 })
430 .collect()
431}
432
433fn too_many(count: usize) -> Error {
434 Error::internal(format!("a column group of {count} is larger than the format allows"))
435}
436
437fn put_u32(out: &mut Vec<u8>, value: u32) {
438 out.extend_from_slice(&value.to_le_bytes());
439}
440
441#[cfg(test)]
442mod tests {
443 use super::*;
444
445 fn urls(host: &str, count: usize, from: usize) -> Vec<Vec<u8>> {
448 let paths = ["/index.html", "/catalog/item", "/search", "/user/profile/settings"];
449 (from..from + count)
450 .map(|index| {
451 let path = paths[(index / 3) % paths.len()];
452 format!("http://{host}{path}?session={}&ref=google", index * 7).into_bytes()
453 })
454 .collect()
455 }
456
457 fn borrow(values: &[Vec<u8>]) -> Vec<&[u8]> {
458 values.iter().map(Vec::as_slice).collect()
459 }
460
461 fn group<'a>(columns: &'a [Vec<&'a [u8]>]) -> Vec<&'a [&'a [u8]]> {
462 columns.iter().map(Vec::as_slice).collect()
463 }
464
465 fn round_trip(columns: &[&[&[u8]]]) -> Vec<u8> {
466 let bytes = encode_group(columns).unwrap();
467 let back = decode_group(&bytes).unwrap();
468 assert_eq!(back.len(), columns.len());
469 for (decoded, original) in back.iter().zip(columns) {
470 assert_eq!(decoded.len(), original.len(), "{}", describe(&bytes).unwrap());
471 for (left, right) in decoded.iter().zip(*original) {
472 assert_eq!(left.as_slice(), *right, "{}", describe(&bytes).unwrap());
473 }
474 }
475 bytes
476 }
477
478 fn strategy_of(bytes: &[u8]) -> Strategy {
479 Strategy::from_tag(bytes[0]).unwrap()
480 }
481
482 fn size_of(sizes: &[(Strategy, usize)], strategy: Strategy) -> usize {
483 sizes
484 .iter()
485 .find(|(kind, _)| *kind == strategy)
486 .map(|(_, size)| *size)
487 .unwrap_or_else(|| panic!("{} did not apply", strategy.name()))
488 }
489
490 #[test]
491 fn two_columns_from_the_same_universe_share_a_dictionary() {
492 let left = urls("www.example.com", 20_000, 0);
495 let right = urls("www.example.com", 20_000, 5_000);
496 let columns = [borrow(&left), borrow(&right)];
497 let group = group(&columns);
498 let bytes = round_trip(&group);
499 assert_eq!(strategy_of(&bytes), Strategy::SharedDict);
500 let sizes = strategy_sizes(&group).unwrap();
501 let independent = size_of(&sizes, Strategy::Independent);
502 assert!(
505 bytes.len() * 4 < independent * 3,
506 "{} against {independent} independent",
507 bytes.len()
508 );
509 }
510
511 #[test]
512 fn two_columns_of_the_same_alphabet_share_a_symbol_table() {
513 let left = urls("www.example.com", 8_000, 0);
516 let right = urls("news.other.example.org", 8_000, 500_000);
517 let columns = [borrow(&left), borrow(&right)];
518 let group = group(&columns);
519 let bytes = round_trip(&group);
520 let sizes = strategy_sizes(&group).unwrap();
521 let shared = size_of(&sizes, Strategy::SharedTable);
522 let independent = size_of(&sizes, Strategy::Independent);
523 assert!(shared < independent, "{shared} against {independent} independent");
524 assert_eq!(strategy_of(&bytes), Strategy::SharedTable);
525 }
526
527 #[test]
528 fn a_small_column_gains_most_from_a_shared_table() {
529 let big = urls("www.example.com", 20_000, 0);
532 let small = urls("www.example.com", 40, 900_000);
533 let columns = [borrow(&big), borrow(&small)];
534 let sizes = strategy_sizes(&group(&columns)).unwrap();
535 let shared = size_of(&sizes, Strategy::SharedTable);
536 let independent = size_of(&sizes, Strategy::Independent);
537 assert!(shared < independent, "group {shared} against {independent} apart");
538 }
539
540 #[test]
541 fn unrelated_columns_are_left_alone() {
542 let urls = urls("www.example.com", 4_000, 0);
543 let numbers: Vec<Vec<u8>> = (0..4_000)
544 .map(|index| format!("{:016x}", index * 2_654_435_761u64).into_bytes())
545 .collect();
546 let columns = [borrow(&urls), borrow(&numbers)];
547 let group = group(&columns);
548 let bytes = round_trip(&group);
549 assert_eq!(strategy_of(&bytes), Strategy::Independent);
550 }
551
552 #[test]
553 fn a_group_of_one_is_the_column_on_its_own() {
554 let column = urls("www.example.com", 2_000, 0);
555 let columns = [borrow(&column)];
556 let bytes = round_trip(&group(&columns));
557 assert_eq!(strategy_of(&bytes), Strategy::Independent);
558 assert_eq!(bytes.len(), 5 + string::encode(&borrow(&column)).unwrap().len());
559 }
560
561 #[test]
562 fn an_empty_group_round_trips() {
563 let bytes = round_trip(&[]);
564 assert_eq!(decode_group(&bytes).unwrap().len(), 0);
565 }
566
567 #[test]
568 fn columns_do_not_have_to_be_the_same_length() {
569 let left = urls("www.example.com", 3_000, 0);
570 let right = urls("www.example.com", 700, 1_000);
571 let columns = [borrow(&left), borrow(&right)];
572 round_trip(&group(&columns));
573 }
574
575 #[test]
576 fn an_empty_column_in_a_group_round_trips() {
577 let left = urls("www.example.com", 1_000, 0);
578 let empty: Vec<Vec<u8>> = Vec::new();
579 let columns = [borrow(&left), borrow(&empty)];
580 round_trip(&group(&columns));
581 }
582
583 #[test]
584 fn every_strategy_that_applies_decodes_to_the_input() {
585 let left = urls("www.example.com", 3_000, 0);
586 let right = urls("www.example.com", 3_000, 1_000);
587 let columns = [borrow(&left), borrow(&right)];
588 let group = group(&columns);
589 for strategy in [Strategy::Independent, Strategy::SharedTable, Strategy::SharedDict] {
590 let bytes = encode_as(strategy, &group).unwrap().unwrap();
591 let back = decode_group(&bytes).unwrap();
592 assert_eq!(back[0].len(), left.len(), "{}", strategy.name());
593 assert_eq!(back[1][7], right[7], "{}", strategy.name());
594 }
595 }
596
597 #[test]
598 fn the_chooser_picks_the_smallest_strategy() {
599 let left = urls("www.example.com", 2_000, 0);
600 let right = urls("www.example.com", 2_000, 1_000);
601 let columns = [borrow(&left), borrow(&right)];
602 let group = group(&columns);
603 let chosen = encode_group(&group).unwrap();
604 for (_, size) in strategy_sizes(&group).unwrap() {
605 assert!(chosen.len() <= size);
606 }
607 }
608
609 #[test]
610 fn describe_says_what_every_column_came_out_as() {
611 let left = urls("www.example.com", 2_000, 0);
612 let right = urls("www.example.com", 2_000, 1_000);
613 let columns = [borrow(&left), borrow(&right)];
614 let group = group(&columns);
615 for strategy in [Strategy::Independent, Strategy::SharedTable, Strategy::SharedDict] {
616 let bytes = encode_as(strategy, &group).unwrap().unwrap();
617 let shape = describe(&bytes).unwrap();
618 assert!(shape.starts_with(strategy.name()), "{shape}");
619 assert!(shape.contains(", "), "{shape}");
620 }
621 }
622
623 #[test]
624 fn a_truncated_group_is_an_error_and_not_a_panic() {
625 let left = urls("www.example.com", 40, 0);
626 let right = urls("www.example.com", 40, 20);
627 let columns = [borrow(&left), borrow(&right)];
628 let group = group(&columns);
629 for strategy in [Strategy::Independent, Strategy::SharedTable, Strategy::SharedDict] {
630 let bytes = encode_as(strategy, &group).unwrap().unwrap();
631 for len in 0..bytes.len() {
632 assert!(
633 decode_group(&bytes[..len]).is_err(),
634 "{} decoded at {len} bytes",
635 strategy.name()
636 );
637 }
638 }
639 }
640
641 #[test]
642 fn trailing_bytes_are_an_error() {
643 let column = urls("www.example.com", 10, 0);
644 let columns = [borrow(&column)];
645 let mut bytes = encode_group(&group(&columns)).unwrap();
646 bytes.push(0);
647 let error = decode_group(&bytes).unwrap_err();
648 assert!(error.message().contains("left over"), "{error}");
649 }
650
651 #[test]
652 fn an_unknown_tag_is_an_error() {
653 let error = decode_group(&[9, 0, 0, 0, 0]).unwrap_err();
654 assert!(error.message().contains("unknown column group tag"), "{error}");
655 }
656
657 #[test]
658 fn a_code_outside_the_shared_dictionary_is_an_error() {
659 let mut bytes = vec![Strategy::SharedDict.tag()];
660 put_u32(&mut bytes, 1);
661 bytes.extend_from_slice(&string::encode(&[b"one".as_slice()]).unwrap());
662 bytes.extend_from_slice(&integer::encode(&[4]).unwrap());
663 let error = decode_group(&bytes).unwrap_err();
664 assert!(error.message().contains("not in the shared dictionary"), "{error}");
665 }
666
667 #[test]
668 fn overlapping_columns_are_grouped_and_the_rest_are_not() {
669 let first = urls("www.example.com", 20_000, 0);
670 let second = urls("news.other.example.org", 20_000, 0);
671 let third = urls("www.example.com", 20_000, 4_000);
672 let fourth = urls("news.other.example.org", 20_000, 4_000);
673 let sketches: Vec<Sketch> = [&first, &second, &third, &fourth]
674 .iter()
675 .map(|column| Sketch::of(&borrow(column)))
676 .collect();
677 let groups = dictionary_groups(&sketches, 0.5).unwrap();
678 assert_eq!(groups, vec![vec![0, 2], vec![1, 3]]);
679 }
680
681 #[test]
682 fn a_column_that_overlaps_nothing_is_a_group_of_one() {
683 let sketches: Vec<Sketch> = (0..4)
684 .map(|index| Sketch::of(&borrow(&urls("www.example.com", 5_000, index * 100_000))))
685 .collect();
686 let groups = dictionary_groups(&sketches, 0.5).unwrap();
687 assert_eq!(groups, vec![vec![0], vec![1], vec![2], vec![3]]);
688 }
689
690 #[test]
691 fn grouping_is_transitive_and_does_not_need_every_pair_to_overlap() {
692 let a = urls("www.example.com", 20_000, 0);
696 let b = urls("www.example.com", 20_000, 10_000);
697 let c = urls("www.example.com", 20_000, 20_000);
698 let sketches: Vec<Sketch> =
699 [&a, &b, &c].iter().map(|column| Sketch::of(&borrow(column))).collect();
700 assert!(sketches[0].jaccard(&sketches[2]).unwrap() < 0.01);
701 let groups = dictionary_groups(&sketches, 0.3).unwrap();
702 assert_eq!(groups, vec![vec![0, 1, 2]]);
703 }
704
705 #[test]
706 fn grouping_needs_sketches_of_the_same_size() {
707 let sketches = [Sketch::new(16).unwrap(), Sketch::new(32).unwrap()];
708 assert!(dictionary_groups(&sketches, 0.5).is_err());
709 }
710}