32 Using Continuous Aggregation

Maintain partition-local aggregation state as cache entries change, then combine that state when an application requests a result.

Continuous Aggregation is available in Coherence 26.1.0.0.0 and later. Use it for frequently repeated aggregation definitions when maintaining their state costs less than rescanning selected entries for every query. The final result is obtained on demand through a map-bound handle.

Choose ordinary aggregation for occasional or changing queries. Choose a Continuous Query Cache (CQC) when the application needs a live filtered view of entries. Continuous Aggregation maintains aggregate state on storage members; it does not push result events or retain the final result on the client, and it does not strengthen the ordinary partitioned-query consistency contract.

• Continuous Aggregation Requirements and Compatibility
  Use compatible storage members, clients, and proxies before registering or querying a maintained aggregation.
• Registering and Querying a Continuous Aggregation
  Register a supported streaming aggregator through the map, query its handle, and coordinate removal of the shared definition.
• Filtering and Indexing a Continuous Aggregation
  Use a stable filter to select contributing entries and indexes to reduce initial-build and fallback costs.
• Choosing a Continuously Maintainable Aggregator
  Use an aggregator that advertises continuous maintenance and account for the size of its retained state.
• Understanding Exact Continuous Aggregation Fallback
  Queries combine ready partials and exact partition scans while maintenance is building or cannot update a result incrementally.
• Projecting a Continuous Aggregation by Data Affinity
  Route a projection to an associated partition when the application needs a result for colocated data.
• Managing Continuous Aggregation Lifecycle and Persistence
  Treat an aggregation definition as shared cache metadata rather than a client-owned subscription.
• Limiting and Monitoring Continuous Aggregations
  Bound registrations and observe maintenance work on each cache’s storage members.
• Implementing a Custom Continuous Aggregator
  Advertise continuous maintenance only when every mutation and partial-result operation preserves the ordinary aggregation result.

Continuous Aggregation Requirements and Compatibility

Use compatible storage members, clients, and proxies before registering or querying a maintained aggregation.

Every storage-enabled member of the cache service must support Continuous Aggregation. During a rolling upgrade or downgrade with older storage members, registration, removal and handle queries fail with UnsupportedOperationException. Ordinary map operations and on-demand aggregations remain available.

The Java API works for in-cluster, Extend, and gRPC clients. Extend requires NamedCache protocol version 13; gRPC requires NamedCache protocol version 2. Both the initiating remote client and its proxy must support the feature. Upgrade clients and proxies as well as storage members before enabling the application path.

Built-in filters, extractors, and aggregators supply the framework metadata for their own code. Custom implementations, projection functions, and captured application data must satisfy application-type and remote-execution rules. See Classifying Remote Executable Objects and Authorizing Access to Server-Side Operations in the Security Guide.

Registering and Querying a Continuous Aggregation

Register a supported streaming aggregator through the map, query its handle, and coordinate removal of the shared definition.

The following complete Java 21 or later application uses only JDK values and Coherence built-ins. Start with an empty amounts cache in an isolated example cluster, using the first-application build and run procedure. No custom model or executable class needs registration for this example.

import com.tangosol.net.Coherence;
import com.tangosol.net.ContinuousAggregator;
import com.tangosol.net.NamedMap;
import com.tangosol.util.Aggregators;
import com.tangosol.util.ValueExtractor;

public class ContinuousTotals {
    public static void main(String[] args) {
        try (var coherence = Coherence.clusterMember().start().join()) {
            NamedMap<String, Long> amounts = coherence.getSession().getMap("amounts");
            ContinuousAggregator<String, Long, Long> total =
                    amounts.addAggregator(Aggregators.longSum(ValueExtractor.identity()));
            try {
                amounts.put("order-1", 120L);
                amounts.put("order-2", 30L);
                System.out.println(total.aggregate()); // 150 with only these entries
                amounts.put("order-1", 100L);
                System.out.println(total.aggregate()); // 130 with only these entries
            } finally {
                amounts.removeAggregator(total);
            }
        }
    }
}

The addAggregator overloads are defined on InvocableMap and available through NamedMap and NamedCache. Each call to aggregate() is a network operation that obtains and combines partition partials. Await each write before querying when the result must include that write.

