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/guides/cost-profiling.md b/guides/cost-profiling.md
new file mode 100644
index 000000000..25f8f9bd3
--- /dev/null
+++ b/guides/cost-profiling.md
@@ -0,0 +1,430 @@
+<!--
+  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.
+-->
+
+# 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 = alpha * number_of_rows + beta
+```
+
+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\<user>\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.<platform>.*.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 = <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 <table>
+```
+
+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/presto-setup/README.md b/presto-setup/README.md
new file mode 100644
index 000000000..023a4e653
--- /dev/null
+++ b/presto-setup/README.md
@@ -0,0 +1,249 @@
+# Presto Local Setup
+
+Local PrestoDB environment using the built-in **memory** connector, completely
+containerised.
+
+The current validation has two parts:
+
+1. Build the Wayang Presto platform and run the shared JDBC SQL-generation tests.
+2. Run the Wayang Presto operator tests against the live local PrestoDB.
+
+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 Presto cost-profiling branch is named `feature/presto-cost-profiling`:
+
+```bash
+git checkout feature/presto-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 |
+|-----------|-------|------|------|
+| **PrestoDB** | `prestodb/presto:0.289` | 8081 | SQL engine and in-memory test storage |
+
+The container listens on port `8080`; Docker exposes it as `8081` to avoid
+clashing with the Trino setup. The `memory` connector needs no metastore,
+database, or object storage. All tables disappear when the container stops.
+
+## Directory Layout
+
+```text
+presto-setup/
+|-- docker-compose.yml
+|-- README.md
+`-- etc/
+    `-- catalog/
+        `-- memory.properties
+
+wayang-platforms/wayang-presto/src/test/java/.../
+`-- PrestoOperatorsIT.java
+```
+
+## 1. Test the Wayang Presto Platform
+
+Build the Presto platform and its required modules:
+
+```bash
+./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am \
+  -DskipTests -Drat.skip=true test
+```
+
+On PowerShell:
+
+```powershell
+.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -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 Presto ... SUCCESS
+Tests run: 4, Failures: 0, Errors: 0, Skipped: 0
+```
+
+## 2. Test Against the Local Presto Stack
+
+### 1. Start Presto
+
+```bash
+docker compose -f presto-setup/docker-compose.yml up -d --wait
+```
+
+Presto can take 20-60 seconds to accept queries. Confirm that it is healthy:
+
+```bash
+docker compose -f presto-setup/docker-compose.yml ps
+```
+
+The Presto web UI is available at <http://localhost:8081>.
+
+### 2. Run the Wayang Presto operator tests
+
+`PrestoOperatorsIT` exercises the Wayang Presto implementation against the live
+container. It checks `TableSource`, `Filter`, `Projection`, `Join`,
+`GlobalReduce`, `ReduceBy`, `Sort`, and `TableSink`, and confirms that the
+expected SQL reached Presto through `system.runtime.queries`.
+
+The standalone join test now runs as a full Wayang plan:
+`PrestoTableSource + PrestoTableSource -> JoinOperator -> MapOperator -> sink`.
+The normalization map accepts both logical `Tuple2<Record, Record>` output and
+pushed-down JDBC flat `Record` output. The suite also includes five
+`JavaPlanBuilder.readTable` combination plans. Together, they cover every
+supported Presto operator through the public API.
+
+The suite is self-contained. It creates `memory.wayang_it`, generates 120,000
+rows so the optimizer selects SQL pushdown, runs the tests, and drops its tables
+afterward.
+
+```bash
+./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am \
+  -Dtest=PrestoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false \
+  -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test
+```
+
+On PowerShell:
+
+```powershell
+.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am -Dtest=PrestoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test
+```
+
+Successful validation must show:
+
+```text
+Tests run: 13, Failures: 0, Errors: 0, Skipped: 0
+BUILD SUCCESS
+```
+
+If Presto is unreachable, the tests are skipped instead of failed. A result
+with skipped tests does not confirm that the operators work. Errors while
+creating the test schema or tables are treated as real failures.
+
+### Verified Result
+
+On June 18, 2026, the suite completed successfully against the local PrestoDB
+0.289 container, including the full-plan join validation:
+
+```text
+[PrestoOperatorsIT] Connected to Presto at jdbc:presto://localhost:8081/memory
+Executed sql sink: CREATE TABLE memory.wayang_it.amer_orders AS SELECT ...
+Tests run: 13, Failures: 0, Errors: 0, Skipped: 0
+BUILD SUCCESS
+```
+
+This verified the complete `Wayang -> Presto JDBC -> live PrestoDB` path,
+including reads, SQL pushdown, join normalization, aggregation, sorting, and
+`CREATE TABLE AS SELECT`.
+
+### 3. Tear down
+
+```bash
+docker compose -f presto-setup/docker-compose.yml down
+```
+
+## Test Coverage
+
+| Test | What it checks |
+|------|----------------|
+| `tableSource` | Full table scan through `PrestoTableSource` |
+| `filter` | Wayang `FilterOperator` and SQL `WHERE` pushdown |
+| `projection` | Column projection pushed into the Presto 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` | Public API filter and projection combination |
+| `javaPlanBuilderReadTableFilterGlobalReduce` | Public API filter and global aggregation combination |
+| `javaPlanBuilderReadTableReduceBySort` | Public API grouped aggregation and sort combination |
+| `javaPlanBuilderReadTableFilterProjectionTableSink` | Public API filtered projection written to a table |
+| `javaPlanBuilderReadTableJoin` | Public API two-table join with pushed-down record output |
+
+## Environment Variables
+
+Override the default endpoint when running against another PrestoDB:
+
+| Variable | Default |
+|----------|---------|
+| `PRESTO_HOST` | `localhost` |
+| `PRESTO_PORT` | `8081` |
+| `PRESTO_USER` | `test` |
+
+Example:
+
+```bash
+PRESTO_HOST=my-presto PRESTO_PORT=8080 PRESTO_USER=wayang \
+  ./mvnw -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am \
+  -Dtest=PrestoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false \
+  -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test
+```
+
+On PowerShell:
+
+```powershell
+$env:PRESTO_HOST="my-presto"
+$env:PRESTO_PORT="8080"
+$env:PRESTO_USER="wayang"
+.\mvnw.cmd --% -Pskip-prerequisite-check -pl wayang-platforms/wayang-presto -am -Dtest=PrestoOperatorsIT -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false -Drat.skip=true -Dlicense.skip=true test
+Remove-Item Env:PRESTO_HOST, Env:PRESTO_PORT, Env:PRESTO_USER
+```
+
+## Cost Profiling
+
+Follow the shared cost-profiling guide in
+[`guides/cost-profiling.md`](../guides/cost-profiling.md). This setup guide
+only covers the Presto stack itself.
+
+Presto-specific profiling values:
+
+| Item | Value |
+|------|-------|
+| Maven module | `wayang-platforms/wayang-presto` |
+| Profiling test | `PrestoCostPilotIT` |
+| Property prefix | `presto.profile.*` |
+| Default output directory | `target/cost-profiling/presto` |
+| Learned parameters file | `wayang-platforms/wayang-presto/src/main/resources/wayang-presto-defaults.properties` |
+
+## Troubleshooting
+
+### `Catalog does not exist: memory`
+
+Check that `presto-setup/etc/catalog/memory.properties` is a regular file
+before starting the container. If Docker first created the container while the
+source file was absent, it may have mounted a directory at the catalog path.
+Recreate the container after checking out the branch:
+
+```bash
+docker compose -f presto-setup/docker-compose.yml down
+docker compose -f presto-setup/docker-compose.yml up -d --force-recreate --wait
+```
+
+Confirm the mounted path inside the container is a file:
+
+```bash
+docker exec presto sh -c \
+  "ls -l /opt/presto-server/etc/catalog/memory.properties"
+```
+
+Then rerun `PrestoOperatorsIT`.
diff --git a/wayang-api/wayang-api-sql/src/test/java/org/apache/wayang/api/sql/SqlToWayangRelTest.java b/wayang-api/wayang-api-sql/src/test/java/org/apache/wayang/api/sql/SqlToWayangRelTest.java
index 77537826d..91c1d7aef 100755
--- a/wayang-api/wayang-api-sql/src/test/java/org/apache/wayang/api/sql/SqlToWayangRelTest.java
+++ b/wayang-api/wayang-api-sql/src/test/java/org/apache/wayang/api/sql/SqlToWayangRelTest.java
@@ -311,7 +311,7 @@ public RelDataType getRowType(final RelDataTypeFactory typeFactory) {
         final StringBuilder query = JdbcExecutor.createSqlString(jdbcExecutor, table, Arrays.asList(), projection, null,
                 null, null, Arrays.asList());
 
-        assertEquals("SELECT ID, NAME FROM T1;", query.toString());
+        assertEquals("SELECT ID, NAME FROM T1", query.toString());
     }
 
     @Test
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 4b816b2eb..dd5792f52 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
@@ -1,4 +1,3 @@
-package org.apache.wayang.jdbc.execution;
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -17,41 +16,8 @@
  * limitations under the License.
  */
 
