<commit_before>#include "vast/query.h"

#include <cppa/cppa.hpp>
#include "vast/event.h"
#include "vast/logger.h"

namespace vast {

namespace {

struct cow_segment_less_than
{
  bool operator()(cow<segment> const& x, cow<segment> const& y) const
  {
    return x->base() < y->base();
  }
};

cow_segment_less_than segment_lt;

} // namespace <anonymous>

query::query(expr::ast ast, std::function<void(event)> fn)
  : ast_{std::move(ast)},
    fn_{fn}
{
  processed_ = bitstream_type{};
  extracted_ = bitstream_type{};
}

void query::update(bitstream result)
{
  assert(result);

  if (hits_)
    hits_ |= result;
  else
    hits_ = std::move(result);

  unprocessed_ = hits_ - processed_;
}

bool query::add(cow<segment> s)
{
  auto i = std::lower_bound(segments_.begin(), segments_.end(), s, segment_lt);
  if (! (i == segments_.end() || segment_lt(s, *i)))
    return false;

  segments_.emplace(i, s);
  return true;
}

size_t query::consolidate(size_t before, size_t after)
{
  auto i = std::lower_bound(
      segments_.begin(), segments_.end(), current_, segment_lt);

  assert(i != segments_.end());
  size_t earlier = std::distance(segments_.begin(), i);
  size_t later = std::distance(i, segments_.end()) - 1;
  size_t n = 0;

  if (earlier > before)
  {
    auto purge = earlier - before;
    VAST_LOG_DEBUG("query removes " << purge << " segments before current");
    segments_.erase(segments_.begin(), segments_.begin() + purge);
    n = purge;
  }

  if (later > after)
  {
    auto purge = later - after;
    VAST_LOG_DEBUG("query removes " << purge << " segments after current");
    segments_.erase(segments_.end() - purge, segments_.end());
    n += purge;
  }

  return n;
}

optional<size_t> query::extract()
{
  if (! unprocessed_)
  {
    VAST_LOG_DEBUG("query has no unprocessed hits available");
    return {};
  }

  if (! reader_ || ! current_->contains(reader_->position()))
  {
    auto last = unprocessed_.find_last();
    if (last == bitstream::npos)
      return {};

    auto found = false;
    for (auto i = segments_.rbegin(); i != segments_.rend(); ++i)
    {
      if ((**i).contains(last))
      {
        found = true;
        current_ = *i;
        break;
      }
    }

    if (! found)
    {
      VAST_LOG_DEBUG("query has no segment for id " << last);
      return {};
    }

    reader_ = make_unique<segment::reader>(&current_.read());
    VAST_LOG_DEBUG("query instantiates reader for segment " <<
                   current_->id() << " covering [" << current_->base() << ','
                   << current_->base() + current_->events() << ')');

    masked_ = bitstream_type{};
    masked_.append(current_->base(), false);
    masked_.append(current_->events(), true);
    masked_ &= unprocessed_;
  }

  size_t n = 0;
  for (auto id : masked_)
  {
    auto e = reader_->read(id);
    if (! e)
    {
      VAST_LOG_ERROR("query failed to extract event " << id);
      break;
    }

    extracted_.append(e->id() - extracted_.size(), false);
    extracted_.push_back(true);

    if (evaluate(ast_, *e).get<bool>())
    {
      fn_(std::move(*e));
      ++n;
    }
    else
    {
      VAST_LOG_WARN("query " << ast_ << " ignores false positive: " << *e);
    }
  }

  if (! extracted_.empty())
  {
    processed_ |= extracted_;
    unprocessed_ -= extracted_;
    masked_ -= extracted_;
    extracted_ = bitstream_type{};
  }

  if (n == 0)
    VAST_LOG_WARN("query could not extract a single result from segment " <<
                  current_->id());

  return n;<commit_msg>Fix to reduce unnecessary warning.<commit_after>#include "vast/query.h"

#include <cppa/cppa.hpp>
#include "vast/event.h"
#include "vast/logger.h"

namespace vast {

namespace {

struct cow_segment_less_than
{
  bool operator()(cow<segment> const& x, cow<segment> const& y) const
  {
    return x->base() < y->base();
  }
};

cow_segment_less_than segment_lt;

} // namespace <anonymous>

query::query(expr::ast ast, std::function<void(event)> fn)
  : ast_{std::move(ast)},
    fn_{fn}
{
  processed_ = bitstream_type{};
  extracted_ = bitstream_type{};
}

void query::update(bitstream result)
{
  assert(result);

  if (hits_)
    hits_ |= result;
  else
    hits_ = std::move(result);

  unprocessed_ = hits_ - processed_;
}

bool query::add(cow<segment> s)
{
  auto i = std::lower_bound(segments_.begin(), segments_.end(), s, segment_lt);
  if (! (i == segments_.end() || segment_lt(s, *i)))
    return false;

  segments_.emplace(i, s);
  return true;
}

size_t query::consolidate(size_t before, size_t after)
{
  auto i = std::lower_bound(
      segments_.begin(), segments_.end(), current_, segment_lt);

  assert(i != segments_.end());
  size_t earlier = std::distance(segments_.begin(), i);
  size_t later = std::distance(i, segments_.end()) - 1;
  size_t n = 0;

  if (earlier > before)
  {
    auto purge = earlier - before;
    VAST_LOG_DEBUG("query removes " << purge << " segments before current");
    segments_.erase(segments_.begin(), segments_.begin() + purge);
    n = purge;
  }

  if (later > after)
  {
    auto purge = later - after;
    VAST_LOG_DEBUG("query removes " << purge << " segments after current");
    segments_.erase(segments_.end() - purge, segments_.end());
    n += purge;
  }

  return n;
}

optional<size_t> query::extract()
{
  if (! unprocessed_)
  {
    VAST_LOG_DEBUG("query has no unprocessed hits available");
    return {};
  }

  if (! reader_ || ! current_->contains(reader_->position()))
  {
    auto last = unprocessed_.find_last();
    if (last == bitstream::npos)
      return {};

    auto found = false;
    for (auto i = segments_.rbegin(); i != segments_.rend(); ++i)
    {
      if ((**i).contains(last))
      {
        found = true;
        current_ = *i;
        break;
      }
    }

    if (! found)
    {
      VAST_LOG_DEBUG("query has no segment for id " << last);
      return {};
    }

    reader_ = make_unique<segment::reader>(&current_.read());
    VAST_LOG_DEBUG("query instantiates reader for segment " <<
                   current_->id() << " covering [" << current_->base() << ','
                   << current_->base() + current_->events() << ')');

    masked_ = bitstream_type{};
    masked_.append(current_->base(), false);
    masked_.append(current_->events(), true);
    masked_ &= unprocessed_;
  }

  if (masked_.empty())
    return {};

  size_t n = 0;
  for (auto id : masked_)
  {
    auto e = reader_->read(id);
    if (! e)
    {
      VAST_LOG_ERROR("query failed to extract event " << id);
      break;
    }

    extracted_.append(e->id() - extracted_.size(), false);
    extracted_.push_back(true);

    if (evaluate(ast_, *e).get<bool>())
    {
      fn_(std::move(*e));
      ++n;
    }
    else
    {
      VAST_LOG_WARN("query " << ast_ << " ignores false positive: " << *e);
    }
  }

  if (! extracted_.empty())
  {
    processed_ |= extracted_;
    unprocessed_ -= extracted_;
    masked_ -= extracted_;
    extracted_ = bitstream_type{};
  }

  if (n == 0)
    VAST_LOG_WARN("query could not find a single result in segment " <<
                  current_->id());

  return n;
{'fraction_non_alphanumeric': 0.16509500351864884, 'fraction_numerical': 0.0011259676284306826, 'mean_word_length': 2.9001097694840836, 'pattern_counts': {'":': 0, '<': 87, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3577951', 'n_tokens_mistral': 2449, 'n_tokens_neox': 2368, 'n_words': 534}