reduce memory usage in queries with 'ORDER BY primary_key'

src/DataStreams/tests/gtest_blocks_size_merging_streams.cpp

@@ -84,7 +84,7 @@ TEST(MergingSortedTest, SimpleBlockSizeTest)
     EXPECT_EQ(pipe.numOutputPorts(), 3);
 
     auto transform = std::make_shared<MergingSortedTransform>(pipe.getHeader(), pipe.numOutputPorts(), sort_description,
-            DEFAULT_MERGE_BLOCK_SIZE, 0, nullptr, false, true);
+            DEFAULT_MERGE_BLOCK_SIZE, 0, false, nullptr, false, true);
 
     pipe.addTransform(std::move(transform));
 
@@ -129,7 +129,7 @@ TEST(MergingSortedTest, MoreInterestingBlockSizes)
     EXPECT_EQ(pipe.numOutputPorts(), 3);
 
     auto transform = std::make_shared<MergingSortedTransform>(pipe.getHeader(), pipe.numOutputPorts(), sort_description,
-            DEFAULT_MERGE_BLOCK_SIZE, 0, nullptr, false, true);
+            DEFAULT_MERGE_BLOCK_SIZE, 0, false, nullptr, false, true);
 
     pipe.addTransform(std::move(transform));
 

src/Processors/Merges/AggregatingSortedTransform.h

@@ -16,7 +16,7 @@ public:
         const Block & header, size_t num_inputs,
         SortDescription description_, size_t max_block_size)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/Merges/CollapsingSortedTransform.h

@@ -20,7 +20,7 @@ public:
         WriteBuffer * out_row_sources_buf_ = nullptr,
         bool use_average_block_sizes = false)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/Merges/FinishAggregatingInOrderTransform.h

@@ -19,7 +19,7 @@ public:
         SortDescription description,
         size_t max_block_size)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             params,

src/Processors/Merges/GraphiteRollupSortedTransform.h

@@ -15,7 +15,7 @@ public:
         SortDescription description_, size_t max_block_size,
         Graphite::Params params_, time_t time_of_merge_)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/Merges/IMergingTransform.cpp

@@ -14,9 +14,11 @@ IMergingTransformBase::IMergingTransformBase(
     size_t num_inputs,
     const Block & input_header,
     const Block & output_header,
-    bool have_all_inputs_)
+    bool have_all_inputs_,
+    bool expected_one_block_)
     : IProcessor(InputPorts(num_inputs, input_header), {output_header})
     , have_all_inputs(have_all_inputs_)
+    , expected_one_block(expected_one_block_)
 {
 }
 
@@ -64,10 +66,7 @@ IProcessor::Status IMergingTransformBase::prepareInitializeInputs()
             continue;
 
         if (input_states[i].is_initialized)
-        {
-            // input.setNotNeeded();
             continue;
-        }
 
         input.setNeeded();
 
@@ -77,12 +76,17 @@ IProcessor::Status IMergingTransformBase::prepareInitializeInputs()
             continue;
         }
 
-        auto chunk = input.pull();
+        /// setNotNeeded after reading first chunk, because in optimismtic case
+        /// (e.g. with optimized 'ORDER BY primary_key LIMIT n' and small 'n')
+        /// we won't have to read any chunks anymore;
+        auto chunk = input.pull(expected_one_block);
         if (!chunk.hasRows())
         {
-
             if (!input.isFinished())
+            {
+                input.setNeeded();
                 all_inputs_has_data = false;
+            }
 
             continue;
         }

src/Processors/Merges/IMergingTransform.h

@@ -16,7 +16,8 @@ public:
         size_t num_inputs,
         const Block & input_header,
         const Block & output_header,
-        bool have_all_inputs_);
+        bool have_all_inputs_,
+        bool expected_one_block_);
 
     OutputPort & getOutputPort() { return outputs.front(); }
 
@@ -66,6 +67,7 @@ private:
     std::vector<InputState> input_states;
     std::atomic<bool> have_all_inputs;
     bool is_initialized = false;
