Skip to content

[SPARK-60745][CONNECT] Keep Arrow-backed pandas string input columnar - #145

Draft
dbtsai wants to merge 23 commits into
apache:masterfrom
dbtsai:codex/SPARK-60745-arrow-string
Draft

dbtsai wants to merge 23 commits into
apache:masterfrom
dbtsai:codex/SPARK-60745-arrow-string

Conversation

@dbtsai

@dbtsai dbtsai commented Oct 30, 2026

Copy link
Copy Markdown
Member

What changes were proposed in this pull request?

Keep canonical Arrow-backed pandas StringDtype columns columnar for an explicit native schema containing only binary-collated StringType fields and at least 10,000 rows. Validate Arrow buffers, cast to standard UTF-8, and reuse the inline IPC bridge. Guards authenticate a finite set of directly skipped normalization/storage hooks; customized or unsupported inputs retain row conversion. This does not guarantee arbitrary pandas customization equivalence.

Depends on PR #144, including its prerequisite stack. This narrow delta compares baseline d31870bc4ceb15c3cadc4bc0a44fcb16f7b5e723 with candidate 28f2b65cfcda7460459e67dcbac7e9437b385c4b: one production helper and two test files. Measurements do not attribute prerequisite-stack changes to this patch.

Why are the changes needed?

Arrow-backed string columns currently become Python objects and rows, discarding their columnar representation. Eligible inputs can retain Arrow storage through explicit-schema IPC.

Does this PR introduce any user-facing change?

Eligible inputs use columnar conversion. Tests preserve values, schemas and checked error types/messages/order against forced row conversion, covering customized direct hooks, manager caches, malformed buffers and public error/RPC ordering.

How was this patch tested?

The selected offline suites passed: pandas 2.3.3, 865 passed/9 skipped; pandas 3.1.0, 863 passed/11 skipped. All 184 added cases pass in each; all 690 baseline outcomes/skip reasons remain.

Original formatting, native client/core units (394/95), native/host-WASM test compilation and wheel builds are explicitly reused under identical production/build source bytes. Baseline Python results are reused; full candidate Python suites ran at the committed test head. Four-environment before/after captures verify source, wheels, installed packages, native and recorded dependency identities. Original/recovery-v2 failures remain FAILED; no fresh Clippy/live Spark result.

Runs use Python 3.12.14, NumPy 2.5.4, PyArrow 26.0.0, grpcio 1.84.1 and pandas 2.3.3/3.1.0 with frozen harness 1bbdec9f3665fcd2ec2ee5129bbcdb95ee4b00c4; Rust packages report 4.3.0, while separate official reference ca1514a2b4e72f9529871f34d09881cd4921df82 reports 5.0.0.dev0 and is measured only with pandas 2.3.3.

Three completed, independently reviewed and root-accepted groups contain 1,175 worker samples: explicit640, object160 and ordinary375. Cohorts remain separate; five samples per role provide descriptive distributions, not significance, confidence intervals or general neutrality.

Explicit640. All 16 eligible cells (two pandas versions × two profiles × two nominal missing slots × exact/Unicode-null-chunked variants) improve median elapsed 88.2–98.9% and CPU 89.3–98.9%. Candidate/baseline ratios are 0.010873–0.117645 and 0.010874–0.106766 respectively. Exact nominal pd_na medians, baseline → candidate:

pandas/profile elapsed ms CPU ms
2.3.3/warm 21.526 → 0.257 21.552 → 0.257
2.3.3/first 25.418 → 2.528 25.105 → 2.058
3.1.0/warm 23.213 → 0.252 23.234 → 0.253
3.1.0/first 26.510 → 2.153 26.172 → 2.148

Fallback/error costs are mixed: 26/64 elapsed and 26/64 CPU medians rise. Warm 3.1.0 hook-error/pd_na elapsed 0.0442 → 0.1036 ms (2.345×, +0.0594), CPU 0.0445 → 0.0772 (1.736×, +0.0327); warm 2.3.3 hook-error elapsed reaches 1.596×. First 3.1.0 hook-error/nan elapsed 1.0639 → 1.4113 ms (+347.38 µs, 1.327×), CPU 0.781 → 0.993. Warm 3.1.0 Python-storage/nan elapsed 20.865 → 23.965 (1.149×), CPU 20.889 → 23.431 (1.122×); first 2.3.3 Python-storage/pd_na elapsed 26.093 → 29.031 (+2.938 ms/+11.26%). First 2.3.3 threshold/nan elapsed 25.594 → 28.164 (1.100×), CPU 25.120 → 27.551 (1.097×). Generic-Arrow worst elapsed/CPU ratios are 1.033×/1.050× in different cells.