The example owns this registration in its isolated cache, so it removes the definition in finally. In a shared application, coordinate registration and removal centrally. Closing a client or losing a handle does not remove a shared registration.

Filtering and Indexing a Continuous Aggregation

Use a stable filter to select contributing entries and indexes to reduce initial-build and fallback costs.

The following method fragment uses a started application session. It maintains the sum of positive values. Inserts, removals, and replacements update the result, including changes into or out of the filter.

NamedMap<String, Long> amounts = session.getMap("amounts");
ValueExtractor<Long, Long> value = ValueExtractor.identity();
Filter<Long> positive = Filters.greater(value, 0L);
amounts.addIndex(value);
ContinuousAggregator<String, Long, Long> positiveTotal =
        amounts.addAggregator(positive, Aggregators.longSum(value));
long result = positiveTotal.aggregate();

Import ValueExtractor, Filter, Filters and Aggregators from com.tangosol.util. Add the indexes justified by the actual filter and aggregation workload before registering. Indexes can help both on-demand aggregation and Continuous Aggregation; compare against an indexed on-demand baseline, including index mutation and memory costs.

A NearCache delegates registration to its backing cache. A non-transforming CQC combines its view filter with the registration filter. A transforming CQC rejects Continuous Aggregation because maintenance sees the backing value type. Register a suitable definition on the backing cache instead.

Choosing a Continuously Maintainable Aggregator

Use an aggregator that advertises continuous maintenance and account for the size of its retained state.

Table 32-1 Built-in continuous aggregators

Family	Supported implementations and considerations
Additive state	Count, LongSum, DoubleSum, BigDecimalSum, DoubleAverage and BigDecimalAverage.
Extrema	Long, double, BigDecimal, and comparable minimum/maximum aggregators. Removal of the last extreme can require exact fallback.
Result-sensitive state	DistinctValues, ReducerAggregator and TopNAggregator; retained state can grow with value or entry cardinality.
Composition	GroupAggregator when its child supports maintenance; CompositeAggregator when all children support maintenance. Group count and nested results affect heap use.

An aggregator must advertise the CONTINUOUS characteristic. Unsupported definitions fail registration with IllegalArgumentException. PriorityAggregator deliberately does not advertise this characteristic: its per-execution scheduling and timeout semantics do not define a long-lived maintenance policy.

Estimate mutation work and storage-member heap as well as query latency. A reducer returning one value per entry can require data-sized state and a large result transfer; maintaining it does not make that result small.

Understanding Exact Continuous Aggregation Fallback

Queries combine ready partials and exact partition scans while maintenance is building or cannot update a result incrementally.

Registration schedules asynchronous initial builds. A returned handle does not mean every partition is ready. A query scans each required partition whose maintained state is not exact and combines those results with exact maintained partials. The result follows the normal aggregation consistency contract; a partition scan fallback is not an approximate result.

Non-invertible state, such as a minimum whose last occurrence was removed, can retain a certified historical bound. Coherence scans that partition only when another exact partial cannot prove that the stale bound is irrelevant. Initial build, ownership movement, backup promotion, and recovery can also require rebuilding.

This mechanism preserves correctness without requiring all queries to wait for background rebuilds. It does not provide a transactionally consistent snapshot across concurrent writes to multiple partitions. Measure fallback frequency and latency when evaluating the benefit of maintenance.

Projecting a Continuous Aggregation by Data Affinity

Route a projection to an associated partition when the application needs a result for colocated data.

A KeyAssociatedFilter restricts a registration to the partition set associated with its host key. The host key and outer filter are part of the registration identity. Its aggregate() method combines all applicable partitions, including broad association strategies.

A cache-wide registration can also be projected with ContinuousAggregator.invoke or key-based invokeAll. Each supplied key identifies one partition. The anchor entry does not have to exist and is neither read nor changed. Coherence finalizes a detached exact result for that partition, then runs the supplied remote function on its owner.

// total is a ContinuousAggregator<String, Long, Long> for the amounts map.
// This is the sum for the partition containing the anchor, not the whole cache.
Long partitionTotal = total.invoke("customer-42-anchor", value -> value);

