diff --git a/conf/wayang-defaults.properties b/conf/wayang-defaults.properties index 736b73ebe..ff2dcf33a 100644 --- a/conf/wayang-defaults.properties +++ b/conf/wayang-defaults.properties @@ -16,7 +16,7 @@ # # Configure statistics collection. -wayang.core.log.enabled = true +wayang.core.log.enabled = false wayang.core.explain.enabled = false wayang.core.explain.directrory = ~/.wayang/ diff --git a/demo-trino.sh b/demo-trino.sh new file mode 100644 index 000000000..45b1fad7a --- /dev/null +++ b/demo-trino.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set -euo pipefail + +WAYANG_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +exec "$WAYANG_ROOT/trino-setup/demo.sh" "$@" diff --git a/guides/cost-profiling.md b/guides/cost-profiling.md new file mode 100644 index 000000000..1e565246c --- /dev/null +++ b/guides/cost-profiling.md @@ -0,0 +1,430 @@ + + +# Cost Profiling Guide + +This document explains why Apache Wayang needs platform-specific cost +profiling, how profiling data is collected, how the genetic optimizer learns +cost parameters, and how users can repeat the profiling workflow on their own +hardware. + +The examples below use Trino, but the same workflow also applies to other +JDBC-based platforms such as Presto and BigQuery. + +Version 2.0 uses S01 through S16 as the profiling workload, including the +join-heavy pipelines S14 through S16. It keeps the guide focused on data +collection and parameter learning, leaving follow-up quality checks out of +scope for now. + +## 1. Why Profiling Is Needed + +Wayang can map the same logical plan to different execution platforms, such as +Java, Spark, Trino, Presto, or BigQuery. For example, a user query may contain: + +```text +TableSource -> Filter -> Projection -> TableSink +``` + +The optimizer needs a cost model to decide whether these operators should stay +on a SQL platform or be moved to another platform. In this context, "cost" does +not mean cloud billing cost. It is the numerical value that Wayang uses to +compare alternative execution plans. + +With the default Trino configuration: + +```properties +wayang.trino.costs.fix = 0.0 +wayang.trino.costs.per-ms = 1.0 +``` + +the optimizer cost can be interpreted approximately as: + +```text +cost = estimated execution time in milliseconds +``` + +However, the real execution time depends on the user's machine, cluster size, +network, database configuration, and workload. Therefore, users should profile +their own environment when they need accurate cost parameters. + +## 2. Load Profile Formulas + +Each execution operator has a load profile. For example, a table source may use +a formula like: + +```properties +wayang.trino.tablesource.load = { + "type":"mathex", + "in":0, + "out":1, + "cpu":"((10)*(out0))+(800000)", + "ram":"0", + "disk":"0", + "net":"0", + "p":0.9 +} +``` + +This can be read as: + +```text +CPU load = 10 * number_of_rows + 800000 +``` + +where: + +- `out0` is the output cardinality. +- `alpha` is the per-row cost. +- `beta` is the fixed overhead, such as query planning, scheduling, and remote + execution startup. +- `p` is the confidence of the estimate. + +The profiling goal is to learn reasonable values for `alpha` and `beta` from +real execution records. + +Wayang can also define templates with unknown parameters: + +```properties +wayang.trino.tablesource.load.template = { + "type":"mathex", + "in":0, + "out":1, + "cpu":"?*out0 + ?", + "ram":"0", + "disk":"0", + "net":"0", + "p":0.9 +} +``` + +The genetic optimizer reads the templates and replaces the `?` placeholders with +learned values. + +## 3. Profiling Workflow + +The expected profiling workflow is: + +```text +Run Wayang jobs with different operators and input cardinalities + | + v +Record platform, operator lineage, cardinalities, and runtime + | + v + executions.json + | + v + GeneticOptimizerApp reads the execution records + | + v +Learn the unknown parameters in *.load.template + | + v +Write learned *.load formulas +``` + +Wayang stores measured executions as `PartialExecution` records. Each record +contains: + +- the measured execution time; +- the platform that executed the stage; +- one or more `ExecutionLineageNode` objects; +- the load profile estimator for each profiled operator; +- input and output cardinalities. + +The default execution log location is usually: + +```text +~/.wayang/executions.json +``` + +For controlled profiling experiments, it is better to write the log to a +dedicated experiment folder, for example: + +```text +C:\Users\\Desktop\Wayang Profiling\trino\week8\executions.json +``` + +## 4. Experiment Design + +Profiling should include both single-operator pipelines and combined pipelines. +Single-operator pipelines help isolate each operator. Combined pipelines help +the optimizer learn parameters from realistic SQL stages, where multiple +operators are executed together. + +Choose input cardinalities according to the machine or cluster being profiled. +The values below are only an example that can run on a laptop-sized local +setup: + +```text +10k, 50k, 100k, 250k +``` + +For a smaller machine, use fewer or smaller cardinalities. For a larger local +or remote platform, add larger cardinalities so the learned model reflects the +scale that users expect to run. + +Recommended repetitions: + +```text +1 warm-up run + 5 measured runs +``` + +Profiling pipelines: + +| Plan | Pipeline | +|------|----------| +| S01 | TableSource -> TableSink | +| S02 | TableSource -> Filter(50%) -> TableSink | +| S03 | TableSource -> Projection(order_id, amount) -> TableSink | +| S04 | TableSource -> Filter(50%) -> Projection(order_id, amount) -> TableSink | +| S05 | TableSource -> GlobalReduce(sum amount) -> TableSink | +| S06 | TableSource -> ReduceBy(bucket) -> TableSink | +| S07 | TableSource -> Sort(amount) -> TableSink | +| S08 | Orders -> Join(Customers 1k) -> Projection -> TableSink | +| S09 | TableSource -> Filter(50%) -> GlobalReduce -> TableSink | +| S10 | TableSource -> Filter(50%) -> ReduceBy -> TableSink | +| S11 | TableSource -> Filter(50%) -> Sort(amount) -> TableSink | +| S12 | TableSource -> Projection(order_id, amount) -> Sort(amount) -> TableSink | +| S13 | TableSource -> Filter(50%) -> Projection(order_id, amount) -> Sort(amount) -> TableSink | +| S14 | Orders -> Filter(50%) -> Join(Customers 1k) -> Projection -> TableSink | +| S15 | Orders -> Join(Customers 1k) -> Projection(order_id, tier, amount) -> Sort(amount) -> TableSink | +| S16 | Orders -> Join(Customers 1k) -> Projection(tier, amount) -> ReduceBy(tier) -> TableSink | + +S01 through S16 should be treated as one profiling workload, including the +join-heavy plans S14 through S16. For example, using 16 plans, 4 cardinalities, +and 6 repetitions produces: + +```text +16 * 4 * 6 = 384 Wayang executions +``` + +If users choose a different number of cardinalities or repetitions, the total +number of executions changes accordingly: + +```text +number_of_plans * number_of_cardinalities * repetitions +``` + +The reference parameters shipped in the platform defaults were learned from our +local Week 8 profiling runs over S01 through S13, with row counts +10k/50k/100k/250k and 1 warm-up plus 5 measured repetitions. S14 through S16 +were added to this guide to document the join-heavy pipelines that users should +include when they rerun profiling in their own environment. The shipped +parameters are intended as reasonable starting values for users who just want +to try Wayang; they are not universal parameters for every deployment. + +## 5. Benchmarking Rules + +To reduce measurement noise: + +1. Create test data before the measured run. Do not include fixture setup time + in operator duration. +2. Run at least one warm-up execution for each plan/cardinality pair. +3. Repeat each measured scenario multiple times. +4. Store every individual measurement instead of storing only averages. +5. Record exact input and output cardinalities. +6. Keep platform settings stable, including worker count, JVM settings, memory + limits, and connector configuration. +7. Record abnormal runs, such as failures caused by GC, cold cache, network + issues, or competing workloads. + +For distributed systems such as Trino, it is also important to define what the +model should predict: + +- If Wayang should predict user-visible runtime, fit wall-clock elapsed time. +- If the platform reports CPU time and the model uses CPU load, make sure the + conversion to Wayang cost is consistent with the resource model. +- Parameters learned on a local Docker setup should be treated as local + reference values, not universal defaults for every deployment. + +## 6. Running a Profiling Experiment + +The exact command depends on the platform module, test class, and property +prefix. + +| Platform | Setup guide | Maven module | Test class | Property prefix | Default output directory | +|----------|-------------|--------------|------------|-----------------|--------------------------| +| Trino | `trino-setup/README.md` | `wayang-platforms/wayang-trino` | `TrinoCostPilotIT` | `trino.profile.*` | `target/cost-profiling/trino` | +| Presto | `presto-setup/README.md` | `wayang-platforms/wayang-presto` | `PrestoCostPilotIT` | `presto.profile.*` | `target/cost-profiling/presto` | +| BigQuery | `bigquery-setup/README.md` | `wayang-platforms/wayang-bigquery` | `BigQueryCostPilotIT` | `bigquery.profile.*` | `target/cost-profiling/bigquery` | + +The commands below use PowerShell. On macOS/Linux, use `./mvnw` instead of +`.\mvnw.cmd` and replace PowerShell backticks with Bash line-continuation +backslashes. + +Trino: + +```powershell +.\mvnw.cmd -Pskip-prerequisite-check -pl wayang-platforms/wayang-trino -am ` + "-Dtest=TrinoCostPilotIT" ` + "-Dsurefire.failIfNoSpecifiedTests=false" ` + "-DfailIfNoTests=false" ` + "-Dtrino.profile.outputDir=target/cost-profiling/trino" ` + "-Dtrino.profile.rowCounts=10000,50000,100000,250000" ` + "-Dtrino.profile.plans=S01,S02,S03,S04,S05,S06,S07,S08,S09,S10,S11,S12,S13,S14,S15,S16" ` + "-Dtrino.profile.repetitions=6" ` + "-Dtrino.profile.reset=true" ` + "-Drat.skip=true" ` + "-Dlicense.skip=true" ` + "-Dmaven.javadoc.skip=true" ` + test +``` + +Presto: + +```powershell +.\mvnw.cmd -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am ` + "-Dtest=PrestoCostPilotIT" ` + "-Dsurefire.failIfNoSpecifiedTests=false" ` + "-DfailIfNoTests=false" ` + "-Dpresto.profile.outputDir=target/cost-profiling/presto" ` + "-Dpresto.profile.rowCounts=10000,50000,100000,250000" ` + "-Dpresto.profile.plans=S01,S02,S03,S04,S05,S06,S07,S08,S09,S10,S11,S12,S13,S14,S15,S16" ` + "-Dpresto.profile.repetitions=6" ` + "-Dpresto.profile.reset=true" ` + "-Drat.skip=true" ` + "-Dlicense.skip=true" ` + "-Dmaven.javadoc.skip=true" ` + test +``` + +BigQuery: + +```powershell +.\mvnw.cmd -Pskip-prerequisite-check -pl wayang-platforms/wayang-bigquery -am ` + "-Dtest=BigQueryCostPilotIT" ` + "-Dsurefire.failIfNoSpecifiedTests=false" ` + "-DfailIfNoTests=false" ` + "-Dbigquery.project=YOUR_PROJECT_ID" ` + "-Dbigquery.saEmail=wayang-bq@YOUR_PROJECT_ID.iam.gserviceaccount.com" ` + "-Dbigquery.keyPath=C:\path\to\wayang-bq-key.json" ` + "-Dbigquery.location=US" ` + "-Dbigquery.profile.outputDir=target/cost-profiling/bigquery" ` + "-Dbigquery.profile.rowCounts=10000,50000,100000,250000" ` + "-Dbigquery.profile.plans=S01,S02,S03,S04,S05,S06,S07,S08,S09,S10,S11,S12,S13,S14,S15,S16" ` + "-Dbigquery.profile.repetitions=6" ` + "-Dbigquery.profile.reset=true" ` + "-Drat.skip=true" ` + "-Dlicense.skip=true" ` + "-Dmaven.javadoc.skip=true" ` + test +``` + +Expected output files: + +| File | Purpose | +|------|---------| +| `executions.json` | Wayang execution records consumed by the GA profiler | +| `manifest.csv` | Human-readable mapping from run ID to plan, cardinality, repetition, and status | + +## 7. Running the Genetic Optimizer + +The entry point for learning cost parameters is: + +```text +org.apache.wayang.profiler.log.GeneticOptimizerApp +``` + +A typical profiling configuration contains: + +- platform default properties; +- `wayang..*.load.template` formulas; +- GA settings; +- the path to `executions.json`; +- the output path for learned parameters. + +Example GA settings: + +```properties +wayang.profiler.ga.timelimit.ms = 120000 +wayang.profiler.ga.maxgenerations = 800 +wayang.profiler.ga.maxstablegenerations = 150 +wayang.profiler.ga.superoptimizations = 1 +wayang.profiler.ga.intermediateupdate = 200 +wayang.profiler.ga.min-exec-time = 1 +wayang.profiler.ga.max-cardinality-spread = 100 +wayang.profiler.ga.min-cardinality-confidence = 0 +wayang.profiler.ga.binning = 1.0 +wayang.profiler.ga.output-file = +``` + +The profiler writes learned formulas such as: + +```properties +wayang.trino.tablesource.load = ... +wayang.trino.filter.load = ... +wayang.trino.join.load = ... +``` + +## 8. Cardinality Estimation Note + +For JDBC-based table sources, `JdbcTableSource#getCardinalityEstimator` may open +a JDBC connection and run: + +```sql +SELECT count(*) FROM +``` + +This is used during Wayang's optimization phase to estimate source +cardinalities. + +Important details: + +- The estimator is not called for every registered platform. +- It is called only for operators that appear in the current Wayang plan or plan + implementation being estimated. +- If the current plan contains a Trino, Presto, or BigQuery table source, the + corresponding JDBC cardinality estimator may run. +- If the count query fails, the current implementation falls back to a + conservative estimate. + +For cloud platforms, this extra count query can add overhead or fail because of +network or authentication issues. For profiling, it can be useful to support +cached or user-provided source cardinalities in the future. + +## 9. Completion Criteria + +A profiling run is complete when: + +- the platform execution stage records a `PartialExecution`; +- `executions.json` contains the expected platform; +- execution records contain estimator keys for relevant operators such as + `tablesource`, `filter`, `projection`, `join`, `reduceby`, `sort`, and + `tablesink`; +- input and output cardinalities are available; +- `GeneticOptimizerApp` can read the execution log; +- the profiler outputs learned platform load formulas; +- the learned formulas and experiment settings are documented together so they + can be interpreted as environment-specific profiling results. + +## 10. Recommended Implementation Order + +When adding profiling support for a new platform, a conservative order is: + +1. Create a minimal proof of concept for one stage, for example + `TableSource -> Filter -> TableSink`. +2. Confirm that `executions.json` contains the correct platform, estimator keys, + cardinalities, and measured duration. +3. Make sure the profiler can initialize the platform and deserialize its + execution records. +4. Run a small benchmark and generate candidate parameters. +5. Extend the workload to all important operators and combined pipelines. +6. Decide whether the learned parameters should become reference defaults or + remain documented as environment-specific profiling results. diff --git a/trino-setup/README.md b/trino-setup/README.md new file mode 100644 index 000000000..56dcca59b --- /dev/null +++ b/trino-setup/README.md @@ -0,0 +1,283 @@ +# Trino Local Setup + +Local Trino environment backed by an **Iceberg** data lake, completely containerised. + +The current validation has three parts: + +1. Build the Wayang Trino platform and run the shared JDBC SQL-generation tests. +2. Run the Wayang Trino operator tests against the live local stack. +3. Run standalone JDBC integration tests against the local Trino, Iceberg, and MinIO stack. + +Run the commands below from the repository root. Java 17 and Docker with +Docker Compose are required; Maven is provided by the repository wrapper. + +The Trino cost-profiling branch is named `feature/trino-cost-profiling`: + +```bash +git checkout feature/trino-cost-profiling +``` + +## Command Conventions + +Use the `bash` blocks on macOS/Linux terminals. Use the `powershell` blocks on +Windows PowerShell from the repository root. Docker Compose commands are the +same on both platforms. + +## Stack + +| Component | Image | Port | Role | +|-----------|-------|------|------| +| **Trino** | `trinodb/trino:435` | 8080 | SQL query engine | +| **Hive Metastore** | `naushadh/hive-metastore:latest` | 9083 | Iceberg table catalog (Thrift) | +| **PostgreSQL** | `postgres:15-alpine` | 5432 | HMS metadata backing store | +| **MinIO** | `minio/minio:latest` | 9000 / 9001 | S3-compatible object storage | + +HMS is the battle-tested Iceberg catalog for Trino. Parquet data files are written by Trino directly to MinIO; HMS only stores schema/table metadata. + +## Directory Layout + +``` +trino-setup/ +|-- docker-compose.yml # Full stack definition +|-- trino/ +| |-- config.properties # Trino node config +| `-- catalog/ +| |-- iceberg.properties # Iceberg via HMS + MinIO +| `-- tpch.properties # Built-in TPC-H (no storage needed) +|-- scripts/ +| |-- init.sql # Creates iceberg.sales.orders + sample rows +| `-- run-init.sh # Helper: waits for Trino then runs init.sql +|-- pom.xml # Standalone Maven project (Java 17) +`-- src/test/java/.../ + `-- TrinoIntegrationTest.java # JUnit 5 integration tests +``` + +## 1. Test the Wayang Trino Platform + +Build the Trino platform and its required modules: + +```bash +./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-trino -am -DskipTests -Drat.skip=true test +``` + +On PowerShell: + +```powershell +.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-trino -am -DskipTests -Drat.skip=true test +``` + +Then run the shared JDBC SQL-generation tests: + +```bash +./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-jdbc-template -am -Dtest=JdbcExecutorTest -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false -Drat.skip=true test +``` + +On PowerShell: + +```powershell +.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-jdbc-template -am -Dtest=JdbcExecutorTest -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false -Drat.skip=true test +``` + +Expected result: + +```text +Wayang Platform Trino ... SUCCESS +Tests run: 4, Failures: 0, Errors: 0, Skipped: 0 +``` + +## 2. Test Against the Local Trino Stack + +### 1. Start the stack + +```bash +docker compose -f trino-setup/docker-compose.yml up -d +``` + +Wait ~30 seconds for all services to become healthy. Check with: + +```bash +docker compose -f trino-setup/docker-compose.yml ps +# or watch the Trino UI at http://localhost:8080 +``` + +### 2. Run the Wayang Trino operator tests + +`TrinoOperatorsIT` exercises the Wayang Trino implementation against the live +Trino stack. It checks `TableSource`, `Filter`, `Projection`, `Join`, +`GlobalReduce`, `ReduceBy`, `Sort`, and `TableSink`, and confirms that the +expected SQL reached Trino. The standalone join test now runs a full Wayang +plan and normalizes both possible join result shapes before collecting records: +logical joins can produce `Tuple2`, while pushed-down JDBC joins +can return a flat `Record`. + +The suite is self-contained: it creates `iceberg.wayang_it`, scales its test +data to 120,000 rows so the optimizer selects SQL pushdown, and drops its test +tables afterward. It does not require `scripts/init.sql`. The suite also +contains five JavaPlanBuilder `readTable` combination tests that cover filter, +projection, global reduce, reduce-by plus sort, table sink, and join through +the public API. + +```bash +./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-trino -am \ + -Dtest=TrinoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false \ + -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test +``` + +On PowerShell: + +```powershell +.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-trino -am -Dtest=TrinoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test +``` + +Expected result: + +```text +Tests run: 13, Failures: 0, Errors: 0, Skipped: 0 +``` + +Verified on June 18, 2026 against the local Docker stack with the full-plan +join test and all five JavaPlanBuilder combination tests enabled. + +If Trino is unreachable, these tests are skipped instead of failed. A result +with skipped tests does not confirm that the operators work. + +### 3. Load sample Iceberg data + +```bash +bash trino-setup/scripts/run-init.sh +``` + +On PowerShell: + +```powershell +Get-Content -Raw trino-setup/scripts/init.sql | docker exec -i trino trino --server http://localhost:8080 --user admin +``` + +This creates the schema `iceberg.sales` and inserts 20 sample orders into +`iceberg.sales.orders` (Parquet files on MinIO). + +### 4. Run the standalone stack integration tests + +```bash +./mvnw -f trino-setup/pom.xml -Pintegration -Dtest=TrinoIntegrationTest test +``` + +On PowerShell: + +```powershell +.\mvnw.cmd --% -f trino-setup/pom.xml -Pintegration -Dtest=TrinoIntegrationTest test +``` + +Tests are skipped by default (no `-Pintegration`) to avoid requiring Docker in CI. +These tests validate the stack and direct JDBC queries independently of the +Wayang operator implementation. + +Expected result: + +```text +Tests run: 10, Failures: 0, Errors: 0, Skipped: 0 +BUILD SUCCESS +``` + +### 5. Manual exploration + +Open the **Trino UI**: http://localhost:8080 + +Or connect via the Trino CLI inside the container: + +```bash +docker exec -it trino trino --catalog iceberg --schema sales +``` + +```sql +-- TPC-H built-in data (no init.sql needed) +SELECT * FROM tpch.tiny.orders LIMIT 5; + +-- Iceberg table +SELECT region, SUM(amount) FROM iceberg.sales.orders GROUP BY region; + +-- Iceberg file metadata +SELECT * FROM iceberg.sales."orders$files"; + +-- Iceberg history +SELECT * FROM iceberg.sales."orders$history"; +``` + +**MinIO console**: http://localhost:9001 (login: `minioadmin` / `minioadmin`) +Look for Parquet files under `warehouse/sales/orders/`. + +### 6. Tear down + +```bash +docker compose -f trino-setup/docker-compose.yml down -v +``` + +The `-v` option removes volumes and clears the local MinIO and PostgreSQL data. + +## Test Coverage + +### Wayang operator integration tests + +| Test | What it checks | +|------|----------------| +| `tableSource` | Full table scan through `TrinoTableSource` | +| `filter` | Wayang `FilterOperator` and SQL `WHERE` pushdown | +| `projection` | Column projection pushed into the Trino query | +| `join` | Full Wayang join plan with normalization before the collecting sink | +| `globalReduce` | Global aggregation such as `SUM` | +| `reduceBy` | Grouped aggregation and SQL `GROUP BY` | +| `sort` | Wayang sort and SQL `ORDER BY` | +| `tableSink` | Filtered result written with `CREATE TABLE AS` | +| `javaPlanBuilderReadTableFilterProjection` | `readTable -> filter -> projection -> collect` | +| `javaPlanBuilderReadTableFilterGlobalReduce` | `readTable -> filter -> globalReduce -> collect` | +| `javaPlanBuilderReadTableReduceBySort` | `readTable -> reduceByKey -> sort -> collect` | +| `javaPlanBuilderReadTableFilterProjectionTableSink` | `readTable -> filter -> projection -> writeTable` | +| `javaPlanBuilderReadTableJoin` | `readTable + readTable -> join -> collect` | + +### Standalone stack integration tests + +| Test | What it checks | +|------|----------------| +| `testConnectivity` | `SELECT 1`, JDBC connection works | +| `testTpchConnector` | TPC-H built-in connector, no storage needed | +| `testTpchTopOrders` | ORDER BY + LIMIT on TPC-H | +| `testIcebergSchemaVisible` | Schema created by `init.sql` is visible | +| `testIcebergSelectAll` | Full table scan, 20 rows | +| `testIcebergFilterByRegion` | WHERE pushdown on string column | +| `testIcebergAggregate` | GROUP BY + SUM aggregation | +| `testIcebergFilterByAmount` | WHERE pushdown on double column | +| `testIcebergProjection` | SELECT subset of columns | +| `testIcebergFilesMetadata` | `$files` system table, confirms Parquet on MinIO | + +## Cost Profiling + +Follow the shared cost-profiling guide in +[`guides/cost-profiling.md`](../guides/cost-profiling.md). This setup guide +only covers the Trino stack itself. + +Trino-specific profiling values: + +| Item | Value | +|------|-------| +| Maven module | `wayang-platforms/wayang-trino` | +| Profiling test | `TrinoCostPilotIT` | +| Property prefix | `trino.profile.*` | +| Default output directory | `target/cost-profiling/trino` | +| Learned parameters file | `wayang-platforms/wayang-trino/src/main/resources/wayang-trino-defaults.properties` | + +## Environment Variables + +Override defaults if running Trino on a different host/port: + +```bash +TRINO_HOST=my-trino-host TRINO_PORT=8080 ./mvnw -f trino-setup/pom.xml -Pintegration -Dtest=TrinoIntegrationTest test +``` + +On PowerShell: + +```powershell +$env:TRINO_HOST="my-trino-host" +$env:TRINO_PORT="8080" +.\mvnw.cmd --% -f trino-setup/pom.xml -Pintegration -Dtest=TrinoIntegrationTest test +Remove-Item Env:TRINO_HOST, Env:TRINO_PORT +``` diff --git a/trino-setup/demo.sh b/trino-setup/demo.sh new file mode 100644 index 000000000..ea3df7892 --- /dev/null +++ b/trino-setup/demo.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +WAYANG_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" +TRINO_SETUP="$SCRIPT_DIR" +TRINO_CONTAINER="trino" +MAVEN_FLAGS="-Pskip-prerequisite-check -Drat.skip=true -Dmaven.javadoc.skip=true" + +banner() { + echo + echo "============================================================" + printf " %s\n" "$*" + echo "============================================================" + echo +} + +step() { + echo + echo "-- $*" + echo +} + +pause() { + if [[ "${WAYANG_DEMO_AUTO:-false}" != "true" ]]; then + echo + read -rp "Press ENTER to continue..." _ || true + echo + fi +} + +run_wayang_demo() { + "$WAYANG_ROOT/mvnw" exec:java -pl wayang-platforms/wayang-trino \ + -Dexec.mainClass="org.apache.wayang.trino.TrinoDemo" \ + ${MAVEN_FLAGS} +} + +banner "ACT 1: Start Trino + Iceberg via Docker" + +step "1a. Starting the stack" +cd "$TRINO_SETUP" +docker compose up -d + +step "1b. Containers running" +docker ps --format "table {{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}" \ + | grep -E "NAMES|trino|minio|metastore|postgres" + +step "1c. Waiting for Trino to be ready" +MAX_WAIT=90 +ELAPSED=0 +until docker exec "$TRINO_CONTAINER" \ + trino --execute "SELECT 1" --output-format ALIGNED >/dev/null 2>&1; do + if [[ "$ELAPSED" -ge "$MAX_WAIT" ]]; then + echo "Timed out waiting for Trino after ${MAX_WAIT}s" + exit 1 + fi + printf ". waiting (%ds elapsed)\r" "$ELAPSED" + sleep 3 + ELAPSED=$((ELAPSED + 3)) +done +echo "Trino is ready at http://localhost:8080" + +step "1d. Initialising Iceberg tables" +docker exec -i "$TRINO_CONTAINER" trino < "$TRINO_SETUP/scripts/init.sql" 2>&1 \ + | grep -v "^WARNING\|jline\|org.jline" || true +echo "iceberg.sales.orders seeded" + +step "1e. Table schema" +docker exec "$TRINO_CONTAINER" \ + trino --execute "DESCRIBE iceberg.sales.orders" \ + --output-format ALIGNED 2>/dev/null + +pause + +banner "ACT 2: Query Iceberg directly via Trino CLI" + +step "2a. Full table scan" +echo "SQL: SELECT * FROM iceberg.sales.orders" +docker exec "$TRINO_CONTAINER" \ + trino --execute "SELECT * FROM iceberg.sales.orders ORDER BY order_id" \ + --output-format ALIGNED 2>/dev/null + +step "2b. Filter: region = 'AMER'" +echo "SQL: SELECT * FROM iceberg.sales.orders WHERE region = 'AMER'" +docker exec "$TRINO_CONTAINER" \ + trino --execute "SELECT * FROM iceberg.sales.orders WHERE region = 'AMER' ORDER BY order_id" \ + --output-format ALIGNED 2>/dev/null + +step "2c. Projection with filter" +echo "SQL: SELECT region, product, amount FROM iceberg.sales.orders WHERE region = 'AMER'" +docker exec "$TRINO_CONTAINER" \ + trino --execute \ + "SELECT region, product, amount + FROM iceberg.sales.orders + WHERE region = 'AMER' + ORDER BY order_id" \ + --output-format ALIGNED 2>/dev/null + +pause + +banner "ACT 3: Wayang API filter + projection pushdown" +cd "$WAYANG_ROOT" +run_wayang_demo + +banner "Demo complete" +echo "Trino UI: http://localhost:8080" +echo "MinIO UI: http://localhost:9001 (minioadmin / minioadmin)" +echo +echo "To stop the stack:" +echo " cd trino-setup && docker compose down" diff --git a/trino-setup/docker-compose.yml b/trino-setup/docker-compose.yml new file mode 100644 index 000000000..0c057540e --- /dev/null +++ b/trino-setup/docker-compose.yml @@ -0,0 +1,141 @@ +--- +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Stack: Trino + Hive Metastore + MinIO (S3 storage) +# +# This is the battle-tested Trino + Iceberg local setup. +# Hive Metastore (HMS) stores Iceberg table metadata over Thrift on port 9083. +# MinIO provides S3-compatible object storage for Parquet data files. +# Trino's Iceberg connector uses HMS as catalog and writes Parquet to MinIO. +# +# Ports: +# Trino: http://localhost:8080 (UI + JDBC) +# MinIO S3: http://localhost:9000 +# MinIO UI: http://localhost:9001 (minioadmin / minioadmin) +# HMS: localhost:9083 (Thrift, internal) +# Postgres: localhost:5432 (HMS backing store) + +services: + + # ── PostgreSQL (Hive Metastore backing database) ─────────────────────────── + postgres: + image: postgres:15-alpine + container_name: trino-postgres + environment: + POSTGRES_DB: metastore + POSTGRES_USER: hive + POSTGRES_PASSWORD: hive + ports: + - "5432:5432" + volumes: + - postgres-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U hive -d metastore"] + interval: 10s + timeout: 5s + retries: 5 + + # ── MinIO (S3-compatible object storage) ────────────────────────────────── + minio: + image: minio/minio:latest + container_name: trino-minio + environment: + MINIO_ROOT_USER: minioadmin + MINIO_ROOT_PASSWORD: minioadmin + ports: + - "9000:9000" + - "9001:9001" + command: server /data --console-address ":9001" + volumes: + - minio-data:/data + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + interval: 10s + timeout: 5s + retries: 5 + + # Create the warehouse bucket before HMS starts + minio-init: + image: minio/mc:latest + container_name: trino-minio-init + depends_on: + minio: + condition: service_healthy + entrypoint: > + /bin/sh -c " + mc alias set local http://minio:9000 minioadmin minioadmin; + mc mb local/warehouse --ignore-existing; + echo 'bucket warehouse ready'; + exit 0; + " + + # ── Hive Metastore ──────────────────────────────────────────────────────── + # naushadh/hive-metastore is a minimal, pre-configured HMS image + # that supports S3-compatible storage via env vars. + metastore: + image: naushadh/hive-metastore:latest + container_name: trino-metastore + depends_on: + postgres: + condition: service_healthy + minio: + condition: service_healthy + minio-init: + condition: service_completed_successfully + ports: + - "9083:9083" + environment: + DATABASE_HOST: postgres + DATABASE_DB: metastore + DATABASE_USER: hive + DATABASE_PASSWORD: hive + # S3 / MinIO + S3_ENDPOINT_URL: http://minio:9000 + S3_BUCKET: warehouse + S3_PREFIX: / + AWS_ACCESS_KEY_ID: minioadmin + AWS_SECRET_ACCESS_KEY: minioadmin + REGION: us-east-1 + # No nc/curl in this image; use bash's /dev/tcp built-in + healthcheck: + test: ["CMD", "/bin/bash", "-c", "exec 3<>/dev/tcp/localhost/9083 2>/dev/null && exit 0 || exit 1"] + interval: 15s + timeout: 10s + retries: 15 + + # ── Trino ───────────────────────────────────────────────────────────────── + trino: + image: trinodb/trino:435 + container_name: trino + depends_on: + metastore: + condition: service_healthy + minio: + condition: service_healthy + ports: + - "8080:8080" + volumes: + - ./trino/catalog:/etc/trino/catalog + - ./trino/config.properties:/etc/trino/config.properties + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8080/v1/info"] + interval: 15s + timeout: 10s + retries: 10 + +volumes: + postgres-data: + minio-data: diff --git a/trino-setup/pom.xml b/trino-setup/pom.xml new file mode 100644 index 000000000..f04955203 --- /dev/null +++ b/trino-setup/pom.xml @@ -0,0 +1,93 @@ + + + + 4.0.0 + + org.apache.wayang + trino-setup + 1.0-SNAPSHOT + jar + + Trino Local Setup — Integration Tests + + Standalone integration tests for a local Trino stack + (Trino + Nessie Iceberg catalog + MinIO S3 storage). + Independent of the Wayang codebase. + + + + 17 + 17 + UTF-8 + 435 + 5.10.2 + + + + + + io.trino + trino-jdbc + ${trino.version} + test + + + + + org.junit.jupiter + junit-jupiter + ${junit.version} + test + + + + + org.slf4j + slf4j-simple + 2.0.12 + test + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + 3.2.5 + + + ${skipIntegrationTests} + + + + + + + + + integration + + false + + + + diff --git a/trino-setup/scripts/init.sql b/trino-setup/scripts/init.sql new file mode 100644 index 000000000..245ffbce3 --- /dev/null +++ b/trino-setup/scripts/init.sql @@ -0,0 +1,66 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +-- Run this after the stack is up to create sample Iceberg tables. +-- Usage: ./scripts/run-init.sh +-- Or manually: docker exec -it trino trino < /scripts/init.sql + +-- Schema +CREATE SCHEMA IF NOT EXISTS iceberg.sales; + +-- Orders table (Iceberg / Parquet on MinIO) +CREATE TABLE IF NOT EXISTS iceberg.sales.orders ( + order_id BIGINT, + region VARCHAR, + product VARCHAR, + amount DOUBLE, + order_date DATE +) +WITH (format = 'PARQUET'); + +-- Idempotent seed: clear before inserting so re-runs do not duplicate rows. +DELETE FROM iceberg.sales.orders; + +-- Sample data: 20 rows, 4 regions (AMER/APAC/EMEA/LATAM), 5 products. +-- AMER rows: 3, 6, 9, 12, 16 -> 5 rows for filter demo +-- Projection demo selects only: region, product, amount +INSERT INTO iceberg.sales.orders VALUES + (1, 'APAC', 'Widget A', 1500.00, DATE '2024-01-15'), + (2, 'EMEA', 'Widget B', 800.50, DATE '2024-01-16'), + (3, 'AMER', 'Widget A', 2200.00, DATE '2024-01-17'), + (4, 'APAC', 'Widget C', 350.75, DATE '2024-01-18'), + (5, 'EMEA', 'Widget A', 1100.00, DATE '2024-01-19'), + (6, 'AMER', 'Widget B', 950.25, DATE '2024-01-20'), + (7, 'APAC', 'Widget B', 1750.00, DATE '2024-01-21'), + (8, 'EMEA', 'Widget C', 420.00, DATE '2024-01-22'), + (9, 'AMER', 'Widget C', 680.50, DATE '2024-01-23'), + (10, 'APAC', 'Widget A', 3000.00, DATE '2024-01-24'), + (11, 'LATAM', 'Widget D', 560.00, DATE '2024-01-25'), + (12, 'AMER', 'Widget D', 1320.75, DATE '2024-01-26'), + (13, 'EMEA', 'Widget D', 990.00, DATE '2024-01-27'), + (14, 'LATAM', 'Widget E', 2100.50, DATE '2024-01-28'), + (15, 'APAC', 'Widget E', 4500.00, DATE '2024-01-29'), + (16, 'AMER', 'Widget E', 3750.00, DATE '2024-01-30'), + (17, 'EMEA', 'Widget E', 1250.00, DATE '2024-01-31'), + (18, 'LATAM', 'Widget A', 870.25, DATE '2024-02-01'), + (19, 'APAC', 'Widget D', 1680.00, DATE '2024-02-02'), + (20, 'LATAM', 'Widget B', 440.50, DATE '2024-02-03'); + +-- Verify +SELECT region, COUNT(*) AS order_count, SUM(amount) AS total_amount +FROM iceberg.sales.orders +GROUP BY region +ORDER BY total_amount DESC; diff --git a/trino-setup/scripts/run-init.sh b/trino-setup/scripts/run-init.sh new file mode 100644 index 000000000..91d279192 --- /dev/null +++ b/trino-setup/scripts/run-init.sh @@ -0,0 +1,39 @@ +#!/bin/bash +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Runs init.sql against the local Trino instance. +# The stack must be fully up before running this. + +set -e + +TRINO_HOST=${TRINO_HOST:-localhost} +TRINO_PORT=${TRINO_PORT:-8080} + +echo "Waiting for Trino to be ready..." +until curl -sf "http://${TRINO_HOST}:${TRINO_PORT}/v1/info" | grep -q '"starting":false'; do + echo " Trino not ready yet, retrying in 5s..." + sleep 5 +done +echo "Trino is ready." + +echo "Running init.sql..." +docker exec -i trino trino \ + --server "http://${TRINO_HOST}:${TRINO_PORT}" \ + --user admin \ + < "$(dirname "$0")/init.sql" + +echo "Done. Sample Iceberg data loaded into iceberg.sales.orders" diff --git a/trino-setup/src/test/java/org/apache/wayang/trino/TrinoIntegrationTest.java b/trino-setup/src/test/java/org/apache/wayang/trino/TrinoIntegrationTest.java new file mode 100644 index 000000000..081beea99 --- /dev/null +++ b/trino-setup/src/test/java/org/apache/wayang/trino/TrinoIntegrationTest.java @@ -0,0 +1,232 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.wayang.trino; + +import org.junit.jupiter.api.*; + +import java.sql.*; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Integration tests for the local Trino stack. + * + * Prerequisites: run `docker-compose up -d` and `./scripts/run-init.sh` first. + * + * Run tests: + * mvn test -Pintegration + * + * Or skip infrastructure setup and run with a custom host: + * TRINO_HOST=localhost TRINO_PORT=8080 mvn test -Pintegration + */ +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +class TrinoIntegrationTest { + + private static final String TRINO_HOST = System.getenv().getOrDefault("TRINO_HOST", "localhost"); + private static final int TRINO_PORT = Integer.parseInt(System.getenv().getOrDefault("TRINO_PORT", "8080")); + private static final String JDBC_URL = String.format("jdbc:trino://%s:%d", TRINO_HOST, TRINO_PORT); + + private static Connection connection; + + // ── Lifecycle ───────────────────────────────────────────────────────── + + @BeforeAll + static void openConnection() throws Exception { + Properties props = new Properties(); + props.setProperty("user", "admin"); // Trino requires a non-empty user + connection = DriverManager.getConnection(JDBC_URL, props); + System.out.printf("Connected to Trino at %s%n", JDBC_URL); + } + + @AfterAll + static void closeConnection() throws Exception { + if (connection != null && !connection.isClosed()) { + connection.close(); + } + } + + // ── Helper ──────────────────────────────────────────────────────────── + + private List> query(String sql) throws SQLException { + List> rows = new ArrayList<>(); + try (Statement stmt = connection.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + int cols = rs.getMetaData().getColumnCount(); + while (rs.next()) { + List row = new ArrayList<>(); + for (int i = 1; i <= cols; i++) row.add(rs.getObject(i)); + rows.add(row); + } + } + return rows; + } + + // ── Test 1: Basic connectivity ──────────────────────────────────────── + + @Test + @Order(1) + @DisplayName("Trino responds to a simple SELECT 1") + void testConnectivity() throws SQLException { + List> rows = query("SELECT 1"); + assertEquals(1, rows.size()); + assertEquals(1L, ((Number) rows.get(0).get(0)).longValue()); + System.out.println("[PASS] Basic connectivity OK"); + } + + // ── Test 2: TPC-H built-in connector ───────────────────────────────── + + @Test + @Order(2) + @DisplayName("TPC-H tiny catalog: count orders") + void testTpchConnector() throws SQLException { + List> rows = query("SELECT COUNT(*) FROM tpch.tiny.orders"); + long count = ((Number) rows.get(0).get(0)).longValue(); + assertTrue(count > 0, "tpch.tiny.orders should have rows"); + System.out.printf("[PASS] TPC-H tiny.orders has %,d rows%n", count); + } + + @Test + @Order(3) + @DisplayName("TPC-H tiny catalog: top 5 orders by total price") + void testTpchTopOrders() throws SQLException { + List> rows = query(""" + SELECT orderkey, totalprice + FROM tpch.tiny.orders + ORDER BY totalprice DESC + LIMIT 5 + """); + assertEquals(5, rows.size(), "Expected exactly 5 rows"); + System.out.println("[PASS] TPC-H top 5 orders:"); + rows.forEach(r -> System.out.printf(" orderkey=%s totalprice=%s%n", r.get(0), r.get(1))); + } + + // ── Test 4: Iceberg — schema exists ────────────────────────────────── + + @Test + @Order(4) + @DisplayName("Iceberg catalog: schema 'sales' is visible") + void testIcebergSchemaVisible() throws SQLException { + List> rows = query("SHOW SCHEMAS IN iceberg LIKE 'sales'"); + assertFalse(rows.isEmpty(), "Schema 'sales' should exist in iceberg catalog. " + + "Did you run scripts/run-init.sh?"); + System.out.println("[PASS] Iceberg schema 'sales' is visible"); + } + + // ── Test 5: Iceberg — full table scan ──────────────────────────────── + + @Test + @Order(5) + @DisplayName("Iceberg table: select all orders") + void testIcebergSelectAll() throws SQLException { + List> rows = query("SELECT * FROM iceberg.sales.orders ORDER BY order_id"); + assertEquals(20, rows.size(), "Expected 20 rows inserted by init.sql"); + System.out.println("[PASS] Iceberg full scan: 20 rows"); + rows.forEach(r -> System.out.printf(" %s%n", r)); + } + + // ── Test 6: Iceberg — pushdown filter ──────────────────────────────── + + @Test + @Order(6) + @DisplayName("Iceberg table: filter by region = APAC") + void testIcebergFilterByRegion() throws SQLException { + List> rows = query(""" + SELECT order_id, region, amount + FROM iceberg.sales.orders + WHERE region = 'APAC' + ORDER BY order_id + """); + assertFalse(rows.isEmpty(), "Should have APAC orders"); + rows.forEach(r -> assertEquals("APAC", r.get(1), "All rows must be APAC")); + System.out.printf("[PASS] Filter pushdown: %d APAC rows%n", rows.size()); + } + + // ── Test 7: Iceberg — aggregation ──────────────────────────────────── + + @Test + @Order(7) + @DisplayName("Iceberg table: aggregate total_amount by region") + void testIcebergAggregate() throws SQLException { + List> rows = query(""" + SELECT region, COUNT(*) AS order_count, SUM(amount) AS total_amount + FROM iceberg.sales.orders + GROUP BY region + ORDER BY total_amount DESC + """); + assertFalse(rows.isEmpty(), "Aggregation should return rows"); + System.out.println("[PASS] Aggregation by region:"); + rows.forEach(r -> System.out.printf(" region=%-5s count=%s total=%.2f%n", + r.get(0), r.get(1), ((Number) r.get(2)).doubleValue())); + } + + // ── Test 8: Iceberg — amount threshold filter ───────────────────────── + + @Test + @Order(8) + @DisplayName("Iceberg table: filter orders with amount > 1000") + void testIcebergFilterByAmount() throws SQLException { + List> rows = query(""" + SELECT order_id, amount + FROM iceberg.sales.orders + WHERE amount > 1000.0 + ORDER BY amount DESC + """); + rows.forEach(r -> assertTrue( + ((Number) r.get(1)).doubleValue() > 1000.0, + "All rows must have amount > 1000" + )); + System.out.printf("[PASS] Amount filter: %d rows with amount > 1000%n", rows.size()); + } + + // ── Test 9: Iceberg — projection (select subset of columns) ─────────── + + @Test + @Order(9) + @DisplayName("Iceberg table: project only region and product columns") + void testIcebergProjection() throws SQLException { + List> rows = query(""" + SELECT region, product + FROM iceberg.sales.orders + LIMIT 5 + """); + assertEquals(5, rows.size()); + rows.forEach(r -> { + assertNotNull(r.get(0), "region should not be null"); + assertNotNull(r.get(1), "product should not be null"); + }); + System.out.println("[PASS] Projection (region, product): 5 rows returned"); + } + + // ── Test 10: Iceberg metadata — table files ────────────────────────── + + @Test + @Order(10) + @DisplayName("Iceberg metadata: $files table lists at least one Parquet file") + void testIcebergFilesMetadata() throws SQLException { + List> rows = query(""" + SELECT file_path, record_count + FROM iceberg.sales."orders$files" + """); + assertFalse(rows.isEmpty(), "There should be at least one data file after inserts"); + System.out.println("[PASS] Iceberg $files metadata:"); + rows.forEach(r -> System.out.printf(" %s (records=%s)%n", r.get(0), r.get(1))); + } +} diff --git a/trino-setup/trino/catalog/iceberg.properties b/trino-setup/trino/catalog/iceberg.properties new file mode 100644 index 000000000..5aabe5275 --- /dev/null +++ b/trino-setup/trino/catalog/iceberg.properties @@ -0,0 +1,32 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Iceberg catalog backed by Hive Metastore (HMS) and MinIO (S3 storage). +# HMS stores Iceberg table metadata; Trino writes Parquet files to MinIO via S3. +connector.name=iceberg +iceberg.catalog.type=hive_metastore +hive.metastore.uri=thrift://metastore:9083 + +# Native S3 filesystem handles both s3:// and s3a:// (which HMS uses internally). +# This avoids the "No FileSystem for scheme s3" error when HMS assigns locations. +fs.native-s3.enabled=true +s3.endpoint=http://minio:9000 +s3.path-style-access=true +s3.aws-access-key=minioadmin +s3.aws-secret-key=minioadmin +s3.region=us-east-1 + +iceberg.file-format=PARQUET diff --git a/trino-setup/trino/catalog/tpch.properties b/trino-setup/trino/catalog/tpch.properties new file mode 100644 index 000000000..ee4f23783 --- /dev/null +++ b/trino-setup/trino/catalog/tpch.properties @@ -0,0 +1,20 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Built-in TPC-H connector with no external dependencies. +# Useful for testing basic Trino connectivity without any storage setup. +# Usage: SELECT * FROM tpch.tiny.orders LIMIT 10; +connector.name=tpch diff --git a/trino-setup/trino/config.properties b/trino-setup/trino/config.properties new file mode 100644 index 000000000..8584e4aea --- /dev/null +++ b/trino-setup/trino/config.properties @@ -0,0 +1,20 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +coordinator=true +node-scheduler.include-coordinator=true +http-server.http.port=8080 +discovery.uri=http://localhost:8080 diff --git a/wayang-commons/wayang-core/src/main/resources/wayang-core-defaults.properties b/wayang-commons/wayang-core/src/main/resources/wayang-core-defaults.properties index 6dd9d1d7a..d0372ed67 100644 --- a/wayang-commons/wayang-core/src/main/resources/wayang-core-defaults.properties +++ b/wayang-commons/wayang-core/src/main/resources/wayang-core-defaults.properties @@ -26,7 +26,7 @@ wayang.core.optimizer.enumeration.invertconcatenations = false wayang.core.optimizer.enumeration.branchesfirst = false # Configure statistics collection. -wayang.core.log.enabled = true +wayang.core.log.enabled = false # wayang.core.log.cardinalities = ~/.wayang/cardinalities.json # wayang.core.log.executions = ~/.wayang/executions.json wayang.core.explain.enabled = false diff --git a/wayang-docs/src/main/resources/index.md b/wayang-docs/src/main/resources/index.md index ab0217dbe..2d16eb1ba 100644 --- a/wayang-docs/src/main/resources/index.md +++ b/wayang-docs/src/main/resources/index.md @@ -106,7 +106,7 @@ $ java -Dwayang.configuration=url://to/my/wayang.properties ... Essential configuration settings: * General settings - * `wayang.core.log.enabled (= true)`: whether to log execution statistics to allow learning better cardinality and cost estimators for the optimizer + * `wayang.core.log.enabled (= false)`: whether to log execution statistics to allow learning better cardinality and cost estimators for the optimizer * `wayang.core.log.executions (= ~/.wayang/executions.json)` where to log execution times of operator groups * `wayang.core.log.cardinalities (= ~/.wayang/cardinalities.json)` where to log cardinality measurements * `wayang.core.optimizer.instrumentation (= org.apache.wayang.core.profiling.OutboundInstrumentationStrategy)`: where to measure cardinalities in Wayang plans; other options are `org.apache.wayang.core.profiling.NoInstrumentationStrategy` and `org.apache.wayang.core.profiling.FullInstrumentationStrategy` diff --git a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java index 6dd59a3a6..ba3d0839d 100644 --- a/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java +++ b/wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java @@ -78,7 +78,9 @@ import org.apache.wayang.core.platform.ExecutionState; import org.apache.wayang.core.platform.Executor; import org.apache.wayang.core.platform.ExecutorTemplate; +import org.apache.wayang.core.platform.PartialExecution; import org.apache.wayang.core.platform.Platform; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; import org.apache.wayang.core.util.WayangCollections; import org.apache.wayang.jdbc.channels.SqlQueryChannel; import org.apache.wayang.jdbc.compiler.FunctionCompiler; @@ -262,7 +264,7 @@ private static ExecutionTask selectStartTask(final Collection startTasks, fin * @param optimizationContext provides optimization information * @param jdbcExecutor the executor with the database connection */ - private static void executeSinkStage(final ExecutionStage stage, final OptimizationContext optimizationContext, + private static long executeSinkStage(final ExecutionStage stage, final OptimizationContext optimizationContext, final JdbcExecutor jdbcExecutor) { final Collection startTasks = stage.getStartTasks(); final Collection termTasks = stage.getTerminalTasks(); @@ -338,15 +340,44 @@ private static void executeSinkStage(final ExecutionStage stage, final Optimizat // Execute the composed query: CREATE TABLE x AS SELECT ... or INSERT INTO x // SELECT ... final String fullSql = sinkClause + " " + selectSql + sinkOp.createSqlSuffix(); + final long startTime = System.currentTimeMillis(); stmt.execute(fullSql); + final long executionDuration = System.currentTimeMillis() - startTime; jdbcExecutor.logger.info("Executed SQL sink: {}", fullSql); System.out.println("Executed sql sink: " + fullSql); + return executionDuration; } catch (final SQLException e) { throw new WayangException("Failed to execute SQL sink on table: " + sinkOp.getTableName(), e); } } + /** + * Creates lineage nodes for the JDBC operators that were executed as one SQL + * statement. Operators without an optimization context or load estimator are + * skipped, so JDBC platforms without cost specifications can still execute. + */ + private Collection createExecutionLineageNodes( + final ExecutionStage stage, + final OptimizationContext optimizationContext) { + final Collection executionLineageNodes = new ArrayList<>(); + for (ExecutionTask task : stage.getAllTasks()) { + final OptimizationContext.OperatorContext operatorContext = + optimizationContext.getOperatorContext(task.getOperator()); + if (operatorContext == null) { + this.logger.warn("Cannot profile {} because its optimization context is missing.", task); + continue; + } + if (operatorContext.getLoadProfileEstimator() == null) { + this.logger.warn("Cannot profile {} because its load profile estimator is missing.", task); + continue; + } + executionLineageNodes.add( + new ExecutionLineageNode(operatorContext).addAtomicExecutionFromOperatorContext()); + } + return executionLineageNodes; + } + /** * Retrieves the follow-up {@link ExecutionTask} of the given {@code task} * unless it is not comprising a {@link JdbcExecutionOperator} and/or not in the @@ -420,7 +451,16 @@ public void execute(final ExecutionStage stage, final OptimizationContext optimi final ExecutionTask termTask = (ExecutionTask) termTasks.toArray()[0]; if (termTask.getOperator() instanceof JdbcTableSinkOperator) { - JdbcExecutor.executeSinkStage(stage, optimizationContext, this); + final long executionDuration = JdbcExecutor.executeSinkStage(stage, optimizationContext, this); + if (this.isProfilingEnabled()) { + final PartialExecution partialExecution = this.createPartialExecution( + this.createExecutionLineageNodes(stage, optimizationContext), + executionDuration + ); + if (partialExecution != null) { + executionState.add(partialExecution); + } + } } else { // If it is normal stage: compose SQL and store in channel for downstream // consumption @@ -433,6 +473,10 @@ public void execute(final ExecutionStage stage, final OptimizationContext optimi } } + private boolean isProfilingEnabled() { + return this.getConfiguration().getBooleanProperty("wayang.core.log.enabled", false); + } + @Override public void dispose() { try { diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcTableSinkExecutorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcTableSinkExecutorTest.java index 263730629..483bd4f81 100644 --- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcTableSinkExecutorTest.java +++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcTableSinkExecutorTest.java @@ -23,7 +23,9 @@ import org.apache.wayang.core.optimizer.DefaultOptimizationContext; import org.apache.wayang.core.plan.executionplan.ExecutionStage; import org.apache.wayang.core.plan.executionplan.ExecutionTask; +import org.apache.wayang.core.platform.AtomicExecution; import org.apache.wayang.core.platform.CrossPlatformExecutor; +import org.apache.wayang.core.platform.PartialExecution; import org.apache.wayang.core.profiling.NoInstrumentationStrategy; import org.apache.wayang.jdbc.channels.SqlQueryChannel; import org.apache.wayang.jdbc.operators.JdbcTableSinkOperator; @@ -37,7 +39,11 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.Mockito.mock; @@ -51,6 +57,17 @@ class JdbcTableSinkExecutorTest { @Test void testOverwriteModeCreatesNewTable() throws SQLException { Configuration configuration = new Configuration(); + configuration.setProperty("wayang.core.log.enabled", "true"); + configuration.setProperty("wayang.hsqldb.cpu.mhz", "2700"); + configuration.setProperty("wayang.hsqldb.cores", "1"); + configuration.setProperty( + "wayang.hsqldb.tablesource.load", + "{\"in\":0,\"out\":1,\"cpu\":\"${1}\",\"ram\":\"0\",\"p\":1.0}" + ); + configuration.setProperty( + "wayang.hsqldb.tablesink.load", + "{\"in\":1,\"out\":0,\"cpu\":\"${1}\",\"ram\":\"0\",\"p\":1.0}" + ); HsqldbPlatform hsqldbPlatform = new HsqldbPlatform(); // Create source table with data @@ -89,10 +106,30 @@ void testOverwriteModeCreatesNewTable() throws SQLException { when(sqlStage.getStartTasks()).thenReturn(Collections.singleton(tableSourceTask)); when(sqlStage.getTerminalTasks()).thenReturn(Collections.singleton(sinkTask)); + when(sqlStage.getAllTasks()).thenReturn(new HashSet<>(Arrays.asList(tableSourceTask, sinkTask))); // Execute JdbcExecutor executor = new JdbcExecutor(HsqldbPlatform.getInstance(), job); - executor.execute(sqlStage, new DefaultOptimizationContext(job), job.getCrossPlatformExecutor()); + DefaultOptimizationContext optimizationContext = new DefaultOptimizationContext(job); + optimizationContext.addOneTimeOperator(tableSource); + optimizationContext.addOneTimeOperator(sinkOp); + executor.execute(sqlStage, optimizationContext, job.getCrossPlatformExecutor()); + + assertEquals(1, job.getCrossPlatformExecutor().getPartialExecutions().size()); + PartialExecution partialExecution = + job.getCrossPlatformExecutor().getPartialExecutions().iterator().next(); + Set estimatorKeys = partialExecution.getAtomicExecutionGroups().stream() + .flatMap(group -> group.getAtomicExecutions().stream()) + .map(AtomicExecution::getLoadProfileEstimator) + .map(estimator -> estimator.getConfigurationKey()) + .collect(Collectors.toSet()); + assertEquals( + new HashSet<>(Arrays.asList( + "wayang.hsqldb.tablesource.load", + "wayang.hsqldb.tablesink.load" + )), + estimatorKeys + ); // Verify table was created and contains all 3 rows try (Connection conn = hsqldbPlatform.createDatabaseDescriptor(configuration).createJdbcConnection()) { @@ -158,6 +195,7 @@ void testOverwriteModeReplacesExistingTable() throws SQLException { JdbcExecutor executor = new JdbcExecutor(HsqldbPlatform.getInstance(), job); executor.execute(sqlStage, new DefaultOptimizationContext(job), job.getCrossPlatformExecutor()); + assertEquals(0, job.getCrossPlatformExecutor().getPartialExecutions().size()); // Verify target was replaced. Old data should be gone, new schema and data present try (Connection conn = hsqldbPlatform.createDatabaseDescriptor(configuration).createJdbcConnection()) { @@ -176,6 +214,7 @@ void testOverwriteModeReplacesExistingTable() throws SQLException { @Test void testAppendModeInsertsIntoExistingTable() throws SQLException { Configuration configuration = new Configuration(); + configuration.setProperty("wayang.core.log.enabled", "false"); HsqldbPlatform hsqldbPlatform = new HsqldbPlatform(); //Create source and target table. Target has existing data. @@ -217,6 +256,7 @@ void testAppendModeInsertsIntoExistingTable() throws SQLException { JdbcExecutor executor = new JdbcExecutor(HsqldbPlatform.getInstance(), job); executor.execute(sqlStage, new DefaultOptimizationContext(job), job.getCrossPlatformExecutor()); + assertEquals(0, job.getCrossPlatformExecutor().getPartialExecutions().size()); // Verify existing data remains and new data is appended try (Connection conn = hsqldbPlatform.createDatabaseDescriptor(configuration).createJdbcConnection()) { @@ -252,4 +292,4 @@ void testAppendClauseGeneration() { sinkOp.setMode("append"); assertEquals("INSERT INTO my_table", sinkOp.createSqlClause(null, null)); } -} \ No newline at end of file +} diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcGlobalReduceOperatorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcGlobalReduceOperatorTest.java index 5850fd2cf..159250354 100644 --- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcGlobalReduceOperatorTest.java +++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcGlobalReduceOperatorTest.java @@ -57,9 +57,18 @@ void testWithHsqldb() throws SQLException { when(job.getCrossPlatformExecutor()) .thenReturn(new CrossPlatformExecutor(job, new NoInstrumentationStrategy())); final SqlQueryChannel.Descriptor sqlChannelDescriptor = HsqldbPlatform.getInstance().getSqlQueryChannelDescriptor(); + final HsqldbPlatform hsqldbPlatform = new HsqldbPlatform(); final ExecutionStage sqlStage = mock(ExecutionStage.class); + try (Connection jdbcConnection = hsqldbPlatform.createDatabaseDescriptor(configuration).createJdbcConnection()) { + final Statement statement = jdbcConnection.createStatement(); + statement.execute("DROP TABLE IF EXISTS testA"); + statement.execute("CREATE TABLE testA (a INT, b VARCHAR(6))"); + statement.execute("INSERT INTO testA VALUES (0, 'zero')"); + statement.execute("INSERT INTO testA VALUES (1, 'one')"); + } + final JdbcTableSource tableSourceA = new HsqldbTableSource("testA"); final ExecutionTask tableSourceATask = new ExecutionTask(tableSourceA); @@ -91,8 +100,6 @@ void testWithHsqldb() throws SQLException { final SqlQueryChannel.Instance sqlQueryChannelInstance = (SqlQueryChannel.Instance) job.getCrossPlatformExecutor() .getChannelInstance(sqlToStreamTask.getInputChannel(0)); - final HsqldbPlatform hsqldbPlatform = new HsqldbPlatform(); - try (Connection jdbcConnection = hsqldbPlatform.createDatabaseDescriptor(configuration).createJdbcConnection()) { final Statement statement = jdbcConnection.createStatement(); statement.execute("DROP TABLE IF EXISTS testA"); diff --git a/wayang-platforms/wayang-trino/src/main/resources/wayang-trino-defaults.properties b/wayang-platforms/wayang-trino/src/main/resources/wayang-trino-defaults.properties index fe0429b85..4c1785e7f 100644 --- a/wayang-platforms/wayang-trino/src/main/resources/wayang-trino-defaults.properties +++ b/wayang-platforms/wayang-trino/src/main/resources/wayang-trino-defaults.properties @@ -28,22 +28,16 @@ wayang.trino.cores = 4 wayang.trino.costs.fix = 0.0 wayang.trino.costs.per-ms = 1.0 -# ── Cost model ──────────────────────────────────────────────────────────────── +# Cost model # -# Formula: cpu = α * rows + β +# Formula: cpu = alpha * rows + beta # -# Trino is a distributed MPP engine: very low per-row cost (small α) because -# scans are parallelised across workers, but noticeable fixed overhead (larger β) -# from query planning and cluster coordination. -# -# Compared to a single-node source (Postgres α=55, β=380k): -# α = 10 — parallel scan makes per-row cost ~5× cheaper -# β = 800k — cluster startup + query dispatch overhead -# -# These are initial estimates; tune after real benchmarks by fitting the -# template formula below on measured data and updating the 'load' key. -# ────────────────────────────────────────────────────────────────────────────── - +# The concrete .load entries below are reference parameters learned from the +# local Week 8-2 Trino profiling run. Scope: S01-S13, row counts +# 10k/50k/100k/250k, 1 warm-up plus 5 measured repetitions. They are useful +# starting values for trying Wayang, but users should rerun profiling on their +# own Trino deployment and machine for accurate optimization. +# Keep the matching .load.template entries so the parameters can be relearned. wayang.trino.tablesource.load.template = {\ "type":"mathex", "in":0, "out":1,\ "cpu":"?*out0 + ?",\ @@ -51,10 +45,14 @@ wayang.trino.tablesource.load.template = {\ "p":0.9\ } wayang.trino.tablesource.load = {\ - "in":0, "out":1,\ - "cpu":"${10*out0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":0,\ + "out":1,\ + "cpu":"((0.04991194623872751)*(out0))+(5.650937751899713E7)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.filter.load.template = {\ @@ -64,10 +62,14 @@ wayang.trino.filter.load.template = {\ "p":0.9\ } wayang.trino.filter.load = {\ - "in":1, "out":1,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":1,\ + "cpu":"((0.1984385279077654)*(in0))+(1.930055572992389E8)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.projection.load.template = {\ @@ -77,10 +79,14 @@ wayang.trino.projection.load.template = {\ "p":0.9\ } wayang.trino.projection.load = {\ - "in":1, "out":1,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":1,\ + "cpu":"((0.0033828158004091907)*(in0))+(9667.384755471103)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.join.load.template = {\ @@ -90,10 +96,14 @@ wayang.trino.join.load.template = {\ "p":0.9\ } wayang.trino.join.load = {\ - "in":2, "out":1,\ - "cpu":"${10*(in0 + in1) + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":2,\ + "out":1,\ + "cpu":"((0.08209057842372317)*((in0)+(in1)))+(7.914300131274152E8)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.globalreduce.load.template = {\ @@ -103,10 +113,14 @@ wayang.trino.globalreduce.load.template = {\ "p":0.9\ } wayang.trino.globalreduce.load = {\ - "in":1, "out":1,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":1,\ + "cpu":"((3.2826093296659775)*(in0))+(418.86246686849927)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.reduceby.load.template = {\ @@ -116,10 +130,14 @@ wayang.trino.reduceby.load.template = {\ "p":0.9\ } wayang.trino.reduceby.load = {\ - "in":1, "out":1,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":1,\ + "cpu":"((0.06770126935475743)*(in0))+(815723.9457107394)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.sort.load.template = {\ @@ -129,10 +147,14 @@ wayang.trino.sort.load.template = {\ "p":0.9\ } wayang.trino.sort.load = {\ - "in":1, "out":1,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":1,\ + "cpu":"((6345.214432640646)*(in0))+(1.5730398856855012E7)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.tablesink.load.template = {\ @@ -142,10 +164,14 @@ wayang.trino.tablesink.load.template = {\ "p":0.9\ } wayang.trino.tablesink.load = {\ - "in":1, "out":0,\ - "cpu":"${10*in0 + 800000}",\ - "ram":"0",\ - "p":0.9\ + "type":"mathex",\ + "in":1,\ + "out":0,\ + "cpu":"((5853.43034225027)*(in0))+(3.5512637525734577E9)",\ + "ram":"0",\ + "disk":"0",\ + "net":"0",\ + "p":0.9\ } wayang.trino.sqltostream.load.query.template = {\ diff --git a/wayang-platforms/wayang-trino/src/test/java/org/apache/wayang/trino/TrinoCostPilotIT.java b/wayang-platforms/wayang-trino/src/test/java/org/apache/wayang/trino/TrinoCostPilotIT.java new file mode 100644 index 000000000..0a12b3b63 --- /dev/null +++ b/wayang-platforms/wayang-trino/src/test/java/org/apache/wayang/trino/TrinoCostPilotIT.java @@ -0,0 +1,820 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. + * See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.wayang.trino; + +import org.apache.wayang.basic.data.Record; +import org.apache.wayang.basic.data.Tuple2; +import org.apache.wayang.basic.function.ProjectionDescriptor; +import org.apache.wayang.basic.operators.FilterOperator; +import org.apache.wayang.basic.operators.GlobalReduceOperator; +import org.apache.wayang.basic.operators.JoinOperator; +import org.apache.wayang.basic.operators.MapOperator; +import org.apache.wayang.basic.operators.ReduceByOperator; +import org.apache.wayang.basic.operators.SortOperator; +import org.apache.wayang.basic.operators.TableSink; +import org.apache.wayang.basic.types.RecordType; +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.api.WayangContext; +import org.apache.wayang.core.function.FunctionDescriptor; +import org.apache.wayang.core.function.PredicateDescriptor; +import org.apache.wayang.core.function.ReduceDescriptor; +import org.apache.wayang.core.function.TransformationDescriptor; +import org.apache.wayang.core.mapping.Mapping; +import org.apache.wayang.core.mapping.OperatorPattern; +import org.apache.wayang.core.mapping.PlanTransformation; +import org.apache.wayang.core.mapping.ReplacementSubplanFactory; +import org.apache.wayang.core.mapping.SubplanPattern; +import org.apache.wayang.core.plan.wayangplan.WayangPlan; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.core.types.DataUnitType; +import org.apache.wayang.trino.operators.TrinoProjectionOperator; +import org.apache.wayang.trino.operators.TrinoTableSource; +import org.apache.wayang.trino.platform.TrinoPlatform; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; + +import java.io.BufferedWriter; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Small Trino cost-profiling pilot. + */ +class TrinoCostPilotIT { + + private static final String HOST = System.getenv().getOrDefault("TRINO_HOST", "localhost"); + private static final int PORT = Integer.parseInt(System.getenv().getOrDefault("TRINO_PORT", "8080")); + private static final String USER = System.getenv().getOrDefault("TRINO_USER", "admin"); + private static final String JDBC_URL = String.format("jdbc:trino://%s:%d", HOST, PORT); + + private static final String SCHEMA = "iceberg.wayang_profile"; + private static final String CUSTOMERS_1K = SCHEMA + ".customers_1k"; + private static final int[] ROW_COUNTS = parseIntList(System.getProperty( + "trino.profile.rowCounts", + "10000,50000,100000,250000" + )); + private static final String[] COLUMNS = {"order_id", "customer_id", "region", "amount", "bucket"}; + private static final String[] JOIN_COLUMNS = { + "order_id", "customer_id", "region", "amount", "bucket", "cust_id", "tier" + }; + private static final String[] JOIN_ORDER_TIER_AMOUNT_COLUMNS = {"order_id", "tier", "amount"}; + private static final String[] JOIN_TIER_AMOUNT_COLUMNS = {"tier", "amount"}; + private static final String JOIN_FLATTEN_NAME = "Trino profile join flatten"; + private static final String JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME = "Trino profile join order tier amount flatten"; + private static final String JOIN_TIER_AMOUNT_FLATTEN_NAME = "Trino profile join tier amount flatten"; + private static final Path OUTPUT_DIR = Paths.get(System.getProperty( + "trino.profile.outputDir", + "target/cost-profiling/trino" + )); + private static final Path EXECUTIONS_PATH = OUTPUT_DIR.resolve("executions.json"); + private static final Path CARDINALITIES_PATH = OUTPUT_DIR.resolve("cardinalities.json"); + private static final Path MANIFEST_PATH = OUTPUT_DIR.resolve("manifest.csv"); + private static final List PLAN_IDS = Arrays.asList( + System.getProperty( + "trino.profile.plans", + "S01,S02,S03,S04,S05,S06,S07,S08,S09,S10,S11,S12,S13,S14,S15,S16" + ).split(",") + ); + private static final int REPETITIONS = Integer.parseInt( + System.getProperty("trino.profile.repetitions", "6") + ); + private static final boolean RESET_OUTPUT = Boolean.parseBoolean( + System.getProperty("trino.profile.reset", "true") + ); + + @Test + void runPilot() throws Exception { + Assumptions.assumeTrue(isTrinoAvailable(), "Trino not reachable"); + Files.createDirectories(OUTPUT_DIR); + initializeOutputFiles(); + + prepareTables(); + + for (int rowCount : ROW_COUNTS) { + for (String planId : PLAN_IDS) { + String normalizedPlanId = planId.trim(); + runPlan( + normalizedPlanId, + getOperatorChain(normalizedPlanId), + rowCount, + getExpectedRows(normalizedPlanId, rowCount) + ); + } + } + } + + private void runPlan(String planId, String operatorChain, int rowCount, long expectedRows) throws Exception { + for (int repetition = 0; repetition < REPETITIONS; repetition++) { + boolean isWarmup = repetition == 0; + String runId = String.format("%s_%s_r%02d", planId, formatRows(rowCount), repetition); + String sourceTable = SCHEMA + ".orders_" + formatRows(rowCount); + String sinkTable = SCHEMA + ".sink_" + runId.toLowerCase(); + + dropTable(sinkTable); + WayangPlan plan = createPlan(planId, sourceTable, sinkTable); + wayangContext().execute(runId, plan); + + long actualRows = queryLong("SELECT count(*) FROM " + sinkTable); + assertEquals(expectedRows, actualRows, runId + " row count"); + appendManifest(runId, planId, operatorChain, rowCount, expectedRows, repetition, isWarmup, sinkTable, "ok", ""); + dropTable(sinkTable); + } + } + + private WayangPlan createPlan(String planId, String sourceTable, String sinkTable) { + TrinoTableSource source = new TrinoTableSource(sourceTable, COLUMNS); + + if ("S01".equals(planId)) { + TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS); + source.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S02".equals(planId)) { + TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS); + FilterOperator filter = createAmerFilter(); + source.connectTo(0, filter, 0); + filter.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S03".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "order_id", "amount"); + MapOperator projection = createOrderAmountProjection(); + source.connectTo(0, projection, 0); + projection.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S04".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "order_id", "amount"); + FilterOperator filter = createAmerFilter(); + MapOperator projection = createOrderAmountProjection(); + source.connectTo(0, filter, 0); + filter.connectTo(0, projection, 0); + projection.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S05".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "total_amount"); + GlobalReduceOperator reduce = new GlobalReduceOperator<>( + new ReduceDescriptor<>((left, right) -> left, Record.class) + .withSqlImplementation("SUM(amount) AS total_amount"), + DataSetType.createDefault(Record.class)); + source.connectTo(0, reduce, 0); + reduce.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S06".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "bucket", "total_amount"); + ReduceByOperator reduceBy = new ReduceByOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(4)), + Record.class, + Record.class + ).withSqlImplementation("bucket", "bucket"), + new ReduceDescriptor<>((left, right) -> left, Record.class) + .withSqlImplementation("SUM(amount) AS total_amount"), + DataSetType.createDefault(Record.class)); + source.connectTo(0, reduceBy, 0); + reduceBy.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S07".equals(planId)) { + TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS); + SortOperator sort = createAmountSortOperator(3); + source.connectTo(0, sort, 0); + sort.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S08".equals(planId)) { + TrinoTableSource customers = new TrinoTableSource(CUSTOMERS_1K, "cust_id", "tier"); + JoinOperator join = new JoinOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(1)), + Record.class, + Record.class + ).withSqlImplementation(sourceTable, "customer_id"), + new TransformationDescriptor<>( + record -> new Record(record.getField(0)), + Record.class, + Record.class + ).withSqlImplementation(CUSTOMERS_1K, "cust_id")); + MapOperator, Record> flatten = createJoinFlattenOperator(); + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, JOIN_COLUMNS); + source.connectTo(0, join, 0); + customers.connectTo(0, join, 1); + join.connectTo(0, flatten, 0); + flatten.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S09".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "total_amount"); + FilterOperator filter = createAmerFilter(); + GlobalReduceOperator reduce = createGlobalAmountReduceOperator(); + source.connectTo(0, filter, 0); + filter.connectTo(0, reduce, 0); + reduce.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S10".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "bucket", "total_amount"); + FilterOperator filter = createAmerFilter(); + ReduceByOperator reduceBy = createBucketReduceByOperator(); + source.connectTo(0, filter, 0); + filter.connectTo(0, reduceBy, 0); + reduceBy.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S11".equals(planId)) { + TableSink sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS); + FilterOperator filter = createAmerFilter(); + SortOperator sort = createAmountSortOperator(3); + source.connectTo(0, filter, 0); + filter.connectTo(0, sort, 0); + sort.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S12".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "order_id", "amount"); + MapOperator projection = createOrderAmountProjection(); + SortOperator sort = createAmountSortOperator(1); + source.connectTo(0, projection, 0); + projection.connectTo(0, sort, 0); + sort.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S13".equals(planId)) { + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "order_id", "amount"); + FilterOperator filter = createAmerFilter(); + MapOperator projection = createOrderAmountProjection(); + SortOperator sort = createAmountSortOperator(1); + source.connectTo(0, filter, 0); + filter.connectTo(0, projection, 0); + projection.connectTo(0, sort, 0); + sort.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S14".equals(planId)) { + TrinoTableSource customers = new TrinoTableSource(CUSTOMERS_1K, "cust_id", "tier"); + FilterOperator filter = createAmerFilter(); + JoinOperator join = createCustomerJoinOperator(sourceTable); + MapOperator, Record> flatten = createJoinFlattenOperator(); + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, JOIN_COLUMNS); + source.connectTo(0, filter, 0); + filter.connectTo(0, join, 0); + customers.connectTo(0, join, 1); + join.connectTo(0, flatten, 0); + flatten.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S15".equals(planId)) { + TrinoTableSource customers = new TrinoTableSource(CUSTOMERS_1K, "cust_id", "tier"); + JoinOperator join = createCustomerJoinOperator(sourceTable); + MapOperator, Record> flatten = createJoinOrderTierAmountFlattenOperator(); + SortOperator sort = createAmountSortOperator(2); + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, JOIN_ORDER_TIER_AMOUNT_COLUMNS); + source.connectTo(0, join, 0); + customers.connectTo(0, join, 1); + join.connectTo(0, flatten, 0); + flatten.connectTo(0, sort, 0); + sort.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + if ("S16".equals(planId)) { + TrinoTableSource customers = new TrinoTableSource(CUSTOMERS_1K, "cust_id", "tier"); + JoinOperator join = createCustomerJoinOperator(sourceTable); + MapOperator, Record> flatten = createJoinTierAmountFlattenOperator(); + ReduceByOperator reduceBy = createTierReduceByOperator(); + TableSink sink = new TableSink<>( + new Properties(), "overwrite", sinkTable, "tier", "total_amount"); + source.connectTo(0, join, 0); + customers.connectTo(0, join, 1); + join.connectTo(0, flatten, 0); + flatten.connectTo(0, reduceBy, 0); + reduceBy.connectTo(0, sink, 0); + return new WayangPlan(sink); + } + + throw new IllegalArgumentException("Unsupported pilot plan: " + planId); + } + + private static GlobalReduceOperator createGlobalAmountReduceOperator() { + return new GlobalReduceOperator<>( + new ReduceDescriptor<>((left, right) -> left, Record.class) + .withSqlImplementation("SUM(amount) AS total_amount"), + DataSetType.createDefault(Record.class)); + } + + private static ReduceByOperator createBucketReduceByOperator() { + return new ReduceByOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(4)), + Record.class, + Record.class + ).withSqlImplementation("bucket", "bucket"), + new ReduceDescriptor<>((left, right) -> left, Record.class) + .withSqlImplementation("SUM(amount) AS total_amount"), + DataSetType.createDefault(Record.class)); + } + + private static ReduceByOperator createTierReduceByOperator() { + return new ReduceByOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(0)), + Record.class, + Record.class + ).withSqlImplementation("tier", "tier"), + new ReduceDescriptor<>((left, right) -> left, Record.class) + .withSqlImplementation("SUM(amount) AS total_amount"), + DataSetType.createDefault(Record.class)); + } + + private static SortOperator createAmountSortOperator(int amountFieldIndex) { + return new SortOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(amountFieldIndex)), + Record.class, + Record.class + ).withSqlImplementation("amount", "ASC"), + DataSetType.createDefault(Record.class)); + } + + private static JoinOperator createCustomerJoinOperator(String sourceTable) { + return new JoinOperator<>( + new TransformationDescriptor<>( + record -> new Record(record.getField(1)), + Record.class, + Record.class + ).withSqlImplementation(sourceTable, "customer_id"), + new TransformationDescriptor<>( + record -> new Record(record.getField(0)), + Record.class, + Record.class + ).withSqlImplementation(CUSTOMERS_1K, "cust_id")); + } + + private static FilterOperator createAmerFilter() { + return new FilterOperator<>( + new PredicateDescriptor<>( + (Record record) -> "AMER".equals(record.getField(2)), + Record.class + ).withSqlImplementation("region = 'AMER'") + ); + } + + private static MapOperator createOrderAmountProjection() { + return new MapOperator<>( + ProjectionDescriptor.createForRecords( + new RecordType(COLUMNS), + "order_id", "amount"), + DataSetType.createDefault(Record.class), + DataSetType.createDefault(Record.class)); + } + + private static MapOperator createOrderTierAmountProjection() { + return new MapOperator<>( + ProjectionDescriptor.createForRecords( + new RecordType(JOIN_COLUMNS), + "order_id", "tier", "amount"), + DataSetType.createDefault(Record.class), + DataSetType.createDefault(Record.class)); + } + + private static MapOperator createTierAmountProjection() { + return new MapOperator<>( + ProjectionDescriptor.createForRecords( + new RecordType(JOIN_COLUMNS), + "tier", "amount"), + DataSetType.createDefault(Record.class), + DataSetType.createDefault(Record.class)); + } + + private static MapOperator, Record> createJoinFlattenOperator() { + return createJoinFlattenOperator(new JoinFlattenFunction(), JOIN_FLATTEN_NAME); + } + + private static MapOperator, Record> createJoinOrderTierAmountFlattenOperator() { + return createJoinFlattenOperator(new JoinOrderTierAmountFlattenFunction(), JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME); + } + + private static MapOperator, Record> createJoinTierAmountFlattenOperator() { + return createJoinFlattenOperator(new JoinTierAmountFlattenFunction(), JOIN_TIER_AMOUNT_FLATTEN_NAME); + } + + private static MapOperator, Record> createJoinFlattenOperator( + FunctionDescriptor.SerializableFunction, Record> function, + String name) { + MapOperator, Record> operator = new MapOperator<>( + new TransformationDescriptor<>( + function, + DataUnitType.createBasicUnchecked(Tuple2.class), + DataUnitType.createBasic(Record.class)), + DataSetType.createDefaultUnchecked(Tuple2.class), + DataSetType.createDefault(Record.class)); + operator.setName(name); + return operator; + } + + private WayangContext wayangContext() { + Configuration configuration = new Configuration(); + configuration.setProperty("wayang.trino.jdbc.url", JDBC_URL); + configuration.setProperty("wayang.trino.jdbc.user", USER); + configuration.setProperty("wayang.trino.jdbc.password", ""); + configuration.setProperty("wayang.core.log.enabled", "true"); + configuration.setProperty("wayang.core.explain.enabled", "false"); + configuration.setProperty("wayang.core.log.executions", EXECUTIONS_PATH.toString().replace('\\', '/')); + configuration.setProperty("wayang.core.log.cardinalities", CARDINALITIES_PATH.toString().replace('\\', '/')); + configuration.getMappingProvider().addAllToWhitelist( + Collections.singleton(new JoinFlattenMapping())); + return new WayangContext(configuration).withPlugin(Trino.plugin()); + } + + private void prepareTables() throws Exception { + try (Connection connection = jdbc(); Statement statement = connection.createStatement()) { + statement.execute("CREATE SCHEMA IF NOT EXISTS " + SCHEMA); + for (int rowCount : ROW_COUNTS) { + String table = SCHEMA + ".orders_" + formatRows(rowCount); + statement.execute("DROP TABLE IF EXISTS " + table); + statement.execute("CREATE TABLE " + table + " WITH (format = 'PARQUET') AS " + + "SELECT " + + "CAST(n AS BIGINT) AS order_id, " + + "CAST(n % 1000 AS BIGINT) AS customer_id, " + + "CASE WHEN n % 2 = 0 THEN 'AMER' ELSE 'EMEA' END AS region, " + + "CAST(n % 10000 AS DOUBLE) AS amount, " + + "CAST(n % 100 AS BIGINT) AS bucket " + + "FROM " + createRowsSql(rowCount)); + assertEquals(rowCount, queryLong("SELECT count(*) FROM " + table), table + " row count"); + assertEquals(rowCount / 2, queryLong("SELECT count(*) FROM " + table + " WHERE region = 'AMER'"), + table + " AMER row count"); + } + statement.execute("DROP TABLE IF EXISTS " + CUSTOMERS_1K); + statement.execute("CREATE TABLE " + CUSTOMERS_1K + " WITH (format = 'PARQUET') AS " + + "SELECT " + + "CAST(n - 1 AS BIGINT) AS cust_id, " + + "CASE WHEN n % 2 = 0 THEN 'GOLD' ELSE 'SILVER' END AS tier " + + "FROM UNNEST(sequence(1, 1000)) AS t(n)"); + assertEquals(1000, queryLong("SELECT count(*) FROM " + CUSTOMERS_1K), CUSTOMERS_1K + " row count"); + } + } + + private static String createRowsSql(int rowCount) { + if (rowCount <= 10000) { + return "UNNEST(sequence(1, " + rowCount + ")) AS t(n)"; + } + + int chunks = (rowCount + 9999) / 10000; + return "(" + + "SELECT chunk * 10000 + offset AS n " + + "FROM UNNEST(sequence(0, " + (chunks - 1) + ")) AS c(chunk) " + + "CROSS JOIN UNNEST(sequence(1, 10000)) AS o(offset) " + + "WHERE chunk * 10000 + offset <= " + rowCount + + ") AS t"; + } + + private static String formatRows(int rowCount) { + if (rowCount % 1000 == 0) { + return (rowCount / 1000) + "k"; + } + return String.valueOf(rowCount); + } + + private void initializeOutputFiles() throws Exception { + if (RESET_OUTPUT) { + Files.deleteIfExists(EXECUTIONS_PATH); + Files.deleteIfExists(CARDINALITIES_PATH); + writeManifestHeader(); + } else if (!Files.exists(MANIFEST_PATH)) { + writeManifestHeader(); + } + } + + private void writeManifestHeader() throws Exception { + try (BufferedWriter writer = Files.newBufferedWriter(MANIFEST_PATH, StandardCharsets.UTF_8)) { + writer.write("run_id,plan_id,operator_chain,input_rows_left,input_rows_right,expected_output_rows," + + "selectivity,repetition,is_warmup,sink_table,status,notes"); + writer.newLine(); + } + } + + private void appendManifest( + String runId, + String planId, + String operatorChain, + int inputRows, + long expectedOutputRows, + int repetition, + boolean isWarmup, + String sinkTable, + String status, + String notes) throws Exception { + try (BufferedWriter writer = Files.newBufferedWriter( + MANIFEST_PATH, + StandardCharsets.UTF_8, + java.nio.file.StandardOpenOption.APPEND)) { + writer.write(String.join(",", + runId, + planId, + operatorChain, + String.valueOf(inputRows), + hasJoin(planId) ? "1000" : "", + String.valueOf(expectedOutputRows), + hasFilter(planId) ? "0.5" : "1.0", + String.valueOf(repetition), + String.valueOf(isWarmup), + sinkTable, + status, + notes)); + writer.newLine(); + } + } + + private static String getOperatorChain(String planId) { + switch (planId) { + case "S01": + return "TableSource->TableSink"; + case "S02": + return "TableSource->Filter(50%)->TableSink"; + case "S03": + return "TableSource->Projection->TableSink"; + case "S04": + return "TableSource->Filter(50%)->Projection->TableSink"; + case "S05": + return "TableSource->GlobalReduce->TableSink"; + case "S06": + return "TableSource->ReduceBy(bucket)->TableSink"; + case "S07": + return "TableSource->Sort(amount)->TableSink"; + case "S08": + return "Orders->Join(Customers 1k)->Projection->TableSink"; + case "S09": + return "TableSource->Filter(50%)->GlobalReduce->TableSink"; + case "S10": + return "TableSource->Filter(50%)->ReduceBy(bucket)->TableSink"; + case "S11": + return "TableSource->Filter(50%)->Sort(amount)->TableSink"; + case "S12": + return "TableSource->Projection(order_id,amount)->Sort(amount)->TableSink"; + case "S13": + return "TableSource->Filter(50%)->Projection(order_id,amount)->Sort(amount)->TableSink"; + case "S14": + return "Orders->Filter(50%)->Join(Customers 1k)->Projection->TableSink"; + case "S15": + return "Orders->Join(Customers 1k)->Projection(order_id,tier,amount)->Sort(amount)->TableSink"; + case "S16": + return "Orders->Join(Customers 1k)->Projection(tier,amount)->ReduceBy(tier)->TableSink"; + default: + throw new IllegalArgumentException("Unsupported pilot plan: " + planId); + } + } + + private static long getExpectedRows(String planId, int rowCount) { + if ("S05".equals(planId) || "S09".equals(planId)) { + return 1; + } + if ("S06".equals(planId)) { + return 100; + } + if ("S10".equals(planId)) { + return 50; + } + if ("S16".equals(planId)) { + return 2; + } + return hasFilter(planId) ? rowCount / 2 : rowCount; + } + + private static boolean hasFilter(String planId) { + return "S02".equals(planId) + || "S04".equals(planId) + || "S09".equals(planId) + || "S10".equals(planId) + || "S11".equals(planId) + || "S13".equals(planId) + || "S14".equals(planId); + } + + private static boolean hasJoin(String planId) { + return "S08".equals(planId) + || "S14".equals(planId) + || "S15".equals(planId) + || "S16".equals(planId); + } + + private static int[] parseIntList(String value) { + return Arrays.stream(value.split(",")) + .map(String::trim) + .filter(token -> !token.isEmpty()) + .mapToInt(Integer::parseInt) + .toArray(); + } + + private long queryLong(String sql) throws Exception { + try (Connection connection = jdbc(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql)) { + resultSet.next(); + return resultSet.getLong(1); + } + } + + private void dropTable(String table) throws Exception { + try (Connection connection = jdbc(); Statement statement = connection.createStatement()) { + statement.execute("DROP TABLE IF EXISTS " + table); + } + } + + private static boolean isTrinoAvailable() { + try (Connection connection = jdbc(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("SELECT 1")) { + return resultSet.next(); + } catch (Exception e) { + return false; + } + } + + private static Connection jdbc() throws Exception { + return DriverManager.getConnection(JDBC_URL, USER, ""); + } + + private static Record flattenJoinResult(Object joinResult) { + if (joinResult instanceof Record) { + return (Record) joinResult; + } + Tuple2 pair = (Tuple2) joinResult; + Record left = (Record) pair.field0; + Record right = (Record) pair.field1; + return new Record( + left.getField(0), + left.getField(1), + left.getField(2), + left.getField(3), + left.getField(4), + right.getField(0), + right.getField(1)); + } + + private static Record flattenJoinOrderTierAmountResult(Object joinResult) { + if (joinResult instanceof Record) { + Record record = (Record) joinResult; + return new Record(record.getField(0), record.getField(6), record.getField(3)); + } + Tuple2 pair = (Tuple2) joinResult; + Record left = (Record) pair.field0; + Record right = (Record) pair.field1; + return new Record(left.getField(0), right.getField(1), left.getField(3)); + } + + private static Record flattenJoinTierAmountResult(Object joinResult) { + if (joinResult instanceof Record) { + Record record = (Record) joinResult; + return new Record(record.getField(6), record.getField(3)); + } + Tuple2 pair = (Tuple2) joinResult; + Record left = (Record) pair.field0; + Record right = (Record) pair.field1; + return new Record(right.getField(1), left.getField(3)); + } + + private static final class JoinFlattenFunction implements + FunctionDescriptor.SerializableFunction, Record> { + + @Override + public Record apply(Tuple2 tuple) { + return flattenJoinResult(tuple); + } + } + + private static final class JoinOrderTierAmountFlattenFunction implements + FunctionDescriptor.SerializableFunction, Record> { + + @Override + public Record apply(Tuple2 tuple) { + return flattenJoinOrderTierAmountResult(tuple); + } + } + + private static final class JoinTierAmountFlattenFunction implements + FunctionDescriptor.SerializableFunction, Record> { + + @Override + public Record apply(Tuple2 tuple) { + return flattenJoinTierAmountResult(tuple); + } + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private static final class JoinFlattenMapping implements Mapping { + + @Override + public java.util.Collection getTransformations() { + OperatorPattern pattern = new OperatorPattern( + "joinFlatten", + new MapOperator(null, DataSetType.none(), DataSetType.createDefault(Record.class)), + false) + .withAdditionalTest(operator -> isJoinFlattenName(((MapOperator) operator).getName())); + + ReplacementSubplanFactory factory = new ReplacementSubplanFactory.OfSingleOperators( + (matchedOperator, epoch) -> createTrinoProjection(matchedOperator.getName()).at(epoch)); + + return Collections.singleton(new PlanTransformation( + SubplanPattern.createSingleton(pattern), + factory, + TrinoPlatform.getInstance())); + } + + private static TrinoProjectionOperator createTrinoProjection(String operatorName) { + ProjectionDescriptor, Record> descriptor = new ProjectionDescriptor<>( + getJoinFlattenFunction(operatorName), + Arrays.asList(getJoinFlattenColumns(operatorName)), + DataUnitType.createBasicUnchecked(Tuple2.class), + DataUnitType.createBasic(Record.class)); + MapOperator, Record> projection = new MapOperator<>( + descriptor, + DataSetType.createDefaultUnchecked(Tuple2.class), + DataSetType.createDefault(Record.class)); + projection.setName(operatorName); + return new TrinoProjectionOperator((MapOperator) (MapOperator) projection); + } + + private static boolean isJoinFlattenName(String operatorName) { + return JOIN_FLATTEN_NAME.equals(operatorName) + || JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName) + || JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName); + } + + private static String[] getJoinFlattenColumns(String operatorName) { + if (JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) { + return JOIN_ORDER_TIER_AMOUNT_COLUMNS; + } + if (JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) { + return JOIN_TIER_AMOUNT_COLUMNS; + } + return JOIN_COLUMNS; + } + + private static FunctionDescriptor.SerializableFunction, Record> getJoinFlattenFunction( + String operatorName) { + if (JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) { + return new JoinOrderTierAmountFlattenFunction(); + } + if (JOIN_TIER_AMOUNT_FLATTEN_NAME.equals(operatorName)) { + return new JoinTierAmountFlattenFunction(); + } + return new JoinFlattenFunction(); + } + } +} diff --git a/wayang-profiler/src/main/java/org/apache/wayang/profiler/log/GeneticOptimizerApp.java b/wayang-profiler/src/main/java/org/apache/wayang/profiler/log/GeneticOptimizerApp.java index 154819b5f..22c016b03 100644 --- a/wayang-profiler/src/main/java/org/apache/wayang/profiler/log/GeneticOptimizerApp.java +++ b/wayang-profiler/src/main/java/org/apache/wayang/profiler/log/GeneticOptimizerApp.java @@ -109,6 +109,7 @@ public GeneticOptimizerApp(Configuration configuration) { Spark.platform(); Sqlite3.platform(); Postgres.platform(); + initializeOptionalPlatform("org.apache.wayang.trino.Trino"); // Load the ExecutionLog. double samplingFactor = this.configuration.getDoubleProperty("wayang.profiler.ga.sampling", 1d); @@ -201,6 +202,20 @@ public GeneticOptimizerApp(Configuration configuration) { ); } + /** + * Initializes a platform integration when it is available on the runtime + * classpath without making it a mandatory profiler dependency. + */ + private static void initializeOptionalPlatform(String platformFacadeClassName) { + try { + Class.forName(platformFacadeClassName).getMethod("platform").invoke(null); + } catch (ClassNotFoundException e) { + logger.debug("Optional platform {} is not on the classpath.", platformFacadeClassName); + } catch (ReflectiveOperationException e) { + throw new WayangException("Could not initialize optional platform " + platformFacadeClassName, e); + } + } + /** * Check if all {@link CardinalityEstimate}s for the {@link PartialExecution} are sufficiently confident. * diff --git a/wayang-tests-integration/src/test/java/org/apache/wayang/tests/FlinkIntegrationIT.java b/wayang-tests-integration/src/test/java/org/apache/wayang/tests/FlinkIntegrationIT.java index cfab48b25..7064f0c0d 100644 --- a/wayang-tests-integration/src/test/java/org/apache/wayang/tests/FlinkIntegrationIT.java +++ b/wayang-tests-integration/src/test/java/org/apache/wayang/tests/FlinkIntegrationIT.java @@ -314,12 +314,13 @@ void testReadAndTransformAndWriteWithIllegalConfiguration3() { // Build a Wayang plan. final WayangPlan wayangPlan = WayangPlans.readTransformWrite(WayangPlans.FILE_SOME_LINES_TXT); - // Instantiate Wayang and activate the Spark backend. - WayangContext wayangContext = makeContext(FLINK); + // Instantiate Wayang and activate the Java/Flink backends. + WayangContext wayangContext = makeContext(BOTH); // Have Wayang execute the plan. final Job job = wayangContext.createJob(null, wayangPlan); - // ILLEGAL: We blacklist the Spark platform, although we need it. + // ILLEGAL: We blacklist all platforms that can execute the plan. + job.getConfiguration().getPlatformProvider().addToBlacklist(Java.platform()); job.getConfiguration().getPlatformProvider().addToBlacklist(Flink.platform()); job.getConfiguration().getPlatformProvider().addToWhitelist(MyMadeUpPlatform.getInstance()); assertThrows(WayangException.class, job::execute);