|
| 1 | +package redis.clients.jedis.search.aggr; |
| 2 | + |
| 3 | +import java.io.Closeable; |
| 4 | +import java.util.Iterator; |
| 5 | +import java.util.NoSuchElementException; |
| 6 | + |
| 7 | +import redis.clients.jedis.Connection; |
| 8 | +import redis.clients.jedis.exceptions.JedisException; |
| 9 | +import redis.clients.jedis.providers.ConnectionProvider; |
| 10 | +import redis.clients.jedis.search.SearchProtocol; |
| 11 | +import redis.clients.jedis.util.IOUtils; |
| 12 | + |
| 13 | +/** |
| 14 | + * Iterator for Redis search aggregation results with cursor support. |
| 15 | + * This class manages the connection to a specific Redis node and handles |
| 16 | + * cursor-based pagination for large aggregation results. |
| 17 | + * |
| 18 | + * <p>The iterator supports the {@link #remove()} method which deletes the cursor |
| 19 | + * on the server and terminates the iteration, freeing server resources immediately. |
| 20 | + * |
| 21 | + * <p>Usage example: |
| 22 | + * <pre>{@code |
| 23 | + * AggregationBuilder aggr = new AggregationBuilder() |
| 24 | + * .groupBy("@field") |
| 25 | + * .cursor(100, 60000); // 100 results per batch, 60 second TTL for the cursor |
| 26 | + * |
| 27 | + * try (AggregateIterator iterator = new AggregateIterator(provider, "myindex", aggr)) { |
| 28 | + * while (iterator.hasNext()) { |
| 29 | + * AggregationResult batch = iterator.next(); |
| 30 | + * // Process batch - access rows via batch.getRows() |
| 31 | + * |
| 32 | + * // Optionally terminate early and free server resources |
| 33 | + * if (someCondition) { |
| 34 | + * iterator.remove(); // Deletes cursor and stops iteration |
| 35 | + * break; |
| 36 | + * } |
| 37 | + * } |
| 38 | + * } |
| 39 | + * }</pre> |
| 40 | + */ |
| 41 | +public class AggregateIterator implements Iterator<AggregationResult>, Closeable { |
| 42 | + |
| 43 | + private final ConnectionProvider connectionProvider; |
| 44 | + private final String indexName; |
| 45 | + private final Integer batchSize; |
| 46 | + |
| 47 | + private Connection connection; |
| 48 | + private Long cursorId = -1L; |
| 49 | + private AggregationResult aggrCommandResult; |
| 50 | + |
| 51 | + /** |
| 52 | + * Creates a new AggregateIterator. |
| 53 | + * |
| 54 | + * @param connectionProvider the connection provider for cluster/standalone Redis |
| 55 | + * @param indexName the search index name |
| 56 | + * @param aggregationBuilder the aggregation query with cursor configuration |
| 57 | + * @throws IllegalArgumentException if aggregation doesn't have cursor configured |
| 58 | + */ |
| 59 | + public AggregateIterator(ConnectionProvider connectionProvider, String indexName, |
| 60 | + AggregationBuilder aggregationBuilder) { |
| 61 | + if (!aggregationBuilder.isWithCursor()) { |
| 62 | + throw new IllegalArgumentException("AggregationBuilder must have cursor configured"); |
| 63 | + } |
| 64 | + |
| 65 | + this.connectionProvider = connectionProvider; |
| 66 | + this.indexName = indexName; |
| 67 | + this.batchSize = aggregationBuilder.getCursorCount(); |
| 68 | + |
| 69 | + // Get a dedicated connection for this cursor session |
| 70 | + this.connection = acquireConnection(aggregationBuilder); |
| 71 | + } |
| 72 | + |
| 73 | + @Override |
| 74 | + public boolean hasNext() { |
| 75 | + // If aggrCommandResult is not null, we have initial results from FT.AGGREGATE that haven't been returned yet |
| 76 | + // For subsequent batches, check if there are more cursor results available |
| 77 | + // If cursor has been removed (cursorId <= 0), no more results are available |
| 78 | + return aggrCommandResult != null || cursorId > 0; |
| 79 | + } |
| 80 | + |
| 81 | + @Override |
| 82 | + public AggregationResult next() { |
| 83 | + if (!hasNext()) { |
| 84 | + throw new NoSuchElementException("No more aggregation results available"); |
| 85 | + } |
| 86 | + |
| 87 | + try { |
| 88 | + if (aggrCommandResult != null) { |
| 89 | + try { |
| 90 | + return aggrCommandResult; |
| 91 | + } finally { |
| 92 | + aggrCommandResult = null; |
| 93 | + } |
| 94 | + } else { |
| 95 | + return doFetch(); |
| 96 | + } |
| 97 | + |
| 98 | + } catch (Exception e) { |
| 99 | + throw new JedisException("Failed to fetch next aggregation batch", e); |
| 100 | + } |
| 101 | + } |
| 102 | + |
| 103 | + /** |
| 104 | + * Returns the current cursor ID. |
| 105 | + * |
| 106 | + * @return cursor ID, or null if not initialized |
| 107 | + */ |
| 108 | + public Long getCursorId() { |
| 109 | + return cursorId; |
| 110 | + } |
| 111 | + |
| 112 | + @Override |
| 113 | + public void remove() { |
| 114 | + if (cursorId == null || cursorId <= 0) { |
| 115 | + // Cursor is already closed or not initialized, nothing to do |
| 116 | + return; |
| 117 | + } |
| 118 | + |
| 119 | + deleteCursor(); |
| 120 | + // Mark cursor as deleted to prevent further operations |
| 121 | + cursorId = -1L; |
| 122 | + } |
| 123 | + |
| 124 | + @Override |
| 125 | + public void close() { |
| 126 | + deleteCursor(); |
| 127 | + // Mark cursor as closed to prevent further operations |
| 128 | + cursorId = -1L; |
| 129 | + IOUtils.closeQuietly(connection); |
| 130 | + } |
| 131 | + |
| 132 | + /** |
| 133 | + * Deletes the cursor on the server to free resources. |
| 134 | + * This method is idempotent and safe to call multiple times. |
| 135 | + */ |
| 136 | + private void deleteCursor() { |
| 137 | + if (cursorId != null && cursorId > 0) { |
| 138 | + try { |
| 139 | + // Delete the cursor to free server resources |
| 140 | + connection.executeCommand( |
| 141 | + new redis.clients.jedis.CommandArguments(SearchProtocol.SearchCommand.CURSOR) |
| 142 | + .add(SearchProtocol.SearchKeyword.DEL) |
| 143 | + .add(indexName) |
| 144 | + .add(cursorId) |
| 145 | + ); |
| 146 | + } catch (Exception e) { |
| 147 | + // Log but don't throw - cursor will expire naturally |
| 148 | + System.err.println("Warning: Failed to delete cursor " + cursorId + ": " + e.getMessage()); |
| 149 | + } |
| 150 | + } |
| 151 | + } |
| 152 | + |
| 153 | + private AggregationResult doFetch() { |
| 154 | + if (cursorId == null || cursorId <= 0) { |
| 155 | + return null; |
| 156 | + } |
| 157 | + |
| 158 | + redis.clients.jedis.CommandArguments args = new redis.clients.jedis.CommandArguments(SearchProtocol.SearchCommand.CURSOR) |
| 159 | + .add(SearchProtocol.SearchKeyword.READ) |
| 160 | + .add(indexName) |
| 161 | + .add(cursorId); |
| 162 | + |
| 163 | + // Only add COUNT argument if a batch size was explicitly specified |
| 164 | + if (batchSize != null) { |
| 165 | + args.add(SearchProtocol.SearchKeyword.COUNT).add(batchSize); |
| 166 | + } |
| 167 | + |
| 168 | + Object rawReply = connection.executeCommand(args); |
| 169 | + AggregationResult result = AggregationResult.SEARCH_AGGREGATION_RESULT_WITH_CURSOR.build(rawReply); |
| 170 | + |
| 171 | + cursorId = result.getCursorId(); |
| 172 | + return result; |
| 173 | + } |
| 174 | + |
| 175 | + private Connection acquireConnection(AggregationBuilder aggregationBuilder) { |
| 176 | + // Create the initial FT.AGGREGATE command |
| 177 | + redis.clients.jedis.CommandArguments args = new redis.clients.jedis.CommandArguments(SearchProtocol.SearchCommand.AGGREGATE) |
| 178 | + .add(indexName) |
| 179 | + .addParams(aggregationBuilder); |
| 180 | + |
| 181 | + Connection conn = null; |
| 182 | + try { |
| 183 | + // Get connection and execute initial command |
| 184 | + conn = connectionProvider.getConnection(args); |
| 185 | + Object rawReply = conn.executeCommand(args); |
| 186 | + aggrCommandResult = AggregationResult.SEARCH_AGGREGATION_RESULT_WITH_CURSOR.build(rawReply); |
| 187 | + |
| 188 | + cursorId = aggrCommandResult.getCursorId(); |
| 189 | + |
| 190 | + return conn; |
| 191 | + |
| 192 | + } catch (Exception e) { |
| 193 | + IOUtils.closeQuietly(conn); |
| 194 | + throw new JedisException("Failed to initialize aggregation cursor", e); |
| 195 | + } |
| 196 | + } |
| 197 | + |
| 198 | + |
| 199 | +} |
0 commit comments