Line data    Source code

       1              : use std::collections::BTreeSet;
       2              : 
       3              : use itertools::Itertools;
       4              : 
       5              : use super::storage_layer::LayerName;
       6              : 
       7              : /// Checks whether a layer map is valid (i.e., is a valid result of the current compaction algorithm if nothing goes wrong).
       8              : /// The function checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
       9              : ///
      10              : /// ```plain
      11              : /// |       |                 |       |
      12              : /// |   1   |    |   2   |    |   3   |
      13              : /// |       |    |       |    |       |
      14              : /// ```
      15              : ///
      16              : /// This is not a valid layer map because the LSN range of layer 1 intersects with the LSN range of layer 2. 1 and 2 should have
      17              : /// the same LSN range.
      18              : ///
      19              : /// The exception is that when layer 2 only contains a single key, it could be split over the LSN range. For example,
      20              : ///
      21              : /// ```plain
      22              : /// |       |    |   2   |    |       |
      23              : /// |   1   |    |-------|    |   3   |
      24              : /// |       |    |   4   |    |       |
      25              : ///
      26              : /// If layer 2 and 4 contain the same single key, this is also a valid layer map.
      27          180 : pub fn check_valid_layermap(metadata: &[LayerName]) -> Option<String> {
      28          180 :     let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
      29          180 :     let mut all_delta_layers = Vec::new();
      30          840 :     for name in metadata {
      31          660 :         if let LayerName::Delta(layer) = name {
      32          318 :             if layer.key_range.start.next() != layer.key_range.end {
      33          204 :                 all_delta_layers.push(layer.clone());
      34          204 :             }
      35          342 :         }
      36              :     }
      37          384 :     for layer in &all_delta_layers {
      38          204 :         let lsn_range = &layer.lsn_range;
      39          204 :         lsn_split_point.insert(lsn_range.start);
      40          204 :         lsn_split_point.insert(lsn_range.end);
      41          204 :     }
      42          384 :     for layer in &all_delta_layers {
      43          204 :         let lsn_range = layer.lsn_range.clone();
      44          204 :         let intersects = lsn_split_point.range(lsn_range).collect_vec();
      45          204 :         if intersects.len() > 1 {
      46            0 :             let err = format!(
      47            0 :                 "layer violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
      48            0 :                 layer,
      49            0 :                 intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
      50            0 :             );
      51            0 :             return Some(err);
      52          204 :         }
      53              :     }
      54          180 :     None
      55          180 : }