R-Tree空间索引算法

B+树是数据库世界中最常见的结构之一,尤其是在索引环境中。 它们很好地映射到用于在硬盘驱动器上保存数据的页面/块模型,并在功能(例如排序、范围扫描)和 HDD/SSD 感知读写性能之间提供了一个很好的“万事通”路线。

话虽如此,B+Tree 构造假设索引数据可以轻松地按单个连续顺序组成。 这并不总是可能的,因为某些类型(例如地图坐标)不提供单一顺序,可用于在使用 B+Tree 构建的索引上有效地执行范围扫描(例如给定区域内的所有点)。

为了解决上述问题,人们提出了一种新的数据结构——称为 R(ange)-Trees。 在这篇博文中,我们将讨论它们如何工作以及如何实现它们。

1、树是如何组织的

从顶层概述来看,R-Tree 结构与传统的平衡树没有太大区别,它是当前使用的一些最流行数据库的基础。 事实上,R-Tree 通常是在 B+Tree 存储层(例如 Postgres)之上实现的。 对于那些对两者都不熟悉的人,让我们先粗略地讨论一下相似之处:

  • 我们的树是一个层次结构,由节点组成,节点可以是叶子(存储在第 0 层)也可以是分支(存储在更高层)。
  • 值总是存储在叶子中,而不是存储在分支中。
  • 节点具有子节点的最大容量。 一旦达到此容量并且必须插入新的子分支,则需要首先将其父分支分成两半,并将新的子分支添加到以这种方式创建的一半中。
  • 分支不存储值,但可以存储键,这些键用于导航我们要在树结构中进一步查找的子节点。

现在有一些针对 R 树的特定内容:

  • R-Tree 允许存储在分支中的键具有与插入时调用的实际键不同的结构。 我们很快就会详细讨论这个问题。
  • 虽然 B 树保证存储在其中的值保持总顺序,但这里并不那么容易。 原因是R-Tree用于存储多维数据,这些数据不容易以线性顺序表示。
  • 由于上述原因,在扫描期间查找要转到哪些子节点需要在下降之前检查特定节点内的所有子节点。 因此,R-Tree 的性能可能比 B+Tree 差。

这里的整个想法是,树中的每个分支键都被定义为最小边界空间(或集合或范围),其中包含其所有子键。 即使叶子和分支存储不同类型的数据,例如,这也有效。 叶子可以存储地理位置数据(点/坐标),然后分支键由限制这些点的矩形区域表示。 这个想法递归地起作用 - 较高级别的分支键是较低级别分支键的边界矩形:

在这里,我们将进行一个简单的 R 树实现。 我们只会覆盖二维空间,但请记住,可以扩展 R-Tree 以覆盖任意数量的维度:如果你有兴趣,可以查看 Swim-rust 的R-Tree 实现,它提供了 在单个代码库上对 2D 和 3D 空间的通用支持,并从那里提供更高维度的泛化。

我们需要提供一些基础数据结构。 出于我们的目的,我们只需要了解点和矩形。

type Point = { x: float; y: float }
type Rect = 
  { low: Point    // lower-left point
    high: Point } // higher-right point

type Spatial =
  abstract Boundary: Rect

我们还定义了一个专用的 Spatial 接口 - 这样我们就可以存储任何数据类型,只要我们可以用最小边界矩形包裹它。 对于单点元素,我们可以想象左下点和右上点彼此相等的矩形。

现在是时候定义我们的树结构了。 我们的 R 树由节点组成,我们可以将这些节点分为叶子(包含键值条目)和包含许多子节点的分支 - 无论是叶子还是其他分支。 我们还将用一个数字来标记每个节点,以告知它在树层次结构中的位置:叶子始终位于 0 级,而树根始终位于最高级别。

type Node<'k,'v when 'k :> Spatial> = 
  { level: int  // node-depth level - 0 for leaf nodes
    entries: Entry<'k,'v>[] } // child entries

and Entry<'k,'v when 'k :> Spatial> =
  | Leaf of key:'k * value:'v
  | Branch of key:Rect * Node<'k,'v>
  
type RTree<'k,'v when 'k :> Spatial> =
  { root: Node<'k,'v>
    config: Config }

