Make Presto integration tests engine-only (no Java plugin)

Run the whole Wayang plan, including the terminal sink, inside Presto:
register only Presto.plugin(), end every PrestoOperatorsIT test in a
TableSink that compiles to CREATE TABLE ... AS SELECT, and assert results
via plain JDBC after execute() returns.

- jdbc-template JdbcExecutor.executeSinkStage: use selectStartTask for
  multi-source joins and collect global-reduce/reduce-by/sort into the
  composed CREATE TABLE AS SELECT (ported from wayang-trino-only-test).
- api-scala-java DataQuantaBuilder: add withSqlUdf/withSqlUdfs on the
  reduce/join/sort builders so JavaPlanBuilder tests can push SQL down.
- PrestoOperatorsIT: 13 engine-only tests (8 operator-level + 5
  JavaPlanBuilder); join Tuple2->Record handled by a test-only flatten
  mapping to PrestoProjectionOperator.
- improvement.md: document the engine-only shape.

Verified against live PrestoDB (memory connector): Tests run: 13,
Failures: 0, Errors: 0, Skipped: 0; 13 CREATE TABLE AS SELECT executed
in Presto.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: JunWang222

improvement.md

@@ -0,0 +1,102 @@
+# Presto engine-only integration test
+
+## 1. What this branch demonstrates
+
+The question this branch answers is **not** "does Presto execute some single
+operator?" but:
+
+> From `WayangContext.execute(...)` to the end of the whole Wayang plan, do all
+> data processing **and** the final sink run inside Presto, **without** registering
+> `Java.basicPlugin()`?
+
+On this branch the answer is **yes**. `PrestoOperatorsIT`:
+
+- registers **only** `Presto.plugin()` — no `Java.basicPlugin()`;
+- ends **every** Wayang plan in a Presto `TableSink`, which compiles to a single
+  `CREATE TABLE ... AS SELECT` executed inside Presto;
+- after `WayangContext.execute(...)` returns, JUnit reads the result table with a
+  plain JDBC query (assertion only — not part of the Wayang plan);
+- handles the join `Tuple2<Record, Record>` vs flat `Record` mismatch with a
+  test-only flatten mapping (see §4). This is a test-only scheme, not a final
+  decision on Tuple-to-Record semantics for JDBC platforms.
+
+This mirrors the Trino-only work on `wayang-trino-only-test`; the contrast is the
+older mixed branch `wayang-presto`, which registered both `Java.basicPlugin()` and
+`Presto.plugin()` and ended most operator tests in a Java `LocalCallbackSink`.
+
+## 2. Execution shape
+
+```text
+Presto TableSource -> Presto operator(s) -> Presto TableSink
+                                               |
+                                               v
+                                CREATE TABLE memory.wayang_it.operator_result AS SELECT ...
+
+WayangContext.execute(...) returns
+                                               |
+                                               v
+                            JUnit queries the result table over JDBC (assertions only)
+```
+
+The final JDBC query is part of the test only: it is not in the Wayang logical
+plan, it is not a Wayang Java execution operator, and it does not process plan
+data on Presto's behalf — it just inspects what Presto already wrote.
+
+## 3. The shared executor change
+
+All JDBC platforms share `wayang-jdbc-template`'s `JdbcExecutor`. When a stage's
+terminal task is a `JdbcTableSinkOperator`, `JdbcExecutor.executeSinkStage(...)`
+composes and runs the `CREATE TABLE ... AS SELECT` directly on the connection.
+
+The previous Presto branch's `executeSinkStage` had two gaps that only surface
+once **every** test ends in a `TableSink`:
+
+1. It asserted a stage has a single source, so a join (orders + customers, two
+   sources) could not be composed. It also lacked the `selectStartTask(...)` helper
+   that picks the correct left/`FROM` table.
+2. It only collected filter, projection and join; it threw `WayangException` for
+   global reduce, reduce-by and sort, and passed `null` for them to
+   `createSqlString(...)`.
+
+This branch ports the engine-only `executeSinkStage` (identical to the file on
+`wayang-trino-only-test`): it uses `selectStartTask(...)` for multi-source joins
+and collects global reduce / reduce-by / sort, passing them into the existing
+`createSqlString(...)`. The file is platform-agnostic. (Assertions are enabled
+under Maven — `pom.xml` `enableAssertions=true` — so without this change a
+join/reduce/sort sink would fail loudly, not silently.)
+
+## 4. The join flatten mapping
+
+A logical `JoinOperator` emits `Tuple2<Record, Record>`, while a pushed-down JDBC
+join already emits a flat `Record`. The test wires an explicit flatten `MapOperator`
+(named `JOIN_FLATTEN_NAME`) and registers a test-only `JoinFlattenMapping` on the
+configuration whitelist; the mapping rewrites that named map into a
+`PrestoProjectionOperator`, so the flatten is also pushed into Presto SQL and the
+plan stays entirely in Presto. (Same approach as the Trino-only test, using
+`PrestoProjectionOperator` + `PrestoPlatform`.)
+
+## 5. Coverage and results
+
+`PrestoOperatorsIT` runs 13 tests (8 operator-level + 5 high-level
+`JavaPlanBuilder`) covering `TableSource`, `Filter`, `Projection`, `Join`,
+`GlobalReduce`, `ReduceBy`, `Sort`, `TableSink`. Each composes a
+`CREATE TABLE ... AS SELECT` and additionally asserts, via
+`system.runtime.queries`, that the expected SQL actually reached Presto.
+
+The high-level tests also rely on the `withSqlUdf` / `withSqlUdfs` additions to
+`DataQuantaBuilder.scala` (ported from `wayang-trino-only-test`) so reduce / join /
+sort builders can carry SQL implementations.
+
+```bash
+docker compose -f presto-setup/docker-compose.yml up -d
+
+JAVA_HOME=<jdk17> mvn test -pl wayang-platforms/wayang-presto -am \
+  -Dtest=PrestoOperatorsIT -DfailIfNoTests=false -Dsurefire.failIfNoSpecifiedTests=false \
+  -Drat.skip=true -Dlicense.skip=true -Pskip-prerequisite-check
+
+docker compose -f presto-setup/docker-compose.yml down
+```
+
+Expected: `Tests run: 13, Failures: 0, Errors: 0, Skipped: 0`. The suite scales
+its fixtures to 120k rows and creates/drops its own `memory.wayang_it` schema.
+If Presto is unreachable the whole class is skipped (not failed).

wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala

@@ -28,7 +28,7 @@ import org.apache.wayang.api.graph.{Edge, EdgeDataQuantaBuilder, EdgeDataQuantaB
 import org.apache.wayang.api.util.{DataQuantaBuilderCache, TypeTrap}
 import org.apache.wayang.basic.data.{Record, Tuple2 => RT2}
 import org.apache.wayang.basic.model.{DLModel, Model, LogisticRegressionModel,DecisionTreeRegressionModel}
-import org.apache.wayang.basic.operators.{DLTrainingOperator, GlobalReduceOperator, LocalCallbackSink, MapOperator, SampleOperator, LogisticRegressionOperator,DecisionTreeRegressionOperator, LinearSVCOperator}
+import org.apache.wayang.basic.operators.{DLTrainingOperator, GlobalReduceOperator, JoinOperator, LocalCallbackSink, MapOperator, ReduceByOperator, SampleOperator, SortOperator, LogisticRegressionOperator,DecisionTreeRegressionOperator, LinearSVCOperator}
 import org.apache.wayang.commons.util.profiledb.model.Experiment
 import org.apache.wayang.core.api.spatial.{SpatialGeometry, SpatialPredicate}
 import org.apache.wayang.core.function.FunctionDescriptor.{SerializableBiFunction, SerializableBinaryOperator, SerializableFunction, SerializableIntUnaryOperator, SerializablePredicate}
@@ -1020,6 +1020,10 @@ class SortDataQuantaBuilder[T, Key](inputDataQuanta: DataQuantaBuilder[_, T],
   /** [[LoadEstimator]] to estimate the RAM load of the [[keyUdf]]. */
   private var keyUdfRamEstimator: LoadEstimator = _
 
+  /** SQL column and direction implementing the sort key. */
+  private var sqlColumnName: String = _
+  private var sqlDirection: String = _
+
 
   // Try to infer the type classes from the UDFs.
   locally {
@@ -1060,8 +1064,27 @@ class SortDataQuantaBuilder[T, Key](inputDataQuanta: DataQuantaBuilder[_, T],
     this
   }
 
-  override protected def build =
-    applyTargetPlatforms(inputDataQuanta.dataQuanta().sortJava(keyUdf)(this.keyTag), this.getTargetPlatforms())
+  /**
+    * Add a SQL implementation of the sort key.
+    *
+    * @param columnName SQL column to sort by
+    * @param direction SQL sort direction, e.g. `ASC` or `DESC`
+    * @return this instance
+    */
+  def withSqlUdf(columnName: String, direction: String) = {
+    this.sqlColumnName = columnName
+    this.sqlDirection = direction
+    this
+  }
+
+  override protected def build = {
+    val result = inputDataQuanta.dataQuanta().sortJava(keyUdf)(this.keyTag)
+    if (this.sqlColumnName != null) {
+      result.operator.asInstanceOf[SortOperator[T, Key]]
+        .getKeyDescriptor.withSqlImplementation(this.sqlColumnName, this.sqlDirection)
+    }
+    applyTargetPlatforms(result, this.getTargetPlatforms())
+  }
 
 }
 
@@ -1283,6 +1306,10 @@ class ReduceByDataQuantaBuilder[Key, T](inputDataQuanta: DataQuantaBuilder[_, T]
   /** [[LoadProfileEstimator]] to estimate the [[LoadProfile]] of the [[udf]]. */
   private var udfLoadProfileEstimator: LoadProfileEstimator = _
 
+  /** SQL implementations of the grouping key and reduction. */
+  private var keySqlUdf: String = _
+  private var reduceSqlUdf: String = _
+
   // TODO: Add these estimators.
   //  /** [[LoadEstimator]] to estimate the CPU load of the [[keyUdf]]. */
   //  private var keyUdfCpuEstimator: LoadEstimator = _
@@ -1322,7 +1349,29 @@ class ReduceByDataQuantaBuilder[Key, T](inputDataQuanta: DataQuantaBuilder[_, T]
     this
   }
 
-  override protected def build = applyTargetPlatforms(inputDataQuanta.dataQuanta().reduceByKeyJava(keyUdf, udf, this.udfLoadProfileEstimator), this.getTargetPlatforms())
+  /**
+    * Add SQL implementations of the grouping key and reduction.
+    *
+    * @param keySqlUdf SQL grouping column
+    * @param reduceSqlUdf SQL aggregate expression
+    * @return this instance
+    */
+  def withSqlUdfs(keySqlUdf: String, reduceSqlUdf: String) = {
+    this.keySqlUdf = keySqlUdf
+    this.reduceSqlUdf = reduceSqlUdf
+    this
+  }
+
+  override protected def build = {
+    val result = inputDataQuanta.dataQuanta()
+      .reduceByKeyJava(keyUdf, udf, this.udfLoadProfileEstimator)
+    if (this.keySqlUdf != null) {
+      val operator = result.operator.asInstanceOf[ReduceByOperator[T, Key]]
+      operator.getKeyDescriptor.withSqlImplementation(this.keySqlUdf, this.keySqlUdf)
+      operator.getReduceDescriptor.withSqlImplementation(this.reduceSqlUdf)
+    }
+    applyTargetPlatforms(result, this.getTargetPlatforms())
+  }
 }
 
 /**
@@ -1402,6 +1451,9 @@ class GlobalReduceDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T],
   /** [[LoadProfileEstimator]] to estimate the [[LoadProfile]] of the [[udf]]. */
   private var udfLoadProfileEstimator: LoadProfileEstimator = _
 
+  /** SQL implementation of the reduction. */
+  private var sqlUdf: String = _
+
   // Try to infer the type classes from the udf.
   locally {
     val parameters = ReflectionUtils.getTypeParameters(udf.getClass, classOf[SerializableBinaryOperator[_]])
@@ -1422,7 +1474,25 @@ class GlobalReduceDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T],
     this
   }
 
-  override protected def build = applyTargetPlatforms(inputDataQuanta.dataQuanta().reduceJava(udf, this.udfLoadProfileEstimator), this.getTargetPlatforms())
+  /**
+    * Add a SQL implementation of the reduction.
+    *
+    * @param sqlUdf SQL aggregate expression
+    * @return this instance
+    */
+  def withSqlUdf(sqlUdf: String) = {
+    this.sqlUdf = sqlUdf
+    this
+  }
+
+  override protected def build = {
+    val result = inputDataQuanta.dataQuanta().reduceJava(udf, this.udfLoadProfileEstimator)
+    if (this.sqlUdf != null) {
+      result.operator.asInstanceOf[GlobalReduceOperator[T]]
+        .getReduceDescriptor.withSqlImplementation(this.sqlUdf)
+    }
+    applyTargetPlatforms(result, this.getTargetPlatforms())
+  }
 
 }
 
