Draft distributed plan

programs/client/Client.cpp

@@ -329,7 +329,7 @@ bool Client::executeMultiQuery(const String & all_queries_text)
 std::vector<String> Client::loadWarningMessages()
 {
     std::vector<String> messages;
-    connection->sendQuery(connection_parameters.timeouts, "SELECT message FROM system.warnings", "" /* query_id */,
+    connection->sendQuery(connection_parameters.timeouts, "SELECT message FROM system.warnings", "zzzzzzzz-zzzz-zzzz-zzzz-zzzzzzzzzzzz" /* query_id */,
                           QueryProcessingStage::Complete, nullptr, nullptr, false);
     while (true)
     {

programs/server/Server.cpp

@@ -509,6 +509,22 @@ if (ThreadFuzzer::instance().isEffective())
     global_context->addWarningMessage("Server was built with sanitizer. It will work slowly.");
 #endif
 
+    if (config().has("running_mode"))
+    {
+        String running_mode = config().getString("running_mode", "compute");
+        if (running_mode == "compute")
+        {
+            global_context->setRunningMode(Context::RunningMode::COMPUTE);
+            LOG_INFO(log, "Running mode is COMPUTE");
+        }
+        else if (running_mode == "store")
+        {
+            global_context->setRunningMode(Context::RunningMode::STORE);
+            LOG_INFO(log, "Running mode is STORE");
+        } else {
+            throw Exception("Invalid running_mode", ErrorCodes::INVALID_CONFIG_PARAMETER);
+        }
+    }
 
     // Initialize global thread pool. Do it before we fetch configs from zookeeper
     // nodes (`from_zk`), because ZooKeeper interface uses the pool. We will
@@ -1546,6 +1562,9 @@ if (ThreadFuzzer::instance().isEffective())
                                                                      &CurrentMetrics::MaxDDLEntryID, &CurrentMetrics::MaxPushedDDLEntryID));
         }
 
+        /// Register with clickhouse-keeper and watch on clusters change.
+        global_context->setClustersWatcher(std::make_unique<ClustersWatcher>(DEFAULT_ZOOKEEPER_CLUSTERS_PATH, global_context));
+
         for (auto & server : *servers)
             server.start();
         LOG_INFO(log, "Ready for connections.");

src/Client/GRPCClient.cpp

