This is an automated email from the ASF dual-hosted git repository.
ringles pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new f70700e feature/GEODE-9902: Modify ZUNIONSTORE and ZINTERSTORE
storing methods (#7219)
f70700e is described below
commit f70700ed8702df331facb9131ce9e2c68ae4f29c
Author: Kris10 <[email protected]>
AuthorDate: Thu Jan 6 06:35:01 2022 -0800
feature/GEODE-9902: Modify ZUNIONSTORE and ZINTERSTORE storing methods
(#7219)
GEODE-9902: Modify ZUNIONSTORE and ZINTERSTORE storing methods
---
.../AbstractZInterStoreIntegrationTest.java | 14 ++++
.../executor/sortedset/ZInterStoreExecutor.java | 5 +-
.../executor/sortedset/ZUnionStoreExecutor.java | 5 +-
.../redis/internal/data/AbstractRedisData.java | 10 +++
.../geode/redis/internal/data/RedisSortedSet.java | 94 +++++++++++++++-------
.../geode/redis/internal/data/delta/DeltaType.java | 1 +
.../data/delta/ReplaceByteArrayDoublePairs.java | 62 ++++++++++++++
.../redis/internal/data/RedisSortedSetTest.java | 56 +++++++++++++
8 files changed, 214 insertions(+), 33 deletions(-)
diff --git
a/geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/sortedset/AbstractZInterStoreIntegrationTest.java
b/geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/sortedset/AbstractZInterStoreIntegrationTest.java
index 3ce1959..7a07499 100755
---
a/geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/sortedset/AbstractZInterStoreIntegrationTest.java
+++
b/geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/commands/executor/sortedset/AbstractZInterStoreIntegrationTest.java
@@ -220,6 +220,20 @@ public abstract class AbstractZInterStoreIntegrationTest
implements RedisIntegra
}
@Test
+ public void shouldOverwriteDestinationKey_givenDestinationExists() {
+ Set<Tuple> expectedResults = new LinkedHashSet<>();
+ expectedResults.add(new Tuple("key1Member", 1.0));
+
+ jedis.zadd(NEW_SET, 1.0, "newSetMember1");
+ jedis.zadd(NEW_SET, 2.0, "newSetMember2");
+ jedis.zadd(KEY1, 1.0, "key1Member");
+
+ assertThat(jedis.zinterstore(NEW_SET, KEY1)).isEqualTo(1L);
+ Set<Tuple> results = jedis.zrangeWithScores(NEW_SET, 0L, 1L);
+ assertThat(results).containsExactlyInAnyOrderElementsOf(expectedResults);
+ }
+
+ @Test
public void shouldStoreIntersection_givenWeightOfOne_andOneRedisSortedSet() {
Map<String, Double> scores = buildMapOfMembersAndScores();
Set<Tuple> expectedResults = convertToTuples(scores, (ignore, value) ->
value);
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZInterStoreExecutor.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZInterStoreExecutor.java
index e21fae3..0ddff26 100644
---
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZInterStoreExecutor.java
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZInterStoreExecutor.java
@@ -16,13 +16,13 @@ package
org.apache.geode.redis.internal.commands.executor.sortedset;
import static
org.apache.geode.redis.internal.RedisConstants.ERROR_KEY_REQUIRED_ZINTERSTORE;
+import static org.apache.geode.redis.internal.data.RedisSortedSet.zinterstore;
import java.util.List;
import org.apache.geode.redis.internal.commands.Command;
import org.apache.geode.redis.internal.commands.executor.RedisResponse;
import org.apache.geode.redis.internal.data.RedisKey;
-import org.apache.geode.redis.internal.data.RedisSortedSet;
import org.apache.geode.redis.internal.netty.ExecutionHandlerContext;
import org.apache.geode.redis.internal.services.RegionProvider;
@@ -35,8 +35,7 @@ public class ZInterStoreExecutor extends ZStoreExecutor {
List<RedisKey> keysToLock = getKeysToLock(regionProvider, key, keyWeights);
return context.lockedExecute(key, keysToLock,
- () -> new RedisSortedSet(0))
- .zinterstore(regionProvider, key, keyWeights, aggregator);
+ () -> zinterstore(regionProvider, key, keyWeights, aggregator));
}
@Override
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZUnionStoreExecutor.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZUnionStoreExecutor.java
index 7214cfa..6d30155 100644
---
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZUnionStoreExecutor.java
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/sortedset/ZUnionStoreExecutor.java
@@ -15,13 +15,13 @@
package org.apache.geode.redis.internal.commands.executor.sortedset;
import static
org.apache.geode.redis.internal.RedisConstants.ERROR_KEY_REQUIRED_ZUNIONSTORE;
+import static org.apache.geode.redis.internal.data.RedisSortedSet.zunionstore;
import java.util.List;
import org.apache.geode.redis.internal.commands.Command;
import org.apache.geode.redis.internal.commands.executor.RedisResponse;
import org.apache.geode.redis.internal.data.RedisKey;
-import org.apache.geode.redis.internal.data.RedisSortedSet;
import org.apache.geode.redis.internal.netty.ExecutionHandlerContext;
import org.apache.geode.redis.internal.services.RegionProvider;
@@ -34,8 +34,7 @@ public class ZUnionStoreExecutor extends ZStoreExecutor {
List<RedisKey> keysToLock = getKeysToLock(regionProvider, key, keyWeights);
return context.lockedExecute(key, keysToLock,
- () -> new RedisSortedSet(0))
- .zunionstore(regionProvider, key, keyWeights, aggregator);
+ () -> zunionstore(regionProvider, key, keyWeights, aggregator));
}
@Override
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/AbstractRedisData.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/AbstractRedisData.java
index 3785784..349bcd7 100644
---
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/AbstractRedisData.java
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/AbstractRedisData.java
@@ -25,6 +25,7 @@ import static
org.apache.geode.redis.internal.data.delta.DeltaType.APPEND_BYTE_A
import static
org.apache.geode.redis.internal.data.delta.DeltaType.REMOVE_BYTE_ARRAYS;
import static
org.apache.geode.redis.internal.data.delta.DeltaType.REPLACE_BYTE_ARRAYS;
import static
org.apache.geode.redis.internal.data.delta.DeltaType.REPLACE_BYTE_ARRAY_AT_OFFSET;
+import static
org.apache.geode.redis.internal.data.delta.DeltaType.REPLACE_BYTE_ARRAY_DOUBLE_PAIRS;
import static
org.apache.geode.redis.internal.data.delta.DeltaType.REPLACE_BYTE_AT_OFFSET;
import static
org.apache.geode.redis.internal.data.delta.DeltaType.SET_BYTE_ARRAY;
import static
org.apache.geode.redis.internal.data.delta.DeltaType.SET_BYTE_ARRAY_AND_TIMESTAMP;
@@ -54,6 +55,7 @@ import org.apache.geode.redis.internal.data.delta.DeltaInfo;
import org.apache.geode.redis.internal.data.delta.DeltaType;
import org.apache.geode.redis.internal.data.delta.RemoveByteArrays;
import org.apache.geode.redis.internal.data.delta.ReplaceByteArrayAtOffset;
+import org.apache.geode.redis.internal.data.delta.ReplaceByteArrayDoublePairs;
import org.apache.geode.redis.internal.data.delta.ReplaceByteArrays;
import org.apache.geode.redis.internal.data.delta.ReplaceByteAtOffset;
import org.apache.geode.redis.internal.data.delta.SetByteArray;
@@ -249,6 +251,9 @@ public abstract class AbstractRedisData implements
RedisData {
case REPLACE_BYTE_ARRAY_AT_OFFSET:
ReplaceByteArrayAtOffset.deserializeFrom(in, this);
break;
+ case REPLACE_BYTE_ARRAY_DOUBLE_PAIRS:
+ ReplaceByteArrayDoublePairs.deserializeFrom(in, this);
+ break;
case REPLACE_BYTE_AT_OFFSET:
ReplaceByteAtOffset.deserializeFrom(in, this);
break;
@@ -295,6 +300,11 @@ public abstract class AbstractRedisData implements
RedisData {
throw new IllegalStateException("unexpected " +
REPLACE_BYTE_ARRAY_AT_OFFSET);
}
+ public void applyReplaceByteArrayDoublePairDelta(RedisSortedSet.MemberMap
members,
+ RedisSortedSet.ScoreSet scoreSet) {
+ throw new IllegalStateException("unexpected " +
REPLACE_BYTE_ARRAY_DOUBLE_PAIRS);
+ }
+
public void applyReplaceByteAtOffsetDelta(int offset, byte bits) {
throw new IllegalStateException("unexpected " + REPLACE_BYTE_AT_OFFSET);
}
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/RedisSortedSet.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/RedisSortedSet.java
index 0af1040..114fd5f 100644
---
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/RedisSortedSet.java
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/RedisSortedSet.java
@@ -58,6 +58,7 @@ import
org.apache.geode.redis.internal.data.collections.OrderStatisticsTree;
import
org.apache.geode.redis.internal.data.collections.SizeableBytes2ObjectOpenCustomHashMapWithCursor;
import org.apache.geode.redis.internal.data.delta.AddByteArrayDoublePairs;
import org.apache.geode.redis.internal.data.delta.RemoveByteArrays;
+import org.apache.geode.redis.internal.data.delta.ReplaceByteArrayDoublePairs;
import org.apache.geode.redis.internal.netty.Coder;
import org.apache.geode.redis.internal.services.RegionProvider;
@@ -66,7 +67,7 @@ public class RedisSortedSet extends AbstractRedisData {
private static final Logger logger = LogService.getLogger();
private MemberMap members;
- private final ScoreSet scoreSet = new ScoreSet();
+ private ScoreSet scoreSet = new ScoreSet();
@Override
public int getSizeInBytes() {
@@ -83,6 +84,11 @@ public class RedisSortedSet extends AbstractRedisData {
}
}
+ RedisSortedSet(MemberMap members, ScoreSet scoreSet) {
+ this.members = members;
+ this.scoreSet = scoreSet;
+ }
+
public RedisSortedSet(int size) {
this.members = new MemberMap(size);
}
@@ -104,6 +110,13 @@ public class RedisSortedSet extends AbstractRedisData {
memberAdd(bytes, score);
}
+ @Override
+ public void applyReplaceByteArrayDoublePairDelta(MemberMap members, ScoreSet
scoreSet) {
+ persistNoDelta();
+ this.members = members;
+ this.scoreSet = scoreSet;
+ }
+
/**
* Since GII (getInitialImage) can come in and call toData while other
threads are modifying this
* object, the striped executor will not protect toData. So any methods that
modify "members"
@@ -274,16 +287,17 @@ public class RedisSortedSet extends AbstractRedisData {
return Coder.doubleToBytes(score);
}
- public long zinterstore(RegionProvider regionProvider, RedisKey key,
List<ZKeyWeight> keyWeights,
+ public static long zinterstore(RegionProvider regionProvider, RedisKey key,
+ List<ZKeyWeight> keyWeights,
ZAggregator aggregator) {
List<RedisSortedSet> sets = new ArrayList<>(keyWeights.size());
+
for (ZKeyWeight keyWeight : keyWeights) {
RedisSortedSet set =
regionProvider.getTypedRedisData(REDIS_SORTED_SET,
keyWeight.getKey(), false);
if (set == NULL_REDIS_SORTED_SET) {
- regionProvider.getLocalDataRegion().remove(key);
- return 0;
+ return sortedSetOpStoreResult(regionProvider, key, new MemberMap(0),
new ScoreSet());
} else {
sets.add(set);
}
@@ -296,6 +310,8 @@ public class RedisSortedSet extends AbstractRedisData {
}
}
+ MemberMap interMembers = new MemberMap(smallestSet.getSortedSetSize());
+ ScoreSet interScores = new ScoreSet();
for (byte[] member : smallestSet.members.keySet()) {
boolean addToSet = true;
double newScore;
@@ -323,18 +339,41 @@ public class RedisSortedSet extends AbstractRedisData {
}
}
}
+
if (addToSet) {
- memberAdd(member, newScore);
+ OrderedSetEntry entry = new OrderedSetEntry(member, newScore);
+ interMembers.put(member, entry);
+ interScores.add(entry);
}
}
- if (removeFromRegion()) {
- regionProvider.getDataRegion().remove(key);
- } else {
- regionProvider.getLocalDataRegion().put(key, this);
+ return sortedSetOpStoreResult(regionProvider, key, interMembers,
interScores);
+ }
+
+ @VisibleForTesting
+ static long sortedSetOpStoreResult(RegionProvider regionProvider, RedisKey
destinationKey,
+ MemberMap interMembers, ScoreSet interScores) {
+ RedisSortedSet destinationSet =
+ regionProvider.getTypedRedisDataElseRemove(REDIS_SORTED_SET,
destinationKey, false);
+
+ if (interMembers.isEmpty() || interScores.isEmpty()) {
+ if (destinationSet != null) {
+ regionProvider.getDataRegion().remove(destinationKey);
+ }
+ return 0;
}
- return getSortedSetSize();
+ if (destinationSet != null) {
+ destinationSet.persistNoDelta();
+ destinationSet.members = interMembers;
+ destinationSet.scoreSet = interScores;
+ destinationSet.storeChanges(regionProvider.getDataRegion(),
destinationKey,
+ new ReplaceByteArrayDoublePairs(interMembers));
+ } else {
+ regionProvider.getDataRegion().put(destinationKey,
+ new RedisSortedSet(interMembers, interScores));
+ }
+ return interMembers.size();
}
public long zlexcount(SortedSetLexRangeOptions lexOptions) {
@@ -463,19 +502,30 @@ public class RedisSortedSet extends AbstractRedisData {
return null;
}
- public long zunionstore(RegionProvider regionProvider, RedisKey key,
List<ZKeyWeight> keyWeights,
+ public static long zunionstore(RegionProvider regionProvider, RedisKey key,
+ List<ZKeyWeight> keyWeights,
ZAggregator aggregator) {
+ MemberMap unionMembers = new MemberMap(0);
+ ScoreSet unionScores = new ScoreSet();
+
for (ZKeyWeight keyWeight : keyWeights) {
RedisSortedSet set =
regionProvider.getTypedRedisData(REDIS_SORTED_SET,
keyWeight.getKey(), false);
if (set == NULL_REDIS_SORTED_SET) {
continue;
}
+
double weight = keyWeight.getWeight();
for (AbstractOrderedSetEntry entry : set.members.values()) {
- OrderedSetEntry existingValue = members.get(entry.getMember());
- if (existingValue == null) {
+ OrderedSetEntry existingValue = unionMembers.get(entry.getMember());
+
+ if (existingValue != null) {
+ unionScores.remove(existingValue);
+
existingValue.updateScore(aggregator.getFunction().apply(existingValue.getScore(),
+ entry.getScore() * weight));
+ unionScores.add(existingValue);
+ } else {
double score;
// Redis math and Java math are different when handling infinity.
Specifically:
// Java: INFINITY * 0 = NaN
@@ -492,23 +542,13 @@ public class RedisSortedSet extends AbstractRedisData {
score = newScore;
}
}
- members.put(entry.getMember(), new
OrderedSetEntry(entry.getMember(), score));
- continue;
+ OrderedSetEntry newUnion = new OrderedSetEntry(entry.getMember(),
score);
+ unionMembers.put(entry.getMember(), newUnion);
+ unionScores.add(newUnion);
}
-
-
existingValue.updateScore(aggregator.getFunction().apply(existingValue.getScore(),
- entry.getScore() * weight));
}
}
-
- if (removeFromRegion()) {
- regionProvider.getDataRegion().remove(key);
- } else {
- scoreSet.addAll(members.values());
- regionProvider.getLocalDataRegion().put(key, this);
- }
-
- return getSortedSetSize();
+ return sortedSetOpStoreResult(regionProvider, key, unionMembers,
unionScores);
}
private List<byte[]> zpop(Iterator<AbstractOrderedSetEntry> scoresIterator,
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/DeltaType.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/DeltaType.java
index 05bef52..c3fef85 100644
---
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/DeltaType.java
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/DeltaType.java
@@ -24,6 +24,7 @@ public enum DeltaType {
REMOVE_BYTE_ARRAYS,
REPLACE_BYTE_ARRAYS,
REPLACE_BYTE_ARRAY_AT_OFFSET,
+ REPLACE_BYTE_ARRAY_DOUBLE_PAIRS,
REPLACE_BYTE_AT_OFFSET,
SET_BYTE_ARRAY,
SET_BYTE_ARRAY_AND_TIMESTAMP,
diff --git
a/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/ReplaceByteArrayDoublePairs.java
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/ReplaceByteArrayDoublePairs.java
new file mode 100644
index 0000000..0b63096
--- /dev/null
+++
b/geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/delta/ReplaceByteArrayDoublePairs.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license
+ * agreements. See the NOTICE file distributed with this work for additional
information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache
License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express
+ * or implied. See the License for the specific language governing permissions
and limitations under
+ * the License.
+ */
+package org.apache.geode.redis.internal.data.delta;
+
+import static org.apache.geode.DataSerializer.readByteArray;
+import static org.apache.geode.DataSerializer.readPrimitiveDouble;
+import static org.apache.geode.internal.InternalDataSerializer.readArrayLength;
+import static
org.apache.geode.redis.internal.data.delta.DeltaType.REPLACE_BYTE_ARRAY_DOUBLE_PAIRS;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.redis.internal.data.AbstractRedisData;
+import org.apache.geode.redis.internal.data.RedisSortedSet;
+
+public class ReplaceByteArrayDoublePairs implements DeltaInfo {
+ private final RedisSortedSet.MemberMap members;
+
+ public ReplaceByteArrayDoublePairs(RedisSortedSet.MemberMap members) {
+ this.members = members;
+ }
+
+ public void serializeTo(DataOutput out) throws IOException {
+ DataSerializer.writeEnum(REPLACE_BYTE_ARRAY_DOUBLE_PAIRS, out);
+ InternalDataSerializer.writeArrayLength(members.size(), out);
+ for (byte[] member : members.keySet()) {
+ DataSerializer.writeByteArray(member, out);
+ DataSerializer.writePrimitiveDouble(members.get(member).getScore(), out);
+ }
+ }
+
+ public static void deserializeFrom(DataInput in, AbstractRedisData
redisData) throws IOException {
+ int size = readArrayLength(in);
+ RedisSortedSet.MemberMap membersMap = new RedisSortedSet.MemberMap(size);
+ RedisSortedSet.ScoreSet scoreSet = new RedisSortedSet.ScoreSet();
+ while (size > 0) {
+ byte[] member = readByteArray(in);
+ double score = readPrimitiveDouble(in);
+ RedisSortedSet.OrderedSetEntry entry = new
RedisSortedSet.OrderedSetEntry(member, score);
+ membersMap.put(member, entry);
+ scoreSet.add(entry);
+ size--;
+ }
+ redisData.applyReplaceByteArrayDoublePairDelta(membersMap, scoreSet);
+
+ }
+}
diff --git
a/geode-for-redis/src/test/java/org/apache/geode/redis/internal/data/RedisSortedSetTest.java
b/geode-for-redis/src/test/java/org/apache/geode/redis/internal/data/RedisSortedSetTest.java
index 14601cf..0d20d15 100644
---
a/geode-for-redis/src/test/java/org/apache/geode/redis/internal/data/RedisSortedSetTest.java
+++
b/geode-for-redis/src/test/java/org/apache/geode/redis/internal/data/RedisSortedSetTest.java
@@ -17,8 +17,11 @@
package org.apache.geode.redis.internal.data;
import static java.util.Collections.singletonList;
+import static
org.apache.geode.redis.internal.data.AbstractRedisData.NO_EXPIRATION;
+import static
org.apache.geode.redis.internal.data.RedisDataType.REDIS_SORTED_SET;
import static
org.apache.geode.redis.internal.data.RedisSortedSet.OrderedSetEntry.ORDERED_SET_ENTRY_OVERHEAD;
import static
org.apache.geode.redis.internal.data.RedisSortedSet.REDIS_SORTED_SET_OVERHEAD;
+import static
org.apache.geode.redis.internal.data.RedisSortedSet.sortedSetOpStoreResult;
import static org.apache.geode.redis.internal.netty.Coder.stringToBytes;
import static
org.apache.geode.redis.internal.netty.StringBytesGlossary.GREATEST_MEMBER_NAME;
import static
org.apache.geode.redis.internal.netty.StringBytesGlossary.LEAST_MEMBER_NAME;
@@ -65,6 +68,7 @@ import
org.apache.geode.redis.internal.commands.executor.sortedset.SortedSetRank
import org.apache.geode.redis.internal.commands.executor.sortedset.ZAddOptions;
import org.apache.geode.redis.internal.data.delta.RemoveByteArrays;
import org.apache.geode.redis.internal.netty.Coder;
+import org.apache.geode.redis.internal.services.RegionProvider;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;
@RunWith(GeodeParamsRunner.class)
@@ -153,6 +157,58 @@ public class RedisSortedSetTest {
assertThat(sortedSet1.hasDelta()).isFalse();
}
+ @Test
+ public void sortedSetOpStoreResult_stores_delta_that_is_stable() {
+ RegionProvider regionProvider = uncheckedCast(mock(RegionProvider.class));
+ Region<RedisKey, RedisData> dataRegion = uncheckedCast(mock(Region.class));
+
+ RedisSortedSet sortedSet1 = createRedisSortedSet("3.14159", "v1",
"2.71828", "v2");
+ when(regionProvider.getTypedRedisDataElseRemove(REDIS_SORTED_SET, null,
false))
+ .thenReturn(sortedSet1);
+ when(regionProvider.getDataRegion()).thenReturn(dataRegion);
+ when(dataRegion.put(any(),
any())).thenAnswer(this::validateDeltaSerialization);
+
+ // Setting up for store operation
+ RedisSortedSet.MemberMap members = new RedisSortedSet.MemberMap(1);
+ RedisSortedSet.ScoreSet scores = new RedisSortedSet.ScoreSet();
+ byte[] member = new byte[] {4};
+ RedisSortedSet.OrderedSetEntry entry = new
RedisSortedSet.OrderedSetEntry(member, 5);
+ members.put(member, entry);
+ scores.add(entry);
+
+ sortedSetOpStoreResult(regionProvider, null, members, scores);
+
+ verify(dataRegion).put(any(), any());
+ assertThat(sortedSet1.hasDelta()).isFalse();
+ }
+
+ @Test
+ public void sortedSetOpStoreResult_sets_expiration_time_to_zero() {
+ RegionProvider regionProvider = uncheckedCast(mock(RegionProvider.class));
+ Region<RedisKey, RedisData> dataRegion = uncheckedCast(mock(Region.class));
+
+ RedisSortedSet setDest = createRedisSortedSet("3.14159", "v1", "2.71828",
"v2");
+ setDest.setExpirationTimestamp(dataRegion, null, 100);
+
+ when(regionProvider.getTypedRedisDataElseRemove(REDIS_SORTED_SET, null,
false))
+ .thenReturn(setDest);
+ when(regionProvider.getDataRegion()).thenReturn(dataRegion);
+ when(dataRegion.put(any(),
any())).thenAnswer(this::validateDeltaSerialization);
+
+ // Setting up for store operation
+ RedisSortedSet.MemberMap members = new RedisSortedSet.MemberMap(1);
+ RedisSortedSet.ScoreSet scores = new RedisSortedSet.ScoreSet();
+ byte[] member = new byte[] {4};
+ RedisSortedSet.OrderedSetEntry entry = new
RedisSortedSet.OrderedSetEntry(member, 5);
+ members.put(member, entry);
+ scores.add(entry);
+
+ sortedSetOpStoreResult(regionProvider, null, members, scores);
+
+ assertThat(setDest.getExpirationTimestamp()).isEqualTo(NO_EXPIRATION);
+ assertThat(setDest.hasDelta()).isFalse();
+ }
+
private Object validateDeltaSerialization(InvocationOnMock invocation)
throws IOException {
RedisSortedSet value = invocation.getArgument(1, RedisSortedSet.class);
assertThat(value.hasDelta()).isTrue();