@@ -1490,6 +1560,12 @@ class JoinDataQuantaBuilder[In0, In1, Key](inputDataQuanta0: DataQuantaBuilder[_
   /** [[LoadEstimator]] to estimate the RAM load of the [[keyUdf1]]. */
   private var keyUdf1RamEstimator: LoadEstimator = _
 
+  /** SQL implementations of both join keys. */
+  private var keyUdf0TableName: String = _
+  private var keyUdf0SqlUdf: String = _
+  private var keyUdf1TableName: String = _
+  private var keyUdf1SqlUdf: String = _
+
   // Try to infer the type classes from the UDFs.
   locally {
     val parameters = ReflectionUtils.getTypeParameters(keyUdf0.getClass, classOf[SerializableFunction[_, _]])
@@ -1568,6 +1644,22 @@ class JoinDataQuantaBuilder[In0, In1, Key](inputDataQuanta0: DataQuantaBuilder[_
     this
   }
 
+  /**
+    * Add SQL implementations of both join keys.
+    *
+    * @return this instance
+    */
+  def withSqlUdfs(thisTableName: String,
+                  thisKeySqlUdf: String,
+                  thatTableName: String,
+                  thatKeySqlUdf: String) = {
+    this.keyUdf0TableName = thisTableName
+    this.keyUdf0SqlUdf = thisKeySqlUdf
+    this.keyUdf1TableName = thatTableName
+    this.keyUdf1SqlUdf = thatKeySqlUdf
+    this
+  }
+
   /**
     * Assemble the joined elements to new elements.
     *
@@ -1579,8 +1671,16 @@ class JoinDataQuantaBuilder[In0, In1, Key](inputDataQuanta0: DataQuantaBuilder[_
       override def apply(joinTuple: RT2[In0, In1]): NewOut = udf.apply(joinTuple.field0, joinTuple.field1)
     })
 
-  override protected def build =
-    applyTargetPlatforms(inputDataQuanta0.dataQuanta().joinJava(keyUdf0, inputDataQuanta1.dataQuanta(), keyUdf1)(inputDataQuanta1.classTag, this.keyTag), this.getTargetPlatforms())
+  override protected def build = {
+    val result = inputDataQuanta0.dataQuanta()
+      .joinJava(keyUdf0, inputDataQuanta1.dataQuanta(), keyUdf1)(inputDataQuanta1.classTag, this.keyTag)
+    if (this.keyUdf0SqlUdf != null) {
+      val operator = result.operator.asInstanceOf[JoinOperator[In0, In1, Key]]
+      operator.getKeyDescriptor0.withSqlImplementation(this.keyUdf0TableName, this.keyUdf0SqlUdf)
+      operator.getKeyDescriptor1.withSqlImplementation(this.keyUdf1TableName, this.keyUdf1SqlUdf)
+    }
+    applyTargetPlatforms(result, this.getTargetPlatforms())
+  }
 
 }