+    bool expected_one_block = false;
 
     IProcessor::Status prepareInitializeInputs();
 };
@@ -81,8 +83,9 @@ public:
         const Block & input_header,
         const Block & output_header,
         bool have_all_inputs_,
+        bool expected_one_block_,
         Args && ... args)
-        : IMergingTransformBase(num_inputs, input_header, output_header, have_all_inputs_)
+        : IMergingTransformBase(num_inputs, input_header, output_header, have_all_inputs_, expected_one_block_)
         , algorithm(std::forward<Args>(args) ...)
     {
     }

src/Processors/Merges/MergingSortedTransform.cpp

@@ -13,12 +13,13 @@ MergingSortedTransform::MergingSortedTransform(
     SortDescription  description_,
     size_t max_block_size,
     UInt64 limit_,
+    bool expected_one_block_,
     WriteBuffer * out_row_sources_buf_,
     bool quiet_,
     bool use_average_block_sizes,
     bool have_all_inputs_)
     : IMergingTransform(
-        num_inputs, header, header, have_all_inputs_,
+        num_inputs, header, header, have_all_inputs_, expected_one_block_,
         header,
         num_inputs,
         std::move(description_),

src/Processors/Merges/MergingSortedTransform.h

@@ -17,6 +17,7 @@ public:
         SortDescription description,
         size_t max_block_size,
         UInt64 limit_ = 0,
+        bool expected_one_block_ = false,
         WriteBuffer * out_row_sources_buf_ = nullptr,
         bool quiet_ = false,
         bool use_average_block_sizes = false,

src/Processors/Merges/ReplacingSortedTransform.h

@@ -18,7 +18,7 @@ public:
         WriteBuffer * out_row_sources_buf_ = nullptr,
         bool use_average_block_sizes = false)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/Merges/SummingSortedTransform.h

@@ -19,7 +19,7 @@ public:
         const Names & partition_key_columns,
         size_t max_block_size)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/Merges/VersionedCollapsingTransform.h

@@ -19,7 +19,7 @@ public:
         WriteBuffer * out_row_sources_buf_ = nullptr,
         bool use_average_block_sizes = false)
         : IMergingTransform(
-            num_inputs, header, header, true,
+            num_inputs, header, header, true, false,
             header,
             num_inputs,
             std::move(description_),

src/Processors/QueryPlan/FinishSortingStep.cpp

@@ -58,11 +58,14 @@ void FinishSortingStep::transformPipeline(QueryPipeline & pipeline, const BuildQ
     if (pipeline.getNumStreams() > 1)
     {
         UInt64 limit_for_merging = (need_finish_sorting ? 0 : limit);
+        bool expected_one_block = limit_for_merging && limit_for_merging < max_block_size;
         auto transform = std::make_shared<MergingSortedTransform>(
                 pipeline.getHeader(),
                 pipeline.getNumStreams(),
                 prefix_description,
-                max_block_size, limit_for_merging);
+                max_block_size,
+                limit_for_merging,
+                expected_one_block);
 
         pipeline.addTransform(std::move(transform));
     }

src/Processors/QueryPlan/ReadFromMergeTree.cpp

@@ -13,7 +13,7 @@
 #include <Processors/Merges/ReplacingSortedTransform.h>
 #include <Processors/Merges/SummingSortedTransform.h>
 #include <Processors/Merges/VersionedCollapsingTransform.h>
-#include <Storages/MergeTree/MergeTreeSelectProcessor.h>
+#include <Storages/MergeTree/MergeTreeInOrderSelectProcessor.h>
 #include <Storages/MergeTree/MergeTreeReverseSelectProcessor.h>
 #include <Storages/MergeTree/MergeTreeThreadSelectBlockInputProcessor.h>
 #include <Storages/MergeTree/MergeTreeDataSelectExecutor.h>
@@ -175,12 +175,13 @@ template<typename TSource>
 ProcessorPtr ReadFromMergeTree::createSource(
     const RangesInDataPart & part,
     const Names & required_columns,
-    bool use_uncompressed_cache)
+    bool use_uncompressed_cache,
+    bool one_range_per_task)
 {
     return std::make_shared<TSource>(
             data, metadata_snapshot, part.data_part, max_block_size, preferred_block_size_bytes,
             preferred_max_column_in_block_size_bytes, required_columns, part.ranges, use_uncompressed_cache,
-            prewhere_info, true, reader_settings, virt_column_names, part.part_index_in_query);
+            prewhere_info, true, reader_settings, virt_column_names, part.part_index_in_query, one_range_per_task);
 }
 
 Pipe ReadFromMergeTree::readInOrder(
@@ -190,11 +191,15 @@ Pipe ReadFromMergeTree::readInOrder(
     bool use_uncompressed_cache)
 {
     Pipes pipes;
+    /// For reading in order it makes sense to read only
+    /// one range per task to reduce number of read rows.
+    bool one_range_per_task = read_type != ReadType::Default;
+
     for (const auto & part : parts_with_range)
     {
         auto source = read_type == ReadType::InReverseOrder
-                    ? createSource<MergeTreeReverseSelectProcessor>(part, required_columns, use_uncompressed_cache)
-                    : createSource<MergeTreeSelectProcessor>(part, required_columns, use_uncompressed_cache);
+                    ? createSource<MergeTreeReverseSelectProcessor>(part, required_columns, use_uncompressed_cache, one_range_per_task)
+                    : createSource<MergeTreeInOrderSelectProcessor>(part, required_columns, use_uncompressed_cache, one_range_per_task);
 
         pipes.emplace_back(std::move(source));
     }
@@ -445,6 +450,7 @@ Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsWithOrder(
                 }
                 parts_with_ranges.emplace_back(part);
             }
+
             ranges_to_get_from_part = split_ranges(ranges_to_get_from_part, input_order_info->direction);
             new_parts.emplace_back(part.data_part, part.part_index_in_query, std::move(ranges_to_get_from_part));
         }
