Add semantic filters with abstractions for semantic models as a platform

guides/semantic-operators-in-Wayang.md

@@ -0,0 +1,156 @@
+<!--
+
+  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.
+
+-->
+
+# 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 is 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 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.
+
+We set up our local model hosting locally 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<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+
+    if (response.statusCode() != 200) {
+        throw new IOException("Ollama API error: " + response.body());
+    }
+
+    return parseOllamaResponse(response.body());
+}
+```
+
+and the semantic UDF:
+
+```java
+public static boolean isPositiveSentiment(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 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()
+        .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<Long> 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.
+

pom.xml

@@ -23,7 +23,7 @@
     <parent>
         <groupId>org.apache</groupId>
         <artifactId>apache</artifactId>
-        <version>32</version>
+        <version>34</version>
     </parent>
 
     <groupId>org.apache.wayang</groupId>

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.
     *

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,

wayang-commons/wayang-basic/src/main/java/org/apache/wayang/basic/operators/SemanticFilterOperator.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.basic.operators;
+
+import java.util.Set;
+
+import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator;
+import org.apache.wayang.core.types.DataSetType;
+
+public class SemanticFilterOperator<T> extends UnaryToUnaryOperator<T, T> implements SemanticOperator {
+
+    private final String prompt;
+
+    private final Set<Object> targetModels;
+
+    public SemanticFilterOperator(final DataSetType<T> type, final String prompt) {
+        super(type, type, false);
+        this.prompt = prompt;
+        this.targetModels = null;
+    }
+
+    public SemanticFilterOperator(final DataSetType<T> type, final String prompt, final Set<Object> targetModels) {
+        super(type, type, false);
+        this.prompt = prompt;
+        this.targetModels = targetModels;
+    }
+
+    public SemanticFilterOperator(final DataSetType<T> type) {
+        super(type, type, false);
+        this.prompt = "";
+        this.targetModels = null;
+    }
+
+    public SemanticFilterOperator(final DataSetType<T> inputType, final DataSetType<T> outputType,
+            final boolean isSupportingBroadcastInputs) {
+        super(inputType, outputType, isSupportingBroadcastInputs);
+        this.prompt = "";
+        this.targetModels = null;
+        }
+
+    public SemanticFilterOperator(final UnaryToUnaryOperator<T, T> that) {
+        super(that);
+        this.prompt = "";
+        this.targetModels = null;
+    }
+
+    @Override
+    public String getPrompt() {
+        return prompt;
+    }
+
+    @Override
+    public Set<Object> getTargetModels() {
+        return targetModels;
+    }
+
+    @Override
+    public void addTargetModel(final Object model) {
+        targetModels.add(model);
+    }
+}