Implement

src/daft-catalog/src/lib.rs

@@ -168,14 +168,14 @@ mod tests {
             ])
             .unwrap(),
         );
-        LogicalPlan::Source(Source {
-            output_schema: schema.clone(),
-            source_info: Arc::new(SourceInfo::PlaceHolder(PlaceHolderInfo {
+        LogicalPlan::Source(Source::new(
+            schema.clone(),
+            Arc::new(SourceInfo::PlaceHolder(PlaceHolderInfo {
                 source_schema: schema,
                 clustering_spec: Arc::new(ClusteringSpec::unknown()),
                 source_id: 0,
             })),
-        })
+        ))
         .arced()
     }
 

src/daft-local-execution/src/pipeline.rs

@@ -319,18 +319,23 @@ pub fn physical_plan_to_pipeline(
             null_equals_null,
             join_type,
             schema,
+            ..
         }) => {
             let left_schema = left.schema();
             let right_schema = right.schema();
+            let left_size = left.get_stats().approx_stats.upper_bound_bytes;
+            let right_size = right.get_stats().approx_stats.upper_bound_bytes;
 
-            // Determine the build and probe sides based on the join type
-            // Currently it is a naive determination, in the future we should leverage the cardinality of the tables
-            // to determine the build and probe sides
+            // Determine the build and probe sides based on the join type and cardinalities.
             let build_on_left = match join_type {
-                JoinType::Inner => true,
-                JoinType::Right => true,
-                JoinType::Outer => true,
+                JoinType::Inner => left_size <= right_size,
+                JoinType::Outer => left_size <= right_size,
+                // TODO(desmond): We might potentially want to flip the probe table side for
+                // left/right outer joins if one side is significantly larger. Needs tuning.
+                // For left outer joins, we build on right so we can stream the left side.
                 JoinType::Left => false,
+                // For right outer joins, we build on left so we can stream the right side.
+                JoinType::Right => true,
                 JoinType::Anti | JoinType::Semi => false,
             };
             let (build_on, probe_on, build_child, probe_child) = match build_on_left {