@@ -482,7 +488,8 @@ Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsWithOrder(
                         pipe.getHeader(),
                         pipe.numOutputPorts(),
                         sort_description,
-                        max_block_size);
+                        max_block_size,
+                        0, true);
 
                 pipe.addTransform(std::move(transform));
             }

src/Processors/QueryPlan/ReadFromMergeTree.h

@@ -111,7 +111,7 @@ private:
     Pipe readInOrder(RangesInDataParts parts_with_range, Names required_columns, ReadType read_type, bool use_uncompressed_cache);
 
     template<typename TSource>
-    ProcessorPtr createSource(const RangesInDataPart & part, const Names & required_columns, bool use_uncompressed_cache);
+    ProcessorPtr createSource(const RangesInDataPart & part, const Names & required_columns, bool use_uncompressed_cache, bool one_range_per_task);
 
     Pipe spreadMarkRangesAmongStreams(
         RangesInDataParts && parts_with_ranges,

src/Processors/Transforms/MergeSortingTransform.cpp

@@ -200,6 +200,7 @@ void MergeSortingTransform::consume(Chunk chunk)
                     description,
                     max_merged_block_size,
                     limit,
+                    false,
                     nullptr,
                     quiet,
                     use_average_block_sizes,

src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp

@@ -891,7 +891,7 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
     {
         case MergeTreeData::MergingParams::Ordinary:
             merged_transform = std::make_unique<MergingSortedTransform>(
-                header, pipes.size(), sort_description, merge_block_size, 0, rows_sources_write_buf.get(), true, blocks_are_granules_size);
+                header, pipes.size(), sort_description, merge_block_size, 0, false, rows_sources_write_buf.get(), true, blocks_are_granules_size);
             break;
 
         case MergeTreeData::MergingParams::Collapsing:

src/Storages/MergeTree/MergeTreeInOrderSelectProcessor.cpp

@@ -0,0 +1,80 @@
+#include <Storages/MergeTree/MergeTreeInOrderSelectProcessor.h>
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int MEMORY_LIMIT_EXCEEDED;
+}
+
+MergeTreeInOrderSelectProcessor::MergeTreeInOrderSelectProcessor(
+    const MergeTreeData & storage_,
+    const StorageMetadataPtr & metadata_snapshot_,
+    const MergeTreeData::DataPartPtr & owned_data_part_,
+    UInt64 max_block_size_rows_,
+    size_t preferred_block_size_bytes_,
+    size_t preferred_max_column_in_block_size_bytes_,
+    Names required_columns_,
+    MarkRanges mark_ranges_,
+    bool use_uncompressed_cache_,
+    const PrewhereInfoPtr & prewhere_info_,
+    bool check_columns_,
+    const MergeTreeReaderSettings & reader_settings_,
+    const Names & virt_column_names_,
+    bool one_range_per_task_,
+    bool quiet)
+    : MergeTreeSelectProcessor{
+        storage_, metadata_snapshot_, owned_data_part_, max_block_size_rows_,
+        preferred_block_size_bytes_, preferred_max_column_in_block_size_bytes_,
+        required_columns_, std::move(mark_ranges_), use_uncompressed_cache_, prewhere_info_,
+        check_columns_, reader_settings_, virt_column_names_, one_range_per_task_}
+{
+    if (!quiet)
+        LOG_DEBUG(log, "Reading {} ranges in order from part {}, approx. {} rows starting from {}",
+            all_mark_ranges.size(), data_part->name, total_rows,
+            data_part->index_granularity.getMarkStartingRow(all_mark_ranges.front().begin));
+}
+
+bool MergeTreeInOrderSelectProcessor::getNewTask()
+try
+{
+    /// Produce no more than one task
+    if (all_mark_ranges.empty())
+    {
+        finish();
+        return false;
+    }
+
+    auto size_predictor = (preferred_block_size_bytes == 0)
+        ? nullptr
+        : std::make_unique<MergeTreeBlockSizePredictor>(data_part, ordered_names, metadata_snapshot->getSampleBlock());
+
+    MarkRanges mark_ranges_for_task;
+    if (one_range_per_task)
+    {
+        mark_ranges_for_task = { std::move(all_mark_ranges.front()) };
+        all_mark_ranges.pop_front();
+    }
+    else
+    {
+        mark_ranges_for_task = std::move(all_mark_ranges);
+        all_mark_ranges.clear();
+    }
+
+    task = std::make_unique<MergeTreeReadTask>(
+        data_part, mark_ranges_for_task, part_index_in_query, ordered_names, column_name_set, task_columns.columns,
+        task_columns.pre_columns, prewhere_info && prewhere_info->remove_prewhere_column,
+        task_columns.should_reorder, std::move(size_predictor));
+
+    return true;
+}
+catch (...)
+{
+    /// Suspicion of the broken part. A part is added to the queue for verification.
+    if (getCurrentExceptionCode() != ErrorCodes::MEMORY_LIMIT_EXCEEDED)
+        storage.reportBrokenPart(data_part->name);
+    throw;
+}
+
+}

