From d7bc1c47485d044dcd46a6482f2d57db62889a52 Mon Sep 17 00:00:00 2001 From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com> Date: Wed, 15 Jul 2026 11:26:23 +0300 Subject: [PATCH] Added support for new COLLECT reducer for FT.AGGREGATE (#1699) * Added support for new COLLECT reducer for FT.AGGREGATE * Added tests cleanup * Updated CHANGELOG.md --- CHANGELOG.md | 1 + .../Argument/Search/AggregateArguments.php | 24 ++ .../Argument/Search/CollectArguments.php | 145 +++++++++++ .../Search/AggregateArgumentsTest.php | 44 ++++ .../Argument/Search/CollectArgumentsTest.php | 129 +++++++++ .../Command/Redis/Search/FTAGGREGATE_Test.php | 246 +++++++++++++++++- 6 files changed, 588 insertions(+), 1 deletion(-) create mode 100644 src/Command/Argument/Search/CollectArguments.php create mode 100644 tests/Predis/Command/Argument/Search/CollectArgumentsTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e8fe436..61bb369b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Added - Make ZMSCORE command Prefixable and add to a ClusterStrategy (#1692) - Added support for new TS commands + Indonesian language support integration test (#1695) +- Added support for new COLLECT reducer for FT.AGGREGATE (#1699) ### Fixed - Fixed Sentinel does not wipe servers on exception caused (#1694) diff --git a/src/Command/Argument/Search/AggregateArguments.php b/src/Command/Argument/Search/AggregateArguments.php index bbf0dab3..6b595fbf 100644 --- a/src/Command/Argument/Search/AggregateArguments.php +++ b/src/Command/Argument/Search/AggregateArguments.php @@ -96,6 +96,30 @@ class AggregateArguments extends CommonArguments return $this; } + /** + * Adds a COLLECT reducer to the current GROUPBY. + * + * Emits "REDUCE COLLECT [AS ]" where is the + * number of tokens produced by the given CollectArguments (the FIELDS, + * DISTINCT, SORTBY and LIMIT clauses) and excludes the trailing AS . + * + * @param CollectArguments $arguments COLLECT clause builder + * @param string|null $alias Optional reducer output column name + * @return $this + */ + public function reduceCollect(CollectArguments $arguments, ?string $alias = null): self + { + $tokens = $arguments->toArray(); + + array_push($this->arguments, 'REDUCE', 'COLLECT', count($tokens), ...$tokens); + + if ($alias !== null) { + array_push($this->arguments, 'AS', $alias); + } + + return $this; + } + /** * Sorts the pipeline up until the point of SORTBY, using a list of properties. * diff --git a/src/Command/Argument/Search/CollectArguments.php b/src/Command/Argument/Search/CollectArguments.php new file mode 100644 index 00000000..4f377bb0 --- /dev/null +++ b/src/Command/Argument/Search/CollectArguments.php @@ -0,0 +1,145 @@ + ... [AS ]" framing is + * added by AggregateArguments::reduceCollect(), which derives from the + * length of this token list. + * + * Field and sort-key names are emitted with a leading "@" on the wire; bare + * names are normalized automatically. Output map keys are the names with the + * "@" stripped by the server. + */ +class CollectArguments implements ArrayableArgument +{ + public const SORT_ASC = 'ASC'; + public const SORT_DESC = 'DESC'; + + /** + * @var array + */ + protected $arguments = []; + + /** + * Projects all fields present in the pipeline at the COLLECT stage (FIELDS *). + * + * This is stage-local: it projects whatever the pipeline has already + * materialized, not a whole-document fetch. Pair with an upstream LOAD * to + * collect the complete document. + * + * @return $this + */ + public function allFields(): self + { + array_push($this->arguments, 'FIELDS', '*'); + + return $this; + } + + /** + * Projects the explicitly listed fields (FIELDS @field ...). + * + * Names may be supplied with or without a leading "@"; each is normalized to + * a single "@" on the wire. + * + * @param string ...$fields + * @return $this + */ + public function fields(string ...$fields): self + { + $names = array_map([$this, 'ensureAtPrefix'], $fields); + + array_push($this->arguments, 'FIELDS', count($names), ...$names); + + return $this; + } + + /** + * Deduplicates entries with identical projected fields. + * + * NOTE: DISTINCT is specified by the product but not yet implemented by the + * server. Sending it currently results in a server error. The option is kept + * for forward-compatibility; do not rely on it until the server ships it. + * + * @return $this + */ + public function distinct(): self + { + $this->arguments[] = 'DISTINCT'; + + return $this; + } + + /** + * Sorts entries within each group by one or more keys (SORTBY @field [ASC|DESC] ...). + * + * Accepts an ordered map of field name to sort direction, e.g. + * ['@rating' => CollectArguments::SORT_DESC, '@name' => CollectArguments::SORT_ASC]. + * Field names are normalized to carry a leading "@". + * + * @param array $sortByFields Ordered map of field => direction (self::SORT_ASC|self::SORT_DESC) + * @return $this + */ + public function sortBy(array $sortByFields): self + { + $tokens = []; + + foreach ($sortByFields as $field => $direction) { + $tokens[] = $this->ensureAtPrefix((string) $field); + $tokens[] = $direction; + } + + array_push($this->arguments, 'SORTBY', count($tokens), ...$tokens); + + return $this; + } + + /** + * Returns at most entries per group after skipping (LIMIT ). + * + * @param int $offset + * @param int $count + * @return $this + */ + public function limit(int $offset, int $count): self + { + array_push($this->arguments, 'LIMIT', $offset, $count); + + return $this; + } + + /** + * {@inheritDoc} + */ + public function toArray(): array + { + return $this->arguments; + } + + /** + * Ensures a field or sort-key name carries a single leading "@" prefix. + * + * @param string $name + * @return string + */ + private function ensureAtPrefix(string $name): string + { + return (strpos($name, '@') === 0) ? $name : '@' . $name; + } +} diff --git a/tests/Predis/Command/Argument/Search/AggregateArgumentsTest.php b/tests/Predis/Command/Argument/Search/AggregateArgumentsTest.php index 4e220d11..315d00af 100644 --- a/tests/Predis/Command/Argument/Search/AggregateArgumentsTest.php +++ b/tests/Predis/Command/Argument/Search/AggregateArgumentsTest.php @@ -62,6 +62,50 @@ class AggregateArgumentsTest extends TestCase $this->assertSame($expectedResponse, $this->arguments->toArray()); } + /** + * @dataProvider reduceCollectProvider + * @param CollectArguments $collectArguments + * @param string|null $alias + * @param array $expectedResponse + * @return void + */ + public function testCreatesArgumentsWithReduceCollectModifier( + CollectArguments $collectArguments, + ?string $alias, + array $expectedResponse + ): void { + $this->arguments->reduceCollect($collectArguments, $alias); + + $this->assertSame($expectedResponse, $this->arguments->toArray()); + } + + public function reduceCollectProvider(): array + { + return [ + 'all fields, no alias' => [ + (new CollectArguments())->allFields(), + null, + ['REDUCE', 'COLLECT', 2, 'FIELDS', '*'], + ], + 'all fields with alias' => [ + (new CollectArguments())->allFields(), + 'members', + ['REDUCE', 'COLLECT', 2, 'FIELDS', '*', 'AS', 'members'], + ], + 'fields, sortby and limit with alias' => [ + (new CollectArguments()) + ->fields('@name', '@sweetness') + ->sortBy(['@sweetness' => CollectArguments::SORT_DESC]) + ->limit(0, 2), + 'fruits', + [ + 'REDUCE', 'COLLECT', 11, 'FIELDS', 2, '@name', '@sweetness', + 'SORTBY', 2, '@sweetness', 'DESC', 'LIMIT', 0, 2, 'AS', 'fruits', + ], + ], + ]; + } + /** * @dataProvider sortByProvider * @param array $arguments diff --git a/tests/Predis/Command/Argument/Search/CollectArgumentsTest.php b/tests/Predis/Command/Argument/Search/CollectArgumentsTest.php new file mode 100644 index 00000000..9df7e2e0 --- /dev/null +++ b/tests/Predis/Command/Argument/Search/CollectArgumentsTest.php @@ -0,0 +1,129 @@ +arguments = new CollectArguments(); + } + + /** + * @return void + */ + public function testAllFields(): void + { + $this->arguments->allFields(); + + $this->assertSame(['FIELDS', '*'], $this->arguments->toArray()); + } + + /** + * @return void + */ + public function testFieldsWithExplicitNames(): void + { + $this->arguments->fields('@name', '@sweetness'); + + $this->assertSame(['FIELDS', 2, '@name', '@sweetness'], $this->arguments->toArray()); + } + + /** + * @return void + */ + public function testFieldsNormalizesMissingAtPrefix(): void + { + $this->arguments->fields('name', '@sweetness', '__key'); + + $this->assertSame(['FIELDS', 3, '@name', '@sweetness', '@__key'], $this->arguments->toArray()); + } + + /** + * @return void + */ + public function testDistinct(): void + { + $this->arguments->allFields()->distinct(); + + $this->assertSame(['FIELDS', '*', 'DISTINCT'], $this->arguments->toArray()); + } + + /** + * @dataProvider sortByProvider + * @param array $sortByFields + * @param array $expectedResponse + * @return void + */ + public function testSortBy(array $sortByFields, array $expectedResponse): void + { + $this->arguments->sortBy($sortByFields); + + $this->assertSame($expectedResponse, $this->arguments->toArray()); + } + + /** + * @return void + */ + public function testLimit(): void + { + $this->arguments->limit(0, 5); + + $this->assertSame(['LIMIT', 0, 5], $this->arguments->toArray()); + } + + /** + * @return void + */ + public function testCombinesAllClausesInCanonicalOrder(): void + { + $this->arguments + ->fields('@name', '@sweetness') + ->sortBy(['@sweetness' => CollectArguments::SORT_DESC]) + ->limit(0, 2); + + $this->assertSame( + ['FIELDS', 2, '@name', '@sweetness', 'SORTBY', 2, '@sweetness', 'DESC', 'LIMIT', 0, 2], + $this->arguments->toArray() + ); + } + + public function sortByProvider(): array + { + return [ + 'single key descending' => [ + ['@sweetness' => CollectArguments::SORT_DESC], + ['SORTBY', 2, '@sweetness', 'DESC'], + ], + 'single key ascending' => [ + ['@name' => CollectArguments::SORT_ASC], + ['SORTBY', 2, '@name', 'ASC'], + ], + 'multiple keys' => [ + ['@sweetness' => CollectArguments::SORT_DESC, '@name' => CollectArguments::SORT_ASC], + ['SORTBY', 4, '@sweetness', 'DESC', '@name', 'ASC'], + ], + 'normalizes missing @ prefix' => [ + ['sweetness' => CollectArguments::SORT_DESC], + ['SORTBY', 2, '@sweetness', 'DESC'], + ], + ]; + } +} diff --git a/tests/Predis/Command/Redis/Search/FTAGGREGATE_Test.php b/tests/Predis/Command/Redis/Search/FTAGGREGATE_Test.php index 98debd50..84a45698 100644 --- a/tests/Predis/Command/Redis/Search/FTAGGREGATE_Test.php +++ b/tests/Predis/Command/Redis/Search/FTAGGREGATE_Test.php @@ -13,6 +13,7 @@ namespace Predis\Command\Redis\Search; use Predis\Command\Argument\Search\AggregateArguments; +use Predis\Command\Argument\Search\CollectArguments; use Predis\Command\Argument\Search\CreateArguments; use Predis\Command\Argument\Search\SchemaFields\AbstractField; use Predis\Command\Argument\Search\SchemaFields\NumericField; @@ -178,7 +179,134 @@ class FTAGGREGATE_Test extends PredisCommandTestCase * @group connected * @group relay-resp3 * @return void - * @requiresRediSearchVersion >= 1.1.0 + * @requiresRedisVersion >= 8.10.0 + */ + public function testAggregatesWithCollectReducer(): void + { + $redis = $this->getClient(); + $this->enableUnstableSearchFeatures($redis); + + $this->createFruitsIndex($redis); + + $ftAggregateArguments = (new AggregateArguments()) + ->groupBy('@color') + ->reduceCollect( + (new CollectArguments()) + ->fields('@name', '@sweetness') + ->sortBy(['@sweetness' => CollectArguments::SORT_DESC]) + ->limit(0, 2), + 'fruits' + ); + + $response = $redis->ftaggregate('idx', '*', $ftAggregateArguments); + + // Top-level group order is not guaranteed; key the collected columns by + // color. Within a group SORTBY makes the entry order deterministic. + $collectedByColor = $this->indexResp2CollectByColor($response); + + $this->assertEquals( + [ + ['name', 'apple', 'sweetness', '4'], + ['name', 'strawberry', 'sweetness', '3'], + ], + $collectedByColor['red'] + ); + $this->assertEquals( + [ + ['name', 'banana', 'sweetness', '4'], + ['name', 'lemon', 'sweetness', '2'], + ], + $collectedByColor['yellow'] + ); + } + + /** + * @group connected + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testAggregatesWithCollectReducerResp3(): void + { + $redis = $this->getResp3Client(); + $this->enableUnstableSearchFeatures($redis); + + $this->createFruitsIndex($redis); + + $ftAggregateArguments = (new AggregateArguments()) + ->groupBy('@color') + ->reduceCollect( + (new CollectArguments()) + ->fields('@name', '@sweetness') + ->sortBy(['@sweetness' => CollectArguments::SORT_DESC]) + ->limit(0, 2), + 'fruits' + ); + + $response = $redis->ftaggregate('idx', '*', $ftAggregateArguments); + + // In RESP3 each entry is a map; key the collected columns by color. + $collectedByColor = []; + foreach ($response['results'] as $result) { + $attributes = $result['extra_attributes']; + $collectedByColor[$attributes['color']] = $attributes['fruits']; + } + + $this->assertEquals( + [ + ['name' => 'apple', 'sweetness' => '4'], + ['name' => 'strawberry', 'sweetness' => '3'], + ], + $collectedByColor['red'] + ); + $this->assertEquals( + [ + ['name' => 'banana', 'sweetness' => '4'], + ['name' => 'lemon', 'sweetness' => '2'], + ], + $collectedByColor['yellow'] + ); + } + + /** + * @group connected + * @group relay-resp3 + * @return void + * @requiresRedisVersion >= 8.10.0 + */ + public function testCollectReducerProjectsAllFields(): void + { + $redis = $this->getClient(); + $this->enableUnstableSearchFeatures($redis); + + $this->createFruitsIndex($redis); + + // FIELDS * projects the fields materialized in the pipeline at this stage. + $ftAggregateArguments = (new AggregateArguments()) + ->load('@name', '@color', '@sweetness') + ->groupBy('@color') + ->reduceCollect( + (new CollectArguments())->allFields()->sortBy(['@sweetness' => CollectArguments::SORT_DESC]), + 'entries' + ); + + $response = $redis->ftaggregate('idx', '*', $ftAggregateArguments); + $collectedByColor = $this->indexResp2CollectByColor($response, 'entries'); + + $this->assertCount(2, $collectedByColor['red']); + $this->assertCount(2, $collectedByColor['yellow']); + + // Each entry carries the projected fields as a flat key/value array. + foreach ($collectedByColor['red'] as $entry) { + $this->assertContains('name', $entry); + $this->assertContains('sweetness', $entry); + } + } + + /** + * @group connected + * @group relay-resp3 + * @return void + * @requiresRedisVersion >= 1.1.0 */ public function testThrowsExceptionOnNonExistingIndex(): void { @@ -189,6 +317,104 @@ class FTAGGREGATE_Test extends PredisCommandTestCase $redis->ftaggregate('index', 'query'); } + /** + * @var \Predis\ClientInterface|null Client used to toggle the unstable-features gate, kept for teardown restoration. + */ + private $unstableFeaturesClient; + + /** + * @var string|null Original value of the unstable-features gate, restored on teardown. + */ + private $originalUnstableFeatures; + + /** + * Restores any global server state mutated by the connected tests. + */ + protected function tearDown(): void + { + if ($this->unstableFeaturesClient !== null && $this->originalUnstableFeatures !== null) { + $this->unstableFeaturesClient->config( + 'set', + 'search-enable-unstable-features', + $this->originalUnstableFeatures + ); + + $this->unstableFeaturesClient = null; + $this->originalUnstableFeatures = null; + } + + parent::tearDown(); + } + + /** + * Enables the unstable-features gate that COLLECT is currently behind. + * + * The previous value is captured and restored on teardown so the connected + * tests do not leave the gate flipped on the shared server. + */ + private function enableUnstableSearchFeatures($redis): void + { + $config = $redis->config('get', 'search-enable-unstable-features'); + $this->originalUnstableFeatures = $config['search-enable-unstable-features'] ?? array_pop($config); + $this->unstableFeaturesClient = $redis; + + $redis->config('set', 'search-enable-unstable-features', 'yes'); + } + + /** + * Creates a fruits HASH index with name, color and a sortable sweetness field. + */ + private function createFruitsIndex($redis): void + { + $schema = [ + new TextField('name'), + new TextField('color'), + new NumericField('sweetness', '', AbstractField::SORTABLE), + ]; + + $this->assertEquals( + 'OK', + $redis->ftcreate('idx', $schema, (new CreateArguments())->prefix(['fruit:'])) + ); + + $this->assertSame(3, $redis->hset('fruit:1', 'name', 'apple', 'color', 'red', 'sweetness', 4)); + $this->assertSame(3, $redis->hset('fruit:2', 'name', 'strawberry', 'color', 'red', 'sweetness', 3)); + $this->assertSame(3, $redis->hset('fruit:3', 'name', 'banana', 'color', 'yellow', 'sweetness', 4)); + $this->assertSame(3, $redis->hset('fruit:4', 'name', 'lemon', 'color', 'yellow', 'sweetness', 2)); + + // Give the index a moment to catch up with the written documents. + usleep(20000); + } + + /** + * Reduces a RESP2 FT.AGGREGATE reply to a map of group value => collected column. + * + * The reply is [count, row, row, ...] where each row is a flat key/value + * array holding the GROUPBY key ("color") and the collect alias column. + */ + private function indexResp2CollectByColor(array $response, string $alias = 'fruits'): array + { + $collectedByColor = []; + + for ($i = 1, $max = count($response); $i < $max; $i++) { + $row = $response[$i]; + $color = null; + $collected = null; + + for ($j = 0, $rowMax = count($row); $j < $rowMax; $j += 2) { + if ($row[$j] === 'color') { + $color = $row[$j + 1]; + } elseif ($row[$j] === $alias) { + $collected = $row[$j + 1]; + } + } + + $collectedByColor[$color] = $collected; + } + + return $collectedByColor; + } + public function argumentsProvider(): array { return [ @@ -224,6 +450,24 @@ class FTAGGREGATE_Test extends PredisCommandTestCase ['index', 'query', (new AggregateArguments())->sortBy(2, 'property1', 'ASC', 'property2', 'DESC')], ['index', 'query', 'SORTBY', 4, 'property1', 'ASC', 'property2', 'DESC', 'MAX', 2, 'DIALECT', 2], ], + 'with REDUCE COLLECT modifier' => [ + [ + 'index', 'query', + (new AggregateArguments()) + ->groupBy('@color') + ->reduceCollect( + (new CollectArguments()) + ->fields('@name', '@sweetness') + ->sortBy(['@sweetness' => CollectArguments::SORT_DESC]) + ->limit(0, 2), + 'fruits' + ), + ], + [ + 'index', 'query', 'GROUPBY', 1, '@color', 'REDUCE', 'COLLECT', 11, 'FIELDS', 2, '@name', + '@sweetness', 'SORTBY', 2, '@sweetness', 'DESC', 'LIMIT', 0, 2, 'AS', 'fruits', 'DIALECT', 2, + ], + ], 'with APPLY modifier' => [ ['index', 'query', (new AggregateArguments())->apply('expression', 'name')], ['index', 'query', 'APPLY', 'expression', 'AS', 'name', 'DIALECT', 2],