query-schedule.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef STATESTORE_QUERY_SCHEDULE_H
#define STATESTORE_QUERY_SCHEDULE_H

#include <vector>
#include <string>
#include <boost/unordered_set.hpp>
#include <boost/unordered_map.hpp>
#include <boost/scoped_ptr.hpp>

#include "common/global-types.h"
#include "common/status.h"
#include "statestore/query-resource-mgr.h"
#include "util/promise.h"
#include "util/runtime-profile.h"
#include "gen-cpp/Types_types.h"  // for TNetworkAddress
#include "gen-cpp/Frontend_types.h"
#include "gen-cpp/ResourceBrokerService_types.h"

namespace impala {

class Coordinator;

typedef std::map<TPlanNodeId, std::vector<TScanRangeParams> > PerNodeScanRanges;
typedef boost::unordered_map<TNetworkAddress, PerNodeScanRanges>
    FragmentScanRangeAssignment;

struct FragmentExecParams {
  std::vector<TNetworkAddress> hosts; // execution backends
  std::vector<TUniqueId> instance_ids;
  std::vector<TPlanFragmentDestination> destinations;
  std::map<PlanNodeId, int> per_exch_num_senders;
  FragmentScanRangeAssignment scan_range_assignment;
  int sender_id_base;
};
class QuerySchedule {
 public:
  QuerySchedule(const TUniqueId& query_id, const TQueryExecRequest& request,
      const TQueryOptions& query_options, const std::string& effective_user,
      RuntimeProfile* summary_profile, RuntimeProfile::EventSequence* query_events);

  Status ValidateReservation();

  const TUniqueId& query_id() const { return query_id_; }
  const TQueryExecRequest& request() const { return request_; }
  const TQueryOptions& query_options() const { return query_options_; }
  const std::string& effective_user() const { return effective_user_; }
  const std::string& request_pool() const { return request_pool_; }
  void set_request_pool(const std::string& pool_name) { request_pool_ = pool_name; }
  bool HasReservation() const { return !reservation_.allocated_resources.empty(); }

  bool NeedsRelease() const { return reservation_.__isset.reservation_id; }

  int64_t GetPerHostMemoryEstimate() const;
  int16_t GetPerHostVCores() const;
  int64_t GetClusterMemoryEstimate() const;
  void GetResourceHostport(const TNetworkAddress& src, TNetworkAddress* dst);

  void AddScanRanges(int64_t delta) { num_scan_ranges_ += delta; }
  void set_num_backends(int64_t num_backends) { num_backends_ = num_backends; }
  void set_num_hosts(int64_t num_hosts) {
    DCHECK_GT(num_hosts, 0);
    num_hosts_ = num_hosts;
  }
  int64_t num_backends() const { return num_backends_; }
  int64_t num_hosts() const { return num_hosts_; }
  int64_t num_scan_ranges() const { return num_scan_ranges_; }
  int32_t GetFragmentIdx(PlanNodeId id) const { return plan_node_to_fragment_idx_[id]; }
  std::vector<FragmentExecParams>* exec_params() { return &fragment_exec_params_; }
  const boost::unordered_set<TNetworkAddress>& unique_hosts() const {
    return unique_hosts_;
  }
  TResourceBrokerReservationResponse* reservation() { return &reservation_; }
  const TResourceBrokerReservationRequest& reservation_request() const {
    return reservation_request_;
  }
  bool is_admitted() const { return is_admitted_; }
  void set_is_admitted(bool is_admitted) { is_admitted_ = is_admitted; }
  RuntimeProfile* summary_profile() { return summary_profile_; }
  RuntimeProfile::EventSequence* query_events() { return query_events_; }

  void SetUniqueHosts(const boost::unordered_set<TNetworkAddress>& unique_hosts);

  void PrepareReservationRequest(const std::string& pool, const std::string& user);

 private:

  const TUniqueId& query_id_;
  const TQueryExecRequest& request_;
  const TQueryOptions& query_options_;
  const std::string effective_user_;
  RuntimeProfile* summary_profile_;
  RuntimeProfile::EventSequence* query_events_;

  std::vector<int32_t> plan_node_to_fragment_idx_;

  std::vector<FragmentExecParams> fragment_exec_params_;

  boost::unordered_set<TNetworkAddress> unique_hosts_;

  int64_t num_backends_;

  int64_t num_hosts_;

  int64_t num_scan_ranges_;

  std::string request_pool_;

  TResourceBrokerReservationRequest reservation_request_;

  TResourceBrokerReservationResponse reservation_;

  bool is_admitted_;

  boost::scoped_ptr<ResourceResolver> resource_resolver_;
};

}

#endif