src/Storages/MergeTree/MergeTreeInOrderSelectProcessor.h

@@ -0,0 +1,39 @@
+#pragma once
+#include <Storages/MergeTree/MergeTreeSelectProcessor.h>
+
+namespace DB
+{
+
+
+/// Used to read data from single part with select query in order of primary key.
+/// Cares about PREWHERE, virtual columns, indexes etc.
+/// To read data from multiple parts, Storage (MergeTree) creates multiple such objects.
+class MergeTreeInOrderSelectProcessor : public MergeTreeSelectProcessor
+{
+public:
+    MergeTreeInOrderSelectProcessor(
+        const MergeTreeData & storage,
+        const StorageMetadataPtr & metadata_snapshot,
+        const MergeTreeData::DataPartPtr & owned_data_part,
+        UInt64 max_block_size_rows,
+        size_t preferred_block_size_bytes,
+        size_t preferred_max_column_in_block_size_bytes,
+        Names required_columns_,
+        MarkRanges mark_ranges,
+        bool use_uncompressed_cache,
+        const PrewhereInfoPtr & prewhere_info,
+        bool check_columns,
+        const MergeTreeReaderSettings & reader_settings,
+        const Names & virt_column_names = {},
+        bool one_range_per_task_ = false,
+        bool quiet = false);
+
+        String getName() const override { return "MergeTreeInOrder"; }
+
+private:
+    bool getNewTask() override;
+
+    Poco::Logger * log = &Poco::Logger::get("MergeTreeInOrderSelectProcessor");
+};
+
+}

