feat(flow-php/etl): schema-driven constructor-less entry instantiation

src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ParquetExtractor.php

@@ -21,8 +21,10 @@
 use Flow\Parquet\Reader;
 use Generator;
 
-use function Flow\ETL\DSL\array_to_row;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\row;
 use function Flow\ETL\DSL\rows;
+use function Flow\Types\DSL\type_string;
 
 final class ParquetExtractor implements Extractor, FileExtractor, LimitableExtractor
 {
@@ -42,6 +44,8 @@ final class ParquetExtractor implements Extractor, FileExtractor, LimitableExtra
 
     private SchemaConverter $schemaConverter;
 
+    private ValueHydrator $valueHydrator;
+
     /**
      * @param Path $path
      */
@@ -50,6 +54,7 @@ public function __construct(
     ) {
         $this->resetLimit();
         $this->schemaConverter = new SchemaConverter();
+        $this->valueHydrator = new ValueHydrator();
         $this->options = Options::default();
     }
 
@@ -79,16 +84,24 @@ public function extract(FlowContext $context): Generator
             }
 
             foreach ($fileData['file']->values($this->columns, $this->limit(), $fileOffset) as $row) {
+                $entries = [];
+
                 if ($shouldPutInputIntoRows) {
-                    $row['_input_file_uri'] = $uri;
+                    $entries[] = $context->entryFactory()->createAs('_input_file_uri', $uri, type_string());
+                }
+
+                // @mago-ignore analysis:mixed-assignment
+                foreach ($row as $entryName => $entryValue) {
+                    $definition = $flowSchema->get(ref($entryName));
+
+                    $entries[] = $context->entryFactory()->instantiate(
+                        $entryName,
+                        $this->valueHydrator->hydrate($entryValue, $definition),
+                        $definition,
+                    );
                 }
 
-                $signal = yield rows(array_to_row(
-                    $row,
-                    $context->entryFactory(),
-                    $fileData['stream']->path()->partitions(),
-                    $flowSchema,
-                ));
+                $signal = yield rows(row(...$entries));
 
                 $this->incrementReturnedRows();
 

src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/ValueHydrator.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\Parquet;
+
+use Flow\ETL\Schema\Definition;
+use Flow\Types\Type\Logical\JsonType;
+use Flow\Types\Type\Logical\OptionalType;
+use Flow\Types\Type\Logical\UuidType;
+
+final readonly class ValueHydrator
+{
+    public function __construct() {}
+
+    /**
+     * @param Definition<mixed> $definition
+     */
+    public function hydrate(mixed $value, Definition $definition): mixed
+    {
+        if ($value === null) {
+            return null;
+        }
+
+        $type = $definition->type();
+
+        if ($type instanceof OptionalType) {
+            $type = $type->base();
+        }
+
+        if ($type instanceof JsonType || $type instanceof UuidType) {
+            return $type->cast($value);
+        }
+
+        return $value;
+    }
+}

src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Unit/ValueHydratorTest.php

@@ -0,0 +1,112 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\Parquet\Tests\Unit;
+
+use DateInterval;
+use DateTimeImmutable;
+use Flow\ETL\Adapter\Parquet\ValueHydrator;
+use Flow\ETL\Tests\FlowTestCase;
+use Flow\Types\Value\Json;
+use Flow\Types\Value\Uuid;
+
+use function Flow\ETL\DSL\bool_schema;
+use function Flow\ETL\DSL\date_schema;
+use function Flow\ETL\DSL\datetime_schema;
+use function Flow\ETL\DSL\float_schema;
+use function Flow\ETL\DSL\int_schema;
+use function Flow\ETL\DSL\json_schema;
+use function Flow\ETL\DSL\list_schema;
+use function Flow\ETL\DSL\map_schema;
+use function Flow\ETL\DSL\string_schema;
+use function Flow\ETL\DSL\structure_schema;
+use function Flow\ETL\DSL\time_schema;
+use function Flow\ETL\DSL\uuid_schema;
+use function Flow\Types\DSL\type_float;
+use function Flow\Types\DSL\type_integer;
+use function Flow\Types\DSL\type_list;
+use function Flow\Types\DSL\type_map;
+use function Flow\Types\DSL\type_string;
+use function Flow\Types\DSL\type_structure;
+
+final class ValueHydratorTest extends FlowTestCase
+{
+    public function test_hydrating_already_hydrated_values_returns_them_as_is(): void
+    {
+        $json = new Json('{"a":1}');
+        $uuid = new Uuid('00000000-0000-0000-0000-000000000000');
+
+        static::assertSame($json, (new ValueHydrator())->hydrate($json, json_schema('json')));
+        static::assertSame($uuid, (new ValueHydrator())->hydrate($uuid, uuid_schema('uuid')));
+    }
+
+    public function test_hydrating_json_string_into_json_value(): void
+    {
+        static::assertEquals(
+            new Json('{"id":1,"name":"flow"}'),
+            (new ValueHydrator())->hydrate('{"id":1,"name":"flow"}', json_schema('json')),
+        );
+    }
+
+    public function test_hydrating_json_string_into_json_value_for_nullable_definition(): void
+    {
+        static::assertEquals(
+            new Json('{"id":1}'),
+            (new ValueHydrator())->hydrate('{"id":1}', json_schema('json', true)),
+        );
+    }
+
+    public function test_hydrating_null(): void
+    {
+        static::assertNull((new ValueHydrator())->hydrate(null, int_schema('int', true)));
+        static::assertNull((new ValueHydrator())->hydrate(null, json_schema('json', true)));
+        static::assertNull((new ValueHydrator())->hydrate(null, uuid_schema('uuid', true)));
+        static::assertNull((new ValueHydrator())->hydrate(null, int_schema('int')));
+    }
+
+    public function test_hydrating_uuid_string_into_uuid_value(): void
+    {
+        static::assertEquals(
+            new Uuid('f47ac10b-58cc-4372-a567-0e02b2c3d479'),
+            (new ValueHydrator())->hydrate('f47ac10b-58cc-4372-a567-0e02b2c3d479', uuid_schema('uuid')),
+        );
+    }
+
+    public function test_hydrating_uuid_string_into_uuid_value_for_nullable_definition(): void
+    {
+        static::assertEquals(
+            new Uuid('f47ac10b-58cc-4372-a567-0e02b2c3d479'),
+            (new ValueHydrator())->hydrate('f47ac10b-58cc-4372-a567-0e02b2c3d479', uuid_schema('uuid', true)),
+        );
+    }
+
+    public function test_values_of_types_without_flow_value_objects_are_passed_through_without_casting(): void
+    {
+        $datetime = new DateTimeImmutable('2024-04-01 10:00:00 UTC');
+        $time = new DateInterval('PT2H30M');
+        $list = [1, 2, 3];
+        $map = ['a' => 1, 'b' => 2];
+        $structure = ['lat' => 1.5, 'lon' => 2.5];
+
+        static::assertSame(1, (new ValueHydrator())->hydrate(1, int_schema('int')));
+        static::assertSame(1.5, (new ValueHydrator())->hydrate(1.5, float_schema('float')));
+        static::assertTrue((new ValueHydrator())->hydrate(true, bool_schema('bool')));
+        static::assertSame('flow', (new ValueHydrator())->hydrate('flow', string_schema('string')));
+        static::assertSame($datetime, (new ValueHydrator())->hydrate($datetime, datetime_schema('datetime')));
+        static::assertSame($datetime, (new ValueHydrator())->hydrate($datetime, date_schema('date')));
+        static::assertSame($time, (new ValueHydrator())->hydrate($time, time_schema('time')));
+        static::assertSame($list, (new ValueHydrator())->hydrate($list, list_schema(
+            'list',
+            type_list(type_integer()),
+        )));
+        static::assertSame($map, (new ValueHydrator())->hydrate($map, map_schema('map', type_map(
+            type_string(),
+            type_integer(),
+        ))));
+        static::assertSame($structure, (new ValueHydrator())->hydrate($structure, structure_schema('struct', type_structure([
+            'lat' => type_float(),
+            'lon' => type_float(),
+        ]))));
+    }
+}

src/core/etl/src/Flow/Floe/EntryFactory.php → src/core/etl/src/Flow/ETL/Row/Entry/EntryInstantiator.php

@@ -2,36 +2,43 @@
 
 declare(strict_types=1);
 
-namespace Flow\Floe;
+namespace Flow\ETL\Row\Entry;
 
 use Closure;
 use Flow\ETL\Row\Entry;
 use Flow\ETL\Schema\Definition;
 use ReflectionClass;
 
-final class EntryFactory
+/**
+ * @template-covariant T of Entry<mixed>
+ */
+final readonly class EntryInstantiator
 {
     /**
-     * @var \Closure(string, mixed, Definition<mixed>): Entry<mixed>
+     * @var \Closure(string, mixed, Definition<mixed>): T
      */
-    private readonly Closure $instantiate;
+    private Closure $instantiate;
 
     /**
-     * @param \Closure(string, mixed, Definition<mixed>): Entry<mixed> $instantiate
+     * @param \Closure(string, mixed, Definition<mixed>): T $instantiate
      */
     private function __construct(Closure $instantiate)
     {
         $this->instantiate = $instantiate;
     }
 
     /**
-     * @param class-string<Entry<mixed>> $entryClass
+     * @template TEntry of Entry<mixed>
+     *
+     * @param class-string<TEntry> $entryClass
+     *
+     * @return self<TEntry>
      */
-    public static function forEntryClass(string $entryClass): self
+    public static function forClass(string $entryClass): self
     {
         $reflection = new ReflectionClass($entryClass);
 
-        /** @var \Closure(string, mixed, Definition<mixed>): Entry<mixed> $instantiate */
+        /** @var \Closure(string, mixed, Definition<mixed>): TEntry $instantiate */
         $instantiate = Closure::bind(
             static function (string $name, mixed $value, Definition $definition) use ($reflection): Entry {
                 $entry = $reflection->newInstanceWithoutConstructor();
@@ -54,9 +61,9 @@ static function (string $name, mixed $value, Definition $definition) use ($refle
     /**
      * @param Definition<mixed> $definition
      *
-     * @return Entry<mixed>
+     * @return T
      */
-    public function create(string $name, mixed $value, Definition $definition): Entry
+    public function instantiate(string $name, mixed $value, Definition $definition): Entry
     {
         return ($this->instantiate)($name, $value, $definition);
     }

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

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Row\Entry;
+
+use Flow\ETL\Row\Entry;
+
+use function array_key_exists;
+
+final class Instantiators
+{
+    /**
+     * @var array<class-string<Entry<mixed>>, EntryInstantiator<Entry<mixed>>>
+     */
+    private array $instantiators = [];
+
+    /**
+     * @template T of Entry<mixed>
+     *
+     * @param class-string<T> $entryClass
+     *
+     * @return EntryInstantiator<T>
+     */
+    public function for(string $entryClass): EntryInstantiator
+    {
+        if (!array_key_exists($entryClass, $this->instantiators)) {
+            $this->instantiators[$entryClass] = EntryInstantiator::forClass($entryClass);
+        }
+
+        /** @var EntryInstantiator<T> */
+        return $this->instantiators[$entryClass];
+    }
+}

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

@@ -11,7 +11,6 @@
 use Flow\ETL\Schema\Definition\ListDefinition;
 use Flow\ETL\Schema\Metadata;
 use Flow\Types\Type\Logical\ListType;
-use Flow\Types\Type\TypeDetector;
 
 use function Flow\Types\DSL\type_equals;
 use function json_encode;
@@ -54,15 +53,6 @@ public function __construct(
 
         $this->value = $value;
 
-        // @mago-ignore analysis:redundant-type-comparison
-        if ($this->value !== null && !$type->isValid($this->value)) {
-            throw InvalidArgumentException::because(
-                'Expected ' . $type->toString() . ' got different types: ' . (new TypeDetector())
-                    ->detectType($this->value)
-                    ->toString(),
-            );
-        }
-
         $this->definition = new ListDefinition(
             $this->name,
             $type,

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

@@ -11,7 +11,6 @@
 use Flow\ETL\Schema\Definition\MapDefinition;
 use Flow\ETL\Schema\Metadata;
 use Flow\Types\Type\Logical\MapType;
-use Flow\Types\Type\TypeDetector;
 
 use function Flow\Types\DSL\type_equals;
 use function json_encode;
@@ -55,17 +54,6 @@ public function __construct(
 
         $this->value = $value;
 
-        // @mago-ignore analysis:redundant-type-comparison
-        if ($this->value !== null && !$type->isValid($this->value)) {
-            throw InvalidArgumentException::because(
-                'Expected ' . $type->toString() . ' got different types: '
-                    . (new TypeDetector())
-                        // @mago-ignore analysis:no-value
-                        ->detectType($this->value)
-                        ->toString(),
-            );
-        }
-
         $this->definition = new MapDefinition(
             $this->name,
             $type,

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

@@ -11,7 +11,6 @@
 use Flow\ETL\Schema\Definition\StructureDefinition;
 use Flow\ETL\Schema\Metadata;
 use Flow\Types\Type\Logical\StructureType;
-use Flow\Types\Type\TypeDetector;
 
 use function count;
 use function Flow\Types\DSL\type_equals;
@@ -60,17 +59,6 @@ public function __construct(
 
         $this->value = $value;
 
-        // @mago-ignore analysis:redundant-type-comparison
-        if ($this->value !== null && !$type->isValid($this->value)) {
-            throw InvalidArgumentException::because(
-                'Expected ' . $type->toString() . ' got different types: '
-                    . (new TypeDetector())
-                        // @mago-ignore analysis:no-value
-                        ->detectType($this->value)
-                        ->toString(),
-            );
-        }
-
         $this->definition = new StructureDefinition(
             $this->name,
             $type,