From 223fafe417869331bfc41f3c404322aba3cb5943 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Wed, 1 Jul 2026 22:31:30 +0800 Subject: [PATCH] [feature](cloud) Support file cache write index only (#64995) This PR adds a new BE config, `enable_file_cache_write_index_file_only`, to make cloud rowset writes and compaction outputs write only index-related content into file cache. This avoids actively putting segment data into file cache during write, reducing cache pollution from large data pages. The main target scenario is a limited local cache where query performance should prioritize indexes and segment metadata. With this mode enabled, the write path protects inverted index files and segment index/footer ranges, while normal data pages are still loaded and cached by the query read path when needed. - Add BE config `enable_file_cache_write_index_file_only`, default `false`. - Make rowset/segment file-cache write decisions file-type aware: - `SEGMENT_FILE` does not create a full-file cache builder and does not allow adaptive write in index-only mode. - `INVERTED_INDEX_FILE` is still written into file cache through the direct write path. - segment index/footer ranges are preloaded into file cache by `SegmentIndexFileCacheLoader` after segment close. - Update packed file handling to honor the original file type when deciding whether to write file cache. - Add profile statistics for segment footer/index reads and separate them from data-page and inverted-index reads. - Replace correlated boolean parameters in `CachedRemoteFileReader::_update_stats` with a read-type enum: - data page - inverted index - segment footer/index - Add BE UTs and cloud regressions for both normal file and packed file paths. When `enable_file_cache_write_index_file_only=true`: - Segment data is not actively written into file cache during write, flush, or compaction output. - Segment index/footer ranges are synchronously preloaded into file cache after segment close. - Independent inverted index files are written into file cache through the direct write path. - Query-side data-page reads keep the existing behavior. The query read path may still fill cache according to current rules. - Segment open, footer/index access, and inverted index access can hit the index-related content preloaded by the write path. - Query profile exposes separate cache/remote read statistics for data pages, inverted indexes, and segment footer/index reads. The write-side file-cache behavior is decided in this order: 1. `enable_file_cache=false` is the highest-priority global switch. - All file-cache writes are disabled. - This includes independent inverted index direct cache and segment index/footer preload. 2. `enable_file_cache=true` and `enable_file_cache_write_index_file_only=true` enables global forced index-only mode. - Segment data does not create a full-file cache builder. - Segment data cannot be written by adaptive write. - Segment index/footer ranges are written into file cache by `SegmentIndexFileCacheLoader` after segment close. - Independent inverted index files are written through the `INVERTED_INDEX_FILE` direct write path. - This config has higher priority than `enable_file_cache_write_base_compaction_index_only` and `enable_file_cache_write_cumu_compaction_index_only`. - These index writes do not depend on `enable_file_cache_adaptive_write`, compaction keep configs, hit-ratio thresholds, or request/session `write_file_cache`. 3. `enable_file_cache=true`, `enable_file_cache_write_index_file_only=false`, and a compaction-level index-only config is enabled applies index-only behavior only to the matching compaction output. - `enable_file_cache_write_base_compaction_index_only=true` only forces base compaction output to use index-only file-cache writes. - `enable_file_cache_write_cumu_compaction_index_only=true` only forces cumulative compaction output to use index-only file-cache writes. - These two configs do not affect load, normal flush, schema change output, or query read-path cache fill. - Compaction types not matched by these configs continue to use existing `write_file_cache`, adaptive write, compaction keep, and hit-ratio decisions. 4. `enable_file_cache=true` and all three index-only configs are `false` preserves existing behavior. - `write_file_cache`, `enable_file_cache_adaptive_write`, compaction keep configs, schema-change hit ratios, and existing rowset/file-writer decisions continue to control whether cache is written. - When index-only is disabled, `enable_file_cache_adaptive_write=true` keeps the existing adaptive write behavior. Segment data may be written into file cache if the cache-space check passes. - When index-only is enabled, adaptive write no longer participates in segment data decisions. - Even if `enable_file_cache_adaptive_write=true`, segment data is still not written into file cache, while segment index/footer and independent inverted index files are still written. Related configs include: - `enable_file_cache_keep_base_compaction_output` - `file_cache_keep_base_compaction_output_min_hit_ratio` - compaction output hit-ratio checks When index-only is disabled, existing compaction output cache behavior is preserved. When index-only is enabled, compaction output is forced into index-only behavior: - Compaction output segment data is not actively written into file cache. - Segment index/footer ranges are synchronously written into file cache. - Independent inverted index files are direct-cached. - Compaction keep configs and hit-ratio thresholds no longer decide whether index content is written into file cache. Related configs include: - `enable_file_cache_write_base_compaction_index_only` - `enable_file_cache_write_cumu_compaction_index_only` These two configs are narrower compaction-output policies and only take effect when `enable_file_cache_write_index_file_only=false`. - When `enable_file_cache_write_base_compaction_index_only=true`, base compaction output uses index-only behavior: output segment data is not actively cached, while output segment index/footer ranges and independent inverted index files are written into file cache. - When `enable_file_cache_write_cumu_compaction_index_only=true`, cumulative compaction output uses index-only behavior: output segment data is not actively cached, while output segment index/footer ranges and independent inverted index files are written into file cache. - If only one of the two configs is enabled, the other compaction type keeps the existing cache-write decision flow. - If `enable_file_cache_write_index_file_only=true`, the global index-only config takes precedence and the base/cumulative compaction-specific configs no longer change the final behavior. - `enable_file_cache=false` is still the highest-priority switch and disables all of these file-cache writes. - When index-only is disabled, `file_cache_keep_schema_change_output_min_hit_ratio` keeps its existing semantics. - When index-only is enabled, schema change output also follows index-only behavior: - segment data is not actively written into file cache; - index-related content is written into file cache; - schema-change hit-ratio thresholds no longer decide whether index content is cached. - `enable_file_cache=false` disables everything and has the highest priority. - When index-only is enabled, the BE config has higher priority than request/session `write_file_cache=false`. - Even if session `disable_file_cache=true` makes request-side `write_file_cache=false`, as long as BE `enable_file_cache=true` and index-only is enabled: - index-related content is still written into file cache; - segment data is still not written into file cache. - When index-only is disabled, `disable_file_cache=true` preserves existing behavior and load output does not actively write file cache. - When index-only is disabled, packed small files keep the existing whole-file cache behavior. - When index-only is enabled: - packed small segment files must not be whole-file cached, preventing segment data from entering file cache; - segment index/footer ranges are written into file cache by `SegmentIndexFileCacheLoader`; - packed independent index small files are independent index files and can be whole-file cached. - Segment index/footer ranges are written through synchronous dry-run reads and do not depend on write-path `enable_flush_file_cache_async`. - Independent inverted index files still use the direct write path, so they continue to follow existing async flush behavior. `SegmentIndexFileCacheLoader` marks index-related reads as index data so that they go to the `FileCacheType::INDEX` queue when possible. After enabling index-only mode, `index_percent` should be reviewed based on index size. The default `index_percent=5` may be insufficient for large indexes. | Scenario | index-only=false | index-only=true | | --- | --- | --- | | load segment data | follows existing cache policy | not actively cached | | load segment index/footer | depends on existing whole-file cache or read path | synchronously range-read into file cache after segment close | | load independent inverted index file | follows existing cache policy | direct-cached by write path | | cumulative compaction segment data | may be cached by existing behavior | not actively cached | | cumulative compaction segment index/footer | may be cached by existing behavior | synchronously range-read into file cache after segment close | | base compaction force-keep output | whole output rowset may be cached | only output segment index/footer is preloaded | | adaptive write has enough space | segment data may be cached | segment data is still not cached | | packed small index file | may be cached | can be whole-file cached | | packed small segment file | may be cached | not whole-file cached; internal index/footer is preloaded | | query read fill cache | unchanged | unchanged | | event-driven warm-up | unchanged | unchanged | Add BE config `enable_file_cache_write_index_file_only` to support index-only file-cache writes for cloud rowset writes and compaction outputs. (cherry picked from commit 73c481f65848ed5f449fad2ac9e7c59cf54bcf9e) --- be/src/cloud/config.cpp | 1 + be/src/cloud/config.h | 1 + be/src/io/cache/block_file_cache_profile.cpp | 48 ++ be/src/io/cache/block_file_cache_profile.h | 10 + be/src/io/cache/cached_remote_file_reader.cpp | 66 +- be/src/io/cache/cached_remote_file_reader.h | 8 +- be/src/io/fs/file_reader.h | 1 + be/src/io/fs/file_writer.h | 5 +- be/src/io/fs/packed_file_system.cpp | 22 +- be/src/io/fs/s3_file_system.cpp | 2 + be/src/io/io_common.h | 10 + be/src/storage/compaction/compaction.cpp | 5 + be/src/storage/rowset/beta_rowset_writer.cpp | 24 +- be/src/storage/rowset/beta_rowset_writer.h | 2 +- be/src/storage/rowset/rowset_writer_context.h | 27 +- be/src/storage/rowset/segment_creator.cpp | 28 +- be/src/storage/rowset/segment_creator.h | 9 + .../rowset/vertical_beta_rowset_writer.cpp | 34 +- .../rowset/vertical_beta_rowset_writer.h | 7 + .../segment_index_file_cache_loader.cpp | 179 ++++ .../segment/segment_index_file_cache_loader.h | 102 +++ be/src/storage/segment/segment_writer.cpp | 25 +- be/src/storage/segment/segment_writer.h | 8 +- .../segment/vertical_segment_writer.cpp | 21 +- .../storage/segment/vertical_segment_writer.h | 8 +- be/test/cloud/cloud_compaction_test.cpp | 17 + ...cloud_file_cache_write_index_only_test.cpp | 799 ++++++++++++++++++ .../storage/compaction_file_cache_test.cpp | 58 +- ...st_file_cache_write_index_file_only.groovy | 344 ++++++++ ...x_file_only_compaction_segment_data.groovy | 239 ++++++ ...e_write_index_file_only_packed_file.groovy | 293 +++++++ 31 files changed, 2333 insertions(+), 70 deletions(-) create mode 100644 be/src/storage/segment/segment_index_file_cache_loader.cpp create mode 100644 be/src/storage/segment/segment_index_file_cache_loader.h create mode 100644 be/test/storage/cloud_file_cache_write_index_only_test.cpp create mode 100644 regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only.groovy create mode 100644 regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_compaction_segment_data.groovy create mode 100644 regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_packed_file.groovy diff --git a/be/src/cloud/config.cpp b/be/src/cloud/config.cpp index 77c1080203ab1e..4e7293b695d70b 100644 --- a/be/src/cloud/config.cpp +++ b/be/src/cloud/config.cpp @@ -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 diff --git a/be/src/cloud/config.h b/be/src/cloud/config.h index 79965117956d31..1b4ffe042fa5f9 100644 --- a/be/src/cloud/config.h +++ b/be/src/cloud/config.h @@ -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); diff --git a/be/src/io/cache/block_file_cache_profile.cpp b/be/src/io/cache/block_file_cache_profile.cpp index 8f9c167c9989e6..43264776a8275f 100644 --- a/be/src/io/cache/block_file_cache_profile.cpp +++ b/be/src/io/cache/block_file_cache_profile.cpp @@ -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; } @@ -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 { @@ -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 diff --git a/be/src/io/cache/block_file_cache_profile.h b/be/src/io/cache/block_file_cache_profile.h index 6c95e49791c054..16286cb52b8b9f 100644 --- a/be/src/io/cache/block_file_cache_profile.h +++ b/be/src/io/cache/block_file_cache_profile.h @@ -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; }; diff --git a/be/src/io/cache/cached_remote_file_reader.cpp b/be/src/io/cache/cached_remote_file_reader.cpp index 32501be65eb128..ed8fc98e4f25b3 100644 --- a/be/src/io/cache/cached_remote_file_reader.cpp +++ b/be/src/io/cache/cached_remote_file_reader.cpp @@ -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 defer((int*)0x01, std::move(defer_func)); @@ -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; } @@ -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; diff --git a/be/src/io/cache/cached_remote_file_reader.h b/be/src/io/cache/cached_remote_file_reader.h index a0037d42c64c19..a978d14e28ba3b 100644 --- a/be/src/io/cache/cached_remote_file_reader.h +++ b/be/src/io/cache/cached_remote_file_reader.h @@ -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). @@ -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; diff --git a/be/src/io/fs/file_reader.h b/be/src/io/fs/file_reader.h index 3df912cbad4af9..d9eb02cc9c2d42 100644 --- a/be/src/io/fs/file_reader.h +++ b/be/src/io/fs/file_reader.h @@ -21,6 +21,7 @@ #include #include +#include #include "common/status.h" #include "io/fs/path.h" diff --git a/be/src/io/fs/file_writer.h b/be/src/io/fs/file_writer.h index 9e3f75525032b6..9402fdef18303c 100644 --- a/be/src/io/fs/file_writer.h +++ b/be/src/io/fs/file_writer.h @@ -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 @@ -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:" diff --git a/be/src/io/fs/packed_file_system.cpp b/be/src/io/fs/packed_file_system.cpp index 9c593d8b2f71da..7c1fa5bf6ffef3 100644 --- a/be/src/io/fs/packed_file_system.cpp +++ b/be/src/io/fs/packed_file_system.cpp @@ -20,6 +20,7 @@ #include #include +#include "cloud/config.h" #include "common/status.h" #include "io/fs/file_reader.h" #include "io/fs/packed_file_reader.h" @@ -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(std::move(inner_writer), file, _append_info); + *writer = std::make_unique(std::move(inner_writer), file, append_info); return Status::OK(); } @@ -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 diff --git a/be/src/io/fs/s3_file_system.cpp b/be/src/io/fs/s3_file_system.cpp index 1ec5b0a83774cd..63be8f1955a59c 100644 --- a/be/src/io/fs/s3_file_system.cpp +++ b/be/src/io/fs/s3_file_system.cpp @@ -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" @@ -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(); diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h index 311fa1b01ded08..031387708ea6c5 100644 --- a/be/src/io/io_common.h +++ b/be/src/io/io_common.h @@ -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 { diff --git a/be/src/storage/compaction/compaction.cpp b/be/src/storage/compaction/compaction.cpp index 3fde9ab2612d41..cbaf83d56726fe 100644 --- a/be/src/storage/compaction/compaction.cpp +++ b/be/src/storage/compaction/compaction.cpp @@ -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" @@ -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; } diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp b/be/src/storage/rowset/beta_rowset_writer.cpp index 34e853f10500b9..614a8de90713ae 100644 --- a/be/src/storage/rowset/beta_rowset_writer.cpp +++ b/be/src/storage/rowset/beta_rowset_writer.cpp @@ -43,6 +43,7 @@ #include "core/block/block.h" #include "core/column/column.h" #include "core/data_type/data_type_factory.hpp" +#include "cpp/sync_point.h" #include "io/fs/file_reader.h" #include "io/fs/file_system.h" #include "io/fs/file_writer.h" @@ -194,6 +195,7 @@ Status SegmentFileCollection::close() { if (writer->state() != io::FileWriter::State::CLOSED) { RETURN_IF_ERROR(writer->close()); } + TEST_SYNC_POINT_CALLBACK("SegmentFileCollection::close_file_writer", writer.get()); } return Status::OK(); @@ -1114,8 +1116,8 @@ Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) { Status BaseBetaRowsetWriter::_create_file_writer(const std::string& path, io::FileWriterPtr& file_writer, - bool is_index_file) { - io::FileWriterOptions opts = _context.get_file_writer_options(is_index_file); + FileType file_type) { + io::FileWriterOptions opts = _context.get_file_writer_options(file_type); Status st = _context.fs()->create_file(path, &file_writer, &opts); if (!st.ok()) { LOG(WARNING) << "failed to create writable file. path=" << path << ", err: " << st; @@ -1123,6 +1125,8 @@ Status BaseBetaRowsetWriter::_create_file_writer(const std::string& path, } DCHECK(file_writer != nullptr); + TEST_SYNC_POINT_CALLBACK("BaseBetaRowsetWriter::_create_file_writer", &path, &file_type, + file_writer.get(), &opts); return Status::OK(); } @@ -1133,9 +1137,9 @@ Status BaseBetaRowsetWriter::create_file_writer(uint32_t segment_id, io::FileWri std::string prefix = std::string {InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)}; std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix); - return _create_file_writer(index_path, file_writer, true /* is_index_file */); + return _create_file_writer(index_path, file_writer, file_type); } else if (file_type == FileType::SEGMENT_FILE) { - return _create_file_writer(segment_path, file_writer, false /* is_index_file */); + return _create_file_writer(segment_path, file_writer, file_type); } return Status::Error( fmt::format("failed to create file = {}, file type = {}", segment_path, file_type)); @@ -1146,7 +1150,9 @@ Status BaseBetaRowsetWriter::create_index_file_writer(uint32_t segment_id, RETURN_IF_ERROR(RowsetWriter::create_index_file_writer(segment_id, index_file_writer)); // used for inverted index format v1 (*index_file_writer) - ->set_file_writer_opts(_context.get_file_writer_options(true /* is_index_file */)); + ->set_file_writer_opts(_context.get_file_writer_options(FileType::INVERTED_INDEX_FILE)); + TEST_SYNC_POINT_CALLBACK("BaseBetaRowsetWriter::create_inverted_index_file_writer", &segment_id, + index_file_writer->get()); return Status::OK(); } @@ -1156,7 +1162,7 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction( std::string path = BetaRowset::local_segment_path_segcompacted(_context.tablet_path, _context.rowset_id, begin, end); io::FileWriterPtr file_writer; - RETURN_IF_ERROR(_create_file_writer(path, file_writer, false /* is_index_file */)); + RETURN_IF_ERROR(_create_file_writer(path, file_writer, FileType::SEGMENT_FILE)); IndexFileWriterPtr index_file_writer; if (_context.tablet_schema->has_inverted_index() || _context.tablet_schema->has_ann_index()) { @@ -1165,13 +1171,15 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction( if (_context.tablet_schema->get_inverted_index_storage_format() != InvertedIndexStorageFormatPB::V1) { std::string index_path = InvertedIndexDescriptor::get_index_file_path_v2(prefix); - RETURN_IF_ERROR( - _create_file_writer(index_path, idx_file_writer, true /* is_index_file */)); + RETURN_IF_ERROR(_create_file_writer(index_path, idx_file_writer, + FileType::INVERTED_INDEX_FILE)); } index_file_writer = std::make_unique( _context.fs(), prefix, _context.rowset_id.to_string(), _num_segcompacted, _context.tablet_schema->get_inverted_index_storage_format(), std::move(idx_file_writer)); + index_file_writer->set_file_writer_opts( + _context.get_file_writer_options(FileType::INVERTED_INDEX_FILE)); } segment_v2::SegmentWriterOptions writer_options; diff --git a/be/src/storage/rowset/beta_rowset_writer.h b/be/src/storage/rowset/beta_rowset_writer.h index 8011864cb58a07..8a8642c2e5b221 100644 --- a/be/src/storage/rowset/beta_rowset_writer.h +++ b/be/src/storage/rowset/beta_rowset_writer.h @@ -200,7 +200,7 @@ class BaseBetaRowsetWriter : public RowsetWriter { Status _generate_delete_bitmap(int32_t segment_id); virtual Status _build_rowset_meta(RowsetMeta* rowset_meta, bool check_segment_num = false); Status _create_file_writer(const std::string& path, io::FileWriterPtr& file_writer, - bool is_index_file = false); + FileType file_type = FileType::SEGMENT_FILE); virtual Status _close_file_writers(); virtual Status _check_segment_number_limit(size_t segnum); virtual int64_t _num_seg() const; diff --git a/be/src/storage/rowset/rowset_writer_context.h b/be/src/storage/rowset/rowset_writer_context.h index ba803a9a118839..b65232f3c1c127 100644 --- a/be/src/storage/rowset/rowset_writer_context.h +++ b/be/src/storage/rowset/rowset_writer_context.h @@ -236,17 +236,26 @@ struct RowsetWriterContext { io::FileSystem& fs_ref() const { return *fs(); } - io::FileWriterOptions get_file_writer_options(bool is_index_file = false) { - bool should_write_cache = write_file_cache; - // If configured to only write index files to cache, skip cache for data files - if (compaction_output_write_index_only && !is_index_file) { - should_write_cache = false; + io::FileWriterOptions get_file_writer_options(FileType file_type = FileType::SEGMENT_FILE) { + io::FileWriterOptions opts {.write_file_cache = write_file_cache, + .is_cold_data = is_hot_data, + .file_cache_expiration_time = file_cache_ttl_sec, + .approximate_bytes_to_write = approximate_bytes_to_write}; + + if (config::enable_file_cache_write_index_file_only) { + opts.allow_adaptive_file_cache_write = false; + opts.approximate_bytes_to_write = 0; + opts.write_file_cache = file_type == FileType::INVERTED_INDEX_FILE; + return opts; } - return io::FileWriterOptions {.write_file_cache = should_write_cache, - .is_cold_data = is_hot_data, - .file_cache_expiration_time = file_cache_ttl_sec, - .approximate_bytes_to_write = approximate_bytes_to_write}; + if (compaction_output_write_index_only && file_type == FileType::SEGMENT_FILE) { + opts.write_file_cache = false; + opts.allow_adaptive_file_cache_write = false; + opts.approximate_bytes_to_write = 0; + } + + return opts; } }; diff --git a/be/src/storage/rowset/segment_creator.cpp b/be/src/storage/rowset/segment_creator.cpp index 5d67ad7b93bb25..2d4bcdd0e2118f 100644 --- a/be/src/storage/rowset/segment_creator.cpp +++ b/be/src/storage/rowset/segment_creator.cpp @@ -40,9 +40,11 @@ #include "core/column/column_variant.h" #include "core/data_type/data_type.h" #include "core/types.h" +#include "cpp/sync_point.h" #include "io/fs/file_writer.h" #include "storage/olap_define.h" #include "storage/rowset/beta_rowset_writer.h" // SegmentStatistics +#include "storage/segment/segment_index_file_cache_loader.h" #include "storage/segment/segment_writer.h" #include "storage/segment/vertical_segment_writer.h" #include "storage/tablet/tablet_schema.h" @@ -85,10 +87,27 @@ Status SegmentFlusher::flush_single_block(const Block* block, int32_t segment_id Status SegmentFlusher::close() { RETURN_IF_ERROR(_seg_files.close()); + RETURN_IF_ERROR(_preload_segment_indexes_to_file_cache()); RETURN_IF_ERROR(_idx_files.finish_close()); return Status::OK(); } +void SegmentFlusher::_record_segment_index_file_cache_preload( + uint32_t segment_id, const segment_v2::SegmentIndexFileCacheInfo& info) { + std::lock_guard lock(_segment_index_file_cache_preloads_lock); + _segment_index_file_cache_preloads.push_back({segment_id, info}); +} + +Status SegmentFlusher::_preload_segment_indexes_to_file_cache() { + std::vector tasks; + { + std::lock_guard lock(_segment_index_file_cache_preloads_lock); + tasks.swap(_segment_index_file_cache_preloads); + } + return segment_v2::SegmentIndexFileCacheLoader::preload_segment_indexes_to_file_cache(_context, + tasks); +} + Status SegmentFlusher::_add_rows(std::unique_ptr& segment_writer, const Block* block, size_t row_pos, size_t num_rows) { RETURN_IF_ERROR(segment_writer->append_block(block, row_pos, num_rows)); @@ -199,7 +218,8 @@ Status SegmentFlusher::_flush_segment_writer( finalize_timer.start(); uint64_t segment_file_size; uint64_t common_index_size; - Status s = writer->finalize(&segment_file_size, &common_index_size); + segment_v2::SegmentIndexFileCacheInfo index_file_cache_info; + Status s = writer->finalize(&segment_file_size, &common_index_size, &index_file_cache_info); finalize_timer.stop(); if (!s.ok()) { @@ -227,6 +247,7 @@ Status SegmentFlusher::_flush_segment_writer( key_bounds.set_max_key(max_key.to_string()); uint32_t segment_id = writer->segment_id(); + TEST_SYNC_POINT_CALLBACK("SegmentFlusher::flush_vertical_segment_writer", &segment_id); SegmentStatistics segstat; segstat.row_num = row_num; segstat.data_size = segment_file_size; @@ -234,6 +255,7 @@ Status SegmentFlusher::_flush_segment_writer( segstat.key_bounds = key_bounds; writer.reset(); + _record_segment_index_file_cache_preload(segment_id, index_file_cache_info); MonotonicStopWatch collector_timer; collector_timer.start(); @@ -277,7 +299,8 @@ Status SegmentFlusher::_flush_segment_writer(std::unique_ptrfinalize(&segment_file_size, &common_index_size); + segment_v2::SegmentIndexFileCacheInfo index_file_cache_info; + Status s = writer->finalize(&segment_file_size, &common_index_size, &index_file_cache_info); finalize_timer.stop(); if (!s.ok()) { @@ -312,6 +335,7 @@ Status SegmentFlusher::_flush_segment_writer(std::unique_ptr #include +#include +#include + #include "common/status.h" #include "core/block/block.h" #include "io/fs/file_reader_writer_fwd.h" #include "storage/index/index_file_writer.h" #include "storage/rowset/rowset_writer_context.h" +#include "storage/segment/segment_index_file_cache_loader.h" #include "storage/tablet/tablet_fwd.h" namespace doris { @@ -149,6 +153,9 @@ class SegmentFlusher { int64_t* flush_size = nullptr); Status _flush_segment_writer(std::unique_ptr& writer, int64_t* flush_size = nullptr); + void _record_segment_index_file_cache_preload( + uint32_t segment_id, const segment_v2::SegmentIndexFileCacheInfo& info); + Status _preload_segment_indexes_to_file_cache(); private: RowsetWriterContext& _context; @@ -161,6 +168,8 @@ class SegmentFlusher { std::atomic _num_rows_new_added = 0; std::atomic _num_rows_deleted = 0; std::atomic _num_rows_filtered = 0; + std::mutex _segment_index_file_cache_preloads_lock; + std::vector _segment_index_file_cache_preloads; }; class SegmentCreator { diff --git a/be/src/storage/rowset/vertical_beta_rowset_writer.cpp b/be/src/storage/rowset/vertical_beta_rowset_writer.cpp index 3736124fc0ff52..c6e7a12b91bc6a 100644 --- a/be/src/storage/rowset/vertical_beta_rowset_writer.cpp +++ b/be/src/storage/rowset/vertical_beta_rowset_writer.cpp @@ -32,11 +32,13 @@ #include "common/compiler_util.h" // IWYU pragma: keep #include "common/logging.h" #include "core/block/block.h" +#include "cpp/sync_point.h" #include "io/fs/file_system.h" #include "io/fs/file_writer.h" #include "storage/rowset/beta_rowset.h" #include "storage/rowset/rowset_meta.h" #include "storage/rowset/rowset_writer_context.h" +#include "storage/segment/segment_index_file_cache_loader.h" #include "util/slice.h" namespace doris { @@ -171,7 +173,7 @@ Status VerticalBetaRowsetWriter::_create_segment_writer( IndexFileWriterPtr index_file_writer; if (context.tablet_schema->has_inverted_index() || context.tablet_schema->has_ann_index()) { - RETURN_IF_ERROR(RowsetWriter::create_index_file_writer(seg_id, &index_file_writer)); + RETURN_IF_ERROR(this->create_index_file_writer(seg_id, &index_file_writer)); } segment_v2::SegmentWriterOptions writer_options; @@ -204,13 +206,18 @@ Status VerticalBetaRowsetWriter::final_flush() { for (auto& segment_writer : _segment_writers) { uint64_t segment_size = 0; //uint64_t footer_position = 0; - auto st = segment_writer->finalize_footer(&segment_size); + segment_v2::SegmentIndexFileCacheInfo index_file_cache_info; + auto segment_id = segment_writer->get_segment_id(); + auto st = segment_writer->finalize_footer(&segment_size, &index_file_cache_info); if (!st.ok()) { LOG(WARNING) << "Fail to finalize segment footer, " << st; return st; } this->_total_data_size += segment_size; + TEST_SYNC_POINT_CALLBACK("VerticalBetaRowsetWriter::final_flush_segment_writer", + &segment_id); segment_writer.reset(); + _record_segment_index_file_cache_preload(segment_id, index_file_cache_info); } return Status::OK(); } @@ -219,7 +226,28 @@ template requires std::is_base_of_v Status VerticalBetaRowsetWriter::_close_file_writers() { RETURN_IF_ERROR(BaseBetaRowsetWriter::_close_inverted_index_file_writers()); - return this->_seg_files.close(); + RETURN_IF_ERROR(this->_seg_files.close()); + return _preload_segment_indexes_to_file_cache(); +} + +template + requires std::is_base_of_v +void VerticalBetaRowsetWriter::_record_segment_index_file_cache_preload( + uint32_t segment_id, const segment_v2::SegmentIndexFileCacheInfo& info) { + std::lock_guard lock(_segment_index_file_cache_preloads_lock); + _segment_index_file_cache_preloads.push_back({segment_id, info}); +} + +template + requires std::is_base_of_v +Status VerticalBetaRowsetWriter::_preload_segment_indexes_to_file_cache() { + std::vector tasks; + { + std::lock_guard lock(_segment_index_file_cache_preloads_lock); + tasks.swap(_segment_index_file_cache_preloads); + } + return segment_v2::SegmentIndexFileCacheLoader::preload_segment_indexes_to_file_cache( + this->_context, tasks); } } // namespace doris diff --git a/be/src/storage/rowset/vertical_beta_rowset_writer.h b/be/src/storage/rowset/vertical_beta_rowset_writer.h index 650cf273478127..ecff7745e8eba2 100644 --- a/be/src/storage/rowset/vertical_beta_rowset_writer.h +++ b/be/src/storage/rowset/vertical_beta_rowset_writer.h @@ -18,11 +18,13 @@ #pragma once #include +#include #include #include #include "common/status.h" #include "storage/rowset/beta_rowset_writer.h" +#include "storage/segment/segment_index_file_cache_loader.h" #include "storage/segment/segment_writer.h" namespace doris { @@ -55,10 +57,15 @@ class VerticalBetaRowsetWriter final : public T { Status _flush_columns(segment_v2::SegmentWriter* segment_writer, bool is_key = false); Status _create_segment_writer(const std::vector& column_ids, bool is_key, std::unique_ptr* writer); + void _record_segment_index_file_cache_preload( + uint32_t segment_id, const segment_v2::SegmentIndexFileCacheInfo& info); + Status _preload_segment_indexes_to_file_cache(); std::vector> _segment_writers; size_t _cur_writer_idx = 0; size_t _total_key_group_rows = 0; + std::mutex _segment_index_file_cache_preloads_lock; + std::vector _segment_index_file_cache_preloads; }; } // namespace doris diff --git a/be/src/storage/segment/segment_index_file_cache_loader.cpp b/be/src/storage/segment/segment_index_file_cache_loader.cpp new file mode 100644 index 00000000000000..13a93460cbc2a3 --- /dev/null +++ b/be/src/storage/segment/segment_index_file_cache_loader.cpp @@ -0,0 +1,179 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "storage/segment/segment_index_file_cache_loader.h" + +#include + +#include "cloud/config.h" +#include "cpp/sync_point.h" +#include "io/fs/file_reader.h" +#include "io/io_common.h" +#include "storage/rowset/rowset_writer_context.h" +#include "util/slice.h" + +namespace doris::segment_v2 { + +namespace { + +bvar::Adder g_segment_index_file_cache_load_total("segment_index_file_cache_load_total"); +bvar::Adder g_segment_index_file_cache_load_failed("segment_index_file_cache_load_failed"); +bvar::Adder g_segment_index_file_cache_load_bytes("segment_index_file_cache_load_bytes"); + +bool enable_cloud_index_only_file_cache() { + return config::is_cloud_mode() && config::enable_file_cache && + config::enable_file_cache_write_index_file_only; +} + +const char* reason_to_string(SegmentIndexFileCacheLoadReason reason) { + switch (reason) { + case SegmentIndexFileCacheLoadReason::LOAD: + return "load"; + case SegmentIndexFileCacheLoadReason::CUMULATIVE_COMPACTION: + return "cumulative_compaction"; + case SegmentIndexFileCacheLoadReason::BASE_COMPACTION: + return "base_compaction"; + case SegmentIndexFileCacheLoadReason::SCHEMA_CHANGE: + return "schema_change"; + } + return "unknown"; +} + +SegmentIndexFileCacheLoadReason reason_from_context(const RowsetWriterContext& context) { + if (context.write_type == DataWriteType::TYPE_SCHEMA_CHANGE) { + return SegmentIndexFileCacheLoadReason::SCHEMA_CHANGE; + } + if (context.write_type == DataWriteType::TYPE_COMPACTION) { + if (context.compaction_type == ReaderType::READER_BASE_COMPACTION || + context.compaction_type == ReaderType::READER_FULL_COMPACTION) { + return SegmentIndexFileCacheLoadReason::BASE_COMPACTION; + } + return SegmentIndexFileCacheLoadReason::CUMULATIVE_COMPACTION; + } + return SegmentIndexFileCacheLoadReason::LOAD; +} + +Status read_range_to_file_cache(io::FileReaderSPtr reader, uint64_t offset, uint64_t size, + const io::IOContext& io_ctx) { + const auto read_size = static_cast(size); + size_t bytes_read = 0; + RETURN_IF_ERROR(reader->read_at(offset, Slice(static_cast(nullptr), read_size), + &bytes_read, &io_ctx)); + if (bytes_read != read_size) { + return Status::InternalError( + "short dry-run read when preloading segment index to file cache, offset={}, " + "expected={}, actual={}", + offset, read_size, bytes_read); + } + return Status::OK(); +} + +} // namespace + +Status SegmentIndexFileCacheLoader::preload_segment_index_to_file_cache( + const RowsetWriterContext& context, uint32_t segment_id, const std::string& segment_path, + const SegmentIndexFileCacheInfo& info) { + if (!enable_cloud_index_only_file_cache() || context.is_local_rowset()) { + return Status::OK(); + } + + for (const auto& range : info.index_ranges) { + auto st = load_segment_index_to_file_cache({ + .fs = context.fs(), + .segment_path = segment_path, + .rowset_id = context.rowset_id, + .tablet_id = context.tablet_id, + .segment_id = segment_id, + .range = range, + .segment_file_size = info.segment_file_size, + .reason = reason_from_context(context), + }); + if (!st.ok()) { + g_segment_index_file_cache_load_failed << 1; + LOG(WARNING) << "failed to preload segment index to file cache, tablet_id=" + << context.tablet_id << ", rowset_id=" << context.rowset_id + << ", segment_id=" << segment_id << ", segment_path=" << segment_path + << ", index_start=" << range.offset << ", index_size=" << range.size + << ", segment_file_size=" << info.segment_file_size << ", status=" << st; + } + } + return Status::OK(); +} + +Status SegmentIndexFileCacheLoader::preload_segment_indexes_to_file_cache( + const RowsetWriterContext& context, + const std::vector& tasks) { + TEST_SYNC_POINT_CALLBACK("SegmentIndexFileCacheLoader::preload_segment_indexes_to_file_cache", + &context, &tasks); + for (const auto& task : tasks) { + RETURN_IF_ERROR(preload_segment_index_to_file_cache( + context, task.segment_id, context.segment_path(task.segment_id), task.info)); + } + return Status::OK(); +} + +Status SegmentIndexFileCacheLoader::load_segment_index_to_file_cache( + const SegmentIndexFileCacheLoadContext& ctx) { + if (!enable_cloud_index_only_file_cache()) { + return Status::OK(); + } + const auto& range = ctx.range; + if (range.empty()) { + return Status::OK(); + } + if (!range.is_valid_for(ctx.segment_file_size)) { + return Status::InvalidArgument( + "invalid segment index cache range, path={}, index_start={}, index_size={}, " + "segment_file_size={}", + ctx.segment_path, range.offset, range.size, ctx.segment_file_size); + } + if (ctx.fs == nullptr) { + return Status::InternalError("file system is null"); + } + + io::IOContext io_ctx { + .reader_type = ReaderType::READER_QUERY, + .is_index_data = true, + .is_dryrun = true, + .is_warmup = false, + }; + TEST_SYNC_POINT_RETURN_WITH_VALUE( + "SegmentIndexFileCacheLoader::load_segment_index_to_file_cache", Status::OK(), &ctx, + &io_ctx); + + io::FileReaderOptions reader_opts; + reader_opts.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE; + reader_opts.is_doris_table = true; + reader_opts.file_size = static_cast(ctx.segment_file_size); + reader_opts.tablet_id = ctx.tablet_id; + + io::FileReaderSPtr reader; + RETURN_IF_ERROR(ctx.fs->open_file(ctx.segment_path, &reader, &reader_opts)); + + RETURN_IF_ERROR(read_range_to_file_cache(reader, range.offset, range.size, io_ctx)); + g_segment_index_file_cache_load_total << 1; + g_segment_index_file_cache_load_bytes << static_cast(range.size); + + VLOG_DEBUG << "preloaded segment index to file cache, tablet_id=" << ctx.tablet_id + << ", rowset_id=" << ctx.rowset_id << ", segment_id=" << ctx.segment_id + << ", segment_path=" << ctx.segment_path << ", index_start=" << range.offset + << ", index_size=" << range.size << ", segment_file_size=" << ctx.segment_file_size + << ", reason=" << reason_to_string(ctx.reason); + return Status::OK(); +} + +} // namespace doris::segment_v2 diff --git a/be/src/storage/segment/segment_index_file_cache_loader.h b/be/src/storage/segment/segment_index_file_cache_loader.h new file mode 100644 index 00000000000000..62d08339405bdd --- /dev/null +++ b/be/src/storage/segment/segment_index_file_cache_loader.h @@ -0,0 +1,102 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include +#include +#include + +#include "common/status.h" +#include "io/fs/file_reader_writer_fwd.h" +#include "io/fs/file_system.h" +#include "storage/olap_common.h" + +namespace doris { + +struct RowsetWriterContext; + +namespace segment_v2 { + +struct SegmentIndexFileCacheRange { + uint64_t offset = 0; + uint64_t size = 0; + + bool empty() const { return size == 0; } + + bool is_valid_for(uint64_t file_size) const { + return !empty() && offset < file_size && size <= file_size - offset; + } +}; + +struct SegmentIndexFileCacheInfo { + uint64_t segment_file_size = 0; + std::vector index_ranges; + + void add_index_range(uint64_t offset, uint64_t size) { + if (size == 0) { + return; + } + index_ranges.push_back({offset, size}); + } + + bool empty() const { return index_ranges.empty(); } + + uint64_t cache_start_offset() const { + return empty() ? segment_file_size : index_ranges.front().offset; + } +}; + +struct SegmentIndexFileCachePreloadTask { + uint32_t segment_id = 0; + SegmentIndexFileCacheInfo info; +}; + +enum class SegmentIndexFileCacheLoadReason { + LOAD, + CUMULATIVE_COMPACTION, + BASE_COMPACTION, + SCHEMA_CHANGE, +}; + +struct SegmentIndexFileCacheLoadContext { + io::FileSystemSPtr fs; + std::string segment_path; + RowsetId rowset_id; + int64_t tablet_id = 0; + uint32_t segment_id = 0; + SegmentIndexFileCacheRange range; + uint64_t segment_file_size = 0; + SegmentIndexFileCacheLoadReason reason = SegmentIndexFileCacheLoadReason::LOAD; +}; + +class SegmentIndexFileCacheLoader { +public: + static Status preload_segment_index_to_file_cache(const RowsetWriterContext& context, + uint32_t segment_id, + const std::string& segment_path, + const SegmentIndexFileCacheInfo& info); + + static Status preload_segment_indexes_to_file_cache( + const RowsetWriterContext& context, + const std::vector& tasks); + + static Status load_segment_index_to_file_cache(const SegmentIndexFileCacheLoadContext& ctx); +}; + +} // namespace segment_v2 +} // namespace doris diff --git a/be/src/storage/segment/segment_writer.cpp b/be/src/storage/segment/segment_writer.cpp index 1b34c24e767efa..438a848c929716 100644 --- a/be/src/storage/segment/segment_writer.cpp +++ b/be/src/storage/segment/segment_writer.cpp @@ -946,6 +946,10 @@ Status SegmentWriter::finalize_columns_data() { Status SegmentWriter::finalize_columns_index(uint64_t* index_size) { uint64_t index_start = _file_writer->bytes_appended(); + // Record each index range separately. Vertical compaction writes column groups as + // data+index pairs, so a single [first index, EOF) range would include later column data. + // This SegmentWriter path is shared by cloud load, non-vertical compaction, schema change + // final output, and vertical compaction via VerticalBetaRowsetWriter. RETURN_IF_ERROR(_write_ordinal_index()); RETURN_IF_ERROR(_write_zone_map()); RETURN_IF_ERROR(_write_inverted_index()); @@ -981,24 +985,36 @@ Status SegmentWriter::finalize_columns_index(uint64_t* index_size) { *index_size = _file_writer->bytes_appended() - index_start; } } + uint64_t file_index_end = _file_writer->bytes_appended(); + _index_file_cache_info.add_index_range(index_start, file_index_end - index_start); // reset all column writers and data_conveter clear(); return Status::OK(); } -Status SegmentWriter::finalize_footer(uint64_t* segment_file_size) { +Status SegmentWriter::finalize_footer(uint64_t* segment_file_size, + SegmentIndexFileCacheInfo* index_file_cache_info) { + uint64_t footer_start = _file_writer->bytes_appended(); RETURN_IF_ERROR(_write_footer()); // finish RETURN_IF_ERROR(_file_writer->close(true)); *segment_file_size = _file_writer->bytes_appended(); + // The closed size completes the preload range recorded above. Local temporary rowsets, such as + // schema-change internal sorting output, are filtered by SegmentIndexFileCacheLoader. + _index_file_cache_info.segment_file_size = *segment_file_size; + _index_file_cache_info.add_index_range(footer_start, *segment_file_size - footer_start); + if (index_file_cache_info != nullptr) { + *index_file_cache_info = _index_file_cache_info; + } if (*segment_file_size == 0) { return Status::Corruption("Bad segment, file size = 0"); } return Status::OK(); } -Status SegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size) { +Status SegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size, + SegmentIndexFileCacheInfo* index_file_cache_info) { MonotonicStopWatch timer; timer.start(); // check disk capacity @@ -1008,12 +1024,10 @@ Status SegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size } // write data RETURN_IF_ERROR(finalize_columns_data()); - // Get the index start before finalize_footer since this function would write new data. - uint64_t index_start = _file_writer->bytes_appended(); // write index RETURN_IF_ERROR(finalize_columns_index(index_size)); // write footer - RETURN_IF_ERROR(finalize_footer(segment_file_size)); + RETURN_IF_ERROR(finalize_footer(segment_file_size, index_file_cache_info)); if (timer.elapsed_time() > 5000000000l) { LOG(INFO) << "segment flush consumes a lot time_ns " << timer.elapsed_time() @@ -1024,6 +1038,7 @@ Status SegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size if (auto* cache_builder = _file_writer->cache_builder(); cache_builder != nullptr && cache_builder->_expiration_time == 0 && config::is_cloud_mode()) { + auto index_start = _index_file_cache_info.cache_start_offset(); auto size = *index_size + *segment_file_size; auto holder = cache_builder->allocate_cache_holder(index_start, size, _tablet->tablet_id()); for (auto& segment : holder->file_blocks) { diff --git a/be/src/storage/segment/segment_writer.h b/be/src/storage/segment/segment_writer.h index d93dff79dbcd92..81ff950cf6c0d3 100644 --- a/be/src/storage/segment/segment_writer.h +++ b/be/src/storage/segment/segment_writer.h @@ -33,6 +33,7 @@ #include "storage/index/index_file_writer.h" #include "storage/olap_define.h" #include "storage/segment/column_writer.h" +#include "storage/segment/segment_index_file_cache_loader.h" #include "storage/tablet/tablet.h" #include "storage/tablet/tablet_schema.h" #include "util/faststring.h" @@ -116,13 +117,15 @@ class SegmentWriter { uint32_t row_count() const { return _row_count; } - Status finalize(uint64_t* segment_file_size, uint64_t* index_size); + Status finalize(uint64_t* segment_file_size, uint64_t* index_size, + SegmentIndexFileCacheInfo* index_file_cache_info = nullptr); uint32_t get_segment_id() const { return _segment_id; } Status finalize_columns_data(); Status finalize_columns_index(uint64_t* index_size); - Status finalize_footer(uint64_t* segment_file_size); + Status finalize_footer(uint64_t* segment_file_size, + SegmentIndexFileCacheInfo* index_file_cache_info = nullptr); void init_column_meta(ColumnMetaPB* meta, uint32_t column_id, const TabletColumn& column, const ColumnWriterOptions& opts); @@ -205,6 +208,7 @@ class SegmentWriter { IndexFileWriter* _index_file_writer = nullptr; SegmentFooterPB _footer; + SegmentIndexFileCacheInfo _index_file_cache_info; // for mow tables with cluster key, the sort key is the cluster keys not unique keys // for other tables, the sort key is the keys size_t _num_sort_key_columns; diff --git a/be/src/storage/segment/vertical_segment_writer.cpp b/be/src/storage/segment/vertical_segment_writer.cpp index 7193cfb0a78d88..bbeafcf8601bc5 100644 --- a/be/src/storage/segment/vertical_segment_writer.cpp +++ b/be/src/storage/segment/vertical_segment_writer.cpp @@ -1324,6 +1324,9 @@ uint64_t VerticalSegmentWriter::_estimated_remaining_size() { Status VerticalSegmentWriter::finalize_columns_index(uint64_t* index_size) { uint64_t index_start = _file_writer->bytes_appended(); + // Record the common index range for cloud index-only file-cache preload. + // This VerticalSegmentWriter path is used when cloud load, compaction, or schema change flushes + // a whole block through SegmentCreator with enable_vertical_segment_writer enabled. RETURN_IF_ERROR(_write_ordinal_index()); RETURN_IF_ERROR(_write_zone_map()); RETURN_IF_ERROR(_write_inverted_index()); @@ -1345,6 +1348,8 @@ Status VerticalSegmentWriter::finalize_columns_index(uint64_t* index_size) { RETURN_IF_ERROR(_write_short_key_index()); *index_size = _file_writer->bytes_appended() - index_start; } + uint64_t file_index_end = _file_writer->bytes_appended(); + _index_file_cache_info.add_index_range(index_start, file_index_end - index_start); // reset all column writers and data_conveter clear(); @@ -1352,18 +1357,28 @@ Status VerticalSegmentWriter::finalize_columns_index(uint64_t* index_size) { return Status::OK(); } -Status VerticalSegmentWriter::finalize_footer(uint64_t* segment_file_size) { +Status VerticalSegmentWriter::finalize_footer(uint64_t* segment_file_size, + SegmentIndexFileCacheInfo* index_file_cache_info) { + uint64_t footer_start = _file_writer->bytes_appended(); RETURN_IF_ERROR(_write_footer()); // finish RETURN_IF_ERROR(_file_writer->close(true)); *segment_file_size = _file_writer->bytes_appended(); + // The closed size completes the preload range recorded above. SegmentIndexFileCacheLoader + // later decides whether this is a remote cloud rowset that should actually be preloaded. + _index_file_cache_info.segment_file_size = *segment_file_size; + _index_file_cache_info.add_index_range(footer_start, *segment_file_size - footer_start); + if (index_file_cache_info != nullptr) { + *index_file_cache_info = _index_file_cache_info; + } if (*segment_file_size == 0) { return Status::Corruption("Bad segment, file size = 0"); } return Status::OK(); } -Status VerticalSegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size) { +Status VerticalSegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* index_size, + SegmentIndexFileCacheInfo* index_file_cache_info) { MonotonicStopWatch timer; timer.start(); // check disk capacity @@ -1377,7 +1392,7 @@ Status VerticalSegmentWriter::finalize(uint64_t* segment_file_size, uint64_t* in // write index RETURN_IF_ERROR(finalize_columns_index(index_size)); // write footer - RETURN_IF_ERROR(finalize_footer(segment_file_size)); + RETURN_IF_ERROR(finalize_footer(segment_file_size, index_file_cache_info)); if (timer.elapsed_time() > 5000000000L) { LOG(INFO) << "segment flush consumes a lot time_ns " << timer.elapsed_time() diff --git a/be/src/storage/segment/vertical_segment_writer.h b/be/src/storage/segment/vertical_segment_writer.h index e6ee4933dd6d19..31de12d772f000 100644 --- a/be/src/storage/segment/vertical_segment_writer.h +++ b/be/src/storage/segment/vertical_segment_writer.h @@ -34,6 +34,7 @@ #include "storage/olap_define.h" #include "storage/partial_update_info.h" #include "storage/segment/column_writer.h" +#include "storage/segment/segment_index_file_cache_loader.h" #include "storage/tablet/tablet.h" #include "storage/tablet/tablet_schema.h" #include "util/faststring.h" @@ -107,10 +108,12 @@ class VerticalSegmentWriter { [[nodiscard]] uint32_t row_count() const { return _row_count; } [[nodiscard]] uint32_t segment_id() const { return _segment_id; } - Status finalize(uint64_t* segment_file_size, uint64_t* index_size); + Status finalize(uint64_t* segment_file_size, uint64_t* index_size, + SegmentIndexFileCacheInfo* index_file_cache_info = nullptr); Status finalize_columns_index(uint64_t* index_size); - Status finalize_footer(uint64_t* segment_file_size); + Status finalize_footer(uint64_t* segment_file_size, + SegmentIndexFileCacheInfo* index_file_cache_info = nullptr); Slice min_encoded_key(); Slice max_encoded_key(); @@ -216,6 +219,7 @@ class VerticalSegmentWriter { IndexFileWriter* _index_file_writer = nullptr; SegmentFooterPB _footer; + SegmentIndexFileCacheInfo _index_file_cache_info; // for mow tables with cluster key, the sort key is the cluster keys not unique keys // for other tables, the sort key is the keys size_t _num_sort_key_columns; diff --git a/be/test/cloud/cloud_compaction_test.cpp b/be/test/cloud/cloud_compaction_test.cpp index 05eec3149f996a..f7916ab511ef55 100644 --- a/be/test/cloud/cloud_compaction_test.cpp +++ b/be/test/cloud/cloud_compaction_test.cpp @@ -28,12 +28,14 @@ #include "cloud/cloud_storage_engine.h" #include "cloud/cloud_tablet.h" #include "cloud/cloud_tablet_mgr.h" +#include "cloud/config.h" #include "json2pb/json_to_pb.h" #include "storage/olap_common.h" #include "storage/rowset/rowset_factory.h" #include "storage/rowset/rowset_meta.h" #include "storage/storage_policy.h" #include "storage/tablet/tablet_meta.h" +#include "util/defer_op.h" #include "util/uid_util.h" namespace doris { @@ -395,6 +397,15 @@ TEST_F(CloudCompactionTest, test_set_storage_resource_from_input_rowsets) { } } TEST_F(CloudCompactionTest, should_cache_compaction_output) { + auto old_write_index_file_only = config::enable_file_cache_write_index_file_only; + auto old_keep_base_compaction_output = config::enable_file_cache_keep_base_compaction_output; + Defer restore_config {[&] { + config::enable_file_cache_write_index_file_only = old_write_index_file_only; + config::enable_file_cache_keep_base_compaction_output = old_keep_base_compaction_output; + }}; + config::enable_file_cache_write_index_file_only = false; + config::enable_file_cache_keep_base_compaction_output = false; + CloudTabletSPtr tablet = std::make_shared(_engine, std::make_shared()); CloudBaseCompaction cloud_base_compaction(_engine, tablet); cloud_base_compaction._input_rowsets_total_size = 0; @@ -436,6 +447,12 @@ TEST_F(CloudCompactionTest, should_cache_compaction_output) { cloud_base_compaction._input_rowsets_cached_data_size = 50; cloud_base_compaction._input_rowsets_cached_index_size = 50; ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), true); + + config::enable_file_cache_keep_base_compaction_output = true; + ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), true); + + config::enable_file_cache_write_index_file_only = true; + ASSERT_EQ(cloud_base_compaction.should_cache_compaction_output(), false); LOG(INFO) << "should_cache_compaction_output done"; } diff --git a/be/test/storage/cloud_file_cache_write_index_only_test.cpp b/be/test/storage/cloud_file_cache_write_index_only_test.cpp new file mode 100644 index 00000000000000..c330b93fee489b --- /dev/null +++ b/be/test/storage/cloud_file_cache_write_index_only_test.cpp @@ -0,0 +1,799 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cloud/config.h" +#include "common/config.h" +#include "core/block/block.h" +#include "cpp/sync_point.h" +#include "io/cache/block_file_cache_factory.h" +#include "io/fs/file_writer.h" +#include "io/fs/local_file_system.h" +#include "io/fs/s3_file_system.h" +#include "io/fs/s3_file_writer.h" +#include "io/fs/s3_obj_storage_client.h" +#include "io/io_common.h" +#include "runtime/exec_env.h" +#include "storage/index/inverted/inverted_index_writer.h" +#include "storage/options.h" +#include "storage/rowset/rowset_factory.h" +#include "storage/rowset/rowset_writer.h" +#include "storage/rowset/rowset_writer_context.h" +#include "storage/segment/segment_index_file_cache_loader.h" +#include "storage/storage_engine.h" +#include "util/threadpool.h" +#include "util/time.h" + +namespace doris { +namespace { + +using segment_v2::SegmentIndexFileCacheLoadContext; +using segment_v2::SegmentIndexFileCacheLoadReason; +using segment_v2::SegmentIndexFileCachePreloadTask; + +constexpr int64_t kIndexOnlyTabletId = 10005; +constexpr int64_t kIndexOnlyPartitionId = 10006; +constexpr int64_t kIndexOnlyTabletSchemaHash = 10007; +constexpr std::string_view kTestDir = "ut_dir/cloud_file_cache_write_index_only_e2e"; +constexpr std::string_view kTmpDir = "ut_dir/cloud_file_cache_write_index_only_e2e/tmp"; + +bool has_suffix(std::string_view value, std::string_view suffix) { + return value.size() >= suffix.size() && value.substr(value.size() - suffix.size()) == suffix; +} + +struct CreatedS3File { + std::string path; + FileType file_type; + bool is_s3_writer = false; + bool has_cache_builder = false; + bool write_file_cache = false; + bool allow_adaptive_file_cache_write = false; + uint64_t approximate_bytes_to_write = 0; + size_t bytes_appended = 0; + bool saw_put_object = false; +}; + +struct ObservedIndexPreload { + SegmentIndexFileCacheLoadReason reason; + uint32_t segment_id = 0; + std::string segment_path; + uint64_t range_offset = 0; + uint64_t range_size = 0; + uint64_t segment_file_size = 0; + int closed_segment_files = 0; +}; + +struct WriterFlushCounters { + int vertical_segment_writer_flush = 0; + int segment_writer_final_flush = 0; +}; + +struct S3WriteCounters { + int segment_file_close = 0; + int open_file = 0; +}; + +} // namespace + +class CloudFileCacheWriteIndexOnlyConfigTest : public testing::Test { +protected: + void SetUp() override { + _origin_index_only = config::enable_file_cache_write_index_file_only; + _origin_enable_file_cache = config::enable_file_cache; + _origin_cloud_unique_id = config::cloud_unique_id; + } + + void TearDown() override { + auto sp = SyncPoint::get_instance(); + sp->disable_processing(); + sp->clear_all_call_backs(); + sp->clear_trace(); + + config::enable_file_cache_write_index_file_only = _origin_index_only; + config::enable_file_cache = _origin_enable_file_cache; + config::cloud_unique_id = _origin_cloud_unique_id; + } + +private: + bool _origin_index_only = false; + bool _origin_enable_file_cache = false; + std::string _origin_cloud_unique_id; +}; + +class CloudFileCacheWriteIndexOnlyTest : public testing::Test { +protected: + void SetUp() override { + _origin_index_only = config::enable_file_cache_write_index_file_only; + _origin_enable_file_cache = config::enable_file_cache; + _origin_enable_flush_file_cache_async = config::enable_flush_file_cache_async; + _origin_cloud_unique_id = config::cloud_unique_id; + _origin_enable_packed_file = config::enable_packed_file; + _origin_enable_vertical_segment_writer = config::enable_vertical_segment_writer; + + config::enable_file_cache_write_index_file_only = true; + config::enable_file_cache = true; + config::enable_flush_file_cache_async = false; + config::cloud_unique_id = "cloud_file_cache_write_index_only_e2e"; + config::enable_packed_file = false; + config::enable_vertical_segment_writer = true; + + ASSERT_TRUE(io::global_local_filesystem()->delete_directory(std::string(kTestDir)).ok()); + ASSERT_TRUE(io::global_local_filesystem()->create_directory(std::string(kTestDir)).ok()); + + _origin_file_cache_factory = ExecEnv::GetInstance()->_file_cache_factory; + _owned_file_cache_factory = std::make_unique(); + ExecEnv::GetInstance()->_file_cache_factory = _owned_file_cache_factory.get(); + io::FileCacheSettings settings; + settings.query_queue_size = 64 * 1024 * 1024; + settings.query_queue_elements = 64; + settings.index_queue_size = 64 * 1024 * 1024; + settings.index_queue_elements = 64; + settings.disposable_queue_size = 1024 * 1024; + settings.disposable_queue_elements = 16; + settings.capacity = 128 * 1024 * 1024; + settings.max_file_block_size = 1024 * 1024; + settings.max_query_cache_size = 0; + settings.storage = "memory"; + ASSERT_TRUE(io::FileCacheFactory::instance()->create_file_cache("memory", settings).ok()); + + std::vector paths; + paths.emplace_back(std::string(kTmpDir), -1); + auto tmp_file_dirs = std::make_unique(paths); + ASSERT_TRUE(tmp_file_dirs->init().ok()); + ExecEnv::GetInstance()->set_tmp_file_dir(std::move(tmp_file_dirs)); + + if (ExecEnv::GetInstance()->s3_file_upload_thread_pool() == nullptr) { + std::unique_ptr pool; + ASSERT_TRUE(ThreadPoolBuilder("cloud_file_cache_write_index_only_s3_upload") + .set_min_threads(1) + .set_max_threads(4) + .build(&pool) + .ok()); + ExecEnv::GetInstance()->_s3_file_upload_thread_pool = std::move(pool); + _created_s3_upload_pool = true; + } + + S3Conf s3_conf; + s3_conf.client_conf.ak = "fake_ak"; + s3_conf.client_conf.sk = "fake_sk"; + s3_conf.client_conf.endpoint = "fake_s3_endpoint"; + s3_conf.client_conf.region = "fake_s3_region"; + s3_conf.bucket = "fake_s3_bucket"; + s3_conf.prefix = "cloud_file_cache_write_index_only_e2e"; + auto fs = io::S3FileSystem::create(std::move(s3_conf), "cloud-file-cache-index-only-ut-fs"); + ASSERT_TRUE(fs.has_value()) << fs.error(); + _remote_fs = fs.value(); + + auto engine = std::make_unique(EngineOptions {}); + _engine = engine.get(); + ExecEnv::GetInstance()->set_storage_engine(std::move(engine)); + } + + void TearDown() override { + auto sp = SyncPoint::get_instance(); + sp->disable_processing(); + sp->clear_all_call_backs(); + sp->clear_trace(); + + _remote_fs.reset(); + _engine = nullptr; + ExecEnv::GetInstance()->set_storage_engine(nullptr); + + if (_created_s3_upload_pool) { + ExecEnv::GetInstance()->_s3_file_upload_thread_pool.reset(); + } + ExecEnv::GetInstance()->set_tmp_file_dir(nullptr); + + _owned_file_cache_factory.reset(); + ExecEnv::GetInstance()->_file_cache_factory = _origin_file_cache_factory; + + ASSERT_TRUE(io::global_local_filesystem()->delete_directory(std::string(kTestDir)).ok()); + + config::enable_file_cache_write_index_file_only = _origin_index_only; + config::enable_file_cache = _origin_enable_file_cache; + config::enable_flush_file_cache_async = _origin_enable_flush_file_cache_async; + config::cloud_unique_id = _origin_cloud_unique_id; + config::enable_packed_file = _origin_enable_packed_file; + config::enable_vertical_segment_writer = _origin_enable_vertical_segment_writer; + } + + TabletSchemaSPtr create_schema(bool with_inverted_index = false, + InvertedIndexStorageFormatPB inverted_index_storage_format = + InvertedIndexStorageFormatPB::V2) { + TabletSchemaPB tablet_schema_pb; + tablet_schema_pb.set_keys_type(KeysType::DUP_KEYS); + tablet_schema_pb.set_num_short_key_columns(1); + tablet_schema_pb.set_num_rows_per_row_block(1024); + tablet_schema_pb.set_compress_kind(COMPRESS_NONE); + tablet_schema_pb.set_next_column_unique_id(3); + + auto* key = tablet_schema_pb.add_column(); + key->set_unique_id(1); + key->set_name("k1"); + key->set_type("INT"); + key->set_is_key(true); + key->set_length(4); + key->set_index_length(4); + key->set_is_nullable(false); + key->set_is_bf_column(false); + + auto* value = tablet_schema_pb.add_column(); + value->set_unique_id(2); + value->set_name("v1"); + value->set_type("INT"); + value->set_is_key(false); + value->set_length(4); + value->set_index_length(4); + value->set_is_nullable(false); + value->set_is_bf_column(false); + + if (with_inverted_index) { + tablet_schema_pb.set_inverted_index_storage_format(inverted_index_storage_format); + auto* index = tablet_schema_pb.add_index(); + index->set_index_id(10000); + index->set_index_name("v1_idx"); + index->set_index_type(IndexType::INVERTED); + index->add_col_unique_id(2); + } + + auto tablet_schema = std::make_shared(); + tablet_schema->init_from_pb(tablet_schema_pb); + return tablet_schema; + } + + RowsetWriterContext create_context(const TabletSchemaSPtr& tablet_schema, + DataWriteType write_type = DataWriteType::TYPE_DEFAULT, + ReaderType compaction_type = ReaderType::UNKNOWN) { + RowsetId rowset_id; + rowset_id.init(_next_rowset_id++); + + RowsetWriterContext context; + context.rowset_id = rowset_id; + context.tablet_id = kIndexOnlyTabletId; + context.partition_id = kIndexOnlyPartitionId; + context.tablet_schema_hash = kIndexOnlyTabletSchemaHash; + context.rowset_type = BETA_ROWSET; + context.tablet_schema = tablet_schema; + context.rowset_state = VISIBLE; + context.version = Version(_next_rowset_id, _next_rowset_id); + context.segments_overlap = OVERLAPPING; + context.max_rows_per_segment = UINT32_MAX; + context.data_dir = nullptr; + context.write_type = write_type; + context.compaction_type = compaction_type; + context.storage_resource = StorageResource(_remote_fs); + context.tablet_path = "unused_local_tablet_path"; + context.write_file_cache = true; + context.approximate_bytes_to_write = 4096; + context.newest_write_timestamp = UnixSeconds(); + context.allow_packed_file = false; + context.encrypt_algorithm = EncryptionAlgorithmPB::PLAINTEXT; + return context; + } + + Block create_full_block(const TabletSchemaSPtr& tablet_schema, int32_t start_key = 1) { + auto block = tablet_schema->create_block(); + auto columns = std::move(block).mutate_columns(); + for (int32_t i = 0; i < 8; ++i) { + int32_t key = start_key + i; + int32_t value = key * 10; + columns[0]->insert_data(reinterpret_cast(&key), sizeof(key)); + columns[1]->insert_data(reinterpret_cast(&value), sizeof(value)); + } + block.set_columns(std::move(columns)); + return block; + } + + Block create_column_block(const TabletSchemaSPtr& tablet_schema, + const std::vector& column_ids, int32_t row_count = 8, + int32_t start_key = 1) { + auto block = tablet_schema->create_block(column_ids); + auto columns = std::move(block).mutate_columns(); + for (int32_t i = 0; i < row_count; ++i) { + int32_t key = start_key + i; + int32_t value = column_ids[0] == 0 ? key : key * 10; + columns[0]->insert_data(reinterpret_cast(&value), sizeof(value)); + } + block.set_columns(std::move(columns)); + return block; + } + + void install_observers( + std::vector* observed, std::vector* created_files, + int* preload_task_count, WriterFlushCounters* writer_flush_counters, + S3WriteCounters* s3_write_counters, SyncPoint::CallbackGuard* load_guard, + SyncPoint::CallbackGuard* task_guard, SyncPoint::CallbackGuard* vertical_writer_guard, + SyncPoint::CallbackGuard* segment_writer_guard, + SyncPoint::CallbackGuard* s3_client_guard, SyncPoint::CallbackGuard* s3_put_guard, + SyncPoint::CallbackGuard* create_file_guard, SyncPoint::CallbackGuard* close_file_guard, + SyncPoint::CallbackGuard* s3_open_file_guard) { + auto sp = SyncPoint::get_instance(); + sp->clear_all_call_backs(); + sp->enable_processing(); + sp->set_call_back( + "s3_client_factory::create", + [](auto&& args) { + auto* ret = try_any_cast_ret>(args); + ret->second = true; + }, + s3_client_guard); + sp->set_call_back( + "S3FileWriter::_put_object", + [created_files](auto&& args) { + auto* writer = try_any_cast(args[0]); + for (auto& file : *created_files) { + if (has_suffix(writer->path().native(), file.path)) { + file.bytes_appended = writer->bytes_appended(); + file.saw_put_object = true; + break; + } + } + auto* should_return = try_any_cast(args.back()); + *should_return = true; + }, + s3_put_guard); + sp->set_call_back( + "BaseBetaRowsetWriter::_create_file_writer", + [created_files](auto&& args) { + auto* path = try_any_cast(args[0]); + auto* file_type = try_any_cast(args[1]); + auto* writer = try_any_cast(args[2]); + auto* opts = try_any_cast(args[3]); + created_files->push_back(CreatedS3File { + .path = *path, + .file_type = *file_type, + .is_s3_writer = dynamic_cast(writer) != nullptr, + .has_cache_builder = writer->cache_builder() != nullptr, + .write_file_cache = opts->write_file_cache, + .allow_adaptive_file_cache_write = + opts->allow_adaptive_file_cache_write, + .approximate_bytes_to_write = opts->approximate_bytes_to_write}); + }, + create_file_guard); + sp->set_call_back( + "SegmentFileCollection::close_file_writer", + [s3_write_counters](auto&& args) { + auto* writer = try_any_cast(args[0]); + if (has_suffix(writer->path().native(), ".dat")) { + ++s3_write_counters->segment_file_close; + } + }, + close_file_guard); + sp->set_call_back( + "S3FileSystem::open_file_internal", + [s3_write_counters](auto&& /*args*/) { ++s3_write_counters->open_file; }, + s3_open_file_guard); + sp->set_call_back( + "SegmentIndexFileCacheLoader::preload_segment_indexes_to_file_cache", + [preload_task_count](auto&& args) { + auto* tasks = + try_any_cast*>( + args[1]); + *preload_task_count += static_cast(tasks->size()); + }, + task_guard); + sp->set_call_back( + "SegmentIndexFileCacheLoader::load_segment_index_to_file_cache", + [observed, s3_write_counters](auto&& args) { + auto* ctx = try_any_cast(args[0]); + auto* io_ctx = try_any_cast(args[1]); + EXPECT_TRUE(io_ctx->is_index_data); + EXPECT_TRUE(io_ctx->is_dryrun); + EXPECT_FALSE(io_ctx->is_warmup); + observed->push_back(ObservedIndexPreload { + .reason = ctx->reason, + .segment_id = ctx->segment_id, + .segment_path = ctx->segment_path, + .range_offset = ctx->range.offset, + .range_size = ctx->range.size, + .segment_file_size = ctx->segment_file_size, + .closed_segment_files = s3_write_counters->segment_file_close}); + + auto* ret = try_any_cast_ret(args); + ret->first = Status::OK(); + ret->second = true; + }, + load_guard); + sp->set_call_back( + "SegmentFlusher::flush_vertical_segment_writer", + [writer_flush_counters](auto&& args) { + static_cast(try_any_cast(args[0])); + ++writer_flush_counters->vertical_segment_writer_flush; + }, + vertical_writer_guard); + sp->set_call_back( + "VerticalBetaRowsetWriter::final_flush_segment_writer", + [writer_flush_counters](auto&& args) { + static_cast(try_any_cast(args[0])); + ++writer_flush_counters->segment_writer_final_flush; + }, + segment_writer_guard); + } + + void expect_segment_write_bypasses_file_cache(const std::vector& created_files) { + bool saw_segment_file = false; + for (const auto& file : created_files) { + if (file.file_type != FileType::SEGMENT_FILE) { + continue; + } + saw_segment_file = true; + EXPECT_TRUE(file.is_s3_writer) << file.path; + EXPECT_FALSE(file.write_file_cache) << file.path; + EXPECT_FALSE(file.allow_adaptive_file_cache_write) << file.path; + EXPECT_EQ(file.approximate_bytes_to_write, 0) << file.path; + EXPECT_FALSE(file.has_cache_builder) << file.path; + + auto cache_key = std::filesystem::path(file.path).filename().native(); + auto cache_blocks = io::FileCacheFactory::instance()->get_cache_data_by_path(cache_key); + EXPECT_TRUE(cache_blocks.empty()) << file.path; + } + EXPECT_TRUE(saw_segment_file); + } + + void expect_inverted_index_writes_file_cache(const std::vector& created_files) { + bool saw_index_file = false; + for (const auto& file : created_files) { + if (file.file_type != FileType::INVERTED_INDEX_FILE) { + continue; + } + saw_index_file = true; + EXPECT_TRUE(file.is_s3_writer) << file.path; + EXPECT_TRUE(file.write_file_cache) << file.path; + EXPECT_FALSE(file.allow_adaptive_file_cache_write) << file.path; + EXPECT_EQ(file.approximate_bytes_to_write, 0) << file.path; + EXPECT_TRUE(file.has_cache_builder) << file.path; + EXPECT_TRUE(file.saw_put_object) << file.path; + + auto cache_key = std::filesystem::path(file.path).filename().native(); + auto cache_blocks = io::FileCacheFactory::instance()->get_cache_data_by_path(cache_key); + if (file.bytes_appended == 0) { + EXPECT_TRUE(cache_blocks.empty()) << file.path; + } else { + EXPECT_FALSE(cache_blocks.empty()) + << file.path << ", bytes_appended=" << file.bytes_appended; + } + } + EXPECT_TRUE(saw_index_file); + } + + void expect_loader_open_file_is_mocked_out(const S3WriteCounters& s3_write_counters) { + EXPECT_EQ(s3_write_counters.open_file, 0); + } + + StorageEngine* _engine = nullptr; + std::shared_ptr _remote_fs; + + io::FileCacheFactory* _origin_file_cache_factory = nullptr; + std::unique_ptr _owned_file_cache_factory; + bool _created_s3_upload_pool = false; + + bool _origin_index_only = false; + bool _origin_enable_file_cache = false; + bool _origin_enable_flush_file_cache_async = false; + std::string _origin_cloud_unique_id; + bool _origin_enable_packed_file = false; + bool _origin_enable_vertical_segment_writer = false; + int64_t _next_rowset_id = 20000; +}; + +TEST_F(CloudFileCacheWriteIndexOnlyConfigTest, FileWriterOptionsKeepLegacyWhenIndexOnlyDisabled) { + config::enable_file_cache_write_index_file_only = false; + + RowsetWriterContext context; + context.write_file_cache = false; + context.approximate_bytes_to_write = 12345; + + auto segment_opts = context.get_file_writer_options(FileType::SEGMENT_FILE); + EXPECT_FALSE(segment_opts.write_file_cache); + EXPECT_TRUE(segment_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(segment_opts.approximate_bytes_to_write, 12345); + + auto index_opts = context.get_file_writer_options(FileType::INVERTED_INDEX_FILE); + EXPECT_FALSE(index_opts.write_file_cache); + EXPECT_TRUE(index_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(index_opts.approximate_bytes_to_write, 12345); +} + +TEST_F(CloudFileCacheWriteIndexOnlyConfigTest, IndexOnlyOptionsSplitSegmentAndInvertedIndexFiles) { + config::enable_file_cache_write_index_file_only = true; + + RowsetWriterContext context; + context.write_file_cache = false; + context.approximate_bytes_to_write = 12345; + + auto segment_opts = context.get_file_writer_options(FileType::SEGMENT_FILE); + EXPECT_FALSE(segment_opts.write_file_cache); + EXPECT_FALSE(segment_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(segment_opts.approximate_bytes_to_write, 0); + + auto index_opts = context.get_file_writer_options(FileType::INVERTED_INDEX_FILE); + EXPECT_TRUE(index_opts.write_file_cache); + EXPECT_FALSE(index_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(index_opts.approximate_bytes_to_write, 0); +} + +TEST_F(CloudFileCacheWriteIndexOnlyConfigTest, + IndexOnlyIgnoresRequestWriteFileCacheForSegmentData) { + config::enable_file_cache_write_index_file_only = true; + + RowsetWriterContext context; + context.write_file_cache = true; + context.approximate_bytes_to_write = 12345; + + auto segment_opts = context.get_file_writer_options(FileType::SEGMENT_FILE); + EXPECT_FALSE(segment_opts.write_file_cache); + EXPECT_FALSE(segment_opts.allow_adaptive_file_cache_write); + + auto index_opts = context.get_file_writer_options(FileType::INVERTED_INDEX_FILE); + EXPECT_TRUE(index_opts.write_file_cache); + EXPECT_FALSE(index_opts.allow_adaptive_file_cache_write); +} + +TEST_F(CloudFileCacheWriteIndexOnlyConfigTest, SegmentIndexFileCacheLoaderSkipsWhenConfigDisabled) { + config::enable_file_cache = false; + config::enable_file_cache_write_index_file_only = true; + + segment_v2::SegmentIndexFileCacheLoadContext context; + context.range = {.offset = 1, .size = 1}; + context.segment_file_size = 2; + + EXPECT_TRUE(segment_v2::SegmentIndexFileCacheLoader::load_segment_index_to_file_cache(context) + .ok()); +} + +TEST_F(CloudFileCacheWriteIndexOnlyConfigTest, + SegmentIndexFileCacheLoaderSkipsEmptyRangeBeforeOpenFile) { + config::enable_file_cache = true; + config::enable_file_cache_write_index_file_only = true; + config::cloud_unique_id = "cloud_file_cache_empty_range_ut"; + + S3Conf s3_conf; + s3_conf.client_conf.ak = "fake_ak"; + s3_conf.client_conf.sk = "fake_sk"; + s3_conf.client_conf.endpoint = "fake_s3_endpoint"; + s3_conf.client_conf.region = "fake_s3_region"; + s3_conf.bucket = "fake_s3_bucket"; + s3_conf.prefix = "cloud_file_cache_empty_range_ut"; + auto fs = io::S3FileSystem::create(std::move(s3_conf), "cloud-file-cache-empty-range-ut-fs"); + ASSERT_TRUE(fs.has_value()) << fs.error(); + + int open_file_count = 0; + SyncPoint::CallbackGuard s3_open_file_guard; + auto sp = SyncPoint::get_instance(); + sp->clear_all_call_backs(); + sp->enable_processing(); + sp->set_call_back( + "S3FileSystem::open_file_internal", + [&open_file_count](auto&& /*args*/) { + ++open_file_count; + ADD_FAILURE() << "empty range should return before opening segment"; + }, + &s3_open_file_guard); + + segment_v2::SegmentIndexFileCacheLoadContext context; + context.fs = fs.value(); + context.segment_path = "empty_range_should_not_open.dat"; + context.tablet_id = kIndexOnlyTabletId; + context.segment_file_size = 2; + + EXPECT_TRUE(segment_v2::SegmentIndexFileCacheLoader::load_segment_index_to_file_cache(context) + .ok()); + EXPECT_EQ(open_file_count, 0); +} + +TEST_F(CloudFileCacheWriteIndexOnlyTest, + LoadUsesVerticalSegmentWriterAndPreloadsAfterAllSegmentFilesClosed) { + auto tablet_schema = create_schema(true); + RowsetWriterContext context = create_context(tablet_schema); + + std::vector observed; + std::vector created_files; + int preload_task_count = 0; + WriterFlushCounters writer_flush_counters; + S3WriteCounters s3_write_counters; + SyncPoint::CallbackGuard load_guard; + SyncPoint::CallbackGuard task_guard; + SyncPoint::CallbackGuard vertical_writer_guard; + SyncPoint::CallbackGuard segment_writer_guard; + SyncPoint::CallbackGuard s3_client_guard; + SyncPoint::CallbackGuard s3_put_guard; + SyncPoint::CallbackGuard create_file_guard; + SyncPoint::CallbackGuard close_file_guard; + SyncPoint::CallbackGuard s3_open_file_guard; + install_observers(&observed, &created_files, &preload_task_count, &writer_flush_counters, + &s3_write_counters, &load_guard, &task_guard, &vertical_writer_guard, + &segment_writer_guard, &s3_client_guard, &s3_put_guard, &create_file_guard, + &close_file_guard, &s3_open_file_guard); + + auto writer_result = RowsetFactory::create_rowset_writer(*_engine, context, false); + ASSERT_TRUE(writer_result.has_value()) << writer_result.error(); + auto rowset_writer = std::move(writer_result).value(); + + auto block = create_full_block(tablet_schema, 1); + auto st = rowset_writer->flush_single_block(&block); + ASSERT_TRUE(st.ok()) << st; + auto second_block = create_full_block(tablet_schema, 100); + st = rowset_writer->flush_single_block(&second_block); + ASSERT_TRUE(st.ok()) << st; + + RowsetSharedPtr rowset; + st = rowset_writer->build(rowset); + ASSERT_TRUE(st.ok()) << st; + ASSERT_NE(rowset, nullptr); + EXPECT_EQ(rowset->rowset_meta()->num_segments(), 2); + + EXPECT_EQ(writer_flush_counters.vertical_segment_writer_flush, 2); + EXPECT_EQ(writer_flush_counters.segment_writer_final_flush, 0); + EXPECT_EQ(preload_task_count, 2); + ASSERT_EQ(observed.size(), 4); + std::vector ranges_per_segment(2, 0); + for (const auto& item : observed) { + ASSERT_LT(item.segment_id, ranges_per_segment.size()); + ++ranges_per_segment[item.segment_id]; + EXPECT_EQ(item.reason, SegmentIndexFileCacheLoadReason::LOAD); + EXPECT_EQ(item.segment_path, context.segment_path(item.segment_id)); + EXPECT_GT(item.range_offset, 0); + EXPECT_GT(item.range_size, 0); + EXPECT_LE(item.range_offset + item.range_size, item.segment_file_size); + EXPECT_EQ(item.closed_segment_files, 2); + } + EXPECT_EQ(ranges_per_segment[0], 2); + EXPECT_EQ(ranges_per_segment[1], 2); + + expect_segment_write_bypasses_file_cache(created_files); + expect_inverted_index_writes_file_cache(created_files); + expect_loader_open_file_is_mocked_out(s3_write_counters); +} + +TEST_F(CloudFileCacheWriteIndexOnlyTest, + VerticalCompactionUsesSegmentWriterAndPreloadsAfterAllSegmentFilesClosed) { + auto tablet_schema = create_schema(true); + RowsetWriterContext context = create_context(tablet_schema, DataWriteType::TYPE_COMPACTION, + ReaderType::READER_CUMULATIVE_COMPACTION); + + std::vector observed; + std::vector created_files; + int preload_task_count = 0; + WriterFlushCounters writer_flush_counters; + S3WriteCounters s3_write_counters; + SyncPoint::CallbackGuard load_guard; + SyncPoint::CallbackGuard task_guard; + SyncPoint::CallbackGuard vertical_writer_guard; + SyncPoint::CallbackGuard segment_writer_guard; + SyncPoint::CallbackGuard s3_client_guard; + SyncPoint::CallbackGuard s3_put_guard; + SyncPoint::CallbackGuard create_file_guard; + SyncPoint::CallbackGuard close_file_guard; + SyncPoint::CallbackGuard s3_open_file_guard; + install_observers(&observed, &created_files, &preload_task_count, &writer_flush_counters, + &s3_write_counters, &load_guard, &task_guard, &vertical_writer_guard, + &segment_writer_guard, &s3_client_guard, &s3_put_guard, &create_file_guard, + &close_file_guard, &s3_open_file_guard); + + auto writer_result = RowsetFactory::create_rowset_writer(*_engine, context, true); + ASSERT_TRUE(writer_result.has_value()) << writer_result.error(); + auto rowset_writer = std::move(writer_result).value(); + + std::vector key_column_ids = {0}; + auto key_block = create_column_block(tablet_schema, key_column_ids, 8, 1); + auto st = rowset_writer->add_columns(&key_block, key_column_ids, true, 4, false); + ASSERT_TRUE(st.ok()) << st; + auto second_key_block = create_column_block(tablet_schema, key_column_ids, 8, 100); + st = rowset_writer->add_columns(&second_key_block, key_column_ids, true, 4, false); + ASSERT_TRUE(st.ok()) << st; + st = rowset_writer->flush_columns(true); + ASSERT_TRUE(st.ok()) << st; + + std::vector value_column_ids = {1}; + auto value_block = create_column_block(tablet_schema, value_column_ids, 16, 1); + st = rowset_writer->add_columns(&value_block, value_column_ids, false, UINT32_MAX, false); + ASSERT_TRUE(st.ok()) << st; + st = rowset_writer->flush_columns(false); + ASSERT_TRUE(st.ok()) << st; + st = rowset_writer->final_flush(); + ASSERT_TRUE(st.ok()) << st; + + RowsetSharedPtr rowset; + st = rowset_writer->build(rowset); + ASSERT_TRUE(st.ok()) << st; + ASSERT_NE(rowset, nullptr); + EXPECT_EQ(rowset->rowset_meta()->num_segments(), 2); + + EXPECT_EQ(writer_flush_counters.vertical_segment_writer_flush, 0); + EXPECT_EQ(writer_flush_counters.segment_writer_final_flush, 2); + EXPECT_EQ(preload_task_count, 2); + ASSERT_EQ(observed.size(), 6); + std::vector ranges_per_segment(2, 0); + for (const auto& item : observed) { + ASSERT_LT(item.segment_id, ranges_per_segment.size()); + ++ranges_per_segment[item.segment_id]; + EXPECT_EQ(item.reason, SegmentIndexFileCacheLoadReason::CUMULATIVE_COMPACTION); + EXPECT_EQ(item.segment_path, context.segment_path(item.segment_id)); + EXPECT_GT(item.range_offset, 0); + EXPECT_GT(item.range_size, 0); + EXPECT_LE(item.range_offset + item.range_size, item.segment_file_size); + EXPECT_EQ(item.closed_segment_files, 2); + } + EXPECT_EQ(ranges_per_segment[0], 3); + EXPECT_EQ(ranges_per_segment[1], 3); + + expect_segment_write_bypasses_file_cache(created_files); + expect_inverted_index_writes_file_cache(created_files); + expect_loader_open_file_is_mocked_out(s3_write_counters); +} + +TEST_F(CloudFileCacheWriteIndexOnlyTest, + VerticalCompactionV1InvertedIndexUsesIndexOnlyFileWriterOptions) { + auto tablet_schema = create_schema(true, InvertedIndexStorageFormatPB::V1); + RowsetWriterContext context = create_context(tablet_schema, DataWriteType::TYPE_COMPACTION, + ReaderType::READER_CUMULATIVE_COMPACTION); + + std::vector observed; + std::vector created_files; + int preload_task_count = 0; + WriterFlushCounters writer_flush_counters; + S3WriteCounters s3_write_counters; + SyncPoint::CallbackGuard load_guard; + SyncPoint::CallbackGuard task_guard; + SyncPoint::CallbackGuard vertical_writer_guard; + SyncPoint::CallbackGuard segment_writer_guard; + SyncPoint::CallbackGuard s3_client_guard; + SyncPoint::CallbackGuard s3_put_guard; + SyncPoint::CallbackGuard create_file_guard; + SyncPoint::CallbackGuard close_file_guard; + SyncPoint::CallbackGuard s3_open_file_guard; + install_observers(&observed, &created_files, &preload_task_count, &writer_flush_counters, + &s3_write_counters, &load_guard, &task_guard, &vertical_writer_guard, + &segment_writer_guard, &s3_client_guard, &s3_put_guard, &create_file_guard, + &close_file_guard, &s3_open_file_guard); + int index_writer_create_count = 0; + SyncPoint::CallbackGuard index_writer_create_guard; + SyncPoint::get_instance()->set_call_back( + "BaseBetaRowsetWriter::create_inverted_index_file_writer", + [&index_writer_create_count](auto&& args) { + static_cast(try_any_cast(args[0])); + ++index_writer_create_count; + }, + &index_writer_create_guard); + + auto writer_result = RowsetFactory::create_rowset_writer(*_engine, context, true); + ASSERT_TRUE(writer_result.has_value()) << writer_result.error(); + auto rowset_writer = std::move(writer_result).value(); + + std::vector key_column_ids = {0}; + auto key_block = create_column_block(tablet_schema, key_column_ids, 8, 1); + auto st = rowset_writer->add_columns(&key_block, key_column_ids, true, 4, false); + ASSERT_TRUE(st.ok()) << st; + + EXPECT_EQ(writer_flush_counters.vertical_segment_writer_flush, 0); + EXPECT_EQ(writer_flush_counters.segment_writer_final_flush, 0); + EXPECT_EQ(preload_task_count, 0); + EXPECT_EQ(index_writer_create_count, 1); + expect_segment_write_bypasses_file_cache(created_files); +} + +} // namespace doris diff --git a/be/test/storage/compaction_file_cache_test.cpp b/be/test/storage/compaction_file_cache_test.cpp index 41303a9ff9e9be..765df31bf1c7b3 100644 --- a/be/test/storage/compaction_file_cache_test.cpp +++ b/be/test/storage/compaction_file_cache_test.cpp @@ -34,17 +34,22 @@ class CompactionFileCacheTest : public testing::Test { public: void SetUp() override { // Save original configuration + _orig_index_file_only_config = config::enable_file_cache_write_index_file_only; _orig_base_config = config::enable_file_cache_write_base_compaction_index_only; _orig_cumu_config = config::enable_file_cache_write_cumu_compaction_index_only; + + config::enable_file_cache_write_index_file_only = false; } void TearDown() override { // Restore original configuration + config::enable_file_cache_write_index_file_only = _orig_index_file_only_config; config::enable_file_cache_write_base_compaction_index_only = _orig_base_config; config::enable_file_cache_write_cumu_compaction_index_only = _orig_cumu_config; } private: + bool _orig_index_file_only_config; bool _orig_base_config; bool _orig_cumu_config; }; @@ -63,7 +68,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_IndexOnly_False_IndexFile) { ctx.compaction_output_write_index_only = false; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should be true EXPECT_TRUE(opts.write_file_cache); @@ -79,7 +84,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_IndexOnly_False_DataFile) { ctx.compaction_output_write_index_only = false; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should be true EXPECT_TRUE(opts.write_file_cache); @@ -95,7 +100,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_IndexOnly_True_IndexFile) { ctx.compaction_output_write_index_only = true; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should be true (index files are always cached) EXPECT_TRUE(opts.write_file_cache); @@ -111,7 +116,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_IndexOnly_True_DataFile) { ctx.compaction_output_write_index_only = true; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should be false (data files are NOT cached when index-only is enabled) EXPECT_FALSE(opts.write_file_cache); @@ -131,7 +136,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_IndexOnly_False_IndexFile) { ctx.compaction_output_write_index_only = false; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should be true EXPECT_TRUE(opts.write_file_cache); @@ -147,7 +152,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_IndexOnly_False_DataFile) { ctx.compaction_output_write_index_only = false; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should be true EXPECT_TRUE(opts.write_file_cache); @@ -163,7 +168,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_IndexOnly_True_IndexFile) { ctx.compaction_output_write_index_only = true; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should be true (index files are always cached) EXPECT_TRUE(opts.write_file_cache); @@ -179,7 +184,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_IndexOnly_True_DataFile) { ctx.compaction_output_write_index_only = true; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should be false (data files are NOT cached when index-only is enabled) EXPECT_FALSE(opts.write_file_cache); @@ -199,7 +204,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_WriteCacheFalse_IndexOnly_False_I ctx.compaction_output_write_index_only = false; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); @@ -215,7 +220,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_WriteCacheFalse_IndexOnly_False_D ctx.compaction_output_write_index_only = false; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); @@ -231,7 +236,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_WriteCacheFalse_IndexOnly_True_In ctx.compaction_output_write_index_only = true; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should remain false (base cache setting takes precedence) EXPECT_FALSE(opts.write_file_cache); @@ -247,7 +252,7 @@ TEST_F(CompactionFileCacheTest, BaseCompaction_WriteCacheFalse_IndexOnly_True_Da ctx.compaction_output_write_index_only = true; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); @@ -267,7 +272,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_WriteCacheFalse_IndexOnly_False_I ctx.compaction_output_write_index_only = false; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); @@ -283,7 +288,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_WriteCacheFalse_IndexOnly_False_D ctx.compaction_output_write_index_only = false; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); @@ -299,7 +304,7 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_WriteCacheFalse_IndexOnly_True_In ctx.compaction_output_write_index_only = true; // Test: Get file writer options for index file - auto opts = ctx.get_file_writer_options(true); + auto opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); // Verify: write_file_cache should remain false (base cache setting takes precedence) EXPECT_FALSE(opts.write_file_cache); @@ -315,12 +320,33 @@ TEST_F(CompactionFileCacheTest, CumuCompaction_WriteCacheFalse_IndexOnly_True_Da ctx.compaction_output_write_index_only = true; // Test: Get file writer options for data file - auto opts = ctx.get_file_writer_options(false); + auto opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); // Verify: write_file_cache should remain false EXPECT_FALSE(opts.write_file_cache); } +TEST_F(CompactionFileCacheTest, GlobalIndexFileOnlyTakesPrecedenceOverCompactionConfigs) { + config::enable_file_cache_write_index_file_only = true; + config::enable_file_cache_write_base_compaction_index_only = true; + config::enable_file_cache_write_cumu_compaction_index_only = true; + + RowsetWriterContext ctx; + ctx.write_file_cache = false; + ctx.compaction_output_write_index_only = false; + ctx.approximate_bytes_to_write = 12345; + + auto segment_opts = ctx.get_file_writer_options(FileType::SEGMENT_FILE); + EXPECT_FALSE(segment_opts.write_file_cache); + EXPECT_FALSE(segment_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(segment_opts.approximate_bytes_to_write, 0); + + auto index_opts = ctx.get_file_writer_options(FileType::INVERTED_INDEX_FILE); + EXPECT_TRUE(index_opts.write_file_cache); + EXPECT_FALSE(index_opts.allow_adaptive_file_cache_write); + EXPECT_EQ(index_opts.approximate_bytes_to_write, 0); +} + // ============================================================================ // Tests for should_enable_compaction_cache_index_only function // ============================================================================ diff --git a/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only.groovy b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only.groovy new file mode 100644 index 00000000000000..647d0936d80d01 --- /dev/null +++ b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only.groovy @@ -0,0 +1,344 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.Http + +suite("test_file_cache_write_index_file_only", "docker") { + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(1) + + options.beConfigs += [ + 'enable_file_cache=true', + 'enable_file_cache_write_index_file_only=true', + 'enable_file_cache_adaptive_write=true', + 'enable_file_cache_keep_base_compaction_output=true', + 'file_cache_keep_base_compaction_output_min_hit_ratio=0', + 'enable_flush_file_cache_async=false', + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'enable_packed_file=false', + 'file_cache_each_block_size=4096', + 'file_cache_path=[{"path":"/opt/apache-doris/be/storage/file_cache","total_size":83886080,"query_limit":83886080}]' + ] + + def tableName = "test_file_cache_write_index_file_only" + def loadRows = 1200 + def compactionBatchRows = 200 + def indexTokenPrefix = "tok" + def backendIdToIp = [:] + def backendIdToHttpPort = [:] + def backendIdToBrpcPort = [:] + + def clearFileCache = { beHost, beHttpPort -> + def result = Http.GET("http://${beHost}:${beHttpPort}/api/file_cache?op=clear&sync=true", true) + logger.info("clear file cache result: ${result}") + } + + def uniqueIndexToken = { int seed -> + def alphabet = "abcdefghijklmnopqrstuvwxyz" + def value = seed + def suffix = new StringBuilder() + for (int i = 0; i < 5; i++) { + suffix.append(alphabet.charAt(value % alphabet.length())) + value = (int) (value / alphabet.length()) + } + return "${indexTokenPrefix}${suffix}" + } + + def insertRows = { int batch, int rowCount -> + def data = new StringBuilder() + (0.. + def payload = (1..24).collect { java.util.UUID.randomUUID().toString() }.join("") + def indexToken = uniqueIndexToken(batch * 100000 + idx) + data.append("${batch * 100000 + idx + 1}\t") + data.append("${rowCount - idx}\t") + data.append("tag_${batch}_${idx}\t") + data.append("quick brown profile text ${indexToken} row ${batch} ${idx}\t") + data.append("${payload}\n") + } + streamLoad { + table "${tableName}" + set 'column_separator', '\t' + set 'columns', 'id,sort_key,tag,body,payload' + inputText data.toString() + time 60000 + check { result, exception, startTime, endTime -> + if (exception != null) { + throw exception + } + logger.info("stream load result: ${result}") + def json = parseJson(result) + assert json.Status.toString().equalsIgnoreCase("success") : result + assert json.NumberLoadedRows.toString().toInteger() == rowCount : result + } + } + } + + def waitForLatestAlterOnTableFinish = { String table -> + for (int i = 0; i < 300; i++) { + def alterResult = sql """ + SHOW ALTER TABLE COLUMN WHERE TableName = "${table}" ORDER BY CreateTime DESC LIMIT 1 + """ + logger.info("latest alter table column result: ${alterResult}") + if (alterResult.size() > 0) { + def alterResultString = alterResult.toString() + assert !alterResultString.contains("CANCELLED") : + "schema change was cancelled, result=${alterResultString}" + if (alterResultString.contains("FINISHED")) { + sleep(3000) + return + } + } + sleep(1000) + } + assert false : "wait alter table column timeout, table=${table}" + } + + def parseProfileCounterValue = { String valueText -> + def exact = (valueText =~ /\((\d+)\)/) + if (exact.find()) { + return exact.group(1).toLong() + } + def number = (valueText =~ /([0-9]+(?:\.[0-9]+)?)\s*(B|KB|MB|GB)?/) + if (!number.find()) { + return 0L + } + BigDecimal value = new BigDecimal(number.group(1)) + long multiplier = 1L + if (number.group(2) == "KB") { + multiplier = 1024L + } else if (number.group(2) == "MB") { + multiplier = 1024L * 1024L + } else if (number.group(2) == "GB") { + multiplier = 1024L * 1024L * 1024L + } + return (value * multiplier).toLong() + } + + def sumProfileCounter = { String profileString, String counterName -> + long total = 0 + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + + java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def sumProfileSummaryCounter = { String profileString, String counterName -> + long total = 0 + def counterRegex = counterName.split(/\s+/).collect { + java.util.regex.Pattern.quote(it) + }.join("\\s+") + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + counterRegex + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def assertIndexOnlyProfile = { String profileString, String label, + boolean requireSegmentFooterCacheBytes, boolean requireDataPageRemoteBytes -> + def invertedIndexCacheBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromCache") + def invertedIndexRemoteBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromRemote") + def segmentFooterIndexCacheBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromCache") + def segmentFooterIndexRemoteBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromRemote") + def totalCacheBytes = sumProfileCounter(profileString, "BytesScannedFromCache") + def totalRemoteBytes = sumProfileCounter(profileString, "BytesScannedFromRemote") + def parallelFragmentExecInstanceNum = + sumProfileSummaryCounter(profileString, "Parallel Fragment Exec Instance Num") + def totalInstancesNum = sumProfileSummaryCounter(profileString, "Total Instances Num") + def classifiedRemoteBytes = invertedIndexRemoteBytes + segmentFooterIndexRemoteBytes + def dataPageRemoteBytes = totalRemoteBytes - classifiedRemoteBytes + + logger.info("${label} profile counters: invertedIndexCacheBytes=${invertedIndexCacheBytes}, " + + "invertedIndexRemoteBytes=${invertedIndexRemoteBytes}, " + + "segmentFooterIndexCacheBytes=${segmentFooterIndexCacheBytes}, " + + "segmentFooterIndexRemoteBytes=${segmentFooterIndexRemoteBytes}, " + + "parallelFragmentExecInstanceNum=${parallelFragmentExecInstanceNum}, " + + "totalInstancesNum=${totalInstancesNum}, totalCacheBytes=${totalCacheBytes}, " + + "totalRemoteBytes=${totalRemoteBytes}, dataPageRemoteBytes=${dataPageRemoteBytes}") + + assert parallelFragmentExecInstanceNum == 1L : + "${label}: expected profile query to use one parallel fragment exec instance, profile=${profileString}" + assert invertedIndexCacheBytes > 0 : + "${label}: expected independent inverted index file to be read from local file cache, profile=${profileString}" + assert invertedIndexCacheBytes > 4096 : + "${label}: expected query to read more than the initial inverted index buffer, profile=${profileString}" + assert invertedIndexRemoteBytes == 0L : + "${label}: independent inverted index should not be read from remote storage, profile=${profileString}" + if (requireSegmentFooterCacheBytes) { + assert segmentFooterIndexCacheBytes > 0 : + "${label}: expected segment footer/index to be read from local file cache, profile=${profileString}" + } + assert segmentFooterIndexRemoteBytes == 0L : + "${label}: segment footer/index should not be read from remote storage, profile=${profileString}" + if (requireDataPageRemoteBytes) { + assert totalRemoteBytes > classifiedRemoteBytes : + "${label}: expected ordinary data pages to be read from remote storage, " + + "totalRemoteBytes=${totalRemoteBytes}, classifiedRemoteBytes=" + + "${classifiedRemoteBytes}, dataPageRemoteBytes=${dataPageRemoteBytes}, " + + "profile=${profileString}" + } + // File cache is block-aligned, so footer/index cache ranges may include nearby data pages. + // Only assert data-page remote reads for stages where the query reaches uncached data pages. + } + + def logRowsetsLayout = { String label -> + def currentTablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert currentTablets.size() == 1 + def currentTablet = currentTablets[0] + def tabletStatus = show_tablet_compaction(currentTablet) + def rowsets = tabletStatus.rowsets == null ? [] : tabletStatus.rowsets + def staleRowsets = tabletStatus.stale_rowsets == null ? [] : tabletStatus.stale_rowsets + logger.info("${label} rowsets layout: tabletId=${currentTablet.TabletId}, " + + "rowsetCount=${rowsets.size()}, staleRowsetCount=${staleRowsets.size()}, " + + "rowsets=${rowsets}, staleRowsets=${staleRowsets}") + } + + docker(options) { + getBackendIpHttpAndBrpcPort(backendIdToIp, backendIdToHttpPort, backendIdToBrpcPort) + + sql """ DROP TABLE IF EXISTS ${tableName} FORCE """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + sort_key INT, + tag VARCHAR(64), + body VARCHAR(2048), + payload STRING, + INDEX body_idx(body) USING INVERTED PROPERTIES("parser" = "english") COMMENT '' + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V2" + ) + """ + + def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert tablets.size() == 1 + def tablet = tablets[0] + def beHost = backendIdToIp[tablet.BackendId] + def beHttpPort = backendIdToHttpPort[tablet.BackendId] + + clearFileCache(beHost, beHttpPort) + insertRows(1, loadRows) + sql """ SYNC """ + + def loadProfileTag = "file_cache_write_index_only_load_profile" + def loadProfileChecked = false + profile(loadProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + sql """ SET inverted_index_max_expansions = 4096 """ + run { + logRowsetsLayout("before load query") + sql """ /* ${loadProfileTag} */ SELECT id + 1 FROM ${tableName} WHERE body MATCH_REGEXP '^${indexTokenPrefix}.*' ORDER BY sort_key LIMIT 10 """ + sleep(500) + } + check { profileString, exception -> + loadProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertIndexOnlyProfile(profileString, "load", true, true) + } + } + assert loadProfileChecked : "load profile check was not executed" + + (2..3).each { batch -> + insertRows(batch, compactionBatchRows) + } + sql """ SYNC """ + + clearFileCache(beHost, beHttpPort) + trigger_and_wait_compaction(tableName, "full") + + def compactionProfileTag = "file_cache_write_index_only_compaction_profile" + def compactionProfileChecked = false + profile(compactionProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + sql """ SET inverted_index_max_expansions = 4096 """ + run { + logRowsetsLayout("before full compaction query") + sql """ + /* ${compactionProfileTag} */ + SELECT COUNT(*), SUM(id), SUM(sort_key), SUM(LENGTH(payload)) + FROM ${tableName} + WHERE body MATCH_REGEXP '^${indexTokenPrefix}.*' + """ + sleep(500) + } + check { profileString, exception -> + compactionProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertIndexOnlyProfile(profileString, "full compaction", false, true) + } + } + assert compactionProfileChecked : "full compaction profile check was not executed" + + clearFileCache(beHost, beHttpPort) + sql """ ALTER TABLE ${tableName} ADD COLUMN sc_key BIGINT KEY DEFAULT "0" AFTER id """ + waitForLatestAlterOnTableFinish(tableName) + sql """ SYNC """ + + def schemaChangeProfileTag = "file_cache_write_index_only_schema_change_profile" + def schemaChangeProfileChecked = false + profile(schemaChangeProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + sql """ SET inverted_index_max_expansions = 4096 """ + run { + logRowsetsLayout("before heavy schema change query") + sql """ + /* ${schemaChangeProfileTag} */ + SELECT COUNT(*), SUM(id), SUM(sort_key), SUM(LENGTH(payload)) + FROM ${tableName} + WHERE body MATCH_REGEXP '^${indexTokenPrefix}.*' + """ + sleep(500) + } + check { profileString, exception -> + schemaChangeProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertIndexOnlyProfile(profileString, "heavy schema change", false, true) + } + } + assert schemaChangeProfileChecked : "heavy schema change profile check was not executed" + } +} diff --git a/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_compaction_segment_data.groovy b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_compaction_segment_data.groovy new file mode 100644 index 00000000000000..7e7dd7b7071632 --- /dev/null +++ b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_compaction_segment_data.groovy @@ -0,0 +1,239 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.Http + +suite("test_file_cache_write_index_file_only_compaction_segment_data", "docker") { + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(1) + + options.beConfigs += [ + 'enable_file_cache=true', + 'enable_file_cache_write_index_file_only=true', + 'enable_file_cache_adaptive_write=true', + 'enable_file_cache_keep_base_compaction_output=true', + 'file_cache_keep_base_compaction_output_min_hit_ratio=0', + 'enable_flush_file_cache_async=false', + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'enable_packed_file=false', + 'file_cache_each_block_size=4096', + 'file_cache_path=[{"path":"/opt/apache-doris/be/storage/file_cache","total_size":83886080,"query_limit":83886080}]' + ] + + def tableName = "test_file_cache_write_index_file_only_compaction_segment_data" + def batchRows = 1200 + def backendIdToIp = [:] + def backendIdToHttpPort = [:] + def backendIdToBrpcPort = [:] + + def clearFileCache = { beHost, beHttpPort -> + def result = Http.GET("http://${beHost}:${beHttpPort}/api/file_cache?op=clear&sync=true", true) + logger.info("clear file cache result: ${result}") + } + + def insertRows = { int batch, int rowCount -> + def data = new StringBuilder() + (0.. + def payload = (1..32).collect { java.util.UUID.randomUUID().toString() }.join("") + def rowId = batch * 100000 + idx + 1 + data.append("${rowId}\t") + data.append("${idx}\t") + data.append("tag_${batch}_${idx}\t") + data.append("scan all segment data row ${batch} ${idx}\t") + data.append("${payload}\n") + } + streamLoad { + table "${tableName}" + set 'column_separator', '\t' + set 'columns', 'id,sort_key,tag,body,payload' + inputText data.toString() + time 60000 + check { result, exception, startTime, endTime -> + if (exception != null) { + throw exception + } + logger.info("stream load result: ${result}") + def json = parseJson(result) + assert json.Status.toString().equalsIgnoreCase("success") : result + assert json.NumberLoadedRows.toString().toInteger() == rowCount : result + } + } + } + + def parseProfileCounterValue = { String valueText -> + def exact = (valueText =~ /\((\d+)\)/) + if (exact.find()) { + return exact.group(1).toLong() + } + def number = (valueText =~ /([0-9]+(?:\.[0-9]+)?)\s*(B|KB|MB|GB)?/) + if (!number.find()) { + return 0L + } + BigDecimal value = new BigDecimal(number.group(1)) + long multiplier = 1L + if (number.group(2) == "KB") { + multiplier = 1024L + } else if (number.group(2) == "MB") { + multiplier = 1024L * 1024L + } else if (number.group(2) == "GB") { + multiplier = 1024L * 1024L * 1024L + } + return (value * multiplier).toLong() + } + + def sumProfileCounter = { String profileString, String counterName -> + long total = 0 + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + + java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def sumProfileSummaryCounter = { String profileString, String counterName -> + long total = 0 + def counterRegex = counterName.split(/\s+/).collect { + java.util.regex.Pattern.quote(it) + }.join("\\s+") + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + counterRegex + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def logRowsetsLayout = { String label -> + def currentTablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert currentTablets.size() == 1 + def currentTablet = currentTablets[0] + def tabletStatus = show_tablet_compaction(currentTablet) + def rowsets = tabletStatus.rowsets == null ? [] : tabletStatus.rowsets + def staleRowsets = tabletStatus.stale_rowsets == null ? [] : tabletStatus.stale_rowsets + logger.info("${label} rowsets layout: tabletId=${currentTablet.TabletId}, " + + "rowsetCount=${rowsets.size()}, staleRowsetCount=${staleRowsets.size()}, " + + "rowsets=${rowsets}, staleRowsets=${staleRowsets}") + } + + def assertCompactionSegmentDataProfile = { String profileString, String label -> + def invertedIndexCacheBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromCache") + def invertedIndexRemoteBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromRemote") + def segmentFooterIndexCacheBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromCache") + def segmentFooterIndexRemoteBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromRemote") + def totalCacheBytes = sumProfileCounter(profileString, "BytesScannedFromCache") + def totalRemoteBytes = sumProfileCounter(profileString, "BytesScannedFromRemote") + def parallelFragmentExecInstanceNum = + sumProfileSummaryCounter(profileString, "Parallel Fragment Exec Instance Num") + def totalInstancesNum = sumProfileSummaryCounter(profileString, "Total Instances Num") + def dataPageRemoteBytes = totalRemoteBytes - segmentFooterIndexRemoteBytes + + logger.info("${label} profile counters: invertedIndexCacheBytes=${invertedIndexCacheBytes}, " + + "invertedIndexRemoteBytes=${invertedIndexRemoteBytes}, " + + "segmentFooterIndexCacheBytes=${segmentFooterIndexCacheBytes}, " + + "segmentFooterIndexRemoteBytes=${segmentFooterIndexRemoteBytes}, " + + "parallelFragmentExecInstanceNum=${parallelFragmentExecInstanceNum}, " + + "totalInstancesNum=${totalInstancesNum}, totalCacheBytes=${totalCacheBytes}, " + + "totalRemoteBytes=${totalRemoteBytes}, dataPageRemoteBytes=${dataPageRemoteBytes}") + + assert parallelFragmentExecInstanceNum == 1L : + "${label}: expected profile query to use one parallel fragment exec instance, profile=${profileString}" + assert invertedIndexCacheBytes == 0L : + "${label}: table has no inverted index, profile=${profileString}" + assert invertedIndexRemoteBytes == 0L : + "${label}: table has no inverted index, profile=${profileString}" + assert segmentFooterIndexRemoteBytes == 0L : + "${label}: segment footer/index should not be read from remote storage, profile=${profileString}" + assert dataPageRemoteBytes > 0L : + "${label}: expected compacted segment data pages to be read from remote storage, " + + "profile=${profileString}" + assert totalRemoteBytes > totalCacheBytes : + "${label}: expected compacted segment data to remain mostly uncached; " + + "totalRemoteBytes=${totalRemoteBytes}, totalCacheBytes=${totalCacheBytes}, " + + "profile=${profileString}" + } + + docker(options) { + getBackendIpHttpAndBrpcPort(backendIdToIp, backendIdToHttpPort, backendIdToBrpcPort) + + sql """ DROP TABLE IF EXISTS ${tableName} FORCE """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + sort_key INT, + tag VARCHAR(64), + body VARCHAR(2048), + payload STRING + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true" + ) + """ + + def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert tablets.size() == 1 + def tablet = tablets[0] + def beHost = backendIdToIp[tablet.BackendId] + def beHttpPort = backendIdToHttpPort[tablet.BackendId] + + clearFileCache(beHost, beHttpPort) + (1..3).each { batch -> + insertRows(batch, batchRows) + } + sql """ SYNC """ + + clearFileCache(beHost, beHttpPort) + trigger_and_wait_compaction(tableName, "full") + + def compactionProfileTag = "file_cache_write_index_only_compaction_segment_data_profile" + def compactionProfileChecked = false + profile(compactionProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + run { + logRowsetsLayout("before full scan query after compaction") + sql """ + /* ${compactionProfileTag} */ + SELECT COUNT(DISTINCT id), SUM(id), SUM(sort_key), SUM(LENGTH(tag)), + SUM(LENGTH(body)), SUM(LENGTH(payload)) + FROM ${tableName} + """ + sleep(500) + } + check { profileString, exception -> + compactionProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertCompactionSegmentDataProfile(profileString, "full compaction segment data") + } + } + assert compactionProfileChecked : "full compaction segment data profile check was not executed" + } +} diff --git a/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_packed_file.groovy b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_packed_file.groovy new file mode 100644 index 00000000000000..d7d797aa9f4454 --- /dev/null +++ b/regression-test/suites/cloud_p0/cache/write_index_only/test_file_cache_write_index_file_only_packed_file.groovy @@ -0,0 +1,293 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.Http + +suite("test_file_cache_write_index_file_only_packed_file", "docker") { + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(1) + + options.beConfigs += [ + 'enable_file_cache=true', + 'enable_file_cache_write_index_file_only=true', + 'enable_file_cache_adaptive_write=true', + 'enable_file_cache_keep_base_compaction_output=true', + 'file_cache_keep_base_compaction_output_min_hit_ratio=0', + 'enable_flush_file_cache_async=false', + 'file_cache_enter_disk_resource_limit_mode_percent=99', + 'enable_evict_file_cache_in_advance=false', + 'enable_packed_file=true', + 'small_file_threshold_bytes=104857600', + 'packed_file_size_threshold_bytes=104857600', + 'file_cache_each_block_size=4096', + 'file_cache_path=[{"path":"/opt/apache-doris/be/storage/file_cache","total_size":83886080,"query_limit":83886080}]' + ] + + def tableName = "test_file_cache_write_index_file_only_packed_file" + def loadRows = 1200 + def compactionBatchRows = 200 + def indexTokenPrefix = "tok" + def backendIdToIp = [:] + def backendIdToHttpPort = [:] + def backendIdToBrpcPort = [:] + + def clearFileCache = { beHost, beHttpPort -> + def result = Http.GET("http://${beHost}:${beHttpPort}/api/file_cache?op=clear&sync=true", true) + logger.info("clear file cache result: ${result}") + } + + def uniqueIndexToken = { int seed -> + def alphabet = "abcdefghijklmnopqrstuvwxyz" + def value = seed + def suffix = new StringBuilder() + for (int i = 0; i < 5; i++) { + suffix.append(alphabet.charAt(value % alphabet.length())) + value = (int) (value / alphabet.length()) + } + return "${indexTokenPrefix}${suffix}" + } + + def insertRows = { int batch, int rowCount -> + def data = new StringBuilder() + (0.. + def payload = (1..24).collect { java.util.UUID.randomUUID().toString() }.join("") + def indexToken = uniqueIndexToken(batch * 100000 + idx) + data.append("${batch * 100000 + idx + 1}\t") + data.append("${rowCount - idx}\t") + data.append("tag_${batch}_${idx}\t") + data.append("quick brown profile text ${indexToken} row ${batch} ${idx}\t") + data.append("${payload}\n") + } + streamLoad { + table "${tableName}" + set 'column_separator', '\t' + set 'columns', 'id,sort_key,tag,body,payload' + inputText data.toString() + time 60000 + check { result, exception, startTime, endTime -> + if (exception != null) { + throw exception + } + logger.info("stream load result: ${result}") + def json = parseJson(result) + assert json.Status.toString().equalsIgnoreCase("success") : result + assert json.NumberLoadedRows.toString().toInteger() == rowCount : result + } + } + } + + def parseProfileCounterValue = { String valueText -> + def exact = (valueText =~ /\((\d+)\)/) + if (exact.find()) { + return exact.group(1).toLong() + } + def number = (valueText =~ /([0-9]+(?:\.[0-9]+)?)\s*(B|KB|MB|GB)?/) + if (!number.find()) { + return 0L + } + BigDecimal value = new BigDecimal(number.group(1)) + long multiplier = 1L + if (number.group(2) == "KB") { + multiplier = 1024L + } else if (number.group(2) == "MB") { + multiplier = 1024L * 1024L + } else if (number.group(2) == "GB") { + multiplier = 1024L * 1024L * 1024L + } + return (value * multiplier).toLong() + } + + def sumProfileCounter = { String profileString, String counterName -> + long total = 0 + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + + java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def sumProfileSummaryCounter = { String profileString, String counterName -> + long total = 0 + def counterRegex = counterName.split(/\s+/).collect { + java.util.regex.Pattern.quote(it) + }.join("\\s+") + def matcher = (profileString =~ ("(?m)^\\s*(?:-\\s*)?" + counterRegex + ":\\s+([^\\n]+)")) + while (matcher.find()) { + total += parseProfileCounterValue(matcher.group(1).toString()) + } + return total + } + + def assertIndexOnlyProfile = { String profileString, String label, + boolean requireSegmentFooterCacheBytes, boolean requireDataPageRemoteBytes -> + def invertedIndexCacheBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromCache") + def invertedIndexRemoteBytes = + sumProfileCounter(profileString, "InvertedIndexBytesScannedFromRemote") + def segmentFooterIndexCacheBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromCache") + def segmentFooterIndexRemoteBytes = + sumProfileCounter(profileString, "SegmentFooterIndexBytesScannedFromRemote") + def totalCacheBytes = sumProfileCounter(profileString, "BytesScannedFromCache") + def totalRemoteBytes = sumProfileCounter(profileString, "BytesScannedFromRemote") + def parallelFragmentExecInstanceNum = + sumProfileSummaryCounter(profileString, "Parallel Fragment Exec Instance Num") + def totalInstancesNum = sumProfileSummaryCounter(profileString, "Total Instances Num") + def classifiedRemoteBytes = invertedIndexRemoteBytes + segmentFooterIndexRemoteBytes + def dataPageRemoteBytes = totalRemoteBytes - classifiedRemoteBytes + + logger.info("${label} profile counters: invertedIndexCacheBytes=${invertedIndexCacheBytes}, " + + "invertedIndexRemoteBytes=${invertedIndexRemoteBytes}, " + + "segmentFooterIndexCacheBytes=${segmentFooterIndexCacheBytes}, " + + "segmentFooterIndexRemoteBytes=${segmentFooterIndexRemoteBytes}, " + + "parallelFragmentExecInstanceNum=${parallelFragmentExecInstanceNum}, " + + "totalInstancesNum=${totalInstancesNum}, totalCacheBytes=${totalCacheBytes}, " + + "totalRemoteBytes=${totalRemoteBytes}, dataPageRemoteBytes=${dataPageRemoteBytes}") + + assert parallelFragmentExecInstanceNum == 1L : + "${label}: expected profile query to use one parallel fragment exec instance, profile=${profileString}" + assert invertedIndexCacheBytes > 0 : + "${label}: expected independent inverted index file to be read from local file cache, profile=${profileString}" + assert invertedIndexCacheBytes > 4096 : + "${label}: expected query to read more than the initial inverted index buffer, profile=${profileString}" + assert invertedIndexRemoteBytes == 0L : + "${label}: independent inverted index should not be read from remote storage, profile=${profileString}" + if (requireSegmentFooterCacheBytes) { + assert segmentFooterIndexCacheBytes > 0 : + "${label}: expected segment footer/index to be read from local file cache, profile=${profileString}" + } + assert segmentFooterIndexRemoteBytes == 0L : + "${label}: segment footer/index should not be read from remote storage, profile=${profileString}" + if (requireDataPageRemoteBytes) { + assert totalRemoteBytes > classifiedRemoteBytes : + "${label}: expected ordinary data pages to be read from remote storage, " + + "totalRemoteBytes=${totalRemoteBytes}, classifiedRemoteBytes=" + + "${classifiedRemoteBytes}, dataPageRemoteBytes=${dataPageRemoteBytes}, " + + "profile=${profileString}" + } + // File cache is block-aligned, so footer/index cache ranges may include nearby data pages. + // Only assert data-page remote reads for stages where the query reaches uncached data pages. + } + + def logRowsetsLayout = { String label -> + def currentTablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert currentTablets.size() == 1 + def currentTablet = currentTablets[0] + def tabletStatus = show_tablet_compaction(currentTablet) + def rowsets = tabletStatus.rowsets == null ? [] : tabletStatus.rowsets + def staleRowsets = tabletStatus.stale_rowsets == null ? [] : tabletStatus.stale_rowsets + logger.info("${label} rowsets layout: tabletId=${currentTablet.TabletId}, " + + "rowsetCount=${rowsets.size()}, staleRowsetCount=${staleRowsets.size()}, " + + "rowsets=${rowsets}, staleRowsets=${staleRowsets}") + } + + docker(options) { + getBackendIpHttpAndBrpcPort(backendIdToIp, backendIdToHttpPort, backendIdToBrpcPort) + + sql """ DROP TABLE IF EXISTS ${tableName} FORCE """ + sql """ + CREATE TABLE ${tableName} ( + id INT, + sort_key INT, + tag VARCHAR(64), + body VARCHAR(2048), + payload STRING, + INDEX body_idx(body) USING INVERTED PROPERTIES("parser" = "english") COMMENT '' + ) + DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true", + "inverted_index_storage_format" = "V2" + ) + """ + + def tablets = sql_return_maparray """ SHOW TABLETS FROM ${tableName} """ + assert tablets.size() == 1 + def tablet = tablets[0] + def beHost = backendIdToIp[tablet.BackendId] + def beHttpPort = backendIdToHttpPort[tablet.BackendId] + + clearFileCache(beHost, beHttpPort) + insertRows(1, loadRows) + sql """ SYNC """ + + def loadProfileTag = "file_cache_write_index_only_packed_file_load_profile" + def loadProfileChecked = false + profile(loadProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + sql """ SET inverted_index_max_expansions = 4096 """ + run { + logRowsetsLayout("before packed file load query") + sql """ /* ${loadProfileTag} */ SELECT id + 1 FROM ${tableName} WHERE body MATCH_REGEXP '^${indexTokenPrefix}.*' ORDER BY sort_key LIMIT 10 """ + sleep(500) + } + check { profileString, exception -> + loadProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertIndexOnlyProfile(profileString, "packed file load", true, true) + } + } + assert loadProfileChecked : "packed file load profile check was not executed" + + (2..3).each { batch -> + insertRows(batch, compactionBatchRows) + } + sql """ SYNC """ + + clearFileCache(beHost, beHttpPort) + trigger_and_wait_compaction(tableName, "full") + + def compactionProfileTag = "file_cache_write_index_only_packed_file_compaction_profile" + def compactionProfileChecked = false + profile(compactionProfileTag) { + sql """ SET enable_profile = true """ + sql """ SET profile_level = 2 """ + sql """ SET parallel_pipeline_task_num = 1 """ + sql """ SET inverted_index_max_expansions = 4096 """ + run { + logRowsetsLayout("before packed file full compaction query") + sql """ + /* ${compactionProfileTag} */ + SELECT COUNT(*), SUM(id), SUM(sort_key), SUM(LENGTH(payload)) + FROM ${tableName} + WHERE body MATCH_REGEXP '^${indexTokenPrefix}.*' + """ + sleep(500) + } + check { profileString, exception -> + compactionProfileChecked = true + if (exception != null) { + throw exception + } + logger.info("profile snippet: {}", profileString.take(3000)) + assertIndexOnlyProfile(profileString, "packed file full compaction", false, true) + } + } + assert compactionProfileChecked : "packed file full compaction profile check was not executed" + } +}