/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANYDa
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.elasticsearch.index.mapper;

import org.apache.lucene.document.DoubleRange;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.FloatRange;
import org.apache.lucene.document.InetAddressPoint;
import org.apache.lucene.document.InetAddressRange;
import org.apache.lucene.document.IntRange;
import org.apache.lucene.document.LongRange;
import org.apache.lucene.document.StoredField;
import org.apache.lucene.index.IndexableField;
import org.apache.lucene.queries.BinaryDocValuesRangeQuery;
import org.apache.lucene.search.IndexOrDocValuesQuery;
import org.apache.lucene.search.MatchNoDocsQuery;
import org.apache.lucene.search.Query;
import org.apache.lucene.util.BytesRef;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.geo.ShapeRelation;
import org.elasticsearch.common.network.InetAddresses;
import org.elasticsearch.common.time.DateMathParser;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.query.QueryShardContext;

import java.io.IOException;
import java.net.InetAddress;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.function.BiFunction;

/** Enum defining the type of range */
public enum RangeType {
    IP("ip_range", LengthType.FIXED_16) {
        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new InetAddressRange(name, (InetAddress)r.from, (InetAddress)r.to);
        }
        @Override
        public InetAddress parseFrom(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce, boolean included)
                throws IOException {
            InetAddress address = InetAddresses.forString(parser.text());
            return included ? address : nextUp(address);
        }
        @Override
        public InetAddress parseTo(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce, boolean included)
                throws IOException {
            InetAddress address = InetAddresses.forString(parser.text());
            return included ? address : nextDown(address);
        }
        @Override
        public InetAddress parse(Object value, boolean coerce) {
            if (value instanceof InetAddress) {
                return (InetAddress) value;
            } else {
                if (value instanceof BytesRef) {
                    value = ((BytesRef) value).utf8ToString();
                }
                return InetAddresses.forString(value.toString());
            }
        }
        @Override
        public InetAddress minValue() {
            return InetAddressPoint.MIN_VALUE;
        }
        @Override
        public InetAddress maxValue() {
            return InetAddressPoint.MAX_VALUE;
        }
        @Override
        public InetAddress nextUp(Object value) {
            return InetAddressPoint.nextUp((InetAddress)value);
        }
        @Override
        public InetAddress nextDown(Object value) {
            return InetAddressPoint.nextDown((InetAddress)value);
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return BinaryRangeUtil.encodeIPRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            // TODO: Implement this.
            throw new UnsupportedOperationException();
        }

