ETL Adapter: Doctrine

Table of Contents

Installation
Description
Loader - DbalLoader
- Types
  - Custom Types Map
  - Bound values
Transactional Loading
Extractor - DbalQuery
Extractor - Keyset Pagination
Schema Converter
- Schema Converter - Types Map

The Doctrine adapter reads and writes database tables through Doctrine DBAL in ETL (Extract, Transform, Load) pipelines.

Installation

For the installation instructions, read the installation page.

Description

Adapter for ETL using bulk operations from Doctrine Dbal Bulk.

Loader - DbalLoader

use Doctrine\DBAL\Tools\DsnParser;

use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
use function Flow\ETL\DSL\{data_frame, from_array};

$params = (new DsnParser(['postgresql' => 'pdo_pgsql']))->parse(\getenv('PGSQL_DATABASE_URL'));

data_frame()
    ->read(from_array([['id' => 1, 'name' => 'norbert']]))
    ->write(to_dbal_table_insert($params, 'users'))
    ->run();

All supported DbalLoader operations, as 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:

use function Flow\ETL\Adapter\Doctrine\{postgresql_insert_options, to_dbal_table_insert};
use function Flow\ETL\DSL\{data_frame, from_array};

data_frame()
    ->read(from_array([['id' => 1, 'name' => 'norbert']]))
    ->write(to_dbal_table_insert(
        $connection,
        'users',
        postgresql_insert_options(conflict_columns: ['id'])
    ))
    ->run();

Types

By default, DbalLoader binds each column with the DBAL type that the map gives for its Flow type. The default map is in Types Map:

use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
use function Flow\ETL\DSL\{data_frame, from_array};

data_frame()
    ->read(from_array([['id' => 1, 'name' => 'norbert']]))
    ->write(to_dbal_table_insert($connection, 'users'))
    ->run();

Custom Types Map

A custom map replaces the default map. To keep the other types, spread TypesMap::FLOW_TYPES:

use Doctrine\DBAL\Types\TextType;
use Flow\ETL\Adapter\Doctrine\Values\Schema\TypesMap;
use Flow\Types\Type\Native\StringType;

use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
use function Flow\ETL\DSL\{data_frame, from_array};

$customTypesMap = new TypesMap([
    ...TypesMap::FLOW_TYPES,
    StringType::class => TextType::class,
]);

data_frame()
    ->read(from_array([['id' => 1, 'name' => 'norbert']]))
    ->write(to_dbal_table_insert($connection, 'users')
        ->withTypesMap($customTypesMap))
    ->run();

Bound values

A column is bound as its values, and its DBAL type converts them. list, map and structure become JSON through JsonType. xml, xml_element, html and html_element are bound as their markup text:

use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;
use function Flow\ETL\DSL\{data_frame, from_array, int_schema, schema, xml_schema};

data_frame()
    ->read(from_array(
        [['id' => 1, 'data' => '<root><item>1</item></root>']],
        schema(int_schema('id'), xml_schema('data')),
    ))
    ->write(to_dbal_table_insert($connection, 'xml_table'))
    ->run();
// data is bound as '<root><item>1</item></root>'

Transactional Loading

to_dbal_transaction() writes one or more children inside transactions. Each batch of rows is written in its own transaction. If a child throws, the open transaction rolls back:

use function Flow\ETL\Adapter\Doctrine\{to_dbal_table_insert, to_dbal_transaction};
use function Flow\ETL\DSL\{data_frame, from_array};

data_frame()
    ->read(from_array([['id' => 1, 'name' => 'norbert']]))
    ->write(to_dbal_transaction(
        $connection,
        to_dbal_table_insert($connection, 'users'),
        to_dbal_table_insert($connection, 'users_audit'),
    ))
    ->run();

A plain loader child is written directly. A to_transformation(...) child delivers inside the same transaction. The loader of each child must use the same connection as the transaction. Give one live Connection to to_dbal_transaction() and to each child. A loader built from array params (for example to_dbal_table_insert($params, 'users')) opens its own connection and is outside the transaction.

Some branches have blocking operations: to_transformation(), also inside to_branch(), with sortBy(), aggregate(), groupBy()->aggregate(), pivot(), window functions, collect() or join(). For more information, read transformations. These branches buffer the stream and deliver it at the end of the run. to_dbal_transaction() opens one final transaction around that delivery. The full drained stream commits atomically. A failure during the delivery rolls back, and run() throws it.

Inside the transaction, each child runs throw-only. A failure first rolls back the batch. Only after that, the onError() handler of the frame decides about the rest of the run.

