Запити

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 вони також уникають зайвої роботи: сегменти, діапазон ідентифікаторів яких виходить за межі інтервалу, навіть не відкриваються, а розріджені індекси для кожного сегмента знаходять перший відповідний запис, майже не скануючи попередні.

where_equal() порівнює поле верхнього рівня в даних із === і приймає скалярні значення або null. Фільтр null знаходить записи, у яких поле присутнє й має значення null; записи без цього поля не відповідають умові.

Сторінкова навігація за курсором

Ідентифікатори сортуються за часом створення, тому останній ідентифікатор сторінки стає курсором для наступної:

$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 місце — це ідентифікатор запису, а для SegmentedLog — segment-file:byte-offset. repair() призначений саме для очищення пошкоджених записів; див. CLI.