src/Storages/MergeTree/MergeTreeReverseSelectProcessor.cpp

@@ -1,8 +1,4 @@
 #include <Storages/MergeTree/MergeTreeReverseSelectProcessor.h>
-#include <Storages/MergeTree/MergeTreeBaseSelectProcessor.h>
-#include <Storages/MergeTree/IMergeTreeReader.h>
-#include <Interpreters/Context.h>
-
 
 namespace DB
 {
@@ -23,62 +19,27 @@ MergeTreeReverseSelectProcessor::MergeTreeReverseSelectProcessor(
     MarkRanges mark_ranges_,
     bool use_uncompressed_cache_,
     const PrewhereInfoPtr & prewhere_info_,
-    bool check_columns,
+    bool check_columns_,
     const MergeTreeReaderSettings & reader_settings_,
     const Names & virt_column_names_,
-    size_t part_index_in_query_,
+    bool one_range_per_task_,
     bool quiet)
-    :
-    MergeTreeBaseSelectProcessor{
-        metadata_snapshot_->getSampleBlockForColumns(required_columns_, storage_.getVirtuals(), storage_.getStorageID()),
-        storage_, metadata_snapshot_, prewhere_info_, max_block_size_rows_,
+    : MergeTreeSelectProcessor{
+        storage_, metadata_snapshot_, owned_data_part_, max_block_size_rows_,
         preferred_block_size_bytes_, preferred_max_column_in_block_size_bytes_,
-        reader_settings_, use_uncompressed_cache_, virt_column_names_},
-    required_columns{std::move(required_columns_)},
-    data_part{owned_data_part_},
-    all_mark_ranges(std::move(mark_ranges_)),
-    part_index_in_query(part_index_in_query_),
-    path(data_part->getFullRelativePath())
+        required_columns_, std::move(mark_ranges_), use_uncompressed_cache_, prewhere_info_,
+        check_columns_, reader_settings_, virt_column_names_, one_range_per_task_}
 {
-    /// Let's estimate total number of rows for progress bar.
-    for (const auto & range : all_mark_ranges)
-        total_marks_count += range.end - range.begin;
-
-    size_t total_rows = data_part->index_granularity.getRowsCountInRanges(all_mark_ranges);
-
     if (!quiet)
         LOG_DEBUG(log, "Reading {} ranges in reverse order from part {}, approx. {} rows starting from {}",
             all_mark_ranges.size(), data_part->name, total_rows,
             data_part->index_granularity.getMarkStartingRow(all_mark_ranges.front().begin));
-
-    addTotalRowsApprox(total_rows);
-
-    ordered_names = header_without_virtual_columns.getNames();
-
-    task_columns = getReadTaskColumns(storage, metadata_snapshot, data_part, required_columns, prewhere_info, check_columns);
-
-    /// will be used to distinguish between PREWHERE and WHERE columns when applying filter
-    const auto & column_names = task_columns.columns.getNames();
-    column_name_set = NameSet{column_names.begin(), column_names.end()};
-
-    if (use_uncompressed_cache)
-        owned_uncompressed_cache = storage.getContext()->getUncompressedCache();
-
-    owned_mark_cache = storage.getContext()->getMarkCache();
-
-    reader = data_part->getReader(task_columns.columns, metadata_snapshot,
-        all_mark_ranges, owned_uncompressed_cache.get(),
-        owned_mark_cache.get(), reader_settings);
-
-    if (prewhere_info)
-        pre_reader = data_part->getReader(task_columns.pre_columns, metadata_snapshot, all_mark_ranges,
-            owned_uncompressed_cache.get(), owned_mark_cache.get(), reader_settings);
 }
 
 bool MergeTreeReverseSelectProcessor::getNewTask()
 try
 {
-    if ((chunks.empty() && all_mark_ranges.empty()) || total_marks_count == 0)
+    if (chunks.empty() && all_mark_ranges.empty())
     {
         finish();
         return false;
@@ -141,17 +102,4 @@ Chunk MergeTreeReverseSelectProcessor::readFromPart()
     return res;
 }
 
-void MergeTreeReverseSelectProcessor::finish()
-{
-    /** Close the files (before destroying the object).
-    * When many sources are created, but simultaneously reading only a few of them,
-    * buffers don't waste memory.
-    */
-    reader.reset();
-    pre_reader.reset();
-    data_part.reset();
-}
-
-MergeTreeReverseSelectProcessor::~MergeTreeReverseSelectProcessor() = default;
-
 }

src/Storages/MergeTree/MergeTreeReverseSelectProcessor.h

@@ -1,19 +1,15 @@
 #pragma once
-#include <DataStreams/IBlockInputStream.h>
-#include <Storages/MergeTree/MergeTreeThreadSelectBlockInputProcessor.h>
-#include <Storages/MergeTree/MergeTreeData.h>
-#include <Storages/MergeTree/MarkRange.h>
-#include <Storages/MergeTree/MergeTreeBlockReadUtils.h>
-#include <Storages/SelectQueryInfo.h>
+#include <Storages/MergeTree/MergeTreeSelectProcessor.h>
 
 namespace DB
 {
 
 
 /// Used to read data from single part with select query
+/// in reverse order of primary key.
 /// Cares about PREWHERE, virtual columns, indexes etc.
 /// To read data from multiple parts, Storage (MergeTree) creates multiple such objects.
-class MergeTreeReverseSelectProcessor : public MergeTreeBaseSelectProcessor
+class MergeTreeReverseSelectProcessor : public MergeTreeSelectProcessor
 {
 public:
     MergeTreeReverseSelectProcessor(
@@ -30,46 +26,16 @@ public:
         bool check_columns,
         const MergeTreeReaderSettings & reader_settings,
         const Names & virt_column_names = {},
-        size_t part_index_in_query = 0,
+        bool one_range_per_task_ = false,
         bool quiet = false);
 
-    ~MergeTreeReverseSelectProcessor() override;
-
     String getName() const override { return "MergeTreeReverse"; }
 
-    /// Closes readers and unlock part locks
-    void finish();
-
-protected:
-
+private:
     bool getNewTask() override;
     Chunk readFromPart() override;
 
-private:
-    Block header;
-
-    /// Used by Task
-    Names required_columns;
-    /// Names from header. Used in order to order columns in read blocks.
-    Names ordered_names;
-    NameSet column_name_set;
-
-    MergeTreeReadTaskColumns task_columns;
-
-    /// Data part will not be removed if the pointer owns it
-    MergeTreeData::DataPartPtr data_part;
-
-    /// Mark ranges we should read (in ascending order)
-    MarkRanges all_mark_ranges;
-    /// Total number of marks we should read
-    size_t total_marks_count = 0;
-    /// Value of _part_index virtual column (used only in SelectExecutor)
-    size_t part_index_in_query = 0;
-
-    String path;
-
     Chunks chunks;
-
     Poco::Logger * log = &Poco::Logger::get("MergeTreeReverseSelectProcessor");
 };
 

src/Storages/MergeTree/MergeTreeSelectProcessor.cpp

@@ -26,10 +26,8 @@ MergeTreeSelectProcessor::MergeTreeSelectProcessor(
     bool check_columns_,
     const MergeTreeReaderSettings & reader_settings_,
     const Names & virt_column_names_,
-    size_t part_index_in_query_,
-    bool quiet)
-    :
-    MergeTreeBaseSelectProcessor{
+    bool one_range_per_task_)
+    : MergeTreeBaseSelectProcessor{
         metadata_snapshot_->getSampleBlockForColumns(required_columns_, storage_.getVirtuals(), storage_.getStorageID()),
         storage_, metadata_snapshot_, prewhere_info_, max_block_size_rows_,
         preferred_block_size_bytes_, preferred_max_column_in_block_size_bytes_,
@@ -37,78 +35,33 @@ MergeTreeSelectProcessor::MergeTreeSelectProcessor(
     required_columns{std::move(required_columns_)},
     data_part{owned_data_part_},
     all_mark_ranges(std::move(mark_ranges_)),
-    part_index_in_query(part_index_in_query_),
-    check_columns(check_columns_)
+    one_range_per_task(one_range_per_task_),
+    check_columns(check_columns_),
+    total_rows(data_part->index_granularity.getRowsCountInRanges(all_mark_ranges))
 {
-    /// Let's estimate total number of rows for progress bar.
-    for (const auto & range : all_mark_ranges)
-        total_marks_count += range.end - range.begin;
-
-    size_t total_rows = data_part->index_granularity.getRowsCountInRanges(all_mark_ranges);
-
-    if (!quiet)
-        LOG_DEBUG(log, "Reading {} ranges from part {}, approx. {} rows starting from {}",
-            all_mark_ranges.size(), data_part->name, total_rows,
-            data_part->index_granularity.getMarkStartingRow(all_mark_ranges.front().begin));
-
-    addTotalRowsApprox(total_rows);
-    ordered_names = header_without_virtual_columns.getNames();
-}
-
-
-bool MergeTreeSelectProcessor::getNewTask()
-try
-{
-    /// Produce no more than one task
-    if (!is_first_task || total_marks_count == 0)
-    {
-        finish();
-        return false;
-    }
-    is_first_task = false;
+    /// will be used to distinguish between PREWHERE and WHERE columns when applying filter
+    const auto & column_names = task_columns.columns.getNames();
+    column_name_set = NameSet{column_names.begin(), column_names.end()};
 
     task_columns = getReadTaskColumns(
         storage, metadata_snapshot, data_part,
         required_columns, prewhere_info, check_columns);
 
-    auto size_predictor = (preferred_block_size_bytes == 0)
-        ? nullptr
-        : std::make_unique<MergeTreeBlockSizePredictor>(data_part, ordered_names, metadata_snapshot->getSampleBlock());
+    if (use_uncompressed_cache)
+        owned_uncompressed_cache = storage.getContext()->getUncompressedCache();
 
-    /// will be used to distinguish between PREWHERE and WHERE columns when applying filter
-    const auto & column_names = task_columns.columns.getNames();
-    column_name_set = NameSet{column_names.begin(), column_names.end()};
+    owned_mark_cache = storage.getContext()->getMarkCache();
 
-    task = std::make_unique<MergeTreeReadTask>(
-        data_part, all_mark_ranges, part_index_in_query, ordered_names, column_name_set, task_columns.columns,
-        task_columns.pre_columns, prewhere_info && prewhere_info->remove_prewhere_column,
-        task_columns.should_reorder, std::move(size_predictor));
+    reader = data_part->getReader(task_columns.columns, metadata_snapshot, all_mark_ranges,
+        owned_uncompressed_cache.get(), owned_mark_cache.get(), reader_settings);
 
-    if (!reader)
-    {
-        if (use_uncompressed_cache)
-            owned_uncompressed_cache = storage.getContext()->getUncompressedCache();
-
-        owned_mark_cache = storage.getContext()->getMarkCache();
-
-        reader = data_part->getReader(task_columns.columns, metadata_snapshot, all_mark_ranges,
+    if (prewhere_info)
+        pre_reader = data_part->getReader(task_columns.pre_columns, metadata_snapshot, all_mark_ranges,
             owned_uncompressed_cache.get(), owned_mark_cache.get(), reader_settings);
 
-        if (prewhere_info)
-            pre_reader = data_part->getReader(task_columns.pre_columns, metadata_snapshot, all_mark_ranges,
-                owned_uncompressed_cache.get(), owned_mark_cache.get(), reader_settings);
-    }
-
-    return true;
+    addTotalRowsApprox(total_rows);
+    ordered_names = header_without_virtual_columns.getNames();
 }
-catch (...)
-{
-    /// Suspicion of the broken part. A part is added to the queue for verification.
-    if (getCurrentExceptionCode() != ErrorCodes::MEMORY_LIMIT_EXCEEDED)
-        storage.reportBrokenPart(data_part->name);
-    throw;
-}
-
 
 void MergeTreeSelectProcessor::finish()
 {
@@ -121,8 +74,6 @@ void MergeTreeSelectProcessor::finish()
     data_part.reset();
 }
 
-
 MergeTreeSelectProcessor::~MergeTreeSelectProcessor() = default;
 
-
 }

src/Storages/MergeTree/MergeTreeSelectProcessor.h

@@ -1,6 +1,6 @@
 #pragma once
 #include <DataStreams/IBlockInputStream.h>
-#include <Storages/MergeTree/MergeTreeThreadSelectBlockInputProcessor.h>
+#include <Storages/MergeTree/MergeTreeBaseSelectProcessor.h>
 #include <Storages/MergeTree/MergeTreeData.h>
 #include <Storages/MergeTree/MarkRange.h>
 #include <Storages/MergeTree/MergeTreeBlockReadUtils.h>
@@ -30,21 +30,18 @@ public:
         bool check_columns,
         const MergeTreeReaderSettings & reader_settings,
         const Names & virt_column_names = {},
-        size_t part_index_in_query = 0,
-        bool quiet = false);
+        bool one_range_per_task_ = false);
 
     ~MergeTreeSelectProcessor() override;
 
-    String getName() const override { return "MergeTree"; }
+    String getName() const override = 0;
 
     /// Closes readers and unlock part locks
     void finish();
 
 protected:
 
-    bool getNewTask() override;
-
-private:
+    bool getNewTask() override = 0;
 
     /// Used by Task
     Names required_columns;
@@ -59,15 +56,14 @@ private:
 
     /// Mark ranges we should read (in ascending order)
     MarkRanges all_mark_ranges;
-    /// Total number of marks we should read
-    size_t total_marks_count = 0;
     /// Value of _part_index virtual column (used only in SelectExecutor)
     size_t part_index_in_query = 0;
+    /// If true, every task will be created only with one range.
+    /// It reduces amount of read data for queries with small LIMIT.
+    bool one_range_per_task = false;
 
     bool check_columns;
-    bool is_first_task = true;
-
-    Poco::Logger * log = &Poco::Logger::get("MergeTreeSelectProcessor");
+    size_t total_rows;
 };
 
 }

