From 9a9ea014b3c45ad2a42180106b3c3536b70fbbfc Mon Sep 17 00:00:00 2001
From: Aleksander Kowalski <mrhdolek14@gmail.com>
Date: Wed, 17 Jun 2026 08:37:16 +0200
Subject: [PATCH] feat:

---
 .../Tests/Unit/SchemaConverterTest.php        |   9 +
 src/core/etl/src/Flow/ETL/Schema.php          | 221 ++++++++++++++++++
 .../Flow/ETL/Tests/Unit/Schema/SchemaTest.php | 174 ++++++++++++++
 3 files changed, 404 insertions(+)

diff --git a/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Unit/SchemaConverterTest.php b/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Unit/SchemaConverterTest.php
index 7ec0ccc044..3a27977afe 100644
--- a/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Unit/SchemaConverterTest.php
+++ b/src/adapter/etl-adapter-postgresql/tests/Flow/ETL/Adapter/PostgreSql/Tests/Unit/SchemaConverterTest.php
@@ -271,6 +271,15 @@ public function test_precision_and_scale_produce_numeric(): void
         static::assertTrue($table->column('amount')->type->isEqual(ColumnType::numeric(10, 2)));
     }
 
+    public function test_prepend_drives_create_table_column_order(): void
+    {
+        $schema = schema(str_schema('name'), str_schema('email'))->prepend(int_schema('id'));
+
+        $table = (new SchemaConverter())->toPostgreSqlTable($schema, 'users');
+
+        static::assertSame(['id', 'name', 'email'], $table->columnNames());
+    }
+
     public function test_primary_key_forces_not_null(): void
     {
         $table = (new SchemaConverter())->toPostgreSqlTable(
diff --git a/src/core/etl/src/Flow/ETL/Schema.php b/src/core/etl/src/Flow/ETL/Schema.php
index ac9bf75e19..2a8b99e88b 100644
--- a/src/core/etl/src/Flow/ETL/Schema.php
+++ b/src/core/etl/src/Flow/ETL/Schema.php
@@ -17,13 +17,17 @@
 use function array_key_exists;
 use function array_map;
 use function array_merge;
+use function array_splice;
 use function array_values;
 use function count;
 use function Flow\ETL\DSL\definition_from_array;
 use function Flow\ETL\DSL\schema;
 use function implode;
 use function is_array;
+use function is_int;
 use function sprintf;
+use function str_starts_with;
+use function substr;
 
 final class Schema implements Countable
 {
@@ -121,6 +125,68 @@ public function add(Definition ...$definitions): self
         return $this;
     }
 
+    /**
+     * Inserts new definitions right after an existing one.
+     *
+     * @param Definition<mixed> ...$definitions
+     *
+     * @throws SchemaDefinitionNotFoundException
+     *
+     * @return Schema
+     */
+    public function addAfter(string|Reference $reference, Definition ...$definitions): self
+    {
+        $this->get($reference);
+
+        $target = EntryReference::init($reference);
+        $result = [];
+
+        foreach (array_values($this->definitions) as $definition) {
+            $result[] = $definition;
+
+            if ($definition->entry()->is($target)) {
+                foreach ($definitions as $new) {
+                    $result[] = $new;
+                }
+            }
+        }
+
+        $this->setDefinitions(...$result);
+
+        return $this;
+    }
+
+    /**
+     * Inserts new definitions right before an existing one.
+     *
+     * @param Definition<mixed> ...$definitions
+     *
+     * @throws SchemaDefinitionNotFoundException
+     *
+     * @return Schema
+     */
+    public function addBefore(string|Reference $reference, Definition ...$definitions): self
+    {
+        $this->get($reference);
+
+        $target = EntryReference::init($reference);
+        $result = [];
+
+        foreach (array_values($this->definitions) as $definition) {
+            if ($definition->entry()->is($target)) {
+                foreach ($definitions as $new) {
+                    $result[] = $new;
+                }
+            }
+
+            $result[] = $definition;
+        }
+
+        $this->setDefinitions(...$result);
+
+        return $this;
+    }
+
     /**
      * Adds metadata to a given definition.
      *
@@ -200,6 +266,34 @@ public function gracefulRemove(string|Reference ...$entries): self
         return $this;
     }
 
+    /**
+     * Inserts new definitions at an explicit position (0 = beginning, count() = end).
+     *
+     * @param Definition<mixed> ...$definitions
+     *
+     * @throws InvalidArgumentException
+     *
+     * @return Schema
+     */
+    public function insertAt(int $index, Definition ...$definitions): self
+    {
+        $current = array_values($this->definitions);
+
+        if ($index < 0 || $index > count($current)) {
+            throw InvalidArgumentException::because(
+                'Cannot insert definitions at index %d, schema has %d definition(s)',
+                $index,
+                count($current),
+            );
+        }
+
+        array_splice($current, $index, 0, $definitions);
+
+        $this->setDefinitions(...$current);
+
+        return $this;
+    }
+
     public function isSame(self $schema): bool
     {
         if (count($this->definitions) !== count($schema->definitions)) {
@@ -300,6 +394,89 @@ public function merge(self $schema): self
         return $this;
     }
 
+    /**
+     * Relocates an existing definition (preserving its metadata) to a new position.
+     * The position is either a numeric index (the final position after removal; use count() - 1 for
+     * the end) or a string anchor prefixed with "before:" / "after:" referencing another column.
+     *
+     * @throws InvalidArgumentException
+     * @throws SchemaDefinitionNotFoundException
+     *
+     * @return Schema
+     */
+    public function move(string|Reference $name, int|string $position): self
+    {
+        $definition = $this->get($name);
+        $movedName = $definition->entry()->name();
+
+        $remaining = [];
+
+        foreach (array_values($this->definitions) as $next) {
+            if ($next->entry()->name() !== $movedName) {
+                $remaining[] = $next;
+            }
+        }
+
+        if (is_int($position)) {
+            if ($position < 0 || $position > count($remaining)) {
+                throw InvalidArgumentException::because(
+                    'Cannot move "%s" to index %d, schema has %d definition(s)',
+                    $movedName,
+                    $position,
+                    count($this->definitions),
+                );
+            }
+
+            array_splice($remaining, $position, 0, [$definition]);
+
+            $this->setDefinitions(...$remaining);
+
+            return $this;
+        }
+
+        if (str_starts_with($position, 'before:')) {
+            $anchor = EntryReference::init(substr($position, 7));
+            $before = true;
+        } elseif (str_starts_with($position, 'after:')) {
+            $anchor = EntryReference::init(substr($position, 6));
+            $before = false;
+        } else {
+            throw InvalidArgumentException::because(
+                'Move position must be an integer index or a string prefixed with "before:" or "after:", given: "%s"',
+                $position,
+            );
+        }
+
+        if ($anchor->name() === $movedName) {
+            throw InvalidArgumentException::because('Cannot move "%s" relative to itself', $movedName);
+        }
+
+        $result = [];
+        $found = false;
+
+        foreach ($remaining as $next) {
+            if ($before && $next->entry()->is($anchor)) {
+                $result[] = $definition;
+                $found = true;
+            }
+
+            $result[] = $next;
+
+            if (!$before && $next->entry()->is($anchor)) {
+                $result[] = $definition;
+                $found = true;
+            }
+        }
+
+        if (!$found) {
+            throw new SchemaDefinitionNotFoundException($anchor->name());
+        }
+
+        $this->setDefinitions(...$result);
+
+        return $this;
+    }
+
     /**
      * @return array<array-key, array<mixed>>
      */
@@ -314,6 +491,20 @@ public function normalize(): array
         return $definitions;
     }
 
+    /**
+     * Inserts new definitions at the beginning of the schema.
+     *
+     * @param Definition<mixed> ...$definitions
+     *
+     * @return Schema
+     */
+    public function prepend(Definition ...$definitions): self
+    {
+        $this->setDefinitions(...array_merge($definitions, array_values($this->definitions)));
+
+        return $this;
+    }
+
     public function references(): References
     {
         $refs = [];
@@ -375,6 +566,36 @@ public function rename(string|Reference $entry, string $newName): self
         return $this;
     }
 
+    /**
+     * Reorders definitions by name. Any unlisted definitions keep their relative order and are
+     * appended after the listed ones.
+     *
+     * @throws SchemaDefinitionNotFoundException
+     *
+     * @return Schema
+     */
+    public function reorder(string|Reference ...$names): self
+    {
+        $ordered = [];
+        $reordered = [];
+
+        foreach ($names as $name) {
+            $definition = $this->get($name);
+            $ordered[] = $definition;
+            $reordered[$definition->entry()->name()] = true;
+        }
+
+        foreach (array_values($this->definitions) as $definition) {
+            if (!array_key_exists($definition->entry()->name(), $reordered)) {
+                $ordered[] = $definition;
+            }
+        }
+
+        $this->setDefinitions(...$ordered);
+
+        return $this;
+    }
+
     /**
      * @param Definition<mixed> $definition
      *
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 4c8435e2b3..ba84b97a2a 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
@@ -192,6 +192,43 @@ public static function provide_is_same_cases(): Generator
         ];
     }
 
+    public function test_add_after_definitions(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('email'))->addAfter('id', str_schema('name'));
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+        static::assertEquals(schema(int_schema('id'), str_schema('name'), str_schema('email')), $schema);
+    }
+
+    public function test_add_after_non_existing_reference(): void
+    {
+        $this->expectException(SchemaDefinitionNotFoundException::class);
+
+        schema(int_schema('id'))->addAfter('not-existing', str_schema('name'));
+    }
+
+    public function test_add_before_definitions(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('email'))->addBefore('email', str_schema('name'));
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+        static::assertEquals(schema(int_schema('id'), str_schema('name'), str_schema('email')), $schema);
+    }
+
+    public function test_add_before_duplicate(): void
+    {
+        $this->expectException(SchemaDefinitionNotUniqueException::class);
+
+        schema(int_schema('id'), str_schema('name'))->addBefore('name', int_schema('id'));
+    }
+
+    public function test_add_before_non_existing_reference(): void
+    {
+        $this->expectException(SchemaDefinitionNotFoundException::class);
+
+        schema(int_schema('id'))->addBefore('not-existing', str_schema('name'));
+    }
+
     public function test_add_metadata(): void
     {
         $schema = schema(int_schema('id'), str_schema('name'));
@@ -274,6 +311,36 @@ public function test_graceful_remove_non_existing_definition(): void
         );
     }
 
+    public function test_insert_at_appends_when_index_equals_count(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('name'))->insertAt(2, bool_schema('active'));
+
+        static::assertSame(['id', 'name', 'active'], array_keys($schema->definitions()));
+    }
+
+    public function test_insert_at_index(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('email'))->insertAt(1, str_schema('name'));
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+        static::assertEquals(schema(int_schema('id'), str_schema('name'), str_schema('email')), $schema);
+    }
+
+    public function test_insert_at_out_of_range(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Cannot insert definitions at index 5, schema has 2 definition(s)');
+
+        schema(int_schema('id'), str_schema('name'))->insertAt(5, bool_schema('active'));
+    }
+
+    public function test_insert_at_prepends_when_index_zero(): void
+    {
+        $schema = schema(str_schema('name'), str_schema('email'))->insertAt(0, int_schema('id'));
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+    }
+
     #[DataProvider('provide_is_same_cases')]
     public function test_is_same(Schema $schema1, Schema $schema2, bool $expected): void
     {
@@ -312,6 +379,67 @@ public function test_merge_returns_self_when_schemas_are_identical(): void
         static::assertSame($schema1, $merged);
     }
 
+    public function test_move_after_reference(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('name'), str_schema('email'))->move('email', 'after:id');
+
+        static::assertSame(['id', 'email', 'name'], array_keys($schema->definitions()));
+    }
+
+    public function test_move_before_reference(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('name'), str_schema('email'))->move('email', 'before:name');
+
+        static::assertSame(['id', 'email', 'name'], array_keys($schema->definitions()));
+    }
+
+    public function test_move_invalid_position_string(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage(
+            'Move position must be an integer index or a string prefixed with "before:" or "after:", given: "name"',
+        );
+
+        schema(int_schema('id'), str_schema('name'))->move('id', 'name');
+    }
+
+    public function test_move_non_existing_column(): void
+    {
+        $this->expectException(SchemaDefinitionNotFoundException::class);
+
+        schema(int_schema('id'), str_schema('name'))->move('not-existing', 0);
+    }
+
+    public function test_move_non_existing_reference(): void
+    {
+        $this->expectException(SchemaDefinitionNotFoundException::class);
+
+        schema(int_schema('id'), str_schema('name'))->move('name', 'after:not-existing');
+    }
+
+    public function test_move_out_of_range_index(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Cannot move "id" to index 5, schema has 2 definition(s)');
+
+        schema(int_schema('id'), str_schema('name'))->move('id', 5);
+    }
+
+    public function test_move_relative_to_itself(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Cannot move "id" relative to itself');
+
+        schema(int_schema('id'), str_schema('name'))->move('id', 'before:id');
+    }
+
+    public function test_move_to_index(): void
+    {
+        $schema = schema(str_schema('name'), str_schema('email'), int_schema('id'))->move('id', 0);
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+    }
+
     public function test_normalizing_and_recreating_schema(): void
     {
         $schema = schema(
@@ -330,6 +458,21 @@ public function test_normalizing_and_recreating_schema(): void
         static::assertEquals($schema, Schema::fromArray($schema->normalize()));
     }
 
+    public function test_prepend_definitions(): void
+    {
+        $schema = schema(str_schema('name'), str_schema('email'))->prepend(int_schema('id'));
+
+        static::assertSame(['id', 'name', 'email'], array_keys($schema->definitions()));
+        static::assertEquals(schema(int_schema('id'), str_schema('name'), str_schema('email')), $schema);
+    }
+
+    public function test_prepend_duplicate(): void
+    {
+        $this->expectException(SchemaDefinitionNotUniqueException::class);
+
+        schema(int_schema('id'), str_schema('name'))->prepend(int_schema('id'));
+    }
+
     public function test_remove_non_existing_definition(): void
     {
         $this->expectException(SchemaDefinitionNotFoundException::class);
@@ -356,6 +499,37 @@ public function test_rename_non_existing(): void
         schema(int_schema('id'), str_schema('name'))->rename('not-existing', 'new_name');
     }
 
+    public function test_reorder_all_columns(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('name'), str_schema('email'))->reorder('email', 'id', 'name');
+
+        static::assertSame(['email', 'id', 'name'], array_keys($schema->definitions()));
+    }
+
+    public function test_reorder_duplicate_name(): void
+    {
+        $this->expectException(SchemaDefinitionNotUniqueException::class);
+
+        schema(int_schema('id'), str_schema('name'))->reorder('id', 'id');
+    }
+
+    public function test_reorder_non_existing(): void
+    {
+        $this->expectException(SchemaDefinitionNotFoundException::class);
+
+        schema(int_schema('id'), str_schema('name'))->reorder('not-existing');
+    }
+
+    public function test_reorder_partial_keeps_unlisted_appended(): void
+    {
+        $schema = schema(int_schema('id'), str_schema('name'), str_schema('email'), bool_schema('active'))->reorder(
+            'email',
+            'id',
+        );
+
+        static::assertSame(['email', 'id', 'name', 'active'], array_keys($schema->definitions()));
+    }
+
     public function test_replace_non_existing_reference(): void
     {
         $this->expectException(SchemaDefinitionNotFoundException::class);