text:
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "kudu/common/generic_iterators.h"

#include <sys/types.h>

#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <iterator>
#include <memory>
#include <mutex>
#include <numeric>
#include <ostream>
#include <string>
#include <unordered_map>
#include <utility>

#include <boost/heap/skew_heap.hpp>
#include <boost/intrusive/list.hpp>
#include <boost/intrusive/list_hook.hpp>
#include <boost/iterator/iterator_facade.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>

#include "kudu/common/column_materialization_context.h"
#include "kudu/common/column_predicate.h"
#include "kudu/common/columnblock.h"
#include "kudu/common/common.pb.h"
#include "kudu/common/iterator.h"
#include "kudu/common/iterator_stats.h"
#include "kudu/common/row.h"
#include "kudu/common/rowblock.h"
#include "kudu/common/scan_spec.h"
#include "kudu/common/schema.h"
#include "kudu/gutil/casts.h"
#include "kudu/gutil/macros.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/locks.h"
#include "kudu/util/memory/arena.h"
#include "kudu/util/object_pool.h"

namespace boost {
namespace heap {
template <class T> struct compare;
}  // namespace heap
}  // namespace boost

using std::deque;
using std::get;
using std::pair;
using std::sort;
using std::string;
using std::unique_ptr;
using std::vector;
using strings::Substitute;

DEFINE_bool(materializing_iterator_do_pushdown, true,
            "Should MaterializingIterator do predicate pushdown");
TAG_FLAG(materializing_iterator_do_pushdown, hidden);
DEFINE_bool(materializing_iterator_decoder_eval, true,
            "Should MaterializingIterator do decoder-level evaluation");
TAG_FLAG(materializing_iterator_decoder_eval, hidden);
TAG_FLAG(materializing_iterator_decoder_eval, runtime);

namespace kudu {
namespace {
void AddIterStats(const RowwiseIterator& iter,
                  vector<IteratorStats>* stats) {
  vector<IteratorStats> iter_stats;
  iter.GetIteratorStats(&iter_stats);
  DCHECK_EQ(stats->size(), iter_stats.size());
  for (int i = 0; i < iter_stats.size(); i++) {
    (*stats)[i] += iter_stats[i];
  }
}
} // anonymous namespace

////////////////////////////////////////////////////////////
// MergeIterator
////////////////////////////////////////////////////////////

// This is sized to a power of 2 to improve BitmapCopy performance when copying
// a RowBlock.
//
// TODO(todd): this should be sized by # bytes, not # rows.
static const int kMergeRowBuffer = 1024;

// MergeIterState wraps a RowwiseIterator for use by the MergeIterator.
class MergeIterState : public boost::intrusive::list_base_hook<> {
 public:
  explicit MergeIterState(IterWithBounds iwb)
      : iwb_(std::move(iwb)),
        arena_(1024),
        next_row_idx_(0)
  {}

  // Fetches the next row from the iterator's current block, or the iterator's
  // absolute lower bound if a block has not yet been pulled.
  //
  // Does not advance the iterator.
  const RowBlockRow& next_row() const {
    if (read_block_) {
      DCHECK_LT(next_row_idx_, read_block_->nrows());
      return next_row_;
    }
    DCHECK(decoded_bounds_);
    return decoded_bounds_->lower;
  }

  // Fetches the last row from the iterator's current block, or the iterator's
  // absolute upper bound if a block has not yet been pulled.
  //
  // Does not advance the iterator.
  const RowBlockRow& last_row() const {
    if (read_block_) {
      return last_row_;
    }
    DCHECK(decoded_bounds_);
    return decoded_bounds_->upper;
  }

  // Finishes construction of the MergeIterState by decoding the bounds if they
  // exist. If not, we have to pull a block immediately: after Init() is
  // finished it must be safe to call next_row() and last_row().
  //
  // Decoded bound allocations are done against 'decoded_bounds_arena'.
  Status Init(Arena* decoded_bounds_arena) {
    DCHECK(!read_block_);

    if (iwb_.encoded_bounds) {
      decoded_bounds_.emplace(&schema(), decoded_bounds_arena);
      decoded_bounds_->lower = decoded_bounds_->block.row(0);
      decoded_bounds_->upper = decoded_bounds_->block.row(1);
      RETURN_NOT_OK(schema().DecodeRowKey(
          iwb_.encoded_bounds->first, &decoded_bounds_->lower, decoded_bounds_arena));
      RETURN_NOT_OK(schema().DecodeRowKey(
          iwb_.encoded_bounds->second, &decoded_bounds_->upper, decoded_bounds_arena));
    } else {
      RETURN_NOT_OK(PullNextBlock());
    }

    return Status::OK();
  }

  // Returns true if the underlying iterator is fully exhausted.
  bool IsFullyExhausted() const {
    return (!read_block_ || read_block_->nrows() == 0) && !iwb_.iter->HasNext();
  }

  // Advances to the next row in the underlying iterator.
  //
  // If successful, 'pulled_new_block' is true if this block was exhausted and a
  // new block was pulled from the underlying iterator.
  Status Advance(size_t num_rows, bool* pulled_new_block);

  // Adds statistics about the underlying iterator to the given vector.
  void AddStats(vector<IteratorStats>* stats) const {
    AddIterStats(*iwb_.iter, stats);
  }

  // Returns the number of rows remaining in the current block.
  size_t remaining_in_block() const {
    DCHECK(read_block_);
    return read_block_->nrows() - next_row_idx_;
  }

  // Returns the schema from the underlying iterator.
  const Schema& schema() const {
    return iwb_.iter->schema();
  }

  // Pulls the next block from the underlying iterator.
  Status PullNextBlock();

  // Copies as many rows as possible from the current block of buffered rows to
  // 'dst' (starting at 'dst_offset').
  //
  // If successful, 'num_rows_copied' will be set to the number of rows copied.
  Status CopyBlock(RowBlock* dst, size_t dst_offset, size_t* num_rows_copied);

  // Returns true if the current block in the underlying iterator is exhausted.
  bool IsBlockExhausted() const {
    return !read_block_ || read_block_->nrows() == next_row_idx_;
  }

  string ToString() const {
    return Substitute("[$0,$1]: $2",
                      schema().DebugRowKey(next_row()),
                      schema().DebugRowKey(last_row()),
                      iwb_.iter->ToString());
  }

 private:
  // The iterator (and optional bounds) whose rows are to be merged with other
  // iterators.
  //
  // Must already be Init'ed at MergeIterState construction time.
  IterWithBounds iwb_;

  // Allocates memory for read_block_.
  Arena arena_;

  // Optional rowset bounds, decoded during Init().
  struct DecodedBounds {
    // 'block' must be constructed immediately; the bounds themselves can be
    // initialized later.
    DecodedBounds(const Schema* schema, Arena* arena)
        : block(schema, /*nrows=*/2, arena) {}

    RowBlock block;
    RowBlockRow lower;
    RowBlockRow upper;
  };
  boost::optional<DecodedBounds> decoded_bounds_;

  // Current block of buffered rows from the iterator.
  //
  // The memory backing the rows was allocated out of the arena.
  unique_ptr<RowBlock> read_block_;

  // The row currently pointed to by the iterator.
  RowBlockRow next_row_;

  // The last row available in read_block_.
  RowBlockRow last_row_;

  // The index of the row currently pointed to by the iterator. Guaranteed to be
  // a selected row.
  size_t next_row_idx_;

