feat: Broadcast plan over port 3238 #3756

raunakab · 2025-02-02T05:05:59Z

Overview

This PR adds the ability of broadcasting the serialized Mermaid query plan over port 3238.
Receiving services can listen to this TCP broadcast over 3238, read the data, and process/present it.

This will be useful for the dashboard, which will (through another proxy service) be listening over 3238 for the plan, so that it can display it to the end-user.

Notes

The number 3238 is daft spelled over the telephone's alphanumeric keypad.

codspeed-hq · 2025-02-02T05:14:52Z

CodSpeed Performance Report

Merging #3756 will improve performances by 10.31%

_{Comparing broadcast-metrics (45a4459) with main (86adc44)}

Summary

⚡ 1 improvements
✅ 26 untouched benchmarks

Benchmarks breakdown

	Benchmark	`BASE`	`HEAD`	Change
⚡	`test_iter_rows_first_row[100 Small Files]`	342.5 ms	310.5 ms	+10.31%

codecov · 2025-02-02T05:29:02Z

Codecov Report

Attention: Patch coverage is 60.00000% with 16 lines in your changes missing coverage. Please review.

Project coverage is 77.66%. Comparing base (6deb87e) to head (cc1d15d).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
daft/dataframe/dataframe.py	66.66%	10 Missing ⚠️
daft/context.py	40.00%	6 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3756      +/-   ##
==========================================
+ Coverage   77.31%   77.66%   +0.34%     
==========================================
  Files         734      735       +1     
  Lines       93737    95495    +1758     
==========================================
+ Hits        72475    74167    +1692     
- Misses      21262    21328      +66

Files with missing lines	Coverage Δ
daft/context.py	`84.88% <40.00%> (-2.78%)`	⬇️
daft/dataframe/dataframe.py	`84.97% <66.66%> (-0.55%)`	⬇️

... and 38 files with indirect coverage changes

daft/dataframe/dataframe.py

jaychia · 2025-02-05T22:14:24Z

daft/dataframe/dataframe.py

