mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
Update
[ghstack-poisoned]
This commit is contained in:
+27
-4
@@ -6708,6 +6708,27 @@ Status DBImpl::IngestExternalFiles(
|
|||||||
"external_files[" + std::to_string(i) + "] is empty";
|
"external_files[" + std::to_string(i) + "] is empty";
|
||||||
return Status::InvalidArgument(err_msg);
|
return Status::InvalidArgument(err_msg);
|
||||||
}
|
}
|
||||||
|
if (!args[i].file_infos.empty()) {
|
||||||
|
if (args[i].file_infos.size() != args[i].external_files.size()) {
|
||||||
|
return Status::InvalidArgument("file_infos[" + std::to_string(i) +
|
||||||
|
"] size must match external_files[" +
|
||||||
|
std::to_string(i) + "] size");
|
||||||
|
}
|
||||||
|
for (const auto* file_info : args[i].file_infos) {
|
||||||
|
if (file_info == nullptr) {
|
||||||
|
return Status::InvalidArgument(
|
||||||
|
"file_infos[" + std::to_string(i) +
|
||||||
|
"] contains a null PreparedFileInfo pointer; each entry must "
|
||||||
|
"point to a handle from SstFileWriter::Finish");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// file_infos avoids opening the file, so it cannot write a global seqno
|
||||||
|
// back into it. (write_global_seqno is deprecated and defaults to false.)
|
||||||
|
if (args[i].options.write_global_seqno) {
|
||||||
|
return Status::InvalidArgument(
|
||||||
|
"write_global_seqno is not supported when file_infos is set");
|
||||||
|
}
|
||||||
|
}
|
||||||
if (i && args[i].options.fill_cache != args[i - 1].options.fill_cache) {
|
if (i && args[i].options.fill_cache != args[i - 1].options.fill_cache) {
|
||||||
return Status::InvalidArgument(
|
return Status::InvalidArgument(
|
||||||
"fill_cache should be the same across ingestion options.");
|
"fill_cache should be the same across ingestion options.");
|
||||||
@@ -6801,8 +6822,9 @@ Status DBImpl::IngestExternalFiles(
|
|||||||
this);
|
this);
|
||||||
Status es = ingestion_jobs[i].Prepare(
|
Status es = ingestion_jobs[i].Prepare(
|
||||||
args[i].external_files, args[i].files_checksums,
|
args[i].external_files, args[i].files_checksums,
|
||||||
args[i].files_checksum_func_names, args[i].atomic_replace_range,
|
args[i].files_checksum_func_names, args[i].file_infos,
|
||||||
args[i].file_temperature, start_file_number, super_version);
|
args[i].atomic_replace_range, args[i].file_temperature,
|
||||||
|
start_file_number, super_version);
|
||||||
// capture first error only
|
// capture first error only
|
||||||
if (!es.ok() && status.ok()) {
|
if (!es.ok() && status.ok()) {
|
||||||
status = es;
|
status = es;
|
||||||
@@ -6817,8 +6839,9 @@ Status DBImpl::IngestExternalFiles(
|
|||||||
this);
|
this);
|
||||||
Status es = ingestion_jobs[0].Prepare(
|
Status es = ingestion_jobs[0].Prepare(
|
||||||
args[0].external_files, args[0].files_checksums,
|
args[0].external_files, args[0].files_checksums,
|
||||||
args[0].files_checksum_func_names, args[0].atomic_replace_range,
|
args[0].files_checksum_func_names, args[0].file_infos,
|
||||||
args[0].file_temperature, next_file_number, super_version);
|
args[0].atomic_replace_range, args[0].file_temperature,
|
||||||
|
next_file_number, super_version);
|
||||||
if (!es.ok()) {
|
if (!es.ok()) {
|
||||||
status = es;
|
status = es;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,6 +19,7 @@
|
|||||||
#include "logging/logging.h"
|
#include "logging/logging.h"
|
||||||
#include "monitoring/statistics_impl.h"
|
#include "monitoring/statistics_impl.h"
|
||||||
#include "table/merging_iterator.h"
|
#include "table/merging_iterator.h"
|
||||||
|
#include "table/prepared_file_info.h"
|
||||||
#include "table/sst_file_writer_collectors.h"
|
#include "table/sst_file_writer_collectors.h"
|
||||||
#include "table/table_builder.h"
|
#include "table/table_builder.h"
|
||||||
#include "table/unique_id_impl.h"
|
#include "table/unique_id_impl.h"
|
||||||
@@ -31,18 +32,22 @@ Status ExternalSstFileIngestionJob::Prepare(
|
|||||||
const std::vector<std::string>& external_files_paths,
|
const std::vector<std::string>& external_files_paths,
|
||||||
const std::vector<std::string>& files_checksums,
|
const std::vector<std::string>& files_checksums,
|
||||||
const std::vector<std::string>& files_checksum_func_names,
|
const std::vector<std::string>& files_checksum_func_names,
|
||||||
|
const std::vector<const PreparedFileInfo*>& file_infos,
|
||||||
const std::optional<RangeOpt>& atomic_replace_range,
|
const std::optional<RangeOpt>& atomic_replace_range,
|
||||||
const Temperature& file_temperature, uint64_t next_file_number,
|
const Temperature& file_temperature, uint64_t next_file_number,
|
||||||
SuperVersion* sv) {
|
SuperVersion* sv) {
|
||||||
Status status;
|
Status status;
|
||||||
|
|
||||||
// Read the information of files we are ingesting
|
// Read the information of files we are ingesting.
|
||||||
for (const std::string& file_path : external_files_paths) {
|
for (size_t i = 0; i < external_files_paths.size(); i++) {
|
||||||
|
const std::string& file_path = external_files_paths[i];
|
||||||
IngestedFileInfo file_to_ingest;
|
IngestedFileInfo file_to_ingest;
|
||||||
// For temperature, first assume it matches provided hint
|
// For temperature, first assume it matches provided hint
|
||||||
file_to_ingest.file_temperature = file_temperature;
|
file_to_ingest.file_temperature = file_temperature;
|
||||||
status =
|
const PreparedFileInfo* file_info =
|
||||||
GetIngestedFileInfo(file_path, next_file_number++, &file_to_ingest, sv);
|
file_infos.empty() ? nullptr : file_infos[i];
|
||||||
|
status = GetIngestedFileInfo(file_path, next_file_number++, file_info,
|
||||||
|
&file_to_ingest, sv);
|
||||||
if (!status.ok()) {
|
if (!status.ok()) {
|
||||||
ROCKS_LOG_WARN(db_options_.info_log,
|
ROCKS_LOG_WARN(db_options_.info_log,
|
||||||
"Failed to get ingested file info: %s: %s",
|
"Failed to get ingested file info: %s: %s",
|
||||||
@@ -970,18 +975,14 @@ Status ExternalSstFileIngestionJob::ResetTableReader(
|
|||||||
}
|
}
|
||||||
|
|
||||||
Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
||||||
const std::string& external_file, uint64_t new_file_number,
|
const std::string& external_file, const TableProperties& props,
|
||||||
SuperVersion* sv, IngestedFileInfo* file_to_ingest,
|
IngestedFileInfo* file_to_ingest) {
|
||||||
std::unique_ptr<TableReader>* table_reader) {
|
const auto& uprops = props.user_collected_properties;
|
||||||
// Get the external file properties
|
|
||||||
auto props = table_reader->get()->GetTableProperties();
|
|
||||||
assert(props.get());
|
|
||||||
const auto& uprops = props->user_collected_properties;
|
|
||||||
|
|
||||||
// Get table version
|
// Get table version
|
||||||
auto version_iter = uprops.find(ExternalSstFilePropertyNames::kVersion);
|
auto version_iter = uprops.find(ExternalSstFilePropertyNames::kVersion);
|
||||||
if (version_iter == uprops.end()) {
|
if (version_iter == uprops.end()) {
|
||||||
assert(!SstFileWriter::CreatedBySstFileWriter(*props));
|
assert(!SstFileWriter::CreatedBySstFileWriter(props));
|
||||||
if (!ingestion_options_.allow_db_generated_files) {
|
if (!ingestion_options_.allow_db_generated_files) {
|
||||||
return Status::Corruption("External file version not found");
|
return Status::Corruption("External file version not found");
|
||||||
} else {
|
} else {
|
||||||
@@ -990,7 +991,7 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
|||||||
file_to_ingest->version = 0;
|
file_to_ingest->version = 0;
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
assert(SstFileWriter::CreatedBySstFileWriter(*props));
|
assert(SstFileWriter::CreatedBySstFileWriter(props));
|
||||||
file_to_ingest->version = DecodeFixed32(version_iter->second.c_str());
|
file_to_ingest->version = DecodeFixed32(version_iter->second.c_str());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1004,12 +1005,17 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
|||||||
|
|
||||||
// Set the global sequence number
|
// Set the global sequence number
|
||||||
file_to_ingest->original_seqno = DecodeFixed64(seqno_iter->second.c_str());
|
file_to_ingest->original_seqno = DecodeFixed64(seqno_iter->second.c_str());
|
||||||
if (props->external_sst_file_global_seqno_offset == 0) {
|
file_to_ingest->global_seqno_offset =
|
||||||
file_to_ingest->global_seqno_offset = 0;
|
static_cast<size_t>(props.external_sst_file_global_seqno_offset);
|
||||||
|
// The on-disk offset is only needed if we will write the global seqno back
|
||||||
|
// into the file (write_global_seqno). The metadata fast-path does not open
|
||||||
|
// the file, and its in-memory table properties do not carry the offset
|
||||||
|
// (it is only computed while reading the file back); that is fine as long
|
||||||
|
// as write_global_seqno is not requested.
|
||||||
|
if (ingestion_options_.write_global_seqno &&
|
||||||
|
file_to_ingest->global_seqno_offset == 0) {
|
||||||
return Status::Corruption("Was not able to find file global seqno field");
|
return Status::Corruption("Was not able to find file global seqno field");
|
||||||
}
|
}
|
||||||
file_to_ingest->global_seqno_offset =
|
|
||||||
static_cast<size_t>(props->external_sst_file_global_seqno_offset);
|
|
||||||
} else if (file_to_ingest->version == 1) {
|
} else if (file_to_ingest->version == 1) {
|
||||||
// SST file V1 should not have global seqno field
|
// SST file V1 should not have global seqno field
|
||||||
assert(seqno_iter == uprops.end());
|
assert(seqno_iter == uprops.end());
|
||||||
@@ -1030,23 +1036,20 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
|||||||
" is not supported");
|
" is not supported");
|
||||||
}
|
}
|
||||||
|
|
||||||
file_to_ingest->cf_id = static_cast<uint32_t>(props->column_family_id);
|
file_to_ingest->cf_id = static_cast<uint32_t>(props.column_family_id);
|
||||||
// This assignment works fine even though `table_reader` may later be reset,
|
file_to_ingest->table_properties = props;
|
||||||
// since that will not affect how table properties are parsed, and this
|
|
||||||
// assignment is making a copy.
|
|
||||||
file_to_ingest->table_properties = *props;
|
|
||||||
|
|
||||||
// Get number of entries in table
|
// Get number of entries in table
|
||||||
file_to_ingest->num_entries = props->num_entries;
|
file_to_ingest->num_entries = props.num_entries;
|
||||||
file_to_ingest->num_range_deletions = props->num_range_deletions;
|
file_to_ingest->num_range_deletions = props.num_range_deletions;
|
||||||
|
|
||||||
// Validate table properties related to comparator name and user defined
|
// Validate table properties related to comparator name and user defined
|
||||||
// timestamps persisted flag.
|
// timestamps persisted flag.
|
||||||
file_to_ingest->user_defined_timestamps_persisted =
|
file_to_ingest->user_defined_timestamps_persisted =
|
||||||
static_cast<bool>(props->user_defined_timestamps_persisted);
|
static_cast<bool>(props.user_defined_timestamps_persisted);
|
||||||
bool mark_sst_file_has_no_udt = false;
|
bool mark_sst_file_has_no_udt = false;
|
||||||
Status s = ValidateUserDefinedTimestampsOptions(
|
Status s = ValidateUserDefinedTimestampsOptions(
|
||||||
cfd_->user_comparator(), props->comparator_name,
|
cfd_->user_comparator(), props.comparator_name,
|
||||||
cfd_->ioptions().persist_user_defined_timestamps,
|
cfd_->ioptions().persist_user_defined_timestamps,
|
||||||
file_to_ingest->user_defined_timestamps_persisted,
|
file_to_ingest->user_defined_timestamps_persisted,
|
||||||
&mark_sst_file_has_no_udt);
|
&mark_sst_file_has_no_udt);
|
||||||
@@ -1054,7 +1057,9 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
|||||||
// A column family that enables user-defined timestamps in Memtable only
|
// A column family that enables user-defined timestamps in Memtable only
|
||||||
// feature can also ingest external files created by a setting that disables
|
// feature can also ingest external files created by a setting that disables
|
||||||
// user-defined timestamps. In that case, we need to re-mark the
|
// user-defined timestamps. In that case, we need to re-mark the
|
||||||
// user_defined_timestamps_persisted flag for the file.
|
// user_defined_timestamps_persisted flag for the file. The open-and-scan
|
||||||
|
// caller is then responsible for reopening its `TableReader` with the
|
||||||
|
// updated flag.
|
||||||
file_to_ingest->user_defined_timestamps_persisted = false;
|
file_to_ingest->user_defined_timestamps_persisted = false;
|
||||||
} else if (!s.ok()) {
|
} else if (!s.ok()) {
|
||||||
ROCKS_LOG_WARN(
|
ROCKS_LOG_WARN(
|
||||||
@@ -1064,22 +1069,88 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
|
|||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|
||||||
// `TableReader` is initialized with `user_defined_timestamps_persisted` flag
|
|
||||||
// to be true. If its value changed to false after this sanity check, we
|
|
||||||
// need to reset the `TableReader`.
|
|
||||||
if (ucmp_->timestamp_size() > 0 &&
|
|
||||||
!file_to_ingest->user_defined_timestamps_persisted) {
|
|
||||||
s = ResetTableReader(external_file, new_file_number,
|
|
||||||
file_to_ingest->user_defined_timestamps_persisted, sv,
|
|
||||||
file_to_ingest, table_reader);
|
|
||||||
}
|
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|
||||||
Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
Status ExternalSstFileIngestionJob::GetIngestedFileInfoFromFileInfo(
|
||||||
|
const std::string& external_file, const PreparedFileInfo& file_info,
|
||||||
|
IngestedFileInfo* file_to_ingest) {
|
||||||
|
// Boundaries, size, and table properties were obtained without opening the
|
||||||
|
// file (e.g. produced by SstFileWriter::Finish). We reuse them directly.
|
||||||
|
file_to_ingest->file_size = file_info.file_size;
|
||||||
|
file_to_ingest->smallest_internal_key.DecodeFrom(
|
||||||
|
file_info.smallest_internal_key);
|
||||||
|
file_to_ingest->largest_internal_key.DecodeFrom(
|
||||||
|
file_info.largest_internal_key);
|
||||||
|
Status status = SanityCheckTableProperties(
|
||||||
|
external_file, file_info.table_properties, file_to_ingest);
|
||||||
|
if (!status.ok()) {
|
||||||
|
ROCKS_LOG_WARN(
|
||||||
|
db_options_.info_log,
|
||||||
|
"Failed to sanity check table properties for external file %s: %s",
|
||||||
|
external_file.c_str(), status.ToString().c_str());
|
||||||
|
return status;
|
||||||
|
}
|
||||||
|
|
||||||
|
// When the column family enables user-defined timestamps but this file was
|
||||||
|
// written without persisting them (the "UDT in Memtables only" feature), the
|
||||||
|
// supplied boundary keys carry no timestamp (SstFileWriter strips it). Pad
|
||||||
|
// them with the min timestamp so they match the format the open-and-scan path
|
||||||
|
// produces (its reopened `TableReader` pads keys the same way) and the bounds
|
||||||
|
// recorded in the new FileMetaData.
|
||||||
|
const size_t ts_sz = ucmp_->timestamp_size();
|
||||||
|
if (ts_sz > 0 && !file_to_ingest->user_defined_timestamps_persisted) {
|
||||||
|
std::string padded;
|
||||||
|
PadInternalKeyWithMinTimestamp(
|
||||||
|
&padded, file_to_ingest->smallest_internal_key.Encode(), ts_sz);
|
||||||
|
file_to_ingest->smallest_internal_key.DecodeFrom(padded);
|
||||||
|
padded.clear();
|
||||||
|
PadInternalKeyWithMinTimestamp(
|
||||||
|
&padded, file_to_ingest->largest_internal_key.Encode(), ts_sz);
|
||||||
|
file_to_ingest->largest_internal_key.DecodeFrom(padded);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Extend the key bounds to cover any range tombstones. Their bounds are not
|
||||||
|
// part of the point-key internal keys, but the file's recorded range must
|
||||||
|
// include them (the open-and-scan path does this while scanning tombstones).
|
||||||
|
// SstFileWriter writes range tombstones at sequence number 0.
|
||||||
|
if (!file_info.smallest_range_del_key.empty()) {
|
||||||
|
RangeTombstone tombstone(Slice(file_info.smallest_range_del_key),
|
||||||
|
Slice(file_info.largest_range_del_key),
|
||||||
|
/*sn=*/0);
|
||||||
|
file_range_checker_.MaybeUpdateRange(
|
||||||
|
tombstone.SerializeKey(), tombstone.SerializeEndKey(), file_to_ingest);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (ingestion_options_.allow_db_generated_files) {
|
||||||
|
// Sequence numbers are preserved (not reassigned), so the bounds come from
|
||||||
|
// the supplied table properties; the GetSeqnoBoundaryForFile scan is
|
||||||
|
// skipped.
|
||||||
|
file_to_ingest->smallest_seqno =
|
||||||
|
file_to_ingest->table_properties.key_smallest_seqno;
|
||||||
|
file_to_ingest->largest_seqno =
|
||||||
|
file_to_ingest->table_properties.key_largest_seqno;
|
||||||
|
} else {
|
||||||
|
// Normal ingestion reassigns a global sequence number later, so the file's
|
||||||
|
// keys must currently be at seqno 0 (mirror of the open-and-scan check).
|
||||||
|
SequenceNumber largest_seqno =
|
||||||
|
file_to_ingest->table_properties.key_largest_seqno;
|
||||||
|
// UINT64_MAX means unknown and the file is generated before table property
|
||||||
|
// `key_largest_seqno` is introduced.
|
||||||
|
if (largest_seqno != UINT64_MAX && largest_seqno > 0) {
|
||||||
|
return Status::Corruption(
|
||||||
|
"External file has non zero largest sequence number " +
|
||||||
|
std::to_string(largest_seqno));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return Status::OK();
|
||||||
|
}
|
||||||
|
|
||||||
|
Status ExternalSstFileIngestionJob::GetIngestedFileInfoFromFile(
|
||||||
const std::string& external_file, uint64_t new_file_number,
|
const std::string& external_file, uint64_t new_file_number,
|
||||||
IngestedFileInfo* file_to_ingest, SuperVersion* sv) {
|
IngestedFileInfo* file_to_ingest, SuperVersion* sv,
|
||||||
file_to_ingest->external_file_path = external_file;
|
std::unique_ptr<TableReader>* out_table_reader) {
|
||||||
|
TEST_SYNC_POINT("ExternalSstFileIngestionJob::GetIngestedFileInfo:ReadPath");
|
||||||
|
|
||||||
// Get external file size
|
// Get external file size
|
||||||
Status status = fs_->GetFileSize(external_file, IOOptions(),
|
Status status = fs_->GetFileSize(external_file, IOOptions(),
|
||||||
@@ -1091,15 +1162,11 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
return status;
|
return status;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Assign FD with number
|
// Create TableReader for external file.
|
||||||
file_to_ingest->fd =
|
|
||||||
FileDescriptor(new_file_number, 0, file_to_ingest->file_size);
|
|
||||||
|
|
||||||
// Create TableReader for external file
|
|
||||||
std::unique_ptr<TableReader> table_reader;
|
std::unique_ptr<TableReader> table_reader;
|
||||||
// Initially create the `TableReader` with flag
|
// Initially create the `TableReader` with flag
|
||||||
// `user_defined_timestamps_persisted` to be true since that's the most common
|
// `user_defined_timestamps_persisted` to be true since that's the most
|
||||||
// case
|
// common case
|
||||||
status = ResetTableReader(external_file, new_file_number,
|
status = ResetTableReader(external_file, new_file_number,
|
||||||
/*user_defined_timestamps_persisted=*/true, sv,
|
/*user_defined_timestamps_persisted=*/true, sv,
|
||||||
file_to_ingest, &table_reader);
|
file_to_ingest, &table_reader);
|
||||||
@@ -1110,8 +1177,8 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
return status;
|
return status;
|
||||||
}
|
}
|
||||||
|
|
||||||
status = SanityCheckTableProperties(external_file, new_file_number, sv,
|
status = SanityCheckTableProperties(
|
||||||
file_to_ingest, &table_reader);
|
external_file, *table_reader->GetTableProperties(), file_to_ingest);
|
||||||
if (!status.ok()) {
|
if (!status.ok()) {
|
||||||
ROCKS_LOG_WARN(
|
ROCKS_LOG_WARN(
|
||||||
db_options_.info_log,
|
db_options_.info_log,
|
||||||
@@ -1120,6 +1187,23 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
return status;
|
return status;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The `TableReader` above was opened with `user_defined_timestamps_persisted`
|
||||||
|
// assumed true. If the sanity check determined the file has no persisted
|
||||||
|
// timestamps (UDT-in-Memtable-only feature), reopen it with the corrected
|
||||||
|
// flag so keys are parsed properly by the scan below.
|
||||||
|
if (ucmp_->timestamp_size() > 0 &&
|
||||||
|
!file_to_ingest->user_defined_timestamps_persisted) {
|
||||||
|
status = ResetTableReader(external_file, new_file_number,
|
||||||
|
file_to_ingest->user_defined_timestamps_persisted,
|
||||||
|
sv, file_to_ingest, &table_reader);
|
||||||
|
if (!status.ok()) {
|
||||||
|
ROCKS_LOG_WARN(db_options_.info_log,
|
||||||
|
"Failed to reset table reader for external file %s: %s",
|
||||||
|
external_file.c_str(), status.ToString().c_str());
|
||||||
|
return status;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
const bool allow_data_in_errors = db_options_.allow_data_in_errors;
|
const bool allow_data_in_errors = db_options_.allow_data_in_errors;
|
||||||
ParsedInternalKey key;
|
ParsedInternalKey key;
|
||||||
if (ingestion_options_.allow_db_generated_files) {
|
if (ingestion_options_.allow_db_generated_files) {
|
||||||
@@ -1141,8 +1225,8 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
} else {
|
} else {
|
||||||
SequenceNumber largest_seqno =
|
SequenceNumber largest_seqno =
|
||||||
table_reader.get()->GetTableProperties()->key_largest_seqno;
|
table_reader.get()->GetTableProperties()->key_largest_seqno;
|
||||||
// UINT64_MAX means unknown and the file is generated before table property
|
// UINT64_MAX means unknown and the file is generated before table
|
||||||
// `key_largest_seqno` is introduced.
|
// property `key_largest_seqno` is introduced.
|
||||||
if (largest_seqno != UINT64_MAX && largest_seqno > 0) {
|
if (largest_seqno != UINT64_MAX && largest_seqno > 0) {
|
||||||
return Status::Corruption(
|
return Status::Corruption(
|
||||||
"External file has non zero largest sequence number " +
|
"External file has non zero largest sequence number " +
|
||||||
@@ -1150,24 +1234,6 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (ingestion_options_.verify_checksums_before_ingest) {
|
|
||||||
// If customized readahead size is needed, we can pass a user option
|
|
||||||
// all the way to here. Right now we just rely on the default readahead
|
|
||||||
// to keep things simple.
|
|
||||||
// TODO: plumb Env::IOActivity, Env::IOPriority
|
|
||||||
ReadOptions ro;
|
|
||||||
ro.readahead_size = ingestion_options_.verify_checksums_readahead_size;
|
|
||||||
ro.fill_cache = ingestion_options_.fill_cache;
|
|
||||||
status = table_reader->VerifyChecksum(
|
|
||||||
ro, TableReaderCaller::kExternalSSTIngestion);
|
|
||||||
if (!status.ok()) {
|
|
||||||
ROCKS_LOG_WARN(db_options_.info_log,
|
|
||||||
"Failed to verify checksum for table reader: %s",
|
|
||||||
status.ToString().c_str());
|
|
||||||
return status;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: plumb Env::IOActivity, Env::IOPriority
|
// TODO: plumb Env::IOActivity, Env::IOPriority
|
||||||
ReadOptions ro;
|
ReadOptions ro;
|
||||||
ro.fill_cache = ingestion_options_.fill_cache;
|
ro.fill_cache = ingestion_options_.fill_cache;
|
||||||
@@ -1254,6 +1320,75 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
*out_table_reader = std::move(table_reader);
|
||||||
|
return Status::OK();
|
||||||
|
}
|
||||||
|
|
||||||
|
Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
|
||||||
|
const std::string& external_file, uint64_t new_file_number,
|
||||||
|
const PreparedFileInfo* file_info, IngestedFileInfo* file_to_ingest,
|
||||||
|
SuperVersion* sv) {
|
||||||
|
file_to_ingest->external_file_path = external_file;
|
||||||
|
|
||||||
|
std::unique_ptr<TableReader> table_reader;
|
||||||
|
Status status =
|
||||||
|
file_info != nullptr
|
||||||
|
? GetIngestedFileInfoFromFileInfo(external_file, *file_info,
|
||||||
|
file_to_ingest)
|
||||||
|
: GetIngestedFileInfoFromFile(external_file, new_file_number,
|
||||||
|
file_to_ingest, sv, &table_reader);
|
||||||
|
if (!status.ok()) {
|
||||||
|
return status;
|
||||||
|
}
|
||||||
|
|
||||||
|
assert(file_to_ingest->file_size > 0);
|
||||||
|
assert(!file_to_ingest->unset());
|
||||||
|
assert(file_to_ingest->table_properties.num_entries > 0 ||
|
||||||
|
file_to_ingest->table_properties.num_range_deletions > 0);
|
||||||
|
if (ingestion_options_.allow_db_generated_files) {
|
||||||
|
// These files keep their original sequence numbers (derived, not
|
||||||
|
// reassigned), so the bounds must be valid here.
|
||||||
|
assert(file_to_ingest->smallest_seqno <= file_to_ingest->largest_seqno);
|
||||||
|
assert(file_to_ingest->largest_seqno < kMaxSequenceNumber);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verify the file checksum if requested. The open-and-scan path already has a
|
||||||
|
// `TableReader` open; the fast-path opens one here only when verification is
|
||||||
|
// requested -- otherwise it performs no file I/O.
|
||||||
|
if (ingestion_options_.verify_checksums_before_ingest) {
|
||||||
|
if (table_reader == nullptr) {
|
||||||
|
status =
|
||||||
|
ResetTableReader(external_file, new_file_number,
|
||||||
|
file_to_ingest->user_defined_timestamps_persisted,
|
||||||
|
sv, file_to_ingest, &table_reader);
|
||||||
|
if (!status.ok()) {
|
||||||
|
ROCKS_LOG_WARN(db_options_.info_log,
|
||||||
|
"Failed to reset table reader for external file %s: %s",
|
||||||
|
external_file.c_str(), status.ToString().c_str());
|
||||||
|
return status;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// If customized readahead size is needed, we can pass a user option all the
|
||||||
|
// way to here. Right now we just rely on the default readahead to keep
|
||||||
|
// things simple.
|
||||||
|
// TODO: plumb Env::IOActivity, Env::IOPriority
|
||||||
|
ReadOptions ro;
|
||||||
|
ro.readahead_size = ingestion_options_.verify_checksums_readahead_size;
|
||||||
|
ro.fill_cache = ingestion_options_.fill_cache;
|
||||||
|
status = table_reader->VerifyChecksum(
|
||||||
|
ro, TableReaderCaller::kExternalSSTIngestion);
|
||||||
|
if (!status.ok()) {
|
||||||
|
ROCKS_LOG_WARN(db_options_.info_log,
|
||||||
|
"Failed to verify checksum for external file %s: %s",
|
||||||
|
external_file.c_str(), status.ToString().c_str());
|
||||||
|
return status;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Assign FD with number.
|
||||||
|
file_to_ingest->fd =
|
||||||
|
FileDescriptor(new_file_number, 0, file_to_ingest->file_size);
|
||||||
|
|
||||||
const size_t ts_sz = ucmp_->timestamp_size();
|
const size_t ts_sz = ucmp_->timestamp_size();
|
||||||
Slice smallest = file_to_ingest->smallest_internal_key.user_key();
|
Slice smallest = file_to_ingest->smallest_internal_key.user_key();
|
||||||
Slice largest = file_to_ingest->largest_internal_key.user_key();
|
Slice largest = file_to_ingest->largest_internal_key.user_key();
|
||||||
|
|||||||
@@ -248,6 +248,7 @@ class ExternalSstFileIngestionJob {
|
|||||||
Status Prepare(const std::vector<std::string>& external_files_paths,
|
Status Prepare(const std::vector<std::string>& external_files_paths,
|
||||||
const std::vector<std::string>& files_checksums,
|
const std::vector<std::string>& files_checksums,
|
||||||
const std::vector<std::string>& files_checksum_func_names,
|
const std::vector<std::string>& files_checksum_func_names,
|
||||||
|
const std::vector<const PreparedFileInfo*>& file_infos,
|
||||||
const std::optional<RangeOpt>& atomic_replace_range,
|
const std::optional<RangeOpt>& atomic_replace_range,
|
||||||
const Temperature& file_temperature, uint64_t next_file_number,
|
const Temperature& file_temperature, uint64_t next_file_number,
|
||||||
SuperVersion* sv);
|
SuperVersion* sv);
|
||||||
@@ -315,17 +316,36 @@ class ExternalSstFileIngestionJob {
|
|||||||
// different options. For example: when external file does not contain
|
// different options. For example: when external file does not contain
|
||||||
// timestamps while column family enables UDT in Memtables only feature.
|
// timestamps while column family enables UDT in Memtables only feature.
|
||||||
Status SanityCheckTableProperties(const std::string& external_file,
|
Status SanityCheckTableProperties(const std::string& external_file,
|
||||||
uint64_t new_file_number, SuperVersion* sv,
|
const TableProperties& props,
|
||||||
IngestedFileInfo* file_to_ingest,
|
IngestedFileInfo* file_to_ingest);
|
||||||
std::unique_ptr<TableReader>* table_reader);
|
|
||||||
|
|
||||||
// Open the external file and populate `file_to_ingest` with all the
|
// Open the external file and populate `file_to_ingest` with all the
|
||||||
// external information we need to ingest this file.
|
// external information we need to ingest this file. When `file_info` is
|
||||||
|
// non-null, its caller-supplied metadata is reused instead of opening and
|
||||||
|
// scanning the file.
|
||||||
Status GetIngestedFileInfo(const std::string& external_file,
|
Status GetIngestedFileInfo(const std::string& external_file,
|
||||||
uint64_t new_file_number,
|
uint64_t new_file_number,
|
||||||
|
const PreparedFileInfo* file_info,
|
||||||
IngestedFileInfo* file_to_ingest,
|
IngestedFileInfo* file_to_ingest,
|
||||||
SuperVersion* sv);
|
SuperVersion* sv);
|
||||||
|
|
||||||
|
// Acquire the per-file metadata from the caller-supplied opaque
|
||||||
|
// `PreparedFileInfo` (produced by SstFileWriter::Finish) instead of opening
|
||||||
|
// the file.
|
||||||
|
Status GetIngestedFileInfoFromFileInfo(const std::string& external_file,
|
||||||
|
const PreparedFileInfo& file_info,
|
||||||
|
IngestedFileInfo* file_to_ingest);
|
||||||
|
|
||||||
|
// Acquire the per-file metadata by opening the external file and scanning it
|
||||||
|
// (table properties, sequence number bounds, and boundary keys including any
|
||||||
|
// range-tombstone extensions). Used when no file_info is available. The
|
||||||
|
// opened `TableReader` is returned via `*table_reader` so the caller can
|
||||||
|
// reuse it (e.g. to verify the file checksum) without re-opening the file.
|
||||||
|
Status GetIngestedFileInfoFromFile(
|
||||||
|
const std::string& external_file, uint64_t new_file_number,
|
||||||
|
IngestedFileInfo* file_to_ingest, SuperVersion* sv,
|
||||||
|
std::unique_ptr<TableReader>* out_table_reader);
|
||||||
|
|
||||||
// If the input files' key range overlaps themselves, this function divides
|
// If the input files' key range overlaps themselves, this function divides
|
||||||
// them in the user specified order into multiple batches. Where the files
|
// them in the user specified order into multiple batches. Where the files
|
||||||
// within a batch do not overlap with each other, but key range could overlap
|
// within a batch do not overlap with each other, but key range could overlap
|
||||||
|
|||||||
@@ -152,7 +152,8 @@ class ExternalSSTFileTest
|
|||||||
bool verify_checksums_before_ingest = true, bool ingest_behind = false,
|
bool verify_checksums_before_ingest = true, bool ingest_behind = false,
|
||||||
bool sort_data = false,
|
bool sort_data = false,
|
||||||
std::map<std::string, std::string>* true_data = nullptr,
|
std::map<std::string, std::string>* true_data = nullptr,
|
||||||
ColumnFamilyHandle* cfh = nullptr, bool fill_cache = false) {
|
ColumnFamilyHandle* cfh = nullptr, bool fill_cache = false,
|
||||||
|
bool ingest_with_file_info = false) {
|
||||||
// Generate a file id if not provided
|
// Generate a file id if not provided
|
||||||
if (file_id == -1) {
|
if (file_id == -1) {
|
||||||
file_id = last_file_id_ + 1;
|
file_id = last_file_id_ + 1;
|
||||||
@@ -188,7 +189,13 @@ class ExternalSSTFileTest
|
|||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
s = sst_file_writer.Finish();
|
ExternalSstFileInfo file_info;
|
||||||
|
std::shared_ptr<const PreparedFileInfo> prepared_file_info;
|
||||||
|
if (ingest_with_file_info) {
|
||||||
|
s = sst_file_writer.Finish(&file_info, &prepared_file_info);
|
||||||
|
} else {
|
||||||
|
s = sst_file_writer.Finish();
|
||||||
|
}
|
||||||
|
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
IngestExternalFileOptions ifo;
|
IngestExternalFileOptions ifo;
|
||||||
@@ -197,7 +204,14 @@ class ExternalSSTFileTest
|
|||||||
ifo.verify_checksums_before_ingest = verify_checksums_before_ingest;
|
ifo.verify_checksums_before_ingest = verify_checksums_before_ingest;
|
||||||
ifo.ingest_behind = ingest_behind;
|
ifo.ingest_behind = ingest_behind;
|
||||||
ifo.fill_cache = fill_cache;
|
ifo.fill_cache = fill_cache;
|
||||||
if (cfh) {
|
if (ingest_with_file_info) {
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = cfh ? cfh : db_->DefaultColumnFamily();
|
||||||
|
arg.external_files = {file_path};
|
||||||
|
arg.file_infos = {prepared_file_info.get()};
|
||||||
|
arg.options = ifo;
|
||||||
|
s = db_->IngestExternalFiles({arg});
|
||||||
|
} else if (cfh) {
|
||||||
s = db_->IngestExternalFile(cfh, {file_path}, ifo);
|
s = db_->IngestExternalFile(cfh, {file_path}, ifo);
|
||||||
} else {
|
} else {
|
||||||
s = db_->IngestExternalFile({file_path}, ifo);
|
s = db_->IngestExternalFile({file_path}, ifo);
|
||||||
@@ -340,6 +354,139 @@ TEST_F(ExternalSSTFileTest, IngestionTimingHistogram) {
|
|||||||
ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_RUN_TIME).count);
|
ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_RUN_TIME).count);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Ingestion that reuses the SstFileWriter's metadata via
|
||||||
|
// IngestExternalFileArg::file_infos skips the open-and-scan path and still
|
||||||
|
// produces correct data, including overlapping data that is reassigned a new
|
||||||
|
// global sequence number.
|
||||||
|
TEST_F(ExternalSSTFileTest, IngestWithFileInfo) {
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
DestroyAndReopen(options);
|
||||||
|
std::map<std::string, std::string> true_data;
|
||||||
|
|
||||||
|
// Count entries into the open-and-scan path so we can assert it is skipped.
|
||||||
|
std::atomic<int> read_path_count{0};
|
||||||
|
SyncPoint::GetInstance()->SetCallBack(
|
||||||
|
"ExternalSstFileIngestionJob::GetIngestedFileInfo:ReadPath",
|
||||||
|
[&](void*) { read_path_count.fetch_add(1); });
|
||||||
|
SyncPoint::GetInstance()->EnableProcessing();
|
||||||
|
|
||||||
|
// Reuse the writer's metadata, with the normal ingestion options: a global
|
||||||
|
// sequence number is assigned (allow_db_generated_files is NOT set) and the
|
||||||
|
// checksum is verified (the fast-path opens the file only to verify it, which
|
||||||
|
// is not a metadata scan).
|
||||||
|
auto ingest_fi = [&](std::vector<std::pair<std::string, std::string>> data) {
|
||||||
|
return GenerateAndAddExternalFile(
|
||||||
|
options, std::move(data), /*file_id=*/-1, /*allow_global_seqno=*/true,
|
||||||
|
/*write_global_seqno=*/false, /*verify_checksums_before_ingest=*/true,
|
||||||
|
/*ingest_behind=*/false, /*sort_data=*/true, &true_data,
|
||||||
|
/*cfh=*/nullptr, /*fill_cache=*/false, /*ingest_with_file_info=*/true);
|
||||||
|
};
|
||||||
|
|
||||||
|
// Non-overlapping ingest: data reads back and the file is not scanned.
|
||||||
|
ASSERT_OK(ingest_fi({{Key(1), "a1"}, {Key(2), "a2"}, {Key(3), "a3"}}));
|
||||||
|
ASSERT_EQ(read_path_count.load(), 0);
|
||||||
|
|
||||||
|
// Overlap existing (flushed) data so a non-zero global seqno is assigned and
|
||||||
|
// the newer values win -- the normal seqno-reassignment path through
|
||||||
|
// file_infos.
|
||||||
|
ASSERT_OK(db_->Put(WriteOptions(), Key(2), "old2"));
|
||||||
|
true_data[Key(2)] = "old2";
|
||||||
|
ASSERT_OK(Flush());
|
||||||
|
ASSERT_OK(ingest_fi({{Key(2), "new2"}, {Key(4), "new4"}}));
|
||||||
|
ASSERT_EQ(read_path_count.load(), 0);
|
||||||
|
|
||||||
|
// Control: ingesting the same way but WITHOUT file_infos does enter the scan.
|
||||||
|
ASSERT_OK(GenerateAndAddExternalFile(
|
||||||
|
options, {{Key(8), "c8"}}, /*file_id=*/-1, /*allow_global_seqno=*/true,
|
||||||
|
/*write_global_seqno=*/false, /*verify_checksums_before_ingest=*/true,
|
||||||
|
/*ingest_behind=*/false, /*sort_data=*/true, &true_data));
|
||||||
|
ASSERT_GT(read_path_count.load(), 0);
|
||||||
|
|
||||||
|
SyncPoint::GetInstance()->DisableProcessing();
|
||||||
|
SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||||
|
size_t kcnt = 0;
|
||||||
|
VerifyDBFromMap(true_data, &kcnt, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Range deletions are honored on the file_infos fast-path: the writer's
|
||||||
|
// range-del bounds are carried in the metadata, so a DeleteRange covering
|
||||||
|
// existing keys still shadows them.
|
||||||
|
TEST_F(ExternalSSTFileTest, IngestWithFileInfoRangeDeletion) {
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
DestroyAndReopen(options);
|
||||||
|
ASSERT_OK(db_->Put(WriteOptions(), Key(20), "live20"));
|
||||||
|
ASSERT_OK(db_->Put(WriteOptions(), Key(25), "live25"));
|
||||||
|
ASSERT_OK(Flush());
|
||||||
|
|
||||||
|
const std::string f = sst_files_dir_ + "rangedel.sst";
|
||||||
|
SstFileWriter w(EnvOptions(), options);
|
||||||
|
ASSERT_OK(w.Open(f));
|
||||||
|
ASSERT_OK(w.Put(Key(30), "v30"));
|
||||||
|
ASSERT_OK(w.DeleteRange(Key(20), Key(26))); // covers keys 20..25
|
||||||
|
std::shared_ptr<const PreparedFileInfo> info;
|
||||||
|
ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
|
||||||
|
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = db_->DefaultColumnFamily();
|
||||||
|
arg.external_files = {f};
|
||||||
|
arg.file_infos = {info.get()};
|
||||||
|
arg.options.allow_global_seqno = true;
|
||||||
|
ASSERT_OK(db_->IngestExternalFiles({arg}));
|
||||||
|
|
||||||
|
ReadOptions ro;
|
||||||
|
std::string val;
|
||||||
|
ASSERT_TRUE(db_->Get(ro, Key(20), &val).IsNotFound());
|
||||||
|
ASSERT_TRUE(db_->Get(ro, Key(25), &val).IsNotFound());
|
||||||
|
ASSERT_OK(db_->Get(ro, Key(30), &val));
|
||||||
|
ASSERT_EQ(val, "v30");
|
||||||
|
}
|
||||||
|
|
||||||
|
// verify_checksums_before_ingest is honored on the file_infos fast-path: it
|
||||||
|
// opens the file just to verify, so corruption is still detected.
|
||||||
|
TEST_F(ExternalSSTFileTest, IngestWithFileInfoVerifiesChecksum) {
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
DestroyAndReopen(options);
|
||||||
|
|
||||||
|
const std::string f = sst_files_dir_ + "corrupt.sst";
|
||||||
|
SstFileWriter w(EnvOptions(), options);
|
||||||
|
ASSERT_OK(w.Open(f));
|
||||||
|
for (int k = 0; k < 100; k++) {
|
||||||
|
ASSERT_OK(w.Put(Key(k), "v" + Key(k)));
|
||||||
|
}
|
||||||
|
std::shared_ptr<const PreparedFileInfo> info;
|
||||||
|
ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
|
||||||
|
ASSERT_OK(test::CorruptFile(options.env, f, /*offset=*/64,
|
||||||
|
/*bytes_to_corrupt=*/8));
|
||||||
|
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = db_->DefaultColumnFamily();
|
||||||
|
arg.external_files = {f};
|
||||||
|
arg.file_infos = {info.get()};
|
||||||
|
arg.options.verify_checksums_before_ingest = true;
|
||||||
|
ASSERT_TRUE(db_->IngestExternalFiles({arg}).IsCorruption());
|
||||||
|
}
|
||||||
|
|
||||||
|
// write_global_seqno is incompatible with file_infos: the file is not opened,
|
||||||
|
// so the seqno cannot be written back into it.
|
||||||
|
TEST_F(ExternalSSTFileTest, IngestWithFileInfoRejectsWriteGlobalSeqno) {
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
DestroyAndReopen(options);
|
||||||
|
|
||||||
|
const std::string f = sst_files_dir_ + "wgs.sst";
|
||||||
|
SstFileWriter w(EnvOptions(), options);
|
||||||
|
ASSERT_OK(w.Open(f));
|
||||||
|
ASSERT_OK(w.Put(Key(1), "v1"));
|
||||||
|
std::shared_ptr<const PreparedFileInfo> info;
|
||||||
|
ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
|
||||||
|
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = db_->DefaultColumnFamily();
|
||||||
|
arg.external_files = {f};
|
||||||
|
arg.file_infos = {info.get()};
|
||||||
|
arg.options.write_global_seqno = true;
|
||||||
|
ASSERT_TRUE(db_->IngestExternalFiles({arg}).IsInvalidArgument());
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(ExternalSSTFileTest, ComparatorMismatch) {
|
TEST_F(ExternalSSTFileTest, ComparatorMismatch) {
|
||||||
Options options = CurrentOptions();
|
Options options = CurrentOptions();
|
||||||
Options options_diff_ucmp = options;
|
Options options_diff_ucmp = options;
|
||||||
@@ -1858,7 +2005,8 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoRandomized) {
|
|||||||
for (int i = 0; i < 500; i++) {
|
for (int i = 0; i < 500; i++) {
|
||||||
std::vector<std::pair<std::string, std::string>> random_data;
|
std::vector<std::pair<std::string, std::string>> random_data;
|
||||||
for (int j = 0; j < 100; j++) {
|
for (int j = 0; j < 100; j++) {
|
||||||
std::string k = rnd.RandomString(rnd.Next() % 20);
|
std::string k =
|
||||||
|
rnd.RandomString(rnd.Next() % 20 + 1); // requires non-empty keys
|
||||||
std::string v = rnd.RandomString(rnd.Next() % 50);
|
std::string v = rnd.RandomString(rnd.Next() % 50);
|
||||||
random_data.emplace_back(k, v);
|
random_data.emplace_back(k, v);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -228,6 +228,7 @@ DECLARE_uint64(backup_max_size);
|
|||||||
DECLARE_int32(checkpoint_one_in);
|
DECLARE_int32(checkpoint_one_in);
|
||||||
DECLARE_int32(ingest_external_file_one_in);
|
DECLARE_int32(ingest_external_file_one_in);
|
||||||
DECLARE_int32(ingest_external_file_width);
|
DECLARE_int32(ingest_external_file_width);
|
||||||
|
DECLARE_int32(ingest_external_file_use_file_info_one_in);
|
||||||
DECLARE_int32(compact_files_one_in);
|
DECLARE_int32(compact_files_one_in);
|
||||||
DECLARE_int32(compact_range_one_in);
|
DECLARE_int32(compact_range_one_in);
|
||||||
DECLARE_int32(promote_l0_one_in);
|
DECLARE_int32(promote_l0_one_in);
|
||||||
|
|||||||
@@ -862,6 +862,12 @@ DEFINE_int32(ingest_external_file_one_in, 0,
|
|||||||
DEFINE_int32(ingest_external_file_width, 100,
|
DEFINE_int32(ingest_external_file_width, 100,
|
||||||
"The width of the ingested external files.");
|
"The width of the ingested external files.");
|
||||||
|
|
||||||
|
DEFINE_int32(ingest_external_file_use_file_info_one_in, 0,
|
||||||
|
"If non-zero, the ingestexternalfile flow reuses each file's "
|
||||||
|
"metadata via IngestExternalFileArg::file_infos (from "
|
||||||
|
"SstFileWriter::Finish) once every N ingestions on average, so "
|
||||||
|
"ingestion skips re-opening and scanning the files.");
|
||||||
|
|
||||||
DEFINE_int32(compact_files_one_in, 0,
|
DEFINE_int32(compact_files_one_in, 0,
|
||||||
"If non-zero, then CompactFiles() will be called once for every N "
|
"If non-zero, then CompactFiles() will be called once for every N "
|
||||||
"operations on average. 0 indicates CompactFiles() is disabled.");
|
"operations on average. 0 indicates CompactFiles() is disabled.");
|
||||||
|
|||||||
@@ -2326,6 +2326,12 @@ class NonBatchedOpsStressTest : public StressTest {
|
|||||||
// deletion file's compaction input optimization.
|
// deletion file's compaction input optimization.
|
||||||
bool test_standalone_range_deletion = thread->rand.OneInOpt(
|
bool test_standalone_range_deletion = thread->rand.OneInOpt(
|
||||||
FLAGS_test_ingest_standalone_range_deletion_one_in);
|
FLAGS_test_ingest_standalone_range_deletion_one_in);
|
||||||
|
// When true, reuse the writer's metadata via IngestExternalFileArg's
|
||||||
|
// file_infos so ingestion skips re-opening and scanning the file. Not
|
||||||
|
// combined with the standalone range deletion mode (a range-del-only file).
|
||||||
|
bool use_file_info =
|
||||||
|
!test_standalone_range_deletion &&
|
||||||
|
thread->rand.OneInOpt(FLAGS_ingest_external_file_use_file_info_one_in);
|
||||||
std::vector<std::string> external_files;
|
std::vector<std::string> external_files;
|
||||||
const std::string sst_filename =
|
const std::string sst_filename =
|
||||||
GetDbPath() + "/." + std::to_string(thread->tid) + ".sst";
|
GetDbPath() + "/." + std::to_string(thread->tid) + ".sst";
|
||||||
@@ -2369,6 +2375,7 @@ class NonBatchedOpsStressTest : public StressTest {
|
|||||||
SstFileWriter sst_file_writer(EnvOptions(options_), options_);
|
SstFileWriter sst_file_writer(EnvOptions(options_), options_);
|
||||||
SstFileWriter standalone_rangedel_sst_file_writer(EnvOptions(options_),
|
SstFileWriter standalone_rangedel_sst_file_writer(EnvOptions(options_),
|
||||||
options_);
|
options_);
|
||||||
|
std::shared_ptr<const PreparedFileInfo> sst_file_info;
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
s = sst_file_writer.Open(sst_filename);
|
s = sst_file_writer.Open(sst_filename);
|
||||||
}
|
}
|
||||||
@@ -2462,7 +2469,7 @@ class NonBatchedOpsStressTest : public StressTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (s.ok() && !keys.empty()) {
|
if (s.ok() && !keys.empty()) {
|
||||||
s = sst_file_writer.Finish();
|
s = sst_file_writer.Finish(/*file_info=*/nullptr, &sst_file_info);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (s.ok() && total_keys != 0 && test_standalone_range_deletion) {
|
if (s.ok() && total_keys != 0 && test_standalone_range_deletion) {
|
||||||
@@ -2494,9 +2501,19 @@ class NonBatchedOpsStressTest : public StressTest {
|
|||||||
<< ingest_options.verify_checksums_readahead_size
|
<< ingest_options.verify_checksums_readahead_size
|
||||||
<< ", fill_cache: " << ingest_options.fill_cache
|
<< ", fill_cache: " << ingest_options.fill_cache
|
||||||
<< ", test_standalone_range_deletion: "
|
<< ", test_standalone_range_deletion: "
|
||||||
<< test_standalone_range_deletion;
|
<< test_standalone_range_deletion
|
||||||
s = db_->IngestExternalFile(column_families_[column_family],
|
<< ", use_file_info: " << use_file_info;
|
||||||
external_files, ingest_options);
|
if (use_file_info && sst_file_info) {
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = column_families_[column_family];
|
||||||
|
arg.external_files = external_files;
|
||||||
|
arg.file_infos = {sst_file_info.get()};
|
||||||
|
arg.options = ingest_options;
|
||||||
|
s = db_->IngestExternalFiles({arg});
|
||||||
|
} else {
|
||||||
|
s = db_->IngestExternalFile(column_families_[column_family],
|
||||||
|
external_files, ingest_options);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
for (PendingExpectedValue& pending_expected_value :
|
for (PendingExpectedValue& pending_expected_value :
|
||||||
|
|||||||
@@ -2871,6 +2871,14 @@ struct IngestExternalFileOptions {
|
|||||||
bool fill_cache = true;
|
bool fill_cache = true;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// PreparedFileInfo is an opaque description of an SST file produced by
|
||||||
|
// SstFileWriter::Finish(). A borrowed pointer to one can be placed in
|
||||||
|
// IngestExternalFileArg::file_infos to let IngestExternalFiles() skip
|
||||||
|
// re-opening and scanning the file when computing its metadata, reducing
|
||||||
|
// ingestion I/O. Its contents are an internal implementation detail; the public
|
||||||
|
// API only ever refers to it through this forward declaration.
|
||||||
|
struct PreparedFileInfo;
|
||||||
|
|
||||||
// It is valid that files_checksums and files_checksum_func_names are both
|
// It is valid that files_checksums and files_checksum_func_names are both
|
||||||
// empty (no checksum information is provided for ingestion). Otherwise,
|
// empty (no checksum information is provided for ingestion). Otherwise,
|
||||||
// their sizes should be the same as external_files. The file order should
|
// their sizes should be the same as external_files. The file order should
|
||||||
@@ -2907,6 +2915,15 @@ struct IngestExternalFileArg {
|
|||||||
// exclusive, so it is best not to depend on one or the other until it is
|
// exclusive, so it is best not to depend on one or the other until it is
|
||||||
// sorted out.
|
// sorted out.
|
||||||
std::optional<RangeOpt> atomic_replace_range;
|
std::optional<RangeOpt> atomic_replace_range;
|
||||||
|
|
||||||
|
// Optional per-file opaque metadata, parallel to `external_files` (same size
|
||||||
|
// and order). Each entry is a borrowed pointer to a PreparedFileInfo produced
|
||||||
|
// by SstFileWriter::Finish; the owning handle must stay alive until ingestion
|
||||||
|
// completes. When provided, ingestion reuses this metadata instead of
|
||||||
|
// re-opening and scanning each file to recompute it, avoiding that extra I/O.
|
||||||
|
// Not compatible with options.write_global_seqno (the file is not opened, so
|
||||||
|
// a global seqno cannot be written back into it).
|
||||||
|
std::vector<const PreparedFileInfo*> file_infos;
|
||||||
};
|
};
|
||||||
|
|
||||||
enum TraceFilterType : uint64_t {
|
enum TraceFilterType : uint64_t {
|
||||||
|
|||||||
@@ -168,11 +168,20 @@ class SstFileWriter {
|
|||||||
Status DeleteRange(const Slice& begin_key, const Slice& end_key,
|
Status DeleteRange(const Slice& begin_key, const Slice& end_key,
|
||||||
const Slice& timestamp);
|
const Slice& timestamp);
|
||||||
|
|
||||||
// Finalize writing to sst file and close file.
|
// Finalize writing to sst file and close file. Calling Finish() with no
|
||||||
|
// argument simply finalizes the file.
|
||||||
//
|
//
|
||||||
// An optional ExternalSstFileInfo pointer can be passed to the function
|
// If `file_info` is non-null, it is populated with human-readable information
|
||||||
// which will be populated with information about the created sst file.
|
// about the created sst file.
|
||||||
Status Finish(ExternalSstFileInfo* file_info = nullptr);
|
//
|
||||||
|
// If `prepared_file_info` is non-null, it receives an opaque owning handle
|
||||||
|
// describing the file. A borrowed pointer to it can be placed in
|
||||||
|
// IngestExternalFileArg::file_infos so ingestion reuses this metadata instead
|
||||||
|
// of re-opening and scanning the file; the handle must be kept alive until
|
||||||
|
// ingestion completes.
|
||||||
|
Status Finish(
|
||||||
|
ExternalSstFileInfo* file_info = nullptr,
|
||||||
|
std::shared_ptr<const PreparedFileInfo>* prepared_file_info = nullptr);
|
||||||
|
|
||||||
// Return the current file size.
|
// Return the current file size.
|
||||||
uint64_t FileSize();
|
uint64_t FileSize();
|
||||||
|
|||||||
@@ -0,0 +1,33 @@
|
|||||||
|
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
||||||
|
// This source code is licensed under both the GPLv2 (found in the
|
||||||
|
// COPYING file in the root directory) and Apache 2.0 License
|
||||||
|
// (found in the LICENSE.Apache file in the root directory).
|
||||||
|
|
||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include <cstdint>
|
||||||
|
#include <string>
|
||||||
|
|
||||||
|
#include "rocksdb/table_properties.h"
|
||||||
|
|
||||||
|
namespace ROCKSDB_NAMESPACE {
|
||||||
|
|
||||||
|
// Definition of the type that is opaque to public API users (declared only as a
|
||||||
|
// forward declaration in rocksdb/options.h and rocksdb/sst_file_writer.h).
|
||||||
|
// Produced by SstFileWriter::Finish and consumed by ExternalSstFileIngestionJob
|
||||||
|
// so ingestion can skip re-opening and scanning the file to recompute its
|
||||||
|
// metadata.
|
||||||
|
struct PreparedFileInfo {
|
||||||
|
uint64_t file_size = 0;
|
||||||
|
// Encoded internal keys (user key + 8-byte footer), with the user-defined
|
||||||
|
// timestamp stripped when timestamps are not persisted.
|
||||||
|
std::string smallest_internal_key;
|
||||||
|
std::string largest_internal_key;
|
||||||
|
// Range-deletion boundary user keys (timestamp-stripped when applicable);
|
||||||
|
// empty when the file has no range deletions.
|
||||||
|
std::string smallest_range_del_key;
|
||||||
|
std::string largest_range_del_key;
|
||||||
|
TableProperties table_properties;
|
||||||
|
};
|
||||||
|
|
||||||
|
} // namespace ROCKSDB_NAMESPACE
|
||||||
@@ -15,6 +15,7 @@
|
|||||||
#include "rocksdb/file_system.h"
|
#include "rocksdb/file_system.h"
|
||||||
#include "rocksdb/table.h"
|
#include "rocksdb/table.h"
|
||||||
#include "table/block_based/block_based_table_builder.h"
|
#include "table/block_based/block_based_table_builder.h"
|
||||||
|
#include "table/prepared_file_info.h"
|
||||||
#include "table/sst_file_writer_collectors.h"
|
#include "table/sst_file_writer_collectors.h"
|
||||||
#include "test_util/sync_point.h"
|
#include "test_util/sync_point.h"
|
||||||
|
|
||||||
@@ -57,6 +58,11 @@ struct SstFileWriter::Rep {
|
|||||||
WriteOptions write_options;
|
WriteOptions write_options;
|
||||||
InternalKeyComparator internal_comparator;
|
InternalKeyComparator internal_comparator;
|
||||||
ExternalSstFileInfo file_info;
|
ExternalSstFileInfo file_info;
|
||||||
|
// Encoded internal keys of the smallest/largest point key, tracked here (not
|
||||||
|
// on the public ExternalSstFileInfo) so a PreparedFileInfo can be built.
|
||||||
|
// Empty when the file has no point keys.
|
||||||
|
std::string smallest_internal_key;
|
||||||
|
std::string largest_internal_key;
|
||||||
InternalKey ikey;
|
InternalKey ikey;
|
||||||
std::string column_family_name;
|
std::string column_family_name;
|
||||||
ColumnFamilyHandle* cfh;
|
ColumnFamilyHandle* cfh;
|
||||||
@@ -124,8 +130,15 @@ struct SstFileWriter::Rep {
|
|||||||
builder->Add(ikey.Encode(), value);
|
builder->Add(ikey.Encode(), value);
|
||||||
|
|
||||||
// update file info
|
// update file info
|
||||||
|
Slice encoded_ikey = ikey.Encode();
|
||||||
|
if (file_info.num_entries == 0) {
|
||||||
|
smallest_internal_key.assign(encoded_ikey.data(), encoded_ikey.size());
|
||||||
|
}
|
||||||
file_info.num_entries++;
|
file_info.num_entries++;
|
||||||
file_info.largest_key.assign(user_key.data(), user_key.size());
|
file_info.largest_key.assign(user_key.data(), user_key.size());
|
||||||
|
// Keys are added in strictly increasing order, so the last point key is the
|
||||||
|
// largest.
|
||||||
|
largest_internal_key.assign(encoded_ikey.data(), encoded_ikey.size());
|
||||||
file_info.file_size = builder->FileSize();
|
file_info.file_size = builder->FileSize();
|
||||||
|
|
||||||
InvalidatePageCache(false /* closing */).PermitUncheckedError();
|
InvalidatePageCache(false /* closing */).PermitUncheckedError();
|
||||||
@@ -436,6 +449,8 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) {
|
|||||||
r->file_info = ExternalSstFileInfo();
|
r->file_info = ExternalSstFileInfo();
|
||||||
r->file_info.file_path = file_path;
|
r->file_info.file_path = file_path;
|
||||||
r->file_info.version = 2;
|
r->file_info.version = 2;
|
||||||
|
r->smallest_internal_key.clear();
|
||||||
|
r->largest_internal_key.clear();
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -476,7 +491,9 @@ Status SstFileWriter::DeleteRange(const Slice& begin_key, const Slice& end_key,
|
|||||||
return rep_->DeleteRange(begin_key, end_key, timestamp);
|
return rep_->DeleteRange(begin_key, end_key, timestamp);
|
||||||
}
|
}
|
||||||
|
|
||||||
Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
|
Status SstFileWriter::Finish(
|
||||||
|
ExternalSstFileInfo* file_info,
|
||||||
|
std::shared_ptr<const PreparedFileInfo>* prepared_file_info) {
|
||||||
Rep* r = rep_.get();
|
Rep* r = rep_.get();
|
||||||
if (!r->builder) {
|
if (!r->builder) {
|
||||||
return Status::InvalidArgument("File is not opened");
|
return Status::InvalidArgument("File is not opened");
|
||||||
@@ -484,12 +501,15 @@ Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
|
|||||||
if (r->file_info.num_entries == 0 &&
|
if (r->file_info.num_entries == 0 &&
|
||||||
r->file_info.num_range_del_entries == 0) {
|
r->file_info.num_range_del_entries == 0) {
|
||||||
r->builder->status().PermitUncheckedError();
|
r->builder->status().PermitUncheckedError();
|
||||||
|
// Leave the builder intact; ~SstFileWriter will Abandon() it.
|
||||||
return Status::InvalidArgument("Cannot create sst file with no entries");
|
return Status::InvalidArgument("Cannot create sst file with no entries");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Finalize the table builder and underlying file. The builder is reset only
|
||||||
|
// at the end of this function, so its table properties stay readable for
|
||||||
|
// prepared_file_info below.
|
||||||
Status s = r->builder->Finish();
|
Status s = r->builder->Finish();
|
||||||
r->file_info.file_size = r->builder->FileSize();
|
r->file_info.file_size = r->builder->FileSize();
|
||||||
|
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
s = WritableFileWriter::PrepareIOOptions(r->write_options, opts);
|
s = WritableFileWriter::PrepareIOOptions(r->write_options, opts);
|
||||||
@@ -541,6 +561,44 @@ Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (s.ok() && prepared_file_info != nullptr) {
|
||||||
|
auto prepared = std::make_shared<PreparedFileInfo>();
|
||||||
|
prepared->file_size = r->file_info.file_size;
|
||||||
|
prepared->table_properties = r->builder->GetTableProperties();
|
||||||
|
prepared->smallest_internal_key = r->smallest_internal_key;
|
||||||
|
prepared->largest_internal_key = r->largest_internal_key;
|
||||||
|
prepared->smallest_range_del_key = r->file_info.smallest_range_del_key;
|
||||||
|
prepared->largest_range_del_key = r->file_info.largest_range_del_key;
|
||||||
|
// Strip user-defined timestamps when they should not be persisted, so the
|
||||||
|
// boundaries match the format the open-and-scan ingestion path produces.
|
||||||
|
if (r->strip_timestamp) {
|
||||||
|
if (!prepared->smallest_internal_key.empty()) {
|
||||||
|
assert(prepared->smallest_internal_key.size() >=
|
||||||
|
kNumInternalBytes + r->ts_sz);
|
||||||
|
assert(prepared->largest_internal_key.size() >=
|
||||||
|
kNumInternalBytes + r->ts_sz);
|
||||||
|
// The timestamp sits just before the 8-byte internal-key footer.
|
||||||
|
prepared->smallest_internal_key.erase(
|
||||||
|
prepared->smallest_internal_key.size() - kNumInternalBytes -
|
||||||
|
r->ts_sz,
|
||||||
|
r->ts_sz);
|
||||||
|
prepared->largest_internal_key.erase(
|
||||||
|
prepared->largest_internal_key.size() - kNumInternalBytes -
|
||||||
|
r->ts_sz,
|
||||||
|
r->ts_sz);
|
||||||
|
}
|
||||||
|
if (!prepared->smallest_range_del_key.empty()) {
|
||||||
|
assert(prepared->smallest_range_del_key.size() >= r->ts_sz);
|
||||||
|
assert(prepared->largest_range_del_key.size() >= r->ts_sz);
|
||||||
|
prepared->smallest_range_del_key.resize(
|
||||||
|
prepared->smallest_range_del_key.size() - r->ts_sz);
|
||||||
|
prepared->largest_range_del_key.resize(
|
||||||
|
prepared->largest_range_del_key.size() - r->ts_sz);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
*prepared_file_info = std::move(prepared);
|
||||||
|
}
|
||||||
|
|
||||||
r->builder.reset();
|
r->builder.reset();
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -62,6 +62,7 @@
|
|||||||
#include "rocksdb/secondary_cache.h"
|
#include "rocksdb/secondary_cache.h"
|
||||||
#include "rocksdb/slice.h"
|
#include "rocksdb/slice.h"
|
||||||
#include "rocksdb/slice_transform.h"
|
#include "rocksdb/slice_transform.h"
|
||||||
|
#include "rocksdb/sst_file_writer.h"
|
||||||
#include "rocksdb/stats_history.h"
|
#include "rocksdb/stats_history.h"
|
||||||
#include "rocksdb/table.h"
|
#include "rocksdb/table.h"
|
||||||
#include "rocksdb/tool_hooks.h"
|
#include "rocksdb/tool_hooks.h"
|
||||||
@@ -235,6 +236,8 @@ DEFINE_string(
|
|||||||
"\tcompact1 -- compact L1 into L2\n"
|
"\tcompact1 -- compact L1 into L2\n"
|
||||||
"\twaitforcompaction - pause until compaction is (probably) done\n"
|
"\twaitforcompaction - pause until compaction is (probably) done\n"
|
||||||
"\tflush - flush the memtable\n"
|
"\tflush - flush the memtable\n"
|
||||||
|
"\tingestexternalfile -- Create external SST files and ingest them into "
|
||||||
|
"the DB via IngestExternalFile\n"
|
||||||
"\topenandcompact -- Open DB and compact all files to bottommost level, "
|
"\topenandcompact -- Open DB and compact all files to bottommost level, "
|
||||||
"writing output to separate directory without modifying source DB. "
|
"writing output to separate directory without modifying source DB. "
|
||||||
"Designed for remote compaction service testing\n"
|
"Designed for remote compaction service testing\n"
|
||||||
@@ -1309,6 +1312,21 @@ DEFINE_string(backup_dir, "",
|
|||||||
DEFINE_string(restore_dir, "",
|
DEFINE_string(restore_dir, "",
|
||||||
"If not empty string, use the given dir for restore.");
|
"If not empty string, use the given dir for restore.");
|
||||||
|
|
||||||
|
DEFINE_int32(ingest_external_file_batch_size, 10,
|
||||||
|
"Number of SST files ingested per IngestExternalFile call (one "
|
||||||
|
"batch) in the ingestexternalfile benchmark.");
|
||||||
|
|
||||||
|
DEFINE_int32(ingest_external_file_num_batches, 1,
|
||||||
|
"Number of IngestExternalFile calls (batches) in the "
|
||||||
|
"ingestexternalfile benchmark. Total files ingested is "
|
||||||
|
"batch_size * num_batches; each file holds --num keys.");
|
||||||
|
|
||||||
|
DEFINE_bool(
|
||||||
|
ingest_external_file_use_file_info, false,
|
||||||
|
"If true, the ingestexternalfile benchmark passes each file's metadata "
|
||||||
|
"(from SstFileWriter::Finish) via IngestExternalFileArg::file_infos, so "
|
||||||
|
"ingestion reuses it instead of re-opening and scanning the files.");
|
||||||
|
|
||||||
DEFINE_uint64(
|
DEFINE_uint64(
|
||||||
initial_auto_readahead_size,
|
initial_auto_readahead_size,
|
||||||
ROCKSDB_NAMESPACE::BlockBasedTableOptions().initial_auto_readahead_size,
|
ROCKSDB_NAMESPACE::BlockBasedTableOptions().initial_auto_readahead_size,
|
||||||
@@ -3972,6 +3990,9 @@ class Benchmark {
|
|||||||
method = &Benchmark::WriteSeqSeekSeq;
|
method = &Benchmark::WriteSeqSeekSeq;
|
||||||
} else if (name == "compact") {
|
} else if (name == "compact") {
|
||||||
method = &Benchmark::Compact;
|
method = &Benchmark::Compact;
|
||||||
|
} else if (name == "ingestexternalfile") {
|
||||||
|
num_threads = 1;
|
||||||
|
method = &Benchmark::IngestExternalFile;
|
||||||
} else if (name == "compactall") {
|
} else if (name == "compactall") {
|
||||||
CompactAll();
|
CompactAll();
|
||||||
} else if (name == "compact0") {
|
} else if (name == "compact0") {
|
||||||
@@ -9288,6 +9309,120 @@ class Benchmark {
|
|||||||
db->CompactRange(cro, nullptr, nullptr);
|
db->CompactRange(cro, nullptr, nullptr);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Creates SST files with SstFileWriter and ingests them into the DB via
|
||||||
|
// IngestExternalFile, reporting throughput. Batching is used only to group
|
||||||
|
// multiple files into a single IngestExternalFile call:
|
||||||
|
// --ingest_external_file_batch_size files per call,
|
||||||
|
// --ingest_external_file_num_batches calls. Each file holds --num keys over
|
||||||
|
// the same key range, so the files overlap and are assigned new global
|
||||||
|
// sequence numbers. With --ingest_external_file_use_file_info the writer's
|
||||||
|
// metadata is passed via IngestExternalFileArg::file_infos so ingestion
|
||||||
|
// reuses it instead of re-opening and scanning the files. Single threaded.
|
||||||
|
//
|
||||||
|
// NOTE: db_bench's reported micros/op includes file generation; use the
|
||||||
|
// rocksdb.ingest.external.file.micros histogram (--statistics) for the
|
||||||
|
// isolated per-call ingest latency.
|
||||||
|
void IngestExternalFile(ThreadState* thread) {
|
||||||
|
DB* db = SelectDB(thread);
|
||||||
|
|
||||||
|
const int batch_size = std::max(1, FLAGS_ingest_external_file_batch_size);
|
||||||
|
const int num_batches = std::max(1, FLAGS_ingest_external_file_num_batches);
|
||||||
|
const bool use_file_info = FLAGS_ingest_external_file_use_file_info;
|
||||||
|
// --num is the number of keys in each file; all files cover this same key
|
||||||
|
// range, so they overlap (which is fine -- a global seqno is assigned).
|
||||||
|
const int64_t keys_per_file = num_;
|
||||||
|
const int64_t key_space = keys_per_file;
|
||||||
|
|
||||||
|
const std::string tmp_dir = FLAGS_db + "/ingest_tmp";
|
||||||
|
Status s = FLAGS_env->CreateDirIfMissing(tmp_dir);
|
||||||
|
if (!s.ok()) {
|
||||||
|
fprintf(stderr, "CreateDirIfMissing(%s) failed: %s\n", tmp_dir.c_str(),
|
||||||
|
s.ToString().c_str());
|
||||||
|
db_bench_exit(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
RandomGenerator gen;
|
||||||
|
std::unique_ptr<const char[]> key_guard;
|
||||||
|
Slice key = AllocateKey(&key_guard);
|
||||||
|
int64_t bytes = 0;
|
||||||
|
|
||||||
|
for (int b = 0; b < num_batches; ++b) {
|
||||||
|
// Write batch_size SstFileWriter files (all covering the same key range);
|
||||||
|
// they are batched into a single ingest call below.
|
||||||
|
std::vector<std::string> files;
|
||||||
|
std::vector<std::shared_ptr<const PreparedFileInfo>> prepared_infos;
|
||||||
|
for (int j = 0; j < batch_size; ++j) {
|
||||||
|
std::string file_path = tmp_dir + "/ingest_" + std::to_string(b) + "_" +
|
||||||
|
std::to_string(j) + ".sst";
|
||||||
|
SstFileWriter sst_file_writer(EnvOptions(), open_options_);
|
||||||
|
s = sst_file_writer.Open(file_path);
|
||||||
|
if (!s.ok()) {
|
||||||
|
fprintf(stderr, "SstFileWriter::Open(%s) failed: %s\n",
|
||||||
|
file_path.c_str(), s.ToString().c_str());
|
||||||
|
db_bench_exit(1);
|
||||||
|
}
|
||||||
|
for (int64_t i = 0; i < keys_per_file; ++i) {
|
||||||
|
GenerateKeyFromInt(i, key_space, &key);
|
||||||
|
Slice value = gen.Generate();
|
||||||
|
s = sst_file_writer.Put(key, value);
|
||||||
|
if (!s.ok()) {
|
||||||
|
sst_file_writer.Finish().PermitUncheckedError();
|
||||||
|
fprintf(stderr, "SstFileWriter::Put failed: %s\n",
|
||||||
|
s.ToString().c_str());
|
||||||
|
db_bench_exit(1);
|
||||||
|
}
|
||||||
|
bytes += key.size() + value.size();
|
||||||
|
}
|
||||||
|
std::shared_ptr<const PreparedFileInfo> info;
|
||||||
|
s = sst_file_writer.Finish(/*file_info=*/nullptr, &info);
|
||||||
|
if (!s.ok()) {
|
||||||
|
fprintf(stderr, "SstFileWriter::Finish(%s) failed: %s\n",
|
||||||
|
file_path.c_str(), s.ToString().c_str());
|
||||||
|
db_bench_exit(1);
|
||||||
|
}
|
||||||
|
files.emplace_back(file_path);
|
||||||
|
prepared_infos.emplace_back(std::move(info));
|
||||||
|
}
|
||||||
|
|
||||||
|
// Link the files into the DB instead of copying them, so the benchmark
|
||||||
|
// measures the ingest path itself (which the file_info fast-path speeds
|
||||||
|
// up by skipping the re-open and scan of each file) rather than file-copy
|
||||||
|
// throughput. allow_global_seqno (default) handles the overlap.
|
||||||
|
IngestExternalFileOptions ingest_options;
|
||||||
|
ingest_options.move_files = true;
|
||||||
|
if (use_file_info) {
|
||||||
|
// Reuse the writer's metadata so ingestion skips re-opening/scanning.
|
||||||
|
IngestExternalFileArg arg;
|
||||||
|
arg.column_family = db->DefaultColumnFamily();
|
||||||
|
arg.external_files = files;
|
||||||
|
for (const auto& prepared : prepared_infos) {
|
||||||
|
arg.file_infos.push_back(prepared.get());
|
||||||
|
}
|
||||||
|
arg.options = ingest_options;
|
||||||
|
s = db->IngestExternalFiles({arg});
|
||||||
|
} else {
|
||||||
|
s = db->IngestExternalFile(files, ingest_options);
|
||||||
|
}
|
||||||
|
if (!s.ok()) {
|
||||||
|
fprintf(stderr, "IngestExternalFile failed: %s\n",
|
||||||
|
s.ToString().c_str());
|
||||||
|
db_bench_exit(1);
|
||||||
|
}
|
||||||
|
thread->stats.FinishedOps(&db_, db, 1, kWrite);
|
||||||
|
|
||||||
|
for (const auto& f : files) {
|
||||||
|
FLAGS_env->DeleteFile(f).PermitUncheckedError();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
thread->stats.AddBytes(bytes);
|
||||||
|
char msg[100];
|
||||||
|
snprintf(msg, sizeof(msg),
|
||||||
|
"(%d batches x %d files, %" PRId64 " keys/file%s)", num_batches,
|
||||||
|
batch_size, keys_per_file, use_file_info ? ", file_info" : "");
|
||||||
|
thread->stats.AddMessage(msg);
|
||||||
|
}
|
||||||
|
|
||||||
void CompactAll() {
|
void CompactAll() {
|
||||||
CompactRangeOptions cro;
|
CompactRangeOptions cro;
|
||||||
cro.max_subcompactions = static_cast<uint32_t>(FLAGS_subcompactions);
|
cro.max_subcompactions = static_cast<uint32_t>(FLAGS_subcompactions);
|
||||||
|
|||||||
@@ -85,6 +85,21 @@ class DBBenchTest : public testing::Test {
|
|||||||
ASSERT_EQ(0, db_bench_tool(argc(), argv()));
|
ASSERT_EQ(0, db_bench_tool(argc(), argv()));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Every flag is passed explicitly because gflags state persists across
|
||||||
|
// db_bench_tool() calls within the same test process.
|
||||||
|
void RunIngestBench(int batch_size, int num_batches, bool use_file_info) {
|
||||||
|
ResetArgs();
|
||||||
|
AppendArgs(
|
||||||
|
{"./db_bench", "--benchmarks=ingestexternalfile", "--use_existing_db=0",
|
||||||
|
"--num=2000", "--compression_type=none", "--db=" + db_path_,
|
||||||
|
"--wal_dir=" + wal_path_,
|
||||||
|
"--ingest_external_file_batch_size=" + std::to_string(batch_size),
|
||||||
|
"--ingest_external_file_num_batches=" + std::to_string(num_batches),
|
||||||
|
"--ingest_external_file_use_file_info=" +
|
||||||
|
std::string(use_file_info ? "true" : "false")});
|
||||||
|
ASSERT_EQ(0, db_bench_tool(argc(), argv()));
|
||||||
|
}
|
||||||
|
|
||||||
void VerifyOptions(const Options& opt) {
|
void VerifyOptions(const Options& opt) {
|
||||||
DBOptions loaded_db_opts;
|
DBOptions loaded_db_opts;
|
||||||
ConfigOptions config_opts;
|
ConfigOptions config_opts;
|
||||||
@@ -144,6 +159,16 @@ TEST_F(DBBenchTest, OptionsFile) {
|
|||||||
VerifyOptions(opt);
|
VerifyOptions(opt);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST_F(DBBenchTest, IngestExternalFile) {
|
||||||
|
RunIngestBench(/*batch_size=*/3, /*num_batches=*/2,
|
||||||
|
/*use_file_info=*/false);
|
||||||
|
}
|
||||||
|
|
||||||
|
TEST_F(DBBenchTest, IngestExternalFileWithFileInfo) {
|
||||||
|
RunIngestBench(/*batch_size=*/3, /*num_batches=*/2,
|
||||||
|
/*use_file_info=*/true);
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(DBBenchTest, OptionsFileUniversal) {
|
TEST_F(DBBenchTest, OptionsFileUniversal) {
|
||||||
const std::string kOptionsFileName = test_path_ + "/OPTIONS_test";
|
const std::string kOptionsFileName = test_path_ + "/OPTIONS_test";
|
||||||
|
|
||||||
|
|||||||
@@ -215,6 +215,7 @@ default_params = {
|
|||||||
"index_block_search_type": lambda: random.choice([0, 1, 2]),
|
"index_block_search_type": lambda: random.choice([0, 1, 2]),
|
||||||
"uniform_cv_threshold": lambda: random.choice([-1, 0.2, 1000]),
|
"uniform_cv_threshold": lambda: random.choice([-1, 0.2, 1000]),
|
||||||
"ingest_external_file_one_in": lambda: random.choice([1000, 1000000]),
|
"ingest_external_file_one_in": lambda: random.choice([1000, 1000000]),
|
||||||
|
"ingest_external_file_use_file_info_one_in": lambda: random.choice([0, 2]),
|
||||||
"test_ingest_standalone_range_deletion_one_in": lambda: random.choice([0, 5, 10]),
|
"test_ingest_standalone_range_deletion_one_in": lambda: random.choice([0, 5, 10]),
|
||||||
"iterpercent": 10,
|
"iterpercent": 10,
|
||||||
"lock_wal_one_in": lambda: random.choice([10000, 1000000]),
|
"lock_wal_one_in": lambda: random.choice([10000, 1000000]),
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
`IngestExternalFileArg` gained a `file_infos` field: a vector of borrowed `const PreparedFileInfo*` pointers, parallel to `external_files`, each referring to an opaque handle produced via the new `std::shared_ptr<const PreparedFileInfo>*` output parameter of `SstFileWriter::Finish` (the handle must outlive the ingestion). When set, `IngestExternalFiles()` reuses the supplied per-file metadata instead of re-opening and scanning each file to recompute it, avoiding that extra I/O.
|
||||||
Reference in New Issue
Block a user