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/tablet/memrowset.h"
  
  #include <memory>
  #include <string>
  #include <utility>
  #include <vector>
  
  #include <gflags/gflags.h>
  #include <glog/logging.h>
  
  #include "kudu/codegen/compilation_manager.h"
  #include "kudu/codegen/row_projector.h"
  #include "kudu/common/columnblock.h"
  #include "kudu/common/encoded_key.h"
  #include "kudu/common/row.h"
  #include "kudu/common/row_changelist.h"
  #include "kudu/common/rowblock.h"
  #include "kudu/common/scan_spec.h"
  #include "kudu/consensus/log_anchor_registry.h"
  #include "kudu/consensus/opid.pb.h"
  #include "kudu/gutil/dynamic_annotations.h"
  #include "kudu/gutil/port.h"
  #include "kudu/gutil/strings/substitute.h"
  #include "kudu/tablet/compaction.h"
  #include "kudu/tablet/mutation.h"
  #include "kudu/tablet/mvcc.h"
  #include "kudu/tablet/tablet.pb.h"
  #include "kudu/util/flag_tags.h"
  #include "kudu/util/mem_tracker.h"
  #include "kudu/util/memory/memory.h"
  
  DEFINE_bool(mrs_use_codegen, true, "whether the memrowset should use code "
              "generation for iteration");
  TAG_FLAG(mrs_use_codegen, hidden);
  
  using std::shared_ptr;
  using std::string;
  using std::unique_ptr;
  using std::vector;
  
  namespace kudu { namespace tablet {
  
  using consensus::OpId;
  using fs::IOContext;
  using log::LogAnchorRegistry;
  using strings::Substitute;
  
  static const int kInitialArenaSize = 16;
  
  bool MRSRow::IsGhost() const {
    const Mutation *mut_tail = header_->redo_tail;
    if (mut_tail == nullptr) {
      return false;
    }
    RowChangeListDecoder decoder(mut_tail->changelist());
    Status s = decoder.Init();
    if (!PREDICT_TRUE(s.ok())) {
      LOG(FATAL) << Substitute("Failed to decode: $0 ($1)",
                               mut_tail->changelist().ToString(*schema()),
                               s.ToString());
    }
    return decoder.is_delete();
  }
  
  namespace {
  
  shared_ptr<MemTracker> CreateMemTrackerForMemRowSet(
      int64_t id, shared_ptr<MemTracker> parent_tracker) {
    string mem_tracker_id = Substitute("MemRowSet-$0", id);
    return MemTracker::CreateTracker(-1, mem_tracker_id, std::move(parent_tracker));
  }
  
  } // anonymous namespace
  
  Status MemRowSet::Create(int64_t id,
                           const Schema &schema,
                           LogAnchorRegistry* log_anchor_registry,
                           shared_ptr<MemTracker> parent_tracker,
                           shared_ptr<MemRowSet>* mrs) {
    auto local_mrs(MemRowSet::make_shared(
        id, schema, log_anchor_registry, std::move(parent_tracker)));
    *mrs = std::move(local_mrs);
    return Status::OK();
  }
  
  MemRowSet::MemRowSet(int64_t id,
                       const Schema &schema,
                       LogAnchorRegistry* log_anchor_registry,
                       shared_ptr<MemTracker> parent_tracker)
    : id_(id),
      schema_(schema),
      allocator_(new MemoryTrackingBufferAllocator(
          HeapBufferAllocator::Get(),
          CreateMemTrackerForMemRowSet(id, std::move(parent_tracker)))),
      arena_(new ThreadSafeMemoryTrackingArena(kInitialArenaSize, allocator_)),
      tree_(arena_),
      debug_insert_count_(0),
      debug_update_count_(0),
      anchorer_(log_anchor_registry, Substitute("MemRowSet-$0", id_)),
      has_been_compacted_(false),
      live_row_count_(0) {
    CHECK(schema.has_column_ids());
    ANNOTATE_BENIGN_RACE(&debug_insert_count_, "insert count isnt accurate");
    ANNOTATE_BENIGN_RACE(&debug_update_count_, "update count isnt accurate");
  }
  
  MemRowSet::~MemRowSet() {
  }
  
  Status MemRowSet::DebugDump(vector<string> *lines) {
    unique_ptr<Iterator> iter(NewIterator());
    RETURN_NOT_OK(iter->Init(nullptr));
    while (iter->HasNext()) {
      MRSRow row = iter->GetCurrentRow();
      LOG_STRING(INFO, lines)
        << "@" << row.insertion_timestamp() << ": row "
        << schema_.DebugRow(row)
        << " mutations=" << Mutation::StringifyMutationList(schema_, row.header_->redo_head)
        << std::endl;
      iter->Next();
    }
  
    return Status::OK();
  }
  
  
  Status MemRowSet::Insert(Timestamp timestamp,
                           const ConstContiguousRow& row,
                           const OpId& op_id) {
    CHECK(row.schema()->has_column_ids());
    DCHECK_SCHEMA_EQ(schema_, *row.schema());
  
    {
      faststring enc_key_buf;
      schema_.EncodeComparableKey(row, &enc_key_buf);
      Slice enc_key(enc_key_buf);
  
      btree::PreparedMutation<MSBTreeTraits> mutation(enc_key);
      mutation.Prepare(&tree_);
  
      // TODO: for now, the key ends up stored doubly --
      // once encoded in the btree key, and again in the value
      // (unencoded).
      // That's not very memory-efficient!
  
      if (mutation.exists()) {
        // It's OK for it to exist if it's just a "ghost" row -- i.e the
        // row is deleted.
        MRSRow ms_row(this, mutation.current_mutable_value());
        if (!ms_row.IsGhost()) {
          return Status::AlreadyPresent("key already present");
        }
  
        // Insert a "reinsert" mutation.
        return Reinsert(timestamp, row, &ms_row);
      }
  
      // Copy the non-encoded key onto the stack since we need
      // to mutate it when we relocate its Slices into our arena.
      DEFINE_MRSROW_ON_STACK(this, mrsrow, mrsrow_slice);
      mrsrow.header_->insertion_timestamp = timestamp;
      mrsrow.header_->redo_head = nullptr;
      mrsrow.header_->redo_tail = nullptr;
      RETURN_NOT_OK(mrsrow.CopyRow(row, arena_.get()));
  
      CHECK(mutation.Insert(mrsrow_slice))
      << "Expected to be able to insert, since the prepared mutation "
      << "succeeded!";
    }
  
    anchorer_.AnchorIfMinimum(op_id.index());
  
    debug_insert_count_++;
    live_row_count_.Increment();
    return Status::OK();
  }
  
  Status MemRowSet::Reinsert(Timestamp timestamp, const ConstContiguousRow& row, MRSRow *ms_row) {
    DCHECK_SCHEMA_EQ(schema_, *row.schema());
  
    // Encode the REINSERT mutation
    faststring buf;
    RowChangeListEncoder encoder(&buf);
    encoder.SetToReinsert(row);
  
    // Move the REINSERT mutation itself into our Arena.
    Mutation *mut = Mutation::CreateInArena(arena_.get(), timestamp, encoder.as_changelist());
  
    // Append the mutation into the row's mutation list.
    // This function has "release" semantics which ensures that the memory writes
    // for the mutation are fully published before any concurrent reader sees
    // the appended mutation.
    mut->AppendToListAtomic(&ms_row->header_->redo_head, &ms_row->header_->redo_tail);
  
    live_row_count_.Increment();
    return Status::OK();
  }
  
  Status MemRowSet::MutateRow(Timestamp timestamp,
                              const RowSetKeyProbe &probe,
                              const RowChangeList &delta,
                              const consensus::OpId& op_id,
                              const IOContext* /*io_context*/,
                              ProbeStats* stats,
                              OperationResultPB *result) {
    {
      btree::PreparedMutation<MSBTreeTraits> mutation(probe.encoded_key_slice());
      mutation.Prepare(&tree_);
  
      if (!mutation.exists()) {
        return Status::NotFound("not in memrowset");
      }
  
      MRSRow row(this, mutation.current_mutable_value());
  
      // If the row exists, it may still be a "ghost" row -- i.e a row
      // that's been deleted. If that's the case, we should treat it as
      // NotFound.
      if (row.IsGhost()) {
        return Status::NotFound("not in memrowset (ghost)");
      }
  
      // Append to the linked list of mutations for this row.
      Mutation *mut = Mutation::CreateInArena(arena_.get(), timestamp, delta);
  
      // This function has "release" semantics which ensures that the memory writes
      // for the mutation are fully published before any concurrent reader sees
      // the appended mutation.
      mut->AppendToListAtomic(&row.header_->redo_head, &row.header_->redo_tail);
  
      MemStoreTargetPB* target = result->add_mutated_stores();
      target->set_mrs_id(id_);
    }
  
    stats->mrs_consulted++;
  
    anchorer_.AnchorIfMinimum(op_id.index());
    debug_update_count_++;
    if (delta.is_delete()) {
      live_row_count_.IncrementBy(-1);
    }
    return Status::OK();
  }
  
  Status MemRowSet::CheckRowPresent(const RowSetKeyProbe &probe, const IOContext* /*io_context*/,
                                    bool* present, ProbeStats* stats) const {
    // Use a PreparedMutation here even though we don't plan to mutate. Even though
    // this takes a lock rather than an optimistic copy, it should be a very short
    // critical section, and this call is only made on updates, which are rare.
  
    stats->mrs_consulted++;
  
    btree::PreparedMutation<MSBTreeTraits> mutation(probe.encoded_key_slice());
    mutation.Prepare(const_cast<MSBTree *>(&tree_));
  
    if (!mutation.exists()) {
      *present = false;
      return Status::OK();
    }
  
    // TODO(perf): using current_mutable_value() will actually change the data's
    // version number, even though we're not going to do any mutation. This would
    // make concurrent readers retry, even though they don't have to (we aren't
    // actually mutating anything here!)
    MRSRow row(this, mutation.current_mutable_value());
  
    // If the row exists, it may still be a "ghost" row -- i.e a row
    // that's been deleted. If that's the case, we should treat it as
    // NotFound.
    *present = !row.IsGhost();
    return Status::OK();
  }
  
  MemRowSet::Iterator *MemRowSet::NewIterator(const RowIteratorOptions& opts) const {
    return new MemRowSet::Iterator(shared_from_this(), tree_.NewIterator(), opts);
  }
  
  MemRowSet::Iterator *MemRowSet::NewIterator() const {
    // TODO(todd): can we kill this function? should be only used by tests?
    RowIteratorOptions opts;
    opts.projection = &schema();
    return NewIterator(opts);
  }
  
  Status MemRowSet::NewRowIterator(const RowIteratorOptions& opts,
                                   unique_ptr<RowwiseIterator>* out) const {
    out->reset(NewIterator(opts));
    return Status::OK();
  }
  
  Status MemRowSet::NewCompactionInput(const Schema* projection,
                                       const MvccSnapshot& snap,
                                       const IOContext* /*io_context*/,
                                       unique_ptr<CompactionInput>* out) const  {
    out->reset(CompactionInput::Create(*this, projection, snap));
    return Status::OK();
  }
  
  Status MemRowSet::GetBounds(string *min_encoded_key,
                              string *max_encoded_key) const {
    return Status::NotSupported("");
  }
  
  // Virtual interface allows two possible row projector implementations
  class MemRowSet::Iterator::MRSRowProjector {
   public:
    typedef RowProjector::ProjectionIdxMapping ProjectionIdxMapping;
    virtual ~MRSRowProjector() {}
    virtual Status ProjectRowForRead(const MRSRow& src_row,
                                     RowBlockRow* dst_row,
                                     Arena* arena) = 0;
    virtual Status ProjectRowForRead(const ConstContiguousRow& src_row,
                                     RowBlockRow* dst_row,
                                     Arena* arena) = 0;
    virtual const vector<ProjectionIdxMapping>& base_cols_mapping() const = 0;
    virtual Status Init() = 0;
  };
  
  namespace {
  
  typedef MemRowSet::Iterator::MRSRowProjector MRSRowProjector;
  
  template<class ActualProjector>
  class MRSRowProjectorImpl : public MRSRowProjector {
   public:
    explicit MRSRowProjectorImpl(unique_ptr<ActualProjector> actual)
      : actual_(std::move(actual)) {}
  
    Status Init() override { return actual_->Init(); }
  
    Status ProjectRowForRead(const MRSRow& src_row, RowBlockRow* dst_row,
                             Arena* arena) override {
      return actual_->ProjectRowForRead(src_row, dst_row, arena);
    }
    Status ProjectRowForRead(const ConstContiguousRow& src_row,
                             RowBlockRow* dst_row,
                             Arena* arena) override {
      return actual_->ProjectRowForRead(src_row, dst_row, arena);
    }
  
    const vector<ProjectionIdxMapping>& base_cols_mapping() const override {
      return actual_->base_cols_mapping();
    }
  
   private:
    unique_ptr<ActualProjector> actual_;
  };
  
  // If codegen is enabled, then generates a codegen::RowProjector;
  // otherwise makes a regular one.
  unique_ptr<MRSRowProjector> GenerateAppropriateProjector(
    const Schema* base, const Schema* projection) {
    // Attempt code-generated implementation
    if (FLAGS_mrs_use_codegen) {
      unique_ptr<codegen::RowProjector> actual;
      if (codegen::CompilationManager::GetSingleton()->RequestRowProjector(
            base, projection, &actual)) {
        return unique_ptr<MRSRowProjector>(
          new MRSRowProjectorImpl<codegen::RowProjector>(std::move(actual)));
      }
    }
  
    // Proceed with default implementation
    unique_ptr<RowProjector> actual(new RowProjector(base, projection));
    return unique_ptr<MRSRowProjector>(
      new MRSRowProjectorImpl<RowProjector>(std::move(actual)));
  }
  
  } // anonymous namespace
  
  MemRowSet::Iterator::Iterator(const std::shared_ptr<const MemRowSet>& mrs,
                                MemRowSet::MSBTIter* iter,
                                RowIteratorOptions opts)
      : memrowset_(mrs),
        iter_(iter),
        opts_(std::move(opts)),
        projector_(
            GenerateAppropriateProjector(&mrs->schema_nonvirtual(), opts_.projection)),
        delta_projector_(&mrs->schema_nonvirtual(), opts_.projection),
        projection_vc_is_deleted_idx_(opts_.projection->first_is_deleted_virtual_column_idx()),
        state_(kUninitialized) {
    // TODO(todd): various code assumes that a newly constructed iterator
    // is pointed at the beginning of the dataset. This causes a redundant
    // seek. Could make this lazy instead, or change the semantics so that
    // a seek is required (probably the latter)
    iter_->SeekToStart();
  }
  
  MemRowSet::Iterator::~Iterator() {}
  
  Status MemRowSet::Iterator::Init(ScanSpec *spec) {
    DCHECK_EQ(state_, kUninitialized);
  
    RETURN_NOT_OK(projector_->Init());
    RETURN_NOT_OK(delta_projector_.Init());
  
    if (spec && spec->lower_bound_key()) {
      bool exact;
      const Slice &lower_bound = spec->lower_bound_key()->encoded_key();
      if (!iter_->SeekAtOrAfter(lower_bound, &exact)) {
        // Lower bound is after the end of the key range, no rows will
        // pass the predicate so we can stop the scan right away.
        state_ = kFinished;
        return Status::OK();
      }
    }
  
    if (spec && spec->exclusive_upper_bound_key()) {
      const Slice &upper_bound = spec->exclusive_upper_bound_key()->encoded_key();
      exclusive_upper_bound_.reset(upper_bound);
    }
  
    state_ = kScanning;
    return Status::OK();
  }
  
  Status MemRowSet::Iterator::SeekAtOrAfter(const Slice &key, bool *exact) {
    DCHECK_NE(state_, kUninitialized) << "not initted";
  
    if (key.size() > 0) {
      ConstContiguousRow row_slice(&memrowset_->schema(), key);
      memrowset_->schema().EncodeComparableKey(row_slice, &tmp_buf);
    } else {
      // Seeking to empty key shouldn't try to run any encoding.
      tmp_buf.resize(0);
    }
  
    if (iter_->SeekAtOrAfter(Slice(tmp_buf), exact) ||
        key.size() == 0) {
      return Status::OK();
    } else {
      return Status::NotFound("no match in memrowset");
    }
  }
  
  Status MemRowSet::Iterator::NextBlock(RowBlock *dst) {
    // TODO: add dcheck that dst->schema() matches our schema
    // also above TODO applies to a lot of other CopyNextRows cases
  
    DCHECK_NE(state_, kUninitialized) << "not initted";
    if (PREDICT_FALSE(!iter_->IsValid())) {
      dst->Resize(0);
      return Status::NotFound("end of iter");
    }
    if (PREDICT_FALSE(state_ != kScanning)) {
      dst->Resize(0);
      return Status::OK();
    }
    if (PREDICT_FALSE(dst->row_capacity() == 0)) {
      return Status::OK();
    }
  
    // Reset rowblock arena to eventually reach appropriate buffer size.
    // Always allocating the full capacity is only a problem for the last block.
    dst->Resize(dst->row_capacity());
    if (dst->arena()) {
      dst->arena()->Reset();
    }
  
    // Fill
    dst->selection_vector()->SetAllTrue();
    size_t fetched;
    RETURN_NOT_OK(FetchRows(dst, &fetched));
    DCHECK_LE(0, fetched);
    DCHECK_LE(fetched, dst->nrows());
  
    // Clear unreached bits by resizing
    dst->Resize(fetched);
  
    return Status::OK();
  }
  
  Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
    *fetched = 0;
    do {
      RowBlockRow dst_row = dst->row(*fetched);
  
      // Copy the row into the destination, including projection
      // and relocating slices.
      // TODO(todd): can we share some code here with CopyRowToArena() from row.h
      // or otherwise put this elsewhere?
      Slice v = iter_->GetCurrentValue();
      MRSRow row(memrowset_.get(), v);
  
      // Short-circuit if we've exceeded the iteration's upper bound.
      if (has_upper_bound() && out_of_bounds(iter_->GetCurrentKey())) {
        state_ = kFinished;
        break;
      }
  
      // The snapshots in 'opts_' represent a time range that this iterator must
      // respect. There are two possible cases:
      //
      // 1. 'snap_to_exclude' is unset but 'snap_to_include' is set. The time
      //    range is [INF, snap_to_include).
      // 2. Both 'snap_to exclude' and 'snap_to_include' are set. The time range
      //    is [snap_to_exclude, snap_to_include).
      //
      // If the row's insertion timestamp is committed in 'snap_to_exclude', it
      // means the insertion was outside this iterator's time range (i.e. the
      // insert was "excluded"). However, subsequent mutations may be inside the
      // time range, so we must still project the row and walk its mutation list.
      bool insert_excluded = opts_.snap_to_exclude &&
                             opts_.snap_to_exclude->IsCommitted(row.insertion_timestamp());
      bool unset_in_sel_vector;
      ApplyStatus apply_status;
      if (insert_excluded || opts_.snap_to_include.IsCommitted(row.insertion_timestamp())) {
        RETURN_NOT_OK(projector_->ProjectRowForRead(row, &dst_row, dst->arena()));
  
        // Roll-forward MVCC for committed updates.
        Mutation* redo_head = reinterpret_cast<Mutation*>(
            base::subtle::Acquire_Load(reinterpret_cast<AtomicWord*>(&row.header_->redo_head)));
        RETURN_NOT_OK(ApplyMutationsToProjectedRow(
            redo_head, &dst_row, dst->arena(), insert_excluded, &apply_status));
        unset_in_sel_vector = (apply_status == APPLIED_AND_DELETED && !opts_.include_deleted_rows) ||
                              (apply_status == NONE_APPLIED && insert_excluded) ||
                              (apply_status == APPLIED_AND_UNOBSERVABLE);
      } else {
        // The insertion is too new; the entire row should be omitted.
        unset_in_sel_vector = true;
      }
  
      if (unset_in_sel_vector) {
        dst->selection_vector()->SetRowUnselected(*fetched);
  
        // In debug mode, fill the row data for easy debugging
        #ifndef NDEBUG
        if (state_ != kFinished) {
          dst_row.OverwriteWithPattern("MVCCMVCCMVCCMVCCMVCCMVCC"
                                       "MVCCMVCCMVCCMVCCMVCCMVCC"
                                       "MVCCMVCCMVCCMVCCMVCCMVCC");
        }
        #endif
      } else if (projection_vc_is_deleted_idx_ != Schema::kColumnNotFound) {
        UnalignedStore(dst_row.mutable_cell_ptr(projection_vc_is_deleted_idx_),
                       apply_status == APPLIED_AND_DELETED);
      }
  
      ++*fetched;
    } while (iter_->Next() && *fetched < dst->nrows());
  
    return Status::OK();
  }
  
  Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
      const Mutation* mutation_head, RowBlockRow* dst_row, Arena* dst_arena,
      bool insert_excluded, ApplyStatus* apply_status) {
    ApplyStatus local_apply_status = NONE_APPLIED;
  
    // Fast short-circuit the likely case of a row which was inserted and never
    // updated.
    if (PREDICT_TRUE(mutation_head == nullptr)) {
      *apply_status = local_apply_status;
      return Status::OK();
    }
  
    // In order to find unobservable rows, we need to track row liveness at the
    // start and end of the time range. If a row was dead at both ends, its
    // lifespan must have been a subset of the time range and it should be
    // excluded from the results.
    //
    // Finding 'is_deleted_end' is relatively straight-forward: we use each
    // relevant mutation to drive a liveness state machine, and after we're done
    // applying, 'is_deleted_end' is just the final value of that state machine.
    //
    // Finding 'is_deleted_start' is trickier. If the insertion was inside the
    // time range, we know the value is true because the row was dead prior to the
    // insertion and the insertion happened after the start of the time range.
    // However, if the insertion was excluded from the time range, the value is
    // going to be whatever the value of the liveness state machine was at the
    // start of the time range.
    bool is_deleted_start = !insert_excluded;
    bool is_deleted_end = false;
  
    for (const Mutation *mut = mutation_head;
         mut != nullptr;
         mut = mut->acquire_next()) {
      if (!opts_.snap_to_include.IsCommitted(mut->timestamp_)) {
        // This mutation is too new and should be omitted.
        //
        // All subsequent mutations are also too new because their timestamps are
        // guaranteed to be equal to or greater than this mutation's timestamp.
        break;
      }
  
      // If the mutation is too old, we still need to apply it (so that the column
      // values are correct if we see a relevant mutation later), but it doesn't
      // count towards the overall "application status".
      if (!opts_.snap_to_exclude ||
          !opts_.snap_to_exclude->IsCommitted(mut->timestamp_)) {
  
        // This is the first mutation within the time range, so we may use it to
        // initialize 'is_deleted_start'.
        if (local_apply_status == NONE_APPLIED && insert_excluded) {
          is_deleted_start = is_deleted_end;
        }
        local_apply_status = APPLIED_AND_PRESENT;
      }
  
      // Apply the mutation.
  
      // Check if it's a deletion.
      RowChangeListDecoder decoder(mut->changelist());
      RETURN_NOT_OK(decoder.Init());
      if (decoder.is_delete()) {
        decoder.TwiddleDeleteStatus(&is_deleted_end);
      } else {
        DCHECK(decoder.is_update() || decoder.is_reinsert());
        if (decoder.is_reinsert()) {
          decoder.TwiddleDeleteStatus(&is_deleted_end);
        }
  
        // TODO(todd): this is slow, since it makes multiple passes through the rowchangelist.
        // Instead, we should keep the backwards mapping of columns.
        for (const RowProjector::ProjectionIdxMapping& mapping : projector_->base_cols_mapping()) {
          RowChangeListDecoder decoder(mut->changelist());
          RETURN_NOT_OK(decoder.Init());
          ColumnBlock dst_col = dst_row->column_block(mapping.first);
          RETURN_NOT_OK(decoder.ApplyToOneColumn(dst_row->row_index(), &dst_col,
                                                 memrowset_->schema_nonvirtual(),
                                                 mapping.second, dst_arena));
        }
      }
    }
  
    if (opts_.snap_to_exclude && is_deleted_start && is_deleted_end) {
      // The row's lifespan was a subset of the time range. It can't be observed,
      // so it should definitely not show up in the results.
      //
      // Note: we condition on 'snap_to_exclude' because although insert_excluded
      // is false on some closed time range scans, it's also false in all open
      // time range scans, and we don't want this heuristic to fire in the latter case.
      local_apply_status = APPLIED_AND_UNOBSERVABLE;
    } else if (is_deleted_end && local_apply_status == APPLIED_AND_PRESENT) {
      // The row was selected and deleted. It may be omitted from the results,
      // depending on whether the results should include deleted rows or not.
      local_apply_status = APPLIED_AND_DELETED;
    }
  
    *apply_status = local_apply_status;
    return Status::OK();
  }
  
  // Copy the current MRSRow to the 'dst_row' provided using the iterator projection schema.
  Status MemRowSet::Iterator::GetCurrentRow(RowBlockRow* dst_row,
                                            Arena* row_arena,
                                            Mutation** redo_head,
                                            Arena* mutation_arena,
                                            Timestamp* insertion_timestamp) {
  
    DCHECK(redo_head != nullptr);
  
    // Get the row from the MemRowSet. It may have a different schema from the iterator projection.
    MRSRow src_row = GetCurrentRow();
  
    *insertion_timestamp = src_row.insertion_timestamp();
  
    // Project the RowChangeList if required
    *redo_head = src_row.acquire_redo_head();
    if (!delta_projector_.is_identity()) {
      DCHECK(mutation_arena != nullptr);
  
      Mutation *prev_redo = nullptr;
      *redo_head = nullptr;
      for (const Mutation *mut = src_row.redo_head();
           mut != nullptr;
           mut = mut->acquire_next()) {
  
        delta_buf_.clear();
        RowChangeListEncoder enc(&delta_buf_);
        RETURN_NOT_OK(RowChangeListDecoder::ProjectChangeList(delta_projector_,
                                                              mut->changelist(),
                                                              &enc));
  
        // The projection resulted in an empty mutation (e.g. update of a removed column)
        if (enc.is_empty()) continue;
  
        Mutation *mutation = Mutation::CreateInArena(mutation_arena,
                                                     mut->timestamp(),
                                                     RowChangeList(delta_buf_));
        if (prev_redo != nullptr) {
          prev_redo->set_next(mutation);
        } else {
          *redo_head = mutation;
        }
        prev_redo = mutation;
      }
    }
  
    // Project the Row
    return projector_->ProjectRowForRead(src_row, dst_row, row_arena);
  }
  
  } // namespace tablet
  } // namespace kudu
  