akshaychitneni opened a new pull request, #2633:
URL: https://github.com/apache/datafusion-ballista/pull/2633
# Which issue does this PR close?
Closes #1829.
# Rationale for this change
`DataFrame::cache()` is not supported in Ballista. This PR adds it as steps
2–4
of the design doc's plan (materialize + read-back, pinning, recovery), on
top of
the cache registry from step 1. A cached dataset is ordinary shuffle output,
pinned past job cleanup and reused through Ballista's existing shuffle reader
and stage recovery.
# What changes are included in this PR?
- **Materialize (cold path):** a cache miss runs the cached subplan as its
own
build job, whose final shuffle output is pinned. The query job is held as
Queued, without blocking, and dispatched once the build completes. Builds
are
shared between queries and run in parallel; a failed build fails the
queries
waiting on it.
- **Read-back (warm path):** a cache node becomes a `CacheReadNode`, planned
as
a `ShuffleReaderExec` over the pinned locations. It is registered as a
complete stage input under a reserved stage id (`EXTERNAL_STAGE_ID_BASE`).
The subplan is not run.
- **Pinning and validate-on-read:** job-data cleanup skips pinned stages,
and a
location on a dead executor turns a hit into a rebuild.
- **Recovery:** losing cache data mid-query rolls the reading stage back
through
Ballista's existing path. The cache is rebuilt and the stage resumes,
bounded
by `stage_max_failures`. The adaptive planner fails the job cleanly
instead.
- **Planning:** the scheduler's default session builder installs
`CacheReadQueryPlanner` (DataFusion's default planner plus
`CacheReadPlanner`)
through `SessionStateBuilderExt`. A session that already has a query
planner
keeps it.
- **Extension points:** `CacheStorage` (`override_cache_storage`) for how a
cache is written and read, `CacheState` for the registry backend, and
`CacheReadQueryPlanner::with_extension_planners` for sessions with their
own
planner. External stage inputs are generic in the graph, not
cache-specific.
- **Tests and docs:** scheduler flow tests, client end-to-end tests, a chaos
scenario, and the design doc
`docs/source/contributors-guide/dataframe-cache.md`.
No proto changes, no changed trait signatures, and existing tests pass
unchanged.
# Are there any user-facing changes?
Yes, opt-in:
- Setting `ballista.cache.noop = false` makes `df.cache()` materialize and
reuse
results. The setting is on by default, so `cache()` stays a no-op unless
the
user changes it.
- New scheduler options: `--disable-dataframe-cache` (operator switch) and
`SchedulerConfig::override_cache_storage`.
- `EXPLAIN` of a cached frame never materializes it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
