Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions documentation/components/adapters/parquet.md
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,7 @@ Flow `float` columns are written as Parquet `DOUBLE`.

## Schema

The file footer is the schema: `from_parquet(...)->withSchema()` throws. Project with `columns:`, change types after
reading:
The file footer is the schema. Project with `columns:`, change types after reading:

```php
<?php
Expand Down Expand Up @@ -91,6 +90,28 @@ use Flow\Parquet\Writer;
);
```

### Declared schema

```php
<?php

use function Flow\ETL\Adapter\Parquet\from_parquet;
use function Flow\ETL\DSL\{data_frame, int_schema, schema, str_schema, to_output};

data_frame()
->read(from_parquet(__DIR__ . '/orders/**/*.parquet')->withSchema(schema(
int_schema('id'),
str_schema('name', nullable: true),
)))
->select('id')
->write(to_output())
->run();
```

- A glob that matches no file reads 0 rows; without `withSchema()` it throws `InferredSchemaException`.
- Columns match by name: an undeclared file column is not read, a declared nullable column a file lacks is `null`, a
declared non-nullable one fails.

## Filter pushdown

A `filter()` directly over `from_parquet()` moves into the reader, which returns only matching rows: row groups whose
Expand Down
14 changes: 14 additions & 0 deletions documentation/components/core/core.md
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,20 @@ $dataFrame = data_frame()

> **Note**: Flow PHP supports many data sources through specialized adapters. See individual adapter documentation for specific extractor usage (CSV, JSON, Parquet, databases, APIs, etc.).

An empty source has no column to infer a schema from and throws `InferredSchemaException`. Declare the schema to read it:

```php
<?php

use function Flow\ETL\DSL\{data_frame, from_array, int_schema, schema, str_schema, to_output};

data_frame()
->read(from_array([], schema(int_schema('id'), str_schema('name'))))
->select('id')
->write(to_output())
->run();
```

## Memory Management Best Practices

1. **Prefer Generator Methods**: Use `get()`, `getEachAsArray()` over `fetch()` for large datasets
Expand Down
3 changes: 2 additions & 1 deletion documentation/components/core/floe.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@ data_frame()

Batches hold at most `withBatchSize()` rows, never spanning two BATCH frames.

The footer is the schema: `from_floe(...)->withSchema()` throws. Change a column's type or zone after reading:
The footer is the schema; `withSchema()` reads every file by name against a declared one, as
[Parquet](/documentation/components/adapters/parquet.md) does. Change a column's type or zone after reading:

```php
<?php
Expand Down
4 changes: 2 additions & 2 deletions documentation/components/core/schema.md
Original file line number Diff line number Diff line change
Expand Up @@ -210,8 +210,8 @@ The HTTP extractors accept a schema the same way - see
[Typing the row with a schema](/documentation/components/adapters/http.md) for declaring the response body as a
`structure`.

Self-descriptive formats carry their schema in the file itself, so their extractors read it from there and have no
`withSchema()` - Parquet and Floe in this repository.
Self-descriptive formats carry their schema in the file itself - Parquet and Floe in this repository. Their extractors
read it from there unless `withSchema()` declares one, which reads every file by name.

### PostgreSQL sources describe themselves, and keep `withSchema()`

Expand Down
9 changes: 9 additions & 0 deletions documentation/upgrading.md
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,15 @@ public function scan(FlowContext $context, FileScan $scan): Generator
| Symfony HttpClient `url.full`, `url.scheme`, `server.address`, `server.port` came from the `$url` argument (`/me`, `unknown` with `base_uri`) | they come from the effective URL, `base_uri` and the `query` option included |
| `exception.stacktrace` was `getTraceAsString()`, with argument values when `zend.exception_ignore_args` is off | `StackTrace::format()`: every frame renders `()`, whatever `zend.exception_ignore_args` is |

### 9) `flow-php/etl`, `flow-php/etl-adapter-*` - an empty source without a declared schema fails at the source; Parquet and Floe accept `withSchema()`

| Before | After |
|---------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------|
| an empty source (`from_all()`, `from_array([])`, a glob that matches no file) described an empty schema | `schema()` throws `InferredSchemaException`; declare the schema with `->withSchema(...)` to read an empty source |
| `df()->read(from_array([]))->fetch()` returned 0 rows | throws `InferredSchemaException`; `from_array([], schema: $schema)` returns 0 rows |
| `from_parquet(...)->withSchema()`, `from_floe(...)->withSchema()` threw `InvalidArgumentException` | read every file by name against the declared schema: undeclared columns are not read, a missing nullable column is null |
| `from_cache($id)` without a cache instance described an empty schema | `SchemaNotDerivableException`; the plan runs unbound |

---

## Upgrading from 0.45.x to 0.46.x
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
namespace Flow\ETL\Adapter\CSV;

use Flow\ETL\Cardinality;
use Flow\ETL\Exception\InferredSchemaException;
use Flow\ETL\Exception\InvalidArgumentException;
use Flow\ETL\Extractor;
use Flow\ETL\Extractor\BatchableExtractor;
Expand Down Expand Up @@ -101,7 +102,7 @@ public function isRepeatable(): bool
public function scan(FlowContext $context, FileScan $scan): Generator
{
$fileColumns = $this->fileColumns($this->filesystem, $this->path);
$schema = $this->schema();
$schema = $this->describe();
$sources = iterator_to_array($this->sourceFiles($this->filesystem, $this->path, $scan->pathFilter), false);

yield from (new FileReadLoop($fileColumns, $schema, $scan->conjuncts))->read(
Expand Down Expand Up @@ -134,6 +135,21 @@ public function inferSchema(SchemaInferenceBuilder $builder): static
}

public function schema(): Schema
{
$schema = $this->describe();
$fileColumns = $this->fileColumns($this->filesystem, $this->path);

if ($this->schema === null && $fileColumns->withoutTail($schema)->count() === 0) {
throw InferredSchemaException::nothingToInferFrom('from_csv()', sprintf(
'no file matching "%s" carries a column',
$this->path->uri(),
));
}

return $schema;
}

private function describe(): Schema
{
$fileColumns = $this->fileColumns($this->filesystem, $this->path);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,39 @@ final class CSVExtractorTest extends FlowTestCase
{
use OperatingSystem;

public function test_an_empty_glob_refuses_to_infer_a_schema(): void
{
$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_csv() has nothing to infer a schema from: no file matching "memory://empty/*.csv" carries a column. '
. 'Declare the schema with ->withSchema(...).',
);

from_csv(path('memory://empty/*.csv'), filesystem: memory_filesystem())->schema();
}

public function test_an_empty_glob_with_metadata_columns_refuses_to_infer_a_schema(): void
{
$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_csv() has nothing to infer a schema from: no file matching "memory://empty/*.csv" carries a column. '
. 'Declare the schema with ->withSchema(...).',
);

from_csv(path('memory://empty/*.csv'), filesystem: memory_filesystem())->withMetadataColumns(true)->schema();
}

public function test_an_empty_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void
{
$extractor = from_csv(
path('memory://empty/*.csv'),
filesystem: memory_filesystem(),
)->withSchema(schema(int_schema('id')));

static::assertEquals(schema(int_schema('id')), $extractor->schema());
static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false));
}

public function test_a_hive_null_partition_value_reads_back_as_null(): void
{
$dir = __DIR__ . '/var/test_a_hive_null_partition_value_reads_back_as_null';
Expand Down Expand Up @@ -484,17 +517,32 @@ public function test_a_header_only_file_infers_every_column_as_string(): void
static::assertSame([], iterator_to_array($extractor->extract(flow_context(config()))));
}

public function test_an_empty_file_infers_an_empty_schema(): void
public function test_an_empty_file_refuses_to_infer_a_schema_and_extracts_nothing(): void
{
$extractor = from_csv(CSVFixtureContext::path('empty.csv'));

static::assertSame([], $extractor->schema()->definitions());
static::assertSame([], iterator_to_array($extractor->extract(flow_context(config()))));

$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_csv() has nothing to infer a schema from: no file matching "'
. path_real(CSVFixtureContext::path('empty.csv'))->uri()
. '" carries a column. Declare the schema with ->withSchema(...).',
);

$extractor->schema();
}

public function test_a_glob_of_only_empty_files_infers_an_empty_schema(): void
public function test_a_glob_of_only_empty_files_refuses_to_infer_a_schema(): void
{
static::assertSame([], from_csv(CSVFixtureContext::path('two_empty/*.csv'))->schema()->definitions());
$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_csv() has nothing to infer a schema from: no file matching "'
. path_real(CSVFixtureContext::path('two_empty/*.csv'))->uri()
. '" carries a column. Declare the schema with ->withSchema(...).',
);

from_csv(CSVFixtureContext::path('two_empty/*.csv'))->schema();
}

public function test_an_empty_file_in_a_glob_is_skipped(): void
Expand Down Expand Up @@ -1224,17 +1272,12 @@ public function test_record_counts_match_fgetcsv_for_every_fixture(string $fixtu

fclose($handle);

static::assertSame(
self::assertExtractedRowsCount(
max(0, $records - 1),
df()
->read(
from_csv($path)
->withSeparator($dialect->separator)
->withEnclosure($dialect->enclosure)
->withEscape($dialect->escape),
)
->fetch()
->count(),
from_csv($path)
->withSeparator($dialect->separator)
->withEnclosure($dialect->enclosure)
->withEscape($dialect->escape),
);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,7 @@ public function test_a_narrow_first_row_does_not_skew_the_estimate(): void
public function test_an_empty_file_declares_zero_rows_exactly(): void
{
$extractor = from_csv(CSVFixtureContext::path('empty.csv'));
$extractor->schema();
self::assertExtractedRowsCount(0, $extractor);

static::assertEquals(Cardinality::exact(0), $extractor->statistics()->rows);
static::assertEquals(Cardinality::unknown(), $extractor->statistics()->size);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
namespace Flow\ETL\Adapter\Excel;

use Flow\ETL\Cardinality;
use Flow\ETL\Exception\InferredSchemaException;
use Flow\ETL\Exception\InvalidArgumentException;
use Flow\ETL\Extractor;
use Flow\ETL\Extractor\BatchableExtractor;
Expand Down Expand Up @@ -125,7 +126,7 @@ public function isRepeatable(): bool
public function scan(FlowContext $context, FileScan $scan): Generator
{
$fileColumns = $this->fileColumns($this->filesystem, $this->path);
$schema = $this->schema();
$schema = $this->describe();
// only the first extract() after an inference reads on from its sample; every later one parses afresh
$batches = new ExcelFileBatches(
new WorkbookReader($this->readOptions, new ExcelFormatDetector($this->filesystem)),
Expand Down Expand Up @@ -163,6 +164,21 @@ public function inferSchema(SchemaInferenceBuilder $builder): static
}

public function schema(): Schema
{
$schema = $this->describe();
$fileColumns = $this->fileColumns($this->filesystem, $this->path);

if ($this->schema === null && $fileColumns->withoutTail($schema)->count() === 0) {
throw InferredSchemaException::nothingToInferFrom('from_excel()', sprintf(
'no file matching "%s" carries a column',
$this->path->uri(),
));
}

return $schema;
}

private function describe(): Schema
{
$fileColumns = $this->fileColumns($this->filesystem, $this->path);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,38 @@ public static function provide_nullable_fixtures(): iterable
yield 'xlsx' => [__DIR__ . '/../Fixtures/nullable_fixture.xlsx'];
}

public function test_an_empty_glob_refuses_to_infer_a_schema(): void
{
$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_excel() has nothing to infer a schema from: no file matching "'
. path_real(ExcelFixtureContext::file('*.no-match'))->uri()
. '" carries a column. Declare the schema with ->withSchema(...).',
);

from_excel(ExcelFixtureContext::file('*.no-match'))->schema();
}

public function test_an_empty_glob_with_metadata_columns_refuses_to_infer_a_schema(): void
{
$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_excel() has nothing to infer a schema from: no file matching "'
. path_real(ExcelFixtureContext::file('*.no-match'))->uri()
. '" carries a column. Declare the schema with ->withSchema(...).',
);

from_excel(ExcelFixtureContext::file('*.no-match'))->withMetadataColumns(true)->schema();
}

public function test_an_empty_glob_with_a_declared_schema_answers_it_and_extracts_nothing(): void
{
$extractor = from_excel(ExcelFixtureContext::file('*.no-match'))->withSchema(schema(int_schema('id')));

static::assertEquals(schema(int_schema('id')), $extractor->schema());
static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false));
}

#[DataProvider('provide_fixtures')]
public function test_extract_excel_file(string $fixtureName): void
{
Expand Down Expand Up @@ -463,18 +495,34 @@ public function test_cross_stream_infers_one_schema_and_the_partition_column():
);
}

#[TestWith(['empty_sheet.xlsx', []])]
#[TestWith(['empty_sheet.ods', []])]
#[TestWith(['header_only.xlsx', ['id', 'name']])]
#[TestWith(['header_only.ods', ['id', 'name']])]
public function test_header_only_and_empty_sheets(string $fixture, array $names): void
public function test_header_only_sheets(string $fixture, array $names): void
{
$extractor = from_excel(ExcelFixtureContext::file($fixture));

static::assertSame($names, $extractor->schema()->references()->names());
static::assertCount(0, df()->extract($extractor)->fetch()->toArray());
}

#[TestWith(['empty_sheet.xlsx'])]
#[TestWith(['empty_sheet.ods'])]
public function test_an_empty_sheet_refuses_to_infer_a_schema_and_extracts_nothing(string $fixture): void
{
$extractor = from_excel(ExcelFixtureContext::file($fixture));

static::assertSame([], iterator_to_array($extractor->extract(flow_context()), false));

$this->expectException(InferredSchemaException::class);
$this->expectExceptionMessage(
'from_excel() has nothing to infer a schema from: no file matching "'
. path_real(ExcelFixtureContext::file($fixture))->uri()
. '" carries a column. Declare the schema with ->withSchema(...).',
);

$extractor->schema();
}

/**
* Guards the three tests below: OpenSpout only writes a shared-strings folder when it picks the file-based
* cache, which depends on the runner's memory_limit and the fixture's unique-string count. If it ever picks
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,14 @@ public function schema(): Schema
$this->gridRowCount = $sampler->gridRowCount();
}

if ($this->schema === null && $base->count() === 0) {
throw InferredSchemaException::nothingToInferFrom('from_google_sheet()', sprintf(
'spreadsheet "%s" sheet "%s" carries no column',
$this->spreadsheetId,
$this->columnRange->sheetName,
));
}

return $this->addMetadataColumns
? $base->add(str_schema('_spread_sheet_id'), str_schema('_sheet_name'))
: $base;
Expand Down
Loading
Loading