RediSearch support (part 1 of 3) (#3325)

* Revert "Revert "Add support for FT.CREATE #2717 (#3150)" (#3160)" (#3161)

This reverts commit 319e315.

* Add support for FT.DROPINDEX #2722 (#3164)

* Add support for FT.DROPINDEX #2722

* Polishing

* FT.SEARCH added

Vibe code the FT.SEARCH command
Pulled latest from the integration branch

Fixed a lot of the hallucinations

RESP2 parser improvements

JSON Indexing tests

Add some advanced use-cases

Fixed all integration tests

Fixed schema fields, added vector tests, but they are all failing

Fixed vector search tests

Improve coverage of integration and unit tests

One unit test fails

* API cleanup, added Kotlin implementation

* Implement the FT.AGGREGATE command

Fixing the integration tests for FT.AGGREGATE

Polishing 1/3

Add more documentation ot the AggregateArgs

* Implement the FT.CURSOR command

Add Integration tests for the FT.CURSOR

Add RESP2 tests for all aggragation integration tests

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

* Fixing unit tests (part1)
Fixing unit tests (part2)
Fixing unit tests (part3)

* Polishing (part 1)

Author: tishun

.github/workflows/integration.yml

@@ -8,10 +8,12 @@ on:
     branches:
       - main
       - '[0-9].*'
+      - 'feature/*'
   pull_request:
     branches:
       - main
       - '[0-9].*'
+      - 'feature/*'
   schedule:
     - cron: '0 1 * * *' # nightly build
   workflow_dispatch:

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

@@ -48,6 +48,12 @@
 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;
+import io.lettuce.core.search.arguments.FieldArgs;
+import io.lettuce.core.search.arguments.SearchArgs;
 import io.lettuce.core.vector.RawVector;
 import io.lettuce.core.vector.VectorMetadata;
 
@@ -82,14 +88,16 @@ public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncC
         RedisSortedSetAsyncCommands<K, V>, RedisScriptingAsyncCommands<K, V>, RedisServerAsyncCommands<K, V>,
         RedisHLLAsyncCommands<K, V>, BaseRedisAsyncCommands<K, V>, RedisTransactionalAsyncCommands<K, V>,
         RedisGeoAsyncCommands<K, V>, RedisClusterAsyncCommands<K, V>, RedisJsonAsyncCommands<K, V>,
-        RedisVectorSetAsyncCommands<K, V> {
+        RedisVectorSetAsyncCommands<K, V>, RediSearchAsyncCommands<K, V> {
 
     private final StatefulConnection<K, V> connection;
 
     private final RedisCommandBuilder<K, V> commandBuilder;
 
     private final RedisJsonCommandBuilder<K, V> jsonCommandBuilder;
 
+    private final RediSearchCommandBuilder<K, V> searchCommandBuilder;
+
     private final RedisVectorSetCommandBuilder<K, V> vectorSetCommandBuilder;
 
     private final Supplier<JsonParser> parser;
@@ -108,6 +116,7 @@ public AbstractRedisAsyncCommands(StatefulConnection<K, V> connection, RedisCode
         this.commandBuilder = new RedisCommandBuilder<>(codec);
         this.jsonCommandBuilder = new RedisJsonCommandBuilder<>(codec, parser);
         this.vectorSetCommandBuilder = new RedisVectorSetCommandBuilder<>(codec, parser);
+        this.searchCommandBuilder = new RediSearchCommandBuilder<>(codec);
     }
 
     /**
@@ -1540,6 +1549,61 @@ public boolean isOpen() {
         return connection.isOpen();
     }
 
+    @Override
+    public RedisFuture<String> ftCreate(K index, CreateArgs<K, V> options, List<FieldArgs<K>> fieldArgs) {
+        return dispatch(searchCommandBuilder.ftCreate(index, options, fieldArgs));
+    }
+
+    @Override
+    public RedisFuture<String> ftCreate(K index, List<FieldArgs<K>> fieldArgs) {
+        return dispatch(searchCommandBuilder.ftCreate(index, null, fieldArgs));
+    }
+
+    @Override
+    public RedisFuture<String> ftDropindex(K index, boolean deleteDocumentKeys) {
+        return dispatch(searchCommandBuilder.ftDropindex(index, deleteDocumentKeys));
+    }
+
+    @Override
+    public RedisFuture<String> ftDropindex(K index) {
+        return dispatch(searchCommandBuilder.ftDropindex(index, false));
+    }
+
+    @Override
+    public RedisFuture<SearchReply<K, V>> ftSearch(K index, V query, SearchArgs<K, V> args) {
+        return dispatch(searchCommandBuilder.ftSearch(index, query, args));
+    }
+
+    @Override
+    public RedisFuture<SearchReply<K, V>> ftSearch(K index, V query) {
+        return dispatch(searchCommandBuilder.ftSearch(index, query, SearchArgs.<K, V> builder().build()));
+    }
+
+    @Override
+    public RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
+        return dispatch(searchCommandBuilder.ftAggregate(index, query, args));
+    }
+
+    @Override
+    public RedisFuture<AggregationReply<K, V>> ftAggregate(K index, V query) {
+        return dispatch(searchCommandBuilder.ftAggregate(index, query, null));
+    }
+
+    @Override
+    public RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId, count));
+    }
+
+    @Override
+    public RedisFuture<AggregationReply<K, V>> ftCursorread(K index, long cursorId) {
+        return dispatch(searchCommandBuilder.ftCursorread(index, cursorId, -1));
+    }
+
+    @Override
+    public RedisFuture<String> ftCursordel(K index, long cursorId) {
+        return dispatch(searchCommandBuilder.ftCursordel(index, cursorId));
+    }
+
     @Override
     public RedisFuture<List<Long>> jsonArrappend(K key, JsonPath jsonPath, JsonValue... values) {
         return dispatch(jsonCommandBuilder.jsonArrappend(key, jsonPath, values));

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

@@ -49,6 +49,12 @@
 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;
+import io.lettuce.core.search.arguments.FieldArgs;
+import io.lettuce.core.search.arguments.SearchArgs;
 import io.lettuce.core.tracing.TraceContext;
 import io.lettuce.core.tracing.TraceContextProvider;
 import io.lettuce.core.tracing.Tracing;
@@ -86,19 +92,22 @@
  * @author Tihomir Mateev
  * @since 4.0
  */
-public abstract class AbstractRedisReactiveCommands<K, V> implements RedisAclReactiveCommands<K, V>,
-        RedisHashReactiveCommands<K, V>, RedisKeyReactiveCommands<K, V>, RedisStringReactiveCommands<K, V>,
-        RedisListReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>, RedisSortedSetReactiveCommands<K, V>,
-        RedisScriptingReactiveCommands<K, V>, RedisServerReactiveCommands<K, V>, RedisHLLReactiveCommands<K, V>,
-        BaseRedisReactiveCommands<K, V>, RedisTransactionalReactiveCommands<K, V>, RedisGeoReactiveCommands<K, V>,
-        RedisClusterReactiveCommands<K, V>, RedisJsonReactiveCommands<K, V>, RedisVectorSetReactiveCommands<K, V> {
+public abstract class AbstractRedisReactiveCommands<K, V>
+        implements RedisAclReactiveCommands<K, V>, RedisHashReactiveCommands<K, V>, RedisKeyReactiveCommands<K, V>,
+        RedisStringReactiveCommands<K, V>, RedisListReactiveCommands<K, V>, RedisSetReactiveCommands<K, V>,
+        RedisSortedSetReactiveCommands<K, V>, RedisScriptingReactiveCommands<K, V>, RedisServerReactiveCommands<K, V>,
+        RedisHLLReactiveCommands<K, V>, BaseRedisReactiveCommands<K, V>, RedisTransactionalReactiveCommands<K, V>,
+        RedisGeoReactiveCommands<K, V>, RedisClusterReactiveCommands<K, V>, RedisJsonReactiveCommands<K, V>,
+        RedisVectorSetReactiveCommands<K, V>, RediSearchReactiveCommands<K, V> {
 
     private final StatefulConnection<K, V> connection;
 
     private final RedisCommandBuilder<K, V> commandBuilder;
 
     private final RedisJsonCommandBuilder<K, V> jsonCommandBuilder;
 
+    private final RediSearchCommandBuilder<K, V> searchCommandBuilder;
+
     private final RedisVectorSetCommandBuilder<K, V> vectorSetCommandBuilder;
 
     private final Supplier<JsonParser> parser;
@@ -123,6 +132,7 @@ public AbstractRedisReactiveCommands(StatefulConnection<K, V> connection, RedisC
         this.commandBuilder = new RedisCommandBuilder<>(codec);
         this.jsonCommandBuilder = new RedisJsonCommandBuilder<>(codec, parser);
         this.vectorSetCommandBuilder = new RedisVectorSetCommandBuilder<>(codec, parser);
+        this.searchCommandBuilder = new RediSearchCommandBuilder<>(codec);
         this.clientResources = connection.getResources();
         this.tracingEnabled = clientResources.tracing().isEnabled();
     }
@@ -1604,6 +1614,61 @@ public boolean isOpen() {
         return connection.isOpen();
     }
 
+    @Override
+    public Mono<String> ftCreate(K index, CreateArgs<K, V> options, List<FieldArgs<K>> fieldArgs) {
+        return createMono(() -> searchCommandBuilder.ftCreate(index, options, fieldArgs));
+    }
+
+    @Override
+    public Mono<String> ftCreate(K index, List<FieldArgs<K>> fieldArgs) {
+        return createMono(() -> searchCommandBuilder.ftCreate(index, null, fieldArgs));
+    }
+
+    @Override
+    public Mono<String> ftCursordel(K index, long cursorId) {
+        return createMono(() -> searchCommandBuilder.ftCursordel(index, cursorId));
+    }
+
+    @Override
+    public Mono<String> ftDropindex(K index, boolean deleteDocumentKeys) {
+        return createMono(() -> searchCommandBuilder.ftDropindex(index, deleteDocumentKeys));
+    }
+
+    @Override
+    public Mono<String> ftDropindex(K index) {
+        return createMono(() -> searchCommandBuilder.ftDropindex(index, false));
+    }
+
+    @Override
+    public Mono<SearchReply<K, V>> ftSearch(K index, V query, SearchArgs<K, V> args) {
+        return createMono(() -> searchCommandBuilder.ftSearch(index, query, args));
+    }
+
+    @Override
+    public Mono<SearchReply<K, V>> ftSearch(K index, V query) {
+        return createMono(() -> searchCommandBuilder.ftSearch(index, query, SearchArgs.<K, V> builder().build()));
+    }
+
+    @Override
+    public Mono<AggregationReply<K, V>> ftAggregate(K index, V query, AggregateArgs<K, V> args) {
+        return createMono(() -> searchCommandBuilder.ftAggregate(index, query, args));
+    }
+
+    @Override
+    public Mono<AggregationReply<K, V>> ftAggregate(K index, V query) {
+        return createMono(() -> searchCommandBuilder.ftAggregate(index, query, null));
+    }
+
+    @Override
+    public Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId, int count) {
+        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId, count));
+    }
+
+    @Override
+    public Mono<AggregationReply<K, V>> ftCursorread(K index, long cursorId) {
+        return createMono(() -> searchCommandBuilder.ftCursorread(index, cursorId, -1));
+    }
+
     @Override
     public Flux<Long> jsonArrappend(K key, JsonPath jsonPath, JsonValue... values) {
         return createDissolvingFlux(() -> jsonCommandBuilder.jsonArrappend(key, jsonPath, values));