Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-647 - Correct SDIFF & SINTER behavior.

Browse Source 
We now consider the last key in SDIFF command execution on Redis Cluster to correctly compute the set difference. 
For reactive connections we collect the result sets entirely before applying further computation. Previously, the zip function could return a previous view of the result that caused too many/too few results.

Original Pull Request: #250
This commit is contained in:
Mark Paluch
2017-05-11 14:15:47 +02:00
committed by Christoph Strobl
parent 95768983b6
commit 78e5b89d1b
6 changed files with 36 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterSetCommands.java
View File

@@ -276,7 +276,7 @@ class JedisClusterSetCommands implements RedisSetCommands {
}
byte[] source = keys[0];
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1);
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length);
ByteArraySet values = new ByteArraySet(sMembers(source));
Collection<Set<byte[]>> resultList = connection.getClusterCommandExecutor()

2
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterSetCommands.java
View File

@@ -177,7 +177,7 @@ class LettuceClusterSetCommands extends LettuceSetCommands {
}
byte[] source = keys[0];
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length - 1);
byte[][] others = Arrays.copyOfRange(keys, 1, keys.length);
ByteArraySet values = new ByteArraySet(sMembers(source));
Collection<Set<byte[]>> nodeResult = connection.getClusterCommandExecutor()

15
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java
View File

@@ -115,9 +115,12 @@ class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands imple
intersectingSets.add(cmd.smembers(command.getKeys().get(i)).distinct().collectList());
}
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets), (source, intersecting) -> {
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets).collectList(),
(source, intersectings) -> {
source.retainAll(intersecting);
for (List<ByteBuffer> intersecting : intersectings) {
source.retainAll(intersecting);
}
return source;
});
@@ -172,9 +175,13 @@ class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands imple
intersectingSets.add(cmd.smembers(command.getKeys().get(i)).distinct().collectList());
}
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets), (source, intersecting) -> {
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets).collectList(),
(source, intersectings) -> {
for (List<ByteBuffer> intersecting : intersectings) {
source.removeAll(intersecting);
}
source.removeAll(intersecting);
return source;
});