refactor(flow-php/etl): splittable sources and the etl layer model

.env.dist

@@ -12,7 +12,7 @@ SFTP_PORT=2222
 SFTP_USER=flow
 SFTP_PASSWORD=flow
 PGSQL_DATABASE_URL=pgsql://postgres:postgres@127.0.0.1:5432/postgres?serverVersion=18&charset=utf8
-MYSQL_DATABASE_URL=mysql://mysql:mysql@127.0.0.1:3306/mysql
+MYSQL_DATABASE_URL=mysql://mysql:mysql@127.0.0.1:3306/flow
 SQLITE_DATABASE_PATH=${FLOW_MONOREPO_PROJECT_ROOT}/var/db/sqlite/test.db
 ELASTICSEARCH_URL=http://127.0.0.1:9200
 OTEL_COLLECTOR_HTTP_ENDPOINT=http://localhost:4318

.github/actions/rust-extension-toolchain/action.yml

@@ -16,6 +16,7 @@ runs:
       uses: dtolnay/rust-toolchain@89b12181fb390509a0842a86cc55eeb8eb928c1d # stable
       with:
         toolchain: stable
+        components: clippy, rustfmt
 
     - name: "Install build dependencies (Ubuntu)"
       if: runner.os == 'Linux'

.github/workflows/job-arrow-extension.yml

@@ -59,6 +59,16 @@ jobs:
           CLANG_PATH: ${{ steps.rust.outputs.clang-path }}
         run: make build
 
+      - name: Test arrow crate
+        working-directory: src/extension/arrow-ext
+        env: *rust-env
+        run: cargo test --locked --features arrow-validate
+
+      - name: Lint arrow crate
+        working-directory: src/extension/arrow-ext
+        env: *rust-env
+        run: cargo fmt --check && cargo clippy --locked --all-targets -- -D warnings
+
       - name: Install Composer Dependencies
         uses: ./.github/actions/composer-install
         with:
@@ -72,7 +82,9 @@ jobs:
 
       - name: Install extension and verify
         working-directory: src/extension/arrow-ext
+        env: *rust-env
         run: |
+          make build
           EXT_DIR=$(php -r 'echo ini_get("extension_dir");')
           sudo cp ext/modules/arrow.so "$EXT_DIR/"
           echo "extension=arrow.so" | sudo tee -a "$(php -r 'echo php_ini_loaded_file();')"

.github/workflows/job-flow-php-extension.yml

@@ -51,7 +51,6 @@ jobs:
           key: php${{ matrix.php }}
           workspaces: |
             src/extension/flow-php-ext -> target
-            src/extension/flow-php-ext/crates/flow-batch-frame -> target
             src/extension/arrow-ext -> target
 
       - name: Build arrow extension
@@ -66,10 +65,15 @@ jobs:
         env: *rust-env
         run: make build
 
-      - name: Test flow-batch-frame crate
-        working-directory: src/extension/flow-php-ext/crates/flow-batch-frame
+      - name: Test flow_php workspace
+        working-directory: src/extension/flow-php-ext
         env: *rust-env
-        run: cargo test --locked
+        run: cargo test --workspace --locked --features arrow-validate
+
+      - name: Lint flow_php workspace
+        working-directory: src/extension/flow-php-ext
+        env: *rust-env
+        run: cargo fmt --all --check && cargo clippy --workspace --locked --all-targets -- -D warnings
 
       - name: Install Composer Dependencies
         uses: ./.github/actions/composer-install
@@ -84,7 +88,9 @@ jobs:
 
       - name: Install extension and verify
         working-directory: src/extension/flow-php-ext
+        env: *rust-env
         run: |
+          make build
           EXT_DIR=$(php -r 'echo ini_get("extension_dir");')
           sudo cp ext/modules/flow_php.so "$EXT_DIR/"
           echo "extension=flow_php.so" | sudo tee -a "$(php -r 'echo php_ini_loaded_file();')"

.github/workflows/job-tests.yml

