Add Options for DataFrameWriter - #167
Open
pequalsnp wants to merge 8 commits into
Open
Conversation
pequalsnp
changed the title
pequalsnp
marked this pull request as ready for review
September 5, 2025 12:49
caldempsey
approved these changes
Dec 3, 2025
caldempsey
suggested changes
Dec 3, 2025
There was a problem hiding this comment.
Can you add an integration test to demonstrate usage and use a Dataframe option which would affect the displayed results? Maybe something like NULLVALUE -> NA.
See https://github.com/apache/spark-connect-go/tree/master/internal/tests/integration
Author
|
Author
I used the |
pequalsnp
force-pushed
the
data-fram-write-options
branch
from
December 3, 2025 19:38
007e7c4 to
c2fc9a0
Compare
…tion Reattachable execution was requested but never carried out: the stream discarded the operationId it was given, and ToTable treated the end of the response stream as the end of the result. The server ends a reattachable stream every senderMaxStreamDuration (2m by default) expecting the client to resume, so any query running longer than that failed. ToTable now resumes on an early stream end and stops only at ResultComplete. Empty rotations are normal -- a write emits nothing until it finishes -- so the loop is bounded by ctx rather than by a retry count. ToTable takes a context now; every caller already had one in scope.
Two behaviours the reference clients have that the reattach commit did not. ReleaseExecute. A reattachable execution buffers responses server-side so a resumed stream can backtrack, and the server only drops them when the client says it is safe. Send release_until as each response is consumed and release_all once the result is complete or the execution is abandoned. It is fire-and-forget on a detached context: releasing is an optimisation, the server ages unreleased executions out anyway, and failing a query because a cleanup RPC failed would be worse than the buffering it avoids. Restart on INVALID_HANDLE.OPERATION_NOT_FOUND / SESSION_NOT_FOUND. The client sets its own operation id precisely so it can tell that the original ExecutePlan never reached the server; when it did not, nothing ran and re-issuing it is safe. That only holds while no response has been consumed -- afterwards a fresh execution would replay them and duplicate rows, so that case fails instead. Re-validated against EMR Serverless: a 400s statement still completes (402.5s, against 402.5s before these additions).
…sumes Follow-up fixes to the two ReattachExecute commits before this one. Correctness: - Only replay a plan that can be shown to have no side effects. Being rooted in a relation is not enough: Relation_Sql carries raw SQL text that the server applies while planning it -- which is why Sql submits the query as a command first -- and Relation_Catalog carries CreateTable and DropTempView. Sql's fallback path wraps the query text as a Relation_Sql, so an INSERT was reachable and would have been executed a second time. replayableRelation lists the shapes that are safe instead, so one it has not been taught about, including one added to the proto later, is refused rather than replayed. The refusal names the shape it saw, so a missing case is not mistaken for a deliberate one. - Release the Arrow records the read loop accumulates. ReadArrowBatchToRecord already retains, and NewTableFromRecords takes its own references to the arrays underneath, so the extra Retain leaked a query's entire result. - Stop reading once ResultComplete arrives. The execution has just been disowned with release_all, so an error on the following read would have failed a query whose result was complete and no longer recoverable. - Restart an execution at most once, and refuse to restart into a session the server has forgotten. The reference clients do restart on SESSION_NOT_FOUND, but they also track the server's own session id across every response and fail if it changes; without that check a replay would silently run in a fresh empty session, without the temporary views, cached tables and SQL confs the caller set up. The divergence and the condition for removing it are recorded at the call site. Resilience: - Resume a stream that breaks mid-flight, not only one the server ends cleanly. Only the first stream comes from the retrying RPC client, so without this a long query was less able to survive a blip after its first rotation than before it. - Bound that path with the retry policy's budget and backoff, reset by every response received. A clean rotation is paced by the server holding each stream open for senderMaxStreamDuration, but a stream that fails on its first Recv costs nothing, so it would otherwise have been reattached to in a tight loop for as long as the caller's context allowed. API: - Add SparkSessionBuilder.WithReattachExecution. Build was the only place client options were constructed for real use and it hardcoded the default, so the feature was unreachable no matter what the client supported. Verified with go build, go vet, go test -race ./spark/... and golangci-lint v2.1.6 (0 issues). Integration tests need a live Spark Connect server and were not run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Implement ReattachExecute so long-running queries survive stream rotation
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
3 participants
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?
Why are the changes needed?
Does this PR introduce any user-facing change?
Yes!
It adds the ability to set options on DataFrameWriter
How was this patch tested?
Added a call to Option in the unit tests.
Added an integration test for writer options