Add aggregateWindow() function to Query Builder

Signed-off-by: davidarendsen <davidarendsen@hey.com>

src/Functions/AggregateWindow.php

@@ -0,0 +1,89 @@
+<?php
+
+namespace Arendsen\FluxQueryBuilder\Functions;
+
+use Arendsen\FluxQueryBuilder\Type;
+use Arendsen\FluxQueryBuilder\Type\ArrayType;
+use Arendsen\FluxQueryBuilder\Type\DurationType;
+use Arendsen\FluxQueryBuilder\Type\FnType;
+
+class AggregateWindow extends Base
+{
+    /**
+     * @var string $every
+     */
+    private $every;
+
+    /**
+     * @var string|null $period
+     */
+    private $period;
+
+    /**
+     * @var string|null $offset
+     */
+    private $offset;
+
+    /**
+     * @var string|null $location
+     */
+    private $location;
+
+    /**
+     * @var string|null $column
+     */
+    private $column;
+
+    /**
+     * @var string|null $timeSrc
+     */
+    private $timeSrc;
+
+    /**
+     * @var string|null $timeDst
+     */
+    private $timeDst;
+
+    /**
+     * @var bool $createEmpty
+     */
+    private $createEmpty;
+
+    public function __construct(
+        $every,
+        ?string $period = null,
+        ?string $offset = null,
+        $fn,
+        ?string $location = null,
+        ?string $column = null,
+        ?string $timeSrc = null,
+        ?string $timeDst = null,
+        bool $createEmpty = true
+    ) {
+        $this->every = $every;
+        $this->period = $period;
+        $this->offset = $offset;
+        $this->fn = $fn;
+        $this->location = $location;
+        $this->column = $column;
+        $this->timeSrc = $timeSrc;
+        $this->timeDst = $timeDst;
+        $this->createEmpty = $createEmpty;
+    }
+
+    public function __toString()
+    {
+        $input = new ArrayType(array_filter([
+            'every' => new DurationType($this->every),
+            'period' => $this->period ? new DurationType($this->period) : null,
+            'offset' => $this->offset ? new DurationType($this->offset) : null,
+            'fn' => new FnType($this->fn),
+            'location' => $this->location ? new Type($this->location) : null,
+            'column' => $this->column ? new Type($this->column) : null,
+            'timeSrc' => $this->timeSrc ? new Type($this->timeSrc) : null,
+            'timeDst' => $this->timeDst ? new Type($this->timeDst) : null,
+            'createEmpty' => !$this->createEmpty ? new Type($this->createEmpty) : null,
+        ]));
+        return '|> aggregateWindow(' . $input . ') ';
+    }
+}

src/QueryBuilder.php

@@ -5,6 +5,7 @@ namespace Arendsen\FluxQueryBuilder;
 use DateTime;
 use Exception;
 use Arendsen\FluxQueryBuilder\Expression\KeyValue;
+use Arendsen\FluxQueryBuilder\Functions\AggregateWindow;
 use Arendsen\FluxQueryBuilder\Functions\Filter;
 use Arendsen\FluxQueryBuilder\Functions\From;
 use Arendsen\FluxQueryBuilder\Functions\Range;
@@ -27,11 +28,13 @@ class QueryBuilder
     public const FLUX_PART_LIMIT = 'limit';
     public const FLUX_PART_WINDOW = 'window';
     public const FLUX_PART_UNWINDOW = 'unwindow';
+    public const FLUX_PART_AGGREGATEWINDOW = 'aggregateWindow';
 
     public const PARTS = [
         self::FLUX_PART_FROM,
         self::FLUX_PART_RANGE,
         self::FLUX_PART_REDUCE,
+        self::FLUX_PART_AGGREGATEWINDOW,
         self::FLUX_PART_WINDOW,
         self::FLUX_PART_FILTERS,
         self::FLUX_PART_MAP,
@@ -198,6 +201,24 @@ class QueryBuilder
         return $this;
     }
 
