feat(workflow-operator): export the CSV sources as Python

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

@@ -21,6 +21,7 @@ package org.apache.texera.amber.operator
 
 import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
 import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 
 import java.net.URLDecoder
 import java.nio.charset.StandardCharsets
@@ -67,21 +68,34 @@ trait StandaloneCodeGenerator {
   }
 
   /**
-    * The file's own name, for a script that reads it from its own directory
-    * rather than through Texera's resolved URI.
+    * The file this operator reads, as Texera resolved it, or None for one that
+    * reads no file.
+    *
+    * The script cannot open a resolved URI, so the body writes
+    * [[StandaloneCodeGenerator.SourceFilePlaceholder]] where the file should be
+    * named and the translator puts a name there. The name has to come from the
+    * plan: two sources reading different files whose paths end in the same
+    * segment both asked for `data.csv`, and the script read one of them twice.
+    */
+  def standaloneSourcePath(): Option[String] = None
+
+  /**
+    * The name to offer for [[standaloneSourcePath]], before the plan has had a
+    * chance to say whether another source already wants it.
     *
     * Taken from the last path segment instead of by parsing the whole string as a
     * URI: the resolver percent-encodes the file-relative segments but leaves the
     * repository and version names as the user typed them, so a dataset version
     * called `v3 - with long text` makes `new URI` throw on the space and no code
     * is generated at all.
     */
-  protected def sourceBasename(rawPath: String): String = {
-    val segment = rawPath.split("/").lastOption.getOrElse("")
-    // Percent-decoding only, matching what `URI.getPath` used to return here: form
-    // decoding would also turn a literal `+` in a file name into a space.
-    URLDecoder.decode(segment.replace("+", "%2B"), StandardCharsets.UTF_8)
-  }
+  final def standaloneSourceName(): Option[String] =
+    standaloneSourcePath().map { rawPath =>
+      val segment = rawPath.split("/").lastOption.getOrElse("")
+      // Percent-decoding only, matching what `URI.getPath` used to return here: form
+      // decoding would also turn a literal `+` in a file name into a space.
+      URLDecoder.decode(segment.replace("+", "%2B"), StandardCharsets.UTF_8)
+    }
 
   def producesDataFrame(): Boolean = true
 
@@ -110,3 +124,75 @@ trait StandaloneCodeGenerator {
     */
   def standaloneImports(): Seq[String] = Seq.empty
 }