Eligible held-RSS medians fall 3,096,576–11,403,264 bytes. Warm 3.1.0 Python-storage/nan nevertheless has absolute held/post-validation increases +5,095,424/+4,571,136 bytes and signed held-minus-baseline +1,638,400 (4,784,128 → 6,422,528). First 2.3.3 generic-Arrow/nan has worst signed post-validation increase +2,555,904 (13,828,096 → 16,384,000), alongside absolute post-validation +2,883,584 and held +278,528. Eligible warm 2.3.3 chunked/pd_na absolute post-validation falls 5,505,024 while its signed increment rises 655,360.

Object160. Pandas 2.3.3 only, eight 10,000×3 fixtures/two profiles. Candidate/baseline median and distribution-maximum timing ratios follow; candidate-minus-baseline absolute held-RSS and signed held-minus-baseline median differences in bytes. Lifetime-peak medians equal held medians.

profile/fixture elapsed × CPU × elapsed max × CPU max × held Δ B signed Δ B
warm/exact 0.932 0.932 0.982 0.979 +65,536 -16,384
warm/all_null 0.983 0.973 0.971 0.970 +98,304 0
warm/early_subclass 1.053 1.049 1.071 1.054 -311,296 +49,152
warm/middle_subclass 0.991 0.991 0.884 0.944 -409,600 -98,304
warm/late_subclass 0.986 0.986 1.013 1.012 +262,144 +376,832
warm/early_surrogate 0.986 0.984 1.003 1.008 +147,456 +16,384
warm/middle_surrogate 1.061 1.061 1.085 1.085 +180,224 +16,384
warm/late_surrogate 1.005 1.006 1.004 1.004 -16,384 0
first/exact 1.010 1.023 0.943 1.078 0 +32,768
first/all_null 1.010 1.039 0.897 0.881 +212,992 -32,768
first/early_subclass 1.043 1.025 1.107 1.074 -294,912 -49,152
first/middle_subclass 1.085 1.045 1.083 1.046 -147,456 +32,768
first/late_subclass 0.998 1.001 0.989 1.013 -114,688 +81,920
first/early_surrogate 0.931 0.952 0.907 0.835 -98,304 +32,768
first/middle_surrogate 1.041 1.014 1.110 1.065 -32,768 -16,384
first/late_surrogate 1.002 1.004 0.943 0.940 +49,152 +65,536

Nine elapsed/ten CPU medians rise. Warm early-subclass elapsed/CPU +5.30%/+4.86%; middle-surrogate +6.06%/+6.11%. First middle-subclass elapsed 24.847 → 26.954 ms (+8.48%), CPU 24.844 → 25.970 (+4.53%). Worst baseline-RSS median increase: first all-null +524,288; held/lifetime-peak: warm late-subclass +262,144; signed increment: the latter +376,832 (5,603,328 → 5,980,160).

Distribution maxima also rise: first early-subclass elapsed +2.970 ms/CPU +2.007 ms; largest maximum ratios first middle-surrogate elapsed 1.110×/warm middle-surrogate CPU 1.085×. Warm early-subclass absolute held/lifetime-peak maxima rise +5,046,272 bytes, signed maxima +5,931,008. These compare cohort maxima, not matched samples or sampled construction peaks.

Ordinary375. Twenty-seven cohorts: 21 pandas-2.3.3 cohorts include official Python plus both Rust roles; six selected pandas-3.1.0 cohorts compare Rust roles only. Versions have different selected coverage; official Python is a reference, not this patch's baseline. All candidate/baseline median and distribution-maximum timing ratios:

