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           60 : pub fn check_valid_layermap(metadata: &[LayerName]) -> Option<String> {
      29           60 :     let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
      30           60 :     let mut all_delta_layers = Vec::new();
      31          280 :     for name in metadata {
      32          220 :         if let LayerName::Delta(layer) = name {
      33          106 :             if layer.key_range.start.next() != layer.key_range.end {
      34           68 :                 all_delta_layers.push(layer.clone());
      35           68 :             }
      36          114 :         }
      37              :     }
      38          128 :     for layer in &all_delta_layers {
      39           68 :         let lsn_range = &layer.lsn_range;
      40           68 :         lsn_split_point.insert(lsn_range.start);
      41           68 :         lsn_split_point.insert(lsn_range.end);
      42           68 :     }
      43          128 :     for layer in &all_delta_layers {
      44           68 :         let lsn_range = layer.lsn_range.clone();
      45           68 :         let intersects = lsn_split_point.range(lsn_range).collect_vec();
      46           68 :         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           68 :         }
      54              :     }
      55           60 :     None
      56           60 : }