@@ -177,6 +199,7 @@ def explain(
            file (Optional[io.IOBase]): Location to print the output to, or defaults to None which defaults to the default location for
                print (in Python, that should be sys.stdout)
        """
+        self.explain_broadcast()


Is this correct? Why would be broadcast to the dashboard if we're not executing a query

Hmm, I was thinking if a user just wanted to see what plan would be produced without executing a query, they could also use the dashboard UI to view the query plan.

In this case, the UI would show:

the query plan

the time it took to produce the plan

"N/A" for the time it took to execute the query (since .explain doesn't execute queries)

jaychia · 2025-02-05T22:15:10Z

daft/dataframe/dataframe.py

@@ -2818,6 +2841,7 @@ def collect(self, num_preview_rows: Optional[int] = 8) -> "DataFrame":
            DataFrame: DataFrame with materialized results.
        """
        self._materialize_results()
+        self.explain_broadcast()


Maybe consider making this a decorator for all "execution" methods (show, collect, write_parquet etc)

Sounds good. A decorator makes sense here. Will update in the coming commit.

Updated to use decorator instead.

I will need to add this to more APIs. Do we have an exhaustive list somewhere of all the APIs we would want to add this decorator to?

jaychia

Can't merge this because of requests dependency

jaychia · 2025-02-06T18:51:55Z

daft/context.py

@@ -125,6 +125,10 @@ class DaftContext:

    _runner: Runner | None = None

+    _enable_broadcast: bool = False
+    _broadcast_addr: str = "127.0.0.1"
+    _broadcast_port: int = 3238


Please put this behind a DashboardConfig dataclass?

jaychia · 2025-02-06T18:52:40Z

daft/context.py

@@ -197,6 +201,31 @@ def get_context() -> DaftContext:
    return _DaftContext


+def broadcast_metrics(


This should really be more of a "enable dashboard" function

jaychia · 2025-02-06T18:53:18Z

daft/dataframe/dataframe.py

@@ -65,6 +67,17 @@
 ManyColumnsInputType = Union[ColumnInputType, Iterable[ColumnInputType]]


+def BroadcastMetrics(func):


nit: functions are usually snake-cased

jaychia · 2025-02-06T18:54:01Z

daft/dataframe/dataframe.py

+def BroadcastMetrics(func):
+    """Calls `self._explain_broadcast()` prior to executing the rest of the function."""
+
+    def wrapper(self, *args, **kwargs):


nit: use @functools.wraps to maintain expected behavior (https://stackoverflow.com/questions/308999/what-does-functools-wraps-do)

jaychia · 2025-02-06T18:54:53Z

daft/dataframe/dataframe.py

@@ -158,6 +171,44 @@ def _result(self) -> Optional[PartitionSet]:
        else:
            return self._result_cache.value

+    def _explain_broadcast(self):
+        """Broadcast the mermaid-formatted plan on the given port (assuming metrics-broadcasting is enabled)."""
+        import requests


I don't think we have requests as a dependency of Daft. This will likely cause issues when run in some users' environments

jaychia · 2025-02-06T19:00:01Z

daft/dataframe/dataframe.py

+
+    def wrapper(self, *args, **kwargs):
+        result = func(self, *args, **kwargs)
+        self._explain_broadcast()


Does this mean that the dashboard only receives the broadcast message after the method finishes execution? Seems wrong.

Another way you can potentially solve this is with a context manager in each dataframe execution method. Something like this:

def write_parquet(): df = ... # build plan with ctx.dashboard.record_execution(df): ... return df

jaychia · 2025-02-06T19:00:21Z

daft/dataframe/dataframe.py

+                    "plan-time-end": str(plan_time_end),
+                },
+            )
+            print(f"Query ID: {id}")


raunakab · 2025-02-06T19:34:48Z

Closing for a couple reasons:

Synced with @jaychia and he mentioned that daft-context has given us problems in the past. Want to move away from a global context in general.
@universalmind303 merged refactor: port DaftContext to rust side #3767 into main. That PR changes daft-context quite a bit, so scrapping this PR and creating a new one seems like it'll be easier.

raunakab added 5 commits January 31, 2025 13:31

Add broadcasting logic

6b28b64

Remove formatting from previous commit

b8e9257

Implement a broadcasting explain function

a029cf0

Add calls to explain_broadcast

4d47847

Add broadcasting of plan

0443d41

github-actions bot added the feat label Feb 2, 2025

Merge branch 'main' into broadcast-metrics

4cf3112

raunakab marked this pull request as ready for review February 5, 2025 21:55

raunakab requested review from jaychia and jessie-young February 5, 2025 21:55

jaychia reviewed Feb 5, 2025

View reviewed changes

raunakab added 5 commits February 5, 2025 14:18

Merge branch 'main' into broadcast-metrics

b654679

Use http posting instead of tcp

0f90721

Rename var

8e3e326

Update to using decorator instead

568f5ce

Add metrics broadcasting to all write functions

0517f7c

raunakab requested a review from jaychia February 5, 2025 22:55

raunakab added 7 commits February 5, 2025 18:28

Send content as json instead of raw text

4d5611b

Update field name

cc1d15d

Edit the structure of data broadcasted on port 3238

74175a5

Add more query-metadata fields

34faa38

Update values passed into broadcast

2b99ef3

Add planning time checkpoints

5e7e2bc

Add printout of query id

45a4459

universalmind303 mentioned this pull request Feb 6, 2025

refactor: port DaftContext to rust side #3767

Merged

jaychia reviewed Feb 6, 2025

View reviewed changes

raunakab closed this Feb 6, 2025

raunakab deleted the broadcast-metrics branch February 6, 2025 19:35

raunakab mentioned this pull request Feb 6, 2025

feat: Add ability to broadcast daft metrics over HTTP #3778

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Broadcast plan over port 3238 #3756

feat: Broadcast plan over port 3238 #3756

raunakab commented Feb 2, 2025

codspeed-hq bot commented Feb 2, 2025 •

edited

Loading

codecov bot commented Feb 2, 2025 •

edited

Loading

jaychia Feb 5, 2025

raunakab Feb 5, 2025 •

edited

Loading

jaychia Feb 5, 2025

raunakab Feb 5, 2025

raunakab Feb 5, 2025

raunakab Feb 5, 2025

jaychia left a comment

jaychia Feb 6, 2025

jaychia Feb 6, 2025

jaychia Feb 6, 2025

jaychia Feb 6, 2025

jaychia Feb 6, 2025

jaychia Feb 6, 2025

jaychia Feb 6, 2025

raunakab commented Feb 6, 2025

		@@ -197,6 +201,31 @@ def get_context() -> DaftContext:
		return _DaftContext


		def broadcast_metrics(

		@@ -65,6 +67,17 @@
		ManyColumnsInputType = Union[ColumnInputType, Iterable[ColumnInputType]]


		def BroadcastMetrics(func):

feat: Broadcast plan over port 3238 #3756

feat: Broadcast plan over port 3238 #3756

Conversation

raunakab commented Feb 2, 2025

Overview

Notes

codspeed-hq bot commented Feb 2, 2025 • edited Loading

Merging #3756 will improve performances by 10.31%

Summary

Benchmarks breakdown

codecov bot commented Feb 2, 2025 • edited Loading

Codecov Report

Choose a reason for hiding this comment

raunakab Feb 5, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jaychia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

raunakab commented Feb 6, 2025

codspeed-hq bot commented Feb 2, 2025 •

edited

Loading

codecov bot commented Feb 2, 2025 •

edited

Loading

raunakab Feb 5, 2025 •

edited

Loading