Skip to content

[SPARK-60727][CONNECT] Build explicit double pandas input from columns - #143

Draft
dbtsai wants to merge 20 commits into
apache:masterfrom
dbtsai:codex/spark-60727-narrow-publication-v8
Draft

dbtsai wants to merge 20 commits into
apache:masterfrom
dbtsai:codex/spark-60727-narrow-publication-v8

Conversation

@dbtsai

@dbtsai dbtsai commented Oct 30, 2026

Copy link
Copy Markdown
Member

What changes were proposed in this pull request?

Build eligible pandas float64 input from its NumPy columns, then encode Arrow IPC directly into the existing inline LocalRelation. This avoids astype(object).where(...).values.tolist() and native row materialization for frames with at least 100,000 rows and an explicit native StructType containing only DoubleType fields. Smaller frames and unsupported or customized inputs retain the existing row path.

This change depends on PR #121, including its columns compatibility follow-up c4284a3e0b71c8d624a1253c52337419761b2131. The proposed change from that base includes the 17-line inline IPC bridge, Float dispatch and helpers, and Float regressions. The bridge is part of this fix and its measured cost; it is absent from the comparison baseline.

The incremental Float delta from that base touches four files (+1,087/-2). While PR #121 and its Arrow prerequisites remain unmerged, GitHub's comparison to master also includes their changes; this draft should be reviewed after those dependencies.

Why are the changes needed?

Large pandas double frames currently become Python object rows and native row values before returning to Arrow. The column path removes those intermediate rows while retaining pandas missing-value semantics, float bits, field names/nullability, logical Spark metadata and inline transport.

Does this PR introduce any user-facing change?

Eligible explicit double input uses the column path. It preserves the existing logical result and row-path fallback behavior, including the legacy pandas infinity-as-missing option where available. No new Python API or configuration is introduced. Direct conversion hooks, custom columns storage and unsupported schema/storage cases retain their existing behavior and error order.

How was this patch tested?

Fresh baseline and candidate wheels were built from the exact PR121 base and candidate f9779461742b83e2e46b97381c054809cacf5965, with separate frozen pandas 2.3.3 and 3.1.0 environments.

  • Both roles pass 394 client and 95 core unit tests, formatting, native integration-test compilation and host compilation with the wasm-udf feature. This is not a WASM cross-target build or execution result.
  • The complete same-source test_arrow_handoff.py and test_dropin_offline.py suites collect 531 baseline cases and 690 candidate cases per pandas version. Across all four environments: 2,404 passed, 38 known skips, zero failures/errors. The candidate's 159 Float cases pass on pandas 2.3.3; 157 pass and two skip on pandas 3.1.0 because it removed mode.use_inf_as_na. Other skips require SPARK_REMOTE or an active-session condition.
  • Received-IPC assertions cover every value/null bit, signed zero, infinities/extremes, all-null columns, renamed fields, nullability, logical metadata and fallback/error order, including custom storage, converter/schema aliases and columns hooks. Source, final wheel, installed package, selected dependency bytes, import origins and protocol descriptors match before/after testing.

The exact current whole-fix comparison contains 1,278 samples: 1,038 received-IPC receipts, 120 predefined construction errors and 120 guard samples, checking 615,959,280 Float cells. Each timing row uses five fresh-process samples per client; three additional cases use one sample per client for correctness only. The six disjoint slices each authenticate the exact fresh source/package/dependency/protocol bindings before and after timing, then confirm the owned process group is gone before releasing the shared CPU lock. No historical timings are transferred.

Representative median construction times in milliseconds, baseline→candidate. The pandas cohorts remain separate:

Float input pandas 2.3.3 wall pandas 2.3.3 CPU pandas 3.1.0 wall pandas 3.1.0 CPU
100,000×3 cold 168.962→12.610 168.885→7.844 168.030→9.262 167.870→6.201
100,000×3 warm 166.593→1.567 165.278→1.566 159.125→1.632 159.130→1.637
1,000,000×3 cold 1649.549→22.597 1647.284→20.938 1699.292→25.925 1654.536→23.071
1,000,000×3 warm 1636.275→8.922 1633.152→8.912 1610.896→9.296 1609.273→9.258

Guard, fallback and error costs are retained. These are the largest observed median increases in each category; wall and CPU maxima can occur in different cases:

