Fix wrong aggregation arguments order

Author: tishun

src/main/java/io/lettuce/core/search/arguments/AggregateArgs.java

@@ -86,15 +86,11 @@ public class AggregateArgs<K, V> {
 
     private Optional<Duration> timeout = Optional.empty();
 
-    private final List<GroupBy<K, V>> groupByList = new ArrayList<>();
-
-    private final List<SortBy<K>> sortByList = new ArrayList<>();
-
-    private final List<Apply<K, V>> applyList = new ArrayList<>();
-
-    private Optional<Limit> limit = Optional.empty();
-
-    private final List<V> filters = new ArrayList<>();
+    /**
+     * Ordered list of pipeline operations (GROUPBY, SORTBY, APPLY, FILTER). These operations must be applied in the order
+     * specified by the user.
+     */
+    private final List<PipelineOperation<K, ?>> pipelineOperations = new ArrayList<>();
 
     private Optional<WithCursor> withCursor = Optional.empty();
 
@@ -214,7 +210,7 @@ public Builder<K, V> timeout(Duration timeout) {
          * @return the builder.
          */
         public Builder<K, V> groupBy(GroupBy<K, V> groupBy) {
-            args.groupByList.add(groupBy);
+            args.pipelineOperations.add(groupBy);
             return this;
         }
 
@@ -225,7 +221,7 @@ public Builder<K, V> groupBy(GroupBy<K, V> groupBy) {
          * @return the builder.
          */
         public Builder<K, V> sortBy(SortBy<K> sortBy) {
-            args.sortByList.add(sortBy);
+            args.pipelineOperations.add(sortBy);
             return this;
         }
 
@@ -236,7 +232,7 @@ public Builder<K, V> sortBy(SortBy<K> sortBy) {
          * @return the builder.
          */
         public Builder<K, V> apply(Apply<K, V> apply) {
-            args.applyList.add(apply);
+            args.pipelineOperations.add(apply);
             return this;
         }
 
@@ -268,7 +264,7 @@ public Builder<K, V> apply(Apply<K, V> apply) {
          * @return the builder.
          */
         public Builder<K, V> limit(long offset, long num) {
-            args.limit = Optional.of(new Limit(offset, num));
+            args.pipelineOperations.add(new Limit<>(offset, num));
             return this;
         }
 
@@ -302,7 +298,7 @@ public Builder<K, V> limit(long offset, long num) {
          * @return the builder.
          */
         public Builder<K, V> filter(V filter) {
-            args.filters.add(filter);
+            args.pipelineOperations.add(new Filter<>(filter));
             return this;
         }
 
@@ -513,32 +509,12 @@ public void build(CommandArgs<K, V> args) {
             args.add(t.toMillis());
         });
 
-        // Add GROUPBY clauses
-        for (GroupBy<K, V> groupBy : groupByList) {
-            groupBy.build(args);
-        }
-
-        // Add SORTBY clauses
-        for (SortBy<K> sortBy : sortByList) {
-            sortBy.build(args);
-        }
-
-        // Add APPLY clauses
-        for (Apply<K, V> apply : applyList) {
-            apply.build(args);
-        }
-
-        // Add LIMIT clause
-        limit.ifPresent(l -> {
-            args.add(CommandKeyword.LIMIT);
-            args.add(l.offset);
-            args.add(l.num);
-        });
-
-        // Add FILTER clauses
-        for (V filter : filters) {
-            args.add(CommandKeyword.FILTER);
-            args.addValue(filter);
+        // Add pipeline operations in user-specified order
+        for (PipelineOperation<K, ?> operation : pipelineOperations) {
+            // Cast is safe because all operations can build with CommandArgs<K, V>
+            @SuppressWarnings("unchecked")
+            PipelineOperation<K, V> typedOperation = (PipelineOperation<K, V>) operation;
+            typedOperation.build(args);
         }
 
         // Add WITHCURSOR clause
@@ -578,6 +554,21 @@ public Optional<WithCursor> getWithCursor() {
         return withCursor;
     }
 
+    /**
+     * Interface for pipeline operations that need to be applied in user-specified order. This includes GROUPBY, SORTBY, APPLY,
+     * and FILTER operations.
+     */
+    public interface PipelineOperation<K, V> {
+
+        /**
+         * Build the operation arguments into the command args.
+         * 
+         * @param args the command args to build into
+         */
+        void build(CommandArgs<K, V> args);
+
+    }
+
     // Helper classes
     public static class LoadField<K> {
 
@@ -592,7 +583,7 @@ public static class LoadField<K> {
 
     }
 
-    public static class Limit {
+    public static class Limit<K, V> implements PipelineOperation<K, V> {
 
         final long offset;
 
@@ -603,6 +594,13 @@ public static class Limit {
             this.num = num;
         }
 
+        @Override
+        public void build(CommandArgs<K, V> args) {
+            args.add(CommandKeyword.LIMIT);
+            args.add(offset);
+            args.add(num);
+        }
+
     }
 
     public static class WithCursor {
@@ -676,7 +674,7 @@ public static WithCursor of(Long count) {
      * performance.
      * </p>
      */
-    public static class GroupBy<K, V> {
+    public static class GroupBy<K, V> implements PipelineOperation<K, V> {
 
         private final List<K> properties;
 
@@ -705,6 +703,7 @@ public static <K, V> GroupBy<K, V> of(K... properties) {
             return new GroupBy<>(Arrays.asList(properties));
         }
 
+        @Override
         public void build(CommandArgs<K, V> args) {
             args.add(CommandKeyword.GROUPBY);
             args.add(properties.size());
@@ -764,7 +763,7 @@ public void build(CommandArgs<K, V> args) {
      * using LIMIT.
      * </p>
      */
-    public static class SortBy<K> {
+    public static class SortBy<K> implements PipelineOperation<K, Object> {
 
         private final List<SortProperty<K>> properties;
 
@@ -810,7 +809,8 @@ public static <K> SortBy<K> of(SortProperty<K>... properties) {
             return new SortBy<>(Arrays.asList(properties));
         }
 
-        public <V> void build(CommandArgs<K, V> args) {
+        @Override
+        public void build(CommandArgs<K, Object> args) {
             args.add(CommandKeyword.SORTBY);
             // Count includes property + direction pairs
             args.add(properties.size() * 2L);
@@ -880,7 +880,7 @@ public <V> void build(CommandArgs<K, V> args) {
      * can be referenced by further operations.
      * </p>
      */
-    public static class Apply<K, V> {
+    public static class Apply<K, V> implements PipelineOperation<K, V> {
 
         private final V expression;
 
@@ -891,6 +891,7 @@ public Apply(V expression, K name) {
             this.name = name;
         }
 
+        @Override
         public void build(CommandArgs<K, V> args) {
             args.add(CommandKeyword.APPLY);
             args.addValue(expression);
@@ -1065,6 +1066,31 @@ public void build(CommandArgs<K, V> args) {
 
     }
 
+    /**
+     * Represents a FILTER clause in an aggregation pipeline.
+     *
+     * <p>
+     * Filters the results using predicate expressions relating to values in each result. Filters are applied after the query
+     * and relate to the current state of the pipeline. This allows filtering on computed fields created by APPLY operations or
+     * reducer results.
+     * </p>
+     */
+    public static class Filter<K, V> implements PipelineOperation<K, V> {
+
+        private final V expression;
+
+        public Filter(V expression) {
+            this.expression = expression;
+        }
+
+        @Override
+        public void build(CommandArgs<K, V> args) {
+            args.add(CommandKeyword.FILTER);
+            args.addValue(expression);
+        }
+
+    }
+
     /**
      * Represents a sort property with direction.
      */

src/test/java/io/lettuce/core/RediSearchCommandBuilderUnitTests.java

@@ -635,6 +635,39 @@ void shouldCorrectlyConstructFtAggregateCommandBasic() {
         assertThat(buf.toString(StandardCharsets.UTF_8)).isEqualTo(result);
     }
 
+    @Test
+    void shouldMaintainPipelineOperationOrder() {
+        // Test that pipeline operations (GROUPBY, SORTBY, APPLY, FILTER, LIMIT)
+        // are output in the order specified by the user, not in a fixed order
+        AggregateArgs<String, String> aggregateArgs = AggregateArgs.<String, String> builder()//
+                .apply("@price * @quantity", "total_value")// First operation
+                .filter("@total_value > 100")// Second operation
+                .groupBy(GroupBy.<String, String> of("category").reduce(Reducer.<String, String> count().as("count")))// Third
+                                                                                                                      // operation
+                .limit(0, 5)// Fourth operation
+                .sortBy(SortBy.of("count", SortDirection.DESC))// Fifth operation
+                .build();
+
+        Command<String, String, AggregationReply<String, String>> command = builder.ftAggregate(MY_KEY, MY_QUERY,
+                aggregateArgs);
+        ByteBuf buf = Unpooled.directBuffer();
+        command.encode(buf);
+
+        // Expected order should match the user's call order: APPLY -> FILTER -> GROUPBY -> LIMIT -> SORTBY
+        String result = "*26\r\n" + "$12\r\n" + "FT.AGGREGATE\r\n" + "$3\r\n" + "idx\r\n" + "$1\r\n" + "*\r\n"//
+                + "$5\r\n" + "APPLY\r\n" + "$18\r\n" + "@price * @quantity\r\n" + "$2\r\n" + "AS\r\n" + "$11\r\n"
+                + "total_value\r\n"//
+                + "$6\r\n" + "FILTER\r\n" + "$18\r\n" + "@total_value > 100\r\n"//
+                + "$7\r\n" + "GROUPBY\r\n" + "$1\r\n" + "1\r\n" + "$9\r\n" + "@category\r\n"//
+                + "$6\r\n" + "REDUCE\r\n" + "$5\r\n" + "COUNT\r\n" + "$1\r\n" + "0\r\n" + "$2\r\n" + "AS\r\n" + "$5\r\n"
+                + "count\r\n"//
+                + "$5\r\n" + "LIMIT\r\n" + "$1\r\n" + "0\r\n" + "$1\r\n" + "5\r\n"//
+                + "$6\r\n" + "SORTBY\r\n" + "$1\r\n" + "2\r\n" + "$6\r\n" + "@count\r\n" + "$4\r\n" + "DESC\r\n"//
+                + "$7\r\n" + "DIALECT\r\n" + "$1\r\n2\r\n";//
+
+        assertThat(buf.toString(StandardCharsets.UTF_8)).isEqualTo(result);
+    }
+
     @Test
     void shouldCorrectlyConstructFtAggregateCommandWithArgs() {
         AggregateArgs<String, String> aggregateArgs = AggregateArgs.<String, String> builder()//