class LuceneReferenceResolver implements ReferenceResolver<LuceneCollectorExpression<?>> {

    private static final Set<Integer> NO_FIELD_TYPES_IDS = Set.of(ObjectType.ID, GeoShapeType.ID);
    private final FieldTypeLookup fieldTypeLookup;
    private final List<Reference> partitionColumns;
    private final String indexName;

    public LuceneReferenceResolver(final String indexName,
                                   final FieldTypeLookup fieldTypeLookup,
                                   final List<Reference> partitionColumns) {
        this.indexName = indexName;
        this.fieldTypeLookup = fieldTypeLookup;
        this.partitionColumns = partitionColumns;
    }

    @Override
    public LuceneCollectorExpression<?> getImplementation(final Reference ref) {<FILL_FUNCTION_BODY>}

    private static LuceneCollectorExpression<?> maybeInjectPartitionValue(LuceneCollectorExpression<?> result,
                                                                          String indexName,
                                                                          List<Reference> partitionColumns,
                                                                          ColumnIdent column) {
        for (int i = 0; i < partitionColumns.size(); i++) {
            final Reference partitionColumn = partitionColumns.get(i);
            final var partitionColumnIdent = partitionColumn.column();
            if (partitionColumnIdent.isChildOf(column)) {
                return new PartitionValueInjectingExpression(
                    PartitionName.fromIndexOrTemplate(indexName),
                    i,
                    partitionColumnIdent.shiftRight(),
                    result
                );
            }
        }
        return result;
    }

    private static LuceneCollectorExpression<?> typeSpecializedExpression(final FieldTypeLookup fieldTypeLookup,
                                                                          final Reference ref) {
        final String fqn = ref.storageIdent();
        final MappedFieldType fieldType = fieldTypeLookup.get(fqn);
        if (fieldType == null) {
            return NO_FIELD_TYPES_IDS.contains(unnest(ref.valueType()).id()) || isIgnoredDynamicReference(ref)
                ? DocCollectorExpression.create(toSourceLookup(ref))
                : new LiteralValueExpression(null);
        }
        if (!fieldType.hasDocValues()) {
            return DocCollectorExpression.create(toSourceLookup(ref));
        }
        switch (ref.valueType().id()) {
            case BitStringType.ID:
                return new BitStringColumnReference(fqn, ((BitStringType) ref.valueType()).length());
            case ByteType.ID:
                return new ByteColumnReference(fqn);
            case ShortType.ID:
                return new ShortColumnReference(fqn);
            case IpType.ID:
                return new IpColumnReference(fqn);
            case StringType.ID:
            case CharacterType.ID:
                return new BytesRefColumnReference(fqn);
            case DoubleType.ID:
                return new DoubleColumnReference(fqn);
            case BooleanType.ID:
                return new BooleanColumnReference(fqn);
            case FloatType.ID:
                return new FloatColumnReference(fqn);
            case LongType.ID:
            case TimestampType.ID_WITH_TZ:
            case TimestampType.ID_WITHOUT_TZ:
                return new LongColumnReference(fqn);
            case IntegerType.ID:
                return new IntegerColumnReference(fqn);
            case GeoPointType.ID:
                return new GeoPointColumnReference(fqn);
            case ArrayType.ID:
                return DocCollectorExpression.create(toSourceLookup(ref));
            case FloatVectorType.ID:
                return new FloatVectorColumnReference(fqn);
            default:
                throw new UnhandledServerException("Unsupported type: " + ref.valueType().getName());
        }
    }

    private static boolean isIgnoredDynamicReference(final Reference ref) {
        return ref.symbolType() == SymbolType.DYNAMIC_REFERENCE && ref.columnPolicy() == ColumnPolicy.IGNORED;
    }

    static class LiteralValueExpression extends LuceneCollectorExpression<Object> {

        private final Object value;

        public LiteralValueExpression(Object value) {
            this.value = value;
        }

        @Override
        public Object value() {
            return value;
        }
    }


    static class PartitionValueInjectingExpression extends LuceneCollectorExpression<Object> {

        private final LuceneCollectorExpression<?> inner;
        private final ColumnIdent partitionPath;
        private final int partitionPos;
        private final PartitionName partitionName;

        public PartitionValueInjectingExpression(PartitionName partitionName,
                                                 int partitionPos,
                                                 ColumnIdent partitionPath,
                                                 LuceneCollectorExpression<?> inner) {
            this.inner = inner;
            this.partitionName = partitionName;
            this.partitionPos = partitionPos;
            this.partitionPath = partitionPath;
        }

        @SuppressWarnings("unchecked")
        @Override
        public Object value() {
            final var object = (Map<String, Object>) inner.value();
            final var partitionValue = partitionName.values().get(partitionPos);
            Maps.mergeInto(
                object,
                partitionPath.name(),
                partitionPath.path(),
                partitionValue
            );
            return object;
        }

        @Override
        public void startCollect(final CollectorContext context) {
            inner.startCollect(context);
        }

        @Override
        public void setNextDocId(final int doc) {
            inner.setNextDocId(doc);
        }

        @Override
        public void setNextReader(ReaderContext context) throws IOException {
            inner.setNextReader(context);
        }

        @Override
        public void setScorer(final Scorable scorer) {
            inner.setScorer(scorer);
        }
    }
}

        final ColumnIdent column = ref.column();
        switch (column.name()) {
            case DocSysColumns.Names.RAW:
                if (column.isRoot()) {
                    return new RawCollectorExpression();
                }
                throw new UnsupportedFeatureException("_raw expression does not support subscripts: " + column);

            case DocSysColumns.Names.UID:
            case DocSysColumns.Names.ID:
                return new IdCollectorExpression();

            case DocSysColumns.Names.FETCHID:
                return new FetchIdCollectorExpression();

            case DocSysColumns.Names.DOCID:
                return new DocIdCollectorExpression();

            case DocSysColumns.Names.SCORE:
                return new ScoreCollectorExpression();

            case DocSysColumns.Names.VERSION:
                return new VersionCollectorExpression();

            case DocSysColumns.Names.SEQ_NO:
                return new SeqNoCollectorExpression();

            case DocSysColumns.Names.PRIMARY_TERM:
                return new PrimaryTermCollectorExpression();

            case DocSysColumns.Names.DOC: {
                var result = DocCollectorExpression.create(ref);
                return maybeInjectPartitionValue(
                    result,
                    indexName,
                    partitionColumns,
                    column.isRoot() ? column : column.shiftRight()  // Remove `_doc` prefix so that it can match the column against partitionColumns
                );
            }

            default: {
                int partitionPos = Reference.indexOf(partitionColumns, column);
                if (partitionPos >= 0) {
                    return new LiteralValueExpression(
                        ref.valueType().implicitCast(PartitionName.fromIndexOrTemplate(indexName).values().get(partitionPos))
                    );
                }
                return maybeInjectPartitionValue(
                    typeSpecializedExpression(fieldTypeLookup, ref),
                    indexName,
                    partitionColumns,
                    column
                );
            }
        }
    
<no_super_class>