Parquet filtering

Justfile

@@ -138,9 +138,7 @@ docs-api:
 
 # Regenerate Parquet Thrift PHP classes.
 gen-thrift: && fix
-    grep -q 'namespace php Flow.Parquet.ThriftModel' src/lib/parquet/src/Flow/Parquet/Resources/Thrift/parquet.thrift || { echo "Flow php namespace not found in thrift definition!"; exit 1; }
-    rm src/lib/parquet/src/Flow/Parquet/ThriftModel/*.php
-    thrift --gen php --out src/lib/parquet/src src/lib/parquet/src/Flow/Parquet/Resources/Thrift/parquet.thrift
+    php bin/thrift.php parquet
 
 # Regenerate PostgreSQL pg_query Protobuf PHP classes.
 gen-protobuf-pg: && fix

bin/docs.php

@@ -73,6 +73,7 @@ public function execute(InputInterface $input, OutputInterface $output): int
             __DIR__ . '/../src/adapter/etl-adapter-xml/src/Flow/ETL/Adapter/XML/functions.php',
             __DIR__ . '/../src/lib/filesystem/src/Flow/Filesystem/DSL/functions.php',
             __DIR__ . '/../src/lib/types/src/Flow/Types/DSL/functions.php',
+            __DIR__ . '/../src/lib/parquet/src/Flow/Parquet/DSL/functions.php',
             __DIR__ . '/../src/lib/postgresql/src/Flow/PostgreSql/DSL/schema.php',
             __DIR__ . '/../src/lib/postgresql/src/Flow/PostgreSql/DSL/query.php',
             __DIR__ . '/../src/lib/postgresql/src/Flow/PostgreSql/DSL/condition.php',

bin/thrift.php

@@ -0,0 +1,100 @@
+#!/usr/bin/env php
+<?php
+
+declare(strict_types=1);
+
+use Symfony\Component\Console\Application;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Process\ExecutableFinder;
+use Symfony\Component\Process\Process;
+
+use function Flow\Types\DSL\type_string;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$application = new Application('Flow-PHP - Thrift');
+
+$application->addCommand(new class extends Command {
+    public function configure(): void
+    {
+        $this
+            ->setName('parquet')
+            ->setDescription('Regenerates the Parquet ThriftModel classes from parquet.thrift.')
+            ->addOption(
+                'thrift',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'Thrift compiler binary, found in PATH when omitted',
+            );
+    }
+
+    public function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $source = __DIR__ . '/../src/lib/parquet/src';
+        $idl = $source . '/Flow/Parquet/Resources/Thrift/parquet.thrift';
+        $models = $source . '/Flow/Parquet/ThriftModel';
+
+        $thrift = $input->getOption('thrift') === null
+            ? (new ExecutableFinder())->find('thrift')
+            : type_string()->assert($input->getOption('thrift'));
+
+        if (!is_string($thrift) || !is_executable($thrift)) {
+            $output->writeln('<error>Thrift compiler not found, pass an executable with --thrift</error>');
+
+            return Command::FAILURE;
+        }
+
+        if (!str_contains((string) file_get_contents($idl), 'namespace php Flow.Parquet.ThriftModel')) {
+            $output->writeln('<error>Flow php namespace not found in thrift definition!</error>');
+
+            return Command::FAILURE;
+        }
+
+        foreach (glob($models . '/*.php') ?: [] as $file) {
+            unlink($file);
+        }
+
+        (new Process([$thrift, '--gen', 'php', '--out', $source, $idl]))->mustRun(
+            static fn(string $type, string $buffer) => $output->write($buffer),
+        );
+
+        // the Thrift compiler types every field non-null; a field the IDL does not require is null when a file leaves it out
+        foreach (glob($models . '/*.php') ?: [] as $file) {
+            $class = 'Flow\\Parquet\\ThriftModel\\' . basename($file, '.php');
+
+            if (!class_exists($class) || !property_exists($class, '_TSPEC')) {
+                continue;
+            }
+
+            /** @var array<int, array{var: string, isRequired: bool}> $spec */
+            $spec = (new ReflectionProperty($class, '_TSPEC'))->getValue();
+            $model = (string) file_get_contents($file);
+            $rewritten = $model;
+
+            foreach ($spec as $field) {
+                if ($field['isRequired']) {
+                    continue;
+                }
+
+                $rewritten = (string) preg_replace(
+                    '/(\* @var )(?!null\|)([^\n]+)(\n\s*\*\/\n\s*public \$'
+                    . preg_quote($field['var'], '/')
+                    . ' = null;)/',
+                    '$1null|$2$3',
+                    $rewritten,
+                );
+            }
+
+            if ($rewritten !== $model) {
+                file_put_contents($file, $rewritten);
+            }
+        }
+
+        return Command::SUCCESS;
+    }
+});
+
+$application->run();