  DISALLOW_COPY_AND_ASSIGN(MergeIterState);
};

Status MergeIterState::Advance(size_t num_rows, bool* pulled_new_block) {
  DCHECK_GE(read_block_->nrows(), next_row_idx_ + num_rows);

  next_row_idx_ += num_rows;

  // If advancing didn't exhaust this block outright, find the next selected row.
  size_t idx;
  if (!IsBlockExhausted() &&
      read_block_->selection_vector()->FindFirstRowSelected(next_row_idx_, &idx)) {
    next_row_idx_ = idx;
    next_row_.Reset(read_block_.get(), next_row_idx_);
    *pulled_new_block = false;
    return Status::OK();
  }

  // We either exhausted the block outright, or all subsequent rows were
  // deselected. Either way, we need to pull the next block.
  next_row_idx_ = read_block_->nrows();
  arena_.Reset();
  RETURN_NOT_OK(PullNextBlock());
  *pulled_new_block = true;
  return Status::OK();
}

Status MergeIterState::PullNextBlock() {
  CHECK(IsBlockExhausted())
      << "should not pull next block until current block is exhausted";

  if (!read_block_) {
    read_block_.reset(new RowBlock(&schema(), kMergeRowBuffer, &arena_));
  }
  while (iwb_.iter->HasNext()) {
    RETURN_NOT_OK(iwb_.iter->NextBlock(read_block_.get()));

    SelectionVector *selection = read_block_->selection_vector();
    DCHECK_EQ(selection->nrows(), read_block_->nrows());
    DCHECK_LE(selection->CountSelected(), read_block_->nrows());
    size_t rows_valid = selection->CountSelected();
    VLOG(2) << Substitute("$0/$1 rows selected", rows_valid, read_block_->nrows());
    if (rows_valid == 0) {
      // Short-circuit: this block is entirely unselected and can be skipped.
      continue;
    }

    // Seek next_row_ and last_row_ to the first and last selected rows
    // respectively (which could be identical).

    CHECK(selection->FindFirstRowSelected(0, &next_row_idx_));
    next_row_.Reset(read_block_.get(), next_row_idx_);

    // We use a signed size_t type to avoid underflowing when finding last_row_.
    //
    // TODO(adar): this can be simplified if there was a BitmapFindLastSet().
    for (ssize_t row_idx = read_block_->nrows() - 1; row_idx >= 0; row_idx--) {
      if (selection->IsRowSelected(row_idx)) {
        last_row_.Reset(read_block_.get(), row_idx);
        VLOG(1) << "Pulled new block: " << ToString();
        return Status::OK();
      }
    }

    LOG(FATAL) << "unreachable code"; // guaranteed by the short-circuit above
  }

  // The underlying iterator is fully exhausted.
  next_row_idx_ = 0;
  read_block_.reset();
  return Status::OK();
}

Status MergeIterState::CopyBlock(RowBlock* dst, size_t dst_offset,
                                 size_t* num_rows_copied) {
  DCHECK(read_block_);
  DCHECK(!IsBlockExhausted());

  size_t num_rows_to_copy = std::min(remaining_in_block(),
                                     dst->nrows() - dst_offset);
  VLOG(3) << Substitute(
      "Copying $0 rows from RowBlock (s:$1,o:$2) to RowBlock (s:$3,o:$4): $5",
      num_rows_to_copy, read_block_->nrows(), next_row_idx_, dst->nrows(),
      dst_offset, ToString());
  RETURN_NOT_OK(read_block_->CopyTo(dst, next_row_idx_,
                                    dst_offset, num_rows_to_copy));
  *num_rows_copied = num_rows_to_copy;
  return Status::OK();
}

// An iterator which merges the results of other iterators, comparing
// based on keys.
//
// Three different heaps are used to optimize the merge process. To explain how
// it works, let's start with an explanation of a traditional heap-based merge:
// there exist N sorted lists of elements and the goal is to produce a single
// sorted list containing all of the elements.
//
// To begin:
// - For each list, peek the first element into a per-list buffer.
// - Add all of the lists to a min-heap ordered on the per-list buffers. This
//   means that the heap's top-most entry (accessible in O(1) time) will be the
//   list containing the smallest not-yet-consumed element.
//
// To perform the merge, loop while the min-heap isn't empty:
// - Pop the top-most list from the min-heap.
// - Copy that list's peeked element to the output.
// - Peek the list's next element. If the list is empty, discard it.
// - If the list has more elements, push it back into the min-heap.
//
// This algorithm runs in O(n log n) time and is generally superior to a naive
// O(n^2) merge. However, it requires peeked elements to remain resident in
// memory during the merge.
//
// The MergeIterator's sub-iterators operate much like the lists described
// above: elements correspond to rows and sorting is based on rows' primary
// keys. However, there are several important differences that open the door for
// further optimization:
// 1.  Each sub-iterator corresponds to a Kudu rowset, and DiskRowSets' smallest
//     and largest possible primary keys (i.e. bounds) are known ahead of time.
// 2.  When iterating on DiskRowSets, peeking even one row means decoding a page
//     of columnar data for each projected column. This decoded data remains
//     resident in memory so that we needn't repeat the decoding for each row,
//     but it means we have a strong motivation to minimize the set of peeked
//     sub-iterators in order to minimize memory consumption.
// 2a. Related to #2, there's little reason not to peek more than row at a time,
//     since the predominant source of memory usage is in the decoded pages
//     rather than the buffered rows.
//
// The aggressive peeking allows us to tweak the list model slightly: instead
// of treating sub-iterators as continuous sequences of single rows, we can
// think of each as a sequence of discrete row "runs". Each run has a lower
// bound (the key of the first row in the run) and an upper bound (the key of
// the last row). One run in each sub-iterator is NEXT in that its rows have
// been peeked and are resident in memory. When we draw rows from the
// sub-iterator, we'll draw them from this run. We can use the bounds to
// establish overlapping relationships between runs across sub-iterators. In
// theory, the less overlap, the fewer runs need to be considered when merging.
//
// To exploit this, we need to formally define the concept of a "merge window".
// The window describes, at any given time, the key space interval where we
// expect to find the row with the smallest key. A sub-iterator whose NEXT
// overlaps with the merge window is said to be actively participating in the merge.
//
// The merge window is defined as follows:
// 1.  The window's start is the smallest lower bound of all sub-iterators. We
//     refer to the sub-iterator that owns this lower bound as LOW.
// 2.  The window’s end is the smallest upper bound of all sub-iterators whose
//     lower bounds are less than or equal to LOW's upper bound.
// 2a. The window's end could be LOW's upper bound itself, if it is the smallest
//     upper bound, but this isn't necessarily the case.
// 3.  The merge window's dimensions change as the merge proceeds, though it
//     only ever moves "to the right" (i.e. the window start/end only increase).
//
// Armed with the merge window concept, we can bifurcate the sub-iterators into
// a set whose NEXTs overlap with the merge window and a set whose NEXTs do not.
// We store the first set in a HOT min-heap (ordered by each's NEXT's lower
// bound). Now the merge steady state resembles that of a traditional heap-based
// merge: the top-most sub-iterator is popped from HOT, the lower bound is
// copied to the output and advanced, and the sub-iterator is pushed back to
// HOT. The bifurcation hasn't yielded any algorithmic improvements, but the
// more non-overlapped the input (i.e. the more compacted the tablet), the fewer
// sub-iterators will be in HOT and thus the fewer comparisons and heap motion
// will take place.
//
// How do sub-iterators move between the two sets? At the outset, we examine all
// sub-iterators to find the initial merge window and bifurcate accordingly. In
// the steady state, we need to move to HOT whenever the end of the merge window
// moves; that's a sign that the window may now overlap with a NEXT belonging to
// a sub-iterator in the second set. The end of the merge window moves when a
// sub-iterator is fully exhausted (i.e. all rows have been copied to the
// output), or when a sub-iterator finishes its NEXT and needs to peek again.
//
// But which sub-iterators should be moved? To answer this question efficiently,
// we need two more heaps:
// - COLD: a min-heap for sub-iterators in the second set, ordered by each's
//   NEXT's lower bound. At any given time, the NEXT belonging to the top-most
//   sub-iterator in COLD is nearest the merge window.
// - HOTMAXES: a min-heap for keys. Each entry corresponds to a sub-iterator
//   present in HOT, and specifically, to its NEXT's upper bound. At any given
//   time, the top-most key in HOTMAXES represents the end of the merge window.
//
// When the merge window's end has moved and we need to refill HOT, the top-most
// sub-iterator in COLD is the best candidate. To figure out whether it should
// be moved, we compare its NEXT's lower bound against the top-most key in
// HOTMAXES: if the lower bound is less than or equal to the key, we move the
// sub-iterator from COLD to HOT. On the flip side, when a sub-iterator from HOT
// finishes its NEXT and peeks again, we also need to check whether it has
// exited the merge window. The approach is similar: if its NEXT's lower bound
// is greater than the top-most key in HOTMAXES, it's time to move it to COLD.
//
// There's one more piece to this puzzle: the absolute bounds that are known
// ahead of time are used as stand-ins for NEXT's lower and upper bounds. This
// helps defer peeking for as long as possible, at least until the sub-iterator
// moves from COLD to HOT. After that, peeked memory must remain resident until
// the sub-iterator is fully exhausted.
//
// For another description of this algorithm including pseudocode, see
// https://docs.google.com/document/d/1uP0ubjM6ulnKVCRrXtwT_dqrTWjF9tlFSRk0JN2e_O0/edit#
class MergeIterator : public RowwiseIterator {
 public:
  // Constructs a MergeIterator of the given iterators.
  //
  // The iterators must have matching schemas and should not yet be initialized.
  //
  // Note: the iterators must be constructed using a projection that includes
  // all key columns; otherwise a CHECK will fire at initialization time.
  MergeIterator(MergeIteratorOptions opts, vector<IterWithBounds> iters);

  virtual ~MergeIterator();

  // The passed-in iterators should be already initialized.
  Status Init(ScanSpec *spec) OVERRIDE;

  virtual bool HasNext() const OVERRIDE;

  virtual string ToString() const OVERRIDE;

  virtual const Schema& schema() const OVERRIDE;

  virtual void GetIteratorStats(vector<IteratorStats>* stats) const OVERRIDE;

  virtual Status NextBlock(RowBlock* dst) OVERRIDE;

 private:
  // Materializes as much of the next block's worth of data into 'dst' at offset
  // 'dst_row_idx' as possible. Only permitted when there's only one
  // sub-iterator in the hot heap.
  //
  // On success, the selection vector in 'dst' and 'dst_row_idx' are both updated.
  Status MaterializeBlock(RowBlock* dst, size_t* dst_row_idx);

  // Finds the next row and materializes it into 'dst' at offset 'dst_row_idx'.
  //
  // On success, the selection vector in 'dst' and 'dst_row_idx' are both updated.
  Status MaterializeOneRow(RowBlock* dst, size_t* dst_row_idx);

  // Calls Init() on all of sub-iterators, wrapping them in predicate evaluating
  // iterators if necessary and setting up additional per-iterator bookkeeping.
  Status InitSubIterators(ScanSpec *spec);

  // Advances 'state' by 'num_rows_to_advance', destroying it and/or updating
  // the three heaps if necessary.
  Status AdvanceAndReheap(MergeIterState* state, size_t num_rows_to_advance);

  // Moves sub-iterators from cold_ to hot_ if they now overlap with the merge
  // window. Should be called whenever the merge window moves.
  Status RefillHotHeap();

  // Destroys a fully exhausted sub-iterator.
  void DestroySubIterator(MergeIterState* state);

  const MergeIteratorOptions opts_;

  // Initialized during Init.
  unique_ptr<Schema> schema_;

  bool initted_;

  // Holds the sub-iterators until Init is called, at which point this is
  // cleared. This is required because we can't create a MergeIterState of an
  // uninitialized sub-iterator.
  vector<IterWithBounds> orig_iters_;

  // See UnionIterator::iters_lock_ for details on locking. This follows the same
  // pattern.
  mutable rw_spinlock states_lock_;
  boost::intrusive::list<MergeIterState> states_; // NOLINT(*)

  // Statistics (keyed by projection column index) accumulated so far by any
  // fully-consumed sub-iterators.
  vector<IteratorStats> finished_iter_stats_by_col_;

  // The number of iterators, used by ToString().
  const int num_orig_iters_;

  // When the underlying iterators are initialized, each needs its own
  // copy of the scan spec in order to do its own pushdown calculations, etc.
  // The copies are allocated from this pool so they can be automatically freed
  // when the UnionIterator goes out of scope.
  ObjectPool<ScanSpec> scan_spec_copies_;

  // Arena dedicated to MergeIterState bounds decoding.
  //
  // Each MergeIterState has an arena for buffered row data, but it is reset
  // every time a new block is pulled. This single arena ensures that a
  // MergeIterState's decoded bounds remain allocated for its lifetime.
  Arena decoded_bounds_arena_;

  // Min-heap that orders rows by their keys. A call to top() will yield the row
  // with the smallest key.
  struct RowComparator {
    bool operator()(const RowBlockRow& a, const RowBlockRow& b) const {
      // This is counter-intuitive, but it's because boost::heap defaults to
      // a max-heap; the comparator must be inverted to yield a min-heap.
      return a.schema()->Compare(a, b) > 0;
    }
  };
  typedef boost::heap::skew_heap<
    RowBlockRow, boost::heap::compare<RowComparator>> RowMinHeap;

  // Min-heap that orders sub-iterators by their next row key. A call to top()
  // will yield the sub-iterator with the smallest next row key.
  struct MergeIterStateComparator {
    bool operator()(const MergeIterState* a, const MergeIterState* b) const {
      // This is counter-intuitive, but it's because boost::heap defaults to
      // a max-heap; the comparator must be inverted to yield a min-heap.
      return a->schema().Compare(a->next_row(), b->next_row()) > 0;
    }
  };
  typedef boost::heap::skew_heap<
    MergeIterState*, boost::heap::compare<MergeIterStateComparator>> MergeStateMinHeap;

