From 22b1fd197a0f684df0264e9714d2e251b02995a8 Mon Sep 17 00:00:00 2001 From: mspruc Date: Fri, 28 Aug 2026 09:35:27 +0200 Subject: [PATCH 01/10] implementation of semantic filters in Wayang --- guides/semantic-operators-in-Wayang.md | 138 +++++++++ .../org/apache/wayang/api/DataQuanta.scala | 33 +++ .../apache/wayang/api/DataQuantaBuilder.scala | 42 ++- .../operators/SemanticFilterOperator.java | 69 +++++ wayang-platforms/pom.xml | 1 + .../java/operators/JavaFilterOperator.java | 11 +- wayang-platforms/wayang-semantic/bin/.project | 23 ++ .../org.eclipse.core.resources.prefs | 2 + .../bin/.settings/org.eclipse.m2e.core.prefs | 4 + wayang-platforms/wayang-semantic/bin/pom.xml | 63 ++++ .../org/apache/wayang/semantic/Semantic.class | Bin 0 -> 1711 bytes .../semantic/mappings/JavaFilterMapping.class | Bin 0 -> 1072 bytes .../wayang/semantic/mappings/Mappings.class | Bin 0 -> 1291 bytes .../operators/SemanticFilterOperator.class | Bin 0 -> 846 bytes .../operators/SemanticMapOperator.class | Bin 0 -> 837 bytes .../semantic/plugin/SemanticPlugin.class | Bin 0 -> 3387 bytes .../semantic/PositiveSentimentUdf.class | Bin 0 -> 4429 bytes .../org/apache/wayang/semantic/Review.class | Bin 0 -> 4940 bytes .../apache/wayang/semantic/SemBenchTest.class | Bin 0 -> 6439 bytes wayang-platforms/wayang-semantic/pom.xml | 75 +++++ .../org/apache/wayang/semantic/Semantic.java | 45 +++ .../semantic/mappings/JavaFilterMapping.java | 64 ++++ .../wayang/semantic/mappings/Mappings.java | 29 ++ .../operators/SemanticMapOperator.java | 24 ++ .../semantic/plugin/SemanticPlugin.java | 77 +++++ .../semantic/udf/SemanticAlgorithm.java | 36 +++ .../apache/wayang/semantic/SemBenchTest.java | 273 ++++++++++++++++++ 27 files changed, 1006 insertions(+), 3 deletions(-) create mode 100644 guides/semantic-operators-in-Wayang.md create mode 100644 wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java create mode 100644 wayang-platforms/wayang-semantic/bin/.project create mode 100644 wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs create mode 100644 wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs create mode 100644 wayang-platforms/wayang-semantic/bin/pom.xml create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/Mappings.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/operators/SemanticFilterOperator.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class create mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/SemBenchTest.class create mode 100644 wayang-platforms/wayang-semantic/pom.xml create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java create mode 100644 wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java diff --git a/guides/semantic-operators-in-Wayang.md b/guides/semantic-operators-in-Wayang.md new file mode 100644 index 000000000..a7de72834 --- /dev/null +++ b/guides/semantic-operators-in-Wayang.md @@ -0,0 +1,138 @@ +# Developing with Semantic Operators in Apache Wayang + +This guide explains how to define semantic operators, provide executable implementations, register multiple implementations, estimate their costs, and let Apache Wayang select an implementation during optimization. + +The example uses a semantic filter that classifies movie reviews as positive or negative through Ollama. + +## 1. Semantic operators + +A semantic operator an operator much like any Wayang operator, however, it takes a prompt as input, +that describes how it should act. + +For example: + +```java +.semanticFilter( + "Analyze the review after the | and write either " + + "\"POSITIVE\" if the review has a positive sentiment and " + + "\"NEGATIVE\" if the review has a negative sentiment." +) +``` + +The prompt describes the task. A `SemanticAlgorithm` provides one concrete implementation of that task. +A `SemanticAlgorithm` could theoretically be any UDF you desire, there are no strict requirements on its implementation. +The implementation just requires the UDF is described as something that takes an input `Record` and a `prompt` and outputs +whatever datatype is required by the operator. + +The ollama local open source model is used as an example for this guide, but may also be useful for quick development. + +We setup our local model hosting locallly using Docker: + +```yaml +ollama: + image: ollama/ollama:latest + container_name: apache-wayang-ollama + ports: + - "11434:11434" + volumes: + - ollama-data:/root/.ollama + - ./docker/ollama-init.sh:/ollama-init.sh + entrypoint: ["/bin/bash", "/ollama-init.sh"] + restart: always + tty: true + networks: + - wayang-network +``` + +```sh +#!/bin/bash +ollama serve & +sleep 10 +ollama pull tinyllama +wait +``` + +We setup the backend call to the model in Wayang: + +```java +private static String callOllama(final String prompt) throws IOException, InterruptedException { + final String requestBody = String.format("{\"model\": \"%s\", \"prompt\": \"%s\", \"stream\": false}", + MODEL_NAME, escapeJson(prompt)); + + final HttpRequest request = HttpRequest.newBuilder().uri(URI.create(OLLAMA_API_URL)) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(requestBody)) + .timeout(Duration.ofMinutes(2)).build(); + + final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() != 200) { + throw new IOException("Ollama API error: " + response.body()); + } + + // Parse JSON response to extract the "response" field + return parseOllamaResponse(response.body()); +} +``` + +and the semantic UDF: + +```java +public static boolean isPositiveSentiment3(final Review review, final String prompt) + throws IOException, InterruptedException { + final String response = callOllama(prompt + " | " + review.getReviewText()); + return response.contains("POSITIVE"); +} +``` + +Now we can define our `SemanticAlgorithm`: + +```java +final SemanticAlgorithm ollamaFilter = new SemanticAlgorithm(); +ollamaFilter.impl = (input, prompt) -> { + try { + return isPositiveSentiment((Review) input); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } +}; +``` + +And also provide a UDF load estimator: + +```java +ollamaFilter.loadProfileEstimator = + LoadProfileEstimators.createFromSpecification( + "wayang.semantic.ollama.model1.load", + configuration + ); +``` + +Note that you should provide multiple configurations if you have more semantic operators. +Now we register these UDFs to the `SemanticPlugin` this will automatically construct a new operator +per UDF, please note this may have implications for optimization time depending on your setup. + +```java +final SemanticPlugin plugin = Semantic.plugin() + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter2) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter3); + +final WayangContext wayangContext = new WayangContext() + .withPlugin(Java.basicPlugin()) + .withPlugin(plugin); +``` + +Currently we only have mappings for semantic operators in Java, so you also need the `Java.basicPlugin()`. +Finally, you can construct your Wayang plan: + +```java +final Collection positiveReviewCnt = planBuilder.loadCollection(loadReviews()) + .filter(review -> "taken_3".equals(review.getId())) + .semanticFilter("Analyze the review after the | and write either \"POSITIVE\" if the review has a positive sentiment and \"NEGATIVE\" if the review has a negative sentiment.") + .withTargetModels(ollamaFilter, ollamaFilter2, ollamaFilter3) + .count() + .collect(); +``` + +You need to provide semantic operators with their target models, even if you plan to use all models you've constructed. + diff --git a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala index fb2dfeb77..f498d3eb4 100644 --- a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala +++ b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuanta.scala @@ -44,6 +44,7 @@ import com.google.protobuf.ByteString import org.apache.wayang.api.python.function._ import org.tensorflow.ndarray.NdArray +import scala.collection.JavaConverters._ import scala.collection.JavaConversions import scala.collection.JavaConversions._ import scala.reflect._ @@ -633,6 +634,38 @@ class DataQuanta[Out: ClassTag](val operator: ElementaryOperator, outputIndex: I joinOperator } + def semanticFilterPrompt(prompt: String): DataQuanta[Out] = { + val dataSetType = org.apache.wayang.core.types.DataSetType.createDefault( + this.output.getType.getDataUnitType.toBasicDataUnitType + ) + + val filterOperator = new SemanticFilterOperator[Out]( + dataSetType, + prompt + ) + + this.connectTo(filterOperator, 0) + wrap[Out](filterOperator) + } + + def semanticFilterPrompt(prompt: String, targetModels: AnyRef * ): DataQuanta[Out] = { + val dataSetType = org.apache.wayang.core.types.DataSetType.createDefault( + this.output.getType.getDataUnitType.toBasicDataUnitType + ) + + val targetModelsSet: java.util.Set[Object] = targetModels.toSet.asJava.asInstanceOf[java.util.Set[Object]] + + val filterOperator = new SemanticFilterOperator[Out]( + dataSetType, + prompt, + targetModelsSet + ) + + this.connectTo(filterOperator, 0) + wrap[Out](filterOperator) + } + + /** * Applies a spatial filter to this instance. * diff --git a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala index 9d37aa930..c399240fb 100644 --- a/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala +++ b/wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/DataQuantaBuilder.scala @@ -45,10 +45,9 @@ import org.apache.iceberg.Schema import org.apache.iceberg.FileFormat import org.apache.iceberg.catalog.{Catalog, TableIdentifier} - - import scala.collection.mutable.ListBuffer import scala.reflect.ClassTag +import scala.annotation.varargs /** * Trait/interface for builders of [[DataQuanta]]. The purpose of the builders is to provide a convenient @@ -282,6 +281,9 @@ trait DataQuantaBuilder[+This <: DataQuantaBuilder[_, Out], Out] extends Logging thatKeyUdf: SerializableFunction[ThatOut, Key]) = new JoinDataQuantaBuilder(this, that, thisKeyUdf, thatKeyUdf) + def semanticFilter(prompt: String) = + new SemanticFilterDataQuantaBuilder[Out](this, prompt) + /** * Feed the built [[DataQuanta]] into a spatial filter operator. * Requires the wayang-spatial plugin to be loaded. @@ -2146,6 +2148,42 @@ class KeyedDataQuantaBuilder[Out, Key](private val dataQuantaBuilder: DataQuanta } +/** + * [[DataQuantaBuilder]] implementation for [[org.apache.wayang.basic.operators.SemanticFilterOperator]]s. + * + * @param inputDataQuanta [[DataQuantaBuilder]] for the input [[DataQuanta]] + * @param udf UDF for the [[SemanticFilterOperator]] + */ +class SemanticFilterDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T], prompt: String) + (implicit javaPlanBuilder: JavaPlanBuilder) + extends BasicDataQuantaBuilder[SemanticFilterDataQuantaBuilder[T], T] { + + // Reuse the input TypeTrap to enforce type equality between input and output. + override def getOutputTypeTrap: TypeTrap = inputDataQuanta.outputTypeTrap + + /** [[LoadProfileEstimator]] to estimate the [[LoadProfile]] of the [[udf]]. */ + private var udfLoadProfileEstimator: LoadProfileEstimator = _ + + /** Selectivity of the filter predicate. */ + private var selectivity: ProbabilisticDoubleInterval = _ + + /* + + */ + private val targetModels: ListBuffer[AnyRef] = ListBuffer() + + @varargs def withTargetModels(models: AnyRef *): SemanticFilterDataQuantaBuilder[T] = { + models.foreach(targetModels.+=_) + this + } + + override protected def build = applyTargetPlatforms( + inputDataQuanta.dataQuanta() + .semanticFilterPrompt(prompt, targetModels: _ *), + this.getTargetPlatforms() + ) + } + class SpatialFilterDataQuantaBuilder[T](inputDataQuanta: DataQuantaBuilder[_, T], keySelector: SerializableFunction[T, _ <: SpatialGeometry], predicateType: SpatialPredicate, diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java new file mode 100644 index 000000000..302da73a1 --- /dev/null +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java @@ -0,0 +1,69 @@ +/* + * 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.basic.operators; + +import java.util.Set; + +import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator; +import org.apache.wayang.core.types.DataSetType; + +public class SemanticFilterOperator extends UnaryToUnaryOperator { + private final String prompt; + + public final Set targetModels; + + public SemanticFilterOperator(final DataSetType type, final String prompt) { + super(type, type, false); + this.prompt = prompt; + this.targetModels = null; + } + + public SemanticFilterOperator(final DataSetType type, final String prompt, final Set targetModels) { + super(type, type, false); + this.prompt = prompt; + this.targetModels = targetModels; + } + + public SemanticFilterOperator(final DataSetType type) { + super(type, type, false); + this.prompt = ""; + this.targetModels = null; + } + + public SemanticFilterOperator(final DataSetType inputType, final DataSetType outputType, + final boolean isSupportingBroadcastInputs) { + super(inputType, outputType, isSupportingBroadcastInputs); + this.prompt = ""; + this.targetModels = null; + } + + public SemanticFilterOperator(final UnaryToUnaryOperator that) { + super(that); + this.prompt = ""; + this.targetModels = null; + } + + public String getPrompt() { + return prompt; + } + + public void addTargetModel(final Object model) { + targetModels.add(model); + } +} diff --git a/wayang-platforms/pom.xml b/wayang-platforms/pom.xml index 9c5e29545..8665ceffb 100644 --- a/wayang-platforms/pom.xml +++ b/wayang-platforms/pom.xml @@ -45,6 +45,7 @@ wayang-generic-jdbc wayang-presto wayang-tensorflow + wayang-semantic diff --git a/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java index 757220e34..a11e7625d 100644 --- a/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java +++ b/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaFilterOperator.java @@ -50,6 +50,16 @@ public class JavaFilterOperator implements JavaExecutionOperator { + + /** + * Creates a new instance. + * + * @param type type of the dataset elements + */ + public JavaFilterOperator(PredicateDescriptor predicateDescriptor) { + super(predicateDescriptor); + } + /** * Creates a new instance. * @@ -119,5 +129,4 @@ public List getSupportedOutputChannels(int index) { assert index <= this.getNumOutputs() || (index == 0 && this.getNumOutputs() == 0); return Collections.singletonList(StreamChannel.DESCRIPTOR); } - } diff --git a/wayang-platforms/wayang-semantic/bin/.project b/wayang-platforms/wayang-semantic/bin/.project new file mode 100644 index 000000000..dc99fdedd --- /dev/null +++ b/wayang-platforms/wayang-semantic/bin/.project @@ -0,0 +1,23 @@ + + + wayang-semantic + + + + + + org.eclipse.jdt.core.javabuilder + + + + + org.eclipse.m2e.core.maven2Builder + + + + + + org.eclipse.jdt.core.javanature + org.eclipse.m2e.core.maven2Nature + + diff --git a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs new file mode 100644 index 000000000..99f26c020 --- /dev/null +++ b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs @@ -0,0 +1,2 @@ +eclipse.preferences.version=1 +encoding/=UTF-8 diff --git a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs new file mode 100644 index 000000000..f897a7f1c --- /dev/null +++ b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs @@ -0,0 +1,4 @@ +activeProfiles= +eclipse.preferences.version=1 +resolveWorkspaceProjects=true +version=1 diff --git a/wayang-platforms/wayang-semantic/bin/pom.xml b/wayang-platforms/wayang-semantic/bin/pom.xml new file mode 100644 index 000000000..721f7ab0b --- /dev/null +++ b/wayang-platforms/wayang-semantic/bin/pom.xml @@ -0,0 +1,63 @@ + + + + 4.0.0 + + + wayang-platforms + org.apache.wayang + 1.1.2-SNAPSHOT + + + wayang-semantic + + Wayang Platform Semantic + + Wayang implementation of semantic operators + + + + org.apache.wayang.platform.semantic + + + + + org.apache.wayang + wayang-api-scala-java + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-basic + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-java + 1.1.2-SNAPSHOT + + + org.junit.jupiter + junit-jupiter + 5.10.2 + test + + + diff --git a/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class b/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class new file mode 100644 index 0000000000000000000000000000000000000000..2f5d301d76f1804651fff792d436a2b732766021 GIT binary patch literal 1711 zcmd5+OK%e~5FV!`n=I*rLU}*j0#e$;t`M9ka6y2I(o(4piKF9n5|>?jvAu;#{4XSs z;LeXijCT`AQACRnh>O?W`R4P?e4hF8^}{CscnG%ws1dkh4UaP>RlLs*SgcJyW;x>n z5x-%FO!Z?cGNzp5@%C^Nz%+sR5}kU$i|)SQjzE2@v-@m)6K$OlbSo|TQmvuwfmG5x zA~4ln*&{IhSSKQY(*)*z@H{a_n+D8~soT5C2&>Zrkx;I)T&B!Pt!Qp^Hx-$^Ps8TU zKu~v>3+f3XA4pUMLOD}PJK7a=L=w?;N2R4wCDLH5=^>hW!U)CjhMv(Go?jPeOK8MX zW-aZ>RM>xcOCs=M9{?8r73&MSj8a6E6j3EbG?HSOCR$kPv&@*|1M09C;%$zpjT4}< zkykbf^+Bk~12_-xcWIfAM$)%&V$b~`aly#=yP8(Wj(_Zm16ecXU9Z*3^WWO zgt-<3Fi&8y-SHyFWkYT|BUOLRH`-hWE4W!?U14?@_6~uCj^-@gV@CS(@aMD}NK4>u z=eLJ}@9n(oa-?pr{M`8sxJcmL3EKIuX-S}w59^JS2_qaY>skY@5>O|IZ`|t-bt4dN z>%wsH6lV;9nNjLxn+8A$w8n1>V41+|2vki}gltG_QRN*RffkkkCo2wh0+-z;95tF_guiWRqjJL3+M3Eghe<%WLPcvM+_Gx nFwl~rR{CASII5eE`dyjexBA_$2G`KL2G`*xe(Siq0ZZQiY6=Lg literal 0 HcmV?d00001 diff --git a/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class b/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class new file mode 100644 index 0000000000000000000000000000000000000000..91f215593034bfebe6c4b7f604c06bf151cc57b3 GIT binary patch literal 1072 zcmcgrO>Yx15FIC>n=FBpKnvVVPw8d%!l?x=RS~HYI5bq^=y)fI+g*FHy@mWTB#_|F zk3x*Ii5j6EI3Nx?-m%|&zL}rDzJ3RQm+&-#27?o4Wm*WO(@SwJQe)?-D1}^L`bjK> zo~I5Ap}h*AvQ&ER(lY|SR=GzzuCF3!G1y(%=fULN^b)0Ku=7G`zZm_4;8rz1uta0>STXZJB#*9l-lqwfq8O=*;ra2bwh{v7D0=Zw7$b*9<1WW2* zTnep?=TqbxCkY>Vt{hi7QtfnE3p$+%Ca0l))*c*&)rt92W8rkD8eZ3LId0Q1z zv=c!C#h^bjQsfuHs-UkgTYjM&gVWJJ*o$KZ@wut2L|PXLxPSKN=B&7(oQFCTt(-^g z-zc=g%EdB*#|)x%0~mD2$~mfLLi6w78PHnM9s`|>D&HX)k@lQ!8J+PrXn*L(U!ePg mv7d*RfCN>v3?r41ZfDsgmnoy5g!FSfUkUxow{ z-1$4)5#!xN2~rRUAufK+=WpKg_4~)q0PqwZ2e3xq(3&*PnNaa1yJE365(somppX@t{o={fmU*&rNjrJRgXaPOpFzC zrQrm2mPGwzM%Oz+)Ouei;f|1_dvHOZaiTLGz$StALd`Q{wAp|*S^M@%8E&<_;u%eK zUI@v&Hd+`xlsvbGG;E$vICay4Q%?|iJEPi8nlh!dqeD*TBoRGzR9GrhCJb6Grzje6 z!&QnJMn>m&e_Ns~qfsiEwR9vTxBpX1#!<010F(5FjXB-Nltp#QqB>>KJY~^5W&1SK z+*0pu&Rja74x_@~7wF?M6P5yg5h|);^(VZw=OQn(adm`CC*+FI{sctke*meB{H40n zypSRlj@q)|2FpWPXUf!RL+ku^4}7XimGO~KKD{WxGb^QAMn`l|TIVH*qD=X))x=Ab zVRKRID?Fo4Q&v@B*ql}SZ&1g^u&G@E3PWhYtrj$)Ltwj`EZXAC8SKuUZ<}_4w!SFy zAvfm?JApta(J7M`%m{y8+n6OIoEalL^rPY^|L4?jWcb?4>> k*!oi8ZsY2Id4@Zx9 z(DP9YTAxE)P*BilPtv<5=}A9-ef5wIFwQ> z$$PaVJY<1nE0Amjl2-!BKLQT`%l3dSjQzX`!PTGrVc4Ir%fiI}$LSIjzcb~&zoHenKe SpY5-mmiZcCmI=dtt1?)UC*2EVAZcPp@1ax1z zf_g9zFuK9M^4|GA+!5mQTk9#f@`?)7uBuI`qHz{$?`9=c;VEW=^Ce+i)r5&ared-! zhFV$cBF+e}NhBUeGy#n*jOTvJ6{qIp$!e}J_pW09L&LcOr7l$nI5#DQwn{-9N-5Uk zy;>0NvpTX>N4Dz7Yjxxwb@s7v6fkvCsi>D2xncTLbNOv5mR%;<<$8>NBF+ZKRb3iw zB8G-_XNhcxzR|oCY&Do1q;VTt(A?NGB4_wpHD%mJOFU^pOc_R@vZ=q8c#AU2pbvM4 za2xIk*qcmm7IqfBv5UjB(9x6`-Za&W{JG*C2^deER^^5ACi$CSH!e*O@Ot`xuI*4j zcIFzd>4?`Q;L%?$axE-820lGLci>m>C_8L>>^4Gt+O2T*L#}{!#1;-EUi&3hh+f&Xb}Kv7OO;yPO6z&uFRKnt18^dyi!7 z7@fJRv~uf=PPJAaGMeAWx-y_9qs0T+d&Xp@L1)?NzrWQ++AJB$E|0S`SBY>c(>yoX zULsTbDG!^w1IgWfF1ZJY{9#db7>|Y4nd5tsk3k~7>A137>8>)+C)vkRPa3IXoX|5_ ziun%(*19~36JagysYKdmc}iE}V6On!{$BJYUjg?~#eGz9AC0+>D(+W!Hp7R;x76jN=fmA@qev$)(C>aZ`#FP9;w#xS8gOik0KG z$fZF}XzNT7yUg$^=v%8kz6;%zJ*9n2V+G%$NZh1!#J39T{0gMF&^}r!;0egES>?+>Lhsw-!RZDdJ;fb{7*otTq&TO!Yuu8X!BDf>jQp)g9IN}Rlc+o(d| zHGSgIz8r^K9Lulbg+*g;5sQ>1&+?0Ww(TJB`ANFI*-Ir=&82Ffz}yz_K@ zYK_s^zI3CQ1K|1AYUjXnSZ?m9z80=9(B#UY#k-x+wzYN0U3*ZV9j0TtDcE)V3HRvj%KabJb#ygH3XEOv(IweT2UZWB402zFQ7|p@ly%o1;a%y$s=5!h4~Tq6@TK{<}z*u=6s$ z=jbJR8Kake1>-d8l^WEWwW!x7P+y&a`dSU@tyPc{vjiq@(YqAUZTi6b=|3u$52^qF literal 0 HcmV?d00001 diff --git a/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class b/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class new file mode 100644 index 0000000000000000000000000000000000000000..6a067c2d324db6f5c4f3b18b68e816b2a43eb69e GIT binary patch literal 4429 zcmeHLOK;RL5T2C6J}C6T`*E=L)S@bT;Y2A2Dv&@6RBQpk%}r)^4UQdbXX*YjB#_|F zFW}c8#tsWx5oi(RC3@H-j%UBI=k<;A?!&7$0I&+@YcLhTsx?X5p>^%2e8}6{^xHh= z$pG6g_=u~1+hWF*6G?krTj9hIyGX-~E)UXP4W=WQ3;&c}_}=DIOk4zqE(;~xl?VqfH$SM@Qd3PzJ4*Lrhtsvs_5-gjOs!dQ)O%FR^B0V}Q(! za%3JL_R8W?8B4fQ+ObV!W008La!gnzR4NRuM~-N!hXz$b8hS=&G=5&tT*~4^a%)*n zNVI=Y%UW2)UT-Bu+MVAedGs0YBWsa=EAdfWB0Vl;<2M!EY)b4w`s;*GGLBO zk6SqzauSK+CiZa?2Zwx(`x&mZs>pI}93#03hwY2}mv#w}Q#c`a3Wr9b2ZCF{r7MFajDpZO+p?rO!8G7c0bd%CCyHi-_XGn^1o?|j}1D!Idd3qm*0$*72PoNB61_J|iF}qGwG6<#)#9}>)pU;G~ zg7gHrj<9Z54k5rmvDai@8{r1o4qN$K<=17=7d!c3y!eVeHXGx~n?kxB1Eo7sj&Cva zeFP)_JN*n5YsIZ)$`Xl=+%=%g3a@_xdYSqNi^N^6m6rFj?@$7I*Kvow5!z3wZS9vg z(zw?Al9cj^RS`audaoJ7N47$!Y&9lICM(l?vFi+F?lLRyOJ&Dco!F(wJ`&k$S9bc% z6j-{a!uOK_vUfs1y1P$AKBjLvoCr0f(nXxEi9@!v2voL0SpK%n{Ilxr4T%Rtihq@* znox%$GjI@&MsQ-WK>0)4aqn4;f7I7H(bJzb=?5#5`wq5gQG`4VQ{2ANs9 vdKsYK99?HhDVT=^8a3b;9H%!P;lehA6BP(5Ab7n~a4K9+!#TQ76T+E~NJH#~ literal 0 HcmV?d00001 diff --git a/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class b/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class new file mode 100644 index 0000000000000000000000000000000000000000..9e48cff47f635c797dc6b5e0b5d4013788cd087a GIT binary patch literal 4940 zcmeHLTW=gS6h2a7JJK% z`VZKY^|cxIS;?{q?>}NQrpA5C3#OdN`uF%$@JDS}B5*a(O22$(?*Y#oftJV#tPTzr zhRz5z-XxHi@Zm1sN3x3_3nkpA1m0g;q+NWoexJb7w$6DQmI#0^&&?l3aCz~^r(T+3;iNoFk_3CZn$xXX6n;%h zInwAHX>^XXL}ib(Dbl%@^k}ZRrCz56bJ>JCR6XA>QIT~dbh^|#T^bCsi2E2WYE=}a zHjW})gTc;8{+D_Qo>Ms?k8+9Uf$;AvxKor;WWrHfmE2%{Q`VU(bK20Dz}D7%uBued zM?(4hge!Q@D(ULfDZO1;=U0%5GUY=u0@olhkT4`0)JdJDEF{pR`*o-a+gdry=ch>f z1!m?3xTNbB?qbTmRSB7U9V}$*I(sLh-quoLX7c6YpJ4}DHKm^V&c4PJ+)Z>&rDl1o zmKN<3N`_B^p#lwgaSKbyB&a$R7VA;`dLgXEk{Rg+!n$J_1OY>R-jZW&gqsvch;mlx z@5*W{jEOVGMp*L_9^#$1V#Hh{g5l#OM4p_C6YTd*90Rgy#5!^ z8`wu!j%rfmaF0V+`P@<{1Po zBSSIk#uJ5((-ED(&3hG|R1|ziO@$RG{+1u^!*KvgpA6<^-t53z1b#XO-}5idlw+VM zLtUASW21u>zlMF!Z&ok=2rIuGd<95N@Sebn_`e+P6SxjH!n>E?Ap50gTbt43Ra`P>xp3QPfA@u9LX++QIgw zm%ob#B#_{pN1pg=5M%F!F3Pr-!qHyUJnVouql_|%%X zX8~bw?+>1`Pshk7d6!Mw>dEBNzVoRvB)!obZIkIvoqioY0DnjXsb( zwV%^wV}Hb{n~pj40FfUo$_Jw{Q%XB}!08MmqPva?ONB~=!THE3b`7|}S>p%;qf=~u zpJ87@qgXO)=|D(sU*IS^VHPL7l!W8z{3_9s5805@7TPt+?Hc8FjmD_#fi`J0GjNM0 znp^61nlcxUs6*BB!!atdh=ltk^7|zU2ARkG8!p$XNXObZigG0eTZ+7rdI>uxb3z`s zC0+)>Cs}Ye9ZL}lM{PFd2Kh}{XR_F7LvsRaZ+NcCRKf>Bd49qkd}o<-Md^s{W!Cu~ zl%h;|m&}0+kmyM0k|pY-PE!^LwAuJB5c-Z*jvt<+?AOT5B||qGzT*y3_N`3F#Oq*= zv5V7tHt8KLB{Gu_7f*-nX;nr&#gToCNx2*8gi6hlyL=+}Ba8M4A;Z&Ph(N^vmNI%6%R|(f&?9LeC1H{{!?U<`I@->}qve-p)Ql3Fv*t98bAI zBQIU5kbZe3&A8@&aVhUuCEKSs?zS;(YD7WO`1P9p61xO$Px+l zwUC~ka|q1MQ)2t#201yQXICuV7W2{$cYA_y zKqUNBX{rfzIKK*K;R1mx?Ve9#JE2a_)!g;9@mde_xo@)cfSY}WeJvfTZ|q&^`G2}~ zc!Q817OByfx##0Uh$7j%pxl$d<$0rjT!**Gne1fDIm_3Und?jw%goi+K6%Yo!o{z4 zF-YKCp|B)SZ*TN`GK=)w=!>CZE;ASax<{XHE>8A&65HKEvZjRmzJPZxlj!-1JrB#L zGO;Jn?CZ?L90>-u$fa2%_XHn~U~BLp9<+O_5?I~K&ms2F6)jM}b9g-J2dLt87FMub z!?#cHE5T>;XQ + + + 4.0.0 + + + wayang-platforms + org.apache.wayang + 1.1.2-SNAPSHOT + + + wayang-semantic + + Wayang Platform Semantic + + Wayang implementation of semantic operators + + + + org.apache.wayang.platform.semantic + + + + + org.apache.wayang + wayang-java + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-basic + 1.1.2-SNAPSHOT + + + org.apache.wayang + wayang-java + 1.1.2-SNAPSHOT + + + org.junit.jupiter + junit-jupiter + 5.10.2 + test + + + com.fasterxml.jackson.core + jackson-core + 2.16.1 + test + + + org.apache.wayang + wayang-api-scala-java + 1.1.2-SNAPSHOT + test + + + diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java new file mode 100644 index 000000000..83f497008 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/Semantic.java @@ -0,0 +1,45 @@ +/* + * 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.semantic; + +import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.apache.wayang.java.platform.JavaPlatform; + +public class Semantic { + private final static SemanticPlugin PLUGIN = new SemanticPlugin(); + + /** + * Retrieve the {@link SemanticPlugin}. + * + * @return the {@link SemanticPlugin} + */ + public static SemanticPlugin plugin() { + return PLUGIN; + } + + + /** + * Retrieve the {@link SemanticPlatform}. + * + * @return the {@link SemanticPlatform} + */ + public static JavaPlatform platform() { + return JavaPlatform.getInstance(); + } +} \ No newline at end of file diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java new file mode 100644 index 000000000..d449fab4b --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java @@ -0,0 +1,64 @@ +/* + * 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.semantic.mappings; + +import java.util.Collection; +import java.util.Collections; + +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.function.PredicateDescriptor; +import org.apache.wayang.core.function.FunctionDescriptor.SerializablePredicate; +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.types.DataSetType; +import org.apache.wayang.java.operators.JavaFilterOperator; +import org.apache.wayang.java.platform.JavaPlatform; +import org.apache.wayang.semantic.udf.SemanticAlgorithm; + +public class JavaFilterMapping implements Mapping { + private final SemanticAlgorithm model; + + public JavaFilterMapping(final SemanticAlgorithm model) { + this.model = model; + } + + @Override + public Collection getTransformations() { + return Collections.singleton(new PlanTransformation(this.createSubplanPattern(), + this.createReplacementSubplanFactory(), JavaPlatform.getInstance())); + } + + private SubplanPattern createSubplanPattern() { + return SubplanPattern.createSingleton(new OperatorPattern>("semantic_filter", + new SemanticFilterOperator<>(DataSetType.NONE), false).withAdditionalTest(op -> op.targetModels != null) + .withAdditionalTest(op -> op.targetModels.contains(model))); + } + + private ReplacementSubplanFactory createReplacementSubplanFactory() { + return new ReplacementSubplanFactory.OfSingleOperators>((matchedOperator, epoch) -> { + final SerializablePredicate predicate = input -> model.impl.apply(input, matchedOperator.getPrompt()); + final PredicateDescriptor predicateDescriptor = new PredicateDescriptor<>(predicate, + matchedOperator.getOutput().getType().getDataUnitType().getTypeClass(), model.loadProfileEstimator); + return new JavaFilterOperator<>(predicateDescriptor).at(epoch); + }); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java new file mode 100644 index 000000000..cc4b494e9 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java @@ -0,0 +1,29 @@ +/* + * 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.semantic.mappings; + +import java.util.Collection; +import org.apache.wayang.core.mapping.Mapping; +import java.util.Arrays; + +public class Mappings { + public static final Collection ALL = Arrays.asList( + new JavaFilterMapping(null) + ); +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java new file mode 100644 index 000000000..0b8ff993a --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java @@ -0,0 +1,24 @@ +/* + * 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.semantic.operators; + +public class SemanticMapOperator { + + +} \ No newline at end of file diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java new file mode 100644 index 000000000..b5d0dc263 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java @@ -0,0 +1,77 @@ +/* + * 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.semantic.plugin; + +import org.apache.wayang.core.plugin.Plugin; +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.mapping.Mapping; +import org.apache.wayang.core.optimizer.channels.ChannelConversion; +import org.apache.wayang.core.platform.Platform; +import org.apache.wayang.semantic.mappings.JavaFilterMapping; +import org.apache.wayang.semantic.udf.SemanticAlgorithm; +import org.apache.wayang.java.Java; +import org.apache.wayang.java.platform.JavaPlatform; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +public class SemanticPlugin implements Plugin { + private final List mappings; + + private SemanticPlugin(final List mappings) { + this.mappings = mappings; + } + + public SemanticPlugin() { + this.mappings = List.of(); + } + + @Override + public Collection getMappings() { + return mappings; + } + + @Override + public Collection getRequiredPlatforms() { + // TODO: maybe we should find another way to handle this? but do Java for now. + return Collections.singleton(JavaPlatform.getInstance()); + } + + @Override + public Collection getChannelConversions() { + return Java.basicPlugin().getChannelConversions(); + } + + @Override + public void setProperties(final Configuration configuration) { + } + + public SemanticPlugin withOperatorMapping(final Class operatorClass, final SemanticAlgorithm model) { + final List nextMappings = new ArrayList<>(this.mappings); + + if (operatorClass.equals(SemanticFilterOperator.class)) { + nextMappings.add(new JavaFilterMapping((SemanticAlgorithm) model)); + } + + return new SemanticPlugin(nextMappings); + } +} \ No newline at end of file diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java new file mode 100644 index 000000000..00549fb86 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java @@ -0,0 +1,36 @@ +/* + * 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.semantic.udf; + +import java.util.function.BiFunction; + +import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; + +/** + * A udf that represents an algorithm you use in semantic queries + */ +public class SemanticAlgorithm { + /** + * An implementation that maps an input T with a prompt String to an output T + * defined via the implementation in the lambda expression. + */ + public BiFunction impl; + + public LoadProfileEstimator loadProfileEstimator; +} diff --git a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java new file mode 100644 index 000000000..929bd60a1 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java @@ -0,0 +1,273 @@ +/* + * 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.semantic; + +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.api.WayangContext; +import org.apache.wayang.core.function.ExecutionContext; +import org.apache.wayang.core.function.FunctionDescriptor; +import org.apache.wayang.core.function.TransformationDescriptor; +import org.apache.wayang.core.function.FunctionDescriptor.SerializablePredicate; +import org.apache.wayang.core.optimizer.costs.EstimationContext; +import org.apache.wayang.core.optimizer.costs.LoadProfile; +import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; +import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators; +import org.apache.wayang.core.plan.wayangplan.WayangPlan; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.core.util.WayangArrays; +import org.apache.wayang.core.util.WayangCollections; +import org.apache.wayang.java.Java; +import org.apache.wayang.java.operators.JavaCollectionSource; +import org.apache.wayang.java.operators.JavaDoWhileOperator; +import org.apache.wayang.java.operators.JavaLocalCallbackSink; +import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.apache.wayang.semantic.Semantic; +import org.apache.wayang.semantic.udf.SemanticAlgorithm; +import org.apache.wayang.semantic.operators.*; +import org.apache.wayang.api.JavaPlanBuilder; +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.java.operators.JavaMapOperator; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedList; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class SemBenchTest { + private static List loadReviews() { + return Arrays.asList(new Review("taken_1", "The movie was fantastic. Great acting and an engaging story."), + new Review("taken_2", "I was disappointed. The plot was boring and too long."), + new Review("taken_3", "Absolutely loved it! One of the best movies I have seen this year."), + new Review("taken_3", "Terrible experience. I would not recommend it to anyone."), + new Review("taken_4", "It was okay. Not great, not terrible.")); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + void testSemBenchMoviesWithOllama() { + final Configuration configuration = new Configuration(); + configuration.setProperty("wayang.java.filter.load", """ + { + "in":1, + "out":1, + "cpu":"${25*in0 + 350000}", + "ram":"100000", + "p":0.9 + } + """ + ); + configuration.setProperty("wayang.semantic.ollama.model1.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${500*in0 + 56789}", + "ram": "10000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """); + configuration.setProperty("wayang.semantic.ollama.model2.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${500*in0 + 56789}", + "ram": "10000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """); + configuration.setProperty("wayang.semantic.ollama.model3.load", + """ + { + "in": 1, + "out": 1, + "cpu": "${50*in0 + 5678}", + "ram": "1000", + "disk": "0", + "net": "0", + "p": 0.9, + "overhead": 0, + "ru": "${wayang:logGrowth(0.1, 0.1, 1000000, in0)}" + } + """ + ); + + final SemanticAlgorithm ollamaFilter = new SemanticAlgorithm(); + ollamaFilter.impl = (input, prompt) -> { + try { + return OllamaSemanticFilter.isPositiveSentiment((Review) input); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }; + ollamaFilter.loadProfileEstimator = + LoadProfileEstimators.createFromSpecification( + "wayang.semantic.ollama.model1.load", + configuration + ); + + final SemanticAlgorithm ollamaFilter2 = new SemanticAlgorithm(); + ollamaFilter2.impl = (input, prompt) -> { + try { + return OllamaSemanticFilter.isPositiveSentiment2((Review) input); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }; + ollamaFilter2.loadProfileEstimator = + LoadProfileEstimators.createFromSpecification( + "wayang.semantic.ollama.model2.load", + configuration + ); + + final SemanticAlgorithm ollamaFilter3 = new SemanticAlgorithm(); + ollamaFilter3.impl = (input, prompt) -> { + try { + return OllamaSemanticFilter.isPositiveSentiment3((Review) input, (String) prompt); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }; + ollamaFilter3.loadProfileEstimator = + LoadProfileEstimators.createFromSpecification( + "wayang.semantic.ollama.model3.load", + configuration + ); + + final SemanticPlugin plugin = Semantic.plugin() + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter2) + .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter3); + + final WayangContext wayangContext = new WayangContext() + .withPlugin(Java.basicPlugin()) + .withPlugin(plugin); + final JavaPlanBuilder planBuilder = new JavaPlanBuilder(wayangContext); + + final Collection positiveReviewCnt = planBuilder.loadCollection(loadReviews()) + .filter(review -> "taken_3".equals(review.getId())) + .semanticFilter("Analyze the review after the | and write either \"POSITIVE\" if the review has a positive sentiment and \"NEGATIVE\" if the review has a negative sentiment.") + .withTargetModels(ollamaFilter, ollamaFilter2, ollamaFilter3) + .count() + .collect(); + } +} + +class Review { + private final String id; + private final String reviewText; + + public Review(final String id, final String reviewText) { + this.id = id; + this.reviewText = reviewText; + } + + public String getId() { + return id; + } + + public String getReviewText() { + return reviewText; + } + + @Override + public String toString() { + return "Review{id='" + id + "', reviewText='" + reviewText + "'}"; + } +} + +final class OllamaSemanticFilter { + private static final String OLLAMA_API_URL = "http://apache-wayang-ollama:11434/api/generate"; + private static final String MODEL_NAME = "tinyllama"; + private static final HttpClient httpClient = HttpClient.newHttpClient(); + + public static boolean isPositiveSentiment(final Review review) throws IOException, InterruptedException { + final String prompt = String.format("Analyze the sentiment of this movie review. " + + "Reply with only 'POSITIVE' or 'NEGATIVE'.\n\n" + "Review: %s", review.getReviewText()); + final String response = callOllama(prompt); + return response.contains("POSITIVE"); + } + + public static boolean isPositiveSentiment2(final Review review) throws IOException, InterruptedException { + final String prompt = String.format( + "Analyze the sentiment of this movie review, words like love, fantastic and great are positive modifiers. " + + "Reply with only 'POSITIVE' or 'NEGATIVE'.\n\n" + "Review: %s", + review.getReviewText()); + final String response = callOllama(prompt); + return response.contains("POSITIVE"); + } + + public static boolean isPositiveSentiment3(final Review review, final String prompt) + throws IOException, InterruptedException { + final String response = callOllama(prompt + " | " + review.getReviewText()); + return response.contains("POSITIVE"); + } + + private static String callOllama(final String prompt) throws IOException, InterruptedException { + final String requestBody = String.format("{\"model\": \"%s\", \"prompt\": \"%s\", \"stream\": false}", + MODEL_NAME, escapeJson(prompt)); + + final HttpRequest request = HttpRequest.newBuilder().uri(URI.create(OLLAMA_API_URL)) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(requestBody)) + .timeout(Duration.ofMinutes(2)).build(); + + final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() != 200) { + throw new IOException("Ollama API error: " + response.body()); + } + + return parseOllamaResponse(response.body()); + } + + private static String parseOllamaResponse(final String jsonResponse) { + int startIdx = jsonResponse.indexOf("\"response\":\""); + + if (startIdx == -1) + return ""; + + startIdx += "\"response\":\"".length(); + + final int endIdx = jsonResponse.indexOf("\"", startIdx); + + return jsonResponse.substring(startIdx, endIdx); + } + + private static String escapeJson(final String str) { + return str.replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r"); + } +} \ No newline at end of file From d5ca01d6e405bb4776ddd2eef2111d7ca3b239a7 Mon Sep 17 00:00:00 2001 From: mspruc Date: Fri, 28 Aug 2026 09:37:13 +0200 Subject: [PATCH 02/10] delete bin --- wayang-platforms/wayang-semantic/bin/.project | 23 ------- .../org.eclipse.core.resources.prefs | 2 - .../bin/.settings/org.eclipse.m2e.core.prefs | 4 -- wayang-platforms/wayang-semantic/bin/pom.xml | 63 ------------------ .../org/apache/wayang/semantic/Semantic.class | Bin 1711 -> 0 bytes .../semantic/mappings/JavaFilterMapping.class | Bin 1072 -> 0 bytes .../wayang/semantic/mappings/Mappings.class | Bin 1291 -> 0 bytes .../operators/SemanticFilterOperator.class | Bin 846 -> 0 bytes .../operators/SemanticMapOperator.class | Bin 837 -> 0 bytes .../semantic/plugin/SemanticPlugin.class | Bin 3387 -> 0 bytes .../semantic/PositiveSentimentUdf.class | Bin 4429 -> 0 bytes .../org/apache/wayang/semantic/Review.class | Bin 4940 -> 0 bytes .../apache/wayang/semantic/SemBenchTest.class | Bin 6439 -> 0 bytes 13 files changed, 92 deletions(-) delete mode 100644 wayang-platforms/wayang-semantic/bin/.project delete mode 100644 wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs delete mode 100644 wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs delete mode 100644 wayang-platforms/wayang-semantic/bin/pom.xml delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/Mappings.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/operators/SemanticFilterOperator.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class delete mode 100644 wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/SemBenchTest.class diff --git a/wayang-platforms/wayang-semantic/bin/.project b/wayang-platforms/wayang-semantic/bin/.project deleted file mode 100644 index dc99fdedd..000000000 --- a/wayang-platforms/wayang-semantic/bin/.project +++ /dev/null @@ -1,23 +0,0 @@ - - - wayang-semantic - - - - - - org.eclipse.jdt.core.javabuilder - - - - - org.eclipse.m2e.core.maven2Builder - - - - - - org.eclipse.jdt.core.javanature - org.eclipse.m2e.core.maven2Nature - - diff --git a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs deleted file mode 100644 index 99f26c020..000000000 --- a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.core.resources.prefs +++ /dev/null @@ -1,2 +0,0 @@ -eclipse.preferences.version=1 -encoding/=UTF-8 diff --git a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs b/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs deleted file mode 100644 index f897a7f1c..000000000 --- a/wayang-platforms/wayang-semantic/bin/.settings/org.eclipse.m2e.core.prefs +++ /dev/null @@ -1,4 +0,0 @@ -activeProfiles= -eclipse.preferences.version=1 -resolveWorkspaceProjects=true -version=1 diff --git a/wayang-platforms/wayang-semantic/bin/pom.xml b/wayang-platforms/wayang-semantic/bin/pom.xml deleted file mode 100644 index 721f7ab0b..000000000 --- a/wayang-platforms/wayang-semantic/bin/pom.xml +++ /dev/null @@ -1,63 +0,0 @@ - - - - 4.0.0 - - - wayang-platforms - org.apache.wayang - 1.1.2-SNAPSHOT - - - wayang-semantic - - Wayang Platform Semantic - - Wayang implementation of semantic operators - - - - org.apache.wayang.platform.semantic - - - - - org.apache.wayang - wayang-api-scala-java - 1.1.2-SNAPSHOT - - - org.apache.wayang - wayang-basic - 1.1.2-SNAPSHOT - - - org.apache.wayang - wayang-java - 1.1.2-SNAPSHOT - - - org.junit.jupiter - junit-jupiter - 5.10.2 - test - - - diff --git a/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class b/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/Semantic.class deleted file mode 100644 index 2f5d301d76f1804651fff792d436a2b732766021..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 1711 zcmd5+OK%e~5FV!`n=I*rLU}*j0#e$;t`M9ka6y2I(o(4piKF9n5|>?jvAu;#{4XSs z;LeXijCT`AQACRnh>O?W`R4P?e4hF8^}{CscnG%ws1dkh4UaP>RlLs*SgcJyW;x>n z5x-%FO!Z?cGNzp5@%C^Nz%+sR5}kU$i|)SQjzE2@v-@m)6K$OlbSo|TQmvuwfmG5x zA~4ln*&{IhSSKQY(*)*z@H{a_n+D8~soT5C2&>Zrkx;I)T&B!Pt!Qp^Hx-$^Ps8TU zKu~v>3+f3XA4pUMLOD}PJK7a=L=w?;N2R4wCDLH5=^>hW!U)CjhMv(Go?jPeOK8MX zW-aZ>RM>xcOCs=M9{?8r73&MSj8a6E6j3EbG?HSOCR$kPv&@*|1M09C;%$zpjT4}< zkykbf^+Bk~12_-xcWIfAM$)%&V$b~`aly#=yP8(Wj(_Zm16ecXU9Z*3^WWO zgt-<3Fi&8y-SHyFWkYT|BUOLRH`-hWE4W!?U14?@_6~uCj^-@gV@CS(@aMD}NK4>u z=eLJ}@9n(oa-?pr{M`8sxJcmL3EKIuX-S}w59^JS2_qaY>skY@5>O|IZ`|t-bt4dN z>%wsH6lV;9nNjLxn+8A$w8n1>V41+|2vki}gltG_QRN*RffkkkCo2wh0+-z;95tF_guiWRqjJL3+M3Eghe<%WLPcvM+_Gx nFwl~rR{CASII5eE`dyjexBA_$2G`KL2G`*xe(Siq0ZZQiY6=Lg diff --git a/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class b/wayang-platforms/wayang-semantic/bin/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.class deleted file mode 100644 index 91f215593034bfebe6c4b7f604c06bf151cc57b3..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 1072 zcmcgrO>Yx15FIC>n=FBpKnvVVPw8d%!l?x=RS~HYI5bq^=y)fI+g*FHy@mWTB#_|F zk3x*Ii5j6EI3Nx?-m%|&zL}rDzJ3RQm+&-#27?o4Wm*WO(@SwJQe)?-D1}^L`bjK> zo~I5Ap}h*AvQ&ER(lY|SR=GzzuCF3!G1y(%=fULN^b)0Ku=7G`zZm_4;8rz1uta0>STXZJB#*9l-lqwfq8O=*;ra2bwh{v7D0=Zw7$b*9<1WW2* zTnep?=TqbxCkY>Vt{hi7QtfnE3p$+%Ca0l))*c*&)rt92W8rkD8eZ3LId0Q1z zv=c!C#h^bjQsfuHs-UkgTYjM&gVWJJ*o$KZ@wut2L|PXLxPSKN=B&7(oQFCTt(-^g z-zc=g%EdB*#|)x%0~mD2$~mfLLi6w78PHnM9s`|>D&HX)k@lQ!8J+PrXn*L(U!ePg mv7d*RfCN>v3?r41ZfDsgmnoy5g!FSfUkUxow{ z-1$4)5#!xN2~rRUAufK+=WpKg_4~)q0PqwZ2e3xq(3&*PnNaa1yJE365(somppX@t{o={fmU*&rNjrJRgXaPOpFzC zrQrm2mPGwzM%Oz+)Ouei;f|1_dvHOZaiTLGz$StALd`Q{wAp|*S^M@%8E&<_;u%eK zUI@v&Hd+`xlsvbGG;E$vICay4Q%?|iJEPi8nlh!dqeD*TBoRGzR9GrhCJb6Grzje6 z!&QnJMn>m&e_Ns~qfsiEwR9vTxBpX1#!<010F(5FjXB-Nltp#QqB>>KJY~^5W&1SK z+*0pu&Rja74x_@~7wF?M6P5yg5h|);^(VZw=OQn(adm`CC*+FI{sctke*meB{H40n zypSRlj@q)|2FpWPXUf!RL+ku^4}7XimGO~KKD{WxGb^QAMn`l|TIVH*qD=X))x=Ab zVRKRID?Fo4Q&v@B*ql}SZ&1g^u&G@E3PWhYtrj$)Ltwj`EZXAC8SKuUZ<}_4w!SFy zAvfm?JApta(J7M`%m{y8+n6OIoEalL^rPY^|L4?jWcb?4>> k*!oi8ZsY2Id4@Zx9 z(DP9YTAxE)P*BilPtv<5=}A9-ef5wIFwQ> z$$PaVJY<1nE0Amjl2-!BKLQT`%l3dSjQzX`!PTGrVc4Ir%fiI}$LSIjzcb~&zoHenKe SpY5-mmiZcCmI=dtt1?)UC*2EVAZcPp@1ax1z zf_g9zFuK9M^4|GA+!5mQTk9#f@`?)7uBuI`qHz{$?`9=c;VEW=^Ce+i)r5&ared-! zhFV$cBF+e}NhBUeGy#n*jOTvJ6{qIp$!e}J_pW09L&LcOr7l$nI5#DQwn{-9N-5Uk zy;>0NvpTX>N4Dz7Yjxxwb@s7v6fkvCsi>D2xncTLbNOv5mR%;<<$8>NBF+ZKRb3iw zB8G-_XNhcxzR|oCY&Do1q;VTt(A?NGB4_wpHD%mJOFU^pOc_R@vZ=q8c#AU2pbvM4 za2xIk*qcmm7IqfBv5UjB(9x6`-Za&W{JG*C2^deER^^5ACi$CSH!e*O@Ot`xuI*4j zcIFzd>4?`Q;L%?$axE-820lGLci>m>C_8L>>^4Gt+O2T*L#}{!#1;-EUi&3hh+f&Xb}Kv7OO;yPO6z&uFRKnt18^dyi!7 z7@fJRv~uf=PPJAaGMeAWx-y_9qs0T+d&Xp@L1)?NzrWQ++AJB$E|0S`SBY>c(>yoX zULsTbDG!^w1IgWfF1ZJY{9#db7>|Y4nd5tsk3k~7>A137>8>)+C)vkRPa3IXoX|5_ ziun%(*19~36JagysYKdmc}iE}V6On!{$BJYUjg?~#eGz9AC0+>D(+W!Hp7R;x76jN=fmA@qev$)(C>aZ`#FP9;w#xS8gOik0KG z$fZF}XzNT7yUg$^=v%8kz6;%zJ*9n2V+G%$NZh1!#J39T{0gMF&^}r!;0egES>?+>Lhsw-!RZDdJ;fb{7*otTq&TO!Yuu8X!BDf>jQp)g9IN}Rlc+o(d| zHGSgIz8r^K9Lulbg+*g;5sQ>1&+?0Ww(TJB`ANFI*-Ir=&82Ffz}yz_K@ zYK_s^zI3CQ1K|1AYUjXnSZ?m9z80=9(B#UY#k-x+wzYN0U3*ZV9j0TtDcE)V3HRvj%KabJb#ygH3XEOv(IweT2UZWB402zFQ7|p@ly%o1;a%y$s=5!h4~Tq6@TK{<}z*u=6s$ z=jbJR8Kake1>-d8l^WEWwW!x7P+y&a`dSU@tyPc{vjiq@(YqAUZTi6b=|3u$52^qF diff --git a/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class b/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/PositiveSentimentUdf.class deleted file mode 100644 index 6a067c2d324db6f5c4f3b18b68e816b2a43eb69e..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 4429 zcmeHLOK;RL5T2C6J}C6T`*E=L)S@bT;Y2A2Dv&@6RBQpk%}r)^4UQdbXX*YjB#_|F zFW}c8#tsWx5oi(RC3@H-j%UBI=k<;A?!&7$0I&+@YcLhTsx?X5p>^%2e8}6{^xHh= z$pG6g_=u~1+hWF*6G?krTj9hIyGX-~E)UXP4W=WQ3;&c}_}=DIOk4zqE(;~xl?VqfH$SM@Qd3PzJ4*Lrhtsvs_5-gjOs!dQ)O%FR^B0V}Q(! za%3JL_R8W?8B4fQ+ObV!W008La!gnzR4NRuM~-N!hXz$b8hS=&G=5&tT*~4^a%)*n zNVI=Y%UW2)UT-Bu+MVAedGs0YBWsa=EAdfWB0Vl;<2M!EY)b4w`s;*GGLBO zk6SqzauSK+CiZa?2Zwx(`x&mZs>pI}93#03hwY2}mv#w}Q#c`a3Wr9b2ZCF{r7MFajDpZO+p?rO!8G7c0bd%CCyHi-_XGn^1o?|j}1D!Idd3qm*0$*72PoNB61_J|iF}qGwG6<#)#9}>)pU;G~ zg7gHrj<9Z54k5rmvDai@8{r1o4qN$K<=17=7d!c3y!eVeHXGx~n?kxB1Eo7sj&Cva zeFP)_JN*n5YsIZ)$`Xl=+%=%g3a@_xdYSqNi^N^6m6rFj?@$7I*Kvow5!z3wZS9vg z(zw?Al9cj^RS`audaoJ7N47$!Y&9lICM(l?vFi+F?lLRyOJ&Dco!F(wJ`&k$S9bc% z6j-{a!uOK_vUfs1y1P$AKBjLvoCr0f(nXxEi9@!v2voL0SpK%n{Ilxr4T%Rtihq@* znox%$GjI@&MsQ-WK>0)4aqn4;f7I7H(bJzb=?5#5`wq5gQG`4VQ{2ANs9 vdKsYK99?HhDVT=^8a3b;9H%!P;lehA6BP(5Ab7n~a4K9+!#TQ76T+E~NJH#~ diff --git a/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class b/wayang-platforms/wayang-semantic/bin/src/test/java/org/apache/wayang/semantic/Review.class deleted file mode 100644 index 9e48cff47f635c797dc6b5e0b5d4013788cd087a..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 4940 zcmeHLTW=gS6h2a7JJK% z`VZKY^|cxIS;?{q?>}NQrpA5C3#OdN`uF%$@JDS}B5*a(O22$(?*Y#oftJV#tPTzr zhRz5z-XxHi@Zm1sN3x3_3nkpA1m0g;q+NWoexJb7w$6DQmI#0^&&?l3aCz~^r(T+3;iNoFk_3CZn$xXX6n;%h zInwAHX>^XXL}ib(Dbl%@^k}ZRrCz56bJ>JCR6XA>QIT~dbh^|#T^bCsi2E2WYE=}a zHjW})gTc;8{+D_Qo>Ms?k8+9Uf$;AvxKor;WWrHfmE2%{Q`VU(bK20Dz}D7%uBued zM?(4hge!Q@D(ULfDZO1;=U0%5GUY=u0@olhkT4`0)JdJDEF{pR`*o-a+gdry=ch>f z1!m?3xTNbB?qbTmRSB7U9V}$*I(sLh-quoLX7c6YpJ4}DHKm^V&c4PJ+)Z>&rDl1o zmKN<3N`_B^p#lwgaSKbyB&a$R7VA;`dLgXEk{Rg+!n$J_1OY>R-jZW&gqsvch;mlx z@5*W{jEOVGMp*L_9^#$1V#Hh{g5l#OM4p_C6YTd*90Rgy#5!^ z8`wu!j%rfmaF0V+`P@<{1Po zBSSIk#uJ5((-ED(&3hG|R1|ziO@$RG{+1u^!*KvgpA6<^-t53z1b#XO-}5idlw+VM zLtUASW21u>zlMF!Z&ok=2rIuGd<95N@Sebn_`e+P6SxjH!n>E?Ap50gTbt43Ra`P>xp3QPfA@u9LX++QIgw zm%ob#B#_{pN1pg=5M%F!F3Pr-!qHyUJnVouql_|%%X zX8~bw?+>1`Pshk7d6!Mw>dEBNzVoRvB)!obZIkIvoqioY0DnjXsb( zwV%^wV}Hb{n~pj40FfUo$_Jw{Q%XB}!08MmqPva?ONB~=!THE3b`7|}S>p%;qf=~u zpJ87@qgXO)=|D(sU*IS^VHPL7l!W8z{3_9s5805@7TPt+?Hc8FjmD_#fi`J0GjNM0 znp^61nlcxUs6*BB!!atdh=ltk^7|zU2ARkG8!p$XNXObZigG0eTZ+7rdI>uxb3z`s zC0+)>Cs}Ye9ZL}lM{PFd2Kh}{XR_F7LvsRaZ+NcCRKf>Bd49qkd}o<-Md^s{W!Cu~ zl%h;|m&}0+kmyM0k|pY-PE!^LwAuJB5c-Z*jvt<+?AOT5B||qGzT*y3_N`3F#Oq*= zv5V7tHt8KLB{Gu_7f*-nX;nr&#gToCNx2*8gi6hlyL=+}Ba8M4A;Z&Ph(N^vmNI%6%R|(f&?9LeC1H{{!?U<`I@->}qve-p)Ql3Fv*t98bAI zBQIU5kbZe3&A8@&aVhUuCEKSs?zS;(YD7WO`1P9p61xO$Px+l zwUC~ka|q1MQ)2t#201yQXICuV7W2{$cYA_y zKqUNBX{rfzIKK*K;R1mx?Ve9#JE2a_)!g;9@mde_xo@)cfSY}WeJvfTZ|q&^`G2}~ zc!Q817OByfx##0Uh$7j%pxl$d<$0rjT!**Gne1fDIm_3Und?jw%goi+K6%Yo!o{z4 zF-YKCp|B)SZ*TN`GK=)w=!>CZE;ASax<{XHE>8A&65HKEvZjRmzJPZxlj!-1JrB#L zGO;Jn?CZ?L90>-u$fa2%_XHn~U~BLp9<+O_5?I~K&ms2F6)jM}b9g-J2dLt87FMub z!?#cHE5T>;XQ Date: Fri, 28 Aug 2026 09:40:45 +0200 Subject: [PATCH 03/10] delete unused file --- .../operators/SemanticMapOperator.java | 24 ------------------- 1 file changed, 24 deletions(-) delete mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java deleted file mode 100644 index 0b8ff993a..000000000 --- a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/SemanticMapOperator.java +++ /dev/null @@ -1,24 +0,0 @@ -/* - * 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.semantic.operators; - -public class SemanticMapOperator { - - -} \ No newline at end of file From b01802d0ade3460e794d8f23f0878b1e7fb9c5ce Mon Sep 17 00:00:00 2001 From: mspruc Date: Fri, 28 Aug 2026 09:49:49 +0200 Subject: [PATCH 04/10] spell check and grammar --- guides/semantic-operators-in-Wayang.md | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/guides/semantic-operators-in-Wayang.md b/guides/semantic-operators-in-Wayang.md index a7de72834..52d4afe3e 100644 --- a/guides/semantic-operators-in-Wayang.md +++ b/guides/semantic-operators-in-Wayang.md @@ -6,7 +6,7 @@ The example uses a semantic filter that classifies movie reviews as positive or ## 1. Semantic operators -A semantic operator an operator much like any Wayang operator, however, it takes a prompt as input, +A semantic operator is an operator much like any Wayang operator, however, it takes a prompt as input, that describes how it should act. For example: @@ -21,12 +21,12 @@ For example: The prompt describes the task. A `SemanticAlgorithm` provides one concrete implementation of that task. A `SemanticAlgorithm` could theoretically be any UDF you desire, there are no strict requirements on its implementation. -The implementation just requires the UDF is described as something that takes an input `Record` and a `prompt` and outputs +The implementation only that the requires the UDF is implemented as something that takes an input `Record` and a `prompt` and outputs whatever datatype is required by the operator. -The ollama local open source model is used as an example for this guide, but may also be useful for quick development. +The Ollama local open-source model is used as an example for this guide, but may also be useful for quick development. -We setup our local model hosting locallly using Docker: +We set up our local model hosting locally using Docker: ```yaml ollama: @@ -69,15 +69,14 @@ private static String callOllama(final String prompt) throws IOException, Interr throw new IOException("Ollama API error: " + response.body()); } - // Parse JSON response to extract the "response" field return parseOllamaResponse(response.body()); } ``` -and the semantic UDF: +and the semantic UDF: ```java -public static boolean isPositiveSentiment3(final Review review, final String prompt) +public static boolean isPositiveSentiment(final Review review, final String prompt) throws IOException, InterruptedException { final String response = callOllama(prompt + " | " + review.getReviewText()); return response.contains("POSITIVE"); @@ -107,9 +106,9 @@ ollamaFilter.loadProfileEstimator = ); ``` -Note that you should provide multiple configurations if you have more semantic operators. -Now we register these UDFs to the `SemanticPlugin` this will automatically construct a new operator -per UDF, please note this may have implications for optimization time depending on your setup. +Note that you should provide a separate configuration for each semantic operator. +Now we register these UDFs with the `SemanticPlugin` this automatically constructs a new operator +per UDF. Please note this may have implications for optimization time depending on your setup. ```java final SemanticPlugin plugin = Semantic.plugin() @@ -122,7 +121,7 @@ final WayangContext wayangContext = new WayangContext() .withPlugin(plugin); ``` -Currently we only have mappings for semantic operators in Java, so you also need the `Java.basicPlugin()`. +Currently, we only have mappings for semantic operators in Java, so you also need the `Java.basicPlugin()`. Finally, you can construct your Wayang plan: ```java From 2a83058ef19a4fdf35f88cd071575ac2af70248e Mon Sep 17 00:00:00 2001 From: mspruc Date: Fri, 28 Aug 2026 09:52:03 +0200 Subject: [PATCH 05/10] Add missing license to guide --- guides/semantic-operators-in-Wayang.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/guides/semantic-operators-in-Wayang.md b/guides/semantic-operators-in-Wayang.md index 52d4afe3e..1a60faa77 100644 --- a/guides/semantic-operators-in-Wayang.md +++ b/guides/semantic-operators-in-Wayang.md @@ -1,3 +1,22 @@ + + # Developing with Semantic Operators in Apache Wayang This guide explains how to define semantic operators, provide executable implementations, register multiple implementations, estimate their costs, and let Apache Wayang select an implementation during optimization. From 7346e70e81438339332e41dcc75b54810477c5c1 Mon Sep 17 00:00:00 2001 From: Juri Petersen Date: Fri, 4 Sep 2026 10:20:42 +0200 Subject: [PATCH 06/10] Add semantic operators with platforms per model --- .project | 28 ++ .settings/org.eclipse.core.resources.prefs | 2 + .settings/org.eclipse.m2e.core.prefs | 4 + conf/flink/default.properties | 4 +- .../operators/SemanticFilterOperator.java | 16 +- .../basic/operators/SemanticOperator.java | 57 ++++ .../semantic/execution/OllamaExecutor.java | 113 +++++++ .../semantic/mappings/JavaFilterMapping.java | 64 ---- .../wayang/semantic/mappings/Mappings.java | 29 -- .../operators/OllamaExecutionOperator.java | 58 ++++ .../operators/OllamaFilterOperator.java | 71 ++++ .../semantic/platform/OllamaPlatform.java | 90 ++++++ .../semantic/plugin/SemanticPlugin.java | 43 +-- .../semantic/udf/SemanticAlgorithm.java | 36 --- ...wayang-semantic-ollama-defaults.properties | 23 ++ .../apache/wayang/semantic/SemBenchTest.java | 306 +++++++++++++----- 16 files changed, 711 insertions(+), 233 deletions(-) create mode 100644 .project create mode 100644 .settings/org.eclipse.core.resources.prefs create mode 100644 .settings/org.eclipse.m2e.core.prefs create mode 100644 wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java delete mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java delete mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java create mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java delete mode 100644 wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java create mode 100644 wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties diff --git a/.project b/.project new file mode 100644 index 000000000..aaad11f4f --- /dev/null +++ b/.project @@ -0,0 +1,28 @@ + + + wayang + + + + + + org.eclipse.m2e.core.maven2Builder + + + + + + org.eclipse.m2e.core.maven2Nature + + + + 1770973809620 + + 30 + + org.eclipse.core.resources.regexFilterMatcher + node_modules|\.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__ + + + + diff --git a/.settings/org.eclipse.core.resources.prefs b/.settings/org.eclipse.core.resources.prefs new file mode 100644 index 000000000..99f26c020 --- /dev/null +++ b/.settings/org.eclipse.core.resources.prefs @@ -0,0 +1,2 @@ +eclipse.preferences.version=1 +encoding/=UTF-8 diff --git a/.settings/org.eclipse.m2e.core.prefs b/.settings/org.eclipse.m2e.core.prefs new file mode 100644 index 000000000..f897a7f1c --- /dev/null +++ b/.settings/org.eclipse.m2e.core.prefs @@ -0,0 +1,4 @@ +activeProfiles= +eclipse.preferences.version=1 +resolveWorkspaceProjects=true +version=1 diff --git a/conf/flink/default.properties b/conf/flink/default.properties index 196b54fb2..8d18d3747 100644 --- a/conf/flink/default.properties +++ b/conf/flink/default.properties @@ -23,8 +23,8 @@ # Local distribute #wayang.flink.mode.run = local -#wayang.flink.paralelism = 1 +#wayang.flink.parallelism = 1 # collection mode wayang.flink.mode.run = collection -wayang.flink.paralelism = 1 +wayang.flink.parallelism = 1 diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java index 302da73a1..f98898428 100644 --- a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.java @@ -23,11 +23,12 @@ import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator; import org.apache.wayang.core.types.DataSetType; -public class SemanticFilterOperator extends UnaryToUnaryOperator { - private final String prompt; +public class SemanticFilterOperator extends UnaryToUnaryOperator implements SemanticOperator { + + private final String prompt; + + private final Set targetModels; - public final Set targetModels; - public SemanticFilterOperator(final DataSetType type, final String prompt) { super(type, type, false); this.prompt = prompt; @@ -59,10 +60,17 @@ public SemanticFilterOperator(final UnaryToUnaryOperator that) { this.targetModels = null; } + @Override public String getPrompt() { return prompt; } + @Override + public Set getTargetModels() { + return targetModels; + } + + @Override public void addTargetModel(final Object model) { targetModels.add(model); } diff --git a/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java new file mode 100644 index 000000000..29c00c80a --- /dev/null +++ b/wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticOperator.java @@ -0,0 +1,57 @@ +/* + * 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.basic.operators; + +import java.util.Set; + +/** + * Common contract for operators whose UDF is not plain Java code but a natural-language + * {@code prompt} that gets executed by a large language model (LLM), e.g. a + * {@link SemanticFilterOperator} or a future {@code SemanticJoinOperator}. + *

