/src/rocksdb/db/version_edit_handler.cc
Line | Count | Source |
1 | | // Copyright (c) 2011-present, Facebook, Inc. All rights reserved. |
2 | | // This source code is licensed under both the GPLv2 (found in the |
3 | | // COPYING file in the root directory) and Apache 2.0 License |
4 | | // (found in the LICENSE.Apache file in the root directory). |
5 | | // |
6 | | // Copyright (c) 2011 The LevelDB Authors. All rights reserved. |
7 | | // Use of this source code is governed by a BSD-style license that can be |
8 | | // found in the LICENSE file. See the AUTHORS file for names of contributors. |
9 | | |
10 | | #include "db/version_edit_handler.h" |
11 | | |
12 | | #include <cinttypes> |
13 | | #include <sstream> |
14 | | |
15 | | #include "db/blob/blob_file_reader.h" |
16 | | #include "db/blob/blob_source.h" |
17 | | #include "db/version_edit.h" |
18 | | #include "logging/logging.h" |
19 | | #include "monitoring/persistent_stats_history.h" |
20 | | #include "util/udt_util.h" |
21 | | |
22 | | namespace ROCKSDB_NAMESPACE { |
23 | | |
24 | | void VersionEditHandlerBase::Iterate(log::Reader& reader, |
25 | 44.8k | Status* log_read_status) { |
26 | 44.8k | Slice record; |
27 | 44.8k | std::string scratch; |
28 | 44.8k | assert(log_read_status); |
29 | 44.8k | assert(log_read_status->ok()); |
30 | | |
31 | 44.8k | [[maybe_unused]] size_t recovered_edits = 0; |
32 | 44.8k | Status s = Initialize(); |
33 | 437k | while (reader.LastRecordEnd() < max_manifest_read_size_ && s.ok() && |
34 | 437k | reader.ReadRecord(&record, &scratch) && log_read_status->ok()) { |
35 | 393k | VersionEdit edit; |
36 | 393k | s = edit.DecodeFrom(record); |
37 | 393k | if (s.ok()) { |
38 | 393k | s = read_buffer_.AddEdit(&edit); |
39 | 393k | } |
40 | 393k | if (s.ok()) { |
41 | 393k | ColumnFamilyData* cfd = nullptr; |
42 | 393k | if (edit.IsInAtomicGroup()) { |
43 | 0 | if (read_buffer_.IsFull()) { |
44 | 0 | last_valid_record_end_ = reader.LastRecordEnd(); |
45 | 0 | s = OnAtomicGroupReplayBegin(); |
46 | 0 | for (size_t i = 0; s.ok() && i < read_buffer_.replay_buffer().size(); |
47 | 0 | i++) { |
48 | 0 | auto& e = read_buffer_.replay_buffer()[i]; |
49 | 0 | s = ApplyVersionEdit(e, &cfd); |
50 | 0 | if (s.ok()) { |
51 | 0 | recovered_edits++; |
52 | 0 | } |
53 | 0 | } |
54 | 0 | if (s.ok()) { |
55 | 0 | read_buffer_.Clear(); |
56 | 0 | s = OnAtomicGroupReplayEnd(); |
57 | 0 | } |
58 | 0 | } |
59 | 393k | } else { |
60 | 393k | last_valid_record_end_ = reader.LastRecordEnd(); |
61 | 393k | s = ApplyVersionEdit(edit, &cfd); |
62 | 393k | if (s.ok()) { |
63 | 393k | recovered_edits++; |
64 | 393k | } |
65 | 393k | } |
66 | 393k | } |
67 | 393k | } |
68 | 44.8k | if (s.ok() && !log_read_status->ok()) { |
69 | 0 | s = *log_read_status; |
70 | 0 | } |
71 | | |
72 | 44.8k | CheckIterationResult(reader, &s); |
73 | | |
74 | 44.8k | if (!s.ok()) { |
75 | 0 | if (s.IsCorruption()) { |
76 | | // when we find a Corruption error, something is |
77 | | // wrong with the underlying file. in this case we |
78 | | // want to report the filename, so in here we append |
79 | | // the filename to the Corruption message |
80 | 0 | assert(reader.file()); |
81 | | |
82 | | // build a new error message |
83 | 0 | std::stringstream message; |
84 | | // append previous dynamic state message |
85 | 0 | const char* state = s.getState(); |
86 | 0 | if (state != nullptr) { |
87 | 0 | message << state; |
88 | 0 | message << ' '; |
89 | 0 | } |
90 | | // append the filename to the corruption message |
91 | 0 | message << " The file " << reader.file()->file_name() |
92 | 0 | << " may be corrupted."; |
93 | | // overwrite the status with the extended status |
94 | 0 | s = Status(s.code(), s.subcode(), s.severity(), message.str()); |
95 | 0 | } |
96 | 0 | status_ = s; |
97 | 0 | } |
98 | 44.8k | TEST_SYNC_POINT_CALLBACK("VersionEditHandlerBase::Iterate:Finish", |
99 | 44.8k | &recovered_edits); |
100 | 44.8k | } |
101 | | |
102 | | Status ListColumnFamiliesHandler::ApplyVersionEdit( |
103 | 0 | VersionEdit& edit, ColumnFamilyData** /*unused*/) { |
104 | 0 | Status s; |
105 | 0 | uint32_t cf_id = edit.GetColumnFamily(); |
106 | 0 | if (edit.IsColumnFamilyAdd()) { |
107 | 0 | if (column_family_names_.find(cf_id) != column_family_names_.end()) { |
108 | 0 | s = Status::Corruption("Manifest adding the same column family twice"); |
109 | 0 | } else { |
110 | 0 | column_family_names_.insert({cf_id, edit.GetColumnFamilyName()}); |
111 | 0 | } |
112 | 0 | } else if (edit.IsColumnFamilyDrop()) { |
113 | 0 | if (column_family_names_.find(cf_id) == column_family_names_.end()) { |
114 | 0 | s = Status::Corruption("Manifest - dropping non-existing column family"); |
115 | 0 | } else { |
116 | 0 | column_family_names_.erase(cf_id); |
117 | 0 | } |
118 | 0 | } |
119 | 0 | return s; |
120 | 0 | } |
121 | | |
122 | | Status FileChecksumRetriever::FetchFileChecksumList( |
123 | 0 | FileChecksumList& file_checksum_list) { |
124 | 0 | Status s = Status::OK(); |
125 | 0 | for (const auto& [cf, file_checksums] : cf_file_checksums_) { |
126 | 0 | [[maybe_unused]] const auto& _ = cf; |
127 | 0 | for (const auto& [file_number, info] : file_checksums) { |
128 | 0 | if (!(s = file_checksum_list.InsertOneFileChecksum( |
129 | 0 | file_number, info.first, info.second)) |
130 | 0 | .ok()) { |
131 | 0 | break; |
132 | 0 | } |
133 | 0 | } |
134 | 0 | } |
135 | 0 | return s; |
136 | 0 | } |
137 | | |
138 | | Status FileChecksumRetriever::ApplyVersionEdit(VersionEdit& edit, |
139 | 0 | ColumnFamilyData** /*unused*/) { |
140 | 0 | uint32_t column_family_id = edit.GetColumnFamily(); |
141 | 0 | if (edit.IsColumnFamilyDrop()) { |
142 | 0 | cf_file_checksums_.erase(column_family_id); |
143 | 0 | } |
144 | 0 | for (const auto& deleted_file : edit.GetDeletedFiles()) { |
145 | 0 | if (cf_file_checksums_.find(column_family_id) == cf_file_checksums_.end()) { |
146 | 0 | return Status::NotFound(); |
147 | 0 | } |
148 | 0 | if (cf_file_checksums_[column_family_id].find(deleted_file.second) == |
149 | 0 | cf_file_checksums_[column_family_id].end()) { |
150 | 0 | return Status::NotFound(); |
151 | 0 | } |
152 | 0 | cf_file_checksums_[column_family_id].erase(deleted_file.second); |
153 | 0 | } |
154 | 0 | for (const auto& new_file : edit.GetNewFiles()) { |
155 | 0 | cf_file_checksums_[column_family_id].emplace( |
156 | 0 | new_file.second.fd.GetNumber(), |
157 | 0 | std::make_pair(new_file.second.file_checksum, |
158 | 0 | new_file.second.file_checksum_func_name)); |
159 | 0 | } |
160 | 0 | for (const auto& new_blob_file : edit.GetBlobFileAdditions()) { |
161 | 0 | std::string checksum_value = new_blob_file.GetChecksumValue(); |
162 | 0 | std::string checksum_method = new_blob_file.GetChecksumMethod(); |
163 | 0 | assert(checksum_value.empty() == checksum_method.empty()); |
164 | 0 | if (checksum_method.empty()) { |
165 | 0 | checksum_value = kUnknownFileChecksum; |
166 | 0 | checksum_method = kUnknownFileChecksumFuncName; |
167 | 0 | } |
168 | 0 | cf_file_checksums_[column_family_id].emplace( |
169 | 0 | new_blob_file.GetBlobFileNumber(), |
170 | 0 | std::make_pair(checksum_value, checksum_method)); |
171 | 0 | } |
172 | 0 | return Status::OK(); |
173 | 0 | } |
174 | | |
175 | | VersionEditHandler::VersionEditHandler( |
176 | | bool read_only, std::vector<ColumnFamilyDescriptor> column_families, |
177 | | VersionSet* version_set, bool track_found_and_missing_files, |
178 | | bool no_error_if_files_missing, const std::shared_ptr<IOTracer>& io_tracer, |
179 | | const ReadOptions& read_options, bool skip_load_table_files, |
180 | | bool allow_incomplete_valid_version, |
181 | | EpochNumberRequirement epoch_number_requirement) |
182 | 44.8k | : VersionEditHandlerBase(read_options), |
183 | 44.8k | read_only_(read_only), |
184 | 44.8k | column_families_(std::move(column_families)), |
185 | 44.8k | version_set_(version_set), |
186 | 44.8k | track_found_and_missing_files_(track_found_and_missing_files), |
187 | 44.8k | no_error_if_files_missing_(no_error_if_files_missing), |
188 | 44.8k | io_tracer_(io_tracer), |
189 | 44.8k | skip_load_table_files_(skip_load_table_files), |
190 | 44.8k | initialized_(false), |
191 | 44.8k | allow_incomplete_valid_version_(allow_incomplete_valid_version), |
192 | 44.8k | epoch_number_requirement_(epoch_number_requirement) { |
193 | 44.8k | assert(version_set_ != nullptr); |
194 | 44.8k | } |
195 | | |
196 | 44.8k | Status VersionEditHandler::Initialize() { |
197 | 44.8k | Status s; |
198 | 44.8k | if (!initialized_) { |
199 | 60.5k | for (const auto& cf_desc : column_families_) { |
200 | 60.5k | name_to_options_.emplace(cf_desc.name, cf_desc.options); |
201 | 60.5k | } |
202 | 44.8k | auto default_cf_iter = name_to_options_.find(kDefaultColumnFamilyName); |
203 | 44.8k | if (default_cf_iter == name_to_options_.end()) { |
204 | 0 | s = Status::InvalidArgument("Default column family not specified"); |
205 | 0 | } |
206 | 44.8k | if (s.ok()) { |
207 | 44.8k | VersionEdit default_cf_edit; |
208 | 44.8k | default_cf_edit.AddColumnFamily(kDefaultColumnFamilyName); |
209 | 44.8k | default_cf_edit.SetColumnFamily(0); |
210 | 44.8k | ColumnFamilyData* cfd = |
211 | 44.8k | CreateCfAndInit(default_cf_iter->second, default_cf_edit); |
212 | 44.8k | assert(cfd != nullptr); |
213 | 44.8k | #ifdef NDEBUG |
214 | 44.8k | (void)cfd; |
215 | 44.8k | #endif |
216 | 44.8k | initialized_ = true; |
217 | 44.8k | } |
218 | 44.8k | } |
219 | 44.8k | return s; |
220 | 44.8k | } |
221 | | |
222 | | Status VersionEditHandler::ApplyVersionEdit(VersionEdit& edit, |
223 | 393k | ColumnFamilyData** cfd) { |
224 | 393k | Status s; |
225 | 393k | if (edit.IsColumnFamilyAdd()) { |
226 | 29.7k | s = OnColumnFamilyAdd(edit, cfd); |
227 | 363k | } else if (edit.IsColumnFamilyDrop()) { |
228 | 14.0k | s = OnColumnFamilyDrop(edit, cfd); |
229 | 349k | } else if (edit.IsWalAddition()) { |
230 | 0 | s = OnWalAddition(edit); |
231 | 349k | } else if (edit.IsWalDeletion()) { |
232 | 26.2k | s = OnWalDeletion(edit); |
233 | 323k | } else { |
234 | 323k | s = OnNonCfOperation(edit, cfd); |
235 | 323k | } |
236 | 393k | if (s.ok()) { |
237 | 393k | assert(cfd != nullptr); |
238 | 393k | s = ExtractInfoFromVersionEdit(*cfd, edit); |
239 | 393k | } |
240 | 393k | return s; |
241 | 393k | } |
242 | | |
243 | | Status VersionEditHandler::OnColumnFamilyAdd(VersionEdit& edit, |
244 | 29.7k | ColumnFamilyData** cfd) { |
245 | 29.7k | bool do_not_open_cf = false; |
246 | 29.7k | bool cf_in_builders = false; |
247 | 29.7k | CheckColumnFamilyId(edit, &do_not_open_cf, &cf_in_builders); |
248 | | |
249 | 29.7k | assert(cfd != nullptr); |
250 | 29.7k | *cfd = nullptr; |
251 | 29.7k | const std::string& cf_name = edit.GetColumnFamilyName(); |
252 | 29.7k | Status s; |
253 | 29.7k | if (cf_in_builders || do_not_open_cf) { |
254 | 0 | s = Status::Corruption("MANIFEST adding the same column family twice: " + |
255 | 0 | cf_name); |
256 | 0 | } |
257 | 29.7k | if (s.ok()) { |
258 | 29.7k | auto cf_options = name_to_options_.find(cf_name); |
259 | | // implicitly add persistent_stats column family without requiring user |
260 | | // to specify |
261 | 29.7k | ColumnFamilyData* tmp_cfd = nullptr; |
262 | 29.7k | bool is_persistent_stats_column_family = |
263 | 29.7k | cf_name.compare(kPersistentStatsColumnFamilyName) == 0; |
264 | 29.7k | if (cf_options == name_to_options_.end() && |
265 | 2.64k | !is_persistent_stats_column_family) { |
266 | 2.64k | do_not_open_column_families_.emplace(edit.GetColumnFamily(), cf_name); |
267 | 27.0k | } else { |
268 | 27.0k | if (is_persistent_stats_column_family) { |
269 | 0 | ColumnFamilyOptions cfo; |
270 | 0 | OptimizeForPersistentStats(&cfo); |
271 | 0 | tmp_cfd = CreateCfAndInit(cfo, edit); |
272 | 27.0k | } else { |
273 | 27.0k | tmp_cfd = CreateCfAndInit(cf_options->second, edit); |
274 | 27.0k | } |
275 | 27.0k | *cfd = tmp_cfd; |
276 | 27.0k | } |
277 | 29.7k | } |
278 | 29.7k | return s; |
279 | 29.7k | } |
280 | | |
281 | | Status VersionEditHandler::OnColumnFamilyDrop(VersionEdit& edit, |
282 | 14.0k | ColumnFamilyData** cfd) { |
283 | 14.0k | bool do_not_open_cf = false; |
284 | 14.0k | bool cf_in_builders = false; |
285 | 14.0k | CheckColumnFamilyId(edit, &do_not_open_cf, &cf_in_builders); |
286 | | |
287 | 14.0k | assert(cfd != nullptr); |
288 | 14.0k | *cfd = nullptr; |
289 | 14.0k | ColumnFamilyData* tmp_cfd = nullptr; |
290 | 14.0k | Status s; |
291 | 14.0k | if (cf_in_builders) { |
292 | 11.3k | tmp_cfd = DestroyCfAndCleanup(edit); |
293 | 11.3k | } else if (do_not_open_cf) { |
294 | 2.64k | do_not_open_column_families_.erase(edit.GetColumnFamily()); |
295 | 2.64k | } else { |
296 | 0 | s = Status::Corruption("MANIFEST - dropping non-existing column family"); |
297 | 0 | } |
298 | 14.0k | *cfd = tmp_cfd; |
299 | 14.0k | return s; |
300 | 14.0k | } |
301 | | |
302 | 0 | Status VersionEditHandler::OnWalAddition(VersionEdit& edit) { |
303 | 0 | assert(edit.IsWalAddition()); |
304 | 0 | return version_set_->wals_.AddWals(edit.GetWalAdditions()); |
305 | 0 | } |
306 | | |
307 | 26.2k | Status VersionEditHandler::OnWalDeletion(VersionEdit& edit) { |
308 | 26.2k | assert(edit.IsWalDeletion()); |
309 | 26.2k | return version_set_->wals_.DeleteWalsBefore( |
310 | 26.2k | edit.GetWalDeletion().GetLogNumber()); |
311 | 26.2k | } |
312 | | |
313 | | Status VersionEditHandler::OnNonCfOperation(VersionEdit& edit, |
314 | 323k | ColumnFamilyData** cfd) { |
315 | 323k | bool do_not_open_cf = false; |
316 | 323k | bool cf_in_builders = false; |
317 | 323k | CheckColumnFamilyId(edit, &do_not_open_cf, &cf_in_builders); |
318 | | |
319 | 323k | assert(cfd != nullptr); |
320 | 323k | *cfd = nullptr; |
321 | 323k | Status s; |
322 | 323k | if (!do_not_open_cf) { |
323 | 317k | if (!cf_in_builders) { |
324 | 0 | s = Status::Corruption( |
325 | 0 | "MANIFEST record referencing unknown column family"); |
326 | 0 | } |
327 | 317k | ColumnFamilyData* tmp_cfd = nullptr; |
328 | 317k | if (s.ok()) { |
329 | 317k | tmp_cfd = version_set_->GetColumnFamilySet()->GetColumnFamily( |
330 | 317k | edit.GetColumnFamily()); |
331 | 317k | assert(tmp_cfd != nullptr); |
332 | | // It's important to handle file boundaries before `MaybeCreateVersion` |
333 | | // because `VersionEditHandlerPointInTime::MaybeCreateVersion` does |
334 | | // `FileMetaData` verification that involves the file boundaries. |
335 | | // All `VersionEditHandlerBase` subclasses that need to deal with |
336 | | // `FileMetaData` for new files are also subclasses of |
337 | | // `VersionEditHandler`, so it's sufficient to do the file boundaries |
338 | | // handling in this method. |
339 | 317k | s = MaybeHandleFileBoundariesForNewFiles(edit, tmp_cfd); |
340 | 317k | if (!s.ok()) { |
341 | 0 | return s; |
342 | 0 | } |
343 | 317k | s = MaybeCreateVersionBeforeApplyEdit(edit, tmp_cfd, |
344 | 317k | /*force_create_version=*/false); |
345 | 317k | } |
346 | 317k | *cfd = tmp_cfd; |
347 | 317k | } |
348 | 323k | return s; |
349 | 323k | } |
350 | | |
351 | | void VersionEditHandler::CheckColumnFamilyId(const VersionEdit& edit, |
352 | | bool* do_not_open_cf, |
353 | 366k | bool* cf_in_builders) const { |
354 | 366k | assert(do_not_open_cf != nullptr); |
355 | 366k | assert(cf_in_builders != nullptr); |
356 | | // Not found means that user didn't supply that column |
357 | | // family option AND we encountered column family add |
358 | | // record. Once we encounter column family drop record, |
359 | | // we will delete the column family from |
360 | | // do_not_open_column_families_. |
361 | 366k | uint32_t cf_id = edit.GetColumnFamily(); |
362 | 366k | bool in_do_not_open = do_not_open_column_families_.find(cf_id) != |
363 | 366k | do_not_open_column_families_.end(); |
364 | | // in builders means that user supplied that column family |
365 | | // option AND that we encountered column family add record |
366 | 366k | bool in_builders = builders_.find(cf_id) != builders_.end(); |
367 | | // They cannot both be true |
368 | 366k | assert(!(in_do_not_open && in_builders)); |
369 | 366k | *do_not_open_cf = in_do_not_open; |
370 | 366k | *cf_in_builders = in_builders; |
371 | 366k | } |
372 | | |
373 | | void VersionEditHandler::CheckIterationResult(const log::Reader& reader, |
374 | 44.8k | Status* s) { |
375 | 44.8k | assert(s != nullptr); |
376 | 44.8k | assert(version_set_ != nullptr); |
377 | 44.8k | if (!s->ok()) { |
378 | | // Do nothing here. |
379 | 44.8k | } else if (!version_edit_params_.HasLogNumber() || |
380 | 44.8k | !version_edit_params_.HasNextFile() || |
381 | 44.8k | !version_edit_params_.HasLastSequence()) { |
382 | 0 | std::string msg("no "); |
383 | 0 | if (!version_edit_params_.HasLogNumber()) { |
384 | 0 | msg.append("log_file_number, "); |
385 | 0 | } |
386 | 0 | if (!version_edit_params_.HasNextFile()) { |
387 | 0 | msg.append("next_file_number, "); |
388 | 0 | } |
389 | 0 | if (!version_edit_params_.HasLastSequence()) { |
390 | 0 | msg.append("last_sequence, "); |
391 | 0 | } |
392 | 0 | msg = msg.substr(0, msg.size() - 2); |
393 | 0 | msg.append(" entry in MANIFEST"); |
394 | 0 | *s = Status::Corruption(msg); |
395 | 0 | } |
396 | | // There were some column families in the MANIFEST that weren't specified |
397 | | // in the argument. This is OK in read_only mode |
398 | 44.8k | if (s->ok() && MustOpenAllColumnFamilies() && |
399 | 44.8k | !do_not_open_column_families_.empty()) { |
400 | 0 | std::string msg; |
401 | 0 | for (const auto& cf : do_not_open_column_families_) { |
402 | 0 | msg.append(", "); |
403 | 0 | msg.append(cf.second); |
404 | 0 | } |
405 | 0 | msg = msg.substr(2); |
406 | 0 | *s = Status::InvalidArgument("Column families not opened: " + msg); |
407 | 0 | } |
408 | 44.8k | if (s->ok()) { |
409 | 44.8k | version_set_->GetColumnFamilySet()->UpdateMaxColumnFamily( |
410 | 44.8k | version_edit_params_.GetMaxColumnFamily()); |
411 | 44.8k | version_set_->MarkMinLogNumberToKeep( |
412 | 44.8k | version_edit_params_.GetMinLogNumberToKeep()); |
413 | 44.8k | version_set_->MarkFileNumberUsed(version_edit_params_.GetPrevLogNumber()); |
414 | 44.8k | version_set_->MarkFileNumberUsed(version_edit_params_.GetLogNumber()); |
415 | 60.5k | for (auto* cfd : *(version_set_->GetColumnFamilySet())) { |
416 | 60.5k | if (cfd->IsDropped()) { |
417 | 0 | continue; |
418 | 0 | } |
419 | 60.5k | auto builder_iter = builders_.find(cfd->GetID()); |
420 | 60.5k | assert(builder_iter != builders_.end()); |
421 | 60.5k | auto* builder = builder_iter->second->version_builder(); |
422 | 60.5k | if (!builder->CheckConsistencyForNumLevels()) { |
423 | 0 | *s = Status::InvalidArgument( |
424 | 0 | "db has more levels than options.num_levels"); |
425 | 0 | break; |
426 | 0 | } |
427 | 60.5k | } |
428 | 44.8k | } |
429 | 44.8k | if (s->ok()) { |
430 | 60.5k | for (auto* cfd : *(version_set_->GetColumnFamilySet())) { |
431 | 60.5k | if (cfd->IsDropped()) { |
432 | 0 | continue; |
433 | 0 | } |
434 | 60.5k | if (version_set_->unchanging()) { |
435 | 0 | cfd->table_cache()->SetTablesAreImmortal(); |
436 | 0 | } |
437 | 60.5k | *s = LoadTables(cfd, /*prefetch_index_and_filter_in_cache=*/false, |
438 | 60.5k | /*is_initial_load=*/true); |
439 | 60.5k | if (!s->ok()) { |
440 | | // If s is IOError::PathNotFound, then we mark the db as corrupted. |
441 | 0 | if (s->IsPathNotFound()) { |
442 | 0 | *s = Status::Corruption("Corruption: " + s->ToString()); |
443 | 0 | } |
444 | 0 | break; |
445 | 0 | } |
446 | 60.5k | } |
447 | 44.8k | } |
448 | | |
449 | 44.8k | if (s->ok()) { |
450 | 60.5k | for (auto* cfd : *(version_set_->column_family_set_)) { |
451 | 60.5k | if (cfd->IsDropped()) { |
452 | 0 | continue; |
453 | 0 | } |
454 | 60.5k | assert(cfd->initialized()); |
455 | 60.5k | VersionEdit edit; |
456 | 60.5k | *s = MaybeCreateVersionBeforeApplyEdit(edit, cfd, |
457 | 60.5k | /*force_create_version=*/true); |
458 | 60.5k | if (!s->ok()) { |
459 | 0 | break; |
460 | 0 | } |
461 | 60.5k | } |
462 | 44.8k | } |
463 | 44.8k | if (s->ok()) { |
464 | 44.8k | version_set_->manifest_file_size_ = reader.GetReadOffset(); |
465 | 44.8k | assert(version_set_->manifest_file_size_ > 0); |
466 | 44.8k | version_set_->manifest_recovery_last_valid_record_end_ = |
467 | 44.8k | last_valid_record_end_; |
468 | 44.8k | version_set_->manifest_last_valid_record_end_ = last_valid_record_end_; |
469 | 44.8k | version_set_->manifest_last_valid_record_end_file_number_ = |
470 | 44.8k | version_set_->manifest_file_number_; |
471 | 44.8k | version_set_->next_file_number_.store(version_edit_params_.GetNextFile() + |
472 | 44.8k | 1); |
473 | 44.8k | SequenceNumber last_seq = version_edit_params_.GetLastSequence(); |
474 | 44.8k | assert(last_seq != kMaxSequenceNumber); |
475 | 44.8k | if (last_seq != kMaxSequenceNumber && |
476 | 44.8k | last_seq > version_set_->last_allocated_sequence_.load()) { |
477 | 24.5k | version_set_->last_allocated_sequence_.store(last_seq); |
478 | 24.5k | } |
479 | 44.8k | if (last_seq != kMaxSequenceNumber && |
480 | 44.8k | last_seq > version_set_->last_published_sequence_.load()) { |
481 | 24.5k | version_set_->last_published_sequence_.store(last_seq); |
482 | 24.5k | } |
483 | 44.8k | if (last_seq != kMaxSequenceNumber && |
484 | 44.8k | last_seq > version_set_->last_sequence_.load()) { |
485 | 24.5k | version_set_->last_sequence_.store(last_seq); |
486 | 24.5k | } |
487 | 44.8k | if (last_seq != kMaxSequenceNumber && |
488 | 44.8k | last_seq > version_set_->descriptor_last_sequence_) { |
489 | | // This is the maximum last sequence of all `VersionEdit`s iterated. It |
490 | | // may be greater than the maximum `largest_seqno` of all files in case |
491 | | // the newest data referred to by the MANIFEST has been dropped or had its |
492 | | // sequence number zeroed through compaction. |
493 | 24.5k | version_set_->descriptor_last_sequence_ = last_seq; |
494 | 24.5k | } |
495 | 44.8k | version_set_->prev_log_number_ = version_edit_params_.GetPrevLogNumber(); |
496 | 44.8k | } |
497 | 44.8k | } |
498 | | |
499 | | ColumnFamilyData* VersionEditHandler::CreateCfAndInit( |
500 | 71.9k | const ColumnFamilyOptions& cf_options, const VersionEdit& edit) { |
501 | 71.9k | uint32_t cf_id = edit.GetColumnFamily(); |
502 | 71.9k | ColumnFamilyData* cfd = version_set_->CreateColumnFamily( |
503 | 71.9k | cf_options, read_options_, &edit, read_only_, |
504 | 71.9k | /*options_are_sanitized=*/false); |
505 | 71.9k | assert(cfd != nullptr); |
506 | 71.9k | cfd->set_initialized(); |
507 | 71.9k | assert(builders_.find(cf_id) == builders_.end()); |
508 | 71.9k | VersionBuilderUPtr builder(new BaseReferencedVersionBuilder( |
509 | 71.9k | cfd, cfd->current(), this, track_found_and_missing_files_, |
510 | 71.9k | allow_incomplete_valid_version_)); |
511 | 71.9k | if (record_missing_blob_files_as_dropped_) { |
512 | 0 | builder->version_builder()->RecordMissingBlobFilesAsDropped(); |
513 | 0 | } |
514 | 71.9k | builders_.emplace(cf_id, std::move(builder)); |
515 | 71.9k | return cfd; |
516 | 71.9k | } |
517 | | |
518 | | ColumnFamilyData* VersionEditHandler::DestroyCfAndCleanup( |
519 | 11.3k | const VersionEdit& edit) { |
520 | 11.3k | uint32_t cf_id = edit.GetColumnFamily(); |
521 | 11.3k | auto builder_iter = builders_.find(cf_id); |
522 | 11.3k | assert(builder_iter != builders_.end()); |
523 | 11.3k | builders_.erase(builder_iter); |
524 | 11.3k | ColumnFamilyData* ret = |
525 | 11.3k | version_set_->GetColumnFamilySet()->GetColumnFamily(cf_id); |
526 | 11.3k | assert(ret != nullptr); |
527 | 11.3k | ret->SetDropped(); |
528 | 11.3k | ret->UnrefAndTryDelete(); |
529 | 11.3k | ret = nullptr; |
530 | 11.3k | return ret; |
531 | 11.3k | } |
532 | | |
533 | | Status VersionEditHandler::MaybeCreateVersionBeforeApplyEdit( |
534 | 378k | const VersionEdit& edit, ColumnFamilyData* cfd, bool force_create_version) { |
535 | 378k | assert(cfd->initialized()); |
536 | 378k | Status s; |
537 | 378k | auto builder_iter = builders_.find(cfd->GetID()); |
538 | 378k | assert(builder_iter != builders_.end()); |
539 | 378k | auto* builder = builder_iter->second->version_builder(); |
540 | 378k | if (force_create_version) { |
541 | 60.5k | auto* v = new Version(cfd, version_set_, version_set_->file_options_, |
542 | 60.5k | cfd->GetLatestMutableCFOptions(), io_tracer_, |
543 | 60.5k | version_set_->current_version_number_++, |
544 | 60.5k | epoch_number_requirement_); |
545 | 60.5k | s = builder->SaveTo(v->storage_info()); |
546 | 60.5k | if (s.ok()) { |
547 | | // Install new version |
548 | 60.5k | v->PrepareAppend( |
549 | 60.5k | read_options_, |
550 | 60.5k | !(version_set_->db_options_->skip_stats_update_on_db_open)); |
551 | 60.5k | version_set_->AppendVersion(cfd, v); |
552 | 60.5k | } else { |
553 | 0 | delete v; |
554 | 0 | } |
555 | 60.5k | } |
556 | 378k | s = builder->Apply(&edit); |
557 | 378k | return s; |
558 | 378k | } |
559 | | |
560 | | Status VersionEditHandler::LoadTables(ColumnFamilyData* cfd, |
561 | | bool prefetch_index_and_filter_in_cache, |
562 | 60.5k | bool is_initial_load) { |
563 | 60.5k | bool skip_load_table_files = skip_load_table_files_; |
564 | 60.5k | TEST_SYNC_POINT_CALLBACK( |
565 | 60.5k | "VersionEditHandler::LoadTables:skip_load_table_files", |
566 | 60.5k | &skip_load_table_files); |
567 | 60.5k | if (skip_load_table_files) { |
568 | 0 | return Status::OK(); |
569 | 0 | } |
570 | 60.5k | assert(cfd != nullptr); |
571 | 60.5k | assert(!cfd->IsDropped()); |
572 | 60.5k | auto builder_iter = builders_.find(cfd->GetID()); |
573 | 60.5k | assert(builder_iter != builders_.end()); |
574 | 60.5k | assert(builder_iter->second != nullptr); |
575 | 60.5k | VersionBuilder* builder = builder_iter->second->version_builder(); |
576 | 60.5k | assert(builder); |
577 | 60.5k | const auto& moptions = cfd->GetLatestMutableCFOptions(); |
578 | 60.5k | Status s = builder->LoadTableHandlers( |
579 | 60.5k | cfd->internal_stats(), |
580 | 60.5k | version_set_->db_options_->max_file_opening_threads, |
581 | 60.5k | prefetch_index_and_filter_in_cache, is_initial_load, moptions, |
582 | 60.5k | MaxFileSizeForL0MetaPin(moptions), read_options_); |
583 | 60.5k | if ((s.IsPathNotFound() || s.IsCorruption()) && no_error_if_files_missing_) { |
584 | 0 | s = Status::OK(); |
585 | 0 | } |
586 | 60.5k | if (!s.ok() && !version_set_->db_options_->paranoid_checks) { |
587 | 0 | s = Status::OK(); |
588 | 0 | } |
589 | 60.5k | return s; |
590 | 60.5k | } |
591 | | |
592 | | Status VersionEditHandler::ExtractInfoFromVersionEdit(ColumnFamilyData* cfd, |
593 | 393k | const VersionEdit& edit) { |
594 | 393k | Status s; |
595 | 393k | if (edit.HasDbId()) { |
596 | 44.8k | version_set_->db_id_ = edit.GetDbId(); |
597 | 44.8k | version_edit_params_.SetDBId(edit.GetDbId()); |
598 | 44.8k | } |
599 | 393k | if (cfd != nullptr) { |
600 | 344k | if (edit.HasLogNumber()) { |
601 | 113k | if (cfd->GetLogNumber() > edit.GetLogNumber()) { |
602 | 0 | ROCKS_LOG_WARN( |
603 | 0 | version_set_->db_options()->info_log, |
604 | 0 | "MANIFEST corruption detected, but ignored - Log numbers in " |
605 | 0 | "records NOT monotonically increasing"); |
606 | 113k | } else { |
607 | 113k | cfd->SetLogNumber(edit.GetLogNumber()); |
608 | 113k | version_edit_params_.SetLogNumber(edit.GetLogNumber()); |
609 | 113k | } |
610 | 113k | } |
611 | 344k | if (edit.HasComparatorName()) { |
612 | 63.7k | bool mark_sst_files_has_no_udt = false; |
613 | | // If `persist_user_defined_timestamps` flag is recorded in manifest, it |
614 | | // is guaranteed to be in the same VersionEdit as comparator. Otherwise, |
615 | | // it's not recorded and it should have default value true. |
616 | 63.7k | s = ValidateUserDefinedTimestampsOptions( |
617 | 63.7k | cfd->user_comparator(), edit.GetComparatorName(), |
618 | 63.7k | cfd->ioptions().persist_user_defined_timestamps, |
619 | 63.7k | edit.GetPersistUserDefinedTimestamps(), &mark_sst_files_has_no_udt); |
620 | 63.7k | if (!s.ok() && cf_to_cmp_names_) { |
621 | 0 | cf_to_cmp_names_->emplace(cfd->GetID(), edit.GetComparatorName()); |
622 | 0 | } |
623 | 63.7k | if (mark_sst_files_has_no_udt) { |
624 | 0 | cfds_to_mark_no_udt_.insert(cfd->GetID()); |
625 | 0 | } |
626 | 63.7k | } |
627 | 344k | if (edit.HasFullHistoryTsLow()) { |
628 | 0 | const std::string& new_ts = edit.GetFullHistoryTsLow(); |
629 | 0 | cfd->SetFullHistoryTsLow(new_ts); |
630 | 0 | } |
631 | 344k | } |
632 | | |
633 | 393k | if (s.ok()) { |
634 | 393k | if (edit.HasPrevLogNumber()) { |
635 | 143k | version_edit_params_.SetPrevLogNumber(edit.GetPrevLogNumber()); |
636 | 143k | } |
637 | 393k | if (edit.HasNextFile()) { |
638 | 181k | version_edit_params_.SetNextFile(edit.GetNextFile()); |
639 | 181k | } |
640 | 393k | if (edit.HasMaxColumnFamily()) { |
641 | 33.6k | version_edit_params_.SetMaxColumnFamily(edit.GetMaxColumnFamily()); |
642 | 33.6k | } |
643 | 393k | if (edit.HasMinLogNumberToKeep()) { |
644 | 56.2k | version_edit_params_.SetMinLogNumberToKeep( |
645 | 56.2k | std::max(version_edit_params_.GetMinLogNumberToKeep(), |
646 | 56.2k | edit.GetMinLogNumberToKeep())); |
647 | 56.2k | } |
648 | 393k | if (edit.HasLastSequence()) { |
649 | | // `VersionEdit::last_sequence_`s are assumed to be non-decreasing. This |
650 | | // is legacy behavior that cannot change without breaking downgrade |
651 | | // compatibility. |
652 | 232k | assert(!version_edit_params_.HasLastSequence() || |
653 | 232k | version_edit_params_.GetLastSequence() <= edit.GetLastSequence()); |
654 | 232k | version_edit_params_.SetLastSequence(edit.GetLastSequence()); |
655 | 232k | } |
656 | 393k | if (edit.HasLastCompactedManifestFileSize()) { |
657 | 36.7k | version_set_->last_compacted_manifest_file_size_ = |
658 | 36.7k | edit.GetLastCompactedManifestFileSize(); |
659 | 36.7k | version_set_->TuneMaxManifestFileSize(); |
660 | 36.7k | } |
661 | 393k | if (edit.HasCommittedOptionsFileNumber()) { |
662 | 0 | assert(version_set_ != nullptr); |
663 | 0 | version_set_->ApplyCommittedOptionsFileNumber( |
664 | 0 | edit.GetCommittedOptionsFileNumber()); |
665 | 0 | } |
666 | 393k | if (!version_edit_params_.HasPrevLogNumber()) { |
667 | 44.8k | version_edit_params_.SetPrevLogNumber(0); |
668 | 44.8k | } |
669 | 393k | } |
670 | 393k | return s; |
671 | 393k | } |
672 | | |
673 | | Status VersionEditHandler::MaybeHandleFileBoundariesForNewFiles( |
674 | 317k | VersionEdit& edit, const ColumnFamilyData* cfd) { |
675 | 317k | if (edit.GetNewFiles().empty()) { |
676 | 281k | return Status::OK(); |
677 | 281k | } |
678 | 36.3k | auto ucmp = cfd->user_comparator(); |
679 | 36.3k | assert(ucmp); |
680 | 36.3k | size_t ts_sz = ucmp->timestamp_size(); |
681 | 36.3k | if (ts_sz == 0) { |
682 | 36.3k | return Status::OK(); |
683 | 36.3k | } |
684 | | |
685 | 0 | VersionEdit::NewFiles& new_files = edit.GetMutableNewFiles(); |
686 | 0 | assert(!new_files.empty()); |
687 | | // If true, enabling user-defined timestamp is detected for this column |
688 | | // family. All its existing SST files need to have the file boundaries handled |
689 | | // and their `persist_user_defined_timestamps` flag set to false regardless of |
690 | | // its existing value. |
691 | 0 | bool mark_existing_ssts_with_no_udt = |
692 | 0 | cfds_to_mark_no_udt_.find(cfd->GetID()) != cfds_to_mark_no_udt_.end(); |
693 | 0 | bool file_boundaries_need_handling = false; |
694 | 0 | for (auto& new_file : new_files) { |
695 | 0 | FileMetaData& meta = new_file.second; |
696 | 0 | if (meta.user_defined_timestamps_persisted && |
697 | 0 | !mark_existing_ssts_with_no_udt) { |
698 | | // `FileMetaData.user_defined_timestamps_persisted` field is the value of |
699 | | // the flag `AdvancedColumnFamilyOptions.persist_user_defined_timestamps` |
700 | | // at the time when the SST file was created. As a result, all added SST |
701 | | // files in one `VersionEdit` should have the same value for it. |
702 | 0 | if (file_boundaries_need_handling) { |
703 | 0 | return Status::Corruption( |
704 | 0 | "New files in one VersionEdit has different " |
705 | 0 | "user_defined_timestamps_persisted value."); |
706 | 0 | } |
707 | 0 | break; |
708 | 0 | } |
709 | 0 | file_boundaries_need_handling = true; |
710 | 0 | assert(!meta.user_defined_timestamps_persisted || |
711 | 0 | mark_existing_ssts_with_no_udt); |
712 | 0 | if (mark_existing_ssts_with_no_udt) { |
713 | 0 | meta.user_defined_timestamps_persisted = false; |
714 | 0 | } |
715 | 0 | std::string smallest_buf; |
716 | 0 | std::string largest_buf; |
717 | 0 | Slice largest_slice = meta.largest.Encode(); |
718 | 0 | PadInternalKeyWithMinTimestamp(&smallest_buf, meta.smallest.Encode(), |
719 | 0 | ts_sz); |
720 | 0 | auto largest_footer = ExtractInternalKeyFooter(largest_slice); |
721 | 0 | if (largest_footer == kRangeTombstoneSentinel) { |
722 | 0 | PadInternalKeyWithMaxTimestamp(&largest_buf, largest_slice, ts_sz); |
723 | 0 | } else { |
724 | 0 | PadInternalKeyWithMinTimestamp(&largest_buf, largest_slice, ts_sz); |
725 | 0 | } |
726 | 0 | meta.smallest.DecodeFrom(smallest_buf); |
727 | 0 | meta.largest.DecodeFrom(largest_buf); |
728 | 0 | } |
729 | 0 | return Status::OK(); |
730 | 0 | } |
731 | | |
732 | | VersionEditHandlerPointInTime::VersionEditHandlerPointInTime( |
733 | | bool read_only, std::vector<ColumnFamilyDescriptor> column_families, |
734 | | VersionSet* version_set, const std::shared_ptr<IOTracer>& io_tracer, |
735 | | const ReadOptions& read_options, bool allow_incomplete_valid_version, |
736 | | bool record_missing_blob_files_as_dropped, bool trust_manifest_recovery, |
737 | | bool defer_sst_file_opening, |
738 | | EpochNumberRequirement epoch_number_requirement) |
739 | 0 | : VersionEditHandler(read_only, std::move(column_families), version_set, |
740 | 0 | /*track_found_and_missing_files=*/true, |
741 | 0 | /*no_error_if_files_missing=*/true, io_tracer, |
742 | 0 | read_options, /*skip_load_table_files=*/false, |
743 | 0 | allow_incomplete_valid_version, |
744 | 0 | epoch_number_requirement), |
745 | 0 | trust_manifest_recovery_(trust_manifest_recovery), |
746 | 0 | defer_sst_file_opening_(defer_sst_file_opening) { |
747 | 0 | assert(!record_missing_blob_files_as_dropped || |
748 | 0 | allow_incomplete_valid_version); |
749 | 0 | record_missing_blob_files_as_dropped_ = record_missing_blob_files_as_dropped; |
750 | 0 | } |
751 | | |
752 | 0 | VersionEditHandlerPointInTime::~VersionEditHandlerPointInTime() { |
753 | 0 | for (const auto& cfid_and_version : atomic_update_versions_) { |
754 | 0 | delete cfid_and_version.second.version; |
755 | 0 | } |
756 | 0 | for (const auto& elem : versions_) { |
757 | 0 | delete elem.second.version; |
758 | 0 | } |
759 | 0 | versions_.clear(); |
760 | 0 | } |
761 | | |
762 | 0 | Status VersionEditHandlerPointInTime::OnAtomicGroupReplayBegin() { |
763 | 0 | if (in_atomic_group_) { |
764 | 0 | return Status::Corruption("unexpected AtomicGroup start"); |
765 | 0 | } |
766 | | |
767 | | // The AtomicGroup that is about to begin may block column families in a valid |
768 | | // state from saving any more updates. So we should save any valid states |
769 | | // before proceeding. |
770 | 0 | for (const auto& cfid_and_builder : builders_) { |
771 | 0 | ColumnFamilyData* cfd = version_set_->GetColumnFamilySet()->GetColumnFamily( |
772 | 0 | cfid_and_builder.first); |
773 | 0 | assert(!cfd->IsDropped()); |
774 | 0 | assert(cfd->initialized()); |
775 | 0 | VersionEdit edit; |
776 | 0 | Status s = MaybeCreateVersionBeforeApplyEdit( |
777 | 0 | edit, cfd, true /* force_create_version */); |
778 | 0 | if (!s.ok()) { |
779 | 0 | return s; |
780 | 0 | } |
781 | 0 | } |
782 | | |
783 | | // An old AtomicGroup is incomplete. Throw away the versions that failed to |
784 | | // complete it. They must not be used for completing the upcoming |
785 | | // AtomicGroup since they are too old. |
786 | 0 | for (auto& cfid_and_version : atomic_update_versions_) { |
787 | 0 | delete cfid_and_version.second.version; |
788 | 0 | } |
789 | |
|
790 | 0 | in_atomic_group_ = true; |
791 | | // We lazily assume the column families that exist at this point are all |
792 | | // involved in the AtomicGroup. Overestimating the scope of the AtomicGroup |
793 | | // will sometimes cause less data to be recovered, which is fine for |
794 | | // best-effort recovery. |
795 | 0 | atomic_update_versions_.clear(); |
796 | 0 | for (const auto& cfid_and_builder : builders_) { |
797 | 0 | atomic_update_versions_[cfid_and_builder.first] = PointInTimeVersion(); |
798 | 0 | } |
799 | 0 | atomic_update_versions_missing_ = atomic_update_versions_.size(); |
800 | 0 | return Status::OK(); |
801 | 0 | } |
802 | | |
803 | 0 | Status VersionEditHandlerPointInTime::OnAtomicGroupReplayEnd() { |
804 | 0 | if (!in_atomic_group_) { |
805 | 0 | return Status::Corruption("unexpected AtomicGroup end"); |
806 | 0 | } |
807 | 0 | in_atomic_group_ = false; |
808 | | |
809 | | // The AtomicGroup must not have changed the column families. We don't support |
810 | | // CF adds or drops in an AtomicGroup. |
811 | 0 | if (builders_.size() != atomic_update_versions_.size()) { |
812 | 0 | return Status::Corruption("unexpected CF change in AtomicGroup"); |
813 | 0 | } |
814 | 0 | for (const auto& cfid_and_builder : builders_) { |
815 | 0 | if (atomic_update_versions_.find(cfid_and_builder.first) == |
816 | 0 | atomic_update_versions_.end()) { |
817 | 0 | return Status::Corruption("unexpected CF add in AtomicGroup"); |
818 | 0 | } |
819 | 0 | } |
820 | 0 | for (const auto& cfid_and_version : atomic_update_versions_) { |
821 | 0 | if (builders_.find(cfid_and_version.first) == builders_.end()) { |
822 | 0 | return Status::Corruption("unexpected CF drop in AtomicGroup"); |
823 | 0 | } |
824 | 0 | } |
825 | 0 | return Status::OK(); |
826 | 0 | } |
827 | | |
828 | | void VersionEditHandlerPointInTime::CheckIterationResult( |
829 | 0 | const log::Reader& reader, Status* s) { |
830 | 0 | VersionEditHandler::CheckIterationResult(reader, s); |
831 | 0 | assert(s != nullptr); |
832 | 0 | if (s->ok()) { |
833 | 0 | for (auto* cfd : *(version_set_->column_family_set_)) { |
834 | 0 | if (cfd->IsDropped()) { |
835 | 0 | continue; |
836 | 0 | } |
837 | 0 | assert(cfd->initialized()); |
838 | 0 | auto v_iter = versions_.find(cfd->GetID()); |
839 | 0 | auto builder_iter = builders_.find(cfd->GetID()); |
840 | 0 | if (v_iter != versions_.end()) { |
841 | 0 | assert(v_iter->second.version != nullptr); |
842 | 0 | assert(builder_iter != builders_.end()); |
843 | |
|
844 | 0 | version_set_->AppendVersion(cfd, v_iter->second.version); |
845 | 0 | installed_version_log_numbers_[cfd->GetID()] = |
846 | 0 | v_iter->second.log_number; |
847 | 0 | versions_.erase(v_iter); |
848 | | // Let's clear found_files, since any files in that are part of the |
849 | | // installed Version. Any files that got obsoleted would have already |
850 | | // been moved to intermediate_files_ |
851 | 0 | builder_iter->second->version_builder()->ClearFoundFiles(); |
852 | 0 | } |
853 | 0 | } |
854 | 0 | } else { |
855 | 0 | for (const auto& elem : versions_) { |
856 | 0 | delete elem.second.version; |
857 | 0 | } |
858 | 0 | versions_.clear(); |
859 | 0 | } |
860 | 0 | } |
861 | | |
862 | | ColumnFamilyData* VersionEditHandlerPointInTime::DestroyCfAndCleanup( |
863 | 0 | const VersionEdit& edit) { |
864 | 0 | ColumnFamilyData* cfd = VersionEditHandler::DestroyCfAndCleanup(edit); |
865 | 0 | uint32_t cfid = edit.GetColumnFamily(); |
866 | 0 | if (AtomicUpdateVersionsContains(cfid)) { |
867 | 0 | AtomicUpdateVersionsDropCf(cfid); |
868 | 0 | if (AtomicUpdateVersionsCompleted()) { |
869 | 0 | AtomicUpdateVersionsApply(); |
870 | 0 | } |
871 | 0 | } |
872 | 0 | auto v_iter = versions_.find(cfid); |
873 | 0 | if (v_iter != versions_.end()) { |
874 | 0 | delete v_iter->second.version; |
875 | 0 | versions_.erase(v_iter); |
876 | 0 | } |
877 | 0 | installed_version_log_numbers_.erase(cfid); |
878 | 0 | return cfd; |
879 | 0 | } |
880 | | |
881 | | Status VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit( |
882 | 0 | const VersionEdit& edit, ColumnFamilyData* cfd, bool force_create_version) { |
883 | 0 | TEST_SYNC_POINT( |
884 | 0 | "VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit:" |
885 | 0 | "Begin1"); |
886 | 0 | TEST_SYNC_POINT( |
887 | 0 | "VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit:" |
888 | 0 | "Begin2"); |
889 | 0 | assert(cfd != nullptr); |
890 | 0 | if (!force_create_version) { |
891 | 0 | assert(edit.GetColumnFamily() == cfd->GetID()); |
892 | 0 | } |
893 | |
|
894 | 0 | bool missing_info = !version_edit_params_.HasLogNumber() || |
895 | 0 | !version_edit_params_.HasNextFile() || |
896 | 0 | !version_edit_params_.HasLastSequence(); |
897 | |
|
898 | 0 | Status s; |
899 | 0 | auto builder_iter = builders_.find(cfd->GetID()); |
900 | 0 | assert(builder_iter != builders_.end()); |
901 | 0 | VersionBuilder* builder = builder_iter->second->version_builder(); |
902 | 0 | const bool holds_point_in_time = BuilderHoldsPointInTime(cfd->GetID()); |
903 | 0 | const bool valid_pit_before_edit = builder->ValidVersionAvailable(); |
904 | 0 | s = builder->Apply(&edit); |
905 | 0 | const bool valid_pit_after_edit = builder->ValidVersionAvailable(); |
906 | | |
907 | | // A new version will be created if: |
908 | | // 1) no error has occurred so far, and |
909 | | // 2) log_number_, next_file_number_ and last_sequence_ are known, and |
910 | | // 3) not in an AtomicGroup, and |
911 | | // 4) the builder held a point in time of the MANIFEST before the edit, and |
912 | | // 5) any of the following: |
913 | | // a) a valid Version is available before applying the edit |
914 | | // and a valid Version is not available after the edit. |
915 | | // b) a valid Version is available after the edit and the |
916 | | // caller explicitly request that a new version be created. |
917 | 0 | if (s.ok() && !missing_info && !in_atomic_group_ && holds_point_in_time && |
918 | 0 | ((!valid_pit_after_edit && valid_pit_before_edit) || |
919 | 0 | (valid_pit_after_edit && force_create_version))) { |
920 | | // The Version to be created reflects the state *before* this edit. On a |
921 | | // negative edge (valid before, invalid after), roll the just-applied edit |
922 | | // back so the builder reflects the pre-edit state, build the Version from |
923 | | // it, then redo to restore the post-edit state and continue replay. In the |
924 | | // force case the edit is empty, so pre-edit == post-edit and no rollback is |
925 | | // needed. |
926 | 0 | const bool negative_edge = valid_pit_before_edit && !valid_pit_after_edit; |
927 | 0 | if (negative_edge) { |
928 | 0 | builder->RollbackLastApply(); |
929 | 0 | } |
930 | | // `cfd->GetLogNumber()` describes exactly the state the Version about to be |
931 | | // built reflects: VersionEditHandler::ExtractInfoFromVersionEdit() applies |
932 | | // `edit`'s own log number only after this function returns, so on a |
933 | | // negative edge it is still the pre-edit value. |
934 | 0 | const uint64_t log_number = cfd->GetLogNumber(); |
935 | 0 | const auto& mopts = cfd->GetLatestMutableCFOptions(); |
936 | 0 | auto* version = new Version( |
937 | 0 | cfd, version_set_, version_set_->file_options_, mopts, io_tracer_, |
938 | 0 | version_set_->current_version_number_++, epoch_number_requirement_); |
939 | 0 | if (!trust_manifest_recovery_ && !defer_sst_file_opening_) { |
940 | 0 | s = builder->LoadTableHandlers( |
941 | 0 | cfd->internal_stats(), |
942 | 0 | version_set_->db_options_->max_file_opening_threads, false, true, |
943 | 0 | mopts, MaxFileSizeForL0MetaPin(mopts), read_options_); |
944 | 0 | if (!s.ok()) { |
945 | 0 | delete version; |
946 | 0 | if (negative_edge) { |
947 | 0 | builder->RedoLastApply(); |
948 | 0 | } |
949 | 0 | if (s.IsCorruption() || |
950 | 0 | (s.IsPathNotFound() && no_error_if_files_missing_)) { |
951 | | // This point in time cannot be recovered; skip it and continue. A |
952 | | // PathNotFound is a file the primary compacted away before this |
953 | | // handler opened it, which open_files_async makes reachable during a |
954 | | // secondary's catch-up by leaving recovery-time files unpinned until |
955 | | // the background opener reaches them. Treat it as missing rather than |
956 | | // latching the error, the same tolerance LoadTables() applies. |
957 | 0 | s = Status::OK(); |
958 | 0 | } |
959 | 0 | builder->CommitLastApply(); |
960 | 0 | return s; |
961 | 0 | } |
962 | 0 | } |
963 | 0 | s = builder->SaveTo(version->storage_info()); |
964 | 0 | if (negative_edge) { |
965 | 0 | builder->RedoLastApply(); |
966 | 0 | } |
967 | 0 | if (s.ok()) { |
968 | 0 | if (AtomicUpdateVersionsContains(cfd->GetID())) { |
969 | 0 | AtomicUpdateVersionsPut(PointInTimeVersion{version, log_number}); |
970 | 0 | if (AtomicUpdateVersionsCompleted()) { |
971 | 0 | AtomicUpdateVersionsApply(); |
972 | 0 | } |
973 | 0 | } else { |
974 | 0 | version->PrepareAppend( |
975 | 0 | read_options_, |
976 | 0 | !version_set_->db_options_->skip_stats_update_on_db_open); |
977 | 0 | auto v_iter = versions_.find(cfd->GetID()); |
978 | 0 | if (v_iter != versions_.end()) { |
979 | 0 | delete v_iter->second.version; |
980 | 0 | v_iter->second = PointInTimeVersion{version, log_number}; |
981 | 0 | } else { |
982 | 0 | versions_.emplace(cfd->GetID(), |
983 | 0 | PointInTimeVersion{version, log_number}); |
984 | 0 | } |
985 | 0 | } |
986 | 0 | } else { |
987 | 0 | delete version; |
988 | 0 | } |
989 | 0 | } |
990 | | |
991 | 0 | builder->CommitLastApply(); |
992 | 0 | return s; |
993 | 0 | } |
994 | | |
995 | | Status VersionEditHandlerPointInTime::VerifyFile(ColumnFamilyData* cfd, |
996 | | const std::string& fpath, |
997 | | int level, |
998 | 0 | const FileMetaData& fmeta) { |
999 | 0 | if (trust_manifest_recovery_ || defer_sst_file_opening_) { |
1000 | | // Trust the MANIFEST: do not stat/open the file to classify it. The |
1001 | | // compaction or asynchronous opener validates files later. |
1002 | 0 | return Status::OK(); |
1003 | 0 | } |
1004 | 0 | return version_set_->VerifyFileMetadata(read_options_, cfd, fpath, level, |
1005 | 0 | fmeta); |
1006 | 0 | } |
1007 | | |
1008 | | Status VersionEditHandlerPointInTime::VerifyBlobFile( |
1009 | | ColumnFamilyData* cfd, uint64_t blob_file_num, |
1010 | 0 | const BlobFileAddition& blob_addition) { |
1011 | 0 | if (trust_manifest_recovery_) { |
1012 | | // Trust the MANIFEST: do not open the blob file to classify it. |
1013 | 0 | return Status::OK(); |
1014 | 0 | } |
1015 | 0 | BlobSource* blob_source = cfd->blob_source(); |
1016 | 0 | assert(blob_source); |
1017 | 0 | CacheHandleGuard<BlobFileReader> blob_file_reader; |
1018 | |
|
1019 | 0 | const BlobFileOpenInfo blob_file{blob_file_num, |
1020 | 0 | blob_addition.GetChecksumValue(), |
1021 | 0 | blob_addition.GetChecksumMethod()}; |
1022 | 0 | Status s = blob_source->GetBlobFileReader(read_options_, blob_file, |
1023 | 0 | &blob_file_reader); |
1024 | 0 | if (!s.ok()) { |
1025 | 0 | return s; |
1026 | 0 | } |
1027 | | // TODO: verify checksum |
1028 | 0 | return s; |
1029 | 0 | } |
1030 | | |
1031 | | Status VersionEditHandlerPointInTime::LoadTables( |
1032 | | ColumnFamilyData* /*cfd*/, bool /*prefetch_index_and_filter_in_cache*/, |
1033 | 0 | bool /*is_initial_load*/) { |
1034 | 0 | return Status::OK(); |
1035 | 0 | } |
1036 | | |
1037 | 0 | bool VersionEditHandlerPointInTime::HasMissingFiles() const { |
1038 | 0 | for (const auto& builder : builders_) { |
1039 | 0 | if (builder.second->version_builder()->HasMissingFiles()) { |
1040 | 0 | return true; |
1041 | 0 | } |
1042 | 0 | } |
1043 | 0 | return false; |
1044 | 0 | } |
1045 | | |
1046 | | uint64_t VersionEditHandlerPointInTime::GetInstalledVersionLogNumber( |
1047 | 0 | uint32_t cf_id) const { |
1048 | 0 | auto iter = installed_version_log_numbers_.find(cf_id); |
1049 | 0 | if (iter == installed_version_log_numbers_.end()) { |
1050 | 0 | return 0; |
1051 | 0 | } |
1052 | 0 | return iter->second; |
1053 | 0 | } |
1054 | | |
1055 | 0 | bool VersionEditHandlerPointInTime::AtomicUpdateVersionsCompleted() { |
1056 | 0 | return atomic_update_versions_missing_ == 0; |
1057 | 0 | } |
1058 | | |
1059 | | bool VersionEditHandlerPointInTime::AtomicUpdateVersionsContains( |
1060 | 0 | uint32_t cfid) { |
1061 | 0 | return atomic_update_versions_.find(cfid) != atomic_update_versions_.end(); |
1062 | 0 | } |
1063 | | |
1064 | 0 | void VersionEditHandlerPointInTime::AtomicUpdateVersionsDropCf(uint32_t cfid) { |
1065 | 0 | assert(!AtomicUpdateVersionsCompleted()); |
1066 | 0 | auto atomic_update_versions_iter = atomic_update_versions_.find(cfid); |
1067 | 0 | assert(atomic_update_versions_iter != atomic_update_versions_.end()); |
1068 | 0 | if (atomic_update_versions_iter->second.version == nullptr) { |
1069 | 0 | atomic_update_versions_missing_--; |
1070 | 0 | } else { |
1071 | 0 | delete atomic_update_versions_iter->second.version; |
1072 | 0 | } |
1073 | 0 | atomic_update_versions_.erase(atomic_update_versions_iter); |
1074 | 0 | } |
1075 | | |
1076 | | void VersionEditHandlerPointInTime::AtomicUpdateVersionsPut( |
1077 | 0 | PointInTimeVersion pit_version) { |
1078 | 0 | assert(!AtomicUpdateVersionsCompleted()); |
1079 | 0 | assert(pit_version.version != nullptr); |
1080 | 0 | auto atomic_update_versions_iter = |
1081 | 0 | atomic_update_versions_.find(pit_version.version->cfd()->GetID()); |
1082 | 0 | assert(atomic_update_versions_iter != atomic_update_versions_.end()); |
1083 | 0 | if (atomic_update_versions_iter->second.version == nullptr) { |
1084 | 0 | atomic_update_versions_missing_--; |
1085 | 0 | } else { |
1086 | 0 | delete atomic_update_versions_iter->second.version; |
1087 | 0 | } |
1088 | 0 | atomic_update_versions_iter->second = pit_version; |
1089 | 0 | } |
1090 | | |
1091 | 0 | void VersionEditHandlerPointInTime::AtomicUpdateVersionsApply() { |
1092 | 0 | assert(AtomicUpdateVersionsCompleted()); |
1093 | 0 | for (const auto& cfid_and_version : atomic_update_versions_) { |
1094 | 0 | uint32_t cfid = cfid_and_version.first; |
1095 | 0 | const PointInTimeVersion& pit_version = cfid_and_version.second; |
1096 | 0 | assert(pit_version.version != nullptr); |
1097 | 0 | pit_version.version->PrepareAppend( |
1098 | 0 | read_options_, |
1099 | 0 | !version_set_->db_options_->skip_stats_update_on_db_open); |
1100 | 0 | auto versions_iter = versions_.find(cfid); |
1101 | 0 | if (versions_iter != versions_.end()) { |
1102 | 0 | delete versions_iter->second.version; |
1103 | 0 | versions_iter->second = pit_version; |
1104 | 0 | } else { |
1105 | 0 | versions_.emplace(cfid, pit_version); |
1106 | 0 | } |
1107 | 0 | } |
1108 | 0 | atomic_update_versions_.clear(); |
1109 | 0 | } |
1110 | | |
1111 | 0 | Status ManifestTailer::Initialize() { |
1112 | 0 | if (Mode::kRecovery == mode_) { |
1113 | 0 | return VersionEditHandler::Initialize(); |
1114 | 0 | } |
1115 | 0 | assert(Mode::kCatchUp == mode_); |
1116 | 0 | Status s; |
1117 | 0 | if (!initialized_) { |
1118 | 0 | ColumnFamilySet* cfd_set = version_set_->GetColumnFamilySet(); |
1119 | 0 | assert(cfd_set); |
1120 | 0 | ColumnFamilyData* default_cfd = cfd_set->GetDefault(); |
1121 | 0 | assert(default_cfd); |
1122 | 0 | auto builder_iter = builders_.find(default_cfd->GetID()); |
1123 | 0 | assert(builder_iter != builders_.end()); |
1124 | |
|
1125 | 0 | Version* dummy_version = default_cfd->dummy_versions(); |
1126 | 0 | assert(dummy_version); |
1127 | 0 | Version* base_version = dummy_version->Next(); |
1128 | 0 | assert(base_version); |
1129 | 0 | VersionBuilderUPtr new_builder(new BaseReferencedVersionBuilder( |
1130 | 0 | default_cfd, base_version, this, track_found_and_missing_files_, |
1131 | 0 | allow_incomplete_valid_version_)); |
1132 | 0 | builder_iter->second = std::move(new_builder); |
1133 | 0 | AwaitFileList(default_cfd->GetID()); |
1134 | |
|
1135 | 0 | initialized_ = true; |
1136 | 0 | } |
1137 | 0 | return s; |
1138 | 0 | } |
1139 | | |
1140 | | Status ManifestTailer::ApplyVersionEdit(VersionEdit& edit, |
1141 | 0 | ColumnFamilyData** cfd) { |
1142 | 0 | Status s = VersionEditHandler::ApplyVersionEdit(edit, cfd); |
1143 | 0 | if (s.ok()) { |
1144 | 0 | assert(cfd); |
1145 | 0 | if (*cfd) { |
1146 | 0 | cfds_changed_.insert(*cfd); |
1147 | | // VersionSet::WriteCurrentStateToManifest() lists a column family's |
1148 | | // files in the first record for it that sets a log number. |
1149 | 0 | if (!edit.IsColumnFamilyAdd() && edit.HasLogNumber()) { |
1150 | 0 | cfs_awaiting_file_list_.erase((*cfd)->GetID()); |
1151 | 0 | } |
1152 | 0 | } |
1153 | | // The snapshot that opens a MANIFEST sets no next file number, and every |
1154 | | // record after it does. The snapshot leaves out only dropped column |
1155 | | // families, so the ones it did not add again never get a file list. |
1156 | 0 | if (edit.HasNextFile()) { |
1157 | 0 | for (uint32_t cf_id : cfs_awaiting_from_previous_manifest_) { |
1158 | 0 | cfs_awaiting_file_list_.erase(cf_id); |
1159 | 0 | } |
1160 | 0 | cfs_awaiting_from_previous_manifest_.clear(); |
1161 | 0 | } |
1162 | 0 | } |
1163 | 0 | return s; |
1164 | 0 | } |
1165 | | |
1166 | | Status ManifestTailer::OnColumnFamilyAdd(VersionEdit& edit, |
1167 | 0 | ColumnFamilyData** cfd) { |
1168 | 0 | if (Mode::kRecovery == mode_) { |
1169 | 0 | return VersionEditHandler::OnColumnFamilyAdd(edit, cfd); |
1170 | 0 | } |
1171 | 0 | assert(Mode::kCatchUp == mode_); |
1172 | 0 | ColumnFamilySet* cfd_set = version_set_->GetColumnFamilySet(); |
1173 | 0 | assert(cfd_set); |
1174 | 0 | ColumnFamilyData* tmp_cfd = cfd_set->GetColumnFamily(edit.GetColumnFamily()); |
1175 | 0 | assert(cfd); |
1176 | 0 | *cfd = tmp_cfd; |
1177 | 0 | if (!tmp_cfd) { |
1178 | | // For now, ignore new column families created after Recover() succeeds. |
1179 | 0 | return Status::OK(); |
1180 | 0 | } |
1181 | 0 | auto builder_iter = builders_.find(edit.GetColumnFamily()); |
1182 | 0 | assert(builder_iter != builders_.end()); |
1183 | |
|
1184 | 0 | Version* dummy_version = tmp_cfd->dummy_versions(); |
1185 | 0 | assert(dummy_version); |
1186 | 0 | Version* base_version = dummy_version->Next(); |
1187 | 0 | assert(base_version); |
1188 | 0 | VersionBuilderUPtr new_builder(new BaseReferencedVersionBuilder( |
1189 | 0 | tmp_cfd, base_version, this, track_found_and_missing_files_)); |
1190 | 0 | builder_iter->second = std::move(new_builder); |
1191 | 0 | AwaitFileList(tmp_cfd->GetID()); |
1192 | |
|
1193 | | #ifndef NDEBUG |
1194 | | auto version_iter = versions_.find(edit.GetColumnFamily()); |
1195 | | assert(version_iter == versions_.end()); |
1196 | | #endif // !NDEBUG |
1197 | 0 | return Status::OK(); |
1198 | 0 | } |
1199 | | |
1200 | 0 | void ManifestTailer::AwaitFileList(uint32_t cf_id) { |
1201 | 0 | cfs_awaiting_file_list_.insert(cf_id); |
1202 | 0 | cfs_awaiting_from_previous_manifest_.erase(cf_id); |
1203 | 0 | } |
1204 | | |
1205 | | void ManifestTailer::CheckIterationResult(const log::Reader& reader, |
1206 | 0 | Status* s) { |
1207 | 0 | VersionEditHandlerPointInTime::CheckIterationResult(reader, s); |
1208 | 0 | assert(s); |
1209 | 0 | if (s->ok()) { |
1210 | 0 | if (Mode::kRecovery == mode_) { |
1211 | 0 | if (defer_sst_file_opening_) { |
1212 | | // Deferred verification treats each added SST as found. Discard files |
1213 | | // also deleted during initial replay so they are not reported as |
1214 | | // obsolete files from a later catch-up. |
1215 | 0 | GetAndClearIntermediateFiles(); |
1216 | 0 | } |
1217 | 0 | defer_sst_file_opening_ = false; |
1218 | 0 | mode_ = Mode::kCatchUp; |
1219 | 0 | } else { |
1220 | 0 | assert(Mode::kCatchUp == mode_); |
1221 | 0 | } |
1222 | 0 | } |
1223 | 0 | } |
1224 | | |
1225 | 0 | std::vector<std::string> ManifestTailer::GetAndClearIntermediateFiles() { |
1226 | 0 | std::vector<std::string> res; |
1227 | 0 | for (const auto& builder : builders_) { |
1228 | 0 | std::vector<std::string> files = |
1229 | 0 | builder.second->version_builder()->GetAndClearIntermediateFiles(); |
1230 | 0 | res.insert(res.end(), std::make_move_iterator(files.begin()), |
1231 | 0 | std::make_move_iterator(files.end())); |
1232 | 0 | } |
1233 | 0 | return res; |
1234 | 0 | } |
1235 | | |
1236 | | Status ManifestTailer::VerifyFile(ColumnFamilyData* cfd, |
1237 | | const std::string& fpath, int level, |
1238 | 0 | const FileMetaData& fmeta) { |
1239 | 0 | Status s = |
1240 | 0 | VersionEditHandlerPointInTime::VerifyFile(cfd, fpath, level, fmeta); |
1241 | | // TODO: Open file or create hard link to prevent the file from being |
1242 | | // deleted. |
1243 | 0 | return s; |
1244 | 0 | } |
1245 | | |
1246 | | void DumpManifestHandler::CheckIterationResult(const log::Reader& reader, |
1247 | 0 | Status* s) { |
1248 | 0 | VersionEditHandler::CheckIterationResult(reader, s); |
1249 | 0 | if (!s->ok()) { |
1250 | 0 | fprintf(stdout, "%s\n", s->ToString().c_str()); |
1251 | 0 | return; |
1252 | 0 | } |
1253 | 0 | assert(cf_to_cmp_names_); |
1254 | 0 | for (auto* cfd : *(version_set_->column_family_set_)) { |
1255 | 0 | fprintf(stdout, |
1256 | 0 | "--------------- Column family \"%s\" (ID %" PRIu32 |
1257 | 0 | ") --------------\n", |
1258 | 0 | cfd->GetName().c_str(), cfd->GetID()); |
1259 | 0 | fprintf(stdout, "log number: %" PRIu64 "\n", cfd->GetLogNumber()); |
1260 | 0 | auto it = cf_to_cmp_names_->find(cfd->GetID()); |
1261 | 0 | if (it != cf_to_cmp_names_->end()) { |
1262 | 0 | fprintf(stdout, |
1263 | 0 | "comparator: <%s>, but the comparator object is not available.\n", |
1264 | 0 | it->second.c_str()); |
1265 | 0 | } else { |
1266 | 0 | fprintf(stdout, "comparator: %s\n", cfd->user_comparator()->Name()); |
1267 | 0 | } |
1268 | 0 | assert(cfd->current()); |
1269 | | |
1270 | | // Print out DebugStrings. Can include non-terminating null characters. |
1271 | 0 | fwrite(cfd->current()->DebugString(hex_).data(), sizeof(char), |
1272 | 0 | cfd->current()->DebugString(hex_).size(), stdout); |
1273 | |
|
1274 | 0 | fprintf(stdout, |
1275 | 0 | "By default, manifest file dump prints LSM trees as if %d levels " |
1276 | 0 | "were configured, " |
1277 | 0 | "which is not necessarily true for the column family (CF) this " |
1278 | 0 | "manifest is associated with. " |
1279 | 0 | "Please consult other DB files, such as the OPTIONS file, to " |
1280 | 0 | "confirm.\n", |
1281 | 0 | cfd->ioptions().num_levels); |
1282 | 0 | } |
1283 | | fprintf(stdout, |
1284 | 0 | "next_file_number %" PRIu64 " last_sequence %" PRIu64 |
1285 | 0 | " prev_log_number %" PRIu64 " max_column_family %" PRIu32 |
1286 | 0 | " min_log_number_to_keep %" PRIu64 "\n", |
1287 | 0 | version_set_->current_next_file_number(), |
1288 | 0 | version_set_->LastSequence(), version_set_->prev_log_number(), |
1289 | 0 | version_set_->column_family_set_->GetMaxColumnFamily(), |
1290 | 0 | version_set_->min_log_number_to_keep()); |
1291 | 0 | } |
1292 | | |
1293 | | } // namespace ROCKSDB_NAMESPACE |