Repository: man-group/ArcticDB (language: python)

Issue:
Add method to `IClause` to filter out unneeded `SliceAndKey`s and structure segments for processing
Currently `read_and_process` assumes:

- That the first clause in the pipeline will always need all of the segments in the pipeline
- That all clauses parallelise over row-slices

Neither of these are true. The structuring and contained elements of the `std::vector<Composite<ProcessingSegment>>` at the start of the pipeline should be delegated to the first clause in the pipeline to remove these assumptions.


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index d37daa2b12..618a139b2e 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -231,7 +231,7 @@ set(arcticdb_srcs
         processing/operation_dispatch_unary.hpp
         processing/operation_types.hpp
         processing/signed_unsigned_comparison.hpp
-        processing/processing_segment.hpp
+        processing/processing_unit.hpp
         processing/bucketizer.hpp
         processing/clause.hpp
         processing/expression_context.hpp
@@ -377,7 +377,7 @@ set(arcticdb_srcs
         pipeline/value_set.cpp
         pipeline/write_frame.cpp
         python/normalization_checks.cpp
-        processing/processing_segment.cpp
+        processing/processing_unit.cpp
         processing/aggregation.cpp
         processing/clause.cpp
         processing/expression_node.cpp
diff --git a/cpp/arcticdb/async/async_store.hpp b/cpp/arcticdb/async/async_store.hpp
index fa1b5cfb08..f8665585d9 100644
--- a/cpp/arcticdb/async/async_store.hpp
+++ b/cpp/arcticdb/async/async_store.hpp
@@ -345,15 +345,15 @@ class AsyncStore : public Store {
         return folly::collect(result).via(&async::io_executor());
     }
 
-    std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
+    std::vector<Composite<ProcessingUnit>> batch_read_uncompressed(
         std::vector<Composite<pipelines::SliceAndKey>> &&sks,
         const std::vector<std::shared_ptr<Clause>>& clauses,
         const std::shared_ptr<std::unordered_set<std::string>>& filter_columns,
         const BatchReadArgs & args) override {
         auto slice_and_keys = std::move(sks);
-        std::vector<Composite<ProcessingSegment>> res;
+        std::vector<Composite<ProcessingUnit>> res;
         res.reserve(slice_and_keys.size());
-        std::vector<folly::Future<Composite<ProcessingSegment>>> batch;
+        std::vector<folly::Future<Composite<ProcessingUnit>>> batch;
         size_t current_size = 0;
         for (auto &&s : slice_and_keys) {
             auto sk = std::move(s);
diff --git a/cpp/arcticdb/async/tasks.hpp b/cpp/arcticdb/async/tasks.hpp
index 4cf0f562b4..d4310e7edb 100644
--- a/cpp/arcticdb/async/tasks.hpp
+++ b/cpp/arcticdb/async/tasks.hpp
@@ -19,7 +19,7 @@
 #include <arcticdb/stream/stream_sink.hpp>
 #include <arcticdb/async/base_task.hpp>
 #include <arcticdb/pipeline/frame_slice.hpp>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/util/constructors.hpp>
 
 #include <type_traits>
@@ -335,24 +335,24 @@ struct SegmentFunctionTask : BaseTask {
 struct MemSegmentProcessingTask : BaseTask {
     std::shared_ptr<Store> store_;
     std::vector<std::shared_ptr<Clause>> clauses_;
-    std::optional<Composite<ProcessingSegment>> procs_;
+    std::optional<Composite<ProcessingUnit>> procs_;
 
     explicit MemSegmentProcessingTask(
            const std::shared_ptr<Store>& store,
            std::vector<std::shared_ptr<Clause>> clauses,
-           std::optional<Composite<ProcessingSegment>>&& procs = std::nullopt) :
+           std::optional<Composite<ProcessingUnit>>&& procs = std::nullopt) :
         store_(store),
         clauses_(std::move(clauses)),
         procs_(std::move(procs)) {
     }
 
-   static ProcessingSegment slice_to_segment(Composite<pipelines::SliceAndKey>&& is) {
+   static ProcessingUnit slice_to_segment(Composite<pipelines::SliceAndKey>&& is) {
         auto inputs = std::move(is);
-        return ProcessingSegment(inputs.as_range());
+        return ProcessingUnit(inputs.as_range());
     }
 
     [[nodiscard]]
-    Composite<ProcessingSegment> process(Composite<ProcessingSegment>&& proc){
+    Composite<ProcessingUnit> process(Composite<ProcessingUnit>&& proc){
         auto procs = std::move(proc);
         for(const auto& clause : clauses_) {
             procs = clause->process(store_, std::move(procs));
@@ -365,14 +365,14 @@ struct MemSegmentProcessingTask : BaseTask {
 
     ARCTICDB_MOVE_ONLY_DEFAULT(MemSegmentProcessingTask)
 
-    Composite<ProcessingSegment> operator()(Composite<pipelines::SliceAndKey>&& sk) {
-        return process(Composite<ProcessingSegment>(slice_to_segment(std::move(sk))));
+    Composite<ProcessingUnit> operator()(Composite<pipelines::SliceAndKey>&& sk) {
+        return process(Composite<ProcessingUnit>(slice_to_segment(std::move(sk))));
     }
 
-    Composite<ProcessingSegment> operator()() {
+    Composite<ProcessingUnit> operator()() {
         internal::check<ErrorCode::E_ASSERTION_FAILURE>(
                 procs_.has_value(),
-                "MemSegmentProcessingTask () operator expects processing segments to be provided in ctor");
+                "MemSegmentProcessingTask () operator expects processing units to be provided in ctor");
         return process(std::move(*procs_));
     }
 
diff --git a/cpp/arcticdb/pipeline/query.hpp b/cpp/arcticdb/pipeline/query.hpp
index bac80a46d9..f86d6a8a0c 100644
--- a/cpp/arcticdb/pipeline/query.hpp
+++ b/cpp/arcticdb/pipeline/query.hpp
@@ -33,14 +33,6 @@ namespace arcticdb::pipelines {
 
 using FilterRange = std::variant<std::monostate, IndexRange, RowRange>;
 
-struct HeadRange {
-    int64_t num_rows_;
-};
-
-struct TailRange {
-    int64_t num_rows_;
-};
-
 struct SignedRowRange {
     int64_t start_;
     int64_t end_;
@@ -48,7 +40,7 @@ struct SignedRowRange {
 
 struct ReadQuery {
     mutable std::vector<std::string> columns; // empty <=> all columns
-    std::variant<std::monostate, HeadRange, TailRange, SignedRowRange> row_range;
+    std::optional<SignedRowRange> row_range;
     FilterRange row_filter; // no filter by default
     std::vector<std::shared_ptr<Clause>> clauses_;
 
@@ -63,32 +55,16 @@ struct ReadQuery {
     }
 
     void calculate_row_filter(int64_t total_rows) {
-        util::variant_match(row_range,
-            [&](const HeadRange &head_range) {
-            if (head_range.num_rows_ >= 0) {
-                row_filter = RowRange(0, std::min(head_range.num_rows_, total_rows));
-            } else {
-                row_filter = RowRange(0, std::max(static_cast<int64_t>(0), total_rows + head_range.num_rows_));
-            }
-            },
-            [&](const TailRange &tail_range) {
-            if (tail_range.num_rows_ >= 0) {
-                row_filter = RowRange(std::max(static_cast<int64_t>(0), total_rows - tail_range.num_rows_), total_rows);
-            } else {
-                row_filter = RowRange(std::min(-tail_range.num_rows_, total_rows), total_rows);
-            }
-            },
-            [&](const SignedRowRange &signed_row_range) {
-            size_t start = signed_row_range.start_ >= 0 ?
-                    std::min(signed_row_range.start_, total_rows) :
-                    std::max(total_rows + signed_row_range.start_, static_cast<int64_t>(0));
-            size_t end = signed_row_range.end_ >= 0 ?
-                    std::min(signed_row_range.end_, total_rows) :
-                    std::max(total_rows + signed_row_range.end_, static_cast<int64_t>(0));
+        if (row_range.has_value()) {
+            size_t start = row_range->start_ >= 0 ?
+                           std::min(row_range->start_, total_rows) :
+                           std::max(total_rows + row_range->start_,
+                                    static_cast<int64_t>(0));
+            size_t end = row_range->end_ >= 0 ?
+                         std::min(row_range->end_, total_rows) :
+                         std::max(total_rows + row_range->end_, static_cast<int64_t>(0));
             row_filter = RowRange(start, end);
-            },
-            [](const auto &) {}
-            );
+        }
     }
 };
 
diff --git a/cpp/arcticdb/processing/aggregation.hpp b/cpp/arcticdb/processing/aggregation.hpp
index 4cb4239458..4cb7c4caa8 100644
--- a/cpp/arcticdb/processing/aggregation.hpp
+++ b/cpp/arcticdb/processing/aggregation.hpp
@@ -8,7 +8,7 @@
 #pragma once
 
 #include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/processing/bucketizer.hpp>
 #include <arcticdb/entity/types.hpp>
 #include <arcticdb/entity/type_utils.hpp>
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 4c90d1202c..978f997c8f 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -7,7 +7,7 @@
 
 #include <vector>
 #include <variant>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <folly/Poly.h>
 #include <arcticdb/util/composite.hpp>
 #include <arcticdb/processing/clause.hpp>
@@ -20,8 +20,53 @@
 
 namespace arcticdb {
 
-std::vector<Composite<ProcessingSegment>> single_partition(std::vector<Composite<ProcessingSegment>> &&comps) {
-    std::vector<Composite<ProcessingSegment>> v;
+std::vector<Composite<SliceAndKey>> structure_by_row_slice(std::vector<SliceAndKey>& slice_and_keys, size_t start_from) {
+    std::sort(std::begin(slice_and_keys), std::end(slice_and_keys), [] (const SliceAndKey& left, const SliceAndKey& right) {
+        return std::tie(left.slice().row_range.first, left.slice().col_range.first) < std::tie(right.slice().row_range.first, right.slice().col_range.first);
+    });
+
+    std::vector<Composite<SliceAndKey>> rows;
+    auto sk_it = std::begin(slice_and_keys);
+    std::advance(sk_it, start_from);
+    while(sk_it != std::end(slice_and_keys)) {
+        pipelines::RowRange row_range{sk_it->slice().row_range};
+        auto sk = Composite{std::move(*sk_it)};
+        // Iterate through all SliceAndKeys that contain data for the same RowRange - i.e., iterate along column segments
+        // for same row group
+        while(++sk_it != std::end(slice_and_keys) && sk_it->slice().row_range == row_range) {
+            sk.push_back(std::move(*sk_it));
+        }
+
+        util::check(!sk.empty(), "Should not push empty slice/key pairs to the pipeline");
+        rows.emplace_back(std::move(sk));
+    }
+    return rows;
+}
+
+std::vector<Composite<SliceAndKey>> structure_by_column_slice(std::vector<SliceAndKey>& slice_and_keys) {
+    std::sort(std::begin(slice_and_keys), std::end(slice_and_keys), [] (const SliceAndKey& left, const SliceAndKey& right) {
+        return std::tie(left.slice().col_range.first, left.slice().row_range.first) < std::tie(right.slice().col_range.first, right.slice().row_range.first);
+    });
+
+    std::vector<Composite<SliceAndKey>> cols;
+    auto sk_it = std::begin(slice_and_keys);
+    while(sk_it != std::end(slice_and_keys)) {
+        pipelines::ColRange col_range{sk_it->slice().col_range};
+        auto sk = Composite{std::move(*sk_it)};
+        // Iterate through all SliceAndKeys that contain data for the same ColRange - i.e., iterate along row segments
+        // for same column group
+        while(++sk_it != std::end(slice_and_keys) && sk_it->slice().col_range == col_range) {
+            sk.push_back(std::move(*sk_it));
+        }
+
+        util::check(!sk.empty(), "Should not push empty slice/key pairs to the pipeline");
+        cols.emplace_back(std::move(sk));
+    }
+    return cols;
+}
+
+std::vector<Composite<ProcessingUnit>> single_partition(std::vector<Composite<ProcessingUnit>> &&comps) {
+    std::vector<Composite<ProcessingUnit>> v;
     v.push_back(merge_composites_shallow(std::move(comps)));
     return v;
 }
@@ -97,18 +142,18 @@ struct SliceAndKeyWrapper {
     }
 };
 
-Composite<ProcessingSegment> PassthroughClause::process(ARCTICDB_UNUSED const std::shared_ptr<Store> &store,
-                                                        Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> PassthroughClause::process(ARCTICDB_UNUSED const std::shared_ptr<Store> &store,
+                                                     Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
     return procs;
 }
 
-Composite<ProcessingSegment> FilterClause::process(
+Composite<ProcessingUnit> FilterClause::process(
         std::shared_ptr<Store> store,
-        Composite<ProcessingSegment> &&p
+        Composite<ProcessingUnit> &&p
         ) const {
     auto procs = std::move(p);
-    Composite<ProcessingSegment> output;
+    Composite<ProcessingUnit> output;
     procs.broadcast([&optimisation=optimisation_, &store, &expression_context = expression_context_, &output](auto &proc) {
         proc.set_expression_context(expression_context);
         auto variant_data = proc.get(expression_context->root_node_name_, store);
@@ -134,10 +179,10 @@ std::string FilterClause::to_string() const {
     return expression_context_ ? fmt::format("WHERE {}", expression_context_->root_node_name_.value) : "";
 }
 
-Composite<ProcessingSegment> ProjectClause::process(std::shared_ptr<Store> store,
-                                                                  Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> ProjectClause::process(std::shared_ptr<Store> store,
+                                                 Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
-    Composite<ProcessingSegment> output;
+    Composite<ProcessingUnit> output;
     procs.broadcast([&store, &expression_context = expression_context_, &output, that = this](auto &proc) {
         proc.set_expression_context(expression_context);
         auto variant_data = proc.get(expression_context->root_node_name_, store);
@@ -199,8 +244,8 @@ AggregationClause::AggregationClause(const std::string& grouping_column,
     }
 }
 
-Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> store,
-                                                        Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> AggregationClause::process(std::shared_ptr<Store> store,
+                                                     Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
     std::vector<GroupingAggregatorData> aggregators_data;
     internal::check<ErrorCode::E_INVALID_ARGUMENT>(
@@ -210,12 +255,11 @@ Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> s
         aggregators_data.emplace_back(agg.get_aggregator_data());
     }
 
-    // Work out the common type between the processing segments for the columns being aggregated
+    // Work out the common type between the processing units for the columns being aggregated
     procs.broadcast([&store, &aggregators_data, &aggregators=aggregators_](auto& proc) {
         for (auto agg_data: folly::enumerate(aggregators_data)) {
             auto input_column_name = aggregators.at(agg_data.index).get_input_column_name();
             auto input_column = proc.get(input_column_name, store);
-            std::optional<ColumnWithStrings> opt_input_column;
             if (std::holds_alternative<ColumnWithStrings>(input_column)) {
                 agg_data->add_data_type(std::get<ColumnWithStrings>(input_column).column_->type().data_type());
             }
@@ -245,7 +289,7 @@ Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> s
                                                 row_to_group.reserve(col.column_->row_count());
                                                 auto input_data = col.column_->data();
                                                 auto hash_to_group = grouping_map.get<RawType>();
-                                                // For string grouping columns, keep a local map within this ProcessingSegment
+                                                // For string grouping columns, keep a local map within this ProcessingUnit
                                                 // from offsets to groups, to avoid needless calls to col.string_at_offset and
                                                 // string_pool->get
                                                 // This could be slower in cases where there aren't many repeats in string
@@ -336,19 +380,19 @@ Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> s
 
     seg.set_string_pool(string_pool);
     seg.set_row_id(num_unique - 1);
-    return Composite{ProcessingSegment{std::move(seg)}};
+    return Composite{ProcessingUnit{std::move(seg)}};
 }
 
 [[nodiscard]] std::string AggregationClause::to_string() const {
     return fmt::format("AGGREGATE {}", aggregation_map_);
 }
 
-[[nodiscard]] Composite<ProcessingSegment> RemoveColumnPartitioningClause::process(std::shared_ptr<Store> store,
-                                                                                   Composite<ProcessingSegment> &&p) const {
+[[nodiscard]] Composite<ProcessingUnit> RemoveColumnPartitioningClause::process(std::shared_ptr<Store> store,
+                                                                                Composite<ProcessingUnit> &&p) const {
     using namespace arcticdb::pipelines;
     auto procs = std::move(p);
-    Composite<ProcessingSegment> output;
-    procs.broadcast([&store, &output](ProcessingSegment &proc) {
+    Composite<ProcessingUnit> output;
+    procs.broadcast([&store, &output](ProcessingUnit &proc) {
         size_t min_start_row = std::numeric_limits<size_t>::max();
         size_t max_end_row = 0;
         size_t min_start_col = std::numeric_limits<size_t>::max();
@@ -370,18 +414,18 @@ Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> s
         if (output_seg.has_value()) {
             const RowRange row_range{min_start_row, max_end_row};
             const ColRange col_range{min_start_col, max_end_col};
-            output.push_back(ProcessingSegment{std::move(*output_seg), FrameSlice{col_range, row_range}});
+            output.push_back(ProcessingUnit{std::move(*output_seg), FrameSlice{col_range, row_range}});
         }
     });
     return output;
 }
 
-Composite<ProcessingSegment> SplitClause::process(std::shared_ptr<Store> store,
-                                                  Composite<ProcessingSegment> &&procs) const {
+Composite<ProcessingUnit> SplitClause::process(std::shared_ptr<Store> store,
+                                               Composite<ProcessingUnit> &&procs) const {
     using namespace arcticdb::pipelines;
 
     auto proc_composite = std::move(procs);
-    Composite<ProcessingSegment> ret;
+    Composite<ProcessingUnit> ret;
     proc_composite.broadcast([&store, rows = rows_, &ret](auto &&p) {
         auto proc = std::forward<decltype(p)>(p);
         auto slice_and_keys = proc.data();
@@ -393,7 +437,7 @@ Composite<ProcessingSegment> SplitClause::process(std::shared_ptr<Store> store,
             for (auto &item : split_segs) {
                 end_row = start_row + item.row_count();
                 const RowRange row_range{start_row, end_row};
-                ret.push_back(ProcessingSegment{std::move(item), FrameSlice{col_range, row_range}});
+                ret.push_back(ProcessingUnit{std::move(item), FrameSlice{col_range, row_range}});
                 start_row = end_row;
             }
         }
@@ -401,8 +445,8 @@ Composite<ProcessingSegment> SplitClause::process(std::shared_ptr<Store> store,
     return ret;
 }
 
-Composite<ProcessingSegment> SortClause::process(std::shared_ptr<Store> store,
-                                                 Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> SortClause::process(std::shared_ptr<Store> store,
+                                              Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
     procs.broadcast([&store, &column = column_](auto &proc) {
         auto slice_and_keys = proc.data();
@@ -415,7 +459,7 @@ Composite<ProcessingSegment> SortClause::process(std::shared_ptr<Store> store,
 
 template<typename IndexType, typename DensityPolicy, typename QueueType, typename Comparator, typename StreamId>
 void merge_impl(
-        Composite<ProcessingSegment> &ret,
+        Composite<ProcessingUnit> &ret,
         QueueType &input_streams,
         bool add_symbol_column,
         StreamId stream_id,
@@ -429,7 +473,7 @@ void merge_impl(
     SegmentationPolicy segmentation_policy{static_cast<size_t>(num_segment_rows)};
 
     auto func = [&ret, &row_range, &col_range](auto &&segment) {
-        ret.push_back(ProcessingSegment{std::forward<SegmentInMemory>(segment), FrameSlice{col_range, row_range}});
+        ret.push_back(ProcessingUnit{std::forward<SegmentInMemory>(segment), FrameSlice{col_range, row_range}});
     };
     
     using AggregatorType = stream::Aggregator<IndexType, stream::DynamicSchema, SegmentationPolicy, DensityPolicy>;
@@ -451,8 +495,8 @@ void merge_impl(
 
 // MergeClause receives a list of DataFrames as input and merge them into a single one where all 
 // the rows are sorted by time stamp
-Composite<ProcessingSegment> MergeClause::process(std::shared_ptr<Store> store,
-                                                  Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> MergeClause::process(std::shared_ptr<Store> store,
+                                               Composite<ProcessingUnit> &&p) const {
     using namespace arcticdb::pipelines;
     auto procs = std::move(p);
 
@@ -491,7 +535,7 @@ Composite<ProcessingSegment> MergeClause::process(std::shared_ptr<Store> store,
     });
     const RowRange row_range{min_start_row, max_end_row};
     const ColRange col_range{min_start_col, max_end_col};
-    Composite<ProcessingSegment> ret;
+    Composite<ProcessingUnit> ret;
     std::visit(
             [&ret, &input_streams, add_symbol_column = add_symbol_column_, &comp = compare, stream_id = stream_id_, &row_range, &col_range, &stream_descriptor = stream_descriptor_](auto idx,
                                                                                             auto density) {
@@ -508,15 +552,15 @@ Composite<ProcessingSegment> MergeClause::process(std::shared_ptr<Store> store,
     return ret;
 }
 
-std::optional<std::vector<Composite<ProcessingSegment>>> MergeClause::repartition(
-        std::vector<Composite<ProcessingSegment>> &&comps) const {
-    std::vector<Composite<ProcessingSegment>> v;
+std::optional<std::vector<Composite<ProcessingUnit>>> MergeClause::repartition(
+        std::vector<Composite<ProcessingUnit>> &&comps) const {
+    std::vector<Composite<ProcessingUnit>> v;
     v.push_back(merge_composites_shallow(std::move(comps)));
     return v;
 }
 
-Composite<ProcessingSegment> ColumnStatsGenerationClause::process(std::shared_ptr<Store> store,
-                                                                  Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingUnit> ColumnStatsGenerationClause::process(std::shared_ptr<Store> store,
+                                                               Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
     std::vector<ColumnStatsAggregatorData> aggregators_data;
     internal::check<ErrorCode::E_INVALID_ARGUMENT>(
@@ -531,7 +575,7 @@ Composite<ProcessingSegment> ColumnStatsGenerationClause::process(std::shared_pt
 
     internal::check<ErrorCode::E_INVALID_ARGUMENT>(
             !procs.empty(),
-            "ColumnStatsGenerationClause::process does not make sense with no processing segments");
+            "ColumnStatsGenerationClause::process does not make sense with no processing units");
     procs.broadcast(
             [&store, &start_indexes, &end_indexes, &aggregators_data, that=this](
                     auto &proc) {
@@ -557,7 +601,7 @@ Composite<ProcessingSegment> ColumnStatsGenerationClause::process(std::shared_pt
 
     internal::check<ErrorCode::E_ASSERTION_FAILURE>(
             start_indexes.size() == 1 && end_indexes.size() == 1,
-            "Expected all data segments in one processing segment to have same start and end indexes");
+            "Expected all data segments in one processing unit to have same start and end indexes");
     auto start_index = *start_indexes.begin();
     auto end_index = *end_indexes.begin();
     schema::check<ErrorCode::E_UNSUPPORTED_INDEX_TYPE>(
@@ -579,26 +623,40 @@ Composite<ProcessingSegment> ColumnStatsGenerationClause::process(std::shared_pt
         seg.concatenate(agg_data->finalize(column_stats_aggregators_->at(agg_data.index).get_output_column_names()));
     }
     seg.set_row_id(0);
-    return Composite{ProcessingSegment{std::move(seg)}};
+    return Composite{ProcessingUnit{std::move(seg)}};
 }
 
-Composite<ProcessingSegment> RowRangeClause::process(std::shared_ptr<Store> store,
-                                                           Composite<ProcessingSegment> &&p) const {
+std::vector<Composite<SliceAndKey>> RowRangeClause::structure_for_processing(
+        std::vector<SliceAndKey>& slice_and_keys, ARCTICDB_UNUSED size_t start_from) const {
+    slice_and_keys.erase(std::remove_if(slice_and_keys.begin(), slice_and_keys.end(), [this](const SliceAndKey& slice_and_key) {
+        return slice_and_key.slice_.row_range.start() >= end_ || slice_and_key.slice_.row_range.end() <= start_;
+    }), slice_and_keys.end());
+    return structure_by_column_slice(slice_and_keys);
+}
+
+Composite<ProcessingUnit> RowRangeClause::process(std::shared_ptr<Store> store,
+                                                  Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
-    procs.broadcast([&store, this](ProcessingSegment &proc) {
-        auto row_range = proc.data()[0].slice_.row_range;
-        if ((start_ > row_range.start() && start_ < row_range.end()) || (end_ > row_range.start() && end_ < row_range.end())) {
-            // Zero-indexed within this ProcessingSegment
-            size_t start_row{0};
-            size_t end_row{row_range.diff()};
-            if (start_ > row_range.start() && start_ < row_range.end()) {
-                start_row = start_ - row_range.start();
-            }
-            if (end_ > row_range.start() && end_ < row_range.end()) {
-                end_row = end_ - (row_range.start());
-            }
-            proc.truncate(start_row, end_row, store);
-        } // else all rows in the processing segment are required, do nothing
+    procs.broadcast([&store, this](ProcessingUnit &proc) {
+        for (auto& slice_and_key: proc.data()) {
+            auto row_range = slice_and_key.slice_.row_range;
+            if ((start_ > row_range.start() && start_ < row_range.end()) ||
+                (end_ > row_range.start() && end_ < row_range.end())) {
+                // Zero-indexed within this slice
+                size_t start_row{0};
+                size_t end_row{row_range.diff()};
+                if (start_ > row_range.start() && start_ < row_range.end()) {
+                    start_row = start_ - row_range.start();
+                }
+                if (end_ > row_range.start() && end_ < row_range.end()) {
+                    end_row = end_ - (row_range.start());
+                }
+                auto seg = truncate_segment(slice_and_key.segment(store), start_row, end_row);
+                slice_and_key.slice_.adjust_rows(seg.row_count());
+                slice_and_key.slice_.adjust_columns(seg.descriptor().field_count() - seg.descriptor().index().field_count());
+                slice_and_key.segment_ = std::move(seg);
+            } // else all rows in the slice and key are required, do nothing
+        }
     });
     return procs;
 }
@@ -631,10 +689,19 @@ std::string RowRangeClause::to_string() const {
     return fmt::format("{} {}", row_range_type_ == RowRangeType::HEAD ? "HEAD" : "TAIL", n_);
 }
 
-Composite<ProcessingSegment> DateRangeClause::process(ARCTICDB_UNUSED std::shared_ptr<Store> store,
-                                                      Composite<ProcessingSegment> &&p) const {
+std::vector<Composite<SliceAndKey>> DateRangeClause::structure_for_processing(
+        std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+    slice_and_keys.erase(std::remove_if(slice_and_keys.begin(), slice_and_keys.end(), [this](const SliceAndKey& slice_and_key) {
+        auto [start_index, end_index] = slice_and_key.key().time_range();
+        return start_index > end_ || end_index <= start_;
+    }), slice_and_keys.end());
+    return structure_by_row_slice(slice_and_keys, start_from);
+}
+
+Composite<ProcessingUnit> DateRangeClause::process(ARCTICDB_UNUSED std::shared_ptr<Store> store,
+                                                   Composite<ProcessingUnit> &&p) const {
     auto procs = std::move(p);
-    procs.broadcast([&store, this](ProcessingSegment &proc) {
+    procs.broadcast([&store, this](ProcessingUnit &proc) {
         // We are only interested in the index, which is in every SegmentInMemory in proc.data(), so just use the first
         auto slice_and_key = proc.data()[0];
         auto row_range = slice_and_key.slice_.row_range;
@@ -649,7 +716,7 @@ Composite<ProcessingSegment> DateRangeClause::process(ARCTICDB_UNUSED std::share
                 end_row = slice_and_key.segment(store).column_ptr(0)->search_sorted<timestamp>(end_, true);
             }
             proc.truncate(start_row, end_row, store);
-        } // else all rows in the processing segment are required, do nothing
+        } // else all rows in the processing unit are required, do nothing
     });
     return procs;
 }
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index 96b61d4d78..821fdc5251 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -9,10 +9,11 @@
 
 #include <arcticdb/entity/key.hpp>
 #include <arcticdb/column_store/column.hpp>
+#include <arcticdb/pipeline/frame_slice.hpp>
 #include <arcticdb/pipeline/value.hpp>
 #include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/entity/types.hpp>
 #include <arcticdb/processing/aggregation.hpp>
 #include <arcticdb/processing/grouper.hpp>
@@ -34,6 +35,8 @@
 
 namespace arcticdb {
 
+using SliceAndKey = pipelines::SliceAndKey;
+
 using NormMetaDescriptor = std::shared_ptr<arcticdb::proto::descriptors::NormalizationMetadata>;
 
 // Contains constant data about the clause identifiable at construction time
@@ -61,25 +64,33 @@ struct ProcessingConfig {
 struct IClause {
     template<class Base>
     struct Interface : Base {
-        [[nodiscard]] Composite<ProcessingSegment>
-        process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&segs) const {
-            return std::move(folly::poly_call<0>(*this, store, std::move(segs)));
+        // (at least) the first clause in the pipeline will be parallelied over elements of the returned vector
+        // TODO #732: Factor out start_from as part of https://github.com/man-group/ArcticDB/issues/732
+        [[nodiscard]] std::vector<Composite<SliceAndKey>>
+        structure_for_processing(std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+            return std::move(folly::poly_call<0>(*this, slice_and_keys, start_from));
+        }
+
+        [[nodiscard]] Composite<ProcessingUnit>
+        process(std::shared_ptr<Store> store, Composite<ProcessingUnit> &&segs) const {
+            return std::move(folly::poly_call<1>(*this, store, std::move(segs)));
         }
 
-        [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-                [[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const {
-            return folly::poly_call<1>(*this, std::move(comps));
+        [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+                [[maybe_unused]] std::vector<Composite<ProcessingUnit>> &&comps) const {
+            return folly::poly_call<2>(*this, std::move(comps));
         }
 
-        [[nodiscard]] const ClauseInfo& clause_info() const { return folly::poly_call<2>(*this); };
+        [[nodiscard]] const ClauseInfo& clause_info() const { return folly::poly_call<3>(*this); };
 
         void set_processing_config(const ProcessingConfig& processing_config) {
-            folly::poly_call<3>(*this, processing_config);
+            folly::poly_call<4>(*this, processing_config);
         }
     };
 
     template<class T>
     using Members = folly::PolyMembers<
+            &T::structure_for_processing,
             &T::process,
             &T::repartition,
             &T::clause_info,
@@ -88,18 +99,26 @@ struct IClause {
 
 using Clause = folly::Poly<IClause>;
 
+std::vector<Composite<SliceAndKey>> structure_by_row_slice(std::vector<SliceAndKey>& slice_and_keys, size_t start_from);
+std::vector<Composite<SliceAndKey>> structure_by_column_slice(std::vector<SliceAndKey>& slice_and_keys);
+
 struct PassthroughClause {
     ClauseInfo clause_info_;
     PassthroughClause() = default;
     ARCTICDB_MOVE_COPY_DEFAULT(PassthroughClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(
             ARCTICDB_UNUSED const std::shared_ptr<Store> &store,
-            Composite<ProcessingSegment> &&p
+            Composite<ProcessingUnit> &&p
             ) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&comps) const {
         return std::nullopt;
     }
 
@@ -127,13 +146,18 @@ struct FilterClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(FilterClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(
             std::shared_ptr<Store> store,
-            Composite<ProcessingSegment> &&p
+            Composite<ProcessingUnit> &&p
             ) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&) const {
         return std::nullopt;
     }
 
@@ -170,11 +194,16 @@ struct ProjectClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(ProjectClause)
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit>
+    process(std::shared_ptr<Store> store, Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&comps) const {
         return std::nullopt;
     }
 
@@ -205,9 +234,14 @@ struct PartitionClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(PartitionClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const {
-        Composite<ProcessingSegment> output;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const {
+        Composite<ProcessingUnit> output;
         auto procs = std::move(p);
         procs.broadcast([&output, &store, that=this](auto &proc) {
             output.push_back(partition_processing_segment<GrouperType, BucketizerType>(store,
@@ -218,12 +252,12 @@ struct PartitionClause {
         return output;
     }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&c) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&c) const {
         auto comps = std::move(c);
-        std::unordered_map<size_t, Composite<ProcessingSegment>> partition_map;
+        std::unordered_map<size_t, Composite<ProcessingUnit>> partition_map;
         schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                std::any_of(comps.begin(), comps.end(), [](const Composite<ProcessingSegment>& proc) {
+                std::any_of(comps.begin(), comps.end(), [](const Composite<ProcessingUnit>& proc) {
                     return !proc.empty();
                 }),
                 "Grouping column {} does not exist or is empty", grouping_column_
@@ -234,14 +268,14 @@ struct PartitionClause {
                 auto bucket_id = proc.get_bucket();
 
                 if (partition_map.find(bucket_id) == partition_map.end()) {
-                    partition_map[bucket_id] = Composite<ProcessingSegment>(std::move(proc));
+                    partition_map[bucket_id] = Composite<ProcessingUnit>(std::move(proc));
                 } else {
                     partition_map[bucket_id].push_back(std::move(proc));
                 }
             });
         }
 
-        std::vector<Composite<ProcessingSegment>> ret;
+        std::vector<Composite<ProcessingUnit>> ret;
         for (auto &[key, value] : partition_map) {
             ret.push_back(std::move(value));
         }
@@ -283,11 +317,18 @@ struct AggregationClause {
                       const std::unordered_map<std::string,
                       std::string>& aggregations);
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            ARCTICDB_UNUSED const std::vector<SliceAndKey>& slice_and_keys, ARCTICDB_UNUSED size_t start_from) const {
+        internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
+                "AggregationClause::structure_for_processing should never be called"
+                );
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&comps
             ) const {
         return std::nullopt;
     }
@@ -312,11 +353,16 @@ struct RemoveColumnPartitioningClause {
     }
     ARCTICDB_MOVE_COPY_DEFAULT(RemoveColumnPartitioningClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&) const {
         return std::nullopt;
     }
 
@@ -336,11 +382,16 @@ struct SplitClause {
             rows_(rows) {
     }
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&procs) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&procs) const;
+
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&comps) const {
         return std::nullopt;
     }
 
@@ -359,11 +410,16 @@ struct SortClause {
             column_(std::move(column)) {
     }
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&comps) const {
         return std::nullopt;
     }
 
@@ -395,11 +451,16 @@ struct MergeClause {
         clause_info_.requires_repartition_ = true;
     }
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            std::vector<Composite<ProcessingSegment>> &&comps) const;
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            std::vector<Composite<ProcessingUnit>> &&comps) const;
 
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
@@ -424,11 +485,16 @@ struct ColumnStatsGenerationClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(ColumnStatsGenerationClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const {
+        return structure_by_row_slice(slice_and_keys, start_from);
+    }
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&) const {
         return std::nullopt;
     }
 
@@ -467,11 +533,14 @@ struct RowRangeClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(RowRangeClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, ARCTICDB_UNUSED size_t start_from) const;
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&) const {
         return std::nullopt;
     }
 
@@ -500,11 +569,14 @@ struct DateRangeClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(DateRangeClause)
 
-    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
-                                                       Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] std::vector<Composite<SliceAndKey>> structure_for_processing(
+            std::vector<SliceAndKey>& slice_and_keys, size_t start_from) const;
+
+    [[nodiscard]] Composite<ProcessingUnit> process(std::shared_ptr<Store> store,
+                                                    Composite<ProcessingUnit> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
-            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingUnit>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingUnit>> &&) const {
         return std::nullopt;
     }
 
diff --git a/cpp/arcticdb/processing/execution_context.hpp b/cpp/arcticdb/processing/execution_context.hpp
deleted file mode 100644
index 2232fb164f..0000000000
--- a/cpp/arcticdb/processing/execution_context.hpp
+++ /dev/null
@@ -1,116 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
- */
-
-#pragma once
-
-#include <memory>
-#include <string>
-#include <unordered_map>
-#include <unordered_set>
-
-#include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/entity/stream_descriptor.hpp>
-#include <arcticdb/pipeline/value.hpp>
-#include <arcticdb/pipeline/value_set.hpp>
-
-namespace arcticdb {
-
-/*
- * Contains an expression tree as a map of string node names to nodes. Nodes contain leaves which may be references
- * to node names.
- *
- * ExecutionContext also contains the name of the root node as well as a map of node names to constant values.
- *
- * This class effectively contains as AST - there is likely a frontend implementation in a higher level language (for
- * example Python).
- */
-struct ExecutionContext {
-    ExecutionContext() = default;
-    using NormMetaDescriptor = std::shared_ptr<arcticdb::proto::descriptors::NormalizationMetadata>;
-
-    ARCTICDB_MOVE_ONLY_DEFAULT(ExecutionContext)
-
-    template <class T>
-    class ConstantMap {
-        std::unordered_map<std::string, std::shared_ptr<T>> map_;
-    public:
-        void set_value(std::string name, std::shared_ptr<T> val) {
-            map_.try_emplace(name, val);
-        }
-        std::shared_ptr<T> get_value(std::string name) {
-            return map_.at(name);
-        }
-    };
-    enum class Optimisation : uint8_t {
-        SPEED,
-        MEMORY
-    };
-    void add_column(const std::string& column_name) {
-        columns_.emplace(column_name);
-    }
-
-    void add_expression_node(const std::string& name, std::shared_ptr<ExpressionNode> expression_node) {
-        expression_nodes_.set_value(name, std::move(expression_node));
-    }
-
-    void set_descriptor(const StreamDescriptor& desc) {
-        output_descriptor_ = desc;
-    }
-
-    void set_norm_meta_descriptor(NormMetaDescriptor& desc) {
-        norm_meta_ = desc;
-    }
-
-    NormMetaDescriptor get_norm_meta_descriptor() {
-        return norm_meta_;
-    }
-
-    void add_value(const std::string& name, std::shared_ptr<Value> value) {
-        values_.set_value(name, std::move(value));
-    }
-
-    void add_value_set(const std::string& name, std::shared_ptr<ValueSet> value_set) {
-        value_sets_.set_value(name, std::move(value_set));
-    }
-
-    void check_output_column(const std::string& name, DataType type) {
-        std::scoped_lock lock{*column_mutex_};
-        auto it = output_columns_.find(name);
-        if(it == std::end(output_columns_)) {
-            output_columns_.try_emplace(name, type);
-            output_descriptor_->add_field(scalar_field(type, name));
-            //TODO check type compatibility in dynamic schema
-        }
-    }
-
-    bool dynamic_schema() const {
-        return dynamic_schema_;
-    }
-
-    void set_dynamic_schema(bool dynamic_schema) {
-        dynamic_schema_ = dynamic_schema;
-    }
-
-    std::unordered_set<std::string> columns_;
-    ConstantMap<ExpressionNode> expression_nodes_;
-    ConstantMap<Value> values_;
-    ConstantMap<ValueSet> value_sets_;
-    ExpressionName root_node_name_;
-    Optimisation optimisation_ = Optimisation::SPEED;
-    // TODO (AN-469): This class should be immutable (and possibly just incorporated into Clause, as there is always
-    //  exactly one execution context per clause). Any mutable state that needs to be maintained as a ProcessingSegment
-    //  moves through the processing pipeline should be stored somewhere else (possibly on the ProcessingSegment itself)
-    std::optional<StreamDescriptor> output_descriptor_;
-    std::optional<StreamDescriptor> orig_output_descriptor_;
-    NormMetaDescriptor norm_meta_;
-    std::unordered_map<std::string, DataType> output_columns_;
-    std::unique_ptr<std::mutex> column_mutex_ = std::make_unique<std::mutex>();
-    std::unique_ptr<std::mutex> name_index_mutex_ = std::make_unique<std::mutex>();
-    bool dynamic_schema_ = false;
-};
-
-}//namespace arcticdb
diff --git a/cpp/arcticdb/processing/expression_node.cpp b/cpp/arcticdb/processing/expression_node.cpp
index 62c9f389f3..aa91ae6018 100644
--- a/cpp/arcticdb/processing/expression_node.cpp
+++ b/cpp/arcticdb/processing/expression_node.cpp
@@ -6,7 +6,7 @@
  */
 
 #include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/processing/operation_types.hpp>
 #include <arcticdb/processing/operation_dispatch_binary.hpp>
 #include <arcticdb/processing/operation_dispatch_unary.hpp>
@@ -26,7 +26,7 @@ ExpressionNode::ExpressionNode(VariantNode left, OperationType op) :
     util::check(!is_binary_operation(op), "Binary expression expects both left and right children");
 }
 
-VariantData ExpressionNode::compute(ProcessingSegment& seg, const std::shared_ptr<Store>& store) const {
+VariantData ExpressionNode::compute(ProcessingUnit& seg, const std::shared_ptr<Store>& store) const {
     if (is_binary_operation(operation_type_)) {
         return dispatch_binary(seg.get(left_, store), seg.get(right_, store), operation_type_);
     } else {
diff --git a/cpp/arcticdb/processing/expression_node.hpp b/cpp/arcticdb/processing/expression_node.hpp
index e36b5b0983..f358c7b0e8 100644
--- a/cpp/arcticdb/processing/expression_node.hpp
+++ b/cpp/arcticdb/processing/expression_node.hpp
@@ -34,7 +34,7 @@ using ExpressionName = util::StringWrappingValue<ExpressionNameTag>;
 
 using VariantNode = std::variant<std::monostate, ColumnName, ValueName, ValueSetName, ExpressionName>;
 
-struct ProcessingSegment;
+struct ProcessingUnit;
 class Store;
 
 /*
@@ -110,7 +110,7 @@ struct ExpressionNode {
 
     ExpressionNode(VariantNode left, OperationType op);
 
-    VariantData compute(ProcessingSegment& seg, const std::shared_ptr<Store>& store) const;
+    VariantData compute(ProcessingUnit& seg, const std::shared_ptr<Store>& store) const;
 };
 
 } //namespace arcticdb
diff --git a/cpp/arcticdb/processing/operation_dispatch.cpp b/cpp/arcticdb/processing/operation_dispatch.cpp
index fc130ae758..e19ca044a1 100644
--- a/cpp/arcticdb/processing/operation_dispatch.cpp
+++ b/cpp/arcticdb/processing/operation_dispatch.cpp
@@ -12,8 +12,8 @@ namespace arcticdb {
 
 VariantData transform_to_placeholder(VariantData data) {
     // There is a further optimization we can do here which is to return FullResult
-    // when the number of set bits is the same as the processing segments row-count,
-    // however that would need to be maintained by the processing segment and
+    // when the number of set bits is the same as the processing units row-count,
+    // however that would need to be maintained by the processing unit and
     // modified with each transformation. It would be worth it in the scenario
     // where a true result is very common, but is a premature change to make at this point.
     return util::variant_match(data,
diff --git a/cpp/arcticdb/processing/processing_segment.cpp b/cpp/arcticdb/processing/processing_unit.cpp
similarity index 92%
rename from cpp/arcticdb/processing/processing_segment.cpp
rename to cpp/arcticdb/processing/processing_unit.cpp
index 90346fb6b3..7cf42c3c04 100644
--- a/cpp/arcticdb/processing/processing_segment.cpp
+++ b/cpp/arcticdb/processing/processing_unit.cpp
@@ -5,10 +5,10 @@
  * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
  */
 
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 
 namespace arcticdb {
-void ProcessingSegment::apply_filter(
+void ProcessingUnit::apply_filter(
     const util::BitSet& bitset,
     const std::shared_ptr<Store>& store,
     PipelineOptimisation optimisation) {
@@ -30,7 +30,7 @@ void ProcessingSegment::apply_filter(
 }
 
 // Inclusive of start_row, exclusive of end_row
-void ProcessingSegment::truncate(size_t start_row, size_t end_row, const std::shared_ptr<Store>& store) {
+void ProcessingUnit::truncate(size_t start_row, size_t end_row, const std::shared_ptr<Store>& store) {
     for (auto& slice_and_key: data_) {
         auto seg = truncate_segment(slice_and_key.segment(store), start_row, end_row);
         if(!seg.is_null()) {
@@ -44,7 +44,7 @@ void ProcessingSegment::truncate(size_t start_row, size_t end_row, const std::sh
     }
 }
 
-VariantData ProcessingSegment::get(const VariantNode &name, const std::shared_ptr<Store> &store) {
+VariantData ProcessingUnit::get(const VariantNode &name, const std::shared_ptr<Store> &store) {
     return util::variant_match(name,
         [&](const ColumnName &column_name) {
         for (auto &slice_and_key: data_) {
@@ -95,7 +95,7 @@ VariantData ProcessingSegment::get(const VariantNode &name, const std::shared_pt
         }
         },
         [&]([[maybe_unused]] const std::monostate &unused) -> VariantData {
-        util::raise_rte("ProcessingSegment::get called with monostate VariantNode");
+        util::raise_rte("ProcessingUnit::get called with monostate VariantNode");
     }
     );
     util::raise_rte("Unexpected value, or expression");
diff --git a/cpp/arcticdb/processing/processing_segment.hpp b/cpp/arcticdb/processing/processing_unit.hpp
similarity index 86%
rename from cpp/arcticdb/processing/processing_segment.hpp
rename to cpp/arcticdb/processing/processing_unit.hpp
index e29b39cec6..50e3e90b66 100644
--- a/cpp/arcticdb/processing/processing_segment.hpp
+++ b/cpp/arcticdb/processing/processing_unit.hpp
@@ -29,28 +29,28 @@ namespace arcticdb {
     };
 
     /*
-     * A processing segment is designed to be used in conjunction with the clause processing framework.
-     * Each clause will execute in turn and will be passed the same mutable ProcessingSegment which it will have free
+     * A processing unit is designed to be used in conjunction with the clause processing framework.
+     * Each clause will execute in turn and will be passed the same mutable ProcessingUnit which it will have free
      * rein to modify prior to passing off to the next clause. In contract, clauses and the expression trees contained
      * within are immutable as they represent predefined behaviour.
      *
-     * ProcessingSegment contains two main member attributes:
+     * ProcessingUnit contains two main member attributes:
      *
      * ExpressionContext: This contains the expression tree for the currently running clause - each clause contains an
      * expression tree containing nodes that map to columns, constant values or other nodes in the tree. This allows
      * for the processing of expressions such as `(df['a'] + 4) < df['b']`. One instance of an expression context can be
-     * shared across many ProcessingSegment's. See ExpressionContext class for more documentation.
+     * shared across many ProcessingUnit's. See ExpressionContext class for more documentation.
      *
      * computed_data_: This contains a map from node name to previously computed result. Should an expression such as
      * df['col1'] + 1 appear multiple times in the tree this allows us to only perform the computation once and spill
      * the result to disk.
      */
-    struct ProcessingSegment {
+    struct ProcessingUnit {
         // We want a collection of SliceAndKeys here so that we have all of the columns for a given row present in a single
-        // ProcessingSegment, for 2 reasons:
+        // ProcessingUnit, for 2 reasons:
         // 1: For binary operations in ExpressionNode involving 2 columns, we need both available in a single processing
         //    segment in order for "get" calls to make sense
-        // 2: As "get" uses cached results where possible, it is not obvious which processing segment should hold the
+        // 2: As "get" uses cached results where possible, it is not obvious which processing unit should hold the
         //    cached result for operations involving 2 columns (probably both). By moving these cached results up a level
         //    to contain all the information for a given row, this becomes unambiguous.
         std::vector<pipelines::SliceAndKey> data_;
@@ -60,29 +60,29 @@ namespace arcticdb {
         // Set by PartitioningClause
         std::optional<size_t> bucket_;
 
-        ProcessingSegment() = default;
+        ProcessingUnit() = default;
 
-        ProcessingSegment(SegmentInMemory &&seg,
-                          pipelines::FrameSlice&& slice,
-                          std::optional<size_t> bucket = std::nullopt) :
+        ProcessingUnit(SegmentInMemory &&seg,
+                       pipelines::FrameSlice&& slice,
+                       std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(seg), std::move(slice)});
         }
 
-        explicit ProcessingSegment(pipelines::SliceAndKey &&sk,
-                                   std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingUnit(pipelines::SliceAndKey &&sk,
+                                std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(sk)});
         }
 
-        explicit ProcessingSegment(std::vector<pipelines::SliceAndKey> &&sk,
-                                   std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingUnit(std::vector<pipelines::SliceAndKey> &&sk,
+                                std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_ = std::move(sk);
         }
 
-        explicit ProcessingSegment(SegmentInMemory &&seg,
-                                   std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingUnit(SegmentInMemory &&seg,
+                                std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(seg)});
         }
@@ -99,13 +99,13 @@ namespace arcticdb {
             return data_;
         }
 
-        ProcessingSegment &self() {
+        ProcessingUnit &self() {
             return *this;
         }
 
         size_t get_bucket(){
             internal::check<ErrorCode::E_ASSERTION_FAILURE>(bucket_.has_value(),
-                                                            "ProcessingSegment::get_bucket called, but bucket_ has no value");
+                                                            "ProcessingUnit::get_bucket called, but bucket_ has no value");
             return *bucket_;
         }
 
@@ -127,12 +127,12 @@ namespace arcticdb {
         VariantData get(const VariantNode &name, const std::shared_ptr<Store> &store);
     };
 
-    inline std::vector<pipelines::SliceAndKey> collect_segments(Composite<ProcessingSegment>&& p) {
+    inline std::vector<pipelines::SliceAndKey> collect_segments(Composite<ProcessingUnit>&& p) {
         auto procs = std::move(p);
         std::vector<pipelines::SliceAndKey> output;
 
         procs.broadcast([&output] (auto&& p) {
-            auto proc = std::forward<ProcessingSegment>(p);
+            auto proc = std::forward<ProcessingUnit>(p);
             auto slice_and_keys = proc.release_data();
             std::move(std::begin(slice_and_keys), std::end(slice_and_keys), std::back_inserter(output));
         });
@@ -153,7 +153,7 @@ namespace arcticdb {
         std::vector<std::optional<uint8_t>> row_to_bucket;
         row_to_bucket.reserve(col.column_->row_count());
         // Tracks how many rows are in each bucket
-        // Use to skip empty buckets, and presize columns in the output ProcessingSegments
+        // Use to skip empty buckets, and presize columns in the output ProcessingUnit
         std::vector<uint64_t> bucket_counts(bucketizer.num_buckets(), 0);
 
         using TypeDescriptorTag = typename Grouper::GrouperDescriptor;
@@ -179,13 +179,13 @@ namespace arcticdb {
     }
 
     template<typename GrouperType, typename BucketizerType>
-    Composite<ProcessingSegment> partition_processing_segment(
+    Composite<ProcessingUnit> partition_processing_segment(
             const std::shared_ptr<Store>& store,
-            ProcessingSegment& input,
+            ProcessingUnit& input,
             const ColumnName& grouping_column_name,
             bool dynamic_schema) {
 
-        Composite<ProcessingSegment> output;
+        Composite<ProcessingUnit> output;
         auto get_result = input.get(ColumnName(grouping_column_name), store);
         if (std::holds_alternative<ColumnWithStrings>(get_result)) {
             auto partitioning_column = std::get<ColumnWithStrings>(get_result);
@@ -201,7 +201,7 @@ namespace arcticdb {
                                         num_buckets);
                     num_buckets = std::numeric_limits<uint8_t>::max();
                 }
-                std::vector<ProcessingSegment> procs{static_cast<size_t>(num_buckets)};
+                std::vector<ProcessingUnit> procs{static_cast<size_t>(num_buckets)};
                 BucketizerType bucketizer(num_buckets);
                 auto [row_to_bucket, bucket_counts] = get_buckets(partitioning_column, grouper, bucketizer);
                 for (auto& seg_slice_and_key : input.data()) {
diff --git a/cpp/arcticdb/stream/stream_source.hpp b/cpp/arcticdb/stream/stream_source.hpp
index 97798dbd5f..8192140fcb 100644
--- a/cpp/arcticdb/stream/stream_source.hpp
+++ b/cpp/arcticdb/stream/stream_source.hpp
@@ -70,7 +70,7 @@ struct StreamSource {
 
     using DecodeContinuation = folly::Function<folly::Unit(SegmentInMemory &&)>;
 
-    virtual std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
+    virtual std::vector<Composite<ProcessingUnit>> batch_read_uncompressed(
         std::vector<Composite<pipelines::SliceAndKey>> &&keys,
         const std::vector<std::shared_ptr<Clause>>& clauses,
         const std::shared_ptr<std::unordered_set<std::string>>& filter_columns,
diff --git a/cpp/arcticdb/version/python_bindings.cpp b/cpp/arcticdb/version/python_bindings.cpp
index 33765d12a6..731f2b8c9b 100644
--- a/cpp/arcticdb/version/python_bindings.cpp
+++ b/cpp/arcticdb/version/python_bindings.cpp
@@ -217,16 +217,6 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
         .def_property_readonly("end", &pipelines::RowRange::end)
         .def_property_readonly("diff", &pipelines::RowRange::diff);
 
-    py::class_<pipelines::HeadRange, std::shared_ptr<pipelines::HeadRange>>(version, "HeadRange")
-    .def(py::init([](int64_t n){
-        return HeadRange{n};
-    }));
-
-    py::class_<pipelines::TailRange, std::shared_ptr<pipelines::TailRange>>(version, "TailRange")
-    .def(py::init([](int64_t n){
-        return TailRange{n};
-    }));
-
     py::class_<pipelines::SignedRowRange, std::shared_ptr<pipelines::SignedRowRange>>(version, "SignedRowRange")
     .def(py::init([](int64_t start, int64_t end){
         return SignedRowRange{start, end};
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 92d253f426..4ce00544c7 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -403,16 +403,16 @@ FrameAndDescriptor read_multi_key(
     return {res.frame_, multi_key_desc, keys, std::shared_ptr<BufferHolder>{}};
 }
 
-Composite<ProcessingSegment> process_remaining_clauses(
+Composite<ProcessingUnit> process_remaining_clauses(
         const std::shared_ptr<Store>& store,
-        std::vector<Composite<ProcessingSegment>>&& procs,
+        std::vector<Composite<ProcessingUnit>>&& procs,
         std::vector<std::shared_ptr<Clause>> clauses ) { // pass by copy deliberately as we don't want to modify read_query
     while (!clauses.empty()) {
         if (clauses[0]->clause_info().requires_repartition_) {
-            std::vector<Composite<ProcessingSegment>> repartitioned_procs = clauses[0]->repartition(std::move(procs)).value();
+            std::vector<Composite<ProcessingUnit>> repartitioned_procs = clauses[0]->repartition(std::move(procs)).value();
             // Erasing from front of vector not ideal, but they're just shared_ptr and there shouldn't be loads of clauses
             clauses.erase(clauses.begin());
-            std::vector<folly::Future<Composite<ProcessingSegment>>> fut_procs;
+            std::vector<folly::Future<Composite<ProcessingUnit>>> fut_procs;
             for (auto&& proc : repartitioned_procs) {
                 fut_procs.emplace_back(
                         async::submit_cpu_task(
@@ -432,7 +432,7 @@ Composite<ProcessingSegment> process_remaining_clauses(
 }
 
 void set_output_descriptors(
-        const Composite<ProcessingSegment>& merged_procs,
+        const Composite<ProcessingUnit>& merged_procs,
         const std::vector<std::shared_ptr<Clause>>& clauses,
         const std::shared_ptr<PipelineContext>& pipeline_context) {
     std::optional<std::string> index_column;
@@ -498,8 +498,9 @@ void set_output_descriptors(
 /*
  * Processes the slices in the given pipeline_context.
  *
- * Slices are processed in order, with slices corresponding to the same row group collected into a single
- * Composite<SliceAndKey>. Slices contained within a single Composite<SliceAndKey> are processed within a single thread.
+ * Slices are processed in an order defined by the first clause in the pipeline, with slices corresponding to the same
+ * processing unit collected into a single Composite<SliceAndKey>. Slices contained within a single
+ * Composite<SliceAndKey> are processed within a single thread.
  *
  * The processing of a Composite<SliceAndKey> is scheduled via the Async Store. Within a single thread, the
  * segments will be retrieved from storage and decompressed before being passed to a MemSegmentProcessingTask which
@@ -512,25 +513,6 @@ std::vector<SliceAndKey> read_and_process(
     const ReadOptions& read_options,
     size_t start_from
     ) {
-    std::sort(std::begin(pipeline_context->slice_and_keys_), std::end(pipeline_context->slice_and_keys_), [] (const SliceAndKey& left, const SliceAndKey& right) {
-        return std::tie(left.slice().row_range.first, left.slice().col_range.first) < std::tie(right.slice().row_range.first, right.slice().col_range.first);
-    });
-
-    std::vector<Composite<SliceAndKey>> rows;
-    auto sk_it = std::begin(pipeline_context->slice_and_keys_);
-    std::advance(sk_it, start_from);
-    while(sk_it != std::end(pipeline_context->slice_and_keys_)) {
-        RowRange row_range{sk_it->slice().row_range};
-        auto sk = Composite{std::move(*sk_it)};
-        // Iterate through all SliceAndKeys that contain data for the same RowRange - i.e., iterate along column segments
-        // for same row group
-        while(++sk_it != std::end(pipeline_context->slice_and_keys_) && sk_it->slice().row_range == row_range) {
-            sk.push_back(std::move(*sk_it));
-        }
-
-        util::check(!sk.empty(), "Should not push empty slice/key pairs to the pipeline");
-        rows.emplace_back(std::move(sk));
-    }
     std::shared_ptr<std::unordered_set<std::string>> filter_columns;
     if(pipeline_context->overall_column_bitset_) {
         filter_columns = std::make_shared<std::unordered_set<std::string>>();
@@ -546,12 +528,16 @@ std::vector<SliceAndKey> read_and_process(
         clause->set_processing_config(processing_config);
     }
 
-    // At this stage, each Composite contains a single ProcessingSegment, which holds a row-slice
-    // Different row-slices are held in different elements of the vector
+    std::vector<Composite<SliceAndKey>> processing_groups = read_query.clauses_[0]->structure_for_processing(
+            pipeline_context->slice_and_keys_, start_from);
+
+    // At this stage, each Composite contains a single ProcessingUnit, which may hold a row-slice, a column-slice, a
+    // general rectangular slice, or some more exotic collection of segments based on the clause's processing
+    // parallelisation.
     // All clauses that do not require repartitioning (e.g. filters and projections) will have already been applied
-    // to these processing segments
-    std::vector<Composite<ProcessingSegment>> procs = store->batch_read_uncompressed(
-            std::move(rows),
+    // to these processing units
+    std::vector<Composite<ProcessingUnit>> procs = store->batch_read_uncompressed(
+            std::move(processing_groups),
             read_query.clauses_,
             filter_columns,
             BatchReadArgs{}
diff --git a/python/arcticdb/version_store/_store.py b/python/arcticdb/version_store/_store.py
index d96c1f8646..c9a071123a 100644
--- a/python/arcticdb/version_store/_store.py
+++ b/python/arcticdb/version_store/_store.py
@@ -38,8 +38,6 @@
 from arcticdb.version_store.read_result import ReadResult
 from arcticdb_ext.version_store import IndexRange as _IndexRange
 from arcticdb_ext.version_store import RowRange as _RowRange
-from arcticdb_ext.version_store import HeadRange as _HeadRange
-from arcticdb_ext.version_store import TailRange as _TailRange
 from arcticdb_ext.version_store import SignedRowRange as _SignedRowRange
 from arcticdb_ext.version_store import PythonVersionStore as _PythonVersionStore
 from arcticdb_ext.version_store import PythonVersionStoreReadQuery as _PythonVersionStoreReadQuery
@@ -1416,9 +1414,6 @@ def _get_read_query(self, date_range: Optional[DateRangeInput], row_range, colum
 
         if date_range is not None:
             read_query.row_filter = _normalize_dt_range(date_range)
-        elif query_builder and query_builder.has_date_range_clause():
-            start, end = query_builder.date_range_boundaries()
-            read_query.row_filter = _IndexRange(start, end)
 
         if row_range is not None:
             read_query.row_range = row_range
@@ -1577,10 +1572,9 @@ def head(
         VersionedItem
         """
 
-        row_range = _HeadRange(n)
         q = QueryBuilder()
         q = q._head(n)
-        version_query, read_options, read_query = self._get_queries(as_of, None, row_range, columns, q, **kwargs)
+        version_query, read_options, read_query = self._get_queries(as_of, None, None, columns, q, **kwargs)
         read_result = self._read_dataframe(symbol, version_query, read_query, read_options)
         return self._post_process_dataframe(read_result, read_query, q)
 
@@ -1607,10 +1601,9 @@ def tail(
         VersionedItem
         """
 
-        row_range = _TailRange(n)
         q = QueryBuilder()
         q = q._tail(n)
-        version_query, read_options, read_query = self._get_queries(as_of, None, row_range, columns, q, **kwargs)
+        version_query, read_options, read_query = self._get_queries(as_of, None, None, columns, q, **kwargs)
         read_result = self._read_dataframe(symbol, version_query, read_query, read_options)
         return self._post_process_dataframe(read_result, read_query, q)
 
diff --git a/python/arcticdb/version_store/processing.py b/python/arcticdb/version_store/processing.py
index 660c78f5a4..c2ebc1c93e 100644
--- a/python/arcticdb/version_store/processing.py
+++ b/python/arcticdb/version_store/processing.py
@@ -626,13 +626,6 @@ def optimise_for_memory(self):
     def needs_post_processing(self):
         return not any(isinstance(clause, (_RowRangeClause, _DateRangeClause)) for clause in self.clauses)
 
-    def has_date_range_clause(self):
-        return len(self.clauses) and isinstance(self.clauses[0], _DateRangeClause)
-
-    def date_range_boundaries(self):
-        check(self.has_date_range_clause(), "QueryBuilder has no date_range clause")
-        return self.clauses[0].start, self.clauses[0].end
-
 
 CONSTRUCTOR_MAP = {
     "u": {1: ValueUint8, 2: ValueUint16, 4: ValueUint32, 8: ValueUint64},


Test patch:
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 4dfddda730..4dc1844570 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -13,7 +13,7 @@
 #include <arcticdb/pipeline/frame_slice.hpp>
 
 template<typename T>
-void segment_scalar_assert_all_values_equal(const arcticdb::ProcessingSegment& segment, const arcticdb::ColumnName& name, const std::unordered_set<T>& expected, size_t expected_row_count) {
+void segment_scalar_assert_all_values_equal(const arcticdb::ProcessingUnit& segment, const arcticdb::ColumnName& name, const std::unordered_set<T>& expected, size_t expected_row_count) {
     const arcticdb::pipelines::SliceAndKey& slice_and_key = segment.data().front();
     std::shared_ptr<arcticdb::Store> empty;
     slice_and_key.ensure_segment(empty);
@@ -31,7 +31,7 @@ void segment_scalar_assert_all_values_equal(const arcticdb::ProcessingSegment& s
     ASSERT_EQ(expected_row_count, row_counter);
 }
 
-void segment_string_assert_all_values_equal(const arcticdb::ProcessingSegment& segment, const arcticdb::ColumnName& name, std::string_view expected, size_t expected_row_count) {
+void segment_string_assert_all_values_equal(const arcticdb::ProcessingUnit& segment, const arcticdb::ColumnName& name, std::string_view expected, size_t expected_row_count) {
     const arcticdb::pipelines::SliceAndKey& slice_and_key = segment.data().front();
     std::shared_ptr<arcticdb::Store> empty;
     slice_and_key.ensure_segment(empty);
@@ -56,10 +56,10 @@ TEST(Clause, Partition) {
 
     ScopedConfig num_buckets("Partition.NumBuckets", 16);
     std::shared_ptr<Store> empty;
-    auto proc_seg = ProcessingSegment{std::move(seg), pipelines::FrameSlice{}};
+    auto proc_seg = ProcessingUnit{std::move(seg), pipelines::FrameSlice{}};
     auto col = std::get<ColumnWithStrings>(proc_seg.get(ColumnName("int8"), empty));
     proc_seg.computed_data_.try_emplace("int8", col);
-    Composite<ProcessingSegment> comp;
+    Composite<ProcessingUnit> comp;
     comp.push_back(std::move(proc_seg));
 
     PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{"int8"};
@@ -80,10 +80,10 @@ TEST(Clause, PartitionString) {
     auto seg = get_groupable_timeseries_segment("groupable", 30, {1,1,3,3,1,1});
     ScopedConfig num_buckets("Partition.NumBuckets", 16);
     std::shared_ptr<Store> empty;
-    auto proc_seg = ProcessingSegment{std::move(seg), pipelines::FrameSlice{}};
+    auto proc_seg = ProcessingUnit{std::move(seg), pipelines::FrameSlice{}};
     auto col = std::get<ColumnWithStrings>(proc_seg.get(ColumnName("strings"), empty));
     proc_seg.computed_data_.try_emplace("strings", col);
-    Composite<ProcessingSegment> comp;
+    Composite<ProcessingUnit> comp;
     comp.push_back(std::move(proc_seg));
 
     PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{"strings"};
@@ -104,8 +104,8 @@ TEST(Clause, Passthrough) {
     auto seg = get_standard_timeseries_segment("passthrough");
 
     PassthroughClause passthrough;
-    auto proc_seg = ProcessingSegment{std::move(seg), pipelines::FrameSlice{}};
-    Composite<ProcessingSegment> comp;
+    auto proc_seg = ProcessingUnit{std::move(seg), pipelines::FrameSlice{}};
+    Composite<ProcessingUnit> comp;
     comp.push_back(std::move(proc_seg));
     std::shared_ptr<Store> empty;
     auto ret = passthrough.process(empty, std::move(comp));
@@ -120,12 +120,12 @@ TEST(Clause, Sort) {
     std::shuffle(seg.begin(), seg.end(), urng);
     std::shared_ptr<Store> empty;
     SortClause sort_clause("time");
-    auto proc_seg = ProcessingSegment{std::move(seg), pipelines::FrameSlice{}};
-    Composite<ProcessingSegment> comp;
+    auto proc_seg = ProcessingUnit{std::move(seg), pipelines::FrameSlice{}};
+    Composite<ProcessingUnit> comp;
     comp.push_back(std::move(proc_seg));
     auto res = sort_clause.process(empty, std::move(comp));
     ASSERT_EQ(res.size(), 1);
-    bool equal = std::get<ProcessingSegment>(res[0]).data_[0].segment(empty) == copied;
+    bool equal = std::get<ProcessingUnit>(res[0]).data_[0].segment(empty) == copied;
     ASSERT_EQ(equal, true);
 }
 
@@ -135,8 +135,8 @@ TEST(Clause, Split) {
     auto seg = get_standard_timeseries_segment(symbol, 100);
     auto copied = seg.clone();
     SplitClause split_clause(10);
-    auto proc_seg = ProcessingSegment{std::move(seg)};
-    Composite<ProcessingSegment> comp;
+    auto proc_seg = ProcessingUnit{std::move(seg)};
+    Composite<ProcessingUnit> comp;
     comp.push_back(std::move(proc_seg));
     std::shared_ptr<Store> empty;
     auto res = split_clause.process(empty, std::move(comp));
@@ -153,7 +153,7 @@ TEST(Clause, Split) {
     SegmentSinkWrapper seg_wrapper(symbol, TimeseriesIndex::default_index(), std::move(desc));
 
     for(auto i = 0u; i < res.level_1_size(); ++i ) {
-        auto item =  std::move(std::get<ProcessingSegment>(res[i]));
+        auto item =  std::move(std::get<ProcessingUnit>(res[i]));
         pipelines::FrameSlice slice(item.data_[0].segment(empty));
         seg_wrapper.aggregator_.add_segment(std::move(item.data_[0].segment(empty)), slice, false);
     }
@@ -169,7 +169,7 @@ TEST(Clause, Merge) {
     using namespace arcticdb;
     const auto seg_size = 5;
     ScopedConfig max_blocks("Merge.SegmentSize", seg_size);
-    Composite<ProcessingSegment> comp;
+    Composite<ProcessingUnit> comp;
     std::vector<SegmentInMemory> copies;
     const auto num_segs = 2;
     const auto num_rows = 5;
@@ -177,7 +177,7 @@ TEST(Clause, Merge) {
         auto symbol = fmt::format("merge_{}", x);
         auto seg = get_standard_timeseries_segment(symbol, 10);
         copies.emplace_back(seg.clone());
-        auto proc_seg = ProcessingSegment{std::move(seg)};
+        auto proc_seg = ProcessingUnit{std::move(seg)};
         comp.push_back(std::move(proc_seg));
     }
 
@@ -189,7 +189,7 @@ TEST(Clause, Merge) {
     auto res = merge_clause.process(empty, std::move(comp));
     ASSERT_EQ(res.size(), 4u);
     for(auto i = 0; i < num_rows * num_segs; ++i) {
-        auto& output_seg = std::get<ProcessingSegment>(res[i / seg_size]).data_[0].segment(empty);
+        auto& output_seg = std::get<ProcessingUnit>(res[i / seg_size]).data_[0].segment(empty);
         auto output_row = i % seg_size;
         const auto& expected_seg = copies[i % num_segs];
         auto expected_row = i / num_segs;
diff --git a/cpp/arcticdb/processing/test/test_expression.cpp b/cpp/arcticdb/processing/test/test_expression.cpp
index ec16035913..c22a03551a 100644
--- a/cpp/arcticdb/processing/test/test_expression.cpp
+++ b/cpp/arcticdb/processing/test/test_expression.cpp
@@ -8,7 +8,7 @@
 #include <gtest/gtest.h>
 #include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/processing_segment.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/util/test/generators.hpp>
 
 TEST(ExpressionNode, AddBasic) {
@@ -28,7 +28,7 @@ TEST(ExpressionNode, AddBasic) {
 
     wrapper.aggregator_.commit();
     auto& seg = wrapper.segment();
-    ProcessingSegment proc(std::move(seg));
+    ProcessingUnit proc(std::move(seg));
     auto node = std::make_shared<ExpressionNode>(ColumnName("thing1"), ColumnName("thing2"), OperationType::ADD);
     auto expression_context = std::make_shared<ExpressionContext>();
     expression_context->root_node_name_ = ExpressionName("new_thing");
diff --git a/cpp/arcticdb/storage/test/in_memory_store.hpp b/cpp/arcticdb/storage/test/in_memory_store.hpp
index c5c31ef3e1..91d90842e8 100644
--- a/cpp/arcticdb/storage/test/in_memory_store.hpp
+++ b/cpp/arcticdb/storage/test/in_memory_store.hpp
@@ -41,7 +41,7 @@ namespace arcticdb {
             return false;
         }
 
-        std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
+        std::vector<Composite<ProcessingUnit>> batch_read_uncompressed(
                 std::vector<Composite<pipelines::SliceAndKey>> &&,
                 const std::vector<std::shared_ptr<Clause>>&,
                 const std::shared_ptr<std::unordered_set<std::string>>&,
diff --git a/python/tests/unit/arcticdb/version_store/test_head.py b/python/tests/unit/arcticdb/version_store/test_head.py
index 26e1138dca..8c03596096 100644
--- a/python/tests/unit/arcticdb/version_store/test_head.py
+++ b/python/tests/unit/arcticdb/version_store/test_head.py
@@ -12,7 +12,6 @@
 import pytest
 
 from arcticdb_ext.exceptions import InternalException
-from arcticdb_ext.version_store import HeadRange as _HeadRange
 
 
 def generic_head_test(version_store, symbol, df, num_rows):
@@ -91,23 +90,6 @@ def test_head_with_column_filter(lmdb_version_store_tiny_segment, three_col_df):
     )
 
 
-# Tests that the frame data returned to the python layer contains at most num_rows + lib segment_row_size
-def test_head_frame_data_size(lmdb_version_store_small_segment):
-    symbol = "test_head_frame_data_size"
-    segment_row_size = lmdb_version_store_small_segment._cfg.write_options.segment_row_size
-    if segment_row_size < 2:
-        raise Exception("test_head_frame_data_size needs segment_row_size to be at least 2")
-    rows_in_table = (2 * segment_row_size) + 1
-    rows_in_head_request = segment_row_size + 1
-    lmdb_version_store_small_segment.write(symbol, DataFrame({"x": np.arange(rows_in_table, dtype=np.int64)}))
-    version_query, read_options, read_query = lmdb_version_store_small_segment._get_queries(
-        None, None, _HeadRange(rows_in_head_request), None, None
-    )
-    read_result = lmdb_version_store_small_segment._read_dataframe(symbol, version_query, read_query, read_options)
-    rows_in_raw_dataframe = read_result.frame_data.value.data[0].size
-    assert rows_in_raw_dataframe <= (1 + rows_in_head_request // segment_row_size) * segment_row_size
-
-
 def test_head_pickled_symbol(lmdb_version_store):
     symbol = "test_head_pickled_symbol"
     lmdb_version_store.write(symbol, np.arange(100).tolist())
diff --git a/python/tests/unit/arcticdb/version_store/test_tail.py b/python/tests/unit/arcticdb/version_store/test_tail.py
index a946eeac26..58d9e47ce1 100644
--- a/python/tests/unit/arcticdb/version_store/test_tail.py
+++ b/python/tests/unit/arcticdb/version_store/test_tail.py
@@ -12,7 +12,6 @@
 import pytest
 
 from arcticdb_ext.exceptions import InternalException
-from arcticdb_ext.version_store import TailRange as _TailRange
 
 
 def generic_tail_test(version_store, symbol, df, num_rows):
@@ -93,23 +92,6 @@ def test_tail_with_column_filter(lmdb_version_store_tiny_segment, three_col_df):
     )
 
 
-# Tests that the frame data returned to the python layer contains at most num_rows + lib segment_row_size
-def test_tail_frame_data_size(lmdb_version_store_small_segment):
-    symbol = "test_tail_frame_data_size"
-    segment_row_size = lmdb_version_store_small_segment._cfg.write_options.segment_row_size
-    if segment_row_size < 2:
-        raise Exception("test_tail_frame_data_size needs segment_row_size to be at least 2")
-    rows_in_table = (2 * segment_row_size) + 1
-    rows_in_tail_request = segment_row_size + 1
-    lmdb_version_store_small_segment.write(symbol, DataFrame({"x": np.arange(rows_in_table, dtype=np.int64)}))
-    version_query, read_options, read_query = lmdb_version_store_small_segment._get_queries(
-        None, None, _TailRange(rows_in_tail_request), None, None
-    )
-    read_result = lmdb_version_store_small_segment._read_dataframe(symbol, version_query, read_query, read_options)
-    rows_in_raw_dataframe = read_result.frame_data.value.data[0].size
-    assert rows_in_raw_dataframe <= (1 + rows_in_tail_request // segment_row_size) * segment_row_size
-
-
 def test_tail_multiple_segments_odd_total_rows(lmdb_version_store_tiny_segment):
     generic_tail_test(
         lmdb_version_store_tiny_segment,