feat: add SearchStream return type support for repository methods

docs/content/modules/ROOT/pages/hash-mappings.adoc

@@ -54,7 +54,7 @@ Redis OM Spring creates full RediSearch indexes using `FT.CREATE` commands, prov
 
 |Query Methods
 |Limited to findBy patterns
-|Complex queries, @Query, Entity Streams
+|Complex queries, @Query, Entity Streams, SearchStream returns
 |===
 
 == Basic Usage
@@ -425,6 +425,75 @@ List<Person> admins = entityStream
   .collect(Collectors.toList());
 ----
 
+=== Entity Streams Integration with Repositories
+
+Repositories can return `SearchStream` for fluent query operations:
+
+[source,java]
+----
+import com.redis.om.spring.search.stream.SearchStream;
+
+public interface PersonRepository extends RedisEnhancedRepository<Person, String> {
+    // Return SearchStream for advanced operations
+    SearchStream<Person> findByDepartment(String department);
+
+    SearchStream<Person> findByAgeGreaterThan(int age);
+
+    SearchStream<Person> findByActive(boolean active);
+
+    // Usage example:
+    // SearchStream<Person> stream = repository.findByDepartment("Engineering");
+    // List<String> names = stream
+    //     .filter(Person$.ACTIVE.eq(true))
+    //     .map(Person$.NAME)
+    //     .collect(Collectors.toList());
+}
+----
+
+This allows you to combine repository query methods with the power of Entity Streams:
+
+[source,java]
+----
+@Service
+public class PersonService {
+    @Autowired
+    PersonRepository repository;
+
+    public List<String> getActiveEngineerNames() {
+        return repository.findByDepartment("Engineering")
+            .filter(Person$.ACTIVE.eq(true))
+            .map(Person$.NAME)
+            .sorted()
+            .collect(Collectors.toList());
+    }
+
+    public long countSeniorEmployees(int minAge) {
+        return repository.findByAgeGreaterThan(minAge)
+            .filter(Person$.DEPARTMENT.in("Engineering", "Management"))
+            .count();
+    }
+
+    public List<Person> getTopPerformers() {
+        return repository.findByActive(true)
+            .filter(Person$.PERFORMANCE_SCORE.gte(90))
+            .sorted(Person$.PERFORMANCE_SCORE, SortOrder.DESC)
+            .limit(10)
+            .collect(Collectors.toList());
+    }
+}
+----
+
+The `SearchStream` returned by repository methods supports all Entity Stream operations:
+
+* **Filtering**: `filter()` with field predicates
+* **Mapping**: `map()` to transform results
+* **Sorting**: `sorted()` with field and order
+* **Limiting**: `limit()` to restrict results
+* **Aggregation**: `count()`, `findFirst()`, `anyMatch()`, `allMatch()`
+* **Collection**: `collect()` to lists, sets, or custom collectors
+
+NOTE: Fields used in SearchStream operations must be properly indexed with `@Indexed`, `@Searchable`, or other indexing annotations.
+
 == Time To Live (TTL)
 
 You can set expiration times for entities:

redis-om-spring/src/main/java/com/redis/om/spring/repository/query/RediSearchQuery.java

@@ -47,6 +47,9 @@
 import com.redis.om.spring.repository.query.countmin.CountMinQueryExecutor;
 import com.redis.om.spring.repository.query.cuckoo.CuckooQueryExecutor;
 import com.redis.om.spring.repository.query.lexicographic.LexicographicQueryExecutor;
+import com.redis.om.spring.search.stream.EntityStream;
+import com.redis.om.spring.search.stream.EntityStreamImpl;
+import com.redis.om.spring.search.stream.SearchStream;
 import com.redis.om.spring.util.ObjectUtils;
 
 import redis.clients.jedis.search.FieldName;
