Merge branch 'main' into batch-resumable-upload

Author: carloea2

.github/dependabot.yml

@@ -50,7 +50,7 @@ import org.apache.texera.amber.error.ErrorUtils.{
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.User
 import org.apache.texera.service.util.LargeBinaryManager
-import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent
+import org.apache.texera.web.model.websocket.event.{TexeraWebSocketEvent, WorkflowErrorEvent}
 import org.apache.texera.web.model.websocket.request.WorkflowExecuteRequest
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource
 import org.apache.texera.web.service.WorkflowService.mkWorkflowStateId
@@ -277,6 +277,14 @@ class WorkflowService(
         }
       }
     }
+    // Once the execution is published via `executionService.onNext`, the normal
+    // state-store path surfaces fatal errors to the UI: `errorHandler` writes
+    // them into `executionStateStore.metadataStore`, whose diff handler (set up
+    // in the WorkflowExecutionService constructor) emits a WorkflowErrorEvent
+    // that `connectToExecution` forwards. Before that point, neither the emitter
+    // nor a subscriber exists yet, so a failure in the constructor itself would
+    // be recorded but never reach the frontend -- see the fallback in `catch`.
+    var executionPublished = false
     try {
       val execution = new WorkflowExecutionService(
         controllerConf,
@@ -290,13 +298,36 @@ class WorkflowService(
       )
       lifeCycleManager.registerCleanUpOnStateChange(executionStateStore)
       executionService.onNext(execution)
+      executionPublished = true
       execution.executeWorkflow()
     } catch {
-      case e: Throwable => errorHandler(e)
+      case e: Throwable =>
+        errorHandler(e)
+        // If the execution was never published, no `connectToExecution`
+        // subscriber is bound to `executionStateStore`, so the state-store path
+        // above cannot deliver the error. Push it directly in that pre-publish
+        // window only; once published, the state-store path already surfaces it
+        // (pushing here too would double-emit).
+        if (!executionPublished) {
+          reportFatalErrorsToSubscribers(executionStateStore)
+        }
     }
 
   }
 
