Add support for continuously starting load jobs as slots free up in the loader

[sh-rp]

Description

In the current implementation we more or less always start n (=max workers) load jobs, let them complete and then rerun the whole loader to schedule the next n jobs. In this PR we submit new load jobs as slots free up.

What is happening here:

• The loader now periodically checks all jobs wether they are done and schedules new ones as needed.
• Runnable Jobs now manage their own internal state (file moving still needs to be done by the loader on the main thread) and have a dedicated "run" method which is the one called on a thread.
• Renaming of the Job base classes to make clearer what is going on:
  • RunnableLoadJob: Jobs that actually do something and should be executed on a thread
  • FollowupJob: Class that creates a new job persisted to the disc
  • HasFollowupJobs (ex-FollowupJob): Trait that tells loader to look for followup Jobs
  • FinalizedJob: Not runnable because it already has an actionable state (completed, failed, retry). Used for indicating failed restored jobs, completed restored jobs and cases where nothing needs to be done
• FollowupJobs always go to the "new_jobs" folder and need to be picked up by the loader as any other jobs do, there were some just directly executed in the main thread: not good! :) We now assume that creation of FollowupJobs does not yield exceptions, I don't think this is needed and it makes the code simpler.
• (Hopefully) simplification of the load class
• Restoring jobs is now handled the same way as creating jobs is, this simplifies the code in a few places. Jobs that are found in the running folder now are simply started again, the bigquery job can figure out wether it needs to be resumed on its own.

Possible FollowupWork:

• Performance Tuning in loader, a lot of time is spent on parsing filenames (I think).

[netlify]

✅ Deploy Preview for dlt-hub-docs canceled.

Name	Link
🔨 Latest commit	2c38f13
🔍 Latest deploy log	https://app.netlify.com/sites/dlt-hub-docs/deploys/66acccc36859290008ab7658

[sh-rp]

is this the correct "package completion" condition? I think so, but am not 100% sure.

[rudolfix]

this was the previous condition:

        if file_count == 0:
            logger.info(f"No new jobs found in {load_id}")
            return 0, []

so checking just new jobs. I do not fully get why you do it here again? the loop above should exit only when all jobs are completed, no?

[sh-rp]

I needed to change a bunch of tests, since we do not rely on multiple executions of the run method anymore. All the changes make sense, it might be good to add a few more tests cases specific to the new implementation.

[sh-rp]

I need to collect the exception to raise in the main loop here now, we could alternatively collect all problems we find and print them out, not just raise on exception.

[sh-rp]

@rudolfix this can go into another review. Two open questions from my side:

1. Should we change the behavior of def is_package_partially_loaded (see above)
2. What exactly is the desired behavior from a conceptual standpoint when creating a followup job fails? In the old version, loading just continues and those jobs are marked as failed. I don't think this makes sense, because the load will be useless if for example a mergejob can't be created and executed. So either we decide

• this is a transient problem (e.g. the user can manually fix the schema and restart he load) , in that case we just raise an exception on the main thread, so that the job that triggered the scheduling of a followupjob remains in "started_jobs" and will be rerun on the pipeline execution including the scheduling of the followupjob (it is implemented and tested like this now)
• or this is a terminal problem, in which case we should also stop the load but mark the loadpackage as failed

[rudolfix]

What exactly is the desired behavior from a conceptual standpoint when creating a followup job fails? In the old version, loading just continues and those jobs are marked as failed. I don't think this makes sense, because the load will be useless if for example a mergejob can't be created and executed. So either we decide
this is a transient problem (e.g. the user can manually fix the schema and restart he load) , in that case we just raise an exception on the main thread, so that the job that triggered the scheduling of a followupjob remains in "started_jobs" and will be rerun on the pipeline execution including the scheduling of the followupjob (it is implemented and tested like this now)
or this is a terminal problem, in which case we should also stop the load but mark the loadpackage as failed

my take:
make it a transient error

and we need how we deal with failed packages. right now we continue load and do not raise exception at the end. IMO we should change that. we should continue load but raise exception at the end automatically.

I'll write a ticket for that - it ia a breaking change

[rudolfix]

We need to change tagging stuff:

1. we tag session in:

def create_load_job(
        self, table: TTableSchema, file_path: str, load_id: str, restore: bool = False
    ) -> LoadJob:
        """Starts SqlLoadJob for files ending with .sql or returns None to let derived classes to handle their specific jobs"""
        self._set_query_tags_for_job(load_id, table)

which is called from main thread and not supposed to open any connection. IDK how it works now :) but even if it does, it will tag a session on main thread and then we immediately close the connection and the reopen it on the worker thread but in that case tagging does not happen.

2. Since you passed all required params in set_vars, we do not need to take any parameters in here:

def _set_query_tags_for_job(self, load_id: str, table: TTableSchema) -> None:

3. this function should be called in run method of all jobs that have sql_client (created by sql_job_client). so we need to move it from client to the job... which is a good move

[rudolfix]

see the above. are we still reading any job listings when looping on idle?

[rudolfix]

I think we need to change how is_package_partially_loaded works. partially loaded -> has packages that are not completed and has packages that are completed.

current implementation assumes that failed packages, started packages and retried packages modify the destination. this is IMO wrong (we assume that jobs are atomic - most of them are). Now I think that was wrong from the start
WDYT?

[rudolfix]

OK! I probably overlooked the lines below!

[rudolfix]

OK. but this idle loop where we sleep 0.1 seconds worries. do we have 100% cpu usage when this test run? maybe it is a little bit faster but we saturate CPU (while having some threads working). pls take a look. with 50k or 100k jobs maybe.

we must avoid starving threads by an idle loop that reads 50k files over and over

[rudolfix]

@property
    def has_pending_data(self) -> bool:
        """Tells if the pipeline contains any extracted files or pending load packages"""
        return (
            len(self.list_normalized_load_packages()) > 0
            or len(self.list_extracted_load_packages()) > 0
        )

it does not even look into package content. any not completed package is pending and will be executed before new package is created. this check is main reason this function exist

I answered is_package_partially_loaded question above

[sh-rp]

my take: make it a transient error

and we need how we deal with failed packages. right now we continue load and do not raise exception at the end. IMO we should change that. we should continue load but raise exception at the end automatically.

I'll write a ticket for that - it ia a breaking change

Ok, I changed it slightly and added new exceptions to indicate what went wrong plus tests. With regards to the load package with failing jobs: I totally agree that that should raise at the end of the load if there were failed jobs. Now these errors are pretty much hidden.

[sh-rp]

the behavior is unified now between the different package states, I'd say this is correct.

[rudolfix]

LGTM!