deprecated statistics API

Author: xudong963

datafusion/core/src/datasource/file_format/csv.rs

@@ -217,8 +217,11 @@ mod tests {
         assert_eq!(tt_batches, 50 /* 100/2 */);
 
         // test metadata
-        assert_eq!(exec.statistics()?.num_rows, Precision::Absent);
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Absent);
+        assert_eq!(exec.partition_statistics(None)?.num_rows, Precision::Absent);
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Absent
+        );
 
         Ok(())
     }

datafusion/core/src/datasource/file_format/json.rs

@@ -75,8 +75,11 @@ mod tests {
         assert_eq!(tt_batches, 6 /* 12/2 */);
 
         // test metadata
-        assert_eq!(exec.statistics()?.num_rows, Precision::Absent);
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Absent);
+        assert_eq!(exec.partition_statistics(None)?.num_rows, Precision::Absent);
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Absent
+        );
 
         Ok(())
     }

datafusion/core/src/datasource/file_format/parquet.rs

@@ -616,9 +616,15 @@ mod tests {
         assert_eq!(tt_batches, 4 /* 8/2 */);
 
         // test metadata
-        assert_eq!(exec.statistics()?.num_rows, Precision::Exact(8));
+        assert_eq!(
+            exec.partition_statistics(None)?.num_rows,
+            Precision::Exact(8)
+        );
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Exact(671));
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Exact(671)
+        );
 
         Ok(())
     }
@@ -659,9 +665,15 @@ mod tests {
             get_exec(&state, "alltypes_plain.parquet", projection, Some(1)).await?;
 
         // note: even if the limit is set, the executor rounds up to the batch size
-        assert_eq!(exec.statistics()?.num_rows, Precision::Exact(8));
+        assert_eq!(
+            exec.partition_statistics(None)?.num_rows,
+            Precision::Exact(8)
+        );
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Exact(671));
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Exact(671)
+        );
         let batches = collect(exec, task_ctx).await?;
         assert_eq!(1, batches.len());
         assert_eq!(11, batches[0].num_columns());

datafusion/core/src/datasource/listing/table.rs

@@ -1294,8 +1294,14 @@ mod tests {
         assert_eq!(exec.output_partitioning().partition_count(), 1);
 
         // test metadata
-        assert_eq!(exec.statistics()?.num_rows, Precision::Exact(8));
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Exact(671));
+        assert_eq!(
+            exec.partition_statistics(None)?.num_rows,
+            Precision::Exact(8)
+        );
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Exact(671)
+        );
 
         Ok(())
     }
@@ -1320,9 +1326,15 @@ mod tests {
         let table = ListingTable::try_new(config)?;
 
         let exec = table.scan(&state, None, &[], None).await?;
-        assert_eq!(exec.statistics()?.num_rows, Precision::Exact(8));
+        assert_eq!(
+            exec.partition_statistics(None)?.num_rows,
+            Precision::Exact(8)
+        );
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Exact(671));
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Exact(671)
+        );
 
         Ok(())
     }
@@ -1348,8 +1360,11 @@ mod tests {
         let table = ListingTable::try_new(config)?;
 
         let exec = table.scan(&state, None, &[], None).await?;
-        assert_eq!(exec.statistics()?.num_rows, Precision::Absent);
-        assert_eq!(exec.statistics()?.total_byte_size, Precision::Absent);
+        assert_eq!(exec.partition_statistics(None)?.num_rows, Precision::Absent);
+        assert_eq!(
+            exec.partition_statistics(None)?.total_byte_size,
+            Precision::Absent
+        );
 
         Ok(())
     }

datafusion/core/tests/custom_sources_cases/statistics.rs

@@ -232,7 +232,7 @@ async fn sql_basic() -> Result<()> {
     let physical_plan = df.create_physical_plan().await.unwrap();
 
     // the statistics should be those of the source
-    assert_eq!(stats, physical_plan.statistics()?);
+    assert_eq!(stats, physical_plan.partition_statistics(None)?);
 
     Ok(())
 }
@@ -248,7 +248,7 @@ async fn sql_filter() -> Result<()> {
         .unwrap();
 
     let physical_plan = df.create_physical_plan().await.unwrap();
-    let stats = physical_plan.statistics()?;
+    let stats = physical_plan.partition_statistics(None)?;
     assert_eq!(stats.num_rows, Precision::Inexact(1));
 
     Ok(())
@@ -270,7 +270,7 @@ async fn sql_limit() -> Result<()> {
             column_statistics: col_stats,
             total_byte_size: Precision::Absent
         },
-        physical_plan.statistics()?
+        physical_plan.partition_statistics(None)?
     );
 
     let df = ctx
