| 382 | } |
| 383 | |
| 384 | Status AdmissionExecRequestCompressed::GetQueryExecRequest( |
| 385 | const TQueryExecRequest** out_req) const { |
| 386 | lock_guard<std::mutex> l(lock_); |
| 387 | if (decompressed_req_ != nullptr) { |
| 388 | *out_req = decompressed_req_.get(); |
| 389 | return Status::OK(); |
| 390 | } |
| 391 | DCHECK(req_ != nullptr); |
| 392 | |
| 393 | int64_t uncomp_size = req_->uncompressed_size; |
| 394 | MemTracker* tracker = admission_controller_->pending_decompression_mem_tracker_.get(); |
| 395 | DCHECK(tracker != nullptr); |
| 396 | tracker->Consume(uncomp_size); |
| 397 | const auto tracker_releaser = |
| 398 | MakeScopeExitTrigger([&]() { tracker->Release(uncomp_size); }); |
| 399 | string rejection_reason; |
| 400 | if (RejectForAdmissionServiceMemory(tracker->consumption(), &rejection_reason)) { |
| 401 | // It is okay we set the pool's name to N/A as the pool is not related. |
| 402 | return Status::Expected( |
| 403 | ErrorMsg(TErrorCode::ADMISSION_REJECTED, "N/A", rejection_reason)); |
| 404 | } |
| 405 | |
| 406 | unique_ptr<TQueryExecRequest> new_req = std::make_unique<TQueryExecRequest>(); |
| 407 | RETURN_IF_ERROR(DecompressThrift(*req_, new_req.get())); |
| 408 | DCHECK(new_req != nullptr); |
| 409 | |
| 410 | int64_t comp_size = req_->compressed_data.size(); |
| 411 | float ratio = (comp_size > 0) ? (static_cast<float>(uncomp_size) / comp_size) : 0.0f; |
| 412 | LOG(INFO) << "Decompress TQueryExecRequest for query " |
| 413 | << PrintId(new_req->query_ctx.query_id) << ": Compressed size=" << comp_size |
| 414 | << " B, Uncompressed size=" << uncomp_size << " B, Ratio=" << ratio << "x"; |
| 415 | |
| 416 | // Update the compression stats only once. |
| 417 | if (first_decompress_) { |
| 418 | cached_query_options_ = new_req->query_ctx.client_request.query_options; |
| 419 | if (admission_controller_ != nullptr) { |
| 420 | admission_controller_->UpdateAdmissionRequestCompressionStats( |
| 421 | comp_size, uncomp_size, ratio); |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | // Store the object in our cache. |
| 426 | decompressed_req_ = std::move(new_req); |
| 427 | *out_req = decompressed_req_.get(); |
| 428 | first_decompress_ = false; |
| 429 | |
| 430 | return Status::OK(); |
| 431 | } |
| 432 | |
| 433 | void AdmissionController::UpdateAdmissionRequestCompressionStats( |
| 434 | int64_t compressed_size, int64_t uncompressed_size, float ratio) const { |
no test coverage detected