Skip to content

branch-4.1: [feature](cloud) Support file cache write index only #65268

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bobhan1 wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
Open

branch-4.1: [feature](cloud) Support file cache write index only #65268

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions be/src/cloud/config.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,7 @@ DEFINE_mBool(enable_warmup_immediately_on_new_rowset, "false");

// Packed file manager config
DEFINE_mBool(enable_packed_file, "true");
DEFINE_mBool(enable_file_cache_write_index_file_only, "false");
DEFINE_mInt64(packed_file_size_threshold_bytes, "5242880"); // 5MB
DEFINE_mInt64(packed_file_time_threshold_ms, "100"); // 100ms
DEFINE_mInt64(packed_file_try_lock_timeout_ms, "5"); // 5ms
Expand Down
1 change: 1 addition & 0 deletions be/src/cloud/config.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,7 @@ DECLARE_mBool(enable_warmup_immediately_on_new_rowset);

// Packed file manager config
DECLARE_mBool(enable_packed_file);
DECLARE_mBool(enable_file_cache_write_index_file_only);
DECLARE_mInt64(packed_file_size_threshold_bytes);
DECLARE_mInt64(packed_file_time_threshold_ms);
DECLARE_mInt64(packed_file_try_lock_timeout_ms);
Expand Down
48 changes: 48 additions & 0 deletions be/src/io/cache/block_file_cache_profile.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,16 @@ FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& curren
SUBTRACT_FIELD(inverted_index_remote_io_timer);
SUBTRACT_FIELD(inverted_index_peer_io_timer);
SUBTRACT_FIELD(inverted_index_io_timer);

SUBTRACT_FIELD(segment_footer_index_num_local_io_total);
SUBTRACT_FIELD(segment_footer_index_num_remote_io_total);
SUBTRACT_FIELD(segment_footer_index_num_peer_io_total);
SUBTRACT_FIELD(segment_footer_index_bytes_read_from_local);
SUBTRACT_FIELD(segment_footer_index_bytes_read_from_remote);
SUBTRACT_FIELD(segment_footer_index_bytes_read_from_peer);
SUBTRACT_FIELD(segment_footer_index_local_io_timer);
SUBTRACT_FIELD(segment_footer_index_remote_io_timer);
SUBTRACT_FIELD(segment_footer_index_peer_io_timer);
#undef SUBTRACT_FIELD
return diff;
}
Expand Down Expand Up @@ -156,6 +166,25 @@ FileCacheProfileReporter::FileCacheProfileReporter(RuntimeProfile* profile) {
ADD_CHILD_TIMER_WITH_LEVEL(profile, "InvertedIndexPeerIOUseTimer", cache_profile, 1);
inverted_index_io_timer =
ADD_CHILD_TIMER_WITH_LEVEL(profile, "InvertedIndexIOTimer", cache_profile, 1);

segment_footer_index_num_local_io_total = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexNumLocalIOTotal", TUnit::UNIT, cache_profile, 1);
segment_footer_index_num_remote_io_total = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexNumRemoteIOTotal", TUnit::UNIT, cache_profile, 1);
segment_footer_index_num_peer_io_total = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexNumPeerIOTotal", TUnit::UNIT, cache_profile, 1);
segment_footer_index_bytes_scanned_from_cache = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexBytesScannedFromCache", TUnit::BYTES, cache_profile, 1);
segment_footer_index_bytes_scanned_from_remote = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexBytesScannedFromRemote", TUnit::BYTES, cache_profile, 1);
segment_footer_index_bytes_scanned_from_peer = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "SegmentFooterIndexBytesScannedFromPeer", TUnit::BYTES, cache_profile, 1);
segment_footer_index_local_io_timer = ADD_CHILD_TIMER_WITH_LEVEL(
profile, "SegmentFooterIndexLocalIOUseTimer", cache_profile, 1);
segment_footer_index_remote_io_timer = ADD_CHILD_TIMER_WITH_LEVEL(
profile, "SegmentFooterIndexRemoteIOUseTimer", cache_profile, 1);
segment_footer_index_peer_io_timer = ADD_CHILD_TIMER_WITH_LEVEL(
profile, "SegmentFooterIndexPeerIOUseTimer", cache_profile, 1);
}

