[CHORE] Enable read_sql for swordfish

[colin-ho]

Enables read_sql tests for swordfish.

Additionally adds an EmptyScan operator. This is needed because some sql tests produce 0 scan tasks (some other IO tests that are skipped also need empty scan, so this PR will enable those to pass as well, I will unskip them in future PRs).

Technically, we don't need a dedicated empty scan operator for this, but it might be useful to be explicit. (For what it's worth the current runner does have an empty scan operator as well)

[codspeed-hq]

CodSpeed Performance Report

Merging #2918 will not alter performance

_{Comparing colin/swordfish-read-sql (df5d4d1) with main (64b8699)}

Summary

✅ 17 untouched benchmarks

[codecov]

Codecov Report

Attention: Patch coverage is 50.94340% with 26 lines in your changes missing coverage. Please review.

Project coverage is 78.47%. Comparing base (3f37a69) to head (df5d4d1).
Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
src/daft-local-execution/src/sources/scan_task.rs	0.00%	24 Missing ⚠️
src/daft-micropartition/src/python.rs	0.00%	1 Missing ⚠️
src/daft-physical-plan/src/local_plan.rs	87.50%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2918      +/-   ##
==========================================
+ Coverage   78.43%   78.47%   +0.03%     
==========================================
  Files         603      610       +7     
  Lines       71504    71742     +238     
==========================================
+ Hits        56086    56299     +213     
- Misses      15418    15443      +25     

Files with missing lines	Coverage Δ
src/daft-local-execution/src/pipeline.rs	91.51% <100.00%> (+0.11%)	⬆️
src/daft-local-execution/src/sources/empty_scan.rs	100.00% <100.00%> (ø)
src/daft-physical-plan/src/translate.rs	91.95% <100.00%> (+0.18%)	⬆️
src/daft-micropartition/src/python.rs	76.74% <0.00%> (ø)
src/daft-physical-plan/src/local_plan.rs	85.80% <87.50%> (+0.09%)	⬆️
src/daft-local-execution/src/sources/scan_task.rs	81.95% <0.00%> (-9.36%)	⬇️

... and 36 files with indirect coverage changes

[colin-ho]

The existing read_sql code for the current executor is here:

Daft/src/daft-micropartition/src/micropartition.rs

Lines 356 to 380 in 3f37a69

	FileFormatConfig::Database(DatabaseSourceConfig { sql, conn }) => {
	let predicate = scan_task
	.pushdowns
	.filters
	.as_ref()
	.map(|p| (*p.as_ref()).clone().into());
	Python::with_gil(|py| {
	let table = crate::python::read_sql_into_py_table(
	py,
	sql,
	conn,
	predicate.clone(),
	scan_task.schema.clone().into(),
	scan_task
	.pushdowns
	.columns
	.as_ref()
	.map(|cols| cols.as_ref().clone()),
	scan_task.pushdowns.limit,
	)
	.map(std::convert::Into::into)
	.context(PyIOSnafu)?;
	Ok(vec![table])
	})?
	}

, i just copied it for swordfish. The only difference is that a scan task for swordfish returns a stream, in the case for sql however, there's no way currently to stream results from a database read, so no choice but to wrap the result in a futures::stream::once

[desmondcheongzx]

Might be nice to leave a comment about this

[desmondcheongzx]

LGTM