Merge branch 'apache:main' into fix_limit_column_ordering_bug

Author: RamakrishnaChilaka

Cargo.lock

@@ -79,7 +79,7 @@ repository = "https://github.com/apache/datafusion"
 # Define Minimum Supported Rust Version (MSRV)
 rust-version = "1.88.0"
 # Define DataFusion version
-version = "52.3.0"
+version = "53.0.0"
 
 [workspace.dependencies]
 # We turn off default-features for some dependencies here so the workspaces which inherit them can
@@ -92,6 +92,13 @@ arrow = { version = "58.1.0", features = [
     "prettyprint",
     "chrono-tz",
 ] }
+arrow-avro = { version = "58.1.0", default-features = false, features = [
+    "deflate",
+    "snappy",
+    "zstd",
+    "bzip2",
+    "xz",
+] }
 arrow-buffer = { version = "58.1.0", default-features = false }
 arrow-flight = { version = "58.1.0", features = [
     "flight-sql-experimental",
@@ -109,43 +116,43 @@ chrono = { version = "0.4.44", default-features = false }
 criterion = "0.8"
 ctor = "0.6.3"
 dashmap = "6.0.1"
-datafusion = { path = "datafusion/core", version = "52.3.0", default-features = false }
-datafusion-catalog = { path = "datafusion/catalog", version = "52.3.0" }
-datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "52.3.0" }
-datafusion-common = { path = "datafusion/common", version = "52.3.0", default-features = false }
-datafusion-common-runtime = { path = "datafusion/common-runtime", version = "52.3.0" }
-datafusion-datasource = { path = "datafusion/datasource", version = "52.3.0", default-features = false }
-datafusion-datasource-arrow = { path = "datafusion/datasource-arrow", version = "52.3.0", default-features = false }
-datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "52.3.0", default-features = false }
-datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "52.3.0", default-features = false }
-datafusion-datasource-json = { path = "datafusion/datasource-json", version = "52.3.0", default-features = false }
-datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "52.3.0", default-features = false }
-datafusion-doc = { path = "datafusion/doc", version = "52.3.0" }
-datafusion-execution = { path = "datafusion/execution", version = "52.3.0", default-features = false }
-datafusion-expr = { path = "datafusion/expr", version = "52.3.0", default-features = false }
-datafusion-expr-common = { path = "datafusion/expr-common", version = "52.3.0" }
-datafusion-ffi = { path = "datafusion/ffi", version = "52.3.0" }
-datafusion-functions = { path = "datafusion/functions", version = "52.3.0" }
-datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "52.3.0" }
-datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "52.3.0" }
-datafusion-functions-nested = { path = "datafusion/functions-nested", version = "52.3.0", default-features = false }
-datafusion-functions-table = { path = "datafusion/functions-table", version = "52.3.0" }
-datafusion-functions-window = { path = "datafusion/functions-window", version = "52.3.0" }
-datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "52.3.0" }
-datafusion-macros = { path = "datafusion/macros", version = "52.3.0" }
-datafusion-optimizer = { path = "datafusion/optimizer", version = "52.3.0", default-features = false }
-datafusion-physical-expr = { path = "datafusion/physical-expr", version = "52.3.0", default-features = false }
-datafusion-physical-expr-adapter = { path = "datafusion/physical-expr-adapter", version = "52.3.0", default-features = false }
-datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "52.3.0", default-features = false }
-datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "52.3.0" }
-datafusion-physical-plan = { path = "datafusion/physical-plan", version = "52.3.0" }
-datafusion-proto = { path = "datafusion/proto", version = "52.3.0" }
-datafusion-proto-common = { path = "datafusion/proto-common", version = "52.3.0" }
-datafusion-pruning = { path = "datafusion/pruning", version = "52.3.0" }
-datafusion-session = { path = "datafusion/session", version = "52.3.0" }
-datafusion-spark = { path = "datafusion/spark", version = "52.3.0" }
-datafusion-sql = { path = "datafusion/sql", version = "52.3.0" }
-datafusion-substrait = { path = "datafusion/substrait", version = "52.3.0" }
+datafusion = { path = "datafusion/core", version = "53.0.0", default-features = false }
+datafusion-catalog = { path = "datafusion/catalog", version = "53.0.0" }
+datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "53.0.0" }
+datafusion-common = { path = "datafusion/common", version = "53.0.0", default-features = false }
+datafusion-common-runtime = { path = "datafusion/common-runtime", version = "53.0.0" }
+datafusion-datasource = { path = "datafusion/datasource", version = "53.0.0", default-features = false }
+datafusion-datasource-arrow = { path = "datafusion/datasource-arrow", version = "53.0.0", default-features = false }
+datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "53.0.0", default-features = false }
+datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "53.0.0", default-features = false }
+datafusion-datasource-json = { path = "datafusion/datasource-json", version = "53.0.0", default-features = false }
+datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "53.0.0", default-features = false }
+datafusion-doc = { path = "datafusion/doc", version = "53.0.0" }
+datafusion-execution = { path = "datafusion/execution", version = "53.0.0", default-features = false }
+datafusion-expr = { path = "datafusion/expr", version = "53.0.0", default-features = false }
+datafusion-expr-common = { path = "datafusion/expr-common", version = "53.0.0" }
+datafusion-ffi = { path = "datafusion/ffi", version = "53.0.0" }
+datafusion-functions = { path = "datafusion/functions", version = "53.0.0" }
+datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "53.0.0" }
+datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "53.0.0" }
+datafusion-functions-nested = { path = "datafusion/functions-nested", version = "53.0.0", default-features = false }
+datafusion-functions-table = { path = "datafusion/functions-table", version = "53.0.0" }
+datafusion-functions-window = { path = "datafusion/functions-window", version = "53.0.0" }
+datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "53.0.0" }
+datafusion-macros = { path = "datafusion/macros", version = "53.0.0" }
+datafusion-optimizer = { path = "datafusion/optimizer", version = "53.0.0", default-features = false }
+datafusion-physical-expr = { path = "datafusion/physical-expr", version = "53.0.0", default-features = false }
+datafusion-physical-expr-adapter = { path = "datafusion/physical-expr-adapter", version = "53.0.0", default-features = false }
+datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "53.0.0", default-features = false }
+datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "53.0.0" }
+datafusion-physical-plan = { path = "datafusion/physical-plan", version = "53.0.0" }
+datafusion-proto = { path = "datafusion/proto", version = "53.0.0" }
+datafusion-proto-common = { path = "datafusion/proto-common", version = "53.0.0" }
+datafusion-pruning = { path = "datafusion/pruning", version = "53.0.0" }
+datafusion-session = { path = "datafusion/session", version = "53.0.0" }
+datafusion-spark = { path = "datafusion/spark", version = "53.0.0" }
+datafusion-sql = { path = "datafusion/sql", version = "53.0.0" }
+datafusion-substrait = { path = "datafusion/substrait", version = "53.0.0" }
 
 doc-comment = "0.3"
 env_logger = "0.11"

