天泣記

bep の資料をしばらく眺めて、よくわからなかったところを元ネタの論文を覗いたら実装できるような気がしてきたので実装してみる。

まず、「二つの独立なハッシュ関数 h1, h2 を用意」とあるので、用意しよう。ここでは、二つのハッシュ関数を組にしたクラス HashPair を定義している。

class HashPair
  @saltcounter = "a"

  def self.gensalt
    ret = @saltcounter.dup
    @saltcounter.succ!
    ret
  end
...

複数のハッシュ関数が必要である。だが、自分でハッシュ関数を用意するのは面倒なので、Ruby の String#hash を利用しよう。でも、String#hash はひとつしかない。なので、ちょっと文字列を加えてから String#hash を使うようにしよう。その文字列をちょっとづつ変えていけば、いくらでもハッシュ関数を用意できる。というわけで、そういう用途のために、gensalt は毎回異なる文字列を生成する。

...
  def initialize(range)
    @range = range
    @m = range * 2
    @min1 = 0
    @min2 = range
    @salt1 = HashPair.gensalt
    @salt2 = HashPair.gensalt
  end
...

HashPair.new はひとつの引数 range をとるが、これは資料では m/2 に対応する。gensalt でふたつの文字列を生成して、準備完了である。

...
  def makehash(str)
    h1 = @min1 + (str + @salt1).hash % @range
    h2 = @min2 + (str + @salt2).hash % @range
    [h1, h2]
  end
end

実際に文字列のハッシュをとるときには makehash を使う。これは 0...(m/2) なハッシュ値と (m/2)...m なハッシュ値を組み合わせて返す。

では最小完全ハッシュ関数を作ってみよう。

class MPHF
  def initialize(keys)
    @n = keys.length
    @m2 = ((@n * 2.09).ceil + 1) / 2
    @m = @m2 * 2
    ordered_edges = mapping(keys)
    assigning ordered_edges
    ranking
  end
...

最小完全ハッシュ関数を作るには三つの段階がある。最初の二つの段階で最小でない完全ハッシュ関数を作り、最後のひとつで最小完全ハッシュ関数を仕立てあげる。それぞれの段階が mapping, assigning, ranking メソッドに対応する。

最小完全ハッシュ関数の値域は 0...keys.length であるが、mapping と assigning で作る最小でないやつの値域はそれよりも大きい。具体的には 2.09倍くらい大きい。この 2.09倍という値には意味があるようだが、その根拠はよくわからないのでさておく。ともかくそのように大きくした値が m である。

最初の段階の mapping では、keys からサイクルのない 2部グラフを生成する。

...
  def mapping(keys)
    begin
      hpair = HashPair.new(@m2)
      hs = keys.map {|key| hpair.makehash(key) }
      ordered_edges = check_ascyclic(hs)
    end until ordered_edges
    @hpair = hpair
    ordered_edges
  end
...

ここで生成するグラフは、枝が各 key に対応し、頂点がハッシュ値に対応する。HashPair#makehash に key を渡すと、ハッシュ値がふたつ返ってくるのでこれらが頂点になる。ふたつのハッシュ値は重ならない値域になっているので、生成されるグラフは 2部グラフになる。

そうやって生成したグラフはサイクルを含むかもしれないし含まないかもしれない。含んでいると都合が悪い (かもしれない) ので、含んでいないグラフが生成されるまでハッシュ関数を作りなおして繰り返す。(ここでサイクルを含んでいないグラフが生成される確率が 2.09倍という倍率に関係しているそうな)

サイクルを検査するメソッドが check_ascyclic である。これはグラフにサイクルがない場合、とある順番で枝を並べ直したものを返す。

無向グラフのサイクルを検査するにはいろいろな方法があるが、他の枝がつながっていない頂点をもつ枝をひとつずつ取り除いていき、グラフが最終的に空になるかどうかで検査する。

...
  def check_ascyclic(edges)
    v2es = Array.new(@m) { [] }
    edges.each_with_index {|e, i|
      n1, n2 = e
      v2es[n1] << e
      v2es[n2] << e
    }
...