void FileCacheProfileReporter::update(const FileCacheStatistics* statistics) const {
Expand Down Expand Up @@ -193,6 +222,25 @@ void FileCacheProfileReporter::update(const FileCacheStatistics* statistics) con
COUNTER_UPDATE(inverted_index_remote_io_timer, statistics->inverted_index_remote_io_timer);
COUNTER_UPDATE(inverted_index_peer_io_timer, statistics->inverted_index_peer_io_timer);
COUNTER_UPDATE(inverted_index_io_timer, statistics->inverted_index_io_timer);

COUNTER_UPDATE(segment_footer_index_num_local_io_total,
statistics->segment_footer_index_num_local_io_total);
COUNTER_UPDATE(segment_footer_index_num_remote_io_total,
statistics->segment_footer_index_num_remote_io_total);
COUNTER_UPDATE(segment_footer_index_num_peer_io_total,
statistics->segment_footer_index_num_peer_io_total);
COUNTER_UPDATE(segment_footer_index_bytes_scanned_from_cache,
statistics->segment_footer_index_bytes_read_from_local);
COUNTER_UPDATE(segment_footer_index_bytes_scanned_from_remote,
statistics->segment_footer_index_bytes_read_from_remote);
COUNTER_UPDATE(segment_footer_index_bytes_scanned_from_peer,
statistics->segment_footer_index_bytes_read_from_peer);
COUNTER_UPDATE(segment_footer_index_local_io_timer,
statistics->segment_footer_index_local_io_timer);
COUNTER_UPDATE(segment_footer_index_remote_io_timer,
statistics->segment_footer_index_remote_io_timer);
COUNTER_UPDATE(segment_footer_index_peer_io_timer,
statistics->segment_footer_index_peer_io_timer);
}

} // namespace doris::io
10 changes: 10 additions & 0 deletions be/src/io/cache/block_file_cache_profile.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,16 @@ struct FileCacheProfileReporter {
RuntimeProfile::Counter* inverted_index_peer_io_timer = nullptr;
RuntimeProfile::Counter* inverted_index_io_timer = nullptr;

RuntimeProfile::Counter* segment_footer_index_num_local_io_total = nullptr;
RuntimeProfile::Counter* segment_footer_index_num_remote_io_total = nullptr;
RuntimeProfile::Counter* segment_footer_index_num_peer_io_total = nullptr;
RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_cache = nullptr;
RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_remote = nullptr;
RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_peer = nullptr;
RuntimeProfile::Counter* segment_footer_index_local_io_timer = nullptr;
RuntimeProfile::Counter* segment_footer_index_remote_io_timer = nullptr;
RuntimeProfile::Counter* segment_footer_index_peer_io_timer = nullptr;

FileCacheProfileReporter(RuntimeProfile* profile);
void update(const FileCacheStatistics* statistics) const;
};
Expand Down
66 changes: 53 additions & 13 deletions be/src/io/cache/cached_remote_file_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,13 +332,23 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
}
if (!io_ctx->is_warmup) {
// update stats increment in this reading procedure for file cache metrics
const auto file_cache_read_type =
io_ctx->is_inverted_index
? FileCacheReadType::INVERTED_INDEX
: (io_ctx->is_index_data ? FileCacheReadType::SEGMENT_FOOTER_INDEX
: FileCacheReadType::DATA);
FileCacheStatistics fcache_stats_increment;
_update_stats(stats, &fcache_stats_increment, io_ctx->is_inverted_index);
_update_stats(stats, &fcache_stats_increment, file_cache_read_type);
io::FileCacheMetrics::instance().update(&fcache_stats_increment);
}
if (io_ctx->file_cache_stats) {
// update stats in io_ctx, for query profile
_update_stats(stats, io_ctx->file_cache_stats, io_ctx->is_inverted_index);
const auto file_cache_read_type =
io_ctx->is_inverted_index
? FileCacheReadType::INVERTED_INDEX
: (io_ctx->is_index_data ? FileCacheReadType::SEGMENT_FOOTER_INDEX
: FileCacheReadType::DATA);
_update_stats(stats, io_ctx->file_cache_stats, file_cache_read_type);
}
};
std::unique_ptr<int, decltype(defer_func)> defer((int*)0x01, std::move(defer_func));
Expand Down Expand Up @@ -621,7 +631,7 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*

