EntryReference expressions (#362)

* First draft of reference expressions

* Allow to chain EntryReference expressions

* Simplified usage of Literal values

* Fixed example

* CS Fixes

Author: norberttech

examples/topics/transformations/filter.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+use function Flow\ETL\DSL\lit;
+use function Flow\ETL\DSL\ref;
+use Flow\ETL\DSL\Entry;
+use Flow\ETL\DSL\From;
+use Flow\ETL\DSL\To;
+use Flow\ETL\Flow;
+use Flow\ETL\Row;
+use Flow\ETL\Rows;
+
+require __DIR__ . '/../../bootstrap.php';
+
+(new Flow())
+    ->read(
+        From::rows(new Rows(
+            Row::with(Entry::int('a', 100), Entry::int('b', 100)),
+            Row::with(Entry::int('a', 100), Entry::int('b', 200))
+        ))
+    )
+    ->filter(ref('b')->divide(lit(2))->equals('a'))
+    ->withEntry('new_b', ref('b')->multiply(lit(2))->multiply(lit(5)))
+    ->write(To::output(false))
+    ->run();
+
+(new Flow())
+    ->read(
+        From::rows(new Rows(
+            Row::with(Entry::int('a', 4), Entry::int('b', 5)),
+            Row::with(Entry::int('a', 3), Entry::int('b', 6))
+        ))
+    )
+    ->filter(ref('b')->mod(lit(2))->equals(lit(0)))
+    ->write(To::output(false))
+    ->run();

examples/topics/transformations/literals.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+use function Flow\ETL\DSL\lit;
+use Flow\ETL\DSL\Entry;
+use Flow\ETL\DSL\From;
+use Flow\ETL\DSL\To;
+use Flow\ETL\Flow;
+use Flow\ETL\Row;
+use Flow\ETL\Rows;
+
+require __DIR__ . '/../../bootstrap.php';
+
+(new Flow())
+    ->read(
+        From::rows(new Rows(
+            Row::create(Entry::string('name', 'Norbert'))
+        ))
+    )
+    ->withEntry('number', lit(1))
+    ->write(To::output(false))
+    ->run();

examples/topics/transformations/math.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+use function Flow\ETL\DSL\ref;
+use Flow\ETL\DSL\Entry;
+use Flow\ETL\DSL\From;
+use Flow\ETL\DSL\To;
+use Flow\ETL\Flow;
+use Flow\ETL\Row;
+use Flow\ETL\Rows;
+
+require __DIR__ . '/../../bootstrap.php';
+
+(new Flow())
+    ->read(
+        From::rows(new Rows(
+            Row::create(Entry::integer('a', 100), Entry::integer('b', 200))
+        ))
+    )
+    ->write(To::output(false))
+    ->withEntry('c', ref('a')->plus('b'))
+    ->withEntry('d', ref('b')->minus('a'))
+    ->write(To::output(false))
+    ->run();

src/adapter/etl-adapter-avro/src/Flow/ETL/Adapter/Avro/FlixTech/AvroLoader.php

@@ -65,7 +65,7 @@ public function destination() : Path
 
     public function load(Rows $rows, FlowContext $context) : void
     {
-        if (\count($context->partitionEntries())) {
+        if ($context->partitionEntries()->count()) {
             throw new RuntimeException('Partitioning is not supported yet');
         }
 

src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVLoader.php

@@ -83,8 +83,8 @@ public function load(Rows $rows, FlowContext $context) : void
 
         $headers = $rows->first()->entries()->map(fn (Entry $entry) => $entry->name());
 
-        if (\count($context->partitionEntries())) {
-            foreach ($rows->partitionBy(...$context->partitionEntries()) as $partition) {
+        if ($context->partitionEntries()->count()) {
+            foreach ($rows->partitionBy(...$context->partitionEntries()->all()) as $partition) {
                 $this->write($partition->rows, $headers, $context, $partition->partitions);
             }
         } else {

src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JsonLoader.php

@@ -59,8 +59,8 @@ public function destination() : Path
 
     public function load(Rows $rows, FlowContext $context) : void
     {
-        if (\count($context->partitionEntries())) {
-            foreach ($rows->partitionBy(...$context->partitionEntries()) as $partitionedRows) {
+        if ($context->partitionEntries()->count()) {
+            foreach ($rows->partitionBy(...$context->partitionEntries()->all()) as $partitionedRows) {
                 $this->write($partitionedRows->rows, $partitionedRows->partitions, $context);
             }
         } else {

src/adapter/etl-adapter-text/src/Flow/ETL/Adapter/Text/TextLoader.php

@@ -58,8 +58,8 @@ public function destination() : Path
 
     public function load(Rows $rows, FlowContext $context) : void
     {
-        if (\count($context->partitionEntries())) {
-            foreach ($rows->partitionBy(...$context->partitionEntries()) as $partition) {
+        if ($context->partitionEntries()->count()) {
+            foreach ($rows->partitionBy(...$context->partitionEntries()->all()) as $partition) {
                 foreach ($partition->rows as $row) {
                     if ($row->entries()->count() > 1) {
                         throw new RuntimeException(\sprintf('Text data loader supports only a single entry rows, and you have %d rows.', $row->entries()->count()));

src/core/etl/README.md

@@ -447,9 +447,15 @@ $flow->process(new Rows(...))
     ->run();
 ```
 
-## Filter
+## Filtering
 
-In order to quickly filter Rows `DataFrame::filter` shortcut function can be used. 
+There are two ways to filter datasets in Flow. 
+
+### Callback Filtering 
+
+This is the most flexible but also the least developer friendly apporach. 
+All you need to do is use callback `fn (Row $row) : bool` that will be executed against each row
+and that will tell filter to keep or skip the row. 
 
 ```php 
 <?php 
@@ -462,6 +468,32 @@ $flow->process(new Rows(...))
     ->run();
 ```
 
+### DSL Filtering 
+
+This approach is way more developer friendly as IDE can autocomplete all filtering functions for you. 
+
+```php
+<?php
+
+(new Flow())
+    ->read(
+        From::rows(new Rows(
+            Row::with(Entry::int('a', 100), Entry::int('b', 100)),
+            Row::with(Entry::int('a', 100), Entry::int('b', 200))
+        ))
+    )
+    ->filter(ref('b')->divide(lit(2))->equals('a'))
+    ->withEntry('new_b', ref('b')->multiply(lit(2))->multiply(lit(5)))
+    ->write(To::output(false))
+    ->run();
+```
+
+So in general all filtering functions are available as [Reference Expression](/src/core/etl/src/Flow/ETL/Row/Reference/Expression.php).
+This means you can chain them creating readable conditions that will significantly improve maintainability of your 
+data processing pipelines. 
+
+All possible expressions are available through [EntryReference.php](/src/core/etl/src/Flow/ETL/Row/EntryReference.php).
+
 ## Group By
 
 Flow allows grouping that is similar to the one known from database engines.

src/core/etl/src/Flow/ETL/Async/Socket/Communication/Message.php

@@ -8,7 +8,7 @@
 use Flow\ETL\FlowContext;
 use Flow\ETL\Partition\PartitionFilter;
 use Flow\ETL\Pipeline\Pipes;
-use Flow\ETL\Row\EntryReference;
+use Flow\ETL\Row\References;
 use Flow\ETL\Rows;
 use Flow\Serializer\Serializable;
 use Flow\Serializer\Serializer;
@@ -22,7 +22,7 @@
  *      cache?: Cache,
  *      cache_id?: string,
  *      rows?: Rows,
- *      partition_entries?: array<EntryReference>,
+ *      partition_entries?: References,
  *      partition_filter?: PartitionFilter
  *     }
  * }>
@@ -37,7 +37,7 @@ final class Message implements Serializable
      *     cache?: Cache,
      *     cache_id?: string,
      *     rows?: Rows,
-     *     partition_entries?: array<EntryReference>,
+     *     partition_entries?: References,
      *     partition_filter?: PartitionFilter
      * } $payload
      */
@@ -110,7 +110,7 @@ public static function setup(Pipes $pipes, FlowContext $context, string $cacheId
      *     cache?: Cache,
      *     cache_id?: string,
      *     rows?: Rows,
-     *     partition_entries?: array<EntryReference>,
+     *     partition_entries?: References,
      *     partition_filter?: PartitionFilter
      * }
      */

src/core/etl/src/Flow/ETL/Async/Socket/Worker/ClientProtocol.php

@@ -26,19 +26,22 @@ public function __construct(private readonly Processor $processor)
 
     public function handle(string $id, Message $message, Server $server) : void
     {
+        $payload = $message->payload();
+
         switch ($message->type()) {
             case Protocol::SERVER_SETUP:
-                $this->processor->setPipes($message->payload()['pipes'] ?? new Pipes([]));
-                $this->processor->setPartitionEntries($message->payload()['partition_entries'] ?? []);
-                $this->processor->setPartitionFilter($message->payload()['partition_filter'] ?? new NoopFilter());
-                $this->cache = $message->payload()['cache'] ?? $this->cache;
-                $this->cacheId = $message->payload()['cache_id'] ?? $this->cacheId;
+                $refs = \array_key_exists('partition_entries', $payload) ? $payload['partition_entries']->all() : [];
+                $this->processor->setPipes($payload['pipes'] ?? new Pipes([]));
+                $this->processor->setPartitionEntries($refs);
+                $this->processor->setPartitionFilter($payload['partition_filter'] ?? new NoopFilter());
+                $this->cache = $payload['cache'] ?? $this->cache;
+                $this->cacheId = $payload['cache_id'] ?? $this->cacheId;
 
                 $server->send(Message::fetch($id));
 
                 break;
             case Protocol::SERVER_PROCESS:
-                $rows = $this->processor->process($message->payload()['rows'] ?? new Rows());
+                $rows = $this->processor->process($payload['rows'] ?? new Rows());
 
                 $server->send(Message::fetch($id));
                 $this->cache->add($this->cacheId, $rows);