+  /**
+    * Push the fatal errors currently recorded in `stateStore` to connected
+    * websocket subscribers (via `errorSubject`).
+    *
+    * Fallback used only when execution initialization fails before the execution
+    * is published (e.g. the WorkflowExecutionService constructor throws): in that
+    * window the per-execution state store has no diff-handler emitter and no
+    * websocket subscriber, so the error -- already recorded by `errorHandler` --
+    * would otherwise be logged but never reach the frontend.
+    */
+  private[service] def reportFatalErrorsToSubscribers(stateStore: ExecutionStateStore): Unit =
+    errorSubject.onNext(WorkflowErrorEvent(stateStore.metadataStore.getState.fatalErrors))
+
   def convertToJson(frontendVersion: String): String = {
     val environmentVersionMap = Map(
       "engine_version" -> Json.toJson(frontendVersion)

amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala

@@ -0,0 +1,88 @@
+/*
+ * 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.texera.amber.engine.architecture.messaginglayer
+
+import org.apache.pekko.actor.{ActorSystem, DeadLetter, Props}
+import org.apache.pekko.testkit.{ImplicitSender, TestKit, TestProbe}
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, TupleLike}
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.engine.architecture.common.WorkflowActor.{
+  MessageBecomesDeadLetter,
+  NetworkMessage
+}
+import org.apache.texera.amber.engine.common.ambermessage.{DataFrame, WorkflowFIFOMessage}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpecLike
+
+import scala.concurrent.duration.DurationInt
+
+class DeadLetterMonitorActorSpec
+    extends TestKit(ActorSystem("DeadLetterMonitorActorSpec"))
+    with ImplicitSender
+    with AnyFlatSpecLike
+    with BeforeAndAfterAll {
+
+  override def afterAll(): Unit = {
+    TestKit.shutdownActorSystem(system)
+  }
+
+  private val channelId =
+    ChannelIdentity(
+      ActorVirtualIdentity("sender"),
+      ActorVirtualIdentity("receiver"),
+      isControl = false
+    )
+
+  private def aNetworkMessage(): NetworkMessage = {
+    val payload = DataFrame(
+      Array(TupleLike(1) enforceSchema Schema().add("field1", AttributeType.INTEGER))
+    )
+    NetworkMessage(0, WorkflowFIFOMessage(channelId, 0, payload))
+  }
+
+  "DeadLetterMonitorActor" should "forward MessageBecomesDeadLetter to the original sender for a NetworkMessage dead letter" in {
+    val monitor = system.actorOf(Props(new DeadLetterMonitorActor()))
+    val originalSender = TestProbe()
+    val recipient = TestProbe()
+    val message = aNetworkMessage()
+
+    monitor ! DeadLetter(message, originalSender.ref, recipient.ref)
+
+    originalSender.expectMsg(MessageBecomesDeadLetter(message))
+  }
+
+  it should "ignore a dead letter whose payload is not a NetworkMessage" in {
+    val monitor = system.actorOf(Props(new DeadLetterMonitorActor()))
+    val originalSender = TestProbe()
+    val recipient = TestProbe()
+
+    monitor ! DeadLetter("not a network message", originalSender.ref, recipient.ref)
+
+    originalSender.expectNoMessage(200.millis)
+  }
+
+  it should "ignore messages that are not dead letters" in {
+    val monitor = system.actorOf(Props(new DeadLetterMonitorActor()))
+
+    monitor ! "some unrelated message"
+
+    expectNoMessage(200.millis)
+  }
+}

amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/DeadLetterMonitorActorSpec.scala

@@ -0,0 +1,85 @@
+/*
+ * 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.texera.web.service
+
+import com.google.protobuf.timestamp.Timestamp
+import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity
+import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.EXECUTION_FAILURE
+import org.apache.texera.amber.core.workflowruntimestate.WorkflowFatalError
+import org.apache.texera.web.model.websocket.event.{TexeraWebSocketEvent, WorkflowErrorEvent}
+import org.apache.texera.web.storage.ExecutionStateStore
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Instant
+import scala.collection.mutable.ArrayBuffer
+
+/**
+  * Unit tests for `WorkflowService.reportFatalErrorsToSubscribers`, the seam
+  * that surfaces init-time fatal errors to the websocket. When execution
+  * initialization fails, the error is recorded in the metadata store; this push
+  * is what makes it visible to connected clients instead of only logged.
+  */
+class WorkflowServiceSpec extends AnyFlatSpec with Matchers {
+
+  private def fatalError(message: String): WorkflowFatalError =
+    WorkflowFatalError(EXECUTION_FAILURE, Timestamp(Instant.now), message, "", "", "")
+
+  /** A WorkflowService with a subscriber collecting every event it pushes. */
+  private def serviceWithCollector(): (WorkflowService, ArrayBuffer[TexeraWebSocketEvent]) = {
+    val service = new WorkflowService(WorkflowIdentity(1), computingUnitId = 1, cleanUpTimeout = 30)
+    val events = ArrayBuffer.empty[TexeraWebSocketEvent]
+    service.connect(evt => events += evt)
+    (service, events)
+  }
+
+  private def errorEventsIn(events: ArrayBuffer[TexeraWebSocketEvent]): Seq[WorkflowErrorEvent] =
+    events.collect { case e: WorkflowErrorEvent => e }.toSeq
+
+  "WorkflowService" should
+    "push a WorkflowErrorEvent carrying the store's fatal error to connected subscribers" in {
+    val (service, events) = serviceWithCollector()
+    val store = new ExecutionStateStore()
+    val err = fatalError("boom during init")
+    store.metadataStore.updateState(_.addFatalErrors(err))
+
+    service.reportFatalErrorsToSubscribers(store)
+
+    val errorEvents = errorEventsIn(events)
+    errorEvents should have size 1
+    // Forwards exactly the store's fatal errors -- no more, no less.
+    errorEvents.head.fatalErrors should contain theSameElementsAs Seq(err)
+  }
+
+  it should "carry every fatal error currently recorded in the store" in {
+    val (service, events) = serviceWithCollector()
+    val store = new ExecutionStateStore()
+    val first = fatalError("first")
+    val second = fatalError("second")
+    store.metadataStore.updateState(_.addFatalErrors(first).addFatalErrors(second))
+
+    service.reportFatalErrorsToSubscribers(store)
+
+    val errorEvents = errorEventsIn(events)
+    errorEvents should have size 1
+    // Exactly the two recorded errors -- no extras.
+    errorEvents.head.fatalErrors should contain theSameElementsAs Seq(first, second)
+  }
+}

amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala

@@ -30,6 +30,7 @@ import org.apache.texera.amber.operator.huggingFace.codegen.{
   ImageTaskCodegen,
   MediaGenCodegen,
   PythonCodegenBase,
+  QaRankingCodegen,
   TaskCodegen,
   TextGenCodegen
 }
@@ -108,6 +109,25 @@ class HuggingFaceInferenceOpDesc extends PythonOperatorDescriptor {
   @AutofillAttributeName
   var inputAudioColumn: EncodableString = ""
 
+  @JsonProperty(value = "contextColumn", required = false)
+  @JsonSchemaTitle("Context Column")
+  @JsonPropertyDescription("Column containing the context passage for question answering")
+  @AutofillAttributeName
+  var contextColumn: EncodableString = ""
+
+  @JsonProperty(value = "candidateLabels", required = false)
+  @JsonSchemaTitle("Candidate Labels")
+  @JsonPropertyDescription("Comma-separated candidate labels for zero-shot classification")
+  var candidateLabels: EncodableString = ""
+
+  @JsonProperty(value = "sentencesColumn", required = false)
+  @JsonSchemaTitle("Sentences Column")
+  @JsonPropertyDescription(
+    "Column with comma-separated sentences for sentence similarity and text ranking"
+  )
+  @AutofillAttributeName
+  var sentencesColumn: EncodableString = ""
+
   @JsonProperty(
     value = "systemPrompt",
     required = false,
@@ -153,6 +173,7 @@ class HuggingFaceInferenceOpDesc extends PythonOperatorDescriptor {
     ImageTaskCodegen.tasks.foreach(t => byTask += (t -> ImageTaskCodegen))
     AudioTaskCodegen.tasks.foreach(t => byTask += (t -> AudioTaskCodegen))
     MediaGenCodegen.tasks.foreach(t => byTask += (t -> MediaGenCodegen))
+    QaRankingCodegen.tasks.foreach(t => byTask += (t -> QaRankingCodegen))
     byTask.toMap
   }
 
@@ -200,6 +221,12 @@ class HuggingFaceInferenceOpDesc extends PythonOperatorDescriptor {
       if (audioInput == null) "" else audioInput
     val safeInputAudioColumn: EncodableString =
       if (inputAudioColumn == null) "" else inputAudioColumn
+    val safeContextColumn: EncodableString =
+      if (contextColumn == null) "" else contextColumn
+    val safeCandidateLabels: EncodableString =
+      if (candidateLabels == null) "" else candidateLabels
+    val safeSentencesColumn: EncodableString =
+      if (sentencesColumn == null) "" else sentencesColumn
 
     val ctx = CodegenContext(
       hfApiToken = safeToken,
@@ -213,7 +240,10 @@ class HuggingFaceInferenceOpDesc extends PythonOperatorDescriptor {
       imageInput = safeImageInput,
       inputImageColumn = safeInputImageColumn,
       audioInput = safeAudioInput,
-      inputAudioColumn = safeInputAudioColumn
+      inputAudioColumn = safeInputAudioColumn,
+      contextColumn = safeContextColumn,
+      candidateLabels = safeCandidateLabels,
+      sentencesColumn = safeSentencesColumn
     )
 
     PythonCodegenBase.render(ctx, codegenForTask(safeTask))

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDesc.scala

@@ -59,6 +59,9 @@ object PythonCodegenBase {
     val inputImageColumn = ctx.inputImageColumn
     val audioInput = ctx.audioInput
     val inputAudioColumn = ctx.inputAudioColumn
+    val contextColumn = ctx.contextColumn
+    val candidateLabels = ctx.candidateLabels
+    val sentencesColumn = ctx.sentencesColumn
     pyb"""import os
        |import re
        |import json
@@ -141,6 +144,9 @@ object PythonCodegenBase {
        |        self.INPUT_IMAGE_COLUMN = $inputImageColumn
        |        self.AUDIO_INPUT = $audioInput
        |        self.INPUT_AUDIO_COLUMN = $inputAudioColumn
+       |        self.CONTEXT_COLUMN = $contextColumn
+       |        self.CANDIDATE_LABELS = $candidateLabels
+       |        self.SENTENCES_COLUMN = $sentencesColumn
        |
        |    def _resolve_providers(self, token):
        |        '''Query the HF Hub API for inference providers serving this model.
@@ -491,6 +497,24 @@ object PythonCodegenBase {
        |                f"Prompt column '{prompt_col}' not found in input table. "
        |                f"Available columns: {list(table.columns)}"
        |            )
+       |        if task == "zero-shot-classification":
+       |            labels = [l.strip() for l in str(self.CANDIDATE_LABELS).split(",") if l.strip()]
+       |            assert labels, (
+       |                "Candidate Labels are required for zero-shot-classification. "
+       |                "Provide a comma-separated list of labels."
+       |            )
+       |        if task == "question-answering":
+       |            ctx_col = self.CONTEXT_COLUMN
+       |            assert ctx_col and ctx_col in table.columns, (
+       |                f"Context column '{ctx_col}' not found in input table. "
+       |                f"Available columns: {list(table.columns)}"
+       |            )
+       |        if task in ("sentence-similarity", "text-ranking"):
+       |            sent_col = self.SENTENCES_COLUMN
+       |            assert sent_col and sent_col in table.columns, (
+       |                f"Sentences column '{sent_col}' not found in input table. "
+       |                f"Available columns: {list(table.columns)}"
+       |            )
        |
        |        # --- handle empty table ---
        |        if table.empty:
@@ -506,6 +530,16 @@ object PythonCodegenBase {
        |            "Authorization": f"Bearer {token}",
        |            "Content-Type": "application/octet-stream",
        |        }
+       |        # --- pre-compute table dict for table-question-answering ---
+       |        table_dict = None
+       |        if task == "table-question-answering":
+       |            table_dict = {}
+       |            for col in table.columns:
+       |                if col != prompt_col and col != result_col:
+       |                    table_dict[col] = [
+       |                        str(v) if not pd.isna(v) else "" for v in table[col].tolist()
+       |                    ]
+       |
        |        # --- resolve image source (upload or column) for image tasks ---
        |        has_image_upload = bool(self.IMAGE_INPUT) and bool(str(self.IMAGE_INPUT).strip())
        |        use_image_column = not has_image_upload and bool(self.INPUT_IMAGE_COLUMN) and self.INPUT_IMAGE_COLUMN in table.columns