Coverage Report

Created: 2026-11-12 08:06

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/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