ClickHouse · alexey-milovidov · Feb 21, 2020 · Jan 31, 2020 · Jan 31, 2020 · Jan 31, 2020
diff --git a/dbms/programs/client/Client.cpp b/dbms/programs/client/Client.cpp
@@ -874,7 +874,7 @@ class Client : public Poco::Util::Application
         if (!select && !external_tables.empty())
             throw Exception("External tables could be sent only with select query", ErrorCodes::BAD_ARGUMENTS);
 
-        std::vector<ExternalTableData> data;
+        std::vector<ExternalTableDataPtr> data;
         for (auto & table : external_tables)
             data.emplace_back(table.getData(context));
 

diff --git a/dbms/src/Client/Connection.cpp b/dbms/src/Client/Connection.cpp
@@ -22,6 +22,9 @@
 #include <Common/config_version.h>
 #include <Interpreters/ClientInfo.h>
 #include <Compression/CompressionFactory.h>
+#include <Processors/Pipe.h>
+#include <Processors/ISink.h>
+#include <Processors/Executors/PipelineExecutor.h>
 
 #include <Common/config.h>
 #if USE_POCO_NETSSL
@@ -534,6 +537,45 @@ void Connection::sendScalarsData(Scalars & data)
     LOG_DEBUG(log_wrapper.get(), msg.rdbuf());
 }
 
