diff --git a/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java b/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
index e7a38811..3fea4c4c 100644
--- a/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
+++ b/src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java
@@ -152,6 +152,10 @@ public class CouchbaseTemplate implements CouchbaseOperations, ApplicationContex
 		return new ExecutableRemoveByQueryOperationSupport(this).removeByQuery(domainType);
 	}
 
+	@Override
+	public <T> ExecutableRangeScan<T> rangeScan(Class<T> domainType) {
+		return new ExecutableRangeScanOperationSupport(this).rangeScan(domainType);
+	}
 	@Override
 	public String getBucketName() {
 		return clientFactory.getBucket().name();
diff --git a/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperation.java b/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperation.java
new file mode 100644
index 00000000..576755ca
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperation.java
@@ -0,0 +1,248 @@
+/*
+ * Copyright 2012-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.couchbase.core;
+
+import java.util.stream.Stream;
+
+import org.springframework.data.couchbase.core.support.ConsistentWith;
+import org.springframework.data.couchbase.core.support.InCollection;
+import org.springframework.data.couchbase.core.support.InScope;
+import org.springframework.data.couchbase.core.support.WithBatchByteLimit;
+import org.springframework.data.couchbase.core.support.WithBatchItemLimit;
+import org.springframework.data.couchbase.core.support.IdsOnly;
+import org.springframework.data.couchbase.core.support.WithLimit;
+import org.springframework.data.couchbase.core.support.WithSampling;
+import org.springframework.data.couchbase.core.support.WithScanOptions;
+import org.springframework.data.couchbase.core.support.WithScanSort;
+
+import com.couchbase.client.java.kv.MutationState;
+import com.couchbase.client.java.kv.ScanOptions;
+import com.couchbase.client.java.kv.ScanSort;
+import org.springframework.data.couchbase.core.support.WithSeed;
+
+/**
+ * Get Operations
+ *
+ * @author Michael Reiche
+ */
+public interface ExecutableRangeScanOperation {
+
+	/**
+	 * Loads a document from a bucket.
+	 *
+	 * @param domainType the entity type to use for the results.
+	 */
+	<T> ExecutableRangeScan<T> rangeScan(Class<T> domainType);
+
+	/**
+	 * Terminating operations invoking the actual execution.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface TerminatingRangeScan<T> /*extends OneAndAllId<T>*/ {
+
+		/**
+		 * Range Scan
+		 *
+		 * @param upper
+		 * @param lower
+		 * @return the list of found entities.
+		 */
+		Stream<T> rangeScan(String lower, String upper);
+
+		/**
+		 * Range Scan Ids
+		 *
+		 * @param upper
+		 * @param lower
+		 * @return the list of found entities.
+		 */
+		Stream<String> rangeScanIds(String lower, String upper);
+
+	}
+
+	/**
+	 * Fluent method to specify options.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanWithOptions<T> extends TerminatingRangeScan<T>, WithScanOptions<T> {
+		/**
+		 * Fluent method to specify options to use for execution
+		 *
+		 * @param options options to use for execution
+		 */
+		@Override
+		TerminatingRangeScan<T> withOptions(ScanOptions options);
+	}
+
+	/**
+	 * Fluent method to specify the collection.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanInCollection<T> extends RangeScanWithOptions<T>, InCollection<T> {
+		/**
+		 * With a different collection
+		 *
+		 * @param collection the collection to use.
+		 */
+		@Override
+		RangeScanWithOptions<T> inCollection(String collection);
+	}
+
+	/**
+	 * Fluent method to specify the scope.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanInScope<T> extends RangeScanInCollection<T>, InScope<T> {
+		/**
+		 * With a different scope
+		 *
+		 * @param scope the scope to use.
+		 */
+		@Override
+		RangeScanInCollection<T> inScope(String scope);
+	}
+
+	interface RangeScanWithSampling<T> extends RangeScanInScope<T>, WithSampling<T> {
+		/**
+		 * sampling
+		 *
+		 * @param isSampling
+		 */
+		@Override
+		RangeScanInScope<T> withSampling(Boolean isSampling);
+	}
+
+	interface RangeScanWithSort<T> extends RangeScanWithSampling<T>, WithScanSort<T> {
+		/**
+		 * sort
+		 *
+		 * @param sort
+		 */
+		@Override
+		RangeScanWithSampling<T> withSort(ScanSort sort);
+	}
+
+	/**
+	 * Fluent method to specify scan consistency. Scan consistency may also come from an annotation.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanConsistentWith<T> extends RangeScanWithSort<T>, ConsistentWith<T> {
+
+		/**
+		 * Allows to override the default scan consistency.
+		 *
+		 * @param mutationState the custom scan consistency to use for this query.
+		 */
+		@Override
+		RangeScanWithSort<T> consistentWith(MutationState mutationState);
+	}
+
+	/**
+	 * Fluent method to specify a return type different than the the entity type to use for the results.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanWithProjection<T> extends RangeScanConsistentWith<T> {
+
+		/**
+		 * Define the target type fields should be mapped to. <br />
+		 * Skip this step if you are only interested in the original the entity type to use for the results.
+		 *
+		 * @param returnType must not be {@literal null}.
+		 * @return new instance of {@link ExecutableFindByQueryOperation.FindByQueryWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		<R> RangeScanConsistentWith<R> as(Class<R> returnType);
+	}
+
+	interface RangeScanIdsOnly<T> extends RangeScanWithProjection<T>, IdsOnly<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param idsOnly must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithProjection<T> idsOnly(Boolean idsOnly);
+	}
+
+	interface RangeScanWithLimit<T> extends RangeScanIdsOnly<T>, WithLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param limit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanIdsOnly<T> withLimit(Long limit);
+	}
+
+	interface RangeScanWithSeed<T> extends RangeScanWithLimit<T>, WithSeed<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param seed must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithLimit<T> withSeed(Long seed);
+	}
+
+	interface RangeScanWithBatchItemLimit<T> extends RangeScanWithSeed<T>, WithBatchItemLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param batchByteLimit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithSeed<T> withBatchItemLimit(Integer batchByteLimit);
+	}
+
+	interface RangeScanWithBatchByteLimit<T> extends RangeScanWithBatchItemLimit<T>, WithBatchByteLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param batchByteLimit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithBatchByteLimit<T> withBatchByteLimit(Integer batchByteLimit);
+	}
+
+	/**
+	 * Provides methods for constructing query operations in a fluent way.
+	 *
+	 * @param <T> the entity type to use for the results
+	 */
+	interface ExecutableRangeScan<T> extends RangeScanWithBatchByteLimit<T> {}
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperationSupport.java
new file mode 100644
index 00000000..3e343898
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/ExecutableRangeScanOperationSupport.java
@@ -0,0 +1,166 @@
+/*
+ * Copyright 2012-2022 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.couchbase.core;
+
+import java.util.stream.Stream;
+
+import org.springframework.data.couchbase.core.ReactiveRangeScanOperationSupport.ReactiveRangeScanSupport;
+import org.springframework.data.couchbase.core.query.OptionsBuilder;
+import org.springframework.util.Assert;
+
+import com.couchbase.client.java.kv.MutationState;
+import com.couchbase.client.java.kv.ScanOptions;
+import com.couchbase.client.java.kv.ScanSort;
+
+public class ExecutableRangeScanOperationSupport implements ExecutableRangeScanOperation {
+
+	private final CouchbaseTemplate template;
+
+	ExecutableRangeScanOperationSupport(CouchbaseTemplate template) {
+		this.template = template;
+	}
+
+	@Override
+	public <T> ExecutableRangeScan<T> rangeScan(Class<T> domainType) {
+		return new ExecutableRangeScanSupport<>(template, domainType, OptionsBuilder.getScopeFrom(domainType),
+				OptionsBuilder.getCollectionFrom(domainType), null, null, null, null, null, null, null, null, null);
+	}
+
+	static class ExecutableRangeScanSupport<T> implements ExecutableRangeScan<T> {
+
+		private final CouchbaseTemplate template;
+		private final Class<T> domainType;
+		private final String scope;
+		private final String collection;
+		private final ScanOptions options;
+		private final Boolean isSamplingScan;
+		private final ScanSort sort;
+		private final MutationState mutationState;
+		private final Boolean withContent;
+		private final Long limit;
+		private final Long seed;
+		private final Integer batchItemLimit;
+		private final Integer batchByteLimit;
+		private final ReactiveRangeScanSupport<T> reactiveSupport;
+
+		ExecutableRangeScanSupport(CouchbaseTemplate template, Class<T> domainType, String scope, String collection,
+				ScanOptions options, Boolean isSamplingScan, ScanSort sort, MutationState mutationState, Boolean withContent,
+				Long seed, Long limit, Integer batchItemLimit, Integer batchByteLimit) {
+			this.template = template;
+			this.domainType = domainType;
+			this.scope = scope;
+			this.collection = collection;
+			this.options = options;
+			this.isSamplingScan = isSamplingScan;
+			this.sort = sort;
+			this.mutationState = mutationState;
+			this.withContent = withContent;
+			this.limit = limit;
+			this.seed = seed;
+			this.batchItemLimit = batchItemLimit;
+			this.batchByteLimit = batchByteLimit;
+			this.reactiveSupport = new ReactiveRangeScanSupport<>(template.reactive(), domainType, scope, collection, options,
+					isSamplingScan, sort, mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit,
+					new NonReactiveSupportWrapper(template.support()));
+		}
+
+		@Override
+		public TerminatingRangeScan<T> withOptions(final ScanOptions options) {
+			Assert.notNull(options, "Options must not be null.");
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithOptions<T> inCollection(final String collection) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope,
+					collection != null ? collection : this.collection, options, isSamplingScan, sort, mutationState, withContent,
+					limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanInCollection<T> inScope(final String scope) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
+					options, isSamplingScan, sort, mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanInScope<T> withSampling(Boolean isSamplingScan) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithSampling<T> withSort(ScanSort sort) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithSort<T> consistentWith(MutationState mutationState) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public <R> RangeScanConsistentWith<R> as(Class<R> returnType) {
+			return new ExecutableRangeScanSupport<>(template, returnType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithProjection<T> idsOnly(Boolean withContent) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanIdsOnly<T> withLimit(Long limit) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+				mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithLimit<T> withSeed(Long seed) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+				mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithSeed<T> withBatchItemLimit(Integer batchItemLimit) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public RangeScanWithBatchByteLimit<T> withBatchByteLimit(Integer batchByteLimit) {
+			return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, withContent, limit, seed, batchItemLimit, batchByteLimit);
+		}
+
+		@Override
+		public Stream<T> rangeScan(String lower, String upper) {
+			return reactiveSupport.rangeScan(lower, upper).toStream();
+		}
+
+		@Override
+		public Stream<String> rangeScanIds(String lower, String upper) {
+			return reactiveSupport.rangeScanIds(lower, upper).toStream();
+		}
+
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/FluentCouchbaseOperations.java b/src/main/java/org/springframework/data/couchbase/core/FluentCouchbaseOperations.java
index 4a4da4de..b67a7a70 100644
--- a/src/main/java/org/springframework/data/couchbase/core/FluentCouchbaseOperations.java
+++ b/src/main/java/org/springframework/data/couchbase/core/FluentCouchbaseOperations.java
@@ -22,4 +22,4 @@ package org.springframework.data.couchbase.core;
 public interface FluentCouchbaseOperations extends ExecutableUpsertByIdOperation, ExecutableInsertByIdOperation,
 		ExecutableReplaceByIdOperation, ExecutableFindByIdOperation, ExecutableFindFromReplicasByIdOperation,
 		ExecutableFindByQueryOperation, ExecutableFindByAnalyticsOperation, ExecutableExistsByIdOperation,
-		ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation {}
+		ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation, ExecutableRangeScanOperation {}
diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java
index 9cbbe31c..593d67a1 100644
--- a/src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java
+++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java
@@ -183,6 +183,12 @@ public class ReactiveCouchbaseTemplate implements ReactiveCouchbaseOperations, A
 		return new ReactiveUpsertByIdOperationSupport(this).upsertById(domainType);
 	}
 
+	@Override
+	public <T> ReactiveRangeScan<T> rangeScan(Class<T> domainType) {
+		return new ReactiveRangeScanOperationSupport(this).rangeScan(domainType);
+	}
+
+
 	@Override
 	public String getBucketName() {
 		return clientFactory.getBucket().name();
diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveFluentCouchbaseOperations.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveFluentCouchbaseOperations.java
index 0ff0a8b8..bcfd6e0e 100644
--- a/src/main/java/org/springframework/data/couchbase/core/ReactiveFluentCouchbaseOperations.java
+++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveFluentCouchbaseOperations.java
@@ -22,4 +22,4 @@ package org.springframework.data.couchbase.core;
 public interface ReactiveFluentCouchbaseOperations extends ReactiveUpsertByIdOperation, ReactiveInsertByIdOperation,
 		ReactiveReplaceByIdOperation, ReactiveFindByIdOperation, ReactiveExistsByIdOperation,
 		ReactiveFindByAnalyticsOperation, ReactiveFindFromReplicasByIdOperation, ReactiveFindByQueryOperation,
-		ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation {}
+		ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation, ReactiveRangeScanOperation {}
diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperation.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperation.java
new file mode 100644
index 00000000..fd626383
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperation.java
@@ -0,0 +1,248 @@
+/*
+ * Copyright 2012-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.couchbase.core;
+
+import org.springframework.data.couchbase.core.support.WithLimit;
+import org.springframework.data.couchbase.core.support.WithSampling;
+import org.springframework.data.couchbase.core.support.WithSeed;
+import reactor.core.publisher.Flux;
+
+import org.springframework.data.couchbase.core.support.ConsistentWith;
+import org.springframework.data.couchbase.core.support.InCollection;
+import org.springframework.data.couchbase.core.support.InScope;
+import org.springframework.data.couchbase.core.support.WithBatchByteLimit;
+import org.springframework.data.couchbase.core.support.WithBatchItemLimit;
+import org.springframework.data.couchbase.core.support.IdsOnly;
+import org.springframework.data.couchbase.core.support.WithScanOptions;
+import org.springframework.data.couchbase.core.support.WithScanSort;
+
+import com.couchbase.client.java.kv.MutationState;
+import com.couchbase.client.java.kv.ScanOptions;
+import com.couchbase.client.java.kv.ScanSort;
+
+/**
+ * Get Operations
+ *
+ * @author Christoph Strobl
+ * @since 2.0
+ */
+public interface ReactiveRangeScanOperation {
+
+	/**
+	 * Loads a document from a bucket.
+	 *
+	 * @param domainType the entity type to use for the results.
+	 */
+	<T> ReactiveRangeScan<T> rangeScan(Class<T> domainType);
+
+	/**
+	 * Terminating operations invoking the actual execution.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface TerminatingRangeScan<T> /*extends OneAndAllId<T>*/ {
+
+		/**
+		 * Finds a list of documents based on the given IDs.
+		 *
+		 * @param lower the lower bound
+		 * @param upper the upper bound
+		 * @return the list of found entities.
+		 */
+		Flux<T> rangeScan(String lower, String upper);
+
+		/**
+		 * Finds a list of documents based on the given IDs.
+		 *
+		 * @param lower the lower bound
+		 * @param upper the upper bound
+		 * @return the list of found entities.
+		 */
+		Flux<String> rangeScanIds(String lower, String upper);
+	}
+
+	/**
+	 * Fluent method to specify options.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanWithOptions<T> extends TerminatingRangeScan<T>, WithScanOptions<T> {
+		/**
+		 * Fluent method to specify options to use for execution
+		 *
+		 * @param options options to use for execution
+		 */
+		@Override
+		TerminatingRangeScan<T> withOptions(ScanOptions options);
+	}
+
+	/**
+	 * Fluent method to specify the collection.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanInCollection<T> extends RangeScanWithOptions<T>, InCollection<T> {
+		/**
+		 * With a different collection
+		 *
+		 * @param collection the collection to use.
+		 */
+		@Override
+		RangeScanWithOptions<T> inCollection(String collection);
+	}
+
+	/**
+	 * Fluent method to specify the scope.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanInScope<T> extends RangeScanInCollection<T>, InScope<T> {
+		/**
+		 * With a different scope
+		 *
+		 * @param scope the scope to use.
+		 */
+		@Override
+		RangeScanInCollection<T> inScope(String scope);
+	}
+
+	interface RangeScanWithSampling<T> extends RangeScanInScope<T>, WithSampling<T> {
+		/**
+		 * sampling
+		 *
+		 * @param isSampling
+		 */
+		@Override
+		RangeScanInScope<T> withSampling(Boolean isSampling);
+	}
+
+	interface RangeScanWithSort<T> extends RangeScanWithSampling<T>, WithScanSort<T> {
+		/**
+		 * sort
+		 *
+		 * @param sort
+		 */
+		@Override
+		RangeScanWithSampling<T> withSort(ScanSort sort);
+	}
+
+	/**
+	 * Fluent method to specify scan consistency. Scan consistency may also come from an annotation.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanConsistentWith<T> extends RangeScanWithSort<T>, ConsistentWith<T> {
+
+		/**
+		 * Allows to override the default scan consistency.
+		 *
+		 * @param mutationState the custom scan consistency to use for this query.
+		 */
+		@Override
+		RangeScanWithSort<T> consistentWith(MutationState mutationState);
+	}
+
+	/**
+	 * Fluent method to specify a return type different than the the entity type to use for the results.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface RangeScanWithProjection<T> extends RangeScanConsistentWith<T> {
+
+		/**
+		 * Define the target type fields should be mapped to. <br />
+		 * Skip this step if you are only interested in the original the entity type to use for the results.
+		 *
+		 * @param returnType must not be {@literal null}.
+		 * @return new instance of {@link ReactiveFindByQueryOperation.FindByQueryWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		<R> RangeScanConsistentWith<R> as(Class<R> returnType);
+	}
+
+	interface RangeScanIdsOnly<T> extends RangeScanWithProjection<T>, IdsOnly<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param idsOnly must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithProjection<T> idsOnly(Boolean idsOnly);
+	}
+
+	interface RangeScanWithLimit<T> extends RangeScanIdsOnly<T>, WithLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param limit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanIdsOnly<T> withLimit(Long limit);
+	}
+
+	interface RangeScanWithSeed<T> extends RangeScanWithLimit<T>, WithSeed<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param seed must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithLimit<T> withSeed(Long seed);
+	}
+
+	interface RangeScanWithBatchItemLimit<T> extends RangeScanWithSeed<T>, WithBatchItemLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param batchByteLimit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithSeed<T> withBatchItemLimit(Integer batchByteLimit);
+	}
+
+	interface RangeScanWithBatchByteLimit<T> extends RangeScanWithBatchItemLimit<T>, WithBatchByteLimit<T> {
+
+		/**
+		 * determines if result are just ids or ids plus contents
+		 *
+		 * @param batchByteLimit must not be {@literal null}.
+		 * @return new instance of {@link RangeScanWithProjection}.
+		 * @throws IllegalArgumentException if returnType is {@literal null}.
+		 */
+		@Override
+		RangeScanWithBatchByteLimit<T> withBatchByteLimit(Integer batchByteLimit);
+	}
+
+	/**
+	 * Provides methods for constructing query operations in a fluent way.
+	 *
+	 * @param <T> the entity type to use for the results
+	 */
+	interface ReactiveRangeScan<T> extends RangeScanWithBatchByteLimit<T> {}
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperationSupport.java b/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperationSupport.java
new file mode 100644
index 00000000..112e9721
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/ReactiveRangeScanOperationSupport.java
@@ -0,0 +1,236 @@
+/*
+ * Copyright 2012-2022 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.couchbase.core;
+
+import reactor.core.publisher.Flux;
+
+import java.nio.charset.StandardCharsets;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.query.OptionsBuilder;
+import org.springframework.data.couchbase.core.support.PseudoArgs;
+import org.springframework.util.Assert;
+
+import com.couchbase.client.java.ReactiveCollection;
+import com.couchbase.client.java.kv.MutationState;
+import com.couchbase.client.java.kv.ScanOptions;
+import com.couchbase.client.java.kv.ScanSort;
+import com.couchbase.client.java.kv.ScanTerm;
+import com.couchbase.client.java.kv.ScanType;
+
+public class ReactiveRangeScanOperationSupport implements ReactiveRangeScanOperation {
+
+	private final ReactiveCouchbaseTemplate template;
+	private static final Logger LOG = LoggerFactory.getLogger(ReactiveRangeScanOperationSupport.class);
+
+	ReactiveRangeScanOperationSupport(ReactiveCouchbaseTemplate template) {
+		this.template = template;
+	}
+
+	@Override
+	public <T> ReactiveRangeScan<T> rangeScan(Class<T> domainType) {
+		return new ReactiveRangeScanSupport<>(template, domainType, OptionsBuilder.getScopeFrom(domainType),
+				OptionsBuilder.getCollectionFrom(domainType), null, null, null, null, null, null, null, null, null,
+				template.support());
+	}
+
+	static class ReactiveRangeScanSupport<T> implements ReactiveRangeScan<T> {
+
+		private final ReactiveCouchbaseTemplate template;
+		private final Class<T> domainType;
+		private final String scope;
+		private final String collection;
+		private final ScanOptions options;
+		private final Boolean isSamplingScan;
+		private final ScanSort sort;
+		private final MutationState mutationState;
+		private final Boolean idsOnly;
+		private final Long limit;
+		private final Long seed;
+		private final Integer batchItemLimit;
+		private final Integer batchByteLimit;
+		private final ReactiveTemplateSupport support;
+
+		ReactiveRangeScanSupport(ReactiveCouchbaseTemplate template, Class<T> domainType, String scope, String collection,
+				ScanOptions options, Boolean isSamplingScan, ScanSort sort, MutationState mutationState, Boolean idsOnly,
+				Long limit, Long seed, Integer batchItemLimit, Integer batchByteLimit, ReactiveTemplateSupport support) {
+			this.template = template;
+			this.domainType = domainType;
+			this.scope = scope;
+			this.collection = collection;
+			this.isSamplingScan = isSamplingScan;
+			this.options = options;
+			this.sort = sort;
+			this.mutationState = mutationState;
+			this.idsOnly = idsOnly;
+			this.limit = limit;
+			this.seed = seed;
+			this.batchItemLimit = batchItemLimit;
+			this.batchByteLimit = batchByteLimit;
+			this.support = support;
+		}
+
+		@Override
+		public TerminatingRangeScan<T> withOptions(final ScanOptions options) {
+			Assert.notNull(options, "Options must not be null.");
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithOptions<T> inCollection(final String collection) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope,
+					collection != null ? collection : this.collection, options, isSamplingScan, sort, mutationState, idsOnly,
+					limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanInCollection<T> inScope(final String scope) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
+					options, isSamplingScan, sort, mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanInScope<T> withSampling(Boolean isSamplingScan) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithSampling<T> withSort(ScanSort sort) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithSort<T> consistentWith(MutationState mutationState) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public <R> RangeScanConsistentWith<R> as(Class<R> returnType) {
+			return new ReactiveRangeScanSupport<>(template, returnType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithProjection<T> idsOnly(Boolean idsOnly) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanIdsOnly<T> withLimit(Long limit) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithLimit<T> withSeed(Long seed) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithSeed<T> withBatchItemLimit(Integer batchItemLimit) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public RangeScanWithBatchByteLimit<T> withBatchByteLimit(Integer batchByteLimit) {
+			return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, isSamplingScan, sort,
+					mutationState, idsOnly, limit, seed, batchItemLimit, batchByteLimit, support);
+		}
+
+		@Override
+		public Flux<T> rangeScan(String lower, String upper) {
+
+			PseudoArgs<ScanOptions> pArgs = new PseudoArgs<>(template, scope, collection, options, domainType);
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("rangeScan lower={} upper={} {}", lower, upper, pArgs);
+			}
+			ReactiveCollection rc = template.getCouchbaseClientFactory().withScope(pArgs.getScope())
+					.getCollection(pArgs.getCollection()).reactive();
+
+			ScanTerm lowerTerm = ScanTerm.minimum();
+			ScanTerm upperTerm = ScanTerm.maximum();
+			if (lower != null) {
+				lowerTerm = ScanTerm.inclusive(lower);
+			}
+			if (upper != null) {
+				upperTerm = ScanTerm.inclusive(upper);
+			}
+
+			ScanType scanType = isSamplingScan ? ScanType.samplingScan(limit != null ? limit : 2, seed != null ? seed : 0)
+					: ScanType.rangeScan(lowerTerm, upperTerm);
+			Flux<T> reactiveEntities = TransactionalSupport.verifyNotInTransaction("rangeScan")
+					.thenMany(rc.scan(scanType, buildScanOptions(pArgs.getOptions(), idsOnly))
+							.flatMap(result -> support.decodeEntity(result.id(),
+									new String(result.contentAsBytes(), StandardCharsets.UTF_8), result.cas(), domainType,
+									pArgs.getScope(), pArgs.getCollection(), null, null)));
+
+			return reactiveEntities.onErrorMap(throwable -> {
+				if (throwable instanceof RuntimeException) {
+					return template.potentiallyConvertRuntimeException((RuntimeException) throwable);
+				} else {
+					return throwable;
+				}
+			});
+
+		}
+
+		@Override
+		public Flux<String> rangeScanIds(String lower, String upper) {
+			PseudoArgs<ScanOptions> pArgs = new PseudoArgs<>(template, scope, collection, options, domainType);
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("rangeScan lower={} upper={} {}", lower, upper, pArgs);
+			}
+			ReactiveCollection rc = template.getCouchbaseClientFactory().withScope(pArgs.getScope())
+					.getCollection(pArgs.getCollection()).reactive();
+
+			ScanTerm lowerTerm = ScanTerm.minimum();
+			ScanTerm upperTerm = ScanTerm.maximum();
+			if (lower != null) {
+				lowerTerm = ScanTerm.inclusive(lower);
+			}
+			if (upper != null) {
+				upperTerm = ScanTerm.inclusive(upper);
+			}
+
+			ScanType scanType = isSamplingScan ? ScanType.samplingScan(limit, seed)
+					: ScanType.rangeScan(lowerTerm, upperTerm);
+			Flux<String> reactiveEntities = TransactionalSupport.verifyNotInTransaction("rangeScanIds")
+					.thenMany(rc.scan(scanType, buildScanOptions(pArgs.getOptions(), true)).map(result -> result.id()));
+
+			return reactiveEntities.onErrorMap(throwable -> {
+				if (throwable instanceof RuntimeException) {
+					return template.potentiallyConvertRuntimeException((RuntimeException) throwable);
+				} else {
+					return throwable;
+				}
+			});
+
+		}
+
+		private ScanOptions buildScanOptions(ScanOptions options, Boolean idsOnly) {
+			return OptionsBuilder.buildScanOptions(options, sort, idsOnly, mutationState, batchByteLimit, batchItemLimit);
+		}
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/convert/AbstractCouchbaseConverter.java b/src/main/java/org/springframework/data/couchbase/core/convert/AbstractCouchbaseConverter.java
index df362f11..1efce4f5 100644
--- a/src/main/java/org/springframework/data/couchbase/core/convert/AbstractCouchbaseConverter.java
+++ b/src/main/java/org/springframework/data/couchbase/core/convert/AbstractCouchbaseConverter.java
@@ -117,10 +117,9 @@ public abstract class AbstractCouchbaseConverter implements CouchbaseConverter,
 			return null;
 		}
 		if (processValueConverter && conversions.hasValueConverter(prop)) {
-			CouchbaseDocument encrypted = (CouchbaseDocument) conversions.getPropertyValueConversions()
+			return conversions.getPropertyValueConversions()
 					.getValueConverter(prop)
 					.write(value, new CouchbaseConversionContext(prop, (MappingCouchbaseConverter) this, accessor));
-			return encrypted;
 		}
 		Class<?> targetClass = this.conversions.getCustomWriteTarget(value.getClass()).orElse(null);
 
diff --git a/src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseCustomConversions.java b/src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseCustomConversions.java
index b663bf9b..87dae22f 100644
--- a/src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseCustomConversions.java
+++ b/src/main/java/org/springframework/data/couchbase/core/convert/CouchbaseCustomConversions.java
@@ -31,6 +31,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.function.Consumer;
 
+import com.fasterxml.jackson.annotation.JsonValue;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.ConverterFactory;
 import org.springframework.core.convert.converter.GenericConverter;
@@ -112,9 +113,6 @@ public class CouchbaseCustomConversions extends org.springframework.data.convert
 
 	@Override
 	public boolean hasValueConverter(PersistentProperty<?> property) {
-		if (property.findAnnotation(Encrypted.class) != null) {
-			return true;
-		}
 		return super.hasValueConverter(property);
 	}
 
diff --git a/src/main/java/org/springframework/data/couchbase/core/query/OptionsBuilder.java b/src/main/java/org/springframework/data/couchbase/core/query/OptionsBuilder.java
index 6214bcbc..2eb6ad4d 100644
--- a/src/main/java/org/springframework/data/couchbase/core/query/OptionsBuilder.java
+++ b/src/main/java/org/springframework/data/couchbase/core/query/OptionsBuilder.java
@@ -43,10 +43,13 @@ import com.couchbase.client.java.json.JsonArray;
 import com.couchbase.client.java.json.JsonObject;
 import com.couchbase.client.java.kv.ExistsOptions;
 import com.couchbase.client.java.kv.InsertOptions;
+import com.couchbase.client.java.kv.MutationState;
 import com.couchbase.client.java.kv.PersistTo;
 import com.couchbase.client.java.kv.RemoveOptions;
 import com.couchbase.client.java.kv.ReplaceOptions;
 import com.couchbase.client.java.kv.ReplicateTo;
+import com.couchbase.client.java.kv.ScanOptions;
+import com.couchbase.client.java.kv.ScanSort;
 import com.couchbase.client.java.kv.UpsertOptions;
 import com.couchbase.client.java.query.QueryOptions;
 import com.couchbase.client.java.query.QueryScanConsistency;
@@ -457,4 +460,24 @@ public class OptionsBuilder {
 		return annotationString(annotation, "value", defaultValue, elements);
 	}
 
+	public static ScanOptions buildScanOptions(ScanOptions options, ScanSort sort, Boolean idsOnly,
+			MutationState mutationState, Integer batchByteLimit, Integer batchItemLimit) {
+		options = options != null ? options : ScanOptions.scanOptions();
+		if (sort != null) {
+			options.sort(sort);
+		}
+		if (idsOnly != null) {
+			options.idsOnly(idsOnly);
+		}
+		if (mutationState != null) {
+			options.consistentWith(mutationState);
+		}
+		if (batchByteLimit != null) {
+			options.batchByteLimit(batchByteLimit);
+		}
+		if (batchItemLimit != null) {
+			options.batchItemLimit(batchItemLimit);
+		}
+		return options;
+	}
 }
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/ConsistentWith.java b/src/main/java/org/springframework/data/couchbase/core/support/ConsistentWith.java
new file mode 100644
index 00000000..c1ea65ca
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/ConsistentWith.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+import com.couchbase.client.java.kv.MutationState;
+
+/**
+ * A common interface for those that support withOptions()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface ConsistentWith<R> {
+	Object consistentWith(MutationState mutationState);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/IdsOnly.java b/src/main/java/org/springframework/data/couchbase/core/support/IdsOnly.java
new file mode 100644
index 00000000..443aafc3
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/IdsOnly.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+/**
+ * A common interface for those that support withContent()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface IdsOnly<R> {
+	Object idsOnly(Boolean withContent);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithBatchByteLimit.java b/src/main/java/org/springframework/data/couchbase/core/support/WithBatchByteLimit.java
new file mode 100644
index 00000000..adde3ca9
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithBatchByteLimit.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+/**
+ * A common interface for those that support withBatchByteLimit()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithBatchByteLimit<R> {
+	Object withBatchByteLimit(Integer batchByteLimit);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithBatchItemLimit.java b/src/main/java/org/springframework/data/couchbase/core/support/WithBatchItemLimit.java
new file mode 100644
index 00000000..7d1b5ba8
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithBatchItemLimit.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+/**
+ * A common interface for those that support withBatchItemLimit()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithBatchItemLimit<R> {
+	Object withBatchItemLimit(Integer batchItemLimit);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithLimit.java b/src/main/java/org/springframework/data/couchbase/core/support/WithLimit.java
new file mode 100644
index 00000000..0fc2a587
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithLimit.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+/**
+ * A common interface for those that support withLimit()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithLimit<R> {
+	Object withLimit(Long limit);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithSampling.java b/src/main/java/org/springframework/data/couchbase/core/support/WithSampling.java
new file mode 100644
index 00000000..5c608da3
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithSampling.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+import com.couchbase.client.java.kv.ScanSort;
+
+/**
+ * A common interface for those that support withSampling()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithSampling<R> {
+	Object withSampling(Boolean isSampling);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithScanOptions.java b/src/main/java/org/springframework/data/couchbase/core/support/WithScanOptions.java
new file mode 100644
index 00000000..3d48de69
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithScanOptions.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+import com.couchbase.client.java.kv.ScanOptions;
+
+/**
+ * A common interface for those that support withOptions()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithScanOptions<R> {
+	Object withOptions(ScanOptions expiry);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithScanSort.java b/src/main/java/org/springframework/data/couchbase/core/support/WithScanSort.java
new file mode 100644
index 00000000..0c97ca1e
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithScanSort.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+import com.couchbase.client.java.kv.ScanSort;
+
+/**
+ * A common interface for those that support withOptions()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithScanSort<R> {
+	Object withSort(ScanSort expiry);
+
+}
diff --git a/src/main/java/org/springframework/data/couchbase/core/support/WithSeed.java b/src/main/java/org/springframework/data/couchbase/core/support/WithSeed.java
new file mode 100644
index 00000000..debe51fc
--- /dev/null
+++ b/src/main/java/org/springframework/data/couchbase/core/support/WithSeed.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020-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.couchbase.core.support;
+
+/**
+ * A common interface for those that support withSeed()
+ *
+ * @author Michael Reiche
+ * @param <R> - the entity class
+ */
+public interface WithSeed<R> {
+	Object withSeed(Long seed);
+
+}
diff --git a/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java b/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java
index 2529d54c..ab005fc8 100644
--- a/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java
+++ b/src/test/java/org/springframework/data/couchbase/core/CouchbaseTemplateKeyValueIntegrationTests.java
@@ -34,6 +34,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+import java.util.stream.Stream;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -63,8 +64,12 @@ import org.springframework.data.couchbase.util.JavaIntegrationTests;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import com.couchbase.client.core.error.CouchbaseException;
+import com.couchbase.client.core.msg.kv.MutationToken;
+import com.couchbase.client.java.json.JsonObject;
+import com.couchbase.client.java.kv.MutationState;
 import com.couchbase.client.java.kv.PersistTo;
 import com.couchbase.client.java.kv.ReplicateTo;