-import org.apache.wayang.basic.channels.FileChannel;
-import org.apache.wayang.basic.data.Tuple2;
-import org.apache.wayang.basic.operators.SpatialFilterOperator;
-import org.apache.wayang.basic.operators.SpatialJoinOperator;
-import org.apache.wayang.basic.operators.FilterOperator;
-import org.apache.wayang.basic.operators.JoinOperator;
-import org.apache.wayang.basic.operators.TableSource;
-import org.apache.wayang.core.api.Job;
-import org.apache.wayang.core.api.exception.WayangException;
-import org.apache.wayang.core.optimizer.OptimizationContext;
-import org.apache.wayang.core.plan.executionplan.Channel;
-import org.apache.wayang.core.plan.executionplan.ExecutionStage;
-import org.apache.wayang.core.plan.executionplan.ExecutionTask;
-import org.apache.wayang.core.platform.ExecutionState;
-import org.apache.wayang.core.platform.ExecutorTemplate;
-import org.apache.wayang.core.platform.Platform;
-import org.apache.wayang.core.util.fs.FileSystem;
-import org.apache.wayang.core.util.fs.FileSystems;
-import org.apache.wayang.jdbc.channels.SqlQueryChannel;
-import org.apache.wayang.jdbc.compiler.FunctionCompiler;
-
-import org.apache.wayang.jdbc.operators.JdbcExecutionOperator;
-import org.apache.wayang.jdbc.operators.JdbcFilterOperator;
-import org.apache.wayang.jdbc.operators.JdbcJoinOperator;
-import org.apache.wayang.jdbc.operators.JdbcProjectionOperator;
-import org.apache.wayang.jdbc.operators.JdbcTableSinkOperator;
-import org.apache.wayang.jdbc.operators.JdbcTableSource;
-
-import org.apache.wayang.jdbc.platform.JdbcPlatformTemplate;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
+package org.apache.wayang.jdbc.execution;
 
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.io.UncheckedIOException;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
@@ -78,7 +44,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;
@@ -96,7 +64,6 @@
 /**
  * {@link Executor} implementation for the {@link JdbcPlatformTemplate}.
  */
