Added String Agg function (#1389)

norberttech · web-flow · commit 6d179598414a · 2025-01-17T22:05:19.000-06:00
diff --git a/documentation/components/core/aggregations.md b/documentation/components/core/aggregations.md
@@ -1,6 +1,6 @@
 # Aggregations
 
-- [⬅️️ Back](group-by.md)
+- [⬅️️ Back](/documentation/components/core/group-by.md)
 
 Each group created by `groupBy` function can be aggregated using one of the aggregation functions.
 
@@ -13,8 +13,9 @@ Each group created by `groupBy` function can be aggregated using one of the aggr
 - [`MAX()`](/src/core/etl/src/Flow/ETL/Function/Max.php)
 - [`COLLECT()`](/src/core/etl/src/Flow/ETL/Function/Collect.php)
 - [`COLLECT_UNIQUE()`](/src/core/etl/src/Flow/ETL/Function/CollectUnique.php)
-- [`FIRST`](/src/core/etl/src/Flow/ETL/Function/First.php)
-- [`LAST`](/src/core/etl/src/Flow/ETL/Function/Last.php)
+- [`FIRST()`](/src/core/etl/src/Flow/ETL/Function/First.php)
+- [`LAST()`](/src/core/etl/src/Flow/ETL/Function/Last.php)
+- [`STRIRNG_AGG()`](/src/core/etl/src/Flow/ETL/Function/StringAggregate.php)
 
 All aggregation functions are implementing [`AggregatingFunction`](/src/core/etl/src/Flow/ETL/Function/AggregatingFunction.php) interface.
 
diff --git a/documentation/components/core/building-blocks.md b/documentation/components/core/building-blocks.md
@@ -1,6 +1,6 @@
 # Building Blocks
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Entries are the columns of the data frame, they are represented by the [Entry](/src/core/etl/src/Flow/ETL/Row/Entry.php) interface.
 Group of Entries is called `Row`, it is represented by the [Row](/src/core/etl/src/Flow/ETL/Row.php) class.
diff --git a/documentation/components/core/caching.md b/documentation/components/core/caching.md
@@ -1,6 +1,6 @@
 # Caching
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 ## Cache
 
diff --git a/documentation/components/core/core.md b/documentation/components/core/core.md
@@ -1,6 +1,6 @@
 # Data Frame
 
-- [⬅️️ Back](../../quick-start.md)
+- [⬅️️ Back](/documentation/quick-start.md)
 
 A Data Frame is a structured collection of tabular data, similar to a spreadsheet.  
 It organizes information into rows and columns, making it easy to understand, filter, and transform.  
diff --git a/documentation/components/core/display.md b/documentation/components/core/display.md
@@ -1,6 +1,6 @@
 # Display
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Displaying/printing data frame is a common operation during development. 
 It's the easiest way to debug data frame.
diff --git a/documentation/components/core/error-handling.md b/documentation/components/core/error-handling.md
@@ -1,6 +1,6 @@
 # Error Handling
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 In case of any exception in transform/load steps, an ETL process will break, 
 to change that behavior, please set custom [ErrorHandler](/src/core/etl/src/Flow/ETL/ErrorHandler.php).
diff --git a/documentation/components/core/filter.md b/documentation/components/core/filter.md
@@ -1,6 +1,6 @@
 # Filter
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 To filter rows from the data frame you can use `DataFrame::filter` function.
 Filter function accepts only one argument which is a `ScalarFunction` that returns `bool` value.
diff --git a/documentation/components/core/group-by.md b/documentation/components/core/group-by.md
@@ -1,6 +1,6 @@
 # Group By
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Grouping is a common operation in data analysis. The concept is no different from the one you know from SQL.
 
diff --git a/documentation/components/core/join.md b/documentation/components/core/join.md
@@ -1,6 +1,6 @@
 # Join
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Joining two data frames is a common operation in data processing. 
 It is used to combine data from two different sources into one data frame. 
diff --git a/documentation/components/core/limit.md b/documentation/components/core/limit.md
@@ -1,6 +1,6 @@
 # Limit
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Sometimes you might just want to process only few first rows, maybe for debugging purpose or you don't want to go above certain number of rows.
 
diff --git a/documentation/components/core/map.md b/documentation/components/core/map.md
@@ -1,6 +1,6 @@
 # Map
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Quick `Row` transformations are available through `DataFrame::map` function
 
diff --git a/documentation/components/core/rename.md b/documentation/components/core/rename.md
@@ -1,6 +1,6 @@
 # Rename
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 There are multiple ways to rename entries in the data frame.
 
diff --git a/documentation/components/core/schema.md b/documentation/components/core/schema.md
@@ -1,6 +1,6 @@
 # Schema
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Schema is a set of rules that defines how data is structured.
 It is used to validate data and to provide information about data structure.
diff --git a/documentation/components/core/select-drop.md b/documentation/components/core/select-drop.md
@@ -1,6 +1,6 @@
 # Select / Drop
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 ## Select
 
diff --git a/documentation/components/core/sort.md b/documentation/components/core/sort.md
@@ -1,6 +1,6 @@
 # Sort
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Sorting is a common operation in data processing.
 Thanks to implementation of External Sort algorithm, sorting as everything else is by default memory-safe. 
diff --git a/documentation/components/core/until.md b/documentation/components/core/until.md
@@ -1,6 +1,6 @@
 # Until
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 `Until` works in a very similar way to [filter](filter.md) function. 
 The difference is that it will send STOP signal to extractor when the condition is met.
diff --git a/documentation/components/core/window-functions.md b/documentation/components/core/window-functions.md
@@ -1,6 +1,6 @@
 # Window Functions
 
-- [⬅️️ Back](core.md)
+- [⬅️️ Back](/documentation/components/core/core.md)
 
 Window functions are a special type of function that perform calculations on a set of rows that are related to the current row. 
 Unlike regular aggregate functions, use of a window function does not cause rows to become grouped into a single output row — the rows retain their separate identities. 
diff --git a/src/core/etl/src/Flow/ETL/DSL/functions.php b/src/core/etl/src/Flow/ETL/DSL/functions.php
@@ -67,6 +67,7 @@
     Size,
     Split,
     Sprintf,
+    StringAggregate,
     StructureFunctions,
     Sum,
     ToDate,
@@ -1196,6 +1197,12 @@ function collect(EntryReference|string $ref) : Collect
     return new Collect(is_string($ref) ? ref($ref) : $ref);
 }
 