+
+object StandaloneCodeGenerator {
+
+  /**
+    * What a source writes where the file it reads should be named.
+    *
+    * A bare identifier rather than a string literal, because the translator only
+    * rewrites the code parts of a body and leaves literals and comments alone.
+    */
+  val SourceFilePlaceholder: String = "sourceFile"
+
+  /** Python that gives an empty read the columns and types its schema declares.
+    *
+    * With no row there is nothing to infer a type from, so pandas leaves every
+    * column an object, and a JSONL read of no lines has no columns at all. The
+    * executor still declares the schema it inferred from the file, so the next
+    * step looks for those columns in those types. A read with rows is left alone.
+    */
+  def typeAnEmptyRead(frame: String, schema: Schema): String = {
+    val names = schema.getAttributes.map(a => pyStringLiteral(a.getName))
+    val dtypes = schema.getAttributes.flatMap { a =>
+      emptyDtypes.get(a.getType).map(d => s"""${pyStringLiteral(a.getName)}: "$d"""")
+    }
+    s"""if $frame.empty:
+       |    $frame = $frame.reindex(columns=[${names.mkString(", ")}]).astype({${dtypes
+      .mkString(", ")}})""".stripMargin
+  }
+
+  /** Python that reads each TIMESTAMP column of `schema`, held as text, the way
+    * the engine reads it, with `_texera_text_to_timestamp` from
+    * [[StandaloneHelpers.AttributeCasts]], which the caller declares.
+    *
+    * pandas parses a date column into nanoseconds, which end in 2262, and left a
+    * later year as text. The engine hands each field to DateParserUtils, so a
+    * row states its own format, any year reads, and the reading is cut to the
+    * millisecond.
+    */
+  def parseTimestamps(frame: String, schema: Schema): String =
+    schema.getAttributes
+      .filter(_.getType == AttributeType.TIMESTAMP)
+      .map { a =>
+        val name = pyStringLiteral(a.getName)
+        s"$frame[$name] = _texera_text_to_timestamp($frame[$name])"
+      }
+      .mkString("\n")
+
+  /** Python that reads each DOUBLE column of `schema` that pandas left as text
+    * the way the engine reads it, with `_texera_read_doubles` from
+    * [[StandaloneHelpers.AttributeCasts]], which the caller declares.
+    *
+    * The engine reads a field with `toDouble`, so "NaN", "2d" and "0x1p3" are
+    * numbers to it, and the column was inferred DOUBLE. pandas' float parser
+    * takes none of them and kept the whole column as strings.
+    */
+  def parseDoubles(frame: String, schema: Schema): String =
+    schema.getAttributes
+      .filter(_.getType == AttributeType.DOUBLE)
+      .map { a =>
+        val name = pyStringLiteral(a.getName)
+        s"$frame[$name] = _texera_read_doubles($frame[$name])"
+      }
+      .mkString("\n")
+
+  private val emptyDtypes: Map[AttributeType, String] = Map(
+    AttributeType.INTEGER -> "Int32",
+    AttributeType.LONG -> "Int64",
+    AttributeType.DOUBLE -> "float64",
+    AttributeType.BOOLEAN -> "boolean",
+    AttributeType.TIMESTAMP -> "datetime64[us]",
+    AttributeType.STRING -> "object"
+  )
+}

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

@@ -144,6 +144,43 @@ object StandaloneHelpers {
       |    return float(x)
       |
       |
+      |def _texera_java_double(text):
+      |    # Double.parseDouble on the trimmed text, which a read field reaches
+      |    # through `toDouble`. It takes a trailing d or f as a type suffix and
+      |    # 0x...p... as a hex literal, both of which float() refuses. A hex
+      |    # literal too large for a double is an infinity in Java, where
+      |    # float.fromhex raises.
+      |    text = text.strip()
+      |    if text[-1:] in ("d", "D", "f", "F"):
+      |        text = text[:-1]
+      |    if text.lstrip("+-")[:2].lower() == "0x":
+      |        try:
+      |            return float.fromhex(text)
+      |        except OverflowError:
+      |            return float("-inf") if text.startswith("-") else float("inf")
+      |    return float(text)
+      |
+      |
+      |def _texera_read_doubles(s):
+      |    # pandas leaves a DOUBLE column as text when a cell reads in Java and
+      |    # not as a plain float: "NaN", "2d", "0x1p3". Read those the engine's
+      |    # way. "NaN" is a value, and float64 reports a NaN as missing to every
+      |    # operator that reads the mask, so a column holding one goes into the
+      |    # nullable Float64, which keeps the empty cells alone null.
+      |    import numpy as np
+      |
+      |    if pd.api.types.is_numeric_dtype(s):
+      |        return s
+      |    holes = s.isna().to_numpy()
+      |    values = np.array(
+      |        [np.nan if hole else _texera_java_double(x) for x, hole in zip(s, holes)],
+      |        dtype="float64",
+      |    )
+      |    if np.isnan(values[~holes]).any():
+      |        values = pd.arrays.FloatingArray(values, holes)
+      |    return pd.Series(values, index=s.index, name=s.name)
+      |
+      |
       |def _texera_cast_string(s):
       |    # `toString` on the field, so the COLUMN's type decides the text and
       |    # not the shape of the value: a double keeps its point, whether or not

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

@@ -82,9 +82,15 @@ abstract class ScanSourceOpDesc extends SourceOperatorDescriptor {
 
   /** Rows actually used for type inference: INFER_READ_LIMIT, capped by `windowLimit`
     * when smaller.
+    *
+    * A window of no rows is still a window on this file, and the file's columns do
+    * not depend on how many of its rows were asked for, so a Limit of 0 samples as
+    * if no limit were set. Capping by it left nothing to infer from, and the
+    * operator declared a schema of no columns at all.
     */
   @JsonIgnore
-  def inferSampleSize: Int = windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)
+  def inferSampleSize: Int =
+    windowLimit.filter(_ > 0).getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)
 
   override def sourceSchema(): Schema = null
 

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala

@@ -28,15 +28,19 @@ 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, StandaloneHelpers}
+import org.apache.texera.amber.operator.StandaloneCodeGenerator.SourceFilePlaceholder
 import org.apache.texera.amber.operator.metadata.annotations.UIWidget
 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.
