@Override
    public int logicalDelete(final List<DataRetentionRuleAction> ruleActions,
                             final TimePeriod period) {

        LOGGER.debug(() ->
                LogUtil.message("logicalDelete called for {} and actions\n{}",
                        period, ruleActions.stream()
                                .map(ruleAction -> ruleAction.getRule().getRuleNumber() + " " +
                                                   ruleAction.getRule().getExpression() + " " +
                                                   ruleAction.getOutcome().name())
                                .collect(Collectors.joining("\n"))));

        final AtomicInteger totalUpdateCount = new AtomicInteger(0);
        if (ruleActions != null && !ruleActions.isEmpty()) {
            final DataRetentionConfig dataRetentionConfig = dataRetentionConfigProvider.get();
            final byte statusIdDeleted = MetaStatusId.getPrimitiveValue(Status.DELETED);

            final List<Condition> baseConditions = createRetentionDeleteConditions(ruleActions);
            final boolean rulesUsePipelineField = ruleActionsContainField(MetaFields.PIPELINE.getFldName(),
                    ruleActions)
                                                  || ruleActionsContainField(MetaFields.PIPELINE_NAME.getFldName(),
                    ruleActions);

            final List<Condition> conditions = new ArrayList<>(baseConditions);

            // Time bound the data we are testing the rules over
            conditions.add(meta.CREATE_TIME.greaterOrEqual(period.getFrom().toEpochMilli()));
            conditions.add(meta.CREATE_TIME.lessThan(period.getTo().toEpochMilli()));

            int lastUpdateCount;
            final AtomicInteger iteration = new AtomicInteger(1);
            Duration totalSelectDuration = Duration.ZERO;
            Duration totalUpdateDuration = Duration.ZERO;
            Instant startTime;
            Instant startTimeInc = period.getFrom();
            final int batchSize = dataRetentionConfig.getDeleteBatchSize();

            // The aim here is to ensure we lock up the meta table for as short a time
            // as possible. As we are updating by non-unique indexes it is likely we will
            // get next-key locks which may impact other writes.
            // At the cost of the whole retention delete taking longer we split the updates
            // up into smaller chunks to reduce the time meta is locked for. To do the chunking
            // we query the table with the same conditions as the update to find a batch
            // of n records and capture the min/max create_time for that batch.
            // Then we update over that sub-period so we can just rang-scan the create_time or
            // feed_id|create_time index.
            do {
                // Get a sub period of period

                startTime = Instant.now();
                final Optional<TimePeriod> optSubPeriod = getTimeSlice(
                        startTimeInc, batchSize, conditions, rulesUsePipelineField);

                totalSelectDuration = totalSelectDuration.plus(Duration.between(startTime,
                        Instant.now().plusMillis(1)));

                if (optSubPeriod.isEmpty()) {
                    LOGGER.debug("No time slice found");
                    break;
                }
                final TimePeriod subPeriod = optSubPeriod.get();
                LOGGER.debug("Iteration {}, using sub-period {}", iteration, subPeriod);

                startTime = Instant.now();
                lastUpdateCount = LOGGER.logDurationIfDebugEnabled(
                        () -> JooqUtil.contextResult(metaDbConnProvider, context -> {

                            // If any of the rules have a predicate on the Pipeline field then we need to
                            // add the join to meta_processor
                            final var tableClause = rulesUsePipelineField
                                    ? meta.leftOuterJoin(metaProcessor).on(meta.PROCESSOR_ID.eq(metaProcessor.ID))
                                    : meta;

                            // We might want to do this delete using a temp table like we do for
                            // MetaDaoImpl#updateStatusWithTempTable. Only if locking other rows is
                            // an issue.
                            final var query = context
                                    .update(tableClause)
                                    .set(meta.STATUS, statusIdDeleted)
                                    .set(meta.STATUS_TIME, Instant.now().toEpochMilli())
                                    .where(conditions)
                                    .and(meta.CREATE_TIME.greaterOrEqual(subPeriod.getFrom().toEpochMilli()))
                                    .and(meta.CREATE_TIME.lessThan(subPeriod.getTo().toEpochMilli()));

                            LOGGER.debug("update:\n{}", query);

                            return query.execute();
                        }),
                        cnt -> LogUtil.message(
                                "Logically deleted {} meta records with approx. limit of {}. Iteration {}",
                                cnt,
                                batchSize,
                                iteration));
                totalUpdateDuration = totalUpdateDuration.plus(Duration.between(startTime,
                        Instant.now().plusMillis(1)));

                // Advance the start time
                startTimeInc = subPeriod.getFrom();
                totalUpdateCount.addAndGet(lastUpdateCount);
                iteration.incrementAndGet();
                LOGGER.debug("Logically deleted {} meta rows (total so far {})", lastUpdateCount, totalUpdateCount);

            } while (lastUpdateCount != 0 && !Thread.currentThread().isInterrupted());

            LOGGER.info("Logically deleted {} meta rows, batchSize {}, iterations {}, select avg {}, update avg {}",
                    totalUpdateCount.get(),
                    batchSize,
                    iteration.decrementAndGet(),
                    iteration.get() != 0
                            ? totalSelectDuration.dividedBy(iteration.get())
                            : Duration.ZERO,
                    iteration.get() != 0
                            ? totalUpdateDuration.dividedBy(iteration.get())
                            : Duration.ZERO);

            if (Thread.currentThread().isInterrupted()) {
                LOGGER.error("Thread interrupted");
            }
        } else {
            LOGGER.debug("Empty ruleActions, nothing to delete");
            totalUpdateCount.set(0);
        }
        return totalUpdateCount.get();
    }
@return True if any term in any rule expression uses at least one of {@code fields}.
stroom-meta/stroom-meta-impl-db/src/main/java/stroom/meta/impl/db/MetaDaoImpl.java