+#[DocumentationDSL(module: Module::CORE, type: DSLType::AGGREGATING_FUNCTION)]
+function string_agg(EntryReference|string $ref, string $separator = ', ', ?Row\SortOrder $sort = null) : StringAggregate
+{
+    return new StringAggregate(is_string($ref) ? ref($ref) : $ref, $separator, $sort);
+}
+
 #[DocumentationDSL(module: Module::CORE, type: DSLType::AGGREGATING_FUNCTION)]
 function collect_unique(EntryReference|string $ref) : CollectUnique
 {
diff --git a/src/core/etl/src/Flow/ETL/Function/StringAggregate.php b/src/core/etl/src/Flow/ETL/Function/StringAggregate.php
@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Function;
+
+use function Flow\ETL\DSL\str_entry;
+use Flow\ETL\Row;
+use Flow\ETL\Row\Entry\StringEntry;
+use Flow\ETL\Row\{Reference, SortOrder};
+
+final class StringAggregate implements AggregatingFunction
+{
+    /**
+     * @var array<string>
+     */
+    private array $values = [];
+
+    public function __construct(private readonly Reference $ref, private readonly string $separator, private readonly ?SortOrder $sort = null)
+    {
+    }
+
+    public function aggregate(Row $row) : void
+    {
+        $stringValue = (new Parameter($this->ref))->asString($row);
+
+        if ($stringValue !== null) {
+            $this->values[] = $stringValue;
+        }
+    }
+
+    public function result() : StringEntry
+    {
+        if (!$this->ref->hasAlias()) {
+            $this->ref->as($this->ref->to() . '_str_agg');
+        }
+
+        if (!\count($this->values)) {
+            return str_entry($this->ref->name(), '');
+        }
+
+        if ($this->sort) {
+            $this->sort === SortOrder::ASC ?
+                \sort($this->values)
+                : \rsort($this->values);
+        }
+
+        return str_entry($this->ref->name(), \implode($this->separator, $this->values));
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/StringAggregateTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/StringAggregateTest.php
@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Unit\Function;
+
+use function Flow\ETL\DSL\row;
+use function Flow\ETL\DSL\{int_entry, ref, str_entry, string_agg};
+use Flow\ETL\Row\SortOrder;
+use Flow\ETL\Tests\FlowTestCase;
+
+final class StringAggregateTest extends FlowTestCase
+{
+    public function test_string_agg() : void
+    {
+        $aggregator = string_agg(ref('data'));
+
+        $aggregator->aggregate(row(str_entry('data', 'a')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'c')));
+
+        self::assertSame(
+            'a, b, b, c',
+            $aggregator->result()->value()
+        );
+        self::assertSame('data_str_agg', $aggregator->result()->name());
+    }
+
+    public function test_string_agg_on_empty_rows() : void
+    {
+        $aggregator = string_agg(ref('data'), sort: SortOrder::DESC);
+
+        self::assertSame(
+            '',
+            $aggregator->result()->value()
+        );
+    }
+
+    public function test_string_agg_on_non_string() : void
+    {
+        $aggregator = string_agg(ref('data'), sort: SortOrder::DESC);
+
+        $aggregator->aggregate(row(str_entry('data', 'a')));
+        $aggregator->aggregate(row(int_entry('data', 1)));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'c')));
+
+        self::assertSame(
+            'c, b, a',
+            $aggregator->result()->value()
+        );
+    }
+
+    public function test_string_agg_with_alias() : void
+    {
+        $aggregator = string_agg(ref('data')->as('string'));
+
+        $aggregator->aggregate(row(str_entry('data', 'a')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'c')));
+
+        self::assertSame(
+            'a, b, b, c',
+            $aggregator->result()->value()
+        );
+        self::assertSame('string', $aggregator->result()->name());
+    }
+
+    public function test_string_agg_with_order() : void
+    {
+        $aggregator = string_agg(ref('data'), sort: SortOrder::DESC);
+
+        $aggregator->aggregate(row(str_entry('data', 'a')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'b')));
+        $aggregator->aggregate(row(str_entry('data', 'c')));
+
+        self::assertSame(
+            'c, b, b, a',
+            $aggregator->result()->value()
+        );
+    }
+}
diff --git a/web/landing/resources/dsl.json b/web/landing/resources/dsl.json
diff --git a/web/landing/templates/documentation/dsl.html.twig b/web/landing/templates/documentation/dsl.html.twig
@@ -54,7 +54,7 @@
                 {% endfor %}
             </ul>
         </div>
