Inside Bitcoin

1251  const bool seek_compaction = (current_->file_to_compact_ != nullptr);
1252  if (size_compaction) {
1253    level = current_->compaction_level_;
1254    assert(level >= 0);
1255    assert(level + 1 < config::kNumLevels);
1256    c = new Compaction(options_, level);
1257
1258    // Pick the first file that comes after compact_pointer_[level]
1259    for (size_t i = 0; i < current_->files_[level].size(); i++) {
1260      FileMetaData* f = current_->files_[level][i];
1261      if (compact_pointer_[level].empty() ||
1262          icmp_.Compare(f->largest.Encode(), compact_pointer_[level]) > 0) {
1263        c->inputs_[0].push_back(f);
1264        break;
1265      }
1266    }
1267    if (c->inputs_[0].empty()) {
1268      // Wrap-around to the beginning of the key space
1269      c->inputs_[0].push_back(current_->files_[level][0]);
1270    }
1271  } else if (seek_compaction) {
1272    level = current_->file_to_compact_level_;
1273    c = new Compaction(options_, level);
1274    c->inputs_[0].push_back(current_->file_to_compact_);
1275  } else {
1276    return nullptr;
1277  }
1278
1279  c->input_version_ = current_;
1280  c->input_version_->Ref();
1281
1282  // Files in level 0 may overlap each other, so pick up all overlapping ones
1283  if (level == 0) {
1284    InternalKey smallest, largest;
1285    GetRange(c->inputs_[0], &smallest, &largest);
1286    // Note that the next call will discard the file we placed in
1287    // c->inputs_[0] earlier and replace it with an overlapping set
1288    // which will include the picked file.
1289    current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]);
1290    assert(!c->inputs_[0].empty());
1291  }
1292
1293  SetupOtherInputs(c);
1294
1295  return c;
1296}
1297
1298// Finds the largest key in a vector of files. Returns true if files it not
1299// empty.
1300bool FindLargestKey(const InternalKeyComparator& icmp,
1301                    const std::vector<FileMetaData*>& files,
1302                    InternalKey* largest_key) {
1303  if (files.empty()) {
1304    return false;
1305  }
1306  *largest_key = files[0]->largest;
1307  for (size_t i = 1; i < files.size(); ++i) {
1308    FileMetaData* f = files[i];
1309    if (icmp.Compare(f->largest, *largest_key) > 0) {
1310      *largest_key = f->largest;
1311    }
1312  }
1313  return true;
1314}
1315
1316// Finds minimum file b2=(l2, u2) in level file for which l2 > u1 and
1317// user_key(l2) = user_key(u1)
1318FileMetaData* FindSmallestBoundaryFile(
1319    const InternalKeyComparator& icmp,
1320    const std::vector<FileMetaData*>& level_files,
1321    const InternalKey& largest_key) {
1322  const Comparator* user_cmp = icmp.user_comparator();
1323  FileMetaData* smallest_boundary_file = nullptr;
1324  for (size_t i = 0; i < level_files.size(); ++i) {
1325    FileMetaData* f = level_files[i];
1326    if (icmp.Compare(f->smallest, largest_key) > 0 &&
1327        user_cmp->Compare(f->smallest.user_key(), largest_key.user_key()) ==
1328            0) {
1329      if (smallest_boundary_file == nullptr ||
1330          icmp.Compare(f->smallest, smallest_boundary_file->smallest) < 0) {
1331        smallest_boundary_file = f;
1332      }
1333    }
1334  }
1335  return smallest_boundary_file;
1336}
1337
1338// Extracts the largest file b1 from |compaction_files| and then searches for a
1339// b2 in |level_files| for which user_key(u1) = user_key(l2). If it finds such a
1340// file b2 (known as a boundary file) it adds it to |compaction_files| and then
1341// searches again using this new upper bound.
1342//
1343// If there are two blocks, b1=(l1, u1) and b2=(l2, u2) and
1344// user_key(u1) = user_key(l2), and if we compact b1 but not b2 then a
1345// subsequent get operation will yield an incorrect result because it will
1346// return the record from b2 in level i rather than from b1 because it searches
1347// level by level for records matching the supplied user key.
1348//
1349// parameters:
1350//   in     level_files:      List of files to search for boundary files.
1351//   in/out compaction_files: List of files to extend by adding boundary files.
1352void AddBoundaryInputs(const InternalKeyComparator& icmp,
1353                       const std::vector<FileMetaData*>& level_files,
1354                       std::vector<FileMetaData*>* compaction_files) {
1355  InternalKey largest_key;
1356
1357  // Quick return if compaction_files is empty.
1358  if (!FindLargestKey(icmp, *compaction_files, &largest_key)) {
1359    return;
1360  }
1361
1362  bool continue_searching = true;
1363  while (continue_searching) {
1364    FileMetaData* smallest_boundary_file =
1365        FindSmallestBoundaryFile(icmp, level_files, largest_key);
1366
1367    // If a boundary file was found advance largest_key, otherwise we're done.
1368    if (smallest_boundary_file != NULL) {
1369      compaction_files->push_back(smallest_boundary_file);
1370      largest_key = smallest_boundary_file->largest;
1371    } else {
1372      continue_searching = false;
1373    }
1374  }
1375}
1376
1377void VersionSet::SetupOtherInputs(Compaction* c) {
1378  const int level = c->level();
1379  InternalKey smallest, largest;
1380
1381  AddBoundaryInputs(icmp_, current_->files_[level], &c->inputs_[0]);
1382  GetRange(c->inputs_[0], &smallest, &largest);
1383
1384  current_->GetOverlappingInputs(level + 1, &smallest, &largest,
1385                                 &c->inputs_[1]);
1386
1387  // Get entire range covered by compaction
1388  InternalKey all_start, all_limit;
1389  GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
1390
1391  // See if we can grow the number of inputs in "level" without
1392  // changing the number of "level+1" files we pick up.
1393  if (!c->inputs_[1].empty()) {
1394    std::vector<FileMetaData*> expanded0;
1395    current_->GetOverlappingInputs(level, &all_start, &all_limit, &expanded0);
1396    AddBoundaryInputs(icmp_, current_->files_[level], &expanded0);
1397    const int64_t inputs0_size = TotalFileSize(c->inputs_[0]);
1398    const int64_t inputs1_size = TotalFileSize(c->inputs_[1]);
1399    const int64_t expanded0_size = TotalFileSize(expanded0);
1400    if (expanded0.size() > c->inputs_[0].size() &&
1401        inputs1_size + expanded0_size <
1402            ExpandedCompactionByteSizeLimit(options_)) {
1403      InternalKey new_start, new_limit;
1404      GetRange(expanded0, &new_start, &new_limit);
1405      std::vector<FileMetaData*> expanded1;
1406      current_->GetOverlappingInputs(level + 1, &new_start, &new_limit,
1407                                     &expanded1);
1408      if (expanded1.size() == c->inputs_[1].size()) {
1409        Log(options_->info_log,
1410            "Expanding@%d %d+%d (%ld+%ld bytes) to %d+%d (%ld+%ld bytes)\n",
1411            level, int(c->inputs_[0].size()), int(c->inputs_[1].size()),
1412            long(inputs0_size), long(inputs1_size), int(expanded0.size()),
1413            int(expanded1.size()), long(expanded0_size), long(inputs1_size));
1414        smallest = new_start;
1415        largest = new_limit;
1416        c->inputs_[0] = expanded0;
1417        c->inputs_[1] = expanded1;
1418        GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
1419      }
1420    }
1421  }
1422
1423  // Compute the set of grandparent files that overlap this compaction
1424  // (parent == level+1; grandparent == level+2)
1425  if (level + 2 < config::kNumLevels) {
1426    current_->GetOverlappingInputs(level + 2, &all_start, &all_limit,
1427                                   &c->grandparents_);
1428  }
1429
1430  // Update the place where we will do the next compaction for this level.
1431  // We update this immediately instead of waiting for the VersionEdit
1432  // to be applied so that if the compaction fails, we will try a different
1433  // key range next time.
1434  compact_pointer_[level] = largest.Encode().ToString();
1435  c->edit_.SetCompactPointer(level, largest);
1436}
1437
1438Compaction* VersionSet::CompactRange(int level, const InternalKey* begin,
1439                                     const InternalKey* end) {
1440  std::vector<FileMetaData*> inputs;
1441  current_->GetOverlappingInputs(level, begin, end, &inputs);
1442  if (inputs.empty()) {
1443    return nullptr;
1444  }
1445
1446  // Avoid compacting too much in one shot in case the range is large.
1447  // But we cannot do this for level-0 since level-0 files can overlap
1448  // and we must not pick one file and drop another older file if the
1449  // two files overlap.
1450  if (level > 0) {
1451    const uint64_t limit = MaxFileSizeForLevel(options_, level);
1452    uint64_t total = 0;
1453    for (size_t i = 0; i < inputs.size(); i++) {
1454      uint64_t s = inputs[i]->file_size;
1455      total += s;
1456      if (total >= limit) {
1457        inputs.resize(i + 1);
1458        break;
1459      }
1460    }
1461  }
1462
1463  Compaction* c = new Compaction(options_, level);
1464  c->input_version_ = current_;
1465  c->input_version_->Ref();
1466  c->inputs_[0] = inputs;
1467  SetupOtherInputs(c);
1468  return c;
1469}
1470
1471Compaction::Compaction(const Options* options, int level)
1472    : level_(level),
1473      max_output_file_size_(MaxFileSizeForLevel(options, level)),
1474      input_version_(nullptr),
1475      grandparent_index_(0),
1476      seen_key_(false),
1477      overlapped_bytes_(0) {
1478  for (int i = 0; i < config::kNumLevels; i++) {
1479    level_ptrs_[i] = 0;
1480  }
1481}
1482
1483Compaction::~Compaction() {
1484  if (input_version_ != nullptr) {
1485    input_version_->Unref();
1486  }
1487}
1488
1489bool Compaction::IsTrivialMove() const {
1490  const VersionSet* vset = input_version_->vset_;
1491  // Avoid a move if there is lots of overlapping grandparent data.
1492  // Otherwise, the move could create a parent file that will require
1493  // a very expensive merge later on.
1494  return (num_input_files(0) == 1 && num_input_files(1) == 0 &&
1495          TotalFileSize(grandparents_) <=
1496              MaxGrandParentOverlapBytes(vset->options_));
1497}
1498
1499void Compaction::AddInputDeletions(VersionEdit* edit) {
1500  for (int which = 0; which < 2; which++) {