Datacouch 1599 add rangescan support (#1738)

* Add RangeScan to Template.

Closes #1599.

* Add RangeScan support.

Closes #1599.
This commit is contained in:
Michael Reiche
2023-05-12 14:00:52 -07:00
committed by GitHub
parent cdf0b4c713
commit cf9bdde6ef
15 changed files with 1113 additions and 19 deletions

View File

@@ -158,6 +158,11 @@ 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();

View File

@@ -0,0 +1,213 @@
/*
* Copyright 2012-2023 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.WithScanOptions;
import org.springframework.data.couchbase.core.support.WithScanSort;
import com.couchbase.client.java.kv.MutationState;
import com.couchbase.client.java.kv.ScanOptions;
/**
* 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 keys.
*/
Stream<String> rangeScanIds(String lower, String upper);
/**
* Range Scan
*
* @param limit
* @param seed
* @return the list of found entities.
*/
Stream<T> samplingScan(Long limit, Long... seed);
/**
* Range Scan Ids
*
* @param limit
* @param seed
* @return the list of keys
*/
Stream<String> samplingScanIds(Long limit, Long... seed);
}
/**
* 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 RangeScanWithSort<T> extends RangeScanInScope<T>, WithScanSort<T> {
/**
* sort
*
* @param sort
*/
@Override
RangeScanInScope<T> withSort(Object 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 RangeScanWithBatchItemLimit<T> extends RangeScanWithProjection<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
RangeScanWithProjection<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
RangeScanWithBatchItemLimit<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> {}
}

View File

@@ -0,0 +1,143 @@
/*
* Copyright 2012-2023 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;
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);
}
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 Object sort;
private final MutationState mutationState;
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, Object sort, MutationState mutationState,
Integer batchItemLimit, Integer batchByteLimit) {
this.template = template;
this.domainType = domainType;
this.scope = scope;
this.collection = collection;
this.options = options;
this.sort = sort;
this.mutationState = mutationState;
this.batchItemLimit = batchItemLimit;
this.batchByteLimit = batchByteLimit;
this.reactiveSupport = new ReactiveRangeScanSupport<>(template.reactive(), domainType, scope, collection, options,
sort, mutationState, 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, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public RangeScanWithOptions<T> inCollection(final String collection) {
return new ExecutableRangeScanSupport<>(template, domainType, scope,
collection != null ? collection : this.collection, options, sort, mutationState,
batchItemLimit, batchByteLimit);
}
@Override
public RangeScanInCollection<T> inScope(final String scope) {
return new ExecutableRangeScanSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
options, sort, mutationState, batchItemLimit, batchByteLimit);
}
@Override
public RangeScanInScope<T> withSort(Object sort) {
return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public RangeScanWithSort<T> consistentWith(MutationState mutationState) {
return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public <R> RangeScanConsistentWith<R> as(Class<R> returnType) {
return new ExecutableRangeScanSupport<>(template, returnType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public RangeScanWithProjection<T> withBatchItemLimit(Integer batchItemLimit) {
return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public RangeScanWithBatchItemLimit<T> withBatchByteLimit(Integer batchByteLimit) {
return new ExecutableRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit);
}
@Override
public Stream<T> rangeScan(String lower, String upper) {
return reactiveSupport.rangeScan(lower, upper, false, null, null).toStream();
}
@Override
public Stream<String> rangeScanIds(String lower, String upper) {
return reactiveSupport.rangeScanIds(lower, upper, false, null, null).toStream();
}
@Override
public Stream<T> samplingScan(Long limit, Long... seed) {
return reactiveSupport.sampleScan(limit, seed).toStream();
}
@Override
public Stream<String> samplingScanIds(Long limit, Long... seed) {
return reactiveSupport.sampleScanIds(limit, seed).toStream();
}
}
}

View File

@@ -22,4 +22,5 @@ package org.springframework.data.couchbase.core;
public interface FluentCouchbaseOperations extends ExecutableUpsertByIdOperation, ExecutableInsertByIdOperation,
ExecutableReplaceByIdOperation, ExecutableFindByIdOperation, ExecutableFindFromReplicasByIdOperation,
ExecutableFindByQueryOperation, ExecutableFindByAnalyticsOperation, ExecutableExistsByIdOperation,
ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation, ExecutableMutateInByIdOperation {}
ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation, ExecutableMutateInByIdOperation,
ExecutableRangeScanOperation {}

View File

@@ -190,6 +190,11 @@ public class ReactiveCouchbaseTemplate implements ReactiveCouchbaseOperations, A
return new ReactiveMutateInByIdOperationSupport(this).mutateInById(domainType);
}
@Override
public <T> ReactiveRangeScan<T> rangeScan(Class<T> domainType) {
return new ReactiveRangeScanOperationSupport(this).rangeScan(domainType);
}
@Override
public String getBucketName() {
return clientFactory.getBucket().name();

View File

@@ -22,4 +22,5 @@ package org.springframework.data.couchbase.core;
public interface ReactiveFluentCouchbaseOperations extends ReactiveUpsertByIdOperation, ReactiveInsertByIdOperation,
ReactiveReplaceByIdOperation, ReactiveFindByIdOperation, ReactiveExistsByIdOperation,
ReactiveFindByAnalyticsOperation, ReactiveFindFromReplicasByIdOperation, ReactiveFindByQueryOperation,
ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation, ReactiveMutateInByIdOperation {}
ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation, ReactiveMutateInByIdOperation,
ReactiveRangeScanOperation {}

View File

@@ -0,0 +1,212 @@
/*
* Copyright 2012-2023 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 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.WithScanOptions;
import org.springframework.data.couchbase.core.support.WithScanSort;
import com.couchbase.client.java.kv.MutationState;
import com.couchbase.client.java.kv.ScanOptions;
/**
* 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 ids.
*/
Flux<String> rangeScanIds(String lower, String upper);
/**
* Finds a list of documents based on the given IDs.
*
* @param limit
* @param seed
* @return the list of found entities.
*/
Flux<T> sampleScan(Long limit, Long... seed);
/**
* Finds a list of documents based on the given IDs.
*
* @param limit
* @param seed
* @return the list of ids.
*/
Flux<String> sampleScanIds(Long limit, Long... seed);
}
/**
* 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 RangeScanWithSort<T> extends RangeScanInScope<T>, WithScanSort<T> {
/**
* sort
*
* @param sort
*/
@Override
RangeScanInScope<T> withSort(Object 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 RangeScanWithBatchItemLimit<T> extends RangeScanWithProjection<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
RangeScanWithProjection<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
RangeScanWithBatchItemLimit<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> {}
}