まず、各頂点についてそれにつながっている枝をリストアップする。

...
    ordered_edges = []
    v2es.each_index {|node|
      next if v2es[node].length != 1
...

各頂点について繰り返すが、頂点につながっている枝がひとつでなければ無視して先を続ける。ひとつならば、以下でそれとそこからつながっているものを可能な限り取り除く。

...
      stack = [node]
...

つながっているのを深さ優先で処理するスタックを作る。

...
      until stack.empty?
        n = stack.pop
        e = v2es[n].first
...

スタックからひとつ頂点を取り出し、それにつながっている枝を取り出す。(スタックの中にはいっている頂点はすべて枝がひとつしかつながっていないので、first とあるけれどそれがすべてである)

...
        ordered_edges << e
        n1, n2 = e
        v2es[n1].delete e
        v2es[n2].delete e
...

枝はその両端の頂点について登録されているので、両方で取り除く。

...
        stack << n1 if v2es[n1].length == 1
        stack << n2 if v2es[n2].length == 1
...

取り除いた結果、一方の頂点につながっている枝は存在しないはずであるが、もう一方にいくつつながっているは不明である。もしひとつしかつながっていなかったらそれを処理するよう、スタックに入れる。(まぁ、スタックじゃなくていいんじゃないかというのはその通りではある。3-ハイパーグラフを使った場合にはスタックが必要なのでこうなっている)

...
      end
    }
    if v2es.any? {|es| !es.empty? }
      return nil # cycle found
    end
...

ループが終わって、枝がひとつでも残っていれば、サイクルが存在するので nil を返す。

...
    ordered_edges
  end
...

サイクルがなければ、ordered_edges を返す。これは枝を取り除いた順番に並べたものである。

このようにしてサイクルがないものが見つかれば、assigning に進む。assigning では、「各枝につき、頂点を1つずつ割り当てる」。

サイクルがない無向グラフなので、これは要するに森である。木がいくつかあるわけである。木に存在する頂点は枝よりもひとつ多いので、割り当てられない頂点がひとつ出てくる。その頂点を根とみなすと、各枝について根につながっていないほうの頂点を割り当てれば良い。

ここで、key に対応しているのが枝で、頂点がハッシュ値に対応しているので、枝に頂点が割り当てられると key にハッシュ値が対応することになる。というわけで、枝を与えられたときにどちらの頂点を選ぶかという情報を記録しておけば、key を与えられたときに二つのハッシュ関数を適用して枝を得て、どちらかを選択することにより他の key と衝突しない値を得ることができる。つまり完全ハッシュ関数ができあがる。

問題はどうやって選択するかであるが、答をいってしまえば、頂点に 0 か 1 の値を割り当て、枝を得たら両端の頂点の値を 2を法として加える。その結果の 0, 1 をふたつのハッシュ関数に対応させる。

根から葉に向かって木をたどると、各頂点でハッシュ関数は交互に使用されるので、たとえば和が 0, 1, 0, 1, 0, 1, ... という数列になるような割り当てにすることができる。そうするための割り当ては 0, 0, 1, 1, 0, 0, 1, 1, ... である。この数列の隣り合ったものを加えると 0, 1, 0, 1, ... となる。

このような割り当てを行うためには根から枝をたどって、割り当てを行う。そのために、check_ascyclic が返した ordered_edges を使う。ordered_edges は、じつは最後の枝の頂点は根とみなしても良いものになっているのである。

というわけで、ordered_edges の最後から割り当てを行う。

...
  def assigning(ordered_edges)
    @g = Array.new(@m, 2)
...

割り当ては @g に記録するが、その初期値は 2 にしておく。2 を法として 2 は 0 であるが、0 と異なり割り当てが行われていないことを示している。

