From efb2dfcd7af4d90dd75a0943635a6c823df345ad Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Sat, 26 Sep 2026 20:14:33 -0700
Subject: [PATCH 1/5] feat(operator): export the line sources and the URL
 fetcher

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../source/fetcher/URLFetcherOpDesc.scala     |  49 ++++-
 .../source/scan/file/FileScanOpDesc.scala     | 111 ++++++++++-
 .../scan/file/FileScanSourceOpDesc.scala      | 113 ++++++++++-
 .../source/scan/file/FileScanUtils.scala      |  35 ++--
 .../scan/text/TextInputSourceOpDesc.scala     |  47 ++++-
 .../source/scan/text/TextSourceOpDesc.scala   |  47 +++++
 .../source/fetcher/URLFetcherOpDescSpec.scala |  43 +++++
 .../source/scan/file/FileScanOpDescSpec.scala | 175 +++++++++++++++++-
 .../scan/file/FileScanSourceOpDescSpec.scala  |  21 ++-
 .../scan/text/TextInputSourceOpDescSpec.scala | 109 ++++++++++-
 10 files changed, 723 insertions(+), 27 deletions(-)

diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDesc.scala
index fc12c11e366..9b15623bfd9 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDesc.scala
+++ b/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")
+  }
+
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
index f83d1d76307..4938d2a9a25 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
+++ b/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,106 @@ 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 {
+        val openArgs = if (isBinary) """"rb"""" 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")
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
index c49a616c339..357c837a04e 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
+++ b/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,105 @@ 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) {
+      val openArgs =
+        if (isBinary) s"""$SourceFilePlaceholder, "rb""""
+        else s"""$SourceFilePlaceholder, "r", encoding=$encLit"""
+      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
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala
index b6cbdc1b936..b6d44d758c8 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtils.scala
+++ b/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)
+              }
         }
       }
 
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
index 50de683f70a..99a25edada2 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
@@ -25,12 +25,18 @@ import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.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.annotations.UIWidget
 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.FileAttributeType
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
-class TextInputSourceOpDesc extends SourceOperatorDescriptor with TextSourceOpDesc {
+class TextInputSourceOpDesc
+    extends SourceOperatorDescriptor
+    with TextSourceOpDesc
+    with StandaloneCodeGenerator {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Text")
   @JsonSchemaInject(json = UIWidget.UIWidgetTextArea)
@@ -68,4 +74,43 @@ class TextInputSourceOpDesc extends SourceOperatorDescriptor with TextSourceOpDe
       inputPorts = List.empty,
       outputPorts = List(OutputPort())
     )
+
+  override def generateStandaloneCode(): String = {
+    val text = objectMapper.writeValueAsString(textInput)
+    val col = attributeName
+    val colLit = pyStringLiteral(col)
+    val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+    buf += s"_text = $text"
+
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+
+    if (attributeType.isSingle) {
+      val valueExpr = if (isBinary) """_text.encode("utf-8")""" else "_text"
+      buf += s"""out1df = pd.DataFrame({$colLit: [$valueExpr]})"""
+    } else {
+      val castExpr = attributeType match {
+        case FileAttributeType.INTEGER   => "int(l)"
+        case FileAttributeType.LONG      => "int(l)"
+        case FileAttributeType.DOUBLE    => "float(l)"
+        case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
+        case FileAttributeType.TIMESTAMP => "l"
+        case _                           => "l"
+      }
+      // 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.
+      val dropped =
+        fileScanOffset.filter(_ > 0).fold("_text.splitlines()")(o => s"_text.splitlines()[$o:]")
+      val linesExpr = fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+      buf += s"""out1df = pd.DataFrame({$colLit: [$castExpr for l in $linesExpr]})"""
+      buf ++= TextSourceOpDesc.parseTimestampColumn(attributeType, colLit)
+    }
+
+    buf.mkString("\n")
+  }
+
+  override def standaloneHelpers(): Seq[String] = TextSourceOpDesc.lineHelpers(attributeType)
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
index 07a773580f6..acd98b10160 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
@@ -26,6 +26,7 @@ import com.kjetland.jackson.jsonSchema.annotations.{
   JsonSchemaString,
   JsonSchemaTitle
 }
+import org.apache.texera.amber.operator.StandaloneHelpers
 import org.apache.texera.amber.operator.metadata.annotations.HideAnnotation
 import org.apache.texera.amber.operator.source.scan.FileAttributeType
 
@@ -48,6 +49,10 @@ trait TextSourceOpDesc {
   @JsonDeserialize(contentAs = classOf[java.lang.String])
   var attributeName: String = "line"
 
+  // Named explicitly so reflection can see it. These are the row-window knobs the
+  // text sources actually read — the ones they inherit are ignored — but they carried
+  // no @JsonProperty, which leaves them invisible to anything walking the config.
+  @JsonProperty
   @JsonSchemaTitle("Limit (lines)")
   @JsonDeserialize(contentAs = classOf[Int])
   @JsonPropertyDescription(
@@ -66,6 +71,7 @@ trait TextSourceOpDesc {
   )
   var fileScanLimit: Option[Int] = None
 
+  @JsonProperty
   @JsonSchemaTitle("Offset (lines)")
   @JsonPropertyDescription(
     "Number of lines to skip from the start before reading. " +
@@ -84,3 +90,44 @@ trait TextSourceOpDesc {
   )
   var fileScanOffset: Option[Int] = None
 }
+
+object TextSourceOpDesc {
+
+  /**
+    * The line-to-boolean rule the engine uses, as Python.
+    *
+    * `parseField` reads "true" and "false" in any case, then falls back to an
+    * integer and calls only 1 true. Anything else ends the run. Comparing the
+    * lowercased line to "true" read 1 as false and let text the engine refuses
+    * through as a row of false.
+    */
+  val BooleanParser: String =
+    """def _texera_parse_bool(line):
+      |    text = line.strip()
+      |    lowered = text.lower()
+      |    if lowered == "true":
+      |        return True
+      |    if lowered == "false":
+      |        return False
+      |    return int(text) == 1""".stripMargin
+
+  /** The call the per-line cast makes, over the loop variable the readers share. */
+  val BooleanParserCall: String = "_texera_parse_bool(l)"
+
+  /** A TIMESTAMP line is kept as text and parsed once the column is built, as
+    * CSV and JSONL parse theirs. `pd.Timestamp` kept the nanoseconds and the
+    * offset that the engine's DateParserUtils does not.
+    */
+  def parseTimestampColumn(attributeType: FileAttributeType, colLit: String): Seq[String] =
+    if (attributeType == FileAttributeType.TIMESTAMP)
+      Seq(s"out1df[$colLit] = _texera_text_to_timestamp(out1df[$colLit])")
+    else Seq.empty
+
+  /** The helpers the Python above calls for this attribute type. */
+  def lineHelpers(attributeType: FileAttributeType): Seq[String] =
+    attributeType match {
+      case FileAttributeType.BOOLEAN   => Seq(BooleanParser)
+      case FileAttributeType.TIMESTAMP => Seq(StandaloneHelpers.AttributeCasts)
+      case _                           => Seq.empty
+    }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDescSpec.scala
index 16b9821cb19..c28dbac9fe5 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/fetcher/URLFetcherOpDescSpec.scala
@@ -92,6 +92,49 @@ class URLFetcherOpDescSpec extends AnyFlatSpec with Matchers {
     }
   }
 
+  "URLFetcherOpDesc.generateStandaloneCode" should
+    "fetch the URL and decode the body as UTF-8 text" in {
+    configured(DecodingMethod.UTF_8).generateStandaloneCode() shouldBe
+      """import http.client
+        |import urllib.request
+        |_url = "https://example.test/data"
+        |try:
+        |    with urllib.request.urlopen(_url) as _resp:
+        |        _content = _resp.read()
+        |except (OSError, http.client.HTTPException):
+        |    _content = f"Fetch failed for URL: {_url}".encode("utf-8")
+        |out1df = pd.DataFrame({"URL content": [_content.decode("utf-8", errors="replace")]})""".stripMargin
+  }
+
+  // IOUtils.toString substitutes U+FFFD for a malformed byte, so the executor returns a
+  // string for any body. A strict decode raised instead, failing the whole script on a
+  // response the executor reads.
+  it should "decode a malformed body the way the executor does, by replacement" in {
+    configured(DecodingMethod.UTF_8).generateStandaloneCode() should
+      include("""_content.decode("utf-8", errors="replace")""")
+  }
+
+  // The executor guards only the fetch, so a value with no scheme stops it. `Exception`
+  // would swallow that too and hand back a row where the platform stopped.
+  it should "let a value that is not a URL through, not just fetch failures" in {
+    val code = configured(DecodingMethod.UTF_8).generateStandaloneCode()
+    code should include("except (OSError, http.client.HTTPException):")
+    code should not include "except Exception:"
+  }
+
+  it should "keep the raw bytes when decoding is not UTF-8" in {
+    configured(DecodingMethod.RAW_BYTES).generateStandaloneCode() should
+      endWith("""out1df = pd.DataFrame({"URL content": [_content]})""")
+  }
+
+  // The URL is user-supplied and lands inside the generated Python source, so
+  // it goes through JSON string encoding rather than raw interpolation.
+  it should "emit the URL as an escaped Python string literal" in {
+    val op = configured(DecodingMethod.UTF_8)
+    op.url = """https://example.test/a"b\c"""
+    op.generateStandaloneCode() should include("""_url = "https://example.test/a\"b\\c"""")
+  }
+
   it should "propagate sourceSchema onto the single output port" in {
     // Exercise propagateSchema.func directly so the test actually proves the
     // sourceSchema gets routed to the output port id, not just that an
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
index c5a8ffc1b7e..30b74f0dc10 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
@@ -28,8 +28,9 @@ import org.apache.texera.amber.core.tuple.{
 }
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
-import org.apache.texera.amber.operator.TestOperators
+import org.apache.texera.amber.operator.{StandaloneHelpers, TestOperators}
 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.util.JSONUtils.objectMapper
 import org.scalatest.BeforeAndAfter
 import org.scalatest.flatspec.AnyFlatSpec
@@ -144,4 +145,176 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     val out = physical.propagateSchema.func(Map.empty)
     assert(out(outPortId) == fileScanOpDesc.sourceSchema())
   }
+
+  "FileScanOpDesc.generateStandaloneCode" should
+    "read every line as text with the configured encoding by default" in {
+    assert(
+      fileScanOpDesc.generateStandaloneCode() ==
+        """def _texera_file_name(row):
+          |    for _v in row:
+          |        if isinstance(_v, str):
+          |            return _v
+          |    raise ValueError(f"no file name in row: {row!r}")
+          |
+          |_rows = []
+          |for _fn in (_texera_file_name(r) for r in in1df.itertuples(index=False)):
+          |    with open(_fn, "r", encoding="utf-8") as _f:
+          |        _rows.extend(l.rstrip("\n") for l in _f)
+          |out1df = pd.DataFrame({"line": _rows})""".stripMargin
+    )
+  }
+
+  // FileScanOpExec takes `tuple.getFields.collectFirst { case s: String => s }`, so a row
+  // carrying an id ahead of the path still finds the path. Reading column 0 opened the id.
+  it should "take the row's first string field as the file name, not its first column" in {
+    val code = fileScanOpDesc.generateStandaloneCode()
+    assert(code.contains("isinstance(_v, str)"))
+    assert(!code.contains("in1df.iloc[:, 0]"))
+  }
+
+  // The enum name is "US_ASCII", not a Python codec name.
+  it should "render the encoding as a Python codec name" in {
+    fileScanOpDesc.fileEncoding = FileDecodingMethod.ASCII
+    assert(fileScanOpDesc.generateStandaloneCode().contains("""encoding="us-ascii""""))
+  }
+
+  it should "read the whole file in single-value mode, keeping the filename only when asked" in {
+    fileScanOpDesc.attributeType = FileAttributeType.SINGLE_STRING
+
+    fileScanOpDesc.outputFileName = true
+    val withName = fileScanOpDesc.generateStandaloneCode()
+    assert(withName.contains("        _rows.append((_fn, _f.read()))"))
+    assert(withName.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
+
+    fileScanOpDesc.outputFileName = false
+    val withoutName = fileScanOpDesc.generateStandaloneCode()
+    assert(withoutName.contains("        _rows.append(_f.read())"))
+    assert(withoutName.endsWith("""out1df = pd.DataFrame({"line": _rows})"""))
+
+    // A line carries the name of the file it came from, as the whole file does,
+    // which is what the platform's line-by-line branch now emits and what the
+    // schema declares either way. It used to emit the value alone, leaving a
+    // one-field row against a two-column schema.
+    fileScanOpDesc.attributeType = FileAttributeType.STRING
+    fileScanOpDesc.outputFileName = true
+    val lines = fileScanOpDesc.generateStandaloneCode()
+    assert(lines.contains("""        _rows.extend((_fn, l.rstrip("\n")) for l in _f)"""))
+    assert(lines.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
+
+    fileScanOpDesc.outputFileName = false
+    val bareLines = fileScanOpDesc.generateStandaloneCode()
+    assert(bareLines.contains("""        _rows.extend(l.rstrip("\n") for l in _f)"""))
+    assert(bareLines.endsWith("""out1df = pd.DataFrame({"line": _rows})"""))
+  }
+
+  it should "open binary attribute types in binary mode" in {
+    Seq(FileAttributeType.BINARY, FileAttributeType.LARGE_BINARY).foreach { attrType =>
+      fileScanOpDesc.attributeType = attrType
+      val code = fileScanOpDesc.generateStandaloneCode()
+      assert(code.contains("""    with open(_fn, "rb") as _f:"""))
+      assert(!code.contains("encoding="))
+      assert(code.contains("        _rows.append(_f.read())"))
+    }
+  }
+
+  it should "cast each line to the configured attribute type" in {
+    val castByType = Seq(
+      FileAttributeType.INTEGER -> "int(l.rstrip())",
+      FileAttributeType.LONG -> "int(l.rstrip())",
+      FileAttributeType.DOUBLE -> "float(l.rstrip())",
+      FileAttributeType.BOOLEAN -> "_texera_parse_bool(l)",
+      FileAttributeType.TIMESTAMP -> "l.rstrip()",
+      FileAttributeType.STRING -> """l.rstrip("\n")"""
+    )
+    castByType.foreach {
+      case (attrType, cast) =>
+        fileScanOpDesc.attributeType = attrType
+        assert(
+          fileScanOpDesc
+            .generateStandaloneCode()
+            .contains(s"        _rows.extend($cast for l in _f)")
+        )
+    }
+    // The boolean cast is the one that calls out of the body, so the script has
+    // to carry the definition.
+    fileScanOpDesc.attributeType = FileAttributeType.BOOLEAN
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(TextSourceOpDesc.BooleanParser))
+    fileScanOpDesc.attributeType = FileAttributeType.STRING
+    assert(fileScanOpDesc.standaloneHelpers().isEmpty)
+    // A timestamp line is read as text and parsed as a column, as CSV parses one.
+    fileScanOpDesc.attributeType = FileAttributeType.TIMESTAMP
+    assert(
+      fileScanOpDesc
+        .generateStandaloneCode()
+        .endsWith("""out1df["line"] = _texera_text_to_timestamp(out1df["line"])""")
+    )
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(StandaloneHelpers.AttributeCasts))
+  }
+
+  it should "slice the raw lines before converting them when a limit or offset is set" in {
+    fileScanOpDesc.attributeType = FileAttributeType.INTEGER
+
+    fileScanOpDesc.fileScanOffset = Option(3)
+    fileScanOpDesc.fileScanLimit = None
+    assert(
+      fileScanOpDesc
+        .generateStandaloneCode()
+        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[3:])")
+    )
+
+    fileScanOpDesc.fileScanOffset = None
+    fileScanOpDesc.fileScanLimit = Option(5)
+    assert(
+      fileScanOpDesc
+        .generateStandaloneCode()
+        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[:5])")
+    )
+
+    fileScanOpDesc.fileScanOffset = Option(3)
+    fileScanOpDesc.fileScanLimit = Option(5)
+    assert(
+      fileScanOpDesc
+        .generateStandaloneCode()
+        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[3:][:5])")
+    )
+  }
+
+  // With extract on, FileScanUtils opens the file as a zip and emits a tuple per
+  // entry. Reading the archive itself as text was the export's answer before, and
+  // it produced whichever bytes the compression happened to leave readable.
+  it should "read the entries inside the archive when extract is on" in {
+    // `extract` and `outputFileName` are vals, so they are set by deserialization.
+    val desc = objectMapper.readValue(
+      """{"operatorType":"FileScanOp","extract":true}""",
+      classOf[FileScanOpDesc]
+    )
+    val code = desc.generateStandaloneCode()
+
+    assert(code.contains("    with zipfile.ZipFile(_fn) as _z:"))
+    assert(code.contains("        for _name in _z.namelist():"))
+    assert(code.contains("""            if _name.startswith("__MACOSX"):"""))
+    assert(code.contains("            with _z.open(_name) as _f:"))
+    // A zip entry opens binary whatever the column holds, so a text line is
+    // decoded rather than read through a text-mode handle.
+    assert(
+      code.contains(
+        """                _rows.extend(l.rstrip("\n") for l in io.TextIOWrapper(_f, encoding="utf-8"))"""
+      )
+    )
+    assert(desc.standaloneImports() == Seq("import io", "import zipfile"))
+    assert(!code.contains("WARNING"))
+  }
+
+  it should "name the entry, not the archive, in the filename column" in {
+    val desc = objectMapper.readValue(
+      """{"operatorType":"FileScanOp","extract":true,"outputFileName":true,
+        |"attributeType":"single string"}""".stripMargin,
+      classOf[FileScanOpDesc]
+    )
+    val code = desc.generateStandaloneCode()
+
+    assert(code.contains("""                _rows.append((_name, _f.read().decode("utf-8")))"""))
+    assert(code.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
+    assert(desc.standaloneImports() == Seq("import zipfile"))
+  }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
index 6ac320e99cf..c0c1df09129 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
@@ -31,7 +31,7 @@ import org.scalatest.BeforeAndAfter
 import org.scalatest.flatspec.AnyFlatSpec
 
 import java.nio.charset.StandardCharsets
-import java.nio.file.Files
+import java.nio.file.{Files, Path, Paths}
 
 class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
 
@@ -187,11 +187,13 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     FileScanSourceOpExec.close()
   }
 
+  // `encoding` and not the inherited `fileEncoding`: the descriptor drops that
+  // one on the way over, so setting it never reached the executor at all.
   it should "read first 5 lines of the input text file with US_ASCII encoding" in {
-    fileScanSourceOpDesc.setResolvedFileName(
-      FileResolver.resolve(TestOperators.TestCRLFTextFilePath)
+    fileScanSourceOpDesc = describing(
+      Paths.get(TestOperators.TestCRLFTextFilePath),
+      """"encoding":"US_ASCII""""
     )
-    fileScanSourceOpDesc.encoding = FileDecodingMethod.ASCII
     fileScanSourceOpDesc.attributeType = FileAttributeType.STRING
     fileScanSourceOpDesc.fileScanLimit = Option(5)
     val FileScanSourceOpExec =
@@ -291,4 +293,15 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(schema.getAttribute("line").getType == AttributeType.STRING)
   }
 
+  /** `extract`, `outputFileName` and `encoding` are vals, so the fields are
+    * deserialized in.
+    */
+  private def describing(file: Path, fields: String*): FileScanSourceOpDesc = {
+    val desc = objectMapper.readValue(
+      (""""operatorType":"FileScan"""" +: fields).mkString("{", ",", "}"),
+      classOf[FileScanSourceOpDesc]
+    )
+    desc.setResolvedFileName(FileResolver.resolve(file.toString))
+    desc
+  }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
index ef8a824849a..7b10e7146cc 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
@@ -19,19 +19,33 @@
 
 package org.apache.texera.amber.operator.source.scan.text
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
-import org.apache.texera.amber.core.tuple.{AttributeType, Schema, SchemaEnforceable, Tuple}
+import org.apache.texera.amber.core.tuple.{
+  AttributeType,
+  AttributeTypeUtils,
+  Schema,
+  SchemaEnforceable,
+  Tuple
+}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.TestOperators
 import org.apache.texera.amber.operator.source.scan.FileAttributeType
+import org.apache.texera.amber.operator.tags.IntegrationTest
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.BeforeAndAfter
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
 
 class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
+
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
   var textInputSourceOpDesc: TextInputSourceOpDesc = _
 
   before {
@@ -276,6 +290,99 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
   }
 
+  // `parseField` reads a BOOLEAN line as "true" or "false" in any case, then as an
+  // integer that is true only at 1, and refuses anything else, which the export
+  // has to refuse too rather than pass off as a row of false.
+  "TextInputSourceOpDesc.generateStandaloneCode" should "refuse a boolean line the engine refuses" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    textInputSourceOpDesc.attributeType = FileAttributeType.BOOLEAN
+    textInputSourceOpDesc.textInput = "true\nyes"
+
+    assertThrows[AttributeTypeUtils.AttributeTypeException](booleansFromEngine())
+    val (exitCode, out) = runStandalone(python)
+    withClue(s"python said:\n$out\n") {
+      assert(exitCode != 0)
+      assert(out.contains("ValueError"))
+    }
+  }
+
+  /** The configured text read as booleans by the executor the engine runs. */
+  private def booleansFromEngine(): Seq[Boolean] = {
+    val exec = new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc))
+    exec.open()
+    try {
+      exec
+        .produceTuple()
+        .map(
+          _.asInstanceOf[SchemaEnforceable]
+            .enforceSchema(textInputSourceOpDesc.sourceSchema())
+            .getField[Boolean]("line")
+        )
+        .toSeq
+    } finally exec.close()
+  }
+
+  /**
+    * The configured operator's exported script, run, printing its one column.
+    *
+    * `bool` on each value because a numpy scalar does not repr as Python's own
+    * True and False.
+    */
+  private def runStandalone(python: String): (Int, String) = {
+    val dir = Files.createTempDirectory("text-input-standalone-")
+    dir.toFile.deleteOnExit()
+    val script = dir.resolve("run.py")
+    Files.write(
+      script,
+      s"""import pandas as pd
+         |${textInputSourceOpDesc.standaloneHelpers().mkString("\n\n")}
+         |${textInputSourceOpDesc.generateStandaloneCode()}
+         |print([bool(v) for v in out1df["line"]])
+         |""".stripMargin.getBytes(StandardCharsets.UTF_8)
+    )
+
+    val process = new ProcessBuilder(python, script.toString)
+      .directory(dir.toFile)
+      .redirectErrorStream(true)
+      .start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    (process.exitValue(), out)
+  }
+
+  // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path
+  // (UDF_PYTHON_PATH), then python3 / python / py.
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas").redirectErrorStream(true).start()
+    ).toOption
+      .exists { p =>
+        if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+        else p.exitValue() == 0
+      }
+
   "TextInputSourceOpDesc.getPhysicalOp" should
     "wire the TextInputSourceOpExec class as a source op with one output port" in {
     val physical =

From 0dcbb8de31a98add0653288a5e1a08bb1d549950 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Thu, 8 Oct 2026 00:52:05 -0700
Subject: [PATCH 2/5] feat(workflow-operator): keep a whole file's CRLF when
 the file scans export it

Read as a single string, the export opened the file in text mode, which turns
a CRLF into a LF. The engine decodes the file's bytes as they are, so the two
disagreed on any file with Windows line endings. Open it with newline="" for
both File Scan operators; reading by line keeps the default, which already
ends a line where the engine does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../source/scan/file/FileScanOpDesc.scala         |  8 +++++++-
 .../source/scan/file/FileScanSourceOpDesc.scala   |  4 +++-
 .../source/scan/file/FileScanOpDescSpec.scala     | 15 +++++++++++++++
 .../scan/file/FileScanSourceOpDescSpec.scala      | 11 +++++++++++
 4 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
index 4938d2a9a25..bc0744591af 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
@@ -136,7 +136,13 @@ class FileScanOpDesc
           s"io.TextIOWrapper(_f, encoding=$encLit)"
         )
       } else {
-        val openArgs = if (isBinary) """"rb"""" else s""""r", encoding=$encLit"""
+        // 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")
       }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
