@Override
    public void search(final ExpressionCriteria criteria,
                       final FieldIndex fieldIndex,
                       final ValuesConsumer consumer) {
        final Set<String> processorFields = Set.of(
                ProcessorTaskFields.PROCESSOR_FILTER_ID.getFldName(),
                ProcessorTaskFields.PROCESSOR_FILTER_PRIORITY.getFldName());

        validateExpressionTerms(criteria.getExpression());

        final String[] fieldNames = fieldIndex.getFields();
        final boolean nodeUsed = isUsed(Set.of(ProcessorTaskFields.NODE_NAME.getFldName()), fieldNames, criteria);
        final boolean feedUsed = isUsed(Set.of(ProcessorTaskFields.FEED.getFldName()), fieldNames, criteria);
        final boolean processorFilterUsed = isUsed(processorFields, fieldNames, criteria);
        final boolean processorUsed =
                isUsed(Set.of(ProcessorTaskFields.PROCESSOR_ID.getFldName()), fieldNames, criteria);
        final boolean pipelineUsed =
                isUsed(Set.of(
                        ProcessorTaskFields.PIPELINE.getFldName(),
                        ProcessorTaskFields.PIPELINE_NAME.getFldName()), fieldNames, criteria);

        final PageRequest pageRequest = criteria.getPageRequest();
        final Condition condition = expressionMapper.apply(criteria.getExpression());
        final Collection<OrderField<?>> orderFields = JooqUtil.getOrderFields(FIELD_MAP, criteria);
        final List<Field<?>> dbFields = new ArrayList<>(valueMapper.getDbFieldsByName(fieldNames));
        final Mapper<?>[] mappers = valueMapper.getMappersForFieldNames(fieldNames);

        JooqUtil.context(processorDbConnProvider, context -> {
            Integer offset = null;
            Integer numberOfRows = null;

            if (pageRequest != null) {
                offset = pageRequest.getOffset();
                numberOfRows = pageRequest.getLength();
            }

            var select = context.select(dbFields).from(PROCESSOR_TASK);
            if (nodeUsed) {
                select = select.leftOuterJoin(PROCESSOR_NODE)
                        .on(PROCESSOR_TASK.FK_PROCESSOR_NODE_ID.eq(PROCESSOR_NODE.ID));
            }
            if (feedUsed) {
                select = select.leftOuterJoin(PROCESSOR_FEED)
                        .on(PROCESSOR_TASK.FK_PROCESSOR_FEED_ID.eq(PROCESSOR_FEED.ID));
            }
            if (processorFilterUsed || processorUsed || pipelineUsed) {
                select = select.join(PROCESSOR_FILTER)
                        .on(PROCESSOR_TASK.FK_PROCESSOR_FILTER_ID.eq(PROCESSOR_FILTER.ID));
            }
            if (processorUsed || pipelineUsed) {
                select = select.join(PROCESSOR)
                        .on(PROCESSOR_FILTER.FK_PROCESSOR_ID.eq(PROCESSOR.ID));
            }

            try (final Cursor<?> cursor = select
                    .where(condition)
                    .orderBy(orderFields)
                    .limit(offset, numberOfRows)
                    .fetchLazy()) {

                while (cursor.hasNext()) {
                    final Result<?> result = cursor.fetchNext(BATCH_SIZE);

                    result.forEach(r -> {
                        final Val[] arr = new Val[fieldNames.length];
                        for (int i = 0; i < fieldNames.length; i++) {
                            Val val = ValNull.INSTANCE;
                            final Mapper<?> mapper = mappers[i];
                            if (mapper != null) {
                                val = mapper.map(r);
                            }
                            arr[i] = val;
                        }
                        consumer.accept(Val.of(arr));
                    });
                }
            }
        });
    }
Count the current number of tasks for a filter matching the specified status.

@param filterId The filter to count tasks for.
@param status   Task status.
@return The number of tasks matching the specified status.
stroom-processor/stroom-processor-impl-db/src/main/java/stroom/processor/impl/db/ProcessorTaskDaoImpl.java