public void restoreRevision(String revision) throws CannotRestoreSiddhiAppStateException {
        PersistenceStore persistenceStore = siddhiAppContext.getSiddhiContext().getPersistenceStore();
        IncrementalPersistenceStore incrementalPersistenceStore =
                siddhiAppContext.getSiddhiContext().getIncrementalPersistenceStore();
        String siddhiAppName = siddhiAppContext.getName();
        if (persistenceStore != null) {
            if (log.isDebugEnabled()) {
                log.debug("Restoring revision: " + revision + " ...");
            }
            byte[] snapshot = persistenceStore.load(siddhiAppContext.getName(), revision);
            if (snapshot != null) {
                restore(snapshot);
                if (log.isDebugEnabled()) {
                    log.debug("Restored revision: " + revision);
                }
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("No data found for revision: " + revision);
                }
                throw new PersistenceStoreException("No data found for revision: " + revision);
            }
        } else if (incrementalPersistenceStore != null) {
            if (log.isDebugEnabled()) {
                log.debug("Restoring revision: " + revision + " ...");
            }
            IncrementalSnapshotInfo restoreSnapshotInfo = PersistenceHelper.convertRevision(revision);
            List<IncrementalSnapshotInfo> incrementalSnapshotInfos =
                    incrementalPersistenceStore.getListOfRevisionsToLoad(
                            restoreSnapshotInfo.getTime(), restoreSnapshotInfo.getSiddhiAppId());
            if (incrementalSnapshotInfos != null) {
                incrementalSnapshotInfos.sort(new Comparator<IncrementalSnapshotInfo>() {
                    @Override
                    public int compare(IncrementalSnapshotInfo o1, IncrementalSnapshotInfo o2) {
                        int results = o1.getId().compareTo(o2.getId());
                        if (results == 0) {
                            results = Long.compare(o2.getTime(), o1.getTime());
                            if (results == 0) {
                                return o2.getType().compareTo(o1.getType());
                            }
                        }
                        return results;
                    }
                });
                String lastId = null;
                boolean baseFound = false;
                boolean perioicFound = false;
                for (Iterator<IncrementalSnapshotInfo> iterator = incrementalSnapshotInfos.iterator();
                     iterator.hasNext(); ) {
                    IncrementalSnapshotInfo snapshotInfo = iterator.next();
                    if (snapshotInfo.getId().equals(lastId)) {
                        if (baseFound && (snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.BASE
                                || snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.INCREMENT)) {
                            iterator.remove();
                        } else if (perioicFound &&
                                snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.PERIODIC) {
                            iterator.remove();
                        } else if (snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.BASE) {
                            baseFound = true;
                        } else if (snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.PERIODIC) {
                            perioicFound = true;
                        }
                    } else {
                        baseFound = snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.BASE;
                        perioicFound = snapshotInfo.getType() == IncrementalSnapshotInfo.SnapshotType.PERIODIC;
                    }
                    lastId = snapshotInfo.getId();
                }
                Map<String, Map<String, Map<String, Map<Long, Map<IncrementalSnapshotInfo, byte[]>>>>>
                        incrementalState = new HashMap<>();
                for (IncrementalSnapshotInfo snapshotInfo : incrementalSnapshotInfos) {
                    Map<String, Map<String, Map<Long, Map<IncrementalSnapshotInfo, byte[]>>>>
                            incrementalStateByPartitionGroupByKey = incrementalState.computeIfAbsent(
                            snapshotInfo.getPartitionId(), k -> new TreeMap<>());
                    Map<String, Map<Long, Map<IncrementalSnapshotInfo, byte[]>>> incrementalStateByTime =
                            incrementalStateByPartitionGroupByKey.computeIfAbsent(snapshotInfo.getPartitionGroupByKey(),
                                    k -> new TreeMap<>());
                    Map<Long, Map<IncrementalSnapshotInfo, byte[]>> idByTime =
                            incrementalStateByTime.computeIfAbsent(snapshotInfo.getId(),
                                    k -> new TreeMap<>());
                    Map<IncrementalSnapshotInfo, byte[]> incrementalStateByInfo = idByTime.
                            computeIfAbsent(snapshotInfo.getTime(), k -> new HashMap<>());
                    incrementalStateByInfo.put(snapshotInfo, incrementalPersistenceStore.load(snapshotInfo));
                }
                restore(incrementalState);
                if (log.isDebugEnabled()) {
                    log.debug("Restored revision: " + revision);
                }
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("No data found for revision: " + revision);
                }
                throw new PersistenceStoreException("No data found for revision: " + revision);
            }
        } else {
            throw new NoPersistenceStoreException("No persistence store assigned for siddhi app " +
                    siddhiAppName);
        }
    }
Service level implementation to take/restore snapshots of processing elements.
Memory   : PartitionId + QueryName + ElementId + PartitionGroupByKey + Item
Snapshot : PartitionId + PartitionGroupByKey + QueryName + ElementId + Item
modules/siddhi-core/src/main/java/io/siddhi/core/util/snapshot/SnapshotService.java