  // The three heaps as described in the algorithm above.
  //
  // Note that the heaps do not "own" any of the objects they contain:
  // - The MergeIterStates in hot_ and cold_ are owned by states_.
  // - The data backing the rows in hotmaxes_ is owned by all states' read_block_.
  //
  // Care must be taken to remove entries from the heaps when the corresponding
  // objects are destroyed.
  //
  // Boost offers a variety of different heap data structures[1]. Perf testing
  // via generic_iterators-test (TestMerge and TestMergeNonOverlapping with
  // num_iters=10, num_lists=1000, and num_rows=1000) shows that while all heaps
  // perform more or less equally well for non-overlapping input, skew heaps
  // outperform the rest for overlapping input. Basic priority queues (i.e.
  // boost::heap::priority_queue and std::priority_queue) were excluded as they
  // do not offer ordered iteration.
  //
  // 1. https://www.boost.org/doc/libs/1_69_0/doc/html/heap/data_structures.html
  MergeStateMinHeap hot_;
  MergeStateMinHeap cold_;
  RowMinHeap hotmaxes_;
};

MergeIterator::MergeIterator(MergeIteratorOptions opts,
                             vector<IterWithBounds> iters)
    : opts_(opts),
      initted_(false),
      orig_iters_(std::move(iters)),
      num_orig_iters_(orig_iters_.size()),
      decoded_bounds_arena_(1024) {
  CHECK_GT(orig_iters_.size(), 0);
}

MergeIterator::~MergeIterator() {
  states_.clear_and_dispose([](MergeIterState* s) { delete s; });
}

Status MergeIterator::Init(ScanSpec *spec) {
  CHECK(!initted_);

  // Initialize the iterators and the per-iterator merge states.
  //
  // When this method finishes, orig_iters_ has been cleared and states_ has
  // been populated.
  RETURN_NOT_OK(InitSubIterators(spec));

  // Verify that the schemas match in debug builds.
  //
  // It's important to do the verification after initializing the iterators, as
  // they may not know their own schemas until they've been initialized (in the
  // case of a union of unions).
  schema_.reset(new Schema(states_.front().schema()));
  CHECK_GT(schema_->num_key_columns(), 0);
  finished_iter_stats_by_col_.resize(schema_->num_columns());
#ifndef NDEBUG
  for (const auto& s : states_) {
    if (!s.schema().Equals(*schema_)) {
      return Status::InvalidArgument(
          Substitute("Schemas do not match: $0 vs. $1",
                     schema_->ToString(), s.schema().ToString()));
    }
  }
#endif

  if (opts_.include_deleted_rows &&
      schema_->first_is_deleted_virtual_column_idx() == Schema::kColumnNotFound) {
    return Status::InvalidArgument("Merge iterator cannot deduplicate deleted "
                                   "rows without an IS_DELETED column");
  }

  // Before we copy any rows, clean up any iterators which were empty
  // to start with. Otherwise, HasNext() won't properly return false
  // if we were passed only empty iterators.
  states_.remove_and_dispose_if(
      [](const MergeIterState& s) { return PREDICT_FALSE(s.IsFullyExhausted()); },
      [](MergeIterState* s) { delete s; });

  // Establish the merge window and initialize the three heaps.
  for (auto& s : states_) {
    cold_.push(&s);
  }
  RETURN_NOT_OK(RefillHotHeap());

  initted_ = true;
  return Status::OK();
}

bool MergeIterator::HasNext() const {
  CHECK(initted_);
  return !states_.empty();
}

Status MergeIterator::InitSubIterators(ScanSpec *spec) {
  // Initialize all the sub iterators.
  for (auto& i : orig_iters_) {
    ScanSpec *spec_copy = spec != nullptr ? scan_spec_copies_.Construct(*spec) : nullptr;
    RETURN_NOT_OK(InitAndMaybeWrap(&i.iter, spec_copy));
    unique_ptr<MergeIterState> state(new MergeIterState(std::move(i)));
    RETURN_NOT_OK(state->Init(&decoded_bounds_arena_));
    states_.push_back(*state.release());
  }
  orig_iters_.clear();

  // Since we handle predicates in all the wrapped iterators, we can clear
  // them here.
  if (spec != nullptr) {
    spec->RemovePredicates();
  }
  return Status::OK();
}

Status MergeIterator::AdvanceAndReheap(MergeIterState* state,
                                       size_t num_rows_to_advance) {
  bool pulled_new_block;
  RETURN_NOT_OK(state->Advance(num_rows_to_advance, &pulled_new_block));
  hot_.pop();

  // Note that hotmaxes_ is not yet popped as it's not necessary to do so if the
  // merge window hasn't changed. Thus, we can avoid some work by deferring it
  // into the cases below.

  if (state->IsFullyExhausted()) {
    hotmaxes_.pop();
    DestroySubIterator(state);

    // This sub-iterator's removal means the end of the merge window may have shifted.
    RETURN_NOT_OK(RefillHotHeap());
  } else if (pulled_new_block) {
    hotmaxes_.pop();

    // This sub-iterator has a new block, which means the end of the merge window
    //may have shifted.
    if (!hotmaxes_.empty() &&
        schema_->Compare(hotmaxes_.top(), state->next_row()) < 0) {
      // The new block lies beyond the new end of the merge window.
      VLOG(2) << "Block finished, became cold: " << state->ToString();
      cold_.push(state);
    } else {
      // The new block is still within the merge window.
      VLOG(2) << "Block finished, still hot: " << state->ToString();
      hot_.push(state);
      hotmaxes_.push(state->last_row());
    }
    RETURN_NOT_OK(RefillHotHeap());
  } else {
    // The sub-iterator's block's upper bound remains the same; the merge window
    // has not changed.
    hot_.push(state);
  }
  return Status::OK();
}

Status MergeIterator::RefillHotHeap() {
  VLOG(2) << "Refilling hot heap";
  while (!cold_.empty() &&
         (hotmaxes_.empty() ||
          schema_->Compare(hotmaxes_.top(), cold_.top()->next_row()) >= 0)) {
    MergeIterState* warmest = cold_.top();
    cold_.pop();

    // This will only be true once per sub-iterator, when it becomes hot for the
    // very first time.
    if (warmest->IsBlockExhausted()) {
      RETURN_NOT_OK(warmest->PullNextBlock());

      // After pulling a block, we can't just assume 'warmest' overlaps the
      // merge window: there could have been a huge gap between the pulled block
      // and the sub-iterator's absolute bounds. In other words, although the
      // bounds told us that 'warmest' was the best candidate, the block is the
      // ultimate source of truth.
      //
      // To deal with this, we pretend 'warmest' doesn't overlap and restart the
      // algorithm. In the worst case (little to no gap between the block and
      // the bounds), we'll pop 'warmest' right back out again.
      if (warmest->IsFullyExhausted()) {
        DestroySubIterator(warmest);
      } else {
        cold_.push(warmest);
      }
      continue;
    }
    VLOG(2) << "Became hot: " << warmest->ToString();
    hot_.push(warmest);
    hotmaxes_.push(warmest->last_row());
  }
  if (VLOG_IS_ON(2)) {
    VLOG(2) << "Done refilling hot heap";
    for (const auto* c : cold_) {
      VLOG(2) << "Still cold: " << c->ToString();
    }
  }
  return Status::OK();
}

void MergeIterator::DestroySubIterator(MergeIterState* state) {
  DCHECK(state->IsFullyExhausted());

  std::lock_guard<rw_spinlock> l(states_lock_);
  state->AddStats(&finished_iter_stats_by_col_);
  states_.erase_and_dispose(states_.iterator_to(*state),
                            [](MergeIterState* s) { delete s; });
}

Status MergeIterator::NextBlock(RowBlock* dst) {
  VLOG(3) << "Called NextBlock (" << dst->nrows() << " rows) on " << ToString();
  CHECK(initted_);
  DCHECK_SCHEMA_EQ(*dst->schema(), schema());

  if (dst->arena()) {
    dst->arena()->Reset();
  }

  size_t dst_row_idx = 0;
  while (dst_row_idx < dst->nrows()) {
    // If the hot heap is empty, we must be out of sub-iterators.
    if (PREDICT_FALSE(hot_.empty())) {
      DCHECK(states_.empty());
      break;
    }

    // If there's just one hot sub-iterator, we can copy its entire block
    // instead of copying row-by-row.
    //
    // When N sub-iterators fully overlap, there'll only be one hot sub-iterator
    // when consuming the very last row. A block copy for this case is more
    // overhead than just copying out the last row.
    //
    // TODO(adar): this can be further optimized by "attaching" data to 'dst'
    // rather than copying it.
    if (hot_.size() == 1 && hot_.top()->remaining_in_block() > 1) {
      RETURN_NOT_OK(MaterializeBlock(dst, &dst_row_idx));
    } else {
      RETURN_NOT_OK(MaterializeOneRow(dst, &dst_row_idx));
    }
  }

  if (dst_row_idx < dst->nrows()) {
    dst->Resize(dst_row_idx);
  }

  return Status::OK();
}

Status MergeIterator::MaterializeBlock(RowBlock* dst, size_t* dst_row_idx) {
  DCHECK_EQ(1, hot_.size());

  MergeIterState* state = hot_.top();
  size_t num_rows_copied;
  RETURN_NOT_OK(state->CopyBlock(dst, *dst_row_idx, &num_rows_copied));
  RETURN_NOT_OK(AdvanceAndReheap(state, num_rows_copied));

  // CopyBlock() already updated dst's SelectionVector.
  *dst_row_idx += num_rows_copied;
  return Status::OK();
}

// TODO(todd): this is an obvious spot to add codegen - there's a ton of branching
// and such around the comparisons. A simple experiment indicated there's some
// 2x to be gained.
Status MergeIterator::MaterializeOneRow(RowBlock* dst, size_t* dst_row_idx) {
  // We need a vector to track the iterators whose next_row() contains the
  // smallest row key at a given moment during the merge because there may be
  // multiple deleted rows with the same row key across multiple rowsets, and up
  // to one live instance, that we have to deduplicate.
  vector<MergeIterState*> smallest;
  smallest.reserve(hot_.size());

  // Find the set of sub-iterators whose matching next row keys are the smallest
  // across all sub-iterators.
  //
  // Note: heap ordered iteration isn't the same as a total ordering. For
  // example, the two absolute smallest keys might be in the same sub-iterator
  // rather than in the first two sub-iterators yielded via ordered iteration.
  // However, the goal here is to identify a group of matching keys for the
  // purpose of deduplication, and we're guaranteed that such matching keys
  // cannot exist in the same sub-iterator (i.e. the same rowset).
  for (auto iter = hot_.ordered_begin(); iter != hot_.ordered_end(); ++iter) {
    MergeIterState* state = *iter;
    if (!smallest.empty() &&
        schema_->Compare(state->next_row(), smallest[0]->next_row()) != 0) {
      break;
    }
    smallest.emplace_back(state);
  }

  MergeIterState* row_to_return_iter = nullptr;
  if (!opts_.include_deleted_rows) {
    // Since deleted rows are not included here, there can only be a single
    // instance of any given row key in 'smallest'.
    CHECK_EQ(1, smallest.size()) << "expected only a single smallest row";
    row_to_return_iter = smallest[0];
  } else {
    // Deduplicate any deleted rows. Row instance de-duplication criteria:
    // 1. If there is a non-deleted instance, return that instance.
    // 2. If all rows are deleted, any instance will suffice because we
    //    don't guarantee that we will return valid field values for deleted
    //    rows.
    int live_rows_found = 0;
    for (const auto& s : smallest) {
      bool is_deleted =
          *schema_->ExtractColumnFromRow<IS_DELETED>(
              s->next_row(), schema_->first_is_deleted_virtual_column_idx());
      if (!is_deleted) {
        // We found the single live instance of the row.
        row_to_return_iter = s;
#ifndef NDEBUG
        live_rows_found++; // In DEBUG mode, do a sanity-check count of the live rows.
#else
        break; // In RELEASE mode, short-circuit the loop.
#endif
      }
    }
    DCHECK_LE(live_rows_found, 1) << "expected at most one live row";

    // If all instances of a given row are deleted then return an arbitrary
    // deleted instance.
    if (row_to_return_iter == nullptr) {
      row_to_return_iter = smallest[0];
      DCHECK(*schema_->ExtractColumnFromRow<IS_DELETED>(
          row_to_return_iter->next_row(), schema_->first_is_deleted_virtual_column_idx()))
          << "expected deleted row";
    }
  }
  VLOG(3) << Substitute("Copying row $0 from $1",
                        *dst_row_idx, row_to_return_iter->ToString());
  RowBlockRow dst_row = dst->row(*dst_row_idx);
  RETURN_NOT_OK(CopyRow(row_to_return_iter->next_row(), &dst_row, dst->arena()));

  // Advance all matching sub-iterators and remove any that are exhausted.
  for (auto& s : smallest) {
    RETURN_NOT_OK(AdvanceAndReheap(s, /*num_rows_to_advance=*/1));
  }

  dst->selection_vector()->SetRowSelected(*dst_row_idx);
  (*dst_row_idx)++;
  return Status::OK();
}

string MergeIterator::ToString() const {
  return Substitute("Merge($0 iters)", num_orig_iters_);
}

const Schema& MergeIterator::schema() const {
  CHECK(initted_);
  return *schema_;
}

void MergeIterator::GetIteratorStats(vector<IteratorStats>* stats) const {
  shared_lock<rw_spinlock> l(states_lock_);
  CHECK(initted_);
  *stats = finished_iter_stats_by_col_;

  for (const auto& s : states_) {
    s.AddStats(stats);
  }
}

unique_ptr<RowwiseIterator> NewMergeIterator(
    MergeIteratorOptions opts, vector<IterWithBounds> iters) {
  return unique_ptr<RowwiseIterator>(new MergeIterator(opts, std::move(iters)));
}

////////////////////////////////////////////////////////////
// UnionIterator
////////////////////////////////////////////////////////////

// An iterator which unions the results of other iterators.
// This is different from MergeIterator in that it lays the results out end-to-end
// rather than merging them based on keys. Hence it is more efficient since there is
// no comparison needed, and the key column does not need to be read if it is not
// part of the projection.
class UnionIterator : public RowwiseIterator {
 public:
  // Constructs a UnionIterator of the given iterators.
  //
  // The iterators must have matching schemas and should not yet be initialized.
  explicit UnionIterator(vector<IterWithBounds> iters);

  Status Init(ScanSpec *spec) OVERRIDE;

  bool HasNext() const OVERRIDE;

  string ToString() const OVERRIDE;

  const Schema &schema() const OVERRIDE {
    CHECK(initted_);
    CHECK(schema_.get() != NULL) << "Bad schema in " << ToString();
    return *CHECK_NOTNULL(schema_.get());
  }

