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              : ///
       9              : /// The function checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
      10              : ///
      11              : /// ```plain
      12              : /// |       |                 |       |
      13              : /// |   1   |    |   2   |    |   3   |
      14              : /// |       |    |       |    |       |
      15              : /// ```
      16              : ///
      17              : /// 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
      18              : /// the same LSN range.
      19              : ///
      20              : /// The exception is that when layer 2 only contains a single key, it could be split over the LSN range. For example,
      21              : ///
      22              : /// ```plain
      23              : /// |       |    |   2   |    |       |
      24              : /// |   1   |    |-------|    |   3   |
      25              : /// |       |    |   4   |    |       |
      26              : ///
      27              : /// If layer 2 and 4 contain the same single key, this is also a valid layer map.
      28           78 : pub fn check_valid_layermap(metadata: &[LayerName]) -> Option<String> {
      29           78 :     let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
      30           78 :     let mut all_delta_layers = Vec::new();
      31          356 :     for name in metadata {
      32          278 :         if let LayerName::Delta(layer) = name {
      33          136 :             if layer.key_range.start.next() != layer.key_range.end {
      34           96 :                 all_delta_layers.push(layer.clone());
      35           96 :             }
      36          142 :         }
      37              :     }
      38          174 :     for layer in &all_delta_layers {
      39           96 :         let lsn_range = &layer.lsn_range;
      40           96 :         lsn_split_point.insert(lsn_range.start);
      41           96 :         lsn_split_point.insert(lsn_range.end);
      42           96 :     }
      43          174 :     for layer in &all_delta_layers {
      44           96 :         let lsn_range = layer.lsn_range.clone();
      45           96 :         let intersects = lsn_split_point.range(lsn_range).collect_vec();
      46           96 :         if intersects.len() > 1 {
      47            0 :             let err = format!(
      48            0 :                 "layer violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
      49            0 :                 layer,
      50            0 :                 intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
      51            0 :             );
      52            0 :             return Some(err);
      53           96 :         }
      54              :     }
      55           78 :     None
      56           78 : }