Dbal bulk data optimization (#1776)

* Convert bulk values to dbal values using Types instead custom BulkData logic

* Create a mechanism allowing to override or pass target table column types to Bulk

* Static analysis

* Updated documentation

* Updated dependecies

Removed problematic column from Dbal Bulk insert integration tests, due
to differences in how different versions of dbal are handling jsonb
columns.

* Removed more fragile tests

Author: norberttech

composer.lock

@@ -29,20 +29,115 @@ Adapter for [ETL](https://github.com/flow-php/etl) using bulk operations from [D
 ## Loader - DbalLoader
 
 ```php
+use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
+
+data_frame()
+    ->read(from_())
+    ->write(to_dbal_table_insert(['url' => \getenv('PGSQL_DATABASE_URL')], 'your-table-name'))
+    ->run();
+```
+
+All supported DbalLoader operations via DSL functions:
+
+- `to_dbal_table_insert(array|Connection $connection, string $table, ?InsertOptions $options = null)` - Insert new rows
+- `to_dbal_table_update(array|Connection $connection, string $table, ?UpdateOptions $options = null)` - Update existing rows
+- `to_dbal_table_delete(array|Connection $connection, string $table)` - Delete rows
+
+You can also configure bulk operations with platform-specific options:
+
+```php
+use function Flow\ETL\Adapter\Doctrine\{to_dbal_table_insert, postgresql_insert_options};
+
+data_frame()
+    ->read(from_())
+    ->write(to_dbal_table_insert(
+        $connection,
+        'users',
+        postgresql_insert_options(conflict_columns: ['id'])
+    ))
+    ->run();
+```
+
+### Type Detection and Optimization
+
+`DbalLoader` now provides advanced type detection capabilities to optimize database operations:
+
+#### Automatic Type Detection from Flow Schema
+
+By default, `DbalLoader` automatically detects column types from the Flow Schema of your data:
+
+```php
+use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
+
+data_frame()
+    ->read(from_())
+    ->write(to_dbal_table_insert($connection, 'users'))
+    ->run();
+// Types are automatically detected from the Flow Schema
+```
+
+#### Manual Type Override
+
+You can override specific column types for fine-grained control:
+
+```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\Types\Type\Native\StringType;
+use Doctrine\DBAL\Types\TextType;
+
+$customTypesMap = new TypesMap([
+    StringType::class => TextType::class, // Map Flow strings to DBAL text type
+]);
+
 data_frame()
     ->read(from_())
-    ->write(new DbalLoader('your-table-name', $bulkSize = 100, ['url' => \getenv('PGSQL_DATABASE_URL')], ['skip_conflicts' => true]))
+    ->write(to_dbal_table_insert($connection, 'users')
+        ->withTypesDetector(new DbalTypesDetector($customTypesMap)))
     ->run();
 ```
 
-All supported types of `DbalBulkLoader` loading:
+#### Data Normalization
 
-- `::insert(Connection $connection, string $table, QueryFactory $queryFactory = null) : self`
-- `::insertOrSkipOnConflict(Connection $connection, string $table, QueryFactory $queryFactory = null) : self`
-- `::insertOrUpdateOnConstraintConflict(Connection $connection, string $table, string $constraint, QueryFactory $queryFactory = null) : self`
+`DbalLoader` automatically handles data normalization for database compatibility:
 
-The `bulkSize` means how many rows you want to push to a database in a single `INSERT` query. Each extracted rows set
-is going to be split before inserting data into the database.
+- **XML Entries**: `XMLEntry` and `XMLElementEntry` objects are automatically converted to their string representation
+- **Complex Types**: Lists, Maps, and Structures are serialized as JSON
+- **Type Safety**: All data is normalized while preserving type information for optimal database performance
+
+```php
+use function Flow\ETL\DSL\{data_frame, from_array, xml_entry};
+use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
+
+data_frame()
+    ->read(from_array([
+        ['id' => 1, 'data' => xml_entry('data', $domDocument)],
+        ['id' => 2, 'data' => xml_entry('data', $domElement)],
+    ]))
+    ->write(to_dbal_table_insert($connection, 'xml_table'))
+    ->run();
+// XML entries are automatically converted to strings before database insertion
+```
 
 ## Extractor - DbalQuery
 
@@ -168,7 +263,7 @@ new Table(
 When types map is not provided, the default one will be used:
 
 ```php
-public const DEFAULT_TYPES = [
+public const FLOW_TYPES = [
     StringType::class => \Doctrine\DBAL\Types\StringType::class,
     IntegerType::class => \Doctrine\DBAL\Types\IntegerType::class,
     FloatType::class => \Doctrine\DBAL\Types\FloatType::class,
@@ -184,4 +279,23 @@ public const DEFAULT_TYPES = [
     MapType::class => \Doctrine\DBAL\Types\JsonType::class,
     StructureType::class => \Doctrine\DBAL\Types\JsonType::class,
 ];
+```
+
+The `TypesMap` class provides bidirectional mapping between Flow types and Doctrine DBAL types, allowing for flexible type conversion in both directions:
+
+```php
+use Flow\ETL\Adapter\Doctrine\TypesMap;
+use Flow\Types\Type\Native\StringType;
+use Doctrine\DBAL\Types\TextType;
+
+// Create custom type mapping
+$customMap = new TypesMap([
+    StringType::class => TextType::class,
+]);
+
+// Convert Flow type to DBAL type
+$dbalType = $customMap->toDbalType(StringType::class);
+
+// Convert DBAL type to Flow type instance
+$flowType = $customMap->toFlowType(TextType::class);
 ```

documentation/components/adapters/doctrine.md

@@ -37,6 +37,39 @@ $bulk->insert(
 
 ```
 
+### Type Optimization with BulkData
+
+`BulkData` can accept optional type information to optimize database operations:
+
+```php
+use Doctrine\DBAL\Types\Type;
+use Doctrine\DBAL\Types\Types;
+
+$bulk = Bulk::create();
+$bulk->insert(
+    $dbalConnection,
+    'your-table-name',
+    new BulkData(
+        [
+            ['id' => 1, 'name' => 'Name One', 'created_at' => new \DateTime()],
+            ['id' => 2, 'name' => 'Name Two', 'created_at' => new \DateTime()],
+        ],
+        [
+            'id' => Type::getType(Types::INTEGER),
+            'name' => Type::getType(Types::STRING),
+            'created_at' => Type::getType(Types::DATETIME_IMMUTABLE),
+        ]
+    )
+);
+```
+
+**Type Detection Behavior:**
+
+- **With Types**: When column types are provided, `BulkData` uses them directly for optimal performance
+- **Without Types**: When no types are provided, `BulkData` automatically queries the database to understand the table structure and determine appropriate types
+
+This automatic type detection ensures data consistency but comes with a performance cost due to the additional database query. For optimal performance in high-throughput scenarios, explicitly providing column types is recommended.
+
 Update:
 
 ```php

documentation/components/libs/doctrine-dbal-bulk.md

@@ -5,18 +5,28 @@
 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};
 
 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;
+
     /**
      * @param array<string, mixed> $connectionParams
      */
@@ -55,14 +65,28 @@ public static function fromConnection(
 
     public function load(Rows $rows, FlowContext $context) : void
     {
-        Bulk::create()->{$this->operation}(
+        $normalizedData = (new RowsNormalizer())->normalize($rows->sortEntries());
+
+        $this->bulk()->{$this->operation}(
             $this->connection(),
             $this->tableName,
-            new BulkData($rows->sortEntries()->toArray()),
+            new BulkData($normalizedData, $this->typesDetector()->convert($rows->schema(), $this->columnTypes ?? [])),
             $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
      */
@@ -84,6 +108,25 @@ public function withOperationOptions(InsertOptions|UpdateOptions|null $operation
         return $this;
     }
 
+    /**
+     * Set custom SchemaToTypesConverter with custom TypesMap.
+     */
+    public function withTypesDetector(DbalTypesDetector $detector) : self
+    {
+        $this->typesDetector = $detector;
+
+        return $this;
+    }
+
+    private function bulk() : Bulk
+    {
+        if ($this->bulk === null) {
+            $this->bulk = Bulk::create();
+        }
+
+        return $this->bulk;
+    }
+
     private function connection() : Connection
     {
         if ($this->connection === null) {
@@ -93,4 +136,15 @@ private function connection() : Connection
 
         return $this->connection;
     }
+
+    private function typesDetector() : DbalTypesDetector
+    {
+        if ($this->typesDetector !== null) {
+            return $this->typesDetector;
+        }
+
+        $this->typesDetector = new DbalTypesDetector();
+
+        return $this->typesDetector;
+    }
 }