[
https://issues.apache.org/jira/browse/SPARK-60606?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
DB Tsai updated SPARK-60606:
----------------------------
Description:
Local createDataFrame conversion builds a second complete coerced-row matrix,
copies every top-level field name into it, and clones each column again before
Arrow conversion. Coercing and encoding one column at a time removes these
intermediate allocations while retaining the existing type coercion and error
order.
Draft PR: https://github.com/apache/spark-connect-rust/pull/109
The PR has been narrowed to this input change after review. The homogeneous
Int64-only decoder is removed; a future decoder optimization should resolve
each column's type once and support mixed batches. The remaining optional PR106
clone-test cleanup is retained. Only top-level name copies in the intermediate
input matrix are removed: nested coerce_value struct names still allocate per
value. The existing width-mismatch panic and unreachable empty-input helper
branch remain follow-up cleanup, outside this measured algorithm change.
Independent input measurements compare 74b8e4125ee8c80f4aad51beabfd37277e39215a
with the input candidate. That baseline has exactly the same complete source
tree as master base c36c846fcb4f22c9995e5de5c69c45ae76a5ad9a. The narrowed PR's
production code matches the input candidate; tests were shortened, with direct
expected values replacing the frozen legacy copies. No combined optimization
remains and no new timing claim is made for the test cleanup.
Observed median changes (screen / independent confirmation; 15 fresh
processes/build/case/phase, rotating order):
- 10,000 x 32 tuple input: CPU -10.10% / -13.56%; wall -10.05% / -13.49%; held
RSS -28.03 / -28.95 MiB (183.91 -> 155.88 / 184.23 -> 155.28 MiB).
- 10,000 x 3 string input: CPU -9.36% / -8.24%; wall -9.29% / -8.24%; held RSS
-3.84 / -3.44 MiB.
Each phase attempted 27 cases x 15 fresh processes x 3 clients = 1,215 samples:
1,185 succeeded, 30 Rust Duration(Microsecond) warmup failures, zero missing.
Official Python succeeded on the Duration case. The existing Rust correctness
gap is now tracked independently by SPARK-60607, and is not caused by this
input change. No timed Rust CPU/wall/RSS result exists for that failed
operation.
The original frozen experiment's CPU and memory gates remain rejected in the
archived evidence; its policy required every raw Rust sample to pass and also
left 7 screening / 3 confirmation timing metrics inconclusive. Available
successful-operation RSS guards passed. Wide-input held/peak savings met the
target-only saving criteria in both phases, but this is not an overall accepted
memory or CPU improvement. The original policies, patches, identities, raw
data, failures and verdicts are preserved unchanged. The unmeasured combination
caveat no longer applies because the decoder optimization has been removed.
Before performance acceptance, run fresh screening and confirmation with all
CPU (maximum +5%), wall (maximum +10%) and RSS (maximum the larger of +5% or 4
MiB) regression confidence bounds passing under a policy frozen before those
runs. Keep pre-existing unsupported Duration failures explicit and separate
from PR causality; do not relabel the old failed runs as passing or drop failed
cases from them. Full live-server parity and green CI remain outstanding.
Fresh narrowed-source validation: 392 library plus 24 golden/sharing checks
(416 total); compile all integration targets with and without wasm-udf; cargo
fmt; core-library Clippy; identical 108-class Python API; 11 independent
untimed baseline/candidate Python fixture comparisons covering local nested
Rows and nested input/return values. The new input-to-decoder round trip
remains. The dated benchmark log and copied legacy implementations have been
removed from the source tree.
Harness and fixture source:
https://github.com/dbtsai/pyspark-connect-benchmark/tree/23bdf9046815ae615c921b2e6bab827eed00aaae
Original frozen experiment evidence and the preserved decoder patch remain
local. The draft PR contains the benchmark table and links here for details.
was:
Local input conversion builds a second complete row matrix, then clones each
column before Arrow conversion. Coerce one column at a time to remove those
intermediate copies while retaining the existing coercion rules. For
homogeneous Int64 result batches, downcast each column once and preserve the
generic decoder for all other batches.
Each optimization was measured independently against `74b8e41`; that baseline
has exactly the same source tree as current master `c36c846`. Each screen and
fresh confirmation used 15 fresh processes per build/case in rotating order.
These are observed median changes; the combined PR head has not been
benchmarked.
Selected independent confirmation observations: wide tuple input CPU −13.56%,
wall −13.49%, held RSS −28.95 MiB; string input CPU/wall −8.24%, held RSS −3.44
MiB; Int64 collect CPU −57.34%, wall −67.50%, held RSS +0.125 MiB; Int64
collect plus full value scan CPU −7.09%, wall −7.27%, held RSS +0.2344 MiB.
Overall CPU and memory acceptance rejected: the retained Duration(Microsecond)
case fails before timing in both Rust builds, and several timing regression
intervals remain uncertain. Wide-input target-only repeated memory support does
not establish overall acceptance. The combined head has no benchmark timing;
live Spark parity and CI remain outstanding.
Fresh combined correctness/API/Python validation is recorded in the draft PR.
Completion requires a fresh frozen combined-head screen and independent
confirmation with every raw Rust sample and CPU/wall/RSS guard passing, plus
live-server parity. Keep the existing Duration failure explicit; exact
timedelta conversion is separate correctness work.
Harness:
https://github.com/dbtsai/pyspark-connect-benchmark/tree/23bdf9046815ae615c921b2e6bab827eed00aaae
Generated with OpenAI Codex.
Draft PR: https://github.com/apache/spark-connect-rust/pull/109
The same draft PR also includes the remaining optional nested-struct clone-test
cleanup from the review of merged PR106, retaining only the two
field-name-sharing assertions. The core createDataFrame name-copy follow-up is
already covered by eliminating the coerced-row matrix.
https://github.com/apache/spark-connect-rust/pull/106#pullrequestreview-5675447129
Summary: Build local Arrow input one column at a time in Spark Connect
Rust (was: Reduce local input allocations and Int64 decoding overhead in Spark
Connect Rust)
> Build local Arrow input one column at a time in Spark Connect Rust
> ------------------------------------------------------------------
>
> Key: SPARK-60606
> URL: https://issues.apache.org/jira/browse/SPARK-60606
> Project: Spark
> Issue Type: Improvement
> Components: Connect
> Affects Versions: connect-rust-4.3.0
> Reporter: DB Tsai
> Priority: Major
> Labels: pull-request-available
> Fix For: connect-rust-4.3.0
>
>
> Local createDataFrame conversion builds a second complete coerced-row matrix,
> copies every top-level field name into it, and clones each column again
> before Arrow conversion. Coercing and encoding one column at a time removes
> these intermediate allocations while retaining the existing type coercion and
> error order.
> Draft PR: https://github.com/apache/spark-connect-rust/pull/109
> The PR has been narrowed to this input change after review. The homogeneous
> Int64-only decoder is removed; a future decoder optimization should resolve
> each column's type once and support mixed batches. The remaining optional
> PR106 clone-test cleanup is retained. Only top-level name copies in the
> intermediate input matrix are removed: nested coerce_value struct names still
> allocate per value. The existing width-mismatch panic and unreachable
> empty-input helper branch remain follow-up cleanup, outside this measured
> algorithm change.
> Independent input measurements compare
> 74b8e4125ee8c80f4aad51beabfd37277e39215a with the input candidate. That
> baseline has exactly the same complete source tree as master base
> c36c846fcb4f22c9995e5de5c69c45ae76a5ad9a. The narrowed PR's production code
> matches the input candidate; tests were shortened, with direct expected
> values replacing the frozen legacy copies. No combined optimization remains
> and no new timing claim is made for the test cleanup.
> Observed median changes (screen / independent confirmation; 15 fresh
> processes/build/case/phase, rotating order):
> - 10,000 x 32 tuple input: CPU -10.10% / -13.56%; wall -10.05% / -13.49%;
> held RSS -28.03 / -28.95 MiB (183.91 -> 155.88 / 184.23 -> 155.28 MiB).
> - 10,000 x 3 string input: CPU -9.36% / -8.24%; wall -9.29% / -8.24%; held
> RSS -3.84 / -3.44 MiB.
> Each phase attempted 27 cases x 15 fresh processes x 3 clients = 1,215
> samples: 1,185 succeeded, 30 Rust Duration(Microsecond) warmup failures, zero
> missing. Official Python succeeded on the Duration case. The existing Rust
> correctness gap is now tracked independently by SPARK-60607, and is not
> caused by this input change. No timed Rust CPU/wall/RSS result exists for
> that failed operation.
> The original frozen experiment's CPU and memory gates remain rejected in the
> archived evidence; its policy required every raw Rust sample to pass and also
> left 7 screening / 3 confirmation timing metrics inconclusive. Available
> successful-operation RSS guards passed. Wide-input held/peak savings met the
> target-only saving criteria in both phases, but this is not an overall
> accepted memory or CPU improvement. The original policies, patches,
> identities, raw data, failures and verdicts are preserved unchanged. The
> unmeasured combination caveat no longer applies because the decoder
> optimization has been removed.
> Before performance acceptance, run fresh screening and confirmation with all
> CPU (maximum +5%), wall (maximum +10%) and RSS (maximum the larger of +5% or
> 4 MiB) regression confidence bounds passing under a policy frozen before
> those runs. Keep pre-existing unsupported Duration failures explicit and
> separate from PR causality; do not relabel the old failed runs as passing or
> drop failed cases from them. Full live-server parity and green CI remain
> outstanding.
> Fresh narrowed-source validation: 392 library plus 24 golden/sharing checks
> (416 total); compile all integration targets with and without wasm-udf; cargo
> fmt; core-library Clippy; identical 108-class Python API; 11 independent
> untimed baseline/candidate Python fixture comparisons covering local nested
> Rows and nested input/return values. The new input-to-decoder round trip
> remains. The dated benchmark log and copied legacy implementations have been
> removed from the source tree.
> Harness and fixture source:
> https://github.com/dbtsai/pyspark-connect-benchmark/tree/23bdf9046815ae615c921b2e6bab827eed00aaae
> Original frozen experiment evidence and the preserved decoder patch remain
> local. The draft PR contains the benchmark table and links here for details.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
