HyukjinKwon commented on PR #54:
URL:
https://github.com/apache/spark-connect-rust/pull/54#issuecomment-5418653077
<!-- ai-code-review -->
## Code review — manual pass
The `/spark-dev:review` harness set up and checked out the PR head fine, but
its `spark-dev page` step can't read its own scratch files under `/tmp` on
macOS (`/tmp` → `/private/tmp` resolves outside the allowlisted roots), so the
automated pipeline can't run here. I reviewed the diff manually against
`apache/master`, focused on correctness. Findings below, most severe first.
Everything not listed here (the 16 variadic `functions.rs` fixes, the `plan.rs`
Sort fallback + `NAFillColumns`, the streaming response-draining fix, the
unified Arrow read decoder incl. all timestamp units / Decimal128&256 / UInt64
guard, the catalog decoder delegation, `Row(**kwargs)`,
`withColumns`/`unpivot`/`rollup`/`cube`/`fillna(dict)` wiring) I read and
believe is correct.
---
### 1. HIGH — `createDataFrame` with a TIMESTAMP column always fails
(regression)
`crates/spark-connect/src/session.rs`
`schema_to_arrow_fields` now declares a `DataType::Timestamp` field as
**`Timestamp(Microsecond, Some("UTC"))`** (~L627), but `build_arrow_array`
builds the column with `TimestampMicrosecondArray::from(values?)` (L837), whose
Arrow type is **`Timestamp(Microsecond, None)`** — the `.from()` constructor
attaches no timezone and has no access to the declared schema.
`RecordBatch::try_new` (L584) validates each column's datatype against the
schema and rejects the mismatch.
**Failure scenario:** `spark.createDataFrame([(datetime(2020, 1, 1),)], "a
timestamp")` — or *any* inferred-schema call containing a naive `datetime`
(Spark infers `TIMESTAMP`/LTZ → `DataType::Timestamp`, reachable straight from
the Python skin's inference path) — fails with:
`Failed to create Arrow batch: column types must match schema types,
expected Timestamp(Microsecond, Some("UTC")) but found Timestamp(Microsecond,
None) for field a`.
This is a **regression**: before this PR both sides were `None` and it
worked. `TIMESTAMP_NTZ` still works (both `None`). No test exercises this path
(the only `.schema("... timestamp ...")` use is a `readStream` that never
materializes), which is why the green parity gate misses it.
**Fix:** thread the target field type into `build_arrow_array` so the
timestamp array's zone matches the schema (`.with_timezone("UTC")` for
`Timestamp`, none for `TimestampNtz`). Minimal stopgap that restores prior
behavior: revert L627 to `Timestamp(Microsecond, None)`.
### 2. MEDIUM — `df.stat.approxQuantile(...)` returns a DataFrame, not a list
`crates/pyspark-rs/src/stat.rs` (core `group.rs::approx_quantile` returns a
`DataFrame`, surfaced as-is)
Reference `DataFrameStatFunctions.approxQuantile(col, probabilities,
relativeError)` returns a **`list[float]`** (or `list[list[float]]` when `col`
is a list). Here it returns a `PyDataFrame`.
**Failure scenario:** `q = df.stat.approxQuantile("age", [0.5], 0.25);
print(q[0])` — reference gives the median as a float; here `q` is a DataFrame
and this breaks. (`corr`/`cov` correctly return floats; `crosstab`/`freqItems`
correctly return DataFrames — only `approxQuantile` is wrong.)
### 3. MEDIUM — `df.persist(storageLevel)` silently ignores the requested
level
`crates/pyspark-rs/src/dataframe.rs` — `persist` accepts `storage_level` but
discards it (`let _ = storage_level;`) and always uses `MEMORY_AND_DISK`.
**Failure scenario:** `df.persist(StorageLevel.DISK_ONLY)` caches in memory
anyway — silently does something other than what was asked, rather than
erroring.
### 4. MEDIUM — `df.repartition("col")` (column-only form) raises TypeError
`crates/pyspark-rs/src/dataframe.rs` — `repartition(num_partitions: i32,
*cols)` forces the first arg to be an int. Reference
`repartition(numPartitions, *cols)` allows the first arg to be a **Column or
str**, used as the first partitioning column with the default partition count.
**Failure scenario:** `df.repartition("country")` and
`df.repartition(col("country"))` are valid in reference but raise a TypeError
here (`"country"` can't parse as `i32`).
### 5. LOW/MEDIUM — boolean options serialize as `"True"`/`"False"` instead
of `"true"`/`"false"`
`crates/pyspark-rs/src/readwriter.rs` (`option`/`options`) and
`crates/pyspark-rs/src/conf.rs` (`set`) coerce values via `value.str()`, which
yields Python's capitalized `str(True) == "True"`. Reference pyspark routes
option values through `to_str`, which **lowercases booleans**.
**Failure scenario:** `spark.read.option("header", True).csv(path)` sends
`"True"` instead of `"true"` — differs from reference and risks breaking
case-sensitive option parsing on the server.
### 6. LOW — `df.hint(name, *params)` rejects non-string parameters
`crates/pyspark-rs/src/dataframe.rs` — `parameters: Vec<String>`. Reference
hint parameters can be ints/lists.
**Failure scenario:** `df.hint("rebalance", 3)` raises a TypeError (int not
a str).
### 7. LOW — `createDataFrame` numeric coercion is incomplete (not a
regression)
`crates/spark-connect/src/session.rs::coerce_value` handles `(Float,
Double)` but not `(Float, Long)` / `(Float, Integer)`.
`spark.createDataFrame([(1,)], "a float")` builds an `Int64Array` against a
`Float32` schema field → `RecordBatch::try_new` mismatch. Reference accepts it.
Pre-existing gap (there was no coercion before), so a follow-up rather than a
blocker.
---
Only **#1** is blocking, in my view — it's a functional regression on a
common path. #2–#6 are drop-in fidelity gaps worth fixing given the PR's stated
goal.
This review was written by Isaac.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