index 357c837a04e..305340f72b5 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
@@ -148,9 +148,11 @@ class FileScanSourceOpDesc
       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"""
+        else s"""$SourceFilePlaceholder, "r", encoding=$encLit, newline="""""
       val dfCols =
         if (emitFilename) s"""{"filename": $SourceFilePlaceholder, $colLit: [_f.read()]}"""
         else s"""{$colLit: [_f.read()]}"""
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
index 30b74f0dc10..c3f78bcec75 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
@@ -207,6 +207,21 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(bareLines.endsWith("""out1df = pd.DataFrame({"line": _rows})"""))
   }
 
+  it should "keep a whole file's CRLF in the export, as the engine does" in {
+    // The engine decodes the bytes as they are. The text reader's default turns a
+    // CRLF into a LF, so `first\r\nsecond\r\n` would export as `first\nsecond\n`.
+    fileScanOpDesc.attributeType = FileAttributeType.SINGLE_STRING
+    assert(
+      fileScanOpDesc
+        .generateStandaloneCode()
+        .contains("""    with open(_fn, "r", encoding="utf-8", newline="") as _f:""")
+    )
+
+    // Lines keep the default, which ends a line where the engine's reader does.
+    fileScanOpDesc.attributeType = FileAttributeType.STRING
+    assert(!fileScanOpDesc.generateStandaloneCode().contains("newline="))
+  }
+
   it should "open binary attribute types in binary mode" in {
     Seq(FileAttributeType.BINARY, FileAttributeType.LARGE_BINARY).foreach { attrType =>
       fileScanOpDesc.attributeType = attrType
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
index c0c1df09129..442e4ca9a15 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDescSpec.scala
@@ -260,6 +260,17 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     }
   }
 
+  it should "keep a whole file's CRLF in the export, as the engine does" in {
+    // The engine decodes the bytes as they are. The text reader's default turns a
+    // CRLF into a LF, so `first\r\nsecond\r\n` would export as `first\nsecond\n`.
+    fileScanSourceOpDesc.attributeType = FileAttributeType.SINGLE_STRING
+    assert(fileScanSourceOpDesc.generateStandaloneCode().contains("""newline="""""))
+
+    // Lines keep the default, which ends a line where the engine's reader does.
+    fileScanSourceOpDesc.attributeType = FileAttributeType.STRING
+    assert(!fileScanSourceOpDesc.generateStandaloneCode().contains("newline="))
+  }
+
   "FileScanSourceOpDesc.getPhysicalOp" should
     "wire the FileScanSourceOpExec class as a source op and propagate its schema" in {
     val physical =

From e9d9824a0bde9f0c3d6e34b15c02ab2e0996bdd9 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Mon, 12 Oct 2026 15:07:42 -0700
Subject: [PATCH 3/5] fix(operator): split Text Input lines only where the
 engine does

The engine splits with linesIterator, which ends a line only at LF, CR and
CRLF. The export used splitlines, which also ends one at eight other
characters such as U+2028 and NEL, so a row holding one came out as two.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../scan/text/TextInputSourceOpDesc.scala     | 10 ++-
 .../scan/text/TextInputSourceOpDescSpec.scala | 65 +++++++++++++++++--
 2 files changed, 69 insertions(+), 6 deletions(-)

diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
index 99a25edada2..706f2f29f0b 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
@@ -98,12 +98,18 @@ class TextInputSourceOpDesc
         case FileAttributeType.TIMESTAMP => "l"
         case _                           => "l"
       }
+      // The engine splits with linesIterator, which ends a line only at LF, CR
+      // and CRLF. splitlines also ends one at eight other characters, U+2028
+      // and NEL among them, so a row holding one came out as two. A newline at
+      // the very end closes the last line rather than opening an empty one.
+      buf += """_lines = _text.replace("\r\n", "\n").replace("\r", "\n").split("\n")"""
+      buf += """if _lines[-1] == "":"""
+      buf += "    _lines.pop()"
       // 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.
-      val dropped =
-        fileScanOffset.filter(_ > 0).fold("_text.splitlines()")(o => s"_text.splitlines()[$o:]")
+      val dropped = fileScanOffset.filter(_ > 0).fold("_lines")(o => s"_lines[$o:]")
       val linesExpr = fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
       buf += s"""out1df = pd.DataFrame({$colLit: [$castExpr for l in $linesExpr]})"""
       buf ++= TextSourceOpDesc.parseTimestampColumn(attributeType, colLit)
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
index 7b10e7146cc..38cfc657594 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
@@ -310,6 +310,60 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     }
   }
 
+  // linesIterator ends a line only at LF, CR and CRLF, while splitlines also
+  // ended one at eight other characters, so text holding U+2028 was one row in
+  // the engine and two in the export.
+  it should "split the text into the lines the engine splits it into" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val separators =
+      Seq(
+        "\n",
+        "\r",
+        "\r\n",
+        "\u000b",
+        "\u000c",
+        "\u001c",
+        "\u001d",
+        "\u001e",
+        "\u0085",
+        "\u2028",
+        "\u2029"
+      )
+    val inputs = separators.map(sep => s"first${sep}second") ++
+      Seq("", "first\n", "first\n\n", "\nfirst", "first\r\r\nsecond")
+
+    textInputSourceOpDesc.attributeType = FileAttributeType.STRING
+    inputs.foreach { input =>
+      textInputSourceOpDesc.textInput = input
+      val (exitCode, out) =
+        runStandalone(python, """import json; print(json.dumps(list(out1df["line"])))""")
+      withClue(s"input ${input.map(c => f"\\u${c.toInt}%04x").mkString}, python said:\n$out\n") {
+        assert(exitCode == 0)
+        assert(objectMapper.readValue(out.trim, classOf[Array[String]]).toSeq == linesFromEngine())
+      }
+    }
+  }
+
+  /** The configured text read as STRING lines by the executor the engine runs. */
+  private def linesFromEngine(): Seq[String] = {
+    val exec = new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc))
+    exec.open()
+    try {
+      exec
+        .produceTuple()
+        .map(
+          _.asInstanceOf[SchemaEnforceable]
+            .enforceSchema(textInputSourceOpDesc.sourceSchema())
+            .getField[String]("line")
+        )
+        .toSeq
+    } finally exec.close()
+  }
+
   /** The configured text read as booleans by the executor the engine runs. */
   private def booleansFromEngine(): Seq[Boolean] = {
     val exec = new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc))
@@ -329,10 +383,13 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
   /**
     * The configured operator's exported script, run, printing its one column.
     *
-    * `bool` on each value because a numpy scalar does not repr as Python's own
-    * True and False.
+    * `bool` on each value by default because a numpy scalar does not repr as
+    * Python's own True and False.
     */
-  private def runStandalone(python: String): (Int, String) = {
+  private def runStandalone(
+      python: String,
+      printLine: String = """print([bool(v) for v in out1df["line"]])"""
+  ): (Int, String) = {
     val dir = Files.createTempDirectory("text-input-standalone-")
     dir.toFile.deleteOnExit()
     val script = dir.resolve("run.py")
@@ -341,7 +398,7 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
       s"""import pandas as pd
          |${textInputSourceOpDesc.standaloneHelpers().mkString("\n\n")}
          |${textInputSourceOpDesc.generateStandaloneCode()}
