Skip to content

Commit d5ffe41

Browse files
norberttechnorberttech
authored
Merge pull request #2710 from flow-php/flow-architecture-refactoring
refactor(flow-php/etl): splittable sources and the etl layer model
2 parents d5aefb9 + c1a25ed commit d5ffe41

4,379 files changed

Lines changed: 200554 additions & 172175 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.env.dist‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ SFTP_PORT=2222
1212
SFTP_USER=flow
1313
SFTP_PASSWORD=flow
1414
PGSQL_DATABASE_URL=pgsql://postgres:postgres@127.0.0.1:5432/postgres?serverVersion=18&charset=utf8
15-
MYSQL_DATABASE_URL=mysql://mysql:mysql@127.0.0.1:3306/mysql
15+
MYSQL_DATABASE_URL=mysql://mysql:mysql@127.0.0.1:3306/flow
1616
SQLITE_DATABASE_PATH=${FLOW_MONOREPO_PROJECT_ROOT}/var/db/sqlite/test.db
1717
ELASTICSEARCH_URL=http://127.0.0.1:9200
1818
OTEL_COLLECTOR_HTTP_ENDPOINT=http://localhost:4318

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ runs:
1616
uses: dtolnay/rust-toolchain@89b12181fb390509a0842a86cc55eeb8eb928c1d # stable
1717
with:
1818
toolchain: stable
19+
components: clippy, rustfmt
1920

2021
- name: "Install build dependencies (Ubuntu)"
2122
if: runner.os == 'Linux'

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,16 @@ jobs:
5959
CLANG_PATH: ${{ steps.rust.outputs.clang-path }}
6060
run: make build
6161

62+
- name: Test arrow crate
63+
working-directory: src/extension/arrow-ext
64+
env: *rust-env
65+
run: cargo test --locked --features arrow-validate
66+
67+
- name: Lint arrow crate
68+
working-directory: src/extension/arrow-ext
69+
env: *rust-env
70+
run: cargo fmt --check && cargo clippy --locked --all-targets -- -D warnings
71+
6272
- name: Install Composer Dependencies
6373
uses: ./.github/actions/composer-install
6474
with:
@@ -72,7 +82,9 @@ jobs:
7282

7383
- name: Install extension and verify
7484
working-directory: src/extension/arrow-ext
85+
env: *rust-env
7586
run: |
87+
make build
7688
EXT_DIR=$(php -r 'echo ini_get("extension_dir");')
7789
sudo cp ext/modules/arrow.so "$EXT_DIR/"
7890
echo "extension=arrow.so" | sudo tee -a "$(php -r 'echo php_ini_loaded_file();')"

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

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@ jobs:
5151
key: php${{ matrix.php }}
5252
workspaces: |
5353
src/extension/flow-php-ext -> target
54-
src/extension/flow-php-ext/crates/flow-batch-frame -> target
5554
src/extension/arrow-ext -> target
5655
5756
- name: Build arrow extension
@@ -66,10 +65,15 @@ jobs:
6665
env: *rust-env
6766
run: make build
6867

69-
- name: Test flow-batch-frame crate
70-
working-directory: src/extension/flow-php-ext/crates/flow-batch-frame
68+
- name: Test flow_php workspace
69+
working-directory: src/extension/flow-php-ext
7170
env: *rust-env
72-
run: cargo test --locked
71+
run: cargo test --workspace --locked --features arrow-validate
72+
73+
- name: Lint flow_php workspace
74+
working-directory: src/extension/flow-php-ext
75+
env: *rust-env
76+
run: cargo fmt --all --check && cargo clippy --workspace --locked --all-targets -- -D warnings
7377

7478
- name: Install Composer Dependencies
7579
uses: ./.github/actions/composer-install
@@ -84,7 +88,9 @@ jobs:
8488

8589
- name: Install extension and verify
8690
working-directory: src/extension/flow-php-ext
91+
env: *rust-env
8792
run: |
93+
make build
8894
EXT_DIR=$(php -r 'echo ini_get("extension_dir");')
8995
sudo cp ext/modules/flow_php.so "$EXT_DIR/"
9096
echo "extension=flow_php.so" | sudo tee -a "$(php -r 'echo php_ini_loaded_file();')"