@@ -0,0 +1,144 @@
+#include <sstream>
+#include <Client/GRPCClient.h>
+#include <Formats/FormatFactory.h>
+#include <Formats/NativeReader.h>
+#include <IO/ReadBufferFromString.h>
+#include <grpcpp/grpcpp.h>
+#include "Common/ErrorCodes.h"
+#include "base/logger_useful.h"
+#include "clickhouse_grpc.grpc.pb.h"
+#include "clickhouse_grpc.pb.h"
+
+namespace Poco
+{
+class Logger;
+}
+
+using GRPCQueryInfo = clickhouse::grpc::QueryInfo;
+using GRPCResult = clickhouse::grpc::Result;
+using GRPCException = clickhouse::grpc::Exception;
+using GRPCTicket = clickhouse::grpc::Ticket;
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int INVALID_GRPC_QUERY_INFO;
+    extern const int GRPC_READ_ERROR;
+    extern const int GRPC_CANCEL_ERROR;
+}
+
+GRPCClient::GRPCClient(const String & addr_, const String & description_)
+{
+    addr = addr_;
+    log = &Poco::Logger::get("GRPCClient(" + description_ + ")");
+}
+
+GRPCResult GRPCClient::executePlanFragment(const GRPCQueryInfo & query_info)
+{
+    auto ch = grpc::CreateChannel(addr, grpc::InsecureChannelCredentials());
+    auto stub = clickhouse::grpc::ClickHouse::NewStub(ch);
+    grpc::ClientContext ctx;
+    GRPCResult result;
+    grpc::Status status = stub->ExecutePlanFragment(&ctx, query_info, &result);
+
+    if (status.ok())
+        return result;
+    else
+    {
+        LOG_ERROR(
+            log,
+            "Send query info to {} failed, code: {}, plan fragment id: {}.",
+            addr,
+            status.error_code(),
+            query_info.initial_query_id() + "/" + toString(query_info.stage_id()) + "/" + query_info.node_id());
+        throw Exception(status.error_message() + ", " + result.exception().display_text(), ErrorCodes::INVALID_GRPC_QUERY_INFO, true);
+    }
+}
+
+void GRPCClient::prepareRead(const GRPCTicket & ticket_)
+{
+    ticket = ticket_;
+    grpc::ChannelArguments arg;
+    arg.SetMaxReceiveMessageSize(-1);
+    auto ch = grpc::CreateCustomChannel(addr, grpc::InsecureChannelCredentials(), arg);
+    std::shared_ptr<grpc::ClientContext> ctx = std::make_shared<grpc::ClientContext>();
+    auto stub = clickhouse::grpc::ClickHouse::NewStub(ch);
+    auto reader = stub->FetchPlanFragmentResult(ctx.get(), ticket);
+    inner_context = std::make_unique<InnerContext>(ch, ctx, stub, reader);
+}
+
+GRPCClient::MessageType GRPCClient::read(Block & block)
+{
+    assert(inner_context);
+
+    GRPCResult result;
+    if (inner_context->reader->Read(&result))
+    {
+        if (result.exception().code() != 0)
+        {
+            LOG_ERROR(
+                log,
+                "Read failed, exception.code: {}, exception.text: {}.",
+                result.exception().code(),
+                result.exception().display_text());
+            throw Exception(result.exception().display_text(), result.exception().code(), true);
+        }
+
+        /// Note: totals and extremes are not used and not tested yet.
+        if (result.totals().size() > 0)
+        {
+            ReadBufferFromString b(result.totals());
+            NativeReader reader(b, 0);
+            block = reader.read();
+            return MessageType::Totals;
+        }
+
+        if (result.extremes().size() > 0)
+        {
+            ReadBufferFromString b(result.extremes());
+            NativeReader reader(b, 0);
+            block = reader.read();
+            return MessageType::Extremes;
+        }
+
+        if (!result.output().empty())
+        {
+            ReadBufferFromString b(result.output());
+            NativeReader reader(b, 0);
+            block = reader.read();
+        }
+        return MessageType::Data;
+    }
+
+    throw Exception(
+        "Read from grpc server " + addr + " failed, " + toString(result.exception().code()) + ", " + result.exception().display_text(),
+        ErrorCodes::GRPC_READ_ERROR,
+        true);
+}
+
+void GRPCClient::cancel()
+{
+    grpc::ClientContext ctx;
+    GRPCResult result;
+
+    auto status = inner_context->stub->CancelPlanFragment(&ctx, ticket, &result);
+
+    auto plan_fragment_id = ticket.initial_query_id() + "/" + toString(ticket.stage_id()) + "/" + ticket.node_id();
+    if (status.ok())
+    {
+        if (result.cancelled())
+            LOG_DEBUG(log, "Cancel success, plan fragment id: {}", plan_fragment_id);
+        else
+        {
+            throw Exception("Cancel failed from " + addr + ", plan fragment id: " + plan_fragment_id + ", code: " + toString(result.exception().code()) + ", " + result.exception().display_text(), ErrorCodes::GRPC_CANCEL_ERROR, true);
+        }
+    }
+    else
+    {
+        LOG_ERROR(
+            log, "Cancel failed, code: {}, plan fragment id: {}.", status.error_code(), plan_fragment_id);
+        throw Exception(status.error_message() + ", " + result.exception().display_text(), ErrorCodes::GRPC_CANCEL_ERROR, true);
+    }
+}
+}

src/Client/GRPCClient.h

