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