viirya commented on PR #50: URL: https://github.com/apache/spark-connect-rust/pull/50#issuecomment-5398143112
Third pass. The crates.io blocker is resolved and I've re-verified it. Remaining items below: one still-open gap from earlier, one new correctness finding, and some structural feedback on the WASM UDF work. ## Confirmed fixed since last pass - **Crate naming.** Everything is renamed to `apache-spark-connect`, `-core`, `-proto`, `-macros`. I checked all four against the crates.io API and they're unclaimed, so `cargo publish` has a clear path now. This closes what I'd flagged as the hard blocker. ## Still open: no test covers any of the five fixed plan-builder bugs I raised this last time and it's the one item I'd hold the PR on, because it's the mechanism that let these bugs ship rather than any individual mistake. The fixes are correct, but nothing asserts the previously-dropped argument reaches the proto: - `test_dropna_golden` (`dataframe_extra_golden.rs:28`) exercises `df.range(5).dropna()` — no `how`, so it passed identically before and after the fix. - The golden `hint` case (`plans_golden.rs:592`) is `plan::hint(range, "broadcast", vec![])` — empty parameters, so likewise. - I found nothing asserting a string `replace`, an explicit-value `pivot`, or `fillna_double`. Reintroduce any of the five and CI stays green. Five golden cases — `dropna(how="all")` asserting `min_non_nulls == 1`, a hint with a parameter, a string replace, a pivot with values, a double fill — would close it. More importantly the shape recurs wherever a method takes a mode/how/parameters argument, and `plan.rs` is now 1,705 lines with `functions.rs` at 4,173, so a convention here is worth more than the five cases. ## New: the retry budget resets on every mid-stream drop `ReattachableResponseStream::message()` constructs `RetryPolicyState::new(...)` *inside* the error arm, so each disconnect gets a fresh budget of `max_retries` (default 15) and the backoff restarts at `initial_backoff_ms` instead of escalating. A repeatedly-flapping stream — a proxy cycling connections, a server shedding load — can therefore reattach without a global bound: 15 attempts, one success, drop, 15 more, indefinitely. The intended ceiling is never reached and a wedged query has no natural end. PySpark scopes one `Retrying` instance to the whole operation for this reason. Hoisting the `RetryPolicyState` to a struct field (built with the stream, reset only on a successful `ResultComplete`) would restore the bound. Related: `DEFAULT_MAX_RETRY_EXCEPTION_ELAPSED_TIME` is defined and exported but never read, so the one-hour elapsed-time ceiling isn't enforced anywhere — either wire it into `next_attempt` or drop it, since right now it reads as a guarantee that doesn't exist. ## `cargo test --workspace` fails without the wasm32 target `wasm-udfs` and `wasm-udf-inline` are unconditional workspace members, and their build scripts hard-`panic!` when the target is absent (`crates/spark-connect-build/src/lib.rs:101`). On a clean checkout: - `cargo test` (default-members) — passes, 100+ tests green. - `cargo test --workspace` — **fails to build**: `error[E0463]: can't find crate for 'std'`. `rust.yml` installs wasm32, so CI is green and this only bites contributors locally — the worst place for it, since the failure looks like a broken checkout rather than an optional feature. Since the feature is explicitly opt-in (`wasm-udf = ["dep:base64"]`, off by default, needs `wasmtime` on executors), the WASM crates shouldn't be unconditional members; failing that, the build script should degrade to a clear skip with a message naming `rustup target add wasm32-unknown-unknown`. Also `lint.yml:49` is `cargo clippy --workspace --all-targets || true`, so clippy can never fail the build. If that's deliberate for now, fine, but it means the lint gate is advisory only. ## On the WASM UDF design To be clear about what I did and didn't check: I read the module docs, the ABI, the packer invocation, and the runner, but I have not run it against a cluster. The design is genuinely clever — Spark Connect has no "run this Rust code" UDF type, so routing a wasm module through the one path that does exist (`PythonUDF`) fills a real gap for Rust-native users, and doing it with zero server-side changes is the right instinct. Two specifics I want to credit because they're easy to get wrong: cloudpickling the runner **by value** so executors need only `wasmtime` and not the `pyspark_wasm_udf` package, and an ABI that covers strings, binary, arrays, and nullability rather than stopping at numbers. I also checked the `SQL_ARROW_BATCHED_UDF` choice against the scalar `__call__`, assumed it was a mismatch, and was wrong — the comment correctly notes that eval type still invokes per row and only the I/O is Arrow. Structural feedback rather than defects: **I'd move this to its own PR on top of the stack.** It's a third thing alongside "Rust core client" (#50) and "drop-in wrapper" (#51): a new execution mechanism spanning a proc-macro crate, a build crate, two example crates, and a Python package, with its own runtime prerequisite. On its own PR the ABI, the by-value pickling, and wasmtime version compatibility each get looked at properly; here they're arriving inside a restructure and will likely get waved through. It would also be independently revertible, and it's the reason for the workspace breakage above. **The tests only cover ABI descriptors.** `wasm_udf.rs`'s test module asserts `AbiType` descriptors, `to_data_type`, and `to_spark_json` — all pure mapping. Nothing exercises the packer, and there's no round-trip (encode args → wasm → decode result) even against a locally compiled module. `rust.yml`'s `wasm-udf` job installs `cloudpickle`/`pyspark`/`wasmtime` and runs `cargo test -p apache-spark-connect --features wasm-udf` plus a build, so it proves things compile — it doesn't prove a UDF computes the right answer. The encode/decode path is hand-rolled pointer arithmetic over linear memory (`packed >> 32`, `& 0xFFFFFFFF`); that's exactly the code that deserves a test per `AbiType`, including the null and empty-array edges. **One documentation point worth being explicit about:** building a Rust UDF now requires spawning a Python interpreter on the client (`python -m pyspark_wasm_udf.pack`, needing `cloudpickle` + `pyspark`). That's a defensible trade to reuse cloudpickle, and the error handling around the subprocess is careful. But it does mean the "pure-Rust client" acquires a Python build-time dependency for this feature, which users will be surprised by unless the README says so plainly. Worth stating too that the goal is capability, not speed — the path is Python worker → wasmtime → linear memory → per-row entrypoint, so nobody should expect it to beat a vectorized Arrow UDF. Unchanged from earlier and not repeated here: the `spark-connect-rs` deprecation path, Issues-off in `.asf.yaml`, and the 4.2.0-versus-"alpha" mismatch. -- 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]