  virtual void GetIteratorStats(vector<IteratorStats>* stats) const OVERRIDE;

  virtual Status NextBlock(RowBlock* dst) OVERRIDE;

 private:
  void PrepareBatch();
  Status MaterializeBlock(RowBlock* dst);
  void FinishBatch();
  Status InitSubIterators(ScanSpec *spec);

  // Pop the front iterator from iters_ and accumulate its statistics into
  // finished_iter_stats_by_col_.
  void PopFront();

  // Schema: initialized during Init()
  unique_ptr<Schema> schema_;

  bool initted_;

  // Lock protecting 'iters_' and 'finished_iter_stats_by_col_'.
  //
  // Scanners are mostly accessed by the thread doing the scanning, but the HTTP endpoint
  // which lists running scans may occasionally need to read as well.
  //
  // The "owner" thread of the scanner doesn't need to acquire this in read mode, since
  // it's the only thread which might write. However, it does need to acquire in write
  // mode when changing 'iters_'.
  mutable rw_spinlock iters_lock_;
  deque<IterWithBounds> iters_;

  // Statistics (keyed by projection column index) accumulated so far by any
  // fully-consumed sub-iterators.
  vector<IteratorStats> finished_iter_stats_by_col_;

  // When the underlying iterators are initialized, each needs its own
  // copy of the scan spec in order to do its own pushdown calculations, etc.
  // The copies are allocated from this pool so they can be automatically freed
  // when the UnionIterator goes out of scope.
  ObjectPool<ScanSpec> scan_spec_copies_;
};

UnionIterator::UnionIterator(vector<IterWithBounds> iters)
  : initted_(false),
    iters_(std::make_move_iterator(iters.begin()),
           std::make_move_iterator(iters.end())) {
  CHECK_GT(iters_.size(), 0);
}

Status UnionIterator::Init(ScanSpec *spec) {
  CHECK(!initted_);

  // Initialize the underlying iterators
  RETURN_NOT_OK(InitSubIterators(spec));

  // Verify that the schemas match in debug builds.
  //
  // It's important to do the verification after initializing the iterators, as
  // they may not know their own schemas until they've been initialized (in the
  // case of a union of unions).
  schema_.reset(new Schema(iters_.front().iter->schema()));
  finished_iter_stats_by_col_.resize(schema_->num_columns());
#ifndef NDEBUG
  for (const auto& i : iters_) {
    if (!i.iter->schema().Equals(*schema_)) {
      return Status::InvalidArgument(
          Substitute("Schemas do not match: $0 vs. $1",
                     schema_->ToString(), i.iter->schema().ToString()));
    }
  }
#endif

  initted_ = true;
  return Status::OK();
}

Status UnionIterator::InitSubIterators(ScanSpec *spec) {
  for (auto& i : iters_) {
    ScanSpec *spec_copy = spec != nullptr ? scan_spec_copies_.Construct(*spec) : nullptr;
    RETURN_NOT_OK(InitAndMaybeWrap(&i.iter, spec_copy));

    // The union iterator doesn't care about these, so let's drop them now to
    // free some memory.
    i.encoded_bounds.reset();
  }
  // Since we handle predicates in all the wrapped iterators, we can clear
  // them here.
  if (spec != nullptr) {
    spec->RemovePredicates();
  }
  return Status::OK();
}

bool UnionIterator::HasNext() const {
  CHECK(initted_);
  for (const auto& i : iters_) {
    if (i.iter->HasNext()) return true;
  }

  return false;
}

Status UnionIterator::NextBlock(RowBlock* dst) {
  CHECK(initted_);
  PrepareBatch();
  RETURN_NOT_OK(MaterializeBlock(dst));
  FinishBatch();
  return Status::OK();
}

void UnionIterator::PrepareBatch() {
  CHECK(initted_);

  while (!iters_.empty() &&
         !iters_.front().iter->HasNext()) {
    PopFront();
  }
}

Status UnionIterator::MaterializeBlock(RowBlock *dst) {
  return iters_.front().iter->NextBlock(dst);
}

void UnionIterator::FinishBatch() {
  if (!iters_.front().iter->HasNext()) {
    // Iterator exhausted, remove it.
    PopFront();
  }
}

void UnionIterator::PopFront() {
  std::lock_guard<rw_spinlock> l(iters_lock_);
  AddIterStats(*iters_.front().iter, &finished_iter_stats_by_col_);
  iters_.pop_front();
}

string UnionIterator::ToString() const {
  string s;
  s.append("Union(");
  s += JoinMapped(iters_, [](const IterWithBounds& i) {
      return i.iter->ToString();
    }, ",");
  s.append(")");
  return s;
}

void UnionIterator::GetIteratorStats(vector<IteratorStats>* stats) const {
  CHECK(initted_);
  shared_lock<rw_spinlock> l(iters_lock_);
  *stats = finished_iter_stats_by_col_;
  if (!iters_.empty()) {
    AddIterStats(*iters_.front().iter, stats);
  }
}

unique_ptr<RowwiseIterator> NewUnionIterator(vector<IterWithBounds> iters) {
  return unique_ptr<RowwiseIterator>(new UnionIterator(std::move(iters)));
}

////////////////////////////////////////////////////////////
// MaterializingIterator
////////////////////////////////////////////////////////////

// An iterator which wraps a ColumnwiseIterator, materializing it into full rows.
//
// Column predicates are pushed down into this iterator. While materializing a
// block, columns with associated predicates are materialized first, and the
// predicates evaluated. If the predicates succeed in filtering out an entire
// batch, then other columns may avoid doing any IO.
class MaterializingIterator : public RowwiseIterator {
 public:
  explicit MaterializingIterator(unique_ptr<ColumnwiseIterator> iter);

  // Initialize the iterator, performing predicate pushdown as described above.
  Status Init(ScanSpec *spec) OVERRIDE;

  bool HasNext() const OVERRIDE;

  string ToString() const OVERRIDE;

  const Schema &schema() const OVERRIDE {
    return iter_->schema();
  }

  virtual void GetIteratorStats(std::vector<IteratorStats>* stats) const OVERRIDE {
    iter_->GetIteratorStats(stats);
  }

  virtual Status NextBlock(RowBlock* dst) OVERRIDE;

 private:
  Status MaterializeBlock(RowBlock *dst);

  unique_ptr<ColumnwiseIterator> iter_;

  // List of (column index, predicate) in order of most to least selective, with
  // ties broken by the index.
  vector<std::pair<int32_t, ColumnPredicate>> col_idx_predicates_;

  // List of column indexes without predicates to materialize.
  vector<int32_t> non_predicate_column_indexes_;

  // Set only by test code to disallow pushdown.
  bool disallow_pushdown_for_tests_;
  bool disallow_decoder_eval_;
};

MaterializingIterator::MaterializingIterator(unique_ptr<ColumnwiseIterator> iter)
    : iter_(std::move(iter)),
      disallow_pushdown_for_tests_(!FLAGS_materializing_iterator_do_pushdown),
      disallow_decoder_eval_(!FLAGS_materializing_iterator_decoder_eval) {
}

Status MaterializingIterator::Init(ScanSpec *spec) {
  RETURN_NOT_OK(iter_->Init(spec));

  const int32_t num_columns = schema().num_columns();
  col_idx_predicates_.clear();
  non_predicate_column_indexes_.clear();

  if (PREDICT_FALSE(!disallow_pushdown_for_tests_) && spec != nullptr) {
    col_idx_predicates_.reserve(spec->predicates().size());
    DCHECK_GE(num_columns, spec->predicates().size());
    non_predicate_column_indexes_.reserve(num_columns - spec->predicates().size());

    for (const auto& col_pred : spec->predicates()) {
      const ColumnPredicate& pred = col_pred.second;
      int col_idx = schema().find_column(pred.column().name());
      if (col_idx == Schema::kColumnNotFound) {
        return Status::InvalidArgument("No such column", col_pred.first);
      }
      VLOG(1) << "Pushing down predicate " << pred.ToString();
      col_idx_predicates_.emplace_back(col_idx, col_pred.second);
    }

    for (int32_t col_idx = 0; col_idx < num_columns; col_idx++) {
      if (!ContainsKey(spec->predicates(), schema().column(col_idx).name())) {
        non_predicate_column_indexes_.emplace_back(col_idx);
      }
    }

    // Since we'll evaluate these predicates ourselves, remove them from the
    // scan spec so higher layers don't repeat our work.
    spec->RemovePredicates();
  } else {
    non_predicate_column_indexes_.resize(num_columns);
    std::iota(non_predicate_column_indexes_.begin(),
              non_predicate_column_indexes_.end(), 0);
  }

  // Sort the predicates by selectivity so that the most selective are evaluated
  // earlier, with ties broken by the column index.
  sort(col_idx_predicates_.begin(), col_idx_predicates_.end(),
       [] (const pair<int32_t, ColumnPredicate>& left,
           const pair<int32_t, ColumnPredicate>& right) {
          int comp = SelectivityComparator(left.second, right.second);
          return comp ? comp < 0 : left.first < right.first;
       });

  return Status::OK();
}

bool MaterializingIterator::HasNext() const {
  return iter_->HasNext();
}

Status MaterializingIterator::NextBlock(RowBlock* dst) {
  size_t n = dst->row_capacity();
  if (dst->arena()) {
    dst->arena()->Reset();
  }

  RETURN_NOT_OK(iter_->PrepareBatch(&n));
  dst->Resize(n);
  RETURN_NOT_OK(MaterializeBlock(dst));
  RETURN_NOT_OK(iter_->FinishBatch());

  return Status::OK();
}

Status MaterializingIterator::MaterializeBlock(RowBlock *dst) {
  // Initialize the selection vector indicating which rows have been
  // been deleted.
  RETURN_NOT_OK(iter_->InitializeSelectionVector(dst->selection_vector()));

  // It's relatively common to delete large sequential chunks of rows.
  // We can fast-path that case and avoid reading any column data.
  if (!dst->selection_vector()->AnySelected()) {
    DVLOG(1) << "Fast path over " << dst->nrows() << " deleted rows";
    return Status::OK();
  }

  for (const auto& col_pred : col_idx_predicates_) {
    // Materialize the column itself into the row block.
    ColumnBlock dst_col(dst->column_block(get<0>(col_pred)));
    ColumnMaterializationContext ctx(get<0>(col_pred),
                                     &get<1>(col_pred),
                                     &dst_col,
                                     dst->selection_vector());
    // None predicates should be short-circuited in scan spec.
    DCHECK(ctx.pred()->predicate_type() != PredicateType::None);
    if (disallow_decoder_eval_) {
      ctx.SetDecoderEvalNotSupported();
    }
    RETURN_NOT_OK(iter_->MaterializeColumn(&ctx));
    if (ctx.DecoderEvalNotSupported()) {
      get<1>(col_pred).Evaluate(dst_col, dst->selection_vector());
    }

    // If after evaluating this predicate the entire row block has been filtered
    // out, we don't need to materialize other columns at all.
    if (!dst->selection_vector()->AnySelected()) {
      DVLOG(1) << "0/" << dst->nrows() << " passed predicate";
      return Status::OK();
    }
  }

  for (size_t col_idx : non_predicate_column_indexes_) {
    // Materialize the column itself into the row block.
    ColumnBlock dst_col(dst->column_block(col_idx));
    ColumnMaterializationContext ctx(col_idx,
                                     nullptr,
                                     &dst_col,
                                     dst->selection_vector());
    RETURN_NOT_OK(iter_->MaterializeColumn(&ctx));
  }

  DVLOG(1) << dst->selection_vector()->CountSelected() << "/"
           << dst->nrows() << " passed predicate";
  return Status::OK();
}

string MaterializingIterator::ToString() const {
  string s;
  s.append("Materializing(").append(iter_->ToString()).append(")");
  return s;
}

unique_ptr<RowwiseIterator> NewMaterializingIterator(
    unique_ptr<ColumnwiseIterator> iter) {
  return unique_ptr<RowwiseIterator>(
      new MaterializingIterator(std::move(iter)));
}

////////////////////////////////////////////////////////////
// PredicateEvaluatingIterator
////////////////////////////////////////////////////////////

// An iterator which wraps another iterator and evaluates any predicates that the
// wrapped iterator did not itself handle during push down.
class PredicateEvaluatingIterator : public RowwiseIterator {
 public:
  // Construct the evaluating iterator.
  // This is only called from ::InitAndMaybeWrap()
  // REQUIRES: base_iter is already Init()ed.
  explicit PredicateEvaluatingIterator(unique_ptr<RowwiseIterator> base_iter);

