Method ReportExecStatus

be/src/runtime/query-state.cc:669–789

}

bool QueryState::ReportExecStatus() {
#ifndef NDEBUG
  if (FLAGS_stress_status_report_delay_ms) {
    LOG(INFO) << "Sleeping " << FLAGS_stress_status_report_delay_ms << "ms before "
              << "reporting for query " << PrintId(query_id());
    SleepForMs(FLAGS_stress_status_report_delay_ms);
  }
#endif
  bool instances_started = fis_map_.size() > 0;

  // This will send a report even if we are cancelled.  If the query completed correctly
  // but fragments still need to be cancelled (e.g. limit reached), the coordinator will
  // be waiting for a final report and profile.
  ReportExecStatusRequestPB report;

  // Gather the statuses and profiles of the fragment instances.
  TRuntimeProfileForest profiles_forest;
  ConstructReport(instances_started, &report, &profiles_forest);

  // Serialize the runtime profile with Thrift to 'profile_buf'. Note that the
  // serialization output is owned by 'serializer' so this must be alive until RPC
  // is done.
  ThriftSerializer serializer(true);
  uint8_t* profile_buf = nullptr;
  uint32_t profile_len = 0;
  Status serialize_status =
      serializer.SerializeToBuffer(&profiles_forest, &profile_len, &profile_buf);
  if (UNLIKELY(!serialize_status.ok() ||
          profile_len > FLAGS_rpc_max_message_size ||
          !DebugAction(query_options(), "REPORT_EXEC_STATUS_PROFILE").ok())) {
    profile_buf = nullptr;
    LOG(ERROR) << Substitute("Failed to create $0profile for query $1: "
        "status=$2 len=$3", IsTerminalState() ? "final " : "", PrintId(query_id()),
        serialize_status.ok() ? "OK" : serialize_status.GetDetail(), profile_len);
  }

  Status rpc_status;
  Status result_status;
  RpcController rpc_controller;

  // The profile is a thrift structure serialized to a string and sent as a sidecar.
  // We keep the runtime profile as Thrift object as Impala client still communicates
  // with Impala server with Thrift RPC.
  //
  // Note that the sidecar is created with faststring so the ownership of the Thrift
  // profile buffer is transferred to RPC layer and it is freed after the RPC payload
  // is sent. If serialization of the profile to RPC sidecar fails, we will proceed
  // without the profile so that the coordinator can still get the status and won't
  // conclude that the backend has hung and cancel the query.
  if (profile_buf != nullptr) {
    kudu::faststring sidecar_buf;
    sidecar_buf.assign_copy(profile_buf, profile_len);
    unique_ptr<RpcSidecar> sidecar = RpcSidecar::FromFaststring(move(sidecar_buf));

    int sidecar_idx;
    kudu::Status sidecar_status =
        rpc_controller.AddOutboundSidecar(move(sidecar), &sidecar_idx);
    if (LIKELY(sidecar_status.ok())) {