+namespace
+{
+/// Sink which sends data for external table.
+class ExternalTableDataSink : public ISink
+{
+public:
+    using OnCancell = std::function<void()>;
+
+    ExternalTableDataSink(Block header, Connection & connection_, ExternalTableData & table_data_, OnCancell callback)
+            : ISink(std::move(header)), connection(connection_), table_data(table_data_),
+              on_cancell(std::move(callback))
+    {}
+
+    String getName() const override { return "ExternalTableSink"; }
+
+    size_t getNumReadRows() const { return num_rows; }
+
+protected:
+    void consume(Chunk chunk) override
+    {
+        if (table_data.is_cancelled)
+        {
+            on_cancell();
+            return;
+        }
+
+        num_rows += chunk.getNumRows();
+
+        auto block = getPort().getHeader().cloneWithColumns(chunk.detachColumns());
+        connection.sendData(block, table_data.table_name);
+    }
+
+private:
+    Connection & connection;
+    ExternalTableData & table_data;
+    OnCancell on_cancell;
+    size_t num_rows = 0;
+};
+}
 
 void Connection::sendExternalTablesData(ExternalTablesData & data)
 {
@@ -553,13 +595,24 @@ void Connection::sendExternalTablesData(ExternalTablesData & data)
 
     for (auto & elem : data)
     {
-        elem.first->readPrefix();
-        while (Block block = elem.first->read())
-        {
-            rows += block.rows();
-            sendData(block, elem.second);
-        }
-        elem.first->readSuffix();
+        PipelineExecutorPtr executor;
+        auto on_cancel = [& executor]() { executor->cancel(); };
+
+        auto sink = std::make_shared<ExternalTableDataSink>(elem->pipe->getHeader(), *this, *elem, std::move(on_cancel));
+        DB::connect(elem->pipe->getPort(), sink->getPort());
+
+        auto processors = std::move(*elem->pipe).detachProcessors();
+        processors.push_back(sink);
+
+        executor = std::make_shared<PipelineExecutor>(processors);
+        executor->execute(/*num_threads = */ 1);
+
+        auto read_rows = sink->getNumReadRows();
+        rows += read_rows;
+
+        /// If table is empty, send empty block with name.
+        if (read_rows == 0)
+            sendData(sink->getPort().getHeader(), elem->table_name);
     }
 
     /// Send empty block, which means end of data transfer.

diff --git a/dbms/src/Client/Connection.h b/dbms/src/Client/Connection.h
@@ -30,11 +30,20 @@ namespace DB
 {
 
 class ClientInfo;
+class Pipe;
 
-/// The stream of blocks reading from the table and its name
-using ExternalTableData = std::pair<BlockInputStreamPtr, std::string>;
-/// Vector of pairs describing tables
-using ExternalTablesData = std::vector<ExternalTableData>;
+/// Struct which represents data we are going to send for external table.
+struct ExternalTableData
+{
+    /// Pipe of data form table;
+    std::unique_ptr<Pipe> pipe;
+    std::string table_name;
+    /// Flag if need to stop reading.
+    std::atomic_bool is_cancelled = false;
+};
+
+using ExternalTableDataPtr = std::unique_ptr<ExternalTableData>;
+using ExternalTablesData = std::vector<ExternalTableDataPtr>;
 
 class Connection;
 

diff --git a/dbms/src/Core/ExternalTable.cpp b/dbms/src/Core/ExternalTable.cpp
@@ -9,8 +9,12 @@
 #include <IO/LimitReadBuffer.h>
 #include <Storages/StorageMemory.h>
 #include <Poco/Net/MessageHeader.h>
+#include <Processors/Sources/SourceFromInputStream.h>
+#include <Processors/Pipe.h>
 
 #include <Core/ExternalTable.h>
+#include <Processors/Sources/SinkToOutputStream.h>
+#include <Processors/Executors/PipelineExecutor.h>
 
 
 namespace DB
@@ -22,12 +26,18 @@ namespace ErrorCodes
 }
 
 
-ExternalTableData BaseExternalTable::getData(const Context & context)
+ExternalTableDataPtr BaseExternalTable::getData(const Context & context)
 {
     initReadBuffer();
     initSampleBlock();
     auto input = context.getInputFormat(format, *read_buffer, sample_block, DEFAULT_BLOCK_SIZE);
-    return std::make_pair(std::make_shared<AsynchronousBlockInputStream>(input), name);
+    auto stream = std::make_shared<AsynchronousBlockInputStream>(input);
+
+    auto data = std::make_unique<ExternalTableData>();
+    data->table_name = name;
+    data->pipe = std::make_unique<Pipe>(std::make_shared<SourceFromInputStream>(std::move(stream)));
+
+    return data;
 }
 
 void BaseExternalTable::clean()
@@ -156,22 +166,24 @@ void ExternalTablesHandler::handlePart(const Poco::Net::MessageHeader & header,
     else
         throw Exception("Neither structure nor types have not been provided for external table " + name + ". Use fields " + name + "_structure or " + name + "_types to do so.", ErrorCodes::BAD_ARGUMENTS);
 
-    ExternalTableData data = getData(context);
+    ExternalTableDataPtr data = getData(context);
 
     /// Create table
     NamesAndTypesList columns = sample_block.getNamesAndTypesList();
-    StoragePtr storage = StorageMemory::create(StorageID("_external", data.second), ColumnsDescription{columns}, ConstraintsDescription{});
+    StoragePtr storage = StorageMemory::create(StorageID("_external", data->table_name), ColumnsDescription{columns}, ConstraintsDescription{});
     storage->startup();
-    context.addExternalTable(data.second, storage);
+    context.addExternalTable(data->table_name, storage);
     BlockOutputStreamPtr output = storage->write(ASTPtr(), context);
 
     /// Write data
-    data.first->readPrefix();
-    output->writePrefix();
-    while (Block block = data.first->read())
-        output->write(block);
-    data.first->readSuffix();
-    output->writeSuffix();
+    auto sink = std::make_shared<SinkToOutputStream>(std::move(output));
+    connect(data->pipe->getPort(), sink->getPort());
+
+    auto processors = std::move(*data->pipe).detachProcessors();
+    processors.push_back(std::move(sink));
+
+    auto executor = std::make_shared<PipelineExecutor>(processors);
+    executor->execute(/*num_threads = */ 1);
 
     /// We are ready to receive the next file, for this we clear all the information received
     clean();

diff --git a/dbms/src/Core/ExternalTable.h b/dbms/src/Core/ExternalTable.h
@@ -57,7 +57,7 @@ class BaseExternalTable
     virtual void initReadBuffer() {}
 
     /// Get the table data - a pair (a stream with the contents of the table, the name of the table)
-    ExternalTableData getData(const Context & context);
+    ExternalTableDataPtr getData(const Context & context);
 
 protected:
     /// Clear all accumulated information

diff --git a/dbms/src/DataStreams/BlockStreamProfileInfo.cpp b/dbms/src/DataStreams/BlockStreamProfileInfo.cpp
@@ -122,6 +122,7 @@ void BlockStreamProfileInfo::calculateRowsBeforeLimit() const
         /// Then the data about `rows_before_limit` can be in `RemoteBlockInputStream` (come from a remote server).
         BlockStreamProfileInfos remotes;
         collectInfosForStreamsWithName("Remote", remotes);
+        collectInfosForStreamsWithName("TreeExecutor", remotes);
 
         if (remotes.empty())
             return;

diff --git a/dbms/src/DataStreams/BlocksBlockInputStream.h b/dbms/src/DataStreams/BlocksBlockInputStream.h
@@ -19,33 +19,31 @@ namespace DB
 
 /** A stream of blocks from a shared vector of blocks
   */
-class BlocksBlockInputStream : public IBlockInputStream
+class BlocksSource : public SourceWithProgress
 {
 public:
     /// Acquires shared ownership of the blocks vector
-    BlocksBlockInputStream(const std::shared_ptr<BlocksPtr> & blocks_ptr_, Block header_)
-        : blocks(*blocks_ptr_), it((*blocks_ptr_)->begin()), end((*blocks_ptr_)->end()), header(std::move(header_)) {}
+    BlocksSource(const std::shared_ptr<BlocksPtr> & blocks_ptr_, Block header)
+        : SourceWithProgress(std::move(header))
+        , blocks(*blocks_ptr_), it((*blocks_ptr_)->begin()), end((*blocks_ptr_)->end()) {}
 
     String getName() const override { return "Blocks"; }
 
-    Block getHeader() const override { return header; }
-
 protected:
-    Block readImpl() override
+    Chunk generate() override
     {
         if (it == end)
-            return Block();
+            return {};
 
         Block res = *it;
         ++it;
-        return res;
+        return Chunk(res.getColumns(), res.rows());
     }
 
 private:
     BlocksPtr blocks;
     Blocks::iterator it;
     const Blocks::iterator end;
-    Block header;
 };
 
 }
diff --git a/dbms/src/DataStreams/RemoteBlockInputStream.cpp b/dbms/src/DataStreams/RemoteBlockInputStream.cpp
@@ -7,8 +7,12 @@
 #include <Interpreters/castColumn.h>
 #include <Interpreters/InternalTextLogsQueue.h>
 #include <Storages/IStorage.h>
+#include <Processors/Sources/SourceFromInputStream.h>
+#include <Processors/ConcatProcessor.h>
+#include <Processors/Pipe.h>
 
 #include <IO/ConnectionTimeouts.h>
+#include <Processors/Sources/SourceFromSingleChunk.h>
 
 
 namespace DB
@@ -112,7 +116,7 @@ void RemoteBlockInputStream::cancel(bool kill)
         /// Stop sending external data.
         for (auto & vec : external_tables_data)
             for (auto & elem : vec)
-                elem.first->cancel(kill);
+                elem->is_cancelled = true;
     }
 
     if (!isQueryPending() || hasThrownException())
