kz930 commented on code in PR #8341:
URL: https://github.com/apache/texera/pull/8341#discussion_r4066569890
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/arrow/ArrowSourceOpDesc.scala:
##########
@@ -38,10 +40,85 @@ 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 pandas reads an Arrow file with; the columns the file
states
+ // are asked of it directly below.
+ 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.
+ val read =
+ s"""out1df = pd.read_feather($SourceFilePlaceholder,
dtype_backend="numpy_nullable")"""
+ // A file pandas wrote from a frame keyed by one of its columns says so in
its
+ // schema, and pandas reads those columns back as the frame's index rather
+ // than as columns. The executor reads the columns the file states and has
no
+ // notion of an index, so a file written from a frame keyed by
`customer_id`
+ // kept that column on the one side and dropped it on the other, where an
+ // operator naming it raised. Put back under the name the file gives it,
which
+ // is `__index_level_0__` for an index that had none, and in the file's own
+ // order, which is where pandas wrote them: last. The Parquet source strips
+ // the same note, which its reader lets it do before the columns are read.
+ val index =
+ s"""|with pa.ipc.open_file($SourceFilePlaceholder) as _file:
+ | _names = _file.schema.names
+ |if list(out1df.columns) != _names:
+ | _index = out1df.index.to_frame(index=False)
+ | _index.columns = [_name for _name in _names if _name not in
out1df.columns]
Review Comment:
Reproduced, thanks. Fixed in ce2e4840e by refusing the pandas note instead
of undoing it: the table is read through pyarrow with `ignore_metadata=True`.
Your case is now a test.
--
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]
