carloea2 commented on code in PR #8341:
URL: https://github.com/apache/texera/pull/8341#discussion_r3973443172
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanSourceOpDesc.scala:
##########
@@ -64,6 +73,65 @@ class FileScanSourceOpDesc extends ScanSourceOpDesc with
TextSourceOpDesc {
fileTypeName = Option("")
+ override def generateStandaloneCode(): String = {
+ val basename = sourceBasename(fileName.getOrElse(""))
+ val col = attributeName
+ val enc = encoding.toString.replace("_", "-").toLowerCase
+ val basenameLit = pyStringLiteral(basename)
+ val colLit = pyStringLiteral(col)
+ val encLit = pyStringLiteral(enc)
+ val buf = scala.collection.mutable.ArrayBuffer[String]()
+
+ if (extract)
+ buf += s"""# WARNING: extract=true is not supported in standalone mode;
provide the unarchived $basenameLit directly."""
+
+ val isBinary =
+ attributeType == FileAttributeType.BINARY || attributeType ==
FileAttributeType.LARGE_BINARY
+
+ if (attributeType.isSingle) {
+ val openArgs =
+ if (isBinary) s"""$basenameLit, "rb""""
+ else s"""$basenameLit, "r", encoding=$encLit"""
+ val dfCols =
+ if (outputFileName) s"""{"filename": $basenameLit, $colLit:
[_f.read()]}"""
+ else s"""{$colLit: [_f.read()]}"""
+ buf += s"""with open($openArgs) as _f:"""
+ buf += s""" out1df = pd.DataFrame($dfCols)"""
+ } else {
+ 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 => """l.rstrip().lower() == "true""""
+ case FileAttributeType.TIMESTAMP => "pd.Timestamp(l.rstrip())"
+ case _ => """l.rstrip("\n")"""
+ }
+ val hasSlice = fileScanOffset.isDefined || fileScanLimit.isDefined
+ if (hasSlice) {
+ val start = fileScanOffset.getOrElse(0)
+ val sliceExpr = fileScanLimit match {
+ case Some(l) => s"_lines[$start:${start + l}]"
+ case None => s"_lines[$start:]"
+ }
+ val dfCols =
+ if (outputFileName) s"""{"filename": $basenameLit, $colLit:
$sliceExpr}"""
+ else s"""{$colLit: $sliceExpr}"""
+ buf += s"""with open($basenameLit, "r", encoding=$encLit) as _f:"""
+ buf += s""" _lines = [$castExpr for l in _f]"""
Review Comment:
Conversion happens before the configured slice. For an INTEGER file
containing `invalid`, `7`, `invalid`, offset 1 and limit 1 work in
FileScanUtils, which slices before parsing. This export raises ValueError on
the first line. Slice the raw lines before conversion. TextInputSourceOpDesc
has the same ordering problem.
--
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]