‎.github/workflows/job-tests.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ jobs:
3030
env:
3131
MYSQL_USER: mysql
3232
MYSQL_PASSWORD: mysql
33-
MYSQL_DATABASE: mysql
33+
MYSQL_DATABASE: flow
3434
MYSQL_ROOT_PASSWORD: root
3535
elasticsearch:
3636
image: elasticsearch:8.19.0
@@ -143,7 +143,7 @@ jobs:
143143
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' || '' }}"
144144
env: &test-env
145145
PGSQL_DATABASE_URL: pgsql://postgres:postgres@127.0.0.1:${{ job.services.postgres.ports['5432'] }}/postgres?serverVersion=18&charset=utf8
146-
MYSQL_DATABASE_URL: mysql://mysql:mysql@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/mysql
146+
MYSQL_DATABASE_URL: mysql://mysql:mysql@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/flow
147147
ELASTICSEARCH_URL: http://127.0.0.1:${{ job.services.elasticsearch.ports['9200'] }}
148148
SQLITE_DATABASE_URL: "sqlite:///:memory:"
149149
AZURITE_HOST: "localhost"

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
opcache.enable=1
2+
opcache.enable_cli=1
3+
opcache.jit=tracing
4+
opcache.jit_buffer_size=128M
5+
opcache.memory_consumption=256
6+
opcache.interned_strings_buffer=32
7+
opcache.max_accelerated_files=20000
8+
opcache.validate_timestamps=1
9+
zend.assertions=-1

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
{
22
php,
33
fetchurl,
4+
stdenv,
45
php-snappy,
56
php-lz4,
67
php-brotli,
@@ -11,6 +12,7 @@
1112
with-pcov ? true,
1213
with-xdebug ? false,
1314
with-blackfire ? false,
15+
with-php-benchmark ? false,
1416
with-pg-query-ext ? false,
1517
with-arrow-ext ? false,
1618
with-flow-php-ext ? false,
@@ -87,5 +89,18 @@ flowPHP.buildEnv {
8789
then builtins.readFile ./../../php/lib/pcov.ini
8890
else builtins.readFile ./../../php/lib/pcov.ini.dist
8991
else ""
92+
)
93+
+ "\n"
94+
+ (
95+
if with-php-benchmark
96+
then
97+
(
98+
if builtins.pathExists ./../../php/lib/benchmark.ini
99+
then builtins.readFile ./../../php/lib/benchmark.ini
100+
else builtins.readFile ./../../php/lib/benchmark.ini.dist
101+
)
102+
# macOS has no huge pages; OPcache warns on every process start, which lands in every worker's stderr.
103+
+ (if stdenv.hostPlatform.isLinux then "\nopcache.huge_code_pages=1\n" else "")
104+
else ""
90105
);
91106
}

‎benchmarks/reference/duckdb/bench.sh‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
#!/usr/bin/env bash
2+
# usage: bench.sh <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir> <threads>
3+
# scan: count the rows that pass a predicate touching every column, so every column is decoded (a bare count(*) on
4+
# Parquet is answered from the footer).
5+
set -euo pipefail
6+
7+
shape="$1"; input="$2"; output="$3"; threads="$4"
8+
9+
case "$input" in
10+
*.parquet) source="read_parquet('$input')" ;;
11+
*) source="read_csv('$input', header = true)" ;;
12+
esac
13+
14+
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"
15+
16+
case "$shape" in
17+
scan) query="SELECT count(*) FROM $source WHERE $scan;" ;;
18+
to_parquet) query="COPY (SELECT * FROM $source) TO '$output/to_parquet.parquet' (FORMAT parquet);" ;;
19+
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);" ;;
20+
sort) query="COPY (SELECT * FROM $source ORDER BY amount DESC) TO '$output/sort.parquet' (FORMAT parquet);" ;;
21+
*) echo "unknown shape: $shape" >&2; exit 2 ;;
22+
esac
23+
24+
mkdir -p "$output"
25+
duckdb -c "SET threads TO $threads; $query"

