feat: Read log signal from DB

master/internal/api_experiment.go

@@ -2706,6 +2706,7 @@ func (a *apiServer) SearchExperiments(
 		Column("searcher_metric_value").
 		Column("trials.external_trial_id").
 		ColumnExpr("nullif(trials.metadata, 'null') as metadata").
+		ColumnExpr("NULL as log_signals").
 		Join("LEFT JOIN validations bv ON trials.best_validation_id = bv.id").
 		Join("LEFT JOIN validations lv ON trials.latest_validation_id = lv.id").
 		Join("LEFT JOIN checkpoints_v2 new_ckpt ON new_ckpt.id = trials.warm_start_checkpoint_id").

master/internal/api_runs.go

@@ -170,11 +170,14 @@ func getRunsColumns(q *bun.SelectQuery) *bun.SelectQuery {
 			'pachyderm_integration', NULLIF(e.config#>'{integrations,pachyderm}', 'null'),
 			'id', e.id) AS experiment`).
 		ColumnExpr("rm.metadata AS metadata").
+		ColumnExpr("ta.log_signal AS log_signal").
 		Join("LEFT JOIN experiments AS e ON r.experiment_id=e.id").
 		Join("LEFT JOIN runs_metadata AS rm ON r.id=rm.run_id").
 		Join("LEFT JOIN users u ON e.owner_id = u.id").
 		Join("LEFT JOIN projects p ON r.project_id = p.id").
-		Join("LEFT JOIN workspaces w ON p.workspace_id = w.id")
+		Join("LEFT JOIN workspaces w ON p.workspace_id = w.id").
+		Join("LEFT JOIN run_id_task_id rt ON r.id=rt.run_id").
+		Join("LEFT JOIN tasks ta ON ta.task_id=rt.task_id")
 }
 
 func sortRuns(sortString *string, runQuery *bun.SelectQuery) error {

master/static/migrations/20240918161807_add-log-signal.tx.up.sql

@@ -0,0 +1,2 @@
+ALTER table public.tasks
+    ADD COLUMN log_signal text;

master/static/srv/proto_get_trials_plus.sql

@@ -128,6 +128,13 @@ SELECT
       WHERE tt.run_id = t.id
       ORDER BY ta.start_time
   ) sub_tasks)) AS task_ids,
+  (
+    (SELECT json_agg(log_signal) FILTER (WHERE log_signal IS NOT NULL) FROM (
+      SELECT ta.log_signal FROM run_id_task_id tt
+      JOIN tasks ta ON tt.task_id = ta.task_id
+      WHERE tt.run_id = t.id
+      ORDER BY ta.start_time
+  ) log_signals)) AS log_signals,
   t.checkpoint_size AS total_checkpoint_size,
   t.checkpoint_count,
   t.total_batches AS total_batches_processed,