src/kudu/tablet/delta_compaction.cc - kudu - Git at Google

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

 #include <cstdint>
 #include <map>
 #include <ostream>
 #include <string>
 #include <utility>
 #include <vector>

 #include <glog/logging.h>

 #include "kudu/common/generic_iterators.h"
 #include "kudu/common/iterator.h"
 #include "kudu/common/row.h"
 #include "kudu/common/row_changelist.h"
 #include "kudu/common/rowblock.h"
 #include "kudu/common/rowblock_memory.h"
 #include "kudu/common/rowid.h"
 #include "kudu/common/scan_spec.h"
 #include "kudu/fs/block_manager.h"
 #include "kudu/fs/fs_manager.h"
 #include "kudu/gutil/casts.h"
 #include "kudu/gutil/map-util.h"
 #include "kudu/gutil/strings/join.h"
 #include "kudu/gutil/strings/substitute.h"
 #include "kudu/tablet/cfile_set.h"
 #include "kudu/tablet/compaction.h"
 #include "kudu/tablet/delta_key.h"
 #include "kudu/tablet/delta_stats.h"
 #include "kudu/tablet/delta_tracker.h"
 #include "kudu/tablet/deltafile.h"
 #include "kudu/tablet/multi_column_writer.h"
 #include "kudu/tablet/mutation.h"
 #include "kudu/tablet/mvcc.h"
 #include "kudu/tablet/rowset_metadata.h"
 #include "kudu/util/memory/arena.h"
 #include "kudu/util/trace.h"

 using kudu::fs::BlockCreationTransaction;
 using kudu::fs::BlockManager;
 using kudu::fs::CreateBlockOptions;
 using kudu::fs::IOContext;
 using kudu::fs::WritableBlock;
 using std::shared_ptr;
 using std::string;
 using std::unique_ptr;
 using std::vector;
 using strings::Substitute;

 namespace kudu {
 namespace tablet {
 namespace {

 const size_t kRowsPerBlock = 100; // Number of rows per block of columns

 } // anonymous namespace

 // TODO: can you major-delta-compact a new column after an alter table in order
 // to materialize it? should write a test for this.
 MajorDeltaCompaction::MajorDeltaCompaction(
     FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
     unique_ptr<DeltaIterator> delta_iter,
     vector<shared_ptr<DeltaStore> > included_stores,
     vector<ColumnId> col_ids,
     HistoryGcOpts history_gc_opts,
     string tablet_id)
     : fs_manager_(fs_manager),
       base_schema_(base_schema),
       column_ids_(std::move(col_ids)),
       history_gc_opts_(std::move(history_gc_opts)),
       base_data_(base_data),
       included_stores_(std::move(included_stores)),
       delta_iter_(std::move(delta_iter)),
       tablet_id_(std::move(tablet_id)),
       redo_delta_mutations_written_(0),
       undo_delta_mutations_written_(0),
       state_(kInitialized) {
   CHECK(!column_ids_.empty());
 }

 MajorDeltaCompaction::~MajorDeltaCompaction() {
 }

 string MajorDeltaCompaction::ColumnNamesToString() const {
   vector<string> col_names;
   col_names.reserve(column_ids_.size());
   for (ColumnId col_id : column_ids_) {
     int col_idx = base_schema_.find_column_by_id(col_id);
     if (col_idx != Schema::kColumnNotFound) {
       col_names.push_back(base_schema_.column_by_id(col_id).ToString());
     } else {
       col_names.push_back(Substitute("[deleted column id $0]", col_id));
     }
   }
   return JoinStrings(col_names, ", ");
 }

 Status MajorDeltaCompaction::FlushRowSetAndDeltas(const IOContext* io_context) {
   CHECK_EQ(state_, kInitialized);

   unique_ptr<ColumnwiseIterator> old_base_data_cwise(base_data_->NewIterator(&partial_schema_,
                                                                              io_context));
   unique_ptr<RowwiseIterator> old_base_data_rwise(
       NewMaterializingIterator(std::move(old_base_data_cwise)));

   ScanSpec spec;
   spec.set_cache_blocks(false);
   RETURN_NOT_OK_PREPEND(
       old_base_data_rwise->Init(&spec),
       "Unable to open iterator for specified columns (" + partial_schema_.ToString() + ")");

   RETURN_NOT_OK(delta_iter_->Init(&spec));
   RETURN_NOT_OK(delta_iter_->SeekToOrdinal(0));

   RowBlockMemory mem(32 * 1024);
   RowBlock block(&partial_schema_, kRowsPerBlock, &mem);

   DVLOG(1) << "Applying deltas and rewriting columns (" << partial_schema_.ToString() << ")";
   unique_ptr<DeltaStats> redo_stats(new DeltaStats);
   unique_ptr<DeltaStats> undo_stats(new DeltaStats);
   size_t nrows = 0;
   // We know that we're reading everything from disk so we're including all transactions.
   MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllOps();
   while (old_base_data_rwise->HasNext()) {

     // 1) Get the next batch of base data for the columns we're compacting.
     mem.Reset();
     RETURN_NOT_OK(old_base_data_rwise->NextBlock(&block));
     size_t n = block.nrows();

     // 2) Fetch all the REDO mutations.
     vector<Mutation *> redo_mutation_block(kRowsPerBlock, static_cast<Mutation *>(nullptr));
     RETURN_NOT_OK(delta_iter_->PrepareBatch(n, DeltaIterator::PREPARE_FOR_COLLECT));
     RETURN_NOT_OK(delta_iter_->CollectMutations(&redo_mutation_block, block.arena()));

     // 3) Write new UNDO mutations for the current block. The REDO mutations
     //    are written out in step 6.
     vector<CompactionInputRow> input_rows;
     input_rows.resize(block.nrows());
     for (int i = 0; i < block.nrows(); i++) {
       CompactionInputRow* input_row = &input_rows[i];
       input_row->row.Reset(&block, i);
       input_row->redo_head = redo_mutation_block[i];
       Mutation::ReverseMutationList(&input_row->redo_head);
       input_row->undo_head = nullptr;

       RowBlockRow dst_row = block.row(i);
       RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));

       Mutation* new_undos_head = nullptr;
       // We're ignoring the result from new_redos_head because we'll find them
       // later at step 5.
       Mutation* new_redos_head = nullptr;

       // Since this is a delta compaction the input and output row id's are the same.
       rowid_t row_id = nrows + input_row->row.row_index();

       DVLOG(3) << "MDC Input Row - RowId: " << row_id << " "
                << CompactionInputRowToString(*input_row);

       // NOTE: This is presently ignored.
       bool is_garbage_collected;

       RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(
           snap, *input_row, &new_undos_head, &new_redos_head, &mem.arena, &dst_row));

       RemoveAncientUndos(history_gc_opts_,
                          &new_undos_head,
                          new_redos_head,
                          &is_garbage_collected);

       DVLOG(3) << "MDC Output Row - RowId: " << row_id << " "
                << RowToString(dst_row, new_undos_head, new_redos_head);

       // We only create a new undo delta file if we need to.
       if (new_undos_head != nullptr && !new_undo_delta_writer_) {
         RETURN_NOT_OK(OpenUndoDeltaFileWriter());
       }
       for (const Mutation *mut = new_undos_head; mut != nullptr; mut = mut->next()) {
         DeltaKey undo_key(nrows + dst_row.row_index(), mut->timestamp());
         RETURN_NOT_OK(new_undo_delta_writer_->AppendDelta<UNDO>(undo_key, mut->changelist()));
         undo_stats->UpdateStats(mut->timestamp(), mut->changelist());
         undo_delta_mutations_written_++;
       }
     }

     // 4) Write the new base data.
     RETURN_NOT_OK(base_data_writer_->AppendBlock(block));

     // 5) Remove the columns that we've done our major REDO delta compaction on
     //    from this delta flush, except keep all the delete and reinsert
     //    mutations.
     mem.Reset();
     vector<DeltaKeyAndUpdate> out;
     RETURN_NOT_OK(delta_iter_->FilterColumnIdsAndCollectDeltas(column_ids_, &out, &mem.arena));

     // We only create a new redo delta file if we need to.
     if (!out.empty() && !new_redo_delta_writer_) {
       RETURN_NOT_OK(OpenRedoDeltaFileWriter());
     }

     // 6) Write the remaining REDO deltas that we haven't compacted away back
     //    into a REDO delta file.
     for (const DeltaKeyAndUpdate& key_and_update : out) {
       RowChangeList update(key_and_update.cell);
       DVLOG(4) << "Keeping delta as REDO: "
                << key_and_update.Stringify(DeltaType::REDO, base_schema_);
       RETURN_NOT_OK_PREPEND(new_redo_delta_writer_->AppendDelta<REDO>(key_and_update.key, update),
                             "Failed to append a delta");
       WARN_NOT_OK(redo_stats->UpdateStats(key_and_update.key.timestamp(), update),
                   "Failed to update stats");
     }
     redo_delta_mutations_written_ += out.size();
     nrows += n;
   }

   BlockManager* bm = fs_manager_->block_manager();
   unique_ptr<BlockCreationTransaction> transaction = bm->NewCreationTransaction();
   RETURN_NOT_OK(base_data_writer_->FinishAndReleaseBlocks(transaction.get()));

   if (redo_delta_mutations_written_ > 0) {
     new_redo_delta_writer_->WriteDeltaStats(std::move(redo_stats));
     RETURN_NOT_OK(new_redo_delta_writer_->FinishAndReleaseBlock(transaction.get()));
   }

   if (undo_delta_mutations_written_ > 0) {
     new_undo_delta_writer_->WriteDeltaStats(std::move(undo_stats));
     RETURN_NOT_OK(new_undo_delta_writer_->FinishAndReleaseBlock(transaction.get()));
   }
   transaction->CommitCreatedBlocks();

   DVLOG(1) << "Applied all outstanding deltas for columns "
            << partial_schema_.ToString()
            << ", and flushed the resulting rowsets and a total of "
            << redo_delta_mutations_written_
            << " REDO delta mutations and "
            << undo_delta_mutations_written_
            << " UNDO delta mutations to disk.";

   state_ = kFinished;
   return Status::OK();
 }

 Status MajorDeltaCompaction::OpenBaseDataWriter() {
   CHECK(!base_data_writer_);

   unique_ptr<MultiColumnWriter> w(new MultiColumnWriter(fs_manager_,
                                                         &partial_schema_,
                                                         tablet_id_));
   RETURN_NOT_OK(w->Open());
   base_data_writer_ = std::move(w);
   return Status::OK();
 }

 Status MajorDeltaCompaction::OpenRedoDeltaFileWriter() {
   unique_ptr<WritableBlock> block;
   CreateBlockOptions opts({ tablet_id_ });
   RETURN_NOT_OK_PREPEND(fs_manager_->CreateNewBlock(opts, &block),
                         "Unable to create REDO delta output block");
   new_redo_delta_block_ = block->id();
   new_redo_delta_writer_.reset(new DeltaFileWriter(std::move(block)));
   return new_redo_delta_writer_->Start();
 }

 Status MajorDeltaCompaction::OpenUndoDeltaFileWriter() {
   unique_ptr<WritableBlock> block;
   CreateBlockOptions opts({ tablet_id_ });
   RETURN_NOT_OK_PREPEND(fs_manager_->CreateNewBlock(opts, &block),
                         "Unable to create UNDO delta output block");
   new_undo_delta_block_ = block->id();
   new_undo_delta_writer_.reset(new DeltaFileWriter(std::move(block)));
   return new_undo_delta_writer_->Start();
 }

 namespace {
 struct DeltaStoreStats {
   int64_t delete_count;
   int64_t reinsert_count;
   int64_t update_count;

  public:
   string ToString() const {
     return Substitute("delete_count=$0, reinsert_count=$1, update_count=$2",
                       delete_count, reinsert_count, update_count);
   }
 };

 DeltaStoreStats ComputeDeltaStoreStats(const SharedDeltaStoreVector& stores) {
   int64_t delete_count = 0;
   int64_t reinsert_count = 0;
   int64_t update_count = 0;
   for (const auto& store : stores) {
     if (!store->Initted()) {
       continue;
     }
     const auto& stats = store->delta_stats();
     delete_count += stats.delete_count();
     reinsert_count += stats.reinsert_count();
     update_count += stats.UpdateCount();
   }
   return DeltaStoreStats{delete_count, reinsert_count, update_count};
 }
 } // anonymous namespace

 Status MajorDeltaCompaction::Compact(const IOContext* io_context) {
   CHECK_EQ(state_, kInitialized);

   VLOG(1) << "Starting major delta compaction for columns " << ColumnNamesToString();
   RETURN_NOT_OK(base_schema_.CreateProjectionByIdsIgnoreMissing(column_ids_, &partial_schema_));

   if (VLOG_IS_ON(1)) {
     for (const auto& ds : included_stores_) {
       VLOG(1) << "Preparing to major compact delta file: " << ds->ToString();
     }
   }

   // We defer calling OpenRedoDeltaFileWriter() since we might not need to flush.
   RETURN_NOT_OK(OpenBaseDataWriter());
   RETURN_NOT_OK(FlushRowSetAndDeltas(io_context));

   TRACE_COUNTER_INCREMENT("delta_blocks_compacted", included_stores_.size());
   const auto stats = ComputeDeltaStoreStats(included_stores_);
   TRACE_COUNTER_INCREMENT("delete_count", stats.delete_count);
   TRACE_COUNTER_INCREMENT("reinsert_count", stats.reinsert_count);
   TRACE_COUNTER_INCREMENT("update_count", stats.update_count);
   VLOG(1) << Substitute("Finished major delta compaction of columns $0. "
                         "Compacted $1 delta files. Overall stats: $2",
                         ColumnNamesToString(),
                         included_stores_.size(),
                         stats.ToString());
   return Status::OK();
 }

 void MajorDeltaCompaction::CreateMetadataUpdate(
     RowSetMetadataUpdate* update) {
   CHECK(update);
   CHECK_EQ(state_, kFinished);

   vector<BlockId> compacted_delta_blocks;
   for (const shared_ptr<DeltaStore>& store : included_stores_) {
     DeltaFileReader* dfr = down_cast<DeltaFileReader*>(store.get());
     compacted_delta_blocks.push_back(dfr->block_id());
   }

   vector<BlockId> new_delta_blocks;
   if (redo_delta_mutations_written_ > 0) {
     new_delta_blocks.push_back(new_redo_delta_block_);
   }

   update->ReplaceRedoDeltaBlocks(compacted_delta_blocks,
                                  new_delta_blocks);

   if (undo_delta_mutations_written_ > 0) {
     update->SetNewUndoBlock(new_undo_delta_block_);
   }

   // Replace old column blocks with new ones
   std::map<ColumnId, BlockId> new_column_blocks;
   base_data_writer_->GetFlushedBlocksByColumnId(&new_column_blocks);

   // NOTE: in the case that one of the columns being compacted is deleted,
   // we may have fewer elements in new_column_blocks compared to 'column_ids'.
   // For those deleted columns, we just remove the old column data.
   CHECK_LE(new_column_blocks.size(), column_ids_.size());

   for (ColumnId col_id : column_ids_) {
     BlockId new_block;
     if (FindCopy(new_column_blocks, col_id, &new_block)) {
       update->ReplaceColumnId(col_id, new_block);
     } else {
       // The column has been deleted.
       // If the base data has a block for this column, we need to remove it.
       // NOTE: It's possible that the base data has no data for this column in the
       // case that the column was added and removed in succession after the base
       // data was flushed.
       CHECK_EQ(base_schema_.find_column_by_id(col_id), Schema::kColumnNotFound)
         << "major compaction removing column " << col_id << " but still present in Schema!";
       if (base_data_->has_data_for_column_id(col_id)) {
         update->RemoveColumnId(col_id);
       }
     }
   }
 }

 // We're called under diskrowset's component_lock_ and delta_tracker's compact_flush_lock_
 // so both AtomicUpdateStores calls can be done separately and still be seen as one atomic
 // operation.
 Status MajorDeltaCompaction::UpdateDeltaTracker(DeltaTracker* tracker,
                                                 const IOContext* io_context) {
   CHECK_EQ(state_, kFinished);

   // 1. Get all the necessary I/O out of the way. It's OK to fail here
   //    because we haven't updated any in-memory state.
   //
   // TODO(awong): pull the OpenDeltaReaders() calls out of the critical path of
   // diskrowset's component_lock_. They touch disk and may block other
   // diskrowset operations.

   // Create blocks for the new redo deltas.
   vector<DeltaBlockIdAndStats> new_redo_blocks;
   if (redo_delta_mutations_written_ > 0) {
     new_redo_blocks.emplace_back(std::make_pair(new_redo_delta_block_,
         new_redo_delta_writer_->release_delta_stats()));
   }
   SharedDeltaStoreVector new_redo_stores;
   RETURN_NOT_OK(tracker->OpenDeltaReaders(std::move(new_redo_blocks), io_context,
                                           &new_redo_stores, REDO));

   // Create blocks for the new undo deltas.
   SharedDeltaStoreVector new_undo_stores;
   if (undo_delta_mutations_written_ > 0) {
     vector<DeltaBlockIdAndStats> new_undo_blocks;
     new_undo_blocks.emplace_back(std::make_pair(new_undo_delta_block_,
         new_undo_delta_writer_->release_delta_stats()));
     RETURN_NOT_OK(tracker->OpenDeltaReaders(std::move(new_undo_blocks), io_context,
                                             &new_undo_stores, UNDO));
   }

   // 2. Only now that we cannot fail do we update the in-memory state.

   // Even if we didn't create any new redo blocks, we still need to update the
   // tracker so it removes the included_stores_.
   tracker->AtomicUpdateStores(included_stores_, new_redo_stores, io_context, REDO);

   // We only call AtomicUpdateStores() for UNDOs if we wrote UNDOs. We're not
   // removing stores so we don't need to call it otherwise.
   if (!new_undo_stores.empty()) {
     tracker->AtomicUpdateStores({}, new_undo_stores, io_context, UNDO);
   }
   return Status::OK();
 }

 } // namespace tablet
 } // namespace kudu
	// 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/delta_compaction.h"

	#include <cstdint>
	#include <map>
	#include <ostream>
	#include <string>
	#include <utility>
	#include <vector>

	#include <glog/logging.h>

	#include "kudu/common/generic_iterators.h"
	#include "kudu/common/iterator.h"
	#include "kudu/common/row.h"
	#include "kudu/common/row_changelist.h"
	#include "kudu/common/rowblock.h"
	#include "kudu/common/rowblock_memory.h"
	#include "kudu/common/rowid.h"
	#include "kudu/common/scan_spec.h"
	#include "kudu/fs/block_manager.h"
	#include "kudu/fs/fs_manager.h"
	#include "kudu/gutil/casts.h"
	#include "kudu/gutil/map-util.h"
	#include "kudu/gutil/strings/join.h"
	#include "kudu/gutil/strings/substitute.h"
	#include "kudu/tablet/cfile_set.h"
	#include "kudu/tablet/compaction.h"
	#include "kudu/tablet/delta_key.h"
	#include "kudu/tablet/delta_stats.h"
	#include "kudu/tablet/delta_tracker.h"
	#include "kudu/tablet/deltafile.h"
	#include "kudu/tablet/multi_column_writer.h"
	#include "kudu/tablet/mutation.h"
	#include "kudu/tablet/mvcc.h"
	#include "kudu/tablet/rowset_metadata.h"
	#include "kudu/util/memory/arena.h"
	#include "kudu/util/trace.h"

	using kudu::fs::BlockCreationTransaction;
	using kudu::fs::BlockManager;
	using kudu::fs::CreateBlockOptions;
	using kudu::fs::IOContext;
	using kudu::fs::WritableBlock;
	using std::shared_ptr;
	using std::string;
	using std::unique_ptr;
	using std::vector;
	using strings::Substitute;

	namespace kudu {
	namespace tablet {
	namespace {

	const size_t kRowsPerBlock = 100; // Number of rows per block of columns

	} // anonymous namespace

	// TODO: can you major-delta-compact a new column after an alter table in order
	// to materialize it? should write a test for this.
	MajorDeltaCompaction::MajorDeltaCompaction(
	FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
	unique_ptr<DeltaIterator> delta_iter,
	vector<shared_ptr<DeltaStore> > included_stores,
	vector<ColumnId> col_ids,
	HistoryGcOpts history_gc_opts,
	string tablet_id)
	: fs_manager_(fs_manager),
	base_schema_(base_schema),
	column_ids_(std::move(col_ids)),
	history_gc_opts_(std::move(history_gc_opts)),
	base_data_(base_data),
	included_stores_(std::move(included_stores)),
	delta_iter_(std::move(delta_iter)),
	tablet_id_(std::move(tablet_id)),
	redo_delta_mutations_written_(0),
	undo_delta_mutations_written_(0),
	state_(kInitialized) {
	CHECK(!column_ids_.empty());
	}

	MajorDeltaCompaction::~MajorDeltaCompaction() {
	}

	string MajorDeltaCompaction::ColumnNamesToString() const {
	vector<string> col_names;
	col_names.reserve(column_ids_.size());
	for (ColumnId col_id : column_ids_) {
	int col_idx = base_schema_.find_column_by_id(col_id);
	if (col_idx != Schema::kColumnNotFound) {
	col_names.push_back(base_schema_.column_by_id(col_id).ToString());
	} else {
	col_names.push_back(Substitute("[deleted column id $0]", col_id));
	}
	}
	return JoinStrings(col_names, ", ");
	}

	Status MajorDeltaCompaction::FlushRowSetAndDeltas(const IOContext* io_context) {
	CHECK_EQ(state_, kInitialized);

	unique_ptr<ColumnwiseIterator> old_base_data_cwise(base_data_->NewIterator(&partial_schema_,
	io_context));
	unique_ptr<RowwiseIterator> old_base_data_rwise(
	NewMaterializingIterator(std::move(old_base_data_cwise)));

	ScanSpec spec;
	spec.set_cache_blocks(false);
	RETURN_NOT_OK_PREPEND(
	old_base_data_rwise->Init(&spec),
	"Unable to open iterator for specified columns (" + partial_schema_.ToString() + ")");

	RETURN_NOT_OK(delta_iter_->Init(&spec));
	RETURN_NOT_OK(delta_iter_->SeekToOrdinal(0));

	RowBlockMemory mem(32 * 1024);
	RowBlock block(&partial_schema_, kRowsPerBlock, &mem);

	DVLOG(1) << "Applying deltas and rewriting columns (" << partial_schema_.ToString() << ")";
	unique_ptr<DeltaStats> redo_stats(new DeltaStats);
	unique_ptr<DeltaStats> undo_stats(new DeltaStats);
	size_t nrows = 0;
	// We know that we're reading everything from disk so we're including all transactions.
	MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllOps();
	while (old_base_data_rwise->HasNext()) {

	// 1) Get the next batch of base data for the columns we're compacting.
	mem.Reset();
	RETURN_NOT_OK(old_base_data_rwise->NextBlock(&block));
	size_t n = block.nrows();

	// 2) Fetch all the REDO mutations.
	vector<Mutation > redo_mutation_block(kRowsPerBlock, static_cast<Mutation >(nullptr));
	RETURN_NOT_OK(delta_iter_->PrepareBatch(n, DeltaIterator::PREPARE_FOR_COLLECT));
	RETURN_NOT_OK(delta_iter_->CollectMutations(&redo_mutation_block, block.arena()));

	// 3) Write new UNDO mutations for the current block. The REDO mutations
	// are written out in step 6.
	vector<CompactionInputRow> input_rows;
	input_rows.resize(block.nrows());
	for (int i = 0; i < block.nrows(); i++) {
	CompactionInputRow* input_row = &input_rows[i];
	input_row->row.Reset(&block, i);
	input_row->redo_head = redo_mutation_block[i];
	Mutation::ReverseMutationList(&input_row->redo_head);
	input_row->undo_head = nullptr;

	RowBlockRow dst_row = block.row(i);
	RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));

	Mutation* new_undos_head = nullptr;
	// We're ignoring the result from new_redos_head because we'll find them
	// later at step 5.
	Mutation* new_redos_head = nullptr;

	// Since this is a delta compaction the input and output row id's are the same.
	rowid_t row_id = nrows + input_row->row.row_index();

	DVLOG(3) << "MDC Input Row - RowId: " << row_id << " "
	<< CompactionInputRowToString(*input_row);

	// NOTE: This is presently ignored.
	bool is_garbage_collected;

	RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(
	snap, *input_row, &new_undos_head, &new_redos_head, &mem.arena, &dst_row));

	RemoveAncientUndos(history_gc_opts_,
	&new_undos_head,
	new_redos_head,
	&is_garbage_collected);

	DVLOG(3) << "MDC Output Row - RowId: " << row_id << " "
	<< RowToString(dst_row, new_undos_head, new_redos_head);

	// We only create a new undo delta file if we need to.
	if (new_undos_head != nullptr && !new_undo_delta_writer_) {
	RETURN_NOT_OK(OpenUndoDeltaFileWriter());
	}
	for (const Mutation *mut = new_undos_head; mut != nullptr; mut = mut->next()) {
	DeltaKey undo_key(nrows + dst_row.row_index(), mut->timestamp());
	RETURN_NOT_OK(new_undo_delta_writer_->AppendDelta<UNDO>(undo_key, mut->changelist()));
	undo_stats->UpdateStats(mut->timestamp(), mut->changelist());
	undo_delta_mutations_written_++;
	}
	}

	// 4) Write the new base data.
	RETURN_NOT_OK(base_data_writer_->AppendBlock(block));

	// 5) Remove the columns that we've done our major REDO delta compaction on
	// from this delta flush, except keep all the delete and reinsert
	// mutations.
	mem.Reset();
	vector<DeltaKeyAndUpdate> out;
	RETURN_NOT_OK(delta_iter_->FilterColumnIdsAndCollectDeltas(column_ids_, &out, &mem.arena));

	// We only create a new redo delta file if we need to.
	if (!out.empty() && !new_redo_delta_writer_) {
	RETURN_NOT_OK(OpenRedoDeltaFileWriter());
	}

	// 6) Write the remaining REDO deltas that we haven't compacted away back
	// into a REDO delta file.
	for (const DeltaKeyAndUpdate& key_and_update : out) {
	RowChangeList update(key_and_update.cell);
	DVLOG(4) << "Keeping delta as REDO: "
	<< key_and_update.Stringify(DeltaType::REDO, base_schema_);
	RETURN_NOT_OK_PREPEND(new_redo_delta_writer_->AppendDelta<REDO>(key_and_update.key, update),
	"Failed to append a delta");
	WARN_NOT_OK(redo_stats->UpdateStats(key_and_update.key.timestamp(), update),
	"Failed to update stats");
	}
	redo_delta_mutations_written_ += out.size();
	nrows += n;
	}

	BlockManager* bm = fs_manager_->block_manager();
	unique_ptr<BlockCreationTransaction> transaction = bm->NewCreationTransaction();
	RETURN_NOT_OK(base_data_writer_->FinishAndReleaseBlocks(transaction.get()));

	if (redo_delta_mutations_written_ > 0) {
	new_redo_delta_writer_->WriteDeltaStats(std::move(redo_stats));
	RETURN_NOT_OK(new_redo_delta_writer_->FinishAndReleaseBlock(transaction.get()));
	}

	if (undo_delta_mutations_written_ > 0) {
	new_undo_delta_writer_->WriteDeltaStats(std::move(undo_stats));
	RETURN_NOT_OK(new_undo_delta_writer_->FinishAndReleaseBlock(transaction.get()));
	}
	transaction->CommitCreatedBlocks();

	DVLOG(1) << "Applied all outstanding deltas for columns "
	<< partial_schema_.ToString()
	<< ", and flushed the resulting rowsets and a total of "
	<< redo_delta_mutations_written_
	<< " REDO delta mutations and "
	<< undo_delta_mutations_written_
	<< " UNDO delta mutations to disk.";

	state_ = kFinished;
	return Status::OK();
	}

	Status MajorDeltaCompaction::OpenBaseDataWriter() {
	CHECK(!base_data_writer_);

	unique_ptr<MultiColumnWriter> w(new MultiColumnWriter(fs_manager_,
	&partial_schema_,
	tablet_id_));
	RETURN_NOT_OK(w->Open());
	base_data_writer_ = std::move(w);
	return Status::OK();
	}

	Status MajorDeltaCompaction::OpenRedoDeltaFileWriter() {
	unique_ptr<WritableBlock> block;
	CreateBlockOptions opts({ tablet_id_ });
	RETURN_NOT_OK_PREPEND(fs_manager_->CreateNewBlock(opts, &block),
	"Unable to create REDO delta output block");
	new_redo_delta_block_ = block->id();
	new_redo_delta_writer_.reset(new DeltaFileWriter(std::move(block)));
	return new_redo_delta_writer_->Start();
	}

	Status MajorDeltaCompaction::OpenUndoDeltaFileWriter() {
	unique_ptr<WritableBlock> block;
	CreateBlockOptions opts({ tablet_id_ });
	RETURN_NOT_OK_PREPEND(fs_manager_->CreateNewBlock(opts, &block),
	"Unable to create UNDO delta output block");
	new_undo_delta_block_ = block->id();
	new_undo_delta_writer_.reset(new DeltaFileWriter(std::move(block)));
	return new_undo_delta_writer_->Start();
	}

	namespace {
	struct DeltaStoreStats {
	int64_t delete_count;
	int64_t reinsert_count;
	int64_t update_count;

	public:
	string ToString() const {
	return Substitute("delete_count=$0, reinsert_count=$1, update_count=$2",
	delete_count, reinsert_count, update_count);
	}
	};

	DeltaStoreStats ComputeDeltaStoreStats(const SharedDeltaStoreVector& stores) {
	int64_t delete_count = 0;
	int64_t reinsert_count = 0;
	int64_t update_count = 0;
	for (const auto& store : stores) {
	if (!store->Initted()) {
	continue;
	}
	const auto& stats = store->delta_stats();
	delete_count += stats.delete_count();
	reinsert_count += stats.reinsert_count();
	update_count += stats.UpdateCount();
	}
	return DeltaStoreStats{delete_count, reinsert_count, update_count};
	}
	} // anonymous namespace

	Status MajorDeltaCompaction::Compact(const IOContext* io_context) {
	CHECK_EQ(state_, kInitialized);

	VLOG(1) << "Starting major delta compaction for columns " << ColumnNamesToString();
	RETURN_NOT_OK(base_schema_.CreateProjectionByIdsIgnoreMissing(column_ids_, &partial_schema_));

	if (VLOG_IS_ON(1)) {
	for (const auto& ds : included_stores_) {
	VLOG(1) << "Preparing to major compact delta file: " << ds->ToString();
	}
	}

	// We defer calling OpenRedoDeltaFileWriter() since we might not need to flush.
	RETURN_NOT_OK(OpenBaseDataWriter());
	RETURN_NOT_OK(FlushRowSetAndDeltas(io_context));

	TRACE_COUNTER_INCREMENT("delta_blocks_compacted", included_stores_.size());
	const auto stats = ComputeDeltaStoreStats(included_stores_);
	TRACE_COUNTER_INCREMENT("delete_count", stats.delete_count);
	TRACE_COUNTER_INCREMENT("reinsert_count", stats.reinsert_count);
	TRACE_COUNTER_INCREMENT("update_count", stats.update_count);
	VLOG(1) << Substitute("Finished major delta compaction of columns $0. "
	"Compacted $1 delta files. Overall stats: $2",
	ColumnNamesToString(),
	included_stores_.size(),
	stats.ToString());
	return Status::OK();
	}

	void MajorDeltaCompaction::CreateMetadataUpdate(
	RowSetMetadataUpdate* update) {
	CHECK(update);
	CHECK_EQ(state_, kFinished);

	vector<BlockId> compacted_delta_blocks;
	for (const shared_ptr<DeltaStore>& store : included_stores_) {
	DeltaFileReader* dfr = down_cast<DeltaFileReader*>(store.get());
	compacted_delta_blocks.push_back(dfr->block_id());
	}

	vector<BlockId> new_delta_blocks;
	if (redo_delta_mutations_written_ > 0) {
	new_delta_blocks.push_back(new_redo_delta_block_);
	}

	update->ReplaceRedoDeltaBlocks(compacted_delta_blocks,
	new_delta_blocks);

	if (undo_delta_mutations_written_ > 0) {
	update->SetNewUndoBlock(new_undo_delta_block_);
	}

	// Replace old column blocks with new ones
	std::map<ColumnId, BlockId> new_column_blocks;
	base_data_writer_->GetFlushedBlocksByColumnId(&new_column_blocks);

	// NOTE: in the case that one of the columns being compacted is deleted,
	// we may have fewer elements in new_column_blocks compared to 'column_ids'.
	// For those deleted columns, we just remove the old column data.
	CHECK_LE(new_column_blocks.size(), column_ids_.size());

	for (ColumnId col_id : column_ids_) {
	BlockId new_block;
	if (FindCopy(new_column_blocks, col_id, &new_block)) {
	update->ReplaceColumnId(col_id, new_block);
	} else {
	// The column has been deleted.
	// If the base data has a block for this column, we need to remove it.
	// NOTE: It's possible that the base data has no data for this column in the
	// case that the column was added and removed in succession after the base
	// data was flushed.
	CHECK_EQ(base_schema_.find_column_by_id(col_id), Schema::kColumnNotFound)
	<< "major compaction removing column " << col_id << " but still present in Schema!";
	if (base_data_->has_data_for_column_id(col_id)) {
	update->RemoveColumnId(col_id);
	}
	}
	}
	}

	// We're called under diskrowset's component_lock_ and delta_tracker's compact_flush_lock_
	// so both AtomicUpdateStores calls can be done separately and still be seen as one atomic
	// operation.
	Status MajorDeltaCompaction::UpdateDeltaTracker(DeltaTracker* tracker,
	const IOContext* io_context) {
	CHECK_EQ(state_, kFinished);

	// 1. Get all the necessary I/O out of the way. It's OK to fail here
	// because we haven't updated any in-memory state.
	//
	// TODO(awong): pull the OpenDeltaReaders() calls out of the critical path of
	// diskrowset's component_lock_. They touch disk and may block other
	// diskrowset operations.

	// Create blocks for the new redo deltas.
	vector<DeltaBlockIdAndStats> new_redo_blocks;
	if (redo_delta_mutations_written_ > 0) {
	new_redo_blocks.emplace_back(std::make_pair(new_redo_delta_block_,
	new_redo_delta_writer_->release_delta_stats()));
	}
	SharedDeltaStoreVector new_redo_stores;
	RETURN_NOT_OK(tracker->OpenDeltaReaders(std::move(new_redo_blocks), io_context,
	&new_redo_stores, REDO));

	// Create blocks for the new undo deltas.
	SharedDeltaStoreVector new_undo_stores;
	if (undo_delta_mutations_written_ > 0) {
	vector<DeltaBlockIdAndStats> new_undo_blocks;
	new_undo_blocks.emplace_back(std::make_pair(new_undo_delta_block_,
	new_undo_delta_writer_->release_delta_stats()));
	RETURN_NOT_OK(tracker->OpenDeltaReaders(std::move(new_undo_blocks), io_context,
	&new_undo_stores, UNDO));
	}

	// 2. Only now that we cannot fail do we update the in-memory state.

	// Even if we didn't create any new redo blocks, we still need to update the
	// tracker so it removes the included_stores_.
	tracker->AtomicUpdateStores(included_stores_, new_redo_stores, io_context, REDO);

	// We only call AtomicUpdateStores() for UNDOs if we wrote UNDOs. We're not
	// removing stores so we don't need to call it otherwise.
	if (!new_undo_stores.empty()) {
	tracker->AtomicUpdateStores({}, new_undo_stores, io_context, UNDO);
	}
	return Status::OK();
	}

	} // namespace tablet
	} // namespace kudu