21.7.3 throws an exception if `_sample_factor` is used in a query:
```
CREATE TABLE sessions
(
  `user_id` UInt64
)
ENGINE = MergeTree
ORDER BY user_id 
SAMPLE BY user_id;

insert into sessions values(1);

SELECT
    sum(user_id * _sample_factor) 
FROM sessions
SAMPLE 10000000

DB::Exception: Cannot find column `user_id` in source stream (version 21.8.1.7409 (official build))

SELECT
    sum(user_id) 
FROM sessions
SAMPLE 10000000

sum(user_id)
------------
           1
```
No exception on 21.6.5
}
};
}
Pipe ReadFromMergeTree::spreadMarkRangesAmongStreams(
    RangesInDataParts && parts_with_ranges,
    const Names & column_names)
{
    const auto & settings = context->getSettingsRef();
    const auto data_settings = data.getSettings();
    PartRangesReadInfo info(parts_with_ranges, settings, *data_settings);
    if (0 == info.sum_marks)
        return {};
    size_t num_streams = requested_num_streams;
    if (num_streams > 1)
    {
        
        if (info.sum_marks < num_streams * info.min_marks_for_concurrent_read && parts_with_ranges.size() < num_streams)
            num_streams = std::max((info.sum_marks + info.min_marks_for_concurrent_read - 1) / info.min_marks_for_concurrent_read, parts_with_ranges.size());
    }
    return read(std::move(parts_with_ranges), column_names, ReadType::Default,
                num_streams, info.min_marks_for_concurrent_read, info.use_uncompressed_cache);
}
static ActionsDAGPtr createProjection(const Block & header)
{
    auto projection = std::make_shared<ActionsDAG>(header.getNamesAndTypesList());
    projection->removeUnusedActions(header.getNames());
    projection->projectInput();
    return projection;
}
Pipe ReadFromMergeTree::spreadMarkRangesAmongStreamsWithOrder(