andygrove commented on code in PR #2568:
URL:
https://github.com/apache/datafusion-ballista/pull/2568#discussion_r4250257102
##########
ballista/scheduler/src/cluster/memory.rs:
##########
@@ -972,3 +978,124 @@ mod test {
Ok(())
}
}
+
+/// In-memory [`CacheState`] backend — the only implementation today.
+///
+/// A thin async wrapper over the synchronous, unit-tested `CacheRegistry`,
+/// mirroring how [`InMemoryJobState`] wraps its maps. A durable backend would
be
+/// a sibling type implementing [`CacheState`] whose `init` loads persisted
+/// entries; no caller changes needed.
+#[derive(Debug, Default)]
+pub struct InMemoryCacheState {
Review Comment:
Nit: `InMemoryCacheState` lands after `mod test` here. Would it read more
naturally next to `InMemoryJobState`, with the two new tests folded into the
existing `mod test`?
##########
ballista/scheduler/src/cluster/mod.rs:
##########
@@ -363,6 +389,82 @@ pub trait JobState: Send + Sync {
fn produce_config(&self) -> SessionConfig;
}
+/// Scheduler-side metadata store for materialized `DataFrame::cache()`
results —
+/// the mapping from a cache key to the shuffle partition locations of its
output.
+///
+/// This is the third piece of scheduler state alongside [`ClusterState`] and
+/// [`JobState`], and follows the same shape: an async trait with an in-memory
+/// implementation today ([`InMemoryCacheState`])
+/// and an [`init`](CacheState::init) hook so a future durable backend (e.g.
one
+/// backed by an object store) can rehydrate entries on startup. The methods
are
+/// `async` even though the in-memory backend does no I/O, so a durable backend
+/// slots in without changing any signatures.
+///
+/// A cache *miss* is simply [`lookup`](CacheState::lookup) returning `None`
(the
+/// key is absent, or materialization is still in flight); a *hit* is `Some`.
+#[async_trait::async_trait]
+pub trait CacheState: Send + Sync + 'static {
+ /// Initializes the backend. A no-op for the in-memory backend; durable
+ /// backends load previously persisted entries here.
+ async fn init(&self) -> Result<()> {
+ Ok(())
+ }
+
+ /// Returns the materialized data for `key`, or `None` on a miss (absent or
+ /// still materializing).
+ async fn lookup(&self, key: &CacheKey) ->
Result<Option<MaterializedCache>>;
+
+ /// Attempts to claim `key` for materialization by `job_id`, deduplicating
+ /// concurrent first-time misses to a single [`BeginOutcome::Claimed`].
+ async fn begin_materialization(
+ &self,
+ key: CacheKey,
+ job_id: String,
Review Comment:
The rest of the scheduler state uses the `JobId` newtype from #1852,
including `JobState` and `pinned_stages` here. Could `begin_materialization`,
`complete_materialization`, `abort_materialization`, and
`MaterializedCache::materializing_job` take `JobId` too? It's a quick change
now, but it would be an API break once the trait ships.
##########
ballista/scheduler/src/state/cache_registry.rs:
##########
@@ -0,0 +1,748 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Cross-job registry of materialized `DataFrame::cache()` results.
+//!
+//! A cached dataset is represented in the logical plan by a
+//! [`BallistaCacheNode`](ballista_core::extension::BallistaCacheNode). When it
+//! is first executed it is materialized as *pinned* shuffle output on the
+//! executors (shuffle files that survive normal job cleanup). This registry
+//! records, per cache key, the schema and the shuffle
+//! [`PartitionLocation`](ballista_core::serde::scheduler::PartitionLocation)s
of
+//! that output so that a later job can read the data back over the existing
+//! shuffle path instead of recomputing the subplan.
+//!
+//! Unlike a job's
[`ExecutionGraph`](crate::state::execution_graph::ExecutionGraph),
+//! which is garbage-collected shortly after the job completes, entries here
+//! outlive the job that produced them — that is the whole point of a cache.
+//!
+//! # Keying
+//!
+//! Entries are keyed by [`CacheKey`](crate::state::cache_registry::CacheKey) =
+//! `(session_id, cache_id)`, where `cache_id` is the per-`cache()`-call UUID
+//! stamped by `BallistaCacheFactory`. This gives reuse whenever the client
+//! holds on to the cached `DataFrame` (its logical plan keeps the same
+//! `cache_id`). Canonical plan-based keying for cross-call deduplication is
+//! intentionally deferred.
+//!
+//! # Lifecycle
+//!
+//! ```text
+//! (absent) --begin_materialization--------------------> Pending{job}
+//! Pending{job} --complete_materialization(job) [match]----> Materialized
+//! Pending{job} --abort_materialization(job) [match]----> (absent)
+//! Pending{job} --complete_materialization(other|absent)---> Stale (no
change)
+//! Materialized --invalidate / invalidate_executor / remove_session-->
(absent)
+//! ```
+//!
+//! Transitions out of `Pending` are **job-id matched**: a late completion
from a
+//! job that no longer owns the claim (because the entry was invalidated and
+//! re-claimed, or the session was removed) is reported
+//!
[`CompleteOutcome::Stale`](crate::state::cache_registry::CompleteOutcome::Stale)
+//! and changes nothing, so stale output cannot resurrect a removed entry or
+//! overwrite a newer claim. Invalidation simply removes the entry; the next
+//! lookup is therefore a miss and the subplan is re-materialized. This
+//! implements the "if an executor holding a cached partition is lost, the
cache
+//! is invalidated" policy.
+
+use std::collections::HashSet;
+
+use ballista_core::JobId;
+use ballista_core::serde::scheduler::PartitionLocation;
+use dashmap::DashMap;
+use datafusion::arrow::datatypes::SchemaRef;
+
+/// Identifies a cached dataset.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CacheKey {
+ /// The owning session.
+ pub session_id: String,
+ /// The per-`cache()`-call identifier carried by `BallistaCacheNode`.
+ pub cache_id: String,
+}
+
+impl CacheKey {
+ /// Creates a new cache key.
+ pub fn new(session_id: impl Into<String>, cache_id: impl Into<String>) ->
Self {
+ Self {
+ session_id: session_id.into(),
+ cache_id: cache_id.into(),
+ }
+ }
+}
+
+/// The materialized output of a cache: its schema and the shuffle locations of
+/// each output partition.
+#[derive(Debug, Clone)]
+pub struct MaterializedCache {
+ /// Schema of the cached data.
+ pub schema: SchemaRef,
+ /// Shuffle locations indexed by output partition. The inner `Vec` allows a
+ /// single output partition to be backed by more than one file (e.g. one
per
+ /// map task), mirroring [`ExecutionStage`] shuffle output.
+ ///
+ /// [`ExecutionStage`]: crate::state::execution_stage
+ pub locations: Vec<Vec<PartitionLocation>>,
+ /// The job that materialized this cache. Used to pin/unpin its shuffle
data.
+ pub materializing_job: String,
+}
+
+/// Internal state of a cache entry.
+#[derive(Debug, Clone)]
+enum EntryState {
+ /// A materialization job has been submitted but has not completed; the
+ /// partition locations are not yet known.
+ Pending { materializing_job: String },
+ /// The cache is fully materialized and readable.
+ Materialized(MaterializedCache),
+}
+
+/// Cross-job registry of materialized cache entries.
+///
+/// Cheap to clone-share via `Arc`; all methods take `&self` and are safe to
call
+/// concurrently. Kept `pub(crate)`: the public surface is the
+/// [`CacheState`](crate::cluster::CacheState) trait, so this type can evolve
+/// freely while the feature grows.
+#[derive(Debug, Default)]
+pub(crate) struct CacheRegistry {
+ entries: DashMap<CacheKey, EntryState>,
+}
+
+/// Outcome of claiming a key for materialization
+///
([`CacheState::begin_materialization`](crate::cluster::CacheState::begin_materialization)).
+#[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum BeginOutcome {
+ /// This caller claimed the key and should submit the materialization job.
+ Claimed,
+ /// A materialization job is already in flight for this key; do not submit.
+ AlreadyPending,
+ /// The key is already materialized; the caller should read it back
instead.
+ AlreadyMaterialized,
+}
+
+/// Outcome of recording a materialization
+///
([`CacheState::complete_materialization`](crate::cluster::CacheState::complete_materialization)).
+#[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum CompleteOutcome {
+ /// The pending claim was promoted to materialized.
+ Completed,
+ /// There was no matching pending claim (the entry was removed while
+ /// materializing, is already materialized, or is owned by a different
job).
+ /// Nothing was recorded; the caller should clean up the shuffle output it
+ /// produced, as it will never be read.
+ Stale,
+}
+
+/// Returns true if `state` is a materialized entry holding a partition on
+/// `executor_id`.
+fn holds_executor(state: &EntryState, executor_id: &str) -> bool {
+ matches!(
+ state,
+ EntryState::Materialized(cache)
+ if cache
+ .locations
+ .iter()
+ .flatten()
+ .any(|loc| loc.executor_meta.id == executor_id)
+ )
+}
+
+impl CacheRegistry {
+ /// Creates an empty registry.
+ #[cfg(test)]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Returns the materialized data for `key`, or `None` if the key is absent
+ /// (a miss) or still pending.
+ pub fn lookup(&self, key: &CacheKey) -> Option<MaterializedCache> {
+ match self.entries.get(key)?.value() {
+ EntryState::Materialized(cache) => Some(cache.clone()),
+ EntryState::Pending { .. } => None,
+ }
+ }
+
+ /// Attempts to claim `key` for materialization by `job_id`.
+ ///
+ /// Returns [`BeginOutcome::Claimed`] only when the caller is responsible
for
+ /// submitting the materialization job. If another job already claimed the
+ /// key, or the key is already materialized, the caller must not submit a
+ /// duplicate. This races safely: concurrent callers for the same key see
+ /// exactly one `Claimed`.
+ pub fn begin_materialization(
+ &self,
+ key: CacheKey,
+ job_id: impl Into<String>,
+ ) -> BeginOutcome {
+ use dashmap::mapref::entry::Entry;
+ match self.entries.entry(key) {
+ Entry::Occupied(occupied) => match occupied.get() {
+ EntryState::Pending { .. } => BeginOutcome::AlreadyPending,
+ EntryState::Materialized(_) =>
BeginOutcome::AlreadyMaterialized,
+ },
+ Entry::Vacant(vacant) => {
+ vacant.insert(EntryState::Pending {
+ materializing_job: job_id.into(),
+ });
+ BeginOutcome::Claimed
+ }
+ }
+ }
+
+ /// Promotes the pending claim for `key` to [`EntryState::Materialized`],
+ /// recording its shuffle output — **only if** `key` is still `Pending` and
+ /// owned by `job_id`. Any other state (absent, already materialized, or a
+ /// different owning job) is reported [`CompleteOutcome::Stale`] and the
entry
+ /// is left untouched. The check and swap are atomic under the entry lock,
so
+ /// a late completion cannot resurrect a removed entry or overwrite a newer
+ /// claim.
+ ///
+ /// Called by the scheduler when a materialization job succeeds (future PR,
+ /// from the `JobFinished` handler); on `Stale` the caller cleans up the
+ /// orphaned shuffle output it produced.
+ pub fn complete_materialization(
+ &self,
+ key: CacheKey,
+ job_id: impl Into<String>,
+ schema: SchemaRef,
+ locations: Vec<Vec<PartitionLocation>>,
+ ) -> CompleteOutcome {
+ use dashmap::mapref::entry::Entry;
+ let job_id = job_id.into();
+ match self.entries.entry(key) {
+ Entry::Occupied(mut occupied) => {
+ let owns_claim = matches!(
+ occupied.get(),
+ EntryState::Pending { materializing_job } if
*materializing_job == job_id
+ );
+ if owns_claim {
+ occupied.insert(EntryState::Materialized(MaterializedCache
{
+ schema,
+ locations,
+ materializing_job: job_id,
+ }));
+ CompleteOutcome::Completed
+ } else {
+ CompleteOutcome::Stale
Review Comment:
Small one on `Stale`. It also covers an entry that this same job already
completed, and the `CompleteOutcome::Stale` docs tell the caller to clean up
its output in that case. But that output is exactly what the entry points at.
So if a completion ever gets delivered twice, the second call would delete the
data behind a live entry. Could a `Materialized` entry with a matching
`materializing_job` return `Completed` (or a new `AlreadyCompleted`)? Then
`Stale` would always mean the output is safe to delete.
##########
ballista/scheduler/src/state/cache_registry.rs:
##########
@@ -0,0 +1,748 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Cross-job registry of materialized `DataFrame::cache()` results.
+//!
+//! A cached dataset is represented in the logical plan by a
+//! [`BallistaCacheNode`](ballista_core::extension::BallistaCacheNode). When it
+//! is first executed it is materialized as *pinned* shuffle output on the
+//! executors (shuffle files that survive normal job cleanup). This registry
+//! records, per cache key, the schema and the shuffle
+//! [`PartitionLocation`](ballista_core::serde::scheduler::PartitionLocation)s
of
+//! that output so that a later job can read the data back over the existing
+//! shuffle path instead of recomputing the subplan.
+//!
+//! Unlike a job's
[`ExecutionGraph`](crate::state::execution_graph::ExecutionGraph),
+//! which is garbage-collected shortly after the job completes, entries here
+//! outlive the job that produced them — that is the whole point of a cache.
+//!
+//! # Keying
+//!
+//! Entries are keyed by [`CacheKey`](crate::state::cache_registry::CacheKey) =
+//! `(session_id, cache_id)`, where `cache_id` is the per-`cache()`-call UUID
+//! stamped by `BallistaCacheFactory`. This gives reuse whenever the client
+//! holds on to the cached `DataFrame` (its logical plan keeps the same
+//! `cache_id`). Canonical plan-based keying for cross-call deduplication is
+//! intentionally deferred.
+//!
+//! # Lifecycle
+//!
+//! ```text
+//! (absent) --begin_materialization--------------------> Pending{job}
+//! Pending{job} --complete_materialization(job) [match]----> Materialized
+//! Pending{job} --abort_materialization(job) [match]----> (absent)
+//! Pending{job} --complete_materialization(other|absent)---> Stale (no
change)
+//! Materialized --invalidate / invalidate_executor / remove_session-->
(absent)
+//! ```
+//!
+//! Transitions out of `Pending` are **job-id matched**: a late completion
from a
+//! job that no longer owns the claim (because the entry was invalidated and
+//! re-claimed, or the session was removed) is reported
+//!
[`CompleteOutcome::Stale`](crate::state::cache_registry::CompleteOutcome::Stale)
+//! and changes nothing, so stale output cannot resurrect a removed entry or
+//! overwrite a newer claim. Invalidation simply removes the entry; the next
+//! lookup is therefore a miss and the subplan is re-materialized. This
+//! implements the "if an executor holding a cached partition is lost, the
cache
+//! is invalidated" policy.
+
+use std::collections::HashSet;
+
+use ballista_core::JobId;
+use ballista_core::serde::scheduler::PartitionLocation;
+use dashmap::DashMap;
+use datafusion::arrow::datatypes::SchemaRef;
+
+/// Identifies a cached dataset.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CacheKey {
+ /// The owning session.
+ pub session_id: String,
+ /// The per-`cache()`-call identifier carried by `BallistaCacheNode`.
+ pub cache_id: String,
+}
+
+impl CacheKey {
+ /// Creates a new cache key.
+ pub fn new(session_id: impl Into<String>, cache_id: impl Into<String>) ->
Self {
+ Self {
+ session_id: session_id.into(),
+ cache_id: cache_id.into(),
+ }
+ }
+}
+
+/// The materialized output of a cache: its schema and the shuffle locations of
+/// each output partition.
+#[derive(Debug, Clone)]
+pub struct MaterializedCache {
+ /// Schema of the cached data.
+ pub schema: SchemaRef,
+ /// Shuffle locations indexed by output partition. The inner `Vec` allows a
+ /// single output partition to be backed by more than one file (e.g. one
per
+ /// map task), mirroring [`ExecutionStage`] shuffle output.
+ ///
+ /// [`ExecutionStage`]: crate::state::execution_stage
+ pub locations: Vec<Vec<PartitionLocation>>,
+ /// The job that materialized this cache. Used to pin/unpin its shuffle
data.
+ pub materializing_job: String,
+}
+
+/// Internal state of a cache entry.
+#[derive(Debug, Clone)]
+enum EntryState {
+ /// A materialization job has been submitted but has not completed; the
+ /// partition locations are not yet known.
+ Pending { materializing_job: String },
+ /// The cache is fully materialized and readable.
+ Materialized(MaterializedCache),
+}
+
+/// Cross-job registry of materialized cache entries.
+///
+/// Cheap to clone-share via `Arc`; all methods take `&self` and are safe to
call
+/// concurrently. Kept `pub(crate)`: the public surface is the
+/// [`CacheState`](crate::cluster::CacheState) trait, so this type can evolve
+/// freely while the feature grows.
+#[derive(Debug, Default)]
+pub(crate) struct CacheRegistry {
+ entries: DashMap<CacheKey, EntryState>,
+}
+
+/// Outcome of claiming a key for materialization
+///
([`CacheState::begin_materialization`](crate::cluster::CacheState::begin_materialization)).
+#[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum BeginOutcome {
+ /// This caller claimed the key and should submit the materialization job.
+ Claimed,
+ /// A materialization job is already in flight for this key; do not submit.
+ AlreadyPending,
+ /// The key is already materialized; the caller should read it back
instead.
+ AlreadyMaterialized,
+}
+
+/// Outcome of recording a materialization
+///
([`CacheState::complete_materialization`](crate::cluster::CacheState::complete_materialization)).
+#[derive(Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum CompleteOutcome {
+ /// The pending claim was promoted to materialized.
+ Completed,
+ /// There was no matching pending claim (the entry was removed while
+ /// materializing, is already materialized, or is owned by a different
job).
+ /// Nothing was recorded; the caller should clean up the shuffle output it
+ /// produced, as it will never be read.
+ Stale,
+}
+
+/// Returns true if `state` is a materialized entry holding a partition on
+/// `executor_id`.
+fn holds_executor(state: &EntryState, executor_id: &str) -> bool {
+ matches!(
+ state,
+ EntryState::Materialized(cache)
+ if cache
+ .locations
+ .iter()
+ .flatten()
+ .any(|loc| loc.executor_meta.id == executor_id)
+ )
+}
+
+impl CacheRegistry {
+ /// Creates an empty registry.
+ #[cfg(test)]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Returns the materialized data for `key`, or `None` if the key is absent
+ /// (a miss) or still pending.
+ pub fn lookup(&self, key: &CacheKey) -> Option<MaterializedCache> {
+ match self.entries.get(key)?.value() {
+ EntryState::Materialized(cache) => Some(cache.clone()),
+ EntryState::Pending { .. } => None,
+ }
+ }
+
+ /// Attempts to claim `key` for materialization by `job_id`.
+ ///
+ /// Returns [`BeginOutcome::Claimed`] only when the caller is responsible
for
+ /// submitting the materialization job. If another job already claimed the
+ /// key, or the key is already materialized, the caller must not submit a
+ /// duplicate. This races safely: concurrent callers for the same key see
+ /// exactly one `Claimed`.
+ pub fn begin_materialization(
+ &self,
+ key: CacheKey,
+ job_id: impl Into<String>,
+ ) -> BeginOutcome {
+ use dashmap::mapref::entry::Entry;
+ match self.entries.entry(key) {
+ Entry::Occupied(occupied) => match occupied.get() {
+ EntryState::Pending { .. } => BeginOutcome::AlreadyPending,
+ EntryState::Materialized(_) =>
BeginOutcome::AlreadyMaterialized,
+ },
+ Entry::Vacant(vacant) => {
+ vacant.insert(EntryState::Pending {
+ materializing_job: job_id.into(),
+ });
+ BeginOutcome::Claimed
+ }
+ }
+ }
+
+ /// Promotes the pending claim for `key` to [`EntryState::Materialized`],
+ /// recording its shuffle output — **only if** `key` is still `Pending` and
+ /// owned by `job_id`. Any other state (absent, already materialized, or a
+ /// different owning job) is reported [`CompleteOutcome::Stale`] and the
entry
+ /// is left untouched. The check and swap are atomic under the entry lock,
so
+ /// a late completion cannot resurrect a removed entry or overwrite a newer
+ /// claim.
+ ///
+ /// Called by the scheduler when a materialization job succeeds (future PR,
+ /// from the `JobFinished` handler); on `Stale` the caller cleans up the
+ /// orphaned shuffle output it produced.
+ pub fn complete_materialization(
+ &self,
+ key: CacheKey,
+ job_id: impl Into<String>,
+ schema: SchemaRef,
+ locations: Vec<Vec<PartitionLocation>>,
+ ) -> CompleteOutcome {
+ use dashmap::mapref::entry::Entry;
+ let job_id = job_id.into();
+ match self.entries.entry(key) {
+ Entry::Occupied(mut occupied) => {
+ let owns_claim = matches!(
+ occupied.get(),
+ EntryState::Pending { materializing_job } if
*materializing_job == job_id
+ );
+ if owns_claim {
+ occupied.insert(EntryState::Materialized(MaterializedCache
{
+ schema,
+ locations,
+ materializing_job: job_id,
+ }));
+ CompleteOutcome::Completed
+ } else {
+ CompleteOutcome::Stale
+ }
+ }
+ Entry::Vacant(_) => CompleteOutcome::Stale,
+ }
+ }
+
+ /// Drops the pending claim for `key` — but **only if** it is still
`Pending`
+ /// and owned by `job_id`. Returns whether a claim was removed.
+ ///
+ /// Called when a materialization job fails or is cancelled (future PR,
from
+ /// the `JobRunningFailed` handler) so a doomed claim does not stay
`Pending`
+ /// forever (which would make every `lookup` miss and every
+ /// `begin_materialization` return `AlreadyPending`). A mismatched `job_id`
+ /// (a newer claim re-took the key) is left untouched.
+ pub fn abort_materialization(&self, key: &CacheKey, job_id: &str) -> bool {
+ self.entries
+ .remove_if(key, |_, state| {
+ matches!(
+ state,
+ EntryState::Pending { materializing_job } if
materializing_job == job_id
+ )
+ })
+ .is_some()
+ }
+
+ /// Removes the entry for `key`. The next [`lookup`](Self::lookup) is a
miss.
+ /// Returns the removed entry's materialized form, if it was materialized,
so
+ /// the caller can unpin its shuffle data.
+ pub fn invalidate(&self, key: &CacheKey) -> Option<MaterializedCache> {
+ match self.entries.remove(key) {
+ Some((_, EntryState::Materialized(cache))) => Some(cache),
+ _ => None,
+ }
+ }
+
+ /// Invalidates every materialized entry that holds a partition on
+ /// `executor_id`. Returns the invalidated entries so the caller can unpin
+ /// their shuffle data. Pending entries are left untouched (their job will
+ /// fail/retry through the normal task-failure path).
Review Comment:
What happens if an executor is lost after the cached stage finishes but
before the job does? `reset_stages_on_lost_executor` only reruns a successful
stage when an unfinished stage still reads from it. So once everything
downstream of the cached stage is done, the job can succeed without rerunning
it.
```text
begin_materialization(k, "j1") // Claimed
// stage 2 (the cached subplan) writes to e1 and e2
// stage 3 reads it and succeeds, stage 4 is still running
invalidate_executor("e1") // skips k, it's still
Pending
// stage 2 isn't rerun, nothing unfinished reads it
complete_materialization(k, "j1", schema, locs) // Completed, locs include
e1
lookup(k) // a hit that points at e1
```
One option is for `invalidate_executor` to record the lost executor on
pending entries. Then `complete_materialization` could drop the entry and
return `Stale` when any location is on one of them, which keeps the check
atomic inside the registry. Happy for the fix to land with the executor-loss
follow-up. In the meantime it might be worth updating this doc so the next PR
doesn't lean on the task-failure path.
--
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]
