Merge branch 'main' into dev

Author: comphead

.asf.yaml

@@ -50,6 +50,9 @@ github:
     main:
       required_pull_request_reviews:
         required_approving_review_count: 1
+  pull_requests:
+    # enable updating head branches of pull requests
+    allow_update_branch: true
 
 # publishes the content of the `asf-site` branch to
 # https://datafusion.apache.org/

Cargo.lock

@@ -48,6 +48,7 @@ members = [
     "datafusion/proto-common",
     "datafusion/proto-common/gen",
     "datafusion/session",
+    "datafusion/spark",
     "datafusion/sql",
     "datafusion/sqllogictest",
     "datafusion/substrait",
@@ -138,6 +139,7 @@ datafusion-physical-plan = { path = "datafusion/physical-plan", version = "47.0.
 datafusion-proto = { path = "datafusion/proto", version = "47.0.0" }
 datafusion-proto-common = { path = "datafusion/proto-common", version = "47.0.0" }
 datafusion-session = { path = "datafusion/session", version = "47.0.0" }
+datafusion-spark = { path = "datafusion/spark", version = "47.0.0" }
 datafusion-sql = { path = "datafusion/sql", version = "47.0.0" }
 doc-comment = "0.3"
 env_logger = "0.11"

Cargo.toml

@@ -1,5 +1,5 @@
 Apache DataFusion
-Copyright 2019-2024 The Apache Software Foundation
+Copyright 2019-2025 The Apache Software Foundation
 
 This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
+The Apache Software Foundation (http://www.apache.org/).

NOTICE.txt

@@ -51,7 +51,7 @@ snmalloc-rs = { version = "0.3", optional = true }
 structopt = { version = "0.3", default-features = false }
 test-utils = { path = "../test-utils/", version = "0.1.0" }
 tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot"] }
-tokio-util = { version = "0.7.14" }
+tokio-util = { version = "0.7.15" }
 
 [dev-dependencies]
 datafusion-proto = { workspace = true }

benchmarks/Cargo.toml

@@ -112,7 +112,7 @@ Results look like
 Note this query is somewhat synthetic as "WatchID" is almost unique (there are a few duplicates)
 
 ```sql
-SELECT "ClientIP", "WatchID",  COUNT(*) c, MIN("ResponseStartTiming") tmin, APPROX_PERCENTILE_CONT("ResponseStartTiming", 0.95) tp95, MAX("ResponseStartTiming") tmax
+SELECT "ClientIP", "WatchID",  COUNT(*) c, MIN("ResponseStartTiming") tmin, APPROX_PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY "ResponseStartTiming") tp95, MAX("ResponseStartTiming") tmax
 FROM 'hits.parquet'
 WHERE "JavaEnable" = 0 -- filters to 32M of 100M rows
 GROUP BY  "ClientIP", "WatchID"
@@ -132,6 +132,7 @@ Results look like
 ```
 
 ### Q6: How many social shares meet complex multi-stage filtering criteria?
+
 **Question**: What is the count of sharing actions from iPhone mobile users on specific social networks, within common timezones, participating in seasonal campaigns, with high screen resolutions and closely matched UTM parameters?
 **Important Query Properties**: Simple filter with high-selectivity, Costly string matching, A large number of filters with high overhead are positioned relatively later in the process
 
@@ -159,6 +160,37 @@ WHERE
 ```
 Result is empty,Since it has already been filtered by `"SocialAction" = 'share'`.
 
+### Q7: Device Resolution and Refresh Behavior Analysis
+
+**Question**: Identify the top 10 WatchIDs with the highest resolution range (min/max "ResolutionWidth") and total refresh count ("IsRefresh") in descending WatchID order
+
+**Important Query Properties**: Primitive aggregation functions, group by single primitive column, high cardinality grouping
+
+```sql
+SELECT "WatchID", MIN("ResolutionWidth") as wmin, MAX("ResolutionWidth") as wmax, SUM("IsRefresh") as srefresh
+FROM hits
+GROUP BY "WatchID"
+ORDER BY "WatchID" DESC
+LIMIT 10;
+```
+
+Results look like
+```
++---------------------+------+------+----------+
+| WatchID             | wmin | wmax | srefresh |
++---------------------+------+------+----------+
+| 9223372033328793741 | 1368 | 1368 | 0        |
+| 9223371941779979288 | 1479 | 1479 | 0        |
+| 9223371906781104763 | 1638 | 1638 | 0        |
+| 9223371803397398692 | 1990 | 1990 | 0        |
+| 9223371799215233959 | 1638 | 1638 | 0        |
+| 9223371785975219972 | 0    | 0    | 0        |
+| 9223371776706839366 | 1368 | 1368 | 0        |
+| 9223371740707848038 | 1750 | 1750 | 0        |
+| 9223371715190479830 | 1368 | 1368 | 0        |
+| 9223371620124912624 | 1828 | 1828 | 0        |
++---------------------+------+------+----------+
+```
 
 ## Data Notes
 

benchmarks/queries/clickbench/README.md

@@ -3,5 +3,6 @@ SELECT COUNT(DISTINCT "HitColor"), COUNT(DISTINCT "BrowserCountry"), COUNT(DISTI
 SELECT "BrowserCountry",  COUNT(DISTINCT "SocialNetwork"), COUNT(DISTINCT "HitColor"), COUNT(DISTINCT "BrowserLanguage"), COUNT(DISTINCT "SocialAction") FROM hits GROUP BY 1 ORDER BY 2 DESC LIMIT 10;
 SELECT "SocialSourceNetworkID", "RegionID", COUNT(*), AVG("Age"), AVG("ParamPrice"), STDDEV("ParamPrice") as s, VAR("ParamPrice")  FROM hits GROUP BY "SocialSourceNetworkID", "RegionID" HAVING s IS NOT NULL ORDER BY s DESC LIMIT 10;
 SELECT "ClientIP", "WatchID",  COUNT(*) c, MIN("ResponseStartTiming") tmin, MEDIAN("ResponseStartTiming") tmed, MAX("ResponseStartTiming") tmax FROM hits WHERE "JavaEnable" = 0  GROUP BY  "ClientIP", "WatchID" HAVING c > 1 ORDER BY tmed DESC LIMIT 10;
-SELECT "ClientIP", "WatchID",  COUNT(*) c, MIN("ResponseStartTiming") tmin, APPROX_PERCENTILE_CONT("ResponseStartTiming", 0.95) tp95, MAX("ResponseStartTiming") tmax FROM 'hits' WHERE "JavaEnable" = 0 GROUP BY  "ClientIP", "WatchID" HAVING c > 1 ORDER BY tp95 DESC LIMIT 10;
+SELECT "ClientIP", "WatchID",  COUNT(*) c, MIN("ResponseStartTiming") tmin, APPROX_PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY "ResponseStartTiming") tp95, MAX("ResponseStartTiming") tmax FROM 'hits' WHERE "JavaEnable" = 0 GROUP BY  "ClientIP", "WatchID" HAVING c > 1 ORDER BY tp95 DESC LIMIT 10;
 SELECT COUNT(*) AS ShareCount FROM hits WHERE "IsMobile" = 1 AND "MobilePhoneModel" LIKE 'iPhone%' AND "SocialAction" = 'share' AND "SocialSourceNetworkID" IN (5, 12) AND "ClientTimeZone" BETWEEN -5 AND 5 AND regexp_match("Referer", '\/campaign\/(spring|summer)_promo') IS NOT NULL AND CASE WHEN split_part(split_part("URL", 'resolution=', 2), '&', 1) ~ '^\d+$' THEN split_part(split_part("URL", 'resolution=', 2), '&', 1)::INT ELSE 0 END > 1920 AND levenshtein(CAST("UTMSource" AS STRING), CAST("UTMCampaign" AS STRING)) < 3;
+SELECT "WatchID", MIN("ResolutionWidth") as wmin, MAX("ResolutionWidth") as wmax, SUM("IsRefresh") as srefresh FROM hits GROUP BY "WatchID" ORDER BY "WatchID" DESC LIMIT 10;

benchmarks/queries/clickbench/extended.sql

@@ -64,7 +64,15 @@ impl Command {
                 let command_batch = all_commands_info();
                 let schema = command_batch.schema();
                 let num_rows = command_batch.num_rows();
-                print_options.print_batches(schema, &[command_batch], now, num_rows)
+                let task_ctx = ctx.task_ctx();
+                let config = &task_ctx.session_config().options().format;
+                print_options.print_batches(
+                    schema,
+                    &[command_batch],
+                    now,
+                    num_rows,
+                    config,
+                )
             }
             Self::ListTables => {
                 exec_and_print(ctx, print_options, "SHOW TABLES".into()).await

datafusion-cli/src/command.rs

@@ -216,7 +216,8 @@ pub(super) async fn exec_and_print(
 ) -> Result<()> {
     let now = Instant::now();
     let task_ctx = ctx.task_ctx();
-    let dialect = &task_ctx.session_config().options().sql_parser.dialect;
+    let options = task_ctx.session_config().options();
+    let dialect = &options.sql_parser.dialect;
     let dialect = dialect_from_str(dialect).ok_or_else(|| {
         plan_datafusion_err!(
             "Unsupported SQL dialect: {dialect}. Available dialects: \
@@ -250,7 +251,9 @@ pub(super) async fn exec_and_print(
             // As the input stream comes, we can generate results.
             // However, memory safety is not guaranteed.
             let stream = execute_stream(physical_plan, task_ctx.clone())?;
-            print_options.print_stream(stream, now).await?;
+            print_options
+                .print_stream(stream, now, &options.format)
+                .await?;
         } else {
             // Bounded stream; collected results size is limited by the maxrows option
             let schema = physical_plan.schema();
@@ -273,9 +276,13 @@ pub(super) async fn exec_and_print(
                 }
                 row_count += curr_num_rows;
             }
-            adjusted
-                .into_inner()
-                .print_batches(schema, &results, now, row_count)?;
+            adjusted.into_inner().print_batches(
+                schema,
+                &results,
+                now,
+                row_count,
+                &options.format,
+            )?;
             reservation.free();
         }
     }

datafusion-cli/src/exec.rs

@@ -177,13 +177,14 @@ async fn main_inner() -> Result<()> {
 
     // set disk limit
     if let Some(disk_limit) = args.disk_limit {
-        let disk_manager = DiskManager::try_new(DiskManagerConfig::NewOs)?;
+        let mut disk_manager = DiskManager::try_new(DiskManagerConfig::NewOs)?;
 
-        let disk_manager = Arc::try_unwrap(disk_manager)
-            .expect("DiskManager should be a single instance")
-            .with_max_temp_directory_size(disk_limit.try_into().unwrap())?;
+        DiskManager::set_arc_max_temp_directory_size(
+            &mut disk_manager,
+            disk_limit.try_into().unwrap(),
+        )?;
 
-        let disk_config = DiskManagerConfig::new_existing(Arc::new(disk_manager));
+        let disk_config = DiskManagerConfig::new_existing(disk_manager);
         rt_builder = rt_builder.with_disk_manager(disk_config);
     }
 
@@ -265,6 +266,11 @@ fn get_session_config(args: &Args) -> Result<SessionConfig> {
         config_options.explain.format = String::from("tree");
     }
 
+    // in the CLI, we want to show NULL values rather the empty strings
+    if env::var_os("DATAFUSION_FORMAT_NULL").is_none() {
+        config_options.format.null = String::from("NULL");
+    }
+
     let session_config =
         SessionConfig::from(config_options).with_information_schema(true);
     Ok(session_config)

datafusion-cli/src/main.rs

@@ -26,7 +26,7 @@ use arrow::datatypes::SchemaRef;
 use arrow::json::{ArrayWriter, LineDelimitedWriter};
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty::pretty_format_batches_with_options;
-use datafusion::common::format::DEFAULT_CLI_FORMAT_OPTIONS;
+use datafusion::config::FormatOptions;
 use datafusion::error::Result;
 
 /// Allow records to be printed in different formats
@@ -110,7 +110,10 @@ fn format_batches_with_maxrows<W: std::io::Write>(
     writer: &mut W,
     batches: &[RecordBatch],
     maxrows: MaxRows,
+    format_options: &FormatOptions,
 ) -> Result<()> {
+    let options: arrow::util::display::FormatOptions = format_options.try_into()?;
+
     match maxrows {
         MaxRows::Limited(maxrows) => {
             // Filter batches to meet the maxrows condition
@@ -131,10 +134,8 @@ fn format_batches_with_maxrows<W: std::io::Write>(
                 }
             }
 
-            let formatted = pretty_format_batches_with_options(
-                &filtered_batches,
-                &DEFAULT_CLI_FORMAT_OPTIONS,
-            )?;
+            let formatted =
+                pretty_format_batches_with_options(&filtered_batches, &options)?;
             if over_limit {
                 let mut formatted_str = format!("{}", formatted);
                 formatted_str = keep_only_maxrows(&formatted_str, maxrows);
@@ -144,8 +145,7 @@ fn format_batches_with_maxrows<W: std::io::Write>(
             }
         }
         MaxRows::Unlimited => {
-            let formatted =
-                pretty_format_batches_with_options(batches, &DEFAULT_CLI_FORMAT_OPTIONS)?;
+            let formatted = pretty_format_batches_with_options(batches, &options)?;
             writeln!(writer, "{}", formatted)?;
         }
     }
@@ -162,6 +162,7 @@ impl PrintFormat {
         batches: &[RecordBatch],
         maxrows: MaxRows,
         with_header: bool,
+        format_options: &FormatOptions,
     ) -> Result<()> {
         // filter out any empty batches
         let batches: Vec<_> = batches
@@ -170,7 +171,7 @@ impl PrintFormat {
             .cloned()
             .collect();
         if batches.is_empty() {
-            return self.print_empty(writer, schema);
+            return self.print_empty(writer, schema, format_options);
         }
 
         match self {
@@ -182,7 +183,7 @@ impl PrintFormat {
                 if maxrows == MaxRows::Limited(0) {
                     return Ok(());
                 }
-                format_batches_with_maxrows(writer, &batches, maxrows)
+                format_batches_with_maxrows(writer, &batches, maxrows, format_options)
             }
             Self::Json => batches_to_json!(ArrayWriter, writer, &batches),
             Self::NdJson => batches_to_json!(LineDelimitedWriter, writer, &batches),
@@ -194,15 +195,17 @@ impl PrintFormat {
         &self,
         writer: &mut W,
         schema: SchemaRef,
+        format_options: &FormatOptions,
     ) -> Result<()> {
         match self {
             // Print column headers for Table format
             Self::Table if !schema.fields().is_empty() => {
+                let format_options: arrow::util::display::FormatOptions =
+                    format_options.try_into()?;
+
                 let empty_batch = RecordBatch::new_empty(schema);
-                let formatted = pretty_format_batches_with_options(
-                    &[empty_batch],
-                    &DEFAULT_CLI_FORMAT_OPTIONS,
-                )?;
+                let formatted =
+                    pretty_format_batches_with_options(&[empty_batch], &format_options)?;
                 writeln!(writer, "{}", formatted)?;
             }
             _ => {}
@@ -644,6 +647,7 @@ mod tests {
                     &self.batches,
                     self.maxrows,
                     with_header,
+                    &FormatOptions::default(),
                 )
                 .unwrap();
             String::from_utf8(buffer).unwrap()

datafusion-cli/src/print_format.rs

@@ -29,6 +29,7 @@ use datafusion::common::DataFusionError;
 use datafusion::error::Result;
 use datafusion::physical_plan::RecordBatchStream;
 
+use datafusion::config::FormatOptions;
 use futures::StreamExt;
 
 #[derive(Debug, Clone, PartialEq, Copy)]
@@ -103,12 +104,19 @@ impl PrintOptions {
         batches: &[RecordBatch],
         query_start_time: Instant,
         row_count: usize,
+        format_options: &FormatOptions,
     ) -> Result<()> {
         let stdout = std::io::stdout();
         let mut writer = stdout.lock();
 
-        self.format
-            .print_batches(&mut writer, schema, batches, self.maxrows, true)?;
+        self.format.print_batches(
+            &mut writer,
+            schema,
+            batches,
+            self.maxrows,
+            true,
+            format_options,
+        )?;
 
         let formatted_exec_details = get_execution_details_formatted(
             row_count,
@@ -132,6 +140,7 @@ impl PrintOptions {
         &self,
         mut stream: Pin<Box<dyn RecordBatchStream>>,
         query_start_time: Instant,
+        format_options: &FormatOptions,
     ) -> Result<()> {
         if self.format == PrintFormat::Table {
             return Err(DataFusionError::External(
@@ -154,6 +163,7 @@ impl PrintOptions {
                 &[batch],
                 MaxRows::Unlimited,
                 with_header,
+                format_options,
             )?;
             with_header = false;
         }

datafusion-cli/src/print_options.rs

@@ -69,6 +69,10 @@ fn init() {
     // can choose the old explain format too
     ["--command", "EXPLAIN FORMAT indent SELECT 123"],
 )]
+#[case::change_format_version(
+    "change_format_version",
+    ["--file", "tests/sql/types_format.sql", "-q"],
+)]
 #[test]
 fn cli_quick_test<'a>(
     #[case] snapshot_name: &'a str,