Allow to pass schema to doctrine dbal extractors

examples/run.php

@@ -32,7 +32,9 @@
     print "\nExample: {$file->getRelativePathname()}\n";
 
     try {
-        include $file->getRealPath();
+        if ($file->getBasename() === 'code.php') {
+            include $file->getRealPath();
+        }
     } catch (Exception $e) {
         print "Example failed: {$e->getMessage()}\n";
 

examples/topics/data_sink/database/code.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
+use function Flow\ETL\DSL\{data_frame, from_array, overwrite};
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Schema\{Column, Table};
+use Doctrine\DBAL\Types\{Type, Types};
+
+require __DIR__ . '/../../../autoload.php';
+
+$generateOrders = require __DIR__ . '/generate_orders.php';
+
+$connection = DriverManager::getConnection([
+    'path' => __DIR__ . '/output/orders.db',
+    'driver' => 'pdo_sqlite',
+]);
+
+$schemaManager = $connection->createSchemaManager();
+
+if ($schemaManager->tablesExist(['orders'])) {
+    $schemaManager->dropTable('orders');
+}
+
+$schemaManager->createTable(new Table(
+    $table = 'orders',
+    [
+        new Column('order_id', Type::getType(Types::GUID), ['notnull' => true]),
+        new Column('created_at', Type::getType(Types::DATETIME_IMMUTABLE), ['notnull' => true]),
+        new Column('updated_at', Type::getType(Types::DATETIME_IMMUTABLE), ['notnull' => false]),
+        new Column('discount', Type::getType(Types::FLOAT), ['notnull' => false]),
+        new Column('email', Type::getType(Types::STRING), ['notnull' => true, 'length' => 255]),
+        new Column('customer', Type::getType(Types::STRING), ['notnull' => true, 'length' => 255]),
+        new Column('address', Type::getType(Types::JSON), ['notnull' => true]),
+        new Column('notes', Type::getType(Types::JSON), ['notnull' => true]),
+        new Column('items', Type::getType(Types::JSON), ['notnull' => true]),
+    ],
+));
+
+data_frame()
+    ->read(from_array(generateOrders(10)))
+    ->saveMode(overwrite())
+    ->write(
+        to_dbal_table_insert(
+            DriverManager::getConnection([
+                'path' => __DIR__ . '/output/orders.db',
+                'driver' => 'pdo_sqlite',
+            ]),
+            'orders',
+        )
+    )
+    ->run();

examples/topics/data_sink/database/generate_orders.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+function generateOrders(int $count) : Generator
+{
+    $faker = Faker\Factory::create();
+
+    $skus = [
+        ['sku' => 'SKU_0001', 'name' => 'Product 1', 'price' => $faker->randomFloat(2, 0, 500)],
+        ['sku' => 'SKU_0002', 'name' => 'Product 2', 'price' => $faker->randomFloat(2, 0, 500)],
+        ['sku' => 'SKU_0003', 'name' => 'Product 3', 'price' => $faker->randomFloat(2, 0, 500)],
+        ['sku' => 'SKU_0004', 'name' => 'Product 4', 'price' => $faker->randomFloat(2, 0, 500)],
+        ['sku' => 'SKU_0005', 'name' => 'Product 5', 'price' => $faker->randomFloat(2, 0, 500)],
+    ];
+
+    for ($i = 0; $i < $count; $i++) {
+        yield [
+            'order_id' => $faker->uuid,
+            'created_at' => $faker->dateTimeThisYear,
+            'updated_at' => \random_int(0, 1) === 1 ? $faker->dateTimeThisMonth : null,
+            'discount' => \random_int(0, 1) === 1 ? $faker->randomFloat(2, 0, 50) : null,
+            'email' => $faker->email,
+            'customer' => $faker->firstName . ' ' . $faker->lastName,
+            'address' => [
+                'street' => $faker->streetAddress,
+                'city' => $faker->city,
+                'zip' => $faker->postcode,
+                'country' => $faker->country,
+            ],
+            'notes' => \array_map(
+                static fn ($i) => $faker->sentence,
+                \range(1, $faker->numberBetween(1, 5))
+            ),
+            'items' => \array_map(
+                static fn (int $index) => [
+                    'sku' => $skus[$skuIndex = $faker->numberBetween(1, 4)]['sku'],
+                    'quantity' => $faker->numberBetween(1, 10),
+                    'price' => $skus[$skuIndex]['price'],
+                ],
+                \range(1, $faker->numberBetween(1, 4))
+            ),
+        ];
+    }
+}

examples/topics/data_sink/database/output/.gitignore

@@ -0,0 +1 @@
+orders.db

examples/topics/data_source/database/code.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+use function Flow\ETL\Adapter\Doctrine\from_dbal_limit_offset;
+use function Flow\ETL\DSL\{data_frame, to_stream};
+use Doctrine\DBAL\DriverManager;
+use Flow\ETL\Adapter\Doctrine\{Order, OrderBy};
+
+require __DIR__ . '/../../../autoload.php';
+
+$connection = DriverManager::getConnection([
+    'path' => __DIR__ . '/input/orders.db',
+    'driver' => 'pdo_sqlite',
+]);
+
+data_frame()
+    ->read(
+        from_dbal_limit_offset(
+            $connection,
+            'orders',
+            new OrderBy('created_at', Order::DESC),
+        )
+    )
+    ->collect()
+    ->write(to_stream(__DIR__ . '/output.txt', truncate: false))
+    ->run();

examples/topics/data_source/database/description.md

@@ -0,0 +1,34 @@
+Read data from Database through Doctrine DBAL
+
+The example below shows how to read from a single table using limit/offset extractor.  
+To read data from more advanced queries, you can use one of the following extractors:
+
+```php
+// read from a single table using limit/offset pagination
+from_dbal_limit_offset(
+    $connection,
+    'orders',
+    new OrderBy('created_at', Order::DESC),
+)
+// read from a query using limit/offset pagination        
+from_dbal_limit_offset_qb(
+    $connection,
+    $connection->createQueryBuilder()->select('*')->from('orders')
+);
+// read from a single query
+from_dbal_query(
+    $connection,
+    'SELECT * FROM orders',
+);
+// read from multiple queries each time using next parameters from the provided set
+from_dbal_queries(
+    $connection,
+    'SELECT * FROM orders LIMIT :limit OFFSET :offset',
+    new \Flow\ETL\Adapter\Doctrine\ParametersSet(
+        ['limit' => 2, 'offset' => 0],
+        ['limit' => 2, 'offset' => 2],
+    )
+);
+```
+
+Additionally, each of them allows setting dataset schema through  `$extractor->withSchema(Schema $schema)` method.