| 98 | } |
| 99 | |
| 100 | Status CatalogOpExecutor::Exec(const TCatalogOpRequest& request) { |
| 101 | Status status; |
| 102 | DCHECK(profile_ != NULL); |
| 103 | RuntimeProfile::Counter* exec_timer = ADD_TIMER(profile_, "CatalogOpExecTimer"); |
| 104 | SCOPED_TIMER(exec_timer); |
| 105 | RETURN_IF_ERROR(status); |
| 106 | switch (request.op_type) { |
| 107 | case TCatalogOpType::DDL: { |
| 108 | // Compute stats stmts must be executed via ExecComputeStats(). |
| 109 | DCHECK(request.ddl_params.ddl_type != TDdlType::COMPUTE_STATS); |
| 110 | |
| 111 | if (!FLAGS_catalogd_deployed) { |
| 112 | return Status("Operation is not supported without CatalogD."); |
| 113 | } |
| 114 | |
| 115 | exec_response_.reset(new TDdlExecResponse()); |
| 116 | int attempt = 0; // Used for debug action only. |
| 117 | CatalogServiceConnection::RpcStatus rpc_status = |
| 118 | CatalogServiceConnection::DoRpcWithRetry(env_->catalogd_client_cache(), |
| 119 | *ExecEnv::GetInstance()->GetCatalogdAddress().get(), |
| 120 | &CatalogServiceClientWrapper::ExecDdl, request.ddl_params, |
| 121 | FLAGS_catalog_client_connection_num_retries, |
| 122 | FLAGS_catalog_client_rpc_retry_interval_ms, |
| 123 | [&attempt]() { return CatalogRpcDebugFn(&attempt); }, exec_response_.get()); |
| 124 | RETURN_IF_ERROR(rpc_status.status); |
| 125 | if (FLAGS_use_local_catalog) VerifyMinimalResponse(exec_response_->result); |
| 126 | catalog_update_result_.reset( |
| 127 | new TCatalogUpdateResult(exec_response_->result)); |
| 128 | Status status = StatusFromThrift(exec_response_->result.status); |
| 129 | if (status.ok()) { |
| 130 | if (request.ddl_params.ddl_type == TDdlType::DROP_FUNCTION) { |
| 131 | HandleDropFunction(request.ddl_params.drop_fn_params); |
| 132 | } else if (request.ddl_params.ddl_type == TDdlType::DROP_DATA_SOURCE) { |
| 133 | HandleDropDataSource(request.ddl_params.drop_data_source_params); |
| 134 | } |
| 135 | } |
| 136 | if (exec_response_->__isset.profile) { |
| 137 | catalog_profile_ = make_unique<TRuntimeProfileNode>(exec_response_->profile); |
| 138 | } |
| 139 | return status; |
| 140 | } |
| 141 | case TCatalogOpType::RESET_METADATA: { |
| 142 | if (!FLAGS_catalogd_deployed) { |
| 143 | if (request.__isset.reset_metadata_params) { |
| 144 | const TResetMetadataRequest& metadata_params = request.reset_metadata_params; |
| 145 | if (metadata_params.authorization) { |
| 146 | return fe_->RefreshAuthorization(); |
| 147 | } |
| 148 | } |
| 149 | // We don't cache metadata without CatalogD. |
| 150 | return Status::OK(); |
| 151 | } |
| 152 | |
| 153 | TResetMetadataResponse response; |
| 154 | int attempt = 0; // Used for debug action only. |
| 155 | CatalogServiceConnection::RpcStatus rpc_status = |
| 156 | CatalogServiceConnection::DoRpcWithRetry(env_->catalogd_client_cache(), |
| 157 | *ExecEnv::GetInstance()->GetCatalogdAddress().get(), |