@@ -30,7 +30,7 @@ jobs:
         env:
           MYSQL_USER: mysql
           MYSQL_PASSWORD: mysql
-          MYSQL_DATABASE: mysql
+          MYSQL_DATABASE: flow
           MYSQL_ROOT_PASSWORD: root
       elasticsearch:
         image: elasticsearch:8.19.0
@@ -143,7 +143,7 @@ jobs:
         run: "just test --exclude-group grpc --log-junit ./var/phpunit/logs/junit.xml ${{ (matrix.php-version == '8.3' && matrix.dependencies == 'locked') && '--coverage-clover=./var/phpunit/coverage/clover/coverage.xml' || '' }}"
         env: &test-env
           PGSQL_DATABASE_URL: pgsql://postgres:postgres@127.0.0.1:${{ job.services.postgres.ports['5432'] }}/postgres?serverVersion=18&charset=utf8
-          MYSQL_DATABASE_URL: mysql://mysql:mysql@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/mysql
+          MYSQL_DATABASE_URL: mysql://mysql:mysql@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/flow
           ELASTICSEARCH_URL: http://127.0.0.1:${{ job.services.elasticsearch.ports['9200'] }}
           SQLITE_DATABASE_URL: "sqlite:///:memory:"
           AZURITE_HOST: "localhost"

.nix/php/lib/benchmark.ini.dist

@@ -0,0 +1,9 @@
+opcache.enable=1
+opcache.enable_cli=1
+opcache.jit=tracing
+opcache.jit_buffer_size=128M
+opcache.memory_consumption=256
+opcache.interned_strings_buffer=32
+opcache.max_accelerated_files=20000
+opcache.validate_timestamps=1
+zend.assertions=-1

.nix/pkgs/flow-php/package.nix

@@ -1,6 +1,7 @@
 {
     php,
     fetchurl,
+    stdenv,
     php-snappy,
     php-lz4,
     php-brotli,
@@ -11,6 +12,7 @@
     with-pcov ? true,
     with-xdebug ? false,
     with-blackfire ? false,
+    with-php-benchmark ? false,
     with-pg-query-ext ? false,
     with-arrow-ext ? false,
     with-flow-php-ext ? false,
@@ -87,5 +89,18 @@ flowPHP.buildEnv {
                 then builtins.readFile ./../../php/lib/pcov.ini
                 else builtins.readFile ./../../php/lib/pcov.ini.dist
         else ""
+      )
+    + "\n"
+    + (
+        if with-php-benchmark
+        then
+            (
+                if builtins.pathExists ./../../php/lib/benchmark.ini
+                    then builtins.readFile ./../../php/lib/benchmark.ini
+                    else builtins.readFile ./../../php/lib/benchmark.ini.dist
+            )
+            # macOS has no huge pages; OPcache warns on every process start, which lands in every worker's stderr.
+            + (if stdenv.hostPlatform.isLinux then "\nopcache.huge_code_pages=1\n" else "")
+        else ""
       );
 }

benchmarks/reference/duckdb/bench.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+# usage: bench.sh <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir> <threads>
+# scan: count the rows that pass a predicate touching every column, so every column is decoded (a bare count(*) on
+# Parquet is answered from the footer).
+set -euo pipefail
+
+shape="$1"; input="$2"; output="$3"; threads="$4"
+
+case "$input" in
+    *.parquet) source="read_parquet('$input')" ;;
+    *)         source="read_csv('$input', header = true)" ;;
+esac
+
+scan="id % 2 >= 0 AND quantity % 2 >= 0 AND amount >= 0 AND email <> '' AND city <> '' AND country <> '' AND first_name <> '' AND last_name <> '' AND order_uuid IS NOT NULL AND created_at IS NOT NULL AND active IS NOT NULL"
+
+case "$shape" in
+    scan)       query="SELECT count(*) FROM $source WHERE $scan;" ;;
+    to_parquet) query="COPY (SELECT * FROM $source) TO '$output/to_parquet.parquet' (FORMAT parquet);" ;;
+    group_by)   query="COPY (SELECT country, sum(amount) AS amount_sum, count(id) AS id_count, avg(amount) AS amount_avg FROM $source GROUP BY country) TO '$output/group_by.parquet' (FORMAT parquet);" ;;
+    sort)       query="COPY (SELECT * FROM $source ORDER BY amount DESC) TO '$output/sort.parquet' (FORMAT parquet);" ;;
+    *) echo "unknown shape: $shape" >&2; exit 2 ;;
+esac
+
+mkdir -p "$output"
+duckdb -c "SET threads TO $threads; $query"

