This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new fb69297540f [fix](filecache) pass tablet_id through FileReaderOptions
instead of parsing from path (#61683) (#66062)
fb69297540f is described below
commit fb69297540fdf3f4fe3697797576a65267deb1e0
Author: deardeng <[email protected]>
AuthorDate: Thu Jul 30 09:07:45 2026 +0800
[fix](filecache) pass tablet_id through FileReaderOptions instead of
parsing from path (#61683) (#66062)
pick from https://github.com/apache/doris/pull/61683
CachedRemoteFileReader::_execute_remote_read previously parsed tablet_id
from file paths at runtime via extract_tablet_id(). This breaks when
enable_packed_file (small file merging) is enabled because packed file
paths don't follow the expected data/{tablet_id}/... format.
Fix: store tablet_id from FileReaderOptions at construction time and use
it directly, eliminating runtime path parsing. Propagate tablet_id
through all code paths: Segment, InvertedIndexFileReader,
FSIndexInput::open, DownloadFileMeta (warmup/preheating), and
beta_rowset consistency checks.
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
None
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [x] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [x] Previous test can cover this change. - [ ] No code files have been
changed. - [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
(cherry picked from commit 58938de7bcf4246650c7b751b00f54af7abc5d9a)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/cloud/cloud_internal_service.cpp | 28 ++++++-----
be/src/cloud/cloud_tablet.cpp | 2 +
be/src/cloud/cloud_warm_up_manager.cpp | 16 ++++--
be/src/cloud/cloud_warm_up_manager.h | 3 +-
be/src/io/cache/block_file_cache_downloader.cpp | 2 +
be/src/io/cache/block_file_cache_downloader.h | 1 +
be/src/io/cache/cached_remote_file_reader.cpp | 58 +++++++++-------------
be/src/io/cache/cached_remote_file_reader.h | 3 +-
be/src/io/tools/file_cache_microbench.cpp | 1 +
.../storage/compaction/collection_statistics.cpp | 2 +-
be/src/storage/compaction/compaction.cpp | 4 +-
be/src/storage/index/index_file_reader.cpp | 10 ++--
be/src/storage/index/index_file_reader.h | 7 ++-
be/src/storage/index/index_file_writer.cpp | 10 ++--
be/src/storage/index/index_file_writer.h | 4 +-
.../index/inverted/inverted_index_fs_directory.cpp | 4 +-
.../index/inverted/inverted_index_fs_directory.h | 3 +-
be/src/storage/rowset/beta_rowset.cpp | 4 +-
be/src/storage/rowset/beta_rowset_writer.cpp | 3 +-
be/src/storage/rowset/rowset_writer.h | 2 +-
be/src/storage/segment/segment.cpp | 11 +++-
be/src/storage/segment/segment.h | 1 +
be/src/storage/task/index_builder.cpp | 10 ++--
be/test/io/cache/block_file_cache_test.cpp | 12 +++++
.../io/cache/block_file_cache_test_lru_dump.cpp | 1 +
be/test/io/fs/packed_file_concurrency_test.cpp | 2 +
26 files changed, 126 insertions(+), 78 deletions(-)
diff --git a/be/src/cloud/cloud_internal_service.cpp
b/be/src/cloud/cloud_internal_service.cpp
index 915f873cf45..28668f1b07e 100644
--- a/be/src/cloud/cloud_internal_service.cpp
+++ b/be/src/cloud/cloud_internal_service.cpp
@@ -723,12 +723,14 @@ void
CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* c
.expiration_time = expiration_time,
.is_dryrun =
config::enable_reader_dryrun_when_download_file_cache,
.is_warmup = true},
- .download_done = [=, version =
rs_meta.version()](Status st) {
- handle_segment_download_done(st, tablet_id,
rowset_id, segment_id,
- tablet, wait,
version, segment_size,
- request_ts,
handle_ts, job_id_str,
-
upstream_trigger_ts_ms);
- }};
+ .download_done =
+ [=, version = rs_meta.version()](Status st) {
+ handle_segment_download_done(
+ st, tablet_id, rowset_id,
segment_id, tablet, wait,
+ version, segment_size, request_ts,
handle_ts,
+ job_id_str,
upstream_trigger_ts_ms);
+ },
+ .tablet_id = tablet_id};
g_file_cache_event_driven_warm_up_submitted_segment_num << 1;
g_file_cache_event_driven_warm_up_submitted_segment_size <<
segment_size;
@@ -752,12 +754,14 @@ void
CloudInternalServiceImpl::warm_up_rowset(google::protobuf::RpcController* c
.expiration_time = expiration_time,
.is_dryrun =
config::enable_reader_dryrun_when_download_file_cache,
.is_warmup = true},
- .download_done = [=, version =
rs_meta.version()](Status st) {
- handle_inverted_index_download_done(
- st, tablet_id, rowset_id, segment_id,
index_path, tablet, wait,
- version, idx_size, request_ts, handle_ts,
job_id_str,
- upstream_trigger_ts_ms);
- }};
+ .download_done =
+ [=, version = rs_meta.version()](Status st) {
+ handle_inverted_index_download_done(
+ st, tablet_id, rowset_id,
segment_id, index_path,
+ tablet, wait, version, idx_size,
request_ts, handle_ts,
+ job_id_str,
upstream_trigger_ts_ms);
+ },
+ .tablet_id = tablet_id};
g_file_cache_event_driven_warm_up_submitted_index_num << 1;
g_file_cache_event_driven_warm_up_submitted_index_size <<
idx_size;
tablet->update_rowset_warmup_state_inverted_idx_num(
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index 364d9455777..271c81dd49c 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -1805,6 +1805,7 @@ void CloudTablet::_submit_segment_download_task(const
RowsetSharedPtr& rs,
LOG_WARNING("add rowset warm up error ").error(st);
}
}},
+ .tablet_id = _tablet_meta->tablet_id(),
});
// clang-format on
}
@@ -1847,6 +1848,7 @@ void
CloudTablet::_submit_inverted_index_download_task(const RowsetSharedPtr& rs
LOG_WARNING("add rowset warm up error ").error(st);
}
}},
+ .tablet_id = _tablet_meta->tablet_id(),
};
self->update_rowset_warmup_state_inverted_idx_num_unlocked(WarmUpTriggerSource::SYNC_ROWSET,
rowset_meta->rowset_id(), 1);
_engine.file_cache_block_downloader().submit_download_task(std::move(meta));
diff --git a/be/src/cloud/cloud_warm_up_manager.cpp
b/be/src/cloud/cloud_warm_up_manager.cpp
index f8107202e4d..9e7275fd260 100644
--- a/be/src/cloud/cloud_warm_up_manager.cpp
+++ b/be/src/cloud/cloud_warm_up_manager.cpp
@@ -157,7 +157,8 @@ void CloudWarmUpManager::submit_download_tasks(io::Path
path, int64_t file_size,
io::FileSystemSPtr file_system,
int64_t expiration_time,
std::shared_ptr<bthread::CountdownEvent> wait,
- bool is_index,
std::function<void(Status)> done_cb) {
+ bool is_index,
std::function<void(Status)> done_cb,
+ int64_t tablet_id) {
VLOG_DEBUG << "submit warm up task for file: " << path << ", file_size: "
<< file_size
<< ", expiration_time: " << expiration_time
<< ", is_index: " << (is_index ? "true" : "false");
@@ -212,6 +213,7 @@ void CloudWarmUpManager::submit_download_tasks(io::Path
path, int64_t file_size,
}
wait->signal();
},
+ .tablet_id = tablet_id,
});
offset += current_chunk_size;
@@ -284,7 +286,8 @@ void CloudWarmUpManager::handle_jobs() {
submit_download_tasks(
storage_resource.value()->remote_segment_path(*rs, seg_id),
rs->segment_file_size(cast_set<int>(seg_id)),
rs->fs(),
- expiration_time, wait, false, [tablet, rs,
seg_id](Status st) {
+ expiration_time, wait, false,
+ [tablet, rs, seg_id](Status st) {
VLOG_DEBUG << "warmup rowset " <<
rs->version() << " segment "
<< seg_id << " completed";
if (tablet->complete_rowset_segment_warmup(
@@ -294,7 +297,8 @@ void CloudWarmUpManager::handle_jobs() {
VLOG_DEBUG << "warmup rowset " <<
rs->version()
<< " completed";
}
- });
+ },
+ tablet_id);
}
// 2nd. download inverted index files
@@ -341,7 +345,8 @@ void CloudWarmUpManager::handle_jobs() {
VLOG_DEBUG << "warmup rowset " <<
rs->version()
<< " completed";
}
- });
+ },
+ tablet_id);
}
} else {
if (schema_ptr->has_inverted_index() ||
schema_ptr->has_ann_index()) {
@@ -364,7 +369,8 @@ void CloudWarmUpManager::handle_jobs() {
VLOG_DEBUG << "warmup rowset " <<
rs->version()
<< " completed";
}
- });
+ },
+ tablet_id);
}
}
}
diff --git a/be/src/cloud/cloud_warm_up_manager.h
b/be/src/cloud/cloud_warm_up_manager.h
index a8530376bf0..f4102915705 100644
--- a/be/src/cloud/cloud_warm_up_manager.h
+++ b/be/src/cloud/cloud_warm_up_manager.h
@@ -137,7 +137,8 @@ private:
void submit_download_tasks(io::Path path, int64_t file_size,
io::FileSystemSPtr file_system,
int64_t expiration_time,
std::shared_ptr<bthread::CountdownEvent> wait,
bool is_index = false,
- std::function<void(Status)> done_cb = nullptr);
+ std::function<void(Status)> done_cb = nullptr,
+ int64_t tablet_id = -1);
std::mutex _mtx;
std::condition_variable _cond;
int64_t _cur_job_id {0};
diff --git a/be/src/io/cache/block_file_cache_downloader.cpp
b/be/src/io/cache/block_file_cache_downloader.cpp
index 04835315b45..bee48c73add 100644
--- a/be/src/io/cache/block_file_cache_downloader.cpp
+++ b/be/src/io/cache/block_file_cache_downloader.cpp
@@ -270,6 +270,7 @@ void FileCacheBlockDownloader::download_file_cache_block(
.is_warmup = true,
},
.download_done = std::move(download_done),
+ .tablet_id = meta.tablet_id(),
};
download_segment_file(download_meta);
});
@@ -284,6 +285,7 @@ void FileCacheBlockDownloader::download_segment_file(const
DownloadFileMeta& met
.is_doris_table = true,
.cache_base_path {},
.file_size = meta.file_size,
+ .tablet_id = meta.tablet_id,
};
auto st = meta.file_system->open_file(meta.path, &file_reader, &opts);
if (!st.ok()) {
diff --git a/be/src/io/cache/block_file_cache_downloader.h
b/be/src/io/cache/block_file_cache_downloader.h
index c9a46891673..29d55a4c6cc 100644
--- a/be/src/io/cache/block_file_cache_downloader.h
+++ b/be/src/io/cache/block_file_cache_downloader.h
@@ -44,6 +44,7 @@ struct DownloadFileMeta {
io::FileSystemSPtr file_system;
IOContext ctx;
std::function<void(Status)> download_done;
+ int64_t tablet_id {-1};
};
struct DownloadTask {
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp
b/be/src/io/cache/cached_remote_file_reader.cpp
index ed8fc98e4f2..f03cbe17248 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -54,7 +54,6 @@
#include "runtime/thread_context.h"
#include "runtime/workload_management/io_throttle.h"
#include "service/backend_options.h"
-#include "storage/storage_policy.h"
#include "util/bit_util.h"
#include "util/brpc_client_cache.h" // BrpcClientCache
#include "util/client_cache.h"
@@ -92,7 +91,8 @@ bvar::Adder<uint64_t> g_failed_get_peer_addr_counter(
CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr
remote_file_reader,
const FileReaderOptions& opts)
- : _remote_file_reader(std::move(remote_file_reader)) {
+ : _tablet_id(opts.tablet_id),
_remote_file_reader(std::move(remote_file_reader)) {
+ DCHECK(!opts.is_doris_table || _tablet_id > 0);
_is_doris_table = opts.is_doris_table;
if (_is_doris_table) {
_cache_hash = BlockFileCache::hash(path().filename().native());
@@ -158,29 +158,30 @@ std::pair<size_t, size_t>
CachedRemoteFileReader::s_align_size(size_t offset, si
}
namespace {
-std::optional<int64_t> extract_tablet_id(const std::string& file_path) {
- return StorageResource::parse_tablet_id_from_path(file_path);
+// Execute S3 read
+Status execute_s3_read(size_t empty_start, size_t& size,
std::unique_ptr<char[]>& buffer,
+ ReadStatistics& stats, const IOContext* io_ctx,
+ FileReaderSPtr remote_file_reader) {
+ s3_read_counter << 1;
+ SCOPED_RAW_TIMER(&stats.remote_read_timer);
+ stats.from_peer_cache = false;
+ return remote_file_reader->read_at(empty_start, Slice(buffer.get(), size),
&size, io_ctx);
}
// Get peer connection info from tablet_id
-std::pair<std::string, int> get_peer_connection_info(const std::string&
file_path) {
+std::pair<std::string, int> get_peer_connection_info(int64_t tablet_id,
+ const std::string&
file_path) {
std::string host = "";
int port = 0;
- // Try to get tablet_id from actual path and lookup tablet info
- if (auto tablet_id = extract_tablet_id(file_path)) {
- auto& manager =
ExecEnv::GetInstance()->storage_engine().to_cloud().cloud_warm_up_manager();
- if (auto tablet_info = manager.get_balanced_tablet_info(*tablet_id)) {
- host = tablet_info->first;
- port = tablet_info->second;
- } else {
- LOG_EVERY_N(WARNING, 100)
- << "get peer connection info not found"
- << ", tablet_id=" << *tablet_id << ", file_path=" <<
file_path;
- }
+ DCHECK(tablet_id > 0);
+ auto& manager =
ExecEnv::GetInstance()->storage_engine().to_cloud().cloud_warm_up_manager();
+ if (auto tablet_info = manager.get_balanced_tablet_info(tablet_id)) {
+ host = tablet_info->first;
+ port = tablet_info->second;
} else {
- LOG_EVERY_N(WARNING, 100) << "parse tablet id from path failed"
- << "tablet_id=null, file_path=" << file_path;
+ LOG_EVERY_N(WARNING, 100) << "get peer connection info not found"
+ << ", tablet_id=" << tablet_id << ",
file_path=" << file_path;
}
DBUG_EXECUTE_IF("PeerFileCacheReader::_fetch_from_peer_cache_blocks", {
@@ -200,8 +201,8 @@ std::pair<std::string, int> get_peer_connection_info(const
std::string& file_pat
Status execute_peer_read(const std::vector<FileBlockSPtr>& empty_blocks,
size_t empty_start,
size_t& size, std::unique_ptr<char[]>& buffer,
const std::string& file_path, size_t file_size, bool
is_doris_table,
- ReadStatistics& stats, const IOContext* io_ctx) {
- auto [host, port] = get_peer_connection_info(file_path);
+ int64_t tablet_id, ReadStatistics& stats, const
IOContext* io_ctx) {
+ auto [host, port] = get_peer_connection_info(tablet_id, file_path);
VLOG_DEBUG << "PeerFileCacheReader read from peer, host=" << host << ",
port=" << port
<< ", file_path=" << file_path;
@@ -226,16 +227,6 @@ Status execute_peer_read(const std::vector<FileBlockSPtr>&
empty_blocks, size_t
return st;
}
-// Execute S3 read
-Status execute_s3_read(size_t empty_start, size_t& size,
std::unique_ptr<char[]>& buffer,
- ReadStatistics& stats, const IOContext* io_ctx,
- FileReaderSPtr remote_file_reader) {
- s3_read_counter << 1;
- SCOPED_RAW_TIMER(&stats.remote_read_timer);
- stats.from_peer_cache = false;
- return remote_file_reader->read_at(empty_start, Slice(buffer.get(), size),
&size, io_ctx);
-}
-
} // anonymous namespace
Status CachedRemoteFileReader::_execute_remote_read(const
std::vector<FileBlockSPtr>& empty_blocks,
@@ -257,7 +248,7 @@ Status CachedRemoteFileReader::_execute_remote_read(const
std::vector<FileBlockS
return execute_s3_read(empty_start, size, buffer, stats, io_ctx,
_remote_file_reader);
} else {
return execute_peer_read(empty_blocks, empty_start, size, buffer,
path().native(),
- this->size(), _is_doris_table, stats,
io_ctx);
+ this->size(), _is_doris_table,
_tablet_id, stats, io_ctx);
}
});
@@ -271,7 +262,7 @@ Status CachedRemoteFileReader::_execute_remote_read(const
std::vector<FileBlockS
// ATTN: Save original size before peer read, as it may be modified by
fetch_blocks, read peer ref size
size_t original_size = size;
auto st = execute_peer_read(empty_blocks, empty_start, size, buffer,
path().native(),
- this->size(), _is_doris_table, stats,
io_ctx);
+ this->size(), _is_doris_table, _tablet_id,
stats, io_ctx);
if (!st.ok()) {
// Restore original size for S3 fallback, as peer read may have
modified it
size = original_size;
@@ -410,8 +401,7 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset,
Slice result, size_t*
s_align_size(offset + already_read, bytes_req - already_read,
size());
CacheContext cache_context(io_ctx);
cache_context.stats = &stats;
- auto tablet_id = get_tablet_id(path().string());
- cache_context.tablet_id = tablet_id.value_or(0);
+ cache_context.tablet_id = _tablet_id;
MonotonicStopWatch sw;
sw.start();
diff --git a/be/src/io/cache/cached_remote_file_reader.h
b/be/src/io/cache/cached_remote_file_reader.h
index a978d14e28b..be9a7ac6726 100644
--- a/be/src/io/cache/cached_remote_file_reader.h
+++ b/be/src/io/cache/cached_remote_file_reader.h
@@ -91,7 +91,8 @@ private:
void _update_stats(const ReadStatistics& stats, FileCacheStatistics* state,
FileCacheReadType read_type) const;
- bool _is_doris_table;
+ bool _is_doris_table = false;
+ int64_t _tablet_id = -1;
FileReaderSPtr _remote_file_reader;
UInt128Wrapper _cache_hash;
BlockFileCache* _cache;
diff --git a/be/src/io/tools/file_cache_microbench.cpp
b/be/src/io/tools/file_cache_microbench.cpp
index 824c4276337..99550c15447 100644
--- a/be/src/io/tools/file_cache_microbench.cpp
+++ b/be/src/io/tools/file_cache_microbench.cpp
@@ -1761,6 +1761,7 @@ private:
doris::io::FileReaderOptions reader_opts;
reader_opts.cache_type =
doris::io::FileCachePolicy::FILE_BLOCK_CACHE;
reader_opts.is_doris_table = true;
+ reader_opts.tablet_id = 1; // microbench placeholder
doris::io::FileDescription fd;
std::string obj_path = "s3://" +
doris::config::test_s3_bucket + "/";
diff --git a/be/src/storage/compaction/collection_statistics.cpp
b/be/src/storage/compaction/collection_statistics.cpp
index 48d7b7134ea..6a680ad036c 100644
--- a/be/src/storage/compaction/collection_statistics.cpp
+++ b/be/src/storage/compaction/collection_statistics.cpp
@@ -165,7 +165,7 @@ Status CollectionStatistics::process_segment(const
RowsetSharedPtr& rowset, int3
rowset_meta->fs(),
std::string
{InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},
tablet_schema->get_inverted_index_storage_format(),
- rowset_meta->inverted_index_file_info(seg_id));
+ rowset_meta->inverted_index_file_info(seg_id),
rowset_meta->tablet_id());
RETURN_IF_ERROR(idx_file_reader->init(config::inverted_index_read_buffer_size,
io_ctx));
int32_t total_seg_num_docs = 0;
diff --git a/be/src/storage/compaction/compaction.cpp
b/be/src/storage/compaction/compaction.cpp
index edfa94e1d8e..37222b29b59 100644
--- a/be/src/storage/compaction/compaction.cpp
+++ b/be/src/storage/compaction/compaction.cpp
@@ -883,7 +883,7 @@ Status Compaction::do_inverted_index_compaction() {
fs,
std::string
{InvertedIndexDescriptor::get_index_file_path_prefix(seg_path.value())},
_cur_tablet_schema->get_inverted_index_storage_format(),
- rowset->rowset_meta()->inverted_index_file_info(seg_id));
+ rowset->rowset_meta()->inverted_index_file_info(seg_id),
_tablet->tablet_id());
auto st =
index_file_reader->init(config::inverted_index_read_buffer_size);
DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_init_inverted_index_file_reader",
{
@@ -1084,7 +1084,7 @@ static bool check_rowset_has_inverted_index(const
RowsetSharedPtr& src_rs, int32
std::string
{InvertedIndexDescriptor::get_index_file_path_prefix(
seg_path.value())},
cur_tablet_schema->get_inverted_index_storage_format(),
- rowset->rowset_meta()->inverted_index_file_info(i));
+ rowset->rowset_meta()->inverted_index_file_info(i),
tablet->tablet_id());
auto st =
index_file_reader->init(config::inverted_index_read_buffer_size);
index_file_path =
index_file_reader->get_index_file_path(index_meta);
DBUG_EXECUTE_IF(
diff --git a/be/src/storage/index/index_file_reader.cpp
b/be/src/storage/index/index_file_reader.cpp
index e4ed32fdc30..348e1399421 100644
--- a/be/src/storage/index/index_file_reader.cpp
+++ b/be/src/storage/index/index_file_reader.cpp
@@ -63,8 +63,9 @@ Status IndexFileReader::_init_from(int32_t read_buffer_size,
const io::IOContext
DCHECK(_fs != nullptr) << "file system is nullptr,
index_file_full_path: "
<< index_file_full_path;
// 2. open file
- auto ok = DorisFSDirectory::FSIndexInput::open(
- _fs, index_file_full_path.c_str(), index_input, err,
read_buffer_size, file_size);
+ auto ok =
+ DorisFSDirectory::FSIndexInput::open(_fs,
index_file_full_path.c_str(), index_input,
+ err, read_buffer_size,
file_size, _tablet_id);
if (!ok) {
if (err.number() == CL_ERR_FileNotFound) {
return Status::Error<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>(
@@ -182,8 +183,9 @@ Result<std::unique_ptr<DorisCompoundReader,
DirectoryDeleter>> IndexFileReader::
DCHECK(_fs != nullptr)
<< "file system is nullptr, index_file_path: " <<
index_file_path;
// 2. open file
- auto ok = DorisFSDirectory::FSIndexInput::open(
- _fs, index_file_path.c_str(), index_input, err,
_read_buffer_size, file_size);
+ auto ok = DorisFSDirectory::FSIndexInput::open(_fs,
index_file_path.c_str(),
+ index_input, err,
_read_buffer_size,
+ file_size,
_tablet_id);
if (!ok) {
// now index_input = nullptr
if (err.number() == CL_ERR_FileNotFound) {
diff --git a/be/src/storage/index/index_file_reader.h
b/be/src/storage/index/index_file_reader.h
index 4912b7ccabf..fb4ec2b9a62 100644
--- a/be/src/storage/index/index_file_reader.h
+++ b/be/src/storage/index/index_file_reader.h
@@ -55,11 +55,13 @@ public:
IndexFileReader(io::FileSystemSPtr fs, std::string index_path_prefix,
InvertedIndexStorageFormatPB storage_format,
- InvertedIndexFileInfo idx_file_info =
InvertedIndexFileInfo())
+ InvertedIndexFileInfo idx_file_info =
InvertedIndexFileInfo(),
+ int64_t tablet_id = -1)
: _fs(std::move(fs)),
_index_path_prefix(std::move(index_path_prefix)),
_storage_format(storage_format),
- _idx_file_info(idx_file_info) {}
+ _idx_file_info(idx_file_info),
+ _tablet_id(tablet_id) {}
virtual ~IndexFileReader() = default;
MOCK_FUNCTION Status init(int32_t read_buffer_size =
config::inverted_index_read_buffer_size,
@@ -93,6 +95,7 @@ private:
mutable std::shared_mutex _mutex; // Use mutable for const read operations
bool _inited = false;
InvertedIndexFileInfo _idx_file_info;
+ int64_t _tablet_id = -1;
};
} // namespace segment_v2
diff --git a/be/src/storage/index/index_file_writer.cpp
b/be/src/storage/index/index_file_writer.cpp
index 279d7bd9def..3c34326cd88 100644
--- a/be/src/storage/index/index_file_writer.cpp
+++ b/be/src/storage/index/index_file_writer.cpp
@@ -41,7 +41,8 @@ namespace doris::segment_v2 {
IndexFileWriter::IndexFileWriter(io::FileSystemSPtr fs, std::string
index_path_prefix,
std::string rowset_id, int64_t seg_id,
InvertedIndexStorageFormatPB storage_format,
- io::FileWriterPtr file_writer, bool
can_use_ram_dir)
+ io::FileWriterPtr file_writer, bool
can_use_ram_dir,
+ int64_t tablet_id)
: _fs(std::move(fs)),
_index_path_prefix(std::move(index_path_prefix)),
_rowset_id(std::move(rowset_id)),
@@ -49,7 +50,8 @@ IndexFileWriter::IndexFileWriter(io::FileSystemSPtr fs,
std::string index_path_p
_storage_format(storage_format),
_local_fs(io::global_local_filesystem()),
_idx_v2_writer(std::move(file_writer)),
- _can_use_ram_dir(can_use_ram_dir) {
+ _can_use_ram_dir(can_use_ram_dir),
+ _tablet_id(tablet_id) {
auto tmp_file_dir =
ExecEnv::GetInstance()->get_tmp_file_dirs()->get_tmp_file_dir();
_tmp_dir = tmp_file_dir.native();
if (_storage_format == InvertedIndexStorageFormatPB::V1) {
@@ -121,8 +123,8 @@ Status IndexFileWriter::delete_index(const TabletIndex*
index_meta) {
}
Status IndexFileWriter::add_into_searcher_cache() {
- auto index_file_reader =
- std::make_unique<IndexFileReader>(_fs, _index_path_prefix,
_storage_format);
+ auto index_file_reader = std::make_unique<IndexFileReader>(
+ _fs, _index_path_prefix, _storage_format, InvertedIndexFileInfo(),
_tablet_id);
auto st = index_file_reader->init();
if (!st.ok()) {
if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) !=
nullptr) {
diff --git a/be/src/storage/index/index_file_writer.h
b/be/src/storage/index/index_file_writer.h
index d968dc5fa12..a303de8b68c 100644
--- a/be/src/storage/index/index_file_writer.h
+++ b/be/src/storage/index/index_file_writer.h
@@ -50,7 +50,8 @@ class IndexFileWriter {
public:
IndexFileWriter(io::FileSystemSPtr fs, std::string index_path_prefix,
std::string rowset_id,
int64_t seg_id, InvertedIndexStorageFormatPB
storage_format,
- io::FileWriterPtr file_writer = nullptr, bool
can_use_ram_dir = true);
+ io::FileWriterPtr file_writer = nullptr, bool
can_use_ram_dir = true,
+ int64_t tablet_id = -1);
virtual ~IndexFileWriter() = default;
MOCK_FUNCTION Result<std::shared_ptr<DorisFSDirectory>> open(const
TabletIndex* index_meta);
@@ -111,6 +112,7 @@ private:
bool _can_use_ram_dir = true;
IndexStorageFormatPtr _index_storage_format;
+ int64_t _tablet_id = -1;
friend class IndexStorageFormatV1;
friend class IndexStorageFormatV2;
diff --git a/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
b/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
index 6b2c4a80b05..e65025b25a4 100644
--- a/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
+++ b/be/src/storage/index/inverted/inverted_index_fs_directory.cpp
@@ -87,7 +87,8 @@ const char* const DorisFSDirectory::WRITE_LOCK_FILE =
"write.lock";
bool DorisFSDirectory::FSIndexInput::open(const io::FileSystemSPtr& fs, const
char* path,
IndexInput*& ret, CLuceneError&
error,
- int32_t buffer_size, int64_t
file_size) {
+ int32_t buffer_size, int64_t
file_size,
+ int64_t tablet_id) {
CND_PRECONDITION(path != nullptr, "path is NULL");
if (buffer_size == -1) {
@@ -100,6 +101,7 @@ bool DorisFSDirectory::FSIndexInput::open(const
io::FileSystemSPtr& fs, const ch
:
io::FileCachePolicy::NO_CACHE;
reader_options.is_doris_table = true;
reader_options.file_size = file_size;
+ reader_options.tablet_id = tablet_id;
Status st = fs->open_file(path, &h->_reader, &reader_options);
DBUG_EXECUTE_IF("inverted file read error: index file not found",
{ st = Status::Error<doris::ErrorCode::NOT_FOUND>("index
file not found"); })
diff --git a/be/src/storage/index/inverted/inverted_index_fs_directory.h
b/be/src/storage/index/inverted/inverted_index_fs_directory.h
index 7b1ac0bdf55..79854df88d2 100644
--- a/be/src/storage/index/inverted/inverted_index_fs_directory.h
+++ b/be/src/storage/index/inverted/inverted_index_fs_directory.h
@@ -190,7 +190,8 @@ protected:
public:
static bool open(const io::FileSystemSPtr& fs, const char* path,
IndexInput*& ret,
- CLuceneError& error, int32_t bufferSize = -1, int64_t
file_size = -1);
+ CLuceneError& error, int32_t bufferSize = -1, int64_t
file_size = -1,
+ int64_t tablet_id = -1);
~FSIndexInput() override;
IndexInput* clone() const override;
diff --git a/be/src/storage/rowset/beta_rowset.cpp
b/be/src/storage/rowset/beta_rowset.cpp
index e94430731b1..96fb7a36889 100644
--- a/be/src/storage/rowset/beta_rowset.cpp
+++ b/be/src/storage/rowset/beta_rowset.cpp
@@ -633,6 +633,7 @@ Status BetaRowset::check_current_rowset_segment() {
.is_doris_table = true,
.cache_base_path {},
.file_size = _rowset_meta->segment_file_size(seg_id),
+ .tablet_id = _rowset_meta->tablet_id(),
};
auto s = segment_v2::Segment::open(fs, seg_path,
_rowset_meta->tablet_id(), seg_id,
@@ -843,7 +844,8 @@ Status BetaRowset::show_nested_index_file(rapidjson::Value*
rowset_value,
auto seg_path = DORIS_TRY(segment_path(seg_id));
auto index_file_path_prefix =
InvertedIndexDescriptor::get_index_file_path_prefix(seg_path);
auto index_file_reader = std::make_unique<IndexFileReader>(
- fs, std::string(index_file_path_prefix), storage_format);
+ fs, std::string(index_file_path_prefix), storage_format,
InvertedIndexFileInfo(),
+ _rowset_meta->tablet_id());
RETURN_IF_ERROR(index_file_reader->init());
auto dirs = index_file_reader->get_all_directories();
diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp
b/be/src/storage/rowset/beta_rowset_writer.cpp
index 708e7f9bfe8..a378c31cae7 100644
--- a/be/src/storage/rowset/beta_rowset_writer.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer.cpp
@@ -502,6 +502,7 @@ Status
BetaRowsetWriter::_load_noncompacted_segment(segment_v2::SegmentSharedPtr
: io::FileCachePolicy::NO_CACHE,
.is_doris_table = true,
.cache_base_path {},
+ .tablet_id = _rowset_meta->tablet_id(),
};
auto s = segment_v2::Segment::open(fs, path, _rowset_meta->tablet_id(),
segment_id, rowset_id(),
_context.tablet_schema, reader_options,
&segment);
@@ -1180,7 +1181,7 @@ Status
BetaRowsetWriter::create_segment_writer_for_segcompaction(
index_file_writer = std::make_unique<IndexFileWriter>(
_context.fs(), prefix, _context.rowset_id.to_string(),
_num_segcompacted,
_context.tablet_schema->get_inverted_index_storage_format(),
- std::move(idx_file_writer));
+ std::move(idx_file_writer), true /* can_use_ram_dir */,
_context.tablet_id);
index_file_writer->set_file_writer_opts(
_context.get_file_writer_options(FileType::INVERTED_INDEX_FILE));
}
diff --git a/be/src/storage/rowset/rowset_writer.h
b/be/src/storage/rowset/rowset_writer.h
index 693ebfe4169..75c0bff084b 100644
--- a/be/src/storage/rowset/rowset_writer.h
+++ b/be/src/storage/rowset/rowset_writer.h
@@ -114,7 +114,7 @@ public:
*index_file_writer = std::make_unique<IndexFileWriter>(
_context.fs(), segment_prefix, _context.rowset_id.to_string(),
segment_id,
_context.tablet_schema->get_inverted_index_storage_format(),
- std::move(idx_file_v2_ptr), can_use_ram_dir);
+ std::move(idx_file_v2_ptr), can_use_ram_dir,
_context.tablet_id);
return Status::OK();
}
diff --git a/be/src/storage/segment/segment.cpp
b/be/src/storage/segment/segment.cpp
index a2e08565e8c..53f06f96913 100644
--- a/be/src/storage/segment/segment.cpp
+++ b/be/src/storage/segment/segment.cpp
@@ -194,8 +194,15 @@ Status Segment::open(io::FileSystemSPtr fs, const
std::string& path, int64_t tab
uint32_t segment_id, RowsetId rowset_id, TabletSchemaSPtr
tablet_schema,
const io::FileReaderOptions& reader_options,
std::shared_ptr<Segment>* output,
InvertedIndexFileInfo idx_file_info,
OlapReaderStatistics* stats) {
- auto s = _open(fs, path, segment_id, rowset_id, tablet_schema,
reader_options, output,
+ // Ensure tablet_id is available in reader_options for
CachedRemoteFileReader peer read.
+ io::FileReaderOptions opts_with_tablet = reader_options;
+ opts_with_tablet.tablet_id = tablet_id;
+
+ auto s = _open(fs, path, segment_id, rowset_id, tablet_schema,
opts_with_tablet, output,
idx_file_info, stats);
+ if (s.ok() && output && *output) {
+ (*output)->_tablet_id = tablet_id;
+ }
if (!s.ok()) {
if (!config::is_cloud_mode()) {
auto res = ExecEnv::get_tablet(tablet_id);
@@ -343,7 +350,7 @@ Status Segment::_open_index_file_reader() {
// latter to an absolute path that won't match the relative keys in
PackedFileSystem's index map.
_index_file_reader = std::make_shared<IndexFileReader>(
_fs, std::string
{InvertedIndexDescriptor::get_index_file_path_prefix(_seg_path)},
- _tablet_schema->get_inverted_index_storage_format(),
_idx_file_info);
+ _tablet_schema->get_inverted_index_storage_format(),
_idx_file_info, _tablet_id);
return Status::OK();
}
diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h
index 66dcc891735..b0f2698e6b3 100644
--- a/be/src/storage/segment/segment.h
+++ b/be/src/storage/segment/segment.h
@@ -313,6 +313,7 @@ private:
DorisCallOnce<Status> _index_file_reader_open;
InvertedIndexFileInfo _idx_file_info;
+ int64_t _tablet_id = -1;
int _be_exec_version = BeExecVersionManager::get_newest_version();
};
diff --git a/be/src/storage/task/index_builder.cpp
b/be/src/storage/task/index_builder.cpp
index a5b937ccde2..172b868efc2 100644
--- a/be/src/storage/task/index_builder.cpp
+++ b/be/src/storage/task/index_builder.cpp
@@ -278,7 +278,8 @@ Status IndexBuilder::update_inverted_index_info() {
auto idx_file_reader = std::make_unique<IndexFileReader>(
context.fs(),
std::string
{InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},
-
output_rs_tablet_schema->get_inverted_index_storage_format());
+
output_rs_tablet_schema->get_inverted_index_storage_format(),
+ InvertedIndexFileInfo(), _tablet->tablet_id());
auto st = idx_file_reader->init();
DBUG_EXECUTE_IF(
"IndexBuilder::update_inverted_index_info_index_file_reader_init_not_ok", {
@@ -374,7 +375,7 @@ Status
IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
fs, std::move(index_path_prefix),
output_rowset_meta->rowset_id().to_string(),
seg_ptr->id(),
output_rowset_schema->get_inverted_index_storage_format(),
- std::move(file_writer));
+ std::move(file_writer), true /* can_use_ram_dir */,
_tablet->tablet_id());
RETURN_IF_ERROR(index_file_writer->initialize(dirs));
// create inverted index writer
for (auto& index_meta : _dropped_inverted_indexes) {
@@ -445,12 +446,13 @@ Status
IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
index_file_writer = std::make_unique<IndexFileWriter>(
fs, index_path_prefix,
output_rowset_meta->rowset_id().to_string(),
seg_ptr->id(),
output_rowset_schema->get_inverted_index_storage_format(),
- std::move(file_writer));
+ std::move(file_writer), true /* can_use_ram_dir */,
_tablet->tablet_id());
RETURN_IF_ERROR(index_file_writer->initialize(dirs));
} else {
index_file_writer = std::make_unique<IndexFileWriter>(
fs, index_path_prefix,
output_rowset_meta->rowset_id().to_string(),
- seg_ptr->id(),
output_rowset_schema->get_inverted_index_storage_format());
+ seg_ptr->id(),
output_rowset_schema->get_inverted_index_storage_format(),
+ nullptr, true /* can_use_ram_dir */,
_tablet->tablet_id());
}
// create inverted index writer, or ann index writer
for (auto inverted_index : _alter_inverted_indexes) {
diff --git a/be/test/io/cache/block_file_cache_test.cpp
b/be/test/io/cache/block_file_cache_test.cpp
index 61df125cf63..a8095389e53 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -3868,6 +3868,7 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
auto key = io::BlockFileCache::hash("tmp_file");
EXPECT_EQ(reader._cache_hash, key);
@@ -4035,6 +4036,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_tail) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
{
std::string buffer;
@@ -4108,6 +4110,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_error_handle) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
auto sp = SyncPoint::get_instance();
sp->enable_processing();
@@ -4429,6 +4432,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_no_self_heal_on_non_not_fou
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
std::string buffer(64_kb, '\0');
@@ -4631,6 +4635,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_concurrent) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
bool flag1 = false;
auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
auto sp = SyncPoint::get_instance();
@@ -4715,6 +4720,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_concurrent_2) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
auto sp = SyncPoint::get_instance();
sp->enable_processing();
@@ -5237,6 +5243,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_opt_lock) {
io::FileReaderOptions opts;
opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
{
FileReaderSPtr local_reader;
ASSERT_TRUE(global_local_filesystem()->open_file(tmp_file,
&local_reader).ok());
@@ -7688,6 +7695,7 @@ TEST_F(BlockFileCacheTest,
reader_dryrun_when_download_file_cache) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
auto key = io::BlockFileCache::hash("tmp_file");
EXPECT_EQ(reader._cache_hash, key);
@@ -8183,6 +8191,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_ttl_index) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
auto key = io::BlockFileCache::hash("tmp_file");
EXPECT_EQ(reader._cache_hash, key);
@@ -8264,6 +8273,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_normal_index) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
CachedRemoteFileReader reader(local_reader, opts);
auto key = io::BlockFileCache::hash("tmp_file");
EXPECT_EQ(reader._cache_hash, key);
@@ -8419,6 +8429,7 @@ TEST_F(BlockFileCacheTest,
DISABLE_cached_remote_file_reader_direct_read_and_evi
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
std::string buffer;
@@ -8514,6 +8525,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_direct_read_bytes_check) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
std::string buffer;
diff --git a/be/test/io/cache/block_file_cache_test_lru_dump.cpp
b/be/test/io/cache/block_file_cache_test_lru_dump.cpp
index b381fc61412..3a581d77c76 100644
--- a/be/test/io/cache/block_file_cache_test_lru_dump.cpp
+++ b/be/test/io/cache/block_file_cache_test_lru_dump.cpp
@@ -593,6 +593,7 @@ TEST_F(BlockFileCacheTest,
cached_remote_file_reader_direct_read_order_check) {
io::FileReaderOptions opts;
opts.cache_type = io::cache_type_from_string("file_block_cache");
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
std::string buffer;
diff --git a/be/test/io/fs/packed_file_concurrency_test.cpp
b/be/test/io/fs/packed_file_concurrency_test.cpp
index 1e581ff59b8..14c41d53a7b 100644
--- a/be/test/io/fs/packed_file_concurrency_test.cpp
+++ b/be/test/io/fs/packed_file_concurrency_test.cpp
@@ -456,6 +456,7 @@ protected:
FileReaderOptions local_opts = opts ? *opts : FileReaderOptions();
local_opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
local_opts.is_doris_table = true;
+ local_opts.tablet_id = 10086;
*reader = std::make_shared<CachedRemoteFileReader>(raw, local_opts);
return Status::OK();
}
@@ -722,6 +723,7 @@ TEST_F(MergeFileConcurrencyTest,
ConcurrentWriteReadCorrectness) {
FileReaderOptions opts;
opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
opts.is_doris_table = true;
+ opts.tablet_id = 10086;
ASSERT_TRUE(reader_fs.open_file(Path(path), &reader,
&opts).ok());
// After the fix, CachedRemoteFileReader wraps
PackedFileReader (not vice versa)
// This ensures cache key uses segment path for proper cleanup
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