  // Initialize the iterator.
  // POSTCONDITION: spec->predicates().empty()
  Status Init(ScanSpec *spec) OVERRIDE;

  virtual Status NextBlock(RowBlock *dst) OVERRIDE;

  bool HasNext() const OVERRIDE;

  string ToString() const OVERRIDE;

  const Schema &schema() const OVERRIDE {
    return base_iter_->schema();
  }

  virtual void GetIteratorStats(std::vector<IteratorStats>* stats) const OVERRIDE {
    base_iter_->GetIteratorStats(stats);
  }

  const vector<ColumnPredicate>& col_predicates() const { return col_predicates_; }

 private:
  unique_ptr<RowwiseIterator> base_iter_;

  // List of predicates in order of most to least selective, with
  // ties broken by the column index.
  vector<ColumnPredicate> col_predicates_;
};

PredicateEvaluatingIterator::PredicateEvaluatingIterator(unique_ptr<RowwiseIterator> base_iter)
    : base_iter_(std::move(base_iter)) {
}

Status PredicateEvaluatingIterator::Init(ScanSpec *spec) {
  // base_iter_ already Init()ed before this is constructed.
  CHECK_NOTNULL(spec);

  // Gather any predicates that the base iterator did not pushdown, and remove
  // the predicates from the spec.
  col_predicates_.clear();
  col_predicates_.reserve(spec->predicates().size());
  for (auto& predicate : spec->predicates()) {
    col_predicates_.emplace_back(predicate.second);
  }
  spec->RemovePredicates();

  // Sort the predicates by selectivity so that the most selective are evaluated
  // earlier, with ties broken by the column index.
  sort(col_predicates_.begin(), col_predicates_.end(),
       [&] (const ColumnPredicate& left, const ColumnPredicate& right) {
          int comp = SelectivityComparator(left, right);
          if (comp != 0) return comp < 0;
          return schema().find_column(left.column().name())
               < schema().find_column(right.column().name());
       });

  return Status::OK();
}

bool PredicateEvaluatingIterator::HasNext() const {
  return base_iter_->HasNext();
}

Status PredicateEvaluatingIterator::NextBlock(RowBlock *dst) {
  RETURN_NOT_OK(base_iter_->NextBlock(dst));

  for (const auto& predicate : col_predicates_) {
    int32_t col_idx = dst->schema()->find_column(predicate.column().name());
    if (col_idx == Schema::kColumnNotFound) {
      return Status::InvalidArgument("Unknown column in predicate", predicate.ToString());
    }
    predicate.Evaluate(dst->column_block(col_idx), dst->selection_vector());

    // If after evaluating this predicate, the entire row block has now been
    // filtered out, we don't need to evaluate any further predicates.
    if (!dst->selection_vector()->AnySelected()) {
      break;
    }
  }

  return Status::OK();
}

string PredicateEvaluatingIterator::ToString() const {
  return Substitute("PredicateEvaluating($0)", base_iter_->ToString());
}

Status InitAndMaybeWrap(unique_ptr<RowwiseIterator>* base_iter,
                        ScanSpec *spec) {
  RETURN_NOT_OK((*base_iter)->Init(spec));

  if (spec != nullptr && !spec->predicates().empty()) {
    // Underlying iterator did not accept all predicates. Wrap it.
    unique_ptr<RowwiseIterator> wrapper(new PredicateEvaluatingIterator(std::move(*base_iter)));
    RETURN_NOT_OK(wrapper->Init(spec));
    *base_iter = std::move(wrapper);
  }
  return Status::OK();
}

const vector<ColumnPredicate>& GetIteratorPredicatesForTests(
    const unique_ptr<RowwiseIterator>& iter) {
  PredicateEvaluatingIterator* pred_eval =
      down_cast<PredicateEvaluatingIterator*>(iter.get());
  return pred_eval->col_predicates();
}

} // namespace kudu

ext: cc

max_stars_repo_name: yuyunliuhen/kudu

max_issues_repo_name: yuyunliuhen/kudu

max_forks_repo_name: yuyunliuhen/kudu


text:
//////////////////////////////////////////////////////////////////////////////
// accumulator::zscore.hpp                                                  //
//                                                                          //
//  (C) Copyright 2010 Erwann Rogard                                        //
//  Use, modification and distribution are subject to the                   //
//  Boost Software License, Version 1.0. (See accompanying file             //
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)        //
//////////////////////////////////////////////////////////////////////////////
#ifndef BOOST_STATISTICS_DETAIL_ACCUMULATOR_ZSCORE_HPP_ER_2010
#define BOOST_STATISTICS_DETAIL_ACCUMULATOR_ZSCORE_HPP_ER_2010
#include <cmath>
#include <boost/accumulators/accumulators.hpp>
#include <boost/accumulators/statistics/mean.hpp>
#include <boost/accumulators/statistics/variance.hpp>
#include <boost/accumulators/framework/extractor.hpp>
#include <boost/accumulators/framework/accumulator_base.hpp>
#include <boost/accumulators/framework/parameters/sample.hpp>
#include <boost/accumulators/framework/parameters/accumulator.hpp>
#include <boost/accumulators/framework/depends_on.hpp>
#include <boost/accumulators/statistics/stats.hpp>

namespace boost{
namespace statistics{
namespace detail{
namespace accumulator{

namespace impl
{

    template<typename T>
    struct zscore : public boost::accumulators::accumulator_base
    {
        typedef boost::accumulators::dont_care dont_care_;

        typedef T result_type;

        zscore(dont_care_){}

        void operator ()(dont_care_)const
        {
        }

        template<typename Args>
        result_type result(const Args& args)const{
        	typedef boost::accumulators::tag::accumulator tag_acc_;
        	typedef boost::parameter::binding<Args,tag_acc_> bind1_;
        	typedef typename bind1_::type cref_acc_;
            cref_acc_ cref_acc = args[boost::accumulators::accumulator];
			T x = args[boost::accumulators::sample];
            T m = boost::accumulators::mean(cref_acc);
            T s = boost::accumulators::variance(cref_acc);
            s = std::sqrt(s);
            return (x - m)/s;
		}
	};

}//impl

namespace tag
{
    struct zscore
        : boost::accumulators::depends_on<
        	boost::accumulators::tag::mean,
        	boost::accumulators::tag::variance
        >
    {
        struct impl{
            template<typename T,typename W>
            struct apply{
                typedef boost::statistics::detail::accumulator::impl
                    	::zscore<T> type;
            };
        };
    };
}

namespace result_of{

    template<typename AccSet>
    struct zscore{
        typedef boost::statistics::detail::accumulator::tag::zscore tag_;
        typedef typename boost::accumulators::detail::extractor_result<
            AccSet,
            tag_
        >::type type;
    };

}

namespace extract
{

    // Usage : zscore(acc,x)
    template<typename AccSet,typename T1>
    typename boost::statistics::detail::accumulator
    		::result_of::template zscore<AccSet>::type
    zscore(AccSet const& acc,const T1& x)
    {
        typedef boost::statistics::detail::accumulator::tag::zscore the_tag;
        return boost::accumulators::extract_result<the_tag>(
            acc,
            (boost::accumulators::sample = x)
        );
    }
}

    using extract::zscore;

}// accumulator
}// detail
}// statistics
}// boost

#endif

ext: hpp


text:
/* Copyright 2003-2013 Joaquin M Lopez Munoz.
 * Distributed under the Boost Software License, Version 1.0.
 * (See accompanying file LICENSE_1_0.txt or copy at
 * http://www.boost.org/LICENSE_1_0.txt)
 *
 * See http://www.boost.org/libs/multi_index for library home page.
 */

#ifndef BOOST_MULTI_INDEX_DETAIL_NO_DUPLICATE_TAGS_HPP
#define BOOST_MULTI_INDEX_DETAIL_NO_DUPLICATE_TAGS_HPP

#if defined(_MSC_VER)
#pragma once
#endif

#include <boost/config.hpp> /* keep it first to prevent nasty warns in MSVC */
#include <boost/mpl/fold.hpp>
#include <boost/mpl/set/set0.hpp>

namespace boost{

namespace multi_index{

namespace detail{

/* no_duplicate_tags check at compile-time that a tag list
 * has no duplicate tags.
 * The algorithm deserves some explanation: tags
 * are sequentially inserted into a mpl::set if they were
 * not already present. Due to the magic of mpl::set
 * (mpl::has_key is contant time), this operation takes linear
 * time, and even MSVC++ 6.5 handles it gracefully (other obvious
 * solutions are quadratic.)
 */

struct duplicate_tag_mark{};

struct duplicate_tag_marker
{
  template <typename MplSet,typename Tag>
  struct apply
  {
    typedef mpl::s_item<
      typename mpl::if_<mpl::has_key<MplSet,Tag>,duplicate_tag_mark,Tag>::type,
      MplSet
    > type;
  };
};

template<typename TagList>
struct no_duplicate_tags
{
  typedef typename mpl::fold<
    TagList,
    mpl::set0<>,
    duplicate_tag_marker
  >::type aux;

  BOOST_STATIC_CONSTANT(
    bool,value=!(mpl::has_key<aux,duplicate_tag_mark>::value));
};

/* Variant for an index list: duplication is checked
 * across all the indices.
 */

struct duplicate_tag_list_marker
{
  template <typename MplSet,typename Index>
  struct apply:mpl::fold<
    BOOST_DEDUCED_TYPENAME Index::tag_list,
    MplSet,
    duplicate_tag_marker>
  {
  };
};

template<typename IndexList>
struct no_duplicate_tags_in_index_list
{
  typedef typename mpl::fold<
    IndexList,
    mpl::set0<>,
    duplicate_tag_list_marker
  >::type aux;

  BOOST_STATIC_CONSTANT(
    bool,value=!(mpl::has_key<aux,duplicate_tag_mark>::value));
};

} /* namespace multi_index::detail */

} /* namespace multi_index */

} /* namespace boost */

#endif


text:
//
// Copyright (c) 2013-2017 Vinnie Falco (vinnie dot falco at gmail dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//

// Test that header file is self-contained.
#include <beast/http/write.hpp>

#include <beast/http/fields.hpp>
#include <beast/http/message.hpp>
#include <beast/http/empty_body.hpp>
#include <beast/http/string_body.hpp>
#include <beast/http/write.hpp>
#include <beast/core/error.hpp>
#include <beast/core/streambuf.hpp>
#include <beast/core/to_string.hpp>
#include <beast/test/fail_stream.hpp>
#include <beast/test/string_ostream.hpp>
#include <beast/test/yield_to.hpp>
#include <beast/unit_test/suite.hpp>
#include <boost/asio/error.hpp>
#include <sstream>
#include <string>

