fix: correct statistics for NestedLoopJoinExec

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -162,7 +162,8 @@ pub struct NestedLoopJoinExec {
     pub(crate) filter: Option<JoinFilter>,
     /// How the join is performed
     pub(crate) join_type: JoinType,
-    /// The schema once the join is applied
+    /// The full concatenated schema of left and right children should be distinct from
+    /// the output schema of the operator
     join_schema: SchemaRef,
     /// Future that consumes left input and buffers it in memory
     ///
@@ -564,7 +565,7 @@ impl ExecutionPlan for NestedLoopJoinExec {
             self.right.partition_statistics(None)?,
             vec![],
             &self.join_type,
-            &self.join_schema,
+            &self.schema(),
         )
     }
 
@@ -1557,6 +1558,26 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn join_has_correct_stats() -> Result<()> {
+        let left = build_left_table();
+        let right = build_right_table();
+        let nested_loop_join = NestedLoopJoinExec::try_new(
+            left,
+            right,
+            None,
+            &JoinType::Left,
+            Some(vec![1, 2]),
+        )?;
+        let stats = nested_loop_join.partition_statistics(None)?;
+        assert_eq!(
+            nested_loop_join.schema().fields().len(),
+            stats.column_statistics.len(),
+        );
+        assert_eq!(2, stats.column_statistics.len());
+        Ok(())
+    }
+
     #[rstest]
     #[tokio::test]
     async fn join_right_semi_with_filter(