tests/queries/0_stateless/01926_order_by_desc_limit.reference

@@ -0,0 +1,2 @@
+1
+1

tests/queries/0_stateless/01926_order_by_desc_limit.sql

@@ -0,0 +1,21 @@
+DROP TABLE IF EXISTS order_by_desc;
+
+CREATE TABLE order_by_desc (u UInt32, s String)
+ENGINE MergeTree ORDER BY u PARTITION BY u % 100
+SETTINGS index_granularity = 1024;
+
+INSERT INTO order_by_desc SELECT number, repeat('a', 1024) FROM numbers(1024 * 300);
+OPTIMIZE TABLE order_by_desc FINAL;
+
+SELECT s FROM order_by_desc ORDER BY u DESC LIMIT 10 FORMAT Null
+SETTINGS max_memory_usage = '400M';
+
+SELECT s FROM order_by_desc ORDER BY u LIMIT 10 FORMAT Null
+SETTINGS max_memory_usage = '400M';
+
+SYSTEM FLUSH LOGS;
+
+SELECT read_rows < 110000 FROM system.query_log
+WHERE type = 'QueryFinish' AND current_database = currentDatabase()
+AND event_time > now() - INTERVAL 10 SECOND
+AND lower(query) LIKE lower('SELECT s FROM order_by_desc ORDER BY u%');