namespace beast {
namespace http {

class write_test
    : public beast::unit_test::suite
    , public test::enable_yield_to
{
public:
    struct unsized_body
    {
        using value_type = std::string;

        class writer
        {
            value_type const& body_;

        public:
            template<bool isRequest, class Allocator>
            explicit
            writer(message<isRequest, unsized_body, Allocator> const& msg) noexcept
                : body_(msg.body)
            {
            }

            void
            init(error_code& ec) noexcept
            {
                beast::detail::ignore_unused(ec);
            }

            template<class WriteFunction>
            bool
            write(error_code&, WriteFunction&& wf) noexcept
            {
                wf(boost::asio::buffer(body_));
                return true;
            }
        };
    };

    struct fail_body
    {
        class writer;

        class value_type
        {
            friend class writer;

            std::string s_;
            test::fail_counter& fc_;
            boost::asio::io_service& ios_;

        public:
            value_type(test::fail_counter& fc,
                    boost::asio::io_service& ios)
                : fc_(fc)
                , ios_(ios)
            {
            }

            boost::asio::io_service&
            get_io_service() const
            {
                return ios_;
            }

            value_type&
            operator=(std::string s)
            {
                s_ = std::move(s);
                return *this;
            }
        };

        class writer
        {
            std::size_t n_ = 0;
            value_type const& body_;

        public:
            template<bool isRequest, class Allocator>
            explicit
            writer(message<isRequest, fail_body, Allocator> const& msg) noexcept
                : body_(msg.body)
            {
            }

            void
            init(error_code& ec) noexcept
            {
                body_.fc_.fail(ec);
            }

            template<class WriteFunction>
            bool
            write(error_code& ec, WriteFunction&& wf) noexcept
            {
                if(body_.fc_.fail(ec))
                    return false;
                if(n_ >= body_.s_.size())
                    return true;
                wf(boost::asio::buffer(body_.s_.data() + n_, 1));
                ++n_;
                return n_ == body_.s_.size();
            }
        };
    };

    template<bool isRequest, class Body, class Fields>
    std::string
    str(message<isRequest, Body, Fields> const& m)
    {
        test::string_ostream ss(ios_);
        write(ss, m);
        return ss.str;
    }

    void
    testAsyncWriteHeaders(yield_context do_yield)
    {
        {
            header<true, fields> m;
            m.version = 11;
            m.method = "GET";
            m.url = "/";
            m.fields.insert("User-Agent", "test");
            error_code ec;
            test::string_ostream ss{ios_};
            async_write(ss, m, do_yield[ec]);
            if(BEAST_EXPECTS(! ec, ec.message()))
                BEAST_EXPECT(ss.str ==
                    "GET / HTTP/1.1\r\n"
                    "User-Agent: test\r\n"
                    "\r\n");
        }
        {
            header<false, fields> m;
            m.version = 10;
            m.status = 200;
            m.reason = "OK";
            m.fields.insert("Server", "test");
            m.fields.insert("Content-Length", "5");
            error_code ec;
            test::string_ostream ss{ios_};
            async_write(ss, m, do_yield[ec]);
            if(BEAST_EXPECTS(! ec, ec.message()))
                BEAST_EXPECT(ss.str ==
                    "HTTP/1.0 200 OK\r\n"
                    "Server: test\r\n"
                    "Content-Length: 5\r\n"
                    "\r\n");
        }
    }

    void
    testAsyncWrite(yield_context do_yield)
    {
        {
            message<false, string_body, fields> m;
            m.version = 10;
            m.status = 200;
            m.reason = "OK";
            m.fields.insert("Server", "test");
            m.fields.insert("Content-Length", "5");
            m.body = "*****";
            error_code ec;
            test::string_ostream ss{ios_};
            async_write(ss, m, do_yield[ec]);
            if(BEAST_EXPECTS(! ec, ec.message()))
                BEAST_EXPECT(ss.str ==
                    "HTTP/1.0 200 OK\r\n"
                    "Server: test\r\n"
                    "Content-Length: 5\r\n"
                    "\r\n"
                    "*****");
        }
        {
            message<false, string_body, fields> m;
            m.version = 11;
            m.status = 200;
            m.reason = "OK";
            m.fields.insert("Server", "test");
            m.fields.insert("Transfer-Encoding", "chunked");
            m.body = "*****";
            error_code ec;
            test::string_ostream ss(ios_);
            async_write(ss, m, do_yield[ec]);
            if(BEAST_EXPECTS(! ec, ec.message()))
                BEAST_EXPECT(ss.str ==
                    "HTTP/1.1 200 OK\r\n"
                    "Server: test\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n"
                    "5\r\n"
                    "*****\r\n"
                    "0\r\n\r\n");
        }
    }

    void
    testFailures(yield_context do_yield)
    {
        static std::size_t constexpr limit = 100;
        std::size_t n;

        for(n = 0; n < limit; ++n)
        {
            test::fail_counter fc(n);
            test::fail_stream<
                test::string_ostream> fs(fc, ios_);
            message<true, fail_body, fields> m(
                std::piecewise_construct,
                    std::forward_as_tuple(fc, ios_));
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.fields.insert("Content-Length", "5");
            m.body = "*****";
            try
            {
                write(fs, m);
                BEAST_EXPECT(fs.next_layer().str ==
                    "GET / HTTP/1.0\r\n"
                    "User-Agent: test\r\n"
                    "Content-Length: 5\r\n"
                    "\r\n"
                    "*****"
                );
                pass();
                break;
            }
            catch(std::exception const&)
            {
            }
        }
        BEAST_EXPECT(n < limit);

        for(n = 0; n < limit; ++n)
        {
            test::fail_counter fc(n);
            test::fail_stream<
                test::string_ostream> fs(fc, ios_);
            message<true, fail_body, fields> m(
                std::piecewise_construct,
                    std::forward_as_tuple(fc, ios_));
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.fields.insert("Transfer-Encoding", "chunked");
            m.body = "*****";
            error_code ec;
            write(fs, m, ec);
            if(ec == boost::asio::error::eof)
            {
                BEAST_EXPECT(fs.next_layer().str ==
                    "GET / HTTP/1.0\r\n"
                    "User-Agent: test\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "0\r\n\r\n"
                );
                break;
            }
        }
        BEAST_EXPECT(n < limit);

        for(n = 0; n < limit; ++n)
        {
            test::fail_counter fc(n);
            test::fail_stream<
                test::string_ostream> fs(fc, ios_);
            message<true, fail_body, fields> m(
                std::piecewise_construct,
                    std::forward_as_tuple(fc, ios_));
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.fields.insert("Transfer-Encoding", "chunked");
            m.body = "*****";
            error_code ec;
            async_write(fs, m, do_yield[ec]);
            if(ec == boost::asio::error::eof)
            {
                BEAST_EXPECT(fs.next_layer().str ==
                    "GET / HTTP/1.0\r\n"
                    "User-Agent: test\r\n"
                    "Transfer-Encoding: chunked\r\n"
                    "\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "1\r\n*\r\n"
                    "0\r\n\r\n"
                );
                break;
            }
        }
        BEAST_EXPECT(n < limit);

        for(n = 0; n < limit; ++n)
        {
            test::fail_counter fc(n);
            test::fail_stream<
                test::string_ostream> fs(fc, ios_);
            message<true, fail_body, fields> m(
                std::piecewise_construct,
                    std::forward_as_tuple(fc, ios_));
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.fields.insert("Content-Length", "5");
            m.body = "*****";
            error_code ec;
            write(fs, m, ec);
            if(! ec)
            {
                BEAST_EXPECT(fs.next_layer().str ==
                    "GET / HTTP/1.0\r\n"
                    "User-Agent: test\r\n"
                    "Content-Length: 5\r\n"
                    "\r\n"
                    "*****"
                );
                break;
            }
        }
        BEAST_EXPECT(n < limit);

        for(n = 0; n < limit; ++n)
        {
            test::fail_counter fc(n);
            test::fail_stream<
                test::string_ostream> fs(fc, ios_);
            message<true, fail_body, fields> m(
                std::piecewise_construct,
                    std::forward_as_tuple(fc, ios_));
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.fields.insert("Content-Length", "5");
            m.body = "*****";
            error_code ec;
            async_write(fs, m, do_yield[ec]);
            if(! ec)
            {
                BEAST_EXPECT(fs.next_layer().str ==
                    "GET / HTTP/1.0\r\n"
                    "User-Agent: test\r\n"
                    "Content-Length: 5\r\n"
                    "\r\n"
                    "*****"
                );
                break;
            }
        }
        BEAST_EXPECT(n < limit);
    }

    void
    testOutput()
    {
        // auto content-length HTTP/1.0
        {
            message<true, string_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m);
            BEAST_EXPECT(str(m) ==
                "GET / HTTP/1.0\r\n"
                "User-Agent: test\r\n"
                "Content-Length: 1\r\n"
                "\r\n"
                "*"
            );
        }
        // keep-alive HTTP/1.0
        {
            message<true, string_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m, connection::keep_alive);
            BEAST_EXPECT(str(m) ==
                "GET / HTTP/1.0\r\n"
                "User-Agent: test\r\n"
                "Content-Length: 1\r\n"
                "Connection: keep-alive\r\n"
                "\r\n"
                "*"
            );
        }
        // upgrade HTTP/1.0
        {
            message<true, string_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            try
            {
                prepare(m, connection::upgrade);
                fail();
            }
            catch(std::exception const&)
            {
                pass();
            }
        }
        // no content-length HTTP/1.0
        {
            message<true, unsized_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 10;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m);
            test::string_ostream ss(ios_);
            error_code ec;
            write(ss, m, ec);
            BEAST_EXPECT(ec == boost::asio::error::eof);
            BEAST_EXPECT(ss.str ==
                "GET / HTTP/1.0\r\n"
                "User-Agent: test\r\n"
                "\r\n"
                "*"
            );
        }
        // auto content-length HTTP/1.1
        {
            message<true, string_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 11;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m);
            BEAST_EXPECT(str(m) ==
                "GET / HTTP/1.1\r\n"
                "User-Agent: test\r\n"
                "Content-Length: 1\r\n"
                "\r\n"
                "*"
            );
        }
        // close HTTP/1.1
        {
            message<true, string_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 11;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m, connection::close);
            test::string_ostream ss(ios_);
            error_code ec;
            write(ss, m, ec);
            BEAST_EXPECT(ec == boost::asio::error::eof);
            BEAST_EXPECT(ss.str ==
                "GET / HTTP/1.1\r\n"
                "User-Agent: test\r\n"
                "Content-Length: 1\r\n"
                "Connection: close\r\n"
                "\r\n"
                "*"
            );
        }
        // upgrade HTTP/1.1
        {
            message<true, empty_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 11;
            m.fields.insert("User-Agent", "test");
            prepare(m, connection::upgrade);
            BEAST_EXPECT(str(m) ==
                "GET / HTTP/1.1\r\n"
                "User-Agent: test\r\n"
                "Connection: upgrade\r\n"
                "\r\n"
            );
        }
        // no content-length HTTP/1.1
        {
            message<true, unsized_body, fields> m;
            m.method = "GET";
            m.url = "/";
            m.version = 11;
            m.fields.insert("User-Agent", "test");
            m.body = "*";
            prepare(m);
            test::string_ostream ss(ios_);
            error_code ec;
            write(ss, m, ec);
            BEAST_EXPECT(ss.str ==
                "GET / HTTP/1.1\r\n"
                "User-Agent: test\r\n"
                "Transfer-Encoding: chunked\r\n"
                "\r\n"
                "1\r\n"
                "*\r\n"
                "0\r\n\r\n"
            );
        }
    }

    void test_std_ostream()
    {
        // Conversion to std::string via operator<<
        message<true, string_body, fields> m;
        m.method = "GET";
        m.url = "/";
        m.version = 11;
        m.fields.insert("User-Agent", "test");
        m.body = "*";
        BEAST_EXPECT(boost::lexical_cast<std::string>(m) ==
            "GET / HTTP/1.1\r\nUser-Agent: test\r\n\r\n*");
        BEAST_EXPECT(boost::lexical_cast<std::string>(m.base()) ==
            "GET / HTTP/1.1\r\nUser-Agent: test\r\n\r\n");
        // Cause exceptions in operator<<
        {
            std::stringstream ss;
            ss.setstate(ss.rdstate() |
                std::stringstream::failbit);
            try
            {
                // header
                ss << m.base();
                fail("", __FILE__, __LINE__);
            }
            catch(std::exception const&)
            {
                pass();
            }
            try
            {
                // message
                ss << m;
                fail("", __FILE__, __LINE__);
            }
            catch(std::exception const&)
            {
                pass();
            }
        }
    }

    void testOstream()
    {
        message<true, string_body, fields> m;
        m.method = "GET";
        m.url = "/";
        m.version = 11;
        m.fields.insert("User-Agent", "test");
        m.body = "*";
        prepare(m);
        std::stringstream ss;
        ss.setstate(ss.rdstate() |
            std::stringstream::failbit);
        try
        {
            ss << m;
            fail();
        }
        catch(std::exception const&)
        {
            pass();
        }
    }

    // Ensure completion handlers are not leaked
    struct handler
    {
        static std::atomic<std::size_t>&
        count() { static std::atomic<std::size_t> n; return n; }
        handler() { ++count(); }
        ~handler() { --count(); }
        handler(handler const&) { ++count(); }
        void operator()(error_code const&) const {}
    };

    void
    testIoService()
    {
        {
            // Make sure handlers are not destroyed
            // after calling io_service::stop
            boost::asio::io_service ios;
            test::string_ostream os{ios};
            BEAST_EXPECT(handler::count() == 0);
            message<true, string_body, fields> m;
            m.method = "GET";
            m.version = 11;
            m.url = "/";
            m.fields["Content-Length"] = "5";
            m.body = "*****";
            async_write(os, m, handler{});
            BEAST_EXPECT(handler::count() > 0);
            ios.stop();
            BEAST_EXPECT(handler::count() > 0);
            ios.reset();
            BEAST_EXPECT(handler::count() > 0);
            ios.run_one();
            BEAST_EXPECT(handler::count() == 0);
        }
        {
            // Make sure uninvoked handlers are
            // destroyed when calling ~io_service
            {
                boost::asio::io_service ios;
                test::string_ostream is{ios};
                BEAST_EXPECT(handler::count() == 0);
                message<true, string_body, fields> m;
                m.method = "GET";
                m.version = 11;
                m.url = "/";
                m.fields["Content-Length"] = "5";
                m.body = "*****";
                async_write(is, m, handler{});
                BEAST_EXPECT(handler::count() > 0);
            }
            BEAST_EXPECT(handler::count() == 0);
        }
    }

    void run() override
    {
        yield_to(&write_test::testAsyncWriteHeaders, this);
        yield_to(&write_test::testAsyncWrite, this);
        yield_to(&write_test::testFailures, this);
        testOutput();
        test_std_ostream();
        testOstream();
        testIoService();
    }
};