To set the isolation level, build the transaction yourself. The level applies to each transaction that opens, also the final transaction. At the end of each transaction, the previous level comes back:

use Doctrine\DBAL\TransactionIsolationLevel;
use Flow\ETL\Adapter\Doctrine\IO\Loader\DbalTransaction;
use Flow\ETL\API\Sink\Transactional;

use function Flow\ETL\Adapter\Doctrine\to_dbal_table_insert;

new Transactional(
    DbalTransaction::fromConnection($connection)->withIsolationLevel(TransactionIsolationLevel::SERIALIZABLE),
    to_dbal_table_insert($connection, 'users'),
);

Extractor - DbalQuery

from_dbal_*() extractors get their schema from the query. ->withSchema() declares a different schema. For how this works and fails, read Doctrine DBAL sources.

Single Query

use function Flow\ETL\Adapter\Doctrine\from_dbal_query;
use function Flow\ETL\DSL\{data_frame, to_output};

data_frame()
    ->read(from_dbal_query($connection, 'SELECT * FROM users ORDER BY id'))
    ->write(to_output())
    ->run();

Single Parametrized Query

use function Flow\ETL\Adapter\Doctrine\from_dbal_query;
use function Flow\ETL\DSL\{data_frame, to_output};

data_frame()
    ->read(from_dbal_query($connection, 'SELECT * FROM users WHERE id = :id', ['id' => 1]))
    ->write(to_output())
    ->run();

Multiple Parametrized Queries

use Flow\ETL\Adapter\Doctrine\IO\Extractor\ParametersSet;

use function Flow\ETL\Adapter\Doctrine\from_dbal_queries;
use function Flow\ETL\DSL\{data_frame, to_output};

data_frame()
    ->read(from_dbal_queries(
        $connection,
        'SELECT * FROM users ORDER BY id LIMIT :limit OFFSET :offset',
        new ParametersSet(
            ['limit' => 2, 'offset' => 0],
            ['limit' => 2, 'offset' => 2],
            ['limit' => 2, 'offset' => 4],
        ),
    ))
    ->write(to_output())
    ->run();

The query runs once for each parameter set, three times here.

Extractor - Keyset Pagination

from_dbal_key_set_qb() reads a query builder page by page. It asks for the rows after the last key that it saw.

The keys must be indexed, non-null and unique together. A single non-unique column such as created_at is not a key. Add the primary key. pagination_key_set() takes the keys from least to most significant, so the last key is the primary sort:

use function Flow\ETL\Adapter\Doctrine\{from_dbal_key_set_qb, pagination_key_asc, pagination_key_set};
use function Flow\ETL\DSL\{data_frame, to_output};

data_frame()
    ->read(from_dbal_key_set_qb(
        $connection,
        $connection->createQueryBuilder()->select('id', 'created_at', 'total')->from('orders'),
        pagination_key_set(pagination_key_asc('id'), pagination_key_asc('created_at')), // ORDER BY created_at, id
    )->withBatchSize(1000))
    ->limit(10_000)
    ->write(to_output())
    ->run();

Before the first page, the extractor sends one check query: the builder plus WHERE (created_at IS NULL) OR (id IS NULL) LIMIT 1. If it finds a row, it throws. On an indexed key, this is an index lookup. On a key without an index, it is one full scan.
Two rows with the same key throw Keyset pagination requires unique keys.
The next page must start at the row that the previous page looked ahead to. If it does not, the read throws. Two causes are possible. The database compares two different values as equal. An example is a case- or accent-insensitive collation such as the MySQL default. Another example is numeric 1.0 and 1.00. Or rows change between pages.
Known limit: two such values on the same page are both read without an error. Nothing is lost.
The query builder must not have an ORDER BY, setMaxResults() or setFirstResult(). Use withMaximum() or DataFrame::limit().
A page shorter than the batch size ends the read without another query.

Schema Converter

The to_dbal_schema_table() function converts a Flow Schema (which represents a dataset) to a Doctrine DBAL Schema Table.

With the metadata in \Flow\ETL\Adapter\Doctrine\Values\Schema\DbalMetadata, you can add more information to the schema, for example length, primary key, index and precision.

use Flow\ETL\Adapter\Doctrine\Values\Schema\DbalMetadata;

use function Flow\ETL\DSL\bool_schema;
use function Flow\ETL\DSL\date_schema;
use function Flow\ETL\DSL\float_schema;
use function Flow\ETL\DSL\int_schema;
use function Flow\ETL\DSL\json_schema;
use function Flow\ETL\DSL\list_schema;
use function Flow\ETL\DSL\map_schema;
use function Flow\ETL\DSL\schema;
use function Flow\ETL\DSL\str_schema;
use function Flow\Types\DSL\type_integer;
use function Flow\Types\DSL\type_list;
use function Flow\Types\DSL\type_map;
use function Flow\Types\DSL\type_string;