@@ -165,6 +168,7 @@ private static FieldType getRedisFieldTypeForMapValue(Class<?> fieldType) {
   private final LexicographicQueryExecutor lexicographicQueryExecutor;
   private final GsonBuilder gsonBuilder;
   private final RediSearchIndexer indexer;
+  private final EntityStream entityStream;
   private RediSearchQueryType type;
   private String value;
   // query fields
@@ -234,6 +238,7 @@ public RediSearchQuery(//
     this.domainType = this.queryMethod.getEntityInformation().getJavaType();
     this.gsonBuilder = gsonBuilder;
     this.redisOMProperties = redisOMProperties;
+    this.entityStream = new EntityStreamImpl(modulesOperations, gsonBuilder, indexer);
 
     bloomQueryExecutor = new BloomQueryExecutor(this, modulesOperations);
     cuckooQueryExecutor = new CuckooQueryExecutor(this, modulesOperations);
@@ -887,8 +892,29 @@ private Object executeQuery(Object[] parameters) {
     // what to return
     Object result = null;
 
-    // Check if this is an exists query
-    if (processor.getReturnedType().getReturnedType() == boolean.class || processor.getReturnedType()
+    // Check if this is a SearchStream query
+    if (SearchStream.class.isAssignableFrom(queryMethod.getReturnedObjectType())) {
+      // For SearchStream, create and configure a stream based on the query
+      @SuppressWarnings(
+        "unchecked"
+      ) SearchStream<?> stream = entityStream.of((Class<Object>) domainType);
+
+      // Build the query string using the existing query builder
+      String queryString = prepareQuery(parameters, true);
+
+      // Apply the filter if it's not a wildcard query
+      if (!queryString.equals("*") && !queryString.isEmpty()) {
+        stream = stream.filter(queryString);
+      }
+
+      // Apply limit if configured
+      if (limit != null && limit > 0) {
+        stream = stream.limit(limit);
+      }
+
+      // Return the configured stream
+      return stream;
+    } else if (processor.getReturnedType().getReturnedType() == boolean.class || processor.getReturnedType()
         .getReturnedType() == Boolean.class) {
       // For exists queries, return true if we have any results, false otherwise
       result = searchResult.getTotalResults() > 0;

redis-om-spring/src/main/java/com/redis/om/spring/repository/query/RedisEnhancedQuery.java

@@ -34,6 +34,7 @@
 import org.springframework.util.ReflectionUtils;
 
 import com.github.f4b6a3.ulid.Ulid;
+import com.google.gson.GsonBuilder;
 import com.redis.om.spring.RedisOMProperties;
 import com.redis.om.spring.annotations.*;
 import com.redis.om.spring.convert.MappingRedisOMConverter;
@@ -45,6 +46,9 @@
 import com.redis.om.spring.repository.query.clause.QueryClause;
 import com.redis.om.spring.repository.query.countmin.CountMinQueryExecutor;
 import com.redis.om.spring.repository.query.cuckoo.CuckooQueryExecutor;
+import com.redis.om.spring.search.stream.EntityStream;
+import com.redis.om.spring.search.stream.EntityStreamImpl;
+import com.redis.om.spring.search.stream.SearchStream;
 import com.redis.om.spring.util.ObjectUtils;
 
 import redis.clients.jedis.search.FieldName;
@@ -115,6 +119,7 @@ public class RedisEnhancedQuery implements RepositoryQuery {
   private final RedisModulesOperations<String> modulesOperations;
   private final MappingRedisOMConverter mappingConverter;
   private final RediSearchIndexer indexer;
+  private final EntityStream entityStream;
   private final BloomQueryExecutor bloomQueryExecutor;
   private final CuckooQueryExecutor cuckooQueryExecutor;
   private final CountMinQueryExecutor countMinQueryExecutor;
@@ -190,6 +195,8 @@ public RedisEnhancedQuery(QueryMethod queryMethod, //
     this.redisOMProperties = redisOMProperties;
     this.redisOperations = redisOperations;
     this.mappingConverter = new MappingRedisOMConverter(null, new ReferenceResolverImpl(redisOperations));
+    // Create EntityStream with a default GsonBuilder since we're dealing with hashes
+    this.entityStream = new EntityStreamImpl(modulesOperations, new GsonBuilder(), indexer);
 
     bloomQueryExecutor = new BloomQueryExecutor(this, modulesOperations);
     cuckooQueryExecutor = new CuckooQueryExecutor(this, modulesOperations);
@@ -577,8 +584,29 @@ private Object executeQuery(Object[] parameters) {
     // what to return
     Object result;
 
-    // Check if this is an exists query
-    if (processor.getReturnedType().getReturnedType() == boolean.class || processor.getReturnedType()
+    // Check if this is a SearchStream query
+    if (SearchStream.class.isAssignableFrom(queryMethod.getReturnedObjectType())) {
+      // For SearchStream, create and configure a stream based on the query
+      @SuppressWarnings(
+        "unchecked"
+      ) SearchStream<?> stream = entityStream.of((Class<Object>) domainType);
+
+      // Build the query string using the existing query builder
+      String queryString = prepareQuery(parameters, true);
+
+      // Apply the filter if it's not a wildcard query
+      if (!queryString.equals("*") && !queryString.isEmpty()) {
+        stream = stream.filter(queryString);
+      }
+
+      // Apply limit if configured
+      if (limit != null && limit > 0) {
+        stream = stream.limit(limit);
+      }
+
+      // Return the configured stream
+      return stream;
+    } else if (processor.getReturnedType().getReturnedType() == boolean.class || processor.getReturnedType()
         .getReturnedType() == Boolean.class) {
       // For exists queries, return true if we have any results, false otherwise
       result = searchResult.getTotalResults() > 0;

tests/src/test/java/com/redis/om/spring/fixtures/hash/model/HashWithSearchStream.java

@@ -0,0 +1,50 @@
+package com.redis.om.spring.fixtures.hash.model;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.redis.core.RedisHash;
+
+import com.redis.om.spring.annotations.Indexed;
+import com.redis.om.spring.annotations.Searchable;
+
+import lombok.*;
+
+/**
+ * Test entity for SearchStream with properly indexed fields
+ */
+@Data
+@NoArgsConstructor(force = true)
+@RequiredArgsConstructor(staticName = "of")
+@AllArgsConstructor(access = AccessLevel.PROTECTED)
+@RedisHash("hash_with_search_stream")
+public class HashWithSearchStream {
+
+  @Id
+  String id;
+
+  @NonNull
+  @Searchable
+  String name;
+
+  @NonNull
+  @Indexed
+  String email;
+
+  @NonNull
+  @Indexed
+  String department;
+
+  @NonNull
+  @Indexed
+  Integer age;
+
+  @NonNull
+  @Indexed
+  Boolean active;
+
+  @NonNull
+  @Indexed
+  Set<String> skills = new HashSet<>();
+}

tests/src/test/java/com/redis/om/spring/fixtures/hash/repository/HashWithSearchStreamRepository.java

@@ -0,0 +1,24 @@
+package com.redis.om.spring.fixtures.hash.repository;
+
+import java.util.Set;
+
+import org.springframework.stereotype.Repository;
+
+import com.redis.om.spring.fixtures.hash.model.HashWithSearchStream;
+import com.redis.om.spring.repository.RedisEnhancedRepository;
+import com.redis.om.spring.search.stream.SearchStream;
+
+@Repository
+public interface HashWithSearchStreamRepository extends RedisEnhancedRepository<HashWithSearchStream, String> {
+
+  // Methods that return SearchStream for testing
+  SearchStream<HashWithSearchStream> findByEmail(String email);
+
+  SearchStream<HashWithSearchStream> findByDepartment(String department);
+
+  SearchStream<HashWithSearchStream> findByAgeGreaterThan(Integer age);
+
+  SearchStream<HashWithSearchStream> findByActive(Boolean active);
+
+  SearchStream<HashWithSearchStream> findBySkills(Set<String> skills);
+}