database/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00
Code Activity
Files
82dfd6a5100e29dc315b71bd38331a2902fd3b3e
rocksdb/table/external_table.cc
Xingbo Wang db1adf47bb Add bootstrap configuration for RocksDB external tables 
Summary:
Add an opaque external_table_config option to the external table adapter. RocksDB persists the string in OPTIONS files and passes it to ExternalTableFactory::Configure during bootstrap, allowing implementations such as Nimble to parse implementation-specific settings without adding RocksDB-specific fields. The option is immutable after DB open, and attempted runtime updates fail safely for non-clonable table factories. A reopen test verifies that callers can supply a different valid configuration between DB opens. Initialize the adapter option metadata lazily as const to avoid namespace-scope dynamic initialization and mutable-global lint issues.

PR https://github.com/facebook/rocksdb/pull/15119

Reviewed By: anand1976

Differential Revision: D116362444

fbshipit-source-id: 225a4aedc455f320b5c90ac2f0862917da477ccb
2026-08-18 10:12:47 -07:00

798 lines
27 KiB
C++
Raw Blame History

// Copyright (c) Meta Platforms, Inc. and affiliates.
// 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).
#include "rocksdb/external_table.h"
#include <chrono>
#include <cstddef>
#include <unordered_map>
#include "db/dbformat.h"
#include "logging/logging.h"
#include "rocksdb/listener.h"
#include "rocksdb/statistics.h"
#include "rocksdb/table.h"
#include "rocksdb/utilities/options_type.h"
#include "table/block_based/block.h"
#include "table/get_context.h"
#include "table/internal_iterator.h"
#include "table/meta_blocks.h"
#include "table/table_builder.h"
#include "table/table_reader.h"
#include "util/string_util.h"
namespace ROCKSDB_NAMESPACE {
namespace {
Histograms GetExternalTableFileReadHistogram(Statistics* stats,
Env::IOActivity io_activity) {
switch (io_activity) {
case Env::IOActivity::kFlush:
return Histograms::FILE_READ_FLUSH_MICROS;
case Env::IOActivity::kCompaction:
return Histograms::FILE_READ_COMPACTION_MICROS;
case Env::IOActivity::kDBOpen:
return Histograms::FILE_READ_DB_OPEN_MICROS;
default:
break;
}
if (stats && stats->get_stats_level() > StatsLevel::kExceptDetailedTimers) {
switch (io_activity) {
case Env::IOActivity::kGet:
return Histograms::FILE_READ_GET_MICROS;
case Env::IOActivity::kMultiGet:
return Histograms::FILE_READ_MULTIGET_MICROS;
case Env::IOActivity::kDBIterator:
return Histograms::FILE_READ_DB_ITERATOR_MICROS;
case Env::IOActivity::kVerifyDBChecksum:
return Histograms::FILE_READ_VERIFY_DB_CHECKSUM_MICROS;
case Env::IOActivity::kVerifyFileChecksums:
return Histograms::FILE_READ_VERIFY_FILE_CHECKSUMS_MICROS;
default:
break;
}
}
return Histograms::HISTOGRAM_ENUM_MAX;
}
class ExternalTableMetricsRandomAccessFile
: public FSRandomAccessFileOwnerWrapper {
public:
ExternalTableMetricsRandomAccessFile(
std::unique_ptr<FSRandomAccessFile>&& file, std::string file_name,
Statistics* statistics,
const std::vector<std::shared_ptr<EventListener>>& listeners,
Temperature file_temperature, bool is_last_level)
: FSRandomAccessFileOwnerWrapper(std::move(file)),
file_name_(std::move(file_name)),
statistics_(statistics),
listeners_(),
file_temperature_(file_temperature),
is_last_level_(is_last_level) {
for (const auto& listener : listeners) {
if (listener->ShouldBeNotifiedOnFileIO()) {
listeners_.emplace_back(listener);
}
}
}
IOStatus Read(uint64_t offset, size_t n, const IOOptions& options,
Slice* result, char* scratch,
IODebugContext* dbg) const override {
const auto start = std::chrono::steady_clock::now();
FileOperationInfo::StartTimePoint start_ts;
if (ShouldNotifyListeners()) {
start_ts = FileOperationInfo::StartNow();
}
auto status = target()->Read(offset, n, options, result, scratch, dbg);
const auto bytes_read = result == nullptr ? 0 : result->size();
if (ShouldNotifyListeners()) {
const auto finish_ts = FileOperationInfo::FinishNow();
NotifyOnFileReadFinish(offset, bytes_read, start_ts, finish_ts, status);
if (!status.ok()) {
NotifyOnIOError(status, FileOperationType::kRead, file_name_,
bytes_read, offset);
}
}
RecordRead(options, bytes_read, start);
return status;
}
IOStatus MultiRead(FSReadRequest* reqs, size_t num_reqs,
const IOOptions& options, IODebugContext* dbg) override {
const auto start = std::chrono::steady_clock::now();
FileOperationInfo::StartTimePoint start_ts;
if (ShouldNotifyListeners()) {
start_ts = FileOperationInfo::StartNow();
}
auto status = target()->MultiRead(reqs, num_reqs, options, dbg);
const auto elapsed = ElapsedMicros(start);
FileOperationInfo::FinishTimePoint finish_ts;
if (ShouldNotifyListeners()) {
finish_ts = FileOperationInfo::FinishNow();
}
RecordReadHistogram(options, elapsed);
for (size_t i = 0; i < num_reqs; ++i) {
const auto bytes_read = reqs[i].result.size();
if (ShouldNotifyListeners()) {
NotifyOnFileReadFinish(reqs[i].offset, bytes_read, start_ts, finish_ts,
reqs[i].status);
if (!reqs[i].status.ok()) {
NotifyOnIOError(reqs[i].status, FileOperationType::kRead, file_name_,
bytes_read, reqs[i].offset);
}
}
RecordReadBytes(bytes_read);
}
return status;
}
private:
static uint64_t ElapsedMicros(std::chrono::steady_clock::time_point start) {
return static_cast<uint64_t>(
std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::steady_clock::now() - start)
.count());
}
void RecordRead(const IOOptions& options, size_t bytes,
std::chrono::steady_clock::time_point start) const {
RecordReadHistogram(options, ElapsedMicros(start));
RecordReadBytes(bytes);
}
void RecordReadHistogram(const IOOptions& options, uint64_t elapsed) const {
if (statistics_ == nullptr) {
return;
}
statistics_->reportTimeToHistogram(Histograms::SST_READ_MICROS, elapsed);
const auto histogram =
GetExternalTableFileReadHistogram(statistics_, options.io_activity);
if (histogram != Histograms::HISTOGRAM_ENUM_MAX) {
statistics_->reportTimeToHistogram(histogram, elapsed);
}
}
void RecordReadBytes(size_t bytes) const {
if (statistics_ == nullptr) {
return;
}
statistics_->recordTick(is_last_level_ ? Tickers::LAST_LEVEL_READ_BYTES
: Tickers::NON_LAST_LEVEL_READ_BYTES,
bytes);
statistics_->recordTick(is_last_level_
? Tickers::LAST_LEVEL_READ_COUNT
: Tickers::NON_LAST_LEVEL_READ_COUNT);
switch (file_temperature_) {
case Temperature::kHot:
statistics_->recordTick(Tickers::HOT_FILE_READ_BYTES, bytes);
statistics_->recordTick(Tickers::HOT_FILE_READ_COUNT);
break;
case Temperature::kWarm:
statistics_->recordTick(Tickers::WARM_FILE_READ_BYTES, bytes);
statistics_->recordTick(Tickers::WARM_FILE_READ_COUNT);
break;
case Temperature::kCool:
statistics_->recordTick(Tickers::COOL_FILE_READ_BYTES, bytes);
statistics_->recordTick(Tickers::COOL_FILE_READ_COUNT);
break;
case Temperature::kCold:
statistics_->recordTick(Tickers::COLD_FILE_READ_BYTES, bytes);
statistics_->recordTick(Tickers::COLD_FILE_READ_COUNT);
break;
case Temperature::kIce:
statistics_->recordTick(Tickers::ICE_FILE_READ_BYTES, bytes);
statistics_->recordTick(Tickers::ICE_FILE_READ_COUNT);
break;
default:
break;
}
}
void NotifyOnFileReadFinish(
uint64_t offset, size_t length,
const FileOperationInfo::StartTimePoint& start_ts,
const FileOperationInfo::FinishTimePoint& finish_ts,
const IOStatus& status) const {
FileOperationInfo info(FileOperationType::kRead, file_name_, start_ts,
finish_ts, status, file_temperature_);
info.offset = offset;
info.length = length;
for (const auto& listener : listeners_) {
listener->OnFileReadFinish(info);
}
info.status.PermitUncheckedError();
}
void NotifyOnIOError(const IOStatus& io_status, FileOperationType operation,
const std::string& file_path, size_t length,
uint64_t offset) const {
IOErrorInfo info(io_status, operation, file_path, length, offset);
for (const auto& listener : listeners_) {
listener->OnIOError(info);
}
info.io_status.PermitUncheckedError();
}
bool ShouldNotifyListeners() const { return !listeners_.empty(); }
const std::string file_name_;
Statistics* statistics_;
std::vector<std::shared_ptr<EventListener>> listeners_;
const Temperature file_temperature_;
const bool is_last_level_;
};
class ExternalTableMetricsFileSystem : public FileSystemWrapper {
public:
ExternalTableMetricsFileSystem(
const std::shared_ptr<FileSystem>& target, Statistics* statistics,
const std::vector<std::shared_ptr<EventListener>>& listeners,
bool is_last_level)
: FileSystemWrapper(target),
statistics_(statistics),
listeners_(listeners),
is_last_level_(is_last_level) {}
const char* Name() const override { return target()->Name(); }
IOStatus NewRandomAccessFile(const std::string& file_name,
const FileOptions& file_options,
std::unique_ptr<FSRandomAccessFile>* result,
IODebugContext* dbg) override {
std::unique_ptr<FSRandomAccessFile> file;
auto status =
target()->NewRandomAccessFile(file_name, file_options, &file, dbg);
if (!status.ok()) {
return status;
}
result->reset(new ExternalTableMetricsRandomAccessFile(
std::move(file), file_name, statistics_, listeners_,
file_options.temperature, is_last_level_));
return status;
}
private:
Statistics* statistics_;
const std::vector<std::shared_ptr<EventListener>>& listeners_;
const bool is_last_level_;
};
class ExternalTableIteratorAdapter : public InternalIterator {
public:
explicit ExternalTableIteratorAdapter(ExternalTableIterator* iterator)
: iterator_(iterator), valid_(false) {}
// No copying allowed
ExternalTableIteratorAdapter(const ExternalTableIteratorAdapter&) = delete;
ExternalTableIteratorAdapter& operator=(const ExternalTableIteratorAdapter&) =
delete;
~ExternalTableIteratorAdapter() override {}
bool Valid() const override { return valid_; }
void SeekToFirst() override {
status_ = Status::OK();
if (iterator_) {
iterator_->SeekToFirst();
UpdateKey(OptSlice());
}
}
void SeekToLast() override {
status_ = Status::OK();
if (iterator_) {
iterator_->SeekToLast();
UpdateKey(OptSlice());
}
}
void Seek(const Slice& target) override {
status_ = Status::OK();
if (iterator_) {
ParsedInternalKey pkey;
status_ = ParseInternalKey(target, &pkey, /*log_err_key=*/false);
if (status_.ok()) {
iterator_->Seek(pkey.user_key);
UpdateKey(OptSlice());
}
}
}
void SeekForPrev(const Slice& target) override {
status_ = Status::OK();
if (iterator_) {
ParsedInternalKey pkey;
status_ = ParseInternalKey(target, &pkey, /*log_err_key=*/false);
if (status_.ok()) {
iterator_->SeekForPrev(pkey.user_key);
UpdateKey(OptSlice());
}
}
}
void Next() override {
if (iterator_) {
iterator_->Next();
UpdateKey(OptSlice());
}
}
bool NextAndGetResult(IterateResult* result) override {
if (iterator_) {
valid_ = iterator_->NextAndGetResult(&result_);
result->value_prepared = result_.value_prepared;
result->bound_check_result = result_.bound_check_result;
if (valid_) {
UpdateKey(result_.key);
result->key = key();
}
} else {
valid_ = false;
}
return valid_;
}
bool PrepareValue() override {
if (iterator_ && !result_.value_prepared) {
valid_ = iterator_->PrepareValue();
result_.value_prepared = true;
}
return valid_;
}
IterBoundCheck UpperBoundCheckResult() override {
if (iterator_) {
result_.bound_check_result = iterator_->UpperBoundCheckResult();
}
return result_.bound_check_result;
}
void Prev() override {
if (iterator_) {
iterator_->Prev();
UpdateKey(OptSlice());
}
}
Slice key() const override {
if (iterator_) {
return key_.GetInternalKey();
}
return Slice();
}
Slice value() const override {
if (iterator_) {
return iterator_->value();
}
return Slice();
}
Status status() const override { return status_; }
void Prepare(const MultiScanArgs* scan_opts) override {
if (iterator_ && scan_opts) {
iterator_->Prepare(scan_opts->GetScanRanges().data(), scan_opts->size());
} else if (iterator_) {
iterator_->Prepare(nullptr, 0);
}
}
private:
std::unique_ptr<ExternalTableIterator> iterator_;
IterKey key_;
bool valid_;
Status status_;
IterateResult result_;
void UpdateKey(OptSlice res) {
if (iterator_) {
valid_ = iterator_->Valid();
status_ = iterator_->status();
if (valid_ && status_.ok()) {
key_.SetInternalKey(res.has_value() ? res.value() : iterator_->key(),
/*s=*/0, ValueType::kTypeValue);
}
}
}
};
class ExternalTableReaderAdapter : public TableReader {
public:
explicit ExternalTableReaderAdapter(
const ImmutableOptions& ioptions,
std::unique_ptr<ExternalTableReader>&& reader)
: ioptions_(ioptions), reader_(std::move(reader)) {}
~ExternalTableReaderAdapter() override {}
// No copying allowed
ExternalTableReaderAdapter(const ExternalTableReaderAdapter&) = delete;
ExternalTableReaderAdapter& operator=(const ExternalTableReaderAdapter&) =
delete;
InternalIterator* NewIterator(
const ReadOptions& read_options, const SliceTransform* prefix_extractor,
Arena* arena, bool /* skip_filters */, TableReaderCaller /* caller */,
size_t /* compaction_readahead_size */ = 0,
bool /* allow_unprepared_value */ = false) override {
auto iterator = reader_->NewIterator(read_options, prefix_extractor);
if (arena == nullptr) {
return new ExternalTableIteratorAdapter(iterator);
} else {
auto* mem = arena->AllocateAligned(sizeof(ExternalTableIteratorAdapter));
return new (mem) ExternalTableIteratorAdapter(iterator);
}
}
uint64_t ApproximateOffsetOf(const ReadOptions&, const Slice&,
TableReaderCaller) override {
return 0;
}
uint64_t ApproximateSize(const ReadOptions&, const Slice&, const Slice&,
TableReaderCaller) override {
return 0;
}
void SetupForCompaction() override {}
std::shared_ptr<const TableProperties> GetTableProperties() const override {
std::shared_ptr<TableProperties> props;
std::unique_ptr<char[]> property_block;
uint64_t property_block_size = 0;
uint64_t property_block_offset = 0;
Status s;
// Get the raw properties block from the external table reader. We don't
// support writing the global sequence number, but we still get and return
// the correct global seqno offset in the file to prevent accidental
// corruption.
s = reader_->GetPropertiesBlock(&property_block, &property_block_size,
&property_block_offset);
if (s.ok()) {
std::unique_ptr<TableProperties> table_properties =
std::make_unique<TableProperties>();
BlockContents block_contents(std::move(property_block),
property_block_size);
Block block(std::move(block_contents));
s = ParsePropertiesBlock(ioptions_, property_block_offset, block,
table_properties);
if (s.ok()) {
props.reset(table_properties.release());
}
} else {
// Fallback to getting a minimal table properties structure from the
// external table reader
props = std::make_shared<TableProperties>(*reader_->GetTableProperties());
props->key_largest_seqno = 0;
props->key_smallest_seqno = 0;
}
return props;
}
size_t ApproximateMemoryUsage() const override { return 0; }
Status Get(const ReadOptions& read_options, const Slice& key,
GetContext* get_context, const SliceTransform* prefix_extractor,
bool /*skip_filters*/ = false) override {
ParsedInternalKey parsed_key;
Status s = ParseInternalKey(key, &parsed_key, /*log_err_key=*/false);
if (!s.ok()) {
return s;
}
PinnableSlice value;
s = reader_->Get(read_options, parsed_key.user_key, prefix_extractor,
&value);
if (!s.ok()) {
if (s.IsNotFound()) {
return Status::OK();
}
return s;
}
get_context->SaveValue(value, /*seq=*/0,
value.IsPinned() ? &value : nullptr);
return s;
}
Status VerifyChecksum(const ReadOptions& /*ro*/, TableReaderCaller /*caller*/,
bool /*meta_blocks_only*/ = false) override {
return Status::OK();
}
private:
const ImmutableOptions& ioptions_;
std::unique_ptr<ExternalTableReader> reader_;
};
class ExternalTableBuilderAdapter : public TableBuilder {
public:
explicit ExternalTableBuilderAdapter(
const TableBuilderOptions& topts,
std::unique_ptr<ExternalTableBuilder>&& builder,
std::unique_ptr<FSWritableFile>&& file)
: builder_(std::move(builder)),
file_(std::move(file)),
ioptions_(topts.ioptions) {
properties_.num_data_blocks = 1;
properties_.index_size = 0;
properties_.filter_size = 0;
properties_.format_version = 0;
properties_.key_largest_seqno = 0;
properties_.key_smallest_seqno = 0;
properties_.column_family_id = topts.column_family_id;
properties_.column_family_name = topts.column_family_name;
properties_.db_id = topts.db_id;
properties_.db_session_id = topts.db_session_id;
properties_.db_host_id = topts.ioptions.db_host_id;
if (!ReifyDbHostIdProperty(topts.ioptions.env, &properties_.db_host_id)
.ok()) {
ROCKS_LOG_INFO(topts.ioptions.logger,
"db_host_id property will not be set");
}
properties_.orig_file_number = topts.cur_file_num;
properties_.comparator_name = topts.ioptions.user_comparator != nullptr
? topts.ioptions.user_comparator->Name()
: "nullptr";
properties_.prefix_extractor_name =
topts.moptions.prefix_extractor != nullptr
? topts.moptions.prefix_extractor->AsString()
: "nullptr";
for (auto& factory : *topts.internal_tbl_prop_coll_factories) {
assert(factory);
std::unique_ptr<InternalTblPropColl> collector{
factory->CreateInternalTblPropColl(topts.column_family_id,
topts.level_at_creation,
topts.ioptions.num_levels)};
if (collector) {
table_properties_collectors_.emplace_back(std::move(collector));
}
}
}
void Add(const Slice& key, const Slice& value) override {
ParsedInternalKey pkey;
status_ = ParseInternalKey(key, &pkey, /*log_err_key=*/false);
if (status_.ok()) {
if (pkey.type != ValueType::kTypeValue) {
status_ = Status::NotSupported(
"Value type " + std::to_string(pkey.type) + "not supported");
} else {
builder_->Add(pkey.user_key, value);
properties_.num_entries++;
properties_.raw_key_size += key.size();
properties_.raw_value_size += value.size();
NotifyCollectTableCollectorsOnAdd(key, value, /*file_size=*/0,
table_properties_collectors_,
ioptions_.logger);
}
}
}
Status status() const override {
if (status_.ok()) {
return builder_->status();
} else {
return status_;
}
}
IOStatus io_status() const override { return status_to_io_status(status()); }
Status Finish() override {
// Approximate the data size
properties_.data_size =
properties_.raw_key_size + properties_.raw_value_size;
PropertyBlockBuilder property_block_builder;
property_block_builder.AddTableProperty(properties_);
UserCollectedProperties more_user_collected_properties;
NotifyCollectTableCollectorsOnFinish(
table_properties_collectors_, ioptions_.logger, &property_block_builder,
more_user_collected_properties, properties_.readable_properties);
properties_.user_collected_properties.insert(
more_user_collected_properties.begin(),
more_user_collected_properties.end());
Slice prop_block = property_block_builder.Finish();
Status s = builder_->PutPropertiesBlock(prop_block);
if (s.ok() || s.IsNotSupported()) {
// If the builder doesn't support writing the properties block,
// we still call Finish() and let the external builder handle it.
s = builder_->Finish();
}
return s;
}
void Abandon() override { builder_->Abandon(); }
uint64_t FileSize() const override { return builder_->FileSize(); }
uint64_t NumEntries() const override { return properties_.num_entries; }
TableProperties GetTableProperties() const override {
return builder_->GetTableProperties();
}
std::string GetFileChecksum() const override {
return builder_->GetFileChecksum();
}
const char* GetFileChecksumFuncName() const override {
return builder_->GetFileChecksumFuncName();
}
private:
Status status_;
std::unique_ptr<ExternalTableBuilder> builder_;
std::unique_ptr<FSWritableFile> file_;
const ImmutableOptions& ioptions_;
TableProperties properties_;
std::vector<std::unique_ptr<InternalTblPropColl>>
table_properties_collectors_;
};
struct ExternalTableFactoryAdapterOptions {
static const char* kName() { return "ExternalTableFactoryAdapterOptions"; }
std::string config;
};
static const std::unordered_map<std::string, OptionTypeInfo>&
GetExternalTableFactoryAdapterOptionsTypeInfo() {
static const std::unordered_map<std::string, OptionTypeInfo> type_info = {
{"external_table_config",
OptionTypeInfo(offsetof(ExternalTableFactoryAdapterOptions, config),
OptionType::kString)
.SetSerializeFunc([](const ConfigOptions&, const std::string&,
const void* addr, std::string* value) {
const std::string* config = static_cast<const std::string*>(addr);
*value = "{" + EscapeOptionString(*config) + "}";
return Status::OK();
})}};
return type_info;
}
class ExternalTableFactoryAdapter : public TableFactory {
public:
explicit ExternalTableFactoryAdapter(
std::shared_ptr<ExternalTableFactory> inner)
: inner_(std::move(inner)) {
RegisterOptions(&options_,
&GetExternalTableFactoryAdapterOptionsTypeInfo());
}
const char* Name() const override { return inner_->Name(); }
Status PrepareOptions(const ConfigOptions& config_options) override {
Status status = TableFactory::PrepareOptions(config_options);
if (status.ok()) {
status = inner_->Configure(options_.config);
}
return status;
}
using TableFactory::NewTableReader;
Status NewTableReader(
const ReadOptions& ro, const TableReaderOptions& topts,
std::unique_ptr<RandomAccessFileReader>&& file, uint64_t /* file_size */,
std::unique_ptr<TableReader>* table_reader,
bool /* prefetch_index_and_filter_in_cache */) const override {
// SstFileReader specifies largest_seqno as kMaxSequenceNumber to denote
// that its unknown
if (topts.largest_seqno > 0 && topts.largest_seqno != kMaxSequenceNumber) {
return Status::NotSupported(
"Ingesting file with sequence number larger than 0");
}
if (topts.ioptions.user_comparator != nullptr &&
topts.ioptions.user_comparator->timestamp_size() != 0) {
return Status::NotSupported(
"External table does not support user-defined timestamps");
}
std::unique_ptr<ExternalTableReader> reader;
FileOptions fopts(topts.env_options);
fopts.io_options.io_activity = ro.io_activity;
std::shared_ptr<FileSystem> fs = topts.ioptions.fs;
if (topts.ioptions.stats != nullptr || !topts.ioptions.listeners.empty()) {
fs = std::make_shared<ExternalTableMetricsFileSystem>(
topts.ioptions.fs, topts.ioptions.stats, topts.ioptions.listeners,
topts.level >= 0 && topts.level == topts.ioptions.num_levels - 1);
}
ExternalTableOptions ext_topts(topts.prefix_extractor,
topts.ioptions.user_comparator, fs, fopts);
auto status =
inner_->NewTableReader(ro, file->file_name(), ext_topts, &reader);
if (!status.ok()) {
return status;
}
table_reader->reset(
new ExternalTableReaderAdapter(topts.ioptions, std::move(reader)));
file.reset();
return Status::OK();
}
using TableFactory::NewTableBuilder;
TableBuilder* NewTableBuilder(const TableBuilderOptions& topts,
WritableFileWriter* file) const override {
std::unique_ptr<ExternalTableBuilder> builder;
ExternalTableBuilderOptions ext_topts(
topts.read_options, topts.write_options,
topts.moptions.prefix_extractor, topts.ioptions.user_comparator,
topts.column_family_name, topts.reason, topts.ioptions.fs);
auto file_wrapper =
std::make_unique<ExternalTableWritableFileWrapper>(file);
builder.reset(inner_->NewTableBuilder(ext_topts, file->file_name(),
file_wrapper.get()));
if (builder) {
return new ExternalTableBuilderAdapter(topts, std::move(builder),
std::move(file_wrapper));
}
return nullptr;
}
std::unique_ptr<TableFactory> Clone() const override { return nullptr; }
private:
// An FSWritableFile subclass for wrapping a WritableFileWriter. The
// latter is private to RocksDB, so we wrap it here in order to pass it
// to the ExternalTableBuilder. This is necessary for WritableFileWriter
// to intercept Append so that it can calculate the file checksum.
class ExternalTableWritableFileWrapper : public FSWritableFile {
public:
explicit ExternalTableWritableFileWrapper(WritableFileWriter* writer)
: writer_(writer) {}
using FSWritableFile::Append;
IOStatus Append(const Slice& data, const IOOptions& options,
IODebugContext* /*dbg*/) override {
return writer_->Append(options, data);
}
IOStatus Close(const IOOptions& options, IODebugContext* /*dbg*/) override {
return writer_->Close(options);
}
IOStatus Flush(const IOOptions& options, IODebugContext* /*dbg*/) override {
return writer_->Flush(options);
}
IOStatus Sync(const IOOptions& options, IODebugContext* /*dbg*/) override {
return writer_->Sync(options, /*use_fsync=*/false);
}
uint64_t GetFileSize(const IOOptions& options,
IODebugContext* dbg) override {
return writer_->writable_file()->GetFileSize(options, dbg);
}
private:
WritableFileWriter* writer_;
};
std::shared_ptr<ExternalTableFactory> inner_;
ExternalTableFactoryAdapterOptions options_;
};
} // namespace
std::unique_ptr<TableFactory> NewExternalTableFactory(
std::shared_ptr<ExternalTableFactory> inner_factory) {
std::unique_ptr<TableFactory> res;
res = std::make_unique<ExternalTableFactoryAdapter>(std::move(inner_factory));
return res;
}
} // namespace ROCKSDB_NAMESPACE
View Git Blame Copy Permalink