Coverage Report

Created: 2026-07-30 17:55

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/packed_file_system.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "io/fs/packed_file_system.h"
19
20
#include <string>
21
#include <string_view>
22
#include <utility>
23
24
#include "cloud/config.h"
25
#include "common/status.h"
26
#include "io/fs/file_reader.h"
27
#include "io/fs/packed_file_reader.h"
28
#include "io/fs/packed_file_writer.h"
29
30
namespace doris::io {
31
32
namespace {
33
34
// Only keep packed-file aggregation for the first segment in a rowset.
35
// The path handled here is expected to look like:
36
//   local/remote segment file: .../{rowset_id}_{segment_id}.dat
37
//   local/remote index file:   .../{rowset_id}_{segment_id}.idx
38
// The .idx case here is V2 only. V1 inverted-index tablets are filtered before
39
// PackedFileSystem is enabled, so V1 names like
40
// {rowset_id}_{segment_id}_{index_id}@{suffix}.idx never reach this helper.
41
// Multi-segment rowsets usually come from large loads or memory-pressure flushes,
42
// and continuing to buffer later segments in packed files can amplify memory usage.
43
// Non-rowset file names keep the legacy behavior to avoid changing unrelated callers.
44
1.00k
bool should_use_packed_writer(std::string_view file_name, int64_t first_segment_id) {
45
1.00k
    DORIS_CHECK_GE(first_segment_id, 0);
46
1.00k
    constexpr std::string_view kSegmentSuffix = ".dat";
47
1.00k
    constexpr std::string_view kIndexSuffix = ".idx";
48
49
1.00k
    size_t suffix_len = 0;
50
1.00k
    if (file_name.ends_with(kSegmentSuffix)) {
51
4
        suffix_len = kSegmentSuffix.size();
52
1.00k
    } else if (file_name.ends_with(kIndexSuffix)) {
53
2
        suffix_len = kIndexSuffix.size();
54
1.00k
    } else {
55
1.00k
        return true;
56
1.00k
    }
57
58
6
    file_name.remove_suffix(suffix_len);
59
6
    size_t pos = file_name.rfind('_');
60
6
    if (pos == std::string_view::npos || pos + 1 >= file_name.size()) {
61
0
        return true;
62
0
    }
63
64
6
    return file_name.substr(pos + 1) == std::to_string(first_segment_id);
65
6
}
66
67
} // namespace
68
69
PackedFileSystem::PackedFileSystem(FileSystemSPtr inner_fs, PackedAppendContext append_info)
70
1.01k
        : FileSystem(inner_fs->id(), inner_fs->type()),
71
1.01k
          _inner_fs(std::move(inner_fs)),
72
1.01k
          _append_info(std::move(append_info)) {
73
1.01k
    if (_append_info.resource_id.empty() && _inner_fs != nullptr) {
74
0
        _append_info.resource_id = _inner_fs->id();
75
0
    }
76
1.01k
}
77
78
PackedFileSystem::PackedFileSystem(FileSystemSPtr inner_fs,
79
                                   std::unordered_map<std::string, PackedSliceLocation> index_map,
80
                                   PackedAppendContext append_info)
81
1.00k
        : FileSystem(inner_fs->id(), inner_fs->type()),
82
1.00k
          _inner_fs(std::move(inner_fs)),
83
1.00k
          _index_map(std::move(index_map)),
84
1.00k
          _append_info(std::move(append_info)) {
85
1.00k
    if (_append_info.resource_id.empty() && _inner_fs != nullptr) {
86
0
        _append_info.resource_id = _inner_fs->id();
87
0
    }
88
1.00k
    _index_map_initialized = true;
89
1.00k
}
90
91
Status PackedFileSystem::create_file_impl(const Path& file, FileWriterPtr* writer,
92
1.00k
                                          const FileWriterOptions* opts) {
93
    // Create file using inner file system
94
1.00k
    FileWriterPtr inner_writer;
95
1.00k
    RETURN_IF_ERROR(_inner_fs->create_file(file, &inner_writer, opts));
96
97
1.00k
    if (!should_use_packed_writer(file.filename().native(), _append_info.first_segment_id)) {
98
3
        *writer = std::move(inner_writer);
99
3
        return Status::OK();
100
3
    }
101
102
1.00k
    auto append_info = _append_info;
103
1.00k
    if (config::enable_file_cache_write_index_file_only && opts != nullptr) {
104
0
        append_info.write_file_cache = opts->write_file_cache;
105
0
    }
106
107
    // Wrap with PackedFileWriter
108
1.00k
    *writer = std::make_unique<PackedFileWriter>(std::move(inner_writer), file, append_info);
109
1.00k
    return Status::OK();
110
1.00k
}
111
112
Status PackedFileSystem::open_file_impl(const Path& file, FileReaderSPtr* reader,
113
1.00k
                                        const FileReaderOptions* opts) {
114
    // Check if this file is in a packed file
115
1.00k
    std::string file_path = file.native();
116
1.00k
    auto it = _index_map.find(file_path);
117
1.00k
    PackedSliceLocation manager_location;
118
1.00k
    const PackedSliceLocation* packed_location = nullptr;
119
1.00k
    if (it != _index_map.end()) {
120
1.00k
        packed_location = &it->second;
121
1.00k
    } else if (PackedFileManager::instance()
122
1
                       ->get_packed_slice_location(file_path, &manager_location)
123
1
                       .ok()) {
124
0
        packed_location = &manager_location;
125
0
    }
126
127
1.00k
    if (packed_location != nullptr) {
128
        // File is in packed file, open packed file and wrap with PackedFileReader
129
1.00k
        const auto& index = *packed_location;
130
1.00k
        FileReaderSPtr inner_reader;
131
132
        // Create options for opening the packed file
133
        // Disable cache at this layer - we'll add cache wrapper around PackedFileReader instead
134
        // This ensures cache key is based on segment path, not packed file path
135
1.00k
        FileReaderOptions inner_opts = opts ? *opts : FileReaderOptions();
136
1.00k
        inner_opts.file_size = index.packed_file_size;
137
1.00k
        inner_opts.cache_type = FileCachePolicy::NO_CACHE;
138
139
1.00k
        VLOG_DEBUG << "open packed file: " << index.packed_file_path << ", file: " << file.native()
140
0
                   << ", offset: " << index.offset << ", size: " << index.size
141
0
                   << ", packed_file_size: " << index.packed_file_size;
142
1.00k
        RETURN_IF_ERROR(
143
1.00k
                _inner_fs->open_file(Path(index.packed_file_path), &inner_reader, &inner_opts));
144
145
        // Create PackedFileReader with segment path
146
        // PackedFileReader.path() returns segment path, not packed file path
147
1.00k
        auto packed_reader = std::make_shared<PackedFileReader>(std::move(inner_reader), file,
148
1.00k
                                                                index.offset, index.size);
149
150
        // If cache is requested, wrap PackedFileReader with CachedRemoteFileReader
151
        // This ensures:
152
        // 1. Cache key = hash(segment_path.filename()) - matches cleanup key
153
        // 2. Cache size = segment size - correct boundary
154
        // 3. Each segment has independent cache entry - no interference during cleanup
155
1.00k
        if (opts && opts->cache_type != FileCachePolicy::NO_CACHE) {
156
999
            FileReaderOptions cache_opts = *opts;
157
999
            cache_opts.file_size = index.size; // Use segment size for cache
158
999
            *reader = DORIS_TRY(create_cached_file_reader(packed_reader, cache_opts));
159
999
        } else {
160
2
            *reader = packed_reader;
161
2
        }
162
1.00k
    } else {
163
1
        RETURN_IF_ERROR(_inner_fs->open_file(file, reader, opts));
164
1
    }
165
1.00k
    return Status::OK();
166
1.00k
}
167
168
4
Status PackedFileSystem::exists_impl(const Path& path, bool* res) const {
169
4
    VLOG_DEBUG << "packed file system exist, rowset id " << _append_info.rowset_id;
170
4
    if (!_index_map_initialized) {
171
1
        return Status::InternalError("PackedFileSystem index map is not initialized");
172
1
    }
173
3
    const auto it = _index_map.find(path.native());
174
3
    if (it != _index_map.end()) {
175
1
        return _inner_fs->exists(Path(it->second.packed_file_path), res);
176
1
    }
177
2
    return _inner_fs->exists(path, res);
178
3
}
179
180
5
Status PackedFileSystem::file_size_impl(const Path& file, int64_t* file_size) const {
181
5
    if (!_index_map_initialized) {
182
1
        return Status::InternalError("PackedFileSystem index map is not initialized");
183
1
    }
184
4
    const auto it = _index_map.find(file.native());
185
4
    if (it != _index_map.end()) {
186
3
        *file_size = it->second.size;
187
3
        return Status::OK();
188
3
    }
189
1
    return _inner_fs->file_size(file, file_size);
190
4
}
191
192
} // namespace doris::io