// 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/deltafile.h"

#include <functional>
#include <memory>
#include <ostream>
#include <string>
#include <utility>

#include <gflags/gflags.h>

#include "kudu/cfile/binary_plain_block.h"
#include "kudu/cfile/block_handle.h"
#include "kudu/cfile/block_pointer.h"
#include "kudu/cfile/cfile_reader.h"
#include "kudu/cfile/cfile_util.h"
#include "kudu/cfile/cfile_writer.h"
#include "kudu/cfile/index_btree.h"
#include "kudu/common/common.pb.h"
#include "kudu/common/row_changelist.h"
#include "kudu/common/rowblock.h"
#include "kudu/common/scan_spec.h"
#include "kudu/common/schema.h"
#include "kudu/common/timestamp.h"
#include "kudu/common/types.h"
#include "kudu/fs/block_id.h"
#include "kudu/fs/block_manager.h"
#include "kudu/fs/fs_manager.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/tablet/delta_relevancy.h"
#include "kudu/tablet/mvcc.h"
#include "kudu/tablet/rowset.h"
#include "kudu/tablet/tablet.pb.h"
#include "kudu/util/compression/compression_codec.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/trace.h"

DECLARE_bool(cfile_lazy_open);
DEFINE_int32(deltafile_default_block_size, 32*1024,
            "Block size for delta files. In the future, this may become configurable "
             "on a per-table basis.");
TAG_FLAG(deltafile_default_block_size, experimental);

DEFINE_string(deltafile_default_compression_codec, "lz4",
              "The compression codec used when writing deltafiles.");
TAG_FLAG(deltafile_default_compression_codec, experimental);

using std::shared_ptr;
using std::string;
using std::unique_ptr;
using std::vector;
using strings::Substitute;

