feat: Create new proxy service to store (and relay) broadcasted daft data

Cargo.toml

@@ -17,6 +17,7 @@ daft-compression = {path = "src/daft-compression", default-features = false}
 daft-connect = {path = "src/daft-connect", optional = true}
 daft-core = {path = "src/daft-core", default-features = false}
 daft-csv = {path = "src/daft-csv", default-features = false}
+daft-dashboard-server = {path = "src/daft-dashboard-server"}
 daft-dsl = {path = "src/daft-dsl", default-features = false}
 daft-functions = {path = "src/daft-functions"}
 daft-functions-json = {path = "src/daft-functions-json", default-features = false}
@@ -170,10 +171,10 @@ members = [
   "src/hyperloglog",
   "src/daft-connect",
   "src/parquet2",
-  # "src/spark-connect-script",
   "src/generated/spark-connect",
   "src/common/partitioning",
-  "src/daft-ray-execution"
+  "src/daft-ray-execution",
+  "src/daft-dashboard-server"
 ]
 
 [workspace.dependencies]
@@ -247,7 +248,8 @@ tokio = {version = "1.37.0", features = [
   "signal",
   "macros",
   "rt",
-  "rt-multi-thread"
+  "rt-multi-thread",
+  "sync"
 ]}
 tokio-stream = {version = "0.1.14", features = ["fs", "io-util", "time"]}
 tokio-util = "0.7.11"

daft/daft/__init__.pyi

@@ -1213,6 +1213,11 @@ def binary_length(expr: PyExpr) -> PyExpr: ...
 def binary_concat(left: PyExpr, right: PyExpr) -> PyExpr: ...
 def binary_slice(expr: PyExpr, start: PyExpr, length: PyExpr | None = None) -> PyExpr: ...
 
+# ---
+# dashboard namespace
+# ---
+def launch(): ...
+
 class PyCatalog:
     @staticmethod
     def new() -> PyCatalog: ...

daft/dashboard.py

@@ -0,0 +1,15 @@
+import os
+
+import daft.daft as native
+
+DAFT_DASHBOARD_ENV_NAME = "DAFT_DASHBOARD"
+DAFT_DASHBOARD_ADDR = "http://localhost:3238"
+
+
+def launch():
+    """Launches the Daft dashboard server on port 3000.
+
+    The server serves HTML/CSS/JS bundles, so you are able to point your browser towards `http://localhost:3000` and view information regarding your queries.
+    """
+    os.environ[DAFT_DASHBOARD_ENV_NAME] = "1"
+    native.launch()

daft/dataframe/dataframe.py

@@ -11,6 +11,7 @@
 import typing
 import warnings
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from functools import partial, reduce
 from typing import (
     TYPE_CHECKING,
@@ -28,6 +29,7 @@
     TypeVar,
     Union,
 )
+from uuid import uuid4
 
 from daft.api_annotations import DataframePublicAPI
 from daft.context import get_context
@@ -158,6 +160,45 @@ def _result(self) -> Optional[PartitionSet]:
         else:
             return self._result_cache.value
 
+    def _explain_broadcast(self):
+        import json
+        from urllib import request
+        from urllib.error import URLError
+
+        from daft.dashboard import DAFT_DASHBOARD_ADDR
+        from daft.dataframe.display import MermaidFormatter
+
+        dashboard_addr = os.environ.get("DAFT_DASHBOARD")
+        if not dashboard_addr:
+            return
+        elif not int(dashboard_addr):
+            return
+
+        is_cached = self._result_cache is not None
+        plan_time_start = datetime.now(timezone.utc)
+        mermaid_plan = MermaidFormatter(
+            builder=self.__builder, show_all=True, simple=False, is_cached=is_cached
+        )._repr_markdown_()
+        plan_time_end = datetime.now(timezone.utc)
+
+        headers = {
+            "Content-Type": "application/json",
+        }
+        data = json.dumps(
+            {
+                "id": str(uuid4()),
+                "mermaid-plan": mermaid_plan,
+                "plan-time-start": str(plan_time_start),
+                "plan-time-end": str(plan_time_end),
+            }
+        ).encode("utf-8")
+        req = request.Request(DAFT_DASHBOARD_ADDR, headers=headers, data=data)
+
+        try:
+            request.urlopen(req, timeout=1)
+        except URLError as e:
+            warnings.warn(f"Failed to broadcast metrics over {DAFT_DASHBOARD_ADDR}: {e}")
+
     @DataframePublicAPI
     def explain(
         self, show_all: bool = False, format: str = "ascii", simple: bool = False, file: Optional[io.IOBase] = None
@@ -2818,6 +2859,7 @@ def collect(self, num_preview_rows: Optional[int] = 8) -> "DataFrame":
             DataFrame: DataFrame with materialized results.
         """
         self._materialize_results()
+        self._explain_broadcast()
 
         assert self._result is not None
         dataframe_len = len(self._result)
@@ -2889,6 +2931,8 @@ def show(self, n: int = 8) -> None:
             n: number of rows to show. Defaults to 8.
         """
         dataframe_display = self._construct_show_display(n)
+        self._explain_broadcast()
+
         try:
             from IPython.display import display
 

src/daft-dashboard-server/Cargo.toml

@@ -0,0 +1,22 @@
+[dependencies]
+anyhow = "1.0"
+chrono = {workspace = true, features = ["serde"]}
+fork = "0.2"
+http-body-util = "0.1"
+hyper = {features = ["full"], version = "1.6"}
+hyper-util = {features = ["full"], version = "0.1"}
+parking_lot = "0.12"
+serde_json = "1.0"
+simdutf8 = "0.1"
+futures.workspace = true
+pyo3.workspace = true
+serde.workspace = true
+tokio.workspace = true
+
+[lints]
+workspace = true
+
+[package]
+name = "daft-dashboard-server"
+edition.workspace = true
+version.workspace = true