-        <div class="order-3 lg:order-1 lg:col-span-4 p-2">
+        <div class="order-3 lg:order-1 lg:col-span-4 py-2 px-4">
             <hr class="text-blue-100 my-4 border-t-2 rounded" />
             <p class="mb-2 text-sm italic">
                 DSL stands for Domain Specific Language. In the case of Flow, the DSL is used to define simple functions that can be used to transform data.
diff --git a/web/landing/templates/documentation/example.html.twig b/web/landing/templates/documentation/example.html.twig
@@ -17,7 +17,7 @@
         <div class="order-1 lg:order-0 lg:col-span-1 lg:sticky lg:top-0 lg:max-h-screen lg:overflow-y-auto">
             {{ render(controller('Flow\\Website\\Controller\\DocumentationController::navigation')) }}
         </div>
-        <div class="order-3 lg:order-1 lg:col-span-4 p-2">
+        <div class="order-3 lg:order-1 lg:col-span-4 py-2 px-4">
             <div id="example">
                 <hr class="lg:hidden"/>
                 <h1>Example: {{ currentExample|humanize }}</h1>
diff --git a/web/landing/templates/documentation/page.html.twig b/web/landing/templates/documentation/page.html.twig
@@ -12,12 +12,12 @@
 {% endblock %}
 
 {% block main %}
-<section id="documentation-page" class="mt-5 mb-5 sm:px-4 mx-auto max-w-screen-xl">
+<section id="documentation-page" class="mt-5 mb-5 px-2 sm:px-4 mx-auto max-w-screen-xl">
     <div class="grid grid-cols-1 lg:grid-cols-6">
         <div class="order-1 lg:order-0 lg:col-span-1 lg:sticky lg:top-0 lg:max-h-screen lg:overflow-y-auto">
             {{ render(controller('Flow\\Website\\Controller\\DocumentationController::navigation')) }}
         </div>
-        <div class="order-3 lg:order-1 lg:col-span-4 p-2">
+        <div class="order-3 lg:order-1 lg:col-span-4 py-2 px-4">
             <hr class="lg:hidden"/>
             <section class="google-ad">
                 <script async src="https://pagead2.googlesyndication.com/pagead/js/adsbygoogle.js?client=ca-pub-2364202911639369" crossorigin="anonymous"></script>