Cargo.toml

@@ -39,7 +39,7 @@ use futures::StreamExt;
 #[derive(Debug, Args, Clone)]
 #[command(verbatim_doc_comment)]
 pub struct RunOpt {
-    /// Query number (between 1 and 20). If not specified, runs all queries
+    /// Query number (between 1 and 23). If not specified, runs all queries
     #[arg(short, long)]
     query: Option<usize>,
 
@@ -410,6 +410,52 @@ const SMJ_QUERIES: &[&str] = &[
         FROM t1_sorted JOIN t2_sorted ON t1_sorted.key = t2_sorted.key
         GROUP BY t1_sorted.key
     "#,
+    // Q21: INNER 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key, t1_sorted.data as d1, t2_sorted.data as d2
+        FROM t1_sorted JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
+    // Q22: LEFT 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key, t1_sorted.data as d1, t2_sorted.data as d2
+        FROM t1_sorted LEFT JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
+    // Q23: FULL 10M x 10M | unique keys (1:1) | 50% join filter
+    r#"
+        WITH t1_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        ),
+        t2_sorted AS (
+            SELECT value as key, value as data
+            FROM range(10000000) ORDER BY value
+        )
+        SELECT t1_sorted.key as k1, t1_sorted.data as d1,
+               t2_sorted.key as k2, t2_sorted.data as d2
+        FROM t1_sorted FULL JOIN t2_sorted
+          ON t1_sorted.key = t2_sorted.key
+         AND t1_sorted.data + t2_sorted.data < 10000000
+    "#,
 ];
 
 impl RunOpt {

benchmarks/src/smj.rs

@@ -41,7 +41,6 @@ workspace = true
 name = "datafusion_common"
 
 [features]
-avro = ["apache-avro"]
 backtrace = []
 parquet_encryption = [
     "parquet",
@@ -66,12 +65,6 @@ harness = false
 name = "stats_merge"
 
 [dependencies]
-apache-avro = { workspace = true, features = [
-    "bzip",
-    "snappy",
-    "xz",
-    "zstandard",
-], optional = true }
 arrow = { workspace = true }
 arrow-ipc = { workspace = true }
 chrono = { workspace = true }

datafusion/common/Cargo.toml

@@ -48,8 +48,6 @@ use std::sync::Arc;
 use crate::utils::datafusion_strsim::normalized_levenshtein;
 use crate::utils::quote_identifier;
 use crate::{Column, DFSchema, Diagnostic, TableReference};
-#[cfg(feature = "avro")]
-use apache_avro::Error as AvroError;
 use arrow::error::ArrowError;
 #[cfg(feature = "parquet")]
 use parquet::errors::ParquetError;
@@ -76,9 +74,6 @@ pub enum DataFusionError {
     /// Error when reading / writing Parquet data.
     #[cfg(feature = "parquet")]
     ParquetError(Box<ParquetError>),
-    /// Error when reading Avro data.
-    #[cfg(feature = "avro")]
-    AvroError(Box<AvroError>),
     /// Error when reading / writing to / from an object_store (e.g. S3 or LocalFile)
     #[cfg(feature = "object_store")]
     ObjectStore(Box<object_store::Error>),
@@ -332,13 +327,6 @@ impl From<ParquetError> for DataFusionError {
     }
 }
 
-#[cfg(feature = "avro")]
-impl From<AvroError> for DataFusionError {
-    fn from(e: AvroError) -> Self {
-        DataFusionError::AvroError(Box::new(e))
-    }
-}
-
 #[cfg(feature = "object_store")]
 impl From<object_store::Error> for DataFusionError {
     fn from(e: object_store::Error) -> Self {
@@ -389,8 +377,6 @@ impl Error for DataFusionError {
             DataFusionError::ArrowError(e, _) => Some(e.as_ref()),
             #[cfg(feature = "parquet")]
             DataFusionError::ParquetError(e) => Some(e.as_ref()),
-            #[cfg(feature = "avro")]
-            DataFusionError::AvroError(e) => Some(e.as_ref()),
             #[cfg(feature = "object_store")]
             DataFusionError::ObjectStore(e) => Some(e.as_ref()),
             DataFusionError::IoError(e) => Some(e),
@@ -520,8 +506,6 @@ impl DataFusionError {
             DataFusionError::ArrowError(_, _) => "Arrow error: ",
             #[cfg(feature = "parquet")]
             DataFusionError::ParquetError(_) => "Parquet error: ",
-            #[cfg(feature = "avro")]
-            DataFusionError::AvroError(_) => "Avro error: ",
             #[cfg(feature = "object_store")]
             DataFusionError::ObjectStore(_) => "Object Store error: ",
             DataFusionError::IoError(_) => "IO error: ",
@@ -561,8 +545,6 @@ impl DataFusionError {
             }
             #[cfg(feature = "parquet")]
             DataFusionError::ParquetError(ref desc) => Cow::Owned(desc.to_string()),
-            #[cfg(feature = "avro")]
-            DataFusionError::AvroError(ref desc) => Cow::Owned(desc.to_string()),
             DataFusionError::IoError(ref desc) => Cow::Owned(desc.to_string()),
             #[cfg(feature = "sql")]
             DataFusionError::SQL(ref desc, ref backtrace) => {

datafusion/common/src/error.rs

@@ -43,7 +43,7 @@ nested_expressions = ["datafusion-functions-nested"]
 # This feature is deprecated. Use the `nested_expressions` feature instead.
 array_expressions = ["nested_expressions"]
 # Used to enable the avro format
-avro = ["datafusion-common/avro", "datafusion-datasource-avro"]
+avro = ["datafusion-datasource-avro"]
 backtrace = ["datafusion-common/backtrace"]
 compression = [
     "liblzma",
@@ -247,6 +247,11 @@ harness = false
 name = "parquet_struct_query"
 required-features = ["parquet"]
 
+[[bench]]
+harness = false
+name = "parquet_struct_projection"
+required-features = ["parquet"]
+
 [[bench]]
 harness = false
 name = "range_and_generate_series"

datafusion/core/Cargo.toml

@@ -295,6 +295,39 @@ fn criterion_benchmark(c: &mut Criterion) {
             )
         })
     });
+
+    c.bench_function("string_agg_query_group_by_few_groups", |b| {
+        b.iter(|| {
+            query(
+                ctx.clone(),
+                &rt,
+                "SELECT u64_narrow, string_agg(utf8, ',') \
+                 FROM t GROUP BY u64_narrow",
+            )
+        })
+    });
+
+    c.bench_function("string_agg_query_group_by_mid_groups", |b| {
+        b.iter(|| {
+            query(
+                ctx.clone(),
+                &rt,
+                "SELECT u64_mid, string_agg(utf8, ',') \
+                 FROM t GROUP BY u64_mid",
+            )
+        })
+    });
+
+    c.bench_function("string_agg_query_group_by_many_groups", |b| {
+        b.iter(|| {
+            query(
+                ctx.clone(),
+                &rt,
+                "SELECT u64_wide, string_agg(utf8, ',') \
+                 FROM t GROUP BY u64_wide",
+            )
+        })
+    });
 }
 
 criterion_group!(benches, criterion_benchmark);