ci: Record info for TPCH benchmarks (#3729)

For TPC-DS benchmarks, we record planning time, execution time, as well
as the query plan. Do the same for TPC-H benchmarks.

benchmarking/tpch/ray_job_runner.py

@@ -10,10 +10,13 @@
 
 import argparse
 import asyncio
+import json
 import os
 import pathlib
 import time
 import uuid
+from datetime import datetime
+from pathlib import Path
 from typing import Callable
 
 from ray.job_submission import JobStatus, JobSubmissionClient
@@ -90,4 +93,29 @@ def _get_df(table_name: str) -> daft.DataFrame:
     get_df = get_df_with_parquet_folder(args.parquet_folder)
     answer = getattr(answers, f"q{args.question_number}")
     daft_df = answer(get_df)
+
+    info_path = Path("/tmp") / "ray" / "session_latest" / "logs" / "info"
+    info_path.mkdir(parents=True, exist_ok=True)
+
+    explain_delta = None
+    with open(info_path / f"plan-{args.question_number}.txt", "w") as f:
+        explain_start = datetime.now()
+        daft_df.explain(show_all=True, file=f, format="mermaid")
+        explain_end = datetime.now()
+        explain_delta = explain_end - explain_start
+
+    execute_delta = None
+    execute_start = datetime.now()
     daft_df.collect()
+    execute_end = datetime.now()
+    execute_delta = execute_end - execute_start
+
+    with open(info_path / f"stats-{args.question_number}.txt", "w") as f:
+        stats = json.dumps(
+            {
+                "question": args.question_number,
+                "planning-time": str(explain_delta),
+                "execution-time": str(execute_delta),
+            }
+        )
+        f.write(stats)