-
 public class JdbcExecutor extends ExecutorTemplate {
     public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, final JdbcTableSource tableOp,
             final Collection<JdbcExecutionOperator> filterTasks, final JdbcProjectionOperator projectionTask, final JdbcGlobalReduceOperator globalReduceTask, final JdbcReduceByOperator reduceByTask, final JdbcSortOperator sortTask, 
@@ -151,20 +118,12 @@ public static StringBuilder createSqlString(final JdbcExecutor jdbcExecutor, fin
             ));
         }
 
-        appendStatementTerminator(sb);
+        // Intentionally no trailing ';'. A trailing semicolon is unnecessary for a
+        // single-statement JDBC executeQuery and is rejected by strict SQL parsers
+        // such as Trino and BigQuery. Postgres/SQLite/HSQLDB accept its absence.
         return sb;
     }
 
-    private static void appendStatementTerminator(final StringBuilder query) {
-        int i = query.length() - 1;
-        while (i >= 0 && Character.isWhitespace(query.charAt(i))) {
-            i--;
-        }
-        if (i < 0 || query.charAt(i) != ';') {
-            query.append(';');
-        }
-    }
-
     /**
      * Creates a query channel and the sql statement
      * 
@@ -202,16 +161,13 @@ protected static Tuple2<String, SqlQueryChannel.Instance> createSqlQuery(final E
             } else if (operator instanceof JdbcProjectionOperator) {
                 assert projectionTask == null; // Allow one projection operator per stage for now.
                 projectionTask = (JdbcProjectionOperator) operator;
-            } else if (operator instanceof JdbcGlobalReduceOperator) {
-                final JdbcGlobalReduceOperator globalReduce = (JdbcGlobalReduceOperator) operator;
+            } else if (operator instanceof final JdbcGlobalReduceOperator globalReduce) {
                 assert globalReduceTask == null; // Allow one projection operator per stage for now.
                 globalReduceTask = globalReduce;
-            } else if (operator instanceof JdbcReduceByOperator) {
-                final JdbcReduceByOperator reduceBy = (JdbcReduceByOperator) operator;
+            } else if (operator instanceof final JdbcReduceByOperator reduceBy) {
                 assert reduceByTask == null; // Allow one projection operator per stage for now.
                 reduceByTask = reduceBy;
-            } else if (operator instanceof JdbcSortOperator) {
-                final JdbcSortOperator sort = (JdbcSortOperator) operator;
+            } else if (operator instanceof final JdbcSortOperator sort) {
                 assert sortTask == null; // Allow one projection operator per stage for now.
                 sortTask = sort;
             } else if (operator instanceof JoinOperator || (operator instanceof SpatialJoinOperator)) {
@@ -270,7 +226,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();
@@ -296,27 +252,21 @@ private static void executeSinkStage(final ExecutionStage stage, final Optimizat
         // Walk through intermediate operators, stopping at the sink
         ExecutionTask nextTask = JdbcExecutor.findJdbcExecutionOperatorTaskInStage(startTask, stage);
         while (nextTask != null && !(nextTask.getOperator() instanceof JdbcTableSinkOperator)) {
-            if (nextTask.getOperator() instanceof JdbcFilterOperator) {
-                final JdbcFilterOperator filterOperator = (JdbcFilterOperator) nextTask.getOperator();
+            if (nextTask.getOperator() instanceof final JdbcFilterOperator filterOperator) {
                 filterTasks.add(filterOperator);
-            } else if (nextTask.getOperator() instanceof JdbcProjectionOperator) {
-                final JdbcProjectionOperator projectionOperator = (JdbcProjectionOperator) nextTask.getOperator();
+            } else if (nextTask.getOperator() instanceof final JdbcProjectionOperator projectionOperator) {
                 assert projectionTask == null;
                 projectionTask = projectionOperator;
-            } else if (nextTask.getOperator() instanceof JdbcGlobalReduceOperator) {
-                final JdbcGlobalReduceOperator globalReduceOperator = (JdbcGlobalReduceOperator) nextTask.getOperator();
+            } else if (nextTask.getOperator() instanceof final JdbcGlobalReduceOperator globalReduceOperator) {
                 assert globalReduceTask == null;
                 globalReduceTask = globalReduceOperator;
-            } else if (nextTask.getOperator() instanceof JdbcReduceByOperator) {
-                final JdbcReduceByOperator reduceByOperator = (JdbcReduceByOperator) nextTask.getOperator();
+            } else if (nextTask.getOperator() instanceof final JdbcReduceByOperator reduceByOperator) {
                 assert reduceByTask == null;
                 reduceByTask = reduceByOperator;
-            } else if (nextTask.getOperator() instanceof JdbcSortOperator) {
-                final JdbcSortOperator sortOperator = (JdbcSortOperator) nextTask.getOperator();
+            } else if (nextTask.getOperator() instanceof final JdbcSortOperator sortOperator) {
                 assert sortTask == null;
                 sortTask = sortOperator;
-            } else if (nextTask.getOperator() instanceof JdbcJoinOperator) {
-                final JdbcJoinOperator joinOperator = (JdbcJoinOperator) nextTask.getOperator();
+            } else if (nextTask.getOperator() instanceof final JdbcJoinOperator joinOperator) {
                 joinTasks.add(joinOperator);
             } else {
                 throw new WayangException(String.format("Unsupported JDBC execution task %s", nextTask.toString()));
@@ -346,13 +296,41 @@ 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<ExecutionLineageNode> createExecutionLineageNodes(
+            final ExecutionStage stage,
+            final OptimizationContext optimizationContext) {
+        final Collection<ExecutionLineageNode> 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;
     }
 
     /**
@@ -368,40 +346,52 @@ private static void executeSinkStage(final ExecutionStage stage, final Optimizat
     private static ExecutionTask findJdbcExecutionOperatorTaskInStage(final ExecutionTask task,
             final ExecutionStage stage) {
         assert task.getNumOuputChannels() == 1;
-
         final Channel outputChannel = task.getOutputChannel(0);
-
-        if (outputChannel.getConsumers().size() != 1) {
-            return null;
-        }
-
-        final ExecutionTask consumer = outputChannel.getConsumers().iterator().next();
-
-        return consumer.getStage() == stage && consumer.getOperator() instanceof JdbcExecutionOperator ? consumer
-                        : null;
+        final ExecutionTask consumer = WayangCollections.getSingle(outputChannel.getConsumers());
+        return consumer.getStage() == stage && consumer.getOperator() instanceof JdbcExecutionOperator 
+                ? consumer
+                : null;
     }
 
+    /**
+     * Instantiates the outbound {@link SqlQueryChannel} of an
+     * {@link ExecutionTask}.
+     *
+     * @param task                whose outbound {@link SqlQueryChannel} should be
+     *                            instantiated
+     * @param optimizationContext provides information about the
+     *                            {@link ExecutionTask}
+     * @return the {@link SqlQueryChannel.Instance}
+     */
     private static SqlQueryChannel.Instance instantiateOutboundChannel(final ExecutionTask task,
             final OptimizationContext optimizationContext, final JdbcExecutor jdbcExecutor) {
-        assert task.getNumOuputChannels() == 1;
-        assert task.getOutputChannel(0) instanceof SqlQueryChannel;
+        assert task.getNumOuputChannels() == 1 : String.format("Illegal task: %s.", task);
+        assert task.getOutputChannel(0) instanceof SqlQueryChannel : String.format("Illegal task: %s.", task);
 
         final SqlQueryChannel outputChannel = (SqlQueryChannel) task.getOutputChannel(0);
-
         final OptimizationContext.OperatorContext operatorContext = optimizationContext
                 .getOperatorContext(task.getOperator());
-
         return outputChannel.createInstance(jdbcExecutor, operatorContext, 0);
     }
 
