[query|batch] collect partition results as they complete (#15316)

## Change Description

Modifies the query `ServiceBackend` to collect partititon results
incrementally, instead of waiting for the job group to complete. It does
this by fetching succeeded jobs whose `end_time` is greater than the
maximum from the previous batch of jobs.

The batch service front end needed the following modifications:

- Include the job's `end_time` in http responses via
`JobListEntryV1Alpha` (and consequently `GetJobResponseV1Alpha`)
- Include milliseconds in date time string format (previously truncated
to seconds).

The first has the performance implication of always joining with the
attempts table. I see no way around this without a more significant
change to enable this new query-on-batch capability. The
`JobEndTimeQuery` has been simplified as a consequence.

Driver memory usage has significantly reduced as a consequence of this
change. I've included before and after screenshots of the driver's cpu
and memory usage in a stress test of force-counting a range table with
10,000 rows and partitions. Before this change, the driver needed 21GB
to collect these results. This change reduces that to <2GB meaning it
can now run on a standard worker.


![image.png](https://app.graphite.com/user-attachments/assets/fdf21f80-cb69-40f4-b60d-a9b32eb1a8bb.png)


![image.png](https://app.graphite.com/user-attachments/assets/89dfb8ad-23cc-4fad-85be-034cb5f95d18.png)

Through this testing, I discovered that `end_time` isn't a very good
index for incrementally collecting jobs. Due to the lack of ordering
guarantees on attempt `end_time`, I noticed a significant number of jobs
were never returned by the batch service while using a polling interval
of `[500, 5000]` ms. Doubling this interval reduced the number of
stragglers by half which is somewhat intuitive. I don't think increasing
futher is a good solution; instead we should consider other ways to get
jobs we haven't processed.

This change reduces the maximum number of parallel reads from 1000 to
50. 50 is the page size batch uses in job responses. After experimenting
a bit I found out that, for this test at least, increasing the
parallelism didn't decrease wall clock time, only memory consumption.

Fixes #15288, #14607

## Security Assessment

This change potentially impacts the Hail Batch instance as deployed by
Broad Institute in GCP

### Impact Rating

This change has a low security impact

### Impact Description

The changes primarily involve database query optimizations and internal
API enhancements for job result collection. The modifications add
time-based filtering capabilities and improve query efficiency without
introducing new external interfaces or changing
authentication/authorization mechanisms. All changes maintain existing
data validation and access control patterns.

### Appsec Review

- [ ] Required: The impact has been assessed and approved by appsec

Author: ehigham

batch/batch/batch.py

@@ -138,8 +138,11 @@ def job_record_to_dict(record: Dict[str, Any], name: Optional[str]) -> JobListEn
         exit_code = None
         duration = None
 
-    if record['cost_breakdown'] is not None:
-        record['cost_breakdown'] = cost_breakdown_to_dict(json.loads(record['cost_breakdown']))
+    cost_breakdown = record['cost_breakdown']
+    cost_breakdown = cost_breakdown_to_dict(json.loads(cost_breakdown)) if cost_breakdown is not None else None
+
+    end_time = record['end_time']
+    end_time = time_msecs_str(end_time) if end_time is not None else None
 
     return cast(
         JobListEntryV1Alpha,
@@ -151,10 +154,11 @@ def job_record_to_dict(record: Dict[str, Any], name: Optional[str]) -> JobListEn
             'billing_project': record['billing_project'],
             'state': record['state'],
             'exit_code': exit_code,
+            'end_time': end_time,
             'duration': duration,
             'cost': coalesce(record['cost'], 0),
             'msec_mcpu': record['msec_mcpu'],
-            'cost_breakdown': record['cost_breakdown'],
+            'cost_breakdown': cost_breakdown,
             'always_run': bool(record['always_run']),
             'display_state': None,
         },

batch/batch/front_end/front_end.py

@@ -2327,7 +2327,13 @@ async def _get_job(app, batch_id, job_id) -> GetJobResponseV1Alpha:
     record = await db.select_and_fetchone(
         """
 WITH base_t AS (
-SELECT jobs.*, user, billing_project, ip_address, format_version, t.attempt_id AS last_cancelled_attempt_id
+SELECT jobs.*
+     , user
+     , billing_project
+     , ip_address
+     , format_version
+     , t.attempt_id AS last_cancelled_attempt_id
+     , attempts.end_time
 FROM jobs
 INNER JOIN batches
   ON jobs.batch_id = batches.id

batch/batch/front_end/query/query.py

@@ -273,11 +273,7 @@ def __init__(self, operator: ComparisonOperator, time_msecs: int):
 
     def query(self) -> Tuple[str, List[int]]:
         op = self.operator.to_sql()
-        sql = f"""
-((jobs.batch_id, jobs.job_id) IN
- (SELECT batch_id, job_id FROM attempts
-  WHERE end_time {op} %s))
-"""
+        sql = f'(latest_attempt.end_time {op} %s)'
         return (sql, [self.time_msecs])
 
 

batch/batch/front_end/query/query_v1.py

@@ -267,11 +267,19 @@ def parse_job_group_jobs_query_v1(
     sql = f"""
 WITH base_t AS
 (
-  SELECT jobs.*, batches.user, batches.billing_project, batches.format_version,
-    job_attributes.value AS name
+  SELECT jobs.*
+       , batches.user
+       , batches.billing_project
+       , batches.format_version
+       , job_attributes.value AS name
+       , latest_attempt.end_time
   FROM jobs
   INNER JOIN batches ON jobs.batch_id = batches.id
   INNER JOIN batch_updates ON jobs.batch_id = batch_updates.batch_id AND jobs.update_id = batch_updates.update_id
+  LEFT JOIN attempts AS latest_attempt
+    ON  jobs.batch_id   = latest_attempt.batch_id
+    AND jobs.job_id     = latest_attempt.job_id
+    AND jobs.attempt_id = latest_attempt.attempt_id
   LEFT JOIN job_attributes
   ON jobs.batch_id = job_attributes.batch_id AND
     jobs.job_id = job_attributes.job_id AND

batch/batch/front_end/query/query_v2.py

@@ -277,15 +277,25 @@ def parse_job_group_jobs_query_v2(
         where_args += args
 
     sql = f"""
-SELECT STRAIGHT_JOIN jobs.*, batches.user, batches.billing_project, batches.format_version, job_attributes.value AS name, cost_t.cost,
-  cost_t.cost_breakdown
+SELECT STRAIGHT_JOIN jobs.*
+                   , batches.user
+                   , batches.billing_project
+                   , batches.format_version
+                   , job_attributes.value AS name
+                   , cost_t.cost
+                   , cost_t.cost_breakdown
+                   , latest_attempt.end_time
 FROM jobs
 INNER JOIN batches ON jobs.batch_id = batches.id
 INNER JOIN batch_updates ON jobs.batch_id = batch_updates.batch_id AND jobs.update_id = batch_updates.update_id
 LEFT JOIN job_attributes
   ON jobs.batch_id = job_attributes.batch_id AND
     jobs.job_id = job_attributes.job_id AND
     job_attributes.`key` = 'name'
+LEFT JOIN attempts AS latest_attempt
+    ON  jobs.batch_id   = latest_attempt.batch_id
+    AND jobs.job_id     = latest_attempt.job_id
+    AND jobs.attempt_id = latest_attempt.attempt_id
 LEFT JOIN LATERAL (
 SELECT COALESCE(SUM(`usage` * rate), 0) AS cost, JSON_OBJECTAGG(resources.resource, COALESCE(`usage` * rate, 0)) AS cost_breakdown
 FROM (SELECT resource_id, CAST(COALESCE(SUM(`usage`), 0) AS SIGNED) AS `usage`