akshaychitneni commented on code in PR #2730:
URL:
https://github.com/apache/datafusion-ballista/pull/2730#discussion_r4542220361
##########
ballista/scheduler/src/state/execution_graph.rs:
##########
@@ -893,69 +935,96 @@ impl ExecutionGraph for StaticExecutionGraph {
.entry(stage_id)
.or_default();
failed_attempts.insert(task_stage_attempt_num);
+ let map_stage_id =
+ fetch_partiton_error.map_stage_id as
usize;
+ let map_partition_id =
+ fetch_partiton_error.map_partition_id
as usize;
+ let executor_id =
fetch_partiton_error.executor_id;
+ // A cache's build stage id can match one
of this
+ // job's: the cache inputs the failure
names, and
+ // whether an own input on that executor
does too.
+ let lost_cache = external_fetch_failures(
+ &running_stage.inputs,
+ &executor_id,
+ map_stage_id,
+ map_partition_id,
+ );
+ let lost_own = lost_cache.is_empty()
Review Comment:
Fixed: a failure counts as own-output loss only if that input still has data
on the executor, so a repeat failure just rolls the stage back; covered by
`a_repeat_fetch_failure_for_lost_cache_data_only_rolls_back`.
##########
ballista/scheduler/src/state/cache_linker.rs:
##########
@@ -734,6 +1005,16 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
}
}
+/// Whether a job with `reads` reads cache `key` more than once: directly, or
+/// through another cache built from it.
+fn read_more_than_once(reads: &[ResolvedRead], key: &CacheKey) -> bool {
Review Comment:
Fixed: a lost read is rebuilt only if it shares no cache, nested ones
included, with another read, checked for every input before anything is
touched; covered by `a_lost_read_is_rebuilt_only_if_it_shares_no_cache`.
##########
ballista/scheduler/src/state/cache_linker.rs:
##########
@@ -98,16 +98,59 @@ enum CacheClaim {
/// A concurrent submission can change a key between lookup and claim.
const CLAIM_ATTEMPTS: usize = 4;
+/// What waits on a build job.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+enum Waiter {
+ /// A held job.
+ Held(JobId),
+ /// A running job's lost cache read, by external stage id.
+ Input(JobId, usize),
+}
+
+/// What a build job's end releases, or fails.
+#[derive(Default)]
+pub(crate) struct Released {
+ /// Held jobs.
+ pub(crate) jobs: Vec<PendingJob>,
+ /// Running jobs' lost reads, by external stage id.
+ pub(crate) inputs: Vec<(JobId, usize)>,
+}
+
+/// A running job's cache reads, kept to rebuild a cache whose data it loses.
+#[derive(Clone)]
+pub(crate) struct JobReads {
+ pub(crate) ctx: Arc<SessionContext>,
+ pub(crate) queued_at: u64,
+ /// The submission's admission, for rebuilds; dropped when the job ends.
+ pub(crate) admission: Admission,
Review Comment:
The record no longer holds the session context or admission: a rebuild
creates its context from the job's frozen config via a new
`JobState::new_context` (defaulted; the in-memory backend touches no session)
and its own admission, and only reads that can be rebuilt keep their subplan.
--
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]