-         |print([bool(v) for v in out1df["line"]])
+         |$printLine
          |""".stripMargin.getBytes(StandardCharsets.UTF_8)
     )
 

From e778d2ab3012d9ff9cd33d889d9d6654f750f472 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Wed, 14 Oct 2026 15:29:31 -0700
Subject: [PATCH 4/5] fix(operator): read a number line the way the engine's
 parseField does

int() and float() read "1_0", "nan", "inf" and an INTEGER past 32 bits,
which end the engine's run, and float() refused "2d" and "0x1p3", which
the engine reads. Each line is now trimmed as Java trims and checked
against the grammar parseInt, parseLong or parseDouble accepts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../source/scan/file/FileScanOpDesc.scala     |   6 +-
 .../scan/file/FileScanSourceOpDesc.scala      |   6 +-
 .../scan/text/TextInputSourceOpDesc.scala     |   6 +-
 .../source/scan/text/TextSourceOpDesc.scala   |  52 +++++++-
 .../source/scan/file/FileScanOpDescSpec.scala |  28 ++--
 .../scan/text/TextInputSourceOpDescSpec.scala | 124 ++++++++++++++++++
 6 files changed, 201 insertions(+), 21 deletions(-)

diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
index bc0744591af..1d47eca7ce3 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
@@ -157,9 +157,9 @@ class FileScanOpDesc
       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.INTEGER   => "_texera_parse_int(l, 32)"
+        case FileAttributeType.LONG      => "_texera_parse_int(l, 64)"
+        case FileAttributeType.DOUBLE    => "_texera_parse_double(l)"
         case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
         case FileAttributeType.TIMESTAMP => "l.rstrip()"
         case _                           => """l.rstrip("\n")"""
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
index 305340f72b5..495b4e018a9 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala
@@ -93,9 +93,9 @@ class FileScanSourceOpDesc
       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.INTEGER   => "_texera_parse_int(l, 32)"
+      case FileAttributeType.LONG      => "_texera_parse_int(l, 64)"
+      case FileAttributeType.DOUBLE    => "_texera_parse_double(l)"
       case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
       case FileAttributeType.TIMESTAMP => "l.rstrip()"
       case _                           => """l.rstrip("\n")"""
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
index 706f2f29f0b..179f49f80f0 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDesc.scala
@@ -91,9 +91,9 @@ class TextInputSourceOpDesc
       buf += s"""out1df = pd.DataFrame({$colLit: [$valueExpr]})"""
     } else {
       val castExpr = attributeType match {
-        case FileAttributeType.INTEGER   => "int(l)"
-        case FileAttributeType.LONG      => "int(l)"
-        case FileAttributeType.DOUBLE    => "float(l)"
+        case FileAttributeType.INTEGER   => "_texera_parse_int(l, 32)"
+        case FileAttributeType.LONG      => "_texera_parse_int(l, 64)"
+        case FileAttributeType.DOUBLE    => "_texera_parse_double(l)"
         case FileAttributeType.BOOLEAN   => TextSourceOpDesc.BooleanParserCall
         case FileAttributeType.TIMESTAMP => "l"
         case _                           => "l"
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
index acd98b10160..1e6fee9f5c3 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala
@@ -93,6 +93,49 @@ trait TextSourceOpDesc {
 
 object TextSourceOpDesc {
 
+  /**
+    * The line-to-number rules the engine uses, as Python.
+    *
+    * `parseField` trims the line and reads it with `toInt`, `toLong` or
+    * `toDouble`. `int()` and `float()` disagree both ways: they read "1_0",
+    * "nan", "inf" and an INTEGER past 32 bits, which end the engine's run, and
+    * `float()` refuses "2d" and "0x1p3", which the engine reads. `trim` drops
+    * only the characters up to U+0020, so a no-break space stays and fails.
+    */
+  val NumberParsers: String =
+    """def _texera_java_trim(line):
+      |    return line.strip("".join(chr(i) for i in range(33)))
+      |
+      |
+      |def _texera_parse_int(line, bits):
+      |    import re
+      |
+      |    # Integer.parseInt and Long.parseLong take a sign and Unicode digits.
+      |    text = _texera_java_trim(line)
+      |    if not re.fullmatch(r"[+-]?\d+", text):
+      |        raise ValueError("For input string: " + repr(text))
+      |    value = int(text)
+      |    if not -(2 ** (bits - 1)) <= value < 2 ** (bits - 1):
+      |        raise ValueError("For input string: " + repr(text))
+      |    return value
+      |
+      |
+      |def _texera_parse_double(line):
+      |    import re
+      |
+      |    # Double.parseDouble's grammar, ASCII digits only. A text it takes
+      |    # is read by _texera_java_double.
+      |    text = _texera_java_trim(line)
+      |    if not re.fullmatch(
+      |        r"[+-]?(NaN|Infinity|"
+      |        r"((\d+\.?\d*|\.\d+)([eE][+-]?\d+)?"
+      |        r"|0[xX]([0-9a-fA-F]+\.?|[0-9a-fA-F]*\.[0-9a-fA-F]+)[pP][+-]?\d+)[fFdD]?)",
+      |        text,
+      |        re.ASCII,
+      |    ):
+      |        raise ValueError("For input string: " + repr(text))
+      |    return _texera_java_double(text)""".stripMargin
+
   /**
     * The line-to-boolean rule the engine uses, as Python.
     *
@@ -103,13 +146,13 @@ object TextSourceOpDesc {
     */
   val BooleanParser: String =
     """def _texera_parse_bool(line):
