kz930 commented on code in PR #8704:
URL: https://github.com/apache/texera/pull/8704#discussion_r4138041956
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala:
##########
@@ -24,25 +24,199 @@ 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,
StandaloneHelpers}
+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 executor reads every value as the text Jackson gives it, so a column
+ // this operator types STRING holds that text whatever the JSON held: 35.0
is
+ // "35.0", true is "true", and a null is "null", which is also why one null
+ // makes its column STRING. read_json keeps the values typed and the null a
+ // null. These columns are rebuilt from the lines the way LONG ones are.
+ val stringColumns: Seq[String] =
+ Try(sourceSchema()).toOption.toSeq.flatMap(
+ _.getAttributes.filter(_.getType ==
AttributeType.STRING).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 || stringColumns.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"). Every column is read as it
+ // was written, and the ones this operator decided were timestamps are
+ // parsed below, the way it parses them, once the frame holds them: under
+ // flattening a nested value is no column until the flattening builds it.
+ readArgs += "convert_dates=False"
+
+ val readExpr = s"pd.read_json(${readArgs.mkString(", ")})"
Review Comment:
Confirmed with pandas 2.2.3. Added `precise_float=True` and a regression
test that reads `1.0000000000000002` back exactly. The test fails without the
option. Fixed in 039d5f9f2.
--
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]
