2014-11-01 00:31:25 +01:00
|
|
|
// Copyright (c) 2013, Facebook, Inc. All rights reserved.
|
|
|
|
// This source code is licensed under the BSD-style license found in the
|
|
|
|
// LICENSE file in the root directory of this source tree. An additional grant
|
|
|
|
// of patent rights can be found in the PATENTS file in the same directory.
|
|
|
|
//
|
|
|
|
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
|
|
|
// Use of this source code is governed by a BSD-style license that can be
|
|
|
|
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
|
|
|
|
|
|
|
#include "db/compaction_job.h"
|
|
|
|
|
|
|
|
#ifndef __STDC_FORMAT_MACROS
|
|
|
|
#define __STDC_FORMAT_MACROS
|
|
|
|
#endif
|
|
|
|
|
|
|
|
#include <inttypes.h>
|
|
|
|
#include <algorithm>
|
|
|
|
#include <vector>
|
|
|
|
#include <memory>
|
|
|
|
#include <list>
|
|
|
|
|
|
|
|
#include "db/builder.h"
|
|
|
|
#include "db/db_iter.h"
|
|
|
|
#include "db/dbformat.h"
|
2015-05-28 22:37:47 +02:00
|
|
|
#include "db/event_helpers.h"
|
2014-11-01 00:31:25 +01:00
|
|
|
#include "db/filename.h"
|
|
|
|
#include "db/log_reader.h"
|
|
|
|
#include "db/log_writer.h"
|
|
|
|
#include "db/memtable.h"
|
|
|
|
#include "db/merge_helper.h"
|
|
|
|
#include "db/memtable_list.h"
|
|
|
|
#include "db/merge_context.h"
|
|
|
|
#include "db/version_set.h"
|
|
|
|
#include "port/port.h"
|
|
|
|
#include "port/likely.h"
|
|
|
|
#include "rocksdb/db.h"
|
|
|
|
#include "rocksdb/env.h"
|
|
|
|
#include "rocksdb/statistics.h"
|
|
|
|
#include "rocksdb/status.h"
|
|
|
|
#include "rocksdb/table.h"
|
|
|
|
#include "table/block.h"
|
|
|
|
#include "table/block_based_table_factory.h"
|
|
|
|
#include "table/merger.h"
|
|
|
|
#include "table/table_builder.h"
|
|
|
|
#include "table/two_level_iterator.h"
|
|
|
|
#include "util/coding.h"
|
|
|
|
#include "util/logging.h"
|
|
|
|
#include "util/log_buffer.h"
|
|
|
|
#include "util/mutexlock.h"
|
|
|
|
#include "util/perf_context_imp.h"
|
|
|
|
#include "util/iostats_context_imp.h"
|
|
|
|
#include "util/stop_watch.h"
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
#include "util/string_util.h"
|
2014-11-01 00:31:25 +01:00
|
|
|
#include "util/sync_point.h"
|
2015-01-13 09:04:08 +01:00
|
|
|
#include "util/thread_status_util.h"
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
namespace rocksdb {
|
|
|
|
|
|
|
|
struct CompactionJob::CompactionState {
|
|
|
|
Compaction* const compaction;
|
|
|
|
|
|
|
|
// Files produced by compaction
|
|
|
|
struct Output {
|
|
|
|
uint64_t number;
|
|
|
|
uint32_t path_id;
|
|
|
|
uint64_t file_size;
|
|
|
|
InternalKey smallest, largest;
|
|
|
|
SequenceNumber smallest_seqno, largest_seqno;
|
2015-06-04 21:03:40 +02:00
|
|
|
bool need_compaction;
|
2014-11-01 00:31:25 +01:00
|
|
|
};
|
|
|
|
std::vector<Output> outputs;
|
|
|
|
|
|
|
|
// State kept for output being generated
|
|
|
|
std::unique_ptr<WritableFile> outfile;
|
|
|
|
std::unique_ptr<TableBuilder> builder;
|
|
|
|
|
|
|
|
uint64_t total_bytes;
|
|
|
|
|
|
|
|
Output* current_output() { return &outputs[outputs.size() - 1]; }
|
|
|
|
|
2014-11-01 02:36:07 +01:00
|
|
|
explicit CompactionState(Compaction* c)
|
|
|
|
: compaction(c),
|
|
|
|
total_bytes(0),
|
|
|
|
num_input_records(0),
|
|
|
|
num_output_records(0) {}
|
2014-11-01 00:31:25 +01:00
|
|
|
|
2014-11-01 02:36:07 +01:00
|
|
|
uint64_t num_input_records;
|
|
|
|
uint64_t num_output_records;
|
2014-11-01 00:31:25 +01:00
|
|
|
};
|
|
|
|
|
|
|
|
CompactionJob::CompactionJob(
|
2015-02-12 18:54:48 +01:00
|
|
|
int job_id, Compaction* compaction, const DBOptions& db_options,
|
2015-05-06 04:01:12 +02:00
|
|
|
const EnvOptions& env_options, VersionSet* versions,
|
|
|
|
std::atomic<bool>* shutting_down, LogBuffer* log_buffer,
|
|
|
|
Directory* db_directory, Directory* output_directory, Statistics* stats,
|
|
|
|
std::vector<SequenceNumber> existing_snapshots,
|
2014-11-08 00:44:12 +01:00
|
|
|
std::shared_ptr<Cache> table_cache,
|
2015-05-06 04:01:12 +02:00
|
|
|
std::function<uint64_t()> yield_callback, EventLogger* event_logger,
|
2015-06-03 02:07:16 +02:00
|
|
|
bool paranoid_file_checks, const std::string& dbname,
|
|
|
|
CompactionJobStats* compaction_job_stats)
|
2015-02-12 18:54:48 +01:00
|
|
|
: job_id_(job_id),
|
|
|
|
compact_(new CompactionState(compaction)),
|
2015-06-03 02:07:16 +02:00
|
|
|
compaction_job_stats_(compaction_job_stats),
|
2014-11-01 00:31:25 +01:00
|
|
|
compaction_stats_(1),
|
2015-06-02 23:12:23 +02:00
|
|
|
dbname_(dbname),
|
2014-11-01 00:31:25 +01:00
|
|
|
db_options_(db_options),
|
|
|
|
env_options_(env_options),
|
|
|
|
env_(db_options.env),
|
|
|
|
versions_(versions),
|
|
|
|
shutting_down_(shutting_down),
|
|
|
|
log_buffer_(log_buffer),
|
|
|
|
db_directory_(db_directory),
|
2015-01-26 22:59:38 +01:00
|
|
|
output_directory_(output_directory),
|
2014-11-01 00:31:25 +01:00
|
|
|
stats_(stats),
|
2015-05-06 04:01:12 +02:00
|
|
|
existing_snapshots_(std::move(existing_snapshots)),
|
2014-11-01 00:31:25 +01:00
|
|
|
table_cache_(std::move(table_cache)),
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
yield_callback_(std::move(yield_callback)),
|
2015-05-06 04:01:12 +02:00
|
|
|
event_logger_(event_logger),
|
|
|
|
paranoid_file_checks_(paranoid_file_checks) {
|
2015-07-15 08:12:34 +02:00
|
|
|
assert(log_buffer_ != nullptr);
|
2015-05-06 04:01:12 +02:00
|
|
|
ThreadStatusUtil::SetColumnFamily(compact_->compaction->column_family_data());
|
2015-03-13 18:45:40 +01:00
|
|
|
ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);
|
2015-05-07 07:50:35 +02:00
|
|
|
ReportStartedCompaction(compaction);
|
2015-03-13 18:45:40 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
CompactionJob::~CompactionJob() {
|
|
|
|
assert(compact_ == nullptr);
|
|
|
|
ThreadStatusUtil::ResetThreadStatus();
|
|
|
|
}
|
2014-11-01 00:31:25 +01:00
|
|
|
|
2015-05-07 07:50:35 +02:00
|
|
|
void CompactionJob::ReportStartedCompaction(
|
|
|
|
Compaction* compaction) {
|
|
|
|
ThreadStatusUtil::SetColumnFamily(
|
|
|
|
compact_->compaction->column_family_data());
|
|
|
|
|
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_JOB_ID,
|
|
|
|
job_id_);
|
|
|
|
|
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_INPUT_OUTPUT_LEVEL,
|
|
|
|
(static_cast<uint64_t>(compact_->compaction->start_level()) << 32) +
|
|
|
|
compact_->compaction->output_level());
|
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
// In the current design, a CompactionJob is always created
|
|
|
|
// for non-trivial compaction.
|
|
|
|
assert(compaction->IsTrivialMove() == false ||
|
2015-07-09 00:21:10 +02:00
|
|
|
compaction->is_manual_compaction() == true);
|
2015-06-03 02:07:16 +02:00
|
|
|
|
2015-05-07 07:50:35 +02:00
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_PROP_FLAGS,
|
2015-07-13 21:11:05 +02:00
|
|
|
compaction->is_manual_compaction() +
|
2015-07-09 00:21:10 +02:00
|
|
|
(compaction->deletion_compaction() << 1));
|
2015-05-07 07:50:35 +02:00
|
|
|
|
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_TOTAL_INPUT_BYTES,
|
|
|
|
compaction->CalculateTotalInputSize());
|
|
|
|
|
|
|
|
IOSTATS_RESET(bytes_written);
|
|
|
|
IOSTATS_RESET(bytes_read);
|
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_BYTES_WRITTEN, 0);
|
|
|
|
ThreadStatusUtil::SetThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_BYTES_READ, 0);
|
|
|
|
|
|
|
|
// Set the thread operation after operation properties
|
|
|
|
// to ensure GetThreadList() can always show them all together.
|
|
|
|
ThreadStatusUtil::SetThreadOperation(
|
|
|
|
ThreadStatus::OP_COMPACTION);
|
2015-06-03 02:07:16 +02:00
|
|
|
|
|
|
|
if (compaction_job_stats_) {
|
|
|
|
compaction_job_stats_->is_manual_compaction =
|
2015-07-13 21:11:05 +02:00
|
|
|
compaction->is_manual_compaction();
|
2015-06-03 02:07:16 +02:00
|
|
|
}
|
2015-05-07 07:50:35 +02:00
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
void CompactionJob::Prepare() {
|
2015-03-13 18:45:40 +01:00
|
|
|
AutoThreadOperationStageUpdater stage_updater(
|
|
|
|
ThreadStatus::STAGE_COMPACTION_PREPARE);
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
// Generate file_levels_ for compaction berfore making Iterator
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
ColumnFamilyData* cfd __attribute__((unused)) =
|
|
|
|
compact_->compaction->column_family_data();
|
2014-11-14 20:35:48 +01:00
|
|
|
assert(cfd != nullptr);
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
assert(cfd->current()->storage_info()->NumLevelFiles(
|
|
|
|
compact_->compaction->level()) > 0);
|
|
|
|
assert(compact_->builder == nullptr);
|
|
|
|
assert(!compact_->outfile);
|
|
|
|
|
|
|
|
visible_at_tip_ = 0;
|
|
|
|
latest_snapshot_ = 0;
|
2015-05-06 04:01:12 +02:00
|
|
|
if (existing_snapshots_.size() == 0) {
|
2014-11-01 00:31:25 +01:00
|
|
|
// optimize for fast path if there are no snapshots
|
|
|
|
visible_at_tip_ = versions_->LastSequence();
|
|
|
|
earliest_snapshot_ = visible_at_tip_;
|
|
|
|
} else {
|
2015-05-06 04:01:12 +02:00
|
|
|
latest_snapshot_ = existing_snapshots_.back();
|
2014-11-01 00:31:25 +01:00
|
|
|
// Add the current seqno as the 'latest' virtual
|
|
|
|
// snapshot to the end of this list.
|
2015-05-06 04:01:12 +02:00
|
|
|
existing_snapshots_.push_back(versions_->LastSequence());
|
|
|
|
earliest_snapshot_ = existing_snapshots_[0];
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Is this compaction producing files at the bottommost level?
|
2015-07-09 00:21:10 +02:00
|
|
|
bottommost_level_ = compact_->compaction->bottommost_level();
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
Status CompactionJob::Run() {
|
2015-03-13 18:45:40 +01:00
|
|
|
AutoThreadOperationStageUpdater stage_updater(
|
|
|
|
ThreadStatus::STAGE_COMPACTION_RUN);
|
2015-03-13 19:59:00 +01:00
|
|
|
TEST_SYNC_POINT("CompactionJob::Run():Start");
|
2014-11-01 00:31:25 +01:00
|
|
|
log_buffer_->FlushBufferToLog();
|
|
|
|
ColumnFamilyData* cfd = compact_->compaction->column_family_data();
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
auto* compaction = compact_->compaction;
|
2015-07-14 09:09:20 +02:00
|
|
|
LogCompaction(cfd, compaction);
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
const uint64_t start_micros = env_->NowMicros();
|
|
|
|
std::unique_ptr<Iterator> input(
|
|
|
|
versions_->MakeInputIterator(compact_->compaction));
|
|
|
|
input->SeekToFirst();
|
|
|
|
|
|
|
|
int64_t imm_micros = 0; // Micros spent doing imm_ compactions
|
2015-07-17 18:59:11 +02:00
|
|
|
auto status = ProcessKeyValueCompaction(&imm_micros, input.get());
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
if (status.ok() &&
|
|
|
|
(shutting_down_->load(std::memory_order_acquire) || cfd->IsDropped())) {
|
|
|
|
status = Status::ShutdownInProgress(
|
|
|
|
"Database shutdown or Column family drop during compaction");
|
|
|
|
}
|
|
|
|
if (status.ok() && compact_->builder != nullptr) {
|
2015-07-15 18:55:45 +02:00
|
|
|
status = FinishCompactionOutputFile(input->status());
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
if (status.ok()) {
|
|
|
|
status = input->status();
|
|
|
|
}
|
|
|
|
input.reset();
|
|
|
|
|
2015-01-26 22:59:38 +01:00
|
|
|
if (output_directory_ && !db_options_.disableDataSync) {
|
|
|
|
output_directory_->Fsync();
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
compaction_stats_.micros = env_->NowMicros() - start_micros - imm_micros;
|
|
|
|
MeasureTime(stats_, COMPACTION_TIME, compaction_stats_.micros);
|
2015-07-14 09:09:20 +02:00
|
|
|
UpdateCompactionStats();
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
RecordCompactionIOStats();
|
|
|
|
|
|
|
|
LogFlush(db_options_.info_log);
|
2015-03-13 19:59:00 +01:00
|
|
|
TEST_SYNC_POINT("CompactionJob::Run():End");
|
2014-11-01 00:31:25 +01:00
|
|
|
return status;
|
|
|
|
}
|
|
|
|
|
2015-05-06 04:01:12 +02:00
|
|
|
void CompactionJob::Install(Status* status,
|
|
|
|
const MutableCFOptions& mutable_cf_options,
|
|
|
|
InstrumentedMutex* db_mutex) {
|
2015-03-13 18:45:40 +01:00
|
|
|
AutoThreadOperationStageUpdater stage_updater(
|
|
|
|
ThreadStatus::STAGE_COMPACTION_INSTALL);
|
2014-11-08 00:44:12 +01:00
|
|
|
db_mutex->AssertHeld();
|
2014-11-01 00:31:25 +01:00
|
|
|
ColumnFamilyData* cfd = compact_->compaction->column_family_data();
|
|
|
|
cfd->internal_stats()->AddCompactionStats(
|
|
|
|
compact_->compaction->output_level(), compaction_stats_);
|
|
|
|
|
2014-11-10 20:57:58 +01:00
|
|
|
if (status->ok()) {
|
2015-05-06 04:01:12 +02:00
|
|
|
*status = InstallCompactionResults(db_mutex, mutable_cf_options);
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
VersionStorageInfo::LevelSummaryStorage tmp;
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
auto vstorage = cfd->current()->storage_info();
|
2014-11-01 00:31:25 +01:00
|
|
|
const auto& stats = compaction_stats_;
|
2015-06-18 08:40:34 +02:00
|
|
|
LogToBuffer(
|
|
|
|
log_buffer_,
|
|
|
|
"[%s] compacted to: %s, MB/sec: %.1f rd, %.1f wr, level %d, "
|
|
|
|
"files in(%d, %d) out(%d) "
|
|
|
|
"MB in(%.1f, %.1f) out(%.1f), read-write-amplify(%.1f) "
|
|
|
|
"write-amplify(%.1f) %s, records in: %d, records dropped: %d\n",
|
|
|
|
cfd->GetName().c_str(), vstorage->LevelSummary(&tmp),
|
|
|
|
(stats.bytes_read_non_output_levels + stats.bytes_read_output_level) /
|
|
|
|
static_cast<double>(stats.micros),
|
|
|
|
stats.bytes_written / static_cast<double>(stats.micros),
|
|
|
|
compact_->compaction->output_level(),
|
|
|
|
stats.num_input_files_in_non_output_levels,
|
|
|
|
stats.num_input_files_in_output_level,
|
|
|
|
stats.num_output_files,
|
|
|
|
stats.bytes_read_non_output_levels / 1048576.0,
|
|
|
|
stats.bytes_read_output_level / 1048576.0,
|
|
|
|
stats.bytes_written / 1048576.0,
|
|
|
|
(stats.bytes_written + stats.bytes_read_output_level +
|
|
|
|
stats.bytes_read_non_output_levels) /
|
|
|
|
static_cast<double>(stats.bytes_read_non_output_levels),
|
|
|
|
stats.bytes_written /
|
|
|
|
static_cast<double>(stats.bytes_read_non_output_levels),
|
|
|
|
status->ToString().c_str(), stats.num_input_records,
|
|
|
|
stats.num_dropped_records);
|
2014-11-01 00:31:25 +01:00
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
UpdateCompactionJobStats(stats);
|
|
|
|
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
auto stream = event_logger_->LogToBuffer(log_buffer_);
|
|
|
|
stream << "job" << job_id_ << "event"
|
|
|
|
<< "compaction_finished"
|
|
|
|
<< "output_level" << compact_->compaction->output_level()
|
|
|
|
<< "num_output_files" << compact_->outputs.size()
|
2015-06-18 08:40:34 +02:00
|
|
|
<< "total_output_size" << compact_->total_bytes
|
|
|
|
<< "num_input_records" << compact_->num_input_records
|
|
|
|
<< "num_output_records" << compact_->num_output_records;
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
2015-04-28 00:20:02 +02:00
|
|
|
stream << "lsm_state";
|
|
|
|
stream.StartArray();
|
|
|
|
for (int level = 0; level < vstorage->num_levels(); ++level) {
|
|
|
|
stream << vstorage->NumLevelFiles(level);
|
|
|
|
}
|
|
|
|
stream.EndArray();
|
|
|
|
|
2014-11-10 20:57:58 +01:00
|
|
|
CleanupCompaction(*status);
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
Status CompactionJob::ProcessKeyValueCompaction(int64_t* imm_micros,
|
2015-07-17 18:59:11 +02:00
|
|
|
Iterator* input) {
|
2015-03-13 18:45:40 +01:00
|
|
|
AutoThreadOperationStageUpdater stage_updater(
|
|
|
|
ThreadStatus::STAGE_COMPACTION_PROCESS_KV);
|
2014-11-01 00:31:25 +01:00
|
|
|
Status status;
|
|
|
|
std::string compaction_filter_value;
|
|
|
|
ParsedInternalKey ikey;
|
|
|
|
IterKey current_user_key;
|
|
|
|
bool has_current_user_key = false;
|
|
|
|
IterKey delete_key;
|
|
|
|
SequenceNumber last_sequence_for_key __attribute__((unused)) =
|
|
|
|
kMaxSequenceNumber;
|
|
|
|
SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
|
|
|
|
ColumnFamilyData* cfd = compact_->compaction->column_family_data();
|
|
|
|
MergeHelper merge(cfd->user_comparator(), cfd->ioptions()->merge_operator,
|
|
|
|
db_options_.info_log.get(),
|
|
|
|
cfd->ioptions()->min_partial_merge_operands,
|
|
|
|
false /* internal key corruption is expected */);
|
|
|
|
auto compaction_filter = cfd->ioptions()->compaction_filter;
|
|
|
|
std::unique_ptr<CompactionFilter> compaction_filter_from_factory = nullptr;
|
|
|
|
if (!compaction_filter) {
|
|
|
|
compaction_filter_from_factory =
|
Allowing L0 -> L1 trivial move on sorted data
Summary:
This diff updates the logic of how we do trivial move, now trivial move can run on any number of files in input level as long as they are not overlapping
The conditions for trivial move have been updated
Introduced conditions:
- Trivial move cannot happen if we have a compaction filter (except if the compaction is not manual)
- Input level files cannot be overlapping
Removed conditions:
- Trivial move only run when the compaction is not manual
- Input level should can contain only 1 file
More context on what tests failed because of Trivial move
```
DBTest.CompactionsGenerateMultipleFiles
This test is expecting compaction on a file in L0 to generate multiple files in L1, this test will fail with trivial move because we end up with one file in L1
```
```
DBTest.NoSpaceCompactRange
This test expect compaction to fail when we force environment to report running out of space, of course this is not valid in trivial move situation
because trivial move does not need any extra space, and did not check for that
```
```
DBTest.DropWrites
Similar to DBTest.NoSpaceCompactRange
```
```
DBTest.DeleteObsoleteFilesPendingOutputs
This test expect that a file in L2 is deleted after it's moved to L3, this is not valid with trivial move because although the file was moved it is now used by L3
```
```
CuckooTableDBTest.CompactionIntoMultipleFiles
Same as DBTest.CompactionsGenerateMultipleFiles
```
This diff is based on a work by @sdong https://reviews.facebook.net/D34149
Test Plan: make -j64 check
Reviewers: rven, sdong, igor
Reviewed By: igor
Subscribers: yhchiang, ott, march, dhruba, sdong
Differential Revision: https://reviews.facebook.net/D34797
2015-06-05 01:51:25 +02:00
|
|
|
compact_->compaction->CreateCompactionFilter();
|
2014-11-01 00:31:25 +01:00
|
|
|
compaction_filter = compaction_filter_from_factory.get();
|
|
|
|
}
|
|
|
|
|
2015-03-14 16:21:53 +01:00
|
|
|
TEST_SYNC_POINT("CompactionJob::Run():Inprogress");
|
2015-03-11 18:31:02 +01:00
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
int64_t key_drop_user = 0;
|
|
|
|
int64_t key_drop_newer_entry = 0;
|
|
|
|
int64_t key_drop_obsolete = 0;
|
|
|
|
int64_t loop_cnt = 0;
|
2015-03-03 19:59:36 +01:00
|
|
|
|
|
|
|
StopWatchNano timer(env_, stats_ != nullptr);
|
|
|
|
uint64_t total_filter_time = 0;
|
2014-11-01 00:31:25 +01:00
|
|
|
while (input->Valid() && !shutting_down_->load(std::memory_order_acquire) &&
|
|
|
|
!cfd->IsDropped() && status.ok()) {
|
2014-11-01 02:36:07 +01:00
|
|
|
compact_->num_input_records++;
|
2014-11-01 00:31:25 +01:00
|
|
|
if (++loop_cnt > 1000) {
|
2015-06-03 02:07:16 +02:00
|
|
|
RecordDroppedKeys(
|
|
|
|
&key_drop_user, &key_drop_newer_entry, &key_drop_obsolete);
|
2014-11-01 00:31:25 +01:00
|
|
|
RecordCompactionIOStats();
|
|
|
|
loop_cnt = 0;
|
|
|
|
}
|
|
|
|
// FLUSH preempts compaction
|
|
|
|
// TODO(icanadi) this currently only checks if flush is necessary on
|
2015-07-14 09:09:20 +02:00
|
|
|
// compacting column family. we should also check if flush is necessary
|
|
|
|
// on other column families, too
|
2014-11-01 00:31:25 +01:00
|
|
|
(*imm_micros) += yield_callback_();
|
|
|
|
|
2015-07-17 18:59:11 +02:00
|
|
|
Slice key = input->key();
|
|
|
|
Slice value = input->value();
|
2014-11-01 00:31:25 +01:00
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
if (compaction_job_stats_ != nullptr) {
|
|
|
|
compaction_job_stats_->total_input_raw_key_bytes +=
|
|
|
|
input->key().size();
|
|
|
|
compaction_job_stats_->total_input_raw_value_bytes +=
|
|
|
|
input->value().size();
|
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
if (compact_->compaction->ShouldStopBefore(key) &&
|
|
|
|
compact_->builder != nullptr) {
|
2015-07-15 18:55:45 +02:00
|
|
|
status = FinishCompactionOutputFile(input->status());
|
2014-11-01 00:31:25 +01:00
|
|
|
if (!status.ok()) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Handle key/value, add to state, etc.
|
|
|
|
bool drop = false;
|
|
|
|
bool current_entry_is_merging = false;
|
|
|
|
if (!ParseInternalKey(key, &ikey)) {
|
|
|
|
// Do not hide error keys
|
|
|
|
// TODO: error key stays in db forever? Figure out the intention/rationale
|
|
|
|
// v10 error v8 : we cannot hide v8 even though it's pretty obvious.
|
|
|
|
current_user_key.Clear();
|
|
|
|
has_current_user_key = false;
|
|
|
|
last_sequence_for_key = kMaxSequenceNumber;
|
|
|
|
visible_in_snapshot = kMaxSequenceNumber;
|
|
|
|
} else {
|
2015-07-15 08:12:34 +02:00
|
|
|
if (compaction_job_stats_ != nullptr && ikey.type == kTypeDeletion) {
|
2015-07-14 00:51:38 +02:00
|
|
|
compaction_job_stats_->num_input_deletion_records++;
|
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
if (!has_current_user_key ||
|
|
|
|
cfd->user_comparator()->Compare(ikey.user_key,
|
|
|
|
current_user_key.GetKey()) != 0) {
|
|
|
|
// First occurrence of this user key
|
|
|
|
current_user_key.SetKey(ikey.user_key);
|
|
|
|
has_current_user_key = true;
|
|
|
|
last_sequence_for_key = kMaxSequenceNumber;
|
|
|
|
visible_in_snapshot = kMaxSequenceNumber;
|
|
|
|
// apply the compaction filter to the first occurrence of the user key
|
2015-07-17 18:59:11 +02:00
|
|
|
if (compaction_filter && ikey.type == kTypeValue &&
|
2014-11-01 00:31:25 +01:00
|
|
|
(visible_at_tip_ || ikey.sequence > latest_snapshot_)) {
|
|
|
|
// If the user has specified a compaction filter and the sequence
|
|
|
|
// number is greater than any external snapshot, then invoke the
|
|
|
|
// filter.
|
|
|
|
// If the return value of the compaction filter is true, replace
|
|
|
|
// the entry with a delete marker.
|
|
|
|
bool value_changed = false;
|
|
|
|
compaction_filter_value.clear();
|
2015-03-03 19:59:36 +01:00
|
|
|
if (stats_ != nullptr) {
|
|
|
|
timer.Start();
|
|
|
|
}
|
2014-11-01 00:31:25 +01:00
|
|
|
bool to_delete = compaction_filter->Filter(
|
|
|
|
compact_->compaction->level(), ikey.user_key, value,
|
|
|
|
&compaction_filter_value, &value_changed);
|
2015-03-03 19:59:36 +01:00
|
|
|
total_filter_time += timer.ElapsedNanos();
|
2014-11-01 00:31:25 +01:00
|
|
|
if (to_delete) {
|
|
|
|
// make a copy of the original key and convert it to a delete
|
|
|
|
delete_key.SetInternalKey(ExtractUserKey(key), ikey.sequence,
|
|
|
|
kTypeDeletion);
|
|
|
|
// anchor the key again
|
|
|
|
key = delete_key.GetKey();
|
|
|
|
// needed because ikey is backed by key
|
|
|
|
ParseInternalKey(key, &ikey);
|
|
|
|
// no value associated with delete
|
|
|
|
value.clear();
|
|
|
|
++key_drop_user;
|
|
|
|
} else if (value_changed) {
|
|
|
|
value = compaction_filter_value;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// If there are no snapshots, then this kv affect visibility at tip.
|
|
|
|
// Otherwise, search though all existing snapshots to find
|
|
|
|
// the earlist snapshot that is affected by this kv.
|
|
|
|
SequenceNumber prev_snapshot = 0; // 0 means no previous snapshot
|
|
|
|
SequenceNumber visible =
|
|
|
|
visible_at_tip_
|
|
|
|
? visible_at_tip_
|
2015-05-06 04:01:12 +02:00
|
|
|
: findEarliestVisibleSnapshot(ikey.sequence, existing_snapshots_,
|
|
|
|
&prev_snapshot);
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
if (visible_in_snapshot == visible) {
|
|
|
|
// If the earliest snapshot is which this key is visible in
|
|
|
|
// is the same as the visibily of a previous instance of the
|
|
|
|
// same key, then this kv is not visible in any snapshot.
|
|
|
|
// Hidden by an newer entry for same user key
|
|
|
|
// TODO: why not > ?
|
|
|
|
assert(last_sequence_for_key >= ikey.sequence);
|
|
|
|
drop = true; // (A)
|
|
|
|
++key_drop_newer_entry;
|
|
|
|
} else if (ikey.type == kTypeDeletion &&
|
|
|
|
ikey.sequence <= earliest_snapshot_ &&
|
|
|
|
compact_->compaction->KeyNotExistsBeyondOutputLevel(
|
|
|
|
ikey.user_key)) {
|
|
|
|
// For this user key:
|
|
|
|
// (1) there is no data in higher levels
|
|
|
|
// (2) data in lower levels will have larger sequence numbers
|
|
|
|
// (3) data in layers that are being compacted here and have
|
|
|
|
// smaller sequence numbers will be dropped in the next
|
|
|
|
// few iterations of this loop (by rule (A) above).
|
|
|
|
// Therefore this deletion marker is obsolete and can be dropped.
|
|
|
|
drop = true;
|
|
|
|
++key_drop_obsolete;
|
|
|
|
} else if (ikey.type == kTypeMerge) {
|
|
|
|
if (!merge.HasOperator()) {
|
|
|
|
LogToBuffer(log_buffer_, "Options::merge_operator is null.");
|
|
|
|
status = Status::InvalidArgument(
|
|
|
|
"merge_operator is not properly initialized.");
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
// We know the merge type entry is not hidden, otherwise we would
|
|
|
|
// have hit (A)
|
|
|
|
// We encapsulate the merge related state machine in a different
|
|
|
|
// object to minimize change to the existing flow. Turn out this
|
|
|
|
// logic could also be nicely re-used for memtable flush purge
|
|
|
|
// optimization in BuildTable.
|
|
|
|
merge.MergeUntil(input, prev_snapshot, bottommost_level_,
|
2015-07-17 18:59:11 +02:00
|
|
|
db_options_.statistics.get(), nullptr, env_);
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
current_entry_is_merging = true;
|
|
|
|
if (merge.IsSuccess()) {
|
|
|
|
// Successfully found Put/Delete/(end-of-key-range) while merging
|
|
|
|
// Get the merge result
|
|
|
|
key = merge.key();
|
|
|
|
ParseInternalKey(key, &ikey);
|
|
|
|
value = merge.value();
|
|
|
|
} else {
|
|
|
|
// Did not find a Put/Delete/(end-of-key-range) while merging
|
|
|
|
// We now have some stack of merge operands to write out.
|
|
|
|
// NOTE: key,value, and ikey are now referring to old entries.
|
|
|
|
// These will be correctly set below.
|
|
|
|
assert(!merge.keys().empty());
|
|
|
|
assert(merge.keys().size() == merge.values().size());
|
|
|
|
|
|
|
|
// Hack to make sure last_sequence_for_key is correct
|
|
|
|
ParseInternalKey(merge.keys().front(), &ikey);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
last_sequence_for_key = ikey.sequence;
|
|
|
|
visible_in_snapshot = visible;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (!drop) {
|
|
|
|
// We may write a single key (e.g.: for Put/Delete or successful merge).
|
|
|
|
// Or we may instead have to write a sequence/list of keys.
|
|
|
|
// We have to write a sequence iff we have an unsuccessful merge
|
2015-07-15 18:55:45 +02:00
|
|
|
if (current_entry_is_merging && !merge.IsSuccess()) {
|
|
|
|
const auto& keys = merge.keys();
|
|
|
|
const auto& values = merge.values();
|
|
|
|
std::deque<std::string>::const_reverse_iterator key_iter =
|
|
|
|
keys.rbegin(); // The back (*rbegin()) is the first key
|
|
|
|
std::deque<std::string>::const_reverse_iterator value_iter =
|
|
|
|
values.rbegin();
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
key = Slice(*key_iter);
|
|
|
|
value = Slice(*value_iter);
|
|
|
|
|
2015-07-15 18:55:45 +02:00
|
|
|
// We have a list of keys to write, traverse the list.
|
|
|
|
while (true) {
|
|
|
|
status = WriteKeyValue(key, value, ikey, input->status());
|
2014-11-01 00:31:25 +01:00
|
|
|
if (!status.ok()) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
++key_iter;
|
|
|
|
++value_iter;
|
|
|
|
|
|
|
|
// If at end of list
|
2015-07-15 18:55:45 +02:00
|
|
|
if (key_iter == keys.rend() || value_iter == values.rend()) {
|
2014-11-01 00:31:25 +01:00
|
|
|
// Sanity Check: if one ends, then both end
|
2015-07-15 18:55:45 +02:00
|
|
|
assert(key_iter == keys.rend() && value_iter == values.rend());
|
2014-11-01 00:31:25 +01:00
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
// Otherwise not at end of list. Update key, value, and ikey.
|
|
|
|
key = Slice(*key_iter);
|
|
|
|
value = Slice(*value_iter);
|
|
|
|
ParseInternalKey(key, &ikey);
|
|
|
|
}
|
2015-07-15 18:55:45 +02:00
|
|
|
} else {
|
|
|
|
// There is only one item to be written out
|
|
|
|
status = WriteKeyValue(key, value, ikey, input->status());
|
|
|
|
}
|
2014-11-01 00:31:25 +01:00
|
|
|
} // if (!drop)
|
|
|
|
|
|
|
|
// MergeUntil has moved input to the next entry
|
|
|
|
if (!current_entry_is_merging) {
|
|
|
|
input->Next();
|
|
|
|
}
|
|
|
|
}
|
2015-03-03 19:59:36 +01:00
|
|
|
RecordTick(stats_, FILTER_OPERATION_TOTAL_TIME, total_filter_time);
|
2015-06-03 02:07:16 +02:00
|
|
|
RecordDroppedKeys(&key_drop_user, &key_drop_newer_entry, &key_drop_obsolete);
|
2014-11-01 00:31:25 +01:00
|
|
|
RecordCompactionIOStats();
|
|
|
|
|
|
|
|
return status;
|
|
|
|
}
|
|
|
|
|
2015-07-15 18:55:45 +02:00
|
|
|
Status CompactionJob::WriteKeyValue(const Slice& key, const Slice& value,
|
|
|
|
const ParsedInternalKey& ikey, const Status& input_status) {
|
|
|
|
Slice newkey(key.data(), key.size());
|
|
|
|
std::string kstr;
|
|
|
|
|
|
|
|
// Zeroing out the sequence number leads to better compression.
|
|
|
|
// If this is the bottommost level (no files in lower levels)
|
|
|
|
// and the earliest snapshot is larger than this seqno
|
|
|
|
// then we can squash the seqno to zero.
|
|
|
|
if (bottommost_level_ && ikey.sequence < earliest_snapshot_ &&
|
|
|
|
ikey.type != kTypeMerge) {
|
|
|
|
assert(ikey.type != kTypeDeletion);
|
|
|
|
// make a copy because updating in place would cause problems
|
|
|
|
// with the priority queue that is managing the input key iterator
|
|
|
|
kstr.assign(key.data(), key.size());
|
|
|
|
UpdateInternalKey(&kstr, (uint64_t)0, ikey.type);
|
|
|
|
newkey = Slice(kstr);
|
|
|
|
}
|
|
|
|
|
|
|
|
// Open output file if necessary
|
|
|
|
if (compact_->builder == nullptr) {
|
|
|
|
Status status = OpenCompactionOutputFile();
|
|
|
|
if (!status.ok()) {
|
|
|
|
return status;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
SequenceNumber seqno = GetInternalKeySeqno(newkey);
|
|
|
|
if (compact_->builder->NumEntries() == 0) {
|
|
|
|
compact_->current_output()->smallest.DecodeFrom(newkey);
|
|
|
|
compact_->current_output()->smallest_seqno = seqno;
|
|
|
|
} else {
|
|
|
|
compact_->current_output()->smallest_seqno =
|
|
|
|
std::min(compact_->current_output()->smallest_seqno, seqno);
|
|
|
|
}
|
|
|
|
compact_->current_output()->largest.DecodeFrom(newkey);
|
|
|
|
compact_->builder->Add(newkey, value);
|
|
|
|
compact_->num_output_records++;
|
|
|
|
compact_->current_output()->largest_seqno =
|
|
|
|
std::max(compact_->current_output()->largest_seqno, seqno);
|
|
|
|
|
|
|
|
// Close output file if it is big enough
|
|
|
|
Status status;
|
|
|
|
if (compact_->builder->FileSize() >=
|
|
|
|
compact_->compaction->max_output_file_size()) {
|
|
|
|
status = FinishCompactionOutputFile(input_status);
|
|
|
|
}
|
|
|
|
|
|
|
|
return status;
|
|
|
|
}
|
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
void CompactionJob::RecordDroppedKeys(
|
|
|
|
int64_t* key_drop_user,
|
|
|
|
int64_t* key_drop_newer_entry,
|
|
|
|
int64_t* key_drop_obsolete) {
|
|
|
|
if (*key_drop_user > 0) {
|
|
|
|
RecordTick(stats_, COMPACTION_KEY_DROP_USER, *key_drop_user);
|
|
|
|
*key_drop_user = 0;
|
|
|
|
}
|
|
|
|
if (*key_drop_newer_entry > 0) {
|
|
|
|
RecordTick(stats_, COMPACTION_KEY_DROP_NEWER_ENTRY, *key_drop_newer_entry);
|
|
|
|
if (compaction_job_stats_) {
|
|
|
|
compaction_job_stats_->num_records_replaced += *key_drop_newer_entry;
|
|
|
|
}
|
|
|
|
*key_drop_newer_entry = 0;
|
|
|
|
}
|
|
|
|
if (*key_drop_obsolete > 0) {
|
|
|
|
RecordTick(stats_, COMPACTION_KEY_DROP_OBSOLETE, *key_drop_obsolete);
|
2015-07-14 00:51:38 +02:00
|
|
|
if (compaction_job_stats_) {
|
|
|
|
compaction_job_stats_->num_expired_deletion_records
|
|
|
|
+= *key_drop_obsolete;
|
|
|
|
}
|
2015-06-03 02:07:16 +02:00
|
|
|
*key_drop_obsolete = 0;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-07-15 18:55:45 +02:00
|
|
|
Status CompactionJob::FinishCompactionOutputFile(const Status& input_status) {
|
2015-03-13 18:45:40 +01:00
|
|
|
AutoThreadOperationStageUpdater stage_updater(
|
|
|
|
ThreadStatus::STAGE_COMPACTION_SYNC_FILE);
|
2014-11-01 00:31:25 +01:00
|
|
|
assert(compact_ != nullptr);
|
|
|
|
assert(compact_->outfile);
|
|
|
|
assert(compact_->builder != nullptr);
|
|
|
|
|
|
|
|
const uint64_t output_number = compact_->current_output()->number;
|
|
|
|
const uint32_t output_path_id = compact_->current_output()->path_id;
|
|
|
|
assert(output_number != 0);
|
|
|
|
|
Add more table properties to EventLogger
Summary:
Example output:
{"time_micros": 1431463794310521, "job": 353, "event": "table_file_creation", "file_number": 387, "file_size": 86937, "table_info": {"data_size": "81801", "index_size": "9751", "filter_size": "0", "raw_key_size": "23448", "raw_average_key_size": "24.000000", "raw_value_size": "990571", "raw_average_value_size": "1013.890481", "num_data_blocks": "245", "num_entries": "977", "filter_policy_name": "", "kDeletedKeys": "0"}}
Also fixed a bug where BuildTable() in recovery was passing Env::IOHigh argument into paranoid_checks_file parameter.
Test Plan: make check + check out the output in the log
Reviewers: sdong, rven, yhchiang
Reviewed By: yhchiang
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D38343
2015-05-13 00:53:55 +02:00
|
|
|
TableProperties table_properties;
|
2014-11-01 00:31:25 +01:00
|
|
|
// Check for iterator errors
|
2015-07-15 18:55:45 +02:00
|
|
|
Status s = input_status;
|
2014-11-01 00:31:25 +01:00
|
|
|
const uint64_t current_entries = compact_->builder->NumEntries();
|
2015-06-04 21:03:40 +02:00
|
|
|
compact_->current_output()->need_compaction =
|
|
|
|
compact_->builder->NeedCompact();
|
2014-11-01 00:31:25 +01:00
|
|
|
if (s.ok()) {
|
|
|
|
s = compact_->builder->Finish();
|
|
|
|
} else {
|
|
|
|
compact_->builder->Abandon();
|
|
|
|
}
|
|
|
|
const uint64_t current_bytes = compact_->builder->FileSize();
|
|
|
|
compact_->current_output()->file_size = current_bytes;
|
|
|
|
compact_->total_bytes += current_bytes;
|
|
|
|
|
|
|
|
// Finish and check for file errors
|
|
|
|
if (s.ok() && !db_options_.disableDataSync) {
|
|
|
|
if (db_options_.use_fsync) {
|
|
|
|
StopWatch sw(env_, stats_, COMPACTION_OUTFILE_SYNC_MICROS);
|
|
|
|
s = compact_->outfile->Fsync();
|
|
|
|
} else {
|
|
|
|
StopWatch sw(env_, stats_, COMPACTION_OUTFILE_SYNC_MICROS);
|
|
|
|
s = compact_->outfile->Sync();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if (s.ok()) {
|
|
|
|
s = compact_->outfile->Close();
|
|
|
|
}
|
|
|
|
compact_->outfile.reset();
|
|
|
|
|
|
|
|
if (s.ok() && current_entries > 0) {
|
|
|
|
// Verify that the table is usable
|
|
|
|
ColumnFamilyData* cfd = compact_->compaction->column_family_data();
|
|
|
|
FileDescriptor fd(output_number, output_path_id, current_bytes);
|
|
|
|
Iterator* iter = cfd->table_cache()->NewIterator(
|
|
|
|
ReadOptions(), env_options_, cfd->internal_comparator(), fd);
|
|
|
|
s = iter->status();
|
2015-04-18 00:26:50 +02:00
|
|
|
|
2015-05-06 04:01:12 +02:00
|
|
|
if (s.ok() && paranoid_file_checks_) {
|
2015-04-18 00:26:50 +02:00
|
|
|
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {}
|
|
|
|
s = iter->status();
|
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
delete iter;
|
|
|
|
if (s.ok()) {
|
2015-06-02 23:12:23 +02:00
|
|
|
TableFileCreationInfo info(compact_->builder->GetTableProperties());
|
|
|
|
info.db_name = dbname_;
|
|
|
|
info.cf_name = cfd->GetName();
|
|
|
|
info.file_path = TableFileName(cfd->ioptions()->db_paths,
|
|
|
|
fd.GetNumber(), fd.GetPathId());
|
|
|
|
info.file_size = fd.GetFileSize();
|
|
|
|
info.job_id = job_id_;
|
2015-02-12 18:54:48 +01:00
|
|
|
Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
|
|
|
|
"[%s] [JOB %d] Generated table #%" PRIu64 ": %" PRIu64
|
Print info message about files need compaction for debuging purpose
Summary:
When there are files marked for compaction after compactions, print extra messages to help debugging. Example:
2015/06/08-23:12:55.212855 7ff5013ff700 [default] [JOB 121] Generated table #75: 54 keys, 4807 bytes (need compaction)
2015/06/08-23:12:55.556194 7ff5013ff700 (Original Log Time 2015/06/08-23:12:55.556160) [default] compacted to: base level 1 max bytes base
10240 files[0 1 9 32 12 0 0 0] max score 0.96 (2 files need compaction), MB/sec: 0.0 rd, 0.1 wr, level 2, files in(1, 3) out(5) MB in(0.0,
0.0) out(0.0), read-write-amplify(11.3) write-amplify(5.7) OK, records in: 40, records dropped: 0
Test Plan:
Run test and see LOG files.
valgrind test DBTest.TablePropertiesNeedCompactTest
Reviewers: rven, yhchiang, kradhakrishnan, IslamAbdelRahman, igor
Reviewed By: igor
Subscribers: yoshinorim, maykov, leveldb, dhruba
Differential Revision: https://reviews.facebook.net/D39771
2015-06-09 08:14:13 +02:00
|
|
|
" keys, %" PRIu64 " bytes%s",
|
2015-02-12 18:54:48 +01:00
|
|
|
cfd->GetName().c_str(), job_id_, output_number, current_entries,
|
Print info message about files need compaction for debuging purpose
Summary:
When there are files marked for compaction after compactions, print extra messages to help debugging. Example:
2015/06/08-23:12:55.212855 7ff5013ff700 [default] [JOB 121] Generated table #75: 54 keys, 4807 bytes (need compaction)
2015/06/08-23:12:55.556194 7ff5013ff700 (Original Log Time 2015/06/08-23:12:55.556160) [default] compacted to: base level 1 max bytes base
10240 files[0 1 9 32 12 0 0 0] max score 0.96 (2 files need compaction), MB/sec: 0.0 rd, 0.1 wr, level 2, files in(1, 3) out(5) MB in(0.0,
0.0) out(0.0), read-write-amplify(11.3) write-amplify(5.7) OK, records in: 40, records dropped: 0
Test Plan:
Run test and see LOG files.
valgrind test DBTest.TablePropertiesNeedCompactTest
Reviewers: rven, yhchiang, kradhakrishnan, IslamAbdelRahman, igor
Reviewed By: igor
Subscribers: yoshinorim, maykov, leveldb, dhruba
Differential Revision: https://reviews.facebook.net/D39771
2015-06-09 08:14:13 +02:00
|
|
|
current_bytes,
|
|
|
|
compact_->current_output()->need_compaction ? " (need compaction)"
|
|
|
|
: "");
|
2015-06-02 23:12:23 +02:00
|
|
|
EventHelpers::LogAndNotifyTableFileCreation(
|
|
|
|
event_logger_, cfd->ioptions()->listeners, fd, info);
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
}
|
2015-06-02 23:12:23 +02:00
|
|
|
compact_->builder.reset();
|
2014-11-01 00:31:25 +01:00
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
2015-05-06 04:01:12 +02:00
|
|
|
Status CompactionJob::InstallCompactionResults(
|
|
|
|
InstrumentedMutex* db_mutex, const MutableCFOptions& mutable_cf_options) {
|
2014-11-08 00:44:12 +01:00
|
|
|
db_mutex->AssertHeld();
|
2014-11-01 00:31:25 +01:00
|
|
|
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
auto* compaction = compact_->compaction;
|
2014-11-01 00:31:25 +01:00
|
|
|
// paranoia: verify that the files that we started with
|
|
|
|
// still exist in the current version and in the same original level.
|
|
|
|
// This ensures that a concurrent compaction did not erroneously
|
|
|
|
// pick the same files to compact_.
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
if (!versions_->VerifyCompactionFileConsistency(compaction)) {
|
|
|
|
Compaction::InputLevelSummaryBuffer inputs_summary;
|
|
|
|
|
2014-11-04 20:07:11 +01:00
|
|
|
Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
"[%s] [JOB %d] Compaction %s aborted",
|
|
|
|
compaction->column_family_data()->GetName().c_str(), job_id_,
|
|
|
|
compaction->InputLevelSummary(&inputs_summary));
|
2014-11-01 00:31:25 +01:00
|
|
|
return Status::Corruption("Compaction input files inconsistent");
|
|
|
|
}
|
|
|
|
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
{
|
|
|
|
Compaction::InputLevelSummaryBuffer inputs_summary;
|
|
|
|
Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
|
|
|
|
"[%s] [JOB %d] Compacted %s => %" PRIu64 " bytes",
|
|
|
|
compaction->column_family_data()->GetName().c_str(), job_id_,
|
|
|
|
compaction->InputLevelSummary(&inputs_summary), compact_->total_bytes);
|
|
|
|
}
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
// Add compaction outputs
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
compaction->AddInputDeletions(compact_->compaction->edit());
|
2014-11-01 00:31:25 +01:00
|
|
|
for (size_t i = 0; i < compact_->outputs.size(); i++) {
|
|
|
|
const CompactionState::Output& out = compact_->outputs[i];
|
2015-06-04 21:03:40 +02:00
|
|
|
compaction->edit()->AddFile(compaction->output_level(), out.number,
|
|
|
|
out.path_id, out.file_size, out.smallest,
|
|
|
|
out.largest, out.smallest_seqno,
|
|
|
|
out.largest_seqno, out.need_compaction);
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
}
|
|
|
|
return versions_->LogAndApply(compaction->column_family_data(),
|
2015-05-06 04:01:12 +02:00
|
|
|
mutable_cf_options, compaction->edit(),
|
options.level_compaction_dynamic_level_bytes to allow RocksDB to pick size bases of levels dynamically.
Summary:
When having fixed max_bytes_for_level_base, the ratio of size of largest level and the second one can range from 0 to the multiplier. This makes LSM tree frequently irregular and unpredictable. It can also cause poor space amplification in some cases.
In this improvement (proposed by Igor Kabiljo), we introduce a parameter option.level_compaction_use_dynamic_max_bytes. When turning it on, RocksDB is free to pick a level base in the range of (options.max_bytes_for_level_base/options.max_bytes_for_level_multiplier, options.max_bytes_for_level_base] so that real level ratios are close to options.max_bytes_for_level_multiplier.
Test Plan: New unit tests and pass tests suites including valgrind.
Reviewers: MarkCallaghan, rven, yhchiang, igor, ikabiljo
Reviewed By: ikabiljo
Subscribers: yoshinorim, ikabiljo, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D31437
2015-02-05 20:44:17 +01:00
|
|
|
db_mutex, db_directory_);
|
2014-11-01 00:31:25 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Given a sequence number, return the sequence number of the
|
|
|
|
// earliest snapshot that this sequence number is visible in.
|
|
|
|
// The snapshots themselves are arranged in ascending order of
|
|
|
|
// sequence numbers.
|
|
|
|
// Employ a sequential search because the total number of
|
|
|
|
// snapshots are typically small.
|
|
|
|
inline SequenceNumber CompactionJob::findEarliestVisibleSnapshot(
|
|
|
|
SequenceNumber in, const std::vector<SequenceNumber>& snapshots,
|
|
|
|
SequenceNumber* prev_snapshot) {
|
2014-11-14 20:35:48 +01:00
|
|
|
assert(snapshots.size());
|
2014-11-01 00:31:25 +01:00
|
|
|
SequenceNumber prev __attribute__((unused)) = 0;
|
|
|
|
for (const auto cur : snapshots) {
|
|
|
|
assert(prev <= cur);
|
|
|
|
if (cur >= in) {
|
|
|
|
*prev_snapshot = prev;
|
|
|
|
return cur;
|
|
|
|
}
|
|
|
|
prev = cur; // assignment
|
|
|
|
assert(prev);
|
|
|
|
}
|
2014-11-04 20:07:11 +01:00
|
|
|
Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
|
|
|
|
"CompactionJob is not able to find snapshot"
|
|
|
|
" with SeqId later than %" PRIu64
|
|
|
|
": current MaxSeqId is %" PRIu64 "",
|
|
|
|
in, snapshots[snapshots.size() - 1]);
|
2014-11-01 00:31:25 +01:00
|
|
|
assert(0);
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
void CompactionJob::RecordCompactionIOStats() {
|
|
|
|
RecordTick(stats_, COMPACT_READ_BYTES, IOSTATS(bytes_read));
|
2015-05-07 07:50:35 +02:00
|
|
|
ThreadStatusUtil::IncreaseThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_BYTES_READ, IOSTATS(bytes_read));
|
2014-11-01 00:31:25 +01:00
|
|
|
IOSTATS_RESET(bytes_read);
|
|
|
|
RecordTick(stats_, COMPACT_WRITE_BYTES, IOSTATS(bytes_written));
|
2015-05-07 07:50:35 +02:00
|
|
|
ThreadStatusUtil::IncreaseThreadOperationProperty(
|
|
|
|
ThreadStatus::COMPACTION_BYTES_WRITTEN, IOSTATS(bytes_written));
|
2014-11-01 00:31:25 +01:00
|
|
|
IOSTATS_RESET(bytes_written);
|
|
|
|
}
|
|
|
|
|
|
|
|
Status CompactionJob::OpenCompactionOutputFile() {
|
|
|
|
assert(compact_ != nullptr);
|
|
|
|
assert(compact_->builder == nullptr);
|
2014-11-08 00:44:12 +01:00
|
|
|
// no need to lock because VersionSet::next_file_number_ is atomic
|
|
|
|
uint64_t file_number = versions_->NewFileNumber();
|
2014-11-01 00:31:25 +01:00
|
|
|
// Make the output file
|
|
|
|
std::string fname = TableFileName(db_options_.db_paths, file_number,
|
2015-07-09 00:21:10 +02:00
|
|
|
compact_->compaction->output_path_id());
|
2014-11-01 00:31:25 +01:00
|
|
|
Status s = env_->NewWritableFile(fname, &compact_->outfile, env_options_);
|
|
|
|
|
|
|
|
if (!s.ok()) {
|
|
|
|
Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
|
2015-02-12 18:54:48 +01:00
|
|
|
"[%s] [JOB %d] OpenCompactionOutputFiles for table #%" PRIu64
|
2014-11-04 20:07:11 +01:00
|
|
|
" fails at NewWritableFile with status %s",
|
2015-02-12 18:54:48 +01:00
|
|
|
compact_->compaction->column_family_data()->GetName().c_str(), job_id_,
|
2014-11-01 00:31:25 +01:00
|
|
|
file_number, s.ToString().c_str());
|
|
|
|
LogFlush(db_options_.info_log);
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
CompactionState::Output out;
|
|
|
|
out.number = file_number;
|
2015-07-09 00:21:10 +02:00
|
|
|
out.path_id = compact_->compaction->output_path_id();
|
2014-11-01 00:31:25 +01:00
|
|
|
out.smallest.Clear();
|
|
|
|
out.largest.Clear();
|
|
|
|
out.smallest_seqno = out.largest_seqno = 0;
|
|
|
|
|
|
|
|
compact_->outputs.push_back(out);
|
|
|
|
compact_->outfile->SetIOPriority(Env::IO_LOW);
|
2015-05-06 04:01:12 +02:00
|
|
|
compact_->outfile->SetPreallocationBlockSize(
|
|
|
|
static_cast<size_t>(compact_->compaction->OutputFilePreallocationSize()));
|
2014-11-01 00:31:25 +01:00
|
|
|
|
|
|
|
ColumnFamilyData* cfd = compact_->compaction->column_family_data();
|
2015-02-17 17:03:45 +01:00
|
|
|
bool skip_filters = false;
|
|
|
|
|
|
|
|
// If the Column family flag is to only optimize filters for hits,
|
|
|
|
// we can skip creating filters if this is the bottommost_level where
|
|
|
|
// data is going to be found
|
|
|
|
//
|
|
|
|
if (cfd->ioptions()->optimize_filters_for_hits && bottommost_level_) {
|
|
|
|
skip_filters = true;
|
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
compact_->builder.reset(NewTableBuilder(
|
A new call back to TablePropertiesCollector to allow users know the entry is add, delete or merge
Summary:
Currently users have no idea a key is add, delete or merge from TablePropertiesCollector call back. Add a new function to add it.
Also refactor the codes so that
(1) make table property collector and internal table property collector two separate data structures with the later one now exposed
(2) table builders only receive internal table properties
Test Plan: Add cases in table_properties_collector_test to cover both of old and new ways of using TablePropertiesCollector.
Reviewers: yhchiang, igor.sugak, rven, igor
Reviewed By: rven, igor
Subscribers: meyering, yoshinorim, maykov, leveldb, dhruba
Differential Revision: https://reviews.facebook.net/D35373
2015-04-06 19:04:30 +02:00
|
|
|
*cfd->ioptions(), cfd->internal_comparator(),
|
|
|
|
cfd->int_tbl_prop_collector_factories(), compact_->outfile.get(),
|
2015-07-09 00:21:10 +02:00
|
|
|
compact_->compaction->output_compression(),
|
2015-02-17 17:03:45 +01:00
|
|
|
cfd->ioptions()->compression_opts, skip_filters));
|
2014-11-01 00:31:25 +01:00
|
|
|
LogFlush(db_options_.info_log);
|
|
|
|
return s;
|
|
|
|
}
|
|
|
|
|
2014-11-10 20:57:58 +01:00
|
|
|
void CompactionJob::CleanupCompaction(const Status& status) {
|
2014-11-01 00:31:25 +01:00
|
|
|
if (compact_->builder != nullptr) {
|
|
|
|
// May happen if we get a shutdown call in the middle of compaction
|
|
|
|
compact_->builder->Abandon();
|
|
|
|
compact_->builder.reset();
|
|
|
|
} else {
|
|
|
|
assert(!status.ok() || compact_->outfile == nullptr);
|
|
|
|
}
|
|
|
|
for (size_t i = 0; i < compact_->outputs.size(); i++) {
|
|
|
|
const CompactionState::Output& out = compact_->outputs[i];
|
|
|
|
|
|
|
|
// If this file was inserted into the table cache then remove
|
|
|
|
// them here because this compaction was not committed.
|
|
|
|
if (!status.ok()) {
|
|
|
|
TableCache::Evict(table_cache_.get(), out.number);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
delete compact_;
|
|
|
|
compact_ = nullptr;
|
|
|
|
}
|
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
#ifndef ROCKSDB_LITE
|
|
|
|
namespace {
|
|
|
|
void CopyPrefix(
|
2015-06-04 21:31:12 +02:00
|
|
|
const Slice& src, size_t prefix_length, std::string* dst) {
|
|
|
|
assert(prefix_length > 0);
|
|
|
|
size_t length = src.size() > prefix_length ? prefix_length : src.size();
|
|
|
|
dst->assign(src.data(), length);
|
2015-06-03 02:07:16 +02:00
|
|
|
}
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
#endif // !ROCKSDB_LITE
|
|
|
|
|
2015-07-14 09:09:20 +02:00
|
|
|
void CompactionJob::UpdateCompactionStats() {
|
|
|
|
size_t num_output_files = compact_->outputs.size();
|
|
|
|
if (compact_->builder != nullptr) {
|
|
|
|
// An error occurred so ignore the last output.
|
|
|
|
assert(num_output_files > 0);
|
|
|
|
--num_output_files;
|
|
|
|
}
|
|
|
|
compaction_stats_.num_output_files = static_cast<int>(num_output_files);
|
|
|
|
|
2015-06-18 08:40:34 +02:00
|
|
|
Compaction* compaction = compact_->compaction;
|
|
|
|
compaction_stats_.num_input_files_in_non_output_levels = 0;
|
|
|
|
compaction_stats_.num_input_files_in_output_level = 0;
|
|
|
|
for (int input_level = 0;
|
|
|
|
input_level < static_cast<int>(compaction->num_input_levels());
|
|
|
|
++input_level) {
|
|
|
|
if (compaction->start_level() + input_level
|
|
|
|
!= compaction->output_level()) {
|
|
|
|
UpdateCompactionInputStatsHelper(
|
|
|
|
&compaction_stats_.num_input_files_in_non_output_levels,
|
|
|
|
&compaction_stats_.bytes_read_non_output_levels,
|
|
|
|
input_level);
|
|
|
|
} else {
|
|
|
|
UpdateCompactionInputStatsHelper(
|
|
|
|
&compaction_stats_.num_input_files_in_output_level,
|
|
|
|
&compaction_stats_.bytes_read_output_level,
|
|
|
|
input_level);
|
|
|
|
}
|
|
|
|
}
|
2015-07-14 09:09:20 +02:00
|
|
|
|
|
|
|
for (size_t i = 0; i < num_output_files; i++) {
|
|
|
|
compaction_stats_.bytes_written += compact_->outputs[i].file_size;
|
|
|
|
}
|
|
|
|
if (compact_->num_input_records > compact_->num_output_records) {
|
|
|
|
compaction_stats_.num_dropped_records +=
|
|
|
|
compact_->num_input_records - compact_->num_output_records;
|
|
|
|
}
|
2015-06-18 08:40:34 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
void CompactionJob::UpdateCompactionInputStatsHelper(
|
|
|
|
int* num_files, uint64_t* bytes_read, int input_level) {
|
|
|
|
const Compaction* compaction = compact_->compaction;
|
|
|
|
auto num_input_files = compaction->num_input_files(input_level);
|
|
|
|
*num_files += static_cast<int>(num_input_files);
|
|
|
|
|
|
|
|
for (size_t i = 0; i < num_input_files; ++i) {
|
|
|
|
const auto* file_meta = compaction->input(input_level, i);
|
|
|
|
*bytes_read += file_meta->fd.GetFileSize();
|
|
|
|
compaction_stats_.num_input_records +=
|
|
|
|
static_cast<uint64_t>(file_meta->num_entries);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-06-03 02:07:16 +02:00
|
|
|
void CompactionJob::UpdateCompactionJobStats(
|
|
|
|
const InternalStats::CompactionStats& stats) const {
|
|
|
|
#ifndef ROCKSDB_LITE
|
|
|
|
if (compaction_job_stats_) {
|
|
|
|
compaction_job_stats_->elapsed_micros = stats.micros;
|
|
|
|
|
|
|
|
// input information
|
|
|
|
compaction_job_stats_->total_input_bytes =
|
2015-06-18 08:40:34 +02:00
|
|
|
stats.bytes_read_non_output_levels +
|
|
|
|
stats.bytes_read_output_level;
|
|
|
|
compaction_job_stats_->num_input_records =
|
|
|
|
compact_->num_input_records;
|
2015-06-03 02:07:16 +02:00
|
|
|
compaction_job_stats_->num_input_files =
|
2015-06-18 08:40:34 +02:00
|
|
|
stats.num_input_files_in_non_output_levels +
|
|
|
|
stats.num_input_files_in_output_level;
|
2015-06-03 02:07:16 +02:00
|
|
|
compaction_job_stats_->num_input_files_at_output_level =
|
2015-06-18 08:40:34 +02:00
|
|
|
stats.num_input_files_in_output_level;
|
2015-06-03 02:07:16 +02:00
|
|
|
|
|
|
|
// output information
|
|
|
|
compaction_job_stats_->total_output_bytes = stats.bytes_written;
|
|
|
|
compaction_job_stats_->num_output_records =
|
|
|
|
compact_->num_output_records;
|
2015-06-18 08:40:34 +02:00
|
|
|
compaction_job_stats_->num_output_files = stats.num_output_files;
|
2015-06-03 02:07:16 +02:00
|
|
|
|
|
|
|
if (compact_->outputs.size() > 0U) {
|
|
|
|
CopyPrefix(
|
2015-06-04 21:31:12 +02:00
|
|
|
compact_->outputs[0].smallest.user_key(),
|
|
|
|
CompactionJobStats::kMaxPrefixLength,
|
|
|
|
&compaction_job_stats_->smallest_output_key_prefix);
|
2015-06-03 02:07:16 +02:00
|
|
|
CopyPrefix(
|
2015-06-04 21:31:12 +02:00
|
|
|
compact_->current_output()->largest.user_key(),
|
|
|
|
CompactionJobStats::kMaxPrefixLength,
|
|
|
|
&compaction_job_stats_->largest_output_key_prefix);
|
2015-06-03 02:07:16 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
#endif // !ROCKSDB_LITE
|
|
|
|
}
|
|
|
|
|
2015-07-14 09:09:20 +02:00
|
|
|
void CompactionJob::LogCompaction(
|
|
|
|
ColumnFamilyData* cfd, Compaction* compaction) {
|
|
|
|
// Let's check if anything will get logged. Don't prepare all the info if
|
|
|
|
// we're not logging
|
|
|
|
if (db_options_.info_log_level <= InfoLogLevel::INFO_LEVEL) {
|
|
|
|
Compaction::InputLevelSummaryBuffer inputs_summary;
|
|
|
|
Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
|
|
|
|
"[%s] [JOB %d] Compacting %s, score %.2f", cfd->GetName().c_str(),
|
|
|
|
job_id_, compaction->InputLevelSummary(&inputs_summary),
|
|
|
|
compaction->score());
|
|
|
|
char scratch[2345];
|
|
|
|
compaction->Summary(scratch, sizeof(scratch));
|
|
|
|
Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
|
|
|
|
"[%s] Compaction start summary: %s\n", cfd->GetName().c_str(), scratch);
|
|
|
|
// build event logger report
|
|
|
|
auto stream = event_logger_->Log();
|
|
|
|
stream << "job" << job_id_ << "event"
|
|
|
|
<< "compaction_started";
|
|
|
|
for (size_t i = 0; i < compaction->num_input_levels(); ++i) {
|
|
|
|
stream << ("files_L" + ToString(compaction->level(i)));
|
|
|
|
stream.StartArray();
|
|
|
|
for (auto f : *compaction->inputs(i)) {
|
|
|
|
stream << f->fd.GetNumber();
|
|
|
|
}
|
|
|
|
stream.EndArray();
|
|
|
|
}
|
|
|
|
stream << "score" << compaction->score() << "input_data_size"
|
|
|
|
<< compaction->CalculateTotalInputSize();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-01 00:31:25 +01:00
|
|
|
} // namespace rocksdb
|