Category pandas 2.3.3 wall / CPU increase pandas 3.1.0 wall / CPU increase
Other input guards +0.10% / +0.10% +5.70% / +5.70%
Below 100,000 rows +4.22% / +4.14% +3.69% / +3.67%
Late unsupported fallback +4.41% / +3.99% +2.15% / +1.88%
Nonnullable NaN error +6.32% / +4.25% +7.26% / +7.76%

The pandas 3.1.0 string guard rises from 21.986→23.240 ms wall and 21.991→23.245 ms CPU. The worst below-threshold wall cases are 99,999×1 cold on pandas 2.3.3 and 99,999×3 warm on pandas 3.1.0. Cold 100,000×1 nonnullable-NaN errors rise from 66.711→70.929 ms and 67.125→72.000 ms; the pandas 3.1.0 warm error also rises 62.849→67.205 ms (+6.93%). These observations do not establish guard or error-path neutrality.

Memory is observational. For warm 1,000,000×3 input, absolute RSS falls while incremental held RSS rises. Values below are candidate-minus-baseline decimal MB; incremental values subtract each sample’s own process baseline before taking medians:

Rust pandas Absolute held Incremental held Absolute post-GC Incremental post-GC Sampled peak
2.3.3 -250.27 +20.71 -339.77 -68.24 -649.28
3.1.0 -252.67 +20.33 -341.54 -67.99 -647.23

Process-peak deltas equal the sampled-peak deltas in these two rows. Adverse RSS observations elsewhere remain: pandas 2.3.3 99,999×8 warm has +8.49 MB absolute held and +8.31 MB sampled peak (+8.49 MB process peak); late unsupported fallback has up to +8.90 MB absolute post-GC and +5.00 MB incremental post-GC. Pandas 3.1.0 has up to +1.87 MB absolute held/post-GC in the string guard and +3.13 MB sampled/process peak in a cold NaN error. Maximum observed external RSS sampling gaps are 6.79 ms and 5.43 ms for the respective cohorts; short peaks can be missed.

Rust construction RPC deltas match baseline and candidate in every paired raw sample. All 386 Rust baseline Float samples and 386 Rust candidate Float samples have zero timed RPCs. All 386 official Float reference samples retain one Config RPC and 771 request bytes per sample; those observed costs are included in the reference measurements. The existing Rust integer guard retains four Config RPCs and 512 request bytes; all other measured Rust construction counters are zero. Received-IPC validation happens after this timed interval. Exhaustive wall ranges, CPU, absolute/incremental RSS, sampling gaps and RPC counters are preserved with the raw receipt hashes in the campaign’s full cost report.

The frozen benchmark's legacy minimal key identifies the V8 pandas 2.3.3 pair and full identifies the V8 pandas 3.1.0 pair. Its historical minimal_dependency_base field is stale metadata; the authoritative per-run bindings identify the exact PR121 baseline and Float candidate. The official reference uses pinned pandas 2.3.3 in each cohort. Cohorts remain separate, with no official pandas 3.1 result or pooled speedup. The Rust wheels identify as 4.3.0; the frozen official pyspark-client reference is 5.0.0.dev0 built from apache/spark revision ca1514a2b4e72f9529871f34d09881cd4921df82. These different versions limit the comparison to the pinned products and workloads; release-matched parity is not established.

Benchmark cold cases time the first createDataFrame after NumPy/pandas/PyArrow imports and spark.version; they do not measure import or process startup. Warm cases use one identical construction warmup. IPC validation follows elapsed/CPU/RPC, held RSS and peak RSS snapshots; external RSS sampling also stops before validation. Post-GC RSS is measured after validation, result deletion and garbage collection, so it includes validation/retention effects. Error cases validate predefined exact construction exceptions and have no IPC claim. RSS observations and their gaps are retained. The receive-only fake server does not execute the submitted relation; no JVM/live Spark is used.

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

Generated-by: OpenAI Codex (GPT-6)

This pull request and its description were written by Isaac.

dbtsai and others added 20 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>
Build on the canonical pandas Index and integer guards from PR121.
Serialize eligible native float64 columns to inline Arrow IPC without row
materialization, preserving explicit schema and pandas missing values.

Include the shared internal inline IPC bridge and public Float coverage.

Co-authored-by: Isaac <no-reply@databricks.com>
Remove the extra blank line at EOF copied from the donor suffix.
All helper function bodies remain byte-identical.

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