fix: autocast handling datetime | numeric type merge (#2196)

- renamed Stages to Segments

Author: norberttech

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

@@ -4,18 +4,18 @@
 
 namespace Flow\ETL;
 
-use Flow\ETL\Pipeline\Stages;
+use Flow\ETL\Pipeline\Segments;
 
 /**
  * @internal
  */
 final readonly class Pipeline
 {
-    private Stages $stages;
+    private Segments $stages;
 
     public function __construct(private Extractor $extractor)
     {
-        $this->stages = new Stages();
+        $this->stages = new Segments();
     }
 
     public function add(Transformer|Loader|Processor $step) : self
@@ -68,7 +68,7 @@ public function process(FlowContext $context) : \Generator
     /**
      * Get the pipeline stages.
      */
-    public function stages() : Stages
+    public function stages() : Segments
     {
         return $this->stages;
     }

…ore/etl/src/Flow/ETL/Pipeline/Stages.php …e/etl/src/Flow/ETL/Pipeline/Segments.phpsrc/core/etl/src/Flow/ETL/Pipeline/Stages.php renamed to src/core/etl/src/Flow/ETL/Pipeline/Segments.php src/core/etl/src/Flow/ETL/Pipeline/Stages.php renamed to src/core/etl/src/Flow/ETL/Pipeline/Segments.php

@@ -11,7 +11,7 @@
  *
  * @internal
  */
-final class Stages
+final class Segments
 {
     private Segment $currentSegment;
 

src/core/etl/src/Flow/ETL/Schema/Definition/DateDefinition.php

@@ -154,6 +154,22 @@ public function merge(Definition $definition) : Definition
             );
         }
 
+        if ($definition instanceof FloatDefinition) {
+            return new FloatDefinition(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
+        if ($definition instanceof IntegerDefinition) {
+            return new IntegerDefinition(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
         throw new RuntimeException(\sprintf(
             'Cannot merge %s with %s',
             self::class,

src/core/etl/src/Flow/ETL/Schema/Definition/DateTimeDefinition.php

@@ -146,6 +146,22 @@ public function merge(Definition $definition) : Definition
             );
         }
 
+        if ($definition instanceof FloatDefinition) {
+            return new FloatDefinition(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
+        if ($definition instanceof IntegerDefinition) {
+            return new IntegerDefinition(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
         throw new RuntimeException(\sprintf(
             'Cannot merge %s with %s',
             self::class,

src/core/etl/src/Flow/ETL/Schema/Definition/FloatDefinition.php

@@ -146,6 +146,14 @@ public function merge(Definition $definition) : Definition
             );
         }
 
+        if ($definition instanceof DateDefinition || $definition instanceof DateTimeDefinition) {
+            return new self(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
         if ($definition instanceof StringDefinition) {
             return new StringDefinition(
                 $this->ref,

src/core/etl/src/Flow/ETL/Schema/Definition/IntegerDefinition.php

@@ -146,6 +146,14 @@ public function merge(Definition $definition) : Definition
             );
         }
 
+        if ($definition instanceof DateDefinition || $definition instanceof DateTimeDefinition) {
+            return new self(
+                $this->ref,
+                $this->nullable || $definition->isNullable(),
+                $this->metadata->merge($definition->metadata())
+            );
+        }
+
         if ($definition instanceof StringDefinition) {
             return new StringDefinition(
                 $this->ref,