@@ -146,4 +150,110 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
 
   }
 
+  override def standaloneSourcePath(): Option[String] = fileName
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
+
+  override def generateStandaloneCode(): String = {
+    // 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=$SourceFilePlaceholder"
+    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 column holding a null has to be asked for, or pandas widens it to carry
+    // the hole: an integer through a float, where a LONG past 2^53 comes back
+    // rounded (9007199254740993 as ...992) and every INTEGER is left a float
+    // that prints 3 as 3.0, and a boolean to an object column. The nullable
+    // dtypes carry the hole and keep the type the schema declared. A timestamp
+    // is read as its text and parsed below, the way the engine parses it.
+    //
+    // By position, header or not, because the schema's names are not pandas'
+    // until the rename below: a blank header is `column-2` here and `Unnamed: 1`
+    // there, and asking for `column-2` ended the read. pandas takes an integer
+    // here as a position even where a header spells one. A schema that cannot be
+    // read (an unresolved file) leaves the argument off rather than failing the
+    // export.
+    val nullableDtypes = Map(
+      AttributeType.INTEGER -> "Int32",
+      AttributeType.LONG -> "Int64",
+      AttributeType.BOOLEAN -> "boolean",
+      AttributeType.TIMESTAMP -> "object"
+    )
+    val typedColumns: Seq[String] =
+      Try(sourceSchema()).toOption.toSeq.flatMap(
+        _.getAttributes.zipWithIndex.flatMap {
+          case (attribute, i) => nullableDtypes.get(attribute.getType).map(d => s"""$i: "$d"""")
+        }
+      )
+    if (typedColumns.nonEmpty) args += s"dtype={${typedColumns.mkString(", ")}}"
+
+    // Clamped: the property editor refuses a negative, but a plan posted to the API
+    // can still carry one, and pandas rejects a negative `nrows` outright where the
+    // executor's `take` simply keeps no rows.
+    offset.map(_.max(0)).foreach { o =>
+      // With a header, skip offset rows after row 0; without, skip offset rows from the start.
+      // The end is counted in Long: the largest offset the operator accepts
+      // overflows an Int on the way past the header, and the range came out
+      // empty, skipping nothing where the executor's `drop` keeps no rows. Asked
+      // of each line rather than listed, since pandas makes a set of a list of
+      // rows to skip, and one of two billion ran the script out of memory.
+      if (hasHeader) args += s"skiprows=lambda _i: 0 < _i <= ${o.toLong}"
+      else args += s"skiprows=$o"
+    }
+    limit.map(_.max(0)).foreach(l => args += s"nrows=$l")
+
+    // Without a header, an offset at or past the last row left pandas no line to
+    // count the columns from, and it raised EmptyDataError where the executor
+    // emits no rows, so the rename and the typing below never ran. Naming the
+    // columns by position gives it the count and keeps the dtype keys above.
+    if (!hasHeader)
+      Try(sourceSchema()).toOption.map(_.getAttributes.size).filter(_ > 0).foreach { n =>
+        args += s"names=[${(0 until n).mkString(", ")}]"
+      }
+
+    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)
+      Seq(
+        readCall,
+        s"out1df.columns = [${schemaNames.mkString(", ")}]",
+        StandaloneCodeGenerator.parseTimestamps("out1df", sourceSchema()),
+        StandaloneCodeGenerator.parseDoubles("out1df", sourceSchema()),
+        StandaloneCodeGenerator.typeAnEmptyRead("out1df", sourceSchema())
+      ).filter(_.nonEmpty).mkString("\n")
+    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
+    }
+  }
 }