请记住,虽然只有叶节点包含值,但叶节点和分支中都存在键。 然而,分支键(对于由矩形表示的二维空间)描述了所谓的最小边界矩形,它足以覆盖存储在该父级中的所有子键。

在这篇博文中,我们将介绍 R-Tree 应该公开的 3 个最关键的操作:搜索区域中的元素、插入/更新和删除。 我们将从搜索开始。

2、范围扫描

R-Tree 武器库中最重要的操作是搜索所有插入元素的能力,这些元素可以在给定区域内找到。 这里我们假设我们的区域是由矩形空间确定的,但实际上你可以通过定义两个操作的任何数据类型抽象此行为:

  • constains a b,它告知形状 b 是否完全属于由形状 a 确定的区域。 如果 a 等于 b,这也成立。
  • insertect a b,表明形状 b 的至少一个点可以在形状 a 内找到。

虽然我们习惯于根据 2D 或 3D 空间的形状来思考这些问题,但这种行为可以泛化 - 我们将对此进行总结讨论。

我们的搜索函数将从根节点向下递归下降,将所有匹配包含条件的叶子条目添加到结果集中。 为了不搜索树中的所有节点,我们只需跳过既不包含也不与我们正在查找的区域相交的节点。

module Node

let search (area: Rect) (n: Node<'k,'v>) =
  let rec loop (result: ResizeArray<'v>) (area: Rect) (n: Node<'k,'v>) =
    for e in n.entries do
      match e with
      | Leaf(key, value) when Rect.contains area key.Boundary ->
        // leaf key fits into `area`, add it to result set
        result.Add value
      | Branch(r, child) when Rect.contains area r || Rect.intersects area r -> 
        // there are children within this branch, that fit into `area`
        loop result area child
      | _ -> () // there are no shared points, skip that node
  let found = ResizeArray()
  loop found area n
  found.ToArray()

3、元素插入

由于我们已经跳过了搜索(这相当简单),现在我们进入深水区并讨论插入/更新算法。 其复杂性背后的原因在于节点分裂语义。 这是强制性的,因为我们不希望节点的子节点无限制地成长。 由于 R-Tree 搜索不仅需要递归地向下遍历节点及其子节点,而且还需要对所选节点的每个子节点执行检查,而 B+Tree 算法仅检查子节点,直到第一个子密钥不满足以下条件: 我们的搜索查询。

在这里,我们使用每个节点可以容纳的最大子节点数量的可配置值,保存在 config.maxCap属性内。 首先,我们假设我们找到一个适合成为其父项的分支,然后添加一个新的叶项:

module Node 

let private addLeaf config item node =
  let rect = Entry.boundary item
  // try find entry matching added item's key
  let idx = 
    node.entries 
    |> Array.tryFindIndex (function Leaf(k, _) -> k.Boundary = rect | _ -> false)
  match idx with
  | None ->
    // item with that key didn't exist in R-Tree, 
    // push it at the end of children list
    let entries = Array.insertAt node.entries.Length item node.entries
    let node = { node with entries = entries }
    if n.entries.Length > config.maxCap then
      // we surpassed capacity of node, we need to split it
      let struct(l, r) = split config node
      Split(l,r)
    else 
      NoSplit node
  | Some idx ->
    // we're updating an existing entry
    let node = { node with entries = Array.replace idx item n.entries }
    NoSplit node

在这里,我们使用 Split/NoSplit 结果值来通知我们是否必须拆分现有节点,以便它在不违反最大子容量的情况下容纳插入的叶子。 有关其发生方式的详细信息隐藏在上面调用的 split 函数后面。 我们稍后会介绍它。

我们介绍了如何将叶项添加到分支节点,但问题仍然存在:给定一个子节点列表,我们如何确定哪一个是插入叶项的正确选择? 同样,R 树键顺序不是全部,因此可以有多个候选键可供选择。

我们这里使用的规则如下:最好的候选者是分支,其最小边界矩形在插入新条目后将按最小因子扩展。 那是因为我们想要保持这些最小边界矩形......好吧,最小。 它们越小,与同一级别的其他人重叠的机会就越小,并且更准确。 我们可以通过获取当前分支边界矩形和封装插入元素的矩形的表面积之间的差值来轻松计算该因子。

module Node

let private addBranch item n =
  let mutable minEntry = n.entries.[0]
  let mutable minEntryIdx = 0
  let itemBoundary = (Entry.boundary item)
  let mutable minRect = Rect.merge (Entry.boundary minEntry) itemBoundary
  let mutable minDiff = Rect.area minRect - Rect.area (Entry.boundary minEntry)
  for i in 1..(n.entries.Length-1) do
    let candidate = n.entries.[i]
    let bound = Entry.boundary candidate
    let expanded = Rect.merge (Entry.boundary item) bound
    // calculate the difference of boundary area increase in case if current
    // candidate would be picked to host an item
    let diff = Rect.area expanded - Rect.area bound
    if diff < minDiff then
      // area difference is smaller than the last best candidate
      minDiff <- diff
      minRect <- expanded
      minEntry <- candidate
      minEntryIdx <- i
  let (Branch(_, child)) = minEntry
  (child, minRect, minEntryIdx)

通过上面定义的两个函数,我们需要以通用和递归的方式将它们组合在一起:

module Node

let rec add (config: Config) level (item: Entry<'k,'v>) (n: Node<'k,'v>) : Split<'k,'v> =
  match item with
  | Branch _ when level = n.level ->
    let n = { n with entries = Array.insertAt n.entries.Length item n.entries }
    if n.entries.Length > config.maxCap then
      // we surpassed capacity of node, we need to split it
      let struct(l, r) = split config n
      Split(l,r)
    else 
      NoSplit n
  | _ ->
    if isLeaf n then
      // add leaf node
      addLeaf config item n
    else
      // descent recursivelly
      let (child, minRect, minEntryIdx) = addBranch item n
      match add config level item child with
      | Split(first, second) ->
        // since our R-Tree is immutable we need produce a new copy of entries
        // array with updated child being removed 
        let entries = Array.removeAt minEntryIdx n.entries
        let n = { n with entries = Array.append entries [|first;second|] }
        if n.entries.Length > config.maxCap then
          // because of the split from children, we surpassed children capacity
          // of a current node, we need to split it and propagate result to a parent
          let struct(l, r) = split config n
          Split(l,r)
        else 
          NoSplit n
      | NoSplit node ->
        let e = Branch(minRect, node)
        let n = { n with entries = Array.replace minEntryIdx e n.entries }
        NoSplit n

该函数的大部分复杂性来自于需要递归地更新子节点列表,这可能需要拆分块,这本身最终也可能导致父节点的递归拆分。

4、分裂节点

到目前为止,我们跳过了节点分裂算法的细节。 现在是时候讨论它了。 正如我们已经目睹的那样,如果添加新的子节点导致超出允许的子节点数量,我们会将父节点分成两半,每一半都包含原始父节点的子节点的(公平)子集。

同样,主要的挑战是缺乏线性顺序,我们可以用它来简单地在子列表中设置分割点。 我们将讨论两种分裂策略:线性和二次。 它们都有共同的特征:*我们首先使用从原始父子项获取的“种子”键定义两个组,然后逐个逐步耗尽子项并将它们分配给“更接近”的组,直到所有子项都被耗尽。

我们可以问自己:这非常简单,那么我们使用的策略有什么区别? 这一切都归结为决定两件事:

  • 如何选取两个种子节点。
  • 确定“更接近”的含义。

稍后我们将介绍这两个内容。 现在让我们展示一下分割函数的基本框架:

module Node

/// Each split group consists of children entries and info about 
/// area that covers all of them
type Group<'k,'v  when 'k :> Spatial>  = (Entry<'k,'v>[] * Rect)

let split (config: Config) (n: Node<'k,'v>) =
  let split config (entries: Entry<'k,'v>[]) : (Group<'k,'v> * Group<'k,'v>) =
    // pick indexes of seed nodes
    let (i1, i2) =
      match config.strategy with
      | SplitStrategy.Linear    -> SplitStrategy.linear entries
      | SplitStrategy.Quadratic -> SplitStrategy.quadratic entries
    
    let entries = ResizeArray(entries)
    
    // pick seed nodes
    // `i2` is always greater than `i1` 
    let seed2 = entries.[i2]
    let seed1 = entries.[i1]
    entries.RemoveAt i2
    entries.RemoveAt i1
    
    // first split group
    let mutable bound1 = Entry.boundary seed1
    let group1 = ResizeArray<_>(config.minCap)
    group1.Add seed1
    
    // second split group
    let mutable bound2 = Entry.boundary seed2
    let group2 = ResizeArray<_>(config.minCap)
    group2.Add seed2
    
    while entries.Count > 0 do
      //TODO: drain entries into groups
         
    // once all entries have been drained let's return newly
    // created split groups
    let first: Group<'k,'v> = (group1.ToArray(), bound1)
    let second: Group<'k,'v> = (group2.ToArray(), bound2)
    (first, second)
    
  // split node entries into two new branch nodes
  let ((group1, bound1), (group2, bound2)) = split config n.entries
  let first  = Entry.Branch(bound1, { level = n.level; entries = group1 })
  let second = Entry.Branch(bound2, { level = n.level; entries = group2 })
  (first, second)

上面的代码基本上负责创建分割组,用种子节点初始化它们,然后在一切完成后组合结果。 我们故意将入口漏极循环的主体留空,所以我们现在可以讨论它。

// drain entries into groups
while entries.Count > 0 do
  // if we have just enough remaining entries to fit
  // minimun capacity of any group, there's no need to pick
  // just push all entries into that group
  if entries.Count + group1.Count = config.minCap then
    for e in entries do
      bound1 <- Rect.merge bound1 (Entry.boundary e)
      group1.Add e
    entries.Clear ()
  elif entries.Count + group2.Count = config.minCap then
    for e in entries do
      bound2 <- Rect.merge bound2 (Entry.boundary e)
      group2.Add e
    entries.Clear ()
  else
    // if we still have plenty of entries, use algorithm to pick a group
    let (idx, expanded, group) =
      match config.strategy with
      | SplitStrategy.Linear -> SplitStrategy.nextLinear entries bound1
      | SplitStrategy.Quadratic -> SplitStrategy.nextQuadratic entries bound1 bound2 group1.Count group2.Count
          
    // pick entry choosen by split strategy and assign it to its new group
    let e = entries.[idx]
    entries.RemoveAt idx
    if group = 0 then
      bound1 <- expanded
      group1.Add e
    else
      bound2 <- expanded
      group2.Add e

让我们从 else 块开始:这里我们使用特定于策略的算法,该算法返回以下三元组:

  • 已选取条目的索引。
  • 该条目的新矩形边界,由所选条目的键扩展。
  • 所选条目应插入的组号。

该循环的另一部分是剩余的 if 语句。 它们来自 B+Tree/R-Tree 的要求之一:将节点一分为二时,我们希望公平地分配子节点,这意味着每个拆分组应至少具有配置的最小( config.minCap)子节点数 - 通常是最大容量的一半,但这不是严格的规则。 如果没有它,我们的算法可能会失控,并将孩子们分成例如。 每组容量的90%/10%,这会导致树不平衡。

4.1 线性分割

现在是讨论不同的分割启发法的时候了。 我们将从线性分割开始。

线性分割背后的核心思想是:从距离最远的元素中找到两个种子节点,然后将下一个条目随机分配给其中一个节点。 为了计算该距离,我们将定义一个支撑结构,我们将使用它来计算每个维度的边界的最小/最大值(每个矩形边界由其左下点和右上点确定) 。 由于我们在二维空间中操作,因此我们将使用两个这样的结构。

/// Struct storing statistics for a single dimension 
[<Struct;IsByRefLike>]
type private DimStats =
  { mutable minLow: float    // minimum value of left-lower point dimension
    mutable maxHigh: float   // maximum value of righ-higher point dimension
    
    mutable maxLow: float    // maximum value of left-lower point dimension
    mutable lowIndex: int    // index of an entry having value above
    
    mutable minHigh: float   // minimum value of left-lower point dimension
    mutable highIndex: int } // index of an entry having value above

[<RequireQualifiedAccess>]
module private DimStats =
  let create () =
    { minLow = Double.MaxValue
      maxHigh = Double.MinValue
      maxLow = Double.MinValue
      minHigh = Double.MaxValue
      lowIndex = 0
      highIndex = 0 }
    
  let inline farest (s: DimStats inref) = Math.Abs(s.maxHigh - s.minLow)
  
  let inline nearest (s: DimStats inref) = Math.Abs(s.minHigh - s.maxLow)

请记住,这种方法并不将我们限制在 2D 空间 - 可以写入立方体形状的 3D 形状,它也只需要两个点来保持正确的形状定义 - 唯一的区别是这些点使用 3 个坐标 (x,y,z) 而不是两个 (x,y)。 我们可以将其抽象为适用于任意数量的维度。

更新我们的维度统计结构相当简单:

/// Update stats based on dimension of lower-left (`lo`) and higher-right (`hi`)
/// corners of a boundary at given index `i`.
let inline computeDim (s: DimStats byref) (lo: float) (hi: float) (i: int) =
  s.minLow <- Math.Min(s.minLow, lo) // update minimum lower-left seen so far
  s.maxHigh <- Math.Max(s.maxHigh, hi) // update maximum higher-right seen so far
  if lo > s.maxLow then
    // if we found new maximum lower-left, cache index of its entry
    s.maxLow <- lo
    s.lowIndex <- i
  if hi < s.minHigh then
    // if we found new minimum lower-left, cache index of its entry
    s.minHigh <- hi
    s.highIndex <- i

我们的线性种子选择依赖于计算给定尺寸的最近和最远距离。 然后我们对这些值进行标准化并比较每个维度,如下所示:

module SplitStrategy
    
let linear (entries: #IReadOnlyList<Entry<'k,'v>>) =
  let mutable x = 0 // index of first seed node
  let mutable y = 1 // index of second seed node
  let mutable dx = DimStats.create () // Entry statistics on `x` dimension
  let mutable dy = DimStats.create () // Entry statistics on `y` dimension
  
  // compute statistics of all entries for each dimension
  if entries.Count > 2 then
    for i in 0..(entries.Count-1) do
      let e = entries.[i]
      let rect = Entry.boundary e
      computeDim &dx rect.low.x rect.high.x i
      computeDim &dy rect.low.y rect.high.y i
  
  // compute normalized ratio between farest and nearest distance
  // for each dimension
  let farX, farY = DimStats.farest &dx, DimStats.farest &dy
  let nearX, nearY = DimStats.nearest &dx, DimStats.nearest &dy
  let normX, normY = (nearX / farX), (nearY / farY)
  let lowIdx, highIdx = if normX > normY then dx.lowIndex, dx.highIndex else dy.lowIndex, dy.highIndex
  x <- lowIdx
  y <- highIdx
  
  let cmp = x.CompareTo y
  if cmp < 0 then (x, y)
  elif cmp > 0 then (y, x)
  elif x = 0 then (0, 1)
  else (0, y)

你可能会注意到,我们选择最远边界的算法可能并不完美 - 获胜种子节点基于单个归一化维度所理解的最大归一化距离。 不过这没关系。 我们不需要一个理想的响应,但需要一个可以快速计算的同时还可以接受的响应。

在为每个分组分配条目方面,我们只是随机分配它们。 在线性方法中,我们优先考虑速度而不是精度。

module SplitStrategy

let inline nextLinear (entries: #IReadOnlyList<Entry<'k,'v'>>) boundary =
  let rect = Rect.merge boundary (Entry.boundary entries.[0])
  (0, rect, 0)

如果你对上面的代码感到困惑,请让我解释一下:条目代表没有建立键顺序的子项。 由于此函数在之前定义的 split 函数的上下文中工作,因此我们总是将下一个条目分配给第一组。 一旦剩余条目的数量足够低以填充第二组以达到最小预期节点容量,拆分函数本身会将它们分配给另一个拆分组。

虽然线性分割在边界分配方面并不完美(由于随机分割组构造),但 nextLinear 函数的简单性使得分配新项目的速度非常快,每当添加/删除条目的流失率很高时,它都是一个很好的策略。

4.2 二次分裂

二次分割策略以形状区域大小的想法为导向 - 你可以将形状区域视为两点之间的距离的等价物。 我们将节点分配给一个组,该组将按最小因子扩展该组中元素的最小边界矩形(这意味着它们距种子的距离最小)。 另一方面,我们选择种子节点,其最小边界矩形面积最高(这相当于它们彼此距离最“远”)。 它是如何工作的?

给定多维空间,我们可以假设两个元素之间的外接矩形面积越大,它们彼此之间的距离就越远。 另一方面,种子本身应该具有尽可能小的面积 - 否则我们可能只是查看一个具有巨大尺寸(例如等于整个搜索空间)的种子节点,这将使我们算法的第二部分产生偏差并导致所有条目 分配给该种子的组(好像不会导致其面积增加,从而使其成为最佳)。

module SplitStrategy

let quadratic (entries: #IReadOnlyList<Entry<'k,'v>>) =
  let mutable x = 0
  let mutable y = 1
  let mutable maxDiff = Double.MinValue
  if entries.Count > 2 then
    for i in 0..(entries.Count-1) do
      for j in 1..(entries.Count-1) do
        let first = Entry.boundary entries.[i]
        let second = Entry.boundary entries.[j]
        let combined = Rect.merge first second
        // calculate the difference between area of minimum bounding rectangle
        // of two elements the the area of the elements themselves
        let diff = (Rect.area combined) - (Rect.area first) - (Rect.area second)
        if diff > maxDiff then
          // the winner is the biggest area difference
          maxDiff <- diff
          x <- i
          y <- j
  (x, y)

第二个函数负责从子项列表中选取下一个条目并将其分配给两组选择之一。 与线性方法不同,二次方法中为每个条目选择正确的组取决于计算每个组边界的大小(提醒:我们使用它们作为分支节点的键)并选择最能扩展其边界的组。

第一:如何计算添加到组边界的条目的扩展比例? 这只是特定组中包含和不包含条目的最小边界矩形的面积之间的差异。

module SplitStrategy

let private preference (e: Entry<'k,'v>) rect1 rect2 =
  let bound = Entry.boundary e // boundary of current entry
  // compute min bounding rectangles that would be required after adding
  // an entry to each of the groups
  let wrap1 = Rect.merge rect1 bound
  let wrap2 = Rect.merge rect2 bound
  // compute how much each of the groups boundaries would grow by adding
  // the current entry
  let diff1 = Rect.area wrap1 - Rect.area rect1
  let diff2 = Rect.area wrap2 - Rect.area rect2
  (diff1, wrap1, diff2, wrap2)

有了这个集合,现在让我们实现一个决策函数:

module SplitStrategy

let private selectGroup rect1 rect2 len1 len2 diff1 diff2 =
  // first pick by minimal bounding area expansion...
  if diff1 < diff2 then 0
  elif diff2 < diff1 then 1
  // ...otherwise pick the group with smaller area...
  elif Rect.area rect1 < Rect.area rect2 then 0
  elif Rect.area rect2 < Rect.area rect1 then 1
  // ...lastly pick the group which has smaller amount of entries assigned
  elif len1 < len2 then 0
  elif len2 < len1 then 1
  else 0

正如你所看到的,我们实际上有三个标准用于选择一个组 - 而我们之前计算的面积扩展参数可能会捕获 9/10 的情况,如果两个组的情况相同,我们就会选择面积最小的组,并且在 面积大小相等的情况下,按组长度计算。

请记住:我们不需要真正担心群体规模不平衡。 我们的拆分函数仅使用组解析算法,直到达到所需的最小容量。 在这里,我们还选择最佳的条目 - 添加它们将导致最少的组边界区域扩展 - 首先:

module SplitStrategy

let inline nextQuadratic (entries: #IReadOnlyList<Entry<'k,'v>>) rect1 rect2 len1 len2 =
  let mutable idx = 0
  let mutable e = entries.[idx]
  let (pref1, exp1, pref2, exp2) = preference e rect1 rect2
  let mutable maxPreferenceDiff = Math.Abs(pref1 - pref2)
  let mutable group = selectGroup rect1 rect2 len1 len2 pref1 pref2
  let mutable expanded = if group = 0 then exp1 else exp2
  
  for i in 1..(entries.Count - 1) do
    let e = entries.[i]
    let struct(pref1, exp1, pref2, exp2) = preference e rect1 rect2
    let preferenceDiff = Math.Abs(pref1 - pref2)
    // pick the entry that causes the smallest min bounding rectangle expansion
    if maxPreferenceDiff <= preferenceDiff then
      maxPreferenceDiff <- preferenceDiff
      idx <- i
      group <- selectGroup rect1 rect2 len1 len2 pref1 pref2
      expanded <- if group = 0 then exp1 else exp2
  
  (idx, expanded, group)

虽然二次分割比线性分割提供更好的质量,但生产成本更高。 一般启发式是,在 R 树相当静态(很少添加条目)的情况下,这是一个不错的选择,因为节点分裂算法比线性分裂使用的随机方法昂贵得多。

5、元素去除

当我们讨论通过键从 R-Tree 中删除项目时,我们可以考虑两种情况:

  • 删除条目,该条目的键等于给定的参数。
  • 删除给定参数边界内存在的所有带有键的条目。

这里我们只考虑第一个。 为了保持我们的 R 树平衡,我们需要考虑一种情况,删除一个项目可能会导致节点子节点计数低于最小允许数量(规范上它是最大允许数量的一半,但这不是一个严格的规则) )。 如果发生这种情况,我们需要考虑将其与另一个节点合并的必要性。

首先,我们从删除叶节点级别的条目开始。

module Node

let rec remove (config: Config) rect (n: Node<'k,'v>) =
  if isLeaf n then
    // handle insertion at leaf
    // check if current leaf node contains expected entry
    let idx =
      n.entries
      |> Array.findIndex (function Leaf(k,_) -> rect = k.Boundary | _ -> false)
    
    if idx = -1 then
      (n, None, [||]) // entry not found in a current leaf node
    else
      // remove entry from the leaf node
      let removed = n.entries.[idx]
      let n = { n with entries = Array.removeAt idx n.entries }
      (n, Some removed, [||])
  else
    //TODO: handle insertion at branch

我们的 Node.remove 函数返回一个已更改的节点、一个已删除的条目(如果有)和一个所谓的“孤儿”列表 - 由于删除而导致其父节点的最小允许容量以下的节点。 我们最终需要在移除过程结束时重新插入它们。

好的,那么在分支机构级别删除怎么样? 嗯,有几件事需要记住:

  • 如果作为分支键的最小边界矩形表示它包含我们要删除的键,那么这只是一个建议而不是承诺。 我们想要删除的项目实际上可能不存在,因此我们可能需要更改其他节点。 这也是为什么我们的递归函数参数之一是已删除的项目 - 所以我们知道它是否确实被发现。
  • 如果我们删除了一个条目,结果可能是我们需要重新计算分支的最小边界矩形。 然而,这种情况并不总是发生,只有当坐标或删除的元素与最小外接矩形本身的边缘相邻时才会发生。

考虑到所有这些挑战,让我们看看我们的分支删除会是什么样子:

let rec remove (config: Config) rect (n: Node<'k,'v>) =
  if isLeaf n then
    //TODO: handle insertion at leaf
  else
    // handle insertion at branch
    let mutable entryIdx = -1
    let mutable removedOption = None
    let mutable entries = Array.copy n.entries
    
    // we're at the branch node, check which child nodes could 
    // potentially contain item we're looking for
    let mutable i = 0
    while i < entries.Length do
      let e = entries.[i]
      let r = Entry.boundary e
      if Rect.contains r rect then
        // we found a potential matching branch
        let (Branch(r, child)) = e
        // try to remove an item from that child
        let (child, removed, orphans) = remove config rect child
        match removed with
        | None -> () // item not found
        | Some key ->
          removedOption <- Some((removed, orphans))
          let removedRect: Rect = Entry.boundary key
          let r =
            // check if removed item's key was adjacent to the
            // edges of a minimum bounding rectangle of its 
            // parent...
            if Point.anyMatch removedRect.low r.low || 
               Point.anyMatch removedRect.high r.high then
              // ...if so, we need to recalculate minimum bounding
              // rectangle, as it may have shrunk
              child.entries
              |> Array.map Entry.boundary
              |> Array.reduce Rect.merge
            else r
          entries.[i] <- Branch(r, child)
          if child.entries.Length < config.minCap then
            // if removed item caused parent to have less than
            // allowed minimum of entries, we'll need to merge it
            entryIdx <- i
            i <- entries.Length // break;
      i <- i + 1
      
    //TODO: remove element from current branch

我们介绍了查找可能需要删除的节点的条目,但删除本身尚未被调用:

module Node

let rec remove (config: Config) rect (n: Node<'k,'v>) =
  if isLeaf n then
    //TODO: handle insertion at leaf
  else
    //TODO: index of candidate having a key to remove ...
       
    // remove element from current branch
    let n = { n with entries = entries } 
    match removedOption with
    | None -> (n, None, None)
    | Some(removed, orphans) when entryIdx < 0 -> 
      (n, removed, orphans)
    | Some(removed, orphans) ->
      // after removing the item, parent capacity gone under minimum
      // allowed number of children. That parent gets orphaned, and
      // will have to be reinserted later on
      let orphan = n.entries.[entryIdx]
      let n = { n with entries = Array.removeAt entryIdx n.entries }        
      let orphans = Array.insertAt orphans.Length orphan orphans
      (n, removed, orphans)

完成 Node.remove 函数后,我们现在准备将其包装成可能面向用户 API 的内容:

module RTree

let remove (key: 'k) (tree: RTree<'k,'v>) =
  let rect = key.Boundary
  let (root, removed, orphans) = Node.remove tree.config rect tree.root
  let mutable root =
    if root.entries.Length = 1 && not (Node.isLeaf root) then
      match root.entries.[0] with
      | Branch(_, n) -> n
      | _ -> root
    else root
  // reinsert orphaned nodes
  for orphan in orphans do
    match orphan with
    | Leaf _ ->
      root <- insert tree.config 0 orphan root
    | Branch(_, n) ->
      for e in n.entries do
        root <- insert tree.config n.level e root
  { tree with root = root }

可以在此处找到上述代码片段的完整源代码。

6、结束语

下一步是什么?

在这里,我们只讨论在 2D 矩形空间中使用 R-Tree。 然而,它们可以在更广泛的背景下使用,例如。 将线性分割中使用的 DimStats 从 2 维扩展到任意维数是相当容易的。 如果我们能够抽象出包含、相交、等于、面积或合并等概念,我们就可以扩展 R 树以处理任何类型的形状。 但这还不是全部! 如果我们将最小边界矩形的概念扩展到最小边界集(紧凑地描述所有嵌套键的数据集),我们可以索引更广泛的结构。

在这篇博客中,你经常会遇到矢量时钟的概念 - 用于时间戳和跟踪操作并发性的数据结构。 主要区别在于思维方式的转变:

  • 如果我们可以将 2D 空间扩展到 N 维空间,我们就有了表示矢量时钟的工具。 在 2D 或 3D 空间中,维度被映射到 x、y(和 z)点坐标,它们可以表示为数组中的索引或...向量时间戳的副本标识符。
  • 类似形状的操作在矢量时钟部分比较空间中也有其等价物:
  • containsequals 对应 >===
  • intersect 对应比较
  • merge 相当于我们过去讨论过的时钟合并

此外,通过最小边界集,我们还可以索引本质上非数字的数据。 作为示例,我们使用 PostgreSQL GiST 索引(它使用这种确切的方法)来存储文本数据。 在这种情况下,我们将叶键表示为索引文本文档中使用的单词(术语)集,而分支键表示为(默认为 64 位长)布隆过滤器:

  • 对于叶键, contains a b / equals a b / intersetcs a b / merge a b 运算符与在单词集上执行的传统 Set 运算符基本相同( isSuperSet a b / setEquals a b / nonEmpty(intersect a b) / union a b)。
  • 对于分支键, contains a b / equals a b / intersects a b / merge a b 运算符可以映射到位掩码操作,例如  (a ||| b = a) / (a = b)/(a &&& b <> 0) / ( a ||| b)

通常,这种方法使我们能够创建比全文搜索解决方案中传统使用的反向文档索引更新速度更快(但读取速度更慢)的索引。

上面定义的所有操作都是可能的,因为与 B+ 树不同,R 树不需要按精确的线性顺序进行操作。 这里缺少的是分割算法的定义,因为我们需要一种方法来表示和计算最小边界集的扩展,而这并不总是容易以通用方式表示。 有关如何处理它的参考,可以查看 Postgres 中文本数据分割算法的实现。


原文链接:R-Tree: algorithm for efficient indexing of spatial data

BimAnt翻译整理,转载请标明出处