From 22aa5df73e785da20174863db72cbac2ca9aa762 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Wed, 26 Nov 2014 08:16:37 +0100 Subject: [PATCH] DATAREDIS-308 - Add support for HyperLogLog. We added methods for PFADD, PFCOUNT and PFMERGE to RedisConnection and StringRedisConnection. HyperLogLogOperations available via RedisTemplate allow more indrect usage of HLL. Please note that currently Jedis is the only driver that can be used for HyperLogLog commands. Original pull request: #116. --- .../DefaultStringRedisConnection.java | 67 ++++++- .../redis/connection/HyperLogLogCommands.java | 51 +++++ .../data/redis/connection/RedisCommands.java | 5 +- .../connection/StringRedisConnection.java | 25 ++- .../connection/jedis/JedisConnection.java | 79 +++++++- .../connection/jredis/JredisConnection.java | 29 ++- .../connection/lettuce/LettuceConnection.java | 33 +++- .../redis/connection/srp/SrpConnection.java | 29 ++- .../core/DefaultHyperLogLogOperations.java | 102 ++++++++++ .../redis/core/HyperLogLogOperations.java | 56 ++++++ .../data/redis/core/RedisOperations.java | 6 + .../data/redis/core/RedisTemplate.java | 15 ++ .../AbstractConnectionIntegrationTests.java | 87 +++++++++ .../DefaultStringRedisConnectionTests.java | 32 ++++ .../connection/RedisConnectionUnitTests.java | 17 +- .../DefaultHyperLogLogOperationsTests.java | 177 ++++++++++++++++++ 16 files changed, 795 insertions(+), 15 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java create mode 100644 src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java create mode 100644 src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java create mode 100644 src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 1bafa35f7..7c0219857 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -51,6 +51,8 @@ import org.springframework.util.Assert; */ public class DefaultStringRedisConnection implements StringRedisConnection { + private static final byte[][] EMPTY_2D_BYTE_ARRAY = new byte[0][]; + private final Log log = LogFactory.getLog(DefaultStringRedisConnection.class); private final RedisConnection delegate; private final RedisSerializer serializer; @@ -1247,6 +1249,11 @@ public class DefaultStringRedisConnection implements StringRedisConnection { } private byte[][] serializeMulti(String... keys) { + + if (keys == null) { + return EMPTY_2D_BYTE_ARRAY; + } + byte[][] ret = new byte[keys.length][]; for (int i = 0; i < ret.length; i++) { @@ -2413,12 +2420,12 @@ public class DefaultStringRedisConnection implements StringRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + Set results = delegate.zRangeByScore(key, min, max); if (isFutureConversion()) { addResultConverter(identityConverter); } - + return results; } @@ -2429,8 +2436,62 @@ public class DefaultStringRedisConnection implements StringRedisConnection { if (isFutureConversion()) { addResultConverter(identityConverter); } - + return results; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + return delegate.pfAdd(key, values); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfAdd(java.lang.String, java.lang.String[]) + */ + @Override + public Long pfAdd(String key, String... values) { + return this.pfAdd(serialize(key), serializeMulti(values)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + return delegate.pfCount(keys); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfCount(java.lang.String[]) + */ + @Override + public Long pfCount(String... keys) { + return this.pfCount(serializeMulti(keys)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + delegate.pfMerge(destinationKey, sourceKeys); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#pfMerge(java.lang.String, java.lang.String[][]) + */ + @Override + public void pfMerge(String destinationKey, String... sourceKeys) { + this.pfMerge(serialize(destinationKey), serializeMulti(sourceKeys)); + } + } diff --git a/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java b/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java new file mode 100644 index 000000000..63cb2de6e --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/HyperLogLogCommands.java @@ -0,0 +1,51 @@ +/* + * Copyright 2014 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 + * + * 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.springframework.data.redis.connection; + +/** + * {@literal HyperLogLog} specific commands supported by Redis. + * + * @author Christoph Strobl + * @since 1.5 + */ +public interface HyperLogLogCommands { + + /** + * Adds given {@literal values} to the HyperLogLog stored at given {@literal key}. + * + * @param key + * @param values + * @return + */ + Long pfAdd(byte[] key, byte[]... values); + + /** + * Return the approximated cardinality of the structures observed by the HyperLogLog at {@literal key(s)}. + * + * @param keys + * @return + */ + Long pfCount(byte[]... keys); + + /** + * Merge N different HyperLogLogs at {@literal sourceKeys} into a single {@literal destinationKey}. + * + * @param destinationKey + * @param sourceKeys + */ + void pfMerge(byte[] destinationKey, byte[]... sourceKeys); + +} diff --git a/src/main/java/org/springframework/data/redis/connection/RedisCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisCommands.java index 24283111a..1da8cda6f 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisCommands.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2013 the original author or authors. + * Copyright 2011-2014 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. @@ -20,10 +20,11 @@ package org.springframework.data.redis.connection; * Interface for the commands supported by Redis. * * @author Costin Leau + * @author Christoph Strobl */ public interface RedisCommands extends RedisKeyCommands, RedisStringCommands, RedisListCommands, RedisSetCommands, RedisZSetCommands, RedisHashCommands, RedisTxCommands, RedisPubSubCommands, RedisConnectionCommands, - RedisServerCommands, RedisScriptingCommands { + RedisServerCommands, RedisScriptingCommands, HyperLogLogCommands { /** * 'Native' or 'raw' execution of the given command along-side the given arguments. The command is executed as is, diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index 9b0c70d5e..a69410be8 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -35,7 +35,6 @@ import org.springframework.data.redis.serializer.RedisSerializer; * @author Christoph Strobl * @author Thomas Darimont * @author David Liu - * * @see RedisCallback * @see RedisSerializer * @see StringRedisTemplate @@ -368,4 +367,28 @@ public interface StringRedisConnection extends RedisConnection { * @return */ Set zRangeByScore(String key, String min, String max, long offset, long count); + + /** + * Adds given {@literal values} to the HyperLogLog stored at given {@literal key}. + * + * @param key + * @param values + * @return + * @since 1.5 + */ + Long pfAdd(String key, String... values); + + /** + * @param keys + * @return + * @since 1.5 + */ + Long pfCount(String... keys); + + /** + * @param destinationKey + * @param sourceKeys + * @since 1.5 + */ + void pfMerge(String destinationKey, String... sourceKeys); } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 4de105420..f3fa64d6d 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -3166,7 +3166,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -3185,7 +3185,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -3196,7 +3196,80 @@ public class JedisConnection extends AbstractRedisConnection { transaction(new JedisResult(transaction.zrangeByScore(keyStr, min, max, (int) offset, (int) count))); return null; } - return JedisConverters.stringSetToByteSet().convert(jedis.zrangeByScore(keyStr, min, max, (int) offset, (int) count)); + return JedisConverters.stringSetToByteSet().convert( + jedis.zrangeByScore(keyStr, min, max, (int) offset, (int) count)); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + + Assert.notEmpty(values, "PFADD requires at least one non 'null' value."); + Assert.noNullElements(values, "Values for PFADD must not contain 'null'."); + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfadd(key, values))); + return null; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfadd(key, values))); + return null; + } + return jedis.pfadd(key, values); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + + Assert.notEmpty(keys, "PFCOUNT requires at least one non 'null' key."); + Assert.noNullElements(keys, "Keys for PFOUNT must not contain 'null'."); + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfcount(keys))); + return null; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfcount(keys))); + return null; + } + return jedis.pfcount(keys); + } catch (Exception ex) { + throw convertJedisAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + + try { + if (isPipelined()) { + pipeline(new JedisResult(pipeline.pfmerge(destinationKey, sourceKeys))); + return; + } + if (isQueueing()) { + transaction(new JedisResult(transaction.pfmerge(destinationKey, sourceKeys))); + return; + } + jedis.pfmerge(destinationKey, sourceKeys); } catch (Exception ex) { throw convertJedisAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java index eb36ea8ef..cca0a97c5 100644 --- a/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java @@ -1282,7 +1282,7 @@ public class JredisConnection extends AbstractRedisConnection { public Cursor> hScan(byte[] key, ScanOptions options) { throw new UnsupportedOperationException("'HSCAN' command is not uspported for jredis"); } - + @Override public Set zRangeByScore(byte[] key, String min, String max) { throw new UnsupportedOperationException("'zRangeByScore' command is not uspported for jredis"); @@ -1292,4 +1292,31 @@ public class JredisConnection extends AbstractRedisConnection { public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { throw new UnsupportedOperationException("'zRangeByScore' command is not uspported for jredis"); } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 30f5b58b1..08fefdad8 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -3618,7 +3618,7 @@ public class LettuceConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { if (isPipelined()) { pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max), @@ -3638,7 +3638,7 @@ public class LettuceConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { - + try { if (isPipelined()) { pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max, offset, count), @@ -3655,5 +3655,32 @@ public class LettuceConnection extends AbstractRedisConnection { throw convertLettuceAccessException(ex); } } - + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } + } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index 2addbea8e..bdd4470e7 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -2474,7 +2474,7 @@ public class SrpConnection extends AbstractRedisConnection { @Override public Set zRangeByScore(byte[] key, String min, String max) { - + try { String keyStr = new String(key, "UTF-8"); if (isPipelined()) { @@ -2503,4 +2503,31 @@ public class SrpConnection extends AbstractRedisConnection { throw convertSrpAccessException(ex); } } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfAdd(byte[], byte[][]) + */ + @Override + public Long pfAdd(byte[] key, byte[]... values) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfCount(byte[][]) + */ + @Override + public Long pfCount(byte[]... keys) { + throw new UnsupportedOperationException(); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.HyperLogLogCommands#pfMerge(byte[], byte[][]) + */ + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + throw new UnsupportedOperationException(); + } } diff --git a/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java new file mode 100644 index 000000000..37b0fe8a0 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/core/DefaultHyperLogLogOperations.java @@ -0,0 +1,102 @@ +/* + * Copyright 2014 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 + * + * 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.springframework.data.redis.core; + +import java.util.Arrays; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.RedisConnection; + +/** + * @author Christoph Strobl + * @since 1.5 + * @param + * @param + */ +public class DefaultHyperLogLogOperations extends AbstractOperations implements HyperLogLogOperations { + + public DefaultHyperLogLogOperations(RedisTemplate template) { + super(template); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#add(java.lang.Object, java.lang.Object[]) + */ + @Override + public Long add(K key, V... values) { + + final byte[] rawKey = rawKey(key); + final byte[][] rawValues = rawValues(values); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + return connection.pfAdd(rawKey, rawValues); + } + }, true); + + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#size(java.lang.Object[]) + */ + @Override + public Long size(K... keys) { + + final byte[][] rawKeys = rawKeys(Arrays.asList(keys)); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + return connection.pfCount(rawKeys); + } + }, true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#union(java.lang.Object, java.lang.Object[]) + */ + @Override + public Long union(K destination, K... sourceKeys) { + + final byte[] rawDestinationKey = rawKey(destination); + final byte[][] rawSourceKeys = rawKeys(Arrays.asList(sourceKeys)); + + return execute(new RedisCallback() { + + @Override + public Long doInRedis(RedisConnection connection) throws DataAccessException { + + connection.pfMerge(rawDestinationKey, rawSourceKeys); + return connection.pfCount(rawDestinationKey); + } + }, true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.HyperLogLogOperations#delete(java.lang.Object) + */ + @Override + public void delete(K key) { + template.delete(key); + } +} diff --git a/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java b/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java new file mode 100644 index 000000000..54ef650e7 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/core/HyperLogLogOperations.java @@ -0,0 +1,56 @@ +/* + * Copyright 2014 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 + * + * 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.springframework.data.redis.core; + +/** + * @author Christoph Strobl + * @since 1.5 + */ +public interface HyperLogLogOperations { + + /** + * Adds the given {@literal values} to the {@literal key}. + * + * @param key must not be {@literal null}. + * @param values must not be {@literal null}. + * @return 1 of at least one of the values was added to the key; 0 otherwise. + */ + Long add(K key, V... values); + + /** + * Gets the current number of elements within the {@literal key}. + * + * @param keys must not be {@literal null) or {@literal empty}. + * @return + */ + Long size(K... keys); + + /** + * Merges all values of given {@literal sourceKeys} into {@literal destination} key. + * + * @param destination key of HyperLogLog to move source keys into. + * @param sourceKeys must not be {@literal null) or {@literal empty}. + */ + Long union(K destination, K... sourceKeys); + + /** + * Removes the given {@literal key}. + * + * @param key must not be {@literal null}. + */ + void delete(K key); + +} diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java index ab49a6564..8cd506d87 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java @@ -250,6 +250,12 @@ public interface RedisOperations { */ ZSetOperations opsForZSet(); + /** + * @return + * @since 1.5 + */ + HyperLogLogOperations opsForHyperLogLog(); + /** * Returns the operations performed on zset values (also known as sorted sets) bound to the given key. * diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index fe7673a51..4086ff5d7 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -96,6 +96,7 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation private ListOperations listOps; private SetOperations setOps; private ZSetOperations zSetOps; + private HyperLogLogOperations hllOps; /** * Constructs a new RedisTemplate instance. @@ -996,6 +997,19 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation return zSetOps; } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.RedisOperations#opsForHyperLogLog() + */ + @Override + public HyperLogLogOperations opsForHyperLogLog() { + + if (hllOps == null) { + hllOps = new DefaultHyperLogLogOperations(this); + } + return hllOps; + } + public BoundHashOperations boundHashOps(K key) { return new DefaultBoundHashOperations(key, this); } @@ -1075,4 +1089,5 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation public void setEnableTransactionSupport(boolean enableTransactionSupport) { this.enableTransactionSupport = enableTransactionSupport; } + } diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 5fc2000fd..7938eed8b 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -2067,6 +2067,93 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(i, is(3)); } + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldAddToNonExistingKeyCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldReturnZeroWhenValueAlreadyExists() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfAdd("hll2", "c", "d", "e")); + actual.add(connection.pfAdd("hll2", "e")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(1L)); + assertThat((Long) results.get(2), is(0L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountShouldReturnCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfCount("hll")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(3L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountWithMultipleKeysShouldReturnCorrectly() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfAdd("hll", "a", "b", "c")); + actual.add(connection.pfAdd("hll2", "d", "e", "f")); + actual.add(connection.pfCount("hll", "hll2")); + + List results = getResults(); + assertThat((Long) results.get(0), is(1L)); + assertThat((Long) results.get(1), is(1L)); + assertThat((Long) results.get(2), is(6L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test(expected = IllegalArgumentException.class) + public void pfCountWithNullKeysShouldThrowIllegalArgumentException() { + + if (!ConnectionUtils.isJedis(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis"); + } + + actual.add(connection.pfCount((String[]) null)); + } + protected void verifyResults(List expected) { assertEquals(expected, getResults()); } diff --git a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java index 162b5f79a..81230bf10 100644 --- a/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/DefaultStringRedisConnectionTests.java @@ -1720,6 +1720,38 @@ public class DefaultStringRedisConnectionTests { verify(nativeConnection, times(1)).setClientName(eq("foo".getBytes())); } + /** + * @see DATAREDIS-308 + */ + @Test + public void pfAddShouldDelegateToNativeConnectionCorrectly() { + + connection.pfAdd("hll", "spring", "data", "redis"); + verify(nativeConnection, times(1)).pfAdd("hll".getBytes(), "spring".getBytes(), "data".getBytes(), + "redis".getBytes()); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfCountShouldDelegateToNativeConnectionCorrectly() { + + connection.pfCount("hll", "hyperLogLog"); + verify(nativeConnection, times(1)).pfCount("hll".getBytes(), "hyperLogLog".getBytes()); + } + + /** + * @see DATAREDIS-308 + */ + @Test + public void pfMergeShouldDelegateToNativeConnectionCorrectly() { + + connection.pfMerge("merged", "spring", "data", "redis"); + verify(nativeConnection, times(1)).pfMerge("merged".getBytes(), "spring".getBytes(), "data".getBytes(), + "redis".getBytes()); + } + /** * @see DATAREDIS-270 */ diff --git a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java index 935023ae4..bb89db8ea 100644 --- a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java @@ -791,7 +791,7 @@ public class RedisConnectionUnitTests { public T evalSha(byte[] scriptSha, ReturnType returnType, int numKeys, byte[]... keysAndArgs) { return delegate.evalSha(scriptSha, returnType, numKeys, keysAndArgs); } - + @Override public Set zRangeByScore(byte[] key, String min, String max) { return delegate.zRangeByScore(key, min, max); @@ -801,5 +801,20 @@ public class RedisConnectionUnitTests { public Set zRangeByScore(byte[] key, String min, String max, long offset, long count) { return delegate.zRangeByScore(key, min, max, offset, count); } + + @Override + public Long pfAdd(byte[] key, byte[]... values) { + return delegate.pfAdd(key, values); + } + + @Override + public Long pfCount(byte[]... keys) { + return delegate.pfCount(keys); + } + + @Override + public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) { + delegate.pfMerge(destinationKey, sourceKeys); + } } } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java new file mode 100644 index 000000000..eebca3412 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java @@ -0,0 +1,177 @@ +/* + * Copyright 2014 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 + * + * 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.springframework.data.redis.core; + +import static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.*; + +import java.util.Collection; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.data.redis.ObjectFactory; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.test.util.MinimumRedisVersionRule; +import org.springframework.test.annotation.IfProfileValue; + +/** + * @author Christoph Strobl + */ +@RunWith(Parameterized.class) +public class DefaultHyperLogLogOperationsTests { + + private RedisTemplate redisTemplate; + + private ObjectFactory keyFactory; + + private ObjectFactory valueFactory; + + private HyperLogLogOperations hyperLogLogOps; + + public @Rule MinimumRedisVersionRule versionRule = new MinimumRedisVersionRule(); + + public DefaultHyperLogLogOperationsTests(RedisTemplate redisTemplate, ObjectFactory keyFactory, + ObjectFactory valueFactory) { + this.redisTemplate = redisTemplate; + this.keyFactory = keyFactory; + this.valueFactory = valueFactory; + } + + @Parameters + public static Collection testParams() { + return AbstractOperationsTestParams.testParams(); + } + + @Before + public void setUp() { + hyperLogLogOps = redisTemplate.opsForHyperLogLog(); + } + + @After + public void tearDown() { + redisTemplate.execute(new RedisCallback() { + public Object doInRedis(RedisConnection connection) { + connection.flushDb(); + return null; + } + }); + } + + /** + * @see DATAREDIS + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void addShouldAddDistinctValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + assertThat(hyperLogLogOps.add(key, v1, v2, v3), equalTo(1L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void addShouldNotAddExistingValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + assertThat(hyperLogLogOps.add(key, v2), equalTo(0L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void sizeShouldCountValuesCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + assertThat(hyperLogLogOps.size(key), equalTo(3L)); + } + + /** + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void sizeShouldCountValuesOfMultipleKeysCorrectly() { + + K key = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + K key2 = keyFactory.instance(); + V v4 = valueFactory.instance(); + + hyperLogLogOps.add(key, v1, v2, v3); + hyperLogLogOps.add(key2, v4); + assertThat(hyperLogLogOps.size(key, key2), equalTo(4L)); + } + + /** + * @throws InterruptedException + * @see DATAREDIS-308 + */ + @Test + @SuppressWarnings("unchecked") + @IfProfileValue(name = "redisVersion", value = "2.8+") + public void unionShouldMergeValuesOfMultipleKeysCorrectly() throws InterruptedException { + + K sourceKey_1 = keyFactory.instance(); + V v1 = valueFactory.instance(); + V v2 = valueFactory.instance(); + V v3 = valueFactory.instance(); + + K sourceKey_2 = keyFactory.instance(); + V v4 = valueFactory.instance(); + + K desinationKey = keyFactory.instance(); + + hyperLogLogOps.add(sourceKey_1, v1, v2, v3); + hyperLogLogOps.add(sourceKey_2, v4); + + Thread.sleep(10); // give redis a little time to catch up + hyperLogLogOps.union(desinationKey, sourceKey_1, sourceKey_2); + Thread.sleep(10); // give redis a little time to catch up + + assertThat(hyperLogLogOps.size(desinationKey), equalTo(4L)); + } +}