carloea2 commented on code in PR #8341:
URL: https://github.com/apache/texera/pull/8341#discussion_r4066223100
##########
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:
A Feather file with integer column labels crashes here. I wrote
`pd.DataFrame({1: [7, 8], 2: [10, 20]}).to_feather(...)`. The physical names
are strings, but pandas restores integer labels, so this assigns two names to
the one-column RangeIndex frame and raises `ValueError: Length mismatch`.
Please preserve physical column names without restoring pandas metadata and add
this case to the tests.
--
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]
