private void purgePartialLoads() {

        final Instant startTime = Instant.now();

        final Predicate<ByteBuffer> isPurgeablePredicate =
                RefDataProcessingInfoSerde.createProcessingStatePredicate(
                        ProcessingState.LOAD_IN_PROGRESS,
                        ProcessingState.PURGE_IN_PROGRESS,
                        ProcessingState.TERMINATED,
                        ProcessingState.READY_FOR_PURGE);

        // Get all the ref stream defs in a partially loaded/purged state
        // It is only in exceptional circumstances that we get streams in this state so the list should
        // not be big.
        final List<Entry<RefStreamDefinition, RefDataProcessingInfo>> purgeableRefStreamDefs =
                lmdbEnvironment.getWithReadTxn(readTxn -> getInvalidStreams(readTxn, isPurgeablePredicate));

        if (!purgeableRefStreamDefs.isEmpty()) {
            AtomicReference<PurgeCounts> purgeCountsRef = new AtomicReference<>(PurgeCounts.ZERO);
            for (final Entry<RefStreamDefinition, RefDataProcessingInfo> entry : purgeableRefStreamDefs) {
                try (final PooledByteBuffer refStreamDefPooledBuf = processingInfoDb.getPooledKeyBuffer()) {
                    final RefStreamDefinition refStreamDefinition = entry.getKey();
                    final RefDataProcessingInfo refDataProcessingInfo = entry.getValue();
                    final Supplier<String> msgSupplier = () -> LogUtil.message(
                            "Purging partially loaded/purged reference stream {}:{} with state {} in store '{}'",
                            refStreamDefinition.getStreamId(),
                            refStreamDefinition.getPartNumber(),
                            refDataProcessingInfo.getProcessingState(),
                            storeName);
                    taskContextFactory.current().info(msgSupplier);
                    LOGGER.info(msgSupplier);
                    processingInfoDb.serializeKey(refStreamDefPooledBuf.getByteBuffer(), refStreamDefinition);

                    // Need to recheck the state under lock in case it has changed
                    final RefStreamPurgeCounts refStreamPurgeCounts = purgeRefStreamIfEligible(
                            isPurgeablePredicate,
                            refStreamDefPooledBuf.getByteBuffer(),
                            refStreamDefinition);

                    purgeCountsRef.set(purgeCountsRef.get().increment(refStreamPurgeCounts));
                }
                if (Thread.currentThread().isInterrupted()) {
                    LOGGER.debug("Thread interrupted during ");
                }
            }

            LOGGER.info(() -> LogUtil.message(
                    "Purge of partial loads/purges {} in store {} - {}",
                    (Thread.currentThread().isInterrupted()
                            ? "interrupted"
                            : "completed successfully"),
                    buildPurgeInfoString(startTime, purgeCountsRef.get()), storeName));
        } else {
            LOGGER.info("Completed check for invalid ref loads/purges in store {} in '{}'",
                    storeName,
                    Duration.between(startTime, Instant.now()));
        }
    }
An Off Heap implementation of {@link RefDataStore} using LMDB.
<p>
Essentially each ref stream (a {@link RefStreamDefinition}) contains 1-* map names.
Multiple ref streams can contain the same map name.
Within a ref stream + map combo (a {@link MapDefinition}) there are 1-* entries.
A lookup is done by creating a key that is the lookup key combined with a {@link MapDefinition}
so only an entry matching the lookup key, the map name and the ref stream will be returned.
stroom-pipeline/src/main/java/stroom/pipeline/refdata/store/offheapstore/RefDataOffHeapStore.java