catalog-server.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 IMPALA_CATALOG_CATALOG_SERVER_H
#define IMPALA_CATALOG_CATALOG_SERVER_H

#include <string>
#include <vector>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/unordered_set.hpp>

#include "gen-cpp/CatalogService.h"
#include "gen-cpp/Frontend_types.h"
#include "gen-cpp/Types_types.h"
#include "catalog/catalog.h"
#include "statestore/statestore-subscriber.h"
#include "util/metrics.h"
#include "rapidjson/rapidjson.h"

namespace impala {

class StatestoreSubscriber;
class Catalog;
class TGetAllCatalogObjectsResponse;

class CatalogServer {
 public:
  static std::string IMPALA_CATALOG_TOPIC;
  CatalogServer(MetricGroup* metrics);

  Status Start();

  void RegisterWebpages(Webserver* webserver);

  const boost::shared_ptr<CatalogServiceIf>& thrift_iface() const {
    return thrift_iface_;
  }
  Catalog* catalog() const { return catalog_.get(); }

 private:
  boost::shared_ptr<CatalogServiceIf> thrift_iface_;
  ThriftSerializer thrift_serializer_;
  MetricGroup* metrics_;
  boost::scoped_ptr<Catalog> catalog_;
  boost::scoped_ptr<StatestoreSubscriber> statestore_subscriber_;

  StatsMetric<double>* topic_processing_time_metric_;

  boost::scoped_ptr<Thread> catalog_update_gathering_thread_;

  boost::unordered_set<std::string> catalog_topic_entry_keys_;

  boost::mutex catalog_lock_;

  boost::condition_variable catalog_update_cv_;

  std::vector<TTopicItem> pending_topic_updates_;

  bool topic_updates_ready_;

  int64_t last_sent_catalog_version_;

  int64_t catalog_objects_min_version_;

  int64_t catalog_objects_max_version_;

  void UpdateCatalogTopicCallback(
      const StatestoreSubscriber::TopicDeltaMap& incoming_topic_deltas,
      std::vector<TTopicDelta>* subscriber_topic_updates);

  void GatherCatalogUpdatesThread();

  void BuildTopicUpdates(const std::vector<TCatalogObject>& catalog_objects);

  void CatalogUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);
};

}

#endif