void CachedRemoteFileReader::_update_stats(const ReadStatistics& read_stats,
FileCacheStatistics* statis,
bool is_inverted_index) const {
FileCacheReadType read_type) const {
if (statis == nullptr) {
return;
}
Expand Down Expand Up @@ -651,22 +661,52 @@ void CachedRemoteFileReader::_update_stats(const ReadStatistics& read_stats,
statis->get_timer += read_stats.get_timer;
statis->set_timer += read_stats.set_timer;

if (is_inverted_index) {
auto update_index_stats = [&](int64_t& num_local_io_total, int64_t& num_remote_io_total,
int64_t& num_peer_io_total, int64_t& bytes_read_from_local,
int64_t& bytes_read_from_remote, int64_t& bytes_read_from_peer,
int64_t& local_io_timer, int64_t& remote_io_timer,
int64_t& peer_io_timer) {
if (read_stats.bytes_read_from_local > 0) {
statis->inverted_index_num_local_io_total++;
statis->inverted_index_bytes_read_from_local += read_stats.bytes_read_from_local;
num_local_io_total++;
bytes_read_from_local += read_stats.bytes_read_from_local;
}
if (read_stats.bytes_read_from_remote > 0) {
statis->inverted_index_num_remote_io_total++;
statis->inverted_index_bytes_read_from_remote += read_stats.bytes_read_from_remote;
statis->inverted_index_remote_io_timer += read_stats.remote_read_timer;
num_remote_io_total++;
bytes_read_from_remote += read_stats.bytes_read_from_remote;
remote_io_timer += read_stats.remote_read_timer;
}
if (read_stats.bytes_read_from_peer > 0) {
statis->inverted_index_num_peer_io_total++;
statis->inverted_index_bytes_read_from_peer += read_stats.bytes_read_from_peer;
statis->inverted_index_peer_io_timer += read_stats.peer_read_timer;
num_peer_io_total++;
bytes_read_from_peer += read_stats.bytes_read_from_peer;
peer_io_timer += read_stats.peer_read_timer;
}
statis->inverted_index_local_io_timer += read_stats.local_read_timer;
local_io_timer += read_stats.local_read_timer;
};

switch (read_type) {
case FileCacheReadType::DATA:
break;
case FileCacheReadType::INVERTED_INDEX:
update_index_stats(
statis->inverted_index_num_local_io_total,
statis->inverted_index_num_remote_io_total,
statis->inverted_index_num_peer_io_total,
statis->inverted_index_bytes_read_from_local,
statis->inverted_index_bytes_read_from_remote,
statis->inverted_index_bytes_read_from_peer, statis->inverted_index_local_io_timer,
statis->inverted_index_remote_io_timer, statis->inverted_index_peer_io_timer);
break;
case FileCacheReadType::SEGMENT_FOOTER_INDEX:
update_index_stats(statis->segment_footer_index_num_local_io_total,
statis->segment_footer_index_num_remote_io_total,
statis->segment_footer_index_num_peer_io_total,
statis->segment_footer_index_bytes_read_from_local,
statis->segment_footer_index_bytes_read_from_remote,
statis->segment_footer_index_bytes_read_from_peer,
statis->segment_footer_index_local_io_timer,
statis->segment_footer_index_remote_io_timer,
statis->segment_footer_index_peer_io_timer);
break;
}

g_skip_cache_sum << read_stats.skip_cache;
Expand Down
8 changes: 7 additions & 1 deletion be/src/io/cache/cached_remote_file_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,12 @@ class CachedRemoteFileReader final : public FileReader,
const IOContext* io_ctx) override;

private:
enum class FileCacheReadType {
DATA,
INVERTED_INDEX,
SEGMENT_FOOTER_INDEX,
};

void _insert_file_reader(FileBlockSPtr file_block);

// Execute remote read (S3 or peer).
Expand All @@ -83,7 +89,7 @@ class CachedRemoteFileReader final : public FileReader,
ReadStatistics& stats, const IOContext* io_ctx);

void _update_stats(const ReadStatistics& stats, FileCacheStatistics* state,
bool is_inverted_index) const;
FileCacheReadType read_type) const;

bool _is_doris_table;
FileReaderSPtr _remote_file_reader;
Expand Down
1 change: 1 addition & 0 deletions be/src/io/fs/file_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#include <stddef.h>

#include <memory>
#include <string>

