48.0.1

Cargo.toml

@@ -76,7 +76,7 @@ repository = "https://github.com/apache/datafusion"
 # Define Minimum Supported Rust Version (MSRV)
 rust-version = "1.82.0"
 # Define DataFusion version
-version = "48.0.0"
+version = "48.0.1"
 
 [workspace.dependencies]
 # We turn off default-features for some dependencies here so the workspaces which inherit them can
@@ -108,39 +108,39 @@ chrono = { version = "0.4.41", default-features = false }
 criterion = "0.5.1"
 ctor = "0.4.0"
 dashmap = "6.0.1"
-datafusion = { path = "datafusion/core", version = "48.0.0", default-features = false }
-datafusion-catalog = { path = "datafusion/catalog", version = "48.0.0" }
-datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "48.0.0" }
-datafusion-common = { path = "datafusion/common", version = "48.0.0", default-features = false }
-datafusion-common-runtime = { path = "datafusion/common-runtime", version = "48.0.0" }
-datafusion-datasource = { path = "datafusion/datasource", version = "48.0.0", default-features = false }
-datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "48.0.0", default-features = false }
-datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "48.0.0", default-features = false }
-datafusion-datasource-json = { path = "datafusion/datasource-json", version = "48.0.0", default-features = false }
-datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "48.0.0", default-features = false }
-datafusion-doc = { path = "datafusion/doc", version = "48.0.0" }
-datafusion-execution = { path = "datafusion/execution", version = "48.0.0" }
-datafusion-expr = { path = "datafusion/expr", version = "48.0.0" }
-datafusion-expr-common = { path = "datafusion/expr-common", version = "48.0.0" }
-datafusion-ffi = { path = "datafusion/ffi", version = "48.0.0" }
-datafusion-functions = { path = "datafusion/functions", version = "48.0.0" }
-datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "48.0.0" }
-datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "48.0.0" }
-datafusion-functions-nested = { path = "datafusion/functions-nested", version = "48.0.0" }
-datafusion-functions-table = { path = "datafusion/functions-table", version = "48.0.0" }
-datafusion-functions-window = { path = "datafusion/functions-window", version = "48.0.0" }
-datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "48.0.0" }
-datafusion-macros = { path = "datafusion/macros", version = "48.0.0" }
-datafusion-optimizer = { path = "datafusion/optimizer", version = "48.0.0", default-features = false }
-datafusion-physical-expr = { path = "datafusion/physical-expr", version = "48.0.0", default-features = false }
-datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "48.0.0", default-features = false }
-datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "48.0.0" }
-datafusion-physical-plan = { path = "datafusion/physical-plan", version = "48.0.0" }
-datafusion-proto = { path = "datafusion/proto", version = "48.0.0" }
-datafusion-proto-common = { path = "datafusion/proto-common", version = "48.0.0" }
-datafusion-session = { path = "datafusion/session", version = "48.0.0" }
-datafusion-spark = { path = "datafusion/spark", version = "48.0.0" }
-datafusion-sql = { path = "datafusion/sql", version = "48.0.0" }
+datafusion = { path = "datafusion/core", version = "48.0.1", default-features = false }
+datafusion-catalog = { path = "datafusion/catalog", version = "48.0.1" }
+datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "48.0.1" }
+datafusion-common = { path = "datafusion/common", version = "48.0.1", default-features = false }
+datafusion-common-runtime = { path = "datafusion/common-runtime", version = "48.0.1" }
+datafusion-datasource = { path = "datafusion/datasource", version = "48.0.1", default-features = false }
+datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "48.0.1", default-features = false }
+datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "48.0.1", default-features = false }
+datafusion-datasource-json = { path = "datafusion/datasource-json", version = "48.0.1", default-features = false }
+datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "48.0.1", default-features = false }
+datafusion-doc = { path = "datafusion/doc", version = "48.0.1" }
+datafusion-execution = { path = "datafusion/execution", version = "48.0.1" }
+datafusion-expr = { path = "datafusion/expr", version = "48.0.1" }
+datafusion-expr-common = { path = "datafusion/expr-common", version = "48.0.1" }
+datafusion-ffi = { path = "datafusion/ffi", version = "48.0.1" }
+datafusion-functions = { path = "datafusion/functions", version = "48.0.1" }
+datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "48.0.1" }
+datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "48.0.1" }
+datafusion-functions-nested = { path = "datafusion/functions-nested", version = "48.0.1" }
+datafusion-functions-table = { path = "datafusion/functions-table", version = "48.0.1" }
+datafusion-functions-window = { path = "datafusion/functions-window", version = "48.0.1" }
+datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "48.0.1" }
+datafusion-macros = { path = "datafusion/macros", version = "48.0.1" }
+datafusion-optimizer = { path = "datafusion/optimizer", version = "48.0.1", default-features = false }
+datafusion-physical-expr = { path = "datafusion/physical-expr", version = "48.0.1", default-features = false }
+datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "48.0.1", default-features = false }
+datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "48.0.1" }
+datafusion-physical-plan = { path = "datafusion/physical-plan", version = "48.0.1" }
+datafusion-proto = { path = "datafusion/proto", version = "48.0.1" }
+datafusion-proto-common = { path = "datafusion/proto-common", version = "48.0.1" }
+datafusion-session = { path = "datafusion/session", version = "48.0.1" }
+datafusion-spark = { path = "datafusion/spark", version = "48.0.1" }
+datafusion-sql = { path = "datafusion/sql", version = "48.0.1" }
 doc-comment = "0.3"
 env_logger = "0.11"
 futures = "0.3"

