void aggregate(final Instant timeNow) {
        final TaskContext taskContext = taskContextFactory.current();
        guard.lock();
        try {
            LOGGER.info("Starting statistics aggregation " +
                            "(stage1BatchSize: {}, stage2BatchSize: {}, timeNow: {})",
                    ModelStringUtil.formatCsv(stage1BatchSize),
                    ModelStringUtil.formatCsv(stage2BatchSize),
                    timeNow);

            long totalStage1Count = 0;
            long totalStage2Count = 0;
            long oldStatsDeletedCount = 0;
            int unusedKeysDeletedCount = 0;
            final LogExecutionTime logExecutionTime = new LogExecutionTime();
            try {
                oldStatsDeletedCount = helper.deleteOldStats(timeNow, taskContext);

                long processedCount;
                int batchNo = 0;
                // Process batches of records until we have processed one
                // that was not a full batch
                do {
                    processedCount = helper.aggregateConfigStage1(
                            taskContext,
                            "Aggregation Stage 1 - batch no: " + ++batchNo + "",
                            stage1BatchSize,
                            timeNow);
                    totalStage1Count += processedCount;
                } while (processedCount > 0 && !Thread.currentThread().isInterrupted());

                LOGGER.info("Completed stage 1 aggregation with {} iterations in {}{}",
                        batchNo,
                        logExecutionTime.getDuration(),
                        (Thread.currentThread().isInterrupted()
                                ? " (INTERRUPTED)"
                                : ""));

                if (!Thread.currentThread().isInterrupted()) {
                    totalStage2Count = helper.aggregateConfigStage2(
                            taskContext,
                            "Aggregation Stage 2",
                            stage2BatchSize,
                            timeNow);
                }

                // We hold the cluster lock so can safely deleted orphaned stat keys
                unusedKeysDeletedCount = helper.deleteUnusedKeys(taskContext);

            } catch (final SQLException ex) {
                throw new RuntimeException(ex.getMessage(), ex);
            } finally {
                LOGGER.info("Finished SQL stats aggregation in {}, " +
                                "oldStatsDeletedCount: {}, " +
                                "totalStage1Count: {}, " +
                                "totalStage2Count: {}, " +
                                "unusedKeysDeletedCount: {} " +
                                "(stage1BatchSize: {}, stage2BatchSize: {}, timeNowOverride: {})",
                        logExecutionTime.getDuration(),
                        ModelStringUtil.formatCsv(oldStatsDeletedCount),
                        ModelStringUtil.formatCsv(totalStage1Count),
                        ModelStringUtil.formatCsv(totalStage2Count),
                        ModelStringUtil.formatCsv(unusedKeysDeletedCount),
                        ModelStringUtil.formatCsv(stage1BatchSize),
                        ModelStringUtil.formatCsv(stage2BatchSize),
                        timeNow);
            }
        } finally {
            guard.unlock();
        }
    }
Step 1 - Move source values into value table with precision 1<br/>
Step 2 - Reduce precisions possibly creating duplicates in same table
<br/>
Step 3 - Remove duplicates using temporary table<br/>
stroom-statistics/stroom-statistics-impl-sql/src/main/java/stroom/statistics/impl/sql/SQLStatisticAggregationManager.java