feat(optimizer): Implement naive join ordering (#3616)

Implements a naive join orderer that simply takes joins relations
arbitrarily (as long as a valid join condition exists).

This is intended as a building block to ensure that our join graphs can
correctly reconstruct into logical plans. The PR that will immediately
follow this will create an optimization rule that applies naive join
ordering. The optimization rule will be hidden behind a config flag, but
will allow us to test logical plan reconstruction on all our integration
tests.

src/common/scan-info/src/test/mod.rs

@@ -17,12 +17,14 @@ use crate::{PartitionField, Pushdowns, ScanOperator, ScanTaskLike, ScanTaskLikeR
 struct DummyScanTask {
     pub schema: SchemaRef,
     pub pushdowns: Pushdowns,
+    pub in_memory_size: Option<usize>,
 }
 
 #[derive(Debug)]
 pub struct DummyScanOperator {
     pub schema: SchemaRef,
     pub num_scan_tasks: u32,
+    pub in_memory_size_per_task: Option<usize>,
 }
 
 #[typetag::serde]
@@ -67,7 +69,7 @@ impl ScanTaskLike for DummyScanTask {
     }
 
     fn estimate_in_memory_size_bytes(&self, _: Option<&DaftExecutionConfig>) -> Option<usize> {
-        None
+        self.in_memory_size
     }
 
     fn file_format_config(&self) -> Arc<FileFormatConfig> {
@@ -136,6 +138,7 @@ impl ScanOperator for DummyScanOperator {
         let scan_task = Arc::new(DummyScanTask {
             schema: self.schema.clone(),
             pushdowns,
+            in_memory_size: self.in_memory_size_per_task,
         });
 
         Ok((0..self.num_scan_tasks)

src/daft-logical-plan/Cargo.toml

@@ -26,6 +26,7 @@ uuid = {version = "1", features = ["v4"]}
 [dev-dependencies]
 daft-dsl = {path = "../daft-dsl", features = ["test-utils"]}
 pretty_assertions = {workspace = true}
+rand = "0.8"
 rstest = {workspace = true}
 test-log = {workspace = true}