BEAST_DEFINE_TESTSUITE(write,http,beast);

} // http
} // beast

ext: cpp

max_stars_repo_name: MassICTBV/casinocoind

max_issues_repo_name: MassICTBV/casinocoind

max_forks_repo_name: MassICTBV/casinocoind


text:
#define BOOST_TEST_MODULE SolutionTest

#include "solution.hpp"

#define BOOST_TEST_DYN_LINK
#include <boost/test/unit_test.hpp>

BOOST_AUTO_TEST_SUITE(SolutionSuite)

BOOST_AUTO_TEST_CASE(PlainTest1)
{
    string s = "aa";
    string p = "a";
    bool results = Solution().isMatch(s, p);

    bool expected = false;
    BOOST_CHECK_EQUAL(results, expected);
}

BOOST_AUTO_TEST_CASE(PlainTest2)
{
    string s = "aa";
    string p = "a*";
    bool results = Solution().isMatch(s, p);

    bool expected = true;
    BOOST_CHECK_EQUAL(results, expected);
}

BOOST_AUTO_TEST_CASE(PlainTest3)
{
    string s = "ab";
    string p = ".*";
    bool results = Solution().isMatch(s, p);

    bool expected = true;
    BOOST_CHECK_EQUAL(results, expected);
}

BOOST_AUTO_TEST_CASE(PlainTest4)
{
    string s = "aab";
    string p = "c*a*b";
    bool results = Solution().isMatch(s, p);

    bool expected = true;
    BOOST_CHECK_EQUAL(results, expected);
}

BOOST_AUTO_TEST_CASE(PlainTest5)
{
    string s = "mississippi";
    string p = "mis*is*p*.";
    bool results = Solution().isMatch(s, p);

    bool expected = false;
    BOOST_CHECK_EQUAL(results, expected);
}

BOOST_AUTO_TEST_SUITE_END()

max_stars_repo_name: johnhany/leetcode

max_issues_repo_name: johnhany/leetcode

max_forks_repo_name: johnhany/leetcode


text:
//              Copyright Toru Niina 2018.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE or copy at
//          https://www.boost.org/LICENSE_1_0.txt)
#ifndef TOML_RESULT_HPP
#define TOML_RESULT_HPP
#include <toml/predefine.hpp>
#include <boost/config.hpp>
#include <boost/type_traits.hpp>
#include <boost/variant.hpp>
#include <utility>
#include <ostream>

namespace toml
{
namespace detail
{

// to contain convertible types (like toml::string and std::string)
// at the same time, add extra type information to each type
template<typename T>
struct success
{
    typedef T value_type;

    explicit success(const value_type& val): value(val) {}
    ~success(){}

    template<typename U>
    explicit success(const U& val, typename boost::enable_if<
            boost::is_convertible<U, T> >::type* = 0)
        : value(value_type(val))
    {}

    success(const success& rhs): value(rhs.value){}
    success& operator=(const success& rhs) {value = rhs.value; return *this;}

#ifdef BOOST_HAS_RVALUE_REFS
    explicit success(success&& rhs): value(std::move(rhs.value)){}
    success& operator=(success&& rhs)
    {value = std::move(rhs.value); return *this;}
#endif

    value_type value;
};

template<typename T>
struct failure
{
    typedef T value_type;

    explicit failure(const value_type& val): value(val) {}
    ~failure(){}

    template<typename U>
    explicit failure(const U& val, typename boost::enable_if<
            boost::is_convertible<U, T> >::type* = 0)
        : value(value_type(val))
    {}

    failure(const failure& rhs): value(rhs.value){}
    failure& operator=(const failure& rhs) {value = rhs.value; return *this;}

#ifdef BOOST_HAS_RVALUE_REFS
    explicit failure(failure&& rhs): value(std::move(rhs.value)){}
    failure& operator=(failure&& rhs)
    {value = std::move(rhs.value); return *this;}
#endif

    value_type value;
};

#ifdef BOOST_HAS_RVALUE_REFS
template<typename T>
success<T> ok(T&& v)  {return success<T>(std::forward<T>(v));}
template<typename T>
failure<T> err(T&& v) {return failure<T>(std::forward<T>(v));}
#else
template<typename T>
success<T> ok(const T& v) {return success<T>(v);}
template<typename T>
failure<T> err(const T& v) {return failure<T>(v);}
#endif

template<typename Ok, typename Err>
struct result
{
    typedef success<Ok>  success_type;
    typedef failure<Err> failure_type;
    typedef boost::variant<success_type, failure_type> storage_type;

    result(const success_type& s): storage_(s){}
    result(const failure_type& f): storage_(f){}
    ~result(){}

    result(const result& rhs): storage_(rhs.storage_){}
    result& operator=(const result& rhs){storage_ = rhs.storage_; return *this;}

    template<typename Ok_, typename Err_>
    result(const result<Ok_, Err_>& rhs)
    {
        if(rhs.is_ok()){this->storage_ = success_type(Ok(rhs.unwrap()));}
        else           {this->storage_ = failure_type(Err(rhs.unwrap_err()));}
    }

    template<typename Ok_>
    result(const success<Ok_>& s, typename boost::enable_if<
            boost::is_convertible<Ok_, Ok> >::type* = 0)
        : storage_(success_type(s.value))
    {}
    template<typename Err_>
    result(const failure<Err_>& f, typename boost::enable_if<
            boost::is_convertible<Err_, Err> >::type* = 0)
        : storage_(failure_type(f.value))
    {}

#ifdef BOOST_HAS_RVALUE_REFS
    result(success_type&& s): storage_(std::move(s)){}
    result(failure_type&& f): storage_(std::move(f)){}

    result(result&& rhs): storage_(std::move(rhs.storage_)){}
    result& operator=(result&& rhs)
    {storage_ = std::move(rhs.storage_); return *this;}
#endif // rvalue_refs

    bool is_ok()  const BOOST_NOEXCEPT_OR_NOTHROW {return storage_.which()==0;}
    bool is_err() const BOOST_NOEXCEPT_OR_NOTHROW {return storage_.which()==1;}

    operator bool() const BOOST_NOEXCEPT_OR_NOTHROW {return this->is_ok();}

    Ok& ok_or(Ok& v) BOOST_NOEXCEPT_OR_NOTHROW
    {
        if(this->is_ok()){return this->unwrap();} else {return v;}
    }
    Ok const& ok_or(const Ok& v) const BOOST_NOEXCEPT_OR_NOTHROW
    {
        if(this->is_ok()){return this->unwrap();} else {return v;}
    }

    Err& err_or(Err& s) BOOST_NOEXCEPT_OR_NOTHROW
    {
        if(this->is_err()){return this->unwrap_err();} else {return s;}
    }
    Err const& err_or(const Err& s) const BOOST_NOEXCEPT_OR_NOTHROW
    {
        if(this->is_err()){return this->unwrap_err();} else {return s;}
    }

    Ok& unwrap()
    {return boost::get<success_type>(this->storage_).value;}
    Ok const& unwrap() const
    {return boost::get<success_type>(this->storage_).value;}

    Err& unwrap_err()
    {return boost::get<failure_type>(this->storage_).value;}
    Err const& unwrap_err() const
    {return boost::get<failure_type>(this->storage_).value;}

  private:
    storage_type storage_;
};

template<typename charT, typename traits, typename Ok, typename Err>
std::basic_ostream<charT, traits>&
operator<<(std::basic_ostream<charT, traits>& os, const result<Ok, Err>& res)
{
    if(res.is_ok()){os << "Ok("      << res.unwrap()     << ')';}
    else           {os << "Error( "  << res.unwrap_err() << ')';}
    return os;
}

} // detail
} // toml
#endif// TOML_RESULT_HPP


text:
#include <ibtk/IBTKInit.h>
#include <ibtk/SAMRAIDataCache.h>

#include <VariableDatabase.h>

#include <Eigen/Core>

#ifdef IBTK_HAVE_LIBMESH
#include <libmesh/point.h>
#endif

#include <iostream>
#include <memory>

// Test IBAMR project. This verifies that we can correctly compile and link against IBAMR

int main(int argc, char **argv)
{
    // Make sure we set up the right C++ version by explicitly using C++11 features:
    auto ibtk_init = std::make_shared<IBTK::IBTKInit>(argc, argv);
#ifdef IBTK_HAVE_LIBMESH
    const libMesh::Point point(1.0, 2.0, 3.0);
#endif

    SAMRAI::hier::VariableDatabase<NDIM>* var_db = SAMRAI::hier::VariableDatabase<NDIM>::getDatabase();
    std::cout << "Number of contexts: " << var_db->getNumberOfRegisteredPatchDataIndices() << '\n';

    std::cout << "hello, world\n";

    std::cout << "eigen version: " << EIGEN_WORLD_VERSION << '.' << EIGEN_MAJOR_VERSION << '.'
              << EIGEN_MINOR_VERSION << '\n';
}


text:
#pragma once

#include <graphene/chain/rock_paper_scissors.hpp>
#include <graphene/chain/game_object.hpp>

#include <graphene/protocol/tournament.hpp>

#include <graphene/db/flat_index.hpp>
#include <graphene/db/generic_index.hpp>

#include <fc/crypto/hex.hpp>

#include <boost/multi_index/composite_key.hpp>

#include <sstream>

namespace graphene { namespace chain {
   class match_object;
} }

namespace fc {
   void to_variant(const graphene::chain::match_object& match_obj, fc::variant& v, uint32_t max_depth = 1);
   void from_variant(const fc::variant& v, graphene::chain::match_object& match_obj, uint32_t max_depth = 1);
} //end namespace fc

namespace graphene { namespace chain {
   class database;
   using namespace graphene::db;

   enum class match_state
   {
      waiting_on_previous_matches,
      match_in_progress,
      match_complete
   };

   class match_object : public graphene::db::abstract_object<match_object>
   {
   public:
      static const uint8_t space_id = protocol_ids;
      static const uint8_t type_id  = match_object_type;