-      |    text = line.strip()
+      |    text = _texera_java_trim(line)
       |    lowered = text.lower()
       |    if lowered == "true":
       |        return True
       |    if lowered == "false":
       |        return False
-      |    return int(text) == 1""".stripMargin
+      |    return _texera_parse_int(text, 32) == 1""".stripMargin
 
   /** The call the per-line cast makes, over the loop variable the readers share. */
   val BooleanParserCall: String = "_texera_parse_bool(l)"
@@ -126,7 +169,10 @@ object TextSourceOpDesc {
   /** The helpers the Python above calls for this attribute type. */
   def lineHelpers(attributeType: FileAttributeType): Seq[String] =
     attributeType match {
-      case FileAttributeType.BOOLEAN   => Seq(BooleanParser)
+      case FileAttributeType.INTEGER   => Seq(NumberParsers)
+      case FileAttributeType.LONG      => Seq(NumberParsers)
+      case FileAttributeType.DOUBLE    => Seq(StandaloneHelpers.AttributeCasts, NumberParsers)
+      case FileAttributeType.BOOLEAN   => Seq(NumberParsers, BooleanParser)
       case FileAttributeType.TIMESTAMP => Seq(StandaloneHelpers.AttributeCasts)
       case _                           => Seq.empty
     }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
index c3f78bcec75..21e89519c8f 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
@@ -234,9 +234,9 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
 
   it should "cast each line to the configured attribute type" in {
     val castByType = Seq(
-      FileAttributeType.INTEGER -> "int(l.rstrip())",
-      FileAttributeType.LONG -> "int(l.rstrip())",
-      FileAttributeType.DOUBLE -> "float(l.rstrip())",
+      FileAttributeType.INTEGER -> "_texera_parse_int(l, 32)",
+      FileAttributeType.LONG -> "_texera_parse_int(l, 64)",
+      FileAttributeType.DOUBLE -> "_texera_parse_double(l)",
       FileAttributeType.BOOLEAN -> "_texera_parse_bool(l)",
       FileAttributeType.TIMESTAMP -> "l.rstrip()",
       FileAttributeType.STRING -> """l.rstrip("\n")"""
@@ -250,10 +250,20 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
             .contains(s"        _rows.extend($cast for l in _f)")
         )
     }
-    // The boolean cast is the one that calls out of the body, so the script has
-    // to carry the definition.
+    // The number and boolean casts call out of the body, so the script has to
+    // carry the definitions.
+    fileScanOpDesc.attributeType = FileAttributeType.INTEGER
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(TextSourceOpDesc.NumberParsers))
+    fileScanOpDesc.attributeType = FileAttributeType.DOUBLE
+    assert(
+      fileScanOpDesc.standaloneHelpers() ==
+        Seq(StandaloneHelpers.AttributeCasts, TextSourceOpDesc.NumberParsers)
+    )
     fileScanOpDesc.attributeType = FileAttributeType.BOOLEAN
