クエリ

RecordQueryはstream()と組み合わせて使うストリーミングフィルターです。追記型のアクセスパターンに対応し、カーソル以降の再開、時間範囲の読み取り、フィールドの等価条件による絞り込み、件数制限による終了ができます。より複雑な条件、並べ替え、インデックスを使った検索にはクエリビルダーを使ってください。ストリームに対する通常のforeachでできる操作だけなら、RecordQueryのほうが低コストです。

RecordQueryは不変です。各修飾メソッドはクエリの複製を返します。

$query = Storh\RecordQuery::all()
    ->after($lastSeenId)
    ->time_range_ms($fromMs, $untilMs)
    ->where_equal('kind', 'page')
    ->limit(100);

foreach ($store->stream($query) as $record) {
    // Records are yielded lazily.
}

after()とtime_range_ms()はUUID順で処理されるため、レコードデータにタイムスタンプフィールドは不要です。SegmentedLogでは処理も省けます。IDの範囲が時間範囲外のセグメントは開かず、セグメントごとのスパースインデックスで該当レコードの近くまでシークするため、先頭からスキャンする必要はありません。

where_equal()はトップレベルのデータフィールドを===と比較し、スカラー値またはnullを受け付けます。nullフィルターは、フィールドが存在して値がnullのレコードに一致します。フィールドがないレコードには一致しません。

カーソルページネーション

IDは作成時刻順に並ぶため、あるページの最後のIDを次のページのカーソルにできます。

$page = iterator_to_array($store->stream(
    Storh\RecordQuery::all()->after($lastSeenId)->limit(100)
), false);

$lastSeenId = [] === $page ? $lastSeenId : end($page)->id();

破損レコードを含めた読み取り

既定では、スキャン中に読み取れないレコードが見つかると、そこで例外が発生します。スキャンを続けて失敗を報告するには、代わりにcontinue_on_error()を使います。

$query = Storh\RecordQuery::all()->continue_on_error(
    static function (string $location, Throwable $error): void {
        error_log($location . ': ' . $error->getMessage());
    },
);

DocStoreとsegment-file:byte-offsetでは位置がレコードIDを示し、SegmentedLogではrepair()を示します。破損レコードを実際にクリーンアップするには、CLIを使用してください。