Simplify extractors output

UPGRADE.md

@@ -0,0 +1,119 @@
+# Upgrade Guide
+
+This document provides guidelines for upgrading between versions of Flow PHP. 
+Please follow the instructions for your specific version to ensure a smooth upgrade process.
+
+---
+
+## Upgrading from 0.3.x to 0.4.x
+
+### 1) `ref` expression nullability 
+
+`ref("entry_name")` is no longer returning null when the entry is not found. Instead, it throws an exception.
+The same behavior can be achieved through using a newly introduced `optional` expression: 
+
+Before:
+```php
+<?php
+
+use function Flow\ETL\DSL\optional;
+use function Flow\ETL\DSL\ref;
+
+ref('non_existing_column')->cast('string'); 
+```
+
+After: 
+```php
+<?php
+
+use function Flow\ETL\DSL\optional;
+use function Flow\ETL\DSL\ref;
+
+optional(ref('non_existing_column'))->cast('string');
+// or  
+optional(ref('non_existing_column')->cast('string'));
+```
+
+### 2) Extractors output
+
+Affected extractors: 
+
+* CSV
+* JSON
+* Avro
+* DBAL
+* GoogleSheet
+* Parquet
+* Text
+* XML
+
+Extractors are no longer returning data under an array entry called `row`, thanks to this unpacking row become redundant. 
+
+Before:
+```php
+<?php 
+
+(new Flow())
+    ->read(From::array([['id' => 1, 'array' => ['a' => 1, 'b' => 2, 'c' => 3]]]))
+    ->withEntry('row', ref('row')->unpack())
+    ->renameAll('row.', '')
+    ->drop('row')
+    ->withEntry('array', ref('array')->arrayMerge(lit(['d' => 4])))
+    ->write(To::memory($memory = new ArrayMemory()))
+    ->run();
+```
+
+After: 
+
+```php
+<?php
+
+(new Flow())
+    ->read(From::array([['id' => 1, 'array' => ['a' => 1, 'b' => 2, 'c' => 3]]]))
+    ->withEntry('array', ref('array')->arrayMerge(lit(['d' => 4])))
+    ->write(To::memory($memory = new ArrayMemory()))
+    ->run();
+```
+
+### 3) ConfigBuilder::putInputIntoRows() output is now prefixed with _ (underscore)
+
+In order to avoid collisions with datasets columns, additional columns created after using putInputIntoRows()
+would now be prefixed with `_` (underscore) symbol. 
+
+Before:
+```php
+<?php
+
+$rows = (new Flow(Config::builder()->putInputIntoRows()))
+            ->read(Json::from(__DIR__ . '/../Fixtures/timezones.json', 5))
+            ->fetch();
+
+foreach ($rows as $row) {
+    $this->assertSame(
+        [
+            ...
+            '_input_file_uri',
+        ],
+        \array_keys($row->toArray())
+    );
+}
+```
+
+After: 
+```php
+<?php
+
+$rows = (new Flow(Config::builder()->putInputIntoRows()))
+            ->read(Json::from(__DIR__ . '/../Fixtures/timezones.json', 5))
+            ->fetch();
+
+foreach ($rows as $row) {
+    $this->assertSame(
+        [
+            ...
+            '_input_file_uri',
+        ],
+        \array_keys($row->toArray())
+    );
+}
+```

examples/topics/aggregations/power_plant.php

@@ -15,8 +15,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_DATA__ . '/power-plant-daily.csv', 10, delimiter: ';'))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
     ->withEntry('production_kwh', ref('Produkcja(kWh)'))
     ->withEntry('consumption_kwh', ref('Zużycie(kWh)'))
     ->withEntry('date', ref('Zaktualizowany czas')->toDate('Y/m/d')->dateFormat('Y/m'))

examples/topics/aggregations/power_plant_bar_chart.php

@@ -15,8 +15,6 @@
 
 $flow = (new Flow)
     ->read(CSV::from(__FLOW_DATA__ . '/power-plant-daily.csv', 10, delimiter: ';'))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
     ->withEntry('production_kwh', ref('Produkcja(kWh)'))
     ->withEntry('consumption_kwh', ref('Zużycie(kWh)'))
     ->withEntry('date', ref('Zaktualizowany czas')->toDate('Y/m/d')->dateFormat('Y/m'))

examples/topics/async/csv_to_db_async_amp.php

@@ -41,9 +41,6 @@
             $workers = 8
         )
     )
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last name')

examples/topics/async/csv_to_db_async_react.php

@@ -41,9 +41,6 @@
             $workers = 8
         )
     )
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last name')

examples/topics/async/csv_to_db_sync.php

@@ -27,9 +27,6 @@
 
 (new Flow())
     ->read(CSV::from($path = __FLOW_OUTPUT__ . '/dataset.csv', 10_000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last name')

examples/topics/async/csv_to_json_async_react.php

@@ -42,9 +42,6 @@
             $workers = 8
         )
     )
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last name')

examples/topics/db/db_source.php

@@ -7,7 +7,6 @@
     exit(1);
 }
 
-use function Flow\ETL\DSL\ref;
 use Doctrine\DBAL\DriverManager;
 use Doctrine\DBAL\Schema\Column;
 use Doctrine\DBAL\Schema\Table;
