geckon01/predis
1
0
Fork 0
mirror of https://github.com/predis/predis.git synced 2026-08-31 12:43:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added support for multiple aggregators for TS range commands (#1670)

Browse Source 
* Added support for multiple aggregators for TS range commands

* Updated CHANGELOG.md
This commit is contained in:
Vladyslav Vildanov
2026-04-23 10:26:00 +03:00
committed by GitHub
parent 6c3909e675
commit 52286f8164
9 changed files with 410 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -7,6 +7,7 @@
- Added FPHA argument for JSON.SET command (#1661)
- Added new COUNT aggregator for Sorted Set commands (#1668)
- Added XNACK support (#1666)
- Added support for multiple aggregators for TS range commands (#1670)
### Changed
- Include command name in unsupported container command error messages (#1653)
src/Command/Argument/TimeSeries/MRangeArguments.php
+32
View File
@@ -12,6 +12,8 @@
namespace Predis\Command\Argument\TimeSeries;
use UnexpectedValueException;
class MRangeArguments extends RangeArguments
{
/**
@@ -27,16 +29,46 @@ class MRangeArguments extends RangeArguments
return $this;
}
/**
* {@inheritDoc}
*
* Multiple aggregators cannot be combined with GROUPBY.
*
* @return $this
*/
public function aggregation($aggregator, int $bucketDuration, int $align = 0, int $bucketTimestamp = 0, bool $empty = false): RangeArguments
{
$isMulti = is_array($aggregator) ? count($aggregator) > 1 : strpos((string) $aggregator, ',') !== false;
if ($isMulti && in_array('GROUPBY', $this->arguments, true)) {
throw new UnexpectedValueException('Multiple aggregators cannot be combined with GROUPBY.');
}
return parent::aggregation($aggregator, $bucketDuration, $align, $bucketTimestamp, $empty);
}
/**
* Splits time series into groups, each group contains time series that share the same
* value for the provided label name, then aggregates results in each group.
*
* GROUPBY cannot be combined with multiple aggregators set via aggregation().
*
* @param string $label
* @param string $reducer
* @return $this
*/
public function groupBy(string $label, string $reducer): self
{
$aggIndex = array_search('AGGREGATION', $this->arguments, true);
if ($aggIndex !== false
&& isset($this->arguments[$aggIndex + 1])
&& is_string($this->arguments[$aggIndex + 1])
&& strpos($this->arguments[$aggIndex + 1], ',') !== false
) {
throw new UnexpectedValueException('GROUPBY cannot be combined with multiple aggregators.');
}
array_push($this->arguments, 'GROUPBY', $label, 'REDUCE', $reducer);
return $this;
src/Command/Argument/TimeSeries/RangeArguments.php
+12 -7
View File
@@ -64,20 +64,25 @@ class RangeArguments extends CommonArguments
/**
* Aggregates samples into time buckets.
*
* @param string $aggregator Aggregation type. Check class constants.
* @param int $bucketDuration Is duration of each bucket, in milliseconds.
* @param int $align It controls the time bucket timestamps by changing the reference timestamp on which a bucket is defined.
* @param int $bucketTimestamp Controls how bucket timestamps are reported.
* @param bool $empty Is a flag, which, when specified, reports aggregations also for empty buckets.
* Multiple aggregators may be specified by passing an array of aggregation types
* (e.g. ['min', 'max']) or a comma-separated string (e.g. "min,max").
*
* @param string|array $aggregator Aggregation type, or list of aggregation types. Check class constants.
* @param int $bucketDuration Is duration of each bucket, in milliseconds.
* @param int $align It controls the time bucket timestamps by changing the reference timestamp on which a bucket is defined.
* @param int $bucketTimestamp Controls how bucket timestamps are reported.
* @param bool $empty Is a flag, which, when specified, reports aggregations also for empty buckets.
* @return $this
*/
public function aggregation(string $aggregator, int $bucketDuration, int $align = 0, int $bucketTimestamp = 0, bool $empty = false): self
public function aggregation($aggregator, int $bucketDuration, int $align = 0, int $bucketTimestamp = 0, bool $empty = false): self
{
$aggString = is_array($aggregator) ? implode(',', $aggregator) : (string) $aggregator;
if ($align > 0) {
array_push($this->arguments, 'ALIGN', $align);
}
array_push($this->arguments, 'AGGREGATION', $aggregator, $bucketDuration);
array_push($this->arguments, 'AGGREGATION', $aggString, $bucketDuration);
if ($bucketTimestamp > 0) {
array_push($this->arguments, 'BUCKETTIMESTAMP', $bucketTimestamp);
tests/Predis/Command/Argument/TimeSeries/MRangeArgumentsTest.php
+55
View File
@@ -13,6 +13,7 @@
namespace Predis\Command\Argument\TimeSeries;
use PHPUnit\Framework\TestCase;
use UnexpectedValueException;
class MRangeArgumentsTest extends TestCase
{
@@ -45,4 +46,58 @@ class MRangeArgumentsTest extends TestCase
$this->assertSame(['GROUPBY', 'label', 'REDUCE', 'reducer'], $this->arguments->toArray());
}
/**
* @return void
*/
public function testThrowsOnGroupByWhenMultipleAggregatorsAlreadySet(): void
{
$this->arguments->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000);
$this->expectException(UnexpectedValueException::class);
$this->expectExceptionMessage('GROUPBY cannot be combined with multiple aggregators.');
$this->arguments->groupBy('label', 'reducer');
}
/**
* @return void
*/
public function testThrowsOnMultipleAggregatorsWhenGroupByAlreadySet(): void
{
$this->arguments->groupBy('label', 'reducer');
$this->expectException(UnexpectedValueException::class);
$this->expectExceptionMessage('Multiple aggregators cannot be combined with GROUPBY.');
$this->arguments->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000);
}
/**
* @return void
*/
public function testAllowsGroupByAfterSingleAggregator(): void
{
$this->arguments->aggregation(RangeArguments::AGG_SUM, 1000);
$this->arguments->groupBy('type', 'max');
$this->assertSame(
['AGGREGATION', RangeArguments::AGG_SUM, 1000, 'GROUPBY', 'type', 'REDUCE', 'max'],
$this->arguments->toArray()
);
}
/**
* @return void
*/
public function testAllowsSingleAggregatorAfterGroupBy(): void
{
$this->arguments->groupBy('type', 'max');
$this->arguments->aggregation(RangeArguments::AGG_SUM, 1000);
$this->assertSame(
['GROUPBY', 'type', 'REDUCE', 'max', 'AGGREGATION', RangeArguments::AGG_SUM, 1000],
$this->arguments->toArray()
);
}
}
tests/Predis/Command/Argument/TimeSeries/RangeArgumentsTest.php
+16
View File
@@ -92,6 +92,22 @@ class RangeArgumentsTest extends TestCase
[RangeArguments::AGG_COUNT_NAN, 1000, 10, 10000, true],
['ALIGN', 10, 'AGGREGATION', RangeArguments::AGG_COUNT_NAN, 1000, 'BUCKETTIMESTAMP', 10000, 'EMPTY'],
],
'with multiple aggregators as array' => [
[[RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000],
['AGGREGATION', 'min,max', 1000],
],
'with multiple aggregators as comma-separated string' => [
['min,max', 1000],
['AGGREGATION', 'min,max', 1000],
],
'with multiple aggregators and all arguments' => [
[[RangeArguments::AGG_MIN, RangeArguments::AGG_MAX, RangeArguments::AGG_SUM], 1000, 10, 10000, true],
['ALIGN', 10, 'AGGREGATION', 'min,max,sum', 1000, 'BUCKETTIMESTAMP', 10000, 'EMPTY'],
],
'with single-element aggregator array' => [
[[RangeArguments::AGG_SUM], 1000],
['AGGREGATION', RangeArguments::AGG_SUM, 1000],
],
];
}
}
tests/Predis/Command/Redis/TimeSeries/TSMRANGE_Test.php
+86
View File
@@ -17,6 +17,7 @@ use Predis\Command\Argument\TimeSeries\CreateArguments;
use Predis\Command\Argument\TimeSeries\MRangeArguments;
use Predis\Command\Argument\TimeSeries\RangeArguments;
use Predis\Command\Redis\PredisCommandTestCase;
use UnexpectedValueException;
/**
* @group commands
@@ -60,6 +61,46 @@ class TSMRANGE_Test extends PredisCommandTestCase
$this->assertSame(1, $this->getCommand()->parseResponse(1));
}
/**
* @group disconnected
* @dataProvider parseResponseProvider
*/
public function testParseResponsePassesThroughSingleAndMultipleAggregatorResults(array $response): void
{
$this->assertSame($response, $this->getCommand()->parseResponse($response));
}
public function parseResponseProvider(): array
{
return [
'single aggregator' => [
[
['stock:A', [['type', 'stock'], ['name', 'A']], [[1000, '110'], [2000, '210']]],
],
],
'multiple aggregators' => [
[
['stock:A', [['type', 'stock'], ['name', 'A']], [[1000, '110', '2'], [2000, '210', '3']]],
],
],
];
}
/**
* @group disconnected
*/
public function testThrowsOnGroupByCombinedWithMultipleAggregators(): void
{
$mrangeArguments = (new MRangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000)
->filter('type=stock');
$this->expectException(UnexpectedValueException::class);
$this->expectExceptionMessage('GROUPBY cannot be combined with multiple aggregators.');
$mrangeArguments->groupBy('type', 'max');
}
/**
* @group connected
* @group relay-resp3
@@ -226,6 +267,39 @@ class TSMRANGE_Test extends PredisCommandTestCase
$this->assertEquals($expectedResponse, $redis->tsmrange(1000, 1001, $mRangeArguments));
}
/**
* @group connected
* @group relay-resp3
* @return void
* @requiresRedisVersion >= 8.7.2
*/
public function testQueryMultipleTimeSeriesWithMultipleAggregators(): void
{
$redis = $this->getClient();
$this->assertEquals(
'OK',
$redis->tscreate('stock:A', (new CreateArguments())->labels('type', 'stock', 'name', 'A'))
);
$this->assertSame(
[1000, 1010, 1020],
$redis->tsmadd('stock:A', 1000, 100, 'stock:A', 1010, 110, 'stock:A', 1020, 120)
);
$mrangeArguments = (new MRangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000)
->filter('type=stock');
$response = $redis->tsmrange('-', '+', $mrangeArguments);
$this->assertCount(1, $response);
$this->assertSame('stock:A', $response[0][0]);
$samples = $response[0][2];
$this->assertCount(1, $samples);
$this->assertCount(3, $samples[0]);
$this->assertSame(1000, $samples[0][0]);
}
public function argumentsProvider(): array
{
return [
@@ -273,6 +347,18 @@ class TSMRANGE_Test extends PredisCommandTestCase
[1000, 1001, (new MRangeArguments())->aggregation('sum', 2, 0, 0, true)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'sum', 2, 'EMPTY', 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators as array' => [
[1000, 1001, (new MRangeArguments())->aggregation(['min', 'max'], 2)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'min,max', 2, 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators as string' => [
[1000, 1001, (new MRangeArguments())->aggregation('min,max', 2)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'min,max', 2, 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators with all options' => [
[1000, 1001, (new MRangeArguments())->aggregation(['min', 'max', 'avg'], 2, 2, 10000, true)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'ALIGN', 2, 'AGGREGATION', 'min,max,avg', 2, 'BUCKETTIMESTAMP', 10000, 'EMPTY', 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with GROUPBY modifier' => [
[1000, 1001, (new MRangeArguments())->filter('filterExpression1', 'filterExpression2')->groupBy('label', 'reducer')],
[1000, 1001, 'FILTER', 'filterExpression1', 'filterExpression2', 'GROUPBY', 'label', 'REDUCE', 'reducer'],
tests/Predis/Command/Redis/TimeSeries/TSMREVRANGE_Test.php
+86
View File
@@ -17,6 +17,7 @@ use Predis\Command\Argument\TimeSeries\CreateArguments;
use Predis\Command\Argument\TimeSeries\MRangeArguments;
use Predis\Command\Argument\TimeSeries\RangeArguments;
use Predis\Command\Redis\PredisCommandTestCase;
use UnexpectedValueException;
/**
* @group commands
@@ -60,6 +61,46 @@ class TSMREVRANGE_Test extends PredisCommandTestCase
$this->assertSame(1, $this->getCommand()->parseResponse(1));
}
/**
* @group disconnected
* @dataProvider parseResponseProvider
*/
public function testParseResponsePassesThroughSingleAndMultipleAggregatorResults(array $response): void
{
$this->assertSame($response, $this->getCommand()->parseResponse($response));
}
public function parseResponseProvider(): array
{
return [
'single aggregator' => [
[
['stock:A', [['type', 'stock'], ['name', 'A']], [[2000, '210'], [1000, '110']]],
],
],
'multiple aggregators' => [
[
['stock:A', [['type', 'stock'], ['name', 'A']], [[2000, '210', '3'], [1000, '110', '2']]],
],
],
];
}
/**
* @group disconnected
*/
public function testThrowsOnGroupByCombinedWithMultipleAggregators(): void
{
$mrangeArguments = (new MRangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000)
->filter('type=stock');
$this->expectException(UnexpectedValueException::class);
$this->expectExceptionMessage('GROUPBY cannot be combined with multiple aggregators.');
$mrangeArguments->groupBy('type', 'max');
}
/**
* @group connected
* @group relay-resp3
@@ -226,6 +267,39 @@ class TSMREVRANGE_Test extends PredisCommandTestCase
$this->assertEquals($expectedResponse, $redis->tsmrevrange(1000, 1001, $mRangeArguments));
}
/**
* @group connected
* @group relay-resp3
* @return void
* @requiresRedisVersion >= 8.7.2
*/
public function testQueryMultipleTimeSeriesInReverseWithMultipleAggregators(): void
{
$redis = $this->getClient();
$this->assertEquals(
'OK',
$redis->tscreate('stock:A', (new CreateArguments())->labels('type', 'stock', 'name', 'A'))
);
$this->assertSame(
[1000, 1010, 1020],
$redis->tsmadd('stock:A', 1000, 100, 'stock:A', 1010, 110, 'stock:A', 1020, 120)
);
$mrangeArguments = (new MRangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000)
->filter('type=stock');
$response = $redis->tsmrevrange('-', '+', $mrangeArguments);
$this->assertCount(1, $response);
$this->assertSame('stock:A', $response[0][0]);
$samples = $response[0][2];
$this->assertCount(1, $samples);
$this->assertCount(3, $samples[0]);
$this->assertSame(1000, $samples[0][0]);
}
public function argumentsProvider(): array
{
return [
@@ -273,6 +347,18 @@ class TSMREVRANGE_Test extends PredisCommandTestCase
[1000, 1001, (new MRangeArguments())->aggregation('sum', 2, 0, 0, true)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'sum', 2, 'EMPTY', 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators as array' => [
[1000, 1001, (new MRangeArguments())->aggregation(['min', 'max'], 2)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'min,max', 2, 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators as string' => [
[1000, 1001, (new MRangeArguments())->aggregation('min,max', 2)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'AGGREGATION', 'min,max', 2, 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with AGGREGATION modifier - multiple aggregators with all options' => [
[1000, 1001, (new MRangeArguments())->aggregation(['min', 'max', 'avg'], 2, 2, 10000, true)->filter('filterExpression1', 'filterExpression2')],
[1000, 1001, 'ALIGN', 2, 'AGGREGATION', 'min,max,avg', 2, 'BUCKETTIMESTAMP', 10000, 'EMPTY', 'FILTER', 'filterExpression1', 'filterExpression2'],
],
'with GROUPBY modifier' => [
[1000, 1001, (new MRangeArguments())->filter('filterExpression1', 'filterExpression2')->groupBy('label', 'reducer')],
[1000, 1001, 'FILTER', 'filterExpression1', 'filterExpression2', 'GROUPBY', 'label', 'REDUCE', 'reducer'],
tests/Predis/Command/Redis/TimeSeries/TSRANGE_Test.php
+61
View File
@@ -61,6 +61,27 @@ class TSRANGE_Test extends PredisCommandTestCase
$this->assertSame(1, $this->getCommand()->parseResponse(1));
}
/**
* @group disconnected
* @dataProvider parseResponseProvider
*/
public function testParseResponsePassesThroughSingleAndMultipleAggregatorResults(array $response): void
{
$this->assertSame($response, $this->getCommand()->parseResponse($response));
}
public function parseResponseProvider(): array
{
return [
'single aggregator' => [
[[1000, '100'], [1020, '120']],
],
'multiple aggregators' => [
[[1000, '100', '200'], [1020, '120', '170']],
],
];
}
/**
* @group disconnected
*/
@@ -184,6 +205,34 @@ class TSRANGE_Test extends PredisCommandTestCase
);
}
/**
* @group connected
* @group relay-resp3
* @return void
* @requiresRedisVersion >= 8.7.2
*/
public function testReturnsQueriedRangeWithMultipleAggregators(): void
{
$redis = $this->getClient();
$createArguments = (new CreateArguments())->labels('type', 'stock', 'name', 'A');
$this->assertEquals('OK', $redis->tscreate('stock:A', $createArguments));
$this->assertSame(
[1000, 1010, 1020],
$redis->tsmadd('stock:A', 1000, 100, 'stock:A', 1010, 110, 'stock:A', 1020, 120)
);
$rangeArguments = (new RangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000);
$response = $redis->tsrange('stock:A', '-', '+', $rangeArguments);
$this->assertCount(1, $response);
$this->assertCount(3, $response[0]);
$this->assertSame(1000, $response[0][0]);
}
/**
* @group connected
* @group relay-resp3
@@ -239,6 +288,18 @@ class TSRANGE_Test extends PredisCommandTestCase
['key', 10000, 10001, (new RangeArguments())->aggregation('sum', 100, 0, 0, true)],
['key', 10000, 10001, 'AGGREGATION', 'sum', 100, 'EMPTY'],
],
'with AGGREGATION modifier - multiple aggregators as array' => [
['key', 10000, 10001, (new RangeArguments())->aggregation(['min', 'max'], 100)],
['key', 10000, 10001, 'AGGREGATION', 'min,max', 100],
],
'with AGGREGATION modifier - multiple aggregators as string' => [
['key', 10000, 10001, (new RangeArguments())->aggregation('min,max', 100)],
['key', 10000, 10001, 'AGGREGATION', 'min,max', 100],
],
'with AGGREGATION modifier - multiple aggregators with all options' => [
['key', 10000, 10001, (new RangeArguments())->aggregation(['min', 'max', 'avg'], 100, 10, 1000, true)],
['key', 10000, 10001, 'ALIGN', 10, 'AGGREGATION', 'min,max,avg', 100, 'BUCKETTIMESTAMP', 1000, 'EMPTY'],
],
'with all modifiers' => [
['key', 10000, 10001, (new RangeArguments())->latest()->filterByTs(1000, 1001)->filterByValue(1000, 1001)->count(100)->aggregation('sum', 100)],
['key', 10000, 10001, 'LATEST', 'FILTER_BY_TS', 1000, 1001, 'FILTER_BY_VALUE', 1000, 1001, 'COUNT', 100, 'AGGREGATION', 'sum', 100],
tests/Predis/Command/Redis/TimeSeries/TSREVRANGE_Test.php
+61
View File
@@ -61,6 +61,27 @@ class TSREVRANGE_Test extends PredisCommandTestCase
$this->assertSame(1, $this->getCommand()->parseResponse(1));
}
/**
* @group disconnected
* @dataProvider parseResponseProvider
*/
public function testParseResponsePassesThroughSingleAndMultipleAggregatorResults(array $response): void
{
$this->assertSame($response, $this->getCommand()->parseResponse($response));
}
public function parseResponseProvider(): array
{
return [
'single aggregator' => [
[[1020, '120'], [1000, '100']],
],
'multiple aggregators' => [
[[1020, '120', '170'], [1000, '100', '200']],
],
];
}
/**
* @group disconnected
*/
@@ -184,6 +205,34 @@ class TSREVRANGE_Test extends PredisCommandTestCase
);
}
/**
* @group connected
* @group relay-resp3
* @return void
* @requiresRedisVersion >= 8.7.2
*/
public function testReturnsQueriedRangeWithMultipleAggregators(): void
{
$redis = $this->getClient();
$createArguments = (new CreateArguments())->labels('type', 'stock', 'name', 'A');
$this->assertEquals('OK', $redis->tscreate('stock:A', $createArguments));
$this->assertSame(
[1000, 1010, 1020],
$redis->tsmadd('stock:A', 1000, 100, 'stock:A', 1010, 110, 'stock:A', 1020, 120)
);
$rangeArguments = (new RangeArguments())
->aggregation([RangeArguments::AGG_MIN, RangeArguments::AGG_MAX], 1000);
$response = $redis->tsrevrange('stock:A', '-', '+', $rangeArguments);
$this->assertCount(1, $response);
$this->assertCount(3, $response[0]);
$this->assertSame(1000, $response[0][0]);
}
/**
* @group connected
* @group relay-resp3
@@ -239,6 +288,18 @@ class TSREVRANGE_Test extends PredisCommandTestCase
['key', 10000, 10001, (new RangeArguments())->aggregation('sum', 100, 0, 0, true)],
['key', 10000, 10001, 'AGGREGATION', 'sum', 100, 'EMPTY'],
],
'with AGGREGATION modifier - multiple aggregators as array' => [
['key', 10000, 10001, (new RangeArguments())->aggregation(['min', 'max'], 100)],
['key', 10000, 10001, 'AGGREGATION', 'min,max', 100],
],
'with AGGREGATION modifier - multiple aggregators as string' => [
['key', 10000, 10001, (new RangeArguments())->aggregation('min,max', 100)],
['key', 10000, 10001, 'AGGREGATION', 'min,max', 100],
],
'with AGGREGATION modifier - multiple aggregators with all options' => [
['key', 10000, 10001, (new RangeArguments())->aggregation(['min', 'max', 'avg'], 100, 10, 1000, true)],
['key', 10000, 10001, 'ALIGN', 10, 'AGGREGATION', 'min,max,avg', 100, 'BUCKETTIMESTAMP', 1000, 'EMPTY'],
],
'with all modifiers' => [
['key', 10000, 10001, (new RangeArguments())->latest()->filterByTs(1000, 1001)->filterByValue(1000, 1001)->count(100)->aggregation('sum', 100)],
['key', 10000, 10001, 'LATEST', 'FILTER_BY_TS', 1000, 1001, 'FILTER_BY_VALUE', 1000, 1001, 'COUNT', 100, 'AGGREGATION', 'sum', 100],