[Hackathon] feat: add image-native workflows

agent-service/src/agent/prompts.test.ts

@@ -0,0 +1,51 @@
+/**
+ * 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.
+ */
+
+import { describe, expect, test } from "bun:test";
+import { buildSystemPrompt } from "./prompts";
+import { WorkflowSystemMetadata } from "./util/workflow-system-metadata";
+
+describe("buildSystemPrompt", () => {
+  test("includes both operator type and display name", () => {
+    const metadata = new WorkflowSystemMetadata();
+    metadata.loadFromMetadata({
+      operators: [
+        {
+          operatorType: "SmartFileScan",
+          operatorVersion: "1",
+          jsonSchema: { properties: { fileName: { type: "string" } }, required: ["fileName"] },
+          additionalMetadata: {
+            userFriendlyName: "Smart Source",
+            operatorGroupName: "Data Input",
+            operatorDescription: "Auto-detects files and folders.",
+            inputPorts: [],
+            outputPorts: [{}],
+          },
+        },
+      ],
+      groups: [],
+    });
+
+    const prompt = buildSystemPrompt(metadata, ["SmartFileScan"]);
+
+    expect(prompt).toContain("## SmartFileScan");
+    expect(prompt).toContain("Display name: Smart Source");
+    expect(prompt).toContain("Description: Auto-detects files and folders.");
+  });
+});

agent-service/src/agent/prompts.ts

