carloea2 commented on code in PR #8341:
URL: https://github.com/apache/texera/pull/8341#discussion_r3975416110
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala:
##########
@@ -24,25 +24,74 @@ 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")
var flatten: Boolean = false
fileTypeName = Option("JSONL")
+ override def generateStandaloneCode(): String = {
+ val basename = sourceBasename(fileName.getOrElse(""))
+ val enc = fileEncoding.toString.replace("_", "-").toLowerCase
+
+ val readArgs = scala.collection.mutable.ArrayBuffer[String]()
+ readArgs += pyStringLiteral(basename)
+ readArgs += "lines=True"
+ 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(", ")}]"
+
+ if (offset.isEmpty) limit.foreach(l => readArgs += s"nrows=$l")
+
+ val readExpr = s"pd.read_json(${readArgs.mkString(", ")})"
Review Comment:
JSONL offset is applied after parsing the entire file. A file containing
`invalid` followed by `{"id":1}` and `{"id":2}`, with offset 1, runs in the
native source but the exported script raises ValueError on the skipped first
line. Please apply the configured line window before JSON parsing.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala:
##########
@@ -145,4 +151,74 @@ 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(", ")})"
+
+ if (hasHeader) {
+ // A blank header position is named by both readers, differently: the
schema above calls
+ // it column-N, pandas calls it "Unnamed: N". A downstream operator
names the column the
+ // schema gave it, so the frame has to carry that name. Only a position
pandas actually
+ // filled in is renamed, which is why the placeholder is matched against
the index rather
+ // than by its prefix: a column genuinely called "Unnamed: 3" keeps its
name anywhere but
+ // position 3.
+ s"""$readCall
+ |out1df.columns = [
+ | f"column-{i + 1}" if c == f"Unnamed: {i}" else c for i, c in
enumerate(out1df.columns)
Review Comment:
This also renames a real header. I tested `id,Unnamed: 1,name` with row
`1,2,alice`: Texera keeps `Unnamed: 1`, but the export changes it to
`column-2`. A downstream operator referencing the declared name then fails.
Please distinguish an explicitly supplied header from a blank one before
renaming.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
