Added execution time to pipeline report (#1380)

Author: norberttech

composer.json

@@ -31,6 +31,7 @@
         "monolog/monolog": "^2.0||^3.0",
         "packaged/thrift": "^0.15.0",
         "php-http/discovery": "^1.0",
+        "psr/clock": "^1.0",
         "psr/http-client": "^1.0",
         "psr/http-message": "^1.0 || ^2.0",
         "psr/log": "^2.0 || ^3.0",

composer.lock

@@ -12,6 +12,7 @@
         "php": "~8.2.0 || ~8.3.0 || ~8.4.0",
         "ext-json": "*",
         "ext-mbstring": "*",
+        "psr/clock": "^1.0",
         "flow-php/array-dot": "^0.10.0 || 1.x-dev",
         "flow-php/rdsl": "^0.10.0 || 1.x-dev",
         "flow-php/filesystem": "^0.10.0 || 1.x-dev",

src/core/etl/composer.json

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Clock;
+
+use Psr\Clock\ClockInterface;
+
+final class FakeClock implements ClockInterface
+{
+    public function __construct(private \DateTimeImmutable $dateTime = new \DateTimeImmutable('now'))
+    {
+    }
+
+    public function modify(string $modify) : void
+    {
+        $this->dateTime = $this->dateTime->modify($modify);
+    }
+
+    public function now() : \DateTimeImmutable
+    {
+        return $this->dateTime;
+    }
+
+    public function set(\DateTimeImmutable $dateTime) : void
+    {
+        $this->dateTime = $dateTime;
+    }
+}

src/core/etl/src/Flow/Clock/FakeClock.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Clock;
+
+use Psr\Clock\ClockInterface;
+
+final readonly class SystemClock implements ClockInterface
+{
+    public function __construct(private \DateTimeZone $timezone)
+    {
+    }
+
+    public static function system() : self
+    {
+        return new self(new \DateTimeZone(date_default_timezone_get()));
+    }
+
+    public static function utc() : self
+    {
+        return new self(new \DateTimeZone('UTC'));
+    }
+
+    public function now() : \DateTimeImmutable
+    {
+        return new \DateTimeImmutable('now', $this->timezone);
+    }
+}

src/core/etl/src/Flow/Clock/SystemClock.php

@@ -13,6 +13,7 @@
 use Flow\ETL\Row\EntryFactory;
 use Flow\Filesystem\{FilesystemTable};
 use Flow\Serializer\Serializer;
+use Psr\Clock\ClockInterface;
 
 /**
  * Immutable configuration that can be used to initialize many contexts.
@@ -33,6 +34,7 @@
     public function __construct(
         private string $id,
         private Serializer $serializer,
+        private ClockInterface $clock,
         private FilesystemTable $filesystemTable,
         private FilesystemStreams $filesystemStreams,
         private Optimizer $optimizer,
@@ -59,6 +61,11 @@ public function caster() : Caster
         return $this->caster;
     }
 
+    public function clock() : ClockInterface
+    {
+        return $this->clock;
+    }
+
     public function entryFactory() : EntryFactory
     {
         return $this->entryFactory;

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

@@ -5,6 +5,7 @@
 namespace Flow\ETL\Config;
 
 use function Flow\Filesystem\DSL\fstab;
+use Flow\Clock\SystemClock;
 use Flow\ETL\Config\Cache\CacheConfigBuilder;
 use Flow\ETL\Config\Sort\SortConfigBuilder;
 use Flow\ETL\Filesystem\FilesystemStreams;
@@ -15,6 +16,7 @@
 use Flow\ETL\{Cache, Config, NativePHPRandomValueGenerator, RandomValueGenerator};
 use Flow\Filesystem\{Filesystem, FilesystemTable};
 use Flow\Serializer\{Base64Serializer, NativePHPSerializer, Serializer};
+use Psr\Clock\ClockInterface;
 
 final class ConfigBuilder
 {
@@ -24,6 +26,8 @@ final class ConfigBuilder
 
     private ?Caster $caster;
 
+    private ?ClockInterface $clock;
+
     private ?FilesystemTable $fstab;
 
     private ?string $id;
@@ -44,6 +48,7 @@ public function __construct()
         $this->putInputIntoRows = false;
         $this->optimizer = null;
         $this->caster = null;
+        $this->clock = null;
         $this->cache = new CacheConfigBuilder();
         $this->sort = new SortConfigBuilder();
         $this->randomValueGenerator = new NativePHPRandomValueGenerator();
@@ -54,7 +59,7 @@ public function build() : Config
         $this->id ??= 'flow_php' . $this->randomValueGenerator->string(32);
         $entryFactory = new NativeEntryFactory();
         $this->serializer ??= new Base64Serializer(new NativePHPSerializer());
-
+        $this->clock ??= SystemClock::utc();
         $this->optimizer ??= new Optimizer(
             new Optimizer\LimitOptimization(),
             new Optimizer\BatchSizeOptimization(batchSize: 1000)
@@ -65,6 +70,7 @@ public function build() : Config
         return new Config(
             $this->id,
             $this->serializer,
+            $this->clock,
             $this->fstab(),
             new FilesystemStreams($this->fstab()),
             $this->optimizer,
@@ -83,6 +89,13 @@ public function cache(Cache $cache) : self
         return $this;
     }
 
+    public function clock(ClockInterface $clocks) : self
+    {
+        $this->clock = $clocks;
+
+        return $this;
+    }
+
     public function dontPutInputIntoRows() : self
     {
         $this->putInputIntoRows = false;

src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php

@@ -745,7 +745,7 @@ public function rows(Transformer|Transformation $transformer) : self
     /**
      * @trigger
      *
-     * @param null|callable(Rows $rows): void $callback
+     * @param null|callable(Rows $rows, FlowContext $context): void $callback
      * @param bool $analyze - when set to true, run will return Report
      */
     #[DSLMethod(exclude: true)]
@@ -756,9 +756,13 @@ public function run(?callable $callback = null, bool $analyze = false) : ?Report
         $totalRows = 0;
         $schema = new Schema();
 
+        if ($analyze) {
+            $startedAt = $this->context->config->clock()->now();
+        }
+
         foreach ($clone->pipeline->process($clone->context) as $rows) {
             if ($callback !== null) {
-                $callback($rows);
+                $callback($rows, $clone->context);
             }
 
             if ($analyze) {
@@ -768,7 +772,9 @@ public function run(?callable $callback = null, bool $analyze = false) : ?Report
         }
 
         if ($analyze) {
-            return new Report($schema, new Statistics($totalRows));
+            $endedAt = $this->context->config->clock()->now();
+
+            return new Report($schema, new Statistics($totalRows, new Statistics\ExecutionTime($startedAt, $endedAt)));
         }
 
         return null;

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

@@ -4,10 +4,13 @@
 
 namespace Flow\ETL\Dataset;
 
+use Flow\ETL\Dataset\Statistics\ExecutionTime;
+
 final readonly class Statistics
 {
     public function __construct(
         private int $totalRows,
+        public readonly ExecutionTime $executionTime,
     ) {
     }
 

src/core/etl/src/Flow/ETL/Dataset/Statistics.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Dataset\Statistics;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+
+final readonly class ExecutionTime
+{
+    public function __construct(public \DateTimeImmutable $startedAt, public \DateTimeImmutable $finishedAt)
+    {
+        if ($startedAt > $finishedAt) {
+            throw new InvalidArgumentException('Execution start date must be before finish date');
+        }
+    }
+
+    public function duration() : \DateInterval
+    {
+        return $this->startedAt->diff($this->finishedAt);
+    }
+
+    public function inSeconds() : int
+    {
+        return $this->finishedAt->getTimestamp() - $this->startedAt->getTimestamp();
+    }
+}