| 2551 | } |
| 2552 | |
| 2553 | void ImpalaServer::WaitForMinCatalogUpdate(const int64_t min_req_catalog_object_version, |
| 2554 | const TUniqueId& catalog_service_id, RuntimeProfile::EventSequence* timeline) { |
| 2555 | unique_lock<mutex> unique_lock(catalog_version_lock_); |
| 2556 | int64_t catalog_object_version_lower_bound = |
| 2557 | catalog_update_info_.catalog_object_version_lower_bound; |
| 2558 | // TODO: Set a timeout to eventually break out of this loop if something goes |
| 2559 | // wrong? |
| 2560 | VLOG_QUERY << "Waiting for local minimum catalog object version to be > " |
| 2561 | << min_req_catalog_object_version << ", current lower bound of local versions: " |
| 2562 | << catalog_object_version_lower_bound; |
| 2563 | while (catalog_update_info_.catalog_service_id == catalog_service_id |
| 2564 | && catalog_update_info_.catalog_object_version_lower_bound <= |
| 2565 | min_req_catalog_object_version) { |
| 2566 | catalog_version_update_cv_.Wait(unique_lock); |
| 2567 | } |
| 2568 | |
| 2569 | if (catalog_update_info_.catalog_service_id != catalog_service_id) { |
| 2570 | MarkTimelineEvent(timeline, "Detected change in catalog service ID"); |
| 2571 | VLOG_QUERY << "Detected change in catalog service ID"; |
| 2572 | } else { |
| 2573 | MarkTimelineEvent(timeline, Substitute("Local min catalog version reached $0", |
| 2574 | min_req_catalog_object_version)); |
| 2575 | VLOG_QUERY << "Updated catalog object version lower bound: " |
| 2576 | << min_req_catalog_object_version; |
| 2577 | } |
| 2578 | } |
| 2579 | |
| 2580 | Status ImpalaServer::ProcessCatalogUpdateResult( |
| 2581 | const TCatalogUpdateResult& catalog_update_result, bool wait_for_all_subscribers, |
nothing calls this directly
no test coverage detected