#include "common/status.h"
#include "io/fs/path.h"
Expand Down
5 changes: 4 additions & 1 deletion be/src/io/fs/file_writer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ struct FileWriterOptions {
// this shortens the inconsistent time window.
bool used_by_s3_committer = false;
bool write_file_cache = false;
bool allow_adaptive_file_cache_write = true;
bool is_cold_data = false;
bool sync_file_data = true; // Whether flush data into storage system
uint64_t file_cache_expiration_time = 0; // Relative time
Expand Down Expand Up @@ -109,13 +110,15 @@ class FileWriter {
io::UInt128Wrapper path_hash = BlockFileCache::hash(path.filename().native());
BlockFileCache* file_cache_ptr = FileCacheFactory::instance()->get_by_path(path_hash);

bool has_enough_file_cache_space = config::enable_file_cache_adaptive_write &&
bool has_enough_file_cache_space = opts->allow_adaptive_file_cache_write &&
config::enable_file_cache_adaptive_write &&
(opts->approximate_bytes_to_write > 0) &&
(file_cache_ptr->approximate_available_cache_size() >
opts->approximate_bytes_to_write);

VLOG_DEBUG << "path:" << path.filename().native()
<< ", write_file_cache:" << opts->write_file_cache
<< ", allow_adaptive_file_cache_write:" << opts->allow_adaptive_file_cache_write
<< ", has_enough_file_cache_space:" << has_enough_file_cache_space
<< ", approximate_bytes_to_write:" << opts->approximate_bytes_to_write
<< ", file_cache_available_size:"
Expand Down
22 changes: 18 additions & 4 deletions be/src/io/fs/packed_file_system.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
#include <string_view>
#include <utility>

#include "cloud/config.h"
#include "common/status.h"
#include "io/fs/file_reader.h"
#include "io/fs/packed_file_reader.h"
Expand Down Expand Up @@ -96,8 +97,13 @@ Status PackedFileSystem::create_file_impl(const Path& file, FileWriterPtr* write
return Status::OK();
}

auto append_info = _append_info;
if (config::enable_file_cache_write_index_file_only && opts != nullptr) {
append_info.write_file_cache = opts->write_file_cache;
}

// Wrap with PackedFileWriter
*writer = std::make_unique<PackedFileWriter>(std::move(inner_writer), file, _append_info);
*writer = std::make_unique<PackedFileWriter>(std::move(inner_writer), file, append_info);
return Status::OK();
}

Expand All @@ -106,11 +112,19 @@ Status PackedFileSystem::open_file_impl(const Path& file, FileReaderSPtr* reader
// Check if this file is in a packed file
std::string file_path = file.native();
auto it = _index_map.find(file_path);
bool is_packed_file = (it != _index_map.end());
PackedSliceLocation manager_location;
const PackedSliceLocation* packed_location = nullptr;
if (it != _index_map.end()) {
packed_location = &it->second;
} else if (PackedFileManager::instance()
->get_packed_slice_location(file_path, &manager_location)
.ok()) {
packed_location = &manager_location;
}

if (is_packed_file) {
if (packed_location != nullptr) {
// File is in packed file, open packed file and wrap with PackedFileReader
const auto& index = it->second;
const auto& index = *packed_location;
FileReaderSPtr inner_reader;

// Create options for opening the packed file
Expand Down
2 changes: 2 additions & 0 deletions be/src/io/fs/s3_file_system.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
#include "common/config.h"
#include "common/logging.h"
#include "common/status.h"
#include "cpp/sync_point.h"
#include "io/fs/err_utils.h"
#include "io/fs/file_system.h"
#include "io/fs/file_writer.h"
Expand Down Expand Up @@ -195,6 +196,7 @@ Status S3FileSystem::create_file_impl(const Path& file, FileWriterPtr* writer,

Status S3FileSystem::open_file_internal(const Path& file, FileReaderSPtr* reader,
const FileReaderOptions& opts) {
TEST_SYNC_POINT_CALLBACK("S3FileSystem::open_file_internal", &file, &opts);
auto key = DORIS_TRY(get_key(file));
*reader = DORIS_TRY(S3FileReader::create(_client, _bucket, key, opts.file_size, nullptr));
return Status::OK();
Expand Down
10 changes: 10 additions & 0 deletions be/src/io/io_common.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,16 @@ struct FileCacheStatistics {
int64_t inverted_index_remote_io_timer = 0;
int64_t inverted_index_peer_io_timer = 0;
int64_t inverted_index_io_timer = 0;

int64_t segment_footer_index_num_local_io_total = 0;
int64_t segment_footer_index_num_remote_io_total = 0;
int64_t segment_footer_index_num_peer_io_total = 0;
int64_t segment_footer_index_bytes_read_from_local = 0;
int64_t segment_footer_index_bytes_read_from_remote = 0;
int64_t segment_footer_index_bytes_read_from_peer = 0;
int64_t segment_footer_index_local_io_timer = 0;
int64_t segment_footer_index_remote_io_timer = 0;
int64_t segment_footer_index_peer_io_timer = 0;
};

struct IOContext {
Expand Down
5 changes: 5 additions & 0 deletions be/src/storage/compaction/compaction.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/config.h"
#include "cloud/pb_convert.h"
#include "common/config.h"
#include "common/metrics/doris_metrics.h"
Expand Down Expand Up @@ -1919,6 +1920,10 @@ int64_t CloudCompactionMixin::num_input_rowsets() const {
}

bool CloudCompactionMixin::should_cache_compaction_output() {
if (config::enable_file_cache_write_index_file_only) {
return false;
}

if (compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION) {
return true;
}
Expand Down
Loading
Loading