Pipeline: Support pipeline table scan fullstack part 1

dbms/src/DataStreams/GeneratedColumnPlaceHolderTransformAction.cpp

@@ -0,0 +1,68 @@
+// Copyright 2023 PingCAP, Ltd.
+//
+// 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.
+
+#include <DataStreams/GeneratedColumnPlaceHolderTransformAction.h>
+
+namespace DB
+{
+GeneratedColumnPlaceHolderTransformAction::GeneratedColumnPlaceHolderTransformAction(
+    const Block & header_,
+    const std::vector<std::tuple<UInt64, String, DataTypePtr>> & generated_column_infos_)
+    : generated_column_infos(generated_column_infos_)
+{
+    header = header_;
+    insertColumns(header, false);
+}
+
+Block GeneratedColumnPlaceHolderTransformAction::getHeader() const
+{
+    return header;
+}
+
+void GeneratedColumnPlaceHolderTransformAction::checkColumn() const
+{
+    RUNTIME_CHECK(!generated_column_infos.empty());
+    // Validation check.
+    for (size_t i = 1; i < generated_column_infos.size(); ++i)
+    {
+        RUNTIME_CHECK(std::get<0>(generated_column_infos[i]) > std::get<0>(generated_column_infos[i - 1]));
+    }
+}
+
+void GeneratedColumnPlaceHolderTransformAction::insertColumns(Block & block, bool insert_data) const
+{
+    if (!block)
+        return;
+
+    for (const auto & ele : generated_column_infos)
+    {
+        const auto & col_index = std::get<0>(ele);
+        const auto & col_name = std::get<1>(ele);
+        const auto & data_type = std::get<2>(ele);
+        ColumnPtr column = nullptr;
+        if (insert_data)
+            column = data_type->createColumnConstWithDefaultValue(block.rows());
+        else
+            column = data_type->createColumn();
+        block.insert(col_index, ColumnWithTypeAndName{column, data_type, col_name});
+    }
+}
+
+bool GeneratedColumnPlaceHolderTransformAction::transform(Block & block)
+{
+    insertColumns(block, true);
+    return true;
+}
+
+} // namespace DB

dbms/src/DataStreams/GeneratedColumnPlaceHolderTransformAction.h

@@ -0,0 +1,40 @@
+// Copyright 2023 PingCAP, Ltd.
+//
+// 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.
+#pragma once
+
+#include <Core/Block.h>
+
+namespace DB
+{
+class GeneratedColumnPlaceHolderTransformAction
+{
+public:
+    GeneratedColumnPlaceHolderTransformAction(
+        const Block & header_,
+        const std::vector<std::tuple<UInt64, String, DataTypePtr>> & generated_column_infos_);
+
+    bool transform(Block & block);
+
+    Block getHeader() const;
+
+    void checkColumn() const;
+
+private:
+    void insertColumns(Block & block, bool insert_data) const;
+
+private:
+    Block header;
+    const std::vector<std::tuple<UInt64, String, DataTypePtr>> generated_column_infos;
+};
+} // namespace DB

dbms/src/DataStreams/GeneratedColumnPlaceholderBlockInputStream.h

@@ -18,6 +18,7 @@
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnsNumber.h>
 #include <Core/ColumnsWithTypeAndName.h>
+#include <DataStreams/GeneratedColumnPlaceHolderTransformAction.h>
 #include <DataStreams/IProfilingBlockInputStream.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypeString.h>
@@ -32,18 +33,17 @@ class GeneratedColumnPlaceholderBlockInputStream : public IProfilingBlockInputSt
         const BlockInputStreamPtr & input,
         const std::vector<std::tuple<UInt64, String, DataTypePtr>> & generated_column_infos_,
         const String & req_id_)
-        : generated_column_infos(generated_column_infos_)
+        : action(input->getHeader(), generated_column_infos_)
         , log(Logger::get(req_id_))
     {
         children.push_back(input);
     }
 
     String getName() const override { return NAME; }
+
     Block getHeader() const override
     {
-        Block block = children.back()->getHeader();
-        insertColumns(block, /*insert_data=*/false);
-        return block;
+        return action.getHeader();
     }
 
     static String getColumnName(UInt64 col_index)