        @Override
        public Double doubleValue (Object endpointValue) {
            throw new UnsupportedOperationException("IP ranges cannot be safely converted to doubles");
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            if (includeFrom == false) {
                from = nextUp(from);
            }

            if (includeTo == false) {
                to = nextDown(to);
            }

            byte[] encodedFrom = InetAddressPoint.encode((InetAddress) from);
            byte[] encodedTo = InetAddressPoint.encode((InetAddress) to);
            return new BinaryDocValuesRangeQuery(field, queryType, LengthType.FIXED_16,
                    new BytesRef(encodedFrom), new BytesRef(encodedTo), from, to);
        }

        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, from, to, includeFrom, includeTo,
                    (f, t) -> InetAddressRange.newWithinQuery(field, f, t));
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, from, to, includeFrom, includeTo,
                    (f, t) -> InetAddressRange.newContainsQuery(field, f, t ));
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, from, to, includeFrom, includeTo,
                    (f, t) -> InetAddressRange.newIntersectsQuery(field, f ,t ));
        }

        private Query createQuery(String field, Object lower, Object upper, boolean includeLower, boolean includeUpper,
                BiFunction<InetAddress, InetAddress, Query> querySupplier) {
            byte[] lowerBytes = InetAddressPoint.encode((InetAddress) lower);
            byte[] upperBytes = InetAddressPoint.encode((InetAddress) upper);
            if (Arrays.compareUnsigned(lowerBytes, 0, lowerBytes.length, upperBytes, 0, upperBytes.length) > 0) {
                throw new IllegalArgumentException(
                        "Range query `from` value (" + lower + ") is greater than `to` value (" + upper + ")");
            }
            InetAddress correctedFrom = includeLower ? (InetAddress) lower : nextUp(lower);
            InetAddress correctedTo = includeUpper ? (InetAddress) upper : nextDown(upper);;
            lowerBytes = InetAddressPoint.encode(correctedFrom);
            upperBytes = InetAddressPoint.encode(correctedTo);
            if (Arrays.compareUnsigned(lowerBytes, 0, lowerBytes.length, upperBytes, 0, upperBytes.length) > 0) {
                return new MatchNoDocsQuery("float range didn't intersect anything");
            } else {
                return querySupplier.apply(correctedFrom, correctedTo);
            }
        }
    },
    DATE("date_range", LengthType.VARIABLE, NumberFieldMapper.NumberType.LONG) {
        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new LongRange(name, new long[] {((Number)r.from).longValue()}, new long[] {((Number)r.to).longValue()});
        }
        private Number parse(DateMathParser dateMathParser, String dateStr) {
            return dateMathParser.parse(dateStr, () -> {throw new IllegalArgumentException("now is not used at indexing time");})
                .toEpochMilli();
        }
        @Override
        public Number parseFrom(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce, boolean included)
                throws IOException {
            Number value = parse(fieldType.dateMathParser, parser.text());
            return included ? value : nextUp(value);
        }
        @Override
        public Number parseTo(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce, boolean included)
                throws IOException{
            Number value = parse(fieldType.dateMathParser, parser.text());
            return included ? value : nextDown(value);
        }
        @Override
        public Long minValue() {
            return Long.MIN_VALUE;
        }
        @Override
        public Long maxValue() {
            return Long.MAX_VALUE;
        }
        @Override
        public Long nextUp(Object value) {
            return (long) LONG.nextUp(value);
        }
        @Override
        public Long nextDown(Object value) {
            return (long) LONG.nextDown(value);
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return LONG.encodeRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            return LONG.decodeRanges(bytes);
        }

        @Override
        public Double doubleValue (Object endpointValue) {
            return LONG.doubleValue(endpointValue);
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            return LONG.dvRangeQuery(field, queryType, from, to, includeFrom, includeTo);
        }

        @Override
        public Query rangeQuery(String field, boolean hasDocValues, Object lowerTerm, Object upperTerm, boolean includeLower,
                                boolean includeUpper, ShapeRelation relation, @Nullable ZoneId timeZone,
                                @Nullable DateMathParser parser, QueryShardContext context) {
            ZoneId zone = (timeZone == null) ? ZoneOffset.UTC : timeZone;

            DateMathParser dateMathParser = (parser == null) ?
                DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.toDateMathParser() : parser;
            Long low = lowerTerm == null ? Long.MIN_VALUE :
                dateMathParser.parse(lowerTerm instanceof BytesRef ? ((BytesRef) lowerTerm).utf8ToString() : lowerTerm.toString(),
                    context::nowInMillis, false, zone).toEpochMilli();
            Long high = upperTerm == null ? Long.MAX_VALUE :
                dateMathParser.parse(upperTerm instanceof BytesRef ? ((BytesRef) upperTerm).utf8ToString() : upperTerm.toString(),
                    context::nowInMillis, false, zone).toEpochMilli();

            return super.rangeQuery(field, hasDocValues, low, high, includeLower, includeUpper, relation, zone,
                dateMathParser, context);
        }
        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeLower, boolean includeUpper) {
            return LONG.withinQuery(field, from, to, includeLower, includeUpper);
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeLower, boolean includeUpper) {
            return LONG.containsQuery(field, from, to, includeLower, includeUpper);
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeLower, boolean includeUpper) {
            return LONG.intersectsQuery(field, from, to, includeLower, includeUpper);
        }
    },
    // todo support half_float
    FLOAT("float_range", LengthType.FIXED_4, NumberFieldMapper.NumberType.FLOAT) {
        @Override
        public Float minValue() {
            return Float.NEGATIVE_INFINITY;
        }
        @Override
        public Float maxValue() {
            return Float.POSITIVE_INFINITY;
        }
        @Override
        public Float nextUp(Object value) {
            return Math.nextUp(((Number)value).floatValue());
        }
        @Override
        public Float nextDown(Object value) {
            return Math.nextDown(((Number)value).floatValue());
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return BinaryRangeUtil.encodeFloatRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            return BinaryRangeUtil.decodeFloatRanges(bytes);
        }

        @Override
        public Double doubleValue(Object endpointValue) {
            assert endpointValue instanceof Float;
            return ((Float) endpointValue).doubleValue();
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            if (includeFrom == false) {
                from = nextUp(from);
            }

            if (includeTo == false) {
                to = nextDown(to);
            }

            byte[] encodedFrom = BinaryRangeUtil.encodeFloat((Float) from);
            byte[] encodedTo = BinaryRangeUtil.encodeFloat((Float) to);
            return new BinaryDocValuesRangeQuery(field, queryType, LengthType.FIXED_4,
                    new BytesRef(encodedFrom), new BytesRef(encodedTo), from, to);
        }

        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new FloatRange(name, new float[] {((Number)r.from).floatValue()}, new float[] {((Number)r.to).floatValue()});
        }
        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Float) from, (Float) to, includeFrom, includeTo,
                    (f, t) -> FloatRange.newWithinQuery(field, new float[] { f }, new float[] { t }), RangeType.FLOAT);
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Float) from, (Float) to, includeFrom, includeTo,
                    (f, t) -> FloatRange.newContainsQuery(field, new float[] { f }, new float[] { t }), RangeType.FLOAT);
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Float) from, (Float) to, includeFrom, includeTo,
                    (f, t) -> FloatRange.newIntersectsQuery(field, new float[] { f }, new float[] { t }), RangeType.FLOAT);
        }
    },
    DOUBLE("double_range", LengthType.FIXED_8, NumberFieldMapper.NumberType.DOUBLE) {
        @Override
        public Double minValue() {
            return Double.NEGATIVE_INFINITY;
        }
        @Override
        public Double maxValue() {
            return Double.POSITIVE_INFINITY;
        }
        @Override
        public Double nextUp(Object value) {
            return Math.nextUp(((Number)value).doubleValue());
        }
        @Override
        public Double nextDown(Object value) {
            return Math.nextDown(((Number)value).doubleValue());
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return BinaryRangeUtil.encodeDoubleRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            return BinaryRangeUtil.decodeDoubleRanges(bytes);
        }

        @Override
        public Double doubleValue(Object endpointValue) {
            assert endpointValue instanceof Double;
            return (Double) endpointValue;
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            if (includeFrom == false) {
                from = nextUp(from);
            }

            if (includeTo == false) {
                to = nextDown(to);
            }

            byte[] encodedFrom = BinaryRangeUtil.encodeDouble((Double) from);
            byte[] encodedTo = BinaryRangeUtil.encodeDouble((Double) to);
            return new BinaryDocValuesRangeQuery(field, queryType, LengthType.FIXED_8,
                    new BytesRef(encodedFrom), new BytesRef(encodedTo), from, to);
        }

        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new DoubleRange(name, new double[] {((Number)r.from).doubleValue()}, new double[] {((Number)r.to).doubleValue()});
        }
        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Double) from, (Double) to, includeFrom, includeTo,
                    (f, t) -> DoubleRange.newWithinQuery(field, new double[] { f }, new double[] { t }), RangeType.DOUBLE);
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Double) from, (Double) to, includeFrom, includeTo,
                    (f, t) -> DoubleRange.newContainsQuery(field, new double[] { f }, new double[] { t }), RangeType.DOUBLE);
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Double) from, (Double) to, includeFrom, includeTo,
                    (f, t) -> DoubleRange.newIntersectsQuery(field, new double[] { f }, new double[] { t }), RangeType.DOUBLE);
        }

    },
    // todo add BYTE support
    // todo add SHORT support
    INTEGER("integer_range", LengthType.VARIABLE, NumberFieldMapper.NumberType.INTEGER) {
        @Override
        public Integer minValue() {
            return Integer.MIN_VALUE;
        }
        @Override
        public Integer maxValue() {
            return Integer.MAX_VALUE;
        }
        @Override
        public Integer nextUp(Object value) {
            return ((Number)value).intValue() + 1;
        }
        @Override
        public Integer nextDown(Object value) {
            return ((Number)value).intValue() - 1;
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return LONG.encodeRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            return LONG.decodeRanges(bytes);
        }

        @Override
        public Double doubleValue(Object endpointValue) {
            return LONG.doubleValue(endpointValue);
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            return LONG.dvRangeQuery(field, queryType, from, to, includeFrom, includeTo);
        }

        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new IntRange(name, new int[] {((Number)r.from).intValue()}, new int[] {((Number)r.to).intValue()});
        }
        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Integer) from, (Integer) to, includeFrom, includeTo,
                    (f, t) -> IntRange.newWithinQuery(field, new int[] { f }, new int[] { t }), RangeType.INTEGER);
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field,  (Integer) from,  (Integer) to, includeFrom, includeTo,
                    (f, t) -> IntRange.newContainsQuery(field, new int[] { f }, new int[] { t }), RangeType.INTEGER);
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field,  (Integer) from,  (Integer) to, includeFrom, includeTo,
                    (f, t) -> IntRange.newIntersectsQuery(field, new int[] { f }, new int[] { t }), RangeType.INTEGER);
        }
    },
    LONG("long_range", LengthType.VARIABLE, NumberFieldMapper.NumberType.LONG) {
        @Override
        public Long minValue() {
            return Long.MIN_VALUE;
        }
        @Override
        public Long maxValue() {
            return Long.MAX_VALUE;
        }
        @Override
        public Long nextUp(Object value) {
            return ((Number)value).longValue() + 1;
        }
        @Override
        public Long nextDown(Object value) {
            return ((Number)value).longValue() - 1;
        }

        @Override
        public BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException {
            return BinaryRangeUtil.encodeLongRanges(ranges);
        }

        @Override
        public List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes) {
            return BinaryRangeUtil.decodeLongRanges(bytes);
        }

        @Override
        public Double doubleValue(Object endpointValue) {
            assert endpointValue instanceof Long;
            return ((Long) endpointValue).doubleValue();
        }

        @Override
        public Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to, boolean includeFrom,
                                  boolean includeTo) {
            if (includeFrom == false) {
                from = nextUp(from);
            }

            if (includeTo == false) {
                to = nextDown(to);
            }

            byte[] encodedFrom = BinaryRangeUtil.encodeLong(((Number) from).longValue());
            byte[] encodedTo = BinaryRangeUtil.encodeLong(((Number) to).longValue());
            return new BinaryDocValuesRangeQuery(field, queryType, LengthType.VARIABLE,
                    new BytesRef(encodedFrom), new BytesRef(encodedTo), from, to);
        }

        @Override
        public Field getRangeField(String name, RangeFieldMapper.Range r) {
            return new LongRange(name, new long[] {((Number)r.from).longValue()},
                new long[] {((Number)r.to).longValue()});
        }
        @Override
        public Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Long) from, (Long) to, includeFrom, includeTo,
                    (f, t) -> LongRange.newWithinQuery(field, new long[] { f }, new long[] { t }), RangeType.LONG);
        }
        @Override
        public Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Long) from, (Long) to, includeFrom, includeTo,
                    (f, t) -> LongRange.newContainsQuery(field, new long[] { f }, new long[] { t }), RangeType.LONG);
        }
        @Override
        public Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo) {
            return createQuery(field, (Long) from, (Long) to, includeFrom, includeTo,
                    (f, t) -> LongRange.newIntersectsQuery(field, new long[] { f }, new long[] { t }), RangeType.LONG);
        }
    };

    RangeType(String name, LengthType lengthType) {
        this.name = name;
        this.numberType = null;
        this.lengthType = lengthType;
    }

    RangeType(String name, LengthType lengthType, NumberFieldMapper.NumberType type) {
        this.name = name;
        this.numberType = type;
        this.lengthType = lengthType;
    }

    /** Get the associated type name. */
    public final String typeName() {
        return name;
    }

    /**
     * Internal helper to create the actual {@link Query} using the provided supplier function. Before creating the query we check if
     * the intervals min &gt; max, in which case an {@link IllegalArgumentException} is raised. The method adapts the interval bounds
     * based on whether the edges should be included or excluded. In case where after this correction the interval would be empty
     * because min &gt; max, we simply return a {@link MatchNoDocsQuery}.
     * This helper handles all {@link Number} cases and dates, the IP range type uses its own logic.
     */
    private static <T extends Comparable<T>> Query createQuery(String field, T from, T to, boolean includeFrom, boolean includeTo,
            BiFunction<T, T, Query> querySupplier, RangeType rangeType) {
        if (from.compareTo(to) > 0) {
            // wrong argument order, this is an error the user should fix
            throw new IllegalArgumentException("Range query `from` value (" + from + ") is greater than `to` value (" + to + ")");
        }

        @SuppressWarnings("unchecked")
        T correctedFrom = includeFrom ? from : (T) rangeType.nextUp(from);
        @SuppressWarnings("unchecked")
        T correctedTo =  includeTo ? to : (T) rangeType.nextDown(to);
        if (correctedFrom.compareTo(correctedTo) > 0) {
            return new MatchNoDocsQuery("range didn't intersect anything");
        } else {
            return querySupplier.apply(correctedFrom, correctedTo);
        }
    }

    public abstract Field getRangeField(String name, RangeFieldMapper.Range range);
    public List<IndexableField> createFields(ParseContext context, String name, RangeFieldMapper.Range range, boolean indexed,
                                             boolean docValued, boolean stored) {
        assert range != null : "range cannot be null when creating fields";
        List<IndexableField> fields = new ArrayList<>();
        if (indexed) {
            fields.add(getRangeField(name, range));
        }
        if (docValued) {
            RangeFieldMapper.BinaryRangesDocValuesField field = (RangeFieldMapper.BinaryRangesDocValuesField) context.doc().getByKey(name);
            if (field == null) {
                field = new RangeFieldMapper.BinaryRangesDocValuesField(name, range, this);
                context.doc().addWithKey(name, field);
            } else {
                field.add(range);
            }
        }
        if (stored) {
            fields.add(new StoredField(name, range.toString()));
        }
        return fields;
    }
    /** parses from value. rounds according to included flag */
    public Object parseFrom(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce,
                            boolean included) throws IOException {
        Number value = numberType.parse(parser, coerce);
        return included ? value : (Number)nextUp(value);
    }
    /** parses to value. rounds according to included flag */
    public Object parseTo(RangeFieldMapper.RangeFieldType fieldType, XContentParser parser, boolean coerce,
                          boolean included) throws IOException {
        Number value = numberType.parse(parser, coerce);
        return included ? value : (Number)nextDown(value);
    }

    public abstract Object minValue();
    public abstract Object maxValue();
    public abstract Object nextUp(Object value);
    public abstract Object nextDown(Object value);
    public abstract Query withinQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo);
    public abstract Query containsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo);
    public abstract Query intersectsQuery(String field, Object from, Object to, boolean includeFrom, boolean includeTo);
    public Object parse(Object value, boolean coerce) {
        return numberType.parse(value, coerce);
    }
    public Query rangeQuery(String field, boolean hasDocValues, Object from, Object to, boolean includeFrom, boolean includeTo,
                            ShapeRelation relation, @Nullable ZoneId timeZone, @Nullable DateMathParser dateMathParser,
                            QueryShardContext context) {
        Object lower = from == null ? minValue() : parse(from, false);
        Object upper = to == null ? maxValue() : parse(to, false);
        Query indexQuery;
        if (relation == ShapeRelation.WITHIN) {
            indexQuery = withinQuery(field, lower, upper, includeFrom, includeTo);
        } else if (relation == ShapeRelation.CONTAINS) {
            indexQuery = containsQuery(field, lower, upper, includeFrom, includeTo);
        } else {
            indexQuery = intersectsQuery(field, lower, upper, includeFrom, includeTo);
        }
        if (hasDocValues) {
            final BinaryDocValuesRangeQuery.QueryType queryType;
            if (relation == ShapeRelation.WITHIN) {
                queryType = BinaryDocValuesRangeQuery.QueryType.WITHIN;
            } else if (relation == ShapeRelation.CONTAINS) {
                queryType = BinaryDocValuesRangeQuery.QueryType.CONTAINS;
            } else {
                queryType = BinaryDocValuesRangeQuery.QueryType.INTERSECTS;
            }
            Query dvQuery = dvRangeQuery(field, queryType, lower, upper, includeFrom, includeTo);
            return new IndexOrDocValuesQuery(indexQuery, dvQuery);
        } else {
            return indexQuery;
        }
    }

    // No need to take into account Range#includeFrom or Range#includeTo, because from and to have already been
    // rounded up via parseFrom and parseTo methods.
    public abstract BytesRef encodeRanges(Set<RangeFieldMapper.Range> ranges) throws IOException;
    public abstract List<RangeFieldMapper.Range> decodeRanges(BytesRef bytes);

    /**
     * Given the Range.to or Range.from Object value from a Range instance, converts that value into a Double.  Before converting, it
     * asserts that the object is of the expected type.  Operation is not supported on IP ranges (because of loss of precision)
     *
     * @param endpointValue Object value for Range.to or Range.from
     * @return endpointValue as a Double
     */
    public abstract Double doubleValue(Object endpointValue);

    public boolean isNumeric() {
        return numberType != null;
    }

    public abstract Query dvRangeQuery(String field, BinaryDocValuesRangeQuery.QueryType queryType, Object from, Object to,
                                       boolean includeFrom, boolean includeTo);

    public final String name;
    private final NumberFieldMapper.NumberType numberType;
    public final LengthType lengthType;

    public enum LengthType {
        FIXED_4 {
            @Override
            public int readLength(byte[] bytes, int offset) {
                return 4;
            }
        },
        FIXED_8 {
            @Override
            public int readLength(byte[] bytes, int offset) {
                return 8;
            }
        },
        FIXED_16 {
            @Override
            public int readLength(byte[] bytes, int offset) {
                return 16;
            }
        },
        VARIABLE {
            @Override
            public int readLength(byte[] bytes, int offset) {
                // the first bit encodes the sign and the next 4 bits encode the number
                // of additional bytes
                int token = Byte.toUnsignedInt(bytes[offset]);
                int length = (token >>> 3) & 0x0f;
                if ((token & 0x80) == 0) {
                    length = 0x0f - length;
                }
                return 1 + length;
            }
        };

        /**
         * Return the length of the value that starts at {@code offset} in {@code bytes}.
         */
        public abstract int readLength(byte[] bytes, int offset);
    }
}
