feat: Add SortOptions builder for better Expr::sort API

benchmarks/src/sort_tpch.rs

@@ -342,7 +342,7 @@ impl RunOpt {
         let options = if self.sorted {
             let key_column_name = schema.fields()[0].name();
             options
-                .with_file_sort_order(vec![vec![col(key_column_name).sort(true, false)]])
+                .with_file_sort_order(vec![vec![col(key_column_name).sort().asc().nulls_last()]])
         } else {
             options
         };

benchmarks/src/tpcds/run.rs

@@ -332,7 +332,7 @@ impl RunOpt {
         let options = if self.sorted {
             let key_column_name = schema.fields()[0].name();
             options
-                .with_file_sort_order(vec![vec![col(key_column_name).sort(true, false)]])
+                .with_file_sort_order(vec![vec![col(key_column_name).sort().asc().nulls_last()]])
         } else {
             options
         };

benchmarks/src/tpch/run.rs

@@ -319,7 +319,7 @@ impl RunOpt {
         let options = if self.sorted {
             let key_column_name = schema.fields()[0].name();
             options
-                .with_file_sort_order(vec![vec![col(key_column_name).sort(true, false)]])
+                .with_file_sort_order(vec![vec![col(key_column_name).sort().asc().nulls_last()]])
         } else {
             options
         };

datafusion-cli/src/main.rs

@@ -835,7 +835,7 @@ mod tests {
                 "file_size_bytes",
                 "etag",
             ])?
-            .sort(vec![col("filename").sort(true, false)])?;
+            .sort(vec![col("filename").sort().asc().nulls_last()])?;
         let rbs = df.collect().await?;
         assert_snapshot!(batches_to_string(&rbs),@r"
         +---------------------+-----------+-----------------+------+

datafusion-examples/examples/custom_data_source/file_stream_provider.rs

@@ -180,7 +180,7 @@ mod non_windows {
         ]));
 
         // Specify the ordering:
-        let order = vec![vec![datafusion::logical_expr::col("a1").sort(true, false)]];
+        let order = vec![vec![datafusion::logical_expr::col("a1").sort().asc().nulls_last()]];
 
         let provider = fifo_table(schema.clone(), fifo_path, order.clone());
         ctx.register_table("fifo", provider)?;

datafusion-examples/examples/dataframe/cache_factory.rs

@@ -70,7 +70,7 @@ pub async fn cache_dataframe_with_custom_logic() -> Result<()> {
         .await?;
 
     let df1 = df_cached.clone().filter(col("car").eq(lit("red")))?;
-    let df2 = df1.clone().sort(vec![col("car").sort(true, false)])?;
+    let df2 = df1.clone().sort(vec![col("car").sort().asc().nulls_last()])?;
 
     // should see log for caching only once
     df_cached.show().await?;

datafusion-examples/examples/query_planning/expr_api.rs

@@ -113,7 +113,7 @@ fn expr_fn_demo() -> Result<()> {
     // such as `FIRST_VALUE(price FILTER quantity > 100 ORDER BY ts )
     let agg = first_value
         .call(vec![col("price")])
-        .order_by(vec![col("ts").sort(false, false)])
+        .order_by(vec![col("ts").sort().desc().nulls_last()])
         .filter(col("quantity").gt(lit(100)))
         .build()?; // build the aggregate
     assert_eq!(

datafusion-examples/examples/query_planning/parse_sql_expr.rs

@@ -115,7 +115,7 @@ async fn query_parquet_demo() -> Result<()> {
         )?
         // Directly parsing the SQL text into a sort expression is not supported yet, so
         // construct it programmatically
-        .sort(vec![col("car").sort(false, false)])?
+        .sort(vec![col("car").sort().desc().nulls_last()])?
         .limit(0, Some(1))?;
 
     let result = df.collect().await?;

datafusion-examples/examples/udf/advanced_udwf.rs

@@ -310,7 +310,7 @@ pub async fn advanced_udwf() -> Result<()> {
     let window_expr = smooth_it
         .call(vec![col("speed")]) // smooth_it(speed)
         .partition_by(vec![col("car")]) // PARTITION BY car
-        .order_by(vec![col("time").sort(true, true)]) // ORDER BY time ASC
+        .order_by(vec![col("time").sort().asc().nulls_first()]) // ORDER BY time ASC
         .window_frame(WindowFrame::new(None))
         .build()?;
     let df = ctx.table("cars").await?.window(vec![window_expr])?;

datafusion-examples/examples/udf/simple_udwf.rs

@@ -155,7 +155,7 @@ pub async fn simple_udwf() -> Result<()> {
     let window_expr = smooth_it
         .call(vec![col("speed")]) // smooth_it(speed)
         .partition_by(vec![col("car")]) // PARTITION BY car
-        .order_by(vec![col("time").sort(true, true)]) // ORDER BY time ASC
+        .order_by(vec![col("time").sort().asc().nulls_first()]) // ORDER BY time ASC
         .window_frame(WindowFrame::new(None))
         .build()?;
     let df = ctx.table("cars").await?.window(vec![window_expr])?;

datafusion-examples/src/utils/csv_to_parquet.rs

@@ -70,7 +70,7 @@ impl ParquetTemp {
 /// let parquet_dir = write_csv_to_parquet(&ctx, &csv_path).await?;
 /// let df = ctx.read_parquet(parquet_dir.path_str()?, ParquetReadOptions::default()).await?;
 /// let rows = df
-///    .sort(vec![col("speed").sort(true, true)])?
+///    .sort(vec![col("speed").sort().asc().nulls_first()])?
 ///    .limit(0, Some(5))?;
 /// assert_batches_eq!(
 ///     &[
@@ -146,7 +146,7 @@ mod tests {
             .read_parquet(parquet_dir.path_str()?, ParquetReadOptions::default())
             .await?;
 
-        let rows = df.sort(vec![col("speed").sort(true, true)])?;
+        let rows = df.sort(vec![col("speed").sort().asc().nulls_first()])?;
         assert_batches_eq!(
             &[
                 "+-------+-------+---------------------+",
@@ -198,7 +198,7 @@ mod tests {
             .read_parquet(parquet_dir.path_str()?, ParquetReadOptions::default())
             .await?;
 
-        let rows = df.sort(vec![col("values").sort(true, true)])?;
+        let rows = df.sort(vec![col("values").sort().asc().nulls_first()])?;
         assert_batches_eq!(
             &[
                 "+------------+--------------------------------------+-------------+-------+",

datafusion/catalog-listing/src/options.rs

@@ -248,7 +248,7 @@ impl ListingOptions {
     /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
     ///
     /// // Tell datafusion that the files are sorted by column "a"
-    /// let file_sort_order = vec![vec![col("a").sort(true, true)]];
+    /// let file_sort_order = vec![vec![col("a").sort().asc().nulls_first()]];
     ///
     /// let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()))
     ///     .with_file_sort_order(file_sort_order.clone());

datafusion/core/benches/preserve_file_partitioning.rs

@@ -510,7 +510,7 @@ fn preserve_order_bench(
                  GROUP BY f_dkey \
                  ORDER BY f_dkey";
 
-    let file_sort_order = vec![vec![col("f_dkey").sort(true, false)]];
+    let file_sort_order = vec![vec![col("f_dkey").sort().asc().nulls_last()]];
 
     run_benchmark(
         c,
@@ -643,7 +643,7 @@ fn preserve_order_join_bench(
                  GROUP BY f.f_dkey \
                  ORDER BY f.f_dkey";
 
-    let file_sort_order = vec![vec![col("f_dkey").sort(true, false)]];
+    let file_sort_order = vec![vec![col("f_dkey").sort().asc().nulls_last()]];
 
     run_benchmark(
         c,
@@ -745,8 +745,8 @@ fn preserve_order_window_bench(
                  LIMIT 1000";
 
     let file_sort_order = vec![vec![
-        col("f_dkey").sort(true, false),
-        col("timestamp").sort(true, false),
+        col("f_dkey").sort().asc().nulls_last(),
+        col("timestamp").sort().asc().nulls_last(),
     ]];
 
     run_benchmark(

datafusion/core/benches/sql_planner.rs

@@ -214,7 +214,7 @@ fn register_union_order_table_generic<T>(
 
     // tell DataFusion that the table is sorted by all columns
     let sort_order = (0..num_columns)
-        .map(|i| col(format!("c{i}")).sort(true, true))
+        .map(|i| col(format!("c{i}")).sort().asc().nulls_first())
         .collect::<Vec<_>>();
 
     // create the table

datafusion/core/src/dataframe/mod.rs

@@ -1179,7 +1179,7 @@ impl DataFrame {
     pub fn sort_by(self, expr: Vec<Expr>) -> Result<DataFrame> {
         self.sort(
             expr.into_iter()
-                .map(|e| e.sort(true, false))
+                .map(|e| e.sort().asc().nulls_last())
                 .collect::<Vec<SortExpr>>(),
         )
     }
@@ -1202,8 +1202,8 @@ impl DataFrame {
     ///     .read_csv("tests/data/example_long.csv", CsvReadOptions::new())
     ///     .await?;
     /// let df = df.sort(vec![
-    ///     col("a").sort(false, true), // a DESC, nulls first
-    ///     col("b").sort(true, false), // b ASC, nulls last
+    ///     col("a").sort().desc().nulls_first(), // a DESC, nulls first
+    ///     col("b").sort().asc().nulls_last(), // b ASC, nulls last
     /// ])?;
     /// let expected = vec![
     ///     "+---+---+---+",
@@ -2023,7 +2023,7 @@ impl DataFrame {
     /// // Sort the data by column "b" and write it to a new location
     /// ctx.read_csv("tests/data/example.csv", CsvReadOptions::new())
     ///     .await?
-    ///     .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
+    ///     .sort(vec![col("b").sort().asc().nulls_first()])? // sort by b asc, nulls first
     ///     .write_csv(
     ///         "output.csv",
     ///         DataFrameWriteOptions::new(),
@@ -2097,7 +2097,7 @@ impl DataFrame {
     /// // Sort the data by column "b" and write it to a new location
     /// ctx.read_csv("tests/data/example.csv", CsvReadOptions::new())
     ///     .await?
-    ///     .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
+    ///     .sort(vec![col("b").sort().asc().nulls_first()])? // sort by b asc, nulls first
     ///     .write_json("output.json", DataFrameWriteOptions::new(), None)
     ///     .await?;
     /// # fs::remove_file("output.json")?;

datafusion/core/src/dataframe/parquet.rs

@@ -44,7 +44,7 @@ impl DataFrame {
     /// // Sort the data by column "b" and write it to a new location
     /// ctx.read_csv("tests/data/example.csv", CsvReadOptions::new())
     ///     .await?
-    ///     .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
+    ///     .sort(vec![col("b").sort().asc().nulls_first()])? // sort by b asc, nulls first
     ///     .write_parquet(
     ///         "output.parquet",
     ///         DataFrameWriteOptions::new(),

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

@@ -281,7 +281,7 @@ mod tests {
             ),
             // sort expr, but non column
             (
-                vec![vec![col("int_col").add(lit(1)).sort(true, true)]],
+                vec![vec![col("int_col").add(lit(1)).sort().asc().nulls_first()]],
                 Ok(vec![
                     [PhysicalSortExpr {
                         expr: binary(
@@ -301,7 +301,7 @@ mod tests {
             ),
             // ok with one column
             (
-                vec![vec![col("string_col").sort(true, false)]],
+                vec![vec![col("string_col").sort().asc().nulls_last()]],
                 Ok(vec![
                     [PhysicalSortExpr {
                         expr: physical_col("string_col", &schema).unwrap(),
@@ -316,8 +316,8 @@ mod tests {
             // ok with two columns, different options
             (
                 vec![vec![
-                    col("string_col").sort(true, false),
-                    col("int_col").sort(false, true),
+                    col("string_col").sort().asc().nulls_last(),
+                    col("int_col").sort().desc().nulls_first(),
                 ]],
                 Ok(vec![
                     [

datafusion/core/src/physical_planner.rs

@@ -2926,7 +2926,7 @@ mod tests {
             .filter(col("c7").lt(lit(5_u8)))?
             .project(vec![col("c1"), col("c2")])?
             .aggregate(vec![col("c1")], vec![sum(col("c2"))])?
-            .sort(vec![col("c1").sort(true, true)])?
+            .sort(vec![col("c1").sort().asc().nulls_first()])?
             .limit(3, Some(10))?
             .build()?;
 

datafusion/core/tests/dataframe/dataframe_functions.rs

@@ -411,7 +411,7 @@ async fn test_fn_approx_median() -> Result<()> {
 
 #[tokio::test]
 async fn test_fn_approx_percentile_cont() -> Result<()> {
-    let expr = approx_percentile_cont(col("b").sort(true, false), lit(0.5), None);
+    let expr = approx_percentile_cont(col("b").sort().asc().nulls_last(), lit(0.5), None);
 
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
@@ -426,7 +426,7 @@ async fn test_fn_approx_percentile_cont() -> Result<()> {
     +---------------------------------------------------------------------------+
     ");
 
-    let expr = approx_percentile_cont(col("b").sort(false, false), lit(0.1), None);
+    let expr = approx_percentile_cont(col("b").sort().desc().nulls_last(), lit(0.1), None);
 
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
@@ -447,7 +447,7 @@ async fn test_fn_approx_percentile_cont() -> Result<()> {
         None::<&str>,
         "arg_2".to_string(),
     ));
-    let expr = approx_percentile_cont(col("b").sort(true, false), alias_expr, None);
+    let expr = approx_percentile_cont(col("b").sort().asc().nulls_last(), alias_expr, None);
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
 
@@ -467,7 +467,7 @@ async fn test_fn_approx_percentile_cont() -> Result<()> {
         None::<&str>,
         "arg_2".to_string(),
     ));
-    let expr = approx_percentile_cont(col("b").sort(false, false), alias_expr, None);
+    let expr = approx_percentile_cont(col("b").sort().desc().nulls_last(), alias_expr, None);
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
 
@@ -483,7 +483,7 @@ async fn test_fn_approx_percentile_cont() -> Result<()> {
     );
 
     // with number of centroids set
-    let expr = approx_percentile_cont(col("b").sort(true, false), lit(0.5), Some(lit(2)));
+    let expr = approx_percentile_cont(col("b").sort().asc().nulls_last(), lit(0.5), Some(lit(2)));
 
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
@@ -499,7 +499,7 @@ async fn test_fn_approx_percentile_cont() -> Result<()> {
     ");
 
     let expr =
-        approx_percentile_cont(col("b").sort(false, false), lit(0.1), Some(lit(2)));
+        approx_percentile_cont(col("b").sort().desc().nulls_last(), lit(0.1), Some(lit(2)));
 
     let df = create_test_table().await?;
     let batches = df.aggregate(vec![], vec![expr]).unwrap().collect().await?;
@@ -1327,7 +1327,7 @@ async fn test_count_wildcard() -> Result<()> {
         .unwrap()
         .project(vec![count_all()])
         .unwrap()
-        .sort(vec![count_all().sort(true, false)])
+        .sort(vec![count_all().sort().asc().nulls_last()])
         .unwrap()
         .build()
         .unwrap();