-    assert(fileScanOpDesc.standaloneHelpers() == Seq(TextSourceOpDesc.BooleanParser))
+    assert(
+      fileScanOpDesc.standaloneHelpers() ==
+        Seq(TextSourceOpDesc.NumberParsers, TextSourceOpDesc.BooleanParser)
+    )
     fileScanOpDesc.attributeType = FileAttributeType.STRING
     assert(fileScanOpDesc.standaloneHelpers().isEmpty)
     // A timestamp line is read as text and parsed as a column, as CSV parses one.
@@ -274,7 +284,7 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[3:])")
+        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[3:])")
     )
 
     fileScanOpDesc.fileScanOffset = None
@@ -282,7 +292,7 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[:5])")
+        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[:5])")
     )
 
     fileScanOpDesc.fileScanOffset = Option(3)
@@ -290,7 +300,7 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(int(l.rstrip()) for l in _f.readlines()[3:][:5])")
+        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[3:][:5])")
     )
   }
 
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
index 38cfc657594..48164d5c207 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/text/TextInputSourceOpDescSpec.scala
@@ -348,6 +348,130 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     }
   }
 
+  // `parseField` trims a line and reads it with toInt, toLong or toDouble.
+  // int() and float() read "1_0", "nan", "inf", an Arabic-Indic digit as a
+  // double and an INTEGER past 32 bits, all of which end the engine's run, and
+  // float() refused "2d" and "0x1p3", which the engine reads. Each line runs
+  // alone through the executor and through the export, and either both read
+  // the same value or both refuse it.
+  it should "read a number line the way the engine's parseField does" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val linesByType = Seq(
+      FileAttributeType.INTEGER -> Seq(
+        "5",
+        " 5",
+        "+5",
+        "1_0",
+        "٣",
+        "５",
+        "-2147483648",
+        "2147483648",
+        "3000000000",
+        " 5",
+        "5\u0000",
+        "1.0"
+      ),
+      FileAttributeType.LONG -> Seq(
+        "9223372036854775807",
+        "9223372036854775808",
+        "1_0",
+        " -7 "
+      ),
+      FileAttributeType.DOUBLE -> Seq(
+        "1.5",
+        "2d",
+        "0x1p3",
+        "0x.8p1",
+        "0x1p1024",
+        "NaN",
+        "nan",
+        "inf",
+        "Infinity",
+        "-Infinity",
+        "NaNd",
+        "1_0",
+        "٣",
+        "1e400",
+        "-0.0",
+        ".5",
+        "5.",
+        " 1.5",
+        "1.5\u0000"
+      ),
+      FileAttributeType.BOOLEAN -> Seq("TRUE", "1", "2", "1_0", "3000000000", " true")
+    )
+
+    val runs = for {
+      (attrType, lines) <- linesByType
+      line <- lines
+    } yield {
+      textInputSourceOpDesc.attributeType = attrType
+      textInputSourceOpDesc.textInput = line
+      val engine = Try {
+        val exec =
+          new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc))
+        exec.open()
+        try {
+          exec
+            .produceTuple()
+            .map(
+              _.asInstanceOf[SchemaEnforceable]
+                .enforceSchema(textInputSourceOpDesc.sourceSchema())
+                .getField[Any]("line")
+            )
+            .toList
+        } finally exec.close()
+      }.toOption.flatMap(_.headOption)
+      val code = (textInputSourceOpDesc.standaloneHelpers() :+
+        textInputSourceOpDesc.generateStandaloneCode()).mkString("\n\n")
+      (s"$attrType ${line.map(c => f"\\u${c.toInt}%04x").mkString}", engine, code)
+    }
+
+    // A float prints as Java's Double.valueOf reads it back, so the two sides
+    // meet on equals, where NaN equals itself and -0.0 is not 0.0.
+    val driver = ("import pandas as pd" +: runs.map {
+      case (_, _, code) =>
+        s"""ns = {"pd": pd}
+           |try:
+           |    exec(${objectMapper.writeValueAsString(code)}, ns)
+           |    v = ns["out1df"]["line"].iloc[0].item()
+           |    if isinstance(v, float):
+           |        print("Infinity" if v == float("inf") else "-Infinity" if v == float("-inf") else repr(v).replace("nan", "NaN"))
+           |    else:
+           |        print(v)
+           |except ValueError:
+           |    print("refused")""".stripMargin
+    }).mkString("\n")
+    val dir = Files.createTempDirectory("text-input-numbers-")
+    dir.toFile.deleteOnExit()
+    val script = dir.resolve("run.py")
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+
+    withClue(s"python said:\n$out\n") {
+      assert(process.exitValue() == 0)
+      val printed = out.linesIterator.toList
+      assert(printed.size == runs.size)
+      runs.zip(printed).foreach {
+        case ((label, engine, _), exported) =>
+          val agreed = engine match {
+            case None => exported == "refused"
+            case Some(d: java.lang.Double) =>
+              Try(java.lang.Double.valueOf(exported)).toOption.exists(_.equals(d))
+            case Some(b: java.lang.Boolean) => exported == (if (b) "True" else "False")
+            case Some(n)                    => exported == n.toString
+          }
+          assert(agreed, s"$label: engine ${engine.getOrElse("refused")}, export $exported")
+      }
+    }
+  }
+
   /** The configured text read as STRING lines by the executor the engine runs. */
   private def linesFromEngine(): Seq[String] = {
     val exec = new TextInputSourceOpExec(objectMapper.writeValueAsString(textInputSourceOpDesc))

From 6fee99b32fc40c54f3adf997a8d5555ffe889a04 Mon Sep 17 00:00:00 2001
From: kary zheng <kary09302005@gmail.com>
Date: Wed, 21 Oct 2026 04:34:50 -0700
Subject: [PATCH 5/5] fix(operator): decode File Scan bytes the way the
 engine's reader does

Every file now opens binary and goes through _texera_decode. Bytes that
cannot be decoded become U+FFFD, and UTF-16 with no byte-order mark reads
as big-endian, as Java's InputStreamReader does. A UTF-8 file read as
UTF-16 used to end the script, while the engine read on.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../source/scan/file/FileScanOpDesc.scala     | 65 +++++++++++------
 .../source/scan/file/FileScanOpDescSpec.scala | 73 +++++++++++--------
 2 files changed, 83 insertions(+), 55 deletions(-)

diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
index 1d47eca7ce3..0457bab2159 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala
@@ -114,6 +114,14 @@ class FileScanOpDesc
     buf += "_rows = []"
     buf += "for _fn in (_texera_file_name(r) for r in in1df.itertuples(index=False)):"
 
+    // Every file opens binary and is decoded by `_texera_decode`, which reads the
+    // bytes the way the engine's reader does (see that helper). A whole file keeps
+    // its text as decoded, so a CRLF stays a CRLF. Lines go through a StringIO in
+    // universal-newline mode, which ends a line where `open(..., "r")` does, and so
+    // where the engine does.
+    val readWholeOf = if (isBinary) "_f.read()" else s"_texera_decode(_f.read(), $encLit)"
+    val linesOf = s"io.StringIO(_texera_decode(_f.read(), $encLit), newline=None)"
+
     // 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
@@ -126,25 +134,10 @@ class FileScanOpDesc
         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)"