+    /**
+     * Instantiates the outbound {@link SqlQueryChannel} of an
+     * {@link ExecutionTask}.
+     *
+     * @param task                       whose outbound {@link SqlQueryChannel}
+     *                                   should be instantiated
+     * @param optimizationContext        provides information about the
+     *                                   {@link ExecutionTask}
+     * @param predecessorChannelInstance preceeding {@link SqlQueryChannel.Instance}
+     *                                   to keep track of lineage
+     * @return the {@link SqlQueryChannel.Instance}
+     */
     private static SqlQueryChannel.Instance instantiateOutboundChannel(final ExecutionTask task,
             final OptimizationContext optimizationContext, final SqlQueryChannel.Instance predecessorChannelInstance,
             final JdbcExecutor jdbcExecutor) {
         final SqlQueryChannel.Instance newInstance = JdbcExecutor.instantiateOutboundChannel(task, optimizationContext,
                 jdbcExecutor);
-
         newInstance.getLineage().addPredecessor(predecessorChannelInstance.getLineage());
-
         return newInstance;
     }
 
@@ -428,7 +418,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
@@ -441,12 +440,16 @@ 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 {
             this.connection.close();
         } catch (final SQLException e) {
-            this.logger.error("Could not close JDBC connection correctly.", e);
+            this.logger.error("Could not close JDBC connection to PostgreSQL correctly.", e);
         }
     }
 
diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcExecutorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcExecutorTest.java
index 0dfd8b698..8f7b3d8a2 100644
--- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcExecutorTest.java
+++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/execution/JdbcExecutorTest.java
@@ -81,7 +81,7 @@ void testExecuteWithPlainTableSource() throws SQLException {
         SqlQueryChannel.Instance sqlQueryChannelInstance =
                 (SqlQueryChannel.Instance) job.getCrossPlatformExecutor().getChannelInstance(sqlToStreamTask.getInputChannel(0));
         assertEquals(
-                "SELECT * FROM customer;",
+                "SELECT * FROM customer",
                 sqlQueryChannelInstance.getSqlQuery()
         );
     }
@@ -130,7 +130,7 @@ void testExecuteWithFilter() throws SQLException {
         SqlQueryChannel.Instance sqlQueryChannelInstance =
                 (SqlQueryChannel.Instance) job.getCrossPlatformExecutor().getChannelInstance(sqlToStreamTask.getInputChannel(0));
         assertEquals(
-                "SELECT * FROM customer WHERE age >= 18;",
+                "SELECT * FROM customer WHERE age >= 18",
                 sqlQueryChannelInstance.getSqlQuery()
         );
     }
@@ -172,7 +172,7 @@ void testExecuteWithProjection() throws SQLException {
         SqlQueryChannel.Instance sqlQueryChannelInstance =
                 (SqlQueryChannel.Instance) job.getCrossPlatformExecutor().getChannelInstance(sqlToStreamTask.getInputChannel(0));
         assertEquals(
-                "SELECT name, age FROM customer;",
+                "SELECT name, age FROM customer",
                 sqlQueryChannelInstance.getSqlQuery()
         );
     }
@@ -240,7 +240,7 @@ void testExecuteWithProjectionAndFilters() throws SQLException {
         SqlQueryChannel.Instance sqlQueryChannelInstance =
                 (SqlQueryChannel.Instance) job.getCrossPlatformExecutor().getChannelInstance(sqlToStreamTask.getInputChannel(0));
         assertEquals(
-                "SELECT name, age FROM customer WHERE age >= 18 AND name IS NOT NULL;",
+                "SELECT name, age FROM customer WHERE age >= 18 AND name IS NOT NULL",
                 sqlQueryChannelInstance.getSqlQuery()
         );
     }
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<String> 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 b5ccb0848..4465751be 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
@@ -112,6 +112,6 @@ void testWithHsqldb() throws SQLException {
             assertTrue(count > 0);
         }
 