+ * This is deliberately kept as a slim interface rather than an abstract base class. Semantic + * operators still need to pick the operator shape that fits their arity/semantics, e.g. a + * filter extends {@link org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator} while a + * join would extend {@link org.apache.wayang.core.plan.wayangplan.BinaryToUnaryOperator}. As + * Java only allows single class inheritance, tying this abstraction to a specific operator + * base class would force every semantic operator into the same shape. Implementors are free + * to (and are expected to) extend whichever operator base class matches their actual + * input/output arity, while still sharing the prompt/target-model contract defined here. + */ +public interface SemanticOperator { + + /** + * @return the prompt describing the semantic algorithm that this operator should execute + */ + String getPrompt(); + + /** + * @return the set of models that are allowed to execute this operator's {@link #getPrompt()}, + * or {@code null} if no target models have been configured + */ + Set getTargetModels(); + + /** + * Adds a model to the set of models that are allowed to execute this operator's + * {@link #getPrompt()}. + * + * @param model the model to add + */ + void addTargetModel(Object model); +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java new file mode 100644 index 000000000..706d90241 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/execution/OllamaExecutor.java @@ -0,0 +1,113 @@ +/* + * 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.semantic.execution; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +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.ExecutionTask; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.Executor; +import org.apache.wayang.core.platform.PartialExecution; +import org.apache.wayang.core.platform.PushExecutorTemplate; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.util.Formats; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.semantic.operators.OllamaExecutionOperator; +import org.apache.wayang.semantic.platform.OllamaPlatform; + +/** + * {@link Executor} implementation for the {@link OllamaPlatform}. + */ +public class OllamaExecutor extends PushExecutorTemplate { + + private final OllamaPlatform platform; + + public OllamaExecutor(final OllamaPlatform platform, final Job job) { + super(job); + this.platform = platform; + } + + @Override + public OllamaPlatform getPlatform() { + return this.platform; + } + + @Override + protected Tuple, PartialExecution> execute( + final ExecutionTask task, + final List inputChannelInstances, + final OptimizationContext.OperatorContext producerOperatorContext, + final boolean isRequestEagerExecution + ) { + final ChannelInstance[] outputChannelInstances = task.getOperator().createOutputChannelInstances( + this, task, producerOperatorContext, inputChannelInstances + ); + + final Collection executionLineageNodes; + final Collection producedChannelInstances; + this.job.reportProgress(task.getOperator().getName(), 50); + final long startTime = System.currentTimeMillis(); + try { + final Tuple, Collection> results = + cast(task.getOperator()).evaluate( + toArray(inputChannelInstances), + outputChannelInstances, + this, + producerOperatorContext + ); + executionLineageNodes = results.getField0(); + producedChannelInstances = results.getField1(); + } catch (Exception e) { + throw new WayangException(String.format("Executing %s failed.", task), e); + } + final long endTime = System.currentTimeMillis(); + final long executionDuration = endTime - startTime; + + this.job.reportProgress(task.getOperator().getName(), 100); + + final PartialExecution partialExecution = this.createPartialExecution(executionLineageNodes, executionDuration); + + if (partialExecution == null && executionDuration > 10) { + this.logger.warn("Execution of {} took suspiciously long ({}).", task, Formats.formatDuration(executionDuration)); + } + + this.registerMeasuredCardinalities(producedChannelInstances); + + if (isRequestEagerExecution && partialExecution == null) { + this.logger.info("{} was not executed eagerly as requested.", task); + } + + return new Tuple<>(Arrays.asList(outputChannelInstances), partialExecution); + } + + private static OllamaExecutionOperator cast(final ExecutionOperator executionOperator) { + return (OllamaExecutionOperator) executionOperator; + } + + private static ChannelInstance[] toArray(final List channelInstances) { + final ChannelInstance[] array = new ChannelInstance[channelInstances.size()]; + return channelInstances.toArray(array); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java deleted file mode 100644 index d449fab4b..000000000 --- a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/JavaFilterMapping.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.semantic.mappings; - -import java.util.Collection; -import java.util.Collections; - -import org.apache.wayang.basic.operators.SemanticFilterOperator; -import org.apache.wayang.core.function.PredicateDescriptor; -import org.apache.wayang.core.function.FunctionDescriptor.SerializablePredicate; -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.types.DataSetType; -import org.apache.wayang.java.operators.JavaFilterOperator; -import org.apache.wayang.java.platform.JavaPlatform; -import org.apache.wayang.semantic.udf.SemanticAlgorithm; - -public class JavaFilterMapping implements Mapping { - private final SemanticAlgorithm model; - - public JavaFilterMapping(final SemanticAlgorithm model) { - this.model = model; - } - - @Override - public Collection getTransformations() { - return Collections.singleton(new PlanTransformation(this.createSubplanPattern(), - this.createReplacementSubplanFactory(), JavaPlatform.getInstance())); - } - - private SubplanPattern createSubplanPattern() { - return SubplanPattern.createSingleton(new OperatorPattern>("semantic_filter", - new SemanticFilterOperator<>(DataSetType.NONE), false).withAdditionalTest(op -> op.targetModels != null) - .withAdditionalTest(op -> op.targetModels.contains(model))); - } - - private ReplacementSubplanFactory createReplacementSubplanFactory() { - return new ReplacementSubplanFactory.OfSingleOperators>((matchedOperator, epoch) -> { - final SerializablePredicate predicate = input -> model.impl.apply(input, matchedOperator.getPrompt()); - final PredicateDescriptor predicateDescriptor = new PredicateDescriptor<>(predicate, - matchedOperator.getOutput().getType().getDataUnitType().getTypeClass(), model.loadProfileEstimator); - return new JavaFilterOperator<>(predicateDescriptor).at(epoch); - }); - } -} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java deleted file mode 100644 index cc4b494e9..000000000 --- a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/mappings/Mappings.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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.semantic.mappings; - -import java.util.Collection; -import org.apache.wayang.core.mapping.Mapping; -import java.util.Arrays; - -public class Mappings { - public static final Collection ALL = Arrays.asList( - new JavaFilterMapping(null) - ); -} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java new file mode 100644 index 000000000..b635a95b4 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaExecutionOperator.java @@ -0,0 +1,58 @@ +/* + * 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.semantic.operators; + +import java.util.Collection; + +import org.apache.wayang.core.optimizer.OptimizationContext; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.semantic.execution.OllamaExecutor; +import org.apache.wayang.semantic.platform.OllamaPlatform; + +/** + * Execution operator for the {@link OllamaPlatform}. + */ +public interface OllamaExecutionOperator extends ExecutionOperator { + + @Override + default OllamaPlatform getPlatform() { + return OllamaPlatform.getInstance(); + } + + /** + * Evaluates this operator by calling out to Ollama. Mirrors + * {@code org.apache.wayang.java.operators.JavaExecutionOperator#evaluate}, just for the + * {@link OllamaPlatform} instead of the Java platform. + * + * @param inputs {@link ChannelInstance}s that satisfy the inputs of this operator + * @param outputs {@link ChannelInstance}s that collect the outputs of this operator + * @param ollamaExecutor that executes this instance + * @param operatorContext optimization information for this instance + * @return {@link Collection}s of what has been executed and produced + */ + Tuple, Collection> evaluate( + ChannelInstance[] inputs, + ChannelInstance[] outputs, + OllamaExecutor ollamaExecutor, + OptimizationContext.OperatorContext operatorContext); + +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java new file mode 100644 index 000000000..40d9ee7ae --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/operators/OllamaFilterOperator.java @@ -0,0 +1,71 @@ +/* + * 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.semantic.operators; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelDescriptor; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.java.channels.CollectionChannel; +import org.apache.wayang.java.channels.StreamChannel; + +/** + * Base for {@link OllamaExecutionOperator}s that implement a {@link SemanticFilterOperator} by calling + * out to one specific model. Every concrete subclass hardcodes exactly one model's evaluation logic + * instead of taking it as an injected UDF, so {@link #getPrompt()} (inherited from the matched logical + * {@link SemanticFilterOperator}) remains the single source of truth for the prompt; a subclass is free + * to use it or to ignore it in favor of its own hardcoded behavior. + *

+ * Input/output channels are the same {@link StreamChannel}/{@link CollectionChannel} that + * {@code JavaFilterOperator} supports, so data can flow between the surrounding Java pipeline and an + * Ollama-backed operator without any channel conversion. + */ +public abstract class OllamaFilterOperator extends SemanticFilterOperator implements OllamaExecutionOperator { + + protected OllamaFilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + /** + * Creates a new instance of the same concrete subclass for the given type/prompt. + */ + protected abstract OllamaFilterOperator newInstance(DataSetType type, String prompt); + + @Override + protected final ExecutionOperator createCopy() { + return this.newInstance(this.getInputType(), this.getPrompt()); + } + + @Override + public List getSupportedInputChannels(final int index) { + assert index <= this.getNumInputs() || (index == 0 && this.getNumInputs() == 0); + if (this.getInput(index).isBroadcast()) return Collections.singletonList(CollectionChannel.DESCRIPTOR); + return Arrays.asList(CollectionChannel.DESCRIPTOR, StreamChannel.DESCRIPTOR); + } + + @Override + public List getSupportedOutputChannels(final int index) { + assert index <= this.getNumOutputs() || (index == 0 && this.getNumOutputs() == 0); + return Collections.singletonList(StreamChannel.DESCRIPTOR); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java new file mode 100644 index 000000000..1cad4d6b1 --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/platform/OllamaPlatform.java @@ -0,0 +1,90 @@ +/* + * 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.semantic.platform; + +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.optimizer.costs.LoadProfileToTimeConverter; +import org.apache.wayang.core.optimizer.costs.LoadToTimeConverter; +import org.apache.wayang.core.optimizer.costs.TimeToCostConverter; +import org.apache.wayang.core.platform.Executor; +import org.apache.wayang.core.platform.Platform; +import org.apache.wayang.core.util.ReflectionUtils; +import org.apache.wayang.semantic.execution.OllamaExecutor; + +/** + * {@link Platform} for operators that are executed by calling out to an Ollama-hosted LLM. + *

+ * Kept as its own {@link Platform} (rather than piggy-backing on + * {@link org.apache.wayang.java.platform.JavaPlatform}) so that the cost of an LLM call is modeled and + * optimized on its own terms, instead of being reported as if it were free Java-Stream work. + */ +public class OllamaPlatform extends Platform { + + private static final String PLATFORM_NAME = "Ollama"; + + private static final String CONFIG_NAME = "semantic.ollama"; + + private static final String DEFAULT_CONFIG_FILE = "wayang-semantic-ollama-defaults.properties"; + + private static OllamaPlatform instance = null; + + public static OllamaPlatform getInstance() { + if (instance == null) { + instance = new OllamaPlatform(); + } + return instance; + } + + private OllamaPlatform() { + super(PLATFORM_NAME, CONFIG_NAME); + } + + @Override + public void configureDefaults(final Configuration configuration) { + configuration.load(ReflectionUtils.loadResource(DEFAULT_CONFIG_FILE)); + } + + @Override + public Executor.Factory getExecutorFactory() { + return job -> new OllamaExecutor(this, job); + } + + @Override + public LoadProfileToTimeConverter createLoadProfileToTimeConverter(final Configuration configuration) { + final int cpuMhz = (int) configuration.getLongProperty("wayang.semantic.ollama.cpu.mhz"); + final int numCores = (int) configuration.getLongProperty("wayang.semantic.ollama.cores"); + final double hdfsMsPerMb = configuration.getDoubleProperty("wayang.semantic.ollama.hdfs.ms-per-mb"); + final double stretch = configuration.getDoubleProperty("wayang.semantic.ollama.stretch"); + return LoadProfileToTimeConverter.createTopLevelStretching( + LoadToTimeConverter.createLinearCoverter(1 / (numCores * cpuMhz * 1000d)), + LoadToTimeConverter.createLinearCoverter(hdfsMsPerMb / 1000000d), + LoadToTimeConverter.createLinearCoverter(0), + (cpuEstimate, diskEstimate, networkEstimate) -> cpuEstimate.plus(diskEstimate).plus(networkEstimate), + stretch + ); + } + + @Override + public TimeToCostConverter createTimeToCostConverter(final Configuration configuration) { + return new TimeToCostConverter( + configuration.getDoubleProperty("wayang.semantic.ollama.costs.fix"), + configuration.getDoubleProperty("wayang.semantic.ollama.costs.per-ms") + ); + } +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java index b5d0dc263..2b0964f55 100644 --- a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java +++ b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/plugin/SemanticPlugin.java @@ -18,22 +18,21 @@ package org.apache.wayang.semantic.plugin; -import org.apache.wayang.core.plugin.Plugin; -import org.apache.wayang.basic.operators.SemanticFilterOperator; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + import org.apache.wayang.core.api.Configuration; import org.apache.wayang.core.mapping.Mapping; +import org.apache.wayang.core.mapping.PlanTransformation; import org.apache.wayang.core.optimizer.channels.ChannelConversion; import org.apache.wayang.core.platform.Platform; -import org.apache.wayang.semantic.mappings.JavaFilterMapping; -import org.apache.wayang.semantic.udf.SemanticAlgorithm; +import org.apache.wayang.core.plugin.Plugin; import org.apache.wayang.java.Java; import org.apache.wayang.java.platform.JavaPlatform; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; - public class SemanticPlugin implements Plugin { private final List mappings; @@ -52,8 +51,15 @@ public Collection getMappings() { @Override public Collection getRequiredPlatforms() { - // TODO: maybe we should find another way to handle this? but do Java for now. - return Collections.singleton(JavaPlatform.getInstance()); + final Set platforms = new LinkedHashSet<>(); + // The surrounding pipeline (sources, sinks, plain transformations) always runs on Java. + platforms.add(JavaPlatform.getInstance()); + for (final Mapping mapping : this.mappings) { + for (final PlanTransformation transformation : mapping.getTransformations()) { + platforms.addAll(transformation.getTargetPlatforms()); + } + } + return platforms; } @Override @@ -65,13 +71,14 @@ public Collection getChannelConversions() { public void setProperties(final Configuration configuration) { } - public SemanticPlugin withOperatorMapping(final Class operatorClass, final SemanticAlgorithm model) { + /** + * Registers a {@link Mapping} that rewrites a semantic operator (e.g. a + * {@link org.apache.wayang.basic.operators.SemanticFilterOperator}) into a physical operator for one + * specific model implementation. + */ + public SemanticPlugin withMapping(final Mapping mapping) { final List nextMappings = new ArrayList<>(this.mappings); - - if (operatorClass.equals(SemanticFilterOperator.class)) { - nextMappings.add(new JavaFilterMapping((SemanticAlgorithm) model)); - } - + nextMappings.add(mapping); return new SemanticPlugin(nextMappings); } -} \ No newline at end of file +} diff --git a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java b/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java deleted file mode 100644 index 00549fb86..000000000 --- a/wayang-platforms/wayang-semantic/src/main/java/org/apache/wayang/semantic/udf/SemanticAlgorithm.java +++ /dev/null @@ -1,36 +0,0 @@ -/* - * 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.semantic.udf; - -import java.util.function.BiFunction; - -import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; - -/** - * A udf that represents an algorithm you use in semantic queries - */ -public class SemanticAlgorithm { - /** - * An implementation that maps an input T with a prompt String to an output T - * defined via the implementation in the lambda expression. - */ - public BiFunction impl; - - public LoadProfileEstimator loadProfileEstimator; -} diff --git a/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties b/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties new file mode 100644 index 000000000..13106b24e --- /dev/null +++ b/wayang-platforms/wayang-semantic/src/main/resources/wayang-semantic-ollama-defaults.properties @@ -0,0 +1,23 @@ +# +# 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. +# + +wayang.semantic.ollama.cpu.mhz = 2700 +wayang.semantic.ollama.cores = 1 +wayang.semantic.ollama.hdfs.ms-per-mb = 2.7 +wayang.semantic.ollama.stretch = 1 +wayang.semantic.ollama.costs.fix = 0.0 +wayang.semantic.ollama.costs.per-ms = 1.0 diff --git a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java index 929bd60a1..e76c1bcdf 100644 --- a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java +++ b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java @@ -18,33 +18,6 @@ package org.apache.wayang.semantic; -import org.apache.wayang.core.api.Configuration; -import org.apache.wayang.core.api.WayangContext; -import org.apache.wayang.core.function.ExecutionContext; -import org.apache.wayang.core.function.FunctionDescriptor; -import org.apache.wayang.core.function.TransformationDescriptor; -import org.apache.wayang.core.function.FunctionDescriptor.SerializablePredicate; -import org.apache.wayang.core.optimizer.costs.EstimationContext; -import org.apache.wayang.core.optimizer.costs.LoadProfile; -import org.apache.wayang.core.optimizer.costs.LoadProfileEstimator; -import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators; -import org.apache.wayang.core.plan.wayangplan.WayangPlan; -import org.apache.wayang.core.types.DataSetType; -import org.apache.wayang.core.util.WayangArrays; -import org.apache.wayang.core.util.WayangCollections; -import org.apache.wayang.java.Java; -import org.apache.wayang.java.operators.JavaCollectionSource; -import org.apache.wayang.java.operators.JavaDoWhileOperator; -import org.apache.wayang.java.operators.JavaLocalCallbackSink; -import org.apache.wayang.semantic.plugin.SemanticPlugin; -import org.apache.wayang.semantic.Semantic; -import org.apache.wayang.semantic.udf.SemanticAlgorithm; -import org.apache.wayang.semantic.operators.*; -import org.apache.wayang.api.JavaPlanBuilder; -import org.apache.wayang.basic.operators.SemanticFilterOperator; -import org.apache.wayang.java.operators.JavaMapOperator; -import org.junit.jupiter.api.Test; - import java.io.IOException; import java.net.URI; import java.net.http.HttpClient; @@ -52,12 +25,34 @@ import java.net.http.HttpResponse; import java.time.Duration; import java.util.Arrays; -import java.util.List; import java.util.Collection; import java.util.Collections; -import java.util.LinkedList; +import java.util.List; +import java.util.stream.Stream; -import static org.junit.jupiter.api.Assertions.assertEquals; +import org.apache.wayang.api.JavaPlanBuilder; +import org.apache.wayang.basic.operators.SemanticFilterOperator; +import org.apache.wayang.core.api.Configuration; +import org.apache.wayang.core.api.WayangContext; +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.optimizer.OptimizationContext; +import org.apache.wayang.core.plan.wayangplan.ExecutionOperator; +import org.apache.wayang.core.platform.ChannelInstance; +import org.apache.wayang.core.platform.lineage.ExecutionLineageNode; +import org.apache.wayang.core.types.DataSetType; +import org.apache.wayang.core.util.Tuple; +import org.apache.wayang.java.Java; +import org.apache.wayang.java.channels.JavaChannelInstance; +import org.apache.wayang.java.channels.StreamChannel; +import org.apache.wayang.semantic.execution.OllamaExecutor; +import org.apache.wayang.semantic.operators.OllamaFilterOperator; +import org.apache.wayang.semantic.platform.OllamaPlatform; +import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.junit.jupiter.api.Test; class SemBenchTest { private static List loadReviews() { @@ -68,13 +63,12 @@ private static List loadReviews() { new Review("taken_4", "It was okay. Not great, not terrible.")); } - @SuppressWarnings({ "unchecked", "rawtypes" }) @Test void testSemBenchMoviesWithOllama() { final Configuration configuration = new Configuration(); configuration.setProperty("wayang.java.filter.load", """ { - "in":1, + "in":1, "out":1, "cpu":"${25*in0 + 350000}", "ram":"100000", @@ -126,54 +120,15 @@ void testSemBenchMoviesWithOllama() { """ ); - final SemanticAlgorithm ollamaFilter = new SemanticAlgorithm(); - ollamaFilter.impl = (input, prompt) -> { - try { - return OllamaSemanticFilter.isPositiveSentiment((Review) input); - } catch (IOException | InterruptedException e) { - throw new RuntimeException("Ollama call failed", e); - } - }; - ollamaFilter.loadProfileEstimator = - LoadProfileEstimators.createFromSpecification( - "wayang.semantic.ollama.model1.load", - configuration - ); - - final SemanticAlgorithm ollamaFilter2 = new SemanticAlgorithm(); - ollamaFilter2.impl = (input, prompt) -> { - try { - return OllamaSemanticFilter.isPositiveSentiment2((Review) input); - } catch (IOException | InterruptedException e) { - throw new RuntimeException("Ollama call failed", e); - } - }; - ollamaFilter2.loadProfileEstimator = - LoadProfileEstimators.createFromSpecification( - "wayang.semantic.ollama.model2.load", - configuration - ); - - final SemanticAlgorithm ollamaFilter3 = new SemanticAlgorithm(); - ollamaFilter3.impl = (input, prompt) -> { - try { - return OllamaSemanticFilter.isPositiveSentiment3((Review) input, (String) prompt); - } catch (IOException | InterruptedException e) { - throw new RuntimeException("Ollama call failed", e); - } - }; - ollamaFilter3.loadProfileEstimator = - LoadProfileEstimators.createFromSpecification( - "wayang.semantic.ollama.model3.load", - configuration - ); - + // Each Ollama model is its own physical operator + Mapping; the optimizer picks among them by cost. + // There is exactly one prompt, owned by the logical SemanticFilterOperator (set via .semanticFilter(...)), + // and each physical operator decides for itself whether it needs that prompt. final SemanticPlugin plugin = Semantic.plugin() - .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter) - .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter2) - .withOperatorMapping(SemanticFilterOperator.class, ollamaFilter3); + .withMapping(new OllamaModel1FilterMapping()) + .withMapping(new OllamaModel2FilterMapping()) + .withMapping(new OllamaModel3FilterMapping()); - final WayangContext wayangContext = new WayangContext() + final WayangContext wayangContext = new WayangContext(configuration) .withPlugin(Java.basicPlugin()) .withPlugin(plugin); final JavaPlanBuilder planBuilder = new JavaPlanBuilder(wayangContext); @@ -181,7 +136,7 @@ void testSemBenchMoviesWithOllama() { final Collection positiveReviewCnt = planBuilder.loadCollection(loadReviews()) .filter(review -> "taken_3".equals(review.getId())) .semanticFilter("Analyze the review after the | and write either \"POSITIVE\" if the review has a positive sentiment and \"NEGATIVE\" if the review has a negative sentiment.") - .withTargetModels(ollamaFilter, ollamaFilter2, ollamaFilter3) + .withTargetModels(OllamaModel1FilterOperator.class, OllamaModel2FilterOperator.class, OllamaModel3FilterOperator.class) .count() .collect(); } @@ -210,6 +165,197 @@ public String toString() { } } +/** + * Physical operator for the "model1" Ollama implementation: a fixed sentiment prompt, ignores the + * logical operator's {@link #getPrompt()}. + */ +final class OllamaModel1FilterOperator extends OllamaFilterOperator { + + OllamaModel1FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment(review); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model1.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel1FilterOperator<>(type, prompt); + } +} + +/** + * Physical operator for the "model2" Ollama implementation: a different fixed sentiment prompt, also + * ignores the logical operator's {@link #getPrompt()}. + */ +final class OllamaModel2FilterOperator extends OllamaFilterOperator { + + OllamaModel2FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment2(review); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model2.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel2FilterOperator<>(type, prompt); + } +} + +/** + * Physical operator for the "model3" Ollama implementation: the only one that actually uses the + * logical operator's {@link #getPrompt()}. + */ +final class OllamaModel3FilterOperator extends OllamaFilterOperator { + + OllamaModel3FilterOperator(final DataSetType type, final String prompt) { + super(type, prompt); + } + + @Override + @SuppressWarnings("unchecked") + public Tuple, Collection> evaluate( + final ChannelInstance[] inputs, + final ChannelInstance[] outputs, + final OllamaExecutor ollamaExecutor, + final OptimizationContext.OperatorContext operatorContext) { + assert inputs.length == this.getNumInputs(); + assert outputs.length == this.getNumOutputs(); + + final Stream filtered = ((JavaChannelInstance) inputs[0]).provideStream().filter(review -> { + try { + return OllamaSemanticFilter.isPositiveSentiment3(review, this.getPrompt()); + } catch (IOException | InterruptedException e) { + throw new RuntimeException("Ollama call failed", e); + } + }); + ((StreamChannel.Instance) outputs[0]).accept(filtered); + + return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext); + } + + @Override + public String getLoadProfileEstimatorConfigurationKey() { + return "wayang.semantic.ollama.model3.load"; + } + + @Override + protected OllamaFilterOperator newInstance(final DataSetType type, final String prompt) { + return new OllamaModel3FilterOperator<>(type, prompt); + } +} + +abstract class AbstractOllamaFilterMapping implements Mapping { + + private final Class targetOperatorClass; + + AbstractOllamaFilterMapping(final Class targetOperatorClass) { + this.targetOperatorClass = targetOperatorClass; + } + + @Override + public Collection getTransformations() { + return Collections.singleton(new PlanTransformation(this.createSubplanPattern(), + this.createReplacementSubplanFactory(), OllamaPlatform.getInstance())); + } + + private SubplanPattern createSubplanPattern() { + return SubplanPattern.createSingleton(new OperatorPattern>("semantic_filter", + new SemanticFilterOperator<>(DataSetType.NONE), false) + .withAdditionalTest(op -> op.getTargetModels() != null) + .withAdditionalTest(op -> op.getTargetModels().contains(this.targetOperatorClass))); + } + + protected abstract OllamaFilterOperator createOperator(DataSetType type, String prompt); + + private ReplacementSubplanFactory createReplacementSubplanFactory() { + return new ReplacementSubplanFactory.OfSingleOperators>((matchedOperator, epoch) -> + this.createOperator(matchedOperator.getInputType(), matchedOperator.getPrompt()).at(epoch)); + } +} + +final class OllamaModel1FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel1FilterMapping() { + super(OllamaModel1FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel1FilterOperator<>(type, prompt); + } +} + +final class OllamaModel2FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel2FilterMapping() { + super(OllamaModel2FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel2FilterOperator<>(type, prompt); + } +} + +final class OllamaModel3FilterMapping extends AbstractOllamaFilterMapping { + OllamaModel3FilterMapping() { + super(OllamaModel3FilterOperator.class); + } + + @Override + protected OllamaFilterOperator createOperator(final DataSetType type, final String prompt) { + return new OllamaModel3FilterOperator<>(type, prompt); + } +} + final class OllamaSemanticFilter { private static final String OLLAMA_API_URL = "http://apache-wayang-ollama:11434/api/generate"; private static final String MODEL_NAME = "tinyllama"; @@ -270,4 +416,4 @@ private static String parseOllamaResponse(final String jsonResponse) { private static String escapeJson(final String str) { return str.replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r"); } -} \ No newline at end of file +} From db219ba6d1ace66a0034950cb072cca7df40a59b Mon Sep 17 00:00:00 2001 From: Juri Petersen Date: Wed, 9 Sep 2026 07:30:22 +0200 Subject: [PATCH 07/10] Update pom and fix address for Ollama --- pom.xml | 2 +- .../src/test/java/org/apache/wayang/semantic/SemBenchTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index f0d5cf77e..22a666e5b 100644 --- a/pom.xml +++ b/pom.xml @@ -23,7 +23,7 @@ org.apache apache - 32 + 34 org.apache.wayang diff --git a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java index e76c1bcdf..38248e27e 100644 --- a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java +++ b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java @@ -357,7 +357,7 @@ protected OllamaFilterOperator createOperator(final DataSetType type, } final class OllamaSemanticFilter { - private static final String OLLAMA_API_URL = "http://apache-wayang-ollama:11434/api/generate"; + private static final String OLLAMA_API_URL = "http://ollama:11434/api/generate"; private static final String MODEL_NAME = "tinyllama"; private static final HttpClient httpClient = HttpClient.newHttpClient(); From 4597c111f2683f4aae8baa202819f089e05f4fdd Mon Sep 17 00:00:00 2001 From: Juri Petersen Date: Thu, 10 Sep 2026 09:53:34 +0200 Subject: [PATCH 08/10] Mock Ollama API for test --- .../apache/wayang/semantic/SemBenchTest.java | 64 ++++++++++++++++++- 1 file changed, 63 insertions(+), 1 deletion(-) diff --git a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java index 38248e27e..30f35ccf2 100644 --- a/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java +++ b/wayang-platforms/wayang-semantic/src/test/java/org/apache/wayang/semantic/SemBenchTest.java @@ -18,11 +18,16 @@ package org.apache.wayang.semantic; +import static org.junit.jupiter.api.Assertions.assertEquals; + import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Arrays; import java.util.Collection; @@ -52,9 +57,52 @@ import org.apache.wayang.semantic.operators.OllamaFilterOperator; import org.apache.wayang.semantic.platform.OllamaPlatform; import org.apache.wayang.semantic.plugin.SemanticPlugin; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import com.sun.net.httpserver.HttpServer; + class SemBenchTest { + + /** + * In-process stand-in for a real Ollama server: it never leaves the JVM and never needs a model + * to be pulled, so the test runs unmodified in CI. It answers with the same response envelope a + * real Ollama server would ("{"response":"..."}"), classifying sentiment by a keyword heuristic + * on the prompt text, so the assertions below exercise genuine end-to-end behavior of the plan + * (filtering, the semantic operator, and JSON parsing) rather than a canned constant. + */ + private static final String[] NEGATIVE_KEYWORDS = {"terrible", "disappointed", "boring", "not recommend"}; + + private HttpServer mockOllamaServer; + + @BeforeEach + void startMockOllamaServer() throws IOException { + this.mockOllamaServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + this.mockOllamaServer.createContext("/api/generate", exchange -> { + final String requestBody = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + final boolean isNegative = Arrays.stream(NEGATIVE_KEYWORDS) + .anyMatch(keyword -> requestBody.toLowerCase().contains(keyword)); + final String responseBody = String.format("{\"response\":\"%s\"}", isNegative ? "NEGATIVE" : "POSITIVE"); + final byte[] responseBytes = responseBody.getBytes(StandardCharsets.UTF_8); + + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, responseBytes.length); + try (OutputStream responseStream = exchange.getResponseBody()) { + responseStream.write(responseBytes); + } + }); + this.mockOllamaServer.start(); + + final int port = this.mockOllamaServer.getAddress().getPort(); + OllamaSemanticFilter.setApiUrl("http://127.0.0.1:" + port + "/api/generate"); + } + + @AfterEach + void stopMockOllamaServer() { + this.mockOllamaServer.stop(0); + } + private static List loadReviews() { return Arrays.asList(new Review("taken_1", "The movie was fantastic. Great acting and an engaging story."), new Review("taken_2", "I was disappointed. The plot was boring and too long."), @@ -139,6 +187,12 @@ void testSemBenchMoviesWithOllama() { .withTargetModels(OllamaModel1FilterOperator.class, OllamaModel2FilterOperator.class, OllamaModel3FilterOperator.class) .count() .collect(); + + // Of the two "taken_3" reviews, only the "Absolutely loved it!" one is positive; the mock + // Ollama server classifies by keyword, so this checks the plan actually ran the semantic + // filter rather than merely completing without error. + assertEquals(1, positiveReviewCnt.size()); + assertEquals(1L, positiveReviewCnt.iterator().next()); } } @@ -357,10 +411,18 @@ protected OllamaFilterOperator createOperator(final DataSetType type, } final class OllamaSemanticFilter { - private static final String OLLAMA_API_URL = "http://ollama:11434/api/generate"; + private static volatile String OLLAMA_API_URL = "http://ollama:11434/api/generate"; private static final String MODEL_NAME = "tinyllama"; private static final HttpClient httpClient = HttpClient.newHttpClient(); + /** + * Points calls at a different Ollama-compatible endpoint, e.g. an in-process mock server for + * tests, so the real HTTP wiring gets exercised without depending on an external Ollama instance. + */ + static void setApiUrl(final String apiUrl) { + OLLAMA_API_URL = apiUrl; + } + public static boolean isPositiveSentiment(final Review review) throws IOException, InterruptedException { final String prompt = String.format("Analyze the sentiment of this movie review. " + "Reply with only 'POSITIVE' or 'NEGATIVE'.\n\n" + "Review: %s", review.getReviewText()); From 26ba63f3f13cab0861ccdc4b736823163d89e2d5 Mon Sep 17 00:00:00 2001 From: Juri Petersen Date: Thu, 10 Sep 2026 10:28:19 +0200 Subject: [PATCH 09/10] Remove dotfiles --- .project | 28 ---------------------- .settings/org.eclipse.core.resources.prefs | 2 -- .settings/org.eclipse.m2e.core.prefs | 4 ---- 3 files changed, 34 deletions(-) delete mode 100644 .project delete mode 100644 .settings/org.eclipse.core.resources.prefs delete mode 100644 .settings/org.eclipse.m2e.core.prefs diff --git a/.project b/.project deleted file mode 100644 index aaad11f4f..000000000 --- a/.project +++ /dev/null @@ -1,28 +0,0 @@ - - - wayang - - - - - - org.eclipse.m2e.core.maven2Builder - - - - - - org.eclipse.m2e.core.maven2Nature - - - - 1770973809620 - - 30 - - org.eclipse.core.resources.regexFilterMatcher - node_modules|\.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__ - - - - diff --git a/.settings/org.eclipse.core.resources.prefs b/.settings/org.eclipse.core.resources.prefs deleted file mode 100644 index 99f26c020..000000000 --- a/.settings/org.eclipse.core.resources.prefs +++ /dev/null @@ -1,2 +0,0 @@ -eclipse.preferences.version=1 -encoding/=UTF-8 diff --git a/.settings/org.eclipse.m2e.core.prefs b/.settings/org.eclipse.m2e.core.prefs deleted file mode 100644 index f897a7f1c..000000000 --- a/.settings/org.eclipse.m2e.core.prefs +++ /dev/null @@ -1,4 +0,0 @@ -activeProfiles= -eclipse.preferences.version=1 -resolveWorkspaceProjects=true -version=1 From c2556729fe5720657e92336f897ff43ae75070d6 Mon Sep 17 00:00:00 2001 From: Juri Petersen Date: Mon, 14 Sep 2026 14:00:41 +0200 Subject: [PATCH 10/10] Remove default.properties for flink --- conf/flink/default.properties | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/conf/flink/default.properties b/conf/flink/default.properties index 8d18d3747..196b54fb2 100644 --- a/conf/flink/default.properties +++ b/conf/flink/default.properties @@ -23,8 +23,8 @@ # Local distribute #wayang.flink.mode.run = local -#wayang.flink.parallelism = 1 +#wayang.flink.paralelism = 1 # collection mode wayang.flink.mode.run = collection -wayang.flink.parallelism = 1 +wayang.flink.paralelism = 1