@@ -142,12 +146,26 @@ void RemoteBlockInputStream::sendExternalTables()
             {
                 StoragePtr cur = table.second;
                 QueryProcessingStage::Enum read_from_table_stage = cur->getQueryProcessingStage(context);
-                BlockInputStreams input = cur->read(cur->getColumns().getNamesOfPhysical(), {}, context,
-                    read_from_table_stage, DEFAULT_BLOCK_SIZE, 1);
-                if (input.size() == 0)
-                    res.push_back(std::make_pair(std::make_shared<OneBlockInputStream>(cur->getSampleBlock()), table.first));
+
+                Pipes pipes;
+
+                pipes = cur->read(cur->getColumns().getNamesOfPhysical(), {}, context,
+                        read_from_table_stage, DEFAULT_BLOCK_SIZE, 1);
+
+                auto data = std::make_unique<ExternalTableData>();
+                data->table_name = table.first;
+
+                if (pipes.empty())
+                    data->pipe = std::make_unique<Pipe>(std::make_shared<SourceFromSingleChunk>(cur->getSampleBlock(), Chunk()));
+                else if (pipes.size() == 1)
+                    data->pipe = std::make_unique<Pipe>(std::move(pipes.front()));
                 else
-                    res.push_back(std::make_pair(input[0], table.first));
+                {
+                    auto concat = std::make_shared<ConcatProcessor>(pipes.front().getHeader(), pipes.size());
+                    data->pipe = std::make_unique<Pipe>(std::move(pipes), std::move(concat));
+                }
+
+                res.emplace_back(std::move(data));
             }
             external_tables_data.push_back(std::move(res));
         }