A grouped result is useful here when every entry in the required group is colocated using key association. Partitioning alone does not filter unrelated entries in that partition: apply the group lookup or selection explicitly. A projection key whose association spans several partitions is rejected with IllegalArgumentException.

Inside an entry processor, BinaryEntry.getContinuousAggregationResult and BackingMapContext.getContinuousAggregationResult provide the same detached exact partition result. The complete filter/aggregator pair must match an active registration. Projection functions use entry-processor authorization and require the corresponding executable metadata.

Managing Continuous Aggregation Lifecycle and Persistence

Treat an aggregation definition as shared cache metadata rather than a client-owned subscription.

• Equal filter and aggregator definitions register idempotently. There is no user-defined name or client reference count.
• Removing any handle for a definition removes that shared definition and its state. Further use of a removed handle throws IllegalStateException.
• clear() and truncate() retain definitions and recreate empty partition results. destroy() removes the cache and definitions.
• Persistent services retain registration definitions as cache metadata. Exact state can be checkpointed with a quiescent partition snapshot. State is restored only when definition, partition, ownership, format, and data identities match; otherwise it is rebuilt.
• Planned primary transfer can carry checkpointable state. Backup owners do not maintain another live aggregation copy on every mutation. Promotion rebuilds from backup data, with exact fallback available while builds complete.

Custom definitions need stable equality and hashing based on serialized definition state. Mutable running totals must not participate in definition identity. Assign application ownership for shared registration changes, especially during deployment overlap and rolling upgrades.

Limiting and Monitoring Continuous Aggregations

Bound registrations and observe maintenance work on each cache’s storage members.

The default limit is 256 registrations per cache. Configure a positive deployment-wide value on every storage member before starting the service:

-Dcoherence.distributed.continuous.max-registrations=128

An equal definition reuses the existing registration. A different filter or aggregator uses another slot. Exceeding the limit throws IllegalStateException. The limit bounds definition count, not total heap or result cardinality.

Table 32-2 StorageManager MBean attributes for Continuous Aggregation

Attribute	Meaning
ContinuousAggregationRegistrationCount	Definitions registered for the cache; aggregated member views use a set collector.
ContinuousAggregationReadyPartitionCount	Partition states with exact maintained results.
ContinuousAggregationBuildingPartitionCount	Partition states currently building.
ContinuousAggregationStalePartitionCount	Inexact partition states that may need fallback.
ContinuousAggregationMutationCount	Selected entry transitions applied since statistics reset.
ContinuousAggregationFallbackCount	Partition scans used to answer maintained queries since reset.
ContinuousAggregationRebuildCount	Completed partition-state builds since reset.
ContinuousAggregationDirtyCount	Exact-to-inexact state transitions since reset.

Partition gauges and work counters sum across storage member views. Building and stale counts should normally fall after startup or transfer. Compare mutation latency, retained heap, query latency, and fallback count against the same indexed workload without maintenance. Remove definitions that no longer justify their cost.

Implementing a Custom Continuous Aggregator

Advertise continuous maintenance only when every mutation and partial-result operation preserves the ordinary aggregation result.

Existing custom StreamingAggregator implementations remain valid for on-demand aggregation. To opt in, implement the CONTINUOUS contract:

• Return a new empty instance from supply(). Do not retain live entries or advertise RETAINS_ENTRIES.
• Return a non-destructive, serializer-compatible partial result from getPartialResult(); repeated reads must not consume the state.
• Retract the old contribution in retract(). Return REBUILD_REQUIRED when an exact incremental answer cannot be maintained.
• Implement replacement in update(), or use its default retract-then-accumulate behavior.
• Keep equality, hashing and serialized definition fields independent of mutable maintenance state.
• For transfer/persistence of exact state, advertise STATE_CHECKPOINTABLE and implement matching snapshotState() and restoreState(). Without this characteristic, state is rebuilt after movement or recovery.

Test insert, replacement, removal, filter transitions, empty partitions, combine, repeated partial reads, and recovery. A correct rebuild request is preferable to an incorrect claimed incremental answer. Classify and deploy the implementation on the members that deserialize and run it before registration.