feat(optimizer): Implement naive join ordering

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

@@ -24,6 +24,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}