diff --git a/dbms/src/DataStreams/narrowBlockInputStreams.cpp b/dbms/src/DataStreams/narrowBlockInputStreams.cpp
@@ -1,12 +1,29 @@
 #include <random>
 #include <Common/thread_local_rng.h>
 #include <DataStreams/ConcatBlockInputStream.h>
+#include <Processors/ConcatProcessor.h>
+#include <Processors/Pipe.h>
 #include "narrowBlockInputStreams.h"
 
 
 namespace DB
 {
 
+namespace
+{
+    using Distribution = std::vector<size_t>;
+    Distribution getDistribution(size_t from, size_t to)
+    {
+        Distribution distribution(from);
+
+        for (size_t i = 0; i < from; ++i)
+            distribution[i] = i % to;
+
+        std::shuffle(distribution.begin(), distribution.end(), thread_local_rng);
+        return distribution;
+    }
+}
+
 BlockInputStreams narrowBlockInputStreams(BlockInputStreams & inputs, size_t width)
 {
     size_t size = inputs.size();
@@ -15,13 +32,7 @@ BlockInputStreams narrowBlockInputStreams(BlockInputStreams & inputs, size_t wid
 
     std::vector<BlockInputStreams> partitions(width);
 
-    using Distribution = std::vector<size_t>;
-    Distribution distribution(size);
-
-    for (size_t i = 0; i < size; ++i)
-        distribution[i] = i % width;
-
-    std::shuffle(distribution.begin(), distribution.end(), thread_local_rng);
+    auto distribution = getDistribution(size, width);
 
     for (size_t i = 0; i < size; ++i)
         partitions[distribution[i]].push_back(inputs[i]);
@@ -33,4 +44,29 @@ BlockInputStreams narrowBlockInputStreams(BlockInputStreams & inputs, size_t wid
     return res;
 }
 
+Pipes narrowPipes(Pipes pipes, size_t width)
+{
+    size_t size = pipes.size();
+    if (size <= width)
+        return pipes;
+
+    std::vector<Pipes> partitions(width);
+
+    auto distribution = getDistribution(size, width);
+
+    for (size_t i = 0; i < size; ++i)
+        partitions[distribution[i]].emplace_back(std::move(pipes[i]));
+
+    Pipes res;
+    res.reserve(width);
+
+    for (size_t i = 0; i < width; ++i)
+    {
+        auto processor = std::make_shared<ConcatProcessor>(partitions[i].at(0).getHeader(), partitions[i].size());
+        res.emplace_back(std::move(partitions[i]), std::move(processor));
+    }
+
+    return res;
+}
+
 }
diff --git a/dbms/src/DataStreams/narrowBlockInputStreams.h b/dbms/src/DataStreams/narrowBlockInputStreams.h
@@ -6,6 +6,9 @@
 namespace DB
 {
 
+class Pipe;
+using Pipes = std::vector<Pipe>;
+
 /** If the number of sources of `inputs` is greater than `width`,
   *  then glues the sources to each other (using ConcatBlockInputStream),
   *  so that the number of sources becomes no more than `width`.
@@ -14,5 +17,6 @@ namespace DB
   *  (to avoid overweighting if the distribution of the amount of data in different sources is subject to some pattern)
   */
 BlockInputStreams narrowBlockInputStreams(BlockInputStreams & inputs, size_t width);
+Pipes narrowPipes(Pipes pipes, size_t width);
 
 }