Lazy callbacks own their backing through captured anchors; dataset removal only drops entries, so move them by value.
| 533 | |
| 534 | // Lazy callbacks own their backing through captured anchors; dataset removal only drops entries, so move them by value. |
| 535 | Expected<ObjectDatasetMergeReport> ObjectStore::mergeDatasets( |
| 536 | DatasetId anchor_id, const std::vector<DatasetMergeSource>& sources) { |
| 537 | std::unique_lock lock(store_mutex_); |
| 538 | |
| 539 | std::unordered_set<DatasetId> seen_sources; |
| 540 | for (const DatasetMergeSource& src : sources) { |
| 541 | if (src.dataset_id == anchor_id) { |
| 542 | return unexpected("mergeDatasets: source dataset " + std::to_string(src.dataset_id) + " is the anchor"); |
| 543 | } |
| 544 | if (!seen_sources.insert(src.dataset_id).second) { |
| 545 | return unexpected("mergeDatasets: source dataset " + std::to_string(src.dataset_id) + " listed more than once"); |
| 546 | } |
| 547 | } |
| 548 | |
| 549 | struct Contributor { |
| 550 | ObjectTopicId topic_id; |
| 551 | ObjectSeries* series; |
| 552 | Timestamp shift; |
| 553 | }; |
| 554 | struct Group { |
| 555 | ObjectTopicId destination_id; |
| 556 | ObjectSeries* destination; |
| 557 | Timestamp destination_shift = 0; |
| 558 | bool destination_needs_reparent = false; |
| 559 | std::vector<Contributor> sources; |
| 560 | }; |
| 561 | |
| 562 | ObjectDatasetMergeReport report; |
| 563 | std::vector<Group> groups; |
| 564 | groups.reserve(topics_.size()); |
| 565 | std::unordered_map<std::string, size_t> group_by_name; |
| 566 | |
| 567 | for (auto& [tid, series] : topics_) { |
| 568 | if (series->descriptor.dataset_id != anchor_id) { |
| 569 | continue; |
| 570 | } |
| 571 | const auto index = groups.size(); |
| 572 | groups.push_back( |
| 573 | Group{ |
| 574 | .destination_id = tid, |
| 575 | .destination = series.get(), |
| 576 | .destination_shift = 0, |
| 577 | .destination_needs_reparent = false, |
| 578 | .sources = {}, |
| 579 | }); |
| 580 | group_by_name.emplace(series->descriptor.topic_name, index); |
| 581 | } |
| 582 | |
| 583 | for (const DatasetMergeSource& src : sources) { |
| 584 | std::vector<std::pair<ObjectTopicId, ObjectSeries*>> source_topics; |
| 585 | for (auto& [tid, series] : topics_) { |
| 586 | if (series->descriptor.dataset_id == src.dataset_id) { |
| 587 | source_topics.emplace_back(tid, series.get()); |
| 588 | } |
| 589 | } |
| 590 | if (source_topics.empty()) { |
| 591 | continue; |
| 592 | } |