[pull] main from apache:main

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -1824,7 +1824,7 @@ STORED AS PARQUET;
     assert!(explain.contains("output_rows=128")); // Read 1 row group
     assert!(explain.contains("t@0 < 1372708809")); // Dynamic filter was applied
     assert!(
-        explain.contains("pushdown_rows_matched=128, pushdown_rows_pruned=99872"),
+        explain.contains("pushdown_rows_matched=128, pushdown_rows_pruned=99.87 K"),
         "{explain}"
     );
     // Pushdown pruned most rows

datafusion/execution/src/memory_pool/mod.rs

@@ -501,6 +501,56 @@ pub fn human_readable_size(size: usize) -> String {
     format!("{value:.1} {unit}")
 }
 
+/// Present count in human-readable form with K, M, B, T suffixes
+pub fn human_readable_count(count: usize) -> String {
+    let count = count as u64;
+    let (value, unit) = {
+        if count >= 1_000_000_000_000 {
+            (count as f64 / 1_000_000_000_000.0, " T")
+        } else if count >= 1_000_000_000 {
+            (count as f64 / 1_000_000_000.0, " B")
+        } else if count >= 1_000_000 {
+            (count as f64 / 1_000_000.0, " M")
+        } else if count >= 1_000 {
+            (count as f64 / 1_000.0, " K")
+        } else {
+            return count.to_string();
+        }
+    };
+
+    // Format with appropriate precision
+    // For values >= 100, show 1 decimal place (e.g., 123.4 K)
+    // For values < 100, show 2 decimal places (e.g., 10.12 K)
+    if value >= 100.0 {
+        format!("{value:.1}{unit}")
+    } else {
+        format!("{value:.2}{unit}")
+    }
+}
+
+/// Present duration in human-readable form with 2 decimal places
+pub fn human_readable_duration(nanos: u64) -> String {
+    const NANOS_PER_SEC: f64 = 1_000_000_000.0;
+    const NANOS_PER_MILLI: f64 = 1_000_000.0;
+    const NANOS_PER_MICRO: f64 = 1_000.0;
+
+    let nanos_f64 = nanos as f64;
+
+    if nanos >= 1_000_000_000 {
+        // >= 1 second: show in seconds
+        format!("{:.2}s", nanos_f64 / NANOS_PER_SEC)
+    } else if nanos >= 1_000_000 {
+        // >= 1 millisecond: show in milliseconds
+        format!("{:.2}ms", nanos_f64 / NANOS_PER_MILLI)
+    } else if nanos >= 1_000 {
+        // >= 1 microsecond: show in microseconds
+        format!("{:.2}µs", nanos_f64 / NANOS_PER_MICRO)
+    } else {
+        // < 1 microsecond: show in nanoseconds
+        format!("{nanos}ns")
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -597,4 +647,57 @@ mod tests {
         assert_eq!(r2.size(), 25);
         assert_eq!(pool.reserved(), 28);
     }
+
+    #[test]
+    fn test_human_readable_count() {
+        // Test small numbers (< 1000) - should display as-is
+        assert_eq!(human_readable_count(0), "0");
+        assert_eq!(human_readable_count(1), "1");
+        assert_eq!(human_readable_count(999), "999");
+
+        // Test thousands (K)
+        assert_eq!(human_readable_count(1_000), "1.00 K");
+        assert_eq!(human_readable_count(10_100), "10.10 K");
+        assert_eq!(human_readable_count(1_532), "1.53 K");
+        assert_eq!(human_readable_count(99_999), "100.00 K");
+
+        // Test millions (M)
+        assert_eq!(human_readable_count(1_000_000), "1.00 M");
+        assert_eq!(human_readable_count(1_532_000), "1.53 M");
+        assert_eq!(human_readable_count(99_000_000), "99.00 M");
+        assert_eq!(human_readable_count(123_456_789), "123.5 M");
+
+        // Test billions (B)
+        assert_eq!(human_readable_count(1_000_000_000), "1.00 B");
+        assert_eq!(human_readable_count(1_532_000_000), "1.53 B");
+        assert_eq!(human_readable_count(999_999_999_999), "1000.0 B");
+
+        // Test trillions (T)
+        assert_eq!(human_readable_count(1_000_000_000_000), "1.00 T");
+        assert_eq!(human_readable_count(42_000_000_000_000), "42.00 T");
+    }
+
+    #[test]
+    fn test_human_readable_duration() {
+        // Test nanoseconds (< 1µs)
+        assert_eq!(human_readable_duration(0), "0ns");
+        assert_eq!(human_readable_duration(1), "1ns");
+        assert_eq!(human_readable_duration(999), "999ns");
+
+        // Test microseconds (1µs to < 1ms)
+        assert_eq!(human_readable_duration(1_000), "1.00µs");
+        assert_eq!(human_readable_duration(1_234), "1.23µs");
+        assert_eq!(human_readable_duration(999_999), "1000.00µs");
+
+        // Test milliseconds (1ms to < 1s)
+        assert_eq!(human_readable_duration(1_000_000), "1.00ms");
+        assert_eq!(human_readable_duration(11_295_377), "11.30ms");
+        assert_eq!(human_readable_duration(1_234_567), "1.23ms");
+        assert_eq!(human_readable_duration(999_999_999), "1000.00ms");
+
+        // Test seconds (>= 1s)
+        assert_eq!(human_readable_duration(1_000_000_000), "1.00s");
+        assert_eq!(human_readable_duration(1_234_567_890), "1.23s");
+        assert_eq!(human_readable_duration(42_000_000_000), "42.00s");
+    }
 }

datafusion/expr/src/logical_plan/plan.rs

@@ -1476,13 +1476,29 @@ impl LogicalPlan {
                     Ok(transformed_expr.update_data(|expr| original_name.restore(expr)))
                 }
             })?
-            // always recompute the schema to ensure the changed in the schema's field should be
-            // poplulated to the plan's parent
-            .map_data(|plan| plan.recompute_schema())
+            .map_data(|plan| plan.update_schema_data_type())
         })
         .map(|res| res.data)
     }
 
+    /// Recompute schema fields' data type after replacing params, ensuring fields data type can be
+    /// updated according to the new parameters.
+    ///
+    /// Unlike `recompute_schema()`, this method rebuilds VALUES plans entirely to properly infer
+    /// types types from literal values after placeholder substitution.
+    fn update_schema_data_type(self) -> Result<LogicalPlan> {
+        match self {
+            // Build `LogicalPlan::Values` from the values for type inference.
+            // We can't use `recompute_schema` because it skips recomputing for
+            // `LogicalPlan::Values`.
+            LogicalPlan::Values(Values { values, schema: _ }) => {
+                LogicalPlanBuilder::values(values)?.build()
+            }
+            // other plans can just use `recompute_schema` directly.
+            plan => plan.recompute_schema(),
+        }
+    }
+
     /// Walk the logical plan, find any `Placeholder` tokens, and return a set of their names.
     pub fn get_parameter_names(&self) -> Result<HashSet<String>> {
         let mut param_names = HashSet::new();