[FEAT] Enable to_json_string for physical plan

[colin-ho]

#2977

Allows physical plan to be serialized to be json strings. The script below shows how you can extract the sql queries from read sql.

import daft
import json

def create_conn():
    return create_engine('sqlite:///test_database.db').connect()

df = daft.read_sql("SELECT * FROM test_data", create_conn, partition_col="id", num_partitions=3)
physical_plan_scheduler = df._builder.to_physical_plan_scheduler(
    daft.context.get_context().daft_execution_config
)

physical_plan_dict = json.loads(physical_plan_scheduler.to_json_string())

# collect all sql queries
sql_queries = []
for task in physical_plan_dict['TabularScan']['scan_tasks']:
    sql_queries.append(task['file_format_config']['Database']['sql'])
print(sql_queries)

['SELECT * FROM (SELECT * FROM test_data) AS subquery WHERE id >= 1 AND id < 34', 'SELECT * FROM (SELECT * FROM test_data) AS subquery WHERE id >= 34 AND id < 67', 'SELECT * FROM (SELECT * FROM test_data) AS subquery WHERE id >= 67 AND id <= 100']

[codspeed-hq]

CodSpeed Performance Report

Merging #3023 will not alter performance

_{Comparing colin/physical-plan-json (0de3933) with main (31e0de2)}

Summary

✅ 17 untouched benchmarks

[codecov]

Codecov Report

Attention: Patch coverage is 16.66667% with 5 lines in your changes missing coverage. Please review.

Project coverage is 78.44%. Comparing base (3f37a69) to head (0de3933).
Report is 20 commits behind head on main.

Files with missing lines	Patch %	Lines
src/daft-scheduler/src/scheduler.rs	0.00%	4 Missing ⚠️
daft/plan_scheduler/physical_plan_scheduler.py	50.00%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main    #3023    +/-   ##
========================================
  Coverage   78.43%   78.44%            
========================================
  Files         603      610     +7     
  Lines       71504    72007   +503     
========================================
+ Hits        56086    56487   +401     
- Misses      15418    15520   +102     

Files with missing lines	Coverage Δ
daft/logical/builder.py	89.61% <ø> (ø)
daft/plan_scheduler/physical_plan_scheduler.py	54.90% <50.00%> (-0.21%)	⬇️
src/daft-scheduler/src/scheduler.rs	92.85% <0.00%> (-0.61%)	⬇️

... and 66 files with indirect coverage changes

[jaehyeon-kim]

Hi @colin-ho and @jaychia

Can this PR be reviewed?

[jaychia]

We should also note that df._builder.to_physical_plan_scheduler is not a stable API, since it does access a private method in the dataframe.

In general I don't think we have any plans to make this a user-accessible feature via a stable user API

[jaehyeon-kim]

@jaychia

Thanks for the reminder. At least, I can get started building the sql connector.

@colin-ho

Thank you so much for adding the feature. Can you please inform me when it may be released?