‎benchmarks/reference/polars/bench.py‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# usage: POLARS_MAX_THREADS=<threads> python bench.py <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir>
2+
# POLARS_MAX_THREADS is read once when polars is imported, so it has to come from the environment.
3+
# scan: count the rows that pass a predicate touching every column, so every column is decoded (pl.len() alone is
4+
# answered from the Parquet footer).
5+
import os
6+
import sys
7+
8+
import polars as pl
9+
10+
shape, source, output = sys.argv[1], sys.argv[2], sys.argv[3]
11+
os.makedirs(output, exist_ok=True)
12+
13+
frame = pl.scan_parquet(source) if source.endswith(".parquet") else pl.scan_csv(source, try_parse_dates=True)
14+
15+
scan = (
16+
(pl.col("id") % 2 >= 0)
17+
& (pl.col("quantity") % 2 >= 0)
18+
& (pl.col("amount") >= 0)
19+
& (pl.col("email") != "")
20+
& (pl.col("city") != "")
21+
& (pl.col("country") != "")
22+
& (pl.col("first_name") != "")
23+
& (pl.col("last_name") != "")
24+
& pl.col("order_uuid").is_not_null()
25+
& pl.col("created_at").is_not_null()
26+
& pl.col("active").is_not_null()
27+
)
28+
29+
match shape:
30+
case "scan":
31+
print(frame.filter(scan).select(pl.len()).collect(engine="streaming").item())
32+
case "to_parquet":
33+
frame.sink_parquet(f"{output}/to_parquet.parquet")
34+
case "group_by":
35+
frame.group_by("country").agg(
36+
pl.col("amount").sum().alias("amount_sum"),
37+
pl.col("id").count().alias("id_count"),
38+
pl.col("amount").mean().alias("amount_avg"),
39+
).sink_parquet(f"{output}/group_by.parquet")
40+
case "sort":
41+
frame.sort("amount", descending=True).sink_parquet(f"{output}/sort.parquet")
42+
case _:
43+
sys.exit(f"unknown shape: {shape}")

‎benchmarks/reference/spark/bench.py‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# usage: spark-submit --master local[<threads>] bench.py <scan|to_parquet|group_by|sort> <input.csv|input.parquet> <output-dir>
2+
# scan: count the rows that pass a predicate touching every column, so every column is decoded.
3+
import sys
4+
5+
from pyspark.sql import SparkSession
6+
from pyspark.sql import functions as F
7+
8+
shape, source, output = sys.argv[1], sys.argv[2], sys.argv[3]
9+
10+
spark = SparkSession.builder.appName(f"flow-reference-{shape}").getOrCreate()
11+
frame = spark.read.parquet(source) if source.endswith(".parquet") else spark.read.csv(source, header=True, inferSchema=True)
12+
13+
scan = (
14+
(F.col("id") % 2 >= 0)
15+
& (F.col("quantity") % 2 >= 0)
16+
& (F.col("amount") >= 0)
17+
& (F.col("email") != "")
18+
& (F.col("city") != "")
19+
& (F.col("country") != "")
20+
& (F.col("first_name") != "")
21+
& (F.col("last_name") != "")
22+
& F.col("order_uuid").isNotNull()
23+
& F.col("created_at").isNotNull()
24+
& F.col("active").isNotNull()
25+
)
26+
27+
match shape:
28+
case "scan":
29+
print(frame.filter(scan).count())
30+
case "to_parquet":
31+
frame.write.mode("overwrite").parquet(f"{output}/to_parquet.parquet")
32+
case "group_by":
33+
frame.groupBy("country").agg(
34+
F.sum("amount").alias("amount_sum"),
35+
F.count("id").alias("id_count"),
36+
F.avg("amount").alias("amount_avg"),
37+
).write.mode("overwrite").parquet(f"{output}/group_by.parquet")
38+
case "sort":
39+
frame.orderBy(F.col("amount").desc()).write.mode("overwrite").parquet(f"{output}/sort.parquet")
40+
case _:
41+
sys.exit(f"unknown shape: {shape}")
42+
43+
spark.stop()

0 commit comments

Comments
 (0)