@@ -279,7 +279,7 @@ async fn sql_limit() -> Result<()> {
         .unwrap();
     let physical_plan = df.create_physical_plan().await.unwrap();
     // when the limit is larger than the original number of lines, statistics remain unchanged
-    assert_eq!(stats, physical_plan.statistics()?);
+    assert_eq!(stats, physical_plan.partition_statistics(None)?);
 
     Ok(())
 }
@@ -296,7 +296,7 @@ async fn sql_window() -> Result<()> {
 
     let physical_plan = df.create_physical_plan().await.unwrap();
 
-    let result = physical_plan.statistics()?;
+    let result = physical_plan.partition_statistics(None)?;
 
     assert_eq!(stats.num_rows, result.num_rows);
     let col_stats = result.column_statistics;

datafusion/core/tests/parquet/file_statistics.rs

@@ -50,13 +50,19 @@ async fn check_stats_precision_with_filter_pushdown() {
     let (_, _, state) = get_cache_runtime_state();
     // Scan without filter, stats are exact
     let exec = table.scan(&state, None, &[], None).await.unwrap();
-    assert_eq!(exec.statistics().unwrap().num_rows, Precision::Exact(8));
+    assert_eq!(
+        exec.partition_statistics(None).unwrap().num_rows,
+        Precision::Exact(8)
+    );
 
     // Scan with filter pushdown, stats are inexact
     let filter = Expr::gt(col("id"), lit(1));
 
     let exec = table.scan(&state, None, &[filter], None).await.unwrap();
-    assert_eq!(exec.statistics().unwrap().num_rows, Precision::Inexact(8));
+    assert_eq!(
+        exec.partition_statistics(None).unwrap().num_rows,
+        Precision::Inexact(8)
+    );
 }
 
 #[tokio::test]
@@ -79,9 +85,12 @@ async fn load_table_stats_with_session_level_cache() {
     assert_eq!(get_static_cache_size(&state1), 0);
     let exec1 = table1.scan(&state1, None, &[], None).await.unwrap();
 
-    assert_eq!(exec1.statistics().unwrap().num_rows, Precision::Exact(8));
     assert_eq!(
-        exec1.statistics().unwrap().total_byte_size,
+        exec1.partition_statistics(None).unwrap().num_rows,
+        Precision::Exact(8)
+    );
+    assert_eq!(
+        exec1.partition_statistics(None).unwrap().total_byte_size,
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
         Precision::Exact(671),
     );
@@ -91,9 +100,12 @@ async fn load_table_stats_with_session_level_cache() {
     //check session 1 cache result not show in session 2
     assert_eq!(get_static_cache_size(&state2), 0);
     let exec2 = table2.scan(&state2, None, &[], None).await.unwrap();
-    assert_eq!(exec2.statistics().unwrap().num_rows, Precision::Exact(8));
     assert_eq!(
-        exec2.statistics().unwrap().total_byte_size,
+        exec2.partition_statistics(None).unwrap().num_rows,
+        Precision::Exact(8)
+    );
+    assert_eq!(
+        exec2.partition_statistics(None).unwrap().total_byte_size,
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
         Precision::Exact(671),
     );
@@ -103,9 +115,12 @@ async fn load_table_stats_with_session_level_cache() {
     //check session 1 cache result not show in session 2
     assert_eq!(get_static_cache_size(&state1), 1);
     let exec3 = table1.scan(&state1, None, &[], None).await.unwrap();
-    assert_eq!(exec3.statistics().unwrap().num_rows, Precision::Exact(8));
     assert_eq!(
-        exec3.statistics().unwrap().total_byte_size,
+        exec3.partition_statistics(None).unwrap().num_rows,
+        Precision::Exact(8)
+    );
+    assert_eq!(
+        exec3.partition_statistics(None).unwrap().total_byte_size,
         // TODO correct byte size: https://github.com/apache/datafusion/issues/14936
         Precision::Exact(671),
     );

datafusion/core/tests/physical_optimizer/enforce_distribution.rs

@@ -170,7 +170,7 @@ impl ExecutionPlan for SortRequiredExec {
     }
 
     fn statistics(&self) -> Result<Statistics> {
-        self.input.statistics()
+        self.input.partition_statistics(None)
     }
 }
 

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -251,11 +251,19 @@ async fn test_join_with_swap() {
         .expect("The type of the plan should not be changed");
 
     assert_eq!(
-        swapped_join.left().statistics().unwrap().total_byte_size,
+        swapped_join
+            .left()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(8192)
     );
     assert_eq!(
-        swapped_join.right().statistics().unwrap().total_byte_size,
+        swapped_join
+            .right()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(2097152)
     );
 }
@@ -291,11 +299,19 @@ async fn test_left_join_no_swap() {
         .expect("The type of the plan should not be changed");
 
     assert_eq!(
-        swapped_join.left().statistics().unwrap().total_byte_size,
+        swapped_join
+            .left()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(8192)
     );
     assert_eq!(
-        swapped_join.right().statistics().unwrap().total_byte_size,
+        swapped_join
+            .right()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(2097152)
     );
 }
@@ -336,11 +352,19 @@ async fn test_join_with_swap_semi() {
 
         assert_eq!(swapped_join.schema().fields().len(), 1);
         assert_eq!(
-            swapped_join.left().statistics().unwrap().total_byte_size,
+            swapped_join
+                .left()
+                .partition_statistics(None)
+                .unwrap()
+                .total_byte_size,
             Precision::Inexact(8192)
         );
         assert_eq!(
-            swapped_join.right().statistics().unwrap().total_byte_size,
+            swapped_join
+                .right()
+                .partition_statistics(None)
+                .unwrap()
+                .total_byte_size,
             Precision::Inexact(2097152)
         );
         assert_eq!(original_schema, swapped_join.schema());
@@ -455,11 +479,19 @@ async fn test_join_no_swap() {
         .expect("The type of the plan should not be changed");
 
     assert_eq!(
-        swapped_join.left().statistics().unwrap().total_byte_size,
+        swapped_join
+            .left()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(8192)
     );
     assert_eq!(
-        swapped_join.right().statistics().unwrap().total_byte_size,
+        swapped_join
+            .right()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(2097152)
     );
 }
@@ -524,11 +556,19 @@ async fn test_nl_join_with_swap(join_type: JoinType) {
     );
 
     assert_eq!(
-        swapped_join.left().statistics().unwrap().total_byte_size,
+        swapped_join
+            .left()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(8192)
     );
     assert_eq!(
-        swapped_join.right().statistics().unwrap().total_byte_size,
+        swapped_join
+            .right()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(2097152)
     );
 }
@@ -589,11 +629,19 @@ async fn test_nl_join_with_swap_no_proj(join_type: JoinType) {
     );
 
     assert_eq!(
-        swapped_join.left().statistics().unwrap().total_byte_size,
+        swapped_join
+            .left()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(8192)
     );
     assert_eq!(
-        swapped_join.right().statistics().unwrap().total_byte_size,
+        swapped_join
+            .right()
+            .partition_statistics(None)
+            .unwrap()
+            .total_byte_size,
         Precision::Inexact(2097152)
     );
 }

datafusion/core/tests/sql/path_partition.rs

@@ -511,7 +511,7 @@ async fn parquet_statistics() -> Result<()> {
     let schema = physical_plan.schema();
     assert_eq!(schema.fields().len(), 4);
 
-    let stat_cols = physical_plan.statistics()?.column_statistics;
+    let stat_cols = physical_plan.partition_statistics(None)?.column_statistics;
     assert_eq!(stat_cols.len(), 4);
     // stats for the first col are read from the parquet file
     assert_eq!(stat_cols[0].null_count, Precision::Exact(3));
@@ -526,7 +526,7 @@ async fn parquet_statistics() -> Result<()> {
     let schema = physical_plan.schema();
     assert_eq!(schema.fields().len(), 2);
 
-    let stat_cols = physical_plan.statistics()?.column_statistics;
+    let stat_cols = physical_plan.partition_statistics(None)?.column_statistics;
     assert_eq!(stat_cols.len(), 2);
     // stats for the first col are read from the parquet file
     assert_eq!(stat_cols[0].null_count, Precision::Exact(1));

datafusion/datasource/src/memory.rs

@@ -976,7 +976,7 @@ mod tests {
         )?;
 
         assert_eq!(
-            values.statistics()?,
+            values.partition_statistics(None)?,
             Statistics {
                 num_rows: Precision::Exact(rows),
                 total_byte_size: Precision::Exact(8), // not important