pandas/profile/case elapsed × CPU × elapsed max × CPU max ×
233/first/explicit-10k×32 0.795 0.990 0.740 0.567
233/first/type-date-10k×3 0.998 1.011 1.022 1.020
233/first/type-string-10k×3 0.923 1.020 1.084 0.926
233/warm/arrow-explicit-10k×3 0.973 1.007 0.632 0.751
233/warm/ddl-10k×3 1.013 1.014 0.986 1.007
233/warm/explicit-10k×3 1.016 1.016 1.046 1.045
233/warm/explicit-10k×32 0.999 1.002 0.117 0.605
233/warm/inferred-10k×3 0.987 0.987 1.005 1.006
233/warm/scale-1M×3 1.017 1.003 0.500 0.844
233/warm/scale-100k×3 0.952 0.948 0.307 1.241
233/warm/scale-1000x3 0.834 1.037 0.179 0.710
233/warm/scale-100x3 1.025 1.026 1.136 1.138
233/warm/tuples-explicit-10k×3 0.996 0.997 0.980 0.980
233/warm/type-array-10k×3 0.997 0.997 0.974 0.974
233/warm/type-date-10k×3 1.002 1.002 1.026 1.023
233/warm/type-decimal-10k×3 1.020 1.020 1.104 1.060
233/warm/type-string-10k×3 0.949 0.949 0.961 0.966
233/warm/type-timestamp-10k×3 0.983 0.983 0.939 0.953
233/warm/return-100k×8-collect 0.999 1.015 0.853 0.979
233/warm/return-100k×8-toArrow 0.967 0.873 1.426 0.942
233/warm/udf-construct-capture-0 0.964 0.968 1.082 1.075
310/first/explicit-10k×32 0.758 0.996 0.693 0.929
310/first/type-date-10k×3 1.003 1.008 0.999 1.007
310/first/type-string-10k×3 0.061 0.058 0.075 0.058
310/warm/explicit-10k×32 0.985 1.016 0.963 1.301
310/warm/type-date-10k×3 0.997 0.996 0.986 0.986
310/warm/type-string-10k×3 0.021 0.021 0.022 0.023

Seven elapsed/14 CPU medians rise. Largest elapsed ratio: 2.3.3/warm/100×3 +2.5325%; CPU: 1000×3 +3.7037%. Decimal has largest absolute median increases: elapsed 24.831 → 25.325 ms (+1.99%), CPU +0.493 ms. Pandas-3.1.0 strings improve elapsed 22.904 → 0.481 ms warm and 22.198 → 1.345 first. Largest elapsed maximum increase: warm 2.3.3 toArrow 27.990 → 39.903 ms (+42.56%); CPU maximum ratio: warm 3.1.0 wide 1.301×; largest absolute CPU-maximum increase: decimal +1.512 ms.

Worst higher absolute RSS medians: baseline +442,368 bytes (2.3.3 first wide); held +507,904/lifetime-peak +606,208 bytes (3.1.0 warm wide); post-GC +475,136 (2.3.3 warm Arrow); sampled peak +884,736 (2.3.3 warm 1M). Worst signed median increases: held +229,376 (warm toArrow), post-GC +1,015,808 (3.1.0 warm strings, despite lower absolute post-GC), sampled peak +950,272 (warm 1M). Decimal has largest absolute held/sampled/lifetime-peak maxima increases +2,408,448, post-GC +2,473,984; signed maxima +2,179,072. Maxima compare distributions, not matched endpoints. Every ordinary cohort has some higher descriptive observation.

Scope and limits. Explicit/object warm uses 20 constructions/three warmups; first uses one/no constructor warmup after interpreter/input/session/server preparation. Ordinary warm uses one operation warmup; first uses zero after untimed spark.version; measured loops have one operation except warm 2.3.3 UDF (five). Automatic GC is not explicitly disabled. No cold-interpreter or per-sample helper-dispatch claim. Explicit/object loop/error-checking overhead is included; preparation/validation IPC decoding is outside construction timing. Ordinary setup/warmup/counters/RSS/validation are outside operation timing; process CPU brackets the wall timer. Explicit/object order alternates deterministically; ordinary retains rotated order, without randomization. Generic Arrow uses actual pd.NA in both nominal slots.

Independent explicit review decodes all 480 retained last-successful DataFrame payloads/14,399,760 cells and verifies 1,680 measured expected-error checks; preceding successful iterations are not individually decoded. Object independently decodes 100 retained UTF-8 IPC/protobuf payloads/3,000,000 cells, only last successful results. Its 630 surrogate attempts/hook totals remain sealed-worker assertions: printed totals/one error fingerprint, no retained per-iteration errors or independent replay. Object has aggregate stdout timing/RSS, without pre-validator construction-measurement files/nanosecond endpoints. Ordinary rehashes/cross-links all 330 typed-validator receipts/requests/payloads covering 72,149,500 cells plus 45 noninput guards; typed value/null-mask/nested-null/schema validation remains a frozen producer assertion, not fresh independent decoding.

Construction RPC zeros apply to explicit/object sole-loopback fixtures; frozen20 zeros are inferred separately. Ordinary directly observes all 20 counter keys in all 375 sample maps, including nonzero input Config traffic (four calls/512 bytes); all 135 Rust-pair maps match. Return/config traffic exists; there is no overall-zero or full-wire claim.