-        )
+        (" " * 16, "_name", readWholeOf, linesOf)
       } 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")
+        buf += "    with open(_fn, \"rb\") as _f:"
+        (" " * 8, "_fn", readWholeOf, linesOf)
       }
 
     // Whatever the flag says, as the platform now reads it: every row carries the
@@ -193,10 +186,36 @@ class FileScanOpDesc
     buf.mkString("\n")
   }
 
-  override def standaloneHelpers(): Seq[String] = TextSourceOpDesc.lineHelpers(attributeType)
+  override def standaloneHelpers(): Seq[String] = {
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+    TextSourceOpDesc.lineHelpers(attributeType) ++
+      (if (isBinary) Seq.empty else Seq(FileScanOpDesc.DecodeHelper))
+  }
+
+  override def standaloneImports(): Seq[String] = {
+    val readsLines =
+      !attributeType.isSingle && attributeType != FileAttributeType.BINARY &&
+        attributeType != FileAttributeType.LARGE_BINARY
+    (if (readsLines) Seq("import io") else Seq.empty) ++
+      (if (extract) Seq("import zipfile") else Seq.empty)
+  }
+}
 
-  override def standaloneImports(): Seq[String] =
-    if (!extract) Seq.empty
-    else if (attributeType.isSingle) Seq("import zipfile")
-    else Seq("import io", "import zipfile")
+object FileScanOpDesc {
+
+  /** The engine decodes through Java's InputStreamReader, which puts U+FFFD in
+    * place of any bytes it cannot decode rather than failing, and reads UTF-16
+    * with no byte-order mark as big-endian. Python's codecs refuse both, so a file
+    * read under the wrong encoding ended the script where the engine read on.
+    */
+  val DecodeHelper: String =
+    """def _texera_decode(raw, encoding):
+      |    if encoding == "utf-16":
+      |        if raw[:2] == b"\xfe\xff":
+      |            return raw[2:].decode("utf-16-be", "replace")
+      |        if raw[:2] == b"\xff\xfe":
+      |            return raw[2:].decode("utf-16-le", "replace")
+      |        return raw.decode("utf-16-be", "replace")
+      |    return raw.decode(encoding, "replace")""".stripMargin
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
index 21e89519c8f..555266278ca 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDescSpec.scala
@@ -39,6 +39,11 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
 
   private val inputSchema = new Schema(new Attribute("filename", AttributeType.STRING))
 
+  // How a UTF-8 file's lines are read: decoded from its bytes, then split where
+  // `open(..., "r")` splits them.
+  private val utf8Lines = """io.StringIO(_texera_decode(_f.read(), "utf-8"), newline=None)"""
+  private val utf8Whole = """_texera_decode(_f.read(), "utf-8")"""
+
   var fileScanOpDesc: FileScanOpDesc = _
 
   before {
@@ -158,8 +163,8 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
           |
           |_rows = []
           |for _fn in (_texera_file_name(r) for r in in1df.itertuples(index=False)):
-          |    with open(_fn, "r", encoding="utf-8") as _f:
-          |        _rows.extend(l.rstrip("\n") for l in _f)
+          |    with open(_fn, "rb") as _f:
+          |        _rows.extend(l.rstrip("\n") for l in io.StringIO(_texera_decode(_f.read(), "utf-8"), newline=None))
           |out1df = pd.DataFrame({"line": _rows})""".stripMargin
     )
   }
@@ -175,7 +180,9 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
   // The enum name is "US_ASCII", not a Python codec name.
   it should "render the encoding as a Python codec name" in {
     fileScanOpDesc.fileEncoding = FileDecodingMethod.ASCII
-    assert(fileScanOpDesc.generateStandaloneCode().contains("""encoding="us-ascii""""))
+    assert(
+      fileScanOpDesc.generateStandaloneCode().contains("""_texera_decode(_f.read(), "us-ascii")""")
+    )
   }
 
   it should "read the whole file in single-value mode, keeping the filename only when asked" in {
@@ -183,12 +190,12 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
 
     fileScanOpDesc.outputFileName = true
     val withName = fileScanOpDesc.generateStandaloneCode()
-    assert(withName.contains("        _rows.append((_fn, _f.read()))"))
+    assert(withName.contains(s"        _rows.append((_fn, $utf8Whole))"))
     assert(withName.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
 
     fileScanOpDesc.outputFileName = false
     val withoutName = fileScanOpDesc.generateStandaloneCode()
-    assert(withoutName.contains("        _rows.append(_f.read())"))
+    assert(withoutName.contains(s"        _rows.append($utf8Whole)"))
     assert(withoutName.endsWith("""out1df = pd.DataFrame({"line": _rows})"""))
 
     // A line carries the name of the file it came from, as the whole file does,
@@ -198,28 +205,28 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     fileScanOpDesc.attributeType = FileAttributeType.STRING
     fileScanOpDesc.outputFileName = true
     val lines = fileScanOpDesc.generateStandaloneCode()
-    assert(lines.contains("""        _rows.extend((_fn, l.rstrip("\n")) for l in _f)"""))
+    assert(lines.contains(s"""        _rows.extend((_fn, l.rstrip("\\n")) for l in $utf8Lines)"""))
     assert(lines.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
 
     fileScanOpDesc.outputFileName = false
     val bareLines = fileScanOpDesc.generateStandaloneCode()
-    assert(bareLines.contains("""        _rows.extend(l.rstrip("\n") for l in _f)"""))
+    assert(bareLines.contains(s"""        _rows.extend(l.rstrip("\\n") for l in $utf8Lines)"""))
     assert(bareLines.endsWith("""out1df = pd.DataFrame({"line": _rows})"""))
   }
 
   it should "keep a whole file's CRLF in the export, as the engine does" in {
     // The engine decodes the bytes as they are. The text reader's default turns a
     // CRLF into a LF, so `first\r\nsecond\r\n` would export as `first\nsecond\n`.
+    // A whole file is decoded from its bytes and goes through no reader at all.
     fileScanOpDesc.attributeType = FileAttributeType.SINGLE_STRING
-    assert(
-      fileScanOpDesc
-        .generateStandaloneCode()
-        .contains("""    with open(_fn, "r", encoding="utf-8", newline="") as _f:""")
-    )
+    val whole = fileScanOpDesc.generateStandaloneCode()
+    assert(whole.contains("""    with open(_fn, "rb") as _f:"""))
+    assert(!whole.contains("StringIO"))
 
-    // Lines keep the default, which ends a line where the engine's reader does.
+    // Lines split in universal-newline mode, which ends a line where the engine's
+    // reader does.
     fileScanOpDesc.attributeType = FileAttributeType.STRING
-    assert(!fileScanOpDesc.generateStandaloneCode().contains("newline="))
+    assert(fileScanOpDesc.generateStandaloneCode().contains(utf8Lines))
   }
 
   it should "open binary attribute types in binary mode" in {
@@ -247,25 +254,26 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
         assert(
           fileScanOpDesc
             .generateStandaloneCode()
-            .contains(s"        _rows.extend($cast for l in _f)")
+            .contains(s"        _rows.extend($cast for l in $utf8Lines)")
         )
     }
     // The number and boolean casts call out of the body, so the script has to
-    // carry the definitions.
+    // carry the definitions, beside the decoder every text read calls.
+    val decode = FileScanOpDesc.DecodeHelper
     fileScanOpDesc.attributeType = FileAttributeType.INTEGER
-    assert(fileScanOpDesc.standaloneHelpers() == Seq(TextSourceOpDesc.NumberParsers))
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(TextSourceOpDesc.NumberParsers, decode))
     fileScanOpDesc.attributeType = FileAttributeType.DOUBLE
     assert(
       fileScanOpDesc.standaloneHelpers() ==
-        Seq(StandaloneHelpers.AttributeCasts, TextSourceOpDesc.NumberParsers)
+        Seq(StandaloneHelpers.AttributeCasts, TextSourceOpDesc.NumberParsers, decode)
     )
     fileScanOpDesc.attributeType = FileAttributeType.BOOLEAN
     assert(
       fileScanOpDesc.standaloneHelpers() ==
-        Seq(TextSourceOpDesc.NumberParsers, TextSourceOpDesc.BooleanParser)
+        Seq(TextSourceOpDesc.NumberParsers, TextSourceOpDesc.BooleanParser, decode)
     )
     fileScanOpDesc.attributeType = FileAttributeType.STRING
-    assert(fileScanOpDesc.standaloneHelpers().isEmpty)
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(decode))
     // A timestamp line is read as text and parsed as a column, as CSV parses one.
     fileScanOpDesc.attributeType = FileAttributeType.TIMESTAMP
     assert(
@@ -273,7 +281,7 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
         .generateStandaloneCode()
         .endsWith("""out1df["line"] = _texera_text_to_timestamp(out1df["line"])""")
     )
-    assert(fileScanOpDesc.standaloneHelpers() == Seq(StandaloneHelpers.AttributeCasts))
+    assert(fileScanOpDesc.standaloneHelpers() == Seq(StandaloneHelpers.AttributeCasts, decode))
   }
 
   it should "slice the raw lines before converting them when a limit or offset is set" in {
@@ -284,7 +292,9 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[3:])")
+        .contains(
+          s"        _rows.extend(_texera_parse_int(l, 32) for l in $utf8Lines.readlines()[3:])"
+        )
     )
 
     fileScanOpDesc.fileScanOffset = None
@@ -292,7 +302,9 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[:5])")
+        .contains(
+          s"        _rows.extend(_texera_parse_int(l, 32) for l in $utf8Lines.readlines()[:5])"
+        )
     )
 
     fileScanOpDesc.fileScanOffset = Option(3)
@@ -300,7 +312,9 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(
       fileScanOpDesc
         .generateStandaloneCode()
-        .contains("        _rows.extend(_texera_parse_int(l, 32) for l in _f.readlines()[3:][:5])")
+        .contains(
+          s"        _rows.extend(_texera_parse_int(l, 32) for l in $utf8Lines.readlines()[3:][:5])"
+        )
     )
   }
 
@@ -319,13 +333,8 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     assert(code.contains("        for _name in _z.namelist():"))
     assert(code.contains("""            if _name.startswith("__MACOSX"):"""))
     assert(code.contains("            with _z.open(_name) as _f:"))
-    // A zip entry opens binary whatever the column holds, so a text line is
-    // decoded rather than read through a text-mode handle.
-    assert(
-      code.contains(
-        """                _rows.extend(l.rstrip("\n") for l in io.TextIOWrapper(_f, encoding="utf-8"))"""
-      )
-    )
+    // A zip entry is read the way a file is: decoded from its bytes, then split.
+    assert(code.contains(s"""                _rows.extend(l.rstrip("\\n") for l in $utf8Lines)"""))
     assert(desc.standaloneImports() == Seq("import io", "import zipfile"))
     assert(!code.contains("WARNING"))
   }
@@ -338,7 +347,7 @@ class FileScanOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     )
     val code = desc.generateStandaloneCode()
 
-    assert(code.contains("""                _rows.append((_name, _f.read().decode("utf-8")))"""))
+    assert(code.contains(s"                _rows.append((_name, $utf8Whole))"))
     assert(code.endsWith("""out1df = pd.DataFrame(_rows, columns=["filename", "line"])"""))
     assert(desc.standaloneImports() == Seq("import zipfile"))
   }