Merge pull request #97 from kununu/composite-aggregations

Add Composite Aggregations Builder and Repository

composer.json

@@ -11,6 +11,7 @@
     "ext-json": "*",
     "ext-mbstring": "*",
     "elasticsearch/elasticsearch": "^7.0",
+    "kununu/collections": "^4.1.0",
     "psr/log": "^1.0|^2.0|^3.0"
   },
   "require-dev": {

src/Exception/MissingAggregationAttributesException.php

@@ -0,0 +1,10 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Exception;
+
+use RuntimeException;
+
+final class MissingAggregationAttributesException extends RuntimeException
+{
+}

src/Query/Aggregation/Builder/CompositeAggregationQueryBuilder.php

@@ -0,0 +1,105 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query\Aggregation\Builder;
+
+use Kununu\Elasticsearch\Exception\MissingAggregationAttributesException;
+use Kununu\Elasticsearch\Query\Aggregation\SourceProperty;
+use Kununu\Elasticsearch\Query\Aggregation\Sources;
+use Kununu\Elasticsearch\Query\CompositeAggregationQueryInterface;
+use Kununu\Elasticsearch\Query\Criteria\Filter;
+use Kununu\Elasticsearch\Query\Criteria\Filters;
+use Kununu\Elasticsearch\Query\QueryInterface;
+use Kununu\Elasticsearch\Query\RawQuery;
+use Kununu\Elasticsearch\Util\ArrayUtilities;
+
+final class CompositeAggregationQueryBuilder implements CompositeAggregationQueryInterface
+{
+    private ?array $afterKey = null;
+    private Filters $filters;
+    private ?string $name = null;
+    private ?Sources $sources = null;
+
+    private function __construct()
+    {
+        $this->filters = new Filters();
+    }
+
+    public static function create(): self
+    {
+        return new self();
+    }
+
+    public function withAfterKey(?array $afterKey): self
+    {
+        $this->afterKey = $afterKey;
+
+        return $this;
+    }
+
+    public function withFilters(Filters $filters): self
+    {
+        $this->filters = $filters;
+
+        return $this;
+    }
+
+    public function withName(string $name): self
+    {
+        $this->name = $name;
+
+        return $this;
+    }
+
+    public function withSources(Sources $sources): self
+    {
+        $this->sources = $sources;
+
+        return $this;
+    }
+
+    public function getName(): string
+    {
+        if (null === $this->name) {
+            throw new MissingAggregationAttributesException('Aggregation name is missing');
+        }
+
+        return $this->name;
+    }
+
+    public function getQuery(int $compositeSize = 100): QueryInterface
+    {
+        return RawQuery::create(
+            ArrayUtilities::filterNullAndEmptyValues([
+                'query' => [
+                    'bool' => [
+                        'must' => $this->filters->map(fn(Filter $filter) => $filter->toArray()),
+                    ],
+                ],
+                'aggs' => [
+                    $this->getName() => [
+                        'composite' => [
+                            'size' => $compositeSize,
+                            'sources' => $this->sources?->map(
+                                fn(SourceProperty $sourceProperty) => [
+                                    $sourceProperty->source => [
+                                        'terms' => [
+                                            'field' => $sourceProperty->property,
+                                            'missing_bucket' => $sourceProperty->missingBucket,
+                                        ]
+                                    ],
+                                ]
+                            ) ?? [],
+                            'after' => $this->afterKey,
+                        ],
+                    ],
+                ],
+            ], true)
+        );
+    }
+
+    public function toArray(): array
+    {
+        return $this->getQuery()->toArray();
+    }
+}

src/Query/Aggregation/SourceProperty.php

@@ -0,0 +1,15 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query\Aggregation;
+
+final class SourceProperty
+{
+    public function __construct(
+        public readonly string $source,
+        public readonly string $property,
+        public readonly bool $missingBucket = false
+    )
+    {
+    }
+}

src/Query/Aggregation/Sources.php

@@ -0,0 +1,37 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query\Aggregation;
+
+use InvalidArgumentException;
+use Kununu\Collection\AbstractCollection;
+
+final class Sources extends AbstractCollection
+{
+    private const INVALID = 'Can only append %s';
+
+    public function __construct(SourceProperty ...$sourceProperties)
+    {
+        parent::__construct();
+
+        foreach ($sourceProperties as $sourceProperty) {
+            $this->append($sourceProperty);
+        }
+    }
+
+    public function current(): ?SourceProperty
+    {
+        $current = parent::current();
+        assert($this->count() > 0 ? $current instanceof SourceProperty : null === $current);
+
+        return $current;
+    }
+
+    public function append($value): void
+    {
+        match (true) {
+            $value instanceof SourceProperty  => parent::append($value),
+            default                           => throw new InvalidArgumentException(sprintf(self::INVALID, SourceProperty::class))
+        };
+    }
+}

src/Query/CompositeAggregationQueryInterface.php

@@ -0,0 +1,17 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query;
+
+interface CompositeAggregationQueryInterface
+{
+    public function getQuery(int $compositeSize = 100): QueryInterface;
+
+    public function getName(): string;
+
+    public function withName(string $name): self;
+
+    public function withAfterKey(?array $afterKey): self;
+
+    public function toArray(): array;
+}

src/Query/Criteria/Filters.php

@@ -0,0 +1,37 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query\Criteria;
+
+use InvalidArgumentException;
+use Kununu\Collection\AbstractCollection;
+
+class Filters extends AbstractCollection
+{
+    private const INVALID = 'Can only append %s';
+
+    public function __construct(Filter ...$propertyFilters)
+    {
+        parent::__construct();
+
+        foreach ($propertyFilters as $propertyFilter) {
+            $this->append($propertyFilter);
+        }
+    }
+
+    public function current(): ?Filter
+    {
+        $current = parent::current();
+        assert($this->count() > 0 ? $current instanceof Filter : null === $current);
+
+        return $current;
+    }
+
+    public function append($value): void
+    {
+        match (true) {
+            $value instanceof Filter  => parent::append($value),
+            default                           => throw new InvalidArgumentException(sprintf(self::INVALID, Filter::class))
+        };
+    }
+}

src/Repository/Repository.php

@@ -5,6 +5,7 @@
 
 use Elasticsearch\Client;
 use Elasticsearch\Common\Exceptions\Missing404Exception;
+use Generator;
 use Kununu\Elasticsearch\Exception\BulkException;
 use Kununu\Elasticsearch\Exception\DeleteException;
 use Kununu\Elasticsearch\Exception\DocumentNotFoundException;
@@ -14,10 +15,12 @@
 use Kununu\Elasticsearch\Exception\UpdateException;
 use Kununu\Elasticsearch\Exception\UpsertException;
 use Kununu\Elasticsearch\Exception\WriteOperationException;
+use Kununu\Elasticsearch\Query\CompositeAggregationQueryInterface;
 use Kununu\Elasticsearch\Query\Query;
 use Kununu\Elasticsearch\Query\QueryInterface;
 use Kununu\Elasticsearch\Result\AggregationResultSet;
 use Kununu\Elasticsearch\Result\AggregationResultSetInterface;
+use Kununu\Elasticsearch\Result\CompositeResult;
 use Kununu\Elasticsearch\Result\ResultIterator;
 use Kununu\Elasticsearch\Result\ResultIteratorInterface;
 use Kununu\Elasticsearch\Util\LoggerAwareTrait;
@@ -283,6 +286,29 @@ function() use ($query) {
         );
     }
 
