FT.AGGREGATE was never working in the first place, now it is working, but for RESP3 only

Author: tishun

src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java

@@ -48,6 +48,7 @@
 import io.lettuce.core.protocol.CommandType;
 import io.lettuce.core.protocol.ProtocolKeyword;
 import io.lettuce.core.protocol.RedisCommand;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -1579,23 +1580,23 @@ public RedisFuture<SearchReply<K, V>> ftSearch(K index, V query) {
     }
 
     @Override
-    public RedisFuture<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
+    public RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
         return dispatch(searchCommandBuilder.ftAggregate(index, query, args));
     }
 
     @Override
-    public RedisFuture<SearchReply<K, V>> ftAggregate(K index, V query) {
+    public RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query) {
         return dispatch(searchCommandBuilder.ftAggregate(index, query, null));
     }
 
     @Override
-    public RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+    public RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count) {
         return dispatch(searchCommandBuilder.ftCursorread(index, cursorId, count));
     }
 
     @Override
-    public RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId) {
-        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId));
+    public RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId) {
+        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId, -1));
     }
 
     @Override

src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java

@@ -49,6 +49,7 @@
 import io.lettuce.core.protocol.RedisCommand;
 import io.lettuce.core.protocol.TracedCommand;
 import io.lettuce.core.resource.ClientResources;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -1649,23 +1650,23 @@ public Mono<SearchReply<K, V>> ftSearch(K index, V query) {
     }
 
     @Override
-    public Mono<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
+    public Mono<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
         return createMono(() -> searchCommandBuilder.ftAggregate(index, query, args));
     }
 
     @Override
-    public Mono<SearchReply<K, V>> ftAggregate(K index, V query) {
+    public Mono<AggregationReply<K, V>> ftAggregate(K index, V query) {
         return createMono(() -> searchCommandBuilder.ftAggregate(index, query, null));
     }
 
     @Override
-    public Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+    public Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count) {
         return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId, count));
     }
 
     @Override
-    public Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId) {
-        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId));
+    public Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId) {
+        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId, -1));
     }
 
     @Override

src/main/java/io/lettuce/core/RediSearchCommandBuilder.java

@@ -13,6 +13,8 @@
 import io.lettuce.core.protocol.Command;
 import io.lettuce.core.protocol.CommandArgs;
 import io.lettuce.core.protocol.CommandKeyword;
+import io.lettuce.core.search.AggregateReplyParser;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.SearchReplyParser;
 import io.lettuce.core.search.arguments.AggregateArgs;
@@ -95,7 +97,7 @@ public Command<K, V, SearchReply<K, V>> ftSearch(K index, V query, SearchArgs<K,
      * @param aggregateArgs the aggregate arguments
      * @return the result of the aggregate command
      */
-    public Command<K, V, SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> aggregateArgs) {
+    public Command<K, V, AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> aggregateArgs) {
         notNullKey(index);
         notNullKey(query);
 
@@ -106,42 +108,42 @@ public Command<K, V, SearchReply<K, V>> ftAggregate(K index, V query, AggregateA
             aggregateArgs.build(args);
         }
 
-        return createCommand(FT_AGGREGATE, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
+        return createCommand(FT_AGGREGATE, new EncodedComplexOutput<>(codec, new AggregateReplyParser<>(codec, null)), args);
     }
 
     /**
      * Read next results from an existing cursor.
      *
      * @param index the index name
      * @param cursorId the cursor id
-     * @param count the number of results to read
      * @return the result of the cursor read command
      */
-    public Command<K, V, SearchReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+    public Command<K, V, AggregationReply<K, V>> ftCursorread(K index, long cursorId) {
         notNullKey(index);
 
         CommandArgs<K, V> args = new CommandArgs<>(codec).add(CommandKeyword.READ).addKey(index);
         args.add(cursorId);
-        args.add(CommandKeyword.COUNT);
-        args.add(count);
 
-        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
+        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new AggregateReplyParser<>(codec, null)), args);
     }
 
     /**
      * Read next results from an existing cursor.
      *
      * @param index the index name
      * @param cursorId the cursor id
+     * @param count the number of results to read
      * @return the result of the cursor read command
      */
-    public Command<K, V, SearchReply<K, V>> ftCursorread(K index, long cursorId) {
+    public Command<K, V, AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count) {
         notNullKey(index);
 
         CommandArgs<K, V> args = new CommandArgs<>(codec).add(CommandKeyword.READ).addKey(index);
         args.add(cursorId);
+        args.add(CommandKeyword.COUNT);
+        args.add(count);
 
-        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new SearchReplyParser<>(codec, null)), args);
+        return createCommand(FT_CURSOR, new EncodedComplexOutput<>(codec, new AggregateReplyParser<>(codec, null)), args);
     }
 
     /**

src/main/java/io/lettuce/core/api/async/RediSearchAsyncCommands.java

@@ -9,6 +9,7 @@
 import java.util.List;
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.annotations.Experimental;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -275,7 +276,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    RedisFuture<SearchReply<K, V>> ftAggregate(K index, V query);
+    RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query);
 
     /**
      * Run a search query on an index and perform advanced aggregate transformations with a processing pipeline.
@@ -331,7 +332,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftCursorread(Object, long)
      */
     @Experimental
