refactor(interpreter): aggreagator batching

Optimize calls to potentially costly aggregators by only extracting the value
from a stateful aggregator when it is needed by subsequent RETE nodes.

This optimization only works with timeless evaluation and delete-and-rederive
evaluation disabled, i.e., only for queries without any recursion. Potentially,
it could also be extended to other mailboxes if needed.

We replace the BehaviorChangingMailbox of ColumnAggregatorNode with a
DefaultMailbox to force update batching. Batched updates only extract the value
from the aggregator when it has been already updated with all received tuples.

2 files changed, 563 insertions, 328 deletions

diff --git a/subprojects/interpreter-rete/src/main/java/tools/refinery/interpreter/rete/aggregation/ColumnAggregatorNode.java b/subprojects/interpreter-rete/src/main/java/tools/refinery/interpreter/rete/aggregation/ColumnAggregatorNode.java
index b1a25807..d5f4a0e4 100644
--- a/subprojects/interpreter-rete/src/main/java/tools/refinery/interpreter/rete/aggregation/ColumnAggregatorNode.java
+++ b/subprojects/interpreter-rete/src/main/java/tools/refinery/interpreter/rete/aggregation/ColumnAggregatorNode.java
@@ -1,5 +1,6 @@
 /*******************************************************************************
  * Copyright (c) 2010-2016, Gabor Bergmann, IncQueryLabs Ltd.
+ * Copyright (c) 2023 The Refinery Authors <https://refinery.tools>
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0 which is available at
  * http://www.eclipse.org/legal/epl-v20.html.
@@ -8,9 +9,6 @@
  *******************************************************************************/
 package tools.refinery.interpreter.rete.aggregation;
 
-import java.util.Map;
-import java.util.Map.Entry;
-
 import tools.refinery.interpreter.matchers.context.IPosetComparator;
 import tools.refinery.interpreter.matchers.psystem.aggregations.IMultisetAggregationOperator;
 import tools.refinery.interpreter.matchers.tuple.Tuple;
@@ -26,8 +24,13 @@ import tools.refinery.interpreter.rete.network.communication.Timestamp;
 import tools.refinery.interpreter.rete.network.communication.timeless.RecursiveCommunicationGroup;
 import tools.refinery.interpreter.rete.network.mailbox.Mailbox;
 import tools.refinery.interpreter.rete.network.mailbox.timeless.BehaviorChangingMailbox;
+import tools.refinery.interpreter.rete.network.mailbox.timeless.DefaultMailbox;
 import tools.refinery.interpreter.rete.network.mailbox.timeless.PosetAwareMailbox;
 
+import java.util.Collection;
+import java.util.Map;
+import java.util.Map.Entry;
+
 /**
  * Timeless implementation of the column aggregator node.
  * <p>
@@ -40,330 +43,376 @@ import tools.refinery.interpreter.rete.network.mailbox.timeless.PosetAwareMailbo
  * @since 1.4
  */
 public class ColumnAggregatorNode<Domain, Accumulator, AggregateResult>