+    public function aggregateCompositeByQuery(CompositeAggregationQueryInterface $query): Generator
+    {
+        $afterKey = null;
+
+        do {
+            $result = $this->aggregateByQuery(
+                $query->withAfterKey($afterKey)->getQuery()
+            )->getResultByName($query->getName());
+
+            foreach ($result?->getFields()['buckets'] ?? [] as $bucket) {
+                if (!empty($bucket['key']) && !empty($bucket['doc_count'])) {
+                    yield new CompositeResult(
+                        $bucket['key'],
+                        $bucket['doc_count'],
+                        $query->getName()
+                    );
+                }
+            }
+
+            $afterKey = $result?->get('after_key') ?? null;
+        } while (null !== $afterKey);
+    }
+
     public function updateByQuery(QueryInterface $query, array $updateScript): array
     {
         return $this->executeWrite(

src/Repository/RepositoryInterface.php

@@ -3,6 +3,8 @@
 
 namespace Kununu\Elasticsearch\Repository;
 
+use Generator;
+use Kununu\Elasticsearch\Query\CompositeAggregationQueryInterface;
 use Kununu\Elasticsearch\Query\QueryInterface;
 use Kununu\Elasticsearch\Result\AggregationResultSetInterface;
 use Kununu\Elasticsearch\Result\ResultIteratorInterface;
@@ -102,6 +104,13 @@ public function countByQuery(QueryInterface $query): int;
      */
     public function aggregateByQuery(QueryInterface $query): AggregationResultSetInterface;
 
+    /**
+     * This method executes a query with composite aggregation, iterates through the results, and retrieves the data.
+     *
+     * @return Generator <CompositeResult>
+     */
+    public function aggregateCompositeByQuery(CompositeAggregationQueryInterface $query): Generator;
+
     /**
      * This method updates all documents matching a given $query using a given $updateScript.
      */

src/Result/CompositeResult.php

@@ -0,0 +1,15 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Result;
+
+final class CompositeResult
+{
+    public function __construct(
+        public readonly array $results,
+        public readonly int $documentsCount,
+        public readonly string $aggregationName
+    )
+    {
+    }
+}

src/Util/ArrayUtilities.php

@@ -0,0 +1,20 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Util;
+
+final class ArrayUtilities
+{
+    public static function filterNullAndEmptyValues(array $values, bool $recursive = false): array
+    {
+        if ($recursive) {
+            foreach ($values as &$value) {
+                if (is_array($value)) {
+                    $value = self::filterNullAndEmptyValues($value, true);
+                }
+            }
+        }
+
+        return array_filter($values, fn ($value) => $value !== null && $value !== []);
+    }
+}