| 2578 | } |
| 2579 | |
| 2580 | Status ImpalaServer::ProcessCatalogUpdateResult( |
| 2581 | const TCatalogUpdateResult& catalog_update_result, bool wait_for_all_subscribers, |
| 2582 | const TQueryOptions& query_options, RuntimeProfile::EventSequence* timeline) { |
| 2583 | if (!FLAGS_catalogd_deployed) return Status::OK(); |
| 2584 | const TUniqueId& catalog_service_id = catalog_update_result.catalog_service_id; |
| 2585 | if (!catalog_update_result.__isset.updated_catalog_objects && |
| 2586 | !catalog_update_result.__isset.removed_catalog_objects) { |
| 2587 | // Operation with no result set. Use the version specified in |
| 2588 | // 'catalog_update_result' to determine when the effects of this operation |
| 2589 | // have been applied to the local catalog cache. |
| 2590 | if (catalog_update_result.is_invalidate) { |
| 2591 | WaitForMinCatalogUpdate(catalog_update_result.version, catalog_service_id, |
| 2592 | timeline); |
| 2593 | } else { |
| 2594 | WaitForCatalogUpdate(catalog_update_result.version, catalog_service_id, timeline); |
| 2595 | } |
| 2596 | if (wait_for_all_subscribers) { |
| 2597 | // Now wait for this update to be propagated to all catalog topic subscribers. |
| 2598 | // If we make it here it implies the first condition was met (the update was |
| 2599 | // processed locally or the catalog service id has changed). |
| 2600 | WaitForCatalogUpdateTopicPropagation(catalog_service_id, timeline); |
| 2601 | } |
| 2602 | } else { |
| 2603 | TUniqueId cur_service_id; |
| 2604 | { |
| 2605 | Status status = DebugAction(query_options, "WAIT_BEFORE_PROCESSING_CATALOG_UPDATE"); |
| 2606 | DCHECK(status.ok()); |
| 2607 | |
| 2608 | unique_lock<mutex> ver_lock(catalog_version_lock_); |
| 2609 | cur_service_id = catalog_update_info_.catalog_service_id; |
| 2610 | if (cur_service_id != catalog_service_id) { |
| 2611 | LOG(INFO) << "Catalog service ID mismatch. Current ID: " |
| 2612 | << PrintId(cur_service_id) << ". ID in response: " |
| 2613 | << PrintId(catalog_service_id) << ". Catalogd may have been restarted. " |
| 2614 | "Waiting for new catalog update from statestore."; |
| 2615 | WaitForNewCatalogServiceId(cur_service_id, &ver_lock); |
| 2616 | cur_service_id = catalog_update_info_.catalog_service_id; |
| 2617 | } |
| 2618 | } |
| 2619 | |
| 2620 | if (cur_service_id == catalog_service_id) { |
| 2621 | CatalogUpdateResultIterator callback_ctx(catalog_update_result); |
| 2622 | TUpdateCatalogCacheRequest update_req; |
| 2623 | update_req.__set_is_delta(true); |
| 2624 | update_req.__set_native_iterator_ptr(reinterpret_cast<int64_t>(&callback_ctx)); |
| 2625 | // The catalog version is updated in WaitForCatalogUpdate below. So we need a |
| 2626 | // standalone field in the request to update the service ID without touching the |
| 2627 | // catalog version. |
| 2628 | update_req.__set_catalog_service_id(catalog_update_result.catalog_service_id); |
| 2629 | // Apply the changes to the local catalog cache. |
| 2630 | TUpdateCatalogCacheResponse resp; |
| 2631 | Status status = exec_env_->frontend()->UpdateCatalogCache(update_req, &resp); |
| 2632 | MarkTimelineEvent(timeline, "Applied catalog updates from DDL"); |
| 2633 | if (!status.ok()) LOG(ERROR) << status.GetDetail(); |
| 2634 | RETURN_IF_ERROR(status); |
| 2635 | } else { |
| 2636 | // We can't apply updates on another service id, because the local catalog is still |
| 2637 | // inconsistent with the catalogd that executes the DDL/DML. |
no test coverage detected