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/arrow/ArrowSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
index ad1d7a34176..5114e3560ed 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala
@@ -25,8 +25,10 @@ import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.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.source.scan.ScanSourceOpDesc
 import org.apache.texera.amber.util.ArrowUtils
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.apache.arrow.memory.RootAllocator
 import org.apache.arrow.vector.ipc.ArrowFileReader
@@ -38,10 +40,31 @@ import java.nio.file.{Files, StandardOpenOption}
 import scala.util.Using
 
 @JsonIgnoreProperties(value = Array("fileEncoding"))
-class ArrowSourceOpDesc extends ScanSourceOpDesc {
+class ArrowSourceOpDesc extends ScanSourceOpDesc with StandaloneCodeGenerator {
 
   fileTypeName = Option("Arrow")
 
+  override def generateStandaloneCode(): String = {
+    val basename = sourceBasename(fileName.getOrElse(""))
+    val read = s"""out1df = pd.read_feather(${pyStringLiteral(basename)})"""
+    // A timestamp column needs nothing here. The file names UTC and holds the
+    // wall clock as UTC, so pd.read_feather and the executor read the same
+    // reading off it — no zone of the reader's own enters either side. The other
+    // scan sources have to name their date columns, CSV and JSONL carrying no
+    // types to go on, but Arrow states its own.
+    //
+    // The executor drops `offset` rows and then takes `limit` of them. Feather has
+    // no row-range read, so the same window is taken once the frame is in memory.
+    val window = (offset, limit) match {
+      case (Some(o), Some(l)) => Some(s"$o:${o + l}")
+      case (Some(o), None)    => Some(s"$o:")
+      case (None, Some(l))    => Some(s":$l")
+      case _                  => None
+    }
+    (read +: window.map(w => s"out1df = out1df.iloc[$w].reset_index(drop=True)").toSeq)
+      .mkString("\n")
+  }
+
   @throws[IOException]
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
index 69088abb6c9..1e35cef8523 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala
@@ -28,22 +28,28 @@ import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 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.source.scan.ScanSourceOpDesc
 import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpExec
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 import java.io.{IOException, InputStreamReader}
 import java.net.URI
+import scala.util.Try
 
-class CSVScanSourceOpDesc extends ScanSourceOpDesc {
+class CSVScanSourceOpDesc extends ScanSourceOpDesc with StandaloneCodeGenerator {
 
   // One character: every reader narrows this with charAt(0), because univocity's
   // setDelimiter and scala-csv's DefaultCSVFormat both take a Char.
+  //
+  // `examples` names a delimiter the fixture's rows do not contain, so the
+  // verification config generator does not pick one that parses them ragged.
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
   @JsonPropertyDescription("single character separating the fields on each line")
   @JsonInclude(JsonInclude.Include.NON_ABSENT)
-  @JsonSchemaInject(json = """{ "maxLength": 1 }""")
+  @JsonSchemaInject(json = """{ "maxLength": 1, "examples": [","] }""")
   var customDelimiter: Option[String] = None
 
   @JsonProperty(defaultValue = "true")
@@ -145,4 +151,77 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
 
   }
 
+  override def generateStandaloneCode(): String = {
+    // Strip to just the basename. The standalone script assumes the CSV
+    // lives in the same directory as the script (Texera's resolved URIs
+    // can't be used directly outside the system).
+    val basename = sourceBasename(fileName.getOrElse(""))
+
+    // Resolve the delimiter the same way the parser above does — first character, empty
+    // means comma — and escape it. Every value the field accepts has to survive this:
+    // pandas reads a separator longer than one character as a REGULAR EXPRESSION, and a
+    // backslash spliced raw produced `sep="\"`, which is not valid Python at all.
+    val sep = customDelimiter.filter(_.nonEmpty).getOrElse(",").charAt(0).toString
+    // Texera's encoding enum uses values like UTF_8; pandas expects utf-8.
+    val encoding = fileEncoding.toString.replace("_", "-").toLowerCase
+    val headerArg = if (hasHeader) "0" else "None"
+
+    val args = scala.collection.mutable.ArrayBuffer[String]()
+    args += s"""filepath_or_buffer=${pyStringLiteral(basename)}"""
+    args += s"sep=${pyStringLiteral(sep)}"
+    args += s"""encoding=${pyStringLiteral(encoding)}"""
+    args += s"header=$headerArg"
+
+    // The parser above sets no null value, so only an empty field is null and every other
+    // text stands for itself. pandas instead reads a list of words as missing by default,
+    // "NA" and "null" among them, which turned a column holding the country code NA into
+    // nulls. Both halves are needed: dropping the default list stops the words, and naming
+    // the empty string keeps the blank cell null.
+    args += "keep_default_na=False"
+    args += """na_values=[""]"""
+
+    // A CSV carries no types, so both readers infer, and they do not infer
+    // alike: the schema above tries TIMESTAMP and parses what it can, while
+    // pd.read_csv leaves a date column as text. Name the columns this operator
+    // decided were timestamps so pandas parses the same ones — by position when
+    // there is no header, the frame's columns having no names until the rename
+    // below. A schema that cannot be read (an unresolved file) leaves the
+    // argument off rather than failing the export.
+    val dateColumns: Seq[String] =
+      Try(sourceSchema()).toOption.toSeq.flatMap(
+        _.getAttributes.zipWithIndex
+          .filter(_._1.getType == AttributeType.TIMESTAMP)
+          .map { case (a, i) => if (hasHeader) pyStringLiteral(a.getName) else i.toString }
+      )
+    if (dateColumns.nonEmpty) args += s"parse_dates=[${dateColumns.mkString(", ")}]"
+
+    offset.foreach { o =>
+      // With a header, skip offset rows after row 0; without, skip offset rows from the start.
+      if (hasHeader) args += s"skiprows=range(1, ${o + 1})"
+      else args += s"skiprows=$o"
+    }
+    limit.foreach(l => args += s"nrows=$l")
+
+    val readCall = s"out1df = pd.read_csv(${args.mkString(", ")})"
+
+    // The schema's own names, which every downstream operator was configured
+    // against. They differ from pandas' in both directions: a blank header is
+    // `column-2` here and `Unnamed: 1` there, and a header the user really did
+    // spell `Unnamed: 1` is kept. Matching the placeholder against the index
+    // cannot tell those two apart when they coincide; taking the names by
+    // position can.
+    val schemaNames: Seq[String] =
+      Try(sourceSchema()).toOption.toSeq
+        .flatMap(_.getAttributes.map(a => pyStringLiteral(a.getName)))
+
+    if (schemaNames.nonEmpty)
+      s"""$readCall
+         |out1df.columns = [${schemaNames.mkString(", ")}]""".stripMargin
+    else if (hasHeader) readCall
+    else {
+      // Unresolved file: fall back to Texera's headerless naming.
+      s"""$readCall
+         |out1df.columns = [f"column-{i + 1}" for i in range(len(out1df.columns))]""".stripMargin
+    }
+  }
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
index cdade62d1e1..902284ce520 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala
@@ -29,20 +29,22 @@ import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 import org.apache.texera.amber.core.tuple.{Attribute, 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.source.scan.ScanSourceOpDesc
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 import java.io.IOException
 import java.net.URI
 
-class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
+class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc with StandaloneCodeGenerator {
 
   // One character -- see CSVScanSourceOpDesc.
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
   @JsonPropertyDescription("single character separating the fields on each line")
   @JsonDeserialize(contentAs = classOf[java.lang.String])
-  @JsonSchemaInject(json = """{ "maxLength": 1 }""")
+  @JsonSchemaInject(json = """{ "maxLength": 1, "examples": [","] }""")
   var customDelimiter: Option[String] = None
 
   @JsonProperty(defaultValue = "true")
@@ -80,6 +82,35 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc {
       )
   }
 
+  override def generateStandaloneCode(): String = {
+    val basename = sourceBasename(fileName.getOrElse(""))
+    // First character, empty means comma — the same resolution the reader below does —
+    // and escaped, so every value the field accepts survives being spliced into Python.
+    // See CSVScanSourceOpDesc for what handing pandas the raw value did.
+    val sep = customDelimiter.filter(_.nonEmpty).getOrElse(",").charAt(0).toString
+    val encoding = fileEncoding.toString.replace("_", "-").toLowerCase
+    val headerArg = if (hasHeader) "0" else "None"
+
+    val args = scala.collection.mutable.ArrayBuffer[String]()
+    args += s"""filepath_or_buffer=${pyStringLiteral(basename)}"""
+    args += s"sep=${pyStringLiteral(sep)}"
+    args += s"""encoding=${pyStringLiteral(encoding)}"""
+    args += s"header=$headerArg"
+
+    offset.foreach { o =>
+      if (hasHeader) args += s"skiprows=range(1, ${o + 1})"
+      else args += s"skiprows=$o"
+    }
+    limit.foreach(l => args += s"nrows=$l")
+
+    val readCall = s"out1df = pd.read_csv(${args.mkString(", ")})"
+
+    if (hasHeader) readCall
+    else
+      s"""$readCall
+         |out1df.columns = [f"column-{i + 1}" for i in range(len(out1df.columns))]""".stripMargin
+  }
+
   override def sourceSchema(): Schema = {
     val delimiterChar = customDelimiter.filter(_.nonEmpty).getOrElse(",").charAt(0)
     require(
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
index 57dec76c556..a7b7607c6bb 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala
@@ -28,19 +28,22 @@ import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
 import org.apache.texera.amber.core.tuple.{Attribute, 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.source.scan.ScanSourceOpDesc
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 import java.io.IOException
 import java.net.URI
+import scala.util.Try
 
-class CSVOldScanSourceOpDesc extends ScanSourceOpDesc {
+class CSVOldScanSourceOpDesc extends ScanSourceOpDesc with StandaloneCodeGenerator {
 
   // One character -- see CSVScanSourceOpDesc.
   @JsonProperty(defaultValue = ",")
   @JsonSchemaTitle("Delimiter")
   @JsonPropertyDescription("single character separating the fields on each line")
-  @JsonSchemaInject(json = """{ "maxLength": 1 }""")
+  @JsonSchemaInject(json = """{ "maxLength": 1, "examples": [","] }""")
   var customDelimiter: Option[String] = Some(",")
 
   @JsonProperty(defaultValue = "true")
@@ -76,6 +79,45 @@ class CSVOldScanSourceOpDesc extends ScanSourceOpDesc {
       )
   }
 
+  override def generateStandaloneCode(): String = {
+    val basename = sourceBasename(fileName.getOrElse(""))
+    // First character, empty means comma — the same resolution the reader below does —
+    // and escaped, so every value the field accepts survives being spliced into Python.
+    // See CSVScanSourceOpDesc for what handing pandas the raw value did.
+    val sep = customDelimiter.filter(_.nonEmpty).getOrElse(",").charAt(0).toString
+    val encoding = fileEncoding.toString.replace("_", "-").toLowerCase
+    val headerArg = if (hasHeader) "0" else "None"
+
+    val args = scala.collection.mutable.ArrayBuffer[String]()
+    args += s"""filepath_or_buffer=${pyStringLiteral(basename)}"""
+    args += s"sep=${pyStringLiteral(sep)}"
+    args += s"""encoding=${pyStringLiteral(encoding)}"""
+    args += s"header=$headerArg"
+
+    // Name the columns this operator inferred as timestamps, so pandas parses
+    // the same ones instead of leaving them as text. See CSVScanSourceOpDesc.
+    val dateColumns: Seq[String] =
+      Try(sourceSchema()).toOption.toSeq.flatMap(
+        _.getAttributes.zipWithIndex
+          .filter(_._1.getType == AttributeType.TIMESTAMP)
+          .map { case (a, i) => if (hasHeader) pyStringLiteral(a.getName) else i.toString }
+      )
+    if (dateColumns.nonEmpty) args += s"parse_dates=[${dateColumns.mkString(", ")}]"
+
+    offset.foreach { o =>
+      if (hasHeader) args += s"skiprows=range(1, ${o + 1})"
+      else args += s"skiprows=$o"
+    }
+    limit.foreach(l => args += s"nrows=$l")
+
+    val readCall = s"out1df = pd.read_csv(${args.mkString(", ")})"
+
+    if (hasHeader) readCall
+    else
+      s"""$readCall
+         |out1df.columns = [f"column-{i + 1}" for i in range(len(out1df.columns))]""".stripMargin
+  }
+
   override def sourceSchema(): Schema = {
     val delimiterChar = customDelimiter.filter(_.nonEmpty).getOrElse(",").charAt(0)
     require(
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..3948298d16e 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,73 @@ 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]()
+
+    if (extract)
+      buf += "# WARNING: extract=true is not supported in standalone mode; files are read as-is, not unpacked from archives."
+
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+    val openArgs =
+      if (isBinary) """"rb""""
+      else s""""r", encoding=${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)):"
+    buf += s"    with open(_fn, $openArgs) as _f:"
+
+    // Match the platform (FileScanUtils.createTuplesFromFile): its line-by-line
+    // branch ignores outputFileName and emits only the value, so the filename
+    // column is added ONLY in single-value mode.
+    val emitFilename = outputFileName && attributeType.isSingle
+
+    if (attributeType.isSingle) {
+      if (emitFilename) buf += "        _rows.append((_fn, _f.read()))"
+      else buf += "        _rows.append(_f.read())"
+    } 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   => """l.rstrip().lower() == "true""""
+        case FileAttributeType.TIMESTAMP => "pd.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.
+      val linesExpr =
+        if (fileScanOffset.isEmpty && fileScanLimit.isEmpty) "_f"
+        else {
+          val dropped =
+            fileScanOffset.filter(_ > 0).fold("_f.readlines()")(o => s"_f.readlines()[$o:]")
+          fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+        }
+      buf += s"        _rows.extend($castExpr 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.mkString("\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 82997632d14..e5a82a1a418 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,22 @@ 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.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(
@@ -64,6 +73,61 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with TextSourceOpDesc {
 
   fileTypeName = Option("")
 
+  override def generateStandaloneCode(): String = {
+    val basename = sourceBasename(fileName.getOrElse(""))
+    val col = attributeName
+    val enc = encoding.toString.replace("_", "-").toLowerCase
+    val basenameLit = pyStringLiteral(basename)
+    val colLit = pyStringLiteral(col)
+    val encLit = pyStringLiteral(enc)
+    val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+    if (extract)
+      buf += s"""# WARNING: extract=true is not supported in standalone mode; provide the unarchived $basenameLit directly."""
+
+    val isBinary =
+      attributeType == FileAttributeType.BINARY || attributeType == FileAttributeType.LARGE_BINARY
+
+    if (attributeType.isSingle) {
+      val openArgs =
+        if (isBinary) s"""$basenameLit, "rb""""
+        else s"""$basenameLit, "r", encoding=$encLit"""
+      val dfCols =
+        if (outputFileName) s"""{"filename": $basenameLit, $colLit: [_f.read()]}"""
+        else s"""{$colLit: [_f.read()]}"""
+      buf += s"""with open($openArgs) as _f:"""
+      buf += s"""    out1df = pd.DataFrame($dfCols)"""
+    } 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   => """l.rstrip().lower() == "true""""
+        case FileAttributeType.TIMESTAMP => "pd.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.
+      val linesExpr =
+        if (fileScanOffset.isEmpty && fileScanLimit.isEmpty) "_f"
+        else {
+          val dropped =
+            fileScanOffset.filter(_ > 0).fold("_f.readlines()")(o => s"_f.readlines()[$o:]")
+          fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+        }
+      val dfCols =
+        if (outputFileName)
+          s"""{"filename": $basenameLit, $colLit: [$castExpr for l in $linesExpr]}"""
+        else s"""{$colLit: [$castExpr for l in $linesExpr]}"""
+      buf += s"""with open($basenameLit, "r", encoding=$encLit) as _f:"""
+      buf += s"""    out1df = pd.DataFrame($dfCols)"""
+    }
+
+    buf.mkString("\n")
+  }
+
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
index faccc76f882..944a8588b4e 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala
@@ -24,18 +24,21 @@ import com.fasterxml.jackson.databind.JsonNode
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.AttributeTypeUtils.inferSchemaFromRows
-import org.apache.texera.amber.core.tuple.{Attribute, Schema}
+import org.apache.texera.amber.core.tuple.{Attribute, 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.source.scan.ScanSourceOpDesc
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.{JSONToMap, objectMapper}
 
 import java.io._
 import java.net.URI
 import scala.collection.mutable.ArrayBuffer
+import scala.util.Try
 import scala.jdk.CollectionConverters.IteratorHasAsScala
 
-class JSONLScanSourceOpDesc extends ScanSourceOpDesc {
+class JSONLScanSourceOpDesc extends ScanSourceOpDesc with StandaloneCodeGenerator {
 
   @JsonProperty(required = true, defaultValue = "false")
   @JsonPropertyDescription("flatten nested objects and arrays")
@@ -43,6 +46,62 @@ class JSONLScanSourceOpDesc extends ScanSourceOpDesc {
 
   fileTypeName = Option("JSONL")
 
+  override def generateStandaloneCode(): String = {
+    val basename = sourceBasename(fileName.getOrElse(""))
+    val enc = fileEncoding.toString.replace("_", "-").toLowerCase
+
+    // The executor drops and takes on the RAW lines, before any of them is
+    // parsed, so a line outside the window is never read as JSON and an
+    // unparseable one there costs nothing. Reading the whole file first and
+    // slicing the frame would end the export on a line the workflow skipped.
+    val windowed = offset.exists(_ > 0) || limit.isDefined
+    val source =
+      if (!windowed) pyStringLiteral(basename)
+      else {
+        val dropped = offset.filter(_ > 0).fold("_lines")(o => s"_lines[$o:]")
+        val taken = limit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+        s"""io.StringIO("".join($taken))"""
+      }
+
+    val readArgs = scala.collection.mutable.ArrayBuffer[String]()
+    readArgs += source
+    readArgs += "lines=True"
+    if (!windowed) readArgs += s"""encoding=${pyStringLiteral(enc)}"""
+
+    // JSON has no timestamp of its own, so both readers infer from the text and
+    // do not infer alike: the schema below tries TIMESTAMP and parses what it
+    // can, while pd.read_json guesses from the COLUMN NAME (anything ending
+    // "_at" or "_time", anything called "date") and leaves the rest as text.
+    // Naming the columns this operator decided were timestamps settles both
+    // halves — the ones it misses and the ones it would have taken on its own.
+    // An unreadable schema leaves the argument off rather than failing the
+    // export.
+    val dateColumns: Seq[String] =
+      Try(sourceSchema()).toOption.toSeq.flatMap(
+        _.getAttributes
+          .filter(_.getType == AttributeType.TIMESTAMP)
+          .map(a => pyStringLiteral(a.getName))
+      )
+    readArgs += s"convert_dates=[${dateColumns.mkString(", ")}]"
+
+    val readExpr = s"pd.read_json(${readArgs.mkString(", ")})"
+    val baseExpr =
+      if (flatten) s"pd.json_normalize($readExpr.to_dict('records'))"
+      else readExpr
+
+    val lines = scala.collection.mutable.ArrayBuffer[String]()
+    if (windowed) {
+      lines += "import io"
+      lines += s"""with open(${pyStringLiteral(basename)}, "r", encoding=${pyStringLiteral(
+        enc
+      )}) as _f:"""
+      lines += "    _lines = _f.readlines()"
+    }
+    lines += s"out1df = $baseExpr"
+
+    lines.mkString("\n")
+  }
+
   @throws[IOException]
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
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..b9ecfa89d1c 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,40 @@ 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   => """l.lower() == "true""""
+        case FileAttributeType.TIMESTAMP => "pd.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.mkString("\n")
+  }
 }
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..654e11c02d2 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
@@ -48,6 +48,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 +70,7 @@ trait TextSourceOpDesc {
   )
   var fileScanLimit: Option[Int] = None
 
+  @JsonProperty
   @JsonSchemaTitle("Offset (lines)")
   @JsonPropertyDescription(
     "Number of lines to skip from the start before reading. " +
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/csv/CSVScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala
index fe426552a96..5fe9b31a919 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDescSpec.scala
@@ -211,6 +211,60 @@ class CSVScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     )
   }
 
+  it should "use the csv basename in standalone code" in {
+    csvScanSourceOpDesc.fileName = Some(TestOperators.CountrySalesSmallMultiLineCsvPath)
+    csvScanSourceOpDesc.customDelimiter = Some(",")
+    csvScanSourceOpDesc.hasHeader = true
+    csvScanSourceOpDesc.setResolvedFileName(FileResolver.resolve(csvScanSourceOpDesc.fileName.get))
+
+    val code = csvScanSourceOpDesc.generateStandaloneCode()
+
+    assert(code.contains("""filepath_or_buffer="country_sales_small_multi_line.csv""""))
+    assert(!code.contains("base64.b64decode"))
+    assert(!code.contains("io.BytesIO"))
+  }
+
+  it should "use the unresolved csv basename in standalone code" in {
+    csvScanSourceOpDesc.fileName = Some(TestOperators.CountrySalesSmallMultiLineCsvPath)
+    csvScanSourceOpDesc.customDelimiter = Some(",")
+    csvScanSourceOpDesc.hasHeader = true
+
+    val code = csvScanSourceOpDesc.generateStandaloneCode()
+
+    assert(code.contains("""filepath_or_buffer="country_sales_small_multi_line.csv""""))
+    assert(!code.contains("base64.b64decode"))
+    assert(!code.contains("io.BytesIO"))
+  }
+
+  // The parser sets no null value, so only an empty field is null. pandas reads a list of
+  // words as missing by default, which turned the country code NA into a null.
+  it should "read only an empty field as null, the way the parser does" in {
+    csvScanSourceOpDesc.fileName = Some(TestOperators.CountrySalesSmallMultiLineCsvPath)
+    csvScanSourceOpDesc.customDelimiter = Some(",")
+    csvScanSourceOpDesc.hasHeader = true
+
+    val code = csvScanSourceOpDesc.generateStandaloneCode()
+
+    assert(code.contains("keep_default_na=False"))
+    assert(code.contains("""na_values=[""]"""))
+  }
+
+  // sourceSchema names a blank header column-N; pandas names it "Unnamed: N". A downstream
+  // operator asks for the name the schema gave, so the frame has to carry that one, and by
+  // position rather than by matching the placeholder: a header the user really did spell
+  // "Unnamed: 1" is kept, and matching cannot tell the two apart where they coincide.
+  it should "give the frame the names the schema gives it" in {
+    val path = writeCsvWithEmptyHeader()
+    csvScanSourceOpDesc.fileName = Some(path)
+    csvScanSourceOpDesc.customDelimiter = Some(",")
+    csvScanSourceOpDesc.hasHeader = true
+    csvScanSourceOpDesc.setResolvedFileName(FileResolver.resolve(path))
+
+    val code = csvScanSourceOpDesc.generateStandaloneCode()
+
+    assert(code.contains("""out1df.columns = ["id", "name", "column-3", "age"]"""))
+  }
+
   it should "use comma as the default delimiter when customDelimiter is not set for parallel CSV" in {
     parallelCsvScanSourceOpDesc.customDelimiter = None
 
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..26ac6372b4a 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
@@ -144,4 +144,123 @@ 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})"""))
+
+    // The platform's line-by-line branch (FileScanUtils.createTuplesFromFile)
+    // emits only the value, so line mode must drop the filename column too.
+    fileScanOpDesc.attributeType = FileAttributeType.STRING
+    fileScanOpDesc.outputFileName = true
+    assert(!fileScanOpDesc.generateStandaloneCode().contains("filename"))
+  }
+
+  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 -> """l.rstrip().lower() == "true"""",
+      FileAttributeType.TIMESTAMP -> "pd.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)")
+        )
+    }
+  }
+
+  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])")
+    )
+  }
+
+  it should "warn that archive extraction is unsupported when extract is on" in {
+    // `extract` is a val, so it can only be set through deserialization.
+    val desc = objectMapper.readValue(
+      """{"operatorType":"FileScanOp","extract":true}""",
+      classOf[FileScanOpDesc]
+    )
+    assert(desc.generateStandaloneCode().startsWith("# WARNING: extract=true is not supported"))
+  }
 }
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 fafb696f131..2f3d76af9de 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
@@ -209,6 +209,35 @@ class FileScanSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     FileScanSourceOpExec.close()
   }
 
+  "FileScanSourceOpDesc.generateStandaloneCode" should
+    "slice the raw lines before converting them" in {
+    fileScanSourceOpDesc.attributeType = FileAttributeType.INTEGER
+
+    fileScanSourceOpDesc.fileScanOffset = Option(3)
+    fileScanSourceOpDesc.fileScanLimit = None
+    assert(
+      fileScanSourceOpDesc
+        .generateStandaloneCode()
+        .contains("""{"line": [int(l.rstrip()) for l in _f.readlines()[3:]]}""")
+    )
+
+    fileScanSourceOpDesc.fileScanOffset = None
+    fileScanSourceOpDesc.fileScanLimit = Option(5)
+    assert(
+      fileScanSourceOpDesc
+        .generateStandaloneCode()
+        .contains("""{"line": [int(l.rstrip()) for l in _f.readlines()[:5]]}""")
+    )
+
+    fileScanSourceOpDesc.fileScanOffset = Option(3)
+    fileScanSourceOpDesc.fileScanLimit = Option(5)
+    assert(
+      fileScanSourceOpDesc
+        .generateStandaloneCode()
+        .contains("""{"line": [int(l.rstrip()) for l in _f.readlines()[3:][:5]]}""")
+    )
+  }
+
   "FileScanSourceOpDesc.getPhysicalOp" should
     "wire the FileScanSourceOpExec class as a source op and propagate its schema" in {
     val physical =
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala
index c0d8cf2e93c..17cc308354f 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDescSpec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.texera.amber.operator.source.scan.json
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.operator.LogicalOp
@@ -28,6 +29,12 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class JSONLScanSourceOpDescSpec extends AnyFlatSpec with Matchers {
 
   private val workflowId = WorkflowIdentity(1L)
@@ -86,4 +93,76 @@ class JSONLScanSourceOpDescSpec extends AnyFlatSpec with Matchers {
     r.limit shouldBe Some(10)
     r.offset shouldBe Some(5)
   }
+
+  // The executor drops and takes on the raw lines, so a line the window skips is
+  // never read as JSON. Reading the file whole and slicing the frame afterwards
+  // ends the export on a line the workflow never looked at.
+  "JSONLScanSourceOpDesc.generateStandaloneCode" should
+    "skip a line the window excludes without parsing it" 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 dir = Files.createTempDirectory("jsonl-window-")
+    dir.toFile.deleteOnExit()
+    val data = dir.resolve("input.jsonl")
+    Files.write(
+      data,
+      "not json at all\n{\"id\":1}\n{\"id\":2}\n".getBytes(StandardCharsets.UTF_8)
+    )
+
+    val op = new JSONLScanSourceOpDesc
+    op.fileName = Some(data.toString)
+    op.offset = Some(1)
+
+    val script = dir.resolve("run.py")
+    Files.write(
+      script,
+      s"""import pandas as pd
+         |${op.generateStandaloneCode()}
+         |print(list(out1df["id"]))
+         |""".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)
+    withClue(s"python said:\n$out\n") {
+      process.exitValue() shouldBe 0
+      out.trim should endWith("[1, 2]")
+    }
+  }
+
+  // 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
+      }
 }
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..dcede030447 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
@@ -276,6 +276,36 @@ class TextInputSourceOpDescSpec extends AnyFlatSpec with BeforeAndAfter {
     new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
   }
 
+  "TextInputSourceOpDesc.generateStandaloneCode" should
+    "slice the raw lines before converting them" in {
+    textInputSourceOpDesc.attributeType = FileAttributeType.INTEGER
+    textInputSourceOpDesc.textInput = "1\n2\n3"
+
+    textInputSourceOpDesc.fileScanOffset = Option(3)
+    textInputSourceOpDesc.fileScanLimit = None
+    assert(
+      textInputSourceOpDesc
+        .generateStandaloneCode()
+        .endsWith("""{"line": [int(l) for l in _text.splitlines()[3:]]})""")
+    )
+
+    textInputSourceOpDesc.fileScanOffset = None
+    textInputSourceOpDesc.fileScanLimit = Option(5)
+    assert(
+      textInputSourceOpDesc
+        .generateStandaloneCode()
+        .endsWith("""{"line": [int(l) for l in _text.splitlines()[:5]]})""")
+    )
+
+    textInputSourceOpDesc.fileScanOffset = Option(3)
+    textInputSourceOpDesc.fileScanLimit = Option(5)
+    assert(
+      textInputSourceOpDesc
+        .generateStandaloneCode()
+        .endsWith("""{"line": [int(l) for l in _text.splitlines()[3:][:5]]})""")
+    )
+  }
+
   "TextInputSourceOpDesc.getPhysicalOp" should
     "wire the TextInputSourceOpExec class as a source op with one output port" in {
     val physical =