@@ -54,43 +54,19 @@ class GeneratedColumnPlaceholderBlockInputStream : public IProfilingBlockInputSt
 protected:
     void readPrefix() override
     {
-        RUNTIME_CHECK(!generated_column_infos.empty());
-        // Validation check.
-        for (size_t i = 1; i < generated_column_infos.size(); ++i)
-        {
-            RUNTIME_CHECK(std::get<0>(generated_column_infos[i]) > std::get<0>(generated_column_infos[i - 1]));
-        }
+        action.checkColumn();
     }
 
     Block readImpl() override
     {
         Block block = children.back()->read();
-        insertColumns(block, /*insert_data=*/true);
+        action.transform(block);
         return block;
     }
 
 private:
-    void insertColumns(Block & block, bool insert_data) const
-    {
-        if (!block)
-            return;
-
-        for (const auto & ele : generated_column_infos)
-        {
-            const auto & col_index = std::get<0>(ele);
-            const auto & col_name = std::get<1>(ele);
-            const auto & data_type = std::get<2>(ele);
-            ColumnPtr column = nullptr;
-            if (insert_data)
-                column = data_type->createColumnConstWithDefaultValue(block.rows());
-            else
-                column = data_type->createColumn();
-            block.insert(col_index, ColumnWithTypeAndName{column, data_type, col_name});
-        }
-    }
-
     static constexpr auto NAME = "GeneratedColumnPlaceholder";
-    const std::vector<std::tuple<UInt64, String, DataTypePtr>> generated_column_infos;
+    GeneratedColumnPlaceHolderTransformAction action;
     const LoggerPtr log;
 };
 

dbms/src/Flash/Coprocessor/CoprocessorReader.h

@@ -87,7 +87,7 @@ class CoprocessorReader
         , has_enforce_encode_type(has_enforce_encode_type_)
         , resp_iter(std::move(tasks), cluster, concurrency, &Poco::Logger::get("pingcap/coprocessor"), tiflash_label_filter_)
         , collected(false)
-        , concurrency_(concurrency)
+        , concurrency(concurrency)
     {}
 
     const DAGSchema & getOutputSchema() const { return schema; }
@@ -115,7 +115,7 @@ class CoprocessorReader
             return detail;
 
         detail.packet_bytes = resp->ByteSizeLong();
-        for (int i = 0; i < chunk_size; i++)
+        for (int i = 0; i < chunk_size; ++i)
         {
             Block block;
             const tipb::Chunk & chunk = resp->chunks(i);
@@ -146,17 +146,27 @@ class CoprocessorReader
         return detail;
     }
 
-    // stream_id, decoder_ptr are only meaningful for ExchagneReceiver.
-    CoprocessorReaderResult nextResult(std::queue<Block> & block_queue, const Block & header, size_t /*stream_id*/, std::unique_ptr<CHBlockChunkDecodeAndSquash> & /*decoder_ptr*/)
+    std::pair<pingcap::coprocessor::ResponseIter::Result, bool> nonBlockingNext()
     {
-        RUNTIME_CHECK(opened == true);
+        return resp_iter.nonBlockingNext();
+    }
+
+    CoprocessorReaderResult toResult(std::pair<pingcap::coprocessor::ResponseIter::Result, bool> & result_pair,
+                                     std::queue<Block> & block_queue,
+                                     const Block & header)
+    {
+        auto && [result, has_next] = result_pair;
 
-        auto && [result, has_next] = resp_iter.next();
         if (!result.error.empty())
             return {nullptr, true, result.error.message(), false};
 
         if (!has_next)
-            return {nullptr, false, "", true};
+        {
+            if (result.finished)
+                return {nullptr, false, "", true};
+            else
+                return {nullptr, false, "", false};
+        }
 
         auto resp = std::make_shared<tipb::SelectResponse>();
         if (resp->ParseFromString(result.data()))
@@ -182,14 +192,24 @@ class CoprocessorReader
         }
     }
 
+    // stream_id, decoder_ptr are only meaningful for ExchagneReceiver.
+    CoprocessorReaderResult nextResult(std::queue<Block> & block_queue, const Block & header, size_t /*stream_id*/, std::unique_ptr<CHBlockChunkDecodeAndSquash> & /*decoder_ptr*/)
+    {
+        RUNTIME_CHECK(opened == true);
+
+        auto && result_pair = resp_iter.next();
+
+        return toResult(result_pair, block_queue, header);
+    }
+
     size_t getSourceNum() const { return 1; }
 
-    int getExternalThreadCnt() const { return concurrency_; }
+    int getExternalThreadCnt() const { return concurrency; }
 
     void close() {}
 
     bool collected = false;
-    int concurrency_;
+    int concurrency;
     bool opened = false;
 };
 } // namespace DB