private Optional<TimePeriod> getTimeSlice(final Instant startTimeInc,
                                              final int batchSize,
                                              final List<Condition> conditions,
                                              final boolean includesMetaProcessorTbl) {
        LOGGER.debug("getTimeSlice({}, {}, {}, {})", startTimeInc, batchSize, conditions, includesMetaProcessorTbl);

        final String createTimeCol = meta.CREATE_TIME.getName();
        final String minCreateTimeCol = "min_create_time";
        final String maxCreateTimeCol = "max_create_time";

        // For a given set of conditions (that may already contain some create_time bounds
        // get the create_time range for a batch n records
        final Optional<TimePeriod> timePeriod =
                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 fromClause = includesMetaProcessorTbl
                                            ? meta.straightJoin(metaProcessor)
                                            .on(meta.PROCESSOR_ID.eq(metaProcessor.ID))
                                            : meta;

                                    final Table<?> orderedFullSet = context
                                            .select(meta.CREATE_TIME)
                                            .from(fromClause)
                                            .where(conditions)
                                            .and(meta.CREATE_TIME.greaterOrEqual(startTimeInc.toEpochMilli()))
                                            .orderBy(meta.CREATE_TIME)
                                            .asTable("orderedFullSet");

                                    final Table<?> limitedSet = context
                                            .select(orderedFullSet.fields())
                                            .from(orderedFullSet)
                                            .limit(batchSize)
                                            .asTable("limitedSet");

                                    final var query = context
                                            .select(
                                                    DSL.min(limitedSet.field(createTimeCol)).as(minCreateTimeCol),
                                                    DSL.max(limitedSet.field(createTimeCol)).as(maxCreateTimeCol))
                                            .from(limitedSet);

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

                                    return query.fetchOne();
                                })
                        .map(record -> {
                            Object min = record.get(minCreateTimeCol);
                            Object max = record.get(maxCreateTimeCol);

                            if (min == null || max == null) {
                                return Optional.empty();
                            } else {
                                // Add one to make it exclusive
                                return Optional.of(
                                        TimePeriod.between((long) min, (long) max + 1));
                            }
                        }), () -> LogUtil.message("Selecting time slice starting at {}, with batch size {}",
                        startTimeInc, batchSize));

        LOGGER.debug("Returning period {}", timePeriod);

        // NOTE The number of records in the slice may differ from the desired batch size if you have
        // multiple records on the boundary with the same create_time (unlikely with milli precision).

        return timePeriod;
    }
@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