@@ -268,10 +268,12 @@ function buildAllowedOperatorSchemas(
   for (const operatorType of operatorTypes) {
     const compactSchema = metadataStore.getCompactSchema(operatorType);
     const description = metadataStore.getDescription(operatorType);
+    const displayName = metadataStore.getAdditionalMetadata(operatorType)?.userFriendlyName;
 
     if (compactSchema) {
       schemas.push(
         `## ${operatorType}\n` +
+          (displayName ? `Display name: ${displayName}\n` : "") +
           (description ? `Description: ${description}\n` : "") +
           `Schema:\n\`\`\`json\n${JSON.stringify(compactSchema, null, 2)}\n\`\`\``
       );

agent-service/src/types/agent.test.ts

@@ -0,0 +1,27 @@
+/**
+ * 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.
+ */
+
+import { describe, expect, test } from "bun:test";
+import { DEFAULT_AGENT_SETTINGS } from "./agent";
+
+describe("DEFAULT_AGENT_SETTINGS", () => {
+  test("allows the smart source operator by default", () => {
+    expect(DEFAULT_AGENT_SETTINGS.allowedOperatorTypes).toContain("SmartFileScan");
+  });
+});

agent-service/src/types/agent.ts

@@ -87,6 +87,7 @@ export const DEFAULT_AGENT_SETTINGS: Omit<AgentSettings, "systemPrompt"> = {
   executionTimeoutMs: 240000,
   maxSteps: 100,
   allowedOperatorTypes: [
+    "SmartFileScan",
     "CSVFileScan",
     "Filter",
     "Projection",

amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala

@@ -130,6 +130,7 @@ class TexeraWebApplication
     environment.servlets.setSessionHandler(new SessionHandler)
 
     environment.jersey.register(classOf[SystemMetadataResource])
+    environment.jersey.register(classOf[SmartFileInferenceResource])
     // environment.jersey().register(classOf[MockKillWorkerResource])
 
     environment.jersey.register(classOf[HealthCheckResource])

amber/src/main/scala/org/apache/texera/web/resource/SmartFileInferenceResource.scala

@@ -0,0 +1,114 @@
+/*
+ * 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.resource
+
+import com.fasterxml.jackson.annotation.{JsonIgnoreProperties, JsonProperty}
+import org.apache.texera.amber.core.storage.FileResolver
+import org.apache.texera.amber.operator.source.scan.FileDecodingMethod
+import org.apache.texera.amber.operator.source.scan.smart.{
+  InferenceOverrides,
+  SmartFileFormat,
+  SmartFileInferencer
+}
+
+import javax.annotation.security.RolesAllowed
+import javax.ws.rs.core.MediaType
+import javax.ws.rs.{Consumes, POST, Path, Produces}
+import scala.jdk.CollectionConverters._
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+case class SmartFileInferenceRequest(
+    @JsonProperty("fileName") fileName: String,
+    @JsonProperty("fileEncoding") fileEncoding: Option[String] = None,
+    @JsonProperty("formatOverride") formatOverride: Option[String] = None,
+    @JsonProperty("customDelimiter") customDelimiter: Option[String] = None,
+    @JsonProperty("hasHeader") hasHeader: Option[Boolean] = None,
+    @JsonProperty("sheetName") sheetName: Option[String] = None,
+    @JsonProperty("flatten") flatten: Option[Boolean] = None
+)
+
+case class SmartFileInferenceColumn(name: String, `type`: String)
+
+case class SmartFileInferenceResponse(
+    detectedFormat: String,
+    schema: java.util.List[SmartFileInferenceColumn],
+    customDelimiter: String,
+    hasHeader: java.lang.Boolean,
+    sheetName: String,
+    availableSheetNames: java.util.List[String],
+    flatten: java.lang.Boolean,
+    isFolder: Boolean,
+    fileCount: Int
+)
+
+@Path("/file-inference")
+@RolesAllowed(Array("REGULAR", "ADMIN"))
+@Consumes(Array(MediaType.APPLICATION_JSON))
+@Produces(Array(MediaType.APPLICATION_JSON))
+class SmartFileInferenceResource {
+
+  @POST
+  @Path("/preview")
+  def preview(request: SmartFileInferenceRequest): SmartFileInferenceResponse = {
+    val uri = FileResolver.resolve(request.fileName)
+    val charset = request.fileEncoding
+      .flatMap(name => tryParseEncoding(name))
+      .getOrElse(FileDecodingMethod.UTF_8.getCharset)
+
+    val overrides = InferenceOverrides(
+      format = request.formatOverride.flatMap(s => tryParseFormat(s)),
+      delimiter = request.customDelimiter.flatMap(_.headOption),
+      hasHeader = request.hasHeader,
+      sheetName = request.sheetName,
+      flatten = request.flatten
+    )
+
+    val result = SmartFileInferencer.infer(uri, charset, overrides)
+    val columns = result.schema.getAttributes
+      .map(a => SmartFileInferenceColumn(a.getName, a.getType.toString))
+      .asJava
+
+    SmartFileInferenceResponse(
+      detectedFormat = result.format.getLabel,
+      schema = columns,
+      customDelimiter = result.csvDelimiter.orNull,
+      hasHeader = result.csvHasHeader.map(java.lang.Boolean.valueOf).orNull,
+      sheetName = result.sheetName.orNull,
+      availableSheetNames = result.availableSheetNames.asJava,
+      flatten = result.flatten.map(java.lang.Boolean.valueOf).orNull,
+      isFolder = result.isFolder,
+      fileCount = result.fileCount
+    )
+  }
+
+  private def tryParseFormat(value: String): Option[SmartFileFormat] = {
+    val upper = value.toUpperCase
+    // Accept both the enum name (CSV, TSV, ...) and the user-facing label ("Plain text", ...).
+    try Some(SmartFileFormat.valueOf(upper))
+    catch {
+      case _: IllegalArgumentException =>
+        SmartFileFormat.values().find(_.getLabel.equalsIgnoreCase(value))
+    }
+  }
+
+  private def tryParseEncoding(value: String): Option[java.nio.charset.Charset] =
+    try Some(FileDecodingMethod.valueOf(value.toUpperCase).getCharset)
+    catch { case _: IllegalArgumentException => None }
+}

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

@@ -46,6 +46,7 @@ import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAg
 import org.apache.texera.amber.engine.common.AmberRuntime
 import org.apache.texera.amber.engine.common.client.AmberClient
 import org.apache.texera.amber.engine.common.executionruntimestate.ExecutionMetadataStore
+import org.apache.texera.amber.util.ImageFormatUtils
 import org.apache.texera.web.SubscriptionManager
 import org.apache.texera.web.model.websocket.event.{
   PaginatedResultEvent,
@@ -59,6 +60,7 @@ import org.apache.texera.web.service.WorkflowExecutionService.getLatestExecution
 import org.apache.texera.web.storage.{ExecutionStateStore, WorkflowStateStore}
 
 import java.lang.Byte.{SIZE => BitsPerByte}
+import java.util.Base64
 import java.util.UUID
 import scala.collection.mutable
 import scala.concurrent.duration.DurationInt
@@ -76,6 +78,11 @@ object ExecutionResultService {
       )
       .mkString("")
 
+  private def bytesToImageDataUrl(bytes: Array[Byte]): Option[String] =
+    ImageFormatUtils
+      .detectMimeType(bytes)
+      .map(mimeType => s"data:$mimeType;base64,${Base64.getEncoder.encodeToString(bytes)}")
+
   /**
     * Converts a collection of Tuples to a list of JSON ObjectNodes.
     *
@@ -107,25 +114,27 @@ object ExecutionResultService {
                   case AttributeType.BINARY =>
                     value match {
                       case byteArray: Array[Byte] =>
-                        val totalSize = byteArray.length
-                        val sizeFormatted = f"$totalSize%,d"
-                        val totalBits = totalSize * BitsPerByte
-                        val preview =
-                          if (totalBits <= binaryPreviewLeadingBits + binaryPreviewTrailingBits)
-                            bytesToBinaryString(byteArray)
-                          else {
-                            val leadingBytesNeeded =
-                              math.ceil(binaryPreviewLeadingBits.toDouble / BitsPerByte).toInt
-                            val trailingBytesNeeded =
-                              math.ceil(binaryPreviewTrailingBits.toDouble / BitsPerByte).toInt
-                            val leading = bytesToBinaryString(byteArray.take(leadingBytesNeeded))
-                              .take(binaryPreviewLeadingBits)
-                            val trailing = bytesToBinaryString(
-                              byteArray.takeRight(trailingBytesNeeded)
-                            ).takeRight(binaryPreviewTrailingBits)
-                            s"$leading...$trailing"
-                          }
-                        s"<binary $preview, size = $sizeFormatted bytes>"
+                        bytesToImageDataUrl(byteArray).getOrElse {
+                          val totalSize = byteArray.length
+                          val sizeFormatted = f"$totalSize%,d"
+                          val totalBits = totalSize * BitsPerByte
+                          val preview =
+                            if (totalBits <= binaryPreviewLeadingBits + binaryPreviewTrailingBits)
+                              bytesToBinaryString(byteArray)
+                            else {
+                              val leadingBytesNeeded =
+                                math.ceil(binaryPreviewLeadingBits.toDouble / BitsPerByte).toInt
+                              val trailingBytesNeeded =
+                                math.ceil(binaryPreviewTrailingBits.toDouble / BitsPerByte).toInt
+                              val leading = bytesToBinaryString(byteArray.take(leadingBytesNeeded))
+                                .take(binaryPreviewLeadingBits)
+                              val trailing = bytesToBinaryString(
+                                byteArray.takeRight(trailingBytesNeeded)
+                              ).takeRight(binaryPreviewTrailingBits)
+                              s"$leading...$trailing"
+                            }
+                          s"<binary $preview, size = $sizeFormatted bytes>"
+                        }
 
                       case _ =>
                         throw new RuntimeException(

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

@@ -23,6 +23,10 @@ import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tup
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.awt.image.BufferedImage
+import java.io.ByteArrayOutputStream
+import javax.imageio.ImageIO
+
 class ExecutionResultServiceSpec extends AnyFlatSpec with Matchers {
 
   "convertTuplesToJson" should "convert tuples with various field types correctly" in {
@@ -181,6 +185,24 @@ class ExecutionResultServiceSpec extends AnyFlatSpec with Matchers {
     emptyBinaryString should include("size = 0 bytes")
   }
 
+  it should "serialize recognized image binaries as data URLs" in {
+    val attributes = List(
+      new Attribute("image", AttributeType.BINARY)
+    )
+    val schema = new Schema(attributes)
+    val imageBytes = pngBytes(width = 2, height = 2)
+
+    val tuple = Tuple
+      .builder(schema)
+      .add("image", AttributeType.BINARY, imageBytes)
+      .build()
+
+    val result = ExecutionResultService.convertTuplesToJson(List(tuple))
+
+    result should have size 1
+    result.head.get("image").asText() should startWith("data:image/png;base64,")
+  }
+
   it should "handle binary data with single ByteBuffer" in {
     val attributes = List(
       new Attribute("singleBufferBinary", AttributeType.BINARY)
@@ -475,4 +497,11 @@ class ExecutionResultServiceSpec extends AnyFlatSpec with Matchers {
     resultsDefault(2).get("value").asText() shouldBe "medium length"
     resultsDefault(3).get("value").asText() should endWith("...")
   }
+
+  private def pngBytes(width: Int, height: Int): Array[Byte] = {
+    val image = new BufferedImage(width, height, BufferedImage.TYPE_INT_RGB)
+    val out = new ByteArrayOutputStream()
+    ImageIO.write(image, "png", out)
+    out.toByteArray
+  }
 }

build.sbt

@@ -50,6 +50,19 @@ lazy val asfLicensingSettingsWithVendored = AddMetaInfLicenseFiles.workflowOpera
 
 val jacksonVersion = "2.18.6"
 
+// Globally exclude transitive Hadoop landmines that conflict with Texera's
+// Dropwizard + Jersey stack. These ride in via Parquet's `parquet-hadoop`,
+// added in common/workflow-operator/build.sbt for SmartFileScan. Defining the
+// excludes at ThisBuild level ensures they apply to every project that
+// transitively pulls Hadoop — most importantly amber.
+ThisBuild / excludeDependencies ++= Seq(
+  ExclusionRule("javax.servlet.jsp", "jsp-api"),
+  ExclusionRule("javax.servlet", "servlet-api"),
+  ExclusionRule(organization = "com.sun.jersey"),
+  ExclusionRule(organization = "com.sun.jersey.contribs"),
+  ExclusionRule("com.github.pjfanning", "jersey-json")
+)
+
 lazy val DAO = (project in file("common/dao")).settings(asfLicensingSettings)
 lazy val Config = (project in file("common/config")).settings(asfLicensingSettings)
 lazy val Auth = (project in file("common/auth"))