Allow to duplicate row (#1541)

* Allow to duplicate rows based on condition, with applying additional transformations on them

* Updated dsl definitions

* Simplified assertions in TransformTest

Author: norberttech

src/core/etl/src/Flow/ETL/DSL/functions.php

@@ -131,7 +131,8 @@
     Rows,
     Transformation,
     Transformer,
-    Window};
+    Window,
+    WithEntry};
 use Flow\Filesystem\Stream\Mode;
 use Flow\Filesystem\{Filesystem, Local\NativeLocalFilesystem, Partition, Partitions, Path};
 use Flow\Serializer\{NativePHPSerializer, Serializer};
@@ -1698,3 +1699,9 @@ function caster(?Options $options = null) : Caster
 {
     return Caster::default($options ?? caster_options());
 }
+
+#[DocumentationDSL(module: Module::CORE, type: DSLType::HELPER)]
+function with_entry(string $name, ScalarFunction $function) : WithEntry
+{
+    return new WithEntry($name, $function);
+}

src/core/etl/src/Flow/ETL/DataFrame.php

@@ -26,13 +26,13 @@
     SortingPipeline,
     VoidPipeline};
 use Flow\ETL\Row\{Reference, References, Schema, Schema\Definition};
-use Flow\ETL\Transformer\{
-    AutoCastTransformer,
+use Flow\ETL\Transformer\{AutoCastTransformer,
     CallbackRowTransformer,
     CrossJoinRowsTransformer,
     DropDuplicatesTransformer,
     DropEntriesTransformer,
     DropPartitionsTransformer,
+    DuplicateRowTransformer,
     EntryNameStyleConverterTransformer,
     JoinEachRowsTransformer,
     LimitTransformer,
@@ -46,8 +46,7 @@
     ScalarFunctionTransformer,
     SelectEntriesTransformer,
     UntilTransformer,
-    WindowFunctionTransformer
-};
+    WindowFunctionTransformer};
 use Flow\Filesystem\Path\Filter;
 
 final class DataFrame
@@ -259,6 +258,13 @@ public function dropPartitions(bool $dropPartitionColumns = false) : self
         return $this;
     }
 
+    public function duplicateRow(mixed $condition, WithEntry ...$entries) : self
+    {
+        $this->pipeline->add(new DuplicateRowTransformer($condition, ...$entries));
+
+        return $this;
+    }
+
     /**
      * Be aware that fetch is not memory safe and will load all rows into memory.
      * If you want to safely iterate over Rows use oe of the following methods:.
@@ -776,7 +782,7 @@ public function sortBy(Reference ...$entries) : self
      *
      * @lazy
      */
-    public function transform(Transformer|Transformation|Transformations $transformer) : self
+    public function transform(Transformer|Transformation|Transformations|WithEntry $transformer) : self
     {
         return $this->with($transformer);
     }
@@ -823,7 +829,7 @@ public function void() : self
     /**
      * @lazy
      */
-    public function with(Transformer|Transformation|Transformations $transformer) : self
+    public function with(Transformer|Transformation|Transformations|WithEntry $transformer) : self
     {
         if ($transformer instanceof Transformer) {
             $this->pipeline->add($transformer);
@@ -837,18 +843,28 @@ public function with(Transformer|Transformation|Transformations $transformer) :
             return $this;
         }
 
+        if ($transformer instanceof WithEntry) {
+            $this->withEntry($transformer->name, $transformer->function);
+
+            return $this;
+        }
+
         return $transformer->transform($this);
     }
 
     /**
      * @lazy
      *
-     * @param array<string, ScalarFunction|WindowFunction> $references
+     * @param array<string, ScalarFunction|WindowFunction|WithEntry> $references
      */
     public function withEntries(array $references) : self
     {
         foreach ($references as $entryName => $ref) {
-            $this->withEntry($entryName, $ref);
+            if ($ref instanceof WithEntry) {
+                $this->withEntry($ref->name, $ref->function);
+            } else {
+                $this->withEntry($entryName, $ref);
+            }
         }
 
         return $this;

src/core/etl/src/Flow/ETL/PHP/Type/Native/BooleanType.php

@@ -64,7 +64,7 @@ public function isValid(mixed $value) : bool
         return \is_bool($value);
     }
 
-    public function makeNullable(bool $nullable) : Type
+    public function makeNullable(bool $nullable) : self
     {
         return new self($nullable);
     }

src/core/etl/src/Flow/ETL/PHP/Type/Native/IntegerType.php

@@ -69,7 +69,7 @@ public function isValid(mixed $value) : bool
         return \is_int($value);
     }
 
-    public function makeNullable(bool $nullable) : Type
+    public function makeNullable(bool $nullable) : self
     {
         return new self($nullable);
     }

src/core/etl/src/Flow/ETL/PHP/Type/Native/StringType.php

@@ -64,7 +64,7 @@ public function isValid(mixed $value) : bool
         return \is_string($value);
     }
 
-    public function makeNullable(bool $nullable) : Type
+    public function makeNullable(bool $nullable) : self
     {
         return new self($nullable);
     }

src/core/etl/src/Flow/ETL/Row.php

@@ -42,6 +42,11 @@ public function add(Entry ...$entries) : self
         return new self($this->entries->add(...$entries));
     }
 
+    public function duplicate() : self
+    {
+        return new self($this->entries()->duplicate());
+    }
+
     public function entries() : Entries
     {
         return $this->entries;

src/core/etl/src/Flow/ETL/Row/Entries.php

@@ -78,6 +78,17 @@ public function count() : int
         return \count($this->entries);
     }
 
+    public function duplicate() : self
+    {
+        $entries = [];
+
+        foreach ($this->all() as $entry) {
+            $entries[$entry->name()] = $entry->duplicate();
+        }
+
+        return self::recreate($entries);
+    }
+
     /**
      * @throws InvalidArgumentException
      *

src/core/etl/src/Flow/ETL/Row/Entry.php

@@ -17,6 +17,11 @@ public function __toString() : string;
 
     public function definition() : Definition;
 
+    /**
+     * @return Entry<TValue, TType>
+     */
+    public function duplicate() : self;
+
     public function is(string|Reference $name) : bool;
 
     /**

src/core/etl/src/Flow/ETL/Row/Entry/BooleanEntry.php

@@ -20,10 +20,7 @@ final class BooleanEntry implements Entry
 
     private Metadata $metadata;
 
-    /**
-     * @var Type<?bool>
-     */
-    private readonly Type $type;
+    private readonly BooleanType $type;
 
     /**
      * @throws InvalidArgumentException
@@ -49,6 +46,11 @@ public function definition() : Definition
         return new Definition($this->name, $this->type, $this->metadata);
     }
 
+    public function duplicate() : Entry
+    {
+        return new self($this->name, $this->value, $this->type, $this->metadata);
+    }
+
     public function is(string|Reference $name) : bool
     {
         if ($name instanceof Reference) {

src/core/etl/src/Flow/ETL/Row/Entry/DateEntry.php

@@ -62,6 +62,11 @@ public function definition() : Definition
         return new Definition($this->name, $this->type, $this->metadata);
     }
 
+    public function duplicate() : Entry
+    {
+        return new self($this->name, $this->value ? clone $this->value : null, $this->type, $this->metadata);
+    }
+
     public function is(string|Reference $name) : bool
     {
         if ($name instanceof Reference) {