-        extends AbstractColumnAggregatorNode<Domain, Accumulator, AggregateResult>
-        implements RederivableNode, PosetAwareReceiver {
-
-    /**
-     * @since 1.6
-     */
-    protected final IPosetComparator posetComparator;
-
-    /**
-     * @since 1.6
-     */
-    protected final boolean deleteRederiveEvaluation;
-
-    // invariant: neutral values are not stored
-    /**
-     * @since 1.6
-     */
-    protected final Map<Tuple, Accumulator> memory;
-    /**
-     * @since 1.6
-     */
-    protected final Map<Tuple, Accumulator> rederivableMemory;
-
-    /**
-     * @since 1.7
-     */
-    protected CommunicationGroup currentGroup;
-
-    /**
-     * Creates a new column aggregator node.
-     *
-     * @param reteContainer
-     *            the RETE container of the node
-     * @param operator
-     *            the aggregation operator
-     * @param deleteRederiveEvaluation
-     *            true if the node should run in DRED mode, false otherwise
-     * @param groupMask
-     *            the mask that masks a tuple to obtain the key that we are grouping-by
-     * @param columnMask
-     *            the mask that masks a tuple to obtain the tuple element(s) that we are aggregating over
-     * @param posetComparator
-     *            the poset comparator for the column, if known, otherwise it can be null
-     * @since 1.6
-     */
-    public ColumnAggregatorNode(final ReteContainer reteContainer,
-            final IMultisetAggregationOperator<Domain, Accumulator, AggregateResult> operator,
-            final boolean deleteRederiveEvaluation, final TupleMask groupMask, final TupleMask columnMask,
-            final IPosetComparator posetComparator) {
-        super(reteContainer, operator, groupMask, columnMask);
-        this.memory = CollectionsFactory.createMap();
-        this.rederivableMemory = CollectionsFactory.createMap();
-        this.deleteRederiveEvaluation = deleteRederiveEvaluation;
-        this.posetComparator = posetComparator;
-        // mailbox MUST be instantiated after the fields are all set
-        this.mailbox = instantiateMailbox();
-    }
-
-    /**
-     * Creates a new column aggregator node.
-     *
-     * @param reteContainer
-     *            the RETE container of the node
-     * @param operator
-     *            the aggregation operator
-     * @param groupMask
-     *            the mask that masks a tuple to obtain the key that we are grouping-by
-     * @param aggregatedColumn
-     *            the index of the column that the aggregator node is aggregating over
-     */
-    public ColumnAggregatorNode(final ReteContainer reteContainer,
-            final IMultisetAggregationOperator<Domain, Accumulator, AggregateResult> operator,
-            final TupleMask groupMask, final int aggregatedColumn) {
-        this(reteContainer, operator, false, groupMask, TupleMask.selectSingle(aggregatedColumn, groupMask.sourceWidth),
-                null);
-    }
-
-    @Override
-    public boolean isInDRedMode() {
-        return this.deleteRederiveEvaluation;
-    }
-
-    @Override
-    protected Mailbox instantiateMailbox() {
-        if (groupMask != null && columnMask != null && posetComparator != null) {
-            return new PosetAwareMailbox(this, this.reteContainer);
-        } else {
-            return new BehaviorChangingMailbox(this, this.reteContainer);
-        }
-    }
-
-    @Override
-    public TupleMask getCoreMask() {
-        return groupMask;
-    }
-
-    @Override
-    public TupleMask getPosetMask() {
-        return columnMask;
-    }
-
-    @Override
-    public IPosetComparator getPosetComparator() {
-        return posetComparator;
-    }
-
-    @Override
-    public void rederiveOne() {
-        final Entry<Tuple, Accumulator> entry = rederivableMemory.entrySet().iterator().next();
-        final Tuple group = entry.getKey();
-        final Accumulator accumulator = entry.getValue();
-        rederivableMemory.remove(group);
-        memory.put(group, accumulator);
-        // unregister the node if there is nothing left to be re-derived
-        if (this.rederivableMemory.isEmpty()) {
-            ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
-        }
-        final AggregateResult value = operator.getAggregate(accumulator);
-        propagateAggregateResultUpdate(group, NEUTRAL, value, Timestamp.ZERO);
-    }
-
-    @Override
-    public void updateWithPosetInfo(final Direction direction, final Tuple update, final boolean monotone) {
-        if (this.deleteRederiveEvaluation) {
-            updateWithDeleteAndRederive(direction, update, monotone);
-        } else {
-            updateDefault(direction, update, Timestamp.ZERO);
-        }
-    }
-
-    @Override
-    public void update(final Direction direction, final Tuple update, final Timestamp timestamp) {
-        updateWithPosetInfo(direction, update, false);
-    }
-
-    /**
-     * @since 2.4
-     */
-    protected void updateDefault(final Direction direction, final Tuple update, final Timestamp timestamp) {
-        final Tuple key = groupMask.transform(update);
-        final Tuple value = columnMask.transform(update);
-        @SuppressWarnings("unchecked")
-        final Domain aggregableValue = (Domain) runtimeContext.unwrapElement(value.get(0));
-        final boolean isInsertion = direction == Direction.INSERT;
-
-        final Accumulator oldMainAccumulator = getMainAccumulator(key);
-        final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
-
-        final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue, isInsertion);
-        storeIfNotNeutral(key, newMainAccumulator, memory);
-        final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
-
-        propagateAggregateResultUpdate(key, oldValue, newValue, timestamp);
-    }
-
-    /**
-     * @since 2.4
-     */
-    protected void updateWithDeleteAndRederive(final Direction direction, final Tuple update, final boolean monotone) {
-        final Tuple group = groupMask.transform(update);
-        final Tuple value = columnMask.transform(update);
-        @SuppressWarnings("unchecked")
-        final Domain aggregableValue = (Domain) runtimeContext.unwrapElement(value.get(0));
-        final boolean isInsertion = direction == Direction.INSERT;
-
-        Accumulator oldMainAccumulator = memory.get(group);
-        Accumulator oldRederivableAccumulator = rederivableMemory.get(group);
-
-        if (direction == Direction.INSERT) {
-            // INSERT
-            if (oldRederivableAccumulator != null) {
-                // the group is in the re-derivable memory
-                final Accumulator newRederivableAccumulator = operator.update(oldRederivableAccumulator,
-                        aggregableValue, isInsertion);
-                storeIfNotNeutral(group, newRederivableAccumulator, rederivableMemory);
-                if (rederivableMemory.isEmpty()) {
-                    // there is nothing left to be re-derived
-                    // this can happen if the accumulator became neutral in response to the INSERT
-                    ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
-                }
-            } else {
-                // the group is in the main memory
-                // at this point, it can happen that we need to initialize with a neutral accumulator
-                if (oldMainAccumulator == null) {
-                    oldMainAccumulator = operator.createNeutral();
-                }
-
-                final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
-                final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue,
-                        isInsertion);
-                storeIfNotNeutral(group, newMainAccumulator, memory);
-                final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
-                propagateAggregateResultUpdate(group, oldValue, newValue, Timestamp.ZERO);
-            }
-        } else {
-            // DELETE
-            if (oldRederivableAccumulator != null) {
-                // the group is in the re-derivable memory
-                if (oldMainAccumulator != null) {
-                    issueError("[INTERNAL ERROR] Inconsistent state for " + update
-                            + " because it is present both in the main and re-derivable memory in the ColumnAggregatorNode "
-                            + this + " for pattern(s) " + getTraceInfoPatternsEnumerated(), null);
-                }
-                try {
-                    final Accumulator newRederivableAccumulator = operator.update(oldRederivableAccumulator,
-                            aggregableValue, isInsertion);
-                    storeIfNotNeutral(group, newRederivableAccumulator, rederivableMemory);
-                    if (rederivableMemory.isEmpty()) {
-                        // there is nothing left to be re-derived
-                        // this can happen if the accumulator became neutral in response to the DELETE
-                        ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
-                    }
-                } catch (final NullPointerException ex) {
-                    issueError("[INTERNAL ERROR] Deleting a domain element in " + update
-                            + " which did not exist before in ColumnAggregatorNode " + this + " for pattern(s) "
-                            + getTraceInfoPatternsEnumerated(), ex);
-                }
-            } else {
-                // the group is in the main memory
-                // at this point, it can happen that we need to initialize with a neutral accumulator
-                if (oldMainAccumulator == null) {
-                    oldMainAccumulator = operator.createNeutral();
-                }
-
-                final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
-                final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue,
-                        isInsertion);
-                final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
-
-                if (monotone) {
-                    storeIfNotNeutral(group, newMainAccumulator, memory);
-                    propagateAggregateResultUpdate(group, oldValue, newValue, Timestamp.ZERO);
-                } else {
-                    final boolean wasEmpty = rederivableMemory.isEmpty();
-                    if (storeIfNotNeutral(group, newMainAccumulator, rederivableMemory) && wasEmpty) {
-                        ((RecursiveCommunicationGroup) currentGroup).addRederivable(this);
-                    }
-                    memory.remove(group);
-                    propagateAggregateResultUpdate(group, oldValue, NEUTRAL, Timestamp.ZERO);
-                }
-            }
-        }
-    }
-
-    @Override
-    public void clear() {
-        this.memory.clear();
-        this.rederivableMemory.clear();
-        this.childMailboxes.clear();
-    }
-
-    /**
-     * Returns true if the accumulator was stored, false otherwise.
-     *
-     * @since 1.6
-     */
-    protected boolean storeIfNotNeutral(final Tuple key, final Accumulator accumulator,
-            final Map<Tuple, Accumulator> memory) {
-        if (operator.isNeutral(accumulator)) {
-            memory.remove(key);
-            return false;
-        } else {
-            memory.put(key, accumulator);
-            return true;
-        }
-    }
-
-    @Override
-    public Tuple getAggregateTuple(final Tuple group) {
-        final Accumulator accumulator = getMainAccumulator(group);
-        final AggregateResult result = operator.getAggregate(accumulator);
-        return tupleFromAggregateResult(group, result);
-    }
-
-    @Override
-    public AggregateResult getAggregateResult(final Tuple group) {
-        final Accumulator accumulator = getMainAccumulator(group);
-        return operator.getAggregate(accumulator);
-    }
-
-    @Override
-    public Map<AggregateResult, Timeline<Timestamp>> getAggregateResultTimeline(Tuple key) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Map<Tuple, Timeline<Timestamp>> getAggregateTupleTimeline(Tuple key) {
-        throw new UnsupportedOperationException();
-    }
-
-    /**
-     * @since 1.6
-     */
-    protected Accumulator getMainAccumulator(final Tuple key) {
-        return getAccumulator(key, memory);
-    }
-
-    /**
-     * @since 1.6
-     */
-    protected Accumulator getRederivableAccumulator(final Tuple key) {
-        return getAccumulator(key, rederivableMemory);
-    }
-
-    /**
-     * @since 1.6
-     */
-    protected Accumulator getAccumulator(final Tuple key, final Map<Tuple, Accumulator> memory) {
-        Accumulator accumulator = memory.get(key);
-        if (accumulator == null) {
-            return operator.createNeutral();
-        } else {
-            return accumulator;
-        }
-    }
-
-    @Override
-    public CommunicationGroup getCurrentGroup() {
-        return currentGroup;
-    }
-
-    @Override
-    public void setCurrentGroup(final CommunicationGroup currentGroup) {
-        this.currentGroup = currentGroup;
-    }
+                extends AbstractColumnAggregatorNode<Domain, Accumulator, AggregateResult>
+                implements RederivableNode, PosetAwareReceiver {
+
+        /**
+         * @since 1.6
+         */
+        protected final IPosetComparator posetComparator;
+
+        /**
+         * @since 1.6
+         */
+        protected final boolean deleteRederiveEvaluation;
+
+        // invariant: neutral values are not stored
+        /**
+         * @since 1.6
+         */
+        protected final Map<Tuple, Accumulator> memory;
+        /**
+         * @since 1.6
+         */
+        protected final Map<Tuple, Accumulator> rederivableMemory;
+
+        /**
+         * @since 1.7
+         */
+        protected CommunicationGroup currentGroup;
+
+        /**
+         * Creates a new column aggregator node.
+         *
+         * @param reteContainer            the RETE container of the node
+         * @param operator                 the aggregation operator
+         * @param deleteRederiveEvaluation true if the node should run in DRED mode, false otherwise
+         * @param groupMask                the mask that masks a tuple to obtain the key that we are grouping-by
+         * @param columnMask               the mask that masks a tuple to obtain the tuple element(s) that we are
+         *                                    aggregating over
+         * @param posetComparator          the poset comparator for the column, if known, otherwise it can be null
+         * @since 1.6
+         */
+        public ColumnAggregatorNode(final ReteContainer reteContainer,
+                                                                final IMultisetAggregationOperator<Domain, Accumulator, AggregateResult> operator,
+                                                                final boolean deleteRederiveEvaluation, final TupleMask groupMask,
+                                                                final TupleMask columnMask,
+                                                                final IPosetComparator posetComparator) {
+                super(reteContainer, operator, groupMask, columnMask);
+                this.memory = CollectionsFactory.createMap();
+                this.rederivableMemory = CollectionsFactory.createMap();
+                this.deleteRederiveEvaluation = deleteRederiveEvaluation;
+                this.posetComparator = posetComparator;
+                // mailbox MUST be instantiated after the fields are all set
+                this.mailbox = instantiateMailbox();
+        }
+
+        /**
+         * Creates a new column aggregator node.
+         *
+         * @param reteContainer    the RETE container of the node
+         * @param operator         the aggregation operator
+         * @param groupMask        the mask that masks a tuple to obtain the key that we are grouping-by
+         * @param aggregatedColumn the index of the column that the aggregator node is aggregating over
+         */
+        public ColumnAggregatorNode(final ReteContainer reteContainer,
+                                                                final IMultisetAggregationOperator<Domain, Accumulator, AggregateResult> operator,
+                                                                final TupleMask groupMask, final int aggregatedColumn) {
+                this(reteContainer, operator, false, groupMask, TupleMask.selectSingle(aggregatedColumn,
+                                                groupMask.sourceWidth),
+                                null);
+        }
+
+        @Override
+        public boolean isInDRedMode() {
+                return this.deleteRederiveEvaluation;
+        }
+
+        @Override
+        protected Mailbox instantiateMailbox() {
+                if (groupMask != null && columnMask != null && posetComparator != null) {
+                        return new PosetAwareMailbox(this, this.reteContainer);
+                } else if (deleteRederiveEvaluation) {
+                        return new BehaviorChangingMailbox(this, this.reteContainer);
+                } else {
+                        // Disable fall-through to enabled batched updates.
+                        return new DefaultMailbox(this, this.reteContainer);
+                }
+        }
+
+        @Override
+        public TupleMask getCoreMask() {
+                return groupMask;
+        }
+
+        @Override
+        public TupleMask getPosetMask() {
+                return columnMask;
+        }
+
+        @Override
+        public IPosetComparator getPosetComparator() {
+                return posetComparator;
+        }
+
+        @Override
+        public void rederiveOne() {
+                final Entry<Tuple, Accumulator> entry = rederivableMemory.entrySet().iterator().next();
+                final Tuple group = entry.getKey();
+                final Accumulator accumulator = entry.getValue();
+                rederivableMemory.remove(group);
+                memory.put(group, accumulator);
+                // unregister the node if there is nothing left to be re-derived
+                if (this.rederivableMemory.isEmpty()) {
+                        ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
+                }
+                final AggregateResult value = operator.getAggregate(accumulator);
+                propagateAggregateResultUpdate(group, NEUTRAL, value, Timestamp.ZERO);
+        }
+
+        @Override
+        public void updateWithPosetInfo(final Direction direction, final Tuple update, final boolean monotone) {
+                if (this.deleteRederiveEvaluation) {
+                        updateWithDeleteAndRederive(direction, update, monotone);
+                } else {
+                        updateDefault(direction, update, Timestamp.ZERO);
+                }
+        }
+
+        @Override
+        public void update(final Direction direction, final Tuple update, final Timestamp timestamp) {
+                updateWithPosetInfo(direction, update, false);
+        }
+
+        /**
+         * @since 2.4
+         */
+        protected void updateDefault(final Direction direction, final Tuple update, final Timestamp timestamp) {
+                final Tuple key = groupMask.transform(update);
+                final Tuple value = columnMask.transform(update);
+                @SuppressWarnings("unchecked") final Domain aggregableValue =
+                                (Domain) runtimeContext.unwrapElement(value.get(0));
+                final boolean isInsertion = direction == Direction.INSERT;
+
+                final Accumulator oldMainAccumulator = getMainAccumulator(key);
+                final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
+
+                final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue, isInsertion);
+                storeIfNotNeutral(key, newMainAccumulator, memory);
+                final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
+
+                propagateAggregateResultUpdate(key, oldValue, newValue, timestamp);
+        }
+
+        /**
+         * @since 2.4
+         */
+        protected void updateWithDeleteAndRederive(final Direction direction, final Tuple update, final boolean monotone) {
+                final Tuple group = groupMask.transform(update);
+                final Tuple value = columnMask.transform(update);
+                @SuppressWarnings("unchecked") final Domain aggregableValue =
+                                (Domain) runtimeContext.unwrapElement(value.get(0));
+                final boolean isInsertion = direction == Direction.INSERT;
+
+                Accumulator oldMainAccumulator = memory.get(group);
+                Accumulator oldRederivableAccumulator = rederivableMemory.get(group);
+
+                if (direction == Direction.INSERT) {
+                        // INSERT
+                        if (oldRederivableAccumulator != null) {
+                                // the group is in the re-derivable memory
+                                final Accumulator newRederivableAccumulator = operator.update(oldRederivableAccumulator,
+                                                aggregableValue, isInsertion);
+                                storeIfNotNeutral(group, newRederivableAccumulator, rederivableMemory);
+                                if (rederivableMemory.isEmpty()) {
+                                        // there is nothing left to be re-derived
+                                        // this can happen if the accumulator became neutral in response to the INSERT
+                                        ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
+                                }
+                        } else {
+                                // the group is in the main memory
+                                // at this point, it can happen that we need to initialize with a neutral accumulator
+                                if (oldMainAccumulator == null) {
+                                        oldMainAccumulator = operator.createNeutral();
+                                }
+
+                                final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
+                                final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue,
+                                                isInsertion);
+                                storeIfNotNeutral(group, newMainAccumulator, memory);
+                                final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
+                                propagateAggregateResultUpdate(group, oldValue, newValue, Timestamp.ZERO);
+                        }
+                } else {
+                        // DELETE
+                        if (oldRederivableAccumulator != null) {
+                                // the group is in the re-derivable memory
+                                if (oldMainAccumulator != null) {
+                                        issueError("[INTERNAL ERROR] Inconsistent state for " + update
+                                                        + " because it is present both in the main and re-derivable memory in the " +
+                                                        "ColumnAggregatorNode "
+                                                        + this + " for pattern(s) " + getTraceInfoPatternsEnumerated(), null);
+                                }
+                                try {
+                                        final Accumulator newRederivableAccumulator = operator.update(oldRederivableAccumulator,
+                                                        aggregableValue, isInsertion);
+                                        storeIfNotNeutral(group, newRederivableAccumulator, rederivableMemory);
+                                        if (rederivableMemory.isEmpty()) {
+                                                // there is nothing left to be re-derived
+                                                // this can happen if the accumulator became neutral in response to the DELETE
+                                                ((RecursiveCommunicationGroup) currentGroup).removeRederivable(this);
+                                        }
+                                } catch (final NullPointerException ex) {
+                                        issueError("[INTERNAL ERROR] Deleting a domain element in " + update
+                                                        + " which did not exist before in ColumnAggregatorNode " + this + " for pattern(s) "
+                                                        + getTraceInfoPatternsEnumerated(), ex);
+                                }
+                        } else {
+                                // the group is in the main memory
+                                // at this point, it can happen that we need to initialize with a neutral accumulator
+                                if (oldMainAccumulator == null) {
+                                        oldMainAccumulator = operator.createNeutral();
+                                }
+
+                                final AggregateResult oldValue = operator.getAggregate(oldMainAccumulator);
+                                final Accumulator newMainAccumulator = operator.update(oldMainAccumulator, aggregableValue,
+                                                isInsertion);
+                                final AggregateResult newValue = operator.getAggregate(newMainAccumulator);
+
+                                if (monotone) {
+                                        storeIfNotNeutral(group, newMainAccumulator, memory);
+                                        propagateAggregateResultUpdate(group, oldValue, newValue, Timestamp.ZERO);
+                                } else {
+                                        final boolean wasEmpty = rederivableMemory.isEmpty();
+                                        if (storeIfNotNeutral(group, newMainAccumulator, rederivableMemory) && wasEmpty) {
+                                                ((RecursiveCommunicationGroup) currentGroup).addRederivable(this);
+                                        }
+                                        memory.remove(group);
+                                        propagateAggregateResultUpdate(group, oldValue, NEUTRAL, Timestamp.ZERO);
+                                }
+                        }
+                }
+        }
+
+        @Override
+        public void batchUpdate(Collection<Entry<Tuple, Integer>> updates, Timestamp timestamp) {
+                if (!Timestamp.ZERO.equals(timestamp)) {
+                        throw new IllegalArgumentException("Timely operation is not supported");
+                }
+                if (deleteRederiveEvaluation || posetComparator != null) {
+                        super.batchUpdate(updates, timestamp);
+                        return;
+                }
+                propagateBatchUpdate(updates, timestamp);
+        }
+
+        private void propagateBatchUpdate(Collection<Entry<Tuple, Integer>> updates, Timestamp timestamp) {
+                if (updates.isEmpty()) {
+                        return;
+                }
+                var oldValues = CollectionsFactory.<Tuple, AggregateResult>createMap();
+                for (var entry : updates) {
+                        var update = entry.getKey();
+                        var key = groupMask.transform(update);
+                        var value = columnMask.transform(update);
+                        @SuppressWarnings("unchecked")
+                        var valueToAggregate = (Domain) runtimeContext.unwrapElement(value.get(0));
+                        int count = entry.getValue();
+                        boolean isInsertion = true;
+                        if (count < 0) {
+                                isInsertion = false;
+                                count = -count;
+                        }
+
+                        var oldMainAccumulator = memory.get(key);
+                        oldValues.computeIfAbsent(key, ignoredKey ->
+                                        oldMainAccumulator == null ? NEUTRAL : operator.getAggregate(oldMainAccumulator));
+                        Accumulator newMainAccumulator = oldMainAccumulator == null ? operator.createNeutral() :
+                                        oldMainAccumulator;
+                        for (int i = 0; i < count; i++) {
+                                newMainAccumulator = operator.update(newMainAccumulator, valueToAggregate, isInsertion);
+                        }
+                        storeIfNotNeutral(key, newMainAccumulator, memory);
+                }
+                for (var entry : oldValues.entrySet()) {
+                        var key = entry.getKey();
+                        var oldValue = entry.getValue();
+                        var newMainAccumulator = getMainAccumulator(key);
+                        var newValue = operator.getAggregate(newMainAccumulator);
+                        propagateAggregateResultUpdate(key, oldValue, newValue, timestamp);
+                }
+        }
+
+        @Override
+        public void clear() {
+                this.memory.clear();
+                this.rederivableMemory.clear();
+                this.childMailboxes.clear();
+        }
+
+        /**
+         * Returns true if the accumulator was stored, false otherwise.
+         *
+         * @since 1.6
+         */
+        protected boolean storeIfNotNeutral(final Tuple key, final Accumulator accumulator,
+                                                                                final Map<Tuple, Accumulator> memory) {
+                if (operator.isNeutral(accumulator)) {
+                        memory.remove(key);
+                        return false;
+                } else {
+                        memory.put(key, accumulator);
+                        return true;
+                }
+        }
+
+        @Override
+        public Tuple getAggregateTuple(final Tuple group) {
+                final Accumulator accumulator = getMainAccumulator(group);
+                final AggregateResult result = operator.getAggregate(accumulator);
+                return tupleFromAggregateResult(group, result);
+        }
+
+        @Override
+        public AggregateResult getAggregateResult(final Tuple group) {
+                final Accumulator accumulator = getMainAccumulator(group);
+                return operator.getAggregate(accumulator);
+        }
+
+        @Override
+        public Map<AggregateResult, Timeline<Timestamp>> getAggregateResultTimeline(Tuple key) {
+                throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Map<Tuple, Timeline<Timestamp>> getAggregateTupleTimeline(Tuple key) {
+                throw new UnsupportedOperationException();
+        }
+
+        /**
+         * @since 1.6
+         */
+        protected Accumulator getMainAccumulator(final Tuple key) {
+                return getAccumulator(key, memory);
+        }
+
+        /**
+         * @since 1.6
+         */
+        protected Accumulator getRederivableAccumulator(final Tuple key) {
+                return getAccumulator(key, rederivableMemory);
+        }
+
+        /**
+         * @since 1.6
+         */
+        protected Accumulator getAccumulator(final Tuple key, final Map<Tuple, Accumulator> memory) {
+                Accumulator accumulator = memory.get(key);
+                if (accumulator == null) {
+                        return operator.createNeutral();
+                } else {
+                        return accumulator;
+                }
+        }
+
+        @Override
+        public CommunicationGroup getCurrentGroup() {
+                return currentGroup;
+        }
+
+        @Override
+        public void setCurrentGroup(final CommunicationGroup currentGroup) {
+                this.currentGroup = currentGroup;
+        }
 
 }
