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

Issue:
Able to limit the rows returned after other filters
Currently, the `row_range/head/tail()` filter is applied before any other filter and conflicts with the `date_range` filter. Most other databases support a `top(n)` or `rownum < n` feature and there's [user demand](https://github.com/man-group/ArcticDB/issues/243#issuecomment-1515133551).

We have to be careful to avoid confusion with the existing `row_range/head/tail()`.

Ideally, we add this to the evolved query builder API so the ordering is clear.


Discussion / hints:
Nice, I would love to see that feature also.
This feature would indeed add flexibility and be helpful in a number of use cases. Appreciate you raising this issue. Have a great weekend!
Same! Please add this feature
I don't mean to be annoying, I know you guys are working hard, but are you able to share an update or ETA for this feature?
id like to see this feature added as well ^^ 
This was discussed privately but I'm moving the discussion here for aync public work.

1. We discussed this issue with @jjerphan and we think that it is a bit underspecified, it needs some clarifications like use cases that should work with the change. We both believe having understood the idea (and Alex clarified most of it in a private meeting) but we have doubts about some details. Providing some examples would be helpful to avoid trying wrong solutions again (there is a similar situation in #976) .

2. Julien and I think that to solve this kind of issue a separation of the request representation and the DAG of tasks to execute to treat the request properly is necessary. The DAG would be built by a query-planner/compositor/whatever-you-name-it which reads the request as value (a sequence of clauses , no processing code in clauses) and output the task DAG to execute from it. That way fine-grained ordering of operations is simple and decorelated from how the request looks like to users.
(the recent switch to an ECS-like system to handle the processing seems to confirm this idea as the task DAG would be the Systems in ECS here).

If we are correct, that is if we are right that we understand the feature correctly and iff there is no long-term-viable shortcut to achieve this feature properly without separating clauses and the task dag,
then this seems like a quite important architectural change and personally I dont feel like my understanding of the overall design of arcticdb's internals is enough to do this kind of change at the moment. It would also affect the whole processing system and several tasks listed in [this epic](https://github.com/man-group/ArcticDB/issues/956).

@alexowens90 Could you help with point 1 and maybe confirm if point 2 seems correct to you?

Also thanks to @alexowens90 @jjerphan @Hind-M and @JohanMabille for helping out on the different aspects of this task.
The way it works at the moment is that the Clause classes (which are perhaps misnamed), are a sequence of stateless behaviours that operate independently on an entity, which might be a row-group, a column group, an individual segment or the all of the segments from a particular stage (the latter in operations that for whatever reason cannot be parallelized).

The DAG of processing stages is currently just a list (actually a std::vector), although it's possible it will evolve into something more complex in the future - probably at the point where we implement joins there will be processing stages that can be executed in parallel. Note that this is not the case for expressions (as in the expression within a ProjectClause, which create a new column on-the-fly), these pull the data up from the left and right expression nodes (in the case of a binary operator) and will use all available cores. For clauses we might always keep things linear as we want to avoid the situation where we read in a lot of data that there aren't cores available to do anything with, and it's easier to apply back-pressure in a simpler DAG.

I think that in this specific case it should be possible to implement it without far-reaching changes. The only tricky bit is that if a head or tail follows (for example) a filter, the slice information will be correct about the number of rows a segment contains, but not necessarily about its offset. As a concrete example, if we have two row-groups with one segment each (i.e. no column slicing), each with 100 rows, a filter clause might reduce them to 20 rows each. The filter clause will adjust the row-range in the slice, but to adjust the offset, the thread executing the filter on rows 100-200 would need to know the result of the filter on rows 0-100, creating a data dependency that we want to avoid. This is particularly important because in the case where we have multiple clauses to execute on a segment that are all independent, for example a filter followed by a column projection, we actually only schedule a single task which does both things one after another while (hopefully) much of the data is in cache. The upshot of this is that the row-ranges will be 0-20 and 100-120, instead of 0-20 and 20-40, which is what they really are. Obviously this needs fixing up at the end of the processing pipeline in order for us to produce a coherent result set, but we try to postpone that work as much as possible.

In the case of a head or tail, the clause should indicate to the pipeline that it needs a barrier (i.e. it needs to wait for all preceding computation to finish) by returning true from the method requires_repartition. This will cause the pipeline to do a collect on all the preceding data, and pass the list of Composite<EntityID> to the repartition method on the clause.(N.B. the composite part is going to go away, in favour of a single entity per processing unit with as many components as required).

This repartition method needs to scan through the list of results from the previous computation, either from the beginning in the case of head or the end in the case of tail, accumulating segments and reading the row information from the slices until it has enough rows to satisfy the head or tail parameter. It can then discard the rest, and pass the required segments into the body of the clause where they can be truncated and coalesced to produce the correct output. The code to truncate a segment and also to take a group of small segments and combine them into a larger one is already available, so the problem is mostly one of getting the wiring correct. 

In terms of testing it would make sense to concentrate on head and tail following a filter, as filters both alter the row-count of a segment and preserve  column slicing, so they should expose most if not all of the edge-cases that you're likely to encounter. There is a series of tests written by @alexowens90 that checks that clauses can be combined with each other, once things are looking good it would be an idea to add some basic tests of head and tail following aggregation and column projection as well.

Let me know if there are any questions.
The most simple use case we would like to support:
```
# Use tiny segment so there are 2 rows per segment
def test_head_after_filter(lmdb_version_store_tiny_segment):
    lib = lmdb_version_store_tiny_segment
    df = pd.DataFrame({"col": [0, 1, 0, 1, 0, 1, 0, 1, 0, 1]}, index=pd.date_range("2023-01-01", periods=10))
    lib.write("sym", df)
    q = QueryBuilder()
    q = q[q["col"] == 0]
    q = q.head(2)
    expected = df[df["col"] == 0].head(2)
    received = lib.read("sym", query_builder=q).data
    assert_frame_equal(expected, received)
>>> expected
                    col
2023-01-01    0
2023-01-03    0
```
i.e. perform a filter as normal, and then only keep the first (or last for `tail`) `n` rows of the result.

Another use case we would like to support (possibly in another PR) is to allow `head/tail` after a `date_range` filter. Using the `QueryBuilder`, this would look like:
```
q = q.date_range(pd.Timestamp("2023-01-03"), pd.Timestamp("2023-01-06")).head(1)
```
i.e. retain only the first `n` rows in the specified date range. There are specialised `head` and `tail` methods on the `Library` class, so we could also make these work with `date_range` parameters similar to `read`. When both `date_range` and `query_builder` parameters are provided to `read`, we convert the `date_range` filter into a clause and prepend this to the provided clauses (there is a zero-copy code path when `query_builder` is not provided that we need to keep).

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index c8cf7264c5..c9824e1895 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -258,8 +258,10 @@ set(arcticdb_srcs
         processing/processing_unit.hpp
         processing/bucketizer.hpp
         processing/clause.hpp
+        processing/clause_utils.hpp
         processing/expression_context.hpp
         processing/expression_node.hpp
+        processing/query_planner.hpp
         processing/sorted_aggregation.hpp
         processing/unsorted_aggregation.hpp
         storage/constants.hpp
@@ -434,6 +436,7 @@ set(arcticdb_srcs
         processing/processing_unit.cpp
         processing/aggregation_utils.cpp
         processing/clause.cpp
+        processing/clause_utils.cpp
         processing/component_manager.cpp
         processing/expression_node.cpp
         processing/operation_dispatch.cpp
@@ -449,6 +452,7 @@ set(arcticdb_srcs
         processing/operation_dispatch_binary_operator_minus.cpp
         processing/operation_dispatch_binary_operator_times.cpp
         processing/operation_dispatch_binary_operator_divide.cpp
+        processing/query_planner.cpp
         processing/sorted_aggregation.cpp
         processing/unsorted_aggregation.cpp
         python/python_to_tensor_frame.cpp
diff --git a/cpp/arcticdb/async/tasks.hpp b/cpp/arcticdb/async/tasks.hpp
index d193eb75f3..a15e5098b5 100644
--- a/cpp/arcticdb/async/tasks.hpp
+++ b/cpp/arcticdb/async/tasks.hpp
@@ -484,11 +484,11 @@ struct MemSegmentProcessingTask : BaseTask {
     ARCTICDB_MOVE_ONLY_DEFAULT(MemSegmentProcessingTask)
 
     std::vector<EntityId> operator()() {
-        std::ranges::reverse_view reversed_clauses{clauses_};
-        for (const auto& clause: reversed_clauses) {
-            entity_ids_ = clause->process(std::move(entity_ids_));
+        for (auto it = clauses_.cbegin(); it != clauses_.cend(); ++it) {
+            entity_ids_ = (*it)->process(std::move(entity_ids_));
 
-            if(clause->clause_info().requires_repartition_)
+            auto next_it = std::next(it);
+            if(next_it != clauses_.cend() && (*it)->clause_info().output_structure_ != (*next_it)->clause_info().input_structure_)
                 break;
         }
         return std::move(entity_ids_);
diff --git a/cpp/arcticdb/pipeline/frame_slice.hpp b/cpp/arcticdb/pipeline/frame_slice.hpp
index 35630222ad..51f21c1892 100644
--- a/cpp/arcticdb/pipeline/frame_slice.hpp
+++ b/cpp/arcticdb/pipeline/frame_slice.hpp
@@ -18,22 +18,22 @@ namespace arcticdb {
 
 namespace arcticdb::pipelines {
 
-struct AxisRange : std::pair<std::size_t, std::size_t> {
-    using std::pair<std::size_t, std::size_t>::pair;
+struct AxisRange : std::pair<size_t, size_t> {
+    using std::pair<size_t, size_t>::pair;
 
-    [[nodiscard]] std::size_t diff() const {
+    [[nodiscard]] size_t diff() const {
         return first > second ? 0 : second - first;
     }
 
-    [[nodiscard]] bool contains(std::size_t v) const {
+    [[nodiscard]] bool contains(size_t v) const {
         return first <= v && v < second;
     }
 
-    [[nodiscard]] std::size_t start() const {
+    [[nodiscard]] size_t start() const {
         return first;
     }
 
-    [[nodiscard]] std::size_t end() const {
+    [[nodiscard]] size_t end() const {
         return second;
     }
 
@@ -186,6 +186,23 @@ struct RangesAndKey {
     RangesAndKey() = delete;
     ARCTICDB_MOVE_COPY_DEFAULT(RangesAndKey)
 
+    const RowRange& row_range() const {
+        return row_range_;
+    }
+
+    const ColRange& col_range() const {
+        return col_range_;
+    }
+
+    timestamp start_time() const {
+        return key_.start_time();
+    }
+
+    timestamp end_time() const {
+        // end_index from the key is 1 nanosecond larger than the index value of the last row in the row-slice
+        return key_.end_time() - 1;
+    }
+
     friend bool operator==(const RangesAndKey& left, const RangesAndKey& right) {
         return left.row_range_ == right.row_range_ && left.col_range_ == right.col_range_ && left.key_ == right.key_;
     }
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index 683dd7134a..ceb508af69 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -26,91 +26,6 @@ namespace arcticdb {
 
 using namespace pipelines;
 
-std::vector<std::vector<size_t>> structure_by_row_slice(
-        std::vector<RangesAndKey>& ranges_and_keys,
-        size_t start_from) {
-    std::sort(std::begin(ranges_and_keys), std::end(ranges_and_keys), [] (const RangesAndKey& left, const RangesAndKey& right) {
-        return std::tie(left.row_range_.first, left.col_range_.first) < std::tie(right.row_range_.first, right.col_range_.first);
-    });
-    ranges_and_keys.erase(ranges_and_keys.begin(), ranges_and_keys.begin() + start_from);
-
-    std::vector<std::vector<size_t>> res;
-    RowRange previous_row_range{-1, -1};
-    for (const auto& [idx, ranges_and_key]: folly::enumerate(ranges_and_keys)) {
-        RowRange current_row_range{ranges_and_key.row_range_};
-        if (current_row_range != previous_row_range) {
-            res.emplace_back();
-        }
-        res.back().emplace_back(idx);
-        previous_row_range = current_row_range;
-    }
-    return res;
-}
-
-std::vector<std::vector<size_t>> structure_by_column_slice(std::vector<RangesAndKey>& ranges_and_keys) {
-    std::sort(std::begin(ranges_and_keys), std::end(ranges_and_keys), [] (const RangesAndKey& left, const RangesAndKey& right) {
-        return std::tie(left.col_range_.first, left.row_range_.first) < std::tie(right.col_range_.first, right.row_range_.first);
-    });
-    std::vector<std::vector<size_t>> res;
-    ColRange previous_col_range;
-    for (const auto& [idx, ranges_and_key]: folly::enumerate(ranges_and_keys)) {
-        ColRange current_col_range{ranges_and_key.col_range_};
-        if (current_col_range != previous_col_range) {
-            res.emplace_back();
-        }
-        res.back().emplace_back(idx);
-        previous_col_range = current_col_range;
-    }
-    return res;
-}
-
-std::vector<std::vector<size_t>> structure_all_together(std::vector<RangesAndKey>& ranges_and_keys) {
-    std::vector<size_t> res;
-    for (const auto& [idx, ranges_and_key]: folly::enumerate(ranges_and_keys))
-        res.emplace_back(idx);
-
-    return {std::move(res)};
-}
-
-/*
- * On exit from a clause, we need to push the elements of the newly created processing unit's into the component
- * manager. These will either be used by the next clause in the pipeline, or to present the output dataframe back to
- * the user if this is the final clause in the pipeline.
- */
-std::vector<EntityId> push_entities(ComponentManager& component_manager, ProcessingUnit&& proc, EntityFetchCount entity_fetch_count) {
-    std::vector<EntityFetchCount> entity_fetch_counts(proc.segments_->size(), entity_fetch_count);
-    std::vector<EntityId> ids;
-    if (proc.bucket_.has_value()) {
-        std::vector<bucket_id> bucket_ids(proc.segments_->size(), *proc.bucket_);
-        ids = component_manager.add_entities(
-                std::move(*proc.segments_),
-                std::move(*proc.row_ranges_),
-                std::move(*proc.col_ranges_),
-                std::move(entity_fetch_counts),
-                std::move(bucket_ids));
-    } else {
-        ids = component_manager.add_entities(
-                std::move(*proc.segments_),
-                std::move(*proc.row_ranges_),
-                std::move(*proc.col_ranges_),
-                std::move(entity_fetch_counts));
-    }
-    return ids;
-}
-
-std::vector<EntityId> flatten_entities(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
-    size_t res_size = std::accumulate(entity_ids_vec.cbegin(),
-                                      entity_ids_vec.cend(),
-                                      size_t(0),
-                                      [](size_t acc, const std::vector<EntityId>& vec) { return acc + vec.size(); });
-    std::vector<EntityId> res;
-    res.reserve(res_size);
-    for (const auto& entity_ids: entity_ids_vec) {
-        res.insert(res.end(), entity_ids.begin(), entity_ids.end());
-    }
-    return res;
-}
-
 class GroupingMap {
     using NumericMapType = std::variant<
             std::monostate,
@@ -232,7 +147,9 @@ std::vector<EntityId> ProjectClause::process(std::vector<EntityId>&& entity_ids)
                             const std::string_view name = output_column_;
 
                             proc.segments_->back()->add_column(scalar_field(data_type, name), col.column_);
-                            ++proc.col_ranges_->back()->second;
+                            auto new_col_range = std::make_shared<ColRange>(*proc.col_ranges_->back());
+                            ++new_col_range->second;
+                            proc.col_ranges_->back() = std::move(new_col_range);
                             output = push_entities(*component_manager_, std::move(proc));
                         },
                         [&proc, &output, this](const EmptyResult &) {
@@ -254,6 +171,7 @@ std::vector<EntityId> ProjectClause::process(std::vector<EntityId>&& entity_ids)
 AggregationClause::AggregationClause(const std::string& grouping_column,
                                      const std::vector<NamedAggregator>& named_aggregators):
         grouping_column_(grouping_column) {
+    clause_info_.input_structure_ = ProcessingStructure::HASH_BUCKETED;
     clause_info_.can_combine_with_column_selection_ = false;
     clause_info_.new_index_ = grouping_column_;
     clause_info_.input_columns_ = std::make_optional<std::unordered_set<std::string>>({grouping_column_});
@@ -284,6 +202,35 @@ AggregationClause::AggregationClause(const std::string& grouping_column,
     str_.append("}");
 }
 
+std::vector<std::vector<EntityId>> AggregationClause::structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+    schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
+            std::any_of(entity_ids_vec.cbegin(), entity_ids_vec.cend(), [](const std::vector<EntityId>& entity_ids) {
+                return !entity_ids.empty();
+            }),
+            "Grouping column {} does not exist or is empty", grouping_column_
+    );
+    // Some could be empty, so actual number may be lower
+    auto max_num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets",
+                                                           async::TaskScheduler::instance()->cpu_thread_count());
+    max_num_buckets = std::min(max_num_buckets, static_cast<int64_t>(std::numeric_limits<bucket_id>::max()));
+    // Preallocate results with expected sizes, erase later if any are empty
+    std::vector<std::vector<EntityId>> res(max_num_buckets);
+    // With an even distribution, expect each element of res to have entity_ids_vec.size() elements
+    for (auto& res_element: res) {
+        res_element.reserve(entity_ids_vec.size());
+    }
+    // Experimentation shows flattening the entities into a single vector and a single call to
+    // component_manager_->get is faster than not flattening and making multiple calls
+    auto entity_ids = flatten_entities(std::move(entity_ids_vec));
+    auto [buckets] = component_manager_->get_entities<bucket_id>(entity_ids, false);
+    for (auto [idx, entity_id]: folly::enumerate(entity_ids)) {
+        res[buckets[idx]].emplace_back(entity_id);
+    }
+    // Get rid of any empty buckets
+    std::erase_if(res, [](const std::vector<EntityId>& entity_ids) { return entity_ids.empty(); });
+    return res;
+}
+
 std::vector<EntityId> AggregationClause::process(std::vector<EntityId>&& entity_ids) const {
     if (entity_ids.empty()) {
         return {};
@@ -517,20 +464,19 @@ void ResampleClause<closed_boundary>::set_processing_config(const ProcessingConf
 
 template<ResampleBoundary closed_boundary>
 std::vector<std::vector<size_t>> ResampleClause<closed_boundary>::structure_for_processing(
-        std::vector<RangesAndKey>& ranges_and_keys,
-        ARCTICDB_UNUSED size_t start_from) {
+        std::vector<RangesAndKey>& ranges_and_keys) {
     if (ranges_and_keys.empty()) {
         return {};
     }
     TimestampRange index_range(
             std::min_element(ranges_and_keys.begin(), ranges_and_keys.end(),
                              [](const RangesAndKey& left, const RangesAndKey& right) {
-                                 return left.key_.start_time() < right.key_.start_time();
-                             })->key_.start_time(),
+                                 return left.start_time() < right.start_time();
+                             })->start_time(),
             std::max_element(ranges_and_keys.begin(), ranges_and_keys.end(),
                              [](const RangesAndKey& left, const RangesAndKey& right) {
-                                 return left.key_.end_time() < right.key_.end_time();
-                             })->key_.end_time() - 1
+                                 return left.end_time() < right.end_time();
+                             })->end_time()
     );
     if (date_range_.has_value()) {
         date_range_->first = std::max(date_range_->first, index_range.first);
@@ -545,84 +491,62 @@ std::vector<std::vector<size_t>> ResampleClause<closed_boundary>::structure_for_
     }
     debug::check<ErrorCode::E_ASSERTION_FAILURE>(std::is_sorted(bucket_boundaries_.begin(), bucket_boundaries_.end()),
                                                  "Resampling expects provided bucket boundaries to be strictly monotonically increasing");
-    std::erase_if(ranges_and_keys, [this](const RangesAndKey &ranges_and_key) {
-        auto [start_index, end_index] = ranges_and_key.key_.time_range();
-        // end_index from the key is 1 nanosecond larger than the index value of the last row in the row-slice
-        end_index--;
-        return index_range_outside_bucket_range(start_index, end_index);
-    });
-    auto res = structure_by_row_slice(ranges_and_keys, 0);
-    // Element i of res also needs the values from element i+1 if there is a bucket which incorporates the last index
-    // value of row-slice i and the first value of row-slice i+1
-    // Element i+1 should be removed if the last bucket involved in element i covers all the index values in element i+1
-    auto bucket_boundaries_it = std::cbegin(bucket_boundaries_);
-    // Exit if res_it == std::prev(res.end()) as this implies the last row slice was not incorporated into an earlier processing unit
-    for (auto res_it = res.begin(); res_it != res.end() && res_it != std::prev(res.end());) {
-        auto last_index_value_in_row_slice = ranges_and_keys[res_it->at(0)].key_.end_time() - 1;
-        advance_boundary_past_value(bucket_boundaries_, bucket_boundaries_it, last_index_value_in_row_slice);
-        // bucket_boundaries_it now contains the end value of the last bucket covering the row-slice in res_it, or an end iterator if the last bucket ends before the end of this row-slice
-        if (bucket_boundaries_it != bucket_boundaries_.end()) {
-            Bucket<closed_boundary> current_bucket{*std::prev(bucket_boundaries_it), *bucket_boundaries_it};
-            auto next_row_slice_it = std::next(res_it);
-            while (next_row_slice_it != res.end()) {
-                // end_index from the key is 1 nanosecond larger than the index value of the last row in the row-slice
-                TimestampRange next_row_slice_timestamp_range{
-                        ranges_and_keys[next_row_slice_it->at(0)].key_.start_time(),
-                        ranges_and_keys[next_row_slice_it->at(0)].key_.end_time() - 1};
-                if (current_bucket.contains(next_row_slice_timestamp_range.first)) {
-                    // The last bucket in the current processing unit overlaps with the first index value in the next row slice, so add segments into current processing unit
-                    res_it->insert(res_it->end(), next_row_slice_it->begin(), next_row_slice_it->end());
-                    if (current_bucket.contains(next_row_slice_timestamp_range.second)) {
-                        // The last bucket in the current processing unit wholly contains the next row slice, so remove it from the result
-                        next_row_slice_it = res.erase(next_row_slice_it);
-                    } else {
-                        break;
-                    }
-                } else {
-                    break;
-                }
-            }
-            // This is the last bucket, and all the required row-slices have been incorporated into the current processing unit, so erase the rest
-            if (bucket_boundaries_it == std::prev(bucket_boundaries_.end())) {
-                res.erase(next_row_slice_it, res.end());
-                break;
-            }
-            res_it = next_row_slice_it;
-        }
-    }
-    return res;
+    return structure_by_time_bucket<closed_boundary>(ranges_and_keys, bucket_boundaries_);
 }
 
 template<ResampleBoundary closed_boundary>
-bool ResampleClause<closed_boundary>::index_range_outside_bucket_range(timestamp start_index, timestamp end_index) const {
-    if constexpr (closed_boundary == ResampleBoundary::LEFT) {
-        return start_index >= bucket_boundaries_.back() || end_index < bucket_boundaries_.front();
-    } else {
-        // closed_boundary == ResampleBoundary::RIGHT
-        return start_index > bucket_boundaries_.back() || end_index <= bucket_boundaries_.front();
+std::vector<std::vector<EntityId>> ResampleClause<closed_boundary>::structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+    auto entity_ids = flatten_entities(std::move(entity_ids_vec));
+    if (entity_ids.empty()) {
+        return {};
+    }
+    auto [segments, row_ranges, col_ranges] = component_manager_->get_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(entity_ids, false);
+    std::vector<RangesAndEntity> ranges_and_entities;
+    ranges_and_entities.reserve(entity_ids.size());
+    timestamp min_start_ts{std::numeric_limits<timestamp>::max()};
+    timestamp max_end_ts{std::numeric_limits<timestamp>::min()};
+    for (size_t idx=0; idx<entity_ids.size(); ++idx) {
+        auto start_ts = std::get<timestamp>(stream::TimeseriesIndex::start_value_for_segment(*segments[idx]));
+        auto end_ts = std::get<timestamp>(stream::TimeseriesIndex::end_value_for_segment(*segments[idx]));
+        min_start_ts = std::min(min_start_ts, start_ts);
+        max_end_ts = std::max(max_end_ts, end_ts);
+        ranges_and_entities.emplace_back(entity_ids[idx], row_ranges[idx], col_ranges[idx], std::make_optional<TimestampRange>(start_ts, end_ts));
     }
-}
 
-template<ResampleBoundary closed_boundary>
-void ResampleClause<closed_boundary>::advance_boundary_past_value(const std::vector<timestamp>& bucket_boundaries,
-                                                                  std::vector<timestamp>::const_iterator& bucket_boundaries_it,
-                                                                  timestamp value) const {
-    // These loops are equivalent to bucket_boundaries_it = std::upper_bound(bucket_boundaries_it, bucket_boundaries.end(), value, std::less[_equal]{})
-    // but optimised for the case where most buckets are non-empty.
-    // Mathematically, this will be faster when b / log_2(b) < n, where b is the number of buckets and n is the number of index values
-    // Even if n is only 1000, this corresponds to 7/8 buckets being empty, rising to 19/20 for n=100,000
-    // Experimentally, this implementation is around 10x faster when every bucket contains values, and 3x slower when 99.9% of buckets are empty
-    // If we wanted to speed this up when most buckets are empty, we could make this method adaptive to the number of buckets and rows
-    if constexpr(closed_boundary == ResampleBoundary::LEFT) {
-        while(bucket_boundaries_it != bucket_boundaries.end() && *bucket_boundaries_it <= value) {
-            ++bucket_boundaries_it;
+    date_range_ = std::make_optional<TimestampRange>(min_start_ts, max_end_ts);
+
+    bucket_boundaries_ = generate_bucket_boundaries_(date_range_->first, date_range_->second, rule_, closed_boundary, offset_);
+    if (bucket_boundaries_.size() < 2) {
+        return {};
+    }
+    debug::check<ErrorCode::E_ASSERTION_FAILURE>(std::is_sorted(bucket_boundaries_.begin(), bucket_boundaries_.end()),
+                                                 "Resampling expects provided bucket boundaries to be strictly monotonically increasing");
+
+    auto new_structure_offsets = structure_by_time_bucket<closed_boundary>(ranges_and_entities, bucket_boundaries_);
+
+    std::vector<EntityFetchCount> expected_fetch_counts(ranges_and_entities.size(), 0);
+    for (const auto& list: new_structure_offsets) {
+        for (auto idx: list) {
+            internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+                    idx < expected_fetch_counts.size(),
+                    "Index {} in new_structure_offsets out of bounds >{}", idx, expected_fetch_counts.size() - 1);
+            expected_fetch_counts[idx]++;
         }
-    } else {
-        // closed_boundary == ResampleBoundary::RIGHT
-        while(bucket_boundaries_it != bucket_boundaries.end() && *bucket_boundaries_it < value) {
-            ++bucket_boundaries_it;
+    }
+    internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+            std::all_of(expected_fetch_counts.begin(), expected_fetch_counts.end(), [](EntityFetchCount fetch_count) {
+                return fetch_count == 1 || fetch_count == 2;
+            }),
+            "ResampleClause::structure_for_processing: invalid expected entity fetch count (should be 1 or 2)"
+            );
+    std::vector<EntityId> entities_to_be_fetched_twice;
+    for (auto&& [idx, ranges_and_entity]: folly::enumerate(ranges_and_entities)) {
+        if (expected_fetch_counts[idx] == 2) {
+            entities_to_be_fetched_twice.emplace_back(ranges_and_entity.id_);
         }
     }
+    component_manager_->replace_entities<EntityFetchCount>(entities_to_be_fetched_twice, EntityFetchCount(2));
+    return offsets_to_entity_ids(new_structure_offsets, ranges_and_entities);
 }
 
 template<ResampleBoundary closed_boundary>
@@ -765,7 +689,7 @@ std::shared_ptr<Column> ResampleClause<closed_boundary>::generate_output_index_c
                     current_bucket_added_to_index = true;
                 }
             } else {
-                advance_boundary_past_value(bucket_boundaries, bucket_end_it, *it);
+                advance_boundary_past_value<closed_boundary>(bucket_boundaries, bucket_end_it, *it);
                 if (ARCTICDB_UNLIKELY(bucket_end_it == bucket_boundaries.end())) {
                     break;
                 } else {
@@ -902,7 +826,8 @@ MergeClause::MergeClause(
         stream_id_(stream_id),
         stream_descriptor_(stream_descriptor),
         dynamic_schema_(dynamic_schema) {
-    clause_info_.requires_repartition_ = true;
+    clause_info_.input_structure_ = ProcessingStructure::ALL;
+    clause_info_.output_structure_ = ProcessingStructure::ALL;
 }
 
 void MergeClause::set_processing_config(const ProcessingConfig&) {}
@@ -915,19 +840,7 @@ const ClauseInfo& MergeClause::clause_info() const {
     return clause_info_;
 }
 
-std::vector<std::vector<size_t>> MergeClause::structure_for_processing(
-    std::vector<RangesAndKey>& ranges_and_keys,
-    size_t) const {
-    return structure_all_together(ranges_and_keys);
-}
-
-// MergeClause receives a list of DataFrames as input and merge them into a single one where all 
-// the rows are sorted by time stamp
-std::vector<EntityId> MergeClause::process(std::vector<EntityId>&& entity_ids) const {
-    return std::move(entity_ids);
-}
-
-std::optional<std::vector<std::vector<EntityId>>> MergeClause::repartition(std::vector<std::vector<EntityId>>&& entity_ids_vec) const {
+std::vector<std::vector<EntityId>> MergeClause::structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
 
     // TODO this is a hack because we don't currently have a way to
     // specify any particular input shape unless a clause is the
@@ -1000,6 +913,12 @@ std::optional<std::vector<std::vector<EntityId>>> MergeClause::repartition(std::
     return ret;
 }
 
+// MergeClause receives a list of DataFrames as input and merge them into a single one where all
+// the rows are sorted by time stamp
+std::vector<EntityId> MergeClause::process(std::vector<EntityId>&& entity_ids) const {
+    return std::move(entity_ids);
+}
+
 
 std::vector<EntityId> ColumnStatsGenerationClause::process(std::vector<EntityId>&& entity_ids) const {
     internal::check<ErrorCode::E_INVALID_ARGUMENT>(
@@ -1063,12 +982,57 @@ std::vector<EntityId> ColumnStatsGenerationClause::process(std::vector<EntityId>
 }
 
 std::vector<std::vector<size_t>> RowRangeClause::structure_for_processing(
-        std::vector<RangesAndKey>& ranges_and_keys,
-        ARCTICDB_UNUSED size_t start_from) {
+        std::vector<RangesAndKey>& ranges_and_keys) {
     ranges_and_keys.erase(std::remove_if(ranges_and_keys.begin(), ranges_and_keys.end(), [this](const RangesAndKey& ranges_and_key) {
         return ranges_and_key.row_range_.start() >= end_ || ranges_and_key.row_range_.end() <= start_;
     }), ranges_and_keys.end());
-    return structure_by_row_slice(ranges_and_keys, start_from);
+    return structure_by_row_slice(ranges_and_keys);
+}
+
+std::vector<std::vector<EntityId>> RowRangeClause::structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+    auto entity_ids = flatten_entities(std::move(entity_ids_vec));
+    if (entity_ids.empty()) {
+        return {};
+    }
+    auto [segments, old_row_ranges, col_ranges] = component_manager_->get_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(entity_ids, false);
+
+    // Map from old row ranges to new ones
+    std::map<RowRange, RowRange> row_range_mapping;
+    for (const auto& row_range: old_row_ranges) {
+        // Value is same as key initially
+        row_range_mapping.insert({*row_range, *row_range});
+    }
+    bool first_range{true};
+    size_t prev_range_end{0};
+    for (auto& [old_range, new_range]: row_range_mapping) {
+        if (first_range) {
+            // Make the first row-range start from zero
+            new_range.first = 0;
+            new_range.second = old_range.diff();
+            first_range = false;
+        } else {
+            new_range.first = prev_range_end;
+            new_range.second = new_range.first + old_range.diff();
+        }
+        prev_range_end = new_range.second;
+    }
+
+    calculate_start_and_end(prev_range_end);
+
+    std::vector<std::shared_ptr<RowRange>> new_row_ranges;
+    new_row_ranges.reserve(old_row_ranges.size());
+    std::vector<RangesAndEntity> ranges_and_entities;
+    ranges_and_entities.reserve(entity_ids.size());
+    for (size_t idx=0; idx<entity_ids.size(); ++idx) {
+        auto new_row_range = std::make_shared<RowRange>(row_range_mapping.at(*old_row_ranges[idx]));
+        ranges_and_entities.emplace_back(entity_ids[idx], new_row_range, col_ranges[idx]);
+        new_row_ranges.emplace_back(std::move(new_row_range));
+    }
+
+    component_manager_->replace_entities<std::shared_ptr<RowRange>>(entity_ids, new_row_ranges);
+
+    auto new_structure_offsets = structure_by_row_slice(ranges_and_entities);
+    return offsets_to_entity_ids(new_structure_offsets, ranges_and_entities);
 }
 
 std::vector<EntityId> RowRangeClause::process(std::vector<EntityId>&& entity_ids) const {
@@ -1077,61 +1041,71 @@ std::vector<EntityId> RowRangeClause::process(std::vector<EntityId>&& entity_ids
     }
     auto proc = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(*component_manager_, std::move(entity_ids));
     std::vector<EntityId> output;
-    for (auto&& [idx, row_range]: folly::enumerate(proc.row_ranges_.value())) {
-        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 truncated_segment = proc.segments_->at(idx)->truncate(start_row, end_row, false);
-            auto num_rows = truncated_segment.is_null() ? 0 : truncated_segment.row_count();
-            proc.row_ranges_->at(idx) = std::make_shared<pipelines::RowRange>(proc.row_ranges_->at(idx)->first, proc.row_ranges_->at(idx)->first + num_rows);
-            auto num_cols = truncated_segment.is_null() ? 0 : truncated_segment.descriptor().field_count() - truncated_segment.descriptor().index().field_count();
-            proc.col_ranges_->at(idx) = std::make_shared<pipelines::ColRange>(proc.col_ranges_->at(idx)->first, proc.col_ranges_->at(idx)->first + num_cols);
-            proc.segments_->at(idx) = std::make_shared<SegmentInMemory>(std::move(truncated_segment));
-        } // else all rows in this segment are required, do nothing
-    }
+    // The processing unit represents one row slice by construction, so just look at the first
+    auto row_range = proc.row_ranges_->at(0);
+    if (start_ >= row_range->end() || end_ <= row_range->start()) {
+        // No rows from this processing unit are required
+        return {};
+    } else 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());
+        }
+        proc.truncate(start_row, end_row);
+    } // else all rows in this segment are required, do nothing
     return push_entities(*component_manager_, std::move(proc));
 }
 
 void RowRangeClause::set_processing_config(const ProcessingConfig& processing_config) {
-    auto total_rows = static_cast<int64_t>(processing_config.total_rows_);
+    calculate_start_and_end(processing_config.total_rows_);
+}
+
+std::string RowRangeClause::to_string() const {
+    if (row_range_type_ == RowRangeType::RANGE) {
+        return fmt::format("ROWRANGE: RANGE, start={}, end ={}", start_, end_);
+    }
+
+    return fmt::format("ROWRANGE: {}, n={}", row_range_type_ == RowRangeType::HEAD ? "HEAD" : "TAIL", n_);
+}
+
+void RowRangeClause::calculate_start_and_end(size_t total_rows) {
+    auto signed_total_rows = static_cast<int64_t>(total_rows);
     switch(row_range_type_) {
         case RowRangeType::HEAD:
             if (n_ >= 0) {
                 start_ = 0;
-                end_ = std::min(n_, total_rows);
+                end_ = std::min(n_, signed_total_rows);
             } else {
                 start_ = 0;
-                end_ = std::max(static_cast<int64_t>(0), total_rows + n_);
+                end_ = std::max(static_cast<int64_t>(0), signed_total_rows + n_);
             }
             break;
         case RowRangeType::TAIL:
             if (n_ >= 0) {
-                start_ = std::max(static_cast<int64_t>(0), total_rows - n_);
-                end_ = total_rows;
+                start_ = std::max(static_cast<int64_t>(0), signed_total_rows - n_);
+                end_ = signed_total_rows;
             } else {
-                start_ = std::min(-n_, total_rows);
-                end_ = total_rows;
+                start_ = std::min(-n_, signed_total_rows);
+                end_ = signed_total_rows;
             }
             break;
         case RowRangeType::RANGE:
             // Wrap around negative indices.
             start_ = (
-                user_provided_start_ >= 0 ?
-                std::min(user_provided_start_, total_rows) :
-                std::max(total_rows + user_provided_start_, static_cast<int64_t>(0))
+                    user_provided_start_ >= 0 ?
+                    std::min(user_provided_start_, signed_total_rows) :
+                    std::max(signed_total_rows + user_provided_start_, static_cast<int64_t>(0))
             );
             end_ = (
-                user_provided_end_ >= 0 ?
-                std::min(user_provided_end_, total_rows) :
-                std::max(total_rows + user_provided_end_, static_cast<int64_t>(0))
+                    user_provided_end_ >= 0 ?
+                    std::min(user_provided_end_, signed_total_rows) :
+                    std::max(signed_total_rows + user_provided_end_, static_cast<int64_t>(0))
             );
             break;
 
@@ -1140,40 +1114,35 @@ void RowRangeClause::set_processing_config(const ProcessingConfig& processing_co
     }
 }
 
-std::string RowRangeClause::to_string() const {
-    if (row_range_type_ == RowRangeType::RANGE) {
-        return fmt::format("ROWRANGE: RANGE, start={}, end ={}", start_, end_);
-    }
-
-    return fmt::format("ROWRANGE: {}, n={}", row_range_type_ == RowRangeType::HEAD ? "HEAD" : "TAIL", n_);
-}
-
 std::vector<std::vector<size_t>> DateRangeClause::structure_for_processing(
-        std::vector<RangesAndKey>& ranges_and_keys,
-        size_t start_from) {
+        std::vector<RangesAndKey>& ranges_and_keys) {
     ranges_and_keys.erase(std::remove_if(ranges_and_keys.begin(), ranges_and_keys.end(), [this](const RangesAndKey& ranges_and_key) {
         auto [start_index, end_index] = ranges_and_key.key_.time_range();
         return start_index > end_ || end_index <= start_;
     }), ranges_and_keys.end());
-    return structure_by_row_slice(ranges_and_keys, start_from);
+    return structure_by_row_slice(ranges_and_keys);
 }
 
 std::vector<EntityId> DateRangeClause::process(std::vector<EntityId> &&entity_ids) const {
     if (entity_ids.empty()) {
         return {};
     }
-    auto proc = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>, std::shared_ptr<AtomKey>>(*component_manager_, std::move(entity_ids));
+    auto proc = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(*component_manager_, std::move(entity_ids));
     std::vector<EntityId> output;
     // We are only interested in the index, which is in every SegmentInMemory in proc.segments_, so just use the first
     auto row_range = proc.row_ranges_->at(0);
-    auto [start_index, end_index] = proc.atom_keys_->at(0)->time_range();
-    if ((start_ > start_index && start_ < end_index) || (end_ >= start_index && end_ < end_index)) {
+    auto start_index = std::get<timestamp>(stream::TimeseriesIndex::start_value_for_segment(*proc.segments_->at(0)));
+    auto end_index = std::get<timestamp>(stream::TimeseriesIndex::end_value_for_segment(*proc.segments_->at(0)));
+    if (start_index > end_ || end_index < start_) {
+        // No rows from this processing unit are required
+        return {};
+    } else if ((start_ > start_index && start_ <= end_index) || (end_ >= start_index && end_ <= end_index)) {
         size_t start_row{0};
         size_t end_row{row_range->diff()};
-        if (start_ > start_index && start_ < end_index) {
+        if (start_ > start_index && start_ <= end_index) {
             start_row = proc.segments_->at(0)->column_ptr(0)->search_sorted<timestamp>(start_);
         }
-        if (end_ >= start_index && end_ < end_index) {
+        if (end_ >= start_index && end_ <= end_index) {
             end_row = proc.segments_->at(0)->column_ptr(0)->search_sorted<timestamp>(end_, true);
         }
         proc.truncate(start_row, end_row);
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index e5ae002eda..ff53585c61 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -14,6 +14,7 @@
 #include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/entity/types.hpp>
+#include <arcticdb/processing/clause_utils.hpp>
 #include <arcticdb/processing/unsorted_aggregation.hpp>
 #include <arcticdb/processing/aggregation_interface.hpp>
 #include <arcticdb/processing/processing_unit.hpp>
@@ -38,50 +39,25 @@ namespace arcticdb {
 using RangesAndKey = pipelines::RangesAndKey;
 using SliceAndKey = pipelines::SliceAndKey;
 
-using NormMetaDescriptor = std::shared_ptr<arcticdb::proto::descriptors::NormalizationMetadata>;
-
-// Contains constant data about the clause identifiable at construction time
-struct ClauseInfo {
-    // Whether processing segments need to be split into new processing segments after this clause's process method has finished
-    bool requires_repartition_{false};
-    // Whether it makes sense to combine this clause with specifying which columns to view in a call to read or similar
-    bool can_combine_with_column_selection_{true};
-    // The names of the columns that are needed for this clause to make sense
-    // Could either be on disk, or columns created by earlier clauses in the processing pipeline
-    std::optional<std::unordered_set<std::string>> input_columns_{std::nullopt};
-    // The name of the index after this clause if it has been modified, std::nullopt otherwise
-    std::optional<std::string> new_index_{std::nullopt};
-    // Whether this clause modifies the output descriptor
-    bool modifies_output_descriptor_{false};
-};
-
-// Changes how the clause behaves based on information only available after it is constructed
-struct ProcessingConfig {
-    bool dynamic_schema_{false};
-    uint64_t total_rows_ = 0;
-};
-
 struct IClause {
     template<class Base>
     struct Interface : Base {
         // Reorders ranges_and_keys into the order they should be queued up to be read from storage.
         // Returns a vector where each element is a vector of indexes into ranges_and_keys representing the segments needed
         // for one ProcessingUnit.
-        // TODO #732: Factor out start_from as part of https://github.com/man-group/ArcticDB/issues/732
         [[nodiscard]] std::vector<std::vector<size_t>>
-        structure_for_processing(std::vector<RangesAndKey>& ranges_and_keys,
-                                 size_t start_from) {
-            return folly::poly_call<0>(*this, ranges_and_keys, start_from);
+        structure_for_processing(std::vector<RangesAndKey>& ranges_and_keys) {
+            return std::move(folly::poly_call<0>(*this, ranges_and_keys));
         }
 
-        [[nodiscard]] std::vector<EntityId>
-        process(std::vector<EntityId>&& entity_ids) const {
-            return folly::poly_call<1>(*this, std::move(entity_ids));
+        [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(
+                std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+            return folly::poly_call<1>(*this, std::move(entity_ids_vec));
         }
 
-        [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(
-                std::vector<std::vector<EntityId>>&& entity_ids_vec) const {
-            return folly::poly_call<2>(*this, std::move(entity_ids_vec));
+        [[nodiscard]] std::vector<EntityId>
+        process(std::vector<EntityId>&& entity_ids) const {
+            return std::move(folly::poly_call<2>(*this, std::move(entity_ids)));
         }
 
         [[nodiscard]] const ClauseInfo& clause_info() const { return folly::poly_call<3>(*this); };
@@ -97,9 +73,9 @@ struct IClause {
 
     template<class T>
     using Members = folly::PolyMembers<
-            &T::structure_for_processing,
+            folly::sig<std::vector<std::vector<size_t>>(std::vector<RangesAndKey>&)>(&T::structure_for_processing),
+            folly::sig<std::vector<std::vector<EntityId>>(std::vector<std::vector<EntityId>>&&)>(&T::structure_for_processing),
             &T::process,
-            &T::repartition,
             &T::clause_info,
             &T::set_processing_config,
             &T::set_component_manager>;
@@ -107,61 +83,23 @@ struct IClause {
 
 using Clause = folly::Poly<IClause>;
 
-std::vector<std::vector<size_t>> structure_by_row_slice(std::vector<RangesAndKey>& ranges_and_keys, size_t start_from);
-
-std::vector<std::vector<size_t>> structure_by_column_slice(std::vector<RangesAndKey>& ranges_and_keys);
-
-std::vector<std::vector<size_t>> structure_all_together(std::vector<RangesAndKey>& ranges_and_keys);
-
-/*
- * On entry to a clause, construct ProcessingUnits from the input entity IDs. These will either be provided by the
- * structure_for_processing method for the first clause in the pipeline, or by the previous clause for all subsequent
- * clauses.
- */
-template<class... Args>
-ProcessingUnit gather_entities(ComponentManager& component_manager, std::vector<EntityId>&& entity_ids) {
-    ProcessingUnit res;
-    auto components = component_manager.get_entities<Args...>(entity_ids);
-    ([&]{
-        auto component = std::move(std::get<std::vector<Args>>(components));
-        if constexpr (std::is_same_v<Args, std::shared_ptr<SegmentInMemory>>) {
-            res.set_segments(std::move(component));
-        } else if constexpr (std::is_same_v<Args, std::shared_ptr<RowRange>>) {
-            res.set_row_ranges(std::move(component));
-        } else if constexpr (std::is_same_v<Args, std::shared_ptr<ColRange>>) {
-            res.set_col_ranges(std::move(component));
-        } else if constexpr (std::is_same_v<Args, std::shared_ptr<AtomKey>>) {
-            res.set_atom_keys(std::move(component));
-        } else if constexpr (std::is_same_v<Args, EntityFetchCount>) {
-            res.set_entity_fetch_count(std::move(component));
-        } else {
-            static_assert(sizeof(Args) == 0, "Unexpected component type provided in gather_entities");
-        }
-    }(), ...);
-    return res;
-}
-
-std::vector<EntityId> push_entities(ComponentManager& component_manager, ProcessingUnit&& proc, EntityFetchCount entity_fetch_count=1);
-
-std::vector<EntityId> flatten_entities(std::vector<std::vector<EntityId>>&& entity_ids_vec);
-
 struct PassthroughClause {
     ClauseInfo clause_info_;
+
     PassthroughClause() = default;
     ARCTICDB_MOVE_COPY_DEFAULT(PassthroughClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys); // TODO: No structuring?
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return entity_ids_vec; // TODO: structure by row slice?
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -190,17 +128,16 @@ struct FilterClause {
     ARCTICDB_MOVE_COPY_DEFAULT(FilterClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>> &&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -240,17 +177,16 @@ struct ProjectClause {
     ARCTICDB_MOVE_COPY_DEFAULT(ProjectClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -277,7 +213,6 @@ struct PartitionClause {
             processing_config_(),
             grouping_column_(grouping_column) {
         clause_info_.input_columns_ = {grouping_column_};
-        clause_info_.requires_repartition_ = true;
         clause_info_.modifies_output_descriptor_ = true;
     }
     PartitionClause() = delete;
@@ -285,9 +220,12 @@ struct PartitionClause {
     ARCTICDB_MOVE_COPY_DEFAULT(PartitionClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys);
+    }
+
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
     [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const {
@@ -301,42 +239,12 @@ struct PartitionClause {
                 processing_config_.dynamic_schema_);
         std::vector<EntityId> output;
         for (auto &&partitioned_proc: partitioned_procs) {
-            // EntityFetchCount is 2, once for the repartition, and once for AggregationClause::process
-            std::vector<EntityId> proc_entity_ids = push_entities(*component_manager_, std::move(partitioned_proc), 2);
+            std::vector<EntityId> proc_entity_ids = push_entities(*component_manager_, std::move(partitioned_proc));
             output.insert(output.end(), proc_entity_ids.begin(), proc_entity_ids.end());
         }
         return output;
     }
 
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(std::vector<std::vector<EntityId>>&& entity_ids_vec) const {
-        schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                std::any_of(entity_ids_vec.cbegin(), entity_ids_vec.cend(), [](const std::vector<EntityId>& entity_ids) {
-                    return !entity_ids.empty();
-                }),
-                "Grouping column {} does not exist or is empty", grouping_column_
-        );
-        // Some could be empty, so actual number may be lower
-        auto max_num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets",
-                                                           async::TaskScheduler::instance()->cpu_thread_count());
-        max_num_buckets = std::min(max_num_buckets, static_cast<int64_t>(std::numeric_limits<bucket_id>::max()));
-        // Preallocate results with expected sizes, erase later if any are empty
-        std::vector<std::vector<EntityId>> res(max_num_buckets);
-        // With an even distribution, expect each element of res to have entity_ids_vec.size() elements
-        for (auto& res_element: res) {
-            res_element.reserve(entity_ids_vec.size());
-        }
-        // Experimentation shows flattening the entities into a single vector and a single call to
-        // component_manager_->get is faster than not flattening and making multiple calls
-        auto entity_ids = flatten_entities(std::move(entity_ids_vec));
-        auto [buckets] = component_manager_->get_entities<bucket_id>(entity_ids);
-        for (auto [idx, entity_id]: folly::enumerate(entity_ids)) {
-            res[buckets[idx]].emplace_back(entity_id);
-        }
-        // Get rid of any empty buckets
-        std::erase_if(res, [](const std::vector<EntityId>& entity_ids) { return entity_ids.empty(); });
-        return res;
-    }
-
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -384,19 +292,13 @@ struct AggregationClause {
     AggregationClause(const std::string& grouping_column,
                       const std::vector<NamedAggregator>& aggregations);
 
-    [[noreturn]] std::vector<std::vector<size_t>> structure_for_processing(
-            ARCTICDB_UNUSED const std::vector<RangesAndKey>&,
-            ARCTICDB_UNUSED size_t) {
-        internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
-                "AggregationClause::structure_for_processing should never be called"
-                );
+    [[noreturn]] std::vector<std::vector<size_t>> structure_for_processing(ARCTICDB_UNUSED std::vector<RangesAndKey>&) {
+        internal::raise<ErrorCode::E_ASSERTION_FAILURE>("AggregationClause should never be first in the pipeline");
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec);
 
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
-    }
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
 
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
@@ -441,19 +343,17 @@ struct ResampleClause {
             label_boundary_(label_boundary),
             generate_bucket_boundaries_(std::move(generate_bucket_boundaries)),
             offset_(offset) {
+        clause_info_.input_structure_ = ProcessingStructure::TIME_BUCKETED;
         clause_info_.can_combine_with_column_selection_ = false;
         clause_info_.modifies_output_descriptor_ = true;
     }
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from);
+            std::vector<RangesAndKey>& ranges_and_keys);
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec);
 
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
-    }
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
 
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
@@ -477,13 +377,6 @@ struct ResampleClause {
         date_range_.emplace(date_range_start, date_range_end);
     }
 
-    bool index_range_outside_bucket_range(timestamp start_index, timestamp end_index) const;
-
-    // Advances the bucket boundary iterator to the end of the last bucket that includes a value from a row slice with the given last index value
-    void advance_boundary_past_value(const std::vector<timestamp>& bucket_boundaries,
-                                     std::vector<timestamp>::const_iterator& bucket_boundaries_it,
-                                     timestamp value) const;
-
     std::vector<timestamp> generate_bucket_boundaries(timestamp first_ts,
                                                       timestamp last_ts,
                                                       bool responsible_for_first_overlapping_bucket) const;
@@ -492,28 +385,36 @@ struct ResampleClause {
                                                          const std::vector<timestamp>& bucket_boundaries) const;
 };
 
+template<typename T>
+struct is_resample: std::false_type{};
+
+template<ResampleBoundary closed_boundary>
+struct is_resample<ResampleClause<closed_boundary>>: std::true_type{};
+
 struct RemoveColumnPartitioningClause {
     ClauseInfo clause_info_;
     std::shared_ptr<ComponentManager> component_manager_;
     mutable bool warning_shown = false; // folly::Poly can't deal with atomic_bool
+    size_t incompletes_after_;
 
-    RemoveColumnPartitioningClause() {
+    RemoveColumnPartitioningClause(size_t incompletes_after=0):
+            incompletes_after_(incompletes_after){
         clause_info_.can_combine_with_column_selection_ = false;
     }
     ARCTICDB_MOVE_COPY_DEFAULT(RemoveColumnPartitioningClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        ranges_and_keys.erase(ranges_and_keys.begin(), ranges_and_keys.begin() + incompletes_after_);
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -536,17 +437,16 @@ struct SplitClause {
     }
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -562,23 +462,25 @@ struct SortClause {
     ClauseInfo clause_info_;
     std::shared_ptr<ComponentManager> component_manager_;
     const std::string column_;
+    size_t incompletes_after_;
 
-    explicit SortClause(std::string column) :
-            column_(std::move(column)) {
+    explicit SortClause(std::string column, size_t incompletes_after) :
+            column_(std::move(column)),
+            incompletes_after_(incompletes_after){
     }
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        ranges_and_keys.erase(ranges_and_keys.begin(), ranges_and_keys.begin() + incompletes_after_);
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -609,13 +511,13 @@ struct MergeClause {
         bool dynamic_schema
     );
 
-    [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t) const;
+    [[noreturn]] std::vector<std::vector<size_t>> structure_for_processing(std::vector<RangesAndKey>&) {
+        internal::raise<ErrorCode::E_ASSERTION_FAILURE>("MergeClause should never be first in the pipeline");
+    }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec);
 
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(std::vector<std::vector<EntityId>>&& entity_ids_vec) const;
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
 
     [[nodiscard]] const ClauseInfo& clause_info() const;
 
@@ -643,17 +545,16 @@ struct ColumnStatsGenerationClause {
     ARCTICDB_MOVE_COPY_DEFAULT(ColumnStatsGenerationClause)
 
     [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from) {
-        return structure_by_row_slice(ranges_and_keys, start_from);
+            std::vector<RangesAndKey>& ranges_and_keys) {
+        return structure_by_row_slice(ranges_and_keys);
     }
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
@@ -697,27 +598,25 @@ struct RowRangeClause {
     explicit RowRangeClause(RowRangeType row_range_type, int64_t n):
             row_range_type_(row_range_type),
             n_(n) {
+        clause_info_.input_structure_ = ProcessingStructure::ALL;
     }
 
     explicit RowRangeClause(int64_t start, int64_t end):
             row_range_type_(RowRangeType::RANGE),
             user_provided_start_(start),
             user_provided_end_(end) {
+        clause_info_.input_structure_ = ProcessingStructure::ALL;
     }
 
     RowRangeClause() = delete;
 
     ARCTICDB_MOVE_COPY_DEFAULT(RowRangeClause)
 
-    [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            ARCTICDB_UNUSED size_t start_from);
+    [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(std::vector<RangesAndKey>& ranges_and_keys);
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec);
 
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
-    }
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
 
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
@@ -730,6 +629,8 @@ struct RowRangeClause {
     }
 
     [[nodiscard]] std::string to_string() const;
+
+    void calculate_start_and_end(size_t total_rows);
 };
 
 struct DateRangeClause {
@@ -749,16 +650,14 @@ struct DateRangeClause {
 
     ARCTICDB_MOVE_COPY_DEFAULT(DateRangeClause)
 
-    [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(
-            std::vector<RangesAndKey>& ranges_and_keys,
-            size_t start_from);
+    [[nodiscard]] std::vector<std::vector<size_t>> structure_for_processing(std::vector<RangesAndKey>& ranges_and_keys);
 
-    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
-
-    [[nodiscard]] std::optional<std::vector<std::vector<EntityId>>> repartition(ARCTICDB_UNUSED std::vector<std::vector<EntityId>>&&) const {
-        return std::nullopt;
+    [[nodiscard]] std::vector<std::vector<EntityId>> structure_for_processing(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+        return structure_by_row_slice(*component_manager_, std::move(entity_ids_vec));
     }
 
+    [[nodiscard]] std::vector<EntityId> process(std::vector<EntityId>&& entity_ids) const;
+
     [[nodiscard]] const ClauseInfo& clause_info() const {
         return clause_info_;
     }
diff --git a/cpp/arcticdb/processing/clause_utils.cpp b/cpp/arcticdb/processing/clause_utils.cpp
new file mode 100644
index 0000000000..854ad5e39b
--- /dev/null
+++ b/cpp/arcticdb/processing/clause_utils.cpp
@@ -0,0 +1,84 @@
+/* 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.
+ */
+
+#include <arcticdb/processing/clause_utils.hpp>
+
+namespace arcticdb {
+
+using namespace pipelines;
+
+std::vector<std::vector<EntityId>> structure_by_row_slice(ComponentManager& component_manager, std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+    auto entity_ids = flatten_entities(std::move(entity_ids_vec));
+    auto [row_ranges, col_ranges] = component_manager.get_entities<std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(entity_ids, false);
+    std::vector<RangesAndEntity> ranges_and_entities;
+    ranges_and_entities.reserve(entity_ids.size());
+    for (size_t idx=0; idx<entity_ids.size(); ++idx) {
+        ranges_and_entities.emplace_back(entity_ids[idx], row_ranges[idx], col_ranges[idx]);
+    }
+    auto new_structure_indices = structure_by_row_slice(ranges_and_entities);
+    std::vector<std::vector<EntityId>> res(new_structure_indices.size());
+    for (const auto&& [outer_idx, vec]: folly::enumerate(new_structure_indices)) {
+        res[outer_idx].reserve(vec.size());
+        for (auto inner_idx: vec) {
+            res[outer_idx].emplace_back(ranges_and_entities[inner_idx].id_);
+        }
+    }
+    return res;
+}
+
+std::vector<std::vector<EntityId>> offsets_to_entity_ids(const std::vector<std::vector<size_t>>& offsets,
+                                                         const std::vector<RangesAndEntity>& ranges_and_entities) {
+    std::vector<std::vector<EntityId>> res(offsets.size());
+    for (const auto&& [outer_idx, vec]: folly::enumerate(offsets)) {
+        res[outer_idx].reserve(vec.size());
+        for (auto inner_idx: vec) {
+            res[outer_idx].emplace_back(ranges_and_entities[inner_idx].id_);
+        }
+    }
+    return res;
+}
+
+/*
+ * On exit from a clause, we need to push the elements of the newly created processing unit's into the component
+ * manager. These will either be used by the next clause in the pipeline, or to present the output dataframe back to
+ * the user if this is the final clause in the pipeline.
+ */
+std::vector<EntityId> push_entities(ComponentManager& component_manager, ProcessingUnit&& proc, EntityFetchCount entity_fetch_count) {
+    std::vector<EntityFetchCount> entity_fetch_counts(proc.segments_->size(), entity_fetch_count);
+    std::vector<EntityId> ids;
+    if (proc.bucket_.has_value()) {
+        std::vector<bucket_id> bucket_ids(proc.segments_->size(), *proc.bucket_);
+        ids = component_manager.add_entities(
+                std::move(*proc.segments_),
+                std::move(*proc.row_ranges_),
+                std::move(*proc.col_ranges_),
+                std::move(entity_fetch_counts),
+                std::move(bucket_ids));
+    } else {
+        ids = component_manager.add_entities(
+                std::move(*proc.segments_),
+                std::move(*proc.row_ranges_),
+                std::move(*proc.col_ranges_),
+                std::move(entity_fetch_counts));
+    }
+    return ids;
+}
+
+std::vector<EntityId> flatten_entities(std::vector<std::vector<EntityId>>&& entity_ids_vec) {
+    size_t res_size = std::accumulate(entity_ids_vec.cbegin(),
+                                      entity_ids_vec.cend(),
+                                      size_t(0),
+                                      [](size_t acc, const std::vector<EntityId>& vec) { return acc + vec.size(); });
+    std::vector<EntityId> res;
+    res.reserve(res_size);
+    for (const auto& entity_ids: entity_ids_vec) {
+        res.insert(res.end(), entity_ids.begin(), entity_ids.end());
+    }
+    return res;
+}
+
+}
diff --git a/cpp/arcticdb/processing/clause_utils.hpp b/cpp/arcticdb/processing/clause_utils.hpp
new file mode 100644
index 0000000000..ebac18ae3c
--- /dev/null
+++ b/cpp/arcticdb/processing/clause_utils.hpp
@@ -0,0 +1,239 @@
+/* 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 <optional>
+#include <string>
+#include <unordered_set>
+
+#include <arcticdb/pipeline/frame_slice.hpp>
+#include <arcticdb/processing/component_manager.hpp>
+#include <arcticdb/processing/processing_unit.hpp>
+#include <arcticdb/processing/sorted_aggregation.hpp>
+
+namespace arcticdb {
+
+using RangesAndKey = pipelines::RangesAndKey;
+using SliceAndKey = pipelines::SliceAndKey;
+
+enum class ProcessingStructure {
+    ROW_SLICE,
+    TIME_BUCKETED,
+    HASH_BUCKETED,
+    ALL
+};
+
+// Contains constant data about the clause identifiable at construction time
+struct ClauseInfo {
+    // The arrangement of segments this clause needs in order for processing to be done correctly
+    ProcessingStructure input_structure_{ProcessingStructure::ROW_SLICE};
+    // The arrangement of segment this clause produces
+    ProcessingStructure output_structure_{ProcessingStructure::ROW_SLICE};
+    // Whether it makes sense to combine this clause with specifying which columns to view in a call to read or similar
+    bool can_combine_with_column_selection_{true};
+    // The names of the columns that are needed for this clause to make sense
+    // Could either be on disk, or columns created by earlier clauses in the processing pipeline
+    std::optional<std::unordered_set<std::string>> input_columns_{std::nullopt};
+    // The name of the index after this clause if it has been modified, std::nullopt otherwise
+    std::optional<std::string> new_index_{std::nullopt};
+    // Whether this clause modifies the output descriptor
+    bool modifies_output_descriptor_{false};
+};
+
+// Changes how the clause behaves based on information only available after it is constructed
+struct ProcessingConfig {
+    bool dynamic_schema_{false};
+    uint64_t total_rows_ = 0;
+};
+
+// Used when restructuring segments inbetween clauses with differing ProcessingStructures
+struct RangesAndEntity {
+    explicit RangesAndEntity(EntityId id, std::shared_ptr<RowRange> row_range, std::shared_ptr<ColRange> col_range, std::optional<TimestampRange>&& timestamp_range=std::nullopt):
+            id_(id),
+            row_range_(std::move(row_range)),
+            col_range_(std::move(col_range)),
+            timestamp_range_(std::move(timestamp_range)) {
+    }
+    ARCTICDB_MOVE_COPY_DEFAULT(RangesAndEntity)
+
+    const RowRange& row_range() const {
+        return *row_range_;
+    }
+
+    const ColRange& col_range() const {
+        return *col_range_;
+    }
+
+    timestamp start_time() const {
+        return timestamp_range_->first;
+    }
+
+    timestamp end_time() const {
+        return timestamp_range_->second;
+    }
+
+    friend bool operator==(const RangesAndEntity& left, const RangesAndEntity& right) {
+        return *left.row_range_ == *right.row_range_ && *left.col_range_ == *right.col_range_ && left.timestamp_range_ == right.timestamp_range_;
+    }
+
+    bool operator!=(const RangesAndEntity& right) const {
+        return !(*this == right);
+    }
+
+    EntityId id_;
+    std::shared_ptr<RowRange> row_range_;
+    std::shared_ptr<ColRange> col_range_;
+    std::optional<TimestampRange> timestamp_range_;
+};
+
+template<typename T>
+requires std::is_same_v<T, RangesAndKey> || std::is_same_v<T, RangesAndEntity>
+std::vector<std::vector<size_t>> structure_by_row_slice(
+        std::vector<T>& ranges) {
+    std::sort(std::begin(ranges), std::end(ranges), [] (const T& left, const T& right) {
+        return std::tie(left.row_range().first, left.col_range().first) < std::tie(right.row_range().first, right.col_range().first);
+    });
+
+    std::vector<std::vector<size_t>> res;
+    RowRange previous_row_range{-1, -1};
+    for (const auto& [idx, ranges_and_key]: folly::enumerate(ranges)) {
+        RowRange current_row_range{ranges_and_key.row_range()};
+        if (current_row_range != previous_row_range) {
+            res.emplace_back();
+        }
+        res.back().emplace_back(idx);
+        previous_row_range = current_row_range;
+    }
+    return res;
+}
+
+template<ResampleBoundary closed_boundary>
+bool index_range_outside_bucket_range(timestamp start_index, timestamp end_index, const std::vector<timestamp>& bucket_boundaries) {
+    if constexpr (closed_boundary == ResampleBoundary::LEFT) {
+        return start_index >= bucket_boundaries.back() || end_index < bucket_boundaries.front();
+    } else {
+        // closed_boundary == ResampleBoundary::RIGHT
+        return start_index > bucket_boundaries.back() || end_index <= bucket_boundaries.front();
+    }
+}
+
+// Advances the bucket boundary iterator to the end of the last bucket that includes a value from a row slice with the given last index value
+template<ResampleBoundary closed_boundary>
+void advance_boundary_past_value(const std::vector<timestamp>& bucket_boundaries,
+                                 std::vector<timestamp>::const_iterator& bucket_boundaries_it,
+                                 timestamp value) {
+    // These loops are equivalent to bucket_boundaries_it = std::upper_bound(bucket_boundaries_it, bucket_boundaries.end(), value, std::less[_equal]{})
+    // but optimised for the case where most buckets are non-empty.
+    // Mathematically, this will be faster when b / log_2(b) < n, where b is the number of buckets and n is the number of index values
+    // Even if n is only 1000, this corresponds to 7/8 buckets being empty, rising to 19/20 for n=100,000
+    // Experimentally, this implementation is around 10x faster when every bucket contains values, and 3x slower when 99.9% of buckets are empty
+    // If we wanted to speed this up when most buckets are empty, we could make this method adaptive to the number of buckets and rows
+    if constexpr(closed_boundary == ResampleBoundary::LEFT) {
+        while(bucket_boundaries_it != bucket_boundaries.end() && *bucket_boundaries_it <= value) {
+            ++bucket_boundaries_it;
+        }
+    } else {
+        // closed_boundary == ResampleBoundary::RIGHT
+        while(bucket_boundaries_it != bucket_boundaries.end() && *bucket_boundaries_it < value) {
+            ++bucket_boundaries_it;
+        }
+    }
+}
+
+template<ResampleBoundary closed_boundary, typename T>
+requires std::is_same_v<T, RangesAndKey> || std::is_same_v<T, RangesAndEntity>
+std::vector<std::vector<size_t>> structure_by_time_bucket(
+        std::vector<T>& ranges,
+        const std::vector<timestamp>& bucket_boundaries) {
+    std::erase_if(ranges, [&bucket_boundaries](const T &range) {
+        auto start_index = range.start_time();
+        auto end_index = range.end_time();
+        return index_range_outside_bucket_range<closed_boundary>(start_index, end_index, bucket_boundaries);
+    });
+    auto res = structure_by_row_slice(ranges);
+    // Element i of res also needs the values from element i+1 if there is a bucket which incorporates the last index
+    // value of row-slice i and the first value of row-slice i+1
+    // Element i+1 should be removed if the last bucket involved in element i covers all the index values in element i+1
+    auto bucket_boundaries_it = std::cbegin(bucket_boundaries);
+    // Exit if res_it == std::prev(res.end()) as this implies the last row slice was not incorporated into an earlier processing unit
+    for (auto res_it = res.begin(); res_it != res.end() && res_it != std::prev(res.end());) {
+        auto last_index_value_in_row_slice = ranges[res_it->at(0)].end_time();
+        advance_boundary_past_value<closed_boundary>(bucket_boundaries, bucket_boundaries_it, last_index_value_in_row_slice);
+        // bucket_boundaries_it now contains the end value of the last bucket covering the row-slice in res_it, or an end iterator if the last bucket ends before the end of this row-slice
+        if (bucket_boundaries_it != bucket_boundaries.end()) {
+            Bucket<closed_boundary> current_bucket{*std::prev(bucket_boundaries_it), *bucket_boundaries_it};
+            auto next_row_slice_it = std::next(res_it);
+            while (next_row_slice_it != res.end()) {
+                // end_index from the key is 1 nanosecond larger than the index value of the last row in the row-slice
+                TimestampRange next_row_slice_timestamp_range{
+                        ranges[next_row_slice_it->at(0)].start_time(),
+                        ranges[next_row_slice_it->at(0)].end_time()};
+                if (current_bucket.contains(next_row_slice_timestamp_range.first)) {
+                    // The last bucket in the current processing unit overlaps with the first index value in the next row slice, so add segments into current processing unit
+                    res_it->insert(res_it->end(), next_row_slice_it->begin(), next_row_slice_it->end());
+                    if (current_bucket.contains(next_row_slice_timestamp_range.second)) {
+                        // The last bucket in the current processing unit wholly contains the next row slice, so remove it from the result
+                        next_row_slice_it = res.erase(next_row_slice_it);
+                    } else {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+            // This is the last bucket, and all the required row-slices have been incorporated into the current processing unit, so erase the rest
+            if (bucket_boundaries_it == std::prev(bucket_boundaries.end())) {
+                res.erase(next_row_slice_it, res.end());
+                break;
+            }
+            res_it = next_row_slice_it;
+        }
+    }
+    return res;
+}
+
+std::vector<std::vector<EntityId>> structure_by_row_slice(ComponentManager& component_manager, std::vector<std::vector<EntityId>>&& entity_ids_vec);
+
+std::vector<std::vector<EntityId>> offsets_to_entity_ids(const std::vector<std::vector<size_t>>& offsets,
+                                                         const std::vector<RangesAndEntity>& ranges_and_entities);
+
+/*
+ * On entry to a clause, construct ProcessingUnits from the input entity IDs. These will be provided by the
+ * structure_for_processing method for the first clause in the pipeline and for clauses whose expected input structure
+ * differs from the previous clause's output structure. Otherwise, these come directly from the previous clause.
+ * clauses.
+ */
+template<class... Args>
+ProcessingUnit gather_entities(ComponentManager& component_manager, std::vector<EntityId>&& entity_ids) {
+    ProcessingUnit res;
+    auto components = component_manager.get_entities<Args...>(entity_ids);
+    ([&]{
+        auto component = std::move(std::get<std::vector<Args>>(components));
+        if constexpr (std::is_same_v<Args, std::shared_ptr<SegmentInMemory>>) {
+            res.set_segments(std::move(component));
+        } else if constexpr (std::is_same_v<Args, std::shared_ptr<RowRange>>) {
+            res.set_row_ranges(std::move(component));
+        } else if constexpr (std::is_same_v<Args, std::shared_ptr<ColRange>>) {
+            res.set_col_ranges(std::move(component));
+        } else if constexpr (std::is_same_v<Args, std::shared_ptr<AtomKey>>) {
+            res.set_atom_keys(std::move(component));
+        } else if constexpr (std::is_same_v<Args, EntityFetchCount>) {
+            res.set_entity_fetch_count(std::move(component));
+        } else {
+            static_assert(sizeof(Args) == 0, "Unexpected component type provided in gather_entities");
+        }
+    }(), ...);
+    return res;
+}
+
+std::vector<EntityId> push_entities(ComponentManager& component_manager, ProcessingUnit&& proc, EntityFetchCount entity_fetch_count=1);
+
+std::vector<EntityId> flatten_entities(std::vector<std::vector<EntityId>>&& entity_ids_vec);
+
+}//namespace arcticdb
diff --git a/cpp/arcticdb/processing/component_manager.hpp b/cpp/arcticdb/processing/component_manager.hpp
index c7410cb53f..406df05c10 100644
--- a/cpp/arcticdb/processing/component_manager.hpp
+++ b/cpp/arcticdb/processing/component_manager.hpp
@@ -77,9 +77,33 @@ class ComponentManager {
         return ids;
     }
 
+    template<typename T>
+    void replace_entities(const std::vector<EntityId>& ids, T value) {
+        for (auto id: ids) {
+            registry_.replace<T>(id, value);
+            if constexpr (std::is_same_v<T, EntityFetchCount>) {
+                auto&& remaining_entity_fetch_count_registry = registry_.storage<EntityFetchCount>(remaining_entity_fetch_count_id);
+                // For some reason named storages don't have a replace API
+                remaining_entity_fetch_count_registry.patch(id, [value](EntityFetchCount& entity_fetch_count){ entity_fetch_count = value; });
+            }
+        }
+    }
+
+    template<typename T>
+    void replace_entities(const std::vector<EntityId>& ids, const std::vector<T>& values) {
+        for (auto [idx, id]: folly::enumerate(ids)) {
+            registry_.replace<T>(id, values[idx]);
+            if constexpr (std::is_same_v<T, EntityFetchCount>) {
+                auto&& remaining_entity_fetch_count_registry = registry_.storage<EntityFetchCount>(remaining_entity_fetch_count_id);
+                // For some reason named storages don't have a replace API
+                remaining_entity_fetch_count_registry.patch(id, [&values, idx](EntityFetchCount& entity_fetch_count){ entity_fetch_count = values[idx]; });
+            }
+        }
+    }
+
     // Get a collection of entities. Returns a tuple of vectors, one for each component requested via Args
     template<class... Args>
-    std::tuple<std::vector<Args>...> get_entities(const std::vector<EntityId>& ids) {
+    std::tuple<std::vector<Args>...> get_entities(const std::vector<EntityId>& ids, const bool decrement_fetch_count=true) {
         std::vector<std::tuple<Args...>> tuple_res;
         tuple_res.reserve(ids.size());
         {
@@ -89,10 +113,14 @@ class ComponentManager {
             auto view = registry_.view<const Args...>();
             for (auto id: ids) {
                 tuple_res.emplace_back(std::move(view.get(id)));
-                auto& remaining_entity_fetch_count = remaining_entity_fetch_count_registry.get(id);
-                // This entity will never be accessed again
-                if (--remaining_entity_fetch_count == 0) {
-                    erase_entity(id);
+            }
+            if (decrement_fetch_count) {
+                for (auto id: ids) {
+                    auto& remaining_entity_fetch_count = remaining_entity_fetch_count_registry.get(id);
+                    // This entity will never be accessed again
+                    if (--remaining_entity_fetch_count == 0) {
+                        erase_entity(id);
+                    }
                 }
             }
         }
diff --git a/cpp/arcticdb/processing/query_planner.cpp b/cpp/arcticdb/processing/query_planner.cpp
new file mode 100644
index 0000000000..fa5e2b7405
--- /dev/null
+++ b/cpp/arcticdb/processing/query_planner.cpp
@@ -0,0 +1,29 @@
+/* Copyright 2024 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.
+ */
+
+#include <arcticdb/processing/query_planner.hpp>
+
+namespace arcticdb {
+
+std::vector<ClauseVariant> plan_query(std::vector<ClauseVariant>&& clauses) {
+    if (clauses.size() >= 2 && std::holds_alternative<std::shared_ptr<DateRangeClause>>(clauses[0])) {
+        util::variant_match(
+                clauses[1],
+                [&clauses](auto&& clause) {
+                    if constexpr (is_resample<typename std::remove_cvref_t<decltype(clause)>::element_type>::value) {
+                        const auto& date_range_clause = *std::get<std::shared_ptr<DateRangeClause>>(clauses[0]);
+                        auto date_range_start = date_range_clause.start_;
+                        auto date_range_end = date_range_clause.end_;
+                        clause->set_date_range(date_range_start, date_range_end);
+                        clauses.erase(clauses.cbegin());
+                    }
+                });
+    }
+    return clauses;
+}
+
+}//namespace arcticdb
diff --git a/cpp/arcticdb/processing/query_planner.hpp b/cpp/arcticdb/processing/query_planner.hpp
new file mode 100644
index 0000000000..99106654e1
--- /dev/null
+++ b/cpp/arcticdb/processing/query_planner.hpp
@@ -0,0 +1,29 @@
+/* Copyright 2024 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 <variant>
+#include <vector>
+
+#include <arcticdb/processing/clause.hpp>
+
+namespace arcticdb {
+
+using GroupByClause = PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer>;
+using ClauseVariant = std::variant<std::shared_ptr<FilterClause>,
+        std::shared_ptr<ProjectClause>,
+        std::shared_ptr<GroupByClause>,
+        std::shared_ptr<AggregationClause>,
+        std::shared_ptr<ResampleClause<ResampleBoundary::LEFT>>,
+        std::shared_ptr<ResampleClause<ResampleBoundary::RIGHT>>,
+        std::shared_ptr<RowRangeClause>,
+        std::shared_ptr<DateRangeClause>>;
+
+std::vector<ClauseVariant> plan_query(std::vector<ClauseVariant>&& clauses);
+
+}//namespace arcticdb
diff --git a/cpp/arcticdb/version/python_bindings.cpp b/cpp/arcticdb/version/python_bindings.cpp
index 0d12aecb33..bd837a4e46 100644
--- a/cpp/arcticdb/version/python_bindings.cpp
+++ b/cpp/arcticdb/version/python_bindings.cpp
@@ -20,6 +20,7 @@
 #include <arcticdb/processing/operation_types.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/processing/expression_context.hpp>
+#include <arcticdb/processing/query_planner.hpp>
 #include <arcticdb/pipeline/value_set.hpp>
 #include <arcticdb/python/adapt_read_dataframe.hpp>
 #include <arcticdb/version/schema_checks.hpp>
@@ -89,9 +90,6 @@ void declare_resample_clause(py::module& version) {
                                         const std::unordered_map<std::string, std::variant<std::string, std::pair<std::string, std::string>>> aggregations) {
                 self.set_aggregations(python_util::named_aggregators_from_dict(aggregations));
             })
-            .def("set_date_range", [](ResampleClause<closed_boundary>& self, timestamp date_range_start, timestamp date_range_end) {
-                self.set_date_range(date_range_start, date_range_end);
-            })
             .def("__str__", &ResampleClause<closed_boundary>::to_string);
 }
 
@@ -318,8 +316,6 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
                     ExpressionContext>())
             .def("__str__", &ProjectClause::to_string);
 
-    using GroupByClause = PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer>;
-
     py::class_<GroupByClause, std::shared_ptr<GroupByClause>>(version, "GroupByClause")
             .def(py::init<std::string>())
             .def_property_readonly("grouping_column", [](const GroupByClause& self) {
@@ -375,26 +371,12 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
                                 std::shared_ptr<ResampleClause<ResampleBoundary::RIGHT>>,
                                 std::shared_ptr<RowRangeClause>,
                                 std::shared_ptr<DateRangeClause>>> clauses) {
+                clauses = plan_query(std::move(clauses));
                 std::vector<std::shared_ptr<Clause>> _clauses;
+                self.needs_post_processing = false;
                 for (auto&& clause: clauses) {
                     util::variant_match(
                         clause,
-                        [&](std::shared_ptr<RowRangeClause> clause) {
-                            self.needs_post_processing = false;
-                            _clauses.emplace_back(std::make_shared<Clause>(*clause));
-                        },
-                        [&](std::shared_ptr<DateRangeClause> clause) {
-                            self.needs_post_processing = false;
-                            _clauses.emplace_back(std::make_shared<Clause>(*clause));
-                        },
-                        [&](std::shared_ptr<ResampleClause<ResampleBoundary::LEFT>> clause) {
-                            self.needs_post_processing = false;
-                            _clauses.emplace_back(std::make_shared<Clause>(*clause));
-                        },
-                        [&](std::shared_ptr<ResampleClause<ResampleBoundary::RIGHT>> clause) {
-                            self.needs_post_processing = false;
-                            _clauses.emplace_back(std::make_shared<Clause>(*clause));
-                        },
                         [&](auto&& clause) {_clauses.emplace_back(std::make_shared<Clause>(*clause));}
                     );
                 }
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 7dfa1dd685..8aaa0cbd25 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -531,8 +531,6 @@ std::vector<EntityId> process_clauses(
     auto slice_added = std::make_shared<std::vector<bool>>(segment_and_slice_futures.size(), false);
     std::vector<folly::Future<std::vector<EntityId>>> futures;
     bool first_clause{true};
-    // Reverse the order of clauses and iterate through them backwards so that the erase is efficient
-    std::reverse(clauses->begin(), clauses->end());
     while (!clauses->empty()) {
         for (auto&& entity_ids: entity_ids_vec) {
             if (first_clause) {
@@ -581,12 +579,19 @@ std::vector<EntityId> process_clauses(
         first_clause = false;
         entity_ids_vec = folly::collect(futures).get();
         futures.clear();
-        while (clauses->size() > 0 && !clauses->back()->clause_info().requires_repartition_) {
-            clauses->erase(clauses->end() - 1);
+        // Erase all the clauses we have already called process on
+        auto it = std::next(clauses->cbegin());
+        while (it != clauses->cend()) {
+            auto prev_it = std::prev(it);
+            if ((*prev_it)->clause_info().output_structure_ == (*it)->clause_info().input_structure_) {
+                ++it;
+            } else {
+                break;
+            }
         }
-        if (clauses->size() > 0 && clauses->back()->clause_info().requires_repartition_) {
-            entity_ids_vec = clauses->back()->repartition(std::move(entity_ids_vec)).value();
-            clauses->erase(clauses->end() - 1);
+        clauses->erase(clauses->cbegin(), it);
+        if (!clauses->empty()) {
+            entity_ids_vec = clauses->front()->structure_for_processing(std::move(entity_ids_vec));
         }
     }
     return flatten_entities(std::move(entity_ids_vec));
@@ -703,8 +708,7 @@ std::vector<SliceAndKey> read_and_process(
     const std::shared_ptr<Store>& store,
     const std::shared_ptr<PipelineContext>& pipeline_context,
     const ReadQuery& read_query,
-    const ReadOptions& read_options,
-    size_t start_from
+    const ReadOptions& read_options
     ) {
     auto component_manager = std::make_shared<ComponentManager>();
     ProcessingConfig processing_config{opt_false(read_options.dynamic_schema_), pipeline_context->rows_};
@@ -719,7 +723,7 @@ std::vector<SliceAndKey> read_and_process(
     // Each element of the vector corresponds to one processing unit containing the list of indexes in ranges_and_keys required for that processing unit
     // i.e. if the first processing unit needs ranges_and_keys[0] and ranges_and_keys[1], and the second needs ranges_and_keys[2] and ranges_and_keys[3]
     // then the structure will be {{0, 1}, {2, 3}}
-    std::vector<std::vector<size_t>> processing_unit_indexes = read_query.clauses_[0]->structure_for_processing(ranges_and_keys, start_from);
+    std::vector<std::vector<size_t>> processing_unit_indexes = read_query.clauses_[0]->structure_for_processing(ranges_and_keys);
 
     // Start reading as early as possible
     auto segment_and_slice_futures = store->batch_read_uncompressed(std::move(ranges_and_keys), columns_to_decode(pipeline_context));
@@ -1251,7 +1255,7 @@ void create_column_stats_impl(
             "Cannot create column stats on pickled data"
             );
 
-    auto segs = read_and_process(store, pipeline_context, read_query, read_options, 0u);
+    auto segs = read_and_process(store, pipeline_context, read_query, read_options);
     schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(!segs.empty(), "Cannot create column stats for nonexistent columns");
 
     // Convert SliceAndKey vector into SegmentInMemory vector
@@ -1358,7 +1362,7 @@ SegmentInMemory do_direct_read_or_process(
     if(!read_query.clauses_.empty()) {
         ARCTICDB_SAMPLE(RunPipelineAndOutput, 0)
         util::check_rte(!pipeline_context->is_pickled(),"Cannot filter pickled data");
-        auto segs = read_and_process(store, pipeline_context, read_query, read_options, 0u);
+        auto segs = read_and_process(store, pipeline_context, read_query, read_options);
         frame = prepare_output_frame(std::move(segs), pipeline_context, store, read_options, handler_data);
     } else {
         ARCTICDB_SAMPLE(MarkAndReadDirect, 0)
@@ -1504,7 +1508,7 @@ VersionedItem sort_merge_impl(
     auto index = stream::index_type_from_descriptor(pipeline_context->descriptor());
     util::variant_match(index,
         [&](const stream::TimeseriesIndex &timeseries_index) {
-            read_query.clauses_.emplace_back(std::make_shared<Clause>(SortClause{timeseries_index.name()}));
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(SortClause{timeseries_index.name(), pipeline_context->incompletes_after()}));
             read_query.clauses_.emplace_back(std::make_shared<Clause>(RemoveColumnPartitioningClause{}));
             //const auto split_size = ConfigsMap::instance()->get_int("Split.RowCount", 10000);
             //read_query.clauses_.emplace_back(std::make_shared<Clause>(SplitClause{static_cast<size_t>(split_size)}));
@@ -1516,7 +1520,7 @@ VersionedItem sort_merge_impl(
                 pipeline_context->descriptor(),
                 write_options.dynamic_schema
             }));
-            auto segments = read_and_process(store, pipeline_context, read_query, ReadOptions{}, pipeline_context->incompletes_after());
+            auto segments = read_and_process(store, pipeline_context, read_query, ReadOptions{});
             if (options.append_ && update_info.previous_index_key_ && !segments.empty()) {
                 const timestamp last_index_on_disc = update_info.previous_index_key_->end_time() - 1;
                 const timestamp incomplete_start =
@@ -1746,8 +1750,8 @@ VersionedItem defragment_symbol_data_impl(
 
     util::variant_match(std::move(policies), [
         &fut_vec, &slices, &store, &options, &pre_defragmentation_info, segment_size=segment_size] (auto &&idx, auto &&schema) {
-        pre_defragmentation_info.read_query.clauses_.emplace_back(std::make_shared<Clause>(RemoveColumnPartitioningClause{}));
-        auto segments = read_and_process(store, pre_defragmentation_info.pipeline_context, pre_defragmentation_info.read_query, defragmentation_read_options_generator(options), pre_defragmentation_info.append_after.value());
+        pre_defragmentation_info.read_query.clauses_.emplace_back(std::make_shared<Clause>(RemoveColumnPartitioningClause{pre_defragmentation_info.append_after.value()}));
+        auto segments = read_and_process(store, pre_defragmentation_info.pipeline_context, pre_defragmentation_info.read_query, defragmentation_read_options_generator(options));
         using IndexType = std::remove_reference_t<decltype(idx)>;
         using SchemaType = std::remove_reference_t<decltype(schema)>;
         do_compact<IndexType, SchemaType, RowCountSegmentPolicy, DenseColumnPolicy>(
diff --git a/python/arcticdb/version_store/_store.py b/python/arcticdb/version_store/_store.py
index 1eec7e3078..fccaf5bf33 100644
--- a/python/arcticdb/version_store/_store.py
+++ b/python/arcticdb/version_store/_store.py
@@ -1535,14 +1535,13 @@ def _get_read_query(
         check(date_range is None or row_range is None, "Date range and row range both specified")
         read_query = _PythonVersionStoreReadQuery()
 
-        if date_range is not None and query_builder is not None and not query_builder.is_resample():
+        if date_range is not None and query_builder is not None:
             query_builder.prepend(QueryBuilder().date_range(date_range))
 
         if row_range is not None and query_builder is not None:
-            query_builder.prepend(QueryBuilder()._row_range(row_range))
+            query_builder.prepend(QueryBuilder().row_range(row_range))
 
-        if query_builder:
-            query_builder.set_date_range(date_range)
+        if query_builder is not None:
             read_query.add_clauses(query_builder.clauses)
 
         if row_range is not None:
@@ -1606,7 +1605,7 @@ def _get_read_queries(
             if columns is not None:
                 these_columns = columns[idx]
             if query_builder is not None:
-                query = query_builder if isinstance(query_builder, QueryBuilder) else query_builder[idx]
+                query = copy.deepcopy(query_builder) if isinstance(query_builder, QueryBuilder) else query_builder[idx]
 
             read_query = self._get_read_query(
                 date_range=date_range,
@@ -1760,7 +1759,7 @@ def head(
         implement_read_index = kwargs.get("implement_read_index", False)
         columns = self._resolve_empty_columns(columns, implement_read_index)
         q = QueryBuilder()
-        q = q._head(n)
+        q = q.head(n)
         version_query, read_options, read_query = self._get_queries(
             as_of=as_of, date_range=None, row_range=None, columns=columns, query_builder=q, **kwargs
         )
@@ -1793,7 +1792,7 @@ def tail(
         implement_read_index = kwargs.get("implement_read_index", False)
         columns = self._resolve_empty_columns(columns, implement_read_index)
         q = QueryBuilder()
-        q = q._tail(n)
+        q = q.tail(n)
         version_query, read_options, read_query = self._get_queries(
             as_of=as_of, date_range=None, row_range=None, columns=columns, query_builder=q, **kwargs
         )
diff --git a/python/arcticdb/version_store/library.py b/python/arcticdb/version_store/library.py
index 514397b343..4239bc8fbb 100644
--- a/python/arcticdb/version_store/library.py
+++ b/python/arcticdb/version_store/library.py
@@ -2029,7 +2029,7 @@ def head(
             If lazy is True, a LazyDataFrame object on which further querying can be performed prior to collect.
         """
         if lazy:
-            q = QueryBuilder()._head(n)
+            q = QueryBuilder().head(n)
             return LazyDataFrame(
                 self,
                 ReadRequest(
@@ -2080,7 +2080,7 @@ def tail(
             If lazy is True, a LazyDataFrame object on which further querying can be performed prior to collect.
         """
         if lazy:
-            q = QueryBuilder()._tail(n)
+            q = QueryBuilder().tail(n)
             return LazyDataFrame(
                 self,
                 ReadRequest(
diff --git a/python/arcticdb/version_store/processing.py b/python/arcticdb/version_store/processing.py
index eb515f21ad..fa0784344b 100644
--- a/python/arcticdb/version_store/processing.py
+++ b/python/arcticdb/version_store/processing.py
@@ -718,7 +718,6 @@ def resample(
                                  agg_1_min  agg_1_max     agg_2
             2024-01-01 00:00:00          1          5      2.75
         """
-        check(not len(self.clauses), "resample only supported as first clause in the pipeline")
         rule = rule.freqstr if isinstance(rule, pd.DateOffset) else rule
         # We use floor and ceiling later to round user-provided date ranges and or start/end index values of the symbol
         # before calling pandas.date_range to generate the bucket boundaries, but floor and ceiling only work with
@@ -756,14 +755,6 @@ def resample(
         self._python_clauses = self._python_clauses + [PythonResampleClause(rule=rule, closed=boundary_map[closed], label=boundary_map[label], offset=offset_ns)]
         return self
 
-    def is_resample(self):
-        return len(self.clauses) and isinstance(self.clauses[0], (_ResampleClauseLeftClosed, _ResampleClauseRightClosed))
-
-    def set_date_range(self, date_range:Optional[DateRangeInput]):
-        if self.is_resample() and date_range is not None:
-            start, end = normalize_dt_range_to_ts(date_range)
-            self.clauses[0].set_date_range(start.value, end.value)
-
 
     # TODO: specify type of other must be QueryBuilder with from __future__ import annotations once only Python 3.7+
     # supported
@@ -781,10 +772,6 @@ def then(self, other):
         QueryBuilder
             Modified QueryBuilder object.
         """
-        check(
-            not len(other.clauses) or not isinstance(other.clauses[0], (_DateRangeClause, _ResampleClauseLeftClosed, _ResampleClauseRightClosed)),
-            "In QueryBuilder.then: Date range and Resample only supported as first clauses in the pipeline",
-        )
         self.clauses = self.clauses + other.clauses
         self._python_clauses = self._python_clauses + other._python_clauses
         return self
@@ -805,28 +792,61 @@ def prepend(self, other):
         QueryBuilder
             Modified QueryBuilder object.
         """
-        check(
-            not len(self.clauses) or not isinstance(self.clauses[0], (_DateRangeClause, _ResampleClauseLeftClosed, _ResampleClauseRightClosed)),
-            "In QueryBuilder.prepend: Date range and Resample only supported as first clauses in the pipeline",
-            )
         self.clauses = other.clauses + self.clauses
         self._python_clauses = other._python_clauses + self._python_clauses
         return self
 
-    def _head(self, n: int):
-        check(not len(self.clauses), "Head only supported as first clause in the pipeline")
+    def head(self, n: int = 5):
+        """
+        Filter out all but the first n rows of data. If n is negative, return all rows except the last n rows.
+
+        Parameters
+        ----------
+        n : int, default=5
+            Number of rows to select if non-negative, otherwise number of rows to exclude.
+
+        Returns
+        -------
+        QueryBuilder
+            Modified QueryBuilder object.
+        """
         self.clauses = self.clauses + [_RowRangeClause(_RowRangeType.HEAD, n)]
         self._python_clauses = self._python_clauses + [PythonRowRangeClause(row_range_type=_RowRangeType.HEAD, n=n)]
         return self
 
-    def _tail(self, n: int):
-        check(not len(self.clauses), "Tail only supported as first clause in the pipeline")
+    def tail(self, n: int = 5):
+        """
+        Filter out all but the last n rows of data. If n is negative, return all rows except the first n rows.
+
+        Parameters
+        ----------
+        n : int, default=5
+            Number of rows to select if non-negative, otherwise number of rows to exclude.
+
+        Returns
+        -------
+        QueryBuilder
+            Modified QueryBuilder object.
+        """
         self.clauses = self.clauses + [_RowRangeClause(_RowRangeType.TAIL, n)]
         self._python_clauses = self._python_clauses + [PythonRowRangeClause(row_range_type=_RowRangeType.TAIL, n=n)]
         return self
 
-    def _row_range(self, row_range):
-        check(not len(self.clauses), "Row range only supported as first clause in the pipeline")
+    def row_range(self, row_range: Tuple[int, int]):
+        """
+        Row range to read data for. Inclusive of the lower bound, exclusive of the upper bound.
+        Should behave the same as df.iloc[start:end], including in the handling of negative start/end values.
+
+        Parameters
+        ----------
+        row_range : Tuple[int, int]
+            Row range to read data for. Inclusive of the lower bound, exclusive of the upper bound.
+
+        Returns
+        -------
+        QueryBuilder
+            Modified QueryBuilder object.
+        """
         start = row_range[0]
         end = row_range[1]
 
@@ -837,9 +857,9 @@ def _row_range(self, row_range):
     def date_range(self, date_range: DateRangeInput):
         """
         DateRange to read data for.  Applicable only for Pandas data with a DateTime index. Returns only the part
-        of the data that falls within the given range. The returned data object will use less memory than passing
-        date_range directly as an argument to the read method, at the cost of being slightly slower.
-        Must be the first clause in the QueryBuilder object.
+        of the data that falls within the given range. If this is the only processing clause being applied, then the
+        returned data object will use less memory than passing date_range directly as an argument to the read method, at
+         the cost of possibly being slightly slower.
 
         Parameters
         ----------
@@ -856,7 +876,6 @@ def date_range(self, date_range: DateRangeInput):
         QueryBuilder
             Modified QueryBuilder object.
         """
-        check(not len(self.clauses), "Date range only supported as first clause in the pipeline")
         start, end = normalize_dt_range_to_ts(date_range)
         self.clauses = self.clauses + [_DateRangeClause(start.value, end.value)]
         self._python_clauses = self._python_clauses + [PythonDateRangeClause(start.value, end.value)]
diff --git a/python/benchmarks/non_asv/profile_resample.py b/python/benchmarks/non_asv/profile_resample.py
index 8d6082d99c..4c3aad0987 100644
--- a/python/benchmarks/non_asv/profile_resample.py
+++ b/python/benchmarks/non_asv/profile_resample.py
@@ -12,7 +12,7 @@
 import pandas as pd
 import pytest
 
-from arcticdb import QueryBuilder, Arctic
+from arcticdb import QueryBuilder, Arctic, LibraryOptions
 from arcticdb.util.test import assert_frame_equal, random_strings_of_length
 
 rows_per_segment = 100_000
@@ -20,7 +20,84 @@
 
 ac = Arctic("lmdb:///tmp/arcticdb")
 lib = ac.get_library("resample_profiling", create_if_missing=True)
+# lib = ac.get_library("bbg_profiling_big_segment", True, LibraryOptions(rows_per_segment=1_000_000))
 lib = lib._nvs
+bbg_sym = "BBG"
+
+
+def test_write_bbg_style_data():
+    lib.version_store.clear()
+    start_time = pd.Timestamp("2020-01-01T08:00:00")
+    num_days = 250
+    ticks_per_day = 24_000_000
+    half_ticks = ticks_per_day // 2
+    for day in range(num_days):
+        print(f"Day {day+1}/{num_days}: {start_time}")
+        index = pd.date_range(start_time, freq="ms", periods=ticks_per_day)
+        # Generate data that is half BID, then half ASK, then randomly permute for final data
+        tick_type_col = np.concatenate([np.repeat(np.array(["BID"]), half_ticks), np.repeat(np.array(["ASK"]), half_ticks)])
+        bid_col = np.concatenate([rng.random(half_ticks), np.repeat(np.array([np.nan]), half_ticks)])
+        ask_col = np.concatenate([np.repeat(np.array([np.nan]), half_ticks), rng.random(half_ticks)])
+        permutation = rng.permutation(ticks_per_day)
+        tick_type_col = tick_type_col[permutation]
+        bid_col = bid_col[permutation]
+        ask_col = ask_col[permutation]
+        df = pd.DataFrame({"tick type": tick_type_col, "bid": bid_col, "ask": ask_col}, index=index)
+        lib.append(bbg_sym, df)
+        start_time += pd.Timedelta(1, "day")
+
+
+def test_read_bbg_style_data():
+    start_time = pd.Timestamp("2020-01-01")
+    num_days = 125
+    end_time = start_time + pd.Timedelta(num_days, "days")
+    start = time.time()
+    # df = lib.read(bbg_sym, date_range=(start_time, end_time)).data
+    df = lib.read(bbg_sym).data
+    end = time.time()
+    print(f"Reading {len(df)} ticks took {end - start}")
+
+
+def test_filter_bbg_style_data():
+    start_time = pd.Timestamp("2020-01-01")
+    num_days = 125
+    end_time = start_time + pd.Timedelta(num_days, "days")
+    q = QueryBuilder()
+    q = q[q["tick type"] == "BID"]
+    start = time.time()
+    # df = lib.read(bbg_sym, date_range=(start_time, end_time), columns=["bid"], query_builder=q).data
+    df = lib.read(bbg_sym, columns=["bid"], query_builder=q).data
+    end = time.time()
+    print(f"Filtering to {len(df)} ticks took {end - start}")
+
+
+def test_resample_bbg_style_data():
+    start_time = pd.Timestamp("2020-01-01")
+    num_days = 125
+    end_time = start_time + pd.Timedelta(num_days, "days")
+    q = QueryBuilder()
+    freq = "min"
+    q = q.resample(freq).agg({"bid": "max"})
+    start = time.time()
+    # df = lib.read(bbg_sym, date_range=(start_time, end_time), query_builder=q).data
+    df = lib.read(bbg_sym, query_builder=q).data
+    end = time.time()
+    print(f"Resampling to {len(df)} {freq} took {end - start}")
+
+
+def test_filter_then_resample_bbg_style_data():
+    start_time = pd.Timestamp("2020-01-01")
+    num_days = 1
+    end_time = start_time + pd.Timedelta(num_days, "days")
+    q = QueryBuilder()
+    q = q[q["tick type"] == "BID"]
+    freq = "min"
+    q = q.resample(freq).agg({"bid": "max"})
+    start = time.time()
+    # df = lib.read(bbg_sym, date_range=(start_time, end_time), query_builder=q).data
+    df = lib.read(bbg_sym, query_builder=q).data
+    end = time.time()
+    print(f"Filtering then resampling to {len(df)} {freq} took {end - start}")
 
 
 @pytest.mark.parametrize("num_rows", [100_000, 1_000_000, 10_000_000, 100_000_000, 1_000_000_000])


Test patch:
diff --git a/cpp/arcticdb/processing/test/rapidcheck_resample.cpp b/cpp/arcticdb/processing/test/rapidcheck_resample.cpp
index 4e3b39ced9..14285d7a91 100644
--- a/cpp/arcticdb/processing/test/rapidcheck_resample.cpp
+++ b/cpp/arcticdb/processing/test/rapidcheck_resample.cpp
@@ -114,11 +114,11 @@ RC_GTEST_PROP(Resample, StructureForProcessing, ()) {
 
     if (left_boundary_closed) {
         ResampleClause<ResampleBoundary::LEFT> resample_clause{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries(std::move(bucket_boundaries)), 0};
-        auto result = resample_clause.structure_for_processing(ranges_and_keys, 0);
+        auto result = resample_clause.structure_for_processing(ranges_and_keys);
         RC_ASSERT(expected_result == result);
     } else {
         ResampleClause<ResampleBoundary::RIGHT> resample_clause{"dummy", ResampleBoundary::RIGHT, generate_bucket_boundaries(std::move(bucket_boundaries)), 0};
-        auto result = resample_clause.structure_for_processing(ranges_and_keys, 0);
+        auto result = resample_clause.structure_for_processing(ranges_and_keys);
         RC_ASSERT(expected_result == result);
     }
 }
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 4cfb15cf69..e21847503a 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -312,7 +312,7 @@ TEST(Clause, Sort) {
     using namespace arcticdb;
     auto component_manager = std::make_shared<ComponentManager>();
 
-    SortClause sort_clause("time");
+    SortClause sort_clause("time", size_t(0));
     sort_clause.set_component_manager(component_manager);
 
     auto seg = get_standard_timeseries_segment("sort");
@@ -408,8 +408,8 @@ TEST(Clause, Merge) {
     std::vector<EntityId> processed_ids = merge_clause.process(std::move(entity_ids));
     std::vector<std::vector<EntityId>> vec;
     vec.emplace_back(std::move(processed_ids));
-    auto repartitioned = merge_clause.repartition(std::move(vec));
-    auto res = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(*component_manager, std::move(repartitioned->at(0)));
+    auto repartitioned = merge_clause.structure_for_processing(std::move(vec));
+    auto res = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(*component_manager, std::move(repartitioned.at(0)));
     ASSERT_TRUE(res.segments_.has_value());
     ASSERT_EQ(res.segments_->size(), 1u);
     ASSERT_EQ(*res.segments_->at(0), seg);
diff --git a/cpp/arcticdb/processing/test/test_resample.cpp b/cpp/arcticdb/processing/test/test_resample.cpp
index 1c8ebc87ed..c2bfa40043 100644
--- a/cpp/arcticdb/processing/test/test_resample.cpp
+++ b/cpp/arcticdb/processing/test/test_resample.cpp
@@ -34,7 +34,7 @@ TEST(Resample, StructureForProcessingBasic) {
     std::vector<RangesAndKey> ranges_and_keys{bottom, top};
 
     ResampleClause<ResampleBoundary::LEFT> resample_clause{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 1500, 2500, 2999}), 0};
-    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys, 0);
+    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 2);
     ASSERT_EQ(ranges_and_keys[0], top);
     ASSERT_EQ(ranges_and_keys[1], bottom);
@@ -62,7 +62,7 @@ TEST(Resample, StructureForProcessingColumnSlicing) {
     std::vector<RangesAndKey> ranges_and_keys{top_right, bottom_left, bottom_right, top_left};
 
     ResampleClause<ResampleBoundary::LEFT> resample_clause{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 1500, 2500, 2999}), 0};
-    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys, 0);
+    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 4);
     ASSERT_EQ(ranges_and_keys[0], top_left);
     ASSERT_EQ(ranges_and_keys[1], top_right);
@@ -87,7 +87,7 @@ TEST(Resample, StructureForProcessingOverlap) {
     std::vector<RangesAndKey> ranges_and_keys{bottom, top};
 
     ResampleClause<ResampleBoundary::LEFT> resample_clause{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 2500, 2999}), 0};
-    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys, 0);
+    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 2);
     ASSERT_EQ(ranges_and_keys[0], top);
     ASSERT_EQ(ranges_and_keys[1], bottom);
@@ -114,7 +114,7 @@ TEST(Resample, StructureForProcessingSubsumed) {
     std::vector<RangesAndKey> ranges_and_keys{bottom, middle, top};
 
     ResampleClause<ResampleBoundary::LEFT> resample_clause{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 4500}), 0};
-    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys, 0);
+    auto proc_unit_ids = resample_clause.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 3);
     ASSERT_EQ(ranges_and_keys[0], top);
     ASSERT_EQ(ranges_and_keys[1], middle);
@@ -139,7 +139,7 @@ TEST(Resample, StructureForProcessingExactBoundary) {
     std::vector<RangesAndKey> ranges_and_keys{bottom, top};
 
     ResampleClause<ResampleBoundary::LEFT> resample_clause_left{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 2000, 2500, 2999}), 0};
-    auto proc_unit_ids = resample_clause_left.structure_for_processing(ranges_and_keys, 0);
+    auto proc_unit_ids = resample_clause_left.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 2);
     ASSERT_EQ(ranges_and_keys[0], top);
     ASSERT_EQ(ranges_and_keys[1], bottom);
@@ -147,7 +147,7 @@ TEST(Resample, StructureForProcessingExactBoundary) {
     ASSERT_EQ(expected_proc_unit_ids_left, proc_unit_ids);
 
     ResampleClause<ResampleBoundary::RIGHT> resample_clause_right{"dummy", ResampleBoundary::LEFT, generate_bucket_boundaries({1, 500, 2000, 2500, 2999}), 0};
-    proc_unit_ids = resample_clause_right.structure_for_processing(ranges_and_keys, 0);
+    proc_unit_ids = resample_clause_right.structure_for_processing(ranges_and_keys);
     ASSERT_EQ(ranges_and_keys.size(), 2);
     ASSERT_EQ(ranges_and_keys[0], top);
     ASSERT_EQ(ranges_and_keys[1], bottom);
diff --git a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
index 95adc1fa26..d08e71ca9f 100644
--- a/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
+++ b/python/tests/integration/arcticdb/version_store/test_basic_version_store.py
@@ -593,6 +593,7 @@ def test_range_index(basic_store, sym):
     assert_frame_equal(expected, vit.data)
 
 
+@pytest.mark.pipeline
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_date_range(basic_store, use_date_range_clause):
     initial_timestamp = pd.Timestamp("2019-01-01")
@@ -640,6 +641,7 @@ def test_date_range(basic_store, use_date_range_clause):
     assert data_closed[data_closed.columns[0]][-1] == end_offset
 
 
+@pytest.mark.pipeline
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_date_range_none(basic_store, use_date_range_clause):
     sym = "date_test2"
@@ -658,6 +660,7 @@ def test_date_range_none(basic_store, use_date_range_clause):
     assert len(data) == rows
 
 
+@pytest.mark.pipeline
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_date_range_start_equals_end(basic_store, use_date_range_clause):
     sym = "date_test2"
@@ -679,6 +682,7 @@ def test_date_range_start_equals_end(basic_store, use_date_range_clause):
     assert data[data.columns[0]][0] == start_offset
 
 
+@pytest.mark.pipeline
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_date_range_row_sliced(basic_store_tiny_segment, use_date_range_clause):
     lib = basic_store_tiny_segment
@@ -2222,6 +2226,7 @@ def test_batch_append_with_throw_exception(basic_store, three_col_df):
         )
 
 
+@pytest.mark.pipeline
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_batch_read_date_range(basic_store_tombstone_and_sync_passive, use_date_range_clause):
     lmdb_version_store = basic_store_tombstone_and_sync_passive
@@ -2286,7 +2291,7 @@ def test_batch_read_row_range(lmdb_version_store_v1, use_row_range_clause):
         qbs = []
         for row_range in row_ranges:
             q = QueryBuilder()
-            q = q._row_range(row_range)
+            q = q.row_range(row_range)
             qbs.append(q)
         result_dict = lib.batch_read(symbols, query_builder=qbs)
     else:
diff --git a/python/tests/unit/arcticdb/test_column_stats.py b/python/tests/unit/arcticdb/test_column_stats.py
index 689d606fae..c4bdc0c93c 100644
--- a/python/tests/unit/arcticdb/test_column_stats.py
+++ b/python/tests/unit/arcticdb/test_column_stats.py
@@ -14,6 +14,9 @@
 from arcticdb_ext.version_store import NoSuchVersionException
 
 
+pytestmark = pytest.mark.pipeline
+
+
 df0 = pd.DataFrame(
     {"col_0": ["a", "b"], "col_1": [1, 2], "col_2": [6, 5]}, index=pd.date_range("2000-01-01", periods=2)
 )
diff --git a/python/tests/unit/arcticdb/version_store/test_aggregation.py b/python/tests/unit/arcticdb/version_store/test_aggregation.py
index 0dad79f295..f6b733fe67 100644
--- a/python/tests/unit/arcticdb/version_store/test_aggregation.py
+++ b/python/tests/unit/arcticdb/version_store/test_aggregation.py
@@ -15,6 +15,9 @@
 from arcticdb.util.test import assert_frame_equal, generic_aggregation_test, make_dynamic
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def test_group_on_float_column_with_nans(lmdb_version_store_v1):
     lib = lmdb_version_store_v1
     symbol = "test_group_on_float_column_with_nans"
diff --git a/python/tests/unit/arcticdb/version_store/test_aggregation_hypothesis.py b/python/tests/unit/arcticdb/version_store/test_aggregation_hypothesis.py
index 5cdd929ce3..6b26962678 100644
--- a/python/tests/unit/arcticdb/version_store/test_aggregation_hypothesis.py
+++ b/python/tests/unit/arcticdb/version_store/test_aggregation_hypothesis.py
@@ -7,6 +7,7 @@
 """
 import pandas as pd
 from hypothesis import assume, given, settings
+import pytest
 
 from arcticdb.util.test import generic_named_aggregation_test
 from arcticdb.util.hypothesis import (
@@ -18,6 +19,9 @@
 )
 
 
+pytestmark = pytest.mark.pipeline
+
+
 @use_of_function_scoped_fixtures_in_hypothesis_checked
 @settings(deadline=None)
 @given(
diff --git a/python/tests/unit/arcticdb/version_store/test_filtering.py b/python/tests/unit/arcticdb/version_store/test_filtering.py
index 4983748d64..6170b29f60 100644
--- a/python/tests/unit/arcticdb/version_store/test_filtering.py
+++ b/python/tests/unit/arcticdb/version_store/test_filtering.py
@@ -35,6 +35,9 @@
 from arcticdb.util._versions import IS_PANDAS_TWO, PANDAS_VERSION
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def test_filter_column_not_present(lmdb_version_store_v1):
     lib = lmdb_version_store_v1
     df = pd.DataFrame({"a": np.arange(2)}, index=np.arange(2))
diff --git a/python/tests/unit/arcticdb/version_store/test_filtering_hypothesis.py b/python/tests/unit/arcticdb/version_store/test_filtering_hypothesis.py
index 2e63a8fcde..bc10c28356 100644
--- a/python/tests/unit/arcticdb/version_store/test_filtering_hypothesis.py
+++ b/python/tests/unit/arcticdb/version_store/test_filtering_hypothesis.py
@@ -11,6 +11,7 @@
 import hypothesis.strategies as st
 import numpy as np
 import pandas as pd
+import pytest
 from pytz import timezone
 
 from arcticdb.version_store.processing import QueryBuilder
@@ -36,6 +37,9 @@
 )
 
 
+pytestmark = pytest.mark.pipeline
+
+
 @use_of_function_scoped_fixtures_in_hypothesis_checked
 @settings(deadline=None)
 @given(
diff --git a/python/tests/unit/arcticdb/version_store/test_head.py b/python/tests/unit/arcticdb/version_store/test_head.py
index 8c03596096..d207e3bce7 100644
--- a/python/tests/unit/arcticdb/version_store/test_head.py
+++ b/python/tests/unit/arcticdb/version_store/test_head.py
@@ -14,6 +14,9 @@
 from arcticdb_ext.exceptions import InternalException
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def generic_head_test(version_store, symbol, df, num_rows):
     version_store.write(symbol, df)
     assert np.array_equal(df.head(num_rows), version_store.head(symbol, num_rows).data)
diff --git a/python/tests/unit/arcticdb/version_store/test_lazy_dataframe.py b/python/tests/unit/arcticdb/version_store/test_lazy_dataframe.py
index 13c82a9dbf..c9d1ee7cbd 100644
--- a/python/tests/unit/arcticdb/version_store/test_lazy_dataframe.py
+++ b/python/tests/unit/arcticdb/version_store/test_lazy_dataframe.py
@@ -8,11 +8,15 @@
 import numpy as np
 import pandas as pd
 import pickle
+import pytest
 
 from arcticdb import col, LazyDataFrame, LazyDataFrameCollection, QueryBuilder, ReadRequest
 from arcticdb.util.test import assert_frame_equal
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def test_lazy_read(lmdb_library):
     lib = lmdb_library
     sym = "test_lazy_read"
diff --git a/python/tests/unit/arcticdb/version_store/test_projection.py b/python/tests/unit/arcticdb/version_store/test_projection.py
index f253a108de..f6f9f9d458 100644
--- a/python/tests/unit/arcticdb/version_store/test_projection.py
+++ b/python/tests/unit/arcticdb/version_store/test_projection.py
@@ -14,6 +14,9 @@
 from arcticdb.util.test import assert_frame_equal, make_dynamic, regularize_dataframe
 
 
+pytestmark = pytest.mark.pipeline
+
+
 @pytest.mark.parametrize("lib_type", ["lmdb_version_store_v1", "lmdb_version_store_dynamic_schema_v1"])
 def test_project_empty_dataframe(request, lib_type):
     lib = request.getfixturevalue(lib_type)
diff --git a/python/tests/unit/arcticdb/version_store/test_projection_hypothesis.py b/python/tests/unit/arcticdb/version_store/test_projection_hypothesis.py
index 860157d7d3..acda7f5a67 100644
--- a/python/tests/unit/arcticdb/version_store/test_projection_hypothesis.py
+++ b/python/tests/unit/arcticdb/version_store/test_projection_hypothesis.py
@@ -8,6 +8,7 @@
 from hypothesis import assume, given, settings
 import numpy as np
 import pandas as pd
+import pytest
 
 from arcticdb.version_store.processing import QueryBuilder
 from arcticdb.util.test import assert_frame_equal
@@ -21,6 +22,9 @@
 )
 
 
+pytestmark = pytest.mark.pipeline
+
+
 @use_of_function_scoped_fixtures_in_hypothesis_checked
 @settings(deadline=None)
 @given(
diff --git a/python/tests/unit/arcticdb/version_store/test_query_builder.py b/python/tests/unit/arcticdb/version_store/test_query_builder.py
index 91049f58df..dd550df539 100644
--- a/python/tests/unit/arcticdb/version_store/test_query_builder.py
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder.py
@@ -6,6 +6,7 @@
 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.
 """
 import copy
+from functools import partial
 import numpy as np
 import pandas as pd
 import pytest
@@ -16,6 +17,8 @@
 from arcticdb.version_store.processing import QueryBuilder
 from arcticdb.util.test import assert_frame_equal
 
+pytestmark = pytest.mark.pipeline
+
 
 def test_query_builder_equality_checks():
     q1 = QueryBuilder()
@@ -188,6 +191,97 @@ def test_reuse_querybuilder_date_range_batch(lmdb_version_store_tiny_segment):
     assert_frame_equal(expected_2, received_2)
 
 
+def test_querybuilder_filter_datetime_with_timezone(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "symbol"
+    def can_read_back(write_with_time, filter_with_time):
+        df = pd.DataFrame({"col": [write_with_time]})
+        lib.delete(symbol)
+        lib.write(symbol, df)
+
+        q = QueryBuilder()
+        q = q[q["col"] == filter_with_time]
+        read_df = lib.read(symbol, query_builder=q).data
+
+        return len(read_df) == 1
+
+    notz_winter_time = datetime.datetime(2024, 1, 1)
+    notz_summer_time = datetime.datetime(2024, 6, 1)
+    utc_time = datetime.datetime(2024, 6, 1, tzinfo=dateutil.tz.tzutc())
+    us_time = datetime.datetime(2024, 6, 1, tzinfo=dateutil.tz.gettz('America/New_York'))
+
+    # Reading back the same time should always succeed
+    assert can_read_back(notz_winter_time, notz_winter_time)
+    assert can_read_back(notz_summer_time, notz_summer_time)
+    assert can_read_back(utc_time, utc_time)
+    assert can_read_back(us_time, us_time)
+
+    # If tzinfo is not specified we assume UTC
+    assert can_read_back(notz_summer_time, utc_time)
+    assert can_read_back(utc_time, notz_summer_time)
+    assert not can_read_back(notz_summer_time, us_time)
+    assert not can_read_back(us_time, notz_summer_time)
+
+
+@pytest.mark.parametrize("batch", [True, False])
+@pytest.mark.parametrize("use_date_range_clause", [True, False])
+def test_querybuilder_date_range_then_date_range(lmdb_version_store_tiny_segment, batch, use_date_range_clause):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_date_range_then_date_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2000-01-01", periods=10))
+    lib.write(symbol, df)
+
+    first_date_range = (pd.Timestamp("2000-01-02"), pd.Timestamp("2000-01-09"))
+    second_date_range = (pd.Timestamp("2000-01-07"), pd.Timestamp("2000-01-08"))
+
+    q = QueryBuilder()
+    if use_date_range_clause:
+        q = q.date_range(first_date_range)
+    q = q.date_range(second_date_range)
+
+    if use_date_range_clause:
+        if batch:
+            received = lib.batch_read([symbol], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, query_builder=q).data
+    else:
+        if batch:
+            received = lib.batch_read([symbol], date_ranges=[first_date_range], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, date_range=first_date_range, query_builder=q).data
+    expected = df.query("col in [7, 8]")
+    assert_frame_equal(expected, received)
+
+
+@pytest.mark.parametrize("batch", [True, False])
+@pytest.mark.parametrize("use_date_range_clause", [True, False])
+def test_querybuilder_date_range_then_row_range(lmdb_version_store_tiny_segment, batch, use_date_range_clause):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_date_range_then_row_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2000-01-01", periods=10))
+    lib.write(symbol, df)
+
+    date_range = (pd.Timestamp("2000-01-02"), pd.Timestamp("2000-01-09"))
+
+    q = QueryBuilder()
+    if use_date_range_clause:
+        q = q.date_range(date_range)
+    q = q.row_range((1, 7))
+
+    if use_date_range_clause:
+        if batch:
+            received = lib.batch_read([symbol], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, query_builder=q).data
+    else:
+        if batch:
+            received = lib.batch_read([symbol], date_ranges=[date_range], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, date_range=date_range, query_builder=q).data
+    expected = df.iloc[2:8]
+    assert_frame_equal(expected, received)
+
+
 @pytest.mark.parametrize("batch", [True, False])
 @pytest.mark.parametrize("use_date_range_clause", [True, False])
 def test_querybuilder_date_range_then_filter(lmdb_version_store_tiny_segment, batch, use_date_range_clause):
@@ -219,36 +313,31 @@ def test_querybuilder_date_range_then_filter(lmdb_version_store_tiny_segment, ba
     assert_frame_equal(expected, received)
 
 
-def test_querybuilder_filter_datetime_with_timezone(lmdb_version_store_tiny_segment):
-    lib = lmdb_version_store_tiny_segment
-    symbol = "symbol"
-    def can_read_back(write_with_time, filter_with_time):
-        df = pd.DataFrame({"col": [write_with_time]})
-        lib.delete(symbol)
-        lib.write(symbol, df)
+def test_querybuilder_date_range_then_filter_then_resample(lmdb_version_store_tiny_segment):
+    # Pandas recommended way to resample and exclude buckets with no index values, which is our behaviour
+    # See https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#sparse-resampling
+    def round(t, freq):
+        freq = pd.tseries.frequencies.to_offset(freq)
+        td = pd.Timedelta(freq)
+        return pd.Timestamp((t.value // td.value) * td.value)
 
-        q = QueryBuilder()
-        q = q[q["col"] == filter_with_time]
-        read_df = lib.read(symbol, query_builder=q).data
-
-        return len(read_df) == 1
-
-    notz_winter_time = datetime.datetime(2024, 1, 1)
-    notz_summer_time = datetime.datetime(2024, 6, 1)
-    utc_time = datetime.datetime(2024, 6, 1, tzinfo=dateutil.tz.tzutc())
-    us_time = datetime.datetime(2024, 6, 1, tzinfo=dateutil.tz.gettz('America/New_York'))
-
-    # Reading back the same time should always succeed
-    assert can_read_back(notz_winter_time, notz_winter_time)
-    assert can_read_back(notz_summer_time, notz_summer_time)
-    assert can_read_back(utc_time, utc_time)
-    assert can_read_back(us_time, us_time)
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_date_range_then_filter_then_resample"
+    rng = np.random.default_rng()
+    df = pd.DataFrame(
+        {"filter_col": rng.integers(0, 2, 100), "agg_col": rng.integers(0, 1000, 100)},
+        index=pd.date_range("2000-01-01", periods=100, freq="h")
+    )
+    lib.write(symbol, df)
 
-    # If tzinfo is not specified we assume UTC
-    assert can_read_back(notz_summer_time, utc_time)
-    assert can_read_back(utc_time, notz_summer_time)
-    assert not can_read_back(notz_summer_time, us_time)
-    assert not can_read_back(us_time, notz_summer_time)
+    date_range=(pd.Timestamp("2000-01-02"), pd.Timestamp("2000-01-04"))
+    q = QueryBuilder()
+    q = q[q["filter_col"] == 0]
+    q = q.resample("3h").agg({"agg_col": "sum"})
+    received = lib.read(symbol, date_range=date_range, query_builder=q).data
+    expected = lib.read(symbol, date_range=date_range).data.query("filter_col == 0")
+    expected = expected.groupby(partial(round, freq="3h")).agg({"agg_col": "sum"})
+    assert_frame_equal(expected, received)
 
 
 @pytest.mark.parametrize("batch", [True, False])
@@ -334,7 +423,7 @@ def test_querybuilder_row_range(lmdb_version_store_tiny_segment, batch, use_row_
 
     q = QueryBuilder()
     if use_row_range_clause:
-        q = q._row_range(row_range)
+        q = q.row_range(row_range)
 
     if use_row_range_clause:
         if batch:
@@ -350,6 +439,65 @@ def test_querybuilder_row_range(lmdb_version_store_tiny_segment, batch, use_row_
     assert_frame_equal(expected, received)
 
 
+@pytest.mark.parametrize("batch", [True, False])
+@pytest.mark.parametrize("use_row_range_clause", [True, False])
+def test_querybuilder_row_range_then_date_range(lmdb_version_store_tiny_segment, batch, use_row_range_clause):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_row_range_then_date_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+
+    row_range = (3, 7)
+
+    q = QueryBuilder()
+    if use_row_range_clause:
+        q = q.row_range(row_range)
+    q = q.date_range((pd.Timestamp("2024-01-04"), pd.Timestamp("2024-01-06")))
+
+    if use_row_range_clause:
+        if batch:
+            received = lib.batch_read([symbol], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, query_builder=q).data
+    else:
+        if batch:
+            received = lib.batch_read([symbol], row_ranges=[row_range], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, row_range=row_range, query_builder=q).data
+    expected = df.query("col in [4, 5, 6]")
+    assert_frame_equal(expected, received)
+
+
+@pytest.mark.parametrize("batch", [True, False])
+@pytest.mark.parametrize("use_row_range_clause", [True, False])
+def test_querybuilder_row_range_then_row_range(lmdb_version_store_tiny_segment, batch, use_row_range_clause):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_row_range_then_row_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+
+    first_row_range = (3, 7)
+    second_row_range = (1, 3)
+
+    q = QueryBuilder()
+    if use_row_range_clause:
+        q = q.row_range(first_row_range)
+    q = q.row_range(second_row_range)
+
+    if use_row_range_clause:
+        if batch:
+            received = lib.batch_read([symbol], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, query_builder=q).data
+    else:
+        if batch:
+            received = lib.batch_read([symbol], row_ranges=[first_row_range], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, row_range=first_row_range, query_builder=q).data
+    expected = df.iloc[4:6]
+    assert_frame_equal(expected, received)
+
+
 @pytest.mark.parametrize("batch", [True, False])
 @pytest.mark.parametrize("use_row_range_clause", [True, False])
 def test_querybuilder_row_range_then_filter(lmdb_version_store_tiny_segment, batch, use_row_range_clause):
@@ -362,7 +510,7 @@ def test_querybuilder_row_range_then_filter(lmdb_version_store_tiny_segment, bat
 
     q = QueryBuilder()
     if use_row_range_clause:
-        q = q._row_range(row_range)
+        q = q.row_range(row_range)
     q = q[q["col1"].isin(0, 3, 6, 9)]
 
     if use_row_range_clause:
@@ -394,7 +542,7 @@ def test_querybuilder_row_range_then_project(lmdb_version_store_tiny_segment, ba
 
     q = QueryBuilder()
     if use_row_range_clause:
-        q = q._row_range(row_range)
+        q = q.row_range(row_range)
     q = q.apply("new_col", (q["col1"] * q["col2"]) + 13)
 
     if use_row_range_clause:
@@ -430,7 +578,7 @@ def test_querybuilder_row_range_then_groupby(lmdb_version_store_tiny_segment, ba
 
     q = QueryBuilder()
     if use_row_range_clause:
-        q = q._row_range(row_range)
+        q = q.row_range(row_range)
     q = q.groupby("col1").agg({"col2": "sum"})
 
     if use_row_range_clause:
@@ -450,6 +598,99 @@ def test_querybuilder_row_range_then_groupby(lmdb_version_store_tiny_segment, ba
     assert_frame_equal(expected, received)
 
 
+@pytest.mark.parametrize("batch", [True, False])
+@pytest.mark.parametrize("use_row_range_clause", [True, False])
+def test_querybuilder_row_range_then_resample(lmdb_version_store_tiny_segment, batch, use_row_range_clause):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_row_range_then_resample"
+    idx = [0, 1, 2, 1000, 1001, 1002]
+    idx = np.array(idx, dtype="datetime64[ns]")
+    df = pd.DataFrame({"col": np.arange(6)}, index=idx)
+    lib.write(symbol, df)
+
+    row_range = (1, 5)
+
+    q = QueryBuilder()
+    if use_row_range_clause:
+        q = q.row_range(row_range)
+    q = q.resample("us").agg({"col": "sum"})
+
+    if use_row_range_clause:
+        if batch:
+            received = lib.batch_read([symbol], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, query_builder=q).data
+    else:
+        if batch:
+            received = lib.batch_read([symbol], row_ranges=[row_range], query_builder=q)[symbol].data
+        else:
+            received = lib.read(symbol, row_range=row_range, query_builder=q).data
+    expected = df.query("col in [1, 2, 3, 4]")
+    expected = expected.resample("us").agg({"col": "sum"})
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_querybuilder_filter_then_date_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_date_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q[q["col"].isin(2, 3, 7)]
+    q = q.date_range((pd.Timestamp("2024-01-03"), pd.Timestamp("2024-01-08")))
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col in [3, 7]")
+    assert_frame_equal(expected, received)
+
+
+@pytest.mark.parametrize("n", range(-7, 8))
+def test_querybuilder_filter_then_head(lmdb_version_store_tiny_segment, n):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_head"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q[q["col"].isin(4, 5, 6, 8, 10)]
+    q = q.head(n)
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col in [4, 5, 6, 8, 10]")
+    expected = expected.head(n)
+    assert_frame_equal(expected, received)
+
+
+@pytest.mark.parametrize("n", range(-7, 8))
+def test_querybuilder_filter_then_tail(lmdb_version_store_tiny_segment, n):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_head"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q[q["col"].isin(4, 5, 6, 8, 10)]
+    q = q.tail(n)
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col in [4, 5, 6, 8, 10]")
+    expected = expected.tail(n)
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_filter_then_row_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_row_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q[q["col"].isin(4, 5, 6, 8, 10)]
+    q = q.row_range((1, 4))
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col in [4, 5, 6, 8, 10]")
+    expected = expected.iloc[1:4]
+    assert_frame_equal(expected, received)
+
+
 def test_querybuilder_filter_then_filter(lmdb_version_store_tiny_segment):
     lib = lmdb_version_store_tiny_segment
     symbol = "test_querybuilder_filter_then_filter"
@@ -500,6 +741,57 @@ def test_querybuilder_filter_then_groupby(lmdb_version_store_tiny_segment):
     assert_frame_equal(expected, received)
 
 
+def test_querybuilder_filter_then_resample(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_resample"
+    idx = [0, 1, 2, 1000, 1001, 1002]
+    idx = np.array(idx, dtype="datetime64[ns]")
+    df = pd.DataFrame({"col": np.arange(6)}, index=idx)
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q[(q["col"] != 1) & (q["col"] != 5)]
+    q = q.resample("us").agg({"col": "sum"})
+
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("(col != 1) & (col != 5)")
+    expected = expected.resample("us").agg({"col": "sum"})
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_querybuilder_project_then_date_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_date_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.apply("new_col", q["col"] * 3)
+    q = q.date_range((pd.Timestamp("2024-01-03"), pd.Timestamp("2024-01-08")))
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df
+    expected["new_col"] = expected["col"] * 3
+    expected = expected.query("col in [3, 4, 5, 6, 7, 8]")
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_querybuilder_project_then_row_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_row_range"
+    df = pd.DataFrame({"col": np.arange(1, 11)}, index=pd.date_range("2024-01-01", periods=10))
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.apply("new_col", q["col"] * 3)
+    q = q.row_range((3, 9))
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df
+    expected["new_col"] = expected["col"] * 3
+    expected = expected.iloc[3:9]
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
 def test_querybuilder_project_then_filter(lmdb_version_store_tiny_segment):
     lib = lmdb_version_store_tiny_segment
     symbol = "test_querybuilder_project_then_filter"
@@ -558,6 +850,26 @@ def test_querybuilder_project_then_groupby(lmdb_version_store_tiny_segment):
     assert_frame_equal(expected, received)
 
 
+def test_querybuilder_project_then_resample(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_resample"
+    idx = [0, 1, 2, 1000, 1001, 1002]
+    idx = np.array(idx, dtype="datetime64[ns]")
+    df = pd.DataFrame({"col": np.arange(6)}, index=idx)
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q.apply("new_col", q["col"] * 3)
+    q = q.resample("us").agg({"new_col": "sum"})
+
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df
+    expected["new_col"] = expected["col"] * 3
+    expected = expected.resample("us").agg({"new_col": "sum"})
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
 def test_querybuilder_groupby_then_filter(lmdb_version_store_tiny_segment):
     lib = lmdb_version_store_tiny_segment
     symbol = "test_querybuilder_groupby_then_filter"
@@ -613,6 +925,40 @@ def test_querybuilder_groupby_then_groupby(lmdb_version_store_tiny_segment):
     assert_frame_equal(expected, received)
 
 
+def test_querybuilder_resample_then_date_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_resample_then_date_range"
+    df = pd.DataFrame({"col": np.arange(30)}, index=pd.date_range("1970-01-01", periods=30, freq="D"))
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q.resample("2D").agg({"col": "sum"})
+    q = q.date_range((pd.Timestamp("1970-01-03"), pd.Timestamp("1970-01-27")))
+
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.resample("2D").agg({"col": "sum"})
+    expected = expected.iloc[1:-1]
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_querybuilder_resample_then_row_range(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_resample_then_row_range"
+    df = pd.DataFrame({"col": np.arange(30)}, index=pd.date_range("1970-01-01", periods=30, freq="D"))
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q.resample("2D").agg({"col": "sum"})
+    q = q.row_range((5, 8))
+
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.resample("2D").agg({"col": "sum"})
+    expected = expected.iloc[5:8]
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
 def test_querybuilder_resample_then_filter(lmdb_version_store_tiny_segment):
     lib = lmdb_version_store_tiny_segment
     symbol = "test_querybuilder_resample_then_filter"
@@ -675,3 +1021,52 @@ def test_querybuilder_resample_then_groupby(lmdb_version_store_tiny_segment):
     expected = df.resample("us").agg({"grouping_col": "sum", "agg_col": "sum"})
     expected = expected.groupby("grouping_col").agg({"agg_col": "sum"})
     assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_querybuilder_resample_then_resample(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_resample_then_resample"
+    df = pd.DataFrame(
+        {
+            "col": np.arange(240),
+        },
+        index=pd.date_range("2024-01-01", periods=240, freq="min")
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.resample("h").agg({"new_col": ("col", "mean")})
+    q = q.resample("2h").agg({"new_col": "mean"})
+    received = lib.read(symbol, query_builder=q).data
+    # Pandas 1.X needs None as the first argument to agg with named aggregators
+    expected = df.resample("h").agg(None, new_col=pd.NamedAgg("col", "mean"))
+    expected = expected.resample("2h").agg({"new_col": "mean"})
+    assert_frame_equal(expected, received, check_dtype=False)
+
+
+def test_query_builder_vwap(lmdb_version_store_v1):
+    lib = lmdb_version_store_v1
+    symbol = "test_query_builder_vwap"
+    rng = np.random.default_rng()
+    index = pd.date_range("2024-01-01", "2024-01-03", freq="min")
+    df = pd.DataFrame(
+        {
+            "price": rng.random(len(index)),
+            "volume": rng.integers(1, 100, len(index)),
+        },
+        index=index
+    )
+    lib.write(symbol, df)
+
+    date_range = (pd.Timestamp("2024-01-01T12:00:00"), pd.Timestamp("2024-01-02T12:00:00"))
+    freq = "h"
+    aggs = {"volume": "sum", "product": "sum"}
+    q = QueryBuilder()
+    q["product"] = q["price"] * q["volume"]
+    q = q.resample(freq).agg(aggs)
+    q["vwap"] = q["product"] / q["volume"]
+    received = lib.read(symbol, date_range=date_range, query_builder=q).data
+    expected = lib.read(symbol, date_range=date_range).data
+    expected["product"] = expected["price"] * expected["volume"]
+    expected = expected.resample(freq).agg(aggs)
+    expected["vwap"] = expected["product"] / expected["volume"]
+    assert_frame_equal(expected, received, check_dtype=False)
diff --git a/python/tests/unit/arcticdb/version_store/test_query_builder_batch.py b/python/tests/unit/arcticdb/version_store/test_query_builder_batch.py
index 5a3cd6ec1f..911d5929e4 100644
--- a/python/tests/unit/arcticdb/version_store/test_query_builder_batch.py
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder_batch.py
@@ -15,6 +15,9 @@
 from arcticdb_ext.exceptions import InternalException, StorageException, UserInputException
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def test_filter_batch_one_query(lmdb_version_store_v1):
     lib = lmdb_version_store_v1
     sym1 = "sym1"
diff --git a/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py b/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py
index 72db930ac3..7e1d83fa05 100644
--- a/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder_sparse.py
@@ -16,6 +16,9 @@
 from arcticdb.util.hypothesis import use_of_function_scoped_fixtures_in_hypothesis_checked
 
 
+pytestmark = pytest.mark.pipeline
+
+
 class TestQueryBuilderSparse:
     sym = "TestQueryBuilderSparse"
     df = None
diff --git a/python/tests/unit/arcticdb/version_store/test_resample.py b/python/tests/unit/arcticdb/version_store/test_resample.py
index 767d478300..1c910d13ca 100644
--- a/python/tests/unit/arcticdb/version_store/test_resample.py
+++ b/python/tests/unit/arcticdb/version_store/test_resample.py
@@ -17,6 +17,8 @@
 from packaging.version import Version
 from arcticdb.util._versions import IS_PANDAS_TWO, PANDAS_VERSION
 
+pytestmark = pytest.mark.pipeline
+
 
 ALL_AGGREGATIONS = ["sum", "mean", "min", "max", "first", "last", "count"]
 
diff --git a/python/tests/unit/arcticdb/version_store/test_row_range.py b/python/tests/unit/arcticdb/version_store/test_row_range.py
index 7e19e1f760..a41430214b 100644
--- a/python/tests/unit/arcticdb/version_store/test_row_range.py
+++ b/python/tests/unit/arcticdb/version_store/test_row_range.py
@@ -12,12 +12,15 @@
 from arcticdb_ext.exceptions import InternalException
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def generic_row_range_test(version_store, symbol, df, start_row, end_row):
     version_store.write(symbol, df)
 
     expected_array = df.iloc[start_row:end_row]
     received_array = version_store.read(symbol, row_range=(start_row, end_row)).data
-    q = QueryBuilder()._row_range((start_row, end_row))
+    q = QueryBuilder().row_range((start_row, end_row))
     received_array_via_querybuilder = version_store.read(symbol, query_builder=q).data
 
     np.testing.assert_array_equal(expected_array, received_array)
@@ -25,7 +28,7 @@ def generic_row_range_test(version_store, symbol, df, start_row, end_row):
 
     expected_array = df.iloc[-end_row:-start_row]
     received_array = version_store.read(symbol, row_range=(-end_row, -start_row)).data
-    q = QueryBuilder()._row_range((-end_row, -start_row))
+    q = QueryBuilder().row_range((-end_row, -start_row))
     received_array_via_querybuilder = version_store.read(symbol, query_builder=q).data
 
     np.testing.assert_array_equal(expected_array, received_array)
diff --git a/python/tests/unit/arcticdb/version_store/test_tail.py b/python/tests/unit/arcticdb/version_store/test_tail.py
index 58d9e47ce1..e13b160b98 100644
--- a/python/tests/unit/arcticdb/version_store/test_tail.py
+++ b/python/tests/unit/arcticdb/version_store/test_tail.py
@@ -14,6 +14,9 @@
 from arcticdb_ext.exceptions import InternalException
 
 
+pytestmark = pytest.mark.pipeline
+
+
 def generic_tail_test(version_store, symbol, df, num_rows):
     version_store.write(symbol, df)
     expected = df.tail(num_rows)