@@ -38,9 +37,6 @@
 
 (new Flow())
     ->read(CSV::from($path = __FLOW_OUTPUT__ . '/dataset.csv', 10_000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->rename('last name', 'last_name')
     ->limit(1_000_000)
     ->load(DbalLoader::fromConnection($dbConnection, 'source_dataset_table', 1000))

examples/topics/db/db_to_db_sync.php

@@ -37,9 +37,6 @@
             new OrderBy('id', Order::DESC)
         )
     )
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last_name')

examples/topics/fs/remote/json_remote_stream.php

@@ -46,9 +46,8 @@
 
 (new Flow())
     ->read(Json::from(new Path('flow-aws-s3://dataset.json', $s3_client_option), 10))
-    ->withEntry('row', ref('row')->unpack())
-    ->withEntry('row.id', ref('row.id')->cast('integer'))
-    ->withEntry('name', concat(ref('row.name'), lit(' '), ref('row.last name')))
-    ->drop('row.last name')
+    ->withEntry('id', ref('id')->cast('integer'))
+    ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
+    ->drop('last name')
     ->write(Json::to(new Path('flow-azure-blob://dataset_test.json', $azure_blob_connection_string)))
     ->run();

examples/topics/fs/remote/json_remote_stream_glob.php

@@ -46,9 +46,6 @@
 
 (new Flow())
     ->read(CSV::from(new Path('flow-aws-s3://nested/**/*.csv', $s3_client_option), 10))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->withEntry('id', ref('id')->cast('int'))
     ->withEntry('name', concat(ref('name'), lit(' '), ref('last name')))
     ->drop('last name')

examples/topics/types/csv/csv_read.php

@@ -2,20 +2,14 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\Flow;
 
 require __DIR__ . '/../../../bootstrap.php';
 
 $flow = (new Flow())
-    ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv', 1000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
-    ->limit(10_000);
+    ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv', 1000));
 
 if ($_ENV['FLOW_PHAR_APP'] ?? false) {
     return $flow;

examples/topics/types/csv/csv_read_partitioned.php

@@ -2,7 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
 use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
@@ -13,9 +12,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_DATA__ . '/partitioned'))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->collect()
     ->sortBy(ref('id'))
     ->write(To::output());

examples/topics/types/csv/csv_read_partitioned_filter.php

@@ -13,9 +13,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_DATA__ . '/partitioned'))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop('row')
     ->collect()
     ->filterPartitions(Partitions::only('t_shirt_color', 'green'))
     ->sortBy(ref('id'))

examples/topics/types/csv/csv_to_avro.php

@@ -2,8 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\Avro;
 use Flow\ETL\DSL\CSV;
@@ -13,9 +11,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv', 10_000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->rename('last name', 'last_name')
     ->write(Avro::to(__FLOW_OUTPUT__ . '/dataset.avro'));
 

examples/topics/types/csv/csv_to_json.php

@@ -1,7 +1,5 @@
 <?php declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\DSL\Json;
 use Flow\ETL\Flow;
@@ -14,9 +12,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv'))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->write(Json::to(__FLOW_OUTPUT__ . '/dataset.json'));
 
 if ($_ENV['FLOW_PHAR_APP'] ?? false) {

examples/topics/types/csv/csv_to_parquet_100k.php

@@ -2,8 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\DSL\Parquet;
@@ -13,9 +11,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv', 10_000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->write(Parquet::to(__FLOW_OUTPUT__ . '/dataset_100k.parquet', 100_000));
 
 if ($_ENV['FLOW_PHAR_APP'] ?? false) {

examples/topics/types/csv/csv_to_parquet_10k.php

@@ -2,8 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\DSL\Parquet;
@@ -13,9 +11,6 @@
 
 $flow = (new Flow())
     ->read(CSV::from(__FLOW_OUTPUT__ . '/dataset.csv', 10_000))
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->write(Parquet::to(__FLOW_OUTPUT__ . '/dataset_10k.parquet', 10_000));
 
 if ($_ENV['FLOW_PHAR_APP'] ?? false) {

examples/topics/types/csv/php_to_csv.php

@@ -2,8 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\Flow;
@@ -15,9 +13,6 @@
 
 $flow = (new Flow())
     ->read($extractor)
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->write(CSV::to(__FLOW_OUTPUT__ . '/dataset.csv'));
 
 if ($_ENV['FLOW_PHAR_APP'] ?? false) {

examples/topics/types/csv/php_to_csv_and_json.php

@@ -2,8 +2,6 @@
 
 declare(strict_types=1);
 
-use function Flow\ETL\DSL\col;
-use function Flow\ETL\DSL\ref;
 use Aeon\Calendar\Stopwatch;
 use Flow\ETL\DSL\CSV;
 use Flow\ETL\DSL\Json;
@@ -16,9 +14,6 @@
 
 $flow = (new Flow())
     ->read($extractor)
-    ->withEntry('unpacked', ref('row')->unpack())
-    ->renameAll('unpacked.', '')
-    ->drop(col('row'))
     ->write(CSV::to(__FLOW_OUTPUT__ . '/dataset.csv'))
     ->write(Json::to(__FLOW_OUTPUT__ . '/dataset.json'));