はじめに

こちらはアドベントカレンダー18日目の記事です。

本記事では、find_eachの内部実装を調査しながら、その処理内容や設計上の意図について考察します。

実装を追う

本記事ではRailsドキュメントに記載がある以下のfind_eachの使用例において、実行されるコードを追っていこうと思います。

Person.find_each do |person|
  person.do_awesome_stuff
end

find_eachメソッド

block_given?メソッドはカレントコンテキストにブロックが渡されていればtrueが返されるのでfind_in_batchesが実行されます。find_in_batchesメソッドへブロックを渡しており、ブロック内部ではfind_each実行時に渡したブロックをeachメソッドに渡しています。つまり、recordsはPersonモデルのオブジェクトであることがわかります。

def find_each(start: nil, finish: nil, batch_size: 1000, error_on_ignore: nil, order: :asc, &block)
  if block_given?
    find_in_batches(start: start, finish: finish, batch_size: batch_size, error_on_ignore: error_on_ignore, order: order) do |records|
      records.each(&block)
    end
  else
    ...
  end
end

考察

find_in_batchesはバッチ処理なので、recordsはバッチ単位ごとのPersonモデルのオブジェクトの配列であるはずです。
find_in_batchesの内部ではrecordsを渡すyieldを実行しており、第一引数としてPersonモデルのオブジェクトがどこかで渡されるはず。
startとfinishでは処理の位置を指定しているがプライマリキーを基準にしていそうです。

find_in_batchesメソッド

引数をそのままリレーして、in_batchesを実行しています。ブロックの内部ではfind_eachで考察した通り、yieldが実行されており、第一引数としてブロック引数のArrayオブジェクトを返しています。

def find_in_batches(start: nil, finish: nil, batch_size: 1000, error_on_ignore: nil, order: :asc)
  ...

  in_batches(of: batch_size, start: start, finish: finish, load: true, error_on_ignore: error_on_ignore, order: order) do |batch|
    yield batch.to_a
  end
end

in_batchesメソッド

find_eachの核となるメソッドです。まとまりごとに一つずつ見ていきましょう。

def in_batches(of: 1000, start: nil, finish: nil, load: false, error_on_ignore: nil, order: :asc)
  relation = self

  ...

  batch_limit = of
  if limit_value
    remaining   = limit_value
    batch_limit = remaining if remaining < batch_limit
  end

  relation = relation.reorder(batch_order(order)).limit(batch_limit)
  relation = apply_limits(relation, start, finish, order)
  relation.skip_query_cache! # Retaining the results in the query cache would undermine the point of batching
  batch_relation = relation

  loop do
    if load
      records = batch_relation.records
      ids = records.map(&:id)
      yielded_relation = where(primary_key => ids)
      yielded_relation.load_records(records)
    else
      ...
    end

    break if ids.empty?

    primary_key_offset = ids.last
    raise ArgumentError.new("Primary key not included in the custom select clause") unless primary_key_offset

    yield yielded_relation

    break if ids.length < batch_limit

    ...

    batch_relation = relation.where(
      predicate_builder[primary_key, primary_key_offset, order == :desc ? :lt : :gt]
    )
  end
end

relation

relationにはfind_eachを呼び出したレシーバ（今回の例ではPerson）がselfとして代入されます。

  relation = self

batch_limitとremaining

batch_limitにはfind_eachからリレーされてきた1000が代入されます。limit_valueについてはlimitの実装を読むとわかりますが、limitメソッドの引数が代入されています。limitが指定されている場合、batch_limitはその上限に調整されます。

今回のケースではlimitをチェーンしたActiveRecordのオブジェクトではないので条件文は実行されません。

  batch_limit = of
  if limit_value
    remaining   = limit_value
    batch_limit = remaining if remaining < batch_limit
  end

batch_relation

先ほどrelationを定義しましたが、このままではbatch単位のオブジェクトになっていないのでrelationをbatch単位にしています。apply_limitsの実装の実装を読むと、limitをかけているだけではなく、プライマリキーについてsortも走っていることがわかります。sortを走らせているのはwhere条件でプライマリキーの範囲をstartとfinishで指定し、limitの処理で適切な数のレコードを取得するために必要で、これは確かにapply_limitsの責務です。

クエリキャッシュを無効にしているのはキャッシュをしてしまうと、batch単位ごとに処理することでメモリ効率化を図るというbatch処理の意義が損なわれてしまうためです。

  relation = relation.reorder(batch_order(order)).limit(batch_limit)
  relation = apply_limits(relation, start, finish, order)
  relation.skip_query_cache! # Retaining the results in the query cache would undermine the point of batching
  batch_relation = relation

loop内部

ここでbatch処理全体を実行しています。注目するべきポイントは3つです。

1つ目はyieldです。yielded_relation(今回のケースではPerson)がin_batchesへ渡しているブロック引数に代入されます。
2つ目はbreakの条件です。ループごとにbatch単位のidsを取得しているのですが、idsが空(処理するべきbatch対象がない)場合とidsがbatchサイズより小さい場合です。
3つ目はprimary_key_offsetです。これでbatch処理のoffsetを指定しています。

  loop do
    if load
      records = batch_relation.records
      ids = records.map(&:id)
      yielded_relation = where(primary_key => ids)
      yielded_relation.load_records(records)
    else
      ...
    end

    break if ids.empty?

    primary_key_offset = ids.last
    raise ArgumentError.new("Primary key not included in the custom select clause") unless primary_key_offset

    yield yielded_relation

    break if ids.length < batch_limit

    ...

    batch_relation = relation.where(
      predicate_builder[primary_key, primary_key_offset, order == :desc ? :lt : :gt]
    )
  end

最後に

メモリ節約によく利用されるfind_eachについて見てきました。内部実装を読むことで、Rubyの知らないメソッドに出会えたり、Railsにおけるインターフェースの思想に触れられて設計力が向上するのでおすすめです。

参考