@Override
    public void search(final ExpressionCriteria criteria,
                       final FieldIndex fieldIndex,
                       final ValuesConsumer consumer) {
        final String[] fieldNames = fieldIndex.getFields();
        final boolean feedUsed = isUsed(Set.of(MetaFields.FEED.getFldName()), fieldNames, criteria);
        final boolean typeUsed = isUsed(Set.of(MetaFields.TYPE.getFldName()), fieldNames, criteria);
        final boolean pipelineUsed = isUsed(Set.of(
                        MetaFields.PIPELINE.getFldName(),
                        MetaFields.PIPELINE_NAME.getFldName()),
                fieldNames, criteria);
        final Set<String> extendedFieldNames = MetaFields
                .getExtendedFields()
                .stream()
                .map(QueryField::getFldName)
                .collect(Collectors.toSet());
        final boolean extendedValuesUsed = isUsed(extendedFieldNames, fieldNames, criteria);

        final PageRequest pageRequest = criteria.getPageRequest();
        final Collection<Condition> conditions = createCondition(criteria.getExpression());
        final Collection<OrderField<?>> orderFields = createOrderFields(criteria);
        final List<Field<?>> dbFields = valueMapper.getDbFieldsByName(fieldNames);
        final Mapper<?>[] mappers = valueMapper.getMappersForFieldNames(fieldNames);

        // Deal with extended fields.
        final int[] extendedFieldKeys = new int[fieldNames.length];
        final List<Integer> extendedFieldKeyIdList = new ArrayList<>();
        final Map<Long, Map<Integer, Long>> extendedFieldValueMap = new HashMap<>();
        for (int i = 0; i < fieldNames.length; i++) {
            final int index = i;
            final String fieldName = fieldNames[i];
            extendedFieldKeys[i] = -1;

            if (extendedFieldNames.contains(fieldName)) {
                final Optional<Integer> keyId = metaKeyDao.getIdForName(fieldName);
                keyId.ifPresent(id -> {
                    extendedFieldKeys[index] = id;
                    extendedFieldKeyIdList.add(id);
                });
            }
        }

        // Need to modify requested fields to include id if we are going to fetch extended attributes.
        if (extendedValuesUsed) {
            if (dbFields.stream().noneMatch(meta.ID::equals)) {
                dbFields.add(meta.ID);
            }
        }

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

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

            var select = context.select(dbFields).from(meta);
            if (feedUsed) {
                select = select.straightJoin(metaFeed).on(meta.FEED_ID.eq(metaFeed.ID));
            }
            if (typeUsed) {
                select = select.straightJoin(metaType).on(meta.TYPE_ID.eq(metaType.ID));
            }
            if (pipelineUsed) {
                select = select.leftOuterJoin(metaProcessor).on(meta.PROCESSOR_ID.eq(metaProcessor.ID));
            }

            // Need to add one join to meta_val for each meta key id used in the criteria
            final Set<Integer> usedValKeys = identifyExtendedAttributesFields(
                    criteria.getExpression(),
                    new HashSet<>());
            select = metaExpressionMapper.addJoins(select, meta.ID, usedValKeys);

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

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

                    // If we require some extended values then perform another query to get them.
                    if (extendedValuesUsed) {
                        final List<Long> idList = result.getValues(meta.ID);
                        fillExtendedFieldValueMap(context, idList, extendedFieldKeyIdList, extendedFieldValueMap);
                    }

                    result.forEach(r -> {
                        final Val[] arr = new Val[fieldNames.length];

                        Map<Integer, Long> extendedValues = null;
                        if (extendedValuesUsed) {
                            extendedValues = extendedFieldValueMap.get(r.get(meta.ID));
                        }

                        for (int i = 0; i < fieldNames.length; i++) {
                            Val val = ValNull.INSTANCE;
                            final Mapper<?> mapper = mappers[i];
                            if (mapper != null) {
                                val = mapper.map(r);

                            } else if (extendedValues != null && extendedFieldKeys[i] != -1) {
                                final Long o = extendedValues.get(extendedFieldKeys[i]);
                                if (o != null) {
                                    val = ValLong.create(o);
                                }
                            }
                            arr[i] = val;
                        }
                        consumer.accept(Val.of(arr));
                    });
                }
            }
        });
    }
@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