Add OffsetsStep.

src/Interpreters/InterpreterSelectQuery.cpp

@@ -94,6 +94,7 @@
 #include <Processors/QueryPlan/CubeStep.h>
 #include <Processors/QueryPlan/FillingStep.h>
 #include <Processors/QueryPlan/ExtremesStep.h>
+#include <Processors/QueryPlan/OffsetsStep.h>
 
 
 namespace DB
@@ -1852,12 +1853,8 @@ void InterpreterSelectQuery::executeOffset(QueryPipeline & pipeline)
         UInt64 limit_offset;
         std::tie(limit_length, limit_offset) = getLimitLengthAndOffset(query, *context);
 
-        pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
-        {
-            if (stream_type != QueryPipeline::StreamType::Main)
-                return nullptr;
-            return std::make_shared<OffsetTransform>(header, limit_offset, 1);
-        });
+        OffsetsStep offsets_step(DataStream{.header = pipeline.getHeader()}, limit_offset);
+        offsets_step.transformPipeline(pipeline);
     }
 }
 

src/Processors/QueryPlan/OffsetsStep.cpp

+#include <Processors/QueryPlan/OffsetsStep.h>
+#include <Processors/OffsetTransform.h>
+#include <Processors/QueryPipeline.h>
+
+namespace DB
+{
+
+OffsetsStep::OffsetsStep(const DataStream & input_stream_, size_t offsets_)
+    : ITransformingStep(input_stream_, input_stream_)
+    , offset(offsets_)
+{
+}
+
+void OffsetsStep::transformPipeline(QueryPipeline & pipeline)
+{
+    pipeline.addSimpleTransform([&](const Block & header, QueryPipeline::StreamType stream_type) -> ProcessorPtr
+    {
+        if (stream_type != QueryPipeline::StreamType::Main)
+            return nullptr;
+
+        return std::make_shared<OffsetTransform>(header, offset, 1);
+    });
+}
+
+}

src/Processors/QueryPlan/OffsetsStep.h

+#pragma once
+#include <Processors/QueryPlan/ITransformingStep.h>
+#include <DataStreams/SizeLimits.h>
+
+namespace DB
+{
+
+class OffsetsStep : public ITransformingStep
+{
+public:
+    OffsetsStep(const DataStream & input_stream_, size_t offset_);
+
+    String getName() const override { return "Offsets"; }
+
+    void transformPipeline(QueryPipeline & pipeline) override;
+
+private:
+    size_t offset;
+};
+
+}

src/Processors/ya.make

@@ -154,6 +154,7 @@ SRCS(
     QueryPlan/MergeSortingStep.cpp
     QueryPlan/MergingAggregatedStep.cpp
     QueryPlan/MergingSortedStep.cpp
+    QueryPlan/OffsetsStep.cpp
     QueryPlan/PartialSortingStep.cpp
     QueryPlan/ReadFromPreparedSource.cpp
     QueryPlan/ReadFromStorageStep.cpp