feat(query): add aggregation and hybrid search queries

Implement aggregation-based queries with full feature parity to Python redisvl:

- Add HybridQuery for text + vector hybrid search with weighted scoring
- Add AggregationQuery base class for aggregation operations
- Add FilterQuery for metadata-only filtering
- Add ReducerFunction enum for all aggregation reducers
- Add TokenEscaper utility for Redis search token escaping
- Update SearchIndex to support new query types
- Add 7 comprehensive integration tests for HybridQuery
- Add SpotBugs exclusions for intentional builder patterns

HybridQuery combines text and vector search using Redis aggregation,
scoring documents as: hybrid_score = alpha * vector_score + (1-alpha) * text_score

Includes support for:
- Custom stopwords with language fallbacks
- Filter expressions (tag, numeric, geo, text)
- Configurable alpha weighting
- Distance threshold tuning
- Return field selection

All 312+ tests passing with no regressions.

Author: bsbodden

core/src/main/java/com/redis/vl/index/SearchIndex.java

@@ -1261,6 +1261,28 @@ public List<Map<String, Object>> query(Object query) {
     } else if (query instanceof TextQuery tq) {
       SearchResult result = search(tq.toString());
       return processSearchResult(result);
+    } else if (query instanceof FilterQuery fq) {
+      // FilterQuery: metadata-only query without vector search
+      // Python: FilterQuery (redisvl/query/query.py:314)
+      redis.clients.jedis.search.Query redisQuery = fq.buildRedisQuery();
+      UnifiedJedis jedis = getUnifiedJedis();
+      SearchResult result = jedis.ftSearch(schema.getName(), redisQuery);
+      return processSearchResult(result);
+    } else if (query instanceof AggregationQuery aq) {
+      // AggregationQuery: HybridQuery and other aggregation-based queries
+      // Python: HybridQuery (redisvl/query/aggregate.py:23)
+      redis.clients.jedis.search.aggr.AggregationBuilder aggregation = aq.buildRedisAggregation();
+      UnifiedJedis jedis = getUnifiedJedis();
+
+      // Add parameters if present (e.g., vector parameter for HybridQuery)
+      Map<String, Object> params = aq.getParams();
+      if (params != null && !params.isEmpty()) {
+        aggregation.params(params);
+      }
+
+      redis.clients.jedis.search.aggr.AggregationResult result =
+          jedis.ftAggregate(schema.getName(), aggregation);
+      return processAggregationResult(result);
     }
 
     // Default: try to convert to string and search
@@ -1323,6 +1345,28 @@ private List<Map<String, Object>> processSearchResult(SearchResult result) {
     return processed;
   }
 
+  /**
+   * Process AggregationResult into List of Maps.
+   *
+   * <p>Converts Redis aggregation results into a list of maps, where each map represents a row from
+   * the aggregation result.
+   *
+   * @param result the AggregationResult from Redis
+   * @return list of maps containing aggregation results
+   */
+  private List<Map<String, Object>> processAggregationResult(
+      redis.clients.jedis.search.aggr.AggregationResult result) {
+    List<Map<String, Object>> processed = new ArrayList<>();
+    if (result != null && result.getResults() != null) {
+      for (Map<String, Object> row : result.getResults()) {
+        // Each row is already a Map<String, Object>
+        // Just add it to the processed list
+        processed.add(new HashMap<>(row));
+      }
+    }
+    return processed;
+  }
+
   public List<SearchResult> batchSearch(List<String> queries) {
     return batchSearch(queries, Integer.MAX_VALUE);
   }