      tournament_id_type tournament_id;

      /// The players in the match
      vector<account_id_type> players;

      /// The list of games in the match
      /// Unlike tournaments where the list of matches is known at the start,
      /// the list of games will start with one game and grow until we have played
      /// enough games to declare a winner for the match.
      vector<game_id_type> games;

      /// A list of the winners of each round of the game.  This information is
      /// also stored in the game object, but is duplicated here to allow displaying
      /// information about a match without having to request all game objects
      vector<flat_set<account_id_type> > game_winners;

      /// A count of the number of wins for each player
      vector<uint32_t> number_of_wins;

      /// the total number of games that ended up in a tie/draw/stalemate
      uint32_t number_of_ties;

      // If the match is not yet complete, this will be empty
      // If the match is in the "match_complete" state, it will contain the
      // list of winners.
      // For Rock-paper-scissors, there will be one winner, unless there is
      // a stalemate (in that case, there are no winners)
      flat_set<account_id_type> match_winners;

      /// the time the match started
      time_point_sec start_time;

      /// If the match has ended, the time it ended
      optional<time_point_sec> end_time;

      match_object();
      match_object(const match_object& rhs);
      ~match_object();
      match_object& operator=(const match_object& rhs);

      match_state get_state() const;

      // serialization functions:
      // for serializing to raw, go through a temporary sstream object to avoid
      // having to implement serialization in the header file
      template<typename Stream>
      friend Stream& operator<<( Stream& s, const match_object& match_obj );

      template<typename Stream>
      friend Stream& operator>>( Stream& s, match_object& match_obj );

      friend void ::fc::to_variant(const graphene::chain::match_object& match_obj, fc::variant& v, uint32_t max_depth);
      friend void ::fc::from_variant(const fc::variant& v, graphene::chain::match_object& match_obj, uint32_t max_depth);

      void pack_impl(std::ostream& stream) const;
      void unpack_impl(std::istream& stream);
      void on_initiate_match(database& db);
      void on_game_complete(database& db, const game_object& game);
      game_id_type start_next_game(database& db, match_id_type match_id);

      class impl;
      std::unique_ptr<impl> my;
   };

   typedef multi_index_container<
      match_object,
      indexed_by<
         ordered_unique< tag<by_id>, member< object, object_id_type, &object::id > >      >
   > match_object_multi_index_type;
   typedef generic_index<match_object, match_object_multi_index_type> match_index;

   template<typename Stream>
   inline Stream& operator<<( Stream& s, const match_object& match_obj )
   {
      // pack all fields exposed in the header in the usual way
      // instead of calling the derived pack, just serialize the one field in the base class
      //   fc::raw::pack<Stream, const graphene::db::abstract_object<match_object> >(s, match_obj);
      fc::raw::pack(s, match_obj.id);
      fc::raw::pack(s, match_obj.tournament_id);
      fc::raw::pack(s, match_obj.players);
      fc::raw::pack(s, match_obj.games);
      fc::raw::pack(s, match_obj.game_winners);
      fc::raw::pack(s, match_obj.number_of_wins);
      fc::raw::pack(s, match_obj.number_of_ties);
      fc::raw::pack(s, match_obj.match_winners);
      fc::raw::pack(s, match_obj.start_time);
      fc::raw::pack(s, match_obj.end_time);

      // fc::raw::pack the contents hidden in the impl class
      std::ostringstream stream;
      match_obj.pack_impl(stream);
      std::string stringified_stream(stream.str());
      fc::raw::pack(s, stream.str());

      return s;
   }

   template<typename Stream>
   inline Stream& operator>>( Stream& s, match_object& match_obj )
   {
      // unpack all fields exposed in the header in the usual way
      //fc::raw::unpack<Stream, graphene::db::abstract_object<match_object> >(s, match_obj);
      fc::raw::unpack(s, match_obj.id);
      fc::raw::unpack(s, match_obj.tournament_id);
      fc::raw::unpack(s, match_obj.players);
      fc::raw::unpack(s, match_obj.games);
      fc::raw::unpack(s, match_obj.game_winners);
      fc::raw::unpack(s, match_obj.number_of_wins);
      fc::raw::unpack(s, match_obj.number_of_ties);
      fc::raw::unpack(s, match_obj.match_winners);
      fc::raw::unpack(s, match_obj.start_time);
      fc::raw::unpack(s, match_obj.end_time);

      // fc::raw::unpack the contents hidden in the impl class
      std::string stringified_stream;
      fc::raw::unpack(s, stringified_stream);
      std::istringstream stream(stringified_stream);
      match_obj.unpack_impl(stream);

      return s;
   }

} }

MAP_OBJECT_ID_TO_TYPE(graphene::chain::match_object)

FC_REFLECT_ENUM(graphene::chain::match_state,
                (waiting_on_previous_matches)
                (match_in_progress)
                (match_complete))

//FC_REFLECT_TYPENAME(graphene::chain::match_object) // manually serialized
FC_REFLECT_DERIVED(graphene::chain::match_object, (graphene::db::object), (players))

max_stars_repo_name: nathanhourt/peerplays

max_issues_repo_name: nathanhourt/peerplays

max_forks_repo_name: nathanhourt/peerplays


text:
#include "changerenderer.h"
#include "ui_changerenderer.h"
#include <boost/filesystem.hpp>

static boost::filesystem::path setting = "overlay_proc.json";

int main(int argc, char**argv)
{
    QApplication app(argc, argv);
	ChangeRenderer* renderer = new ChangeRenderer();
	renderer->show();
    return app.exec();
}

ChangeRenderer::ChangeRenderer(QWidget* parent)
	: QWidget(parent)
	,ui(new Ui::ChangeRenderer)
{
	ui->setupUi(this);
	if (boost::filesystem::exists(setting))
	{
		Json::Reader reader;
		Json::Value zero(0);
#ifdef _WIN32
		std::ifstream fin(setting.wstring().c_str());
#else
		std::ifstream fin(setting.string().c_str());
#endif
		if (reader.parse(fin, value))
		{
			ui->renderer->setCurrentIndex(value.get("renderer", zero).asInt());
		}
		fin.close();
	}
}
ChangeRenderer::~ChangeRenderer() {

}
void ChangeRenderer::on_buttonOK_clicked()
{
	value["renderer"] = ui->renderer->currentIndex();

	try
	{
#ifdef _WIN32
		std::ofstream fout(setting.wstring().c_str());
#else
		std::ofstream fout(setting.string().c_str());
#endif
		Json::StyledStreamWriter writer;
		writer.write(fout, value);
		fout.close();

		QApplication::instance()->quit();
	}
	catch (std::exception& e)
	{
	}

}

void ChangeRenderer::on_buttonCancel_clicked()
{
	QApplication::instance()->quit();
}

max_stars_repo_name: ZCube/OverlayProc

max_issues_repo_name: ZCube/OverlayProc

max_forks_repo_name: ZCube/OverlayProc


text:
#include "mod_list.hpp"
#include "download.hpp"

#include <SimpleJSON/stringify/jss.hpp>
#include <SimpleJSON/parse/jsd.hpp>
#include <SimpleJSON/parse/jsd_convenience.hpp>
#include <SimpleJSON/utility/beauty_stream.hpp>
#include <SimpleJSON/utility/fill_missing.hpp>

#include <curl/curl.h>
#include <attendee/attendee.hpp>
#include <fstream>

#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/device/file.hpp>
#include <iostream>

using namespace std::string_literals;

//#####################################################################################################################
std::pair <std::string, std::string> FileDescription::download() const
{
    auto url = "https://minecraft.curseforge.com" + href + "/download";

    int code;
    std::string content;
    std::string redirect = url;
    do {
        auto [result, codeIn, redirectIn, contentStr] = ::followRedirect(redirect);
        code = codeIn;

        if (result != 0)
            throw std::runtime_error(url + " could not be downloaded: " + curl_easy_strerror(static_cast <CURLcode>(result)));

        if (code == 200)
            content = std::move(contentStr);
        else
            redirect = std::move(redirectIn);
    } while (code >= 300 && code < 400);
    return {content, redirect};
}
//---------------------------------------------------------------------------------------------------------------------
void FileDescription::download(std::filesystem::path const& to)
{
    auto [content, href] = download();
    auto pos = href.find_last_of('/');

    name = std::string{std::begin(href) + pos + 1, std::end(href)};
    name = attendee::response::url_decode(name);
    std::ofstream writer{to / name, std::ios_base::binary};
    writer << content;
}
//#####################################################################################################################
void Mod::readName()
{
    std::string href = "https://www.curseforge.com/minecraft/mc-mods/" + id;
    auto [result, content] = download(href);

    if (result != 0)
        throw std::runtime_error(href + " could not be downloaded: " + curl_easy_strerror(static_cast <CURLcode>(result)));

    auto getName = [&]()
    {
        auto p1 = content.find("
");
        auto p2 = content.find("
", p1) + 17;
        auto p3 = content.find("
", p2);
        return std::string{std::begin(content) + p2, std::begin(content) + p3};
    };

    name = getName();
}
//---------------------------------------------------------------------------------------------------------------------
std::string Mod::getProjectUrl() const
{
    std::string href = "https://www.curseforge.com/minecraft/mc-mods/" + id;
    auto [result, content] = download(href);

    if (result != 0)
        throw std::runtime_error(href + " could not be downloaded: " + curl_easy_strerror(static_cast <CURLcode>(result)));

    auto pos = content.find("
");
    pos = content.find("<a href=\"", pos) + 9;
    auto pos2 = content.find("\"", pos);

    return std::string{std::begin(content) + pos, std::begin(content) + pos2};
}
//---------------------------------------------------------------------------------------------------------------------
std::string Mod::getProjectUrlFiles() const
{
    return getProjectUrl() + "/files";
}
//---------------------------------------------------------------------------------------------------------------------
bool Mod::isNewer(long long epoch)
{
    if (!newestInstalled)
        return true;
    else
        return newestInstalled.value() < epoch;
}
//---------------------------------------------------------------------------------------------------------------------
std::vector <FileDescription> Mod::getFiles(int page) const
{
    std::string url = getProjectUrlFiles();
    if (page > 1)
        url += "?page="s + std::to_string(page);

    auto [result, content] = download(url);

    if (result != 0)
        throw std::runtime_error(url + " could not be downloaded: " + curl_easy_strerror(static_cast <CURLcode>(result)));

    auto extract = [&content](int trPos, std::string term, char const* end)
    {
        auto pos = content.find(term, trPos) + term.length();
        if (pos == std::string::npos)
            return std::string{""};
        return std::string{std::begin(content) + pos, std::begin(content) + content.find(end, pos)};
    };

    auto tbody = content.find("
");
    std::vector <FileDescription> list;
    for (int trPos = content.find("
", tbody); static_cast <std::size_t> (trPos) != std::string::npos; ++trPos)
    {
        FileDescription descr;
        descr.href = extract(trPos, R"()", "<");
        descr.timeEpoch = std::stoll(extract(trPos, R"(data-epoch=")", "\""));
        descr.timeHuman = extract(trPos, R"(time-processed=")", "\"");
        descr.mcVersion = extract(trPos, R"()", "<");

        list.push_back(descr);

        // next
        trPos = content.find("
", trPos + 1);
        if (trPos == 0 || static_cast <std::size_t> (trPos) == std::string::npos)
            break;
    }
    return list;
}
//#####################################################################################################################
void ModList::load(std::filesystem::path const& path)
{
    auto stream = std::ifstream{path.string(), std::ios_base::binary};
    auto tree = JSON::parse_json(stream);
    JSON::fill_missing <decltype(*this)> ("", tree);
    JSON::parse(*this, "", tree);
}
//---------------------------------------------------------------------------------------------------------------------
void ModList::save(std::filesystem::path const& path) const
{
    namespace io = boost::iostreams;
    io::filtering_ostream out;
    out.push(JSON::BeautifiedStreamWrapper{});
    out.push(io::file_sink(path.string()));
    JSON::stringify(out, "modlist", *this);
}
//#####################################################################################################################