$flowSchema = schema(
    int_schema('int', nullable: false, metadata: DbalMetadata::primaryKey('pk_test')),
    str_schema('str', nullable: true, metadata: DbalMetadata::primaryKey('pk_test')),
    str_schema('str_with_length', true, DbalMetadata::length(255)),
    str_schema('str_unique', true, DbalMetadata::indexUnique('idx_str_unique')),
    date_schema('date', nullable: true, metadata: DbalMetadata::index('idx_date')),
    float_schema('float', nullable: true, metadata: DbalMetadata::precision(10)->merge(DbalMetadata::scale(2))),
    bool_schema('bool', nullable: true, metadata: DbalMetadata::default(true)),
    json_schema('json', nullable: true, metadata: DbalMetadata::platformOptions(['jsonb' => true])),
    list_schema('list', type_list(type_integer()), metadata: DbalMetadata::columnDefinition('integer[]')),
    map_schema('map', type_map(type_integer(), type_string()), metadata: DbalMetadata::comment('test comment!')),
);

Convert it to a Doctrine DBAL Schema Table like this:

use function Flow\ETL\Adapter\Doctrine\to_dbal_schema_table;

to_dbal_schema_table($flowSchema, 'test');

This generates:


use Doctrine\DBAL\Schema\Column;
use Doctrine\DBAL\Schema\Index;
use Doctrine\DBAL\Schema\Table;
use Doctrine\DBAL\Types\Type;

new Table(
    'test',
    [
        new Column('int', Type::getType('integer'), ['notnull' => true]),
        new Column('str', Type::getType('string'), ['notnull' => true]),
        new Column('str_with_length', Type::getType('string'), ['notnull' => false, 'length' => 255]),
        new Column('str_unique', Type::getType('string'), ['notnull' => false]),
        new Column('date', Type::getType('date_immutable'), ['notnull' => false]),
        new Column('float', Type::getType('float'), ['notnull' => false, 'precision' => 10, 'scale' => 2]),
        new Column('bool', Type::getType('boolean'), ['notnull' => false, 'default' => true]),
        new Column('json', Type::getType('json'), ['notnull' => false, 'platformOptions' => ['jsonb' => true]]),
        new Column('list', Type::getType('json'), ['notnull' => true, 'columnDefinition' => 'integer[]']),
        new Column('map', Type::getType('json'), ['notnull' => true, 'comment' => 'test comment!']),
    ],
    [
        new Index('pk_test', ['int', 'str'], true, true),
        new Index('idx_date', ['date'], false, false),
        new Index('idx_str_unique', ['str_unique'], true, false),
    ]
);

Schema Converter - Types Map

Without a types map, the converter uses the default map:

public const array FLOW_TYPES = [
    StringType::class => \Doctrine\DBAL\Types\StringType::class,
    IntegerType::class => \Doctrine\DBAL\Types\IntegerType::class,
    FloatType::class => \Doctrine\DBAL\Types\FloatType::class,
    BooleanType::class => \Doctrine\DBAL\Types\BooleanType::class,
    DateType::class => \Doctrine\DBAL\Types\DateImmutableType::class,
    TimeType::class => \Doctrine\DBAL\Types\TimeImmutableType::class,
    DateTimeType::class => \Doctrine\DBAL\Types\DateTimeImmutableType::class,
    UuidType::class => \Doctrine\DBAL\Types\GuidType::class,
    TimeZoneType::class => \Doctrine\DBAL\Types\StringType::class,
    JsonType::class => \Doctrine\DBAL\Types\JsonType::class,
    XMLType::class => \Doctrine\DBAL\Types\StringType::class,
    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,
];

The TypesMap class maps Flow types to Doctrine DBAL types, and Doctrine DBAL types back to Flow types:

use Flow\ETL\Adapter\Doctrine\Values\Schema\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);

Found a typo or an outdated section? Edit this page on GitHub

#ETL Adapter: Doctrine

#Installation

#Description

#Loader - DbalLoader

#Types

#Custom Types Map

#Bound values

#Transactional Loading

#Extractor - DbalQuery

#Single Query

#Single Parametrized Query

#Multiple Parametrized Queries

#Extractor - Keyset Pagination

#Schema Converter

#Schema Converter - Types Map

Built in the open.