@@ -1487,6 +1531,9 @@ private String buildQueryString(Object query) {
               return vq.toQueryString();
             } else if (query instanceof Filter) {
               return ((Filter) query).build();
+            } else if (query instanceof FilterQuery fq) {
+              // FilterQuery: extract filter expression or use "*"
+              return (fq.getFilterExpression() != null) ? fq.getFilterExpression().build() : "*";
             } else if (query instanceof TextQuery) {
               return query.toString();
             } else {

core/src/main/java/com/redis/vl/query/AggregationQuery.java

@@ -0,0 +1,54 @@
+package com.redis.vl.query;
+
+import java.util.Map;
+import redis.clients.jedis.search.aggr.AggregationBuilder;
+
+/**
+ * Base class for aggregation queries used to create aggregation queries for Redis.
+ *
+ * <p>Ported from Python: redisvl/query/aggregate.py:14
+ *
+ * <p>Python equivalent:
+ *
+ * <pre>
+ * class AggregationQuery(AggregateRequest):
+ *     """
+ *     Base class for aggregation queries used to create aggregation queries for Redis.
+ *     """
+ *
+ *     def __init__(self, query_string):
+ *         super().__init__(query_string)
+ * </pre>
+ *
+ * <p>This is a base class for queries that use Redis aggregation capabilities. Aggregation queries
+ * can perform complex data analysis operations like grouping, filtering, sorting, and applying
+ * reducer functions.
+ *
+ * @see HybridQuery
+ * @since 0.1.0
+ */
+public abstract class AggregationQuery {
+
+  /**
+   * Build the Redis AggregationBuilder for this query.
+   *
+   * @return the Jedis AggregationBuilder configured for this query
+   */
+  public abstract AggregationBuilder buildRedisAggregation();
+
+  /**
+   * Build the base query string for the aggregation.
+   *
+   * @return the query string
+   */
+  public abstract String buildQueryString();
+
+  /**
+   * Get the parameters for the aggregation query.
+   *
+   * <p>Used for parameterized queries (e.g., vector parameter in HybridQuery).
+   *
+   * @return a map of parameter names to values
+   */
+  public abstract Map<String, Object> getParams();
+}

core/src/main/java/com/redis/vl/query/FilterQuery.java

@@ -0,0 +1,107 @@
+package com.redis.vl.query;
+
+import java.util.List;
+import java.util.Map;
+import lombok.Builder;
+import lombok.Getter;
+import redis.clients.jedis.search.Query;
+
+/**
+ * A query for running a filtered search with a filter expression (no vector search).
+ *
+ * <p>Ported from Python: redisvl/query/query.py:314 (FilterQuery class)
+ *
+ * <p>Python equivalent:
+ *
+ * <pre>
+ * FilterQuery(
+ *     filter_expression=Tag("credit_score") == "high",
+ *     return_fields=["user", "age"],
+ *     num_results=10,
+ *     sort_by="age"
+ * )
+ * </pre>
+ *
+ * Java equivalent:
+ *
+ * <pre>
+ * FilterQuery.builder()
+ *     .filterExpression(Filter.tag("credit_score", "high"))
+ *     .returnFields(List.of("user", "age"))
+ *     .numResults(10)
+ *     .sortBy("age")
+ *     .build()
+ * </pre>
+ */
+@Getter
+@Builder
+public class FilterQuery {
+
+  /**
+   * The filter expression to query with. Python: filter_expression (Optional[Union[str,
+   * FilterExpression]]) Defaults to '*' if null.
+   */
+  private final Filter filterExpression;
+
+  /** The fields to return in results. Python: return_fields (Optional[List[str]]) */
+  @Builder.Default private final List<String> returnFields = List.of();
+
+  /** The number of results to return. Python: num_results (int) - defaults to 10 */
+  @Builder.Default private final int numResults = 10;
+
+  /** The query dialect (RediSearch version). Python: dialect (int) - defaults to 2 */
+  @Builder.Default private final int dialect = 2;
+
+  /**
+   * Field to sort results by. Python: sort_by (Optional[SortSpec]) - can be str, Tuple[str, str],
+   * or List Note: Redis Search only supports single-field sorting, so only first field is used.
+   * Defaults to ascending order.
+   */
+  private final String sortBy;
+
+  /**
+   * Whether to require terms in field to have same order as in query filter. Python: in_order
+   * (bool) - defaults to False
+   */
+  @Builder.Default private final boolean inOrder = false;
+
+  /** Additional parameters for the query. Python: params (Optional[Dict[str, Any]]) */
+  @Builder.Default private final Map<String, Object> params = Map.of();
+
+  /**
+   * Build Redis Query object from FilterQuery.
+   *
+   * <p>Python equivalent: _build_query_string() method (line 368-372) Returns the filter expression
+   * string or '*' if no filter.
+   *
+   * @return Jedis Query object
+   */
+  public Query buildRedisQuery() {
+    // Python: if isinstance(self._filter_expression, FilterExpression):
+    //             return str(self._filter_expression)
+    //         return self._filter_expression
+    String filterStr = (filterExpression != null) ? filterExpression.build() : "*";
+
+    // Python: super().__init__("*")
+    // Python: self.paging(0, self._num_results).dialect(dialect)
+    Query query = new Query(filterStr).limit(0, numResults).dialect(dialect);
+
+    // Python: if return_fields: self.return_fields(*return_fields)
+    if (!returnFields.isEmpty()) {
+      query.returnFields(returnFields.toArray(String[]::new));
+    }
+
+    // Python: if sort_by: self.sort_by(sort_by)
+    // Note: Python accepts tuple for DESC, but here we default to ASC
+    if (sortBy != null && !sortBy.isEmpty()) {
+      query.setSortBy(sortBy, true); // true = ascending
+    }
+
+    // Python: if in_order: self.in_order()
+    if (inOrder) {
+      query.setInOrder();
+    }
+
+    return query;
+  }
+}