@@ -0,0 +1,78 @@
+#pragma once
+//#if USE_GRPC
+#include <map>
+#include <memory>
+#include <shared_mutex>
+
+#include <grpcpp/grpcpp.h>
+#include <grpcpp/impl/codegen/proto_utils.h>
+#include <grpcpp/impl/codegen/sync_stream.h>
+
+#include <Core/Block.h>
+#include <base/types.h>
+#include <Poco/Net/SocketAddress.h>
+#include "clickhouse_grpc.grpc.pb.h"
+#include "clickhouse_grpc.pb.h"
+
+using GRPCQueryInfo = clickhouse::grpc::QueryInfo;
+using GRPCResult = clickhouse::grpc::Result;
+using GRPCTicket = clickhouse::grpc::Ticket;
+using GRPCStub = clickhouse::grpc::ClickHouse::Stub;
+
+namespace DB
+{
+using ReadDataCallback = std::function<void(const Block & block)>;
+
+class GRPCClient
+{
+public:
+    enum MessageType
+    {
+        Data = 1,
+        Totals = 2,
+        Extremes = 3,
+        MAX = Extremes,
+    };
+public:
+    GRPCClient(const String & addr_, const String & description_);
+    ~GRPCClient() = default;
+
+    /// Send params of plan fragment to remote, and execute it.
+    GRPCResult executePlanFragment(const GRPCQueryInfo & query_info);
+
+    /// Initialize reader and inner context.
+    void prepareRead(const GRPCTicket & ticket_);
+
+    /// Try to read a block from remote.
+    /// If got EOF, an empty block will be returned, you can use if (!block) to check it.
+    MessageType read(Block & bock);
+
+    /// Cancel plan fragment (ticket associated with the prepareRead)
+    void cancel();
+
+private:
+    struct InnerContext
+    {
+        InnerContext(
+            std::shared_ptr<grpc::Channel> & ch_,
+            std::shared_ptr<grpc::ClientContext> & ctx_,
+            std::unique_ptr<GRPCStub> & stub_,
+            std::unique_ptr<grpc::ClientReader<GRPCResult>> & reader_)
+            : ch(ch_), ctx(ctx_), stub(std::move(stub_)), reader(std::move(reader_))
+        {
+        }
+        ~InnerContext() { }
+
+        std::shared_ptr<grpc::Channel> ch;
+        std::shared_ptr<grpc::ClientContext> ctx;
+        std::unique_ptr<GRPCStub> stub;
+        std::unique_ptr<grpc::ClientReader<GRPCResult>> reader;
+    };
+
+    Poco::Logger * log;
+    String addr;
+    std::unique_ptr<InnerContext> inner_context;
+    GRPCTicket ticket;
+};
+}
+//#endif

src/Client/Suggest.cpp

@@ -141,7 +141,7 @@ void Suggest::load(ContextPtr context, const ConnectionParameters & connection_p
 
 void Suggest::fetch(IServerConnection & connection, const ConnectionTimeouts & timeouts, const std::string & query)
 {
-    connection.sendQuery(timeouts, query, "" /* query_id */, QueryProcessingStage::Complete, nullptr, nullptr, false);
+    connection.sendQuery(timeouts, query, "zzzzzzzz-zzzz-zzzz-zzzz-zzzzzzzzzzzz" /* query_id */, QueryProcessingStage::Complete, nullptr, nullptr, false);
 
     while (true)
     {

src/Common/ErrorCodes.cpp

@@ -593,6 +593,9 @@
     M(623, CAPN_PROTO_BAD_CAST) \
     M(624, BAD_FILE_TYPE) \
     M(625, IO_SETUP_ERROR) \
+    M(702, GRPC_READ_ERROR) \
+    M(703, GRPC_CANCEL_ERROR) \
+    M(704, CANNOT_DROP_DATABASE) \
     \
     M(999, KEEPER_EXCEPTION) \
     M(1000, POCO_EXCEPTION) \

src/Common/ProfileEvents.cpp

@@ -89,6 +89,8 @@
     M(ZooKeeperCheck, "") \
     M(ZooKeeperClose, "") \
     M(ZooKeeperWatchResponse, "") \
+    M(ZooKeeperAddWatch, "") \
+    M(ZooKeeperRemomveWatches, "") \
     M(ZooKeeperUserExceptions, "") \
     M(ZooKeeperHardwareExceptions, "") \
     M(ZooKeeperOtherExceptions, "") \