Repository navigation
Spark 4.2 follow-ups: review fixes, release drain, Windows paths, protobuf advisory - #55
Merged
Merged
Conversation
`StreamReader.changes/2` had only unit encoding coverage. Drive it end to end on the Spark 4.2 fixture server: read `cdc_e2e.p1_cdc` from startingVersion 1 with an available-now trigger into a memory sink, then assert the replayed change types and commit versions match the seeded insert/update/delete history and that query progress reports the changelog micro-batch stream as its source. No fixture change was needed. Spark42SeededChangelogCatalog inherits InMemoryTableWithChangelog, which already serves the seeded history as a streaming change scan via InMemoryChangelogMicroBatchStream; document that in the fixture README and note the coverage in docs/spark_42.md. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The {:one_col_defaults, _} and {:two_col_defaults, _} generators only took
positional arguments, so a keyword list landed in the first optional slot and
encoded as a literal keyword tuple, which PlanEncoder rejects. Resolve keyword
lists against the registry defaults instead, keeping the positional form.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The specs allowed a %SparkEx.Reader{} builder as the first argument, but only
the %DataFrame{} branch handled one, so a builder fell into data_source/4 and
crashed in GenServer.whereis/1. Pin the format and reuse load_from_builder/3 so
builder schema/options are preserved and call-time options override them, and
de-duplicate the csv separator normalization.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
protobuf 0.16.0 carries GHSA-rv48-qqj5-crxg (HIGH): unbounded recursion when decoding deeply nested embedded messages. grpc 0.11.5 requires ~> 0.14, so 0.17.0 resolves without moving grpc. Regenerated with protoc-gen-elixir 0.17.0; the generated diff is limited to the generator version stamp. protoc-gen-elixir 0.17 nests output under the package module path, so gen_proto.sh now stages the output and keeps the existing lib/spark_ex/proto/spark/connect layout. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Point the manual integration-test setup and proto-regeneration snippet at the 4.2.0 distribution (the vendored proto target) instead of 4.1.1, and mention the Spark 4.2 provider fixtures used for CDC/real-time/pending- operation tests.
Mirror upstream PySpark's SPARK-55071 fix: an absolute Windows drive path (C:\foo\bar.jar) or a UNC path (\\host\share\file.jar) is a local path, not a URI with a single-letter scheme, and file: URIs (with or without the extra authority slash, percent-encoded or not) should resolve to the underlying local path rather than being treated as a literal, unparseable filename. Add SparkEx.Artifacts.classify_local_path/1 as a pure, filesystem-free classifier so Windows-shaped inputs can be unit-tested from any host OS, and wire it into Artifacts.prepare/2's extension check and file stat/basename resolution, and into Session.copy_from_local_to_fs/3's local-path handling. Ordinary POSIX paths and remote URI schemes (hdfs://, s3://, ...) are untouched, passed through exactly as before. Update the pre-existing Spark42SmallFixesTest, which had asserted the old failure-on-file-URI behavior, to reflect the new decode-and- resolve behavior instead.
PySpark 4.2 (SPARK-55406 / SPARK-55362, reattach.py) keeps ReleaseExecute
on a dedicated executor and `SparkConnectClient.close()` waits up to 10s
for the pending release futures *before* sending ReleaseSession, so the
server is not left with orphaned executions. spark_ex fired those
releases as fire-and-forget tasks on the global SparkEx.TaskSupervisor
with no way for Session.terminate/2 to know about them.
Add SparkEx.Internal.ReleaseTracker: an ETS row ({session_id, ref} ->
task pid) inserted by the caller *before* the task is spawned and removed
by the task when it finishes. No new supervisor or registry process — the
existing Task.Supervisor still runs the tasks and EtsTableOwner owns the
table. Wire it into the three release-task call sites: the reattach
stream's release_all and release_until checkpoints (via the stream ctx's
session) and the ManagedStream controller's release (via a new
:session_id option threaded from Client.execute_plan_managed_stream/3).
Session.terminate/2 now drains this session's pending releases before the
ReleaseSession task. The wait is bounded by :release_drain_timeout_ms
(new session option, default 10_000 like PySpark, also settable via the
:spark_ex app env), returns immediately when nothing is pending, only
ever waits on rows keyed by this session's id, and prunes rows whose task
died without cleanup. A drain timeout logs and emits
[:spark_ex, :session, :release_drain, :timeout]. Terminate stays
idempotent: the released/no-channel clauses return before the drain, so
no RPC is issued after channel closure.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Adds test/unit/session_shutdown_test.exs, exercising SparkEx.Session against a local TCP listener that accepts and never answers (the new test/support/blackhole_server.ex helper) plus a closed port. Covers Session.stop/1, Supervisor.stop/1 with the session as a supervised child, and Process.exit(pid, :shutdown), asserting each returns/terminates well inside 15s — the worst case implied by the existing timeouts: the release drain (:release_drain_timeout_ms) plus the ReleaseSession Task.yield (5s) plus Task.shutdown's grace (5s). Also pins that a hanging ReleaseExecute cannot push stop past the drain timeout, and that session creation against a closed port fails fast (~5s, gun's connect timeout) rather than hanging. No production change was needed: stop already returns in ~5s against an unresponsive server (the ReleaseSession task is yielded for 5s and then shut down), and the drain added in the previous commit is bounded. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
- Drain in-flight ReleaseExecute tasks on the explicit Session.release/1
path too, not only on stop/terminate.
- Size SparkEx.Session.child_spec shutdown to the drain budget so a default
supervisor no longer brutal-kills the session mid-drain.
- Reject mixing positional and keyword options in tuple-sketch wrappers
instead of failing at plan-encoding time.
- Accept a %Reader{} builder with a path in parquet/text/orc/avro readers.
- Loosen the protobuf requirement to ~> 0.16.1 or ~> 0.17 (the advisory fix
landed in 0.16.1) and correct the CHANGELOG wording.
- Tighten the fail-fast connect assertion in the shutdown test.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
There was a problem hiding this comment.
🟡 Changes recommended
There are a couple of correctness/portability issues in the updated helper scripts and shutdown configuration logic that should be addressed before approval.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR consolidates post-Spark-4.2 follow-ups across the SparkEx client, focusing on correctness fixes (function wrapper option encoding, Reader builder call shapes, artifact path parsing), more robust session shutdown behavior (bounded drain of in-flight ReleaseExecute tasks before ReleaseSession), and maintenance updates (protobuf advisory bump + proto regeneration workflow), with expanded unit/integration coverage and updated documentation.
Changes:
- Add session-scoped
ReleaseExecutetask tracking + bounded release-drain beforeReleaseSessionon shutdown/release, with targeted tests for unreachable/unresponsive servers. - Fix API call shapes and encoding: tuple-sketch wrappers accept keyword defaults safely;
Reader.*format entrypoints accept a%Reader{}builder with path(s); artifact paths classify Windows paths andfile:URIs correctly. - Update protobuf dependency/lock and adjust proto regeneration script to preserve historical output layout; update README/CHANGELOG/docs and add Spark 4.2 streaming CDC integration coverage.
File summaries
| File | Description |
|---|---|
| test/unit/spark42_small_fixes_test.exs | Updates artifact handling expectation for file: URIs to resolve to a local path. |
| test/unit/spark_42_functions_test.exs | Adds unit coverage for tuple-sketch keyword defaults, mixed-args rejection, and encodability. |
| test/unit/session_shutdown_test.exs | New unit tests ensuring shutdown/stop are bounded against black-hole servers and closed ports. |
| test/unit/release_drain_test.exs | New unit tests for session-scoped release tracking and drain ordering/timeouts. |
| test/unit/reader_test.exs | Adds coverage for Reader builder + path(s) support and option/schema precedence. |
| test/unit/artifact_test.exs | Adds coverage for file:// path resolution and classify_local_path/1 behavior. |
| test/support/spark42/README.md | Documents seeded changelog fixture behavior and streaming CDC coverage rationale. |
| test/support/blackhole_server.ex | New test helper TCP server to simulate “accepting but unresponsive” endpoints. |
| test/integration/spark42_p1_readwrite_test.exs | Adds live streaming CDC replay test via memory sink + available-now trigger. |
| README.md | Updates local dev instructions to Spark 4.2.0 + fixture-provider server guidance. |
| priv/scripts/gen_proto.sh | Adjusts proto generation to handle protoc-gen-elixir output nesting while keeping layout. |
| mix.lock | Locks protobuf to 0.17.0. |
| mix.exs | Raises protobuf requirement to ~> 0.16.1 or ~> 0.17. |
| lib/spark_ex/session.ex | Adds release_drain_timeout_ms, drains tracked releases before ReleaseSession, sizes child shutdown, cleans tracker rows. |
| lib/spark_ex/reader.ex | Extends format entrypoints to accept %Reader{} builders with path(s) and pins formats. |
| lib/spark_ex/proto/spark/connect/types.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/relations.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/pipelines.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/ml.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/ml_common.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/expressions.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/example_plugins.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/common.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/commands.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/catalog.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/proto/spark/connect/base.pb.ex | Regenerated protos: updates protoc-gen-elixir version stamps. |
| lib/spark_ex/managed_stream.ex | Tracks release tasks by session_id so shutdown can drain them before releasing session. |
| lib/spark_ex/macros/function_gen.ex | Adds keyword-default resolution + mixed-args rejection for *_defaults wrappers. |
| lib/spark_ex/internal/release_tracker.ex | New ETS-backed per-session tracking/drain of in-flight release tasks. |
| lib/spark_ex/ets_table_owner.ex | Ensures release-task ETS table exists for process-lifetime management. |
| lib/spark_ex/connect/client.ex | Routes release tasks through ReleaseTracker, keyed by session_id. |
| lib/spark_ex/artifacts.ex | Adds classify_local_path/1 and uses it when validating/stat’ing artifact inputs. |
| docs/spark_42.md | Documents new streaming CDC integration coverage. |
| CHANGELOG.md | Adds an Unreleased entry summarizing the Spark 4.2 follow-up fixes and updates. |
Review details
- Files reviewed: 33/34 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+2244
to
+2255
| drain_ms = | ||
| case Keyword.get(opts, :release_drain_timeout_ms) do | ||
| ms when is_integer(ms) and ms >= 0 -> | ||
| ms | ||
|
|
||
| _ -> | ||
| Application.get_env( | ||
| :spark_ex, | ||
| :release_drain_timeout_ms, | ||
| @default_release_drain_timeout_ms | ||
| ) | ||
| end |
Owner
Author
There was a problem hiding this comment.
Fixed in d1b8d9f: child_spec/1 now calls the same normalize_release_drain_timeout_opt/1 used by start_link/1, so an invalid value raises the same ArgumentError (test added).
| # protoc-gen-elixir >= 0.17 nests its output under the package module path | ||
| # ("spark/connect") on top of the .proto path, so generate into a staging | ||
| # directory and keep the historical lib/spark_ex/proto/spark/connect layout. | ||
| STAGING_DIR="$(mktemp -d)" |
Owner
Author
There was a problem hiding this comment.
Fixed in d1b8d9f: mktemp now receives an explicit template (${TMPDIR:-/tmp}/spark_ex_gen_proto.XXXXXX), which works on both GNU and BSD/macOS mktemp. Verified the script regenerates an identical tree on macOS.
… portable mktemp - SparkEx.Session.child_spec/1 resolves :release_drain_timeout_ms through the same normalizer as start_link/1, so invalid values raise the same ArgumentError instead of being silently ignored. - gen_proto.sh passes an explicit template to mktemp for BSD/macOS portability. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.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.
Closes the items left open after the Spark 4.2 PRs (#51–#54) and the post-merge review recorded in
SPARK_42_FABLE.txt§10.Fixes
tuple_sketch_agg_*,tuple_union_agg_*,tuple_union_*,tuple_union_theta_*acceptlg_nom_entries:/mode:keyword options; a keyword list was previously encoded as a literal and failed at plan-encoding time. Unknown keys, or mixing positional and keyword options, raiseArgumentError. Positional form unchanged (still sendslit(12), lit("sum")like PySpark).Reader.csv/json/xml(andparquet/text/orc/avro) accept a%SparkEx.Reader{}builder with a path or list of paths; the advertised call shape previously crashed withFunctionClauseError. Builder schema/options are honoured, call-time options win.C:\x.jar) and UNC paths are local paths, not URIs;file:URIs (percent-encoded, with/without authority slash, Windows or POSIX targets) resolve to the underlying file. POSIX paths and remote URIs are byte-for-byte unchanged. Pure classifierSparkEx.Artifacts.classify_local_path/1with filesystem-free tests.Lifecycle
Session.stop/1,terminate/2and the explicitSession.release/1wait (bounded by new:release_drain_timeout_ms, default 10 s as in PySpark'sclose()) for this session's in-flight ReleaseExecute tasks, tracked in an ETS table owned by the existingEtsTableOwner. Other sessions' releases are never awaited.Session.child_spec/1now sizes the supervisorshutdownto the drain budget so a default supervisor no longer brutal-kills the session mid-drain.stop/1, supervised shutdown andProcess.exit(:shutdown)complete within the bounded timeouts, and connecting fails fast. No production change was needed for this beyond keeping the drain bounded.Coverage / deps / docs
StreamReader.changes/2against the seeded changelog fixture (the upstreamInMemoryTableWithChangelogalready supports a streaming change scan; no fixture change needed).protobufrequirement~> 0.16.1 or ~> 0.17, lock at 0.17.0 (GHSA-rv48-qqj5-crxg fixed from 0.16.1); protos regenerated with the matching plugin — the diff is version stamps only, no message/field changes.gen_proto.shkeeps the historical output layout.Validation
tools/check.sh(format, compile--warnings-as-errors, credo, unit tests, dialyzer): green — 2735 passed, dialyzer 0 errors.tools/check.sh integrationon Spark 4.2.0 with provider fixtures (SPARK_EX_TEST_PROVIDERS=1): 3460 passed, 2 skipped, 3 excluded.tools/check.sh integrationon Spark 4.1.1: 3422 passed, 2 skipped, 41 excluded.file://host/shareauthority drop (PySpark parity),:bagETS layout for the tracker.🤖 Generated with Claude Code