-    RedisFuture<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
+    RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
 
     /**
      * Read next results from an existing cursor.
@@ -365,7 +366,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count);
+    RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count);
 
     /**
      * Read next results from an existing cursor using the default batch size.
@@ -398,7 +399,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    RedisFuture<SearchReply<K, V>> ftCursorread(K index, long cursorId);
+    RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId);
 
     /**
      * Delete a cursor and free its associated resources.

src/main/java/io/lettuce/core/api/reactive/RediSearchReactiveCommands.java

@@ -9,11 +9,13 @@
 import java.util.List;
 
 import io.lettuce.core.annotations.Experimental;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
 import io.lettuce.core.search.arguments.FieldArgs;
 import io.lettuce.core.search.arguments.SearchArgs;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -276,7 +278,7 @@ public interface RediSearchReactiveCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Mono<SearchReply<K, V>> ftAggregate(K index, V query);
+    Mono<AggregationReply<K, V>> ftAggregate(K index, V query);
 
     /**
      * Run a search query on an index and perform advanced aggregate transformations with a processing pipeline.
@@ -332,7 +334,7 @@ public interface RediSearchReactiveCommands<K, V> {
      * @see #ftCursorread(Object, long)
      */
     @Experimental
-    Mono<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
+    Mono<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
 
     /**
      * Read next results from an existing cursor.
@@ -366,7 +368,7 @@ public interface RediSearchReactiveCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count);
+    Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count);
 
     /**
      * Read next results from an existing cursor using the default batch size.
@@ -399,7 +401,7 @@ public interface RediSearchReactiveCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Mono<SearchReply<K, V>> ftCursorread(K index, long cursorId);
+    Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId);
 
     /**
      * Delete a cursor and free its associated resources.

src/main/java/io/lettuce/core/api/sync/RediSearchCommands.java

@@ -9,6 +9,7 @@
 import java.util.List;
 
 import io.lettuce.core.annotations.Experimental;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -275,7 +276,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    SearchReply<K, V> ftAggregate(K index, V query);
+    AggregationReply<K, V> ftAggregate(K index, V query);
 
     /**
      * Run a search query on an index and perform advanced aggregate transformations with a processing pipeline.
@@ -331,7 +332,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftCursorread(Object, long)
      */
     @Experimental
-    SearchReply<K, V> ftAggregate(K index, V query, AggregateArgs<K, V> args);
+    AggregationReply<K, V> ftAggregate(K index, V query, AggregateArgs<K, V> args);
 
     /**
      * Read next results from an existing cursor.
@@ -365,7 +366,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    SearchReply<K, V> ftCursorread(K index, long cursorId, int count);
+    AggregationReply<K, V> ftCursorread(K index, long cursorId, int count);
 
     /**
      * Read next results from an existing cursor using the default batch size.
@@ -398,7 +399,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    SearchReply<K, V> ftCursorread(K index, long cursorId);
+    AggregationReply<K, V> ftCursorread(K index, long cursorId);
 
     /**
      * Delete a cursor and free its associated resources.

src/main/java/io/lettuce/core/cluster/api/async/RediSearchAsyncCommands.java

@@ -9,6 +9,7 @@
 import java.util.List;
 
 import io.lettuce.core.annotations.Experimental;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -275,7 +276,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    AsyncExecutions<SearchReply<K, V>> ftAggregate(K index, V query);
+    AsyncExecutions<AggregationReply<K, V>> ftAggregate(K index, V query);
 
     /**
      * Run a search query on an index and perform advanced aggregate transformations with a processing pipeline.
@@ -331,7 +332,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftCursorread(Object, long)
      */
     @Experimental
-    AsyncExecutions<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
+    AsyncExecutions<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
 
     /**
      * Read next results from an existing cursor.
@@ -365,7 +366,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    AsyncExecutions<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count);
+    AsyncExecutions<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count);
 
     /**
      * Read next results from an existing cursor using the default batch size.
@@ -398,7 +399,7 @@ public interface RediSearchAsyncCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    AsyncExecutions<SearchReply<K, V>> ftCursorread(K index, long cursorId);
+    AsyncExecutions<AggregationReply<K, V>> ftCursorread(K index, long cursorId);
 
     /**
      * Delete a cursor and free its associated resources.

src/main/java/io/lettuce/core/cluster/api/sync/RediSearchCommands.java

@@ -9,6 +9,7 @@
 import java.util.List;
 
 import io.lettuce.core.annotations.Experimental;
+import io.lettuce.core.search.AggregationReply;
 import io.lettuce.core.search.SearchReply;
 import io.lettuce.core.search.arguments.AggregateArgs;
 import io.lettuce.core.search.arguments.CreateArgs;
@@ -275,7 +276,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Executions<SearchReply<K, V>> ftAggregate(K index, V query);
+    Executions<AggregationReply<K, V>> ftAggregate(K index, V query);
 
     /**
      * Run a search query on an index and perform advanced aggregate transformations with a processing pipeline.
@@ -331,7 +332,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftCursorread(Object, long)
      */
     @Experimental
-    Executions<SearchReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
+    Executions<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args);
 
     /**
      * Read next results from an existing cursor.
@@ -365,7 +366,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Executions<SearchReply<K, V>> ftCursorread(K index, long cursorId, int count);
+    Executions<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count);
 
     /**
      * Read next results from an existing cursor using the default batch size.
@@ -398,7 +399,7 @@ public interface RediSearchCommands<K, V> {
      * @see #ftAggregate(Object, Object, AggregateArgs)
      */
     @Experimental
-    Executions<SearchReply<K, V>> ftCursorread(K index, long cursorId);
+    Executions<AggregationReply<K, V>> ftCursorread(K index, long cursorId);
 
     /**
      * Delete a cursor and free its associated resources.