+    public function addAggregateWindow(
+        $every,
+        ?string $period = null,
+        ?string $offset = null,
+        $fn,
+        ?string $location = null,
+        ?string $column = null,
+        ?string $timeSrc = null,
+        ?string $timeDst = null,
+        bool $createEmpty = true
+    ): QueryBuilder {
+        $this->addToQuery(
+            self::FLUX_PART_AGGREGATEWINDOW,
+            new AggregateWindow($every, $period, $offset, $fn, $location, $column, $timeSrc, $timeDst, $createEmpty)
+        );
+        return $this;
+    }
+
     protected function addToQuery($key, $query)
     {
         $this->fluxQueryParts[$key] = $query;

src/Type/FnType.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace Arendsen\FluxQueryBuilder\Type;
+
+class FnType implements TypeInterface
+{
+    public function __construct(string $value)
+    {
+        $this->value = $value;
+    }
+
+    public function __toString(): string
+    {
+        return $this->value;
+    }
+}

tests/Functions/AggregateWindowTest.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Functions;
+
+use Arendsen\FluxQueryBuilder\Functions\AggregateWindow;
+use PHPUnit\Framework\TestCase;
+
+final class AggregateWindowFunctionTest extends TestCase
+{
+    public function testSimpleWindow()
+    {
+        $expression = new AggregateWindow('20s', null, null, 'mean');
+
+        $query = '|> aggregateWindow(every: 20s, fn: mean) ';
+
+        $this->assertEquals($query, $expression->__toString());
+    }
+
+    public function testAllParameters()
+    {
+        $expression = new AggregateWindow('20s', 'every', '0s', 'mean', 'location', '_value', '_stop', '_time', false);
+
+        $query = '|> aggregateWindow(every: 20s, period: every, offset: 0s, fn: mean, location: "location", ' .
+            'column: "_value", timeSrc: "_stop", timeDst: "_time", createEmpty: false) ';
+
+        $this->assertEquals($query, $expression->__toString());
+    }
+}

tests/Generic/QueryBuilderTest.php

@@ -122,7 +122,6 @@ final class QueryBuilderTest extends TestCase
         $queryBuilder->fromBucket('test_bucket')
             ->fromMeasurement('test_measurement')
             ->addRangeStart(new DateTime('2022-08-12 17:31:00'))
-            ->addFieldFilter(['username', 'ip'])
             ->addWindow('20s')
             ->addReduce(['count' => new MathType('accumulator.count + 1')], ['count' => 0])
             ->addFilter(KeyValue::setGreaterOrEqualTo('count', 1)->andGreaterOrEqualTo('count2', 2))
@@ -131,9 +130,24 @@ final class QueryBuilderTest extends TestCase
         $expectedQuery = 'from(bucket: "test_bucket") |> range(start: time(v: 2022-08-12T17:31:00Z)) ' .
         '|> reduce(fn: (r, accumulator) => ({count: accumulator.count + 1}), identity: {count: 0}) ' .
         '|> window(every: 20s) |> filter(fn: (r) => r._measurement == "test_measurement") ' .
-        '|> filter(fn: (r) => r._field == "username" or r._field == "ip") ' .
         '|> filter(fn: (r) => r.count >= 1 and r.count2 >= 2) |> window(every: inf) ';
 
         $this->assertEquals($expectedQuery, $queryBuilder->build());
     }
+
+    public function testQueryWithAggregateWindow()
+    {
+        $queryBuilder = new QueryBuilder();
+        $queryBuilder->fromBucket('test_bucket')
+            ->fromMeasurement('test_measurement')
+            ->addRangeStart(new DateTime('2022-08-12 17:31:00'))
+            ->addAggregateWindow('20s', null, null, 'mean')
+            ->addReduce(['count' => new MathType('accumulator.count + 1')], ['count' => 0]);
+
+        $expectedQuery = 'from(bucket: "test_bucket") |> range(start: time(v: 2022-08-12T17:31:00Z)) ' .
+        '|> reduce(fn: (r, accumulator) => ({count: accumulator.count + 1}), identity: {count: 0}) ' .
+        '|> aggregateWindow(every: 20s, fn: mean) |> filter(fn: (r) => r._measurement == "test_measurement") ';
+
+        $this->assertEquals($expectedQuery, $queryBuilder->build());
+    }
 }