Preserve null values when parsing SearchReplies (#3518)

EncodedComplexOutput was skipping null values instead of passing them on. Then SearchReplyParser needs to store null values as they are and not try to decode them.
This affected both RESP2 and RESP3 parsing.

Added two integration tests in RediSearchAggregateIntegrationTests to verify that nulls in JSON documents are parsed correctly.

Author: mhyllander

src/main/java/io/lettuce/core/output/EncodedComplexOutput.java

@@ -25,16 +25,12 @@ public EncodedComplexOutput(RedisCodec<K, V> codec, ComplexDataParser<T> parser)
 
     @Override
     public void set(ByteBuffer bytes) {
-        if (bytes != null) {
-            data.storeObject(bytes.asReadOnlyBuffer());
-        }
+        data.storeObject(bytes == null ? null : bytes.asReadOnlyBuffer());
     }
 
     @Override
     public void setSingle(ByteBuffer bytes) {
-        if (bytes != null) {
-            data.storeObject(bytes.asReadOnlyBuffer());
-        }
+        data.storeObject(bytes == null ? null : bytes.asReadOnlyBuffer());
     }
 
 }

src/main/java/io/lettuce/core/search/SearchReplyParser.java

@@ -186,11 +186,13 @@ private void parseResults(SearchReply<K, V> searchReply, List<Object> resultsLis
                     ComplexData resultData = (ComplexData) resultsList.get(i);
                     List<Object> resultEntries = resultData.getDynamicList();
 
-                    Map<K, V> resultEntriesProcessed = IntStream.range(0, resultEntries.size() / 2).boxed()
-                            .collect(Collectors.toMap(idx -> codec.decodeKey((ByteBuffer) resultEntries.get(idx * 2)),
-                                    idx -> codec.decodeValue((ByteBuffer) resultEntries.get(idx * 2 + 1))));
+                    for (int idx = 0; idx < resultEntries.size(); idx += 2) {
+                        K decodedKey = codec.decodeKey((ByteBuffer) resultEntries.get(idx));
+                        Object value = resultEntries.get(idx + 1);
+                        V decodedValue = value == null ? null : codec.decodeValue((ByteBuffer) value);
+                        searchResult.addFields(decodedKey, decodedValue);
+                    }
 
-                    searchResult.addFields(resultEntriesProcessed);
                     i++;
                 }
 
@@ -265,7 +267,7 @@ public SearchReply<K, V> parse(ComplexData data) {
                         ComplexData extraAttributes = (ComplexData) resultEntry.get(EXTRA_ATTRIBUTES_KEY);
                         extraAttributes.getDynamicMap().forEach((key, value) -> {
                             K decodedKey = codec.decodeKey((ByteBuffer) key);
-                            V decodedValue = codec.decodeValue((ByteBuffer) value);
+                            V decodedValue = value == null ? null : codec.decodeValue((ByteBuffer) value);
                             searchResult.addFields(decodedKey, decodedValue);
                         });
                     }