datafusion-cli/src/functions.rs

@@ -322,7 +322,7 @@ pub struct ParquetMetadataFunc {}
 impl TableFunctionImpl for ParquetMetadataFunc {
     fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
         let filename = match exprs.first() {
-            Some(Expr::Literal(ScalarValue::Utf8(Some(s)))) => s, // single quote: parquet_metadata('x.parquet')
+            Some(Expr::Literal(ScalarValue::Utf8(Some(s)), _)) => s, // single quote: parquet_metadata('x.parquet')
             Some(Expr::Column(Column { name, .. })) => name, // double quote: parquet_metadata("x.parquet")
             _ => {
                 return plan_err!(

datafusion-examples/examples/expr_api.rs

@@ -65,7 +65,7 @@ async fn main() -> Result<()> {
     let expr2 = Expr::BinaryExpr(BinaryExpr::new(
         Box::new(col("a")),
         Operator::Plus,
-        Box::new(Expr::Literal(ScalarValue::Int32(Some(5)))),
+        Box::new(Expr::Literal(ScalarValue::Int32(Some(5)), None)),
     ));
     assert_eq!(expr, expr2);
 

datafusion-examples/examples/optimizer_rule.rs

@@ -171,7 +171,7 @@ fn is_binary_eq(binary_expr: &BinaryExpr) -> bool {
 
 /// Return true if the expression is a literal or column reference
 fn is_lit_or_col(expr: &Expr) -> bool {
-    matches!(expr, Expr::Column(_) | Expr::Literal(_))
+    matches!(expr, Expr::Column(_) | Expr::Literal(_, _))
 }
 
 /// A simple user defined filter function

datafusion-examples/examples/planner_api.rs

@@ -80,9 +80,35 @@ async fn to_physical_plan_in_one_api_demo(
         displayable(physical_plan.as_ref()).indent(false)
     );
 
+    let traversal = extract_node_ids_from_execution_plan_tree(physical_plan.as_ref());
+    let expected_traversal = vec![
+        Some(0),
+        Some(1),
+        Some(2),
+        Some(3),
+        Some(4),
+        Some(5),
+        Some(6),
+        Some(7),
+        Some(8),
+        Some(9),
+    ];
+    assert_eq!(expected_traversal, traversal);
     Ok(())
 }
 
+fn extract_node_ids_from_execution_plan_tree(
+    physical_plan: &dyn ExecutionPlan,
+) -> Vec<Option<usize>> {
+    let mut traversed_nodes: Vec<Option<usize>> = vec![];
+    for child in physical_plan.children() {
+        let node_ids = extract_node_ids_from_execution_plan_tree(child.as_ref());
+        traversed_nodes.extend(node_ids);
+    }
+    traversed_nodes.push(physical_plan.properties().node_id());
+    traversed_nodes
+}
+
 /// Converts a logical plan into a physical plan by utilizing the analyzer,
 /// optimizer, and query planner APIs separately. This flavor gives more
 /// control over the planning process.

datafusion-examples/examples/simple_udtf.rs

@@ -133,7 +133,8 @@ struct LocalCsvTableFunc {}
 
 impl TableFunctionImpl for LocalCsvTableFunc {
     fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
-        let Some(Expr::Literal(ScalarValue::Utf8(Some(ref path)))) = exprs.first() else {
+        let Some(Expr::Literal(ScalarValue::Utf8(Some(ref path)), _)) = exprs.first()
+        else {
             return plan_err!("read_csv requires at least one string argument");
         };
 
@@ -145,7 +146,7 @@ impl TableFunctionImpl for LocalCsvTableFunc {
                 let info = SimplifyContext::new(&execution_props);
                 let expr = ExprSimplifier::new(info).simplify(expr.clone())?;
 
-                if let Expr::Literal(ScalarValue::Int64(Some(limit))) = expr {
+                if let Expr::Literal(ScalarValue::Int64(Some(limit)), _) = expr {
                     Ok(limit as usize)
                 } else {
                     plan_err!("Limit must be an integer")

datafusion/catalog-listing/src/helpers.rs

@@ -61,7 +61,7 @@ pub fn expr_applicable_for_cols(col_names: &[&str], expr: &Expr) -> bool {
                 Ok(TreeNodeRecursion::Stop)
             }
         }
-        Expr::Literal(_)
+        Expr::Literal(_, _)
         | Expr::Alias(_)
         | Expr::OuterReferenceColumn(_, _)
         | Expr::ScalarVariable(_, _)
@@ -346,8 +346,8 @@ fn populate_partition_values<'a>(
     {
         match op {
             Operator::Eq => match (left.as_ref(), right.as_ref()) {
-                (Expr::Column(Column { ref name, .. }), Expr::Literal(val))
-                | (Expr::Literal(val), Expr::Column(Column { ref name, .. })) => {
+                (Expr::Column(Column { ref name, .. }), Expr::Literal(val, _))
+                | (Expr::Literal(val, _), Expr::Column(Column { ref name, .. })) => {
                     if partition_values
                         .insert(name, PartitionValue::Single(val.to_string()))
                         .is_some()
@@ -984,7 +984,7 @@ mod tests {
         assert_eq!(
             evaluate_partition_prefix(
                 partitions,
-                &[col("a").eq(Expr::Literal(ScalarValue::Date32(Some(3))))],
+                &[col("a").eq(Expr::Literal(ScalarValue::Date32(Some(3)), None))],
             ),
             Some(Path::from("a=1970-01-04")),
         );
@@ -993,9 +993,10 @@ mod tests {
         assert_eq!(
             evaluate_partition_prefix(
                 partitions,
-                &[col("a").eq(Expr::Literal(ScalarValue::Date64(Some(
-                    4 * 24 * 60 * 60 * 1000
-                )))),],
+                &[col("a").eq(Expr::Literal(
+                    ScalarValue::Date64(Some(4 * 24 * 60 * 60 * 1000)),
+                    None
+                )),],
             ),
             Some(Path::from("a=1970-01-05")),
         );

datafusion/common/src/config.rs

@@ -294,8 +294,8 @@ config_namespace! {
 
         /// Should DataFusion collect statistics when first creating a table.
         /// Has no effect after the table is created. Applies to the default
-        /// `ListingTableProvider` in DataFusion. Defaults to false.
-        pub collect_statistics: bool, default = false
+        /// `ListingTableProvider` in DataFusion. Defaults to true.
+        pub collect_statistics: bool, default = true
 
         /// Number of partitions for query execution. Increasing partitions can increase
         /// concurrency.
@@ -457,7 +457,7 @@ config_namespace! {
 
         /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
         /// and `Binary/BinaryLarge` with `BinaryView`.
-        pub schema_force_view_types: bool, default = true
+        pub schema_force_view_types: bool, default = false
 
         /// (reading) If true, parquet reader will read columns of
         /// `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`.

datafusion/core/benches/map_query_sql.rs

@@ -71,8 +71,11 @@ fn criterion_benchmark(c: &mut Criterion) {
     let mut value_buffer = Vec::new();
 
     for i in 0..1000 {
-        key_buffer.push(Expr::Literal(ScalarValue::Utf8(Some(keys[i].clone()))));
-        value_buffer.push(Expr::Literal(ScalarValue::Int32(Some(values[i]))));
+        key_buffer.push(Expr::Literal(
+            ScalarValue::Utf8(Some(keys[i].clone())),
+            None,
+        ));
+        value_buffer.push(Expr::Literal(ScalarValue::Int32(Some(values[i])), None));
     }
     c.bench_function("map_1000_1", |b| {
         b.iter(|| {

datafusion/core/src/dataframe/mod.rs

@@ -1337,7 +1337,10 @@ impl DataFrame {
     /// ```
     pub async fn count(self) -> Result<usize> {
         let rows = self
-            .aggregate(vec![], vec![count(Expr::Literal(COUNT_STAR_EXPANSION))])?
+            .aggregate(
+                vec![],
+                vec![count(Expr::Literal(COUNT_STAR_EXPANSION, None))],
+            )?
             .collect()
             .await?;
         let len = *rows

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

@@ -2230,7 +2230,7 @@ mod tests {
         let filter_predicate = Expr::BinaryExpr(BinaryExpr::new(
             Box::new(Expr::Column("column1".into())),
             Operator::GtEq,
-            Box::new(Expr::Literal(ScalarValue::Int32(Some(0)))),
+            Box::new(Expr::Literal(ScalarValue::Int32(Some(0)), None)),
         ));
 
         // Create a new batch of data to insert into the table

datafusion/core/src/execution/context/mod.rs

@@ -1214,7 +1214,7 @@ impl SessionContext {
         let mut params: Vec<ScalarValue> = parameters
             .into_iter()
             .map(|e| match e {
-                Expr::Literal(scalar) => Ok(scalar),
+                Expr::Literal(scalar, _) => Ok(scalar),
                 _ => not_impl_err!("Unsupported parameter type: {}", e),
             })
             .collect::<Result<_>>()?;

datafusion/core/src/execution/context/parquet.rs

@@ -34,13 +34,12 @@ impl SessionContext {
     ///
     /// # Note: Statistics
     ///
-    /// NOTE: by default, statistics are not collected when reading the Parquet
-    /// files as this can slow down the initial DataFrame creation. However,
-    /// collecting statistics can greatly accelerate queries with certain
-    /// filters.
+    /// NOTE: by default, statistics are collected when reading the Parquet
+    /// files This can slow down the initial DataFrame creation while
+    /// greatly accelerating queries with certain filters.
     ///
-    /// To enable collect statistics, set the [config option]
-    /// `datafusion.execution.collect_statistics` to `true`. See
+    /// To disable statistics collection, set the [config option]
+    /// `datafusion.execution.collect_statistics` to `false`. See
     /// [`ConfigOptions`] and [`ExecutionOptions::collect_statistics`] for more
     /// details.
     ///
@@ -171,28 +170,28 @@ mod tests {
 
     #[tokio::test]
     async fn register_parquet_respects_collect_statistics_config() -> Result<()> {
-        // The default is false
+        // The default is true
         let mut config = SessionConfig::new();
         config.options_mut().explain.physical_plan_only = true;
         config.options_mut().explain.show_statistics = true;
         let content = explain_query_all_with_config(config).await?;
-        assert_contains!(content, "statistics=[Rows=Absent,");
+        assert_contains!(content, "statistics=[Rows=Exact(");
 
-        // Explicitly set to false
+        // Explicitly set to true
         let mut config = SessionConfig::new();
         config.options_mut().explain.physical_plan_only = true;
         config.options_mut().explain.show_statistics = true;
-        config.options_mut().execution.collect_statistics = false;
+        config.options_mut().execution.collect_statistics = true;
         let content = explain_query_all_with_config(config).await?;
-        assert_contains!(content, "statistics=[Rows=Absent,");
+        assert_contains!(content, "statistics=[Rows=Exact(");
 
-        // Explicitly set to true
+        // Explicitly set to false
         let mut config = SessionConfig::new();
         config.options_mut().explain.physical_plan_only = true;
         config.options_mut().explain.show_statistics = true;
-        config.options_mut().execution.collect_statistics = true;
+        config.options_mut().execution.collect_statistics = false;
         let content = explain_query_all_with_config(config).await?;
-        assert_contains!(content, "statistics=[Rows=Exact(10),");
+        assert_contains!(content, "statistics=[Rows=Absent,");
 
         Ok(())
     }

datafusion/core/src/execution/session_state.rs

@@ -67,6 +67,9 @@ use datafusion_physical_expr::create_physical_expr;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_optimizer::optimizer::PhysicalOptimizer;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
+use datafusion_physical_plan::node_id::{
+    annotate_node_id_for_execution_plan, NodeIdAnnotator,
+};
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_session::Session;
 use datafusion_sql::parser::{DFParserBuilder, Statement};
@@ -654,9 +657,12 @@ impl SessionState {
         logical_plan: &LogicalPlan,
     ) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
         let logical_plan = self.optimize(logical_plan)?;
-        self.query_planner
+        let physical_plan = self
+            .query_planner
             .create_physical_plan(&logical_plan, self)
-            .await
+            .await?;
+        let mut id_annotator = NodeIdAnnotator::new();
+        annotate_node_id_for_execution_plan(&physical_plan, &mut id_annotator)
     }
 
     /// Create a [`PhysicalExpr`] from an [`Expr`] after applying type