diff --git a/.github/workflows/job-benchmark-tests.yml b/.github/workflows/job-benchmark-tests.yml
index 17c9fdd71..dbebd8a51 100644
--- a/.github/workflows/job-benchmark-tests.yml
+++ b/.github/workflows/job-benchmark-tests.yml
@@ -55,34 +55,11 @@ jobs:
       - name: "Execute benchmarks"
         id: init_comment
         run: |
-          # Run all benchmarks in parallel
-          composer test:benchmark:extractor -- --ref=1.x --progress=none --iterations=1 > ./var/phpbench/extractor.txt 2>&1 &
-          PID_EXTRACTOR=$!
-
-          composer test:benchmark:transformer -- --ref=1.x --progress=none --iterations=1 > ./var/phpbench/transformer.txt 2>&1 &
-          PID_TRANSFORMER=$!
-
-          composer test:benchmark:loader -- --ref=1.x --progress=none --iterations=1 > ./var/phpbench/loader.txt 2>&1 &
-          PID_LOADER=$!
-
-          composer test:benchmark:building_blocks -- --ref=1.x --progress=none --iterations=1 > ./var/phpbench/building_blocks.txt 2>&1 &
-          PID_BUILDING=$!
-
-          composer test:benchmark:parquet-library -- --ref=1.x --progress=none --iterations=1 > ./var/phpbench/parquet.txt 2>&1 &
-          PID_PARQUET=$!
-
-          # Wait for all to complete and capture exit codes
-          EXIT_CODE=0
-          wait $PID_EXTRACTOR || EXIT_CODE=$?
-          wait $PID_TRANSFORMER || EXIT_CODE=$?
-          wait $PID_LOADER || EXIT_CODE=$?
-          wait $PID_BUILDING || EXIT_CODE=$?
-          wait $PID_PARQUET || EXIT_CODE=$?
-
-          if [ $EXIT_CODE -ne 0 ]; then
-            echo "One or more benchmarks failed with exit code $EXIT_CODE"
-            exit $EXIT_CODE
-          fi
+          composer test:benchmark:extractor -- --ref=1.x --progress=none > ./var/phpbench/extractor.txt
+          composer test:benchmark:transformer -- --ref=1.x --progress=none > ./var/phpbench/transformer.txt
+          composer test:benchmark:loader -- --ref=1.x --progress=none > ./var/phpbench/loader.txt
+          composer test:benchmark:building_blocks -- --ref=1.x --progress=none > ./var/phpbench/building_blocks.txt
+          composer test:benchmark:parquet-library -- --ref=1.x --progress=none > ./var/phpbench/parquet.txt
 
           # Build the summary file
           {
diff --git a/documentation/components/adapters/doctrine.md b/documentation/components/adapters/doctrine.md
index d64a5aee6..929d01e31 100644
--- a/documentation/components/adapters/doctrine.md
+++ b/documentation/components/adapters/doctrine.md
@@ -80,33 +80,13 @@ data_frame()
 // Types are automatically detected from the Flow Schema
 ```
 
-#### Manual Type Override
+#### Custom Types Map
 
-You can override specific column types for fine-grained control:
+For advanced scenarios, you can provide a custom type mapping to control how Flow types are converted to DBAL types:
 
 ```php
 use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
-use Doctrine\DBAL\Types\Type;
-use Doctrine\DBAL\Types\Types;
-
-data_frame()
-    ->read(from_())
-    ->write(to_dbal_table_insert($connection, 'users')
-        ->withColumnTypes([
-            'id' => Type::getType(Types::INTEGER),
-            'email' => Type::getType(Types::STRING),
-            'created_at' => Type::getType(Types::DATETIME_IMMUTABLE),
-        ]))
-    ->run();
-```
-
-#### Custom Type Detector
-
-For advanced scenarios, you can provide a custom type detector with your own type mapping:
-
-```php
-use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
-use Flow\ETL\Adapter\Doctrine\{DbalTypesDetector, TypesMap};
+use Flow\ETL\Adapter\Doctrine\TypesMap;
 use Flow\Types\Type\Native\StringType;
 use Doctrine\DBAL\Types\TextType;
 
@@ -117,7 +97,7 @@ $customTypesMap = new TypesMap([
 data_frame()
     ->read(from_())
     ->write(to_dbal_table_insert($connection, 'users')
-        ->withTypesDetector(new DbalTypesDetector($customTypesMap)))
+        ->withTypesMap($customTypesMap))
     ->run();
 ```
 
diff --git a/phpbench.json.dist b/phpbench.json.dist
index c9b4a5b88..6ef59153e 100644
--- a/phpbench.json.dist
+++ b/phpbench.json.dist
@@ -27,11 +27,11 @@
     "src/core/etl/tests/Flow/ETL/Tests/Benchmark/",
     "src/lib/parquet/tests/Flow/Parquet/Tests/Benchmark/"
   ],
-  "runner.php_config": { "memory_limit": "1G" },
+  "runner.php_config": { "memory_limit": "2G" },
   "runner.php_env": {
     "PGSQL_DATABASE_URL": "pgsql://postgres:postgres@127.0.0.1:5432/postgres?serverVersion=11&charset=utf8"
   },
-  "runner.iterations": 3,
-  "runner.retry_threshold": 5,
+  "runner.iterations": 2,
+  "runner.retry_threshold": 10,
   "storage.xml_storage_path": "var/phpbench"
 }
diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Benchmark/CSVLoaderBench.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Benchmark/CSVLoaderBench.php
index e9884eac1..e2fe04354 100644
--- a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Benchmark/CSVLoaderBench.php
+++ b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Benchmark/CSVLoaderBench.php
@@ -4,9 +4,9 @@
 
 namespace Flow\ETL\Adapter\CSV\Tests\Benchmark;
 
-use function Flow\ETL\Adapter\CSV\{from_csv, to_csv};
+use function Flow\ETL\Adapter\CSV\to_csv;
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\{FlowContext, Rows};
+use Flow\ETL\{FlowContext, Rows, Tests\Double\FakeStaticOrdersExtractor};
 use PhpBench\Attributes\Groups;
 
 #[Groups(['loader'])]
@@ -22,11 +22,7 @@ public function __construct()
     {
         $this->context = flow_context(config());
         $this->outputPath = \tempnam(\sys_get_temp_dir(), 'etl_csv_loader_bench') . '.csv';
-        $this->rows = \Flow\ETL\DSL\rows();
-
-        foreach (from_csv(__DIR__ . '/Fixtures/orders_flow.csv')->extract($this->context) as $rows) {
-            $this->rows = $this->rows->merge($rows);
-        }
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
     }
 
     public function __destruct()
diff --git a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalLoader.php b/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalLoader.php
index e437fc9bf..dcb4095fd 100644
--- a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalLoader.php
+++ b/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalLoader.php
@@ -5,7 +5,6 @@
 namespace Flow\ETL\Adapter\Doctrine;
 
 use Doctrine\DBAL\{Connection, DriverManager};
-use Doctrine\DBAL\Types\Type;
 use Flow\Doctrine\Bulk\{Bulk, BulkData, InsertOptions, UpdateOptions};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\{FlowContext, Loader, Rows};
@@ -14,18 +13,13 @@ final class DbalLoader implements Loader
 {
     private ?Bulk $bulk = null;
 
-    /**
-     * @var null|array<string, Type>
-     */
-    private ?array $columnTypes = null;
-
     private ?Connection $connection = null;
 
     private string $operation = 'insert';
 
     private InsertOptions|UpdateOptions|null $operationOptions = null;
 
-    private ?DbalTypesDetector $typesDetector = null;
+    private ?TypesMap $typesMap = null;
 
     /**
      * @param array<string, mixed> $connectionParams
@@ -65,28 +59,21 @@ public static function fromConnection(
 
     public function load(Rows $rows, FlowContext $context) : void
     {
-        $normalizedData = (new RowsNormalizer())->normalize($rows->sortEntries());
+        if ($rows->count() === 0) {
+            return;
+        }
+
+        $sortedRows = $rows->sortEntries();
+        $normalizedData = (new RowsNormalizer())->normalize($sortedRows);
 
         $this->bulk()->{$this->operation}(
             $this->connection(),
             $this->tableName,
-            new BulkData($normalizedData, $this->typesDetector()->convert($rows->schema(), $this->columnTypes ?? [])),
+            new BulkData($normalizedData, $this->typesMap()->flowRowTypes($sortedRows->first())),
             $this->operationOptions
         );
     }
 
-    /**
-     * Override types taken from Flow Schema with explicitly provided DBAL types.
-     *
-     * @param array<string, Type> $types Column name => DBAL Type instance
-     */
-    public function withColumnTypes(array $types) : self
-    {
-        $this->columnTypes = $types;
-
-        return $this;
-    }
-
     /**
      * @throws InvalidArgumentException
      */
@@ -109,11 +96,11 @@ public function withOperationOptions(InsertOptions|UpdateOptions|null $operation
     }
 
     /**
-     * Set custom SchemaToTypesConverter with custom TypesMap.
+     * Set custom types map for Flow Type to DBAL Type conversion.
      */
-    public function withTypesDetector(DbalTypesDetector $detector) : self
+    public function withTypesMap(TypesMap $typesMap) : self
     {
-        $this->typesDetector = $detector;
+        $this->typesMap = $typesMap;
 
         return $this;
     }
@@ -137,14 +124,8 @@ private function connection() : Connection
         return $this->connection;
     }
 
-    private function typesDetector() : DbalTypesDetector
+    private function typesMap() : TypesMap
     {
-        if ($this->typesDetector !== null) {
-            return $this->typesDetector;
-        }
-
-        $this->typesDetector = new DbalTypesDetector();
-
-        return $this->typesDetector;
+        return $this->typesMap ??= new TypesMap([]);
     }
 }
diff --git a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalTypesDetector.php b/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalTypesDetector.php
deleted file mode 100644
index 2893b5bb1..000000000
--- a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalTypesDetector.php
+++ /dev/null
@@ -1,89 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Flow\ETL\Adapter\Doctrine;
-
-use Doctrine\DBAL\Types\{BigIntType, BlobType, BooleanType, DateImmutableType, DateTimeImmutableType, DecimalType, FloatType, GuidType, IntegerType, JsonType, SmallIntType, StringType, TextType, TimeImmutableType, Type, Types};
-use Flow\ETL\Schema;
-use Flow\Types\Type as FlowType;
-use Flow\Types\Type\Logical\{ListType, MapType, StructureType};
-
-final readonly class DbalTypesDetector
-{
-    public function __construct(private TypesMap $typesMap = new TypesMap([]))
-    {
-    }
-
-    /**
-     * Converts Flow Schema to DBAL Types array.
-     * Detects types for columns not already provided in $typesOverride.
-     *
-     * @param Schema $schema
-     * @param array<string, Type> $typesOverride column types that take priority
-     *
-     * @return array<string, Type> Column name => DBAL Type instance
-     */
-    public function convert(Schema $schema, array $typesOverride = []) : array
-    {
-        $detectedTypes = [];
-
-        foreach ($schema->definitions() as $definition) {
-            $flowType = $definition->type();
-            $columnName = $definition->entry()->name();
-
-            if (array_key_exists($columnName, $typesOverride)) {
-                continue;
-            }
-
-            if ($this->isNestedType($flowType)) {
-                $detectedTypes[$columnName] = Type::getType(Types::JSON);
-
-                continue;
-            }
-
-            $dbalTypeClass = $this->typesMap->toDbalType($flowType::class);
-            $detectedTypes[$columnName] = Type::getType($this->getTypeConstant($dbalTypeClass));
-        }
-
-        return array_merge($detectedTypes, $typesOverride);
-    }
-
-    /**
-     * Maps DBAL type class to DBAL Types constant.
-     *
-     * @param class-string<Type> $dbalTypeClass
-     */
-    private function getTypeConstant(string $dbalTypeClass) : string
-    {
-        return match ($dbalTypeClass) {
-            StringType::class => Types::STRING,
-            IntegerType::class => Types::INTEGER,
-            FloatType::class => Types::FLOAT,
-            BooleanType::class => Types::BOOLEAN,
-            DateTimeImmutableType::class => Types::DATETIME_IMMUTABLE,
-            DateImmutableType::class => Types::DATE_IMMUTABLE,
-            TimeImmutableType::class => Types::TIME_IMMUTABLE,
-            GuidType::class => Types::GUID,
-            JsonType::class => Types::JSON,
-            TextType::class => Types::TEXT,
-            BigIntType::class => Types::BIGINT,
-            SmallIntType::class => Types::SMALLINT,
-            DecimalType::class => Types::DECIMAL,
-            BlobType::class => Types::BLOB,
-            default => throw new \InvalidArgumentException("Unsupported DBAL type class: {$dbalTypeClass}"),
-        };
-    }
-
-    /**
-     * Checks if a Flow type is a nested type (List, Map, Structure).
-     *
-     * @param FlowType<mixed> $flowType
-     */
-    private function isNestedType(FlowType $flowType) : bool
-    {
-        return $flowType instanceof ListType
-            || $flowType instanceof MapType
-            || $flowType instanceof StructureType;
-    }
-}
diff --git a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/TypesMap.php b/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/TypesMap.php
index 2a1dc6052..090ceebd7 100644
--- a/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/TypesMap.php
+++ b/src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/TypesMap.php
@@ -7,6 +7,7 @@
 use Doctrine\DBAL\Types\{BigIntType, BlobType, DateImmutableType, DateTimeImmutableType, DateTimeTzImmutableType, DateTimeTzType, DecimalType, GuidType, SmallFloatType, SmallIntType, TextType, TimeImmutableType};
 use Doctrine\DBAL\Types\Type as DbalType;
 use Flow\ETL\Exception\InvalidArgumentException;
+use Flow\ETL\Row;
 use Flow\Types\Type as FlowType;
 use Flow\Types\Type\Logical\{DateTimeType,
     DateType,
@@ -20,7 +21,7 @@
     UuidType,
     XMLElementType,
     XMLType};
-use Flow\Types\Type\Native\{BooleanType, FloatType, IntegerType, StringType};
+use Flow\Types\Type\Native\{BooleanType, EnumType, FloatType, IntegerType, StringType};
 
 final class TypesMap
 {
@@ -67,6 +68,7 @@ final class TypesMap
         XMLElementType::class => \Doctrine\DBAL\Types\StringType::class,
         HTMLType::class => \Doctrine\DBAL\Types\StringType::class,
         HTMLElementType::class => \Doctrine\DBAL\Types\StringType::class,
+        EnumType::class => \Doctrine\DBAL\Types\StringType::class,
         ListType::class => \Doctrine\DBAL\Types\JsonType::class,
         MapType::class => \Doctrine\DBAL\Types\JsonType::class,
         StructureType::class => \Doctrine\DBAL\Types\JsonType::class,
@@ -99,6 +101,29 @@ public function __construct(array $map)
         }
     }
 
+    /**
+     * Build DBAL types array from a row's entries.
+     *
+     * @return array<string, DbalType> Column name => DBAL Type instance
+     */
+    public function flowRowTypes(Row $row) : array
+    {
+        $types = [];
+        $typeClassToName = \array_flip(DbalType::getTypesMap());
+
+        foreach ($row->entries() as $entry) {
+            $dbalTypeClass = $this->toDbalType($entry->type()::class);
+
+            if (!\array_key_exists($dbalTypeClass, $typeClassToName)) {
+                throw new \InvalidArgumentException(\sprintf('DBAL type "%s" is not registered.', $dbalTypeClass));
+            }
+
+            $types[$entry->name()] = DbalType::getType($typeClassToName[$dbalTypeClass]);
+        }
+
+        return $types;
+    }
+
     /**
      * @param class-string<FlowType<mixed>> $flowType
      *
diff --git a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Benchmark/DbalLoaderBench.php b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Benchmark/DbalLoaderBench.php
index d4d5e1e0a..80f8af168 100644
--- a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Benchmark/DbalLoaderBench.php
+++ b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Benchmark/DbalLoaderBench.php
@@ -5,9 +5,10 @@
 namespace Flow\ETL\Adapter\Doctrine\Tests\Benchmark;
 
 use function Flow\ETL\Adapter\Doctrine\{to_dbal_schema_table, to_dbal_table_insert};
-use function Flow\ETL\DSL\df;
-use Doctrine\DBAL\DriverManager;
+use function Flow\ETL\DSL\flow_context;
+use Doctrine\DBAL\{Connection, DriverManager};
 use Doctrine\DBAL\Tools\DsnParser;
+use Flow\ETL\{FlowContext, Rows};
 use Flow\ETL\Tests\Double\FakeStaticOrdersExtractor;
 use PhpBench\Attributes\{BeforeMethods, Groups};
 
@@ -16,7 +17,11 @@ final class DbalLoaderBench
 {
     private const TABLE_NAME = 'benchmark_orders_loader';
 
-    private \Doctrine\DBAL\Connection $connection;
+    private Connection $connection;
+
+    private readonly FlowContext $context;
+
+    private Rows $rows;
 
     public function __construct()
     {
@@ -29,6 +34,8 @@ public function __construct()
         $params = (new DsnParser(['postgresql' => 'pdo_pgsql']))->parse($dsn);
 
         $this->connection = DriverManager::getConnection($params);
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
+        $this->context = flow_context();
     }
 
     public function __destruct()
@@ -58,9 +65,8 @@ public function setUp() : void
     #[BeforeMethods('setUp')]
     public function bench_load_10k() : void
     {
-        df()
-            ->read(new FakeStaticOrdersExtractor(10_000))
-            ->write(to_dbal_table_insert($this->connection, self::TABLE_NAME))
-            ->run();
+        foreach ($this->rows->chunks(1_000) as $chunk) {
+            to_dbal_table_insert($this->connection, self::TABLE_NAME)->load($chunk, $this->context);
+        }
     }
 }
diff --git a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLimitOffsetExtractorTest.php b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLimitOffsetExtractorTest.php
index ce9a46a1f..b48480914 100644
--- a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLimitOffsetExtractorTest.php
+++ b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLimitOffsetExtractorTest.php
@@ -8,7 +8,7 @@
 use function Flow\ETL\DSL\{data_frame, flow_context, from_array};
 use Doctrine\DBAL\Schema\Column;
 use Doctrine\DBAL\Types\{TextType, Type, Types};
-use Flow\ETL\Adapter\Doctrine\{DbalLoader, DbalTypesDetector, Order, OrderBy, Table, TypesMap};
+use Flow\ETL\Adapter\Doctrine\{DbalLoader, Order, OrderBy, Table, TypesMap};
 use Flow\ETL\Adapter\Doctrine\Tests\IntegrationTestCase;
 use Flow\ETL\{Config, Rows};
 use Flow\Types\Type\Native\{IntegerType, StringType};
@@ -31,10 +31,8 @@ public function test_creating_limit_offset_extractor_for_table() : void
             IntegerType::class => \Doctrine\DBAL\Types\IntegerType::class,
         ]);
 
-        $customConverter = new DbalTypesDetector($customTypesMap);
-
         $loader = (new DbalLoader($table, $this->postgresqlConnectionParams()))
-            ->withTypesDetector($customConverter);
+            ->withTypesMap($customTypesMap);
 
         (data_frame())
             ->read(from_array([
diff --git a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLoaderTest.php b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLoaderTest.php
index a5f9d2ef8..1c83efcf5 100644
--- a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLoaderTest.php
+++ b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Integration/DbalLoaderTest.php
@@ -9,7 +9,7 @@
 use Doctrine\DBAL\Types\TextType;
 use Doctrine\DBAL\Types\{Type, Types};
 use Flow\Doctrine\Bulk\Dialect\PostgreSQLInsertOptions;
-use Flow\ETL\Adapter\Doctrine\{DbalLoader, DbalTypesDetector, TypesMap};
+use Flow\ETL\Adapter\Doctrine\{DbalLoader, TypesMap};
 use Flow\ETL\Adapter\Doctrine\Tests\IntegrationTestCase;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\Types\Type\Native\{IntegerType, StringType};
@@ -57,7 +57,7 @@ public function test_create_loader_with_invalid_operation_from_connection() : vo
         );
     }
 
-    public function test_loader_with_custom_schema_to_types_converter() : void
+    public function test_loader_with_custom_types_map() : void
     {
         $this->pgsqlDatabaseContext->createTable((new Table(
             $table = 'flow_doctrine_bulk_test',
@@ -74,53 +74,8 @@ public function test_loader_with_custom_schema_to_types_converter() : void
             IntegerType::class => \Doctrine\DBAL\Types\IntegerType::class,
         ]);
 
-        $customConverter = new DbalTypesDetector($customTypesMap);
-
-        $loader = (new DbalLoader($table, $this->postgresqlConnectionParams()))
-            ->withTypesDetector($customConverter);
-
-        (data_frame())
-            ->read(from_array([
-                ['id' => 1, 'name' => 'Name One', 'description' => 'Description One'],
-                ['id' => 2, 'name' => 'Name Two', 'description' => 'Description Two'],
-            ]))
-            ->load($loader)
-            ->run();
-
-        self::assertEquals(2, $this->pgsqlDatabaseContext->tableCount($table));
-        self::assertEquals(
-            [
-                ['id' => 1, 'name' => 'Name One', 'description' => 'Description One'],
-                ['id' => 2, 'name' => 'Name Two', 'description' => 'Description Two'],
-            ],
-            $this->pgsqlDatabaseContext->selectAll($table)
-        );
-    }
-
-    public function test_loader_with_custom_schema_to_types_converter_and_manual_column_types() : void
-    {
-        $this->pgsqlDatabaseContext->createTable((new Table(
-            $table = 'flow_doctrine_bulk_test',
-            [
-                new Column('id', Type::getType(Types::INTEGER), ['notnull' => true]),
-                new Column('name', Type::getType(Types::STRING), ['notnull' => true, 'length' => 255]),
-                new Column('description', Type::getType(Types::TEXT), ['notnull' => false]),
-            ],
-        ))
-            ->setPrimaryKey(['id']));
-
-        $customTypesMap = new TypesMap([
-            StringType::class => TextType::class,
-            IntegerType::class => \Doctrine\DBAL\Types\IntegerType::class,
-        ]);
-
-        $customConverter = new DbalTypesDetector($customTypesMap);
-
         $loader = (new DbalLoader($table, $this->postgresqlConnectionParams()))
-            ->withTypesDetector($customConverter)
-            ->withColumnTypes([
-                'name' => Type::getType(Types::STRING),
-            ]);
+            ->withTypesMap($customTypesMap);
 
         (data_frame())
             ->read(from_array([
diff --git a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/DbalTypesDetectorTest.php b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/DbalTypesDetectorTest.php
deleted file mode 100644
index f0a40c1c0..000000000
--- a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/DbalTypesDetectorTest.php
+++ /dev/null
@@ -1,68 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Flow\ETL\Adapter\Doctrine\Tests\Unit;
-
-use function Flow\ETL\DSL\{bool_schema, datetime_schema, float_schema, integer_schema, json_schema, list_schema, map_schema, schema, string_schema, structure_schema};
-use function Flow\Types\DSL\{type_integer, type_list, type_map, type_string, type_structure};
-use Doctrine\DBAL\Types\{Type, Types};
-use Flow\ETL\Adapter\Doctrine\DbalTypesDetector;
-use PHPUnit\Framework\TestCase;
-
-final class DbalTypesDetectorTest extends TestCase
-{
-    public function test_converts_empty_schema() : void
-    {
-        $converter = new DbalTypesDetector();
-
-        $schema = schema();
-
-        $types = $converter->convert($schema);
-
-        self::assertCount(0, $types);
-    }
-
-    public function test_converts_flow_schema_to_dbal_types() : void
-    {
-        $converter = new DbalTypesDetector();
-
-        $schema = schema(string_schema('name'), integer_schema('age'), float_schema('score'), bool_schema('active'), datetime_schema('created_at'));
-
-        $types = $converter->convert($schema);
-
-        self::assertCount(5, $types);
-        self::assertInstanceOf(Type::class, $types['name']);
-        self::assertSame(Types::STRING, Type::getTypeRegistry()->lookupName($types['name']));
-        self::assertSame(Types::INTEGER, Type::getTypeRegistry()->lookupName($types['age']));
-        self::assertSame(Types::FLOAT, Type::getTypeRegistry()->lookupName($types['score']));
-        self::assertSame(Types::BOOLEAN, Type::getTypeRegistry()->lookupName($types['active']));
-        self::assertSame(Types::DATETIME_IMMUTABLE, Type::getTypeRegistry()->lookupName($types['created_at']));
-    }
-
-    public function test_converts_json_type_to_json() : void
-    {
-        $converter = new DbalTypesDetector();
-
-        $schema = schema(json_schema('data'));
-
-        $types = $converter->convert($schema);
-
-        self::assertCount(1, $types);
-        self::assertSame(Types::JSON, Type::getTypeRegistry()->lookupName($types['data']));
-    }
-
-    public function test_converts_nested_types_to_json() : void
-    {
-        $converter = new DbalTypesDetector();
-
-        $schema = schema(list_schema('items', type_list(type_string())), map_schema('metadata', type_map(type_string(), type_string())), structure_schema('config', type_structure(['field1' => type_string(), 'field2' => type_integer()])));
-
-        $types = $converter->convert($schema);
-
-        self::assertCount(3, $types);
-        self::assertSame(Types::JSON, Type::getTypeRegistry()->lookupName($types['items']));
-        self::assertSame(Types::JSON, Type::getTypeRegistry()->lookupName($types['metadata']));
-        self::assertSame(Types::JSON, Type::getTypeRegistry()->lookupName($types['config']));
-    }
-}
diff --git a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/TypesMapTest.php b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/TypesMapTest.php
index 541f99ba0..54933202f 100644
--- a/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/TypesMapTest.php
+++ b/src/adapter/etl-adapter-doctrine/tests/Flow/ETL/Adapter/Doctrine/Tests/Unit/TypesMapTest.php
@@ -20,7 +20,7 @@
     UuidType,
     XMLElementType,
     XMLType};
-use Flow\Types\Type\Native\{BooleanType, FloatType, IntegerType, StringType};
+use Flow\Types\Type\Native\{BooleanType, EnumType, FloatType, IntegerType, StringType};
 use PHPUnit\Framework\TestCase;
 
 final class TypesMapTest extends TestCase
@@ -188,6 +188,7 @@ public function test_default_flow_types_constant_mapping() : void
             XMLElementType::class => \Doctrine\DBAL\Types\StringType::class,
             HTMLType::class => \Doctrine\DBAL\Types\StringType::class,
             HTMLElementType::class => \Doctrine\DBAL\Types\StringType::class,
+            EnumType::class => \Doctrine\DBAL\Types\StringType::class,
             ListType::class => DbalJsonType::class,
             MapType::class => DbalJsonType::class,
             StructureType::class => DbalJsonType::class,
diff --git a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Benchmark/ExcelLoaderBench.php b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Benchmark/ExcelLoaderBench.php
index 8ce281376..c9b839142 100644
--- a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Benchmark/ExcelLoaderBench.php
+++ b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Benchmark/ExcelLoaderBench.php
@@ -4,14 +4,20 @@
 
 namespace Flow\ETL\Adapter\Excel\Tests\Benchmark;
 
-use function Flow\ETL\Adapter\Excel\DSL\{from_excel, to_excel};
-use function Flow\ETL\DSL\df;
+use function Flow\ETL\Adapter\Excel\DSL\to_excel;
+use function Flow\ETL\DSL\flow_context;
 use Flow\ETL\Adapter\Excel\ExcelWriter;
+use Flow\ETL\{FlowContext, Rows};
+use Flow\ETL\Tests\Double\FakeStaticOrdersExtractor;
 use PhpBench\Attributes\Groups;
 
 #[Groups(['loader'])]
 final readonly class ExcelLoaderBench
 {
+    private FlowContext $context;
+
+    private Rows $rows;
+
     private string $tempDir;
 
     public function __construct()
@@ -21,35 +27,26 @@ public function __construct()
         if (!\is_dir($this->tempDir)) {
             \mkdir($this->tempDir, 0777, true);
         }
+
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
+        $this->context = flow_context();
     }
 
     public function bench_load_10k_ods() : void
     {
-        $inputPath = __DIR__ . '/../Fixtures/orders_flow.ods';
         $outputPath = $this->tempDir . '/output_bench.ods';
 
-        if (\file_exists($outputPath)) {
-            \unlink($outputPath);
-        }
-
-        df()
-            ->read(from_excel($inputPath))
-            ->write(to_excel($outputPath)->withWriter(ExcelWriter::ODS))
-            ->run();
+        to_excel($outputPath)->withWriter(ExcelWriter::ODS)->load($this->rows, $this->context);
     }
 
     public function bench_load_10k_xlsx() : void
     {
-        $inputPath = __DIR__ . '/../Fixtures/orders_flow.xlsx';
         $outputPath = $this->tempDir . '/output_bench.xlsx';
 
         if (\file_exists($outputPath)) {
             \unlink($outputPath);
         }
 
-        df()
-            ->read(from_excel($inputPath))
-            ->write(to_excel($outputPath)->withWriter(ExcelWriter::XLSX))
-            ->run();
+        to_excel($outputPath)->withWriter(ExcelWriter::XLSX)->load($this->rows, $this->context);
     }
 }
diff --git a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Benchmark/JsonLoaderBench.php b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Benchmark/JsonLoaderBench.php
index f816ea37b..232099d28 100644
--- a/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Benchmark/JsonLoaderBench.php
+++ b/src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Benchmark/JsonLoaderBench.php
@@ -4,9 +4,9 @@
 
 namespace Flow\ETL\Adapter\JSON\Tests\Benchmark;
 
-use function Flow\ETL\Adapter\JSON\{from_json, to_json};
+use function Flow\ETL\Adapter\JSON\to_json;
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\{FlowContext, Rows};
+use Flow\ETL\{FlowContext, Rows, Tests\Double\FakeStaticOrdersExtractor};
 use PhpBench\Attributes\Groups;
 
 #[Groups(['loader'])]
@@ -22,11 +22,7 @@ public function __construct()
     {
         $this->context = flow_context(config());
         $this->outputPath = \tempnam(\sys_get_temp_dir(), 'etl_json_loader_bench') . '.json';
-        $this->rows = \Flow\ETL\DSL\rows();
-
-        foreach (from_json(__DIR__ . '/../Fixtures/orders_flow.json')->extract($this->context) as $rows) {
-            $this->rows = $this->rows->merge($rows);
-        }
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
     }
 
     public function __destruct()
diff --git a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/ParquetLoaderBench.php b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/ParquetLoaderBench.php
index a4ecb777c..67f2d6745 100644
--- a/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/ParquetLoaderBench.php
+++ b/src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Benchmark/ParquetLoaderBench.php
@@ -4,9 +4,9 @@
 
 namespace Flow\ETL\Adapter\Parquet\Tests\Benchmark;
 
-use function Flow\ETL\Adapter\Parquet\{from_parquet, to_parquet};
+use function Flow\ETL\Adapter\Parquet\to_parquet;
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\{FlowContext, Rows};
+use Flow\ETL\{FlowContext, Rows, Tests\Double\FakeStaticOrdersExtractor};
 use PhpBench\Attributes\Groups;
 
 #[Groups(['loader'])]
@@ -22,11 +22,7 @@ public function __construct()
     {
         $this->context = flow_context(config());
         $this->outputPath = \tempnam(\sys_get_temp_dir(), 'etl_parquet_loader_bench') . '.parquet';
-        $this->rows = \Flow\ETL\DSL\rows();
-
-        foreach (from_parquet(__DIR__ . '/Fixtures/orders_10k.parquet')->extract($this->context) as $rows) {
-            $this->rows = $this->rows->merge($rows);
-        }
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
     }
 
     public function __destruct()
diff --git a/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Benchmark/PostgreSqlLoaderBench.php b/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Benchmark/PostgreSqlLoaderBench.php
index bc84c11ed..c9265fe4c 100644
--- a/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Benchmark/PostgreSqlLoaderBench.php
+++ b/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Benchmark/PostgreSqlLoaderBench.php
@@ -5,8 +5,9 @@
 namespace Flow\ETL\Adapter\PostgreSql\Tests\Benchmark;
 
 use function Flow\ETL\Adapter\PostgreSql\to_pgsql_table;
-use function Flow\ETL\DSL\df;
+use function Flow\ETL\DSL\flow_context;
 use function Flow\PostgreSql\DSL\{column, create, data_type_double_precision, data_type_integer, data_type_jsonb, data_type_text, data_type_timestamptz, data_type_uuid, drop, pgsql_client, pgsql_connection_dsn, pgsql_mapper};
+use Flow\ETL\{FlowContext, Rows};
 use Flow\ETL\Tests\Double\FakeStaticOrdersExtractor;
 use Flow\PostgreSql\Client\Client;
 use PhpBench\Attributes\{BeforeMethods, Groups};
@@ -18,6 +19,10 @@ final class PostgreSqlLoaderBench
 
     private Client $client;
 
+    private FlowContext $context;
+
+    private Rows $rows;
+
     public function __construct()
     {
         $dsn = \getenv('PGSQL_DATABASE_URL');
@@ -38,6 +43,8 @@ public function __construct()
             pgsql_connection_dsn($dsn),
             mapper: pgsql_mapper(),
         );
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
+        $this->context = flow_context();
     }
 
     public function __destruct()
@@ -72,9 +79,8 @@ public function setUp() : void
     #[BeforeMethods('setUp')]
     public function bench_load_10k() : void
     {
-        df()
-            ->read(new FakeStaticOrdersExtractor(10_000))
-            ->write(to_pgsql_table($this->client, self::TABLE_NAME))
-            ->run();
+        foreach ($this->rows->chunks(1_000) as $chunk) {
+            to_pgsql_table($this->client, self::TABLE_NAME)->load($chunk, $this->context);
+        }
     }
 }
diff --git a/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Benchmark/TextLoaderBench.php b/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Benchmark/TextLoaderBench.php
index 8eceba544..6db6d93aa 100644
--- a/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Benchmark/TextLoaderBench.php
+++ b/src/adapter/etl-adapter-text/tests/Flow/ETL/Adapter/Text/Tests/Benchmark/TextLoaderBench.php
@@ -4,9 +4,9 @@
 
 namespace Flow\ETL\Adapter\Text\Tests\Benchmark;
 
-use function Flow\ETL\Adapter\Text\{from_text, to_text};
+use function Flow\ETL\Adapter\Text\to_text;
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\{FlowContext, Rows};
+use Flow\ETL\{FlowContext, Row, Rows, Tests\Double\FakeStaticOrdersExtractor};
 use PhpBench\Attributes\Groups;
 
 #[Groups(['loader'])]
@@ -22,11 +22,9 @@ public function __construct()
     {
         $this->context = flow_context(config());
         $this->outputPath = \tempnam(\sys_get_temp_dir(), 'etl_txt_loader_bench') . '.txt';
-        $this->rows = \Flow\ETL\DSL\rows();
-
-        foreach (from_text(__DIR__ . '/../Fixtures/orders_flow.csv')->extract($this->context) as $rows) {
-            $this->rows = $this->rows->merge($rows);
-        }
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows()->map(
+            fn (Row $r) => \Flow\ETL\DSL\row($r->get('order_id'))
+        );
     }
 
     public function __destruct()
diff --git a/src/core/etl/src/Flow/ETL/Row.php b/src/core/etl/src/Flow/ETL/Row.php
index c36d38f69..8004dd416 100644
--- a/src/core/etl/src/Flow/ETL/Row.php
+++ b/src/core/etl/src/Flow/ETL/Row.php
@@ -8,9 +8,11 @@
 use Flow\ETL\Hash\{Algorithm, NativePHPHash};
 use Flow\ETL\Row\{Entries, Entry, Reference};
 
-final readonly class Row
+final class Row
 {
-    public function __construct(private Entries $entries)
+    private ?Schema $schema = null;
+
+    public function __construct(private readonly Entries $entries)
     {
     }
 
@@ -137,13 +139,19 @@ public function rename(string $currentName, string $newName) : self
      */
     public function schema() : Schema
     {
+        if ($this->schema !== null) {
+            return $this->schema;
+        }
+
         $definitions = [];
 
         foreach ($this->entries->all() as $entry) {
             $definitions[] = $entry->definition();
         }
 
-        return new Schema(...$definitions);
+        $this->schema = new Schema(...$definitions);
+
+        return $this->schema;
     }
 
     /**
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/BooleanEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/BooleanEntry.php
index ce35ba06b..6e91b08ed 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/BooleanEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/BooleanEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_boolean, type_equals, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\BooleanDefinition;
@@ -18,12 +18,7 @@ final class BooleanEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<bool>
-     */
-    private readonly Type $type;
+    private BooleanDefinition $definition;
 
     /**
      * @throws InvalidArgumentException
@@ -34,8 +29,7 @@ public function __construct(private readonly string $name, private readonly ?boo
             throw InvalidArgumentException::because('Entry name cannot be empty');
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_boolean();
+        $this->definition = new BooleanDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -45,12 +39,12 @@ public function __toString() : string
 
     public function definition() : BooleanDefinition
     {
-        return new BooleanDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->metadata);
+        return new self($this->name, $this->value, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -64,7 +58,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value() === $entry->value();
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value() === $entry->value();
     }
 
     public function map(callable $mapper) : static
@@ -96,7 +90,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?bool
@@ -106,6 +100,6 @@ public function value() : ?bool
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/DateEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/DateEntry.php
index d96a1c598..2bbf4b1d7 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/DateEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/DateEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_date, type_equals, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\DateDefinition;
@@ -18,12 +18,7 @@ final class DateEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<\DateTimeInterface>
-     */
-    private readonly Type $type;
+    private DateDefinition $definition;
 
     private readonly ?\DateTimeInterface $value;
 
@@ -50,8 +45,7 @@ public function __construct(private readonly string $name, \DateTimeInterface|st
             $this->value = $value;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_date();
+        $this->definition = new DateDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -61,12 +55,12 @@ public function __toString() : string
 
     public function definition() : DateDefinition
     {
-        return new DateDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? clone $this->value : null, $this->metadata);
+        return new self($this->name, $this->value ? clone $this->value : null, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -80,7 +74,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value() == $entry->value();
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value() == $entry->value();
     }
 
     public function map(callable $mapper) : static
@@ -111,7 +105,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\DateTimeInterface
@@ -121,6 +115,6 @@ public function value() : ?\DateTimeInterface
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/DateTimeEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/DateTimeEntry.php
index a38e80169..a03d4eb46 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/DateTimeEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/DateTimeEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_datetime, type_equals, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\DateTimeDefinition;
@@ -18,12 +18,7 @@ final class DateTimeEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<\DateTimeInterface>
-     */
-    private readonly Type $type;
+    private DateTimeDefinition $definition;
 
     private readonly ?\DateTimeInterface $value;
 
@@ -51,8 +46,7 @@ public function __construct(
             $this->value = $value;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_datetime();
+        $this->definition = new DateTimeDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -62,12 +56,12 @@ public function __toString() : string
 
     public function definition() : DateTimeDefinition
     {
-        return new DateTimeDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? clone $this->value : null, $this->metadata);
+        return new self($this->name, $this->value ? clone $this->value : null, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -81,7 +75,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value() == $entry->value();
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value() == $entry->value();
     }
 
     public function map(callable $mapper) : static
@@ -112,7 +106,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\DateTimeInterface
@@ -122,6 +116,6 @@ public function value() : ?\DateTimeInterface
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php
index 18ee6b205..8b2b7e21d 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/EnumEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_enum, type_equals, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\EnumDefinition;
 use Flow\ETL\Schema\Metadata;
@@ -17,22 +17,19 @@ final class EnumEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
     /**
-     * @var EnumType<\UnitEnum>
+     * @var EnumDefinition<\UnitEnum>
      */
-    private readonly EnumType $type;
+    private EnumDefinition $definition;
 
     public function __construct(
         private readonly string $name,
         private readonly ?\UnitEnum $value,
         ?Metadata $metadata = null,
     ) {
-        $this->metadata = $metadata ?: Metadata::empty();
-        /** @var EnumType<\UnitEnum> $type */
-        $type = type_enum($this->value === null ? \UnitEnum::class : $this->value::class);
-        $this->type = $type;
+        /** @var class-string<\UnitEnum>&literal-string $enumClass */
+        $enumClass = $this->value === null ? \UnitEnum::class : $this->value::class;
+        $this->definition = new EnumDefinition($this->name, $enumClass, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -49,15 +46,12 @@ public function __toString() : string
      */
     public function definition() : EnumDefinition
     {
-        /** @var class-string<\UnitEnum>&literal-string $enumClass */
-        $enumClass = $this->value === null ? \UnitEnum::class : $this->value::class;
-
-        return new EnumDefinition($this->name, $enumClass, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->metadata);
+        return new self($this->name, $this->value, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -71,7 +65,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $entry instanceof self && type_equals($this->type, $entry->type) && $this->value === $entry->value;
+        return $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value === $entry->value;
     }
 
     public function map(callable $mapper) : static
@@ -103,7 +97,7 @@ public function toString() : string
      */
     public function type() : EnumType
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\UnitEnum
@@ -113,6 +107,6 @@ public function value() : ?\UnitEnum
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/FloatEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/FloatEntry.php
index d1130fd23..a72a252a8 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/FloatEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/FloatEntry.php
@@ -5,7 +5,7 @@
 namespace Flow\ETL\Row\Entry;
 
 use function Flow\ETL\DSL\is_type;
-use function Flow\Types\DSL\{type_equals, type_float, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Brick\Math\BigDecimal;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
@@ -20,12 +20,7 @@ final class FloatEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<float>
-     */
-    private readonly Type $type;
+    private FloatDefinition $definition;
 
     private readonly ?float $value;
 
@@ -38,9 +33,8 @@ public function __construct(
             throw InvalidArgumentException::because('Entry name cannot be empty');
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
         $this->value = $value !== null ? BigDecimal::of($value)->toFloat() : null;
-        $this->type = type_float();
+        $this->definition = new FloatDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -50,12 +44,12 @@ public function __toString() : string
 
     public function definition() : FloatDefinition
     {
-        return new FloatDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->metadata);
+        return new self($this->name, $this->value, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -83,12 +77,12 @@ public function isEqual(Entry $entry) : bool
         if ($entryValue === null && $thisValue === null) {
             return $this->is($entry->name())
                 && $entry instanceof self
-                && is_type($this->type, $entry->type);
+                && is_type($this->type(), $entry->type());
         }
 
         return $this->is($entry->name())
             && $entry instanceof self
-            && type_equals($this->type, $entry->type)
+            && type_equals($this->type(), $entry->type())
             /** @phpstan-ignore-next-line */
             && \bccomp((string) $thisValue, (string) $entryValue) === 0;
     }
@@ -122,7 +116,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?float
@@ -132,6 +126,6 @@ public function value() : ?float
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php
index 115bdf126..c468e06c2 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLElementEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_html_element, type_instance_of, type_optional};
+use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional};
 use Dom\{HTMLDocument, HTMLElement};
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\HTMLElementDefinition;
@@ -18,12 +18,7 @@ final class HTMLElementEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<HTMLElement>
-     */
-    private readonly Type $type;
+    private HTMLElementDefinition $definition;
 
     private readonly ?HTMLElement $value;
 
@@ -38,9 +33,8 @@ public function __construct(
             $value = $document->documentElement;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
         $this->value = $value;
-        $this->type = type_html_element();
+        $this->definition = new HTMLElementDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -54,12 +48,12 @@ public function __toString() : string
 
     public function definition() : HTMLElementDefinition
     {
-        return new HTMLElementDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, type_optional(type_instance_of(HTMLElement::class))->assert($this->value ? $this->value->cloneNode(true) : null), $this->metadata);
+        return new self($this->name, type_optional(type_instance_of(HTMLElement::class))->assert($this->value ? $this->value->cloneNode(true) : null), $this->definition->metadata());
     }
 
     public function is(Reference|string $name) : bool
@@ -77,7 +71,7 @@ public function isEqual(Entry $entry) : bool
             return false;
         }
 
-        if (!type_equals($this->type, $entry->type)) {
+        if (!type_equals($this->type(), $entry->type())) {
             return false;
         }
 
@@ -113,7 +107,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?HTMLElement
@@ -123,6 +117,6 @@ public function value() : ?HTMLElement
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php
index 444f848ba..9a255d0ba 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/HTMLEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_html, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Dom\HTMLDocument;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\HTMLDefinition;
@@ -18,12 +18,7 @@ final class HTMLEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<HTMLDocument>
-     */
-    private readonly Type $type;
+    private HTMLDefinition $definition;
 
     private ?HTMLDocument $value;
 
@@ -38,8 +33,7 @@ public function __construct(
             $this->value = $value;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_html();
+        $this->definition = new HTMLDefinition($this->name, null === $this->value, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -49,12 +43,12 @@ public function __toString() : string
 
     public function definition() : HTMLDefinition
     {
-        return new HTMLDefinition($this->name, null === $this->value, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? clone $this->value : null, $this->metadata);
+        return new self($this->name, $this->value ? clone $this->value : null, $this->definition->metadata());
     }
 
     public function is(Reference|string $name) : bool
@@ -72,7 +66,7 @@ public function isEqual(Entry $entry) : bool
             return false;
         }
 
-        if (!type_equals($this->type, $entry->type)) {
+        if (!type_equals($this->type(), $entry->type())) {
             return false;
         }
 
@@ -105,7 +99,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?HTMLDocument
@@ -115,6 +109,6 @@ public function value() : ?HTMLDocument
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/IntegerEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/IntegerEntry.php
index a0540af4e..0cae5684b 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/IntegerEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/IntegerEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_integer, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\IntegerDefinition;
@@ -18,12 +18,7 @@ final class IntegerEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<int>
-     */
-    private readonly Type $type;
+    private IntegerDefinition $definition;
 
     /**
      * @throws InvalidArgumentException
@@ -37,8 +32,7 @@ public function __construct(
             throw InvalidArgumentException::because('Entry name cannot be empty');
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_integer();
+        $this->definition = new IntegerDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -48,12 +42,12 @@ public function __toString() : string
 
     public function definition() : IntegerDefinition
     {
-        return new IntegerDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->metadata);
+        return new self($this->name, $this->value, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -67,7 +61,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value() === $entry->value();
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value() === $entry->value();
     }
 
     public function map(callable $mapper) : static
@@ -99,7 +93,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?int
@@ -109,6 +103,6 @@ public function value() : ?int
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/JsonEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/JsonEntry.php
index 4cdd7d914..01410289e 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/JsonEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/JsonEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_json, type_optional};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\JsonDefinition;
@@ -19,14 +19,9 @@ final class JsonEntry implements Entry
 {
     use EntryRef;
 
-    private readonly ?Json $json;
-
-    private Metadata $metadata;
+    private JsonDefinition $definition;
 
-    /**
-     * @var Type<Json>
-     */
-    private readonly Type $type;
+    private readonly ?Json $json;
 
     /**
      * @param null|array<array-key, mixed>|Json|string $value
@@ -56,8 +51,7 @@ public function __construct(
             $this->json = null;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_json();
+        $this->definition = new JsonDefinition($this->name, $this->json === null, $metadata ?: Metadata::empty());
     }
 
     /**
@@ -91,12 +85,12 @@ public function __toString() : string
 
     public function definition() : JsonDefinition
     {
-        return new JsonDefinition($this->name, $this->json === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->json, $this->metadata);
+        return new self($this->name, $this->json, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -118,7 +112,7 @@ public function isEqual(Entry $entry) : bool
             return false;
         }
 
-        if (!type_equals($this->type, $entry->type)) {
+        if (!type_equals($this->type(), $entry->type())) {
             return false;
         }
 
@@ -138,7 +132,7 @@ public function isEqual(Entry $entry) : bool
 
     public function map(callable $mapper) : static
     {
-        return new self($this->name, $mapper($this->json), $this->metadata);
+        return new self($this->name, $mapper($this->json), $this->definition->metadata());
     }
 
     public function name() : string
@@ -148,7 +142,7 @@ public function name() : string
 
     public function rename(string $name) : static
     {
-        return new self($name, $this->json, $this->metadata);
+        return new self($name, $this->json, $this->definition->metadata());
     }
 
     public function toString() : string
@@ -165,7 +159,7 @@ public function toString() : string
      */
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?Json
@@ -175,6 +169,6 @@ public function value() : ?Json
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->cast($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->cast($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/ListEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/ListEntry.php
index 883e987b4..c15503478 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/ListEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/ListEntry.php
@@ -22,12 +22,10 @@ final class ListEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
     /**
-     * @var ListType<T>
+     * @var ListDefinition<T>
      */
-    private readonly ListType $type;
+    private ListDefinition $definition;
 
     /**
      * @param ?list<T> $value
@@ -49,8 +47,7 @@ public function __construct(
             throw InvalidArgumentException::because('Expected ' . $type->toString() . ' got different types: ' . (new TypeDetector())->detectType($this->value)->toString());
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = $type;
+        $this->definition = new ListDefinition($this->name, $type, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -63,12 +60,12 @@ public function __toString() : string
      */
     public function definition() : ListDefinition
     {
-        return new ListDefinition($this->name, $this->type, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->type, $this->metadata);
+        return new self($this->name, $this->value, $this->type(), $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -96,18 +93,18 @@ public function isEqual(Entry $entry) : bool
         if ($entryValue === null && $thisValue === null) {
             return $this->is($entry->name())
                 && $entry instanceof self
-                && type_equals($this->type, $entry->type);
+                && type_equals($this->type(), $entry->type());
         }
 
         return $this->is($entry->name())
             && $entry instanceof self
-            && type_equals($this->type, $entry->type)
+            && type_equals($this->type(), $entry->type())
             && (new ArrayComparison())->equals($thisValue, \is_array($entryValue) ? $entryValue : null);
     }
 
     public function map(callable $mapper) : static
     {
-        return new self($this->name, $mapper($this->value), $this->type);
+        return new self($this->name, $mapper($this->value), $this->type());
     }
 
     public function name() : string
@@ -117,7 +114,7 @@ public function name() : string
 
     public function rename(string $name) : static
     {
-        return new self($name, $this->value, $this->type);
+        return new self($name, $this->value, $this->type());
     }
 
     public function toString() : string
@@ -134,7 +131,7 @@ public function toString() : string
      */
     public function type() : ListType
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?array
@@ -144,6 +141,6 @@ public function value() : ?array
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->type);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->type());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/MapEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/MapEntry.php
index 0afa159a0..340ac5cd4 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/MapEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/MapEntry.php
@@ -23,12 +23,10 @@ final class MapEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
     /**
-     * @var MapType<TKey, TValue>
+     * @var MapDefinition<TKey, TValue>
      */
-    private MapType $type;
+    private MapDefinition $definition;
 
     /**
      * @param ?array<array-key, mixed> $value
@@ -50,8 +48,7 @@ public function __construct(
             throw InvalidArgumentException::because('Expected ' . $type->toString() . ' got different types: ' . (new TypeDetector())->detectType($this->value)->toString());
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = $type;
+        $this->definition = new MapDefinition($this->name, $type, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -64,12 +61,12 @@ public function __toString() : string
      */
     public function definition() : MapDefinition
     {
-        return new MapDefinition($this->name, $this->type, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->type, $this->metadata);
+        return new self($this->name, $this->value, $this->type(), $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -97,18 +94,18 @@ public function isEqual(Entry $entry) : bool
         if ($entryValue === null && $thisValue === null) {
             return $this->is($entry->name())
                 && $entry instanceof self
-                && type_equals($this->type, $entry->type);
+                && type_equals($this->type(), $entry->type());
         }
 
         return $this->is($entry->name())
             && $entry instanceof self
-            && type_equals($this->type, $entry->type)
+            && type_equals($this->type(), $entry->type())
             && (new ArrayComparison())->equals($thisValue, \is_array($entryValue) ? $entryValue : null);
     }
 
     public function map(callable $mapper) : static
     {
-        return new self($this->name, $mapper($this->value), $this->type);
+        return new self($this->name, $mapper($this->value), $this->type());
     }
 
     public function name() : string
@@ -118,7 +115,7 @@ public function name() : string
 
     public function rename(string $name) : static
     {
-        return new self($name, $this->value, $this->type);
+        return new self($name, $this->value, $this->type());
     }
 
     public function toString() : string
@@ -135,7 +132,7 @@ public function toString() : string
      */
     public function type() : MapType
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?array
@@ -145,6 +142,6 @@ public function value() : ?array
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->type);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->type());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/StringEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/StringEntry.php
index d1330633b..d4d36dd1a 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/StringEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/StringEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_optional, type_string};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\StringDefinition;
@@ -18,14 +18,7 @@ final class StringEntry implements Entry
 {
     use EntryRef;
 
-    private bool $fromNull = false;
-
-    private Metadata $metadata;
-
-    /**
-     * @var Type<string>
-     */
-    private readonly Type $type;
+    private StringDefinition $definition;
 
     /**
      * @throws InvalidArgumentException
@@ -34,21 +27,25 @@ public function __construct(
         private readonly string $name,
         private readonly ?string $value,
         ?Metadata $metadata = null,
+        bool $fromNull = false,
     ) {
         if ('' === $name) {
             throw InvalidArgumentException::because('Entry name cannot be empty');
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_string();
+        $metadata = $metadata ?: Metadata::empty();
+        $this->definition = new StringDefinition(
+            $this->name,
+            $this->value === null,
+            $fromNull
+                ? $metadata->merge(Metadata::fromArray([Metadata::FROM_NULL => true]))
+                : $metadata
+        );
     }
 
     public static function fromNull(string $name, ?Metadata $metadata = null) : self
     {
-        $entry = new self($name, null, $metadata);
-        $entry->fromNull = true;
-
-        return $entry;
+        return new self($name, null, $metadata, fromNull: true);
     }
 
     /**
@@ -74,18 +71,12 @@ public function __toString() : string
 
     public function definition() : StringDefinition
     {
-        return new StringDefinition(
-            $this->name,
-            $this->value === null,
-            $this->fromNull
-                ? $this->metadata->merge(Metadata::fromArray([Metadata::FROM_NULL => true]))
-                : $this->metadata
-        );
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->metadata);
+        return new self($this->name, $this->value, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -99,7 +90,7 @@ public function is(string|Reference $name) : bool
 
     public function isEqual(Entry $entry) : bool
     {
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value() === $entry->value();
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value() === $entry->value();
     }
 
     public function map(callable $mapper) : static
@@ -138,7 +129,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?string
@@ -148,6 +139,6 @@ public function value() : ?string
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/StructureEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/StructureEntry.php
index 57011e20e..d0f65865f 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/StructureEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/StructureEntry.php
@@ -22,12 +22,10 @@ final class StructureEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
     /**
-     * @var StructureType<T>
+     * @var StructureDefinition<T>
      */
-    private readonly StructureType $type;
+    private StructureDefinition $definition;
 
     /**
      * @param ?array<array-key, mixed> $value
@@ -53,8 +51,7 @@ public function __construct(
             throw InvalidArgumentException::because('Expected ' . $type->toString() . ' got different types: ' . (new TypeDetector())->detectType($this->value)->toString());
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = $type;
+        $this->definition = new StructureDefinition($this->name, $type, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __toString() : string
@@ -67,12 +64,12 @@ public function __toString() : string
      */
     public function definition() : StructureDefinition
     {
-        return new StructureDefinition($this->name, $this->type, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value, $this->type, $this->metadata);
+        return new self($this->name, $this->value, $this->type(), $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -98,15 +95,15 @@ public function isEqual(Entry $entry) : bool
         }
 
         if ($entryValue === null && $thisValue === null) {
-            return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type);
+            return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type());
         }
 
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && (new ArrayComparison())->equals($thisValue, \is_array($entryValue) ? $entryValue : null);
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && (new ArrayComparison())->equals($thisValue, \is_array($entryValue) ? $entryValue : null);
     }
 
     public function map(callable $mapper) : static
     {
-        return new self($this->name, $mapper($this->value), $this->type);
+        return new self($this->name, $mapper($this->value), $this->type());
     }
 
     public function name() : string
@@ -116,7 +113,7 @@ public function name() : string
 
     public function rename(string $name) : static
     {
-        return new self($name, $this->value, $this->type);
+        return new self($name, $this->value, $this->type());
     }
 
     public function toString() : string
@@ -133,7 +130,7 @@ public function toString() : string
      */
     public function type() : StructureType
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?array
@@ -143,6 +140,6 @@ public function value() : ?array
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->type);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->type());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/TimeEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/TimeEntry.php
index 97db7d479..28d0ba8d2 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/TimeEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/TimeEntry.php
@@ -5,7 +5,7 @@
 namespace Flow\ETL\Row\Entry;
 
 use function Flow\ETL\DSL\date_interval_to_microseconds;
-use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional, type_time};
+use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\TimeDefinition;
@@ -19,12 +19,7 @@ final class TimeEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<\DateInterval>
-     */
-    private readonly Type $type;
+    private TimeDefinition $definition;
 
     /**
      * Time represented php \DateInterval.
@@ -87,8 +82,7 @@ public function __construct(
             $this->value = null;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_time();
+        $this->definition = new TimeDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public static function fromDays(string $name, int $days) : self
@@ -145,12 +139,12 @@ public function __toString() : string
 
     public function definition() : TimeDefinition
     {
-        return new TimeDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? clone $this->value : null, $this->metadata);
+        return new self($this->name, $this->value ? clone $this->value : null, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -180,7 +174,7 @@ public function isEqual(Entry $entry) : bool
 
         return $this->is($entry->name())
             && $entry instanceof self
-            && type_equals($this->type, $entry->type)
+            && type_equals($this->type(), $entry->type())
             && date_interval_to_microseconds($thisValue) == date_interval_to_microseconds($entryValue);
     }
 
@@ -218,7 +212,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\DateInterval
@@ -228,6 +222,6 @@ public function value() : ?\DateInterval
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/UuidEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/UuidEntry.php
index 3fdc1c460..2b6ac71eb 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/UuidEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/UuidEntry.php
@@ -4,7 +4,7 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_optional, type_uuid};
+use function Flow\Types\DSL\{type_equals, type_optional};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\UuidDefinition;
@@ -19,12 +19,7 @@ final class UuidEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<Uuid>
-     */
-    private readonly Type $type;
+    private UuidDefinition $definition;
 
     private ?Uuid $value;
 
@@ -46,8 +41,7 @@ public function __construct(
             $this->value = $value;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_uuid();
+        $this->definition = new UuidDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public static function from(string $name, string $value) : self
@@ -62,12 +56,12 @@ public function __toString() : string
 
     public function definition() : UuidDefinition
     {
-        return new UuidDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? new Uuid($this->value->toString()) : null, $this->metadata);
+        return new self($this->name, $this->value ? new Uuid($this->value->toString()) : null, $this->definition->metadata());
     }
 
     public function is(string|Reference $name) : bool
@@ -95,7 +89,7 @@ public function isEqual(Entry $entry) : bool
         /**
          * @var Uuid $entryValue
          */
-        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type, $entry->type) && $this->value?->isEqual($entryValue);
+        return $this->is($entry->name()) && $entry instanceof self && type_equals($this->type(), $entry->type()) && $this->value?->isEqual($entryValue);
     }
 
     public function map(callable $mapper) : static
@@ -127,7 +121,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?Uuid
@@ -137,6 +131,6 @@ public function value() : ?Uuid
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php
index 92df7b191..cd5d82d73 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php
@@ -4,13 +4,12 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional, type_string, type_xml_element};
+use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional, type_string};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\XMLElementDefinition;
 use Flow\ETL\Schema\Metadata;
 use Flow\Types\Type;
-use Flow\Types\Type\Logical\XMLElementType;
 
 /**
  * @implements Entry<?\DOMElement>
@@ -19,12 +18,7 @@ final class XMLElementEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<\DOMElement>
-     */
-    private readonly Type $type;
+    private XMLElementDefinition $definition;
 
     private readonly ?\DOMElement $value;
 
@@ -43,9 +37,8 @@ public function __construct(
             $value = $doc->documentElement;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
         $this->value = $value;
-        $this->type = type_xml_element();
+        $this->definition = new XMLElementDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __serialize() : array
@@ -53,7 +46,6 @@ public function __serialize() : array
         return [
             'name' => $this->name,
             'value' => $this->value === null ? null : \base64_encode(\gzcompress($this->toString()) ?: ''),
-            'type' => $this->type,
         ];
     }
 
@@ -73,13 +65,12 @@ public function __toString() : string
     public function __unserialize(array $data) : void
     {
         type_string()->assert($data['name']);
-        type_instance_of(XMLElementType::class)->assert($data['type']);
 
         $this->name = $data['name'];
-        $this->type = $data['type'];
 
         if ($data['value'] === null) {
             $this->value = null;
+            $this->definition = new XMLElementDefinition($this->name, true, Metadata::empty());
 
             return;
         }
@@ -93,16 +84,17 @@ public function __unserialize(array $data) : void
          * @phpstan-ignore-next-line
          */
         $this->value = (new \DOMDocument())->importNode($domDocument->documentElement, true);
+        $this->definition = new XMLElementDefinition($this->name, false, Metadata::empty());
     }
 
     public function definition() : XMLElementDefinition
     {
-        return new XMLElementDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, type_optional(type_instance_of(\DOMElement::class))->assert($this->value ? $this->value->cloneNode(true) : null), $this->metadata);
+        return new self($this->name, type_optional(type_instance_of(\DOMElement::class))->assert($this->value ? $this->value->cloneNode(true) : null), $this->definition->metadata());
     }
 
     public function is(Reference|string $name) : bool
@@ -120,7 +112,7 @@ public function isEqual(Entry $entry) : bool
             return false;
         }
 
-        if (!type_equals($this->type, $entry->type)) {
+        if (!type_equals($this->type(), $entry->type())) {
             return false;
         }
 
@@ -157,7 +149,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\DOMElement
@@ -167,6 +159,6 @@ public function value() : ?\DOMElement
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php b/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php
index a5cbd094a..f8955c560 100644
--- a/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php
+++ b/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php
@@ -4,13 +4,12 @@
 
 namespace Flow\ETL\Row\Entry;
 
-use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional, type_string, type_xml};
+use function Flow\Types\DSL\{type_equals, type_instance_of, type_optional, type_string};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Row\{Entry, Reference};
 use Flow\ETL\Schema\Definition\XMLDefinition;
 use Flow\ETL\Schema\Metadata;
 use Flow\Types\Type;
-use Flow\Types\Type\Logical\XMLType;
 
 /**
  * @implements Entry<?\DOMDocument>
@@ -19,12 +18,7 @@ final class XMLEntry implements Entry
 {
     use EntryRef;
 
-    private Metadata $metadata;
-
-    /**
-     * @var Type<\DOMDocument>
-     */
-    private readonly Type $type;
+    private XMLDefinition $definition;
 
     private readonly ?\DOMDocument $value;
 
@@ -45,8 +39,7 @@ public function __construct(
             $this->value = $value;
         }
 
-        $this->metadata = $metadata ?: Metadata::empty();
-        $this->type = type_xml();
+        $this->definition = new XMLDefinition($this->name, $this->value === null, $metadata ?: Metadata::empty());
     }
 
     public function __serialize() : array
@@ -55,7 +48,6 @@ public function __serialize() : array
             'name' => $this->name,
             /** @phpstan-ignore-next-line  */
             'value' => $this->value === null ? null : \base64_encode(\gzcompress($this->toString())),
-            'type' => $this->type,
         ];
     }
 
@@ -74,13 +66,12 @@ public function __toString() : string
     public function __unserialize(array $data) : void
     {
         type_string()->assert($data['name']);
-        type_instance_of(XMLType::class)->assert($data['type']);
 
         $this->name = $data['name'];
-        $this->type = $data['type'];
 
         if ($data['value'] === null) {
             $this->value = null;
+            $this->definition = new XMLDefinition($this->name, true, Metadata::empty());
 
             return;
         }
@@ -95,16 +86,17 @@ public function __unserialize(array $data) : void
         }
 
         $this->value = $doc;
+        $this->definition = new XMLDefinition($this->name, false, Metadata::empty());
     }
 
     public function definition() : XMLDefinition
     {
-        return new XMLDefinition($this->name, $this->value === null, $this->metadata);
+        return $this->definition;
     }
 
     public function duplicate() : static
     {
-        return new self($this->name, $this->value ? clone $this->value : null, $this->metadata);
+        return new self($this->name, $this->value ? clone $this->value : null, $this->definition->metadata());
     }
 
     public function is(Reference|string $name) : bool
@@ -122,7 +114,7 @@ public function isEqual(Entry $entry) : bool
             return false;
         }
 
-        if (!type_equals($this->type, $entry->type)) {
+        if (!type_equals($this->type(), $entry->type())) {
             return false;
         }
 
@@ -163,7 +155,7 @@ public function toString() : string
 
     public function type() : Type
     {
-        return $this->type;
+        return $this->definition->type();
     }
 
     public function value() : ?\DOMDocument
@@ -173,6 +165,6 @@ public function value() : ?\DOMDocument
 
     public function withValue(mixed $value) : static
     {
-        return new self($this->name, type_optional($this->type())->assert($value), $this->metadata);
+        return new self($this->name, type_optional($this->type())->assert($value), $this->definition->metadata());
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/Rows.php b/src/core/etl/src/Flow/ETL/Rows.php
index d7ca5d0cf..c63960cc3 100644
--- a/src/core/etl/src/Flow/ETL/Rows.php
+++ b/src/core/etl/src/Flow/ETL/Rows.php
@@ -28,6 +28,8 @@ final class Rows implements \ArrayAccess, \Countable, \IteratorAggregate
      */
     private readonly array $rows;
 
+    private ?Schema $schema = null;
+
     public function __construct(Row ...$rows)
     {
         $this->rows = \array_values($rows);
@@ -689,6 +691,10 @@ public function reverse() : self
      */
     public function schema() : Schema
     {
+        if ($this->schema !== null) {
+            return $this->schema;
+        }
+
         if (!$this->count()) {
             return new Schema();
         }
@@ -697,13 +703,17 @@ public function schema() : Schema
         $schema = null;
 
         foreach ($this->rows as $row) {
-            $schema = $schema === null
-                ? $row->schema()
-                : $schema->merge($row->schema());
+            if ($schema === null) {
+                $schema = $row->schema();
+            } else {
+                $schema = $schema->merge($row->schema());
+            }
         }
 
         /** @var Schema $schema */
-        return $schema;
+        $this->schema = $schema;
+
+        return $this->schema;
     }
 
     /**
diff --git a/src/core/etl/src/Flow/ETL/Schema.php b/src/core/etl/src/Flow/ETL/Schema.php
index 1042800c5..bb3351892 100644
--- a/src/core/etl/src/Flow/ETL/Schema.php
+++ b/src/core/etl/src/Flow/ETL/Schema.php
@@ -207,6 +207,25 @@ public function gracefulRemove(string|Reference ...$entries) : self
         return $this;
     }
 
+    public function isSame(self $schema) : bool
+    {
+        if (\count($this->definitions) !== \count($schema->definitions)) {
+            return false;
+        }
+
+        foreach ($this->definitions as $entry => $definition) {
+            if (!\array_key_exists($entry, $schema->definitions)) {
+                return false;
+            }
+
+            if (!$definition->isSame($schema->definitions[$entry])) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
     /**
      * @return Schema
      */
@@ -255,8 +274,6 @@ public function makeNullable() : self
 
     public function merge(self $schema) : self
     {
-        $newDefinitions = $this->definitions;
-
         if (!$this->count()) {
             return $schema;
         }
@@ -265,6 +282,12 @@ public function merge(self $schema) : self
             return $this;
         }
 
+        if ($this->isSame($schema)) {
+            return $this;
+        }
+
+        $newDefinitions = $this->definitions;
+
         foreach ($schema->definitions as $entry => $definition) {
             if (!\array_key_exists($definition->entry()->name(), $newDefinitions)) {
                 $newDefinitions[$entry] = $definition->makeNullable();
diff --git a/src/core/etl/src/Flow/ETL/Schema/Definition/EnumDefinition.php b/src/core/etl/src/Flow/ETL/Schema/Definition/EnumDefinition.php
index 08b7973c1..798c0f58d 100644
--- a/src/core/etl/src/Flow/ETL/Schema/Definition/EnumDefinition.php
+++ b/src/core/etl/src/Flow/ETL/Schema/Definition/EnumDefinition.php
@@ -35,7 +35,7 @@ public function __construct(
         private readonly bool $nullable = false,
         ?Metadata $metadata = null,
     ) {
-        if (!\enum_exists($enumClass)) {
+        if ($enumClass !== \UnitEnum::class && !\enum_exists($enumClass)) {
             throw new InvalidArgumentException(\sprintf('Enum of type "%s" not found', $enumClass));
         }
 
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Benchmark/RowsBench.php b/src/core/etl/tests/Flow/ETL/Tests/Benchmark/RowsBench.php
index 83c3f0cdb..ebab0921f 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Benchmark/RowsBench.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Benchmark/RowsBench.php
@@ -4,170 +4,118 @@
 
 namespace Flow\ETL\Tests\Benchmark;
 
-use function Flow\ETL\DSL\{array_to_rows, config, flow_context, ref, string_entry};
-use Flow\ETL\{Row, Rows};
-use PhpBench\Attributes\{BeforeMethods, Groups, Revs};
+use function Flow\ETL\DSL\ref;
+use Flow\ETL\{Row, Rows, Tests\Double\FakeStaticOrdersExtractor};
+use PhpBench\Attributes\{BeforeMethods, Groups};
 
 #[BeforeMethods('setUp')]
-#[Revs(2)]
 #[Groups(['building_blocks'])]
 final class RowsBench
 {
-    private Rows $reducedRows;
-
     private Rows $rows;
 
-    public function setUp() : void
-    {
-        $this->rows = array_to_rows(
-            \array_merge(...\array_map(static fn () : array => [
-                ['id' => 1, 'random' => false, 'text' => null, 'from' => 666],
-                ['id' => 2, 'random' => true, 'text' => null, 'from' => 666],
-                ['id' => 3, 'random' => false, 'text' => null, 'from' => 666],
-                ['id' => 4, 'random' => true, 'text' => null, 'from' => 666],
-                ['id' => 5, 'random' => false, 'text' => null, 'from' => 666],
-            ], \range(0, 10_000))),
-            flow_context(config())->entryFactory(),
-        );
-
-        $this->reducedRows = array_to_rows(
-            \array_merge(...\array_map(static fn () : array => [
-                ['id' => 1, 'random' => false, 'text' => null, 'from' => 666],
-                ['id' => 2, 'random' => true, 'text' => null, 'from' => 666],
-                ['id' => 3, 'random' => false, 'text' => null, 'from' => 666],
-                ['id' => 4, 'random' => true, 'text' => null, 'from' => 666],
-                ['id' => 5, 'random' => false, 'text' => null, 'from' => 666],
-            ], \range(0, 1000))),
-            flow_context(config())->entryFactory(),
-        );
-    }
-
-    public function bench_chunk_10_on_10k() : void
-    {
-        foreach ($this->rows->chunks(10) as $chunk) {
+    private Rows $rows100;
 
-        }
-    }
+    private Rows $rows1k;
 
-    public function bench_diff_left_1k_on_10k() : void
+    public function setUp() : void
     {
-        $this->rows->diffLeft($this->reducedRows);
+        $this->rows = (new FakeStaticOrdersExtractor(10_000))->toRows();
+        $this->rows1k = (new FakeStaticOrdersExtractor(1_000))->toRows();
+        $this->rows100 = (new FakeStaticOrdersExtractor(100))->toRows();
     }
 
-    public function bench_diff_right_1k_on_10k() : void
+    public function bench_chunk_1_000_on_10k() : void
     {
-        $this->rows->diffRight($this->reducedRows);
+        foreach ($this->rows->chunks(1_000) as $chunk) {
+
+        }
     }
 
-    public function bench_drop_1k_on_10k() : void
+    public function bench_diff_left_100_on_1k() : void
     {
-        $this->rows->drop(1000);
+        $this->rows1k->diffLeft($this->rows100);
     }
 
-    public function bench_drop_right_1k_on_10k() : void
+    public function bench_diff_right_100_on_1k() : void
     {
-        $this->rows->dropRight(1000);
+        $this->rows1k->diffRight($this->rows100);
     }
 
-    public function bench_entries_on_10k() : void
+    public function bench_drop_100_on_1k() : void
     {
-        foreach ($this->rows->entries() as $entries) {
-
-        }
+        $this->rows1k->drop(100);
     }
 
-    public function bench_filter_on_10k() : void
+    public function bench_drop_right_10_on_1k() : void
     {
-        $this->rows->filter(fn (Row $row) : bool => $row->valueOf('random') === true);
+        $this->rows1k->dropRight(100);
     }
 
-    public function bench_find_on_10k() : void
+    public function bench_entries_on_1k() : void
     {
-        $this->rows->find(fn (Row $row) : bool => $row->valueOf('random') === true);
+        foreach ($this->rows1k->entries() as $entries) {
+
+        }
     }
 
-    #[Revs(10)]
-    public function bench_find_one_on_10k() : void
+    public function bench_filter_on_1k() : void
     {
-        $this->rows->findOne(fn (Row $row) : bool => $row->valueOf('random') === true);
+        $this->rows1k->filter(fn (Row $row) : bool => $row->valueOf('order_id') === true);
     }
 
-    #[Revs(10)]
-    public function bench_first_on_10k() : void
+    public function bench_find_on_1k() : void
     {
-        $this->rows->first();
+        $this->rows1k->find(fn (Row $row) : bool => $row->valueOf('order_id') === true);
     }
 
-    public function bench_flat_map_on_1k() : void
+    public function bench_find_one_on_1k() : void
     {
-        $this->reducedRows->flatMap(fn (Row $row) : array => [
-            /** @phpstan-ignore-next-line */
-            $row->add(string_entry('name', $row->valueOf('id') . '-name-01')),
-            /** @phpstan-ignore-next-line */
-            $row->add(string_entry('name', $row->valueOf('id') . '-name-02')),
-        ]);
+        $this->rows1k->findOne(fn (Row $row) : bool => $row->valueOf('order_id') === true);
     }
 
-    public function bench_map_on_10k() : void
+    public function bench_first_on_1k() : void
     {
-        $this->rows->map(fn (Row $row) : Row => $row->rename('random', 'whatever'));
+        $this->rows1k->first();
     }
 
-    public function bench_merge_1k_on_10k() : void
+    public function bench_merge_100_on_1k() : void
     {
-        $this->rows->merge($this->reducedRows);
+        $this->rows1k->merge($this->rows100);
     }
 
-    public function bench_partition_by_on_10k() : void
+    public function bench_partition_by_on_1k() : void
     {
-        $this->rows->partitionBy(ref('from'));
+        $this->rows1k->partitionBy(ref('order_id'));
     }
 
-    public function bench_remove_on_10k() : void
+    public function bench_schema_on_1k_identical_rows() : void
     {
-        $this->rows->remove(1001);
+        $this->rows->schema();
     }
 
     public function bench_sort_asc_on_1k() : void
     {
-        $this->reducedRows->sortAscending(ref('random'));
+        $this->rows1k->sortAscending(ref('order_id'));
     }
 
     public function bench_sort_by_on_1k() : void
     {
-        $this->reducedRows->sortBy(ref('random'));
+        $this->rows1k->sortBy(ref('order_id'));
     }
 
     public function bench_sort_desc_on_1k() : void
     {
-        $this->reducedRows->sortDescending(ref('random'));
+        $this->rows1k->sortDescending(ref('order_id'));
     }
 
     public function bench_sort_entries_on_1k() : void
     {
-        $this->reducedRows->sortEntries();
-    }
-
-    public function bench_sort_on_1k() : void
-    {
-        /** @phpstan-ignore-next-line */
-        $this->reducedRows->sort(fn (Row $row, Row $nextRow) : int => $row->valueOf('random') <=> $nextRow->valueOf('random'));
-    }
-
-    #[Revs(10)]
-    public function bench_take_1k_on_10k() : void
-    {
-        $this->rows->take(1000);
-    }
-
-    #[Revs(10)]
-    public function bench_take_right_1k_on_10k() : void
-    {
-        $this->rows->takeRight(1000);
+        $this->rows1k->sortEntries();
     }
 
     public function bench_unique_on_1k() : void
     {
-        $this->rows->unique();
+        $this->rows1k->unique();
     }
 }
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/FakeStaticOrdersExtractor.php b/src/core/etl/tests/Flow/ETL/Tests/Double/FakeStaticOrdersExtractor.php
index aa52f6caa..aa3114e50 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Double/FakeStaticOrdersExtractor.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Double/FakeStaticOrdersExtractor.php
@@ -9,12 +9,13 @@
     float_schema,
     integer_schema,
     list_schema,
+    rows,
     schema,
     string_schema,
     structure_schema,
     uuid_schema};
 use function Flow\Types\DSL\{type_float, type_integer, type_list, type_string, type_structure};
-use Flow\ETL\{Extractor, FlowContext, Schema};
+use Flow\ETL\{Extractor, FlowContext, Row\EntryFactory, Rows, Schema};
 
 final readonly class FakeStaticOrdersExtractor implements Extractor
 {
@@ -107,4 +108,15 @@ public function rawData() : \Generator
             ];
         }
     }
+
+    public function toRows(EntryFactory $entryFactory = new EntryFactory()) : Rows
+    {
+        $rows = rows();
+
+        foreach ($this->rawData() as $row) {
+            $rows = $rows->merge(array_to_rows($row, entryFactory: $entryFactory, schema: self::schema()));
+        }
+
+        return $rows;
+    }
 }
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/DataFrameTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/DataFrameTest.php
index 1bec62a43..4f77b588c 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Unit/DataFrameTest.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/DataFrameTest.php
@@ -444,9 +444,9 @@ public function test_selective_validation_against_schema() : void
             new SelectiveValidator()
         )->fetch();
 
-        self::assertEquals(
-            rows(row(int_entry('id', 1), str_entry('name', 'foo'), bool_entry('active', true)), row(int_entry('id', 2), str_entry('name', null), json_entry('tags', ['foo', 'bar'])), row(int_entry('id', 2), str_entry('name', 'bar'), bool_entry('active', false))),
-            $rows
+        self::assertSame(
+            rows(row(int_entry('id', 1), str_entry('name', 'foo'), bool_entry('active', true)), row(int_entry('id', 2), str_entry('name', null), json_entry('tags', ['foo', 'bar'])), row(int_entry('id', 2), str_entry('name', 'bar'), bool_entry('active', false)))->toArray(),
+            $rows->toArray()
         );
     }
 
@@ -458,9 +458,9 @@ public function test_strict_validation_against_schema() : void
             schema(integer_schema('id', $nullable = false), string_schema('name', $nullable = true), bool_schema('active', $nullable = false))
         )->fetch();
 
-        self::assertEquals(
-            rows(row(int_entry('id', 1), str_entry('name', 'foo'), bool_entry('active', true)), row(int_entry('id', 2), str_entry('name', null), bool_entry('active', false)), row(int_entry('id', 2), str_entry('name', 'bar'), bool_entry('active', false))),
-            $rows
+        self::assertSame(
+            rows(row(int_entry('id', 1), str_entry('name', 'foo'), bool_entry('active', true)), row(int_entry('id', 2), str_entry('name', null), bool_entry('active', false)), row(int_entry('id', 2), str_entry('name', 'bar'), bool_entry('active', false)))->toArray(),
+            $rows->toArray()
         );
     }
 
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Schema/SchemaTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Schema/SchemaTest.php
index 92c049d95..5e814477b 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Schema/SchemaTest.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Schema/SchemaTest.php
@@ -14,9 +14,163 @@
 use Flow\ETL\Schema;
 use Flow\ETL\Schema\Metadata;
 use Flow\ETL\Tests\FlowTestCase;
+use PHPUnit\Framework\Attributes\DataProvider;
 
 final class SchemaTest extends FlowTestCase
 {
+    public static function provide_is_same_cases() : \Generator
+    {
+        yield 'identical simple schemas' => [
+            schema(int_schema('id'), str_schema('name')),
+            schema(int_schema('id'), str_schema('name')),
+            true,
+        ];
+
+        yield 'different column count' => [
+            schema(int_schema('id'), str_schema('name')),
+            schema(int_schema('id')),
+            false,
+        ];
+
+        yield 'different column names' => [
+            schema(int_schema('id'), str_schema('name')),
+            schema(int_schema('id'), str_schema('surname')),
+            false,
+        ];
+
+        yield 'different column types' => [
+            schema(int_schema('id'), str_schema('name')),
+            schema(int_schema('id'), int_schema('name')),
+            false,
+        ];
+
+        yield 'different nullable flags' => [
+            schema(int_schema('id'), str_schema('name', nullable: false)),
+            schema(int_schema('id'), str_schema('name', nullable: true)),
+            false,
+        ];
+
+        yield 'different metadata' => [
+            schema(int_schema('id', metadata: Metadata::fromArray(['foo' => 'bar'])), str_schema('name')),
+            schema(int_schema('id', metadata: Metadata::fromArray(['foo' => 'baz'])), str_schema('name')),
+            false,
+        ];
+
+        yield 'empty schemas' => [
+            schema(),
+            schema(),
+            true,
+        ];
+
+        yield 'identical nested structure schemas' => [
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'city' => type_string()]))),
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'city' => type_string()]))),
+            true,
+        ];
+
+        yield 'different nested structure field types' => [
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'city' => type_string()]))),
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'city' => type_integer()]))),
+            false,
+        ];
+
+        yield 'different nested structure field names' => [
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'city' => type_string()]))),
+            schema(structure_schema('address', type_structure(['street' => type_string(), 'town' => type_string()]))),
+            false,
+        ];
+
+        yield 'identical list schemas' => [
+            schema(list_schema('tags', type_list(type_string()))),
+            schema(list_schema('tags', type_list(type_string()))),
+            true,
+        ];
+
+        yield 'different list element types' => [
+            schema(list_schema('tags', type_list(type_string()))),
+            schema(list_schema('tags', type_list(type_integer()))),
+            false,
+        ];
+
+        yield 'identical map schemas' => [
+            schema(map_schema('metadata', type_map(type_string(), type_integer()))),
+            schema(map_schema('metadata', type_map(type_string(), type_integer()))),
+            true,
+        ];
+
+        yield 'different map key types' => [
+            schema(map_schema('metadata', type_map(type_string(), type_integer()))),
+            schema(map_schema('metadata', type_map(type_integer(), type_integer()))),
+            false,
+        ];
+
+        yield 'different map value types' => [
+            schema(map_schema('metadata', type_map(type_string(), type_integer()))),
+            schema(map_schema('metadata', type_map(type_string(), type_string()))),
+            false,
+        ];
+
+        yield 'identical map of list of structure' => [
+            schema(map_schema('complex', type_map(
+                type_string(),
+                type_list(type_structure(['id' => type_integer(), 'name' => type_string()]))
+            ))),
+            schema(map_schema('complex', type_map(
+                type_string(),
+                type_list(type_structure(['id' => type_integer(), 'name' => type_string()]))
+            ))),
+            true,
+        ];
+
+        yield 'different nested element in map of list of structure' => [
+            schema(map_schema('complex', type_map(
+                type_string(),
+                type_list(type_structure(['id' => type_integer(), 'name' => type_string()]))
+            ))),
+            schema(map_schema('complex', type_map(
+                type_string(),
+                type_list(type_structure(['id' => type_integer(), 'name' => type_integer()]))
+            ))),
+            false,
+        ];
+
+        yield 'deeply nested structure' => [
+            schema(structure_schema('root', type_structure([
+                'level1' => type_structure([
+                    'level2' => type_structure([
+                        'value' => type_string(),
+                    ]),
+                ]),
+            ]))),
+            schema(structure_schema('root', type_structure([
+                'level1' => type_structure([
+                    'level2' => type_structure([
+                        'value' => type_string(),
+                    ]),
+                ]),
+            ]))),
+            true,
+        ];
+
+        yield 'different deeply nested structure' => [
+            schema(structure_schema('root', type_structure([
+                'level1' => type_structure([
+                    'level2' => type_structure([
+                        'value' => type_string(),
+                    ]),
+                ]),
+            ]))),
+            schema(structure_schema('root', type_structure([
+                'level1' => type_structure([
+                    'level2' => type_structure([
+                        'value' => type_integer(),
+                    ]),
+                ]),
+            ]))),
+            false,
+        ];
+    }
+
     public function test_add_metadata() : void
     {
         $schema = schema(
@@ -124,6 +278,12 @@ public function test_graceful_remove_non_existing_definition() : void
         );
     }
 
+    #[DataProvider('provide_is_same_cases')]
+    public function test_is_same(Schema $schema1, Schema $schema2, bool $expected) : void
+    {
+        self::assertSame($expected, $schema1->isSame($schema2));
+    }
+
     public function test_keep_non_existing_entries() : void
     {
         $this->expectException(SchemaDefinitionNotFoundException::class);
@@ -164,6 +324,23 @@ public function test_making_whole_schema_nullable() : void
         );
     }
 
+    public function test_merge_returns_self_when_schemas_are_identical() : void
+    {
+        $schema1 = schema(
+            int_schema('id'),
+            str_schema('name'),
+        );
+
+        $schema2 = schema(
+            int_schema('id'),
+            str_schema('name'),
+        );
+
+        $merged = $schema1->merge($schema2);
+
+        self::assertSame($schema1, $merged);
+    }
+
     public function test_normalizing_and_recreating_schema() : void
     {
         $schema = schema(