feat(workflow-operator): export the source operators as Python

common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtils.scala

@@ -144,8 +144,14 @@ object AttributeTypeUtils extends Serializable {
           } else {
             str.trim.toInt
           }
-        case int: Integer               => int
-        case long: java.lang.Long       => long.toInt
+        case int: Integer         => int
+        case long: java.lang.Long => long.toInt
+        // An Arrow file states the width of its own integers, and a narrow
+        // column hands its values over as Shorts or Bytes. Without this the parse
+        // threw, the Arrow source caught it, and every value in the column
+        // arrived null.
+        case short: java.lang.Short     => short.toInt
+        case byte: java.lang.Byte       => byte.toInt
         case double: java.lang.Double   => double.toInt
         case boolean: java.lang.Boolean => if (boolean) 1 else 0
         // Timestamp and Binary are considered to be illegal here.
@@ -232,10 +238,15 @@ object AttributeTypeUtils extends Serializable {
   def parseDouble(fieldValue: Any): java.lang.Double = {
     val attempt: Try[Double] = Try {
       fieldValue match {
-        case str: String                => str.trim.toDouble
-        case int: Integer               => int.toDouble
-        case long: java.lang.Long       => long.toDouble
-        case double: java.lang.Double   => double
+        case str: String              => str.trim.toDouble
+        case int: Integer             => int.toDouble
+        case long: java.lang.Long     => long.toDouble
+        case double: java.lang.Double => double
+        // A single-precision column hands its values over as Floats, which only
+        // an Arrow file produces: Texera writes every double it owns as eight
+        // bytes. Without this the parse threw, the Arrow source caught it, and
+        // the whole column arrived null.
+        case float: java.lang.Float     => float.toDouble
         case boolean: java.lang.Boolean => if (boolean) 1 else 0
         // Timestamp and Binary are considered to be illegal here.
         case _ =>

common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala

@@ -84,11 +84,16 @@ object ArrowUtils extends LazyLogging {
               // Use the attribute type from the schema (which includes metadata)
               // instead of deriving it from the Arrow type
               val attributeType = schema.getAttributes(index).getType
-              // A timestamp is the one type whose field says more than the
-              // schema does, so it is the only one that reads the field.
+              // A timestamp and an unsigned integer are the types whose field
+              // says more than the schema does, so they are the ones that read
+              // the field.
               attributeType match {
                 case AttributeType.TIMESTAMP => wallClockOf(value, fieldVector.getField.getType)
-                case _                       => AttributeTypeUtils.parseField(value, attributeType)
+                case _ =>
+                  AttributeTypeUtils.parseField(
+                    unsignedValueOf(value, fieldVector.getField.getType),
+                    attributeType
+                  )
               }
             } catch {
               case e: Exception =>
@@ -101,6 +106,29 @@ object ArrowUtils extends LazyLogging {
       .build()
   }
 
+  /** The number an unsigned column counts to, out of the storage it counts in.
+    *
+    * Arrow's unsigned vectors hand back the raw storage, signed: `UInt1Vector` a
+    * Byte, `UInt4Vector` an Int, each reading the file's largest value as -1.
+    * `UInt2Vector` is the exception and hands back a Character, which is already
+    * the number but is no kind of integer to the parse below. Every other value
+    * passes through untouched.
+    *
+    * Nothing wider is handled because nothing wider arrives: [[toAttributeType]]
+    * refuses an unsigned 64-bit column, having no Texera type to hold it.
+    */
+  private def unsignedValueOf(value: AnyRef, arrowType: ArrowType): AnyRef =
+    arrowType match {
+      case int: ArrowType.Int if !int.getIsSigned =>
+        value match {
+          case byte: java.lang.Byte      => Int.box(byte & 0xff)
+          case char: java.lang.Character => Int.box(char.toInt)
+          case integer: Integer          => Long.box(integer.toLong & 0xffffffffL)
+          case other                     => other
+        }
+      case _ => value
+    }
+
   /** The wall clock a timestamp column holds, read the way its own field states.
     *
     * A Texera TIMESTAMP carries no zone, so the wall clock is the whole of what
@@ -182,15 +210,23 @@ object ArrowUtils extends LazyLogging {
   @throws[AttributeTypeException]
   def toAttributeType(srcType: ArrowType): AttributeType = {
     srcType match {
+      // An unsigned column counts up where its storage counts down: the largest
+      // unsigned 32-bit value is stored as -1, so read as its storage it would
+      // arrive as -1 where the file means 4294967295. The next Texera integer up
+      // holds it, and [[unsignedValueOf]] does the reading. Past 64 bits there is
+      // no next one. The same widening covers the narrow widths, Texera having no
+      // column shorter than a 32-bit integer.
       case int: ArrowType.Int =>
-        int.getBitWidth match {
-          case 16 | 32 =>
-            AttributeType.INTEGER
-
-          case 64 =>
-            AttributeType.LONG
-
-          case other =>
+        (int.getBitWidth, int.getIsSigned) match {
+          case (8 | 16 | 32, true) => AttributeType.INTEGER
+          case (8 | 16, false)     => AttributeType.INTEGER
+          case (64, true)          => AttributeType.LONG
+          case (32, false)         => AttributeType.LONG
+          case (64, false) =>
+            throw new AttributeTypeUtils.AttributeTypeException(
+              "Unsupported unsigned 64-bit Int, which is wider than any Texera column"
+            )
+          case (other, _) =>
             throw new AttributeTypeUtils.AttributeTypeException(
               s"Unsupported Int bit width: $other"
             )

common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala

@@ -50,15 +50,27 @@ class ArrowUtilsSpec extends AnyFlatSpec with Matchers {
     ArrowUtils.toAttributeType(new ArrowType.Int(64, true)) shouldBe AttributeType.LONG
   }
 
-  it should "throw AttributeTypeException for non-standard Int bit-widths" in {
-    // Only 16/32 (INTEGER) and 64 (LONG) are supported. Other widths used to
-    // be silently coerced to LONG by a `case 64 | _` catch-all; they now
-    // raise rather than masquerade as Int64.
+  it should "map every integer to the narrowest Texera type that holds it" in {
+    // Texera has no column shorter than a 32-bit integer, so a narrower width
+    // reads as one. An unsigned column needs the width above its own, counting
+    // up where its storage counts down: the largest unsigned 32-bit value is
+    // past what an INTEGER holds.
+    ArrowUtils.toAttributeType(new ArrowType.Int(8, true)) shouldBe AttributeType.INTEGER
+    ArrowUtils.toAttributeType(new ArrowType.Int(8, false)) shouldBe AttributeType.INTEGER
+    ArrowUtils.toAttributeType(new ArrowType.Int(16, false)) shouldBe AttributeType.INTEGER
+    ArrowUtils.toAttributeType(new ArrowType.Int(32, false)) shouldBe AttributeType.LONG
+  }
+
+  it should "throw AttributeTypeException for an integer no Texera type holds" in {
+    // A width above 64 used to be silently coerced to LONG by a `case 64 | _`
+    // catch-all; it raises rather than masquerade as Int64. An unsigned 64-bit
+    // column raises for the same reason, there being no width above it to read
+    // it as.
     assertThrows[AttributeTypeException] {
-      ArrowUtils.toAttributeType(new ArrowType.Int(8, true))
+      ArrowUtils.toAttributeType(new ArrowType.Int(128, true))
     }
     assertThrows[AttributeTypeException] {
-      ArrowUtils.toAttributeType(new ArrowType.Int(128, true))
+      ArrowUtils.toAttributeType(new ArrowType.Int(64, false))
     }
   }
 

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

@@ -67,21 +67,34 @@ trait StandaloneCodeGenerator {
   }
 
   /**
-    * The file's own name, for a script that reads it from its own directory
-    * rather than through Texera's resolved URI.
+    * The file this operator reads, as Texera resolved it, or None for one that
+    * reads no file.
+    *
+    * The script cannot open a resolved URI, so the body writes
+    * [[StandaloneCodeGenerator.SourceFilePlaceholder]] where the file should be
+    * named and the translator puts a name there. The name has to come from the
+    * plan: two sources reading different files whose paths end in the same
+    * segment both asked for `data.csv`, and the script read one of them twice.
+    */
+  def standaloneSourcePath(): Option[String] = None
+
+  /**
+    * The name to offer for [[standaloneSourcePath]], before the plan has had a
+    * chance to say whether another source already wants it.
     *
     * Taken from the last path segment instead of by parsing the whole string as a
     * URI: the resolver percent-encodes the file-relative segments but leaves the
     * repository and version names as the user typed them, so a dataset version
     * called `v3 - with long text` makes `new URI` throw on the space and no code
     * is generated at all.
     */
-  protected def sourceBasename(rawPath: String): String = {
-    val segment = rawPath.split("/").lastOption.getOrElse("")
-    // Percent-decoding only, matching what `URI.getPath` used to return here: form
-    // decoding would also turn a literal `+` in a file name into a space.
-    URLDecoder.decode(segment.replace("+", "%2B"), StandardCharsets.UTF_8)
-  }
+  final def standaloneSourceName(): Option[String] =
+    standaloneSourcePath().map { rawPath =>
+      val segment = rawPath.split("/").lastOption.getOrElse("")
+      // Percent-decoding only, matching what `URI.getPath` used to return here: form
+      // decoding would also turn a literal `+` in a file name into a space.
+      URLDecoder.decode(segment.replace("+", "%2B"), StandardCharsets.UTF_8)
+    }
 
   def producesDataFrame(): Boolean = true
 
@@ -110,3 +123,14 @@ trait StandaloneCodeGenerator {
     */
   def standaloneImports(): Seq[String] = Seq.empty
 }
+
+object StandaloneCodeGenerator {
+
+  /**
+    * What a source writes where the file it reads should be named.
+    *
+    * A bare identifier rather than a string literal, because the translator only
+    * rewrites the code parts of a body and leaves literals and comments alone.
+    */
+  val SourceFilePlaceholder: String = "sourceFile"
+}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDesc.scala

@@ -20,22 +20,33 @@
 package org.apache.texera.amber.operator.source.fetcher
 
 import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{OutputPort, PhysicalOp, SchemaPropagationFunc}
+import org.apache.texera.amber.operator.StandaloneCodeGenerator
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
-class URLFetcherOpDesc extends SourceOperatorDescriptor {
+class URLFetcherOpDesc extends SourceOperatorDescriptor with StandaloneCodeGenerator {
 
+  // No `pattern`: the reader is `java.net.URL`, which asks only that the value carry
+  // a scheme its JVM has a handler for. That is not something a regex can state --
+  // one excluding `www.example.com` would still pass `htp://x`, so it would advertise
+  // a validation the field does not have. `examples` offers a realistic value without
+  // claiming to constrain anything.
   @JsonProperty(required = true)
   @JsonSchemaTitle("URL")
   @JsonPropertyDescription(
     "Only accepts standard URL format"
   )
+  @JsonSchemaInject(json = """
+{
+  "examples": ["https://example.com"]
+}
+""")
   var url: String = _
 
   @JsonProperty(required = true)
@@ -87,4 +98,38 @@ class URLFetcherOpDesc extends SourceOperatorDescriptor {
       outputPorts = List(OutputPort())
     )
 
+  // The generated snippet uses `urllib.request`, which the translator's shared
+  // imports don't include. Following the per-operator convention (e.g. Split
+  // emits `import numpy as np`), the code block prepends its own import so the
+  // generated script is self-contained.
+  override def generateStandaloneCode(): String = {
+    val urlLiteral = objectMapper.writeValueAsString(url)
+    val isUtf8 = decodingMethod == DecodingMethod.UTF_8
+    // IOUtils.toString decodes through a reader that substitutes U+FFFD for a malformed
+    // byte, so the executor returns a string for any body at all. Python's decode raises
+    // instead, which turned a response the executor reads into a failed export.
+    val valueExpr =
+      if (isUtf8) """_content.decode("utf-8", errors="replace")""" else "_content"
+    val buf = scala.collection.mutable.ArrayBuffer[String]()
+    buf += "import http.client"
+    buf += "import urllib.request"
+    buf += s"_url = $urlLiteral"
+    // Catch the fetch failures, not everything: the executor guards only the fetch, so
+    // a value with no scheme stops it, and `except Exception` here would swallow that
+    // and hand back a row instead. The two are told apart by type -- a fetch raises
+    // OSError (URLError, HTTPError, TimeoutError) or HTTPException on a malformed
+    // response, a missing scheme raises ValueError.
+    // Still divergent, and not fixable this way: a scheme that merely is not RECOGNISED
+    // ("htp://x") stops the executor but reaches Python as URLError, and the two
+    // languages do not recognise the same schemes anyway -- Java takes `mailto:`,
+    // urlopen does not -- so no list of schemes is right on both sides.
+    buf += "try:"
+    buf += "    with urllib.request.urlopen(_url) as _resp:"
+    buf += "        _content = _resp.read()"
+    buf += "except (OSError, http.client.HTTPException):"
+    buf += """    _content = f"Fetch failed for URL: {_url}".encode("utf-8")"""
+    buf += s"""out1df = pd.DataFrame({"URL content": [$valueExpr]})"""
+    buf.mkString("\n")
+  }
+
 }