Запросы

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