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

Issue:
Composite::empty() not equivalent to Composite::size() == 0
`Composite<T>::empty()` does not correctly handle the case where (possibly recursively) the `values_` vector contains another Composite<T> which is empty.
i.e. if the `values_` vector of the root of the composite contains a single unique pointer to another `Composite<T>`, and the child's `values_` vector is empty, then `empty()` will return `false`, but `size()` will return 0.
Refactor processing pipeline
- [x] Remove `ExecutionContext` in favour of keeping const information on clauses themselves
- [x] Add methods to `IClause` to retrieve the required input and created output columns
- [x] Make clauses directly constructable from Python layer, and remove `ClauseBuilder` class entirely
- [x] Ensure arbitrary chaining of all supported processing clauses works as expected (fixes #206 )


Discussion / hints:

> in favour of keeping const information on clauses themselves

I actually tried that in the past and encountered two folly poly limitations:
* An `IClause` can be easily copied, so has to be careful about the cost, uniqueness etc.
* Not all types can be members of a folly poly class

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index 1b06561f69..1989881853 100755
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -220,7 +220,7 @@ set(arcticdb_srcs
         processing/processing_segment.hpp
         processing/bucketizer.hpp
         processing/clause.hpp
-        processing/execution_context.hpp
+        processing/expression_context.hpp
         processing/expression_node.hpp
         storage/common.hpp
         storage/config_resolvers.hpp
diff --git a/cpp/arcticdb/async/async_store.hpp b/cpp/arcticdb/async/async_store.hpp
index 14d91e3c7b..6ffbe1c6b1 100644
--- a/cpp/arcticdb/async/async_store.hpp
+++ b/cpp/arcticdb/async/async_store.hpp
@@ -305,8 +305,7 @@ class AsyncStore : public Store {
 
     std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
         std::vector<Composite<pipelines::SliceAndKey>> &&slice_and_keys,
-        const std::shared_ptr<std::vector<Clause>>& query,
-        const StreamDescriptor& desc,
+        const std::vector<std::shared_ptr<Clause>>& clauses,
         const std::shared_ptr<std::unordered_set<std::string>>& filter_columns,
         const BatchReadArgs & args) override {
 
@@ -321,15 +320,15 @@ class AsyncStore : public Store {
                 batch.push_back(
                     async::submit_io_task(ReadCompressedSlicesTask(std::move(sk), library_))
                         .via(&async::cpu_executor())
-                        .thenValue(DecodeSlicesTask{desc, filter_columns})
-                        .thenValue(MemSegmentProcessingTask{shared_from_this(),query}));
+                        .thenValue(DecodeSlicesTask{filter_columns})
+                        .thenValue(MemSegmentProcessingTask{shared_from_this(), clauses}));
             }
             // IO option will execute all work in the same Folly thread potentially limiting context switches.
             else {
                 batch.push_back(
                     async::submit_io_task(ReadCompressedSlicesTask(std::move(sk), library_))
-                        .thenValue(DecodeSlicesTask{desc, filter_columns})
-                        .thenValue(MemSegmentProcessingTask{shared_from_this(),query}));
+                        .thenValue(DecodeSlicesTask{filter_columns})
+                        .thenValue(MemSegmentProcessingTask{shared_from_this(), clauses}));
             }
 
             if(++current_size == args.batch_size_) {
diff --git a/cpp/arcticdb/async/task_scheduler.hpp b/cpp/arcticdb/async/task_scheduler.hpp
index 5165794991..0cea307e45 100644
--- a/cpp/arcticdb/async/task_scheduler.hpp
+++ b/cpp/arcticdb/async/task_scheduler.hpp
@@ -235,6 +235,14 @@ class TaskScheduler {
         cpu_exec_.setNumThreads(cpu_thread_count_);
     }
 
+    size_t cpu_thread_count() const {
+        return cpu_thread_count_;
+    }
+
+    size_t io_thread_count() const {
+        return io_thread_count_;
+    }
+
 private:
     size_t cpu_thread_count_;
     size_t io_thread_count_;
diff --git a/cpp/arcticdb/async/tasks.hpp b/cpp/arcticdb/async/tasks.hpp
index 5bef472a61..3caa64d8b8 100644
--- a/cpp/arcticdb/async/tasks.hpp
+++ b/cpp/arcticdb/async/tasks.hpp
@@ -303,13 +303,10 @@ struct DecodeSegmentTask : BaseTask {
 struct DecodeSlicesTask : BaseTask {
     ARCTICDB_MOVE_ONLY_DEFAULT(DecodeSlicesTask)
 
-    StreamDescriptor desc_;
     std::shared_ptr<std::unordered_set<std::string>> filter_columns_;
 
     DecodeSlicesTask(
-            const StreamDescriptor& desc,
             const std::shared_ptr<std::unordered_set<std::string>>& filter_columns)  :
-                desc_(desc),
                 filter_columns_(filter_columns) {
             }
 
@@ -342,49 +339,18 @@ struct SegmentFunctionTask : BaseTask {
     }
 };
 
-// This class is used to restart the pipeline following a repartition
-struct MemSegmentPassthroughProcessingTask : BaseTask {
-    std::shared_ptr<Store> store_;
-    std::shared_ptr<std::vector<Clause>> clauses_;
-    std::optional<Composite<ProcessingSegment>> starting_segments_;
-
-    explicit MemSegmentPassthroughProcessingTask(
-            const std::shared_ptr<Store> store,
-            std::shared_ptr<std::vector<Clause>> clauses,
-            Composite<ProcessingSegment>&& starting_segments) :
-            store_(store),
-            clauses_(std::move(clauses)),
-            starting_segments_(std::move(starting_segments)){
-    }
-
-    [[nodiscard]]
-    Composite<ProcessingSegment> process(Composite<ProcessingSegment>&& proc){
-        auto procs = std::move(proc);
-        for(const auto& clause : *clauses_) {
-            if(clause.requires_repartition())
-                break;
-
-            procs = clause.process(store_, std::move(procs));
-        }
-        return procs;
-    }
-
-    ARCTICDB_MOVE_ONLY_DEFAULT(MemSegmentPassthroughProcessingTask)
-
-    Composite<ProcessingSegment> operator()() {
-        return process(std::move(starting_segments_.value()));
-    }
-};
-
 struct MemSegmentProcessingTask : BaseTask {
     std::shared_ptr<Store> store_;
-    std::shared_ptr<std::vector<Clause>> clauses_;
+    std::vector<std::shared_ptr<Clause>> clauses_;
+    std::optional<Composite<ProcessingSegment>> procs_;
 
     explicit MemSegmentProcessingTask(
            const std::shared_ptr<Store> store,
-           std::shared_ptr<std::vector<Clause>> clauses) :
+           std::vector<std::shared_ptr<Clause>> clauses,
+           std::optional<Composite<ProcessingSegment>>&& procs = std::nullopt) :
         store_(store),
-        clauses_(std::move(clauses)) {
+        clauses_(std::move(clauses)),
+        procs_(std::move(procs)) {
     }
 
     ProcessingSegment slice_to_segment(Composite<pipelines::SliceAndKey>&& is) {
@@ -395,10 +361,10 @@ struct MemSegmentProcessingTask : BaseTask {
     [[nodiscard]]
     Composite<ProcessingSegment> process(Composite<ProcessingSegment>&& proc){
         auto procs = std::move(proc);
-        for(const auto& clause : *clauses_) {
-            procs = clause.process(store_, std::move(procs));
+        for(const auto& clause : clauses_) {
+            procs = clause->process(store_, std::move(procs));
 
-            if(clause.requires_repartition())
+            if(clause->clause_info().requires_repartition_)
                 break;
         }
         return procs;
@@ -410,6 +376,13 @@ struct MemSegmentProcessingTask : BaseTask {
         return process(Composite<ProcessingSegment>(slice_to_segment(std::move(sk))));
     }
 
+    Composite<ProcessingSegment> operator()() {
+        internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+                procs_.has_value(),
+                "MemSegmentProcessingTask () operator expects processing segments to be provided in ctor");
+        return process(std::move(*procs_));
+    }
+
 };
 
 struct MemSegmentFunctionTask : BaseTask {
diff --git a/cpp/arcticdb/column_store/memory_segment_impl.hpp b/cpp/arcticdb/column_store/memory_segment_impl.hpp
index fa62bad921..c4db9c9eb1 100644
--- a/cpp/arcticdb/column_store/memory_segment_impl.hpp
+++ b/cpp/arcticdb/column_store/memory_segment_impl.hpp
@@ -417,7 +417,7 @@ class SegmentInMemoryImpl {
         columns_.reserve(descriptor_->field_count());
         for (size_t i = old_size; i < size_t(descriptor_->field_count()); ++i) {
             auto type = type_desc_from_proto(descriptor_->fields(i).type_desc());
-            util::check(type.data_type() != DataType::UNKNOWN, "Can't create column with unknown data type");
+            util::check(type.data_type() != DataType::UNKNOWN, "Can't create column in create_columns with unknown data type");
             columns_.emplace_back(
                 std::make_shared<Column>(type_desc_from_proto(descriptor_->fields(i).type_desc()), expected_column_size, presize, allow_sparse));
         }
diff --git a/cpp/arcticdb/entity/types.hpp b/cpp/arcticdb/entity/types.hpp
index d9bf29ef76..63b70a88ce 100644
--- a/cpp/arcticdb/entity/types.hpp
+++ b/cpp/arcticdb/entity/types.hpp
@@ -742,7 +742,7 @@ struct StreamDescriptor {
     }
 
     const FieldDescriptor::Proto& add_field(const FieldDescriptor::Proto& field) {
-        util::check(type_desc_from_proto(field.type_desc()).data_type() != DataType::UNKNOWN, "Can't create column with unknown data type");
+        util::check(type_desc_from_proto(field.type_desc()).data_type() != DataType::UNKNOWN, "Can't create column in add_field with unknown data type");
         auto new_field = data_->mutable_fields()->Add();
         new_field->CopyFrom(field);
         return *new_field;
diff --git a/cpp/arcticdb/pipeline/column_stats.cpp b/cpp/arcticdb/pipeline/column_stats.cpp
index 5e0137c1a1..0e4f944d47 100644
--- a/cpp/arcticdb/pipeline/column_stats.cpp
+++ b/cpp/arcticdb/pipeline/column_stats.cpp
@@ -2,7 +2,6 @@
 
 #include <arcticdb/processing/aggregation_interface.hpp>
 #include <arcticdb/processing/aggregation.hpp>
-#include <arcticdb/processing/execution_context.hpp>
 #include <arcticdb/entity/type_utils.hpp>
 #include <arcticdb/util/preconditions.hpp>
 
@@ -223,24 +222,26 @@ std::optional<Clause> ColumnStats::clause() const {
     if (column_stats_.empty()) {
         return std::nullopt;
     }
-    auto execution_context = std::make_shared<ExecutionContext>();
+    std::unordered_set<std::string> input_columns;
     auto index_generation_aggregators = std::make_shared<std::vector<ColumnStatsAggregator>>();
     for (const auto& [column, column_stat_types]: column_stats_) {
-        execution_context->columns_.emplace(column);
+        input_columns.emplace(column);
 
         for (const auto& column_stat_type: column_stat_types) {
             switch (column_stat_type) {
                 case ColumnStatType::MINMAX:
-                    index_generation_aggregators->emplace_back(MinMaxAggregator(ColumnName(column),
-                                                                               ColumnName(to_segment_column_name(column, ColumnStatTypeInternal::MIN)),
-                                                                               ColumnName(to_segment_column_name(column, ColumnStatTypeInternal::MAX))));
+                    index_generation_aggregators->emplace_back(
+                            MinMaxAggregator(ColumnName(column),
+                                             ColumnName(to_segment_column_name(column, ColumnStatTypeInternal::MIN)),
+                                             ColumnName(to_segment_column_name(column, ColumnStatTypeInternal::MAX)))
+                                             );
                     break;
                 default:
                     internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unrecognised ColumnStatType");
             }
         }
     }
-    return ColumnStatsGenerationClause(execution_context, index_generation_aggregators);
+    return ColumnStatsGenerationClause(std::move(input_columns), index_generation_aggregators);
 }
 
 bool ColumnStats::operator==(const ColumnStats& right) const {
diff --git a/cpp/arcticdb/pipeline/query.hpp b/cpp/arcticdb/pipeline/query.hpp
index 62bb6212a3..2fb700f273 100644
--- a/cpp/arcticdb/pipeline/query.hpp
+++ b/cpp/arcticdb/pipeline/query.hpp
@@ -9,7 +9,7 @@
 
 #include <arcticdb/util/bitset.hpp>
 #include <arcticdb/entity/index_range.hpp>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/entity/versioned_item.hpp>
 #include <arcticdb/pipeline/python_output_frame.hpp>
 #include <arcticdb/pipeline/write_frame.hpp>
@@ -50,19 +50,16 @@ struct ReadQuery {
     mutable std::vector<std::string> columns; // empty <=> all columns
     std::variant<std::monostate, HeadRange, TailRange, SignedRowRange> row_range;
     FilterRange row_filter; // no filter by default
-    std::shared_ptr<std::vector<Clause>> query_ = std::make_shared<std::vector<Clause>>();
+    std::vector<std::shared_ptr<Clause>> clauses_;
 
     ReadQuery() = default;
 
-    explicit ReadQuery(std::vector<Clause>&& query) {
-        query_ = std::make_shared<std::vector<Clause>>(std::move(query));
+    explicit ReadQuery(std::vector<std::shared_ptr<Clause>>&& clauses):
+            clauses_(std::move(clauses)) {
     }
 
-    void set_clause_builder(ClauseBuilder& builder) {
-        ClauseBuilder b = std::move(builder);
-        for (auto&& c: b.get_clauses()) {
-            query_->emplace_back(std::move(c));
-        }
+    void add_clauses(std::vector<std::shared_ptr<Clause>>& clauses) {
+        clauses_ = clauses;
     }
 
     void calculate_row_filter(int64_t total_rows) {
@@ -73,9 +70,9 @@ struct ReadQuery {
             } else {
                 row_filter = RowRange(0, std::max(static_cast<int64_t>(0), total_rows + head_range.num_rows_));
             }
-            if (query_->empty() && columns.empty() && head_range.num_rows_ > 0) {
+            if (clauses_.empty() && columns.empty() && head_range.num_rows_ > 0) {
                 // TODO: columns aren't supported due to AN-334
-                query_->emplace_back(RowNumberLimitClause{static_cast<size_t>(head_range.num_rows_)});
+                clauses_.emplace_back(std::make_shared<Clause>(RowNumberLimitClause{static_cast<size_t>(head_range.num_rows_)}));
             } else {
                 log::version().info("Arguments not compatible with head() memory usage optimisation");
             }
diff --git a/cpp/arcticdb/pipeline/read_pipeline.hpp b/cpp/arcticdb/pipeline/read_pipeline.hpp
index 69d0dc6b5c..26af284726 100644
--- a/cpp/arcticdb/pipeline/read_pipeline.hpp
+++ b/cpp/arcticdb/pipeline/read_pipeline.hpp
@@ -128,34 +128,33 @@ inline std::optional<util::BitSet> requested_column_bitset_including_index(const
 }
 
 inline std::optional<util::BitSet> clause_column_bitset(const StreamDescriptor::Proto& desc,
-                                                        const std::shared_ptr<std::vector<Clause>>& clauses) {
-    std::optional<util::BitSet> res;
-    if (clauses) {
-        folly::F14FastSet<std::string_view> column_set;
-        for (const auto& clause: *clauses) {
-            std::shared_ptr<ExecutionContext> execution_context = clause.execution_context();
-            if (execution_context) {
-                for (const auto& column: execution_context->columns_) {
-                    column_set.insert(std::string_view(column));
-                }
+                                                        const std::vector<std::shared_ptr<Clause>>& clauses) {
+    folly::F14FastSet<std::string_view> column_set;
+    for (const auto& clause: clauses) {
+        auto opt_columns = clause->clause_info().input_columns_;
+        if (opt_columns.has_value()) {
+            for (const auto& column: *clause->clause_info().input_columns_) {
+                column_set.insert(std::string_view(column));
             }
         }
-        // If only needed because the hacky clause for head memory usage doesn't specify columns
-        if (!column_set.empty()) {
-            res = build_column_bitset(desc, column_set);
-        }
     }
-    return res;
+    // If only needed because the hacky clause for head memory usage doesn't specify columns
+    if (!column_set.empty()) {
+        return build_column_bitset(desc, column_set);
+    } else {
+        return std::nullopt;
+    }
 }
 
 // Returns std::nullopt if all columns are required, which is the case if requested_columns is std::nullopt
 // Otherwise augment the requested_columns bitset with columns that are required by any of the clauses
 inline std::optional<util::BitSet> overall_column_bitset(const StreamDescriptor::Proto& desc,
-                                                         const std::shared_ptr<std::vector<Clause>>& clauses,
+                                                         const std::vector<std::shared_ptr<Clause>>& clauses,
                                                          const std::optional<util::BitSet>& requested_columns) {
     // std::all_of returns true if the range is empty
-    auto clauses_can_combine_with_column_selection = std::all_of(clauses->begin(), clauses->end(), [](const Clause& clause){
-        return clause.can_combine_with_column_selection();
+    auto clauses_can_combine_with_column_selection = std::all_of(clauses.begin(), clauses.end(),
+                                                                 [](const std::shared_ptr<Clause>& clause){
+        return clause->clause_info().can_combine_with_column_selection_;
     });
     user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(
             !requested_columns.has_value() || clauses_can_combine_with_column_selection,
@@ -209,7 +208,7 @@ inline std::vector<FilterQuery<ContainerType>> get_column_bitset_and_query_funct
     if(!dynamic_schema || column_groups) {
         pipeline_context->set_selected_columns(query.columns);
         pipeline_context->overall_column_bitset_ = overall_column_bitset(pipeline_context->descriptor().proto(),
-                                                                         query.query_,
+                                                                         query.clauses_,
                                                                          pipeline_context->selected_columns_);
     }
     return build_read_query_filters<ContainerType>(pipeline_context->overall_column_bitset_, pipeline_context, query.row_filter, dynamic_schema, column_groups);
diff --git a/cpp/arcticdb/processing/aggregation.cpp b/cpp/arcticdb/processing/aggregation.cpp
index f5722ad9c0..ba63158b70 100644
--- a/cpp/arcticdb/processing/aggregation.cpp
+++ b/cpp/arcticdb/processing/aggregation.cpp
@@ -10,6 +10,20 @@
 
 namespace arcticdb {
 
+void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type) {
+    if (current_data_type.has_value()) {
+        auto common_type = has_valid_common_type(entity::TypeDescriptor(*current_data_type, 0),
+                                                 entity::TypeDescriptor(data_type, 0));
+        schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+                common_type.has_value(),
+                "Cannot perform aggregation on column, incompatible types present: {} and {}",
+                entity::TypeDescriptor(*current_data_type, 0), entity::TypeDescriptor(data_type, 0));
+        current_data_type = common_type->data_type();
+    } else {
+        current_data_type = data_type;
+    }
+}
+
 void MinMaxAggregatorData::aggregate(const ColumnWithStrings& input_column) {
     entity::details::visit_type(input_column.column_->type().data_type(), [&input_column, that=this] (auto type_desc_tag) {
         using InputType = decltype(type_desc_tag);
@@ -59,24 +73,26 @@ SegmentInMemory MinMaxAggregatorData::finalize(const std::vector<ColumnName>& ou
     return seg;
 }
 
-void Sum::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
-    entity::details::visit_type(data_type_, [&input_column, unique_values, &groups, that=this] (auto global_type_desc_tag) {
+void SumAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
+    // If data_type_ has no value, it means there is no data for this aggregation
+    // For sums, we want this to display as zero rather than NaN
+    data_type_ = data_type_.value_or(DataType::FLOAT64);
+    entity::details::visit_type(*data_type_, [&input_column, unique_values, &groups, that=this] (auto global_type_desc_tag) {
         using GlobalInputType = decltype(global_type_desc_tag);
         if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
             using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
             using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
             that->aggregated_.resize(sizeof(GlobalRawType)* unique_values);
             auto out_ptr = reinterpret_cast<GlobalRawType*>(that->aggregated_.data());
-            if(input_column.has_value()) {
+            if (input_column.has_value()) {
                 entity::details::visit_type(input_column->column_->type().data_type(), [&input_column, &groups, &out_ptr] (auto type_desc_tag) {
-                    using InputType = decltype(type_desc_tag);
-                    if constexpr(!is_sequence_type(InputType::DataTypeTag::data_type)) {
-                        using TypeDescriptorTag =  typename OutputType<InputType>::type;
-                        using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
+                    using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
+                    using ColumnType =  typename ColumnTagType::raw_type;
+                    if constexpr(!is_sequence_type(ColumnTagType::data_type)) {
                         auto col_data = input_column->column_->data();
                         auto groups_pos = 0;
-                        while (auto block = col_data.next<TypeDescriptorTag>()) {
-                            auto ptr = reinterpret_cast<const RawType *>(block.value().data());
+                        while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
+                            auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
                             for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr) {
                                 out_ptr[groups[groups_pos++]] += GlobalRawType(*ptr);
                             }
@@ -90,117 +106,22 @@ void Sum::aggregate(const std::optional<ColumnWithStrings>& input_column, const
     });
 }
 
-std::optional<DataType> Sum::finalize(SegmentInMemory& seg, bool, size_t unique_values) {
+SegmentInMemory SumAggregatorData::finalize(const ColumnName& output_column_name, bool, size_t unique_values) {
+    SegmentInMemory res;
     if(!aggregated_.empty()) {
-        entity::details::visit_type(data_type_, [that=this, &seg, unique_values] (auto type_desc_tag) {
+        entity::details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto type_desc_tag) {
             using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
             that->aggregated_.resize(sizeof(RawType)* unique_values);
-            auto col = std::make_shared<Column>(make_scalar_type(that->data_type_), unique_values, true, false);
+            auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
             memcpy(col->ptr(), that->aggregated_.data(), that->aggregated_.size());
-            seg.add_column(scalar_field_proto(that->data_type_, that->get_output_column_name().value), col);
+            res.add_column(scalar_field_proto(that->data_type_.value(), output_column_name.value), col);
             col->set_row_data(unique_values - 1);
         });
-        return data_type_;
-    } else {
-        return std::nullopt;
     }
+    return res;
 }
 
-void MaxOrMin::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
-    if(input_column.has_value()) {
-        entity::details::visit_type(data_type_, [&input_column, unique_values, &groups, that=this] (auto global_type_desc_tag) {
-            using GlobalInputType = decltype(global_type_desc_tag);
-            if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
-                using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
-                using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
-                auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<GlobalRawType>);
-                that->aggregated_.resize(sizeof(MaybeValue<GlobalRawType>) * unique_values);
-                auto col_data = input_column->column_->data();
-                auto out_ptr = reinterpret_cast<MaybeValue<GlobalRawType>*>(that->aggregated_.data());
-                std::fill(out_ptr + prev_size, out_ptr + unique_values, MaybeValue<GlobalRawType>{});
-                entity::details::visit_type(input_column->column_->type().data_type(), [&groups, &out_ptr, &col_data, that=that] (auto type_desc_tag) {
-                    using InputType = decltype(type_desc_tag);
-                    if constexpr(!is_sequence_type(InputType::DataTypeTag::data_type)) {
-                        using TypeDescriptorTag =  typename OutputType<InputType>::type;
-                        using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
-                        auto groups_pos = 0;
-                        while (auto block = col_data.next<TypeDescriptorTag>()) {
-                            auto ptr = reinterpret_cast<const RawType *>(block.value().data());
-                            for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr) {
-                                auto& val = out_ptr[groups[groups_pos]];
-                                if constexpr(std::is_floating_point_v<RawType>) {
-                                    const auto& curr = GlobalRawType(*ptr);
-                                    if (!val.written_ || std::isnan(static_cast<RawType>(val.value_))) {
-                                        val.value_ = curr;
-                                        val.written_ = true;
-                                    } else if (!std::isnan(static_cast<RawType>(curr))) {
-                                        if (that->extremum_ == Extremum::max) {
-                                            val.value_ = std::max(val.value_, curr);
-                                        } else {
-                                            val.value_ = std::min(val.value_, curr);
-                                        }
-                                    }
-                                } else {
-                                    if (that->extremum_ == Extremum::max) {
-                                        val.value_ = std::max(val.value_, GlobalRawType(*ptr));
-                                    } else {
-                                        // If unwritten, MaybeValue begins life with MIN_INT. That's always going to "win" when
-                                        // passed into `std::min`!
-                                        val.value_ = !val.written_ ? GlobalRawType(*ptr) : std::min(val.value_, GlobalRawType(*ptr));
-                                    }
-                                    val.written_ = true;
-                                }
-                                ++groups_pos;
-                            }
-                        }
-                    } else {
-                        util::raise_rte("String aggregations not currently supported");
-                    }
-                });
-            }
-        });
-    }
-}
-
-std::optional<DataType> MaxOrMin::finalize(SegmentInMemory& seg, bool dynamic_schema, size_t unique_values) {
-    if(!aggregated_.empty()) {
-        if(dynamic_schema) {
-            entity::details::visit_type(data_type_, [that=this, &seg, unique_values] (auto type_desc_tag) {
-                using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
-                auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<RawType>);
-                auto new_size = sizeof(MaybeValue<RawType>) * unique_values;
-                that->aggregated_.resize(new_size);
-                auto in_ptr =  reinterpret_cast<MaybeValue<RawType>*>(that->aggregated_.data());
-                std::fill(in_ptr + prev_size, in_ptr + unique_values, MaybeValue<RawType>{});
-                auto col = std::make_shared<Column>(make_scalar_type(DataType::FLOAT64), unique_values, true, false);
-                auto out_ptr = reinterpret_cast<double*>(col->ptr());
-                for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
-                    *out_ptr = in_ptr->written_ ? static_cast<double>(in_ptr->value_) : std::numeric_limits<double>::quiet_NaN();                }
-
-                col->set_row_data(unique_values - 1);
-                seg.add_column(scalar_field_proto(DataType::FLOAT64, that->get_output_column_name().value), col);
-            });
-            return DataType::FLOAT64;
-        } else {
-            entity::details::visit_type(data_type_, [that=this, &seg, unique_values] (auto type_desc_tag) {
-                using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
-                auto col = std::make_shared<Column>(make_scalar_type(that->data_type_), unique_values, true, false);
-                const auto* in_ptr =  reinterpret_cast<const MaybeValue<RawType>*>(that->aggregated_.data());
-                auto out_ptr = reinterpret_cast<RawType*>(col->ptr());
-                for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
-                    *out_ptr = in_ptr->value_;
-                }
-                col->set_row_data(unique_values - 1);
-                seg.add_column(scalar_field_proto(that->data_type_, that->get_output_column_name().value), col);
-            });
-            return data_type_;
-        }
-    } else {
-        return std::nullopt;
-    }
-}
-
-void Mean::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
+void MeanAggregatorData::aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
     if(input_column.has_value()) {
         input_column->column_->type().visit_tag([&] (auto type_desc_tag) {
             using TypeDescriptorTag =  decltype(type_desc_tag);
@@ -222,21 +143,20 @@ void Mean::aggregate(const std::optional<ColumnWithStrings>& input_column, const
     }
 }
 
-std::optional<DataType> Mean::finalize(SegmentInMemory& seg, bool, size_t unique_values) {
+SegmentInMemory MeanAggregatorData::finalize(const ColumnName& output_column_name,  bool, size_t unique_values) {
+    SegmentInMemory res;
     if(!data_.fractions_.empty()) {
         data_.fractions_.resize(unique_values);
-        auto grouping_desc = scalar_field_proto(arcticdb::entity::DataType::FLOAT64, get_output_column_name().value);
-        auto pos = seg.add_column(grouping_desc, data_.fractions_.size(), true);
-        auto& column = seg.column(pos);
+        auto grouping_desc = scalar_field_proto(arcticdb::entity::DataType::FLOAT64, output_column_name.value);
+        auto pos = res.add_column(grouping_desc, data_.fractions_.size(), true);
+        auto& column = res.column(pos);
         auto ptr = reinterpret_cast<double*>(column.ptr());
         column.set_row_data(data_.fractions_.size() - 1);
 
         for (const auto& fraction : folly::enumerate(data_.fractions_)) {
             ptr[fraction.index] = fraction->to_double();
         }
-        return arcticdb::entity::DataType::FLOAT64;
-    } else {
-        return std::nullopt;
     }
+    return res;
 }
 } //namespace arcticdb
diff --git a/cpp/arcticdb/processing/aggregation.hpp b/cpp/arcticdb/processing/aggregation.hpp
index 016485df33..b429cc7f0b 100644
--- a/cpp/arcticdb/processing/aggregation.hpp
+++ b/cpp/arcticdb/processing/aggregation.hpp
@@ -11,6 +11,7 @@
 #include <arcticdb/processing/processing_segment.hpp>
 #include <arcticdb/processing/bucketizer.hpp>
 #include <arcticdb/entity/types.hpp>
+#include <arcticdb/entity/type_utils.hpp>
 #include <arcticdb/pipeline/value.hpp>
 #include <arcticdb/pipeline/value_set.hpp>
 #include <arcticdb/processing/aggregation_interface.hpp>
@@ -47,9 +48,15 @@ struct OutputType<DataTypeTag<DataType::MICROS_UTC64>, void> {
     using type = ScalarTagType<DataTypeTag<DataType::MICROS_UTC64>>;
 };
 
+void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type);
+
 struct MinMaxAggregatorData {
     std::optional<Value> min_;
     std::optional<Value> max_;
+
+    MinMaxAggregatorData() = default;
+    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregatorData)
+
     void aggregate(const ColumnWithStrings& input_column);
     SegmentInMemory finalize(const std::vector<ColumnName>& output_column_names) const;
 };
@@ -59,39 +66,57 @@ struct MinMaxAggregator {
     ColumnName output_column_name_min_;
     ColumnName output_column_name_max_;
 
-    MinMaxAggregator(ColumnName column_name, ColumnName output_column_name_min, ColumnName output_column_name_max) :
+    explicit MinMaxAggregator(ColumnName column_name, ColumnName output_column_name_min, ColumnName output_column_name_max) :
             column_name_(std::move(column_name)),
             output_column_name_min_(std::move(output_column_name_min)),
             output_column_name_max_(std::move(output_column_name_max)){
     }
+    ARCTICDB_MOVE_COPY_DEFAULT(MinMaxAggregator)
+
     [[nodiscard]] ColumnName get_input_column_name() const { return column_name_; }
     [[nodiscard]] std::vector<ColumnName> get_output_column_names() const { return {output_column_name_min_, output_column_name_max_}; }
     [[nodiscard]] ColumnStatsAggregatorData get_aggregator_data() const { return MinMaxAggregatorData(); }
 };
 
-struct Sum {
+struct SumAggregatorData {
     std::vector<uint8_t> aggregated_;
+    std::optional<DataType> data_type_;
+
+    SumAggregatorData() = default;
+    // Warn on copies as aggregated_ could be a large buffer
+    SumAggregatorData(const SumAggregatorData& other):
+    aggregated_(other.aggregated_),
+    data_type_(other.data_type_) {
+        log::version().warn("Copying potentially large buffer in SumAggregatorData");
+    }
+    SumAggregatorData& operator=(const SumAggregatorData& other) {
+        aggregated_ = other.aggregated_;
+        data_type_ = other.data_type_;
+        log::version().warn("Copying potentially large buffer in SumAggregatorData");
+        return *this;
+    }
+    ARCTICDB_MOVE(SumAggregatorData)
+
+    void add_data_type(DataType data_type) {
+        add_data_type_impl(data_type, data_type_);
+    }
+    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
+    SegmentInMemory finalize(const ColumnName& output_column_name,  bool dynamic_schema, size_t unique_values);
+};
+
+struct SumAggregator {
     ColumnName input_column_name_;
     ColumnName output_column_name_;
-    DataType data_type_ = {};
 
-    Sum(ColumnName input_column_name, ColumnName output_column_name) :
+    explicit SumAggregator(ColumnName input_column_name, ColumnName output_column_name) :
         input_column_name_(std::move(input_column_name)),
         output_column_name_(std::move(output_column_name)) {
     }
-
-    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
-
-    std::optional<DataType> finalize(SegmentInMemory& seg, bool dynamic_schema, size_t unique_values);
+    ARCTICDB_MOVE_COPY_DEFAULT(SumAggregator)
 
     [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-
     [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-
-    [[nodiscard]] Sum construct() const { return {input_column_name_, output_column_name_}; }
-
-    void set_data_type(DataType data_type) { data_type_ = data_type; }
-
+    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return SumAggregatorData(); }
 };
 
 template <typename T>
@@ -101,60 +126,185 @@ struct MaybeValue {
 };
 
 enum class Extremum {
-    max, min
+    MAX,
+    MIN
 };
 
-struct MaxOrMin {
+template<Extremum T>
+struct MaxOrMinAggregatorData {
     std::vector<uint8_t> aggregated_;
+    std::optional<DataType> data_type_;
+
+    MaxOrMinAggregatorData() = default;
+    // Warn on copies as aggregated_ could be a large buffer
+    MaxOrMinAggregatorData(const MaxOrMinAggregatorData& other):
+            aggregated_(other.aggregated_),
+            data_type_(other.data_type_) {
+        log::version().warn("Copying potentially large buffer in MaxOrMinAggregatorData");
+    }
+    MaxOrMinAggregatorData& operator=(const MaxOrMinAggregatorData& other) {
+        aggregated_ = other.aggregated_;
+        data_type_ = other.data_type_;
+        log::version().warn("Copying potentially large buffer in MaxOrMinAggregatorData");
+        return *this;
+    }
+    ARCTICDB_MOVE(MaxOrMinAggregatorData)
+
+    void add_data_type(DataType data_type) {
+        add_data_type_impl(data_type, data_type_);
+    }
+    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
+        if(data_type_.has_value() && input_column.has_value()) {
+            entity::details::visit_type(*data_type_, [&input_column, unique_values, &groups, that=this] (auto global_type_desc_tag) {
+                using GlobalInputType = decltype(global_type_desc_tag);
+                if constexpr(!is_sequence_type(GlobalInputType::DataTypeTag::data_type)) {
+                    using GlobalTypeDescriptorTag =  typename OutputType<GlobalInputType>::type;
+                    using GlobalRawType = typename GlobalTypeDescriptorTag::DataTypeTag::raw_type;
+                    auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<GlobalRawType>);
+                    that->aggregated_.resize(sizeof(MaybeValue<GlobalRawType>) * unique_values);
+                    auto col_data = input_column->column_->data();
+                    auto out_ptr = reinterpret_cast<MaybeValue<GlobalRawType>*>(that->aggregated_.data());
+                    std::fill(out_ptr + prev_size, out_ptr + unique_values, MaybeValue<GlobalRawType>{});
+                    entity::details::visit_type(input_column->column_->type().data_type(), [&groups, &out_ptr, &col_data, that=that] (auto type_desc_tag) {
+                        using ColumnTagType = std::decay_t<decltype(type_desc_tag)>;
+                        using ColumnType =  typename ColumnTagType::raw_type;
+                        if constexpr(!is_sequence_type(ColumnTagType::data_type)) {
+                            auto groups_pos = 0;
+                            while (auto block = col_data.next<TypeDescriptorTag<ColumnTagType, DimensionTag<entity::Dimension::Dim0>>>()) {
+                                auto ptr = reinterpret_cast<const ColumnType *>(block.value().data());
+                                for (auto i = 0u; i < block.value().row_count(); ++i, ++ptr) {
+                                    auto& val = out_ptr[groups[groups_pos]];
+                                    if constexpr(std::is_floating_point_v<ColumnType>) {
+                                        const auto& curr = GlobalRawType(*ptr);
+                                        if (!val.written_ || std::isnan(static_cast<ColumnType>(val.value_))) {
+                                            val.value_ = curr;
+                                            val.written_ = true;
+                                        } else if (!std::isnan(static_cast<ColumnType>(curr))) {
+                                            if constexpr(T == Extremum::MAX) {
+                                                val.value_ = std::max(val.value_, curr);
+                                            } else {
+                                                val.value_ = std::min(val.value_, curr);
+                                            }
+                                        }
+                                    } else {
+                                        if constexpr(T == Extremum::MAX) {
+                                            val.value_ = std::max(val.value_, GlobalRawType(*ptr));
+                                        } else {
+                                            // If unwritten, MaybeValue begins life with MIN_INT. That's always going to "win" when
+                                            // passed into `std::min`!
+                                            val.value_ = !val.written_ ? GlobalRawType(*ptr) : std::min(val.value_, GlobalRawType(*ptr));
+                                        }
+                                        val.written_ = true;
+                                    }
+                                    ++groups_pos;
+                                }
+                            }
+                        } else {
+                            util::raise_rte("String aggregations not currently supported");
+                        }
+                    });
+                }
+            });
+        }
+    }
+    [[nodiscard]] SegmentInMemory finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values) {
+        SegmentInMemory res;
+        if(!aggregated_.empty()) {
+            if(dynamic_schema) {
+                entity::details::visit_type(*data_type_, [that=this, &res, &output_column_name, unique_values] (auto type_desc_tag) {
+                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+                    auto prev_size = that->aggregated_.size() / sizeof(MaybeValue<RawType>);
+                    auto new_size = sizeof(MaybeValue<RawType>) * unique_values;
+                    that->aggregated_.resize(new_size);
+                    auto in_ptr =  reinterpret_cast<MaybeValue<RawType>*>(that->aggregated_.data());
+                    std::fill(in_ptr + prev_size, in_ptr + unique_values, MaybeValue<RawType>{});
+                    auto col = std::make_shared<Column>(make_scalar_type(DataType::FLOAT64), unique_values, true, false);
+                    auto out_ptr = reinterpret_cast<double*>(col->ptr());
+                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
+                        *out_ptr = in_ptr->written_ ? static_cast<double>(in_ptr->value_) : std::numeric_limits<double>::quiet_NaN();                }
+
+                    col->set_row_data(unique_values - 1);
+                    res.add_column(scalar_field_proto(DataType::FLOAT64, output_column_name.value), col);
+                });
+            } else {
+                entity::details::visit_type(*data_type_, [that=this, &res, output_column_name, unique_values] (auto type_desc_tag) {
+                    using RawType = typename decltype(type_desc_tag)::DataTypeTag::raw_type;
+                    auto col = std::make_shared<Column>(make_scalar_type(that->data_type_.value()), unique_values, true, false);
+                    const auto* in_ptr =  reinterpret_cast<const MaybeValue<RawType>*>(that->aggregated_.data());
+                    auto out_ptr = reinterpret_cast<RawType*>(col->ptr());
+                    for(auto i = 0u; i < unique_values; ++i, ++in_ptr, ++out_ptr) {
+                        *out_ptr = in_ptr->value_;
+                    }
+                    col->set_row_data(unique_values - 1);
+                    res.add_column(scalar_field_proto(that->data_type_.value(), output_column_name.value), col);
+                });
+            }
+        }
+        return res;
+    }
+};
+
+struct MaxOrMinAggregator {
     ColumnName input_column_name_;
     ColumnName output_column_name_;
-    DataType data_type_ = {};
     Extremum extremum_;
 
-    MaxOrMin(ColumnName input_column_name, ColumnName output_column_name, Extremum extremum) :
+    explicit MaxOrMinAggregator(ColumnName input_column_name, ColumnName output_column_name, Extremum extremum) :
         input_column_name_(std::move(input_column_name)),
         output_column_name_(std::move(output_column_name)),
         extremum_(extremum){
     }
-
-    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
-
-    std::optional<DataType> finalize(SegmentInMemory& seg, bool dynamic_schema, size_t unique_values);
+    ARCTICDB_MOVE_COPY_DEFAULT(MaxOrMinAggregator)
 
     [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-
     [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
+    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const {
+        switch(extremum_) {
+            case Extremum::MAX:
+                return MaxOrMinAggregatorData<Extremum::MAX>();
+            case Extremum::MIN:
+                return MaxOrMinAggregatorData<Extremum::MIN>();
+            default:
+                internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Unexpected Extremum enum value in MaxOrMinAggregator");
+        }
+    }
+};
 
-    [[nodiscard]] MaxOrMin construct() const { return {input_column_name_, output_column_name_, extremum_}; }
+struct MeanAggregatorData {
+    CountAndTotals data_;
 
-    void set_data_type(DataType data_type) { data_type_ = data_type; }
+    MeanAggregatorData() = default;
+    // Warn on copies as data_ could be a large buffer
+    MeanAggregatorData(const MeanAggregatorData& other):
+            data_(other.data_) {
+        log::version().warn("Copying potentially large buffer in MeanAggregatorData");
+    }
+    MeanAggregatorData& operator=(const MeanAggregatorData& other) {
+        data_ = other.data_;
+        log::version().warn("Copying potentially large buffer in MeanAggregatorData");
+        return *this;
+    }
+    ARCTICDB_MOVE(MeanAggregatorData)
 
+    // Mean values are always doubles so this is a no-op
+    void add_data_type(ARCTICDB_UNUSED DataType data_type) {}
+    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
+    SegmentInMemory finalize(const ColumnName& output_column_name,  bool dynamic_schema, size_t unique_values);
 };
 
-struct Mean {
-    ColumnName input_;
-    CountAndTotals data_;
+struct MeanAggregator {
     ColumnName input_column_name_;
     ColumnName output_column_name_;
-    DataType data_type_ = {};
 
-    explicit Mean(ColumnName input_column_name, ColumnName output_column_name) :
-        input_column_name_(std::move(input_column_name)),
-        output_column_name_(std::move(output_column_name)) {
+    explicit MeanAggregator(ColumnName input_column_name, ColumnName output_column_name) :
+            input_column_name_(std::move(input_column_name)),
+            output_column_name_(std::move(output_column_name)) {
     }
-
-    void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values);
-
-    std::optional<DataType> finalize(SegmentInMemory& seg, bool dynamic_schema, size_t unique_values);
+    ARCTICDB_MOVE_COPY_DEFAULT(MeanAggregator)
 
     [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
-
     [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
-
-    [[nodiscard]] Mean construct() const { return Mean(input_column_name_, output_column_name_); }
-
-    void set_data_type(DataType data_type) { data_type_ = data_type; }
-
+    [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return MeanAggregatorData(); }
 };
 
 } //namespace arcticdb
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/aggregation_interface.hpp b/cpp/arcticdb/processing/aggregation_interface.hpp
index 99f3be90cc..8dd45b3409 100644
--- a/cpp/arcticdb/processing/aggregation_interface.hpp
+++ b/cpp/arcticdb/processing/aggregation_interface.hpp
@@ -11,6 +11,39 @@
 
 namespace arcticdb{
 
+struct IGroupingAggregatorData {
+    template<class Base>
+    struct Interface : Base {
+        void add_data_type(DataType data_type) { folly::poly_call<0>(*this, data_type); }
+
+        void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) {
+            folly::poly_call<1>(*this, input_column, groups, unique_values);
+        }
+        [[nodiscard]] SegmentInMemory finalize(const ColumnName& output_column_name, bool dynamic_schema, size_t unique_values) {
+            return folly::poly_call<2>(*this, output_column_name, dynamic_schema, unique_values);
+        }
+    };
+
+    template<class T>
+    using Members = folly::PolyMembers<&T::add_data_type, &T::aggregate, &T::finalize>;
+};
+
+using GroupingAggregatorData = folly::Poly<IGroupingAggregatorData>;
+
+struct IGroupingAggregator {
+    template<class Base>
+    struct Interface : Base {
+        [[nodiscard]] ColumnName get_input_column_name() const { return folly::poly_call<0>(*this); };
+        [[nodiscard]] ColumnName get_output_column_name() const { return folly::poly_call<1>(*this); };
+        [[nodiscard]] GroupingAggregatorData get_aggregator_data() const { return folly::poly_call<2>(*this); }
+    };
+
+    template<class T>
+    using Members = folly::PolyMembers<&T::get_input_column_name, &T::get_output_column_name, &T::get_aggregator_data>;
+};
+
+using GroupingAggregator = folly::Poly<IGroupingAggregator>;
+
 struct IColumnStatsAggregatorData {
     template<class Base>
     struct Interface : Base {
@@ -53,38 +86,4 @@ struct CountAndTotals {
     std::vector<Fraction> fractions_;
 };
 
-struct IAggregation {
-    template<class Base>
-    struct Interface : Base {
-        void aggregate(const std::optional<ColumnWithStrings>& input_column, const std::vector<size_t>& groups, size_t unique_values) { folly::poly_call<0>(*this, input_column, groups, unique_values); }
-
-        std::optional<DataType> finalize(SegmentInMemory& seg, bool dynamic_schema, size_t unique_values) { return folly::poly_call<1>(*this, seg, dynamic_schema, unique_values); }
-
-        [[nodiscard]] ColumnName get_input_column_name() const { return folly::poly_call<2>(*this); };
-
-        [[nodiscard]] ColumnName get_output_column_name() const { return folly::poly_call<3>(*this); };
-
-        void set_data_type(DataType data_type_) { folly::poly_call<4>(*this, data_type_); }
-    };
-
-    template<class T>
-    using Members = folly::PolyMembers<&T::aggregate, &T::finalize, &T::get_input_column_name, &T::get_output_column_name, &T::set_data_type>;
-};
-
-using Aggregation = folly::Poly<IAggregation>;
-
-struct IAggregationFactory {
-    template<class Base>
-    struct Interface : Base {
-
-        [[nodiscard]] Aggregation construct() const { return folly::poly_call<0>(*this); }
-
-    };
-
-    template<class T>
-    using Members = folly::PolyMembers<&T::construct>;
-};
-
-using AggregationFactory = folly::Poly<IAggregationFactory>;
-
 } //namespace arcticdb
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
index a99740d9a6..4bac8a6d83 100644
--- a/cpp/arcticdb/processing/clause.cpp
+++ b/cpp/arcticdb/processing/clause.cpp
@@ -20,80 +20,101 @@
 
 namespace arcticdb {
 
-[[nodiscard]] Composite<ProcessingSegment>
-ColumnStatsGenerationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-    auto procs = std::move(p);
-    std::vector<ColumnStatsAggregatorData> aggregators_data;
-    internal::check<ErrorCode::E_INVALID_ARGUMENT>(
-            static_cast<bool>(column_stats_aggregators_),
-            "ColumnStatsGenerationClause::process does not make sense with no aggregators");
-    for (const auto &agg : *column_stats_aggregators_){
-        aggregators_data.emplace_back(agg.get_aggregator_data());
+std::vector<Composite<ProcessingSegment>> single_partition(std::vector<Composite<ProcessingSegment>> &&comps) {
+    std::vector<Composite<ProcessingSegment>> v;
+    v.push_back(merge_composites_shallow(std::move(comps)));
+    return v;
+}
+
+class GroupingMap {
+    using NumericMapType = std::variant<
+            std::monostate,
+            std::shared_ptr<std::unordered_map<bool, size_t>>,
+            std::shared_ptr<std::unordered_map<uint8_t, size_t>>,
+            std::shared_ptr<std::unordered_map<uint16_t, size_t>>,
+            std::shared_ptr<std::unordered_map<uint32_t, size_t>>,
+            std::shared_ptr<std::unordered_map<uint64_t, size_t>>,
+            std::shared_ptr<std::unordered_map<int8_t, size_t>>,
+            std::shared_ptr<std::unordered_map<int16_t, size_t>>,
+            std::shared_ptr<std::unordered_map<int32_t, size_t>>,
+            std::shared_ptr<std::unordered_map<int64_t, size_t>>,
+            std::shared_ptr<std::unordered_map<float, size_t>>,
+            std::shared_ptr<std::unordered_map<double, size_t>>>;
+
+    NumericMapType map_;
+
+public:
+    size_t size() const {
+        return util::variant_match(map_,
+                                   [](const std::monostate &) {
+                                       return size_t(0);
+                                   },
+                                   [](const auto &other) {
+                                       return other->size();
+                                   });
     }
 
-    emilib::HashSet<IndexValue> start_indexes;
-    emilib::HashSet<IndexValue> end_indexes;
+    template<typename T>
+    std::shared_ptr<std::unordered_map<T, size_t>> get() {
+        return util::variant_match(map_,
+                                   [that = this](const std::monostate &) {
+                                       that->map_ = std::make_shared<std::unordered_map<T, size_t>>();
+                                       return std::get<std::shared_ptr<std::unordered_map<T, size_t>>>(that->map_);
+                                   },
+                                   [](const std::shared_ptr<std::unordered_map<T, size_t>> &ptr) {
+                                       return ptr;
+                                   },
+                                   [](const auto &) -> std::shared_ptr<std::unordered_map<T, size_t>> {
+                                       schema::raise<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(
+                                               "GroupBy does not support the grouping column type changing with dynamic schema");
+                                   });
+    }
+};
 
-    internal::check<ErrorCode::E_INVALID_ARGUMENT>(
-            !procs.empty(),
-            "ColumnStatsGenerationClause::process does not make sense with no processing segments");
-    procs.broadcast(
-            [&store, &start_indexes, &end_indexes, &aggregators_data, that = this](
-                    auto &proc) {
-                proc.set_execution_context(that->execution_context_);
-                for (const auto& slice_and_key: proc.data_) {
-                    start_indexes.insert(slice_and_key.key_->start_index());
-                    end_indexes.insert(slice_and_key.key_->end_index());
-                }
-                for (auto agg_data : folly::enumerate(aggregators_data)) {
-                    auto input_column_name = that->column_stats_aggregators_->at(agg_data.index).get_input_column_name();
-                    auto input_column = proc.get(input_column_name, store);
-                    if (std::holds_alternative<ColumnWithStrings>(input_column)) {
-                        auto input_column_with_strings = std::get<ColumnWithStrings>(input_column);
-                        agg_data->aggregate(input_column_with_strings);
-                    } else {
-                        if(!that->execution_context_->dynamic_schema())
-                            internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
-                                    "Unable to resolve column denoted by aggregation operator: '{}'",
-                                    input_column_name);
-                    }
-                }
-            });
+struct SliceAndKeyWrapper {
+    pipelines::SliceAndKey seg_;
+    std::shared_ptr<Store> store_;
+    SegmentInMemory::iterator it_;
+    const StreamId id_;
+
+    explicit SliceAndKeyWrapper(pipelines::SliceAndKey &&seg, std::shared_ptr<Store> store) :
+            seg_(std::move(seg)),
+            store_(std::move(store)),
+            it_(seg_.segment(store_).begin()),
+            id_(seg_.segment(store_).descriptor().id()) {
+    }
 
-    internal::check<ErrorCode::E_ASSERTION_FAILURE>(
-            start_indexes.size() == 1 && end_indexes.size() == 1,
-            "Expected all data segments in one processing segment to have same start and end indexes");
-    auto start_index = *start_indexes.begin();
-    auto end_index = *end_indexes.begin();
-    schema::check<ErrorCode::E_UNSUPPORTED_INDEX_TYPE>(
-            std::holds_alternative<NumericIndex>(start_index) && std::holds_alternative<NumericIndex>(end_index),
-            "Cannot build column stats over string-indexed symbol"
-            );
-    auto start_index_col = std::make_shared<Column>(make_scalar_type(DataType::MICROS_UTC64), true);
-    auto end_index_col = std::make_shared<Column>(make_scalar_type(DataType::MICROS_UTC64), true);
-    start_index_col->template push_back<NumericIndex>(std::get<NumericIndex>(start_index));
-    end_index_col->template push_back<NumericIndex>(std::get<NumericIndex>(end_index));
+    bool advance() {
+        return ++it_ != seg_.segment(store_).end();
+    }
 
-    SegmentInMemory seg;
-    seg.add_column(scalar_field_proto(DataType::MICROS_UTC64, start_index_column_name), start_index_col);
-    seg.add_column(scalar_field_proto(DataType::MICROS_UTC64, end_index_column_name), end_index_col);
-    for (const auto& agg_data: folly::enumerate(aggregators_data)) {
-       seg.concatenate(agg_data->finalize(column_stats_aggregators_->at(agg_data.index).get_output_column_names()));
+    SegmentInMemory::Row &row() {
+        return *it_;
     }
-    seg.set_row_id(0);
-    return Composite{ProcessingSegment{std::move(seg)}};
+
+    const StreamId &id() const {
+        return id_;
+    }
+};
+
+Composite<ProcessingSegment> PassthroughClause::process(ARCTICDB_UNUSED const std::shared_ptr<Store> &store,
+                                                        Composite<ProcessingSegment> &&p) const {
+    auto procs = std::move(p);
+    return procs;
 }
 
-[[nodiscard]] Composite<ProcessingSegment>
-FilterClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
+Composite<ProcessingSegment> FilterClause::process(
+        std::shared_ptr<Store> store,
+        Composite<ProcessingSegment> &&p
+        ) const {
     auto procs = std::move(p);
     Composite<ProcessingSegment> output;
-    procs.broadcast([&store, &execution_context = execution_context_, &output](auto &proc) {
-        proc.set_execution_context(execution_context);
-        auto variant_data = proc.get(execution_context->root_node_name_, store);
+    procs.broadcast([&optimisation=optimisation_, &store, &expression_context = expression_context_, &output](auto &proc) {
+        proc.set_expression_context(expression_context);
+        auto variant_data = proc.get(expression_context->root_node_name_, store);
         util::variant_match(variant_data,
-                            [&proc, &output, &store](const std::shared_ptr<util::BitSet> &bitset) {
-                                proc.apply_filter(*bitset, store);
+                            [&optimisation, &proc, &output, &store](const std::shared_ptr<util::BitSet> &bitset) {
+                                proc.apply_filter(*bitset, store, optimisation);
                                 output.push_back(std::move(proc));
                             },
                             [](EmptyResult) {
@@ -109,131 +130,113 @@ FilterClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment>
     return output;
 }
 
-[[nodiscard]] Composite<ProcessingSegment> ProjectClause::process(std::shared_ptr<Store> store,
+std::string FilterClause::to_string() const {
+    return expression_context_ ? fmt::format("WHERE {}", expression_context_->root_node_name_.value) : "";
+}
+
+Composite<ProcessingSegment> ProjectClause::process(std::shared_ptr<Store> store,
                                                                   Composite<ProcessingSegment> &&p) const {
     auto procs = std::move(p);
     Composite<ProcessingSegment> output;
-    procs.broadcast([&store, &execution_context = execution_context_, &output, that = this](auto &proc) {
-        proc.set_execution_context(execution_context);
-        auto variant_data = proc.get(execution_context->root_node_name_, store);
+    procs.broadcast([&store, &expression_context = expression_context_, &output, that = this](auto &proc) {
+        proc.set_expression_context(expression_context);
+        auto variant_data = proc.get(expression_context->root_node_name_, store);
         util::variant_match(variant_data,
-                            [&proc, &output, &store, &execution_context, &that](ColumnWithStrings &col) {
+                            [&proc, &output, &store, &that](ColumnWithStrings &col) {
 
                                 const auto data_type = col.column_->type().data_type();
-                                const std::string_view name = that->column_name_;
+                                const std::string_view name = that->output_column_;
 
-                                std::call_once(*that->add_column_, [context = execution_context, &data_type, &name]() {
-                                    std::scoped_lock lock{*context->column_mutex_};
-                                    context->output_descriptor_->add_scalar_field(data_type, name);
-                                });
                                 auto &slice_and_keys = proc.data();
                                 auto &last = *slice_and_keys.rbegin();
                                 last.segment(store).add_column(scalar_field_proto(data_type, name), col.column_);
                                 ++last.slice().col_range.second;
                                 output.push_back(std::move(proc));
                             },
-                            [&proc, &output, &execution_context](const EmptyResult&) {
-                                if(execution_context->dynamic_schema())
+                            [&proc, &output, &expression_context](const EmptyResult&) {
+                                if(expression_context->dynamic_schema_)
                                     output.push_back(std::move(proc));
                                 else
                                     util::raise_rte("Cannot project from empty column with static schema");
                             },
                             [](const auto &) {
-                                util::raise_rte("Expected bitset from filter clause");
+                                util::raise_rte("Expected column from projection clause");
                             });
     });
     return output;
 }
 
-class GroupingMap {
-    using NumericMapType = std::variant<
-        std::monostate,
-        std::shared_ptr<std::unordered_map<bool, size_t>>,
-        std::shared_ptr<std::unordered_map<uint8_t, size_t>>,
-        std::shared_ptr<std::unordered_map<uint16_t, size_t>>,
-        std::shared_ptr<std::unordered_map<uint32_t, size_t>>,
-        std::shared_ptr<std::unordered_map<uint64_t, size_t>>,
-        std::shared_ptr<std::unordered_map<int8_t, size_t>>,
-        std::shared_ptr<std::unordered_map<int16_t, size_t>>,
-        std::shared_ptr<std::unordered_map<int32_t, size_t>>,
-        std::shared_ptr<std::unordered_map<int64_t, size_t>>,
-        std::shared_ptr<std::unordered_map<float, size_t>>,
-        std::shared_ptr<std::unordered_map<double, size_t>>>;
-
-    NumericMapType map_;
-
-public:
-    size_t size() const {
-        return util::variant_match(map_,
-                                   [](const std::monostate &) {
-                                       return size_t(0);
-                                   },
-                                   [](const auto &other) {
-                                       return other->size();
-                                   });
-    }
+[[nodiscard]] std::string ProjectClause::to_string() const {
+    return expression_context_ ? fmt::format("PROJECT Column[\"{}\"] = {}", output_column_, expression_context_->root_node_name_.value) : "";
+}
 
-    template<typename T>
-    std::shared_ptr<std::unordered_map<T, size_t>> get() {
-        return util::variant_match(map_,
-                                   [that = this](const std::monostate &) {
-                                       that->map_ = std::make_shared<std::unordered_map<T, size_t>>();
-                                       return std::get<std::shared_ptr<std::unordered_map<T, size_t>>>(that->map_);
-                                   },
-                                   [](const std::shared_ptr<std::unordered_map<T, size_t>> &ptr) {
-                                       return ptr;
-                                   },
-                                   [](const auto &) -> std::shared_ptr<std::unordered_map<T, size_t>> {
-                                       util::raise_rte("Grouping column type change");
-                                   });
+AggregationClause::AggregationClause(const std::string& grouping_column,
+                                     const std::unordered_map<std::string,
+                                     std::string>& aggregations):
+        grouping_column_(grouping_column),
+        aggregation_map_(aggregations) {
+    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_});
+    clause_info_.modifies_output_descriptor_ = true;
+    for (const auto& [column_name, aggregation_operator]: aggregations) {
+        auto [_, inserted] = clause_info_.input_columns_->insert(column_name);
+        user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(inserted,
+                                                              "Cannot perform two aggregations over the same column: {}",
+                                                              column_name);
+        auto typed_column_name = ColumnName(column_name);
+        if (aggregation_operator == "sum") {
+            aggregators_.emplace_back(SumAggregator(typed_column_name, typed_column_name));
+        } else if (aggregation_operator == "mean") {
+            aggregators_.emplace_back(MeanAggregator(typed_column_name, typed_column_name));
+        } else if (aggregation_operator == "max") {
+            aggregators_.emplace_back(MaxOrMinAggregator(typed_column_name, typed_column_name, Extremum::MAX));
+        } else if (aggregation_operator == "min") {
+            aggregators_.emplace_back(MaxOrMinAggregator(typed_column_name, typed_column_name, Extremum::MIN));
+        } else {
+            user_input::raise<ErrorCode::E_INVALID_USER_ARGUMENT>("Unknown aggregation operator provided: {}", aggregation_operator);
+        }
     }
-};
-
-std::vector<Composite<ProcessingSegment>> single_partition(std::vector<Composite<ProcessingSegment>> &&comps) {
-    std::vector<Composite<ProcessingSegment>> v;
-    v.push_back(merge_composites_shallow(std::move(comps)));
-    return v;
 }
 
-[[nodiscard]] Composite<ProcessingSegment>
-AggregationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-    std::call_once(*reset_descriptor_, [context = execution_context_]() {
-        std::scoped_lock lock{*context->column_mutex_};
-        context->orig_output_descriptor_ = context->output_descriptor_;
-        context->output_descriptor_ = empty_descriptor();
-    });
-    SegmentInMemory seg{empty_descriptor()};
+Composite<ProcessingSegment> AggregationClause::process(std::shared_ptr<Store> store,
+                                                        Composite<ProcessingSegment> &&p) const {
     auto procs = std::move(p);
-    size_t num_unique = 0;
-    std::vector<Aggregation> aggregators;
-    auto desc = execution_context_->orig_output_descriptor_;
-    for (const auto &agg : aggregation_operators_){
-        auto agg_construct = agg.construct();
-        const auto& agg_field_pos = desc->find_field(agg_construct.get_input_column_name().value);
-        util::check(agg_field_pos.has_value(), "Field {} not found in aggregation", agg_construct.get_input_column_name().value);
-        auto agg_field = desc->field(agg_field_pos.value());
-        agg_construct.set_data_type(data_type_from_proto(agg_field.type_desc()));
-        aggregators.emplace_back(agg_construct);
+    std::vector<GroupingAggregatorData> aggregators_data;
+    internal::check<ErrorCode::E_INVALID_ARGUMENT>(
+            !aggregators_.empty(),
+            "AggregationClause::process does not make sense with no aggregators");
+    for (const auto &agg: aggregators_){
+        aggregators_data.emplace_back(agg.get_aggregator_data());
     }
 
-    auto offset = 0;
-    auto grouping_column_name = execution_context_->root_node_name_.value;
+    // Work out the common type between the processing segments for the columns being aggregated
+    procs.broadcast([&store, &aggregators_data, &aggregators=aggregators_](auto& proc) {
+        for (auto agg_data: folly::enumerate(aggregators_data)) {
+            auto input_column_name = aggregators.at(agg_data.index).get_input_column_name();
+            auto input_column = proc.get(input_column_name, store);
+            std::optional<ColumnWithStrings> opt_input_column;
+            if (std::holds_alternative<ColumnWithStrings>(input_column)) {
+                agg_data->add_data_type(std::get<ColumnWithStrings>(input_column).column_->type().data_type());
+            }
+        }
+    });
+
+    size_t num_unique{0};
+    size_t next_group_id{0};
     auto string_pool = std::make_shared<StringPool>();
     DataType grouping_data_type;
     GroupingMap grouping_map;
     procs.broadcast(
-        [&store, &num_unique, &execution_context =
-        execution_context_, &grouping_data_type, &grouping_map, &offset, &aggregators, &string_pool, &grouping_column_name](
+        [&store, &num_unique,
+        &grouping_data_type, &grouping_map, &next_group_id, &aggregators_data, &string_pool, that=this](
             auto &proc) {
-            proc.set_execution_context(execution_context);
-            //TODO this is a hack, ideally the grouping should be able to be an expression
-            auto partitioning_column = proc.get(ColumnName(grouping_column_name), store);
+            auto partitioning_column = proc.get(ColumnName(that->grouping_column_), store);
             if (std::holds_alternative<ColumnWithStrings>(partitioning_column)) {
                 ColumnWithStrings col = std::get<ColumnWithStrings>(partitioning_column);
                 entity::details::visit_type(col.column_->type().data_type(),
-                                            [&proc_ =
-                                            proc, &grouping_map, &offset, &aggregators, &string_pool, &col, &num_unique, &store, &grouping_data_type](
-                                                auto data_type_tag) {
+                                            [&proc_=proc, &grouping_map, &next_group_id, &aggregators_data, &string_pool, &col,
+                                             &num_unique, &store, &grouping_data_type, that](auto data_type_tag) {
                                                 using DataTypeTagType = decltype(data_type_tag);
                                                 using RawType = typename DataTypeTagType::raw_type;
                                                 constexpr auto data_type = DataTypeTagType::data_type;
@@ -241,6 +244,7 @@ AggregationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSeg
                                                 std::vector<size_t> row_to_group;
                                                 row_to_group.reserve(col.column_->row_count());
                                                 auto input_data = col.column_->data();
+                                                auto hash_to_group = grouping_map.get<RawType>();
                                                 while (auto block = input_data.next<ScalarTagType<DataTypeTagType>>()) {
                                                     const auto row_count = block->row_count();
                                                     auto ptr = block->data();
@@ -256,29 +260,25 @@ AggregationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSeg
                                                         } else {
                                                             val = *ptr;
                                                         }
-                                                        auto hashes = grouping_map.get<RawType>();
-                                                        if (auto it = hashes->find(val); it == hashes->end()) {
-                                                            row_to_group.push_back(offset);
-                                                            hashes->try_emplace(val, offset++);
+                                                        if (auto it = hash_to_group->find(val); it == hash_to_group->end()) {
+                                                            row_to_group.push_back(next_group_id);
+                                                            hash_to_group->try_emplace(val, next_group_id++);
                                                         } else {
                                                             row_to_group.push_back(it->second);
                                                         }
                                                     }
                                                 }
 
-                                                num_unique = offset;
+                                                num_unique = next_group_id;
                                                 util::check(num_unique != 0, "Got zero unique values");
-                                                for (Aggregation &agg : aggregators) {
-                                                    auto input_column = proc_.get(agg.get_input_column_name(), store);
+                                                for (auto agg_data: folly::enumerate(aggregators_data)) {
+                                                    auto input_column_name = that->aggregators_.at(agg_data.index).get_input_column_name();
+                                                    auto input_column = proc_.get(input_column_name, store);
                                                     std::optional<ColumnWithStrings> opt_input_column;
                                                     if (std::holds_alternative<ColumnWithStrings>(input_column)) {
                                                         opt_input_column.emplace(std::get<ColumnWithStrings>(input_column));
                                                     }
-                                                    agg.aggregate(opt_input_column,
-                                                                    row_to_group,
-                                                                    num_unique);
-
-
+                                                    agg_data->aggregate(opt_input_column, row_to_group, num_unique);
                                                 }
                                             });
             } else {
@@ -286,9 +286,11 @@ AggregationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSeg
             }
         });
 
-    auto index_pos =
-        seg.add_column(scalar_field_proto(grouping_data_type, grouping_column_name), grouping_map.size(), true);
-    execution_context_->check_output_column(grouping_column_name, grouping_data_type);
+    SegmentInMemory seg;
+    auto index_col = std::make_shared<Column>(make_scalar_type(grouping_data_type), grouping_map.size(), true, false);
+    auto index_pos = seg.add_column(scalar_field_proto(grouping_data_type, grouping_column_), index_col);
+    seg.descriptor().set_index(IndexDescriptor(0, IndexDescriptor::ROWCOUNT));
+
     entity::details::visit_type(grouping_data_type, [&seg, &grouping_map, index_pos](auto data_type_tag) {
         using DataTypeTagType = decltype(data_type_tag);
         using RawType = typename DataTypeTagType::raw_type;
@@ -307,25 +309,91 @@ AggregationClause::process(std::shared_ptr<Store> store, Composite<ProcessingSeg
         for (const auto &element : elements)
             *index_ptr++ = element.first;
     });
+    index_col->set_row_data(grouping_map.size() - 1);
 
-    for (auto &agg : aggregators) {
-        auto data_type = agg.finalize(seg, execution_context_->dynamic_schema(), num_unique);
-        if(data_type)
-            execution_context_->check_output_column(agg.get_output_column_name().value, data_type.value());
+    for (auto agg_data: folly::enumerate(aggregators_data)) {
+        seg.concatenate(agg_data->finalize(aggregators_.at(agg_data.index).get_output_column_name(), processing_config_.dynamic_schema_, num_unique));
     }
 
     seg.set_string_pool(string_pool);
     seg.set_row_id(num_unique - 1);
-    std::call_once(*set_name_index_, [context = execution_context_, grouping_column_name]() {
-        std::scoped_lock lock{*context->name_index_mutex_};
-        auto norm_desc = context->get_norm_meta_descriptor();
-        norm_desc->mutable_df()->mutable_common()->mutable_index()->set_name(grouping_column_name);
-        norm_desc->mutable_df()->mutable_common()->mutable_index()->clear_fake_name();
-        norm_desc->mutable_df()->mutable_common()->mutable_index()->set_is_not_range_index(true);
-    });
     return Composite{ProcessingSegment{std::move(seg)}};
 }
 
+[[nodiscard]] std::string AggregationClause::to_string() const {
+    return fmt::format("AGGREGATE {}", aggregation_map_);
+}
+
+[[nodiscard]] Composite<ProcessingSegment> RemoveColumnPartitioningClause::process(std::shared_ptr<Store> store,
+                                                                                   Composite<ProcessingSegment> &&p) const {
+    using namespace arcticdb::pipelines;
+    auto procs = std::move(p);
+    Composite<ProcessingSegment> output;
+    procs.broadcast([&store, &output](ProcessingSegment &proc) {
+        size_t min_start_row = std::numeric_limits<size_t>::max();
+        size_t max_end_row = 0;
+        size_t min_start_col = std::numeric_limits<size_t>::max();
+        size_t max_end_col = 0;
+        std::optional<SegmentInMemory> output_seg;
+        for (auto& slice_and_key: proc.data()) {
+            min_start_row = std::min(min_start_row, slice_and_key.slice().row_range.start());
+            max_end_row = std::max(max_end_row, slice_and_key.slice().row_range.end());
+            min_start_col = std::min(min_start_col, slice_and_key.slice().col_range.start());
+            max_end_col = std::max(max_end_col, slice_and_key.slice().col_range.end());
+            auto segment = std::move(slice_and_key.segment(store));
+            if (output_seg.has_value()) {
+                stream::merge_string_columns(segment, output_seg->string_pool_ptr(), false);
+                output_seg->concatenate(std::move(segment), true);
+            } else {
+                output_seg = std::make_optional<SegmentInMemory>(std::move(segment));
+            }
+        }
+        if (output_seg.has_value()) {
+            const RowRange row_range{min_start_row, max_end_row};
+            const ColRange col_range{min_start_col, max_end_col};
+            output.push_back(ProcessingSegment{std::move(*output_seg), FrameSlice{col_range, row_range}});
+        }
+    });
+    return output;
+}
+
+Composite<ProcessingSegment> SplitClause::process(std::shared_ptr<Store> store,
+                                                  Composite<ProcessingSegment> &&procs) const {
+    using namespace arcticdb::pipelines;
+
+    auto proc_composite = std::move(procs);
+    Composite<ProcessingSegment> ret;
+    proc_composite.broadcast([&store, rows = rows_, &ret](auto &&p) {
+        auto proc = std::forward<decltype(p)>(p);
+        auto slice_and_keys = proc.data();
+        for (auto &slice_and_key: slice_and_keys) {
+            auto split_segs = slice_and_key.segment(store).split(rows);
+            const ColRange col_range{slice_and_key.slice().col_range};
+            size_t start_row = slice_and_key.slice().row_range.start();
+            size_t end_row = 0;
+            for (auto &item : split_segs) {
+                end_row = start_row + item.row_count();
+                const RowRange row_range{start_row, end_row};
+                ret.push_back(ProcessingSegment{std::move(item), FrameSlice{col_range, row_range}});
+                start_row = end_row;
+            }
+        }
+    });
+    return ret;
+}
+
+Composite<ProcessingSegment> SortClause::process(std::shared_ptr<Store> store,
+                                                 Composite<ProcessingSegment> &&p) const {
+    auto procs = std::move(p);
+    procs.broadcast([&store, &column = column_](auto &proc) {
+        auto slice_and_keys = proc.data();
+        for (auto &slice_and_key: slice_and_keys) {
+            slice_and_key.segment(store).sort(column);
+        }
+    });
+    return procs;
+}
+
 template<typename IndexType, typename DensityPolicy, typename QueueType, typename Comparator, typename StreamId>
 void merge_impl(
         Composite<ProcessingSegment> &ret,
@@ -335,8 +403,7 @@ void merge_impl(
         const arcticdb::pipelines::RowRange row_range,
         const arcticdb::pipelines::ColRange col_range,
         IndexType index,
-        std::shared_ptr<ExecutionContext> execution_context
-) {
+        const StreamDescriptor& stream_descriptor) {
     using namespace arcticdb::pipelines;
     auto num_segment_rows = ConfigsMap::instance()->get_int("Merge.SegmentSize", 100000);
     using SegmentationPolicy = stream::RowCountSegmentPolicy;
@@ -347,7 +414,7 @@ void merge_impl(
     };
     
     using AggregatorType = stream::Aggregator<IndexType, stream::DynamicSchema, SegmentationPolicy, DensityPolicy>;
-    auto fields = execution_context->output_descriptor_->fields();
+    auto fields = stream_descriptor.fields();
     StreamDescriptor::FieldsCollection new_fields{};
     auto field_name = fields[0].name();
     auto new_field = new_fields.Add();
@@ -369,115 +436,151 @@ void merge_impl(
 
 // MergeClause receives a list of DataFrames as input and merge them into a single one where all 
 // the rows are sorted by time stamp
-[[nodiscard]] Composite<ProcessingSegment>
-MergeClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-        using namespace arcticdb::pipelines;
-        auto procs = std::move(p);
-
-        auto compare =
-                [](const std::unique_ptr<SliceAndKeyWrapper> &left,
-                   const std::unique_ptr<SliceAndKeyWrapper> &right) {
-                    const auto left_index = pipelines::index::index_value_from_row(left->row(),
-                                                                                   IndexDescriptor::TIMESTAMP, 0);
-                    const auto right_index = pipelines::index::index_value_from_row(right->row(),
-                                                                                    IndexDescriptor::TIMESTAMP, 0);
-                    return left_index > right_index;
-                };
-
-        movable_priority_queue<std::unique_ptr<SliceAndKeyWrapper>, std::vector<std::unique_ptr<SliceAndKeyWrapper>>, decltype(compare)> input_streams{
-                compare};
+Composite<ProcessingSegment> MergeClause::process(std::shared_ptr<Store> store,
+                                                  Composite<ProcessingSegment> &&p) const {
+    using namespace arcticdb::pipelines;
+    auto procs = std::move(p);
 
-        size_t min_start_row = std::numeric_limits<size_t>::max();
-        size_t max_end_row = 0;
-        size_t min_start_col = std::numeric_limits<size_t>::max();
-        size_t max_end_col = 0;
-        procs.broadcast([&input_streams, &store, &min_start_row, &max_end_row, &min_start_col, &max_end_col](auto &&proc) {
-            auto slice_and_keys = proc.release_data();
-            for (auto &&slice_and_key: slice_and_keys) {
-                size_t start_row = slice_and_key.slice().row_range.start();
-                min_start_row = start_row < min_start_row ? start_row : min_start_row;
-                size_t end_row = slice_and_key.slice().row_range.end();
-                max_end_row = end_row > max_end_row ? end_row : max_end_row;
-                size_t start_col = slice_and_key.slice().col_range.start();
-                min_start_col = start_col < min_start_col ? start_col : min_start_col;
-                size_t end_col = slice_and_key.slice().col_range.end();
-                max_end_col = end_col > max_end_col ? end_col : max_end_col;
-                input_streams.push(
-                        std::make_unique<SliceAndKeyWrapper>(std::forward<pipelines::SliceAndKey>(slice_and_key),
-                                                             store));
-            }
-        });
-        const RowRange row_range{min_start_row, max_end_row};
-        const ColRange col_range{min_start_col, max_end_col};
-        Composite<ProcessingSegment> ret;
-        std::visit(
-                [&ret, &input_streams, add_symbol_column = add_symbol_column_, &comp = compare, stream_id = stream_id_, &row_range, &col_range, execution_context = execution_context_](auto idx,
-                                                                                                auto density) {
-                    merge_impl<decltype(idx), decltype(density), decltype(input_streams), decltype(comp), decltype(stream_id)>(ret,
-                                                                                                          input_streams,
-                                                                                                          add_symbol_column,
-                                                                                                          stream_id,
-                                                                                                          row_range,
-                                                                                                          col_range,
-                                                                                                          idx,
-                                                                                                          execution_context);
-                }, index_, density_policy_);
-
-        return ret;
+    auto compare =
+            [](const std::unique_ptr<SliceAndKeyWrapper> &left,
+               const std::unique_ptr<SliceAndKeyWrapper> &right) {
+                const auto left_index = pipelines::index::index_value_from_row(left->row(),
+                                                                               IndexDescriptor::TIMESTAMP, 0);
+                const auto right_index = pipelines::index::index_value_from_row(right->row(),
+                                                                                IndexDescriptor::TIMESTAMP, 0);
+                return left_index > right_index;
+            };
+
+    movable_priority_queue<std::unique_ptr<SliceAndKeyWrapper>, std::vector<std::unique_ptr<SliceAndKeyWrapper>>, decltype(compare)> input_streams{
+            compare};
+
+    size_t min_start_row = std::numeric_limits<size_t>::max();
+    size_t max_end_row = 0;
+    size_t min_start_col = std::numeric_limits<size_t>::max();
+    size_t max_end_col = 0;
+    procs.broadcast([&input_streams, &store, &min_start_row, &max_end_row, &min_start_col, &max_end_col](auto &&proc) {
+        auto slice_and_keys = proc.release_data();
+        for (auto &&slice_and_key: slice_and_keys) {
+            size_t start_row = slice_and_key.slice().row_range.start();
+            min_start_row = start_row < min_start_row ? start_row : min_start_row;
+            size_t end_row = slice_and_key.slice().row_range.end();
+            max_end_row = end_row > max_end_row ? end_row : max_end_row;
+            size_t start_col = slice_and_key.slice().col_range.start();
+            min_start_col = start_col < min_start_col ? start_col : min_start_col;
+            size_t end_col = slice_and_key.slice().col_range.end();
+            max_end_col = end_col > max_end_col ? end_col : max_end_col;
+            input_streams.push(
+                    std::make_unique<SliceAndKeyWrapper>(std::forward<pipelines::SliceAndKey>(slice_and_key),
+                                                         store));
+        }
+    });
+    const RowRange row_range{min_start_row, max_end_row};
+    const ColRange col_range{min_start_col, max_end_col};
+    Composite<ProcessingSegment> ret;
+    std::visit(
+            [&ret, &input_streams, add_symbol_column = add_symbol_column_, &comp = compare, stream_id = stream_id_, &row_range, &col_range, &stream_descriptor = stream_descriptor_](auto idx,
+                                                                                            auto density) {
+                merge_impl<decltype(idx), decltype(density), decltype(input_streams), decltype(comp), decltype(stream_id)>(ret,
+                                                                                                      input_streams,
+                                                                                                      add_symbol_column,
+                                                                                                      stream_id,
+                                                                                                      row_range,
+                                                                                                      col_range,
+                                                                                                      idx,
+                                                                                                      stream_descriptor);
+            }, index_, density_policy_);
+
+    return ret;
+}
+
+std::optional<std::vector<Composite<ProcessingSegment>>> MergeClause::repartition(
+        std::vector<Composite<ProcessingSegment>> &&comps) const {
+    std::vector<Composite<ProcessingSegment>> v;
+    v.push_back(merge_composites_shallow(std::move(comps)));
+    return v;
+}
+
+Composite<ProcessingSegment> ColumnStatsGenerationClause::process(std::shared_ptr<Store> store,
+                                                                  Composite<ProcessingSegment> &&p) const {
+    auto procs = std::move(p);
+    std::vector<ColumnStatsAggregatorData> aggregators_data;
+    internal::check<ErrorCode::E_INVALID_ARGUMENT>(
+            static_cast<bool>(column_stats_aggregators_),
+            "ColumnStatsGenerationClause::process does not make sense with no aggregators");
+    for (const auto &agg : *column_stats_aggregators_){
+        aggregators_data.emplace_back(agg.get_aggregator_data());
     }
 
-[[nodiscard]] Composite<ProcessingSegment>
-RemoveColumnPartitioningClause::process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-        using namespace arcticdb::pipelines;
-        auto procs = std::move(p);
-        Composite<ProcessingSegment> output;
-        auto index = stream::index_type_from_descriptor(execution_context_->output_descriptor_.value());
-        util::variant_match(index,
-        [&](const stream::TimeseriesIndex &) {
-            size_t num_index_columns = stream::TimeseriesIndex::field_count();
-            procs.broadcast([&store, &output, &num_index_columns, this](ProcessingSegment &proc) {
-                SegmentInMemory new_segment{empty_descriptor(descriptor_type_, proc.data()[0].segment_->get_index_col_name())};
-                new_segment.set_row_id(proc.data()[0].segment_->get_row_id());
-                size_t min_start_row = std::numeric_limits<size_t>::max();
-                size_t max_end_row = 0;
-                size_t min_start_col = std::numeric_limits<size_t>::max();
-                size_t max_end_col = 0;
-                for (auto &slice_and_key: proc.data()) {
-                    size_t start_row = slice_and_key.slice().row_range.start();
-                    min_start_row = start_row < min_start_row ? start_row : min_start_row;
-                    size_t end_row = slice_and_key.slice().row_range.end();
-                    max_end_row = end_row > max_end_row ? end_row : max_end_row;
-                    size_t start_col = slice_and_key.slice().col_range.start();
-                    min_start_col = start_col < min_start_col ? start_col : min_start_col;
-                    size_t end_col = slice_and_key.slice().col_range.end();
-                    max_end_col = end_col > max_end_col ? end_col : max_end_col;
-                    size_t column_idx = 0;
-                    for(auto field : folly::enumerate(slice_and_key.segment(store).descriptor().fields())) {
-                        const auto column_name = field->name();
-                        auto column = proc.get(ColumnName(column_name), store);
-                        if (std::holds_alternative<ColumnWithStrings>(column)) {
-                            ColumnWithStrings column_strings = std::get<ColumnWithStrings>(column);
-                            const auto data_type = column_strings.column_->type().data_type();
-                            if(((start_col - num_index_columns) == 0) || column_idx >= num_index_columns){
-                                new_segment.add_column(scalar_field_proto(data_type, column_name),  slice_and_key.segment(store).column_ptr(field.index));
-                            }
-                        }else {
-                            util::raise_rte("Expected single column from expression");
-                        }
-                        column_idx++;
+    emilib::HashSet<IndexValue> start_indexes;
+    emilib::HashSet<IndexValue> end_indexes;
+
+    internal::check<ErrorCode::E_INVALID_ARGUMENT>(
+            !procs.empty(),
+            "ColumnStatsGenerationClause::process does not make sense with no processing segments");
+    procs.broadcast(
+            [&store, &start_indexes, &end_indexes, &aggregators_data, that=this](
+                    auto &proc) {
+                for (const auto& slice_and_key: proc.data_) {
+                    start_indexes.insert(slice_and_key.key_->start_index());
+                    end_indexes.insert(slice_and_key.key_->end_index());
+                }
+                for (auto agg_data : folly::enumerate(aggregators_data)) {
+                    auto input_column_name = that->column_stats_aggregators_->at(agg_data.index).get_input_column_name();
+                    auto input_column = proc.get(input_column_name, store);
+                    if (std::holds_alternative<ColumnWithStrings>(input_column)) {
+                        auto input_column_with_strings = std::get<ColumnWithStrings>(input_column);
+                        agg_data->aggregate(input_column_with_strings);
+                    } else {
+                        if(!that->processing_config_.dynamic_schema_)
+                            internal::raise<ErrorCode::E_ASSERTION_FAILURE>(
+                                    "Unable to resolve column denoted by aggregation operator: '{}'",
+                                    input_column_name);
                     }
-                    stream::merge_string_columns(slice_and_key.segment(store), new_segment.string_pool_ptr(), dedup_rows_);
                 }
-                const RowRange row_range{min_start_row, max_end_row};
-                const ColRange col_range{min_start_col, max_end_col};
-                output.push_back(ProcessingSegment{std::move(new_segment), FrameSlice{col_range, row_range}});
             });
-        },
-        [&](const auto &) {
-            util::raise_rte("Column partition removal only supports datetime indexed data. You data does not have a datetime index.");
+
+    internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+            start_indexes.size() == 1 && end_indexes.size() == 1,
+            "Expected all data segments in one processing segment to have same start and end indexes");
+    auto start_index = *start_indexes.begin();
+    auto end_index = *end_indexes.begin();
+    schema::check<ErrorCode::E_UNSUPPORTED_INDEX_TYPE>(
+            std::holds_alternative<NumericIndex>(start_index) && std::holds_alternative<NumericIndex>(end_index),
+            "Cannot build column stats over string-indexed symbol"
+    );
+    auto start_index_col = std::make_shared<Column>(make_scalar_type(DataType::MICROS_UTC64), true);
+    auto end_index_col = std::make_shared<Column>(make_scalar_type(DataType::MICROS_UTC64), true);
+    start_index_col->template push_back<NumericIndex>(std::get<NumericIndex>(start_index));
+    end_index_col->template push_back<NumericIndex>(std::get<NumericIndex>(end_index));
+    start_index_col->set_row_data(0);
+    end_index_col->set_row_data(0);
+
+    SegmentInMemory seg;
+    seg.descriptor().set_index(IndexDescriptor(0, IndexDescriptor::ROWCOUNT));
+    seg.add_column(scalar_field_proto(DataType::MICROS_UTC64, start_index_column_name), start_index_col);
+    seg.add_column(scalar_field_proto(DataType::MICROS_UTC64, end_index_column_name), end_index_col);
+    for (const auto& agg_data: folly::enumerate(aggregators_data)) {
+        seg.concatenate(agg_data->finalize(column_stats_aggregators_->at(agg_data.index).get_output_column_names()));
+    }
+    seg.set_row_id(0);
+    return Composite{ProcessingSegment{std::move(seg)}};
+}
+
+Composite<ProcessingSegment> RowNumberLimitClause::process(std::shared_ptr<Store> store,
+                                                           Composite<ProcessingSegment> &&p) const {
+    auto procs = std::move(p);
+    procs.broadcast([&store, this](ProcessingSegment &proc) {
+        auto row_range = proc.data()[0].slice_.row_range;
+        if (row_range.start() == 0ULL && row_range.end() > n) {
+            util::BitSet bv(static_cast<util::BitSet::size_type>(row_range.diff()));
+            bv.set_range(0, bv_size(n));
+            proc.apply_filter(bv, store, PipelineOptimisation::MEMORY);
+        } else if (!warning_shown) {
+            warning_shown = true;
+            log::version().info("RowNumberLimitFilter bypassed because rows.start() == {}", row_range.start());
         }
-        );
-        return output;
+    });
+    return procs;
 }
 
 }
\ No newline at end of file
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
index 0061ac8552..05dc3efe01 100644
--- a/cpp/arcticdb/processing/clause.hpp
+++ b/cpp/arcticdb/processing/clause.hpp
@@ -10,7 +10,7 @@
 #include <arcticdb/entity/key.hpp>
 #include <arcticdb/column_store/column.hpp>
 #include <arcticdb/pipeline/value.hpp>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/processing/processing_segment.hpp>
 #include <arcticdb/entity/types.hpp>
@@ -34,6 +34,29 @@
 
 namespace arcticdb {
 
+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};
+};
+
+
 struct IClause {
     template<class Base>
     struct Interface : Base {
@@ -47,153 +70,163 @@ struct IClause {
             return folly::poly_call<1>(*this, std::move(comps));
         }
 
-        [[nodiscard]] bool requires_repartition() const { return folly::poly_call<2>(*this); }
+        [[nodiscard]] const ClauseInfo& clause_info() const { return folly::poly_call<2>(*this); };
 
-        [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return folly::poly_call<3>(*this); }
-
-        [[nodiscard]] bool can_combine_with_column_selection() const { return folly::poly_call<4>(*this); }
+        void set_processing_config(const ProcessingConfig& processing_config) {
+            folly::poly_call<3>(*this, processing_config);
+        }
     };
 
     template<class T>
-    using Members = folly::PolyMembers<&T::process, &T::repartition, &T::requires_repartition, &T::execution_context, &T::can_combine_with_column_selection>;
+    using Members = folly::PolyMembers<
+            &T::process,
+            &T::repartition,
+            &T::clause_info,
+            &T::set_processing_config>;
 };
 
 using Clause = folly::Poly<IClause>;
 
-std::vector<Composite<ProcessingSegment>> single_partition(std::vector<Composite<ProcessingSegment>> &&segs);
-
-struct ColumnStatsGenerationClause {
-    std::shared_ptr<ExecutionContext> execution_context_;
-    std::shared_ptr<std::vector<ColumnStatsAggregator>> column_stats_aggregators_;
+struct PassthroughClause {
+    ClauseInfo clause_info_;
+    PassthroughClause() = default;
+    ARCTICDB_MOVE_COPY_DEFAULT(PassthroughClause)
+
+    [[nodiscard]] Composite<ProcessingSegment> process(
+            ARCTICDB_UNUSED const std::shared_ptr<Store> &store,
+            Composite<ProcessingSegment> &&p
+            ) const;
+
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+        return std::nullopt;
+    }
 
-    explicit ColumnStatsGenerationClause(std::shared_ptr<ExecutionContext> execution_context,
-                                         std::shared_ptr<std::vector<ColumnStatsAggregator>> column_stats_aggregators) :
-            execution_context_(std::move(execution_context)),
-            column_stats_aggregators_(std::move(column_stats_aggregators)) {
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {}
+};
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return execution_context_; }
+struct FilterClause {
+    ClauseInfo clause_info_;
+    std::shared_ptr<ExpressionContext> expression_context_;
+    PipelineOptimisation optimisation_;
+
+    explicit FilterClause(std::unordered_set<std::string> input_columns,
+                          ExpressionContext expression_context,
+                          std::optional<PipelineOptimisation> optimisation) :
+            expression_context_(std::make_shared<ExpressionContext>(std::move(expression_context))),
+            optimisation_(optimisation.value_or(PipelineOptimisation::SPEED)) {
+        clause_info_.input_columns_ = std::move(input_columns);
+    }
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
+    FilterClause() = delete;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&) const { return std::nullopt; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return false; }
-};
+    ARCTICDB_MOVE_COPY_DEFAULT(FilterClause)
 
-struct FilterClause {
-    std::shared_ptr<ExecutionContext> execution_context_;
+    [[nodiscard]] Composite<ProcessingSegment> process(
+            std::shared_ptr<Store> store,
+            Composite<ProcessingSegment> &&p
+            ) const;
 
-    explicit FilterClause(std::shared_ptr<ExecutionContext> execution_context) :
-        execution_context_(std::move(execution_context)) {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+        return std::nullopt;
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return execution_context_; }
+    void set_processing_config(const ProcessingConfig& processing_config) {
+        expression_context_->dynamic_schema_ = processing_config.dynamic_schema_;
+    }
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
+    [[nodiscard]] std::string to_string() const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&) const { return std::nullopt; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    void set_pipeline_optimisation(PipelineOptimisation pipeline_optimisation) {
+        optimisation_ = pipeline_optimisation;
+    }
 };
 
-/**
- * Hacky short-cut to reduce memory usage for head(). Very similar to how FilterClause above works.
- * Limitations:
- * - Must be the only Clause (otherwise may return wrong results)
- * - If the n is so big that multiple RowRanges/ProcessingSegments are passed in then much more state tracking is
- * required than what is reasonable for a stop-gap, so this optimisation just disables itself.
- */
-struct RowNumberLimitClause {
-    size_t n;
+struct ProjectClause {
+    ClauseInfo clause_info_;
+    std::string output_column_;
+    std::shared_ptr<ExpressionContext> expression_context_;
+
+    explicit ProjectClause(std::unordered_set<std::string> input_columns,
+                           std::string output_column,
+                           ExpressionContext expression_context) :
+            output_column_(output_column),
+            expression_context_(std::make_shared<ExpressionContext>(std::move(expression_context))) {
+        clause_info_.input_columns_ = std::move(input_columns);
+        clause_info_.modifies_output_descriptor_ = true;
+    }
 
-    explicit RowNumberLimitClause(size_t n) : n(n) {}
+    ProjectClause() = delete;
+
+    ARCTICDB_MOVE_COPY_DEFAULT(ProjectClause)
 
     [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-        auto procs = std::move(p);
-        procs.broadcast([&store, this](ProcessingSegment &proc) {
-            auto row_range = proc.data()[0].slice_.row_range;
-            if (row_range.start() == 0ULL && row_range.end() > n) {
-                util::BitSet bv(static_cast<util::BitSet::size_type>(row_range.diff()));
-                bv.set_range(0, bv_size(n));
-                proc.apply_filter(bv, store);
-            } else if (!warning_shown) {
-                warning_shown = true;
-                log::version().info("RowNumberLimitFilter bypassed because rows.start() == {}", row_range.start());
-            }
-        });
-        return procs;
+    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
+
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+        return std::nullopt;
     }
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    mutable bool warning_shown = false; // folly::Poly can't deal with atomic_bool
-    [[nodiscard]] bool requires_repartition() const { return false; }
+    void set_processing_config(const ProcessingConfig& processing_config) {
+        expression_context_->dynamic_schema_ = processing_config.dynamic_schema_;
+    }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&) const { return std::nullopt; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    [[nodiscard]] std::string to_string() const;
 };
 
 template<typename GrouperType, typename BucketizerType>
 struct PartitionClause {
-    std::shared_ptr<ExecutionContext> execution_context_;
-
-    explicit PartitionClause(std::shared_ptr<ExecutionContext> execution_context) :
-            execution_context_(std::move(execution_context)) {
+    ClauseInfo clause_info_;
+    ProcessingConfig processing_config_;
+    std::string grouping_column_;
+
+    explicit PartitionClause(const std::string& grouping_column) :
+            grouping_column_(grouping_column) {
+        clause_info_.input_columns_ = {grouping_column_};
+        clause_info_.requires_repartition_ = true;
+        clause_info_.modifies_output_descriptor_ = true;
     }
+    PartitionClause() = delete;
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-        Composite<ProcessingSegment> output;
+    ARCTICDB_MOVE_COPY_DEFAULT(PartitionClause)
 
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const {
+        Composite<ProcessingSegment> output;
         auto procs = std::move(p);
-        procs.broadcast([&output, &store, &execution_context = execution_context_](auto &proc) {
-            proc.set_execution_context(execution_context);
-            // TODO (AN-469): I would push all of the logic between here and calling partition_processing_segment down
-            // into partition_processing_segment, just passing the column name to group on, processing segment, and
-            // store in
-            auto partitioning_column = proc.get(ColumnName(execution_context->root_node_name_.value), store);
-            if (std::holds_alternative<ColumnWithStrings>(partitioning_column)) {
-                ColumnWithStrings col = std::get<ColumnWithStrings>(partitioning_column);
-
-                col.column_->type().visit_tag([&output, &proc, &col, &store](auto type_desc_tag) {
-                    using TypeDescriptorTag = decltype(type_desc_tag);
-
-                    using ResolvedGrouperType = typename GrouperType::template Grouper<TypeDescriptorTag>;
-
-                    auto grouper = std::make_shared<ResolvedGrouperType>(ResolvedGrouperType());
-                    // TODO (AN-469): We should put some thought into how to pick an appropriate value for num_buckets
-                    auto num_cores = std::thread::hardware_concurrency() == 0 ? 16 : std::thread::hardware_concurrency();
-                    auto num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets", num_cores);
-                    auto bucketizer = std::make_shared<BucketizerType>(num_buckets);
-
-                    output.push_back(
-                            partition_processing_segment<ResolvedGrouperType, BucketizerType>(proc, col, store, grouper, bucketizer));
-                });
-            } else {
-                util::raise_rte("Expected single column from expression");
-            }
+        procs.broadcast([&output, &store, that=this](auto &proc) {
+            output.push_back(partition_processing_segment<GrouperType, BucketizerType>(store,
+                                                                                       proc,
+                                                                                       ColumnName(that->grouping_column_),
+                                                                                       that->processing_config_.dynamic_schema_));
         });
-
         return output;
     }
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return execution_context_; }
-
-    [[nodiscard]] bool requires_repartition() const { return true; }
-
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&c) const {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&c) const {
         auto comps = std::move(c);
         std::unordered_map<size_t, Composite<ProcessingSegment>> partition_map;
+        schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
+                std::any_of(comps.begin(), comps.end(), [](const Composite<ProcessingSegment>& proc) {
+                    return !proc.empty();
+                }),
+                "Grouping column {} does not exist or is empty", grouping_column_
+        );
 
         for (auto &comp : comps) {
             comp.broadcast([&partition_map](auto &proc) {
@@ -214,285 +247,222 @@ struct PartitionClause {
 
         return ret;
     }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
-};
-
-struct AggregationClause {
-    std::shared_ptr<ExecutionContext> execution_context_;
-    std::vector<AggregationFactory> aggregation_operators_;
-    std::shared_ptr<std::once_flag> reset_descriptor_ = std::make_shared<std::once_flag>();
-    std::shared_ptr<std::once_flag> set_name_index_ = std::make_shared<std::once_flag>();
-    AggregationClause() = delete;
-
-    ARCTICDB_MOVE_COPY_DEFAULT(AggregationClause)
 
-    AggregationClause(std::shared_ptr<ExecutionContext> execution_context,
-                               std::vector<AggregationFactory> &&aggregation_operators) :
-            execution_context_(std::move(execution_context)),
-            aggregation_operators_(std::forward<std::vector<AggregationFactory>>(aggregation_operators)) {
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
-
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return execution_context_; }
-
-    [[nodiscard]] bool requires_repartition() const { return false; }
-
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const { return std::nullopt; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return false; }
-};
+    void set_processing_config(const ProcessingConfig& processing_config) {
+        processing_config_ = processing_config;
+    }
 
-struct PassthroughClause {
-    [[nodiscard]] Composite<ProcessingSegment>
-    process([[maybe_unused]] const std::shared_ptr<Store> &store, Composite<ProcessingSegment> &&p) const {
-        auto procs = std::move(p);
-        return procs;
+    [[nodiscard]] std::string to_string() const {
+        return fmt::format("GROUPBY Column[\"{}\"]", grouping_column_);
     }
+};
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
+struct AggregationClause {
+    ClauseInfo clause_info_;
+    ProcessingConfig processing_config_;
+    std::string grouping_column_;
+    std::unordered_map<std::string, std::string> aggregation_map_;
+    std::vector<GroupingAggregator> aggregators_;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const { return std::nullopt; }
+    AggregationClause() = delete;
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    ARCTICDB_MOVE_COPY_DEFAULT(AggregationClause)
 
-};
+    AggregationClause(const std::string& grouping_column,
+                      const std::unordered_map<std::string,
+                      std::string>& aggregations);
 
-struct SliceAndKeyWrapper {
-    pipelines::SliceAndKey seg_;
-    std::shared_ptr<Store> store_;
-    SegmentInMemory::iterator it_;
-    const StreamId id_;
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-    explicit SliceAndKeyWrapper(pipelines::SliceAndKey &&seg, std::shared_ptr<Store> store) :
-            seg_(std::move(seg)),
-            store_(std::move(store)),
-            it_(seg_.segment(store_).begin()),
-            id_(seg_.segment(store_).descriptor().id()) {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps
+            ) const {
+        return std::nullopt;
     }
 
-    bool advance() {
-        return ++it_ != seg_.segment(store_).end();
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
     }
 
-    SegmentInMemory::Row &row() {
-        return *it_;
+    void set_processing_config(const ProcessingConfig& processing_config) {
+        processing_config_ = processing_config;
     }
 
-    const StreamId &id() const {
-        return id_;
-    }
+    [[nodiscard]] std::string to_string() const;
 };
 
-struct MergeClause {
-    stream::Index index_;
-    stream::VariantColumnPolicy density_policy_;
-    StreamId stream_id_;
-    bool add_symbol_column_ = false;
-    StreamId target_id_;
-    const std::shared_ptr<ExecutionContext> execution_context_;
+struct RemoveColumnPartitioningClause {
+    ClauseInfo clause_info_;
+    mutable bool warning_shown = false; // folly::Poly can't deal with atomic_bool
 
-    MergeClause(
-            stream::Index index,
-            const stream::VariantColumnPolicy &density_policy,
-            const StreamId& stream_id,
-            std::shared_ptr<ExecutionContext> execution_context ) :
-            index_(std::move(index)),
-            density_policy_(density_policy),
-            stream_id_(stream_id),
-            execution_context_(std::move(execution_context)) {
+    RemoveColumnPartitioningClause() {
+        clause_info_.can_combine_with_column_selection_ = false;
     }
+    ARCTICDB_MOVE_COPY_DEFAULT(RemoveColumnPartitioningClause)
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
-
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+        return std::nullopt;
+    }
 
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const {
-        return single_partition(std::move(comps));
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
     }
 
-    [[nodiscard]] bool requires_repartition() const { return true; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {
+    }
 };
 
 struct SplitClause {
+    ClauseInfo clause_info_;
     const size_t rows_;
 
     explicit SplitClause(size_t rows) :
             rows_(rows) {
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&procs) const {
-        using namespace arcticdb::pipelines;
-
-        auto proc_composite = std::move(procs);
-        Composite<ProcessingSegment> ret;
-        proc_composite.broadcast([&store, rows = rows_, &ret](auto &&p) {
-            auto proc = std::forward<decltype(p)>(p);
-            auto slice_and_keys = proc.data();
-            for (auto &slice_and_key: slice_and_keys) {
-                auto split_segs = slice_and_key.segment(store).split(rows);
-                const ColRange col_range{slice_and_key.slice().col_range};
-                size_t start_row = slice_and_key.slice().row_range.start();
-                size_t end_row = 0;
-                for (auto &item : split_segs) {
-                    end_row = start_row + item.row_count();
-                    const RowRange row_range{start_row, end_row};
-                    ret.push_back(ProcessingSegment{std::move(item), FrameSlice{col_range, row_range}});
-                    start_row = end_row;
-                }
-            }
-        });
-        return ret;
-    }
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&procs) const;
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+        return std::nullopt;
+    }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const { return std::nullopt; }
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {}
 };
 
 struct SortClause {
+    ClauseInfo clause_info_;
     const std::string column_;
 
     explicit SortClause(std::string column) :
             column_(std::move(column)) {
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const {
-        auto procs = std::move(p);
-        procs.broadcast([&store, &column = column_](auto &proc) {
-            auto slice_and_keys = proc.data();
-            for (auto &slice_and_key: slice_and_keys) {
-                slice_and_key.segment(store).sort(column);
-            }
-        });
-        return procs;
-    }
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&comps) const {
+        return std::nullopt;
+    }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const { return std::nullopt; }
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {}
 };
 
-struct ProjectClause {
-    //TODO can't use root node name all the time, do something like this
-    //const std::shared_ptr<std::map<std::string, std::string>> projections_;
-    std::string column_name_;
-    const std::shared_ptr<ExecutionContext> execution_context_;
-    std::shared_ptr<std::once_flag> add_column_ = std::make_shared<std::once_flag>();
+struct MergeClause {
+    ClauseInfo clause_info_;
+    stream::Index index_;
+    stream::VariantColumnPolicy density_policy_;
+    StreamId stream_id_;
+    bool add_symbol_column_ = false;
+    StreamId target_id_;
+    StreamDescriptor stream_descriptor_;
 
-    //TODO support multiple projections
-    //explicit ProjectClause(std::shared_ptr<std::map<std::string, std::string>> projections) :
-    explicit ProjectClause(const std::string& column_name, std::shared_ptr<ExecutionContext> execution_context) :
-        column_name_(column_name),
-        execution_context_(std::move(execution_context)) {
+    MergeClause(
+            stream::Index index,
+            const stream::VariantColumnPolicy &density_policy,
+            const StreamId& stream_id,
+            const StreamDescriptor& stream_descriptor) :
+            index_(std::move(index)),
+            density_policy_(density_policy),
+            stream_id_(stream_id),
+            stream_descriptor_(stream_descriptor) {
+        clause_info_.requires_repartition_ = true;
     }
 
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return execution_context_; }
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            std::vector<Composite<ProcessingSegment>> &&comps) const;
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&comps) const { return std::nullopt; }
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    [[nodiscard]] bool requires_repartition() const { return false; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return true; }
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {
+    }
 };
 
-class ClauseBuilder {
-private:
-    std::vector<Clause> clauses_;
-    std::shared_ptr<ExecutionContext> execution_context_;
-    std::vector<AggregationFactory> operators_;
+struct ColumnStatsGenerationClause {
+    ClauseInfo clause_info_;
+    ProcessingConfig processing_config_;
+    std::shared_ptr<std::vector<ColumnStatsAggregator>> column_stats_aggregators_;
 
-public:
-    void add_ProjectClause(const std::string& column_name, const std::shared_ptr<ExecutionContext>& ec) {
-        clauses_.emplace_back(ProjectClause(column_name, ec));
+    explicit ColumnStatsGenerationClause(std::unordered_set<std::string>&& input_columns,
+                                         std::shared_ptr<std::vector<ColumnStatsAggregator>> column_stats_aggregators) :
+            column_stats_aggregators_(std::move(column_stats_aggregators)) {
+        clause_info_.input_columns_ = std::move(input_columns);
+        clause_info_.can_combine_with_column_selection_ = false;
+        clause_info_.modifies_output_descriptor_ = true;
     }
 
-    void add_FilterClause(const std::shared_ptr<ExecutionContext>& ec) {
-        clauses_.emplace_back(FilterClause(ec));
-    }
+    ARCTICDB_MOVE_COPY_DEFAULT(ColumnStatsGenerationClause)
 
-    void prepare_AggregationClause(const std::shared_ptr<ExecutionContext>& ec) {
-        execution_context_ = ec;
-        if (!operators_.empty()) {
-            util::raise_rte("Prior aggregation clause not finalised before adding new aggregation clause!");
-        }
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-        operators_ = std::vector<AggregationFactory>{};
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+        return std::nullopt;
     }
 
-    void add_SumAggregationOperator(ColumnName &&input, ColumnName &&output) {
-        operators_.emplace_back(Sum{std::move(input), std::move(output)});
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
     }
 
-    void add_MeanAggregationOperator(ColumnName &&input, ColumnName &&output) {
-        operators_.emplace_back(Mean{std::move(input), std::move(output)});
+    void set_processing_config(const ProcessingConfig& processing_config) {
+        processing_config_ = processing_config;
     }
+};
 
-    void add_MaxAggregationOperator(ColumnName &&input, ColumnName &&output) {
-        operators_.emplace_back(MaxOrMin{std::move(input), std::move(output), Extremum::max});
-    }
+/**
+ * Hacky short-cut to reduce memory usage for head(). Very similar to how FilterClause above works.
+ * Limitations:
+ * - Must be the only Clause (otherwise may return wrong results)
+ * - If the n is so big that multiple RowRanges/ProcessingSegments are passed in then much more state tracking is
+ * required than what is reasonable for a stop-gap, so this optimisation just disables itself.
+ */
+struct RowNumberLimitClause {
+    ClauseInfo clause_info_;
+    size_t n;
+    mutable bool warning_shown = false; // folly::Poly can't deal with atomic_bool
 
-    void add_MinAggregationOperator(ColumnName &&input, ColumnName &&output) {
-        operators_.emplace_back(MaxOrMin{std::move(input), std::move(output), Extremum::min});
-    }
+    explicit RowNumberLimitClause(size_t n) : n(n) {}
 
-    void finalize_AggregationClause() {
-        // TODO: Complete hack. Two clauses shouldn't share an EC.
-        clauses_.emplace_back(
-                 PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer>{execution_context_});
-        clauses_.emplace_back(AggregationClause{execution_context_, std::move(operators_)});
-        execution_context_.reset();
-    }
+    ARCTICDB_MOVE_COPY_DEFAULT(RowNumberLimitClause)
 
-    [[nodiscard]] std::vector<Clause> get_clauses() const {
-        return clauses_;
-    }
-};
+    [[nodiscard]] Composite<ProcessingSegment> process(std::shared_ptr<Store> store,
+                                                       Composite<ProcessingSegment> &&p) const;
 
-struct RemoveColumnPartitioningClause {
-    std::shared_ptr<ExecutionContext> execution_context_;
-    const bool dedup_rows_;
-    const arcticdb::proto::descriptors::IndexDescriptor::Type descriptor_type_;
-
-    RemoveColumnPartitioningClause(
-            std::shared_ptr<ExecutionContext> execution_context,
-            const arcticdb::proto::descriptors::IndexDescriptor::Type descriptor_type = IndexDescriptor::ROWCOUNT,
-            bool dedup_rows = false) :
-            execution_context_(std::move(execution_context)),
-            dedup_rows_(dedup_rows),
-            descriptor_type_(descriptor_type) {
+    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>> repartition(
+            ARCTICDB_UNUSED std::vector<Composite<ProcessingSegment>> &&) const {
+        return std::nullopt;
     }
-    [[nodiscard]] Composite<ProcessingSegment>
-    process(std::shared_ptr<Store> store, Composite<ProcessingSegment> &&p) const;
-
-    [[nodiscard]] std::shared_ptr<ExecutionContext> execution_context() const { return nullptr; }
 
-    mutable bool warning_shown = false; // folly::Poly can't deal with atomic_bool
-    [[nodiscard]] bool requires_repartition() const { return false; }
+    [[nodiscard]] const ClauseInfo& clause_info() const {
+        return clause_info_;
+    }
 
-    [[nodiscard]] std::optional<std::vector<Composite<ProcessingSegment>>>
-    repartition([[maybe_unused]] std::vector<Composite<ProcessingSegment>> &&) const { return std::nullopt; }
-    [[nodiscard]] bool can_combine_with_column_selection() const { return false; }
+    void set_processing_config(ARCTICDB_UNUSED const ProcessingConfig& processing_config) {}
 };
 
 }//namespace arcticdb
diff --git a/cpp/arcticdb/processing/execution_context.hpp b/cpp/arcticdb/processing/execution_context.hpp
deleted file mode 100644
index af91635d09..0000000000
--- a/cpp/arcticdb/processing/execution_context.hpp
+++ /dev/null
@@ -1,115 +0,0 @@
-/* Copyright 2023 Man Group Operations Limited
- *
- * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
- *
- * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
- */
-
-#pragma once
-
-#include <memory>
-#include <string>
-#include <unordered_map>
-#include <unordered_set>
-
-#include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/pipeline/value.hpp>
-#include <arcticdb/pipeline/value_set.hpp>
-
-namespace arcticdb {
-
-/*
- * Contains an expression tree as a map of string node names to nodes. Nodes contain leaves which may be references
- * to node names.
- *
- * ExecutionContext also contains the name of the root node as well as a map of node names to constant values.
- *
- * This class effectively contains as AST - there is likely a frontend implementation in a higher level language (for
- * example Python).
- */
-struct ExecutionContext {
-    ExecutionContext() = default;
-    using NormMetaDescriptor = std::shared_ptr<arcticdb::proto::descriptors::NormalizationMetadata>;
-
-    ARCTICDB_MOVE_ONLY_DEFAULT(ExecutionContext)
-
-    template <class T>
-    class ConstantMap {
-        std::unordered_map<std::string, std::shared_ptr<T>> map_;
-    public:
-        void set_value(std::string name, std::shared_ptr<T> val) {
-            map_.try_emplace(name, val);
-        }
-        std::shared_ptr<T> get_value(std::string name) {
-            return map_.at(name);
-        }
-    };
-    enum class Optimisation : uint8_t {
-        SPEED,
-        MEMORY
-    };
-    void add_column(const std::string& column_name) {
-        columns_.emplace(column_name);
-    }
-
-    void add_expression_node(const std::string& name, std::shared_ptr<ExpressionNode> expression_node) {
-        expression_nodes_.set_value(name, std::move(expression_node));
-    }
-
-    void set_descriptor(const StreamDescriptor& desc) {
-        output_descriptor_ = desc;
-    }
-
-    void set_norm_meta_descriptor(NormMetaDescriptor& desc) {
-        norm_meta_ = desc;
-    }
-
-    NormMetaDescriptor get_norm_meta_descriptor() {
-        return norm_meta_;
-    }
-
-    void add_value(const std::string& name, std::shared_ptr<Value> value) {
-        values_.set_value(name, std::move(value));
-    }
-
-    void add_value_set(const std::string& name, std::shared_ptr<ValueSet> value_set) {
-        value_sets_.set_value(name, std::move(value_set));
-    }
-
-    void check_output_column(const std::string& name, DataType type) {
-        std::scoped_lock lock{*column_mutex_};
-        auto it = output_columns_.find(name);
-        if(it == std::end(output_columns_)) {
-            output_columns_.try_emplace(name, type);
-            output_descriptor_->add_field(scalar_field_proto(type, name));
-            //TODO check type compatibility in dynamic schema
-        }
-    }
-
-    bool dynamic_schema() const {
-        return dynamic_schema_;
-    }
-
-    void set_dynamic_schema(bool dynamic_schema) {
-        dynamic_schema_ = dynamic_schema;
-    }
-
-    std::unordered_set<std::string> columns_;
-    ConstantMap<ExpressionNode> expression_nodes_;
-    ConstantMap<Value> values_;
-    ConstantMap<ValueSet> value_sets_;
-    ExpressionName root_node_name_;
-    Optimisation optimisation_ = Optimisation::SPEED;
-    // TODO (AN-469): This class should be immutable (and possibly just incorporated into Clause, as there is always
-    //  exactly one execution context per clause). Any mutable state that needs to be maintained as a ProcessingSegment
-    //  moves through the processing pipeline should be stored somewhere else (possibly on the ProcessingSegment itself)
-    std::optional<StreamDescriptor> output_descriptor_;
-    std::optional<StreamDescriptor> orig_output_descriptor_;
-    NormMetaDescriptor norm_meta_;
-    std::unordered_map<std::string, DataType> output_columns_;
-    std::unique_ptr<std::mutex> column_mutex_ = std::make_unique<std::mutex>();
-    std::unique_ptr<std::mutex> name_index_mutex_ = std::make_unique<std::mutex>();
-    bool dynamic_schema_ = false;
-};
-
-}//namespace arcticdb
diff --git a/cpp/arcticdb/processing/expression_context.hpp b/cpp/arcticdb/processing/expression_context.hpp
new file mode 100644
index 0000000000..9ea1ccf058
--- /dev/null
+++ b/cpp/arcticdb/processing/expression_context.hpp
@@ -0,0 +1,65 @@
+/* Copyright 2023 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#include <arcticdb/processing/expression_node.hpp>
+#include <arcticdb/pipeline/value.hpp>
+#include <arcticdb/pipeline/value_set.hpp>
+
+namespace arcticdb {
+
+/*
+ * Contains an expression tree as a map of string node names to nodes. Nodes contain leaves which may be references
+ * to node names.
+ *
+ * ExpressionContext also contains the name of the root node as well as a map of node names to constant values.
+ *
+ * This class effectively contains as AST - there is likely a frontend implementation in a higher level language (for
+ * example Python).
+ */
+struct ExpressionContext {
+    ExpressionContext() = default;
+
+    ARCTICDB_MOVE_COPY_DEFAULT(ExpressionContext)
+
+    template <class T>
+    class ConstantMap {
+        std::unordered_map<std::string, std::shared_ptr<T>> map_;
+    public:
+        void set_value(std::string name, std::shared_ptr<T> val) {
+            map_.try_emplace(name, val);
+        }
+        std::shared_ptr<T> get_value(std::string name) {
+            return map_.at(name);
+        }
+    };
+
+    void add_expression_node(const std::string& name, std::shared_ptr<ExpressionNode> expression_node) {
+        expression_nodes_.set_value(name, std::move(expression_node));
+    }
+
+    void add_value(const std::string& name, std::shared_ptr<Value> value) {
+        values_.set_value(name, std::move(value));
+    }
+
+    void add_value_set(const std::string& name, std::shared_ptr<ValueSet> value_set) {
+        value_sets_.set_value(name, std::move(value_set));
+    }
+
+    ConstantMap<ExpressionNode> expression_nodes_;
+    ConstantMap<Value> values_;
+    ConstantMap<ValueSet> value_sets_;
+    ExpressionName root_node_name_;
+    bool dynamic_schema_{false};
+};
+
+}//namespace arcticdb
diff --git a/cpp/arcticdb/processing/grouper.hpp b/cpp/arcticdb/processing/grouper.hpp
index a45d53995b..2c8d1c1fea 100644
--- a/cpp/arcticdb/processing/grouper.hpp
+++ b/cpp/arcticdb/processing/grouper.hpp
@@ -13,7 +13,7 @@
 #include <fmt/core.h>
 
 #include <arcticdb/column_store/memory_segment.hpp>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/pipeline/frame_slice.hpp>
 #include <arcticdb/pipeline/filter_segment.hpp>
@@ -35,12 +35,15 @@ class HashingGroupers {
         using DataTypeTag = typename GrouperDescriptor::DataTypeTag;
         using RawType = typename DataTypeTag::raw_type;
 
-        size_t group(RawType key, std::shared_ptr<StringPool> sp) const {
+        std::optional<size_t> group(RawType key, std::shared_ptr<StringPool> sp) const {
             constexpr DataType dt = DataTypeTag::data_type;
-            HashedValue hash_result;
+            std::optional<HashedValue> hash_result;
             if constexpr(dt == DataType::ASCII_FIXED64 || dt == DataType::ASCII_DYNAMIC64 || dt == DataType::UTF_FIXED64 || dt == DataType::UTF_DYNAMIC64) {
-                // TODO (AN-468): This will throw on Nones/NaNs
-                hash_result = hash(sp->get_view(key));
+                if (is_a_string(key)) {
+                    hash_result = hash(sp->get_view(key));
+                } else {
+                    hash_result = std::nullopt;
+                }
             } else {
                 hash_result = hash<RawType>(&key, 1);
             }
diff --git a/cpp/arcticdb/processing/processing_segment.cpp b/cpp/arcticdb/processing/processing_segment.cpp
index f97a2efc8c..614d9b4894 100644
--- a/cpp/arcticdb/processing/processing_segment.cpp
+++ b/cpp/arcticdb/processing/processing_segment.cpp
@@ -8,12 +8,11 @@
 #include <arcticdb/processing/processing_segment.hpp>
 
 namespace arcticdb {
-void ProcessingSegment::apply_filter(const util::BitSet& bitset,
-                  const std::shared_ptr<Store>& store) {
-    // execution_context_ is null with the legacy filtering implementation
-    // Remove null check once legacy filtering is deprecated
-    auto filter_down_stringpool = execution_context_ &&
-        execution_context_->optimisation_ == ExecutionContext::Optimisation::MEMORY;
+void ProcessingSegment::apply_filter(
+    const util::BitSet& bitset,
+    const std::shared_ptr<Store>& store,
+    PipelineOptimisation optimisation) {
+    auto filter_down_stringpool = optimisation == PipelineOptimisation::MEMORY;
 
     for (auto& slice_and_key: data_) {
         auto seg = filter_segment(slice_and_key.segment(store),
@@ -31,46 +30,49 @@ void ProcessingSegment::apply_filter(const util::BitSet& bitset,
 }
 
 VariantData ProcessingSegment::get(const VariantNode &name, const std::shared_ptr<Store> &store) {
-    return util::variant_match(name, [&](const ColumnName &column_name) {
+    return util::variant_match(name,
+        [&](const ColumnName &column_name) {
         for (auto &slice_and_key: data_) {
             slice_and_key.segment(store).init_column_map();
-            if (auto opt_idx = slice_and_key.segment(store).column_index(
-                column_name.value))
+            if (auto opt_idx = slice_and_key.segment(store).column_index(column_name.value)) {
                 return VariantData(ColumnWithStrings(
-                    slice_and_key.segment(store).column_ptr(
+                        slice_and_key.segment(store).column_ptr(
                         position_t(position_t(opt_idx.value()))),
                         slice_and_key.segment(store).string_pool_ptr()));
+            }
         }
         // Try multi-index column names
         std::string multi_index_column_name = fmt::format("__idx__{}",
                                                           column_name.value);
         for (auto &slice_and_key: data_) {
-            if (auto opt_idx = slice_and_key.segment(store).column_index(
-                multi_index_column_name))
+            if (auto opt_idx = slice_and_key.segment(store).column_index(multi_index_column_name)) {
                 return VariantData(ColumnWithStrings(
-                    slice_and_key.segment(store).column_ptr(
+                        slice_and_key.segment(store).column_ptr(
                         position_t(opt_idx.value())),
                         slice_and_key.segment(store).string_pool_ptr()));
+            }
         }
-        if(execution_context_->dynamic_schema()) {
-            log::version().debug("Column not found in {}", data_[0].segment(store).descriptor());
+        if (expression_context_ && !expression_context_->dynamic_schema_) {
+            internal::raise<ErrorCode::E_ASSERTION_FAILURE>("Column {} not found in {}",
+                                                            column_name,
+                                                            data_[0].segment(store).descriptor());
+        } else {
+            log::version().debug("Column {} not found in {}", column_name, data_[0].segment(store).descriptor());
             return VariantData{EmptyResult{}};
         }
-        else
-            util::raise_rte("Unexpected column name");
         },
         [&](const ValueName &value_name) {
-        return VariantData(execution_context_->values_.get_value(value_name.value));
+        return VariantData(expression_context_->values_.get_value(value_name.value));
         },
         [&](const ValueSetName &value_set_name) {
-        return VariantData(execution_context_->value_sets_.get_value(value_set_name.value));
+        return VariantData(expression_context_->value_sets_.get_value(value_set_name.value));
         },
         [&](const ExpressionName &expression_name) {
         if (auto computed = computed_data_.find(expression_name.value);
         computed != std::end(computed_data_)) {
             return computed->second;
         } else {
-            auto expr = execution_context_->expression_nodes_.get_value(expression_name.value);
+            auto expr = expression_context_->expression_nodes_.get_value(expression_name.value);
             auto data = expr->compute(self(), store);
             computed_data_.try_emplace(expression_name.value, data);
             return data;
diff --git a/cpp/arcticdb/processing/processing_segment.hpp b/cpp/arcticdb/processing/processing_segment.hpp
index 106f891163..7a326e179f 100644
--- a/cpp/arcticdb/processing/processing_segment.hpp
+++ b/cpp/arcticdb/processing/processing_segment.hpp
@@ -12,8 +12,9 @@
 
 #include <fmt/core.h>
 
+#include <arcticdb/async/task_scheduler.hpp>
 #include <arcticdb/column_store/memory_segment.hpp>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/pipeline/frame_slice.hpp>
 #include <arcticdb/pipeline/filter_segment.hpp>
@@ -22,6 +23,11 @@
 #include <arcticdb/util/variant.hpp>
 
 namespace arcticdb {
+    enum class PipelineOptimisation : uint8_t {
+        SPEED,
+        MEMORY
+    };
+
     /*
      * A processing segment is designed to be used in conjunction with the clause processing framework.
      * Each clause will execute in turn and will be passed the same mutable ProcessingSegment which it will have free
@@ -30,10 +36,10 @@ namespace arcticdb {
      *
      * ProcessingSegment contains two main member attributes:
      *
-     * ExecutionContext: This contains the expression tree for the currently running clause - each clause contains an
+     * ExpressionContext: This contains the expression tree for the currently running clause - each clause contains an
      * expression tree containing nodes that map to columns, constant values or other nodes in the tree. This allows
-     * for the processing of expressions such as `(df['a'] + 4) < df['b']`. One instance of an execution context can be
-     * shared across many ProcessingSegment's. See ExecutionContext class for more documentation.
+     * for the processing of expressions such as `(df['a'] + 4) < df['b']`. One instance of an expression context can be
+     * shared across many ProcessingSegment's. See ExpressionContext class for more documentation.
      *
      * computed_data_: This contains a map from node name to previously computed result. Should an expression such as
      * df['col1'] + 1 appear multiple times in the tree this allows us to only perform the computation once and spill
@@ -46,9 +52,9 @@ namespace arcticdb {
         //    segment in order for "get" calls to make sense
         // 2: As "get" uses cached results where possible, it is not obvious which processing segment should hold the
         //    cached result for operations involving 2 columns (probably both). By moving these cached results up a level
-        //    to contain all of the information for a given row, this becomes unambiguous.
+        //    to contain all the information for a given row, this becomes unambiguous.
         std::vector<pipelines::SliceAndKey> data_;
-        std::shared_ptr<ExecutionContext> execution_context_;
+        std::shared_ptr<ExpressionContext> expression_context_;
         std::unordered_map<std::string, VariantData> computed_data_;
 
         // Set by PartitioningClause
@@ -56,22 +62,27 @@ namespace arcticdb {
 
         ProcessingSegment() = default;
 
-        ProcessingSegment(SegmentInMemory &&seg, pipelines::FrameSlice&& slice, std::optional<size_t> bucket = std::nullopt) :
+        ProcessingSegment(SegmentInMemory &&seg,
+                          pipelines::FrameSlice&& slice,
+                          std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(seg), std::move(slice)});
         }
 
-        explicit ProcessingSegment(pipelines::SliceAndKey &&sk, std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingSegment(pipelines::SliceAndKey &&sk,
+                                   std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(sk)});
         }
 
-        explicit ProcessingSegment(std::vector<pipelines::SliceAndKey> &&sk, std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingSegment(std::vector<pipelines::SliceAndKey> &&sk,
+                                   std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_ = std::move(sk);
         }
 
-        explicit ProcessingSegment(SegmentInMemory &&seg, std::optional<size_t> bucket = std::nullopt) :
+        explicit ProcessingSegment(SegmentInMemory &&seg,
+                                   std::optional<size_t> bucket = std::nullopt) :
                 bucket_(bucket) {
             data_.emplace_back(pipelines::SliceAndKey{std::move(seg)});
         }
@@ -100,10 +111,10 @@ namespace arcticdb {
             bucket_ = bucket;
         }
 
-        void apply_filter(const util::BitSet& bitset, const std::shared_ptr<Store>& store);
+        void apply_filter(const util::BitSet& bitset, const std::shared_ptr<Store>& store, PipelineOptimisation optimisation);
 
-        void set_execution_context(const std::shared_ptr<ExecutionContext>& execution_context) {
-            execution_context_ = execution_context;
+        void set_expression_context(const std::shared_ptr<ExpressionContext>& expression_context) {
+            expression_context_ = expression_context;
         }
 
         // The name argument to this function is either a column/value name, or uniquely identifies an ExpressionNode object.
@@ -125,63 +136,85 @@ namespace arcticdb {
         return output;
     }
 
-    using BucketVectorType = std::vector<size_t>;
-
     template<typename Grouper, typename Bucketizer>
-    BucketVectorType get_buckets(const ColumnWithStrings& col, std::shared_ptr<Grouper> grouper, std::shared_ptr<Bucketizer> bucketizer) {
+    std::vector<std::optional<size_t>> get_buckets(const ColumnWithStrings& col, std::shared_ptr<Grouper> grouper, std::shared_ptr<Bucketizer> bucketizer) {
         auto input_data = col.column_->data();
 
-        BucketVectorType output;
+        std::vector<std::optional<size_t>> output;
         output.reserve(col.column_->row_count());
-        col.column_->type().visit_tag([&input_data, &grouper, &bucketizer, &col, &output] (auto type_desc_tag) {
-            using TypeDescriptorTag =  decltype(type_desc_tag);
-            using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
-
-            while (auto block = input_data.next<TypeDescriptorTag>()) {
-                const auto row_count = block->row_count();
-                auto ptr = reinterpret_cast<const RawType*>(block->data());
-                for(auto i = 0u; i < row_count; ++i, ++ptr){
-                    if constexpr(std::is_same_v<typename Grouper::GrouperDescriptor, TypeDescriptorTag>) {
-                        auto group = grouper->group(*ptr, col.string_pool_);
-                        output.emplace_back(bucketizer->bucket(group));
+        using TypeDescriptorTag = typename Grouper::GrouperDescriptor;
+        using RawType = typename TypeDescriptorTag::DataTypeTag::raw_type;
+
+        while (auto block = input_data.next<TypeDescriptorTag>()) {
+            const auto row_count = block->row_count();
+            auto ptr = reinterpret_cast<const RawType*>(block->data());
+            for(auto i = 0u; i < row_count; ++i, ++ptr){
+                if constexpr(std::is_same_v<typename Grouper::GrouperDescriptor, TypeDescriptorTag>) {
+                    auto opt_group = grouper->group(*ptr, col.string_pool_);
+                    if (opt_group.has_value()) {
+                        output.emplace_back(bucketizer->bucket(*opt_group));
+                    } else {
+                        output.emplace_back(std::nullopt);
                     }
                 }
             }
-        });
+        }
         return output;
     }
 
-    template<typename Grouper, typename Bucketizer>
-    Composite<ProcessingSegment> partition_processing_segment(const ProcessingSegment& input, const ColumnWithStrings& col, const std::shared_ptr<Store>& store, std::shared_ptr<Grouper> grouper, std::shared_ptr<Bucketizer> bucketizer) {
+    template<typename GrouperType, typename BucketizerType>
+    Composite<ProcessingSegment> partition_processing_segment(
+            const std::shared_ptr<Store>& store,
+            ProcessingSegment& input,
+            const ColumnName& grouping_column_name,
+            bool dynamic_schema) {
         Composite<ProcessingSegment> output;
-        auto bucket_vec = get_buckets(col, grouper, bucketizer);
-        std::vector<util::BitSet> bitsets;
-        bitsets.resize(bucketizer->num_buckets());
-        std::vector<util::BitSet::bulk_insert_iterator> iterators;
-        for(auto& bitset : bitsets)
-            iterators.emplace_back(util::BitSet::bulk_insert_iterator(bitset));
-
-        for(auto val : folly::enumerate(bucket_vec))
-            iterators[*val] = val.index;
-
-        for(auto& iterator : iterators)
-            iterator.flush();
-
-        for(auto bitset : folly::enumerate(bitsets)) {
-            if(bitset->count() != 0) {
-                ProcessingSegment proc;
-                proc.set_bucket(bitset.index);
-                for (auto& seg_slice_and_key : input.data()) {
-                    const SegmentInMemory& seg = seg_slice_and_key.segment(store);
-                    bitset->resize(seg.row_count());
-                    auto new_seg = filter_segment(seg, *bitset);
-                    pipelines::FrameSlice new_slice{seg_slice_and_key.slice()};
-                    new_slice.adjust_rows(new_seg.row_count());
-
-                    proc.data().emplace_back(pipelines::SliceAndKey{std::move(new_seg), std::move(new_slice)});
+        auto get_result = input.get(ColumnName(grouping_column_name), store);
+        if (std::holds_alternative<ColumnWithStrings>(get_result)) {
+            auto partitioning_column = std::get<ColumnWithStrings>(get_result);
+            partitioning_column.column_->type().visit_tag([&output, &input, &partitioning_column, &store](auto type_desc_tag) {
+                using TypeDescriptorTag = decltype(type_desc_tag);
+                using ResolvedGrouperType = typename GrouperType::template Grouper<TypeDescriptorTag>;
+
+                auto grouper = std::make_shared<ResolvedGrouperType>(ResolvedGrouperType());
+                auto num_buckets = ConfigsMap::instance()->get_int("Partition.NumBuckets", async::TaskScheduler::instance()->cpu_thread_count());
+                auto bucketizer = std::make_shared<BucketizerType>(num_buckets);
+                auto bucket_vec = get_buckets(partitioning_column, grouper, bucketizer);
+                std::vector<util::BitSet> bitsets;
+                bitsets.resize(bucketizer->num_buckets());
+                std::vector<util::BitSet::bulk_insert_iterator> iterators;
+                for(auto& bitset : bitsets)
+                    iterators.emplace_back(util::BitSet::bulk_insert_iterator(bitset));
+
+                for(auto val : folly::enumerate(bucket_vec))
+                    if (val->has_value())
+                        iterators[val->value()] = val.index;
+
+                for(auto& iterator : iterators)
+                    iterator.flush();
+
+                for(auto bitset : folly::enumerate(bitsets)) {
+                    if(bitset->count() != 0) {
+                        ProcessingSegment proc;
+                        proc.set_bucket(bitset.index);
+                        for (auto& seg_slice_and_key : input.data()) {
+                            const SegmentInMemory& seg = seg_slice_and_key.segment(store);
+                            bitset->resize(seg.row_count());
+                            auto new_seg = filter_segment(seg, *bitset);
+                            pipelines::FrameSlice new_slice{seg_slice_and_key.slice()};
+                            new_slice.adjust_rows(new_seg.row_count());
+
+                            proc.data().emplace_back(pipelines::SliceAndKey{std::move(new_seg), std::move(new_slice)});
+                        }
+                        output.push_back(std::move(proc));
+                    }
                 }
-                output.push_back(std::move(proc));
-            }
+            });
+        } else {
+            internal::check<ErrorCode::E_ASSERTION_FAILURE>(
+                    dynamic_schema,
+                    "Grouping column missing from row-slice in static schema symbol"
+            );
         }
         return output;
     }
diff --git a/cpp/arcticdb/stream/stream_source.hpp b/cpp/arcticdb/stream/stream_source.hpp
index fca7a5acb3..6e33d34e31 100644
--- a/cpp/arcticdb/stream/stream_source.hpp
+++ b/cpp/arcticdb/stream/stream_source.hpp
@@ -72,8 +72,7 @@ struct StreamSource {
 
     virtual std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
         std::vector<Composite<pipelines::SliceAndKey>> &&keys,
-        const std::shared_ptr<std::vector<Clause>>& query,
-        const StreamDescriptor& desc,
+        const std::vector<std::shared_ptr<Clause>>& clauses,
         const std::shared_ptr<std::unordered_set<std::string>>& filter_columns,
         const BatchReadArgs &args) = 0;
 
diff --git a/cpp/arcticdb/util/composite.hpp b/cpp/arcticdb/util/composite.hpp
index cb2818a939..f6383f2c2c 100644
--- a/cpp/arcticdb/util/composite.hpp
+++ b/cpp/arcticdb/util/composite.hpp
@@ -134,7 +134,12 @@ struct Composite {
         }
 
         [[nodiscard]] bool empty() const {
-            return values_.empty();
+            return values_.empty() || std::all_of(values_.begin(), values_.end(), [](const ValueType& v) {
+                return util::variant_match(v,
+                                           [](const T &) { return false; },
+                                           [](const std::unique_ptr<Composite<T>> &c) { return c->empty(); }
+                );
+            });
         }
 
         auto as_range() {
diff --git a/cpp/arcticdb/version/local_versioned_engine.cpp b/cpp/arcticdb/version/local_versioned_engine.cpp
index a870783d31..3856c07e61 100644
--- a/cpp/arcticdb/version/local_versioned_engine.cpp
+++ b/cpp/arcticdb/version/local_versioned_engine.cpp
@@ -974,7 +974,7 @@ std::vector<std::pair<VersionedItem, FrameAndDescriptor>> LocalVersionedEngine::
     const ReadOptions& read_options) {
 
     if(std::none_of(std::begin(read_queries), std::end(read_queries), [] (const auto& read_query) {
-        return !read_query.query_->empty();
+        return !read_query.clauses_.empty();
     })) {
         return temp_batch_read_internal_direct(stream_ids, version_queries, read_queries, read_options);
     }
diff --git a/cpp/arcticdb/version/python_bindings.cpp b/cpp/arcticdb/version/python_bindings.cpp
index 61e42cd330..b8d865c39e 100644
--- a/cpp/arcticdb/version/python_bindings.cpp
+++ b/cpp/arcticdb/version/python_bindings.cpp
@@ -18,7 +18,7 @@
 #include <arcticdb/storage/mongo/mongo_instance.hpp>
 #include <arcticdb/processing/operation_types.hpp>
 #include <arcticdb/processing/expression_node.hpp>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/pipeline/value_set.hpp>
 #include <arcticdb/python/adapt_read_dataframe.hpp>
 #include <arcticdb/version/schema_checks.hpp>
@@ -92,25 +92,6 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
     }))
     ;
 
-    py::class_<ClauseBuilder>(version, "ClauseBuilder")
-            .def(py::init())
-            .def("add_ProjectClause", &ClauseBuilder::add_ProjectClause)
-            .def("add_FilterClause", &ClauseBuilder::add_FilterClause)
-            .def("prepare_AggregationClause", &ClauseBuilder::prepare_AggregationClause)
-            .def("add_MeanAggregationOperator", [&](ClauseBuilder& v,  std::string input_column, std::string output_column) {
-                return v.add_MeanAggregationOperator(ColumnName(input_column), ColumnName(output_column));
-            })
-            .def("add_SumAggregationOperator", [&](ClauseBuilder& v,  std::string input_column, std::string output_column) {
-                return v.add_SumAggregationOperator(ColumnName(input_column), ColumnName(output_column));
-            })
-            .def("add_MaxAggregationOperator", [&](ClauseBuilder& v,  std::string input_column, std::string output_column) {
-              return v.add_MaxAggregationOperator(ColumnName(input_column), ColumnName(output_column));
-            })
-            .def("add_MinAggregationOperator", [&](ClauseBuilder& v,  std::string input_column, std::string output_column) {
-                return v.add_MinAggregationOperator(ColumnName(input_column), ColumnName(output_column));
-            })
-        .def("finalize_AggregationClause", &ClauseBuilder::finalize_AggregationClause);
-
     py::class_<VersionQuery>(version, "PythonVersionStoreVersionQuery")
         .def(py::init())
         .def("set_snap_name", &VersionQuery::set_snap_name)
@@ -211,12 +192,55 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
                 return std::get<timestamp>(self.end_);
             });
 
+    py::class_<FilterClause, std::shared_ptr<FilterClause>>(version, "FilterClause")
+            .def(py::init<
+                    std::unordered_set<std::string>,
+                    ExpressionContext,
+                    std::optional<PipelineOptimisation>>())
+            .def("__str__", &FilterClause::to_string)
+            .def("set_pipeline_optimisation", &FilterClause::set_pipeline_optimisation);
+
+    py::class_<ProjectClause, std::shared_ptr<ProjectClause>>(version, "ProjectClause")
+            .def(py::init<
+                    std::unordered_set<std::string>,
+                    std::string,
+                    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) {
+                return self.grouping_column_;
+            })
+            .def("__str__", &GroupByClause::to_string);
+
+    py::class_<AggregationClause, std::shared_ptr<AggregationClause>>(version, "AggregationClause")
+            .def(py::init<std::string, std::unordered_map<std::string, std::string>>())
+            .def("__str__", &AggregationClause::to_string);
+
     py::class_<ReadQuery>(version, "PythonVersionStoreReadQuery")
             .def(py::init())
             .def_readwrite("columns",&ReadQuery::columns)
             .def_readwrite("row_range",&ReadQuery::row_range)
             .def_readwrite("row_filter",&ReadQuery::row_filter)
-            .def("set_clause_builder", &ReadQuery::set_clause_builder);
+            // Unsurprisingly, pybind11 doesn't understand folly::poly, so use vector of variants here
+            .def("add_clauses",
+                 [](ReadQuery& self,
+                    std::vector<std::variant<std::shared_ptr<FilterClause>,
+                                std::shared_ptr<ProjectClause>,
+                                std::shared_ptr<GroupByClause>,
+                                std::shared_ptr<AggregationClause>>> clauses) {
+                std::vector<std::shared_ptr<Clause>> _clauses;
+                for (auto&& clause: clauses) {
+                    util::variant_match(
+                        clause,
+                        [&](auto&& clause) {_clauses.emplace_back(std::make_shared<Clause>(*clause));}
+                    );
+                }
+                self.add_clauses(_clauses);
+            });
 
     py::enum_<OperationType>(version, "OperationType")
             .value("ABS", OperationType::ABS)
@@ -277,18 +301,16 @@ void register_bindings(py::module &version, py::exception<arcticdb::ArcticExcept
                 return ExpressionNode(left, operation_type);
             }));
 
-    py::enum_<ExecutionContext::Optimisation>(version, "ExecutionContextOptimisation")
-            .value("SPEED", ExecutionContext::Optimisation::SPEED)
-            .value("MEMORY", ExecutionContext::Optimisation::MEMORY);
+    py::enum_<PipelineOptimisation>(version, "PipelineOptimisation")
+            .value("SPEED", PipelineOptimisation::SPEED)
+            .value("MEMORY", PipelineOptimisation::MEMORY);
 
-    py::class_<ExecutionContext, std::shared_ptr<ExecutionContext>>(version, "ExecutionContext")
+    py::class_<ExpressionContext, std::shared_ptr<ExpressionContext>>(version, "ExpressionContext")
             .def(py::init())
-            .def("add_column", &ExecutionContext::add_column)
-            .def("add_expression_node", &ExecutionContext::add_expression_node)
-            .def("add_value", &ExecutionContext::add_value)
-            .def("add_value_set", &ExecutionContext::add_value_set)
-            .def_readwrite("root_node_name", &ExecutionContext::root_node_name_)
-            .def_readwrite("optimisation", &ExecutionContext::optimisation_);
+            .def("add_expression_node", &ExpressionContext::add_expression_node)
+            .def("add_value", &ExpressionContext::add_value)
+            .def("add_value_set", &ExpressionContext::add_value_set)
+            .def_readwrite("root_node_name", &ExpressionContext::root_node_name_);
 
     py::class_<UpdateQuery>(version, "PythonVersionStoreUpdateQuery")
             .def(py::init())
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index 83d798acc3..0e396cb766 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -410,6 +410,98 @@ FrameAndDescriptor read_multi_key(
     return {res.frame_, multi_key_desc, keys, std::shared_ptr<BufferHolder>{}};
 }
 
+Composite<ProcessingSegment> process_remaining_clauses(
+        const std::shared_ptr<Store>& store,
+        std::vector<Composite<ProcessingSegment>>&& procs,
+        std::vector<std::shared_ptr<Clause>> clauses ) { // pass by copy deliberately as we don't want to modify read_query
+    while (!clauses.empty()) {
+        if (clauses[0]->clause_info().requires_repartition_) {
+            std::vector<Composite<ProcessingSegment>> repartitioned_procs = clauses[0]->repartition(std::move(procs)).value();
+            // Erasing from front of vector not ideal, but they're just shared_ptr and there shouldn't be loads of clauses
+            clauses.erase(clauses.begin());
+            std::vector<folly::Future<Composite<ProcessingSegment>>> fut_procs;
+            for (auto&& proc : repartitioned_procs) {
+                fut_procs.emplace_back(
+                        async::submit_cpu_task(
+                                async::MemSegmentProcessingTask(store,
+                                                                clauses,
+                                                                std::move(proc))
+                        )
+                );
+            }
+            procs = folly::collect(fut_procs).get();
+        } else {
+            // Erasing from front of vector not ideal, but they're just shared_ptr and there shouldn't be loads of clauses
+            clauses.erase(clauses.begin());
+        }
+    }
+    return merge_composites(std::move(procs));
+}
+
+void set_output_descriptors(
+        const Composite<ProcessingSegment>& merged_procs,
+        const std::vector<std::shared_ptr<Clause>>& clauses,
+        const std::shared_ptr<PipelineContext>& pipeline_context) {
+    std::optional<std::string> index_column;
+    for (auto clause = clauses.rbegin(); clause != clauses.rend(); ++clause) {
+        if (auto new_index = (*clause)->clause_info().new_index_; new_index.has_value()) {
+            index_column = new_index;
+            pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index()->set_name(*new_index);
+            pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index()->clear_fake_name();
+            pipeline_context->norm_meta_->mutable_df()->mutable_common()->mutable_index()->set_is_not_range_index(
+                    true);
+            break;
+        }
+    }
+    std::optional<StreamDescriptor> new_stream_descriptor;
+    merged_procs.broadcast([&new_stream_descriptor](const auto& proc) {
+        if (!new_stream_descriptor.has_value()) {
+            if (proc.data_.size() > 0) {
+                new_stream_descriptor = std::make_optional<StreamDescriptor>();
+                new_stream_descriptor->set_index(proc.data_[0].segment_->descriptor().index());
+                for (size_t idx = 0; idx < new_stream_descriptor->index().field_count(); idx++) {
+                    new_stream_descriptor->add_field(proc.data_[0].segment_->descriptor().field(idx));
+                }
+            }
+        }
+        if (new_stream_descriptor.has_value()) {
+            std::vector<StreamDescriptor::FieldsCollection> fields;
+            for (const auto &slice_and_key: proc.data_) {
+                fields.push_back(slice_and_key.segment_->descriptor().fields());
+            }
+            new_stream_descriptor = merge_descriptors(*new_stream_descriptor,
+                                                      fields,
+                                                      std::vector<std::string>{});
+        }
+    });
+    if (new_stream_descriptor.has_value()) {
+        // Columns might be in a different order to the original dataframe, so reorder here
+        auto original_stream_descriptor = pipeline_context->descriptor();
+        StreamDescriptor final_stream_descriptor{original_stream_descriptor.id()};
+        final_stream_descriptor.set_index(new_stream_descriptor->index());
+        // Erase field from new_stream_descriptor as we add them to final_stream_descriptor, as all fields left in new_stream_descriptor
+        // after these operations were created by the processing pipeline, and so should be appended
+        // Index columns should always appear first
+        if (index_column.has_value()) {
+            auto opt_idx = new_stream_descriptor->find_field(*index_column);
+            internal::check<ErrorCode::E_ASSERTION_FAILURE>(opt_idx.has_value(), "New index column not found in processing pipeline");
+            final_stream_descriptor.add_field(new_stream_descriptor->field(*opt_idx));
+            new_stream_descriptor->erase_field(*opt_idx);
+        }
+        for (const auto& field: original_stream_descriptor.fields()) {
+            if (auto position = new_stream_descriptor->find_field(field.name()); position.has_value()) {
+                final_stream_descriptor.add_field(new_stream_descriptor->field(*position));
+
+                new_stream_descriptor->erase_field(*position);
+            }
+        }
+        for (const auto& field: new_stream_descriptor->fields()) {
+            final_stream_descriptor.add_field(field);
+        }
+        pipeline_context->set_descriptor(final_stream_descriptor);
+    }
+}
+
 /*
  * Processes the slices in the given pipeline_context.
  *
@@ -427,23 +519,11 @@ std::vector<SliceAndKey> read_and_process(
     const ReadOptions& read_options,
     size_t start_from
     ) {
-    std::vector<Composite<SliceAndKey>> rows;
-    if(!read_query.query_->empty()) {
-        if(auto execution_context = read_query.query_->begin()->execution_context(); execution_context) {
-            execution_context->set_descriptor(pipeline_context->descriptor());
-            execution_context->set_norm_meta_descriptor(pipeline_context->norm_meta_);
-        }
-
-        for(auto& clause : *read_query.query_) {
-            if(auto execution_context = clause.execution_context(); execution_context)
-                execution_context->set_dynamic_schema(opt_false(read_options.dynamic_schema_));
-        }
-    }
-
     std::sort(std::begin(pipeline_context->slice_and_keys_), std::end(pipeline_context->slice_and_keys_), [] (const SliceAndKey& left, const SliceAndKey& right) {
         return std::tie(left.slice().row_range.first, left.slice().col_range.first) < std::tie(right.slice().row_range.first, right.slice().col_range.first);
     });
 
+    std::vector<Composite<SliceAndKey>> rows;
     auto sk_it = std::begin(pipeline_context->slice_and_keys_);
     std::advance(sk_it, start_from);
     while(sk_it != std::end(pipeline_context->slice_and_keys_)) {
@@ -468,44 +548,30 @@ std::vector<SliceAndKey> read_and_process(
         }
     }
 
-    auto parallel_output = store->batch_read_uncompressed(std::move(rows), read_query.query_, pipeline_context->descriptor(), filter_columns, BatchReadArgs{});
-
-    size_t clause_index = 0;
-    for (const auto& clause : *read_query.query_) {
-        if (clause.requires_repartition()) {
-            std::vector<Composite<ProcessingSegment>> composites = clause.repartition(std::move(parallel_output)).value();
-
-            std::vector<folly::Future<Composite<ProcessingSegment>>> batch;
-            std::vector<Composite<ProcessingSegment>> res;
-            res.reserve(composites.size());
-            for (auto&& val : composites){
-                auto clause_copy = std::make_shared<std::vector<Clause>>(read_query.query_->begin() + clause_index + 1, read_query.query_->end());
-                batch.emplace_back(
-                        async::submit_cpu_task(
-                                async::MemSegmentPassthroughProcessingTask(store, clause_copy, std::move(val))
-                        )
-                );
-            }
-
-            if(!batch.empty()) {
-                auto segments = folly::collect(batch).get();
-                res.insert(std::end(res), std::make_move_iterator(std::begin(segments)), std::make_move_iterator(std::end(segments)));
-            }
-
-            parallel_output = std::move(res);
-        }
-
-        clause_index++;
+    ProcessingConfig processing_config{opt_false(read_options.dynamic_schema_)};
+    for (auto& clause: read_query.clauses_) {
+        clause->set_processing_config(processing_config);
     }
 
+    // At this stage, each Composite contains a single ProcessingSegment, which holds a row-slice
+    // Different row-slices are held in different elements of the vector
+    // All clauses that do not require repartitioning (e.g. filters and projections) will have already been applied
+    // to these processing segments
+    std::vector<Composite<ProcessingSegment>> procs = store->batch_read_uncompressed(
+            std::move(rows),
+            read_query.clauses_,
+            filter_columns,
+            BatchReadArgs{}
+            );
+
+    auto merged_procs = process_remaining_clauses(store, std::move(procs), read_query.clauses_);
     //TODO split pipeline context into load_context and output_context
-        for(auto clause  = read_query.query_->rbegin(); clause != read_query.query_->rend(); ++clause ) {
-        if(clause->execution_context() && clause->execution_context()->output_descriptor_)
-            pipeline_context->set_descriptor(clause->execution_context()->output_descriptor_.value());
+    if (std::any_of(read_query.clauses_.begin(), read_query.clauses_.end(), [](const std::shared_ptr<Clause>& clause) {
+        return clause->clause_info().modifies_output_descriptor_;
+    })) {
+        set_output_descriptors(merged_procs, read_query.clauses_, pipeline_context);
     }
-
-    auto merged = merge_composites(std::move(parallel_output));
-    return collect_segments(std::move(merged));
+    return collect_segments(std::move(merged_procs));
 }
 
 SegmentInMemory read_direct(const std::shared_ptr<Store>& store,
@@ -773,7 +839,7 @@ void create_column_stats_impl(
         log::version().warn("Cannot create empty column stats");
         return;
     }
-    ReadQuery read_query({*clause});
+    ReadQuery read_query({std::make_shared<Clause>(std::move(*clause))});
 
     auto column_stats_key = index_key_to_column_stats_key(versioned_item.key_);
     std::optional<SegmentInMemory> old_segment;
@@ -930,7 +996,7 @@ FrameAndDescriptor read_dataframe_impl(
     ARCTICDB_DEBUG(log::version(), "Fetching data to frame");
     SegmentInMemory frame;
     auto buffers = std::make_shared<BufferHolder>();
-    if(!read_query.query_->empty()) {
+    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);
@@ -1017,15 +1083,11 @@ 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.query_->emplace_back(SortClause{timeseries_index.name()});
-            ExecutionContext remove_column_partition_context{};
-            remove_column_partition_context.set_descriptor(pipeline_context->descriptor());
-            read_query.query_->emplace_back(RemoveColumnPartitioningClause{std::make_shared<ExecutionContext>(std::move(remove_column_partition_context))});
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(SortClause{timeseries_index.name()}));
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(RemoveColumnPartitioningClause{}));
             const auto split_size = ConfigsMap::instance()->get_int("Split.RowCount", 10000);
-            read_query.query_->emplace_back(SplitClause{static_cast<size_t>(split_size)});
-            ExecutionContext merge_clause_context{};
-            merge_clause_context.set_descriptor(pipeline_context->descriptor());
-            read_query.query_->emplace_back(MergeClause{timeseries_index, DenseColumnPolicy{}, stream_id, std::make_shared<ExecutionContext>(std::move(merge_clause_context))});
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(SplitClause{static_cast<size_t>(split_size)}));
+            read_query.clauses_.emplace_back(std::make_shared<Clause>(MergeClause{timeseries_index, DenseColumnPolicy{}, stream_id, pipeline_context->descriptor()}));
             auto segments = read_and_process(store, pipeline_context, read_query, ReadOptions{}, pipeline_context->incompletes_after());
             pipeline_context->total_rows_ = num_versioned_rows + get_slice_rowcounts(segments);
 
@@ -1220,9 +1282,7 @@ 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) {
-        ExecutionContext remove_column_partition_context{};
-        remove_column_partition_context.set_descriptor(pre_defragmentation_info.pipeline_context->descriptor());
-        pre_defragmentation_info.read_query.query_->emplace_back(RemoveColumnPartitioningClause{std::make_shared<ExecutionContext>(std::move(remove_column_partition_context))});
+        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());
         using IndexType = std::remove_reference_t<decltype(idx)>;
         using SchemaType = std::remove_reference_t<decltype(schema)>;
diff --git a/python/arcticdb/version_store/_store.py b/python/arcticdb/version_store/_store.py
index be338b665c..2ca765c36e 100644
--- a/python/arcticdb/version_store/_store.py
+++ b/python/arcticdb/version_store/_store.py
@@ -1391,7 +1391,7 @@ def _get_read_query(self, date_range: Optional[DateRangeInput], row_range, colum
         read_query = _PythonVersionStoreReadQuery()
 
         if query_builder:
-            read_query.set_clause_builder(query_builder.finalize_clause_builder())
+            read_query.add_clauses(query_builder.clauses)
 
         if date_range is not None:
             read_query.row_filter = _normalize_dt_range(date_range)
diff --git a/python/arcticdb/version_store/processing.py b/python/arcticdb/version_store/processing.py
index 22b998cc63..6a94f058a5 100644
--- a/python/arcticdb/version_store/processing.py
+++ b/python/arcticdb/version_store/processing.py
@@ -5,20 +5,25 @@
 
 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 collections import namedtuple
 import datetime
 from math import inf
 
 import numpy as np
 import pandas as pd
 
-from abc import ABC, abstractmethod
+from typing import Dict
 
 from arcticdb.exceptions import ArcticNativeException, UserInputException
+from arcticdb.preconditions import check
 from arcticdb.supported_types import time_types as supported_time_types
 
-from arcticdb_ext.version_store import ExecutionContextOptimisation as _Optimisation
-from arcticdb_ext.version_store import ExecutionContext as _ExecutionContext
+from arcticdb_ext.version_store import PipelineOptimisation as _Optimisation
+from arcticdb_ext.version_store import ExpressionContext as _ExpressionContext
+from arcticdb_ext.version_store import FilterClause as _FilterClause
+from arcticdb_ext.version_store import ProjectClause as _ProjectClause
+from arcticdb_ext.version_store import GroupByClause as _GroupByClause
+from arcticdb_ext.version_store import AggregationClause as _AggregationClause
 from arcticdb_ext.version_store import ExpressionName as _ExpressionName
 from arcticdb_ext.version_store import ColumnName as _ColumnName
 from arcticdb_ext.version_store import ValueName as _ValueName
@@ -41,8 +46,6 @@
 from arcticdb_ext.version_store import ExpressionNode as _ExpressionNode
 from arcticdb_ext.version_store import OperationType as _OperationType
 
-from arcticdb_ext.version_store import ClauseBuilder as _ClauseBuilder
-
 COLUMN = "COLUMN"
 
 
@@ -254,85 +257,11 @@ def value_list_from_args(*args):
     return value_list
 
 
-class PyClauseBase(ABC):
-    @abstractmethod
-    def to_cpp(self, clause_builder) -> None:
-        pass
-
-
-class WhereClause(PyClauseBase):
-    def __init__(self, expr):
-        self.expr = expr
-
-    def __str__(self):
-        return "WhereClause: {}".format(str(self.expr))
-
-    def to_cpp(self, clause_builder):
-        clause_builder.add_FilterClause(visit_expression(self.expr))
-
-
-class ProjectClause(PyClauseBase):
-    def __init__(self, name, expr):
-        self.name = name
-        self.expr = expr
-
-    def __str__(self):
-        return "ProjectClause:{} -> {}".format(str(self.expr), self.name)
-
-    def to_cpp(self, clause_builder):
-        clause_builder.add_ProjectClause(self.name, visit_expression(self.expr))
-
-
-class Aggregation:
-    def __init__(self, source, operator):
-        self.source = source
-        self.operator = operator
-
-    def __str__(self):
-        return "{}({})".format(self.operator, self.source)
-
-    def to_cpp(self, clause_builder):
-        # TODO: Move to dictionary
-        if self.operator.lower() == "sum":
-            clause_builder.add_SumAggregationOperator(self.source, self.source)
-        elif self.operator.lower() == "mean":
-            clause_builder.add_MeanAggregationOperator(self.source, self.source)
-        elif self.operator.lower() == "max":
-            clause_builder.add_MaxAggregationOperator(self.source, self.source)
-        elif self.operator.lower() == "min":
-            clause_builder.add_MinAggregationOperator(self.source, self.source)
-        else:
-            raise ValueError("Aggregation operators are limited to 'sum', 'mean', 'max' and 'min'.")
-
-
-class GroupByClause(PyClauseBase):
-    def __init__(self, key, query_builder):
-        self.key = key
-        self.query_builder = query_builder
-        self.aggregations = {}
-
-    def __str__(self):
-        return "GroupByClause: key={}, [{}]".format(
-            str(self.key), ", ".join(["{} <- {}".format(k, v) for k, v in self.aggregations.items()])
-        )
-
-    def agg(self, aggregations):
-        for key, value in aggregations.items():
-            self.aggregations[key] = Aggregation(key, value)
-
-        return self.query_builder
-
-    def to_cpp(self, clause_builder):
-        def _expression_root_only(col_name: str):
-            _ec = _ExecutionContext()
-            _ec.root_node_name = _ExpressionName(col_name)
-
-            return _ec
-
-        clause_builder.prepare_AggregationClause(_expression_root_only(self.key))
-        for agg in self.aggregations.values():
-            agg.to_cpp(clause_builder)
-        clause_builder.finalize_AggregationClause()
+# These are just used for shallow/deep copying, pickling, and equality checks
+PythonFilterClause = namedtuple("PythonFilterClause", ["expr"])
+PythonProjectionClause = namedtuple("PythonProjectionClause", ["name", "expr"])
+PythonGroupByClause = namedtuple("PythonGroupByClause", ["name"])
+PythonAggregationClause = namedtuple("PythonAggregationClause", ["aggregations"])
 
 
 class QueryBuilder:
@@ -343,23 +272,21 @@ class QueryBuilder:
         >>> q = q[q["a"] < 5] (equivalent to q = q[q.a < 5] provided the column name is also a valid Python variable name)
         >>> dataframe = lib.read(symbol, query_builder=q).data
 
-    QueryBuilder objects are stateful, and so should not be reused without reinitialising:
-
-    >>> q = QueryBuilder()
-
     For Group By and Aggregation functionality please see the documentation for the `groupby`. For projection
     functionality, see the documentation for the `apply` method.
 
-    Supported numeric operations when filtering:
-
-    * Binary comparisons: <, <=, >, >=, ==, !=
-
-    * Unary NOT: ~
+    Supported arithmetic operations when projection or filtering:
 
     * Binary arithmetic: +, -, *, /
 
     * Unary arithmetic: -, abs
 
+    Supported filtering operations:
+
+    * Binary comparisons: <, <=, >, >=, ==, !=
+
+    * Unary NOT: ~
+
     * Binary combinators: &, |, ^
 
     * List membership: isin, isnotin (also accessible with == and !=)
@@ -408,11 +335,12 @@ class QueryBuilder:
     """
 
     def __init__(self):
-        self.stages = []
+        self.clauses = []
+        # This is hacky, but the alternative is implementing pickle for the C++ classes of all the clauses, and the tree
+        # of classes these depend on, which is A LOT
+        self._python_clauses = []
         self._optimisation = _Optimisation.SPEED
 
-        self._clause_builder = _ClauseBuilder()
-
     def apply(self, name, expr):
         """
         Apply enables new columns to be created using supported QueryBuilder numeric operations. See the documentation for the
@@ -458,10 +386,12 @@ def apply(self, name, expr):
         QueryBuilder
             Modified QueryBuilder object.
         """
-        self.stages.append(ProjectClause(name, expr))
+        input_columns, expression_context = visit_expression(expr)
+        self.clauses.append(_ProjectClause(input_columns, name, expression_context))
+        self._python_clauses.append(PythonProjectionClause(name, expr))
         return self
 
-    def groupby(self, expr: str):
+    def groupby(self, name: str):
         """
         Group symbol by column name. GroupBy operations must be followed by an aggregation operator. Currently the following four aggregation
         operators are supported:
@@ -474,8 +404,8 @@ def groupby(self, expr: str):
 
         Parameters
         ----------
-        expr: `str`
-            Name of the symbol to group on. Note that currently GroupBy only supports single-column groupings.
+        name: `str`
+            Name of the column to group on. Note that currently GroupBy only supports single-column groupings.
 
         Examples
         --------
@@ -534,14 +464,27 @@ def groupby(self, expr: str):
         QueryBuilder
             Modified QueryBuilder object.
         """
-        self.stages.append(GroupByClause(expr, self))
-        return self.stages[-1]
+        self.clauses.append(_GroupByClause(name))
+        self._python_clauses.append(PythonGroupByClause(name))
+        return self
+
+    def agg(self, aggregations: Dict[str, str]):
+        # Only makes sense if previous stage is a group-by
+        check(
+            len(self.clauses) and isinstance(self.clauses[-1], _GroupByClause),
+            f"Aggregation only makes sense after groupby",
+        )
+        for v in aggregations.values():
+            v = v.lower()
+        self.clauses.append(_AggregationClause(self.clauses[-1].grouping_column, aggregations))
+        self._python_clauses.append(PythonAggregationClause(aggregations))
+        return self
 
     def __eq__(self, right):
-        return str(self) == str(right)
+        return self._optimisation == right._optimisation and self._python_clauses == right._python_clauses
 
     def __str__(self):
-        return " | ".join(str(e) for e in self.stages)
+        return " | ".join(str(clause) for clause in self.clauses)
 
     def __getitem__(self, item):
         if isinstance(item, str):
@@ -551,7 +494,9 @@ def __getitem__(self, item):
             # e.g. q = q[q["col"]]
             if isinstance(item, ExpressionNode) and item.operator == COLUMN:
                 item = ExpressionNode.compose(item, _OperationType.IDENTITY, None)
-            self.stages.append(WhereClause(item))
+            input_columns, expression_context = visit_expression(item)
+            self.clauses.append(_FilterClause(input_columns, expression_context, self._optimisation))
+            self._python_clauses.append(PythonFilterClause(item))
             return self
 
     def __getattr__(self, key):
@@ -559,33 +504,41 @@ def __getattr__(self, key):
 
     def __getstate__(self):
         rv = vars(self).copy()
-        del rv["_clause_builder"]
+        del rv["clauses"]
         return rv
 
     def __setstate__(self, state):
         vars(self).update(state)
-        self._clause_builder = _ClauseBuilder()
-
-    def __copy__(self):
-        cls = self.__class__
-        result = cls.__new__(cls)
-        result.__dict__.update(self.__dict__)
-        return result
+        self.clauses = []
+        for python_clause in self._python_clauses:
+            if isinstance(python_clause, PythonFilterClause):
+                input_columns, expression_context = visit_expression(python_clause.expr)
+                self.clauses.append(_FilterClause(input_columns, expression_context, self._optimisation))
+            elif isinstance(python_clause, PythonProjectionClause):
+                input_columns, expression_context = visit_expression(python_clause.expr)
+                self.clauses.append(_ProjectClause(input_columns, python_clause.name, expression_context))
+            elif isinstance(python_clause, PythonGroupByClause):
+                self.clauses.append(_GroupByClause(python_clause.name))
+            elif isinstance(python_clause, PythonAggregationClause):
+                self.clauses.append(_AggregationClause(self.clauses[-1].grouping_column, python_clause.aggregations))
+            else:
+                raise ArcticNativeException(
+                    f"Unrecognised clause type {type(python_clause)} when unpickling QueryBuilder"
+                )
 
     def __deepcopy__(self, memo):
         cls = self.__class__
         result = cls.__new__(cls)
-        memo[id(self)] = result
-        for k, v in self.__dict__.items():
-            if k != "_clause_builder":
-                setattr(result, k, copy.deepcopy(v, memo))
-        result._clause_builder = _ClauseBuilder()
+        result.__setstate__(self.__getstate__())
         return result
 
     # Might want to apply different optimisations to different clauses once projections/group-bys are implemented
     def optimise_for_speed(self):
         """Process query as fast as possible (the default behaviour)"""
         self._optimisation = _Optimisation.SPEED
+        for clause in self.clauses:
+            if hasattr(clause, "set_pipeline_optimisation"):
+                clause.set_pipeline_optimisation(_Optimisation.SPEED)
 
     def optimise_for_memory(self):
         """Reduce peak memory usage during the query, at the expense of some performance.
@@ -595,18 +548,9 @@ def optimise_for_memory(self):
         * Memory used by strings that are present in segments read from storage, but are not required in the final dataframe that will be presented back to the user, is reclaimed earlier in the processing pipeline.
         """
         self._optimisation = _Optimisation.MEMORY
-
-    def execution_contexts(self):
-        res = [visit_expression(stage.expr) for stage in self.stages]
-        for execution_context in res:
-            execution_context.optimisation = self._optimisation
-        return res
-
-    def finalize_clause_builder(self):
-        for py_clause in self.stages:
-            py_clause.to_cpp(self._clause_builder)
-
-        return self._clause_builder
+        for clause in self.clauses:
+            if hasattr(clause, "set_pipeline_optimisation"):
+                clause.set_pipeline_optimisation(_Optimisation.MEMORY)
 
 
 CONSTRUCTOR_MAP = {
@@ -670,15 +614,15 @@ def _handle_leaf(node):
                     else:
                         valueset_keys[key] += 1
                     key = key + "-v" + str(valueset_keys[key])
-                    execution_context.add_value_set(key, _ValueSet(node))
+                    expression_context.add_value_set(key, _ValueSet(node))
                     return _ValueSetName(key)
                 else:
-                    execution_context.add_value(key, create_value(node))
+                    expression_context.add_value(key, create_value(node))
                     return _ValueName(key)
 
             if isinstance(node, ExpressionNode):
                 if node.operator == COLUMN:
-                    execution_context.add_column(node.left)
+                    input_columns.add(node.left)
                     return _ColumnName(node.left)
                 else:
                     _visit(node)
@@ -695,10 +639,11 @@ def _handle_leaf(node):
             expression_node = _ExpressionNode(left, right, node.operator)
         else:
             expression_node = _ExpressionNode(left, node.operator)
-        execution_context.add_expression_node(node.get_name(), expression_node)
+        expression_context.add_expression_node(node.get_name(), expression_node)
 
-    execution_context = _ExecutionContext()
+    expression_context = _ExpressionContext()
+    input_columns = set()
     valueset_keys = dict()
     _visit(expr)
-    execution_context.root_node_name = _ExpressionName(expr.get_name())
-    return execution_context
+    expression_context.root_node_name = _ExpressionName(expr.get_name())
+    return input_columns, expression_context


Test patch:
diff --git a/cpp/arcticdb/processing/test/test_clause.cpp b/cpp/arcticdb/processing/test/test_clause.cpp
index 14bbd2c578..6b814a9122 100644
--- a/cpp/arcticdb/processing/test/test_clause.cpp
+++ b/cpp/arcticdb/processing/test/test_clause.cpp
@@ -62,9 +62,7 @@ TEST(Clause, Partition) {
     Composite<ProcessingSegment> comp;
     comp.push_back(std::move(proc_seg));
 
-    ExecutionContext context{};
-    context.root_node_name_ = ExpressionName("int8");
-    PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{std::make_shared<ExecutionContext>(std::move(context))};
+    PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{"int8"};
 
     auto partitioned = partition.process(empty, std::move(comp));
 
@@ -88,9 +86,7 @@ TEST(Clause, PartitionString) {
     Composite<ProcessingSegment> comp;
     comp.push_back(std::move(proc_seg));
 
-    ExecutionContext context{};
-    context.root_node_name_ = ExpressionName("strings");
-    PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{std::make_shared<ExecutionContext>(std::move(context))};
+    PartitionClause<arcticdb::grouping::HashingGroupers, arcticdb::grouping::ModuloBucketizer> partition{"strings"};
 
     auto partitioned = partition.process(empty, std::move(comp));
 
@@ -189,10 +185,8 @@ TEST(Clause, Merge) {
 
     StreamDescriptor descriptor{};
     descriptor.add_field(scalar_field_proto(DataType::MICROS_UTC64, "time"));
-    ExecutionContext merge_clause_context{};
-    merge_clause_context.set_descriptor(descriptor);
     auto stream_id = StreamId("Merge");
-    MergeClause merge_clause{TimeseriesIndex{"time"}, DenseColumnPolicy{}, stream_id, std::make_shared<ExecutionContext>(std::move(merge_clause_context))};
+    MergeClause merge_clause{TimeseriesIndex{"time"}, DenseColumnPolicy{}, stream_id, descriptor};
     std::shared_ptr<Store> empty;
     auto res = merge_clause.process(empty, std::move(comp));
     ASSERT_EQ(res.size(), 4u);
diff --git a/cpp/arcticdb/processing/test/test_expression.cpp b/cpp/arcticdb/processing/test/test_expression.cpp
index 06e981be9e..c58c251d93 100644
--- a/cpp/arcticdb/processing/test/test_expression.cpp
+++ b/cpp/arcticdb/processing/test/test_expression.cpp
@@ -6,7 +6,7 @@
  */
 
 #include <gtest/gtest.h>
-#include <arcticdb/processing/execution_context.hpp>
+#include <arcticdb/processing/expression_context.hpp>
 #include <arcticdb/processing/expression_node.hpp>
 #include <arcticdb/processing/processing_segment.hpp>
 #include <arcticdb/util/test/generators.hpp>
@@ -30,10 +30,10 @@ TEST(ExpressionNode, AddBasic) {
     auto& seg = wrapper.segment();
     ProcessingSegment proc(std::move(seg));
     auto node = std::make_shared<ExpressionNode>(ColumnName("thing1"), ColumnName("thing2"), OperationType::ADD);
-    auto execution_context = std::make_shared<ExecutionContext>();
-    execution_context->root_node_name_ = ExpressionName("new_thing");
-    execution_context->add_expression_node("new_thing", node);
-    proc.set_execution_context(execution_context);
+    auto expression_context = std::make_shared<ExpressionContext>();
+    expression_context->root_node_name_ = ExpressionName("new_thing");
+    expression_context->add_expression_node("new_thing", node);
+    proc.set_expression_context(expression_context);
     std::shared_ptr<Store> empty;
     auto ret = proc.get(ExpressionName("new_thing"), empty);
     const auto& col = std::get<ColumnWithStrings>(ret).column_;
diff --git a/cpp/arcticdb/storage/test/in_memory_store.hpp b/cpp/arcticdb/storage/test/in_memory_store.hpp
index e2bd6885e1..fe8f2e1840 100644
--- a/cpp/arcticdb/storage/test/in_memory_store.hpp
+++ b/cpp/arcticdb/storage/test/in_memory_store.hpp
@@ -41,8 +41,7 @@ namespace arcticdb {
 
         std::vector<Composite<ProcessingSegment>> batch_read_uncompressed(
                 std::vector<Composite<pipelines::SliceAndKey>> &&,
-                const std::shared_ptr<std::vector<Clause>>&,
-                const StreamDescriptor&,
+                const std::vector<std::shared_ptr<Clause>>&,
                 const std::shared_ptr<std::unordered_set<std::string>>&,
                 const BatchReadArgs &) override {
             throw std::runtime_error("Not implemented for tests");
diff --git a/cpp/arcticdb/util/test/test_composite.cpp b/cpp/arcticdb/util/test/test_composite.cpp
index 0c050954c6..020944e8b5 100644
--- a/cpp/arcticdb/util/test/test_composite.cpp
+++ b/cpp/arcticdb/util/test/test_composite.cpp
@@ -43,3 +43,18 @@ TEST(Composite, IterationSimple) {
     for(auto x : comp.as_range())
         ASSERT_EQ(x, expected++);
 }
+
+TEST(Composite, Empty) {
+    using namespace arcticdb;
+    Composite<int> comp1;
+    Composite<int> comp2;
+    comp1.push_back(std::move(comp2));
+    ASSERT_EQ(comp1.size(), 0);
+    ASSERT_TRUE(comp1.empty());
+
+    Composite<int> comp3;
+    Composite<int> comp4(0);
+    comp3.push_back(std::move(comp4));
+    ASSERT_EQ(comp3.size(), 1);
+    ASSERT_FALSE(comp3.empty());
+}
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 839717f635..bffa3447bf 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
@@ -1725,7 +1725,7 @@ def do(op, start, n):
 def test_diff_long_stream_descriptor_mismatch(lmdb_version_store, method, num):
     lib: NativeVersionStore = lmdb_version_store
     lib.write("x", pd.DataFrame({f"col{i}": [i, i + 1, i + 2] for i in range(num)}, index=pd.date_range(0, periods=3)))
-    bad_row = {f"col{i}": (["a"] if i % 20 == 4 else [i]) for i in (0, *range(3, num + 1))}
+    bad_row = {f"col{i}": ["a"] if i % 20 == 4 else [i] for i in (0, *range(3, num + 1))}
     try:
         if method == "append":
             lib.append("x", pd.DataFrame(bad_row, index=pd.date_range("1970-01-04", periods=1)))
diff --git a/python/tests/unit/arcticdb/version_store/test_aggregation.py b/python/tests/unit/arcticdb/version_store/test_aggregation.py
index 8440b076dc..a3580dca6a 100644
--- a/python/tests/unit/arcticdb/version_store/test_aggregation.py
+++ b/python/tests/unit/arcticdb/version_store/test_aggregation.py
@@ -11,7 +11,7 @@
 from pandas import DataFrame
 
 from arcticdb.version_store.processing import QueryBuilder
-from arcticdb_ext.exceptions import InternalException
+from arcticdb_ext.exceptions import InternalException, SchemaException
 from arcticdb.util.test import assert_frame_equal
 from arcticdb.util.hypothesis import (
     use_of_function_scoped_fixtures_in_hypothesis_checked,
@@ -112,7 +112,7 @@ def test_hypothesis_max_min_agg(lmdb_version_store, df):
     assert_frame_equal(expected, vit.data)
 
 
-def test_sum_aggregation(s3_version_store):
+def test_sum_aggregation(lmdb_version_store):
     df = DataFrame(
         {"grouping_column": ["group_1", "group_1", "group_1", "group_2", "group_2"], "to_sum": [1, 1, 2, 2, 2]},
         index=np.arange(5),
@@ -120,9 +120,9 @@ def test_sum_aggregation(s3_version_store):
     q = QueryBuilder()
     q = q.groupby("grouping_column").agg({"to_sum": "sum"})
     symbol = "test_sum_aggregation"
-    s3_version_store.write(symbol, df)
+    lmdb_version_store.write(symbol, df)
 
-    res = s3_version_store.read(symbol, query_builder=q)
+    res = lmdb_version_store.read(symbol, query_builder=q)
     res.data.sort_index(inplace=True)
 
     df = pd.DataFrame({"to_sum": [4, 4]}, index=["group_1", "group_2"])
@@ -131,7 +131,7 @@ def test_sum_aggregation(s3_version_store):
     assert_frame_equal(res.data, df)
 
 
-def test_mean_aggregation(s3_version_store):
+def test_mean_aggregation(lmdb_version_store):
     df = DataFrame(
         {"grouping_column": ["group_1", "group_1", "group_1", "group_2", "group_2"], "to_mean": [1, 1, 2, 2, 2]},
         index=np.arange(5),
@@ -139,9 +139,9 @@ def test_mean_aggregation(s3_version_store):
     q = QueryBuilder()
     q = q.groupby("grouping_column").agg({"to_mean": "mean"})
     symbol = "test_aggregation"
-    s3_version_store.write(symbol, df)
+    lmdb_version_store.write(symbol, df)
 
-    res = s3_version_store.read(symbol, query_builder=q)
+    res = lmdb_version_store.read(symbol, query_builder=q)
     res.data.sort_index(inplace=True)
 
     df = pd.DataFrame({"to_mean": [4 / 3, 2]}, index=["group_1", "group_2"])
@@ -150,7 +150,7 @@ def test_mean_aggregation(s3_version_store):
     assert_frame_equal(res.data, df)
 
 
-def test_mean_aggregation_float(s3_version_store):
+def test_mean_aggregation_float(lmdb_version_store):
     df = DataFrame(
         {
             "grouping_column": ["group_1", "group_1", "group_1", "group_2", "group_2"],
@@ -161,9 +161,9 @@ def test_mean_aggregation_float(s3_version_store):
     q = QueryBuilder()
     q = q.groupby("grouping_column").agg({"to_mean": "mean"})
     symbol = "test_aggregation"
-    s3_version_store.write(symbol, df)
+    lmdb_version_store.write(symbol, df)
 
-    res = s3_version_store.read(symbol, query_builder=q)
+    res = lmdb_version_store.read(symbol, query_builder=q)
     res.data.sort_index(inplace=True)
 
     df = pd.DataFrame({"to_mean": [(1.1 + 1.4 + 2.5) / 3, 2.2]}, index=["group_1", "group_2"])
@@ -221,8 +221,8 @@ def test_group_empty_dataframe(lmdb_version_store):
 
     symbol = "test_group_empty_dataframe"
     lmdb_version_store.write(symbol, df)
-    vit = lmdb_version_store.read(symbol, query_builder=q)
-    assert vit.data.empty
+    with pytest.raises(SchemaException):
+        _ = lmdb_version_store.read(symbol, query_builder=q)
 
 
 def test_group_pickled_symbol(lmdb_version_store):
@@ -291,6 +291,48 @@ def test_group_column_splitting_strings(lmdb_version_store_tiny_segment):
     assert_frame_equal(expected, vit.data)
 
 
+def test_aggregation_with_nones_and_nans_in_string_grouping_column(version_store_factory):
+    lib = version_store_factory(column_group_size=2, segment_row_size=2, dynamic_strings=True)
+    symbol = "test_aggregation_with_nones_and_nans_in_string_grouping_column"
+    # Structured so that the row-slices of the grouping column contain:
+    # 1 - All strings
+    # 2 - Strings and Nones
+    # 3 - Strings and NaNs
+    # 4 - All Nones
+    # 5 - All NaNs
+    # 6 - Nones and NaNs
+    df = DataFrame(
+        {
+            "grouping_column": [
+                "group_1",
+                "group_2",
+                "group_1",
+                None,
+                np.nan,
+                "group_2",
+                None,
+                None,
+                np.nan,
+                np.nan,
+                None,
+                np.nan,
+            ],
+            "to_sum": np.arange(12),
+        },
+        index=np.arange(12),
+    )
+    lib.write(symbol, df, dynamic_strings=True)
+
+    expected = df.groupby("grouping_column").agg({"to_sum": "sum"})
+
+    q = QueryBuilder()
+    q = q.groupby("grouping_column").agg({"to_sum": "sum"})
+    res = lib.read(symbol, query_builder=q)
+    res.data.sort_index(inplace=True)
+
+    assert_frame_equal(res.data, expected)
+
+
 def test_docstring_example_query_builder_apply(lmdb_version_store):
     lib = lmdb_version_store
     df = pd.DataFrame(
@@ -334,6 +376,6 @@ def test_docstring_example_query_builder_groupby_max_and_mean(lmdb_version_store
 
     lmdb_version_store.write("symbol", df)
     res = lmdb_version_store.read("symbol", query_builder=q)
-    df = pd.DataFrame({"to_max": [2.5], "to_mean": (1.1 + 1.4 + 2.5) / 3}, index=["group_1"])
+    df = pd.DataFrame({"to_mean": (1.1 + 1.4 + 2.5) / 3, "to_max": [2.5]}, index=["group_1"])
     df.index.rename("grouping_column", inplace=True)
     assert_frame_equal(res.data, df)
diff --git a/python/tests/unit/arcticdb/version_store/test_aggregation_dynamic.py b/python/tests/unit/arcticdb/version_store/test_aggregation_dynamic.py
index 903db5bcb5..3bd0d3a3c6 100644
--- a/python/tests/unit/arcticdb/version_store/test_aggregation_dynamic.py
+++ b/python/tests/unit/arcticdb/version_store/test_aggregation_dynamic.py
@@ -19,7 +19,7 @@
     from pandas.core.base import SpecificationError
 
 from arcticdb.version_store.processing import QueryBuilder
-from arcticdb_ext.exceptions import InternalException
+from arcticdb_ext.exceptions import InternalException, SchemaException
 from arcticdb.util.test import make_dynamic, assert_frame_equal
 from arcticdb.util.hypothesis import (
     use_of_function_scoped_fixtures_in_hypothesis_checked,
@@ -174,8 +174,8 @@ def test_group_empty_dataframe_dynamic(lmdb_version_store_dynamic_schema):
 
     symbol = "test_group_empty_dataframe"
     lmdb_version_store_dynamic_schema.write(symbol, df)
-    vit = lmdb_version_store_dynamic_schema.read(symbol, query_builder=q)
-    assert vit.data.empty
+    with pytest.raises(SchemaException):
+        _ = lmdb_version_store_dynamic_schema.read(symbol, query_builder=q)
 
 
 def test_group_pickled_symbol_dynamic(lmdb_version_store_dynamic_schema):
@@ -194,7 +194,7 @@ def test_group_column_not_present_dynamic(lmdb_version_store_dynamic_schema):
     q = q.groupby("grouping_column").agg({"to_mean": "mean"})
     symbol = "test_group_column_not_present"
     lmdb_version_store_dynamic_schema.write(symbol, df)
-    with pytest.raises(InternalException) as e_info:
+    with pytest.raises(SchemaException):
         _ = lmdb_version_store_dynamic_schema.read(symbol, query_builder=q)
 
 
@@ -266,7 +266,6 @@ def test_segment_without_aggregation_column(lmdb_version_store_dynamic_schema):
         assert_equal_value(received, expected)
 
 
-@pytest.mark.xfail(reason="ArcticDB/issues/130")
 def test_minimal_repro_type_change(lmdb_version_store_dynamic_schema):
     lib = lmdb_version_store_dynamic_schema
     sym = "test_minimal_repro_type_change"
@@ -284,7 +283,6 @@ def test_minimal_repro_type_change(lmdb_version_store_dynamic_schema):
     assert_equal_value(received, expected)
 
 
-@pytest.mark.xfail(reason="ArcticDB/issues/130")
 def test_minimal_repro_type_change_max(lmdb_version_store_dynamic_schema):
     lib = lmdb_version_store_dynamic_schema
     sym = "test_minimal_repro_type_change_max"
@@ -316,3 +314,28 @@ def test_minimal_repro_type_sum_similar_string_group_values(lmdb_version_store_d
     q = q.groupby("grouping_column").agg({"to_sum": "sum"})
     received = lib.read(sym, query_builder=q).data
     assert_equal_value(received, expected)
+
+
+def test_aggregation_grouping_column_missing_from_row_group(lmdb_version_store_dynamic_schema):
+    lib = lmdb_version_store_dynamic_schema
+    df0 = DataFrame(
+        {"to_sum": [1, 2], "grouping_column": ["group_1", "group_2"]},
+        index=np.arange(2),
+    )
+    df1 = DataFrame(
+        {"to_sum": [3, 4]},
+        index=np.arange(2, 4),
+    )
+    expected = df0.append(df1).groupby("grouping_column").agg({"to_sum": "sum"})
+
+    symbol = "test_aggregation_grouping_column_missing_from_row_group"
+    lib.write(symbol, df0)
+    lib.append(symbol, df1)
+
+    q = QueryBuilder()
+    q = q.groupby("grouping_column").agg({"to_sum": "sum"})
+
+    res = lib.read(symbol, query_builder=q)
+    res.data.sort_index(inplace=True)
+
+    assert_frame_equal(res.data, expected)
diff --git a/python/tests/unit/arcticdb/version_store/test_filtering.py b/python/tests/unit/arcticdb/version_store/test_filtering.py
index ccbe37dd9b..6f00568183 100644
--- a/python/tests/unit/arcticdb/version_store/test_filtering.py
+++ b/python/tests/unit/arcticdb/version_store/test_filtering.py
@@ -67,6 +67,7 @@ def test_querybuilder_shallow_copy(lmdb_version_store):
     q = q[q["a"] > 1]
     q_copy = copy.copy(q)
     pandas_query = "a > 1"
+    generic_filter_test(lmdb_version_store, "test_querybuilder_shallow_copy", df, q, pandas_query)
     generic_filter_test(lmdb_version_store, "test_querybuilder_shallow_copy", df, q_copy, pandas_query)
 
 
@@ -76,6 +77,8 @@ def test_querybuilder_deepcopy(lmdb_version_store):
     q = q[q["a"] > 1]
     q_copy = copy.deepcopy(q)
     pandas_query = "a > 1"
+    generic_filter_test(lmdb_version_store, "test_querybuilder_deepcopy", df, q, pandas_query)
+    del q
     generic_filter_test(lmdb_version_store, "test_querybuilder_deepcopy", df, q_copy, pandas_query)
 
 
@@ -84,8 +87,10 @@ def test_querybuilder_pickle(lmdb_version_store):
     q = QueryBuilder()
     q = q[q["a"] > 1]
     q_pickled = pickle.dumps(q)
-    q_unpickled = pickle.loads(q_pickled)
     pandas_query = "a > 1"
+    generic_filter_test(lmdb_version_store, "test_querybuilder_pickle", df, q, pandas_query)
+    del q
+    q_unpickled = pickle.loads(q_pickled)
     generic_filter_test(lmdb_version_store, "test_querybuilder_pickle", df, q_unpickled, pandas_query)
 
 
@@ -142,16 +147,10 @@ def test_filter_explicit_index(request, lib_type):
     assert_frame_equal(df.query(pandas_query), lib.read(symbol, query_builder=q).data)
 
 
-@pytest.mark.parametrize("lib_type", ["lmdb_version_store", "lmdb_version_store_dynamic_schema"])
-def test_filter_infinite_value(request, lib_type):
-    lib = request.getfixturevalue(lib_type)
-    df = DataFrame({"a": np.arange(1)})
+def test_filter_infinite_value():
     q = QueryBuilder()
-    q = q[q["a"] < inf]
-    symbol = "test_filter_infinite_value"
-    lib.write(symbol, df)
-    with pytest.raises(ArcticNativeException) as e_info:
-        _ = lib.read(symbol, query_builder=q)
+    with pytest.raises(ArcticNativeException):
+        q = q[q["a"] < inf]
 
 
 @pytest.mark.parametrize("lib_type", ["lmdb_version_store", "lmdb_version_store_dynamic_schema"])
@@ -1062,7 +1061,6 @@ def test_filter_stringpool_shrinking_block_alignment(lmdb_version_store):
     q = QueryBuilder()
     string_to_find = data[3]
     q = q[q["a"] == string_to_find]
-    q.optimise_for_memory()
     pandas_query = f"a == '{string_to_find}'"
     generic_filter_test_strings(
         lmdb_version_store, "test_filter_stringpool_shrinking_block_alignment", df, q, pandas_query
@@ -2018,12 +2016,14 @@ def test_filter_numeric_membership_equivalence():
 
 def test_filter_bool_short_circuiting():
     def _clear(first, second):
-        first.stages.clear()
-        second.stages.clear()
+        first.clauses.clear()
+        first._python_clauses.clear()
+        second.clauses.clear()
+        second._python_clauses.clear()
 
     # Original query
     q1 = QueryBuilder()
-    # Expected short circuited version
+    # Expected short-circuited version
     q2 = QueryBuilder()
     errors = []
 
@@ -2033,10 +2033,8 @@ def _clear(first, second):
         errors.append("and true")
     _clear(q1, q2)
 
-    q1 = q1[(q1["a"] < 5) & False]
-    q2 = q2[False]
-    if q1 != q2:
-        errors.append("and false")
+    with pytest.raises(ArcticNativeException):
+        q1 = q1[(q1["a"] < 5) & False]
     _clear(q1, q2)
 
     q1 = q1[True & (q1["a"] < 5)]
@@ -2045,16 +2043,12 @@ def _clear(first, second):
         errors.append("rand true")
     _clear(q1, q2)
 
-    q1 = q1[False & (q1["a"] < 5)]
-    q2 = q2[False]
-    if q1 != q2:
-        errors.append("rand false")
+    with pytest.raises(ArcticNativeException):
+        q1 = q1[False & (q1["a"] < 5)]
     _clear(q1, q2)
 
-    q1 = q1[(q1["a"] < 5) | True]
-    q2 = q2[True]
-    if q1 != q2:
-        errors.append("or true")
+    with pytest.raises(ArcticNativeException):
+        q1 = q1[(q1["a"] < 5) | True]
     _clear(q1, q2)
 
     q1 = q1[(q1["a"] < 5) | False]
@@ -2063,10 +2057,8 @@ def _clear(first, second):
         errors.append("or false")
     _clear(q1, q2)
 
-    q1 = q1[True | (q1["a"] < 5)]
-    q2 = q2[True]
-    if q1 != q2:
-        errors.append("ror true")
+    with pytest.raises(ArcticNativeException):
+        q1 = q1[True | (q1["a"] < 5)]
     _clear(q1, q2)
 
     q1 = q1[False | (q1["a"] < 5)]
diff --git a/python/tests/unit/arcticdb/version_store/test_parallel.py b/python/tests/unit/arcticdb/version_store/test_parallel.py
index 39656cb625..9118d48a84 100644
--- a/python/tests/unit/arcticdb/version_store/test_parallel.py
+++ b/python/tests/unit/arcticdb/version_store/test_parallel.py
@@ -90,7 +90,6 @@ def test_floats_to_nans(lmdb_version_store):
     assert_frame_equal(vit.data, df)
 
 
-@pytest.mark.skip("index refact")
 def test_sort_merge_write(lmdb_version_store):
     num_rows_per_day = 10
     num_days = 10
@@ -123,7 +122,6 @@ def test_sort_merge_write(lmdb_version_store):
     assert_frame_equal(vit.data, df)
 
 
-@pytest.mark.skip("index refact")
 def test_sort_merge_append(lmdb_version_store_dynamic_schema):
     lib = lmdb_version_store_dynamic_schema
     num_rows_per_day = 10
diff --git a/python/tests/unit/arcticdb/version_store/test_projection_dynamic.py b/python/tests/unit/arcticdb/version_store/test_projection_dynamic.py
index 489293938d..65ad39a8ab 100644
--- a/python/tests/unit/arcticdb/version_store/test_projection_dynamic.py
+++ b/python/tests/unit/arcticdb/version_store/test_projection_dynamic.py
@@ -46,6 +46,34 @@ def test_project_dynamic(lmdb_version_store_dynamic_schema):
     assert_frame_equal(expected, received)
 
 
+def test_project_column_types_changing_and_missing(lmdb_version_store_dynamic_schema):
+    lib = lmdb_version_store_dynamic_schema
+    symbol = "test_project_column_types_changing_and_missing"
+    # Floats
+    expected = pd.DataFrame({"col_to_project": [0.5, 1.5], "data_col": [0, 1]}, index=np.arange(0, 2))
+    lib.write(symbol, expected)
+    # uint8
+    df = pd.DataFrame({"col_to_project": np.arange(2, dtype=np.uint8), "data_col": [2, 3]}, index=np.arange(2, 4))
+    lib.append(symbol, df)
+    expected = expected.append(df)
+    # Missing
+    df = pd.DataFrame({"data_col": [4, 5]}, index=np.arange(4, 6))
+    lib.append(symbol, df)
+    expected = expected.append(df)
+    # int16
+    df = pd.DataFrame(
+        {"col_to_project": np.arange(200, 202, dtype=np.int16), "data_col": [6, 7]}, index=np.arange(6, 8)
+    )
+    lib.append(symbol, df)
+
+    expected = expected.append(df)
+    expected["projected_col"] = expected["col_to_project"] * 2
+    q = QueryBuilder()
+    q = q.apply("projected_col", q["col_to_project"] * 2)
+    received = lib.read(symbol, query_builder=q).data
+    assert_frame_equal(expected, received)
+
+
 @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
new file mode 100644
index 0000000000..daa953402d
--- /dev/null
+++ b/python/tests/unit/arcticdb/version_store/test_query_builder.py
@@ -0,0 +1,211 @@
+"""
+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.
+"""
+import numpy as np
+import pandas as pd
+
+from arcticdb.version_store.processing import QueryBuilder
+from arcticdb.util.test import assert_frame_equal
+
+
+def test_reuse_querybuilder(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_reuse_querybuilder"
+    df = pd.DataFrame(
+        {"col1": np.arange(10, dtype=np.int64), "col2": np.arange(100, 110, dtype=np.int64)}, index=np.arange(10)
+    )
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q[q["col1"].isin(2, 3, 7)]
+    expected = df.query("col1 in [2, 3, 7]")
+    received = lib.read(symbol, query_builder=q).data
+    assert_frame_equal(expected, received)
+
+    q = q.apply("new_col", (q["col1"] * q["col2"]) + 13)
+    expected = df.query("col1 in [2, 3, 7]")
+    received = lib.read(symbol, query_builder=q).data
+
+    expected["new_col"] = (expected["col1"] * expected["col2"]) + 13
+    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"
+    df = pd.DataFrame({"col1": np.arange(10), "col2": np.arange(100, 110)}, index=np.arange(10))
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q[q["col1"].isin(2, 3, 7)]
+    q = q[q["col1"].isin(2, 3)]
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col1 in [2, 3]")
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_filter_then_project(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_project"
+    df = pd.DataFrame(
+        {"col1": np.arange(10, dtype=np.int64), "col2": np.arange(100, 110, dtype=np.int64)}, index=np.arange(10)
+    )
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q[q["col1"].isin(2, 3, 7)]
+    q = q.apply("new_col", (q["col1"] * q["col2"]) + 13)
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df.query("col1 in [2, 3, 7]")
+    expected["new_col"] = expected["col1"] * expected["col2"] + 13
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_filter_then_groupby(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_filter_then_groupby"
+    df = pd.DataFrame(
+        {
+            "col1": ["a", "b", "c", "a", "b", "c", "a", "b", "c", "d"],
+            "col2": [1, 2, 3, 2, 1, 3, 1, 1, 3, 4],
+        },
+        index=np.arange(10),
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q[q["col1"] != "b"]
+    q = q.groupby("col1").agg({"col2": "sum"})
+    received = lib.read(symbol, query_builder=q).data
+    received.sort_index(inplace=True)
+    expected = df.query("col1 != 'b'").groupby("col1").agg({"col2": "sum"})
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_project_then_filter(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_filter"
+    df = pd.DataFrame(
+        {"col1": np.arange(10, dtype=np.int64), "col2": np.arange(100, 110, dtype=np.int64)}, index=np.arange(10)
+    )
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q.apply("new_col", (q["col1"] * q["col2"]) + 13)
+    q = q[q["new_col"].isin(13, 114, 538)]
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df
+    expected["new_col"] = expected["col1"] * expected["col2"] + 13
+    expected = expected.query("new_col in [13, 114, 538]")
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_project_then_project(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_project"
+    df = pd.DataFrame(
+        {"col1": np.arange(10, dtype=np.int64), "col2": np.arange(100, 110, dtype=np.int64)}, index=np.arange(10)
+    )
+    lib.write(symbol, df)
+
+    q = QueryBuilder()
+    q = q.apply("new_col1", (q["col1"] * q["col2"]) + 13)
+    q = q.apply("new_col2", (q["new_col1"] * q["col2"]) - 5)
+    received = lib.read(symbol, query_builder=q).data
+
+    expected = df
+    expected["new_col1"] = expected["col1"] * expected["col2"] + 13
+    expected["new_col2"] = expected["new_col1"] * expected["col2"] - 5
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_project_then_groupby(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_project_then_groupby"
+    df = pd.DataFrame(
+        {
+            "col1": [1, 2, 2, 3, 3, 3, 4, 4, 4, 4],
+            "col2": np.arange(0, 1, 0.1, dtype=np.float64),
+        },
+        index=np.arange(10),
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.apply("new_col1", q["col1"] * 3)
+    q = q.apply("new_col2", q["col2"] + 2.5)
+    q = q.groupby("new_col1").agg({"new_col2": "sum"})
+    received = lib.read(symbol, query_builder=q).data
+    received.sort_index(inplace=True)
+    expected = df
+    expected["new_col1"] = expected["col1"] * 3
+    expected["new_col2"] = expected["col2"] + 2.5
+    expected = expected.groupby("new_col1").agg({"new_col2": "sum"})
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_groupby_then_filter(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_groupby_then_filter"
+    df = pd.DataFrame(
+        {
+            "col1": ["a", "b", "c", "a", "b", "c", "a", "b", "c", "d"],
+            "col2": [1, 2, 3, 2, 1, 3, 1, 1, 3, 4],
+        },
+        index=np.arange(10),
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.groupby("col1").agg({"col2": "sum"})
+    q = q[(q["col1"] != "b") & (q["col2"] != 9)]
+    received = lib.read(symbol, query_builder=q).data
+    received.sort_index(inplace=True)
+    expected = df.groupby("col1").agg({"col2": "sum"}).query("(col1 != 'b') & (col2 != 9)")
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_groupby_then_project(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_groupby_then_project"
+    df = pd.DataFrame(
+        {
+            "col1": [5, 23, 42, 5, 23, 42, 5, 23, 42, 0],
+            "col2": [1, 2, 3, 2, 1, 3, 1, 1, 3, 4],
+        },
+        index=np.arange(10),
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.groupby("col1").agg({"col2": "sum"})
+    q = q.apply("new_col", q["col2"] * 3)
+    received = lib.read(symbol, query_builder=q).data
+    received.sort_index(inplace=True)
+    expected = df.groupby("col1").agg({"col2": "sum"})
+    expected["new_col"] = expected["col2"] * 3
+    assert_frame_equal(expected, received)
+
+
+def test_querybuilder_groupby_then_groupby(lmdb_version_store_tiny_segment):
+    lib = lmdb_version_store_tiny_segment
+    symbol = "test_querybuilder_groupby_then_groupby"
+    df = pd.DataFrame(
+        {
+            "col1": ["a", "b", "c", "a", "b", "c", "a", "b", "c", "d"],
+            "col2": [1, 2, 3, 2, 1, 3, 1, 1, 3, 4],
+            "col3": np.arange(100, 110),
+        },
+        index=np.arange(10),
+    )
+    lib.write(symbol, df)
+    q = QueryBuilder()
+    q = q.groupby("col1").agg({"col2": "sum", "col3": "mean"})
+    q = q.groupby("col2").agg({"col3": "mean"})
+    received = lib.read(symbol, query_builder=q).data
+    received.sort_index(inplace=True)
+    expected = df.groupby("col1").agg({"col2": "sum", "col3": "mean"}).groupby("col2").agg({"col3": "mean"})
+    assert_frame_equal(expected, received)