Clean up job data on both Scheduler and Executor

[mingmwang]

Which issue does this PR close?

Closes #9 and #185.

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

[mingmwang]

@thinkharderdev @yahoNanJing @Ted-Jiang @andygrove

[Ted-Jiang]

I think this will lock the whole 🤔 in standalone

 async fn lock(&self, keyspace: Keyspace, key: &str) -> Result<Box<dyn Lock>> {
        let mut mlock = self.locks.lock().await;
        let lock_key = format!("/{:?}/{}", keyspace, key);
        if let Some(lock) = mlock.get(&lock_key) {
            Ok(Box::new(lock.clone().lock_owned().await))
        } else {
            let new_lock = Arc::new(Mutex::new(()));
            mlock.insert(lock_key, new_lock.clone());
            Ok(Box::new(new_lock.lock_owned().await))
        }
    }

[Ted-Jiang]

Oh, this lock only works in func async fn lock(&self, keyspace: Keyspace, key: &str), will release after this func quick call.

[Ted-Jiang]

I think each SQL sends an RPC to all executors is not a good idea (for interactive query finish in ms)🤔

[Ted-Jiang]

Maybe we can do some improvement later !

[mingmwang]

Yes, in current code base, the TaskManager/ExecutionGraph do not track the tasks are executed by which executors.

[Ted-Jiang]

(for interactive query finish in ms) i think this will store a lot feature in heap. Just some opinion.

[yahoNanJing]

Nice refinement. It's better to have only one entrance to modify the scheduler state and all of the state changes should have a related event and be dealt with in the event loop.

[andygrove]

I am constantly filling my disk up with shuffle files so would love to see us get this merged before the 0.9.0 release.

@mingmwang Could you rebase when you get a chance and I will test this out and review the PR as well.

[mingmwang]

I am constantly filling my disk up with shuffle files so would love to see us get this merged before the 0.9.0 release.

@mingmwang Could you rebase when you get a chance and I will test this out and review the PR as well.

Sure, working on it.

[mingmwang]

@andygrove @yahoNanJing @Ted-Jiang @yahoNanJing

BTW, in this PR, the job data in the state store will also be deleted after 300s.
I think we need a following PR to move the completed(Success or Failed) job data from state store to ObjectStore
for long time storing purpose, and Scheduler UI can read from the ObjectStore.

Please share your thoughts.

const CLEANUP_FINISHED_JOB_DELAY_SECS: u64 = 300;

async fn clean_up_job_data(
        state: Arc<dyn StateBackendClient>,
        active_job_cache: ExecutionGraphCache,
        failed: bool,
        job_id: String,
        executor_manager: Option<ExecutorManager>,
    ) -> Result<()> {
        let mut active_graph_cache = active_job_cache.write().await;
        active_graph_cache.remove(&job_id);

        let keyspace = if failed {
            Keyspace::FailedJobs
        } else {
            Keyspace::CompletedJobs
        };

        let lock = state.lock(keyspace.clone(), "").await?;
        with_lock(lock, state.delete(keyspace, &job_id)).await?;

        executor_manager
            .map(|em| async { Self::clean_up_executors_data(job_id.clone(), em).await });
        Ok(())
    }

[andygrove]

We should make this configurable. Some of the queries I am testing take much longer than 300 seconds. We already have the ability to set configs on the context.

[andygrove]

Never mind, this is a delay after the job completes. I would still like to see this configurable but we could do that as a follow in PR.

[andygrove]

LGTM. Thanks @mingmwang

[andygrove]

@mingmwang could you fix the conflicts here when you have the time so that we can merge this?

[mingmwang]

@mingmwang could you fix the conflicts here when you have the time so that we can merge this?

Sure, I will fix the conflicts tomorrow.

[mingmwang]

Resolved conflicts.

[andygrove]

Thanks again @mingmwang