-        assertEquals("SELECT COUNT(*) FROM testA;", sqlQueryChannelInstance.getSqlQuery());
+        assertEquals("SELECT COUNT(*) FROM testA", sqlQueryChannelInstance.getSqlQuery());
     }
 }
diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcJoinOperatorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcJoinOperatorTest.java
index 875a7a47b..d56405b19 100644
--- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcJoinOperatorTest.java
+++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcJoinOperatorTest.java
@@ -39,7 +39,9 @@
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.LinkedHashSet;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.mock;
@@ -116,7 +118,12 @@ void testWithHsqldb() throws SQLException {
         joinTask.setOutputChannel(0, new SqlQueryChannel(sqlChannelDescriptor, joinOperator.getOutput(0)));
         joinTask.setStage(sqlStage);
 
-        when(sqlStage.getStartTasks()).thenReturn(Collections.singleton(tableSourceATask));
+        // Deliberately list the right source first: JdbcExecutor must still choose
+        // the join's left source for the FROM clause.
+        when(sqlStage.getStartTasks()).thenReturn(new LinkedHashSet<>(Arrays.asList(
+                tableSourceBTask, tableSourceATask)));
+        when(sqlStage.getAllTasks()).thenReturn(new LinkedHashSet<>(Arrays.asList(
+                tableSourceBTask, tableSourceATask, joinTask)));
         when(sqlStage.getTerminalTasks()).thenReturn(Collections.singleton(joinTask));
 
         ExecutionStage nextStage = mock(ExecutionStage.class);
@@ -135,7 +142,7 @@ void testWithHsqldb() throws SQLException {
         System.out.println();
 
         assertEquals(
-            "SELECT * FROM testA JOIN testB ON testB.a=testA.a;",
+            "SELECT * FROM testA JOIN testB ON testB.a=testA.a",
             sqlQueryChannelInstance.getSqlQuery()
         );
     }
@@ -213,7 +220,7 @@ void testMultiConditionJoinWithHsqldb() throws SQLException {
 
             String generatedSql = sqlQueryChannelInstance.getSqlQuery();
             assertEquals(
-                "SELECT * FROM orders JOIN shipments ON orders.order_id=shipments.order_id AND orders.customer_id=shipments.customer_id;",
+                "SELECT * FROM orders JOIN shipments ON orders.order_id=shipments.order_id AND orders.customer_id=shipments.customer_id",
                 generatedSql
             );
 
diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcReduceByOperatorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcReduceByOperatorTest.java
index f00f4020e..556224027 100644
--- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcReduceByOperatorTest.java
+++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcReduceByOperatorTest.java
@@ -91,6 +91,6 @@ void testWithHsqldb() throws SQLException {
         final SqlQueryChannel.Instance sqlQueryChannelInstance = (SqlQueryChannel.Instance) job.getCrossPlatformExecutor()
                 .getChannelInstance(sqlToStreamTask.getInputChannel(0));
 
-        assertEquals("SELECT col0,COUNT(*) FROM testA GROUP BY col0;", sqlQueryChannelInstance.getSqlQuery());
+        assertEquals("SELECT col0,COUNT(*) FROM testA GROUP BY col0", sqlQueryChannelInstance.getSqlQuery());
     }
 }
diff --git a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcSortOperatorTest.java b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcSortOperatorTest.java
index 118fb7efa..1dc2fe12f 100644
--- a/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcSortOperatorTest.java
+++ b/wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/JdbcSortOperatorTest.java
@@ -86,6 +86,6 @@ void testWithHsqldb() throws SQLException {
         final SqlQueryChannel.Instance sqlQueryChannelInstance = (SqlQueryChannel.Instance) job.getCrossPlatformExecutor()
                 .getChannelInstance(sqlToStreamTask.getInputChannel(0));
 
-        assertEquals("SELECT * FROM testA ORDER BY col0 DESC;", sqlQueryChannelInstance.getSqlQuery());
+        assertEquals("SELECT * FROM testA ORDER BY col0 DESC", sqlQueryChannelInstance.getSqlQuery());
     }
 }
diff --git a/wayang-platforms/wayang-presto/src/main/resources/wayang-presto-defaults.properties b/wayang-platforms/wayang-presto/src/main/resources/wayang-presto-defaults.properties
index f0e165964..d614fa087 100644
--- a/wayang-platforms/wayang-presto/src/main/resources/wayang-presto-defaults.properties
+++ b/wayang-platforms/wayang-presto/src/main/resources/wayang-presto-defaults.properties
@@ -28,21 +28,16 @@ wayang.presto.cores    = 4
 wayang.presto.costs.fix    = 0.0
 wayang.presto.costs.per-ms = 1.0
 
-# ── Cost model ────────────────────────────────────────────────────────────────
+# Cost model
 #
-# Formula: cpu = α * rows + β
+# Formula: cpu = alpha * rows + beta
 #