benchmarks/reference/polars/bench.py

@@ -0,0 +1,43 @@
+# usage: POLARS_MAX_THREADS=<threads> python bench.py <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir>
+# POLARS_MAX_THREADS is read once when polars is imported, so it has to come from the environment.
+# scan: count the rows that pass a predicate touching every column, so every column is decoded (pl.len() alone is
+# answered from the Parquet footer).
+import os
+import sys
+
+import polars as pl
+
+shape, source, output = sys.argv[1], sys.argv[2], sys.argv[3]
+os.makedirs(output, exist_ok=True)
+
+frame = pl.scan_parquet(source) if source.endswith(".parquet") else pl.scan_csv(source, try_parse_dates=True)
+
+scan = (
+    (pl.col("id") % 2 >= 0)
+    & (pl.col("quantity") % 2 >= 0)
+    & (pl.col("amount") >= 0)
+    & (pl.col("email") != "")
+    & (pl.col("city") != "")
+    & (pl.col("country") != "")
+    & (pl.col("first_name") != "")
+    & (pl.col("last_name") != "")
+    & pl.col("order_uuid").is_not_null()
+    & pl.col("created_at").is_not_null()
+    & pl.col("active").is_not_null()
+)
+
+match shape:
+    case "scan":
+        print(frame.filter(scan).select(pl.len()).collect(engine="streaming").item())
+    case "to_parquet":
+        frame.sink_parquet(f"{output}/to_parquet.parquet")
+    case "group_by":
+        frame.group_by("country").agg(
+            pl.col("amount").sum().alias("amount_sum"),
+            pl.col("id").count().alias("id_count"),
+            pl.col("amount").mean().alias("amount_avg"),
+        ).sink_parquet(f"{output}/group_by.parquet")
+    case "sort":
+        frame.sort("amount", descending=True).sink_parquet(f"{output}/sort.parquet")
+    case _:
+        sys.exit(f"unknown shape: {shape}")

benchmarks/reference/spark/bench.py

@@ -0,0 +1,43 @@
+# usage: spark-submit --master local[<threads>] bench.py <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir>
+# scan: count the rows that pass a predicate touching every column, so every column is decoded.
+import sys
+
+from pyspark.sql import SparkSession
+from pyspark.sql import functions as F
+
+shape, source, output = sys.argv[1], sys.argv[2], sys.argv[3]
+
+spark = SparkSession.builder.appName(f"flow-reference-{shape}").getOrCreate()
+frame = spark.read.parquet(source) if source.endswith(".parquet") else spark.read.csv(source, header=True, inferSchema=True)
+
+scan = (
+    (F.col("id") % 2 >= 0)
+    & (F.col("quantity") % 2 >= 0)
+    & (F.col("amount") >= 0)
+    & (F.col("email") != "")
+    & (F.col("city") != "")
+    & (F.col("country") != "")
+    & (F.col("first_name") != "")
+    & (F.col("last_name") != "")
+    & F.col("order_uuid").isNotNull()
+    & F.col("created_at").isNotNull()
+    & F.col("active").isNotNull()
+)
+
+match shape:
+    case "scan":
+        print(frame.filter(scan).count())
+    case "to_parquet":
+        frame.write.mode("overwrite").parquet(f"{output}/to_parquet.parquet")
+    case "group_by":
+        frame.groupBy("country").agg(
+            F.sum("amount").alias("amount_sum"),
+            F.count("id").alias("id_count"),
+            F.avg("amount").alias("amount_avg"),
+        ).write.mode("overwrite").parquet(f"{output}/group_by.parquet")
+    case "sort":
+        frame.orderBy(F.col("amount").desc()).write.mode("overwrite").parquet(f"{output}/sort.parquet")
+    case _:
+        sys.exit(f"unknown shape: {shape}")
+
+spark.stop()

