kz930 commented on code in PR #8341:
URL: https://github.com/apache/texera/pull/8341#discussion_r4088288648
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala:
##########
@@ -38,10 +40,91 @@ 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")
+ // pyarrow is what reads an Arrow file, pandas going through it; the block
+ // below asks it for the file's own columns.
+ override def standaloneImports(): Seq[String] = Seq("import pyarrow as pa")
+
+ override def standaloneSourcePath(): Option[String] = fileName
+
+ override def generateStandaloneCode(): String = {
+ // Arrow says of every value whether it is there, and a numpy column has
+ // nowhere to put that: a missing double and a stored NaN both land on NaN.
+ // The nullable dtypes keep a holed integer integral too, so a long past
2^53
+ // is not rounded on its way through a float. Named here because the read
+ // below cannot ask pd.read_feather for them, and named for the types
+ // ArrowUtils.toAttributeType reads; a timestamp needs none, datetime64
+ // having a slot of its own for a missing value.
+ val dtypes =
+ """|_nullable = {
+ | pa.bool_(): pd.BooleanDtype(),
+ | pa.string(): pd.StringDtype(),
+ | pa.float32(): pd.Float32Dtype(),
+ | pa.float64(): pd.Float64Dtype(),
+ | pa.int8(): pd.Int8Dtype(),
+ | pa.int16(): pd.Int16Dtype(),
+ | pa.int32(): pd.Int32Dtype(),
+ | pa.int64(): pd.Int64Dtype(),
+ | pa.uint8(): pd.UInt8Dtype(),
+ | pa.uint16(): pd.UInt16Dtype(),
+ | pa.uint32(): pd.UInt32Dtype(),
+ |}""".stripMargin
+ // A file pandas wrote notes in its schema what the frame it came from
looked
+ // like, and pandas reads that note back: a column the frame was keyed by
+ // returns as the index, and numbered labels return as numbers where the
file
+ // says "1" and "2". The executor reads the columns the file states, so the
+ // note is refused rather than undone, as the Parquet source refuses it.
+ val read =
+ s"""|with pa.ipc.open_file($SourceFilePlaceholder) as _file:
+ | _table = _file.read_all()
+ |out1df = _table.to_pandas(ignore_metadata=True,
types_mapper=_nullable.get)""".stripMargin
Review Comment:
Fixed in 719cb935d. A zoned timestamp now drops its zone and keeps the
file's wall clock. The spec uses an Asia/Tokyo column and checks that the
engine and the generated script both read 09:00 and survive the datetime64 cast.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala:
##########
@@ -145,4 +156,97 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc {
}
+ override def standaloneSourcePath(): Option[String] = fileName
+
+ 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 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(",
")}]"
Review Comment:
Fixed in 5bc1a0df2. All three CSV scans now pass parse_dates and dtype by
position, so a blank header no longer misses. The spec checks the generated
read for the `id,` header.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala:
##########
@@ -24,25 +24,151 @@ 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.StandaloneCodeGenerator.SourceFilePlaceholder
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 standaloneSourcePath(): Option[String] = fileName
+
+ override def generateStandaloneCode(): String = {
+ 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
+
+ // read_json parses a JSON number into a float before any dtype it is
handed
+ // can apply, so a LONG past 2^53 is already rounded by the time the column
+ // exists: 9007199254740993 arrives as ...992, a value the file never held.
+ // Reading the lines a second time with Python's own parser, which keeps an
+ // integer exact, is the only way to put the column back.
+ val longColumns: Seq[String] =
+ Try(sourceSchema()).toOption.toSeq.flatMap(
+ _.getAttributes.filter(_.getType == AttributeType.LONG).map(_.getName)
+ )
+
+ // The lines are read into the script whenever something below needs them,
+ // and read_json is then fed from those rather than from the file, so the
+ // file is opened once either way.
+ val readsLines = windowed || longColumns.nonEmpty
+ val dropped = offset.filter(_ > 0).fold("_lines")(o => s"_lines[$o:]")
+ val taken = limit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+
+ val readArgs = scala.collection.mutable.ArrayBuffer[String]()
+ readArgs += (if (readsLines) s"""io.StringIO("".join($taken))""" else
SourceFilePlaceholder)
+ readArgs += "lines=True"
+ // Text already decoded by the open above carries no encoding of its own.
+ if (!readsLines) 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))
+ )
+ // Under flattening the schema names a nested value for the column the
+ // flattening is about to build, and read_json is asked about that name
+ // while the file still holds the object around it. It finds no such
column,
+ // converts nothing, and the value reaches the plan as text, where a sort
+ // puts a 2025 date before a 2024 one. Those columns are converted once the
+ // frame that holds them exists, below.
+ if (!flatten) readArgs += s"convert_dates=[${dateColumns.mkString(", ")}]"
+
+ val readExpr = s"pd.read_json(${readArgs.mkString(", ")})"
Review Comment:
Fixed in 053a6fdf0. A boolean column with a missing key is cast to pandas'
nullable boolean. The spec runs both paths on the true / missing / false rows
and checks a column without a hole stays bool.
--
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]