+import com.couchbase.client.java.kv.ScanSort;
 import com.couchbase.client.java.query.QueryOptions;
 import com.couchbase.client.java.query.QueryScanConsistency;
 
@@ -435,7 +440,11 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
 		assertEquals(upserted, foundUpserted);
 
 		// upsert will replace
-		upserted = couchbaseTemplate.upsertById(PersonValue.class).one(inserted);
+		try {
+			upserted = couchbaseTemplate.upsertById(PersonValue.class).one(inserted);
+		} catch(Exception e){
+			e.printStackTrace();
+		}
 		assertNotEquals(0, upserted.getVersion());
 		PersonValue foundUpserted2 = couchbaseTemplate.findById(PersonValue.class).one(upserted.getId());
 		assertNotNull(foundUpserted2, "upserted personValue not found");
@@ -449,6 +458,109 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
 		couchbaseTemplate.removeById(PersonValue.class).one(replaced.getId());
 	}
 
+	@Test
+	void rangeScan() {
+		String id = "A";
+		String lower = null;
+		String upper = null;
+		for (int i = 0; i < 10; i++) {
+			if (lower == null) {
+				lower = "" + i;
+			}
+			User inserted = couchbaseTemplate.insertById(User.class).one(new User("" + i, "fn_" + i, "ln_" + i));
+			upper = "" + i;
+		}
+		MutationToken mt = couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection()
+				.upsert(id, JsonObject.create().put("id", id)).mutationToken().get();
+		Stream<User> users = couchbaseTemplate.rangeScan(User.class).consistentWith(MutationState.from(mt)).withSort(ScanSort.ASCENDING).rangeScan(lower,
+				upper);
+		for (User u : users.toList()) {
+			System.err.print(u);
+			System.err.println(",");
+			assertTrue(u.getId().compareTo(lower) >= 0 && u.getId().compareTo(upper) <= 0);
+			couchbaseTemplate.removeById(User.class).one(u.getId());
+		}
+		couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection().remove(id);
+	}
+
+	@Test
+	void rangeScanId() {
+		String id = "A";
+		String lower = null;
+		String upper = null;
+		for (int i = 0; i < 10; i++) {
+			if (lower == null) {
+				lower = "" + i;
+			}
+			User inserted = couchbaseTemplate.insertById(User.class).one(new User("" + i, "fn_" + i, "ln_" + i));
+			upper = "" + i;
+		}
+		MutationToken mt = couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection()
+				.upsert(id, JsonObject.create().put("id", id)).mutationToken().get();
+		Stream<String> userIds = couchbaseTemplate.rangeScan(User.class).consistentWith(MutationState.from(mt))
+				.withSort(ScanSort.ASCENDING).rangeScanIds(lower, upper);
+		for (String userId : userIds.toList()) {
+			System.err.print(userId);
+			System.err.println(",");
+			assertTrue(userId.compareTo(lower) >= 0 && userId.compareTo(upper) <= 0);
+			couchbaseTemplate.removeById(User.class).one(userId);
+		}
+		couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection().remove(id);
+
+	}
+
+	@Test
+	void sampleScan() {
+		String id = "A";
+		String lower = null;
+		String upper = null;
+		for (int i = 0; i < 10; i++) {
+			if (lower == null) {
+				lower = "" + i;
+			}
+			User inserted = couchbaseTemplate.insertById(User.class).one(new User("" + i, "fn_" + i, "ln_" + i));
+			upper = "" + i;
+		}
+		MutationToken mt = couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection()
+			.upsert(id, JsonObject.create().put("id", id)).mutationToken().get();
+		Stream<User> users = couchbaseTemplate.rangeScan(User.class).consistentWith(MutationState.from(mt)).withSort(ScanSort.ASCENDING).withSampling(true).rangeScan(lower,
+			upper);
+		for (User u : users.toList()) {
+			System.err.print(u);
+			System.err.println(",");
+			assertTrue(u.getId().compareTo(lower) >= 0 && u.getId().compareTo(upper) <= 0);
+			couchbaseTemplate.removeById(User.class).one(u.getId());
+		}
+		couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection().remove(id);
+	}
+
+	@Test
+	void sampleScanId() {
+		String id = "A";
+		String lower = null;
+		String upper = null;
+		for (int i = 0; i < 10; i++) {
+			if (lower == null) {
+				lower = "" + i;
+			}
+			User inserted = couchbaseTemplate.insertById(User.class).one(new User("" + i, "fn_" + i, "ln_" + i));
+			upper = "" + i;
+		}
+		MutationToken mt = couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection()
+			.upsert(id, JsonObject.create().put("id", id)).mutationToken().get();
+		Stream<String> userIds = couchbaseTemplate.rangeScan(User.class).consistentWith(MutationState.from(mt))
+			.withSort(ScanSort.ASCENDING).rangeScanIds(lower, upper);
+		for (String userId : userIds.toList()) {
+			System.err.print(userId);
+			System.err.println(",");
+			assertTrue(userId.compareTo(lower) >= 0 && userId.compareTo(upper) <= 0);
+			couchbaseTemplate.removeById(User.class).one(userId);
+		}
+		couchbaseTemplate.getCouchbaseClientFactory().getDefaultCollection().remove(id);
+
+	}
+
+
 	private void sleepSecs(int i) {
 		try {
 			Thread.sleep(i * 1000);