carloea2 commented on code in PR #8705:
URL: https://github.com/apache/texera/pull/8705#discussion_r4210118522
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala:
##########
@@ -68,6 +78,105 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with
TextSourceOpDesc {
fileTypeName = Option("")
+ override def standaloneSourcePath(): Option[String] = fileName
+
+ override def generateStandaloneCode(): String = {
+ val col = attributeName
+ // `encoding` is the charset the panel offers, and the one the executor now
+ // decodes with.
+ val enc = encoding.toString.replace("_", "-").toLowerCase
+ val colLit = pyStringLiteral(col)
+ val encLit = pyStringLiteral(enc)
+ val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+ val isBinary =
+ attributeType == FileAttributeType.BINARY || attributeType ==
FileAttributeType.LARGE_BINARY
+
+ 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 => TextSourceOpDesc.BooleanParserCall
+ case FileAttributeType.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. With extract on, the window is per
entry,
+ // as the engine's is: it drops and takes inside the flatMap over entries.
+ def windowed(lines: String): String =
+ if (fileScanOffset.isEmpty && fileScanLimit.isEmpty) lines
+ else {
+ val dropped =
+ fileScanOffset.filter(_ > 0).fold(s"$lines.readlines()")(o =>
s"$lines.readlines()[$o:]")
+ fileScanLimit.fold(dropped)(l => s"$dropped[:${l.max(0)}]")
+ }
+
+ // Whatever the flag says, as the platform now reads it: every row carries
+ // the name of the file its value came from, a line's as much as a whole
+ // file's. See FileScanUtils.createTuplesFromFile.
+ val emitFilename = outputFileName
+
+ if (extract) {
+ // The engine reads the files INSIDE the archive, one tuple per entry,
and
+ // skips the entries macOS adds. `zipfile` is the reader that matches:
the
+ // platform casts what it opened to a ZipArchiveInputStream, so an
archive
+ // that is not a zip fails on both sides. A directory entry is read, not
+ // skipped, there and here alike — it yields nothing. The filename column
+ // carries the ENTRY's name, which is what the engine puts there.
+ buf += "_rows = []"
+ buf += s"with zipfile.ZipFile($SourceFilePlaceholder) as _z:"
+ buf += " for _name in _z.namelist():"
+ buf += """ if _name.startswith("__MACOSX"):"""
+ buf += " continue"
+ buf += " with _z.open(_name) as _f:"
+ if (attributeType.isSingle) {
+ // A zip entry only opens binary, so text is decoded here rather than
by
+ // the reader.
+ val readWhole = if (isBinary) "_f.read()" else
s"_f.read().decode($encLit)"
+ if (emitFilename) buf += s" _rows.append((_name,
$readWhole))"
+ else buf += s" _rows.append($readWhole)"
+ } else {
+ // TextIOWrapper, not splitlines: it ends a line where `open(..., "r")`
+ // does, which is what the branch below reads with.
+ val linesExpr = windowed(s"io.TextIOWrapper(_f, encoding=$encLit)")
+ val row = if (emitFilename) s"(_name, $castExpr)" else castExpr
+ buf += s" _rows.extend($row for l in $linesExpr)"
+ }
+ if (emitFilename) buf += s"""out1df = pd.DataFrame(_rows,
columns=["filename", $colLit])"""
+ else buf += s"""out1df = pd.DataFrame({$colLit: _rows})"""
+ } else if (attributeType.isSingle) {
+ val openArgs =
+ if (isBinary) s"""$SourceFilePlaceholder, "rb""""
+ else s"""$SourceFilePlaceholder, "r", encoding=$encLit"""
Review Comment:
For SINGLE_STRING, opening in text mode normalizes CRLF to LF. A file
containing `first\r\nsecond\r\n` exports as `first\nsecond\n`, while
FileScanUtils decodes the bytes and preserves CRLF. Please disable newline
translation for this mode and cover it in a parity test. The non-archive
FileScanOpDesc path does the same thing.
--
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]