-# Presto is a distributed MPP engine: low per-row cost (small α) because scans
-# are parallelised across workers, but a noticeable fixed overhead (larger β)
-# from query planning and cluster coordination.
-#
-#   α  = 10   — parallel scan; per-row cost is cheap
-#   β  = 800k — cluster startup + query dispatch overhead
-#
-# These are initial estimates; tune after real benchmarks by fitting the
-# template formula on measured data and updating the 'load' key.
-# ──────────────────────────────────────────────────────────────────────────────
-
+# The concrete .load entries below are reference parameters learned from the
+# local Week 8 Presto 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 Presto deployment and machine for accurate optimization.
+# Keep the matching .load.template entries so the parameters can be relearned.
 wayang.presto.tablesource.load.template = {\
   "type":"mathex", "in":0, "out":1,\
   "cpu":"?*out0 + ?",\
@@ -50,10 +45,14 @@ wayang.presto.tablesource.load.template = {\
   "p":0.9\
 }
 wayang.presto.tablesource.load = {\
-  "in":0, "out":1,\
-  "cpu":"${10*out0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":0,\
+ "out":1,\
+ "cpu":"((3.8231612775422605)*(out0))+(3.486271520327456E8)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.filter.load.template = {\
@@ -63,10 +62,14 @@ wayang.presto.filter.load.template = {\
   "p":0.9\
 }
 wayang.presto.filter.load = {\
-  "in":1, "out":1,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((1084.9370069592017)*(in0))+(78.40893517692683)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.projection.load.template = {\
@@ -76,10 +79,14 @@ wayang.presto.projection.load.template = {\
   "p":0.9\
 }
 wayang.presto.projection.load = {\
-  "in":1, "out":1,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((100.92159156541014)*(in0))+(16.018722027525722)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.join.load.template = {\
@@ -89,10 +96,14 @@ wayang.presto.join.load.template = {\
   "p":0.9\
 }
 wayang.presto.join.load = {\
-  "in":2, "out":1,\
-  "cpu":"${10*in0 + 10*in1 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":2,\
+ "out":1,\
+ "cpu":"(((1.6962577903997843E-4)*(in0))+((366170.0267543196)*(in1)))+(131.93596176659847)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.globalreduce.load.template = {\
@@ -102,10 +113,14 @@ wayang.presto.globalreduce.load.template = {\
   "p":0.9\
 }
 wayang.presto.globalreduce.load = {\
-  "in":1, "out":1,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((0.00485443619692481)*(in0))+(0.0037433905360257595)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.reduceby.load.template = {\
@@ -115,10 +130,14 @@ wayang.presto.reduceby.load.template = {\
   "p":0.9\
 }
 wayang.presto.reduceby.load = {\
-  "in":1, "out":1,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((0.0014566422213525138)*(in0))+(6071.424576432309)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.sort.load.template = {\
@@ -128,10 +147,14 @@ wayang.presto.sort.load.template = {\
   "p":0.9\
 }
 wayang.presto.sort.load = {\
-  "in":1, "out":1,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":1,\
+ "cpu":"((10594.203699004942)*(in0))+(4.062163618916201E8)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.tablesink.load.template = {\
@@ -141,10 +164,14 @@ wayang.presto.tablesink.load.template = {\
   "p":0.9\
 }
 wayang.presto.tablesink.load = {\
-  "in":1, "out":0,\
-  "cpu":"${10*in0 + 800000}",\
-  "ram":"0",\
-  "p":0.9\
+ "type":"mathex",\
+ "in":1,\
+ "out":0,\
+ "cpu":"((599.3170069949873)*(in0))+(8.530778184751565E8)",\
+ "ram":"0",\
+ "disk":"0",\
+ "net":"0",\
+ "p":0.9\
 }
 
 wayang.presto.sqltostream.load.query.template = {\
diff --git a/wayang-platforms/wayang-presto/src/test/java/org/apache/wayang/presto/PrestoCostPilotIT.java b/wayang-platforms/wayang-presto/src/test/java/org/apache/wayang/presto/PrestoCostPilotIT.java
new file mode 100644
index 000000000..0020a77c2
--- /dev/null
+++ b/wayang-platforms/wayang-presto/src/test/java/org/apache/wayang/presto/PrestoCostPilotIT.java
@@ -0,0 +1,818 @@
+/*
+ * 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.presto;
+
+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.presto.operators.PrestoProjectionOperator;
+import org.apache.wayang.presto.operators.PrestoTableSource;
+import org.apache.wayang.presto.platform.PrestoPlatform;
+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 Presto cost-profiling pilot against the local memory connector.
+ */
+class PrestoCostPilotIT {
+
+    private static final String HOST = System.getenv().getOrDefault("PRESTO_HOST", "localhost");
+    private static final int PORT = Integer.parseInt(System.getenv().getOrDefault("PRESTO_PORT", "8081"));
+    private static final String USER = System.getenv().getOrDefault("PRESTO_USER", "test");
+    private static final String JDBC_URL = String.format("jdbc:presto://%s:%d/memory", HOST, PORT);
+
+    private static final String SCHEMA = "memory.wayang_profile";
+    private static final String CUSTOMERS_1K = SCHEMA + ".customers_1k";
+    private static final int[] ROW_COUNTS = parseIntList(System.getProperty(
+            "presto.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 = "Presto profile join flatten";
+    private static final String JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME = "Presto profile join order tier amount flatten";
+    private static final String JOIN_TIER_AMOUNT_FLATTEN_NAME = "Presto profile join tier amount flatten";
+    private static final Path OUTPUT_DIR = Paths.get(System.getProperty(
+            "presto.profile.outputDir",
+            "target/cost-profiling/presto"
+    ));
+    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<String> PLAN_IDS = Arrays.asList(
+            System.getProperty(
+                    "presto.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("presto.profile.repetitions", "6")
+    );
+    private static final boolean RESET_OUTPUT = Boolean.parseBoolean(
+            System.getProperty("presto.profile.reset", "true")
+    );
+
+    @Test
+    void runPilot() throws Exception {
+        Assumptions.assumeTrue(isPrestoAvailable(), "Presto 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);
+            dropTable(sinkTable);
+        }
+    }
+
+    private WayangPlan createPlan(String planId, String sourceTable, String sinkTable) {
+        PrestoTableSource source = new PrestoTableSource(sourceTable, COLUMNS);
+
+        if ("S01".equals(planId)) {
+            TableSink<Record> sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+            source.connectTo(0, sink, 0);
+            return new WayangPlan(sink);
+        }
+
+        if ("S02".equals(planId)) {
+            TableSink<Record> sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+            FilterOperator<Record> filter = createAmerFilter();
+            source.connectTo(0, filter, 0);
+            filter.connectTo(0, sink, 0);
+            return new WayangPlan(sink);
+        }
+
+        if ("S03".equals(planId)) {
+            TableSink<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "order_id", "amount");
+            MapOperator<Record, Record> projection = createOrderAmountProjection();
+            source.connectTo(0, projection, 0);
+            projection.connectTo(0, sink, 0);
+            return new WayangPlan(sink);
+        }
+
+        if ("S04".equals(planId)) {
+            TableSink<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "order_id", "amount");
+            FilterOperator<Record> filter = createAmerFilter();
+            MapOperator<Record, Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "total_amount");
+            GlobalReduceOperator<Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "bucket", "total_amount");
+            ReduceByOperator<Record, Record> 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<Record> sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+            SortOperator<Record, Record> sort = createAmountSortOperator(3);
+            source.connectTo(0, sort, 0);
+            sort.connectTo(0, sink, 0);
+            return new WayangPlan(sink);
+        }
+
+        if ("S08".equals(planId)) {
+            PrestoTableSource customers = new PrestoTableSource(CUSTOMERS_1K, "cust_id", "tier");
+            JoinOperator<Record, Record, Record> 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<Tuple2<Record, Record>, Record> flatten = createJoinFlattenOperator();
+            TableSink<Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "total_amount");
+            FilterOperator<Record> filter = createAmerFilter();
+            GlobalReduceOperator<Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "bucket", "total_amount");
+            FilterOperator<Record> filter = createAmerFilter();
+            ReduceByOperator<Record, Record> 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<Record> sink = new TableSink<>(new Properties(), "overwrite", sinkTable, COLUMNS);
+            FilterOperator<Record> filter = createAmerFilter();
+            SortOperator<Record, Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "order_id", "amount");
+            MapOperator<Record, Record> projection = createOrderAmountProjection();
+            SortOperator<Record, Record> 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<Record> sink = new TableSink<>(
+                    new Properties(), "overwrite", sinkTable, "order_id", "amount");
+            FilterOperator<Record> filter = createAmerFilter();
+            MapOperator<Record, Record> projection = createOrderAmountProjection();
+            SortOperator<Record, Record> 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)) {
+            PrestoTableSource customers = new PrestoTableSource(CUSTOMERS_1K, "cust_id", "tier");
+            FilterOperator<Record> filter = createAmerFilter();
+            JoinOperator<Record, Record, Record> join = createCustomerJoinOperator(sourceTable);
+            MapOperator<Tuple2<Record, Record>, Record> flatten = createJoinFlattenOperator();
+            TableSink<Record> 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)) {
+            PrestoTableSource customers = new PrestoTableSource(CUSTOMERS_1K, "cust_id", "tier");
+            JoinOperator<Record, Record, Record> join = createCustomerJoinOperator(sourceTable);
+            MapOperator<Tuple2<Record, Record>, Record> flatten = createJoinOrderTierAmountFlattenOperator();
+            SortOperator<Record, Record> sort = createAmountSortOperator(2);
+            TableSink<Record> 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)) {
+            PrestoTableSource customers = new PrestoTableSource(CUSTOMERS_1K, "cust_id", "tier");
+            JoinOperator<Record, Record, Record> join = createCustomerJoinOperator(sourceTable);
+            MapOperator<Tuple2<Record, Record>, Record> flatten = createJoinTierAmountFlattenOperator();
+            ReduceByOperator<Record, Record> reduceBy = createTierReduceByOperator();
+            TableSink<Record> 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<Record> createGlobalAmountReduceOperator() {
+        return new GlobalReduceOperator<>(
+                new ReduceDescriptor<>((left, right) -> left, Record.class)
+                        .withSqlImplementation("SUM(amount) AS total_amount"),
+                DataSetType.createDefault(Record.class));
+    }
+
+    private static ReduceByOperator<Record, Record> 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<Record, Record> 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<Record, Record> 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<Record, Record, Record> 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<Record> createAmerFilter() {
+        return new FilterOperator<>(
+                new PredicateDescriptor<>(
+                        (Record record) -> "AMER".equals(record.getField(2)),
+                        Record.class
+                ).withSqlImplementation("region = 'AMER'")
+        );
+    }
+
+    private static MapOperator<Record, Record> createOrderAmountProjection() {
+        return new MapOperator<>(
+                ProjectionDescriptor.createForRecords(
+                        new RecordType(COLUMNS),
+                        "order_id", "amount"),
+                DataSetType.createDefault(Record.class),
+                DataSetType.createDefault(Record.class));
+    }
+
+    private static MapOperator<Record, Record> 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<Record, Record> createTierAmountProjection() {
+        return new MapOperator<>(
+                ProjectionDescriptor.createForRecords(
+                        new RecordType(JOIN_COLUMNS),
+                        "tier", "amount"),
+                DataSetType.createDefault(Record.class),
+                DataSetType.createDefault(Record.class));
+    }
+
+    private static MapOperator<Tuple2<Record, Record>, Record> createJoinFlattenOperator() {
+        return createJoinFlattenOperator(new JoinFlattenFunction(), JOIN_FLATTEN_NAME);
+    }
+
+    private static MapOperator<Tuple2<Record, Record>, Record> createJoinOrderTierAmountFlattenOperator() {
+        return createJoinFlattenOperator(new JoinOrderTierAmountFlattenFunction(), JOIN_ORDER_TIER_AMOUNT_FLATTEN_NAME);
+    }
+
+    private static MapOperator<Tuple2<Record, Record>, Record> createJoinTierAmountFlattenOperator() {
+        return createJoinFlattenOperator(new JoinTierAmountFlattenFunction(), JOIN_TIER_AMOUNT_FLATTEN_NAME);
+    }
+
+    private static MapOperator<Tuple2<Record, Record>, Record> createJoinFlattenOperator(
+            FunctionDescriptor.SerializableFunction<Tuple2<Record, Record>, Record> function,
+            String name) {
+        MapOperator<Tuple2<Record, Record>, 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.presto.jdbc.url", JDBC_URL);
+        configuration.setProperty("wayang.presto.jdbc.user", USER);
+        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(Presto.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 + " 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 + " 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) 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,
+                    "ok",
+                    ""));
+            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 isPrestoAvailable() {
+        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 {
+        Properties properties = new Properties();
+        properties.setProperty("user", USER);
+        return DriverManager.getConnection(JDBC_URL, properties);
+    }
+
+    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<Tuple2<Record, Record>, Record> {
+
+        @Override
+        public Record apply(Tuple2<Record, Record> tuple) {
+            return flattenJoinResult(tuple);
+        }
+    }
+
+    private static final class JoinOrderTierAmountFlattenFunction implements
+            FunctionDescriptor.SerializableFunction<Tuple2<Record, Record>, Record> {
+
+        @Override
+        public Record apply(Tuple2<Record, Record> tuple) {
+            return flattenJoinOrderTierAmountResult(tuple);
+        }
+    }
+
+    private static final class JoinTierAmountFlattenFunction implements
+            FunctionDescriptor.SerializableFunction<Tuple2<Record, Record>, Record> {
+
+        @Override
+        public Record apply(Tuple2<Record, Record> tuple) {
+            return flattenJoinTierAmountResult(tuple);
+        }
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private static final class JoinFlattenMapping implements Mapping {
+
+        @Override
+        public java.util.Collection<PlanTransformation> getTransformations() {
+            OperatorPattern<MapOperator> 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<MapOperator>(
+                    (matchedOperator, epoch) -> createPrestoProjection(matchedOperator.getName()).at(epoch));
+
+            return Collections.singleton(new PlanTransformation(
+                    SubplanPattern.createSingleton(pattern),
+                    factory,
+                    PrestoPlatform.getInstance()));
+        }
+
+        private static PrestoProjectionOperator createPrestoProjection(String operatorName) {
+            ProjectionDescriptor<Tuple2<Record, Record>, Record> descriptor = new ProjectionDescriptor<>(
+                    getJoinFlattenFunction(operatorName),
+                    Arrays.asList(getJoinFlattenColumns(operatorName)),
+                    DataUnitType.createBasicUnchecked(Tuple2.class),
+                    DataUnitType.createBasic(Record.class));
+            MapOperator<Tuple2<Record, Record>, Record> projection = new MapOperator<>(
+                    descriptor,
+                    DataSetType.createDefaultUnchecked(Tuple2.class),
+                    DataSetType.createDefault(Record.class));
+            projection.setName(operatorName);
+            return new PrestoProjectionOperator((MapOperator<Record, Record>) (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<Tuple2<Record, Record>, 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-plugins/wayang-spatial/src/test/java/org/apache/wayang/spatial/operators/jdbc/JdbcSpatialJoinOperatorTest.java b/wayang-plugins/wayang-spatial/src/test/java/org/apache/wayang/spatial/operators/jdbc/JdbcSpatialJoinOperatorTest.java
index 313679f65..77cf23537 100644
--- a/wayang-plugins/wayang-spatial/src/test/java/org/apache/wayang/spatial/operators/jdbc/JdbcSpatialJoinOperatorTest.java
+++ b/wayang-plugins/wayang-spatial/src/test/java/org/apache/wayang/spatial/operators/jdbc/JdbcSpatialJoinOperatorTest.java
@@ -78,8 +78,8 @@ void testSpatialJoinIntersectsGeneratesCorrectSql() throws SQLException {
         String sql = buildSpatialJoinSql(SpatialPredicate.INTERSECTS);
 
         assertEquals(
-                "SELECT * FROM testA JOIN testB ON ST_Intersects(testA.geom, testB.geom);",
-                sql
+                "SELECT * FROM testA JOIN testB ON ST_Intersects(testA.geom, testB.geom)",
+                stripTrailingSemicolon(sql)
         );
     }
 
@@ -88,8 +88,8 @@ void testSpatialJoinContainsGeneratesCorrectSql() throws SQLException {
         String sql = buildSpatialJoinSql(SpatialPredicate.CONTAINS);
 
         assertEquals(
-                "SELECT * FROM testA JOIN testB ON ST_Contains(testA.geom, testB.geom);",
-                sql
+                "SELECT * FROM testA JOIN testB ON ST_Contains(testA.geom, testB.geom)",
+                stripTrailingSemicolon(sql)
         );
     }
 
@@ -98,11 +98,15 @@ void testSpatialJoinWithinGeneratesCorrectSql() throws SQLException {
         String sql = buildSpatialJoinSql(SpatialPredicate.WITHIN);
 
         assertEquals(
-                "SELECT * FROM testA JOIN testB ON ST_Within(testA.geom, testB.geom);",
-                sql
+                "SELECT * FROM testA JOIN testB ON ST_Within(testA.geom, testB.geom)",
+                stripTrailingSemicolon(sql)
         );
     }
 
+    private static String stripTrailingSemicolon(String sql) {
+        return sql.endsWith(";") ? sql.substring(0, sql.length() - 1) : sql;
+    }
+
     /**
      * Sets up a JDBC execution pipeline (two table sources -> spatial join -> SqlToStream)
      * and returns the generated SQL query string.
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..e297cf9c6 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.presto.Presto");
 
         // 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.
      *