RSS baseline/held precedes validation; lifetime ru_maxrss includes preparation/warmups. Explicit post-validation RSS follows output/mock-holder release/one GC; inputs/expected/schema/session/server and native arenas/internal frames may remain. Ordinary post-GC includes additive validation allocations. Neither proves complete reclamation/construction-only retention. Ordinary external RSS targets 2 ms but records 1–59 observations/largest gap 5.520125 ms; observation starts before go and stops on measured message, not exact timer endpoints; short transients can be missed. Object has no post-GC/sampled-peak/count/gap fields. Signed deltas can be nonpositive; ratios require positive denominators; medians of signed deltas differ from differences of absolute medians.

Ordinary harness truncates stderr at 8,000 and exception stdout at 2,000 characters. Object streams/receipts exist before launch; interruption may leave RUNNING receipts/only written bytes. Selected dependency RECORD/PySpark-origin inventories do not establish shadow-import/loaded-dependency/continuous isolation. Inherited-group supervision does not cover arbitrary detach/setuid/host crash/realtime/external-supervisor-SIGKILL guarantees.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Codex (GPT-6)

This pull request and its description were written by Isaac.

dbtsai and others added 23 commits October 28, 2026 17:17
Provide an alternative to the per-batch C Data handoff in PR apache#103 while retaining PyO3 0.28.3. Add ownership, buffer-sharing, compatibility, and public client handoff tests.

Generated-by: OpenAI Codex
Apply LC's suggestion to export one boxed RecordBatchReader through
arrow-pyarrow and materialize it with read_all. Remove the custom stream
callbacks, unused converter crate, and dedicated converter workflow.

Keep schema consistency checks, standalone schema preflight, and the NUL
schema IPC fallback. Retain handoff tests in the coverage workflow and
keep PyO3 at 0.28.3.

Generated-by: OpenAI Codex (application version unavailable).
Keep public behavior coverage and compile mock protocol descriptors from source.
Add native toArrow smoke cases, remove the unproven schema preflight, and avoid
the arrow module name collision. Leave CI workflows unchanged.

Co-authored-by: Isaac <no-reply@databricks.com>
Document the Arrow 58.3 NUL export panic at its upstream source and wrap the
architecture paragraph. Keep CI workflows unchanged.

Co-authored-by: Isaac <no-reply@databricks.com>
…am interface

Normalize Arrow schemas and timestamps with the existing PySpark helpers, then serialize imported Rust batches directly for local relations without Python row materialization.

Co-authored-by: Isaac <no-reply@databricks.com>
Honor server local-relation settings, split Arrow IPC into independently readable chunks, and upload session-scoped cache artifacts with SHA-256 deduplication and CRC validation. Keep large input payloads out of query plans and test uploads above the gRPC message limit.

Co-authored-by: Isaac <no-reply@databricks.com>
…mantics

Use explicit LongType schema names when constructing Arrow columns. Limit the
columnar path to concrete pandas/NumPy arrays and native schema descriptors,
including authentic converter aliases retained by already imported modules.
Keep nullable values in non-nullable fields on the native row path so local
validation still precedes configuration RPCs. Support warm Arrow importers that
resolve converters locally.

Add mock-RPC regressions for all emitted integer cells, schema metadata and
nullability, custom conversion hooks, and repeated warm construction.

Co-authored-by: Isaac <no-reply@databricks.com>
Build the subclass-label fallback fixture with an explicit object Index and
assert its concrete label type. Pandas 3 otherwise normalizes constructor
labels to builtin strings before the unchanged integer helper sees them.

The original regression passes on pandas 2.3.3 and fails on 3.1.0; the fixed
regression and full CI-selected Arrow/drop-in suites pass on both versions.
Runtime Python and native sources remain unchanged.

Co-authored-by: Isaac <no-reply@databricks.com>
Merge current Apache master d9defae while retaining all pandas integer input guards and the pandas 3 fixture repair. Keep upstream Arrow output regressions, CI protoc requirement and live smoke step; preserve fork input bridge/docs and stable-buffer assertions.

Co-authored-by: Isaac <no-reply@databricks.com>
Decline noncanonical columns indexes and their direct string-storage hooks before the integer fast path inspects labels or normalizes cells. Keep native tolist errors and mutations in their original order and cover public construction on both pandas versions.

Co-authored-by: Isaac <no-reply@databricks.com>
Decline class-level columns storage descriptors before reading labels. Keep the single native columns storage read before pandas normalization, and cover public construction with inferred, named and explicit schemas on both pandas versions.

Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Reuse inline IPC for explicit native string schemas after authenticating ordinary pandas Arrow string storage and normalization hooks. Retain the row path for customized and unsupported inputs.

Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant