diff --git a/src/daft-catalog/src/lib.rs b/src/daft-catalog/src/lib.rs
index e01c6f7acd..73f75864c8 100644
--- a/src/daft-catalog/src/lib.rs
+++ b/src/daft-catalog/src/lib.rs
@@ -182,7 +182,7 @@ mod tests {
     #[test]
     fn test_register_and_unregister_named_table() {
         let mut catalog = DaftMetaCatalog::new_from_env();
-        let plan = LogicalPlanBuilder::new(mock_plan(), None, None);
+        let plan = LogicalPlanBuilder::from(mock_plan());
 
         // Register a table
         assert!(catalog
@@ -198,7 +198,7 @@ mod tests {
     #[test]
     fn test_read_registered_table() {
         let mut catalog = DaftMetaCatalog::new_from_env();
-        let plan = LogicalPlanBuilder::new(mock_plan(), None, None);
+        let plan = LogicalPlanBuilder::from(mock_plan());
 
         catalog.register_named_table("test_table", plan).unwrap();
 
diff --git a/src/daft-local-execution/src/pipeline.rs b/src/daft-local-execution/src/pipeline.rs
index ba7119d6f5..e294d7ebc1 100644
--- a/src/daft-local-execution/src/pipeline.rs
+++ b/src/daft-local-execution/src/pipeline.rs
@@ -333,6 +333,7 @@ pub fn physical_plan_to_pipeline(
                     left_stats.approx_stats.upper_bound_bytes
                         <= right_stats.approx_stats.upper_bound_bytes
                 }
+                // If stats are not available, we fall back and build on the left.
                 _ => true,
             };
 
diff --git a/src/daft-logical-plan/src/builder.rs b/src/daft-logical-plan/src/builder.rs
index 7ab1561cb5..0d2effa0cf 100644
--- a/src/daft-logical-plan/src/builder.rs
+++ b/src/daft-logical-plan/src/builder.rs
@@ -152,7 +152,7 @@ impl LogicalPlanBuilder {
         ));
         let logical_plan: LogicalPlan = ops::Source::new(schema, source_info.into()).into();
 
-        Ok(Self::new(logical_plan.into(), None, None))
+        Ok(Self::from(Arc::new(logical_plan)))
     }
 
     pub fn table_scan(
@@ -204,7 +204,7 @@ impl LogicalPlanBuilder {
             schema_with_generated_fields
         };
         let logical_plan: LogicalPlan = ops::Source::new(output_schema, source_info.into()).into();
-        Ok(Self::new(logical_plan.into(), None, None))
+        Ok(Self::from(Arc::new(logical_plan)))
     }
 
     pub fn select(&self, to_select: Vec<ExprRef>) -> DaftResult<Self> {
diff --git a/src/daft-logical-plan/src/display.rs b/src/daft-logical-plan/src/display.rs
index 17559d8b8e..b2340ef616 100644
--- a/src/daft-logical-plan/src/display.rs
+++ b/src/daft-logical-plan/src/display.rs
@@ -85,11 +85,11 @@ mod test {
     #[test]
     // create a random, complex plan and check if it can be displayed as expected
     fn test_mermaid_display() -> DaftResult<()> {
-        let subplan = LogicalPlanBuilder::new(plan_1(), None, None)
+        let subplan = LogicalPlanBuilder::from(plan_1())
             .filter(col("id").eq(lit(1)))?
             .build();
 
-        let subplan2 = LogicalPlanBuilder::new(plan_2(), None, None)
+        let subplan2 = LogicalPlanBuilder::from(plan_2())
             .filter(
                 startswith(col("last_name"), lit("S")).and(endswith(col("last_name"), lit("n"))),
             )?
@@ -159,11 +159,11 @@ Project1 --> Limit0
     #[test]
     // create a random, complex plan and check if it can be displayed as expected
     fn test_mermaid_display_simple() -> DaftResult<()> {
-        let subplan = LogicalPlanBuilder::new(plan_1(), None, None)
+        let subplan = LogicalPlanBuilder::from(plan_1())
             .filter(col("id").eq(lit(1)))?
             .build();
 
-        let subplan2 = LogicalPlanBuilder::new(plan_2(), None, None)
+        let subplan2 = LogicalPlanBuilder::from(plan_2())
             .filter(
                 startswith(col("last_name"), lit("S")).and(endswith(col("last_name"), lit("n"))),
             )?
@@ -173,7 +173,7 @@ Project1 --> Limit0
             .sort(vec![col("last_name")], vec![false], vec![false])?
             .build();
 
-        let plan = LogicalPlanBuilder::new(subplan, None, None)
+        let plan = LogicalPlanBuilder::from(subplan)
             .join_with_null_safe_equal(
                 subplan2,
                 vec![col("id")],
diff --git a/src/daft-logical-plan/src/ops/source.rs b/src/daft-logical-plan/src/ops/source.rs
index a2212d8c94..dbd214ee06 100644
--- a/src/daft-logical-plan/src/ops/source.rs
+++ b/src/daft-logical-plan/src/ops/source.rs
@@ -29,6 +29,7 @@ impl Source {
         }
     }
 
+    // Helper function that converts ScanOperatorRef inside a the Source node's PhysicalScanInfo into scan tasks.
     pub(crate) fn build_materialized_scan_source(
         mut self,
         execution_config: Option<&DaftExecutionConfig>,
diff --git a/src/daft-sql/src/lib.rs b/src/daft-sql/src/lib.rs
index 97e06a2533..485dcf0aeb 100644
--- a/src/daft-sql/src/lib.rs
+++ b/src/daft-sql/src/lib.rs
@@ -191,7 +191,7 @@ mod tests {
         let sql = "select test as a from tbl1";
         let plan = planner.plan_sql(sql).unwrap();
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .select(vec![col("test").alias("a")])
             .unwrap()
             .build();
@@ -203,7 +203,7 @@ mod tests {
         let sql = "select test as a from tbl1 where test = 'a'";
         let plan = planner.plan_sql(sql)?;
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .filter(col("test").eq(lit("a")))?
             .select(vec![col("test").alias("a")])?
             .build();
@@ -216,7 +216,7 @@ mod tests {
         let sql = "select test as a from tbl1 limit 10";
         let plan = planner.plan_sql(sql)?;
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .select(vec![col("test").alias("a")])?
             .limit(10, true)?
             .build();
@@ -230,7 +230,7 @@ mod tests {
         let sql = "select utf8 from tbl1 order by utf8 desc";
         let plan = planner.plan_sql(sql)?;
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .select(vec![col("utf8")])?
             .sort(vec![col("utf8")], vec![true], vec![true])?
             .build();
@@ -241,7 +241,7 @@ mod tests {
 
     #[rstest]
     fn test_cast(mut planner: SQLPlanner, tbl_1: LogicalPlanRef) -> SQLPlannerResult<()> {
-        let builder = LogicalPlanBuilder::new(tbl_1, None, None);
+        let builder = LogicalPlanBuilder::from(tbl_1);
         let cases = vec![
             (
                 "select bool::text from tbl1",
@@ -285,7 +285,7 @@ mod tests {
             if null_equals_null { "<=>" } else { "=" }
         );
         let plan = planner.plan_sql(&sql)?;
-        let expected = LogicalPlanBuilder::new(tbl_2, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_2)
             .join_with_null_safe_equal(
                 tbl_3,
                 vec![col("id")],
@@ -312,7 +312,7 @@ mod tests {
         let sql = "select * from tbl2 join tbl3 on tbl2.id = tbl3.id and tbl2.val > 0";
         let plan = planner.plan_sql(&sql)?;
 
-        let expected = LogicalPlanBuilder::new(tbl_2, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_2)
             .filter(col("val").gt(lit(0 as i64)))?
             .join_with_null_safe_equal(
                 tbl_3,
@@ -394,7 +394,7 @@ mod tests {
         let sql = "select max(i32) from tbl1";
         let plan = planner.plan_sql(sql)?;
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .aggregate(vec![col("i32").max()], vec![])?
             .select(vec![col("i32")])?
             .build();
@@ -469,7 +469,7 @@ mod tests {
             field: Field::new("i32", DataType::Int32),
             depth: 1,
         }));
-        let subquery = LogicalPlanBuilder::new(tbl_2, None, None)
+        let subquery = LogicalPlanBuilder::from(tbl_2)
             .filter(col("id").eq(outer_col))?
             .aggregate(vec![col("id").max()], vec![])?
             .select(vec![col("id")])?
@@ -477,7 +477,7 @@ mod tests {
 
         let subquery = Arc::new(Expr::Subquery(Subquery { plan: subquery }));
 
-        let expected = LogicalPlanBuilder::new(tbl_1, None, None)
+        let expected = LogicalPlanBuilder::from(tbl_1)
             .filter(col("i64").gt(subquery))?
             .select(vec![col("utf8")])?
             .build();