Method rangeQuery

pj_datastore/src/reader.cpp:154–168

}

Expected<RangeCursor> DataReader::rangeQuery(const QueryRange& range) const {
  // Take the read-lock BEFORE the lookup (a concurrent createTopic rehash could
  // otherwise move storages mid-lookup) and MOVE it into the cursor, which holds
  // it for its lazy-iteration lifetime.
  auto lock = engine_.lockEngine();
  const TopicStorage* storage = engine_.getTopicStorage(range.topic_id);
  if (storage == nullptr) {
    return PJ::unexpected(fmt::format("Topic {} not found", range.topic_id));
  }
  // Raise the lower bound to the topic's retention floor so a straddling chunk's
  // logically-evicted rows are never returned (RangeCursor skips rows < t_min,
  // and yields nothing when t_min > t_max — i.e. a window entirely below floor).
  const Timestamp t_min = std::max(range.t_min, storage->retentionFloor());
  return RangeCursor(storage->sealedChunks(), t_min, range.t_max, std::move(lock));
}

PJ::Expected<std::optional<MaterializedSample>> DataReader::latestAt(const QueryPoint& point) const {
  auto lock = engine_.lockEngine();