benchmarks/src/BenchmarkConfig.php

@@ -5,7 +5,7 @@
 namespace Flow\Benchmarks;
 
 use Flow\Benchmarks\Datasets\Paths;
-use Flow\ETL\Config\ConfigBuilder;
+use Flow\ETL\API\Config\ConfigBuilder;
 
 use function Flow\ETL\DSL\config_builder;
 

benchmarks/src/Datasets/FixtureFormat.php

@@ -5,7 +5,7 @@
 namespace Flow\Benchmarks\Datasets;
 
 /**
- * Named FixtureFormat rather than Format so the short name does not collide with Flow\Floe\Format.
+ * Named FixtureFormat rather than Format so the short name does not collide with Flow\ETL\Storage\Floe\Format.
  */
 enum FixtureFormat: string
 {

benchmarks/src/Joining/JoinEachScenario.php

@@ -25,7 +25,7 @@ public function run(): void
             ->read(from_floe(Datasets::orders($this->rows)->floe()))
             ->batchSize(1000)
             ->withColumn('seller_id', ref('seller_id')->cast(type_string()))
-            ->joinEach(new SellersDataFrameFactory(Datasets::sellers($this->rows)->parquet()), join_on([
+            ->joinEach(new SellersExtractorFactory(Datasets::sellers($this->rows)->parquet()), join_on([
                 'seller_id' => 'id',
             ], join_prefix: 'sellers_'))
             ->run();

benchmarks/src/Joining/JoinOrdersWithSellersScenario.php

@@ -6,7 +6,7 @@
 
 use Flow\Benchmarks\BenchmarkConfig;
 use Flow\Benchmarks\Datasets\Datasets;
-use Flow\ETL\Join\Join;
+use Flow\ETL\Values\Join\JoinType;
 
 use function Flow\ETL\Adapter\Parquet\from_parquet;
 use function Flow\ETL\DSL\data_frame;
@@ -18,7 +18,7 @@
 final readonly class JoinOrdersWithSellersScenario
 {
     public function __construct(
-        private Join $type,
+        private JoinType $type,
         private int $rows,
     ) {}
 

benchmarks/src/Joining/SellersExtractorFactory.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Joining;
+
+use Flow\ETL\IO\Extractor;
+use Flow\ETL\IO\Extractor\ExtractorFactory;
+use Flow\ETL\Values\Rows;
+
+use function Flow\ETL\Adapter\Parquet\from_parquet;
+
+final readonly class SellersExtractorFactory implements ExtractorFactory
+{
+    public function __construct(
+        private string $sellersPath,
+    ) {}
+
+    public function from(Rows $rows): Extractor
+    {
+        return from_parquet($this->sellersPath);
+    }
+}

benchmarks/src/Partitioning/PartitionCardinality.php

@@ -4,8 +4,8 @@
 
 namespace Flow\Benchmarks\Partitioning;
 
-use Flow\ETL\DataFrame;
-use Flow\ETL\Row\Reference;
+use Flow\ETL\API\DataFrame;
+use Flow\ETL\Functions\Reference\Reference;
 
 use function Flow\ETL\DSL\ref;
 

benchmarks/src/Partitioning/PartitionedReadScenario.php

@@ -5,7 +5,7 @@
 namespace Flow\Benchmarks\Partitioning;
 
 use Flow\Benchmarks\BenchmarkConfig;
-use Flow\ETL\Rows;
+use Flow\ETL\Values\Rows;
 
 use function Flow\ETL\Adapter\CSV\from_csv;
 use function Flow\ETL\DSL\data_frame;