diff --git a/src/main/java/org/springframework/data/elasticsearch/core/Range.java b/src/main/java/org/springframework/data/elasticsearch/core/Range.java
new file mode 100644
index 000000000..762fa758b
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/Range.java
@@ -0,0 +1,444 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core;
+
+import java.util.Optional;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Simple value object to work with ranges and boundaries.
+ * 
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class Range<T> {
+
+	private final static Range<?> UNBOUNDED = Range.of(Bound.unbounded(), Bound.UNBOUNDED);
+
+	/**
+	 * The lower bound of the range.
+	 */
+	private Bound<T> lowerBound;
+
+	/**
+	 * The upper bound of the range.
+	 */
+	private Bound<T> upperBound;
+
+	/**
+	 * Creates a new {@link Range} with inclusive bounds for both values.
+	 *
+	 * @param <T>
+	 * @param from must not be {@literal null}.
+	 * @param to must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> closed(T from, T to) {
+		return new Range<>(Bound.inclusive(from), Bound.inclusive(to));
+	}
+
+	/**
+	 * Creates a new Range with the given value as sole member.
+	 *
+	 * @param <T>
+	 * @param value must not be {@literal null}.
+	 * @return
+	 * @see Range#closed(T, T)
+	 */
+	public static <T> Range<T> just(T value) {
+		return Range.closed(value, value);
+	}
+
+	/**
+	 * Creates a new left-open {@link Range}, i.e. left exclusive, right inclusive.
+	 *
+	 * @param <T>
+	 * @param from must not be {@literal null}.
+	 * @param to must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> leftOpen(T from, T to) {
+		return new Range<>(Bound.exclusive(from), Bound.inclusive(to));
+	}
+
+	/**
+	 * Creates a left-unbounded {@link Range} (the left bound set to {@link Bound#unbounded()}) with the given right
+	 * bound.
+	 *
+	 * @param <T>
+	 * @param to the right {@link Bound}, must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> leftUnbounded(Bound<T> to) {
+		return new Range<>(Bound.unbounded(), to);
+	}
+
+	/**
+	 * Creates a new {@link Range} with the given lower and upper bound. Prefer {@link #from(Bound)} for a more builder
+	 * style API.
+	 *
+	 * @param lowerBound must not be {@literal null}.
+	 * @param upperBound must not be {@literal null}.
+	 * @see #from(Bound)
+	 */
+	public static <T> Range<T> of(Bound<T> lowerBound, Bound<T> upperBound) {
+		return new Range<>(lowerBound, upperBound);
+	}
+
+	/**
+	 * Creates a new {@link Range} with exclusive bounds for both values.
+	 *
+	 * @param <T>
+	 * @param from must not be {@literal null}.
+	 * @param to must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> open(T from, T to) {
+		return new Range<>(Bound.exclusive(from), Bound.exclusive(to));
+	}
+
+	/**
+	 * Creates a new right-open {@link Range}, i.e. left inclusive, right exclusive.
+	 *
+	 * @param <T>
+	 * @param from must not be {@literal null}.
+	 * @param to must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> rightOpen(T from, T to) {
+		return new Range<>(Bound.inclusive(from), Bound.exclusive(to));
+	}
+
+	/**
+	 * Creates a right-unbounded {@link Range} (the right bound set to {@link Bound#unbounded()}) with the given left
+	 * bound.
+	 *
+	 * @param <T>
+	 * @param from the left {@link Bound}, must not be {@literal null}.
+	 * @return
+	 */
+	public static <T> Range<T> rightUnbounded(Bound<T> from) {
+		return new Range<>(from, Bound.unbounded());
+	}
+
+	/**
+	 * Returns an unbounded {@link Range}.
+	 *
+	 * @return
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> Range<T> unbounded() {
+		return (Range<T>) UNBOUNDED;
+	}
+
+	private Range() {
+	}
+
+	private Range(Bound<T> lowerBound, Bound<T> upperBound) {
+
+		Assert.notNull(lowerBound, "Lower bound must not be null!");
+		Assert.notNull(upperBound, "Upper bound must not be null!");
+
+		this.lowerBound = lowerBound;
+		this.upperBound = upperBound;
+	}
+
+	/**
+	 * Returns whether the {@link Range} contains the given value.
+	 *
+	 * @param value must not be {@literal null}.
+	 * @return
+	 */
+	public boolean contains(T value) {
+
+		Assert.notNull(value, "Reference value must not be null!");
+		Assert.isInstanceOf(Comparable.class, value, "value must implements Comparable!");
+
+		boolean greaterThanLowerBound = lowerBound.getValue() //
+				.map(it -> lowerBound.isInclusive() ? ((Comparable<? super T>) it).compareTo(value) <= 0
+						: ((Comparable<? super T>) it).compareTo(value) < 0) //
+				.orElse(true);
+
+		boolean lessThanUpperBound = upperBound.getValue() //
+				.map(it -> upperBound.isInclusive() ? ((Comparable<? super T>) it).compareTo(value) >= 0
+						: ((Comparable<? super T>) it).compareTo(value) > 0) //
+				.orElse(true);
+
+		return greaterThanLowerBound && lessThanUpperBound;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#equals(java.lang.Object)
+	 */
+	@Override
+	public boolean equals(Object o) {
+
+		if (this == o) {
+			return true;
+		}
+
+		if (!(o instanceof Range)) {
+			return false;
+		}
+
+		Range<?> range = (Range<?>) o;
+
+		if (!ObjectUtils.nullSafeEquals(lowerBound, range.lowerBound)) {
+			return false;
+		}
+
+		return ObjectUtils.nullSafeEquals(upperBound, range.upperBound);
+	}
+
+	public Range.Bound<T> getLowerBound() {
+		return this.lowerBound;
+	}
+
+	public Range.Bound<T> getUpperBound() {
+		return this.upperBound;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#hashCode()
+	 */
+	@Override
+	public int hashCode() {
+		int result = ObjectUtils.nullSafeHashCode(lowerBound);
+		result = 31 * result + ObjectUtils.nullSafeHashCode(upperBound);
+		return result;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+		return String.format("%s-%s", lowerBound.toPrefixString(), upperBound.toSuffixString());
+	}
+
+	/**
+	 * Value object representing a boundary. A boundary can either be {@link #unbounded() unbounded}, {@link #inclusive(T)
+	 * including its value} or {@link #exclusive(T) its value}.
+	 */
+	public static final class Bound<T> {
+
+		@SuppressWarnings({ "rawtypes", "unchecked" }) //
+		private static final Bound<?> UNBOUNDED = new Bound(Optional.empty(), true);
+
+		private final Optional<T> value;
+		private final boolean inclusive;
+
+		/**
+		 * Creates a boundary excluding {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Double> exclusive(double value) {
+			return exclusive((Double) value);
+		}
+
+		/**
+		 * Creates a boundary excluding {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Float> exclusive(float value) {
+			return exclusive((Float) value);
+		}
+
+		/**
+		 * Creates a boundary excluding {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Integer> exclusive(int value) {
+			return exclusive((Integer) value);
+		}
+
+		/**
+		 * Creates a boundary excluding {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Long> exclusive(long value) {
+			return exclusive((Long) value);
+		}
+
+		/**
+		 * Creates a boundary excluding {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static <T> Bound<T> exclusive(T value) {
+
+			Assert.notNull(value, "Value must not be null!");
+			Assert.isInstanceOf(Comparable.class, value, "value must implements Comparable!");
+			return new Bound<>(Optional.of(value), false);
+		}
+
+		/**
+		 * Creates a boundary including {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Double> inclusive(double value) {
+			return inclusive((Double) value);
+		}
+
+		/**
+		 * Creates a boundary including {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Float> inclusive(float value) {
+			return inclusive((Float) value);
+		}
+
+		/**
+		 * Creates a boundary including {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Integer> inclusive(int value) {
+			return inclusive((Integer) value);
+		}
+
+		/**
+		 * Creates a boundary including {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static Bound<Long> inclusive(long value) {
+			return inclusive((Long) value);
+		}
+
+		/**
+		 * Creates a boundary including {@code value}.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public static <T> Bound<T> inclusive(T value) {
+
+			Assert.notNull(value, "Value must not be null!");
+			Assert.isInstanceOf(Comparable.class, value, "value must implements Comparable!");
+			return new Bound<>(Optional.of(value), true);
+		}
+
+		/**
+		 * Creates an unbounded {@link Bound}.
+		 */
+		@SuppressWarnings("unchecked")
+		public static <T> Bound<T> unbounded() {
+			return (Bound<T>) UNBOUNDED;
+		}
+
+		private Bound(Optional<T> value, boolean inclusive) {
+			this.value = value;
+			this.inclusive = inclusive;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Object#equals(java.lang.Object)
+		 */
+		@Override
+		public boolean equals(Object o) {
+
+			if (this == o) {
+				return true;
+			}
+
+			if (!(o instanceof Bound)) {
+				return false;
+			}
+
+			Bound<?> bound = (Bound<?>) o;
+
+			if (inclusive != bound.inclusive)
+				return false;
+
+			return ObjectUtils.nullSafeEquals(value, bound.value);
+		}
+
+		public Optional<T> getValue() {
+			return this.value;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Object#hashCode()
+		 */
+		@Override
+		public int hashCode() {
+			int result = ObjectUtils.nullSafeHashCode(value);
+			result = 31 * result + (inclusive ? 1 : 0);
+			return result;
+		}
+
+		/**
+		 * Returns whether this boundary is bounded.
+		 *
+		 * @return
+		 */
+		public boolean isBounded() {
+			return value.isPresent();
+		}
+
+		public boolean isInclusive() {
+			return this.inclusive;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Object#toString()
+		 */
+		@Override
+		public String toString() {
+			return value.map(Object::toString).orElse("unbounded");
+		}
+
+		String toPrefixString() {
+
+			return getValue() //
+					.map(Object::toString) //
+					.map(it -> isInclusive() ? "[".concat(it) : "(".concat(it)) //
+					.orElse("unbounded");
+		}
+
+		String toSuffixString() {
+
+			return getValue() //
+					.map(Object::toString) //
+					.map(it -> isInclusive() ? it.concat("]") : it.concat(")")) //
+					.orElse("unbounded");
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractPersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractPersistentPropertyConverter.java
new file mode 100644
index 000000000..44a06beb5
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractPersistentPropertyConverter.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersistentPropertyConverter;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.util.Assert;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public abstract class AbstractPersistentPropertyConverter implements ElasticsearchPersistentPropertyConverter {
+
+	private final PersistentProperty<?> property;
+
+	public AbstractPersistentPropertyConverter(PersistentProperty<?> property) {
+
+		Assert.notNull(property, "property must not be null.");
+		this.property = property;
+	}
+
+	protected PersistentProperty<?> getProperty() {
+		return property;
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractRangePersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractRangePersistentPropertyConverter.java
new file mode 100644
index 000000000..223f82f4f
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/AbstractRangePersistentPropertyConverter.java
@@ -0,0 +1,121 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.data.elasticsearch.core.Range;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.util.Assert;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public abstract class AbstractRangePersistentPropertyConverter<T> extends AbstractPersistentPropertyConverter {
+
+	protected static final String LT_FIELD = "lt";
+	protected static final String LTE_FIELD = "lte";
+	protected static final String GT_FIELD = "gt";
+	protected static final String GTE_FIELD = "gte";
+
+	public AbstractRangePersistentPropertyConverter(PersistentProperty<?> property) {
+		super(property);
+	}
+
+	@Override
+	public Object read(Object value) {
+
+		Assert.notNull(value, "value must not be null.");
+		Assert.isInstanceOf(Map.class, value, "value must be instance of Map.");
+
+		try {
+			Map<String, Object> source = (Map<String, Object>) value;
+			Range.Bound<T> lowerBound;
+			Range.Bound<T> upperBound;
+
+			if (source.containsKey(GTE_FIELD)) {
+				lowerBound = Range.Bound.inclusive(parse((String) source.get(GTE_FIELD)));
+			} else if (source.containsKey(GT_FIELD)) {
+				lowerBound = Range.Bound.exclusive(parse((String) source.get(GT_FIELD)));
+			} else {
+				lowerBound = Range.Bound.unbounded();
+			}
+
+			if (source.containsKey(LTE_FIELD)) {
+				upperBound = Range.Bound.inclusive(parse((String) source.get(LTE_FIELD)));
+			} else if (source.containsKey(LT_FIELD)) {
+				upperBound = Range.Bound.exclusive(parse((String) source.get(LT_FIELD)));
+			} else {
+				upperBound = Range.Bound.unbounded();
+			}
+
+			return Range.of(lowerBound, upperBound);
+
+		} catch (Exception e) {
+			throw new ConversionException(
+					String.format("Unable to convert value '%s' of property '%s'", value, getProperty().getName()), e);
+		}
+	}
+
+	@Override
+	public Object write(Object value) {
+
+		Assert.notNull(value, "value must not be null.");
+		Assert.isInstanceOf(Range.class, value, "value must be instance of Range.");
+
+		try {
+			Range<T> range = (Range<T>) value;
+			Range.Bound<T> lowerBound = range.getLowerBound();
+			Range.Bound<T> upperBound = range.getUpperBound();
+			Map<String, Object> target = new LinkedHashMap<>();
+
+			if (lowerBound.isBounded()) {
+				String lowerBoundValue = format(lowerBound.getValue().get());
+				if (lowerBound.isInclusive()) {
+					target.put(GTE_FIELD, lowerBoundValue);
+				} else {
+					target.put(GT_FIELD, lowerBoundValue);
+				}
+			}
+
+			if (upperBound.isBounded()) {
+				String upperBoundValue = format(upperBound.getValue().get());
+				if (upperBound.isInclusive()) {
+					target.put(LTE_FIELD, upperBoundValue);
+				} else {
+					target.put(LT_FIELD, upperBoundValue);
+				}
+			}
+
+			return target;
+
+		} catch (Exception e) {
+			throw new ConversionException(
+					String.format("Unable to convert value '%s' of property '%s'", value, getProperty().getName()), e);
+		}
+	}
+
+	protected abstract String format(T value);
+
+	protected Class<?> getGenericType() {
+		return getProperty().getTypeInformation().getTypeArguments().get(0).getType();
+	}
+
+	protected abstract T parse(String value);
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/DatePersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/DatePersistentPropertyConverter.java
new file mode 100644
index 000000000..27b0bc635
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/DatePersistentPropertyConverter.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import java.util.Date;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.mapping.PersistentProperty;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class DatePersistentPropertyConverter extends AbstractPersistentPropertyConverter {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(DatePersistentPropertyConverter.class);
+
+	private final List<ElasticsearchDateConverter> dateConverters;
+
+	public DatePersistentPropertyConverter(PersistentProperty<?> property,
+			List<ElasticsearchDateConverter> dateConverters) {
+
+		super(property);
+		this.dateConverters = dateConverters;
+	}
+
+	@Override
+	public Object read(Object value) {
+
+		String s = value.toString();
+
+		for (ElasticsearchDateConverter dateConverter : dateConverters) {
+			try {
+				return dateConverter.parse(s);
+			} catch (Exception e) {
+				LOGGER.trace(e.getMessage(), e);
+			}
+		}
+
+		throw new ConversionException(String.format("Unable to convert value '%s' to %s for property '%s'", s,
+				getProperty().getActualType().getTypeName(), getProperty().getName()));
+	}
+
+	@Override
+	public Object write(Object value) {
+
+		try {
+			return dateConverters.get(0).format((Date) value);
+		} catch (Exception e) {
+			throw new ConversionException(
+					String.format("Unable to convert value '%s' of property '%s'", value, getProperty().getName()), e);
+		}
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/DateRangePersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/DateRangePersistentPropertyConverter.java
new file mode 100644
index 000000000..b628a79f1
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/DateRangePersistentPropertyConverter.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import java.util.Date;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.mapping.PersistentProperty;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class DateRangePersistentPropertyConverter extends AbstractRangePersistentPropertyConverter<Date> {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(DateRangePersistentPropertyConverter.class);
+
+	private final List<ElasticsearchDateConverter> dateConverters;
+
+	public DateRangePersistentPropertyConverter(PersistentProperty<?> property,
+			List<ElasticsearchDateConverter> dateConverters) {
+
+		super(property);
+		this.dateConverters = dateConverters;
+	}
+
+	@Override
+	protected String format(Date value) {
+		return dateConverters.get(0).format(value);
+	}
+
+	@Override
+	protected Date parse(String value) {
+
+		for (ElasticsearchDateConverter converters : dateConverters) {
+			try {
+				return converters.parse(value);
+			} catch (Exception e) {
+				LOGGER.trace(e.getMessage(), e);
+			}
+		}
+
+		throw new ConversionException(String.format("Unable to convert value '%s' to %s for property '%s'", value,
+				getGenericType().getTypeName(), getProperty().getName()));
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverter.java
index 86643e544..f69d4563a 100644
--- a/src/main/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverter.java
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverter.java
@@ -482,10 +482,7 @@ private Object propertyConverterRead(ElasticsearchPersistentProperty property, O
 		}
 
 		private Object convertOnRead(ElasticsearchPersistentPropertyConverter propertyConverter, Object source) {
-			if (String.class.isAssignableFrom(source.getClass())) {
-				source = propertyConverter.read((String) source);
-			}
-			return source;
+			return propertyConverter.read(source);
 		}
 
 		/**
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/NumberRangePersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/NumberRangePersistentPropertyConverter.java
new file mode 100644
index 000000000..76b819e69
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/NumberRangePersistentPropertyConverter.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import org.springframework.data.mapping.PersistentProperty;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class NumberRangePersistentPropertyConverter extends AbstractRangePersistentPropertyConverter<Number> {
+
+	public NumberRangePersistentPropertyConverter(PersistentProperty<?> property) {
+		super(property);
+	}
+
+	@Override
+	protected String format(Number number) {
+		return String.valueOf(number);
+	}
+
+	@Override
+	protected Number parse(String value) {
+
+		Class<?> type = getGenericType();
+		if (Integer.class.isAssignableFrom(type)) {
+			return Integer.valueOf(value);
+		} else if (Float.class.isAssignableFrom(type)) {
+			return Float.valueOf(value);
+		} else if (Long.class.isAssignableFrom(type)) {
+			return Long.valueOf(value);
+		} else if (Double.class.isAssignableFrom(type)) {
+			return Double.valueOf(value);
+		}
+
+		throw new ConversionException(String.format("Unable to convert value '%s' to %s for property '%s'", value,
+				type.getTypeName(), getProperty().getName()));
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalPersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalPersistentPropertyConverter.java
new file mode 100644
index 000000000..2cdd167e9
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalPersistentPropertyConverter.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.mapping.PersistentProperty;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class TemporalPersistentPropertyConverter extends AbstractPersistentPropertyConverter {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(TemporalPersistentPropertyConverter.class);
+
+	private final List<ElasticsearchDateConverter> dateConverters;
+
+	public TemporalPersistentPropertyConverter(PersistentProperty<?> property,
+			List<ElasticsearchDateConverter> dateConverters) {
+
+		super(property);
+		this.dateConverters = dateConverters;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Object read(Object value) {
+
+		String s = value.toString();
+		Class<?> actualType = getProperty().getActualType();
+
+		for (ElasticsearchDateConverter dateConverter : dateConverters) {
+			try {
+				return dateConverter.parse(s, (Class<? extends TemporalAccessor>) actualType);
+			} catch (Exception e) {
+				LOGGER.trace(e.getMessage(), e);
+			}
+		}
+
+		throw new ConversionException(String.format("Unable to convert value '%s' to %s for property '%s'", s,
+				getProperty().getActualType().getTypeName(), getProperty().getName()));
+	}
+
+	@Override
+	public Object write(Object value) {
+
+		try {
+			return dateConverters.get(0).format((TemporalAccessor) value);
+		} catch (Exception e) {
+			throw new ConversionException(
+					String.format("Unable to convert value '%s' of property '%s'", value, getProperty().getName()), e);
+		}
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalRangePersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalRangePersistentPropertyConverter.java
new file mode 100644
index 000000000..6111846eb
--- /dev/null
+++ b/src/main/java/org/springframework/data/elasticsearch/core/convert/TemporalRangePersistentPropertyConverter.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.convert;
+
+import java.time.temporal.TemporalAccessor;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.util.Assert;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class TemporalRangePersistentPropertyConverter
+		extends AbstractRangePersistentPropertyConverter<TemporalAccessor> {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(TemporalRangePersistentPropertyConverter.class);
+
+	private final List<ElasticsearchDateConverter> dateConverters;
+
+	public TemporalRangePersistentPropertyConverter(PersistentProperty<?> property,
+			List<ElasticsearchDateConverter> dateConverters) {
+
+		super(property);
+
+		Assert.notEmpty(dateConverters, "dateConverters must not be empty.");
+		this.dateConverters = dateConverters;
+	}
+
+	@Override
+	protected String format(TemporalAccessor temporal) {
+		return dateConverters.get(0).format(temporal);
+	}
+
+	@Override
+	protected TemporalAccessor parse(String value) {
+
+		Class<?> type = getGenericType();
+		for (ElasticsearchDateConverter converters : dateConverters) {
+			try {
+				return converters.parse(value, (Class<? extends TemporalAccessor>) type);
+			} catch (Exception e) {
+				LOGGER.trace(e.getMessage(), e);
+			}
+		}
+
+		throw new ConversionException(String.format("Unable to convert value '%s' to %s for property '%s'", value,
+				type.getTypeName(), getProperty().getName()));
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchPersistentPropertyConverter.java b/src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchPersistentPropertyConverter.java
index 0f3183975..630f949af 100644
--- a/src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchPersistentPropertyConverter.java
+++ b/src/main/java/org/springframework/data/elasticsearch/core/mapping/ElasticsearchPersistentPropertyConverter.java
@@ -16,25 +16,26 @@
 package org.springframework.data.elasticsearch.core.mapping;
 
 /**
- * Interface defining methods to convert a property value to a String and back.
+ * Interface defining methods to convert a persistent property value to an elasticsearch property value and back.
  *
  * @author Peter-Josef Meisch
+ * @author Sascha Woo
  */
 public interface ElasticsearchPersistentPropertyConverter {
 
 	/**
-	 * converts the property value to a String.
+	 * Converts a persistent property value to an elasticsearch property value.
 	 *
-	 * @param property the property value to convert, must not be {@literal null}
-	 * @return String representation.
+	 * @param value the persistent property value to convert, must not be {@literal null}
+	 * @return The elasticsearch property value.
 	 */
-	String write(Object property);
+	Object write(Object value);
 
 	/**
-	 * converts a property value from a String.
+	 * Converts an elasticsearch property value to a persistent property value.
 	 *
-	 * @param s the property to convert, must not be {@literal null}
-	 * @return property value
+	 * @param value the elasticsearch property value to convert, must not be {@literal null}
+	 * @return The persistent property value.
 	 */
-	Object read(String s);
+	Object read(Object value);
 }
diff --git a/src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentProperty.java b/src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentProperty.java
index 1ecb9dcb1..d13dfc3c1 100644
--- a/src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentProperty.java
+++ b/src/main/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentProperty.java
@@ -29,9 +29,14 @@
 import org.springframework.data.elasticsearch.annotations.GeoPointField;
 import org.springframework.data.elasticsearch.annotations.GeoShapeField;
 import org.springframework.data.elasticsearch.annotations.MultiField;
+import org.springframework.data.elasticsearch.core.Range;
 import org.springframework.data.elasticsearch.core.completion.Completion;
-import org.springframework.data.elasticsearch.core.convert.ConversionException;
+import org.springframework.data.elasticsearch.core.convert.DatePersistentPropertyConverter;
+import org.springframework.data.elasticsearch.core.convert.DateRangePersistentPropertyConverter;
 import org.springframework.data.elasticsearch.core.convert.ElasticsearchDateConverter;
+import org.springframework.data.elasticsearch.core.convert.NumberRangePersistentPropertyConverter;
+import org.springframework.data.elasticsearch.core.convert.TemporalPersistentPropertyConverter;
+import org.springframework.data.elasticsearch.core.convert.TemporalRangePersistentPropertyConverter;
 import org.springframework.data.elasticsearch.core.geo.GeoJson;
 import org.springframework.data.elasticsearch.core.geo.GeoPoint;
 import org.springframework.data.elasticsearch.core.join.JoinField;
@@ -39,6 +44,7 @@
 import org.springframework.data.mapping.Association;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.model.AnnotationBasedPersistentProperty;
 import org.springframework.data.mapping.model.FieldNamingStrategy;
 import org.springframework.data.mapping.model.Property;
@@ -92,7 +98,7 @@ public SimpleElasticsearchPersistentProperty(Property property,
 			throw new MappingException("@Field annotation must not be used on a @MultiField property.");
 		}
 
-		initDateConverter();
+		initPropertyConverter();
 
 		storeNullValue = isField && getRequiredAnnotation(Field.class).storeNullValue();
 	}
@@ -128,102 +134,127 @@ protected boolean hasExplicitFieldName() {
 	}
 
 	/**
-	 * Initializes an {@link ElasticsearchPersistentPropertyConverter} if this property is annotated as a Field with type
-	 * {@link FieldType#Date}, has a {@link DateFormat} set and if the type of the property is one of the Java8 temporal
-	 * classes or java.util.Date.
+	 * Initializes the property converter for this {@link PersistentProperty}, if any.
 	 */
-	private void initDateConverter() {
-		Field field = findAnnotation(Field.class);
+	private void initPropertyConverter() {
 
 		Class<?> actualType = getActualTypeOrNull();
-
 		if (actualType == null) {
 			return;
 		}
 
-		boolean isTemporalAccessor = TemporalAccessor.class.isAssignableFrom(actualType);
-		boolean isDate = Date.class.isAssignableFrom(actualType);
+		Field field = findAnnotation(Field.class);
+		if (field == null) {
+			return;
+		}
 
-		if (field != null && (field.type() == FieldType.Date || field.type() == FieldType.Date_Nanos)
-				&& (isTemporalAccessor || isDate)) {
+		switch (field.type()) {
+			case Date:
+			case Date_Nanos: {
+				List<ElasticsearchDateConverter> dateConverters = getDateConverters(field, actualType);
+				if (dateConverters.isEmpty()) {
+					LOGGER.warn("No date formatters configured for property '{}'.", getName());
+					return;
+				}
 
-			DateFormat[] dateFormats = field.format();
-			String[] dateFormatPatterns = field.pattern();
+				if (TemporalAccessor.class.isAssignableFrom(actualType)) {
+					propertyConverter = new TemporalPersistentPropertyConverter(this, dateConverters);
+				} else if (Date.class.isAssignableFrom(actualType)) {
+					propertyConverter = new DatePersistentPropertyConverter(this, dateConverters);
+				} else {
+					LOGGER.warn("Unsupported type '{}' for date property '{}'.", actualType, getName());
+				}
+				break;
+			}
+			case Date_Range: {
+				if (!Range.class.isAssignableFrom(actualType)) {
+					return;
+				}
 
-			String property = getOwner().getType().getSimpleName() + "." + getName();
+				List<ElasticsearchDateConverter> dateConverters = getDateConverters(field, actualType);
+				if (dateConverters.isEmpty()) {
+					LOGGER.warn("No date formatters configured for property '{}'.", getName());
+					return;
+				}
 
-			if (dateFormats.length == 0 && dateFormatPatterns.length == 0) {
-				LOGGER.warn(
-						"Property '{}' has @Field type '{}' but has no built-in format or custom date pattern defined. Make sure you have a converter registered for type {}.",
-						property, field.type().name(), actualType.getSimpleName());
-				return;
+				Class<?> genericType = getTypeInformation().getTypeArguments().get(0).getType();
+				if (TemporalAccessor.class.isAssignableFrom(genericType)) {
+					propertyConverter = new TemporalRangePersistentPropertyConverter(this, dateConverters);
+				} else if (Date.class.isAssignableFrom(genericType)) {
+					propertyConverter = new DateRangePersistentPropertyConverter(this, dateConverters);
+				} else {
+					LOGGER.warn("Unsupported generic type '{}' for date range property '{}'.", genericType, getName());
+				}
+				break;
 			}
+			case Integer_Range:
+			case Float_Range:
+			case Long_Range:
+			case Double_Range: {
+				if (!Range.class.isAssignableFrom(actualType)) {
+					return;
+				}
 
-			List<ElasticsearchDateConverter> converters = new ArrayList<>();
-
-			// register converters for built-in formats
-			for (DateFormat dateFormat : dateFormats) {
-				switch (dateFormat) {
-					case none:
-					case custom:
-						break;
-					case weekyear:
-					case weekyear_week:
-					case weekyear_week_day:
-						LOGGER.warn("No default converter available for '{}' and date format '{}'. Use a custom converter instead.",
-								actualType.getName(), dateFormat.name());
-						break;
-					default:
-						converters.add(ElasticsearchDateConverter.of(dateFormat));
-						break;
+				Class<?> genericType = getTypeInformation().getTypeArguments().get(0).getType();
+				if ((field.type() == FieldType.Integer_Range && !Integer.class.isAssignableFrom(genericType))
+						|| (field.type() == FieldType.Float_Range && !Float.class.isAssignableFrom(genericType))
+						|| (field.type() == FieldType.Long_Range && !Long.class.isAssignableFrom(genericType))
+						|| (field.type() == FieldType.Double_Range && !Double.class.isAssignableFrom(genericType))) {
+					LOGGER.warn("Unsupported generic type '{}' for range field type '{}' of property '{}'.", genericType,
+							field.type(), getName());
+					return;
 				}
+
+				propertyConverter = new NumberRangePersistentPropertyConverter(this);
+				break;
+			}
+			case Ip_Range: {
+				// TODO currently unsupported, needs a library like https://seancfoley.github.io/IPAddress/
 			}
+			default:
+				break;
+		}
+	}
 
-			// register converters for custom formats
-			for (String dateFormatPattern : dateFormatPatterns) {
-				if (!StringUtils.hasText(dateFormatPattern)) {
-					throw new MappingException(String.format("Date pattern of property '%s' must not be empty", property));
-				}
-				converters.add(ElasticsearchDateConverter.of(dateFormatPattern));
+	private List<ElasticsearchDateConverter> getDateConverters(Field field, Class<?> actualType) {
+
+		DateFormat[] dateFormats = field.format();
+		String[] dateFormatPatterns = field.pattern();
+		List<ElasticsearchDateConverter> converters = new ArrayList<>();
+
+		if (dateFormats.length == 0 && dateFormatPatterns.length == 0) {
+			LOGGER.warn(
+					"Property '{}' has @Field type '{}' but has no built-in format or custom date pattern defined. Make sure you have a converter registered for type {}.",
+					getName(), field.type().name(), actualType.getSimpleName());
+			return converters;
+		}
+
+		// register converters for built-in formats
+		for (DateFormat dateFormat : dateFormats) {
+			switch (dateFormat) {
+				case none:
+				case custom:
+					break;
+				case weekyear:
+				case weekyear_week:
+				case weekyear_week_day:
+					LOGGER.warn("No default converter available for '{}' and date format '{}'. Use a custom converter instead.",
+							actualType.getName(), dateFormat.name());
+					break;
+				default:
+					converters.add(ElasticsearchDateConverter.of(dateFormat));
+					break;
 			}
+		}
 
-			if (!converters.isEmpty()) {
-				propertyConverter = new ElasticsearchPersistentPropertyConverter() {
-					final List<ElasticsearchDateConverter> dateConverters = converters;
-
-					@SuppressWarnings("unchecked")
-					@Override
-					public Object read(String s) {
-						for (ElasticsearchDateConverter dateConverter : dateConverters) {
-							try {
-								if (isTemporalAccessor) {
-									return dateConverter.parse(s, (Class<? extends TemporalAccessor>) actualType);
-								} else { // must be date
-									return dateConverter.parse(s);
-								}
-							} catch (Exception e) {
-								LOGGER.trace(e.getMessage(), e);
-							}
-						}
-
-						throw new ConversionException(String
-								.format("Unable to parse date value '%s' of property '%s' with configured converters", s, property));
-					}
-
-					@Override
-					public String write(Object property) {
-						ElasticsearchDateConverter dateConverter = dateConverters.get(0);
-						if (isTemporalAccessor && TemporalAccessor.class.isAssignableFrom(property.getClass())) {
-							return dateConverter.format((TemporalAccessor) property);
-						} else if (isDate && Date.class.isAssignableFrom(property.getClass())) {
-							return dateConverter.format((Date) property);
-						} else {
-							return property.toString();
-						}
-					}
-				};
+		for (String dateFormatPattern : dateFormatPatterns) {
+			if (!StringUtils.hasText(dateFormatPattern)) {
+				throw new MappingException(String.format("Date pattern of property '%s' must not be empty", getName()));
 			}
+			converters.add(ElasticsearchDateConverter.of(dateFormatPattern));
 		}
+
+		return converters;
 	}
 
 	@SuppressWarnings("ConstantConditions")
@@ -303,4 +334,5 @@ public boolean isJoinFieldProperty() {
 	public boolean isCompletionProperty() {
 		return getActualType() == Completion.class;
 	}
+
 }
diff --git a/src/test/java/org/springframework/data/elasticsearch/core/CriteriaQueryMappingUnitTests.java b/src/test/java/org/springframework/data/elasticsearch/core/CriteriaQueryMappingUnitTests.java
index 294e99866..cb63c81ee 100644
--- a/src/test/java/org/springframework/data/elasticsearch/core/CriteriaQueryMappingUnitTests.java
+++ b/src/test/java/org/springframework/data/elasticsearch/core/CriteriaQueryMappingUnitTests.java
@@ -196,7 +196,7 @@ void shouldMapNamesAndConvertValuesInCriteriaQueryForSubCriteriaWithDate() throw
 		CriteriaQuery criteriaQuery = new CriteriaQuery( //
 				Criteria.or().subCriteria(Criteria.where("birthDate") //
 						.between(LocalDate.of(1989, 11, 9), LocalDate.of(1990, 11, 9))) //
-						.subCriteria(Criteria.where("createdDate").is(383745721653L)) //
+						.subCriteria(Criteria.where("createdDate").is(new Date(383745721653L))) //
 		);
 
 		// mapped field name and converted parameter
diff --git a/src/test/java/org/springframework/data/elasticsearch/core/RangeTests.java b/src/test/java/org/springframework/data/elasticsearch/core/RangeTests.java
new file mode 100644
index 000000000..8d631f08b
--- /dev/null
+++ b/src/test/java/org/springframework/data/elasticsearch/core/RangeTests.java
@@ -0,0 +1,177 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.time.LocalDate;
+import java.util.Arrays;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Sascha Woo
+ * @since 4.3
+ */
+public class RangeTests {
+
+	@Test
+	public void shouldContainsLocalDate() {
+
+		// given
+		// when
+		// then
+		assertThat(Range.open(LocalDate.of(2021, 1, 1), LocalDate.of(2021, 2, 1)).contains(LocalDate.of(2021, 1, 10)))
+				.isTrue();
+	}
+
+	@Test
+	public void shouldEqualToSameRange() {
+
+		// given
+		Range<LocalDate> range1 = Range.open(LocalDate.of(2021, 1, 1), LocalDate.of(2021, 2, 1));
+		Range<LocalDate> range2 = Range.open(LocalDate.of(2021, 1, 1), LocalDate.of(2021, 2, 1));
+		// when
+		// then
+		assertThat(range1).isEqualTo(range2);
+	}
+
+	@Test
+	public void shouldHaveClosedBoundaries() {
+
+		// given
+		Range<Integer> range = Range.closed(1, 3);
+		// when
+		// then
+		assertThat(range.contains(1)).isTrue();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isTrue();
+	}
+
+	@Test
+	public void shouldHaveJustOneValue() {
+
+		// given
+		Range<Integer> range = Range.just(2);
+		// when
+		// then
+		assertThat(range.contains(1)).isFalse();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isFalse();
+	}
+
+	@Test
+	public void shouldHaveLeftOpenBoundary() {
+
+		// given
+		Range<Integer> range = Range.leftOpen(1, 3);
+		// when
+		// then
+		assertThat(range.contains(1)).isFalse();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isTrue();
+	}
+
+	@Test
+	public void shouldHaveLeftUnboundedAndRightExclusive() {
+
+		// given
+		Range<Integer> range = Range.leftUnbounded(Range.Bound.exclusive(3));
+		// when
+		// then
+		assertThat(range.contains(0)).isTrue();
+		assertThat(range.contains(1)).isTrue();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isFalse();
+	}
+
+	@Test
+	public void shouldHaveLeftUnboundedAndRightInclusive() {
+
+		// given
+		Range<Integer> range = Range.leftUnbounded(Range.Bound.inclusive(3));
+		// when
+		// then
+		assertThat(range.contains(0)).isTrue();
+		assertThat(range.contains(1)).isTrue();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isTrue();
+	}
+
+	@Test
+	public void shouldHaveOpenBoundaries() {
+
+		// given
+		Range<Integer> range = Range.open(1, 3);
+		// when
+		// then
+		assertThat(range.contains(1)).isFalse();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isFalse();
+	}
+
+	@Test
+	public void shouldHaveRightOpenBoundary() {
+
+		// given
+		Range<Integer> range = Range.rightOpen(1, 3);
+		// when
+		// then
+		assertThat(range.contains(1)).isTrue();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isFalse();
+	}
+
+	@Test
+	public void shouldHaveRightUnboundedAndLeftExclusive() {
+
+		// given
+		Range<Integer> range = Range.rightUnbounded(Range.Bound.exclusive(1));
+		// when
+		// then
+		assertThat(range.contains(1)).isFalse();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isTrue();
+		assertThat(range.contains(4)).isTrue();
+	}
+
+	@Test
+	public void shouldHaveRightUnboundedAndLeftInclusive() {
+
+		// given
+		Range<Integer> range = Range.rightUnbounded(Range.Bound.inclusive(1));
+		// when
+		// then
+		assertThat(range.contains(1)).isTrue();
+		assertThat(range.contains(2)).isTrue();
+		assertThat(range.contains(3)).isTrue();
+		assertThat(range.contains(4)).isTrue();
+	}
+
+	@Test
+	public void shouldThrowExceptionIfNotComparable() {
+
+		// given
+		// when
+		Throwable thrown = catchThrowable(() -> {
+			Range.just(Arrays.asList("test"));
+		});
+		// then
+		assertThat(thrown).isInstanceOf(IllegalArgumentException.class)
+				.hasMessageContaining("value must implements Comparable!");
+	}
+
+}
diff --git a/src/test/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverterUnitTests.java b/src/test/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverterUnitTests.java
index 1d5e456eb..4e0a469d4 100644
--- a/src/test/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverterUnitTests.java
+++ b/src/test/java/org/springframework/data/elasticsearch/core/convert/MappingElasticsearchConverterUnitTests.java
@@ -20,9 +20,15 @@
 import static org.skyscreamer.jsonassert.JSONAssert.*;
 
 import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetTime;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -46,6 +52,7 @@
 import org.springframework.data.elasticsearch.annotations.Field;
 import org.springframework.data.elasticsearch.annotations.FieldType;
 import org.springframework.data.elasticsearch.annotations.GeoPointField;
+import org.springframework.data.elasticsearch.core.Range;
 import org.springframework.data.elasticsearch.core.document.Document;
 import org.springframework.data.elasticsearch.core.geo.GeoJsonEntity;
 import org.springframework.data.elasticsearch.core.geo.GeoJsonGeometryCollection;
@@ -882,6 +889,200 @@ void shouldWriteNullValueIfConfigured() throws JSONException {
 		assertEquals(expected, document.toJson(), false);
 	}
 
+	@Nested
+	class RangeTests {
+
+		static final String JSON = "{"
+				+ "\"_class\":\"org.springframework.data.elasticsearch.core.convert.MappingElasticsearchConverterUnitTests$RangeTests$RangeEntity\","
+				+ "\"integerRange\":{\"gt\":\"1\",\"lt\":\"10\"}," //
+				+ "\"floatRange\":{\"gte\":\"1.2\",\"lte\":\"2.5\"}," //
+				+ "\"longRange\":{\"gt\":\"2\",\"lte\":\"5\"}," //
+				+ "\"doubleRange\":{\"gte\":\"3.2\",\"lt\":\"7.4\"}," //
+				+ "\"dateRange\":{\"gte\":\"1970-01-01T00:00:00.000Z\",\"lte\":\"1970-01-01T01:00:00.000Z\"}," //
+				+ "\"localDateRange\":{\"gte\":\"2021-07-06\"}," //
+				+ "\"localTimeRange\":{\"gte\":\"00:30:00.000\",\"lt\":\"02:30:00.000\"}," //
+				+ "\"localDateTimeRange\":{\"gt\":\"2021-01-01T00:30:00.000\",\"lt\":\"2021-01-01T02:30:00.000\"}," //
+				+ "\"offsetTimeRange\":{\"gte\":\"00:30:00.000+02:00\",\"lt\":\"02:30:00.000+02:00\"}," //
+				+ "\"zonedDateTimeRange\":{\"gte\":\"2021-01-01T00:30:00.000+02:00\",\"lte\":\"2021-01-01T00:30:00.000+02:00\"}," //
+				+ "\"nullRange\":null}";
+
+		@Test
+		public void shouldReadRanges() throws JSONException {
+
+			// given
+			Document source = Document.parse(JSON);
+
+			// when
+			RangeEntity entity = mappingElasticsearchConverter.read(RangeEntity.class, source);
+
+			// then
+			assertThat(entity) //
+					.isNotNull() //
+					.satisfies(e -> {
+						assertThat(e.getIntegerRange()).isEqualTo(Range.open(1, 10));
+						assertThat(e.getFloatRange()).isEqualTo(Range.closed(1.2f, 2.5f));
+						assertThat(e.getLongRange()).isEqualTo(Range.leftOpen(2l, 5l));
+						assertThat(e.getDoubleRange()).isEqualTo(Range.rightOpen(3.2d, 7.4d));
+						assertThat(e.getDateRange()).isEqualTo(Range.closed(new Date(0), new Date(60 * 60 * 1000)));
+						assertThat(e.getLocalDateRange())
+								.isEqualTo(Range.rightUnbounded(Range.Bound.inclusive(LocalDate.of(2021, 7, 6))));
+						assertThat(e.getLocalTimeRange()).isEqualTo(Range.rightOpen(LocalTime.of(0, 30), LocalTime.of(2, 30)));
+						assertThat(e.getLocalDateTimeRange())
+								.isEqualTo(Range.open(LocalDateTime.of(2021, 1, 1, 0, 30), LocalDateTime.of(2021, 1, 1, 2, 30)));
+						assertThat(e.getOffsetTimeRange())
+								.isEqualTo(Range.rightOpen(OffsetTime.of(LocalTime.of(0, 30), ZoneOffset.ofHours(2)),
+										OffsetTime.of(LocalTime.of(2, 30), ZoneOffset.ofHours(2))));
+						assertThat(e.getZonedDateTimeRange()).isEqualTo(
+								Range.just(ZonedDateTime.of(LocalDate.of(2021, 1, 1), LocalTime.of(0, 30), ZoneOffset.ofHours(2))));
+						assertThat(e.getNullRange()).isNull();
+					});
+		}
+
+		@Test
+		public void shouldWriteRanges() throws JSONException {
+
+			// given
+			Document source = Document.parse(JSON);
+			RangeEntity entity = new RangeEntity();
+			entity.setIntegerRange(Range.open(1, 10));
+			entity.setFloatRange(Range.closed(1.2f, 2.5f));
+			entity.setLongRange(Range.leftOpen(2l, 5l));
+			entity.setDoubleRange(Range.rightOpen(3.2d, 7.4d));
+			entity.setDateRange(Range.closed(new Date(0), new Date(60 * 60 * 1000)));
+			entity.setLocalDateRange(Range.rightUnbounded(Range.Bound.inclusive(LocalDate.of(2021, 7, 6))));
+			entity.setLocalTimeRange(Range.rightOpen(LocalTime.of(0, 30), LocalTime.of(2, 30)));
+			entity
+					.setLocalDateTimeRange(Range.open(LocalDateTime.of(2021, 1, 1, 0, 30), LocalDateTime.of(2021, 1, 1, 2, 30)));
+			entity.setOffsetTimeRange(Range.rightOpen(OffsetTime.of(LocalTime.of(0, 30), ZoneOffset.ofHours(2)),
+					OffsetTime.of(LocalTime.of(2, 30), ZoneOffset.ofHours(2))));
+			entity.setZonedDateTimeRange(
+					Range.just(ZonedDateTime.of(LocalDate.of(2021, 1, 1), LocalTime.of(0, 30), ZoneOffset.ofHours(2))));
+			entity.setNullRange(null);
+
+			// when
+			Document document = mappingElasticsearchConverter.mapObject(entity);
+
+			// then
+			assertThat(document).isEqualTo(source);
+		}
+
+		@org.springframework.data.elasticsearch.annotations.Document(indexName = "test-index-range-entity-mapper")
+		class RangeEntity {
+
+			@Id private String id;
+			@Field(type = FieldType.Integer_Range) private Range<Integer> integerRange;
+			@Field(type = FieldType.Float_Range) private Range<Float> floatRange;
+			@Field(type = FieldType.Long_Range) private Range<Long> longRange;
+			@Field(type = FieldType.Double_Range) private Range<Double> doubleRange;
+			@Field(type = FieldType.Date_Range) private Range<Date> dateRange;
+			@Field(type = FieldType.Date_Range, format = DateFormat.year_month_day) private Range<LocalDate> localDateRange;
+			@Field(type = FieldType.Date_Range,
+					format = DateFormat.hour_minute_second_millis) private Range<LocalTime> localTimeRange;
+			@Field(type = FieldType.Date_Range,
+					format = DateFormat.date_hour_minute_second_millis) private Range<LocalDateTime> localDateTimeRange;
+			@Field(type = FieldType.Date_Range, format = DateFormat.time) private Range<OffsetTime> offsetTimeRange;
+			@Field(type = FieldType.Date_Range) private Range<ZonedDateTime> zonedDateTimeRange;
+			@Field(type = FieldType.Date_Range, storeNullValue = true) private Range<ZonedDateTime> nullRange;
+
+			public String getId() {
+				return id;
+			}
+
+			public Range<Integer> getIntegerRange() {
+				return integerRange;
+			}
+
+			public Range<Float> getFloatRange() {
+				return floatRange;
+			}
+
+			public Range<Long> getLongRange() {
+				return longRange;
+			}
+
+			public Range<Double> getDoubleRange() {
+				return doubleRange;
+			}
+
+			public Range<Date> getDateRange() {
+				return dateRange;
+			}
+
+			public Range<LocalDate> getLocalDateRange() {
+				return localDateRange;
+			}
+
+			public Range<LocalTime> getLocalTimeRange() {
+				return localTimeRange;
+			}
+
+			public Range<LocalDateTime> getLocalDateTimeRange() {
+				return localDateTimeRange;
+			}
+
+			public Range<OffsetTime> getOffsetTimeRange() {
+				return offsetTimeRange;
+			}
+
+			public Range<ZonedDateTime> getZonedDateTimeRange() {
+				return zonedDateTimeRange;
+			}
+
+			public Range<ZonedDateTime> getNullRange() {
+				return nullRange;
+			}
+
+			public void setId(String id) {
+				this.id = id;
+			}
+
+			public void setIntegerRange(Range<Integer> integerRange) {
+				this.integerRange = integerRange;
+			}
+
+			public void setFloatRange(Range<Float> floatRange) {
+				this.floatRange = floatRange;
+			}
+
+			public void setLongRange(Range<Long> longRange) {
+				this.longRange = longRange;
+			}
+
+			public void setDoubleRange(Range<Double> doubleRange) {
+				this.doubleRange = doubleRange;
+			}
+
+			public void setDateRange(Range<Date> dateRange) {
+				this.dateRange = dateRange;
+			}
+
+			public void setLocalDateRange(Range<LocalDate> localDateRange) {
+				this.localDateRange = localDateRange;
+			}
+
+			public void setLocalTimeRange(Range<LocalTime> localTimeRange) {
+				this.localTimeRange = localTimeRange;
+			}
+
+			public void setLocalDateTimeRange(Range<LocalDateTime> localDateTimeRange) {
+				this.localDateTimeRange = localDateTimeRange;
+			}
+
+			public void setOffsetTimeRange(Range<OffsetTime> offsetTimeRange) {
+				this.offsetTimeRange = offsetTimeRange;
+			}
+
+			public void setZonedDateTimeRange(Range<ZonedDateTime> zonedDateTimeRange) {
+				this.zonedDateTimeRange = zonedDateTimeRange;
+			}
+
+			public void setNullRange(Range<ZonedDateTime> nullRange) {
+				this.nullRange = nullRange;
+			}
+
+		}
+	}
+
 	@Nested
 	class GeoJsonUnitTests {
 		private GeoJsonEntity entity;
@@ -2074,7 +2275,8 @@ public void setSaved(@Nullable String saved) {
 		}
 	}
 
-	private static class ElectricCar extends Car {}
+	private static class ElectricCar extends Car {
+	}
 
 	private static class PersonWithCars {
 		@Id @Nullable String id;
diff --git a/src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentPropertyUnitTests.java b/src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentPropertyUnitTests.java
index 8ac934880..fcf02f7c5 100644
--- a/src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentPropertyUnitTests.java
+++ b/src/test/java/org/springframework/data/elasticsearch/core/mapping/SimpleElasticsearchPersistentPropertyUnitTests.java
@@ -114,7 +114,7 @@ void shouldConvertFromLocalDate() {
 		ElasticsearchPersistentProperty persistentProperty = persistentEntity.getRequiredPersistentProperty("localDate");
 		LocalDate localDate = LocalDate.of(2019, 12, 27);
 
-		String converted = persistentProperty.getPropertyConverter().write(localDate);
+		String converted = persistentProperty.getPropertyConverter().write(localDate).toString();
 
 		assertThat(converted).isEqualTo("27.12.2019");
 	}
@@ -138,7 +138,7 @@ void shouldConvertFromLegacyDate() {
 				.from(ZonedDateTime.of(LocalDateTime.of(2020, 4, 19, 19, 44), ZoneId.of("UTC")));
 		Date legacyDate = calendar.getTime();
 
-		String converted = persistentProperty.getPropertyConverter().write(legacyDate);
+		String converted = persistentProperty.getPropertyConverter().write(legacyDate).toString();
 
 		assertThat(converted).isEqualTo("20200419T194400.000Z");
 	}