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
This commit is contained in:
Vladyslav Vildanov
2026-07-15 11:26:23 +03:00
committed by GitHub
parent 1d7f035ec0
commit d7bc1c4748
6 changed files with 588 additions and 1 deletions
+1
View File
@@ -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)
@@ -96,6 +96,30 @@ class AggregateArguments extends CommonArguments
return $this;
}
/**
* Adds a COLLECT reducer to the current GROUPBY.
*
* Emits "REDUCE COLLECT <narg> <tokens...> [AS <alias>]" where <narg> is the
* number of tokens produced by the given CollectArguments (the FIELDS,
* DISTINCT, SORTBY and LIMIT clauses) and excludes the trailing AS <alias>.
*
* @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.
*
@@ -0,0 +1,145 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2026 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Command\Argument\Search;
use Predis\Command\Argument\ArrayableArgument;
/**
* Builds the argument tokens for the COLLECT reducer of FT.AGGREGATE GROUPBY.
*
* The produced token list covers only the FIELDS, DISTINCT, SORTBY and LIMIT
* clauses. The enclosing "REDUCE COLLECT <narg> ... [AS <alias>]" framing is
* added by AggregateArguments::reduceCollect(), which derives <narg> 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 <num_fields> @field ...).
*
* Names may be supplied with or without a leading "@"; each is normalized to
* a single "@<name>" 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 <n> @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 <count> entries per group after skipping <offset> (LIMIT <offset> <count>).
*
* @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;
}
}
@@ -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
@@ -0,0 +1,129 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2026 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Command\Argument\Search;
use PHPUnit\Framework\TestCase;
class CollectArgumentsTest extends TestCase
{
/**
* @var CollectArguments
*/
private $arguments;
protected function setUp(): void
{
$this->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'],
],
];
}
}
@@ -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],