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.

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<String> 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<byte[]> zRangeByScore(byte[] key, String min, String max) {
-		
+
 		Set<byte[]> 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));
+	}
+
 }

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);
+
+}

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,

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<byte[]> 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);
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -3166,7 +3166,7 @@ public class JedisConnection extends AbstractRedisConnection {
 
 	@Override
 	public Set<byte[]> 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<byte[]> 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);
 		}

src/main/java/org/springframework/data/redis/connection/jredis/JredisConnection.java

@@ -1282,7 +1282,7 @@ public class JredisConnection extends AbstractRedisConnection {
 	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
 		throw new UnsupportedOperationException("'HSCAN' command is not uspported for jredis");
 	}
-	
+
 	@Override
 	public Set<byte[]> 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<byte[]> 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();
+	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -3618,7 +3618,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 
 	@Override
 	public Set<byte[]> 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<byte[]> 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();
+	}
+
 }

src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java

@@ -2474,7 +2474,7 @@ public class SrpConnection extends AbstractRedisConnection {
 
 	@Override
 	public Set<byte[]> 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();
+	}
 }

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 <K>
+ * @param <V>
+ */
+public class DefaultHyperLogLogOperations<K, V> extends AbstractOperations<K, V> implements HyperLogLogOperations<K, V> {
+
+	public DefaultHyperLogLogOperations(RedisTemplate<K, V> 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<Long>() {
+
+			@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<Long>() {
+
+			@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<Long>() {
+
+			@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);
+	}
+}

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<K, V> {
+
+	/**
+	 * 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);
+
+}

src/main/java/org/springframework/data/redis/core/RedisOperations.java

@@ -250,6 +250,12 @@ public interface RedisOperations<K, V> {
 	 */
 	ZSetOperations<K, V> opsForZSet();
 
+	/**
+	 * @return
+	 * @since 1.5
+	 */
+	HyperLogLogOperations<K, V> opsForHyperLogLog();
+
 	/**
 	 * Returns the operations performed on zset values (also known as sorted sets) bound to the given key.
 	 * 

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -96,6 +96,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	private ListOperations<K, V> listOps;
 	private SetOperations<K, V> setOps;
 	private ZSetOperations<K, V> zSetOps;
+	private HyperLogLogOperations<K, V> hllOps;
 
 	/**
 	 * Constructs a new <code>RedisTemplate</code> instance.
@@ -996,6 +997,19 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return zSetOps;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#opsForHyperLogLog()
+	 */
+	@Override
+	public HyperLogLogOperations<K, V> opsForHyperLogLog() {
+
+		if (hllOps == null) {
+			hllOps = new DefaultHyperLogLogOperations<K, V>(this);
+		}
+		return hllOps;
+	}
+
 	public <HK, HV> BoundHashOperations<K, HK, HV> boundHashOps(K key) {
 		return new DefaultBoundHashOperations<K, HK, HV>(key, this);
 	}
@@ -1075,4 +1089,5 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 	public void setEnableTransactionSupport(boolean enableTransactionSupport) {
 		this.enableTransactionSupport = enableTransactionSupport;
 	}
+
 }

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<Object> 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<Object> 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<Object> 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<Object> 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<Object> expected) {
 		assertEquals(expected, getResults());
 	}

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
 	 */

src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java

@@ -791,7 +791,7 @@ public class RedisConnectionUnitTests {
 		public <T> T evalSha(byte[] scriptSha, ReturnType returnType, int numKeys, byte[]... keysAndArgs) {
 			return delegate.evalSha(scriptSha, returnType, numKeys, keysAndArgs);
 		}
-		
+
 		@Override
 		public Set<byte[]> zRangeByScore(byte[] key, String min, String max) {
 			return delegate.zRangeByScore(key, min, max);
@@ -801,5 +801,20 @@ public class RedisConnectionUnitTests {
 		public Set<byte[]> 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);
+		}
 	}
 }

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<K, V> {
+
+	private RedisTemplate<K, V> redisTemplate;
+
+	private ObjectFactory<K> keyFactory;
+
+	private ObjectFactory<V> valueFactory;
+
+	private HyperLogLogOperations<K, V> hyperLogLogOps;
+
+	public @Rule MinimumRedisVersionRule versionRule = new MinimumRedisVersionRule();
+
+	public DefaultHyperLogLogOperationsTests(RedisTemplate<K, V> redisTemplate, ObjectFactory<K> keyFactory,
+			ObjectFactory<V> valueFactory) {
+		this.redisTemplate = redisTemplate;
+		this.keyFactory = keyFactory;
+		this.valueFactory = valueFactory;
+	}
+
+	@Parameters
+	public static Collection<Object[]> testParams() {
+		return AbstractOperationsTestParams.testParams();
+	}
+
+	@Before
+	public void setUp() {
+		hyperLogLogOps = redisTemplate.opsForHyperLogLog();
+	}
+
+	@After
+	public void tearDown() {
+		redisTemplate.execute(new RedisCallback<Object>() {
+			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));
+	}
+}