...
    visited = Array.new(@m, false)
    ordered_edges.reverse_each {|e|
      u = e.find {|n| !visited[n] }
...

それぞれの枝について、まだ訪れていない頂点を探す。ここで訪れていない頂点は必ず残っているように枝が並んでいる。

...
      j = e.index(u)
...

その頂点が枝のどっち側の頂点かで使用するハッシュ関数が決まる。j が使いたいハッシュ関数を示している。

...
      e.each {|v|
        next if !visited[v]
        j -= @g[v]
      }
...

各頂点について g の値を足したものが j になって欲しいわけである。なので、j からここで割り当てを行う以外の値を引いて、値をあわせる。

...
      @g[u] = j % 2
...

そんで、2を法としているので 0, 1 に直して割り当てる。

...
      e.each {|v|
        visited[v] = true
      }
...

枝の両端を訪れたものとしてフラグを立てる。

...
    }
  end
...

というようにして割り当てが行われ、完全ハッシュ関数に必要な情報が揃う。

完全ハッシュ関数 (perfect hash function) は以下である。

...
  def phf(key)
    h1, h2 = @hpair.makehash(key)
    i = @g[h1] + @g[h2]
    case (i % 2)
    when 0 then h1
    when 1 then h2
    end
  end
...

ハッシュ値を二つ求め、g でそれぞれに割り当てられた値を加え、ハッシュ値のどちらかを選ぶ。key の長さを気にしないことにすれば、これは定数時間で済む。

こうやってできた完全ハッシュ関数は最小完全ハッシュ関数の 2.09 倍くらい大きな値域をもつので、これをどうにかして縮めれば最小完全ハッシュ関数ができる。要するに、使用していない値を除いた値にすればいいわけで、phf が返した値と 0 の間に有効な値がいくつあるかを数えて、その値を返せば良い。

しかし、ハッシュは定数時間で動いてほしいので、key の数に比例してしまうような数え方は許されない。なので、事前に 256個毎に 0から累積した有効な値の数を記録してテーブルを作る。それが ranking である。

...
  RANK_BLOCKSIZE = 256
  def ranking
    @ranking = []
    k = 0
    @g.each_with_index {|j, i|
      @ranking << k if i % RANK_BLOCKSIZE == 0
      next if j == 2
      k += 1
    }
  end
...

ここで @g に 2 として残っている、割り当てが行われていない、という情報を使っている。

そうやってできた @ranking テーブルを使って、最小完全ハッシュ関数 (minimal perfect hash function) が作れる。

...
  def mphf(key)
    h = phf(key)
    a = h / RANK_BLOCKSIZE
    result = @ranking[a]
    (a * RANK_BLOCKSIZE).upto(h-1) {|i|
      result += 1 if @g[i] != 2
    }
    result
  end
end
...

まぁ、phf の結果を 256で割って、テーブルから累積したのをとってきて、残りを数えているわけである。このループは 256という定数で抑えられているので、定数時間で動くことには変わりない。

いちおう確認してみよう。

...
keys = (1..10).map { rand.to_s }
...

key を適当に用意する。

...
mphf = MPHF.new(keys)
...

最小完全ハッシュ関数を生成する。

...
check = {}
keys.each {|key|
  hash = mphf.mphf(key)
  p [key, hash]
  if check[hash]
    raise "collision: #{hash} : #{check[hash].inspect} #{key.inspect}"
  end
  check[hash] = key
}

key のそれぞれについて最小完全ハッシュ関数を適用して、返ってくる値に重複がないか検査する。

実行すると、例えば以下のようになる。

% ruby mphf.rb
["0.723613576386467", 1]
["0.576711772900182", 5]
["0.705918310484599", 2]
["0.547451826293092", 4]
["0.593221727615754", 6]
["0.444904732250504", 8]
["0.140848444162556", 3]
["0.578232164791411", 0]
["0.986732992090755", 9]
["0.631578866781294", 7]

適当に生成した key の文字列 10個に対し、0 から 9 までの整数を重複なく割り当てる最小完全ハッシュ関数になっている。

天泣記

2008-01-02 (Wed)

2008-01-03 (Thu)

2008-01-08 (Tue)

2008-01-09 (Wed)

2008-01-10 (Thu)

2008-01-11 (Fri)

2008-01-13 (Sun)

2008-01-18 (Fri)

2008-01-25 (Fri)

2008-01-28 (Mon)

2008-01-29 (Tue)

2008-01-31 (Thu)