namespace kudu {

class MemTracker;

using cfile::BinaryPlainBlockDecoder;
using cfile::BlockPointer;
using cfile::CFileReader;
using cfile::IndexTreeIterator;
using cfile::ReaderOptions;
using fs::BlockCreationTransaction;
using fs::BlockManager;
using fs::IOContext;
using fs::ReadableBlock;
using fs::WritableBlock;

namespace tablet {

const char * const DeltaFileReader::kDeltaStatsEntryName = "deltafilestats";

DeltaFileWriter::DeltaFileWriter(unique_ptr<WritableBlock> block)
#ifndef NDEBUG
 : has_appended_(false)
#endif
{ // NOLINT(*)
  cfile::WriterOptions opts;
  opts.write_validx = true;
  opts.storage_attributes.cfile_block_size = FLAGS_deltafile_default_block_size;
  opts.storage_attributes.encoding = PLAIN_ENCODING;
  opts.storage_attributes.compression =
      GetCompressionCodecType(FLAGS_deltafile_default_compression_codec);

  // The CFile value index is 'compressed' by truncating delta values to only
  // contain the delta key. The entire deltakey is required in order to support
  // efficient seeks without deserializing the entire block. The generic value
  // index optimization is disabled, since it could truncate portions of the
  // deltakey.
  //
  // Note: The deltafile usage of the CFile value index is irregular, since it
  // inserts values in non-sorted order (the timestamp portion of the deltakey
  // in UNDO files is sorted in descending order). This doesn't appear to cause
  // problems in practice.
  opts.optimize_index_keys = false;
  opts.validx_key_encoder = [] (const void* value, faststring* buffer) {
    buffer->clear();
    const Slice* s1 = static_cast<const Slice*>(value);
    Slice s2(*s1);
    DeltaKey key;
    CHECK_OK(key.DecodeFrom(&s2));
    key.EncodeTo(buffer);
  };

  writer_.reset(new cfile::CFileWriter(std::move(opts),
                                       GetTypeInfo(BINARY),
                                       false,
                                       std::move(block)));
}

Status DeltaFileWriter::Start() {
  return writer_->Start();
}

Status DeltaFileWriter::Finish() {
  BlockManager* bm = writer_->block()->block_manager();
  unique_ptr<BlockCreationTransaction> transaction = bm->NewCreationTransaction();
  RETURN_NOT_OK(FinishAndReleaseBlock(transaction.get()));
  return transaction->CommitCreatedBlocks();
}

Status DeltaFileWriter::FinishAndReleaseBlock(BlockCreationTransaction* transaction) {
  if (writer_->written_value_count() == 0) {
    return Status::Aborted("no deltas written");
  }
  return writer_->FinishAndReleaseBlock(transaction);
}

Status DeltaFileWriter::DoAppendDelta(const DeltaKey &key,
                                      const RowChangeList &delta) {
  Slice delta_slice(delta.slice());
  tmp_buf_.clear();

  // Write the encoded form of the key to the file.
  key.EncodeTo(&tmp_buf_);

  tmp_buf_.append(delta_slice.data(), delta_slice.size());
  Slice tmp_buf_slice(tmp_buf_);

  return writer_->AppendEntries(&tmp_buf_slice, 1);
}

template<>
Status DeltaFileWriter::AppendDelta<REDO>(
  const DeltaKey &key, const RowChangeList &delta) {

#ifndef NDEBUG
  // Sanity check insertion order in debug mode.
  if (has_appended_) {
    DCHECK(last_key_.CompareTo<REDO>(key) <= 0)
      << "must insert redo deltas in sorted order (ascending key, then ascending ts): "
      << "got key " << key.ToString() << " after "
      << last_key_.ToString();
  }
  has_appended_ = true;
  last_key_ = key;
#endif

  return DoAppendDelta(key, delta);
}

template<>
Status DeltaFileWriter::AppendDelta<UNDO>(
  const DeltaKey &key, const RowChangeList &delta) {

#ifndef NDEBUG
  // Sanity check insertion order in debug mode.
  if (has_appended_) {
    DCHECK(last_key_.CompareTo<UNDO>(key) <= 0)
      << "must insert undo deltas in sorted order (ascending key, then descending ts): "
      << "got key " << key.ToString() << " after "
      << last_key_.ToString();
  }
  has_appended_ = true;
  last_key_ = key;
#endif

  return DoAppendDelta(key, delta);
}

void DeltaFileWriter::WriteDeltaStats(std::unique_ptr<DeltaStats> stats) {
  DeltaStatsPB delta_stats_pb;
  stats->ToPB(&delta_stats_pb);

  faststring buf;
  pb_util::SerializeToString(delta_stats_pb, &buf);
  writer_->AddMetadataPair(DeltaFileReader::kDeltaStatsEntryName, buf.ToString());
  delta_stats_ = std::move(stats);
}


////////////////////////////////////////////////////////////
// Reader
////////////////////////////////////////////////////////////

Status DeltaFileReader::Open(unique_ptr<ReadableBlock> block,
                             DeltaType delta_type,
                             ReaderOptions options,
                             shared_ptr<DeltaFileReader>* reader_out) {
  shared_ptr<DeltaFileReader> df_reader;
  const IOContext* io_context = options.io_context;
  RETURN_NOT_OK(DeltaFileReader::OpenNoInit(std::move(block),
                                            delta_type,
                                            std::move(options),
                                            /*delta_stats*/nullptr,
                                            &df_reader));
  RETURN_NOT_OK(df_reader->Init(io_context));

  *reader_out = df_reader;
  return Status::OK();
}

Status DeltaFileReader::OpenNoInit(unique_ptr<ReadableBlock> block,
                                   DeltaType delta_type,
                                   ReaderOptions options,
                                   unique_ptr<DeltaStats> delta_stats,
                                   shared_ptr<DeltaFileReader>* reader_out) {
  unique_ptr<CFileReader> cf_reader;
  const IOContext* io_context = options.io_context;
  RETURN_NOT_OK(CFileReader::OpenNoInit(std::move(block),
                                        std::move(options),
                                        &cf_reader));
  unique_ptr<DeltaFileReader> df_reader(
      new DeltaFileReader(std::move(cf_reader), std::move(delta_stats), delta_type));
  if (!FLAGS_cfile_lazy_open) {
    RETURN_NOT_OK(df_reader->Init(io_context));
  }

  reader_out->reset(df_reader.release());

  return Status::OK();
}

DeltaFileReader::DeltaFileReader(unique_ptr<CFileReader> cf_reader,
                                 unique_ptr<DeltaStats> delta_stats,
                                 DeltaType delta_type)
    : reader_(cf_reader.release()),
      delta_stats_(std::move(delta_stats)),
      delta_type_(delta_type) {}

Status DeltaFileReader::Init(const IOContext* io_context) {
  return init_once_.Init([this, io_context] { return InitOnce(io_context); });
}

Status DeltaFileReader::InitOnce(const IOContext* io_context) {
  // Fully open the CFileReader if it was lazily opened earlier.
  //
  // If it's already initialized, this is a no-op.
  RETURN_NOT_OK(reader_->Init(io_context));

  if (!reader_->has_validx()) {
    return Status::NotSupported("file does not have a value index!");
  }

  // Initialize delta file stats
  if (!has_delta_stats()) {
    RETURN_NOT_OK(ReadDeltaStats());
  }
  return Status::OK();
}

Status DeltaFileReader::ReadDeltaStats() {
  string filestats_pb_buf;
  if (!reader_->GetMetadataEntry(kDeltaStatsEntryName, &filestats_pb_buf)) {
    return Status::NotSupported("missing delta stats from the delta file metadata");
  }

  DeltaStatsPB deltastats_pb;
  if (!deltastats_pb.ParseFromString(filestats_pb_buf)) {
    return Status::Corruption("unable to parse the delta stats protobuf");
  }
  unique_ptr<DeltaStats> stats(new DeltaStats());
  RETURN_NOT_OK(stats->InitFromPB(deltastats_pb));
  std::lock_guard<simple_spinlock> l(stats_lock_);
  delta_stats_ = std::move(stats);
  return Status::OK();
}

bool DeltaFileReader::IsRelevantForSnapshots(
    const std::optional<MvccSnapshot>& snap_to_exclude,
    const MvccSnapshot& snap_to_include) const {
  if (!init_once_.init_succeeded()) {
    // If we're not initted, it means we have no delta stats and must
    // assume that this file is relevant for every snapshot.
    return true;
  }

  // We don't know whether the caller's intent is to apply deltas, to select
  // them, or both. As such, we must be conservative and assume 'both', which
  // means the file is relevant if any relevancy criteria is true.
  bool relevant = delta_type_ == REDO ?
                  IsDeltaRelevantForApply<REDO>(snap_to_include,
                                                delta_stats_->min_timestamp()) :
                  IsDeltaRelevantForApply<UNDO>(snap_to_include,
                                                delta_stats_->max_timestamp());
  if (snap_to_exclude) {
    // The select criteria is the same regardless of delta_type_.
    relevant |= IsDeltaRelevantForSelect(*snap_to_exclude, snap_to_include,
                                         delta_stats_->min_timestamp(),
                                         delta_stats_->max_timestamp());
  }
  return relevant;
}

Status DeltaFileReader::CloneForDebugging(FsManager* fs_manager,
                                          const shared_ptr<MemTracker>& parent_mem_tracker,
                                          shared_ptr<DeltaFileReader>* out) const {
  unique_ptr<ReadableBlock> block;
  RETURN_NOT_OK(fs_manager->OpenBlock(reader_->block_id(), &block));
  ReaderOptions options;
  options.parent_mem_tracker = parent_mem_tracker;
  return DeltaFileReader::OpenNoInit(std::move(block), delta_type_, options,
                                     /*delta_stats*/nullptr, out);
}

Status DeltaFileReader::NewDeltaIterator(const RowIteratorOptions& opts,
                                         unique_ptr<DeltaIterator>* iterator) const {
  if (IsRelevantForSnapshots(opts.snap_to_exclude, opts.snap_to_include)) {
    if (VLOG_IS_ON(2)) {
      if (!init_once_.init_succeeded()) {
        TRACE_COUNTER_INCREMENT("delta_iterators_lazy_initted", 1);

        VLOG(2) << (delta_type_ == REDO ? "REDO" : "UNDO") << " delta " << ToString()
                << " has no delta stats"
                << ": can't cull for " << opts.snap_to_include.ToString();
      } else if (delta_type_ == REDO) {
        VLOG(2) << "REDO delta " << ToString()
                << " has min ts " << delta_stats_->min_timestamp().ToString()
                << ": can't cull for " << opts.snap_to_include.ToString();
      } else {
        VLOG(2) << "UNDO delta " << ToString()
                << " has max ts " << delta_stats_->max_timestamp().ToString()
                << ": can't cull for " << opts.snap_to_include.ToString();
      }
    }

    TRACE_COUNTER_INCREMENT("delta_iterators_relevant", 1);
    // Ugly cast, but it lets the iterator fully initialize the reader
    // during its first seek.
    auto s_this = const_cast<DeltaFileReader*>(this)->shared_from_this();
    if (delta_type_ == REDO) {
      iterator->reset(new DeltaFileIterator<REDO>(std::move(s_this), opts));
    } else {
      iterator->reset(new DeltaFileIterator<UNDO>(std::move(s_this), opts));
    }
    return Status::OK();
  }
  VLOG(2) << "Culling "
          << ((delta_type_ == REDO) ? "REDO":"UNDO")
          << " delta " << ToString() << " for " << opts.snap_to_include.ToString();
  return Status::NotFound("MvccSnapshot outside the range of this delta.");
}

Status DeltaFileReader::CheckRowDeleted(rowid_t row_idx, const IOContext* io_context,
                                        bool* deleted) const {
  RETURN_NOT_OK(const_cast<DeltaFileReader*>(this)->Init(io_context));
  // If there are no deletes in the delta file at all, we can short-circuit
  // the seek.
  if (delta_stats_->delete_count() == 0) {
    *deleted = false;
    return Status::OK();
  }

  // TODO(todd): would be nice to avoid allocation here, but we don't want to
  // duplicate all the logic from NewDeltaIterator. So, we'll heap-allocate
  // for now.
  Schema empty_schema;
  RowIteratorOptions opts;
  opts.projection = &empty_schema;
  opts.io_context = io_context;
  unique_ptr<DeltaIterator> iter;
  Status s = NewDeltaIterator(opts, &iter);
  if (s.IsNotFound()) {
    *deleted = false;
    return Status::OK();
  }
  RETURN_NOT_OK(s);
  ScanSpec spec;
  RETURN_NOT_OK(iter->Init(&spec));
  RETURN_NOT_OK(iter->SeekToOrdinal(row_idx));
  RETURN_NOT_OK(iter->PrepareBatch(1, DeltaIterator::PREPARE_FOR_APPLY));

  // TODO: this does an allocation - can we stack-allocate the bitmap
  // and make SelectionVector able to "release" its buffer?
  SelectionVector sel_vec(1);
  sel_vec.SetAllTrue();
  RETURN_NOT_OK(iter->ApplyDeletes(&sel_vec));
  *deleted = !sel_vec.IsRowSelected(0);
  return Status::OK();
}

uint64_t DeltaFileReader::EstimateSize() const {
  return reader_->file_size();
}


////////////////////////////////////////////////////////////
// DeltaFileIterator
////////////////////////////////////////////////////////////

template<DeltaType Type>
DeltaFileIterator<Type>::DeltaFileIterator(shared_ptr<DeltaFileReader> dfr,
                                           RowIteratorOptions opts)
    : dfr_(std::move(dfr)),
      preparer_(std::move(opts)),
      prepared_(false),
      exhausted_(false),
      initted_(false),
      cache_blocks_(CFileReader::CACHE_BLOCK) {}

template<DeltaType Type>
Status DeltaFileIterator<Type>::Init(ScanSpec* spec) {
  DCHECK(!initted_) << "Already initted";

  if (spec) {
    cache_blocks_ = spec->cache_blocks() ? CFileReader::CACHE_BLOCK :
                                           CFileReader::DONT_CACHE_BLOCK;
  }

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

template<DeltaType Type>
Status DeltaFileIterator<Type>::SeekToOrdinal(rowid_t idx) {
  DCHECK(initted_) << "Must call Init()";

  // Finish the initialization of any lazily-initialized state.
  RETURN_NOT_OK(dfr_->Init(preparer_.opts().io_context));

  // Check again whether this delta file is relevant given the snapshots
  // that we are querying. We did this already before creating the
  // DeltaFileIterator, but due to lazy initialization, it's possible
  // that we weren't able to check at that time.
  if (!dfr_->IsRelevantForSnapshots(preparer_.opts().snap_to_exclude,
                                    preparer_.opts().snap_to_include)) {
    exhausted_ = true;
    delta_blocks_.clear();
    return Status::OK();
  }

  if (!index_iter_) {
    index_iter_.reset(IndexTreeIterator::Create(
        preparer_.opts().io_context,
        dfr_->cfile_reader().get(),
        dfr_->cfile_reader()->validx_root()));
  }

  tmp_buf_.clear();
  DeltaKey(idx, Timestamp(0)).EncodeTo(&tmp_buf_);
  Slice key_slice(tmp_buf_);

  Status s = index_iter_->SeekAtOrBefore(key_slice);
  if (PREDICT_FALSE(s.IsNotFound())) {
    // Seeking to a value before the first value in the file
    // will return NotFound, due to the way the index seek
    // works. We need to special-case this and have the
    // iterator seek all the way down its leftmost branches
    // to get the correct result.
    s = index_iter_->SeekToFirst();
  }
  RETURN_NOT_OK(s);

  preparer_.Seek(idx);
  prepared_ = false;
  delta_blocks_.clear();
  exhausted_ = false;
  return Status::OK();
}

struct PreparedDeltaBlock {
  // The pointer from which this block was read. This is only used for
  // logging, etc.
  cfile::BlockPointer block_ptr_;

  // Handle to the block, so it doesn't get freed from underneath us.
  scoped_refptr<cfile::BlockHandle> block_;

  // The block decoder, to avoid having to re-parse the block header
  // on every ApplyUpdates() call
  std::unique_ptr<cfile::BinaryPlainBlockDecoder> decoder_;

  // The first row index for which there is an update in this delta block.
  rowid_t first_updated_idx_;

  // The last row index for which there is an update in this delta block.
  rowid_t last_updated_idx_;

  // Within this block, the index of the update which is the first one that
  // needs to be consulted. This allows deltas to be skipped at the beginning
  // of the block when the row block starts towards the end of the delta block.
  // For example:
  // <-- delta block ---->
  //                   <--- prepared row block --->
  // Here, we can skip a bunch of deltas at the beginning of the delta block
  // which we know don't apply to the prepared row block.
  rowid_t prepared_block_start_idx_;

  // Return a string description of this prepared block, for logging.
  std::string ToString() const {
    return StringPrintf("%d-%d (%s)", first_updated_idx_, last_updated_idx_,
                        block_ptr_.ToString().c_str());
  }
};


template<DeltaType Type>
Status DeltaFileIterator<Type>::ReadCurrentBlockOntoQueue() {
  DCHECK(initted_) << "Must call Init()";
  DCHECK(index_iter_) << "Must call SeekToOrdinal()";

  unique_ptr<PreparedDeltaBlock> pdb(new PreparedDeltaBlock());
  BlockPointer dblk_ptr = index_iter_->GetCurrentBlockPointer();
  shared_ptr<CFileReader> reader = dfr_->cfile_reader();
  RETURN_NOT_OK(reader->ReadBlock(preparer_.opts().io_context,
                                  dblk_ptr, cache_blocks_, &pdb->block_));

  // The data has been successfully read. Finish creating the decoder.
  pdb->prepared_block_start_idx_ = 0;
  pdb->block_ptr_ = dblk_ptr;

  // Decode the block.
  pdb->decoder_.reset(new BinaryPlainBlockDecoder(pdb->block_));
  RETURN_NOT_OK_PREPEND(pdb->decoder_->ParseHeader(),
                        Substitute("unable to decode data block header in delta block $0 ($1)",
                                   dfr_->cfile_reader()->block_id().ToString(),
                                   dblk_ptr.ToString()));

  RETURN_NOT_OK(GetFirstRowIndexInCurrentBlock(&pdb->first_updated_idx_));
  RETURN_NOT_OK(GetLastRowIndexInDecodedBlock(*pdb->decoder_, &pdb->last_updated_idx_));

  #ifndef NDEBUG
  VLOG(2) << "Read delta block which updates " <<
    pdb->first_updated_idx_ << " through " <<
    pdb->last_updated_idx_;
  #endif

  delta_blocks_.emplace_back(std::move(pdb));
  return Status::OK();
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::GetFirstRowIndexInCurrentBlock(rowid_t *idx) {
  DCHECK(index_iter_) << "Must call SeekToOrdinal()";

  Slice index_entry = index_iter_->GetCurrentKey();
  DeltaKey k;
  RETURN_NOT_OK(k.DecodeFrom(&index_entry));
  *idx = k.row_idx();
  return Status::OK();
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::GetLastRowIndexInDecodedBlock(const BinaryPlainBlockDecoder &dec,
                                                              rowid_t *idx) {
  DCHECK_GT(dec.Count(), 0);
  Slice s(dec.string_at_index(dec.Count() - 1));
  DeltaKey k;
  RETURN_NOT_OK(k.DecodeFrom(&s));
  *idx = k.row_idx();
  return Status::OK();
}


template<DeltaType Type>
Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, int prepare_flags) {
  DCHECK(initted_) << "Must call Init()";
  DCHECK(exhausted_ || index_iter_) << "Must call SeekToOrdinal()";

  CHECK_GT(nrows, 0);

  rowid_t start_row = preparer_.cur_prepared_idx();
  rowid_t stop_row = start_row + nrows - 1;

  // Remove blocks from our list which are no longer relevant to the range
  // being prepared.
  while (!delta_blocks_.empty() &&
         delta_blocks_.front()->last_updated_idx_ < start_row) {
    delta_blocks_.pop_front();
  }

  while (!exhausted_) {
    rowid_t next_block_rowidx = 0;
    RETURN_NOT_OK(GetFirstRowIndexInCurrentBlock(&next_block_rowidx));
    VLOG(2) << "Current delta block starting at row " << next_block_rowidx;

    if (next_block_rowidx > stop_row) {
      break;
    }

    RETURN_NOT_OK(ReadCurrentBlockOntoQueue());

    Status s = index_iter_->Next();
    if (s.IsNotFound()) {
      exhausted_ = true;
      break;
    }
    RETURN_NOT_OK(s);
  }

  if (!delta_blocks_.empty()) {
    PreparedDeltaBlock& block = *delta_blocks_.front();
    int i = 0;
    for (i = block.prepared_block_start_idx_;
         i < block.decoder_->Count();
         i++) {
      Slice s(block.decoder_->string_at_index(i));
      DeltaKey key;
      RETURN_NOT_OK(key.DecodeFrom(&s));
      if (key.row_idx() >= start_row) break;
    }
    block.prepared_block_start_idx_ = i;
  }

  #ifndef NDEBUG
  VLOG(2) << "Done preparing deltas for " << start_row << "-" << stop_row
          << ": row block spans " << delta_blocks_.size() << " delta blocks";
  #endif
  prepared_ = true;

  preparer_.Start(nrows, prepare_flags);
  RETURN_NOT_OK(AddDeltas(start_row, stop_row));
  preparer_.Finish(nrows);
  return Status::OK();
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::AddDeltas(rowid_t start_row, rowid_t stop_row) {
  DCHECK(prepared_) << "must Prepare";

  for (auto& block : delta_blocks_) {
    const BinaryPlainBlockDecoder& bpd = *block->decoder_;
    DVLOG(2) << "Adding deltas from delta block " << block->first_updated_idx_ << "-"
             << block->last_updated_idx_ << " for row block starting at " << start_row;

    if (PREDICT_FALSE(start_row > block->last_updated_idx_)) {
      // The block to be updated completely falls after this delta block:
      //  <-- delta block -->      <-- delta block -->
      //                      <-- block to update     -->
      // This can happen because we don't know the block's last entry until after
      // we queued it in PrepareBatch(). We could potentially remove it at that
      // point during the prepare step, but for now just skip it here.
      continue;
    }

    bool finished_row = false;
    for (int i = block->prepared_block_start_idx_; i < bpd.Count(); i++) {
      Slice slice = bpd.string_at_index(i);

      // Decode and check the ID of the row we're going to update.
      DeltaKey key;
      RETURN_NOT_OK(key.DecodeFrom(&slice));

      // If this delta is for the same row as before, skip it if the previous
      // AddDelta() call told us that we're done with this row.
      if (preparer_.last_added_idx() &&
          preparer_.last_added_idx() == key.row_idx() &&
          finished_row) {
        continue;
      }
      finished_row = false;

      // Check that the delta is within the block we're currently processing.
      if (key.row_idx() > stop_row) {
        // Delta is for a row which comes after the block we're processing.
        return Status::OK();
      }
      if (key.row_idx() < start_row) {
        // Delta is for a row which comes before the block we're processing.
        continue;
      }

      // Note: if AddDelta sets 'finished_row' to true, we could skip the
      // remaining deltas for this row by seeking the block decoder. This trades
      // off the cost of a seek against the cost of decoding some irrelevant delta keys.
      //
      // Given that updates are expected to be uncommon and that most scans are
      // _not_ historical, the current implementation eschews seeking in favor of
      // skipping irrelevant deltas one by one.
      RETURN_NOT_OK(preparer_.AddDelta(key, slice, &finished_row));
      if (VLOG_IS_ON(3)) {
        RowChangeList rcl(slice);
        DVLOG(3) << "Visited " << DeltaType_Name(DeltaTypeSelector<Type>::kTag)
                 << " delta for key: " << key.ToString() << " Mut: "
                 << rcl.ToString(*preparer_.opts().projection)
                 << " Continue?: " << (!finished_row ? "TRUE" : "FALSE");
      }
    }
  }

  return Status::OK();
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
                                             const SelectionVector& filter) {
  return preparer_.ApplyUpdates(col_to_apply, dst, filter);
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::ApplyDeletes(SelectionVector* sel_vec) {
  return preparer_.ApplyDeletes(sel_vec);
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::SelectDeltas(SelectedDeltas* deltas) {
  return preparer_.SelectDeltas(deltas);
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
  return preparer_.CollectMutations(dst, arena);
}

template<DeltaType Type>
bool DeltaFileIterator<Type>::HasNext() const {
  return !exhausted_ || !delta_blocks_.empty();
}

template<DeltaType Type>
bool DeltaFileIterator<Type>::MayHaveDeltas() const {
  return preparer_.MayHaveDeltas();
}

template<DeltaType Type>
string DeltaFileIterator<Type>::ToString() const {
  return "DeltaFileIterator(" + dfr_->ToString() + ")";
}

template<DeltaType Type>
Status DeltaFileIterator<Type>::FilterColumnIdsAndCollectDeltas(
    const vector<ColumnId>& col_ids,
    vector<DeltaKeyAndUpdate>* out,
    Arena* arena) {
  return preparer_.FilterColumnIdsAndCollectDeltas(col_ids, out, arena);
}

template<DeltaType Type>
void DeltaFileIterator<Type>::FatalUnexpectedDelta(const DeltaKey &key, const Slice &deltas,
                                             const string &msg) {
  LOG(FATAL) << "Saw unexpected delta type in deltafile " << dfr_->ToString() << ": "
             << " rcl=" << RowChangeList(deltas).ToString(*preparer_.opts().projection)
             << " key=" << key.ToString() << " (" << msg << ")";
}

} // namespace tablet
} // namespace kudu