feat(workflow-operator): export the line sources and the URL fetcher as Python

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")
+  }
+
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala

@@ -30,13 +30,18 @@ import org.apache.texera.amber.core.workflow.{
   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.operator.source.scan.FileDecodingMethod
+import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
 import org.apache.texera.amber.operator.source.scan.text.TextSourceOpDesc
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
-class FileScanOpDesc extends SourceOperatorDescriptor with TextSourceOpDesc {
+class FileScanOpDesc
+    extends SourceOperatorDescriptor
+    with TextSourceOpDesc
+    with StandaloneCodeGenerator {
   @JsonProperty(defaultValue = "UTF_8", required = true)
   @JsonSchemaTitle("Encoding")
   var fileEncoding: FileDecodingMethod = FileDecodingMethod.UTF_8
@@ -86,4 +91,112 @@ class FileScanOpDesc extends SourceOperatorDescriptor with TextSourceOpDesc {
       inputPorts = List(InputPort(displayName = "Filename")),
       outputPorts = List(OutputPort())
     )
+
+  override def generateStandaloneCode(): String = {
+    val col = attributeName
+    val enc = fileEncoding.toString.replace("_", "-").toLowerCase
+    val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+    val encLit = pyStringLiteral(enc)
+
+    // The executor takes the row's first String field, not its first column, so a row that
+    // carries an id ahead of the path still finds the path. Reading column 0 opened the id.
+    // A row with no string at all makes the executor's `.get` throw, so this raises too
+    // rather than quietly skipping the row.
+    buf += "def _texera_file_name(row):"
+    buf += "    for _v in row:"
+    buf += "        if isinstance(_v, str):"
+    buf += "            return _v"
+    buf += """    raise ValueError(f"no file name in row: {row!r}")"""
+    buf += ""
+    buf += "_rows = []"
+    buf += "for _fn in (_texera_file_name(r) for r in in1df.itertuples(index=False)):"
+
+    // With extract on the engine reads the files INSIDE the archive, one tuple
+    // per entry, and skips the entries macOS adds. `zipfile` is the reader that
+    // matches: the platform casts what it opened to a ZipArchiveInputStream, so
+    // an archive that is not a zip fails on both sides. A directory entry is
+    // read, not skipped, there and here alike — it yields nothing.
+    val (indent, nameExpr, readWhole, lines) =
+      if (extract) {
+        buf += "    with zipfile.ZipFile(_fn) as _z:"
+        buf += "        for _name in _z.namelist():"
+        buf += """            if _name.startswith("__MACOSX"):"""
+        buf += "                continue"
+        buf += "            with _z.open(_name) as _f:"
+        // A zip entry only opens binary, so text is decoded here rather than by
+        // the reader. TextIOWrapper, not splitlines: it ends a line where
+        // `open(..., "r")` does, which is what the non-extract branch reads with.
+        (
+          " " * 16,
+          "_name",
+          if (isBinary) "_f.read()" else s"_f.read().decode($encLit)",
+          s"io.TextIOWrapper(_f, encoding=$encLit)"
+        )
+      } else {
+        // A whole file reads with `newline=""`: the engine decodes its bytes as they
+        // are, so a CRLF stays a CRLF, where the text reader's default would turn it
+        // into a LF. Lines keep the default, which ends a line where the engine does.
+        val openArgs =
+          if (isBinary) """"rb""""
+          else if (attributeType.isSingle) s""""r", encoding=$encLit, newline="""""
+          else s""""r", encoding=$encLit"""
+        buf += s"    with open(_fn, $openArgs) as _f:"
+        (" " * 8, "_fn", "_f.read()", "_f")
+      }
+
+    // Whatever the flag says, as the platform now reads it: every row carries the
+    // name of the file its value came from, a line's as much as a whole file's.
+    // See FileScanUtils.createTuplesFromFile.
+    val emitFilename = outputFileName
+
+    if (attributeType.isSingle) {
+      if (emitFilename) buf += s"${indent}_rows.append(($nameExpr, $readWhole))"
+      else buf += s"${indent}_rows.append($readWhole)"
+    } else {
+      val castExpr = attributeType match {
+        case FileAttributeType.INTEGER   => "int(l.rstrip())"
+        case FileAttributeType.LONG      => "int(l.rstrip())"
+        case FileAttributeType.DOUBLE    => "float(l.rstrip())"
+        case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
+        case FileAttributeType.TIMESTAMP => "l.rstrip()"
+        case _                           => """l.rstrip("\n")"""
+      }
+      // The slice applies to the raw lines, as the engine drops and takes
+      // before parsing: a line outside the window is never converted, so an
+      // unparseable one there costs nothing. Taking after dropping also keeps
+      // a large limit from overflowing the end index. The window is per entry,
+      // as the engine's is: it drops and takes inside the flatMap over entries.
+      val linesExpr =
+        if (fileScanOffset.isEmpty && fileScanLimit.isEmpty) lines
+        else {
+          val dropped =
+            fileScanOffset
+              .filter(_ > 0)
+              .fold(s"$lines.readlines()")(o => s"$lines.readlines()[$o:]")
+          fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+        }
+      val row = if (emitFilename) s"($nameExpr, $castExpr)" else castExpr
+      buf += s"${indent}_rows.extend($row for l in $linesExpr)"
+    }
+
+    val colLit = pyStringLiteral(col)
+    if (emitFilename) {
+      buf += s"""out1df = pd.DataFrame(_rows, columns=["filename", $colLit])"""
+    } else {
+      buf += s"""out1df = pd.DataFrame({$colLit: _rows})"""
+    }
+    buf ++= TextSourceOpDesc.parseTimestampColumn(attributeType, colLit)
+
+    buf.mkString("\n")
+  }
+
+  override def standaloneHelpers(): Seq[String] = TextSourceOpDesc.lineHelpers(attributeType)
+
+  override def standaloneImports(): Seq[String] =
+    if (!extract) Seq.empty
+    else if (attributeType.isSingle) Seq("import zipfile")
+    else Seq("import io", "import zipfile")
 }

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala

@@ -29,13 +29,23 @@ 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.{PhysicalOp, SchemaPropagationFunc}
+import org.apache.texera.amber.operator.StandaloneCodeGenerator
+import org.apache.texera.amber.operator.StandaloneCodeGenerator.SourceFilePlaceholder
 import org.apache.texera.amber.operator.metadata.annotations.HideAnnotation
 import org.apache.texera.amber.operator.source.scan.text.TextSourceOpDesc
-import org.apache.texera.amber.operator.source.scan.{FileDecodingMethod, ScanSourceOpDesc}
+import org.apache.texera.amber.operator.source.scan.{
+  FileAttributeType,
+  FileDecodingMethod,
+  ScanSourceOpDesc
+}
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 @JsonIgnoreProperties(value = Array("limit", "offset", "fileEncoding"))
-class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc {
+class FileScanSourceOpDesc
+    extends ScanSourceOpDesc
+    with TextSourceOpDesc
+    with StandaloneCodeGenerator {
   @JsonProperty(defaultValue = "UTF_8", required = true)
   @JsonSchemaTitle("Encoding")
   @JsonSchemaInject(
@@ -68,6 +78,107 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc {
 
   fileTypeName = Option("")
 
+  override def standaloneSourcePath(): Option[String] = fileName
+
+  override def generateStandaloneCode(): String = {
+    val col = attributeName
+    // `encoding` is the charset the panel offers, and the one the executor now
+    // decodes with.
+    val enc = encoding.toString.replace("_", "-").toLowerCase
+    val colLit = pyStringLiteral(col)
+    val encLit = pyStringLiteral(enc)
+    val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+
+    val castExpr = attributeType match {
+      case FileAttributeType.INTEGER   => "int(l.rstrip())"
+      case FileAttributeType.LONG      => "int(l.rstrip())"
+      case FileAttributeType.DOUBLE    => "float(l.rstrip())"
+      case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
+      case FileAttributeType.TIMESTAMP => "l.rstrip()"
+      case _                           => """l.rstrip("\n")"""
+    }
+
+    // The slice applies to the raw lines, as the engine drops and takes before
+    // parsing: a line outside the window is never converted, so an unparseable
+    // one there costs nothing. Taking after dropping also keeps a large limit
+    // from overflowing the end index. With extract on, the window is per entry,
+    // as the engine's is: it drops and takes inside the flatMap over entries.
+    def windowed(lines: String): String =
+      if (fileScanOffset.isEmpty && fileScanLimit.isEmpty) lines
+      else {
+        val dropped =
+          fileScanOffset.filter(_ > 0).fold(s"$lines.readlines()")(o => s"$lines.readlines()[$o:]")
+        fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+      }
+
+    // Whatever the flag says, as the platform now reads it: every row carries
+    // the name of the file its value came from, a line's as much as a whole
+    // file's. See FileScanUtils.createTuplesFromFile.
+    val emitFilename = outputFileName
+
+    if (extract) {
+      // The engine reads the files INSIDE the archive, one tuple per entry, and
+      // skips the entries macOS adds. `zipfile` is the reader that matches: the
+      // platform casts what it opened to a ZipArchiveInputStream, so an archive
+      // that is not a zip fails on both sides. A directory entry is read, not
+      // skipped, there and here alike — it yields nothing. The filename column
+      // carries the ENTRY's name, which is what the engine puts there.
+      buf += "_rows = []"
+      buf += s"with zipfile.ZipFile($SourceFilePlaceholder) as _z:"
+      buf += "    for _name in _z.namelist():"
+      buf += """        if _name.startswith("__MACOSX"):"""
+      buf += "            continue"
+      buf += "        with _z.open(_name) as _f:"
+      if (attributeType.isSingle) {
+        // A zip entry only opens binary, so text is decoded here rather than by
+        // the reader.
+        val readWhole = if (isBinary) "_f.read()" else s"_f.read().decode($encLit)"
+        if (emitFilename) buf += s"            _rows.append((_name, $readWhole))"
+        else buf += s"            _rows.append($readWhole)"
+      } else {
+        // TextIOWrapper, not splitlines: it ends a line where `open(..., "r")`
+        // does, which is what the branch below reads with.
+        val linesExpr = windowed(s"io.TextIOWrapper(_f, encoding=$encLit)")
+        val row = if (emitFilename) s"(_name, $castExpr)" else castExpr
+        buf += s"            _rows.extend($row for l in $linesExpr)"
+      }
+      if (emitFilename) buf += s"""out1df = pd.DataFrame(_rows, columns=["filename", $colLit])"""
+      else buf += s"""out1df = pd.DataFrame({$colLit: _rows})"""
+    } else if (attributeType.isSingle) {
+      // `newline=""`: the engine decodes the file's bytes as they are, so a CRLF
+      // stays a CRLF. The text reader's default would turn it into a LF.
+      val openArgs =
+        if (isBinary) s"""$SourceFilePlaceholder, "rb""""
+        else s"""$SourceFilePlaceholder, "r", encoding=$encLit, newline="""""
+      val dfCols =
+        if (emitFilename) s"""{"filename": $SourceFilePlaceholder, $colLit: [_f.read()]}"""
+        else s"""{$colLit: [_f.read()]}"""
+      buf += s"""with open($openArgs) as _f:"""
+      buf += s"""    out1df = pd.DataFrame($dfCols)"""
+    } else {
+      val linesExpr = windowed("_f")
+      val dfCols =
+        if (emitFilename)
+          s"""{"filename": $SourceFilePlaceholder, $colLit: [$castExpr for l in $linesExpr]}"""
+        else s"""{$colLit: [$castExpr for l in $linesExpr]}"""
+      buf += s"""with open($SourceFilePlaceholder, "r", encoding=$encLit) as _f:"""
+      buf += s"""    out1df = pd.DataFrame($dfCols)"""
+    }
+    buf ++= TextSourceOpDesc.parseTimestampColumn(attributeType, colLit)
+
+    buf.mkString("\n")
+  }
+
+  override def standaloneHelpers(): Seq[String] = TextSourceOpDesc.lineHelpers(attributeType)
+
+  override def standaloneImports(): Seq[String] =
+    if (!extract) Seq.empty
+    else if (attributeType.isSingle) Seq("import zipfile")
+    else Seq("import io", "import zipfile")
+
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala

@@ -138,20 +138,27 @@ private[file] object FileScanUtils {
             TupleLike(fields.toSeq: _*)
         }
       } else {
-        fileEntries.flatMap { entry =>
-          val lines = new BufferedReader(new InputStreamReader(entry, fileEncoding.getCharset))
-            .lines()
-            .iterator()
-            .asScala
-            .drop(fileScanOffset.getOrElse(0))
-          fileScanLimit
-            .fold(lines)(lines.take)
-            .map(line =>
-              TupleLike(attributeType match {
-                case FileAttributeType.SINGLE_STRING => line
-                case _                               => parseField(line, attributeType.getType)
-              })
-            )
+        // Paired with the entry names the same way the single-value branch is:
+        // a row carries the name of the file its line came from, which is what
+        // the schema declares when the filename was asked for. Emitting the value
+        // alone left a one-field row against a two-column schema, and the tuple
+        // could not be built at all.
+        fileEntries.zipAll(filenameIt, null, null).flatMap {
+          case (entry, entryFileName) =>
+            val lines = new BufferedReader(new InputStreamReader(entry, fileEncoding.getCharset))
+              .lines()
+              .iterator()
+              .asScala
+              .drop(fileScanOffset.getOrElse(0))
+            fileScanLimit
+              .fold(lines)(lines.take)
+              .map { line =>
+                val value = attributeType match {
+                  case FileAttributeType.SINGLE_STRING => line
+                  case _                               => parseField(line, attributeType.getType)
+                }
+                if (outputFileName) TupleLike(entryFileName, value) else TupleLike(value)
+              }
         }
       }