diff --git a/subprojects/store-query-interpreter/src/test/java/tools/refinery/store/query/interpreter/AggregatorBatchingTest.java b/subprojects/store-query-interpreter/src/test/java/tools/refinery/store/query/interpreter/AggregatorBatchingTest.java
new file mode 100644
index 00000000..d8e06d82
--- /dev/null
+++ b/subprojects/store-query-interpreter/src/test/java/tools/refinery/store/query/interpreter/AggregatorBatchingTest.java
@@ -0,0 +1,186 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The Refinery Authors <https://refinery.tools/>
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package tools.refinery.store.query.interpreter;
+
+import org.junit.jupiter.api.Test;
+import tools.refinery.store.model.Model;
+import tools.refinery.store.model.ModelStore;
+import tools.refinery.store.query.ModelQueryAdapter;
+import tools.refinery.store.query.dnf.Query;
+import tools.refinery.store.query.term.StatefulAggregate;
+import tools.refinery.store.query.term.StatefulAggregator;
+import tools.refinery.store.query.term.Variable;
+import tools.refinery.store.query.view.AnySymbolView;
+import tools.refinery.store.query.view.FunctionView;
+import tools.refinery.store.query.view.KeyOnlyView;
+import tools.refinery.store.representation.Symbol;
+import tools.refinery.store.tuple.Tuple;
+
+import java.util.Map;
+import java.util.Optional;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static tools.refinery.store.query.interpreter.tests.QueryAssertions.assertNullableResults;
+
+class AggregatorBatchingTest {
+        private static final Symbol<Boolean> person = Symbol.of("Person", 1);
+        private static final Symbol<Integer> values = Symbol.of("values", 2, Integer.class, null);
+        private static final AnySymbolView personView = new KeyOnlyView<>(person);
+        private static final FunctionView<Integer> valuesView = new FunctionView<>(values);
+
+        private final Query<Integer> query = Query.of(Integer.class, (builder, p1, output) -> builder
+                        .clause(
+                                        personView.call(p1),
+                                        output.assign(valuesView.aggregate(new InstrumentedAggregator(), p1, Variable.of()))
+                        ));
+
+        private int extractCount = 0;
+
+        @Test
+        void batchTest() {
+                var model = createModel();
+                var personInterpretation = model.getInterpretation(person);
+                var valuesInterpretation = model.getInterpretation(values);
+                var queryEngine = model.getAdapter(ModelQueryAdapter.class);
+                var resultSet = queryEngine.getResultSet(query);
+
+                assertThat(extractCount, is(1));
+
+                personInterpretation.put(Tuple.of(0), true);
+                personInterpretation.put(Tuple.of(1), true);
+
+                valuesInterpretation.put(Tuple.of(0, 0), 1);
+                valuesInterpretation.put(Tuple.of(0, 1), 2);
+                valuesInterpretation.put(Tuple.of(0, 2), 3);
+                valuesInterpretation.put(Tuple.of(1, 0), 1);
+                valuesInterpretation.put(Tuple.of(1, 1), -1);
+
+                queryEngine.flushChanges();
+
+                assertThat(extractCount, is(5));
+
+                assertNullableResults(Map.of(
+                                Tuple.of(0), Optional.of(6),
+                                Tuple.of(1), Optional.of(0),
+                                Tuple.of(2), Optional.empty()
+                ), resultSet);
+        }
+
+        @Test
+        void separateTest() {
+                var model = createModel();
+                var personInterpretation = model.getInterpretation(person);
+                var valuesInterpretation = model.getInterpretation(values);
+                var queryEngine = model.getAdapter(ModelQueryAdapter.class);
+                var resultSet = queryEngine.getResultSet(query);
+
+                assertThat(extractCount, is(1));
+
+                personInterpretation.put(Tuple.of(0), true);
+                personInterpretation.put(Tuple.of(1), true);
+
+                queryEngine.flushChanges();
+                assertThat(extractCount, is(3));
+
+                valuesInterpretation.put(Tuple.of(0, 0), 1);
+                valuesInterpretation.put(Tuple.of(1, 0), 1);
+
+                queryEngine.flushChanges();
+                assertThat(extractCount, is(5));
+                assertNullableResults(Map.of(
+                                Tuple.of(0), Optional.of(1),
+                                Tuple.of(1), Optional.of(1),
+                                Tuple.of(2), Optional.empty()
+                ), resultSet);
+
+                valuesInterpretation.put(Tuple.of(0, 1), 2);
+                valuesInterpretation.put(Tuple.of(1, 1), -1);
+
+                queryEngine.flushChanges();
+                assertThat(extractCount, is(9));
+                assertNullableResults(Map.of(
+                                Tuple.of(0), Optional.of(3),
+                                Tuple.of(1), Optional.of(0),
+                                Tuple.of(2), Optional.empty()
+                ), resultSet);
+
+                valuesInterpretation.put(Tuple.of(0, 2), 3);
+
+                queryEngine.flushChanges();
+                assertThat(extractCount, is(11));
+                assertNullableResults(Map.of(
+                                Tuple.of(0), Optional.of(6),
+                                Tuple.of(1), Optional.of(0),
+                                Tuple.of(2), Optional.empty()
+                ), resultSet);
+        }
+
+        private Model createModel() {
+                var store = ModelStore.builder()
+                                .symbols(person, values)
+                                .with(QueryInterpreterAdapter.builder()
+                                                .query(query))
+                                .build();
+                return store.createEmptyModel();
+        }
+
+        class InstrumentedAggregator implements StatefulAggregator<Integer, Integer> {
+                @Override
+                public Class<Integer> getResultType() {
+                        return Integer.class;
+                }
+
+                @Override
+                public Class<Integer> getInputType() {
+                        return Integer.class;
+                }
+
+                @Override
+                public StatefulAggregate<Integer, Integer> createEmptyAggregate() {
+                        return new InstrumentedAggregate();
+                }
+        }
+
+        class InstrumentedAggregate implements StatefulAggregate<Integer, Integer> {
+                private int sum;
+
+                public InstrumentedAggregate() {
+                        this(0);
+                }
+
+                private InstrumentedAggregate(int sum) {
+                        this.sum = sum;
+                }
+
+
+                @Override
+                public void add(Integer value) {
+                        sum += value;
+                }
+
+                @Override
+                public void remove(Integer value) {
+                        sum -= value;
+                }
+
+                @Override
+                public Integer getResult() {
+                        extractCount++;
+                        return sum;
+                }
+
+                @Override
+                public boolean isEmpty() {
+                        return sum == 0;
+                }
+
+                @Override
+                public StatefulAggregate<Integer, Integer> deepCopy() {
+                        return new InstrumentedAggregate(sum);
+                }
+        }
+}