Added Implementation for Composite Aggregations.

composer.json

@@ -4,13 +4,21 @@
   "type": "library",
   "license": "proprietary",
   "minimum-stability": "stable",
+  "repositories": [
+    {
+      "type": "vcs",
+      "url": "https://github.com/kununu/utilities.git",
+      "no-api": true
+    }
+  ],
   "require": {
     "php": ">=8.1",
     "ext-ctype": "*",
     "ext-iconv": "*",
     "ext-json": "*",
     "ext-mbstring": "*",
     "elasticsearch/elasticsearch": "^7.0",
+    "kununu/utilities": "^4.8.0",
     "psr/log": "^1.0|^2.0|^3.0"
   },
   "require-dev": {

src/Query/Aggregation/Builder/CompositeAggregationBuilder.php

@@ -0,0 +1,109 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Query\Aggregation\Builder;
+
+use Kununu\Elasticsearch\Query\Aggregation\SourceProperty;
+use Kununu\Elasticsearch\Query\Aggregation\Sources;
+use Kununu\Elasticsearch\Query\AggregationInterface;
+use Kununu\Elasticsearch\Query\Criteria\Filter;
+use Kununu\Elasticsearch\Query\Criteria\Filters;
+use Kununu\Elasticsearch\Query\QueryInterface;
+use Kununu\Elasticsearch\Query\RawQuery;
+use Kununu\Utilities\Arrays\ArrayUtilities;
+use Kununu\Utilities\Elasticsearch\Q;
+use RuntimeException;
+
+class CompositeAggregationBuilder implements AggregationInterface
+{
+    private ?array $afterKey;
+    private Filters $filters;
+    private ?string $name;
+    private ?Sources $sources;
+
+    private function __construct()
+    {
+        $this->afterKey = null;
+        $this->filters = new Filters();
+        $this->name = null;
+        $this->sources = null;
+    }
+
+    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 RuntimeException('Aggregation name is required');
+        }
+
+        return $this->name;
+    }
+
+    public function getQuery(int $compositeSize = 100): QueryInterface
+    {
+        return RawQuery::create(
+            ArrayUtilities::filterNullAndEmptyValues([
+                Q::query() => [
+                    Q::bool() => [
+                        Q::must() => $this->filters->map(fn(Filter $filter) => $filter->toArray()),
+                    ],
+                ],
+                Q::aggs() => [
+                    $this->getName() => [
+                        Q::composite() => [
+                            Q::size() => $compositeSize,
+                            Q::sources() => $this->sources?->map(
+                                    fn(SourceProperty $sourceProperty) => [
+                                        $sourceProperty->source => [
+                                            Q::terms() => [
+                                                Q::field() => $sourceProperty->property,
+                                                Q::missingBucket() => $sourceProperty->missingBucket,
+                                            ]
+                                        ],
+                                    ]
+                                ) ?? [],
+                            Q::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/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/CompositeAggregationRepository.php

@@ -0,0 +1,40 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Repository;
+
+use Generator;
+use Kununu\Elasticsearch\Query\Aggregation\Builder\CompositeAggregationBuilder;
+use Kununu\Elasticsearch\Query\Aggregation\Sources;
+use Kununu\Elasticsearch\Query\Criteria\Filters;
+use Kununu\Elasticsearch\Result\CompositeResult;
+use Kununu\Utilities\Elasticsearch\Q;
+
+final class CompositeAggregationRepository extends Repository implements CompositeAggregationRepositoryInterface
+{
+    public function lookup(Filters $filters, Sources $sources, string $aggregationName): Generator
+    {
+        $elasticsearchQuery = CompositeAggregationBuilder::create()
+            ->withName($aggregationName)
+            ->withFilters($filters)
+            ->withSources($sources);
+
+        do {
+            $result = $this->aggregateByQuery(
+                $elasticsearchQuery->getQuery()->limit(0)
+            )->getResultByName($aggregationName);
+
+            foreach ($result?->getFields()[Q::buckets()] ?? [] as $bucket) {
+                if (!empty($bucket[Q::key()]) && !empty($bucket[Q::docCount()])) {
+                    yield new CompositeResult(
+                        $bucket[Q::key()],
+                        $bucket[Q::docCount()],
+                        $aggregationName
+                    );
+                }
+            }
+
+            $elasticsearchQuery->withAfterKey($afterKey = ($result?->get(Q::afterKey()) ?? null));
+        } while (null !== $afterKey);
+    }
+}

src/Repository/CompositeAggregationRepositoryInterface.php

@@ -0,0 +1,13 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Repository;
+
+use Generator;
+use Kununu\Elasticsearch\Query\Aggregation\Sources;
+use Kununu\Elasticsearch\Query\Criteria\Filters;
+
+interface CompositeAggregationRepositoryInterface
+{
+    public function lookup(Filters $filters, Sources $sources, string $aggregationName): Generator;
+}

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
+    )
+    {
+    }
+}

tests/Query/Aggregation/Builder/CompositeAggregationBuilderTest.php

@@ -0,0 +1,62 @@
+<?php
+declare(strict_types=1);
+
+namespace Kununu\Elasticsearch\Tests\Query\Aggregation\Builder;
+
+use Kununu\Elasticsearch\Query\Aggregation\Builder\CompositeAggregationBuilder;
+use Kununu\Elasticsearch\Query\Aggregation\SourceProperty;
+use Kununu\Elasticsearch\Query\Aggregation\Sources;
+use Kununu\Elasticsearch\Query\Criteria\Filter;
+use Kununu\Elasticsearch\Query\Criteria\Filters;
+use PHPUnit\Framework\TestCase;
+
+class CompositeAggregationBuilderTest extends TestCase
+{
+    public function testCompositeAggregationBuilder(): void
+    {
+        $compositeAggregation = CompositeAggregationBuilder::create()
+            ->withName('agg')
+            ->withFilters(new Filters(
+                new Filter('field', 'value')
+            ))
+            ->withSources(
+                new Sources(
+                    new SourceProperty('field', 'value')
+                )
+            );
+
+        self::assertEquals(
+            [
+                'query' => [
+                    'bool' => [
+                        'must' => [
+                            [
+                                'term' => [
+                                    'field' => 'value'
+                                ]
+                            ]
+                        ]
+                    ]
+                ],
+                'aggs' => [
+                    'agg' => [
+                        'composite' => [
+                            'size' => 100,
+                            'sources' => [
+                                [
+                                    'field' => [
+                                        'terms' => [
+                                            'field' => 'value',
+                                            'missing_bucket' => false,
+                                        ]
+                                    ]
+                                ]
+                            ]
+                        ]
+                    ]
+                ],
+            ],
+            $compositeAggregation->getQuery()->toArray()
+        );
+    }
+}