composer.json

@@ -260,6 +260,7 @@
             "src/lib/filesystem/src/Flow/Filesystem/DSL/functions.php",
             "src/lib/parquet/src/Flow/Parquet/functions.php",
             "src/lib/parquet/src/Flow/Parquet/Binary/functions.php",
+            "src/lib/parquet/src/Flow/Parquet/DSL/functions.php",
             "src/lib/parquet/src/stubs.php",
             "src/lib/postgresql/src/Flow/PostgreSql/DSL/client.php",
             "src/lib/postgresql/src/Flow/PostgreSql/DSL/condition.php",

documentation/components/adapters/parquet.md

@@ -73,3 +73,72 @@ data_frame()
     ->write(to_output())
     ->run();
 ```
+
+A `UINT_64` column reads as `numeric_string` (`'18446744073709551615'` is past `PHP_INT_MAX`). A `numeric_string`
+column is written as a parquet `STRING`; write `UINT_64` through the lib with an explicit schema:
+
+```php
+<?php
+
+use Flow\Parquet\ParquetFile\Schema;
+use Flow\Parquet\ParquetFile\Schema\{FlatColumn, LogicalType, PhysicalType};
+use Flow\Parquet\Writer;
+
+(new Writer())->write(
+    __DIR__ . '/ids.parquet',
+    Schema::with(new FlatColumn('id', PhysicalType::INT64, logicalType: LogicalType::integer(64, false))),
+    [['id' => 7], ['id' => '18446744073709551615']],
+);
+```
+
+## Filter pushdown
+
+A `filter()` directly over `from_parquet()` moves into the reader, which returns only matching rows: row groups whose
+statistics prove that no row matches are skipped, and a `limit()` after the filter stops the read at that many
+matching rows:
+
+```php
+<?php
+
+use function Flow\ETL\Adapter\Parquet\from_parquet;
+use function Flow\ETL\DSL\{data_frame, lit, ref, to_output};
+
+data_frame()
+    ->read(from_parquet(__DIR__ . '/orders/*.parquet'))
+    ->filter(ref('id')->greaterThan(lit(1000))->and(ref('status')->equals(lit('paid'))))
+    ->limit(10)
+    ->write(to_output())
+    ->run();
+```
+
+Each part of an `and()` is pushed on its own when it compares a file column with a literal it can compare exactly.
+
+Pushed:
+
+- `equals`, `notEquals`, `same`, `notSame`, `lessThan`, `lessThanEqual`, `greaterThan`, `greaterThanEqual`,
+  `between`, `isIn` (literal list), `startsWith`, `isNull`, `isNotNull`, and `not()`/`and()`/`or()` of those
+- an int literal on a float column up to 2^53, an integral float on an int column below 2^53
+- a `date` literal on a `datetime` column and the reverse
+- a digit string (`lit('1000')`) on a `numeric_string` (`UINT_64`) column
+
+Evaluated on the rows read:
+
+- other functions, two columns compared, a literal of another type
+- `json` columns (except `isNull`/`isNotNull`), ordering on `uuid`
+- a column the file stores as another type than the read
+
+Partition columns list fewer files instead ([partitioning](/documentation/components/core/partitioning.md)).
+
+`explain()` shows the filter and the limit under `Read`, not what the reader skipped:
+
+```
+#3 Result
+│  Rows this plan hands out: to the trigger, or to the node reading it
+└─ #2 Limit
+   │  Limit: 10
+   └─ #1 Read
+         Extractor: ParquetExtractor
+         Source: file:///orders/*.parquet
+         Limit: 10
+         Filter: All
+```

documentation/components/core/core.md

@@ -67,7 +67,7 @@ Outputs
 ├─ #4 Result
 │  │  Rows this plan hands out: to the trigger, or to the node reading it
 │  └─ #3 Select
-│     │  Columns: id, email
+│     │  Keeps only: id, email
 │     └─ #2 Filter
 │        │  Condition: IsNotNull
 │        └─ #1 Read
@@ -129,7 +129,7 @@ echo $dataFrame->explain()->toString(format: Format::flow);
 └─ #2 Filter
    │  Condition: IsNotNull
    ├─ #3 Select
-   │  │  Columns: id, email
+   │  │  Keeps only: id, email
    │  └─ #4 Result
    │        Rows this plan hands out: to the trigger, or to the node reading it
    └─ #5 Write
@@ -154,7 +154,7 @@ echo $dataFrame->explain()->toString(format: Format::boxes);
 ┌─────────────┴─────────────┐┌─────────────┴─────────────┐
 │         #3 Select         ││         #2 Filter         │
 │   ────────────────────    ││         (shared)          │
-│    Columns: id, email     ││                           │
+│   Keeps only: id, email   ││                           │
 └─────────────┬─────────────┘└───────────────────────────┘
 ┌─────────────┴─────────────┐
 │         #2 Filter         │

documentation/components/core/partitioning.md

@@ -268,8 +268,10 @@ it from the file schema (Parquet, Floe), so `filter(ref('date')->between(lit($fr
 
 ### Partition Pruning
 
-The optimizer pushes a `filter()` that reads only partition columns into the source, so whole directories
-are skipped; a filter on a body column still reads everything and then discards. A partition column renamed
+The optimizer moves a `filter()` into the file source, which returns only matching rows. A filter that reads only
+partition columns skips whole directories; a filter on a body column reads every file and drops the rows it does not
+match (Parquet skips row groups, see [filter pushdown](/documentation/components/adapters/parquet.md#filter-pushdown)).
+A `limit()` after the filter reaches the source with it ([filter pushdown](/documentation/components/adapters/parquet.md#filter-pushdown)). A partition column renamed
 with `rename()` or copied with `withEntry('day', ref('date'))` is still pruned; any other redefinition of it
 stops the push. A filter whose answer can change between two evaluations - `call()`, `now()`, `uuid()`,
 `to_date_time()` with a format that leaves fields out (`'Y-m-d'`) - still filters rows but prunes nothing.
@@ -283,6 +285,7 @@ use function Flow\ETL\DSL\{data_frame, lit, ref, to_output};
 data_frame()
     ->read(from_csv(__DIR__ . '/output/date=*/department=*/*.csv'))
     ->filter(ref('date')->greaterThanEqual(lit('2024-01-01')))
+    ->limit(10)
     ->write(to_output())
     ->run();
 ```

documentation/components/extensions/arrow-ext.md

@@ -92,18 +92,23 @@ use Flow\Parquet\Engine\{RustParquetFileReader, RustParquetFileWriter};
 use Flow\Parquet\Engine\Arrow\{OptionsConverter, SchemaConverter};
 use Flow\Parquet\Options;
 use Flow\Parquet\ParquetFile\Compressions;
+use Flow\Parquet\ParquetFile\Index\StreamIndexes;
+use Flow\Parquet\Reader\ReadPlan;
 
 use function Flow\Filesystem\DSL\{native_local_filesystem, path};
 
 $filesystem = native_local_filesystem();
 
 $file = new RustParquetFileReader($filesystem->readFrom(path('data.parquet'))); // Flow\Parquet\ParquetFileReader, one footer read
-$file->rowsNumber();
-$file->schema();                                                      // Flow\Parquet\ParquetFile\Schema
-$file->metadata();                                                    // Flow\Parquet\ParquetFile\Metadata
-$file->thrift();                                                      // Flow\Parquet\ThriftModel\FileMetaData
-
-foreach ($file->readColumns(['id', 'name'], 1_000, null, null) as $chunk) { // RustColumnsReader, an Iterator that streams once
+$footer = $file->footer();                                            // Flow\Parquet\Engine\RustParquetFooter, readable after close()
+$footer->rowsNumber();
+$footer->schema();                                                    // Flow\Parquet\ParquetFile\Schema
+$footer->metadata();                                                  // Flow\Parquet\ParquetFile\Metadata
+$footer->thrift();                                                    // Flow\Parquet\ThriftModel\FileMetaData
+$indexes = new StreamIndexes($footer, $filesystem->readFrom(path('data.parquet')));  // page indexes, read on demand
+$plan = ReadPlan::window($footer->rowsNumber(), $indexes, offset: 10, limit: 100);  // rows 10-109
+
+foreach ($file->readColumns(['id', 'name'], 1_000, $plan) as $chunk) { // RustColumnsReader, an Iterator that streams once
     // ['id' => [1, 2, ...], 'name' => ['a', 'b', ...]]
 }
 
@@ -118,7 +123,7 @@ $writer->writeBatch([['id' => 1, 'name' => 'a']]);   // an array or any Traversa
 $writer->writeColumns(['id' => [2, 3], 'name' => ['b', null]]);
 $writer->close();                              // footer, then the stream closed
 
-$batches = new RustBatchReader($file, ['id', 'name'], 1_000, null, null); // batch size, offset, limit
+$batches = new RustBatchReader($file, ['id', 'name'], 1_000, $plan); // batch size, rows to read
 $batches->schema();                            // Flow\Arrow\RustArrowSchema
 $batches->next();                              // Flow\Arrow\RustParquetBatch, null after the last
 ```