Repository navigation
Conversation
This was referenced Oct 29, 2026
HyukjinKwon
added a commit
that referenced
this pull request
Nov 5, 2026
…am interface ### What changes were proposed in this pull request? [SPARK-60675](https://issues.apache.org/jira/browse/SPARK-60675) adds `spark.createDataFrame(pyarrow_table)` to the native Rust-backed Python client. Previously, a PyArrow Table reached the Python-list cast and failed. The client uses PySpark's Arrow schema/timestamp conversion helpers, imports the normalized reader through arrow-rs's owned C Stream importer, and serializes the Rust batches directly to IPC without materializing Python rows. Empty input requires an explicit schema and no configuration RPC. Non-empty Arrow input fetches all eight normalization/upload settings in one strict `Get` request and parses them into named fields. The supported server floor remains Spark 4.2; there are no missing-setting fallbacks or byte-unit compatibility parser. Small inputs stay inline. Inputs at or above `spark.sql.session.localRelationCacheThreshold` become session-scoped cache artifacts referenced by `ChunkedCachedLocalRelation`, allowing input larger than the 128 MiB message limit. Inferred schemas stay in Arrow IPC: the inline Spark schema and cached schema hash are omitted, with column names applied using `toDF`, matching upstream. Explicit schemas are retained. Chunk serialization checks the schema-plus-batch-data estimate before writing, as upstream does; the row setting caps record batches rather than entire streams. The decimal relation-size check is retained, as in official PySpark 4.2. Artifact uploads use the existing 32 KiB implementation, with SHA-256 through `sha2`, cache lookup/deduplication, CRC validation, and acknowledgement checks. Generic artifact framing/batching is unchanged. The 1 MiB frames, shared-buffer changes, direct `ring` dependency, and batching-overhead change have been removed from this PR. User-context handling is left to #140. The batched configuration work overlaps and supersedes #134. #108 is already merged and is no longer a dependency. This replaces the simpler `createDataFrame(pyarrow.Table)` path from #141 ([SPARK-60739](https://issues.apache.org/jira/browse/SPARK-60739)), which serialized the table with PyArrow and always sent it inline. Its IPC-bytes `SparkSession::create_dataframe_from_arrow` and the `RuntimeConf::get_many` helper only it used are removed; neither was released. Its `is_loaded_instance` check is kept, and treats a `None` entry in `sys.modules` (a blocked import) as not loaded. ### Why are the changes needed? PySpark Connect accepts Arrow Tables directly. The native client needs equivalent input support while preserving Arrow's columnar representation. Cached relations keep large input out of subsequent query plans. Schema casts, alignment, and IPC serialization can still copy data; this PR makes no zero-copy or general performance claim. ### Does this PR introduce any user-facing change? `createDataFrame` accepts a `pyarrow.Table` with inferred or explicit schemas, atomic schemas, column-name lists, nested values, dictionary arrays, slices/chunks, decimals, nulls, and timestamp conversion. Empty tables require an explicit schema. Explicit schemas with incompatible column counts raise `AXIS_LENGTH_MISMATCH` before Arrow casts. Unsupported top-level schema types, including ArrayType/MapType objects and their DDL equivalents, raise `NOT_EXPECTED_TYPE`; those types remain valid struct fields. `isLocal()` matches PySpark: it is true for input with an explicit schema, inline or cached, and false when inferred or listed column names are applied with `toDF`, which the server resolves into a `Project`. A zero-column table, whose rows the input cast drops, keeps its Arrow schema, so the relation stays valid. The public Python entry point remains `createDataFrame`. No vendored files or CI workflows change. The shared Rust settings snapshot and importer entry point are hidden from documentation and documented as internal, unstable APIs. ### How was this patch tested? After merging master, which now tracks Spark 4.3.0, the final revision was checked locally on macOS arm64 (Python 3.11, PyArrow 22) against a Spark 4.3.0 server. The Rust lib tests, all integration targets with and without `wasm-udf`, the server-gated Rust e2e suites, the offline Python suites, and the live drop-in and Arrow suites passed. `isLocal()` and zero-column input were compared with official PySpark 4.3.0. CI passed against Spark 4.3.0, including the official Connect, pandas, and pipelines parity jobs. The results below are from before the merge. Local verification of `0c335de` on macOS arm64, Python 3.12.14, PyArrow 26, Spark Connect 4.2.0: - 403 Rust client and 95 transport release unit tests passed. All client integration targets compiled with and without `wasm-udf`; formatting and whitespace checks passed. - 393 Python tests passed, 8 live-server cases skipped, across the offline and mock Arrow suites. Tests cover inferred/explicit wire schemas on inline/cached paths, one strict configuration RPC, chunk boundaries/order, deduplication, size limits, and CRC/missing-ack failures. There is one mock input just above 128 MiB rather than repeated large allocations. - All 34 direct Arrow cases passed against a real local Spark server, including the large binary payload's count and checksum. - Eight native Arrow end-to-end tests passed, including a 2 MiB cached upload and timestamp/NTZ values in a non-UTC session through both inline and cached paths. Tests create and close isolated child sessions. Client-only width/type checks stay offline or use the mock service; the live test does not assert a particular server configuration representation. - Official creation/collection parity through the Rust transport: 33 passed, 2 skipped. API introspection retains the existing `SparkSession.Hook` gap and intentionally absent `Column.to_plan`. On `0c335de`, Rust stable/WASM, Rust checks and behavioral E2E, coverage, docs, lint, pipelines parity, vendored drift, and the E2E benchmark checks have passed. Local all-target Clippy still reports existing baseline errors; comparison with the previous run found no new error categories/sites or diagnostics in the changed Arrow importer. PyArrow 10 compatibility has not been rerun. The pre-existing minimum-Rust mismatch remains: verified Cargo 1.82 cannot parse the locked edition-2024 dependency manifest; existing dependencies require Rust 1.88 despite lower manifest declarations. **Real local Spark benchmark, `0c335de`** Measured the final 32 KiB implementation against official PySpark 4.2.0 using a real Spark Connect 4.2.0 JVM on the same Apple M3 Max / 64 GiB machine (`local[2]`, 3 GiB driver heap, loopback without TLS). Both clients used Python 3.12.14, PyArrow 26, pandas 2.3.3, and stock Arrow upload settings, including the 1 GiB upload-batch cap. Each client/input has ten fresh-process, fresh-session samples, run serially with alternating client order; six JVM/JIT preflight runs are excluded. All 60 measured runs verified row count and sums of all eight int64 columns. Input preparation, session creation, metadata reads, and unrelated warm queries are outside timing; `createDataFrame`'s own RPCs remain timed. Cells are median [minimum–maximum] milliseconds. Creation plus the first materializing aggregation: | Input | Official PySpark | Native 32 KiB | | --- | ---: | ---: | | 0.8 MB | 89.99 [68.41–111.87] | 99.58 [67.28–177.19] | | 16 MB | 509.01 [429.79–552.63] | 525.43 [481.12–707.01] | | 136 MB | 3324.11 [3079.14–3815.65] | 3513.54 [3299.49–3945.67] | Warmed aggregation, using each fresh session's median of three queries after an untimed query: | Input | Official PySpark | Native 32 KiB | | --- | ---: | ---: | | 0.8 MB | 38.77 [27.65–54.79] | 37.30 [28.16–44.10] | | 16 MB | 224.19 [204.97–254.42] | 275.67 [246.35–301.71] | | 136 MB | 2042.50 [1986.79–2180.77] | 3043.90 [2732.53–3534.39] | For 136 MB, creation/upload alone was 528.30 ms (official) versus 774.34 ms (native). Sampled peak extra client RSS above the prepared input was 189.34 MiB (official) versus 288.18 MiB (native); 5 ms sampling can miss short peaks. The large warmed-query gap remains after omitting inferred schemas: native was about 49.0% slower in this run. Removing the extra schema is the correct wire behavior, but these results do not show that it resolves the regression. A separate source/jar check confirmed that fresh native plan IDs prevent reuse of Spark 4.2's full-protobuf plan-cache keys, while official PySpark retains stable IDs. Its timing contribution has not been isolated, and generic plan-ID behavior is outside this PR's scope. These results replace the earlier fixture/1 MiB-frame measurements. Query timings include Spark execution and transport; isolated server execution time, network bytes, and per-query upload counts were not measured. Performance against a server on another machine over normal network/TLS conditions remains unmeasured. Benchmark methodology, results, and remaining validation notes are kept here rather than in a merge-gating repository document. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (application version unavailable). This pull request and its description were written by Isaac. Closes #118 from dbtsai/codex/arrow-table-input. Lead-authored-by: DB Tsai <dbtsai@dbtsai.com> Co-authored-by: Hyukjin Kwon <hyukjin.kwon@databricks.com> Signed-off-by: Hyukjin Kwon <hyukjin.kwon@databricks.com>
Read the three mutable Arrow normalization settings in one Get request per nonempty construction while retaining empty-input and transport behavior. Co-authored-by: Isaac <no-reply@databricks.com>
dbtsai
force-pushed
the
codex/spark-60698-batch-arrow-config
branch
from
November 6, 2026 02:09
5737f1d to
13e3558
Compare
dbtsai
changed the title
Keep the tests of both sides. The configuration snapshot tests move before test_non_arrow_input_does_not_import_pyarrow, apart from the tests appended at the end. Co-authored-by: Isaac <no-reply@databricks.com>
HyukjinKwon
marked this pull request as ready for review
November 6, 2026 06:37
HyukjinKwon
added a commit
to dbtsai/spark-connect-rust
that referenced
this pull request
Nov 6, 2026
Stack this on the replacements of apache#143 and apache#144 (apache#207 and apache#208), on the current master, which has apache#121, apache#128, apache#130 and apache#134 that this branch carried copies of: take the stacked branch's files, and add the date hook after the string one, and the date functions and tests after theirs. Co-authored-by: Isaac <no-reply@databricks.com>
HyukjinKwon
added a commit
that referenced
this pull request
Nov 6, 2026
…rrow ### What changes were proposed in this pull request? Following #121 (SPARK-60676) and #128 (SPARK-60685), `createDataFrame(pandas_df, schema)` with an explicit `StructType` of `DateType` fields now converts object columns of exact `datetime.date` values and `None`, of at least 10,000 rows, through the Arrow importer (`date32`), instead of materializing Python rows. `NaN`, `NaT`, `pd.NA`, `datetime` and other objects, `None` in a non-nullable field, smaller frames, inferred schemas and customized inputs keep the row path, as for #121. It is stacked on #208 (SPARK-60720), which is stacked on #207 (SPARK-60727): the four pandas paths add their hooks to the same dispatcher and their functions and tests to the same files. Until those merge, its diff includes theirs. The copies of #128's and #134's commits that this branch carried are on master now, so they no longer show in the diff. In the merge with master and #208: the capture mock of #128's calendar-range test now serves the Arrow importer's settings, as its explicit pandas case has more than 10,000 rows and goes through the importer; and, as for #121, a test that required a second row conversion to read `Index._data` exactly as often as the first now checks that the path declines without more reads than the row path. ### Why are the changes needed? [SPARK-60716](https://issues.apache.org/jira/browse/SPARK-60716): converting each date through Python rows dominates `createDataFrame` of large date frames. Measured on macOS against a local Spark 4.3.0 server, as the client CPU time of `createDataFrame` of a pandas DataFrame of 200,000 rows of two date columns with an explicit schema (median of 5): 213 ms through the row path with #128, 39 ms with this change (official PySpark 4.3.0: 34 ms). The results are the same, `date.min`, `date.max` and nulls included. ### Does this PR introduce _any_ user-facing change? No, other than speed. ### How was this patch tested? - New tests in `python/tests/test_arrow_handoff.py`, against the mock Arrow input service, check the `date32` values of the IPC, nulls, names, nullability and metadata, the row threshold, and that each unsupported or customized input keeps the row path with its errors. - Against the current master, merged with #207 and #208 into this branch, and a local Spark 4.3.0 server: the lib and every integration test of `apache-spark-connect` and `apache-spark-connect-core`, e2e included (789 passed), the build of all tests with `wasm-udf`, the offline Python suites (1,775 passed, 8 skipped), the live drop-in and Arrow suites (86 passed) and the live cases (10 passed). `cargo fmt --all --check` passes. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6); the merge with master and the stacked PRs, the review follow-ups and the measurement by Claude Opus 5.5 This pull request and its description were written by Isaac. Closes #138 from dbtsai/codex/spark-60716-columnar-pandas-date. Lead-authored-by: DB Tsai <dbtsai@dbtsai.com> Co-authored-by: DB Tsai <db.tsai@databricks.com> Co-authored-by: Hyukjin Kwon <hyukjin.kwon@databricks.com> Signed-off-by: Hyukjin Kwon <hyukjin.kwon@databricks.com>
HyukjinKwon
pushed a commit
that referenced
this pull request
Nov 7, 2026
…tings in tests ### What changes were proposed in this pull request? The Arrow input settings test parametrized missing and null values, but its mock omitted both from `ConfigResponse`. Preserve null-valued keys with the optional protobuf value unset and assert the two response shapes. Cover missing and null values for all eight required settings, document the one-shot configuration error hook, and assert the exact snapshot request in the schema handoff test. ### Why are the changes needed? Follow-up to #134 ([SPARK-60698](https://issues.apache.org/jira/browse/SPARK-60698)). The two previous cases produced identical wire responses, leaving the null-value path untested. Only three of the eight required settings had missing-value coverage. ### Does this PR introduce _any_ user-facing change? No, tests only. ### How was this patch tested? A private wire-level reproduction fails on current master and passes after this change. It executes the actual nested mock Config handler through installed official protobuf classes; all 16 missing/null response cases pass, including response ordering and optional-value presence. Python syntax and `git diff --check` pass. Tooling unit tests: 12 passed, 2 skipped. Draft: native runtime and mandatory pre-push validation remain pending. This host has no Cargo/Rust/protoc, so the five required Cargo checks and extension build could not start. The Arrow suite fails at collection without a current-built extension (`Row` cannot be imported); API introspection and the official parity gate remain unrun. Wire-level checks do not prove the Rust error path. No full validation convergence is claimed. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6). Independently source-reviewed using installed local Claude Code. This pull request and its description were written by Isaac. Closes #290 from dbtsai/codex/postmerge-pr-134-config-snapshot-tests. Authored-by: DB Tsai <dbtsai@dbtsai.com> Signed-off-by: Hyukjin Kwon <hyukjin.kwon@databricks.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Tests for the snapshot of Arrow input settings that
createDataFramereads for Arrow input: oneConfigRequest.getof eight keys (spark.sql.timestampType,spark.sql.session.timeZone,spark.sql.execution.arrow.useLargeVarTypesand the fivespark.sql.session.localRelation*limits). The runtime batching is already on master; this adds only the tests. They check:useLargeVarTypes;SQL_CONF_NOT_FOUNDfor a missing or null setting, with no upload;createDataFrame, after the settings change;The mock service in
test_arrow_handoff.pygains the hooks for these: settings that tests can change, an injectable configuration error, reversed answers, and the decoded input tables.Why are the changes needed?
SPARK-60698: the snapshot had no tests of its own, so a change to its keys, its order mapping or its error handling would go unnoticed.
Does this PR introduce any user-facing change?
No, tests only.
How was this patch tested?
Against the current master, merged into this branch, and a local Spark 4.3.0 server: the lib and every integration test of
apache-spark-connectandapache-spark-connect-core, e2e included (771 passed), the build of all tests withwasm-udf, the offline Python suites (738 passed, 9 skipped), the live drop-in and Arrow suites (76 passed) and the live cases (10 passed).cargo fmt --all --checkpasses.Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6); the merge with master by Claude Opus 5.5
This pull request and its description were written by Isaac.