View File

@@ -0,0 +1,230 @@
/*
* Copyright 2012-2023 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.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,
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 Object sort;
private final MutationState mutationState;
private final Integer batchItemLimit;
private final Integer batchByteLimit;
private final ReactiveTemplateSupport support;
ReactiveRangeScanSupport(ReactiveCouchbaseTemplate template, Class<T> domainType, String scope, String collection,
ScanOptions options, Object sort, MutationState mutationState,
Integer batchItemLimit, Integer batchByteLimit, ReactiveTemplateSupport support) {
this.template = template;
this.domainType = domainType;
this.scope = scope;
this.collection = collection;
this.options = options;
this.sort = sort;
this.mutationState = mutationState;
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, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanWithOptions<T> inCollection(final String collection) {
return new ReactiveRangeScanSupport<>(template, domainType, scope,
collection != null ? collection : this.collection, options, sort, mutationState,
batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanInCollection<T> inScope(final String scope) {
return new ReactiveRangeScanSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
options, sort, mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanInScope<T> withSort(Object sort) {
return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanWithSort<T> consistentWith(MutationState mutationState) {
return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public <R> RangeScanConsistentWith<R> as(Class<R> returnType) {
return new ReactiveRangeScanSupport<>(template, returnType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanWithProjection<T> withBatchItemLimit(Integer batchItemLimit) {
return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public RangeScanWithBatchItemLimit<T> withBatchByteLimit(Integer batchByteLimit) {
return new ReactiveRangeScanSupport<>(template, domainType, scope, collection, options, sort,
mutationState, batchItemLimit, batchByteLimit, support);
}
@Override
public Flux<T> rangeScan(String lower, String upper) {
return rangeScan(lower, upper, false, null, null);
}
@Override
public Flux<T> sampleScan(Long limit, Long... seed) {
return rangeScan(null, null, true, limit, seed!= null && seed.length > 0 ? seed[0] : null);
}
Flux<T> rangeScan(String lower, String upper, boolean isSamplingScan, Long limit, Long seed) {
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();
ScanType scanType = null;
if(isSamplingScan){
scanType = ScanType.samplingScan(limit, seed != null ? seed : 0);
} else {
ScanTerm lowerTerm = ScanTerm.minimum();
ScanTerm upperTerm = ScanTerm.maximum();
if (lower != null) {
lowerTerm = ScanTerm.inclusive(lower);
}
if (upper != null) {
upperTerm = ScanTerm.inclusive(upper);
}
scanType = ScanType.rangeScan(lowerTerm, upperTerm);
}
Flux<T> reactiveEntities = TransactionalSupport.verifyNotInTransaction("rangeScan")
.thenMany(rc.scan(scanType, buildScanOptions(pArgs.getOptions(), false))
.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 upper, String lower) {
return rangeScanIds(upper, lower, false, null, null);
}
@Override
public Flux<String> sampleScanIds(Long limit, Long... seed) {
return rangeScanIds(null, null, true, limit, seed!= null && seed.length > 0 ? seed[0] : null);
}
Flux<String> rangeScanIds(String lower, String upper, boolean isSamplingScan, Long limit, Long seed) {
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();
ScanType scanType = null;
if(isSamplingScan){
scanType = ScanType.samplingScan(limit, seed != null ? seed : 0);
} else {
ScanTerm lowerTerm = ScanTerm.minimum();
ScanTerm upperTerm = ScanTerm.maximum();
if (lower != null) {
lowerTerm = ScanTerm.inclusive(lower);
}
if (upper != null) {
upperTerm = ScanTerm.inclusive(upper);
}
scanType = 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);
}
}
}

View File

@@ -15,7 +15,6 @@
*/
package org.springframework.data.couchbase.core.query;
import static com.couchbase.client.core.util.Validators.notNull;
import static org.springframework.data.couchbase.core.query.Meta.MetaKey.RETRY_STRATEGY;
import static org.springframework.data.couchbase.core.query.Meta.MetaKey.SCAN_CONSISTENCY;
import static org.springframework.data.couchbase.core.query.Meta.MetaKey.TIMEOUT;
@@ -25,13 +24,9 @@ import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import com.couchbase.client.core.api.query.CoreQueryScanConsistency;
import com.couchbase.client.core.classic.query.ClassicCoreQueryOps;
import com.couchbase.client.core.error.InvalidArgumentException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -44,6 +39,9 @@ import org.springframework.data.couchbase.repository.ScanConsistency;
import org.springframework.data.couchbase.repository.Scope;
import org.springframework.data.couchbase.repository.query.CouchbaseQueryMethod;
import com.couchbase.client.core.api.query.CoreQueryScanConsistency;
import com.couchbase.client.core.classic.query.ClassicCoreQueryOps;
import com.couchbase.client.core.error.InvalidArgumentException;
import com.couchbase.client.core.io.CollectionIdentifier;
import com.couchbase.client.core.msg.kv.DurabilityLevel;
import com.couchbase.client.core.retry.RetryStrategy;
@@ -51,11 +49,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.PersistTo;
import com.couchbase.client.java.kv.MutateInOptions;
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.UpsertOptions;
import com.couchbase.client.java.query.QueryOptions;
import com.couchbase.client.java.query.QueryScanConsistency;
@@ -545,4 +545,24 @@ public class OptionsBuilder {
return annotationString(annotation, "value", defaultValue, elements);
}
public static ScanOptions buildScanOptions(ScanOptions options, Object 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;
}
}

View File

@@ -0,0 +1,29 @@
/*
* Copyright 2020-2023 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);
}

View File

@@ -0,0 +1,26 @@
/*
* Copyright 2020-2023 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);
}

View File

@@ -0,0 +1,26 @@
/*
* Copyright 2020-2023 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);
}

View File

@@ -0,0 +1,28 @@
/*
* Copyright 2020-2023 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);
}

View File

@@ -0,0 +1,26 @@
/*
* Copyright 2020-2023 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 withOptions()
*
* @author Michael Reiche
* @param <R> - the entity class
*/
public interface WithScanSort<R> {
Object withSort(Object expiry);
}

View File

@@ -27,11 +27,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.time.Duration;
import java.util.*;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Stream;
import com.couchbase.client.core.error.TimeoutException;
import com.couchbase.client.core.msg.kv.DurabilityLevel;
import com.couchbase.client.core.retry.RetryReason;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -41,8 +46,26 @@ import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.data.couchbase.core.ExecutableFindByIdOperation.ExecutableFindById;
import org.springframework.data.couchbase.core.ExecutableRemoveByIdOperation.ExecutableRemoveById;
import org.springframework.data.couchbase.core.ExecutableReplaceByIdOperation.ExecutableReplaceById;
import org.springframework.data.couchbase.core.support.*;
import org.springframework.data.couchbase.domain.*;
import org.springframework.data.couchbase.core.support.OneAndAllEntity;
import org.springframework.data.couchbase.core.support.OneAndAllId;
import org.springframework.data.couchbase.core.support.WithDurability;
import org.springframework.data.couchbase.core.support.WithExpiry;
import org.springframework.data.couchbase.domain.Address;
import org.springframework.data.couchbase.domain.Config;
import org.springframework.data.couchbase.domain.MutableUser;
import org.springframework.data.couchbase.domain.NaiveAuditorAware;
import org.springframework.data.couchbase.domain.PersonValue;
import org.springframework.data.couchbase.domain.Submission;
import org.springframework.data.couchbase.domain.User;
import org.springframework.data.couchbase.domain.UserAnnotated;
import org.springframework.data.couchbase.domain.UserAnnotated2;
import org.springframework.data.couchbase.domain.UserAnnotated3;
import org.springframework.data.couchbase.domain.UserAnnotatedDurability;
import org.springframework.data.couchbase.domain.UserAnnotatedDurabilityExpression;
import org.springframework.data.couchbase.domain.UserAnnotatedPersistTo;
import org.springframework.data.couchbase.domain.UserAnnotatedReplicateTo;
import org.springframework.data.couchbase.domain.UserAnnotatedTouchOnRead;
import org.springframework.data.couchbase.domain.UserSubmission;
import org.springframework.data.couchbase.util.ClusterType;
import org.springframework.data.couchbase.util.IgnoreWhen;
import org.springframework.data.couchbase.util.JavaIntegrationTests;
@@ -50,6 +73,12 @@ import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.couchbase.client.core.error.CouchbaseException;
import com.couchbase.client.core.error.TimeoutException;
import com.couchbase.client.core.msg.kv.DurabilityLevel;
import com.couchbase.client.core.msg.kv.MutationToken;
import com.couchbase.client.core.retry.RetryReason;
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.query.QueryOptions;
@@ -1051,11 +1080,6 @@ class CouchbaseTemplateKeyValueIntegrationTests extends JavaIntegrationTests {
assertEquals(user, inserted);
User found = couchbaseTemplate.findById(User.class).one(user.getId());
assertEquals(inserted, found);
System.err.println("inserted: "+inserted);
System.err.println("found: "+found);
System.err.println("found:jsonNode "+found.jsonNode.toPrettyString());
System.err.println("found:jsonObject "+found.jsonObject.toString());
System.err.println("found:jsonArray "+found.jsonArray.toString());
assertThrows(DuplicateKeyException.class, () -> couchbaseTemplate.insertById(User.class).one(user));
couchbaseTemplate.removeById(User.class).one(user.getId());
}
@@ -1209,6 +1233,111 @@ 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)*/.samplingScan(5l, null);
List<User> usersList = users.toList();
assertEquals(5, usersList.size(), "number in sample");
for (User u : usersList) {
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)*/.samplingScanIds(5l);
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);