Lint fix

Author: Subham Singhal

datafusion/core/examples/h2o_window_topn_bench.rs

@@ -1,3 +1,20 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
 // Standalone H2O groupby Q8 benchmark: PartitionedTopKExec enabled vs disabled
 //
 // Usage:
@@ -10,7 +27,7 @@ use datafusion::prelude::*;
 use datafusion_common::instant::Instant;
 use std::sync::Arc;
 
-use arrow::array::{Int64Array, Float64Array, RecordBatch};
+use arrow::array::{Float64Array, Int64Array, RecordBatch};
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion::datasource::MemTable;
 use rand::rngs::StdRng;
@@ -49,10 +66,7 @@ fn generate_data(num_partitions: i64) -> Arc<MemTable> {
             .collect();
 
         batches.push(
-            RecordBatch::try_new(Arc::clone(&schema), vec![
-                Arc::new(id6),
-                Arc::new(v3),
-            ])
+            RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(id6), Arc::new(v3)])
                 .unwrap(),
         );
     }
@@ -116,4 +130,4 @@ async fn main() {
         }
         println!();
     }
-}
+}

datafusion/core/tests/physical_optimizer/window_topn.rs

@@ -20,24 +20,24 @@
 use std::sync::Arc;
 
 use arrow::datatypes::{DataType, Field, Schema};
-use datafusion_common::config::ConfigOptions;
 use datafusion_common::Result;
 use datafusion_common::ScalarValue;
+use datafusion_common::config::ConfigOptions;
+use datafusion_expr::Operator;
 use datafusion_expr::{WindowFrame, WindowFrameBound, WindowFrameUnits};
 use datafusion_functions_window::row_number::row_number_udwf;
 use datafusion_physical_expr::expressions::{BinaryExpr, Column, col, lit};
 use datafusion_physical_expr::window::StandardWindowExpr;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
-use datafusion_physical_optimizer::window_topn::WindowTopN;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
+use datafusion_physical_optimizer::window_topn::WindowTopN;
 use datafusion_physical_plan::displayable;
 use datafusion_physical_plan::filter::FilterExec;
 use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::windows::{BoundedWindowAggExec, create_udwf_window_expr};
 use datafusion_physical_plan::{ExecutionPlan, InputOrderMode};
-use datafusion_expr::Operator;
 use insta::assert_snapshot;
 
 fn schema() -> Arc<Schema> {
@@ -68,18 +68,17 @@ fn build_window_topn_plan(
     op: Operator,
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let s = schema();
-    let input: Arc<dyn ExecutionPlan> =
-        Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
+    let input: Arc<dyn ExecutionPlan> = Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
 
     // Sort by pk ASC, val ASC
     let ordering = LexOrdering::new(vec![
         PhysicalSortExpr::new_default(col("pk", &s)?).asc(),
         PhysicalSortExpr::new_default(col("val", &s)?).asc(),
-    ]).unwrap();
+    ])
+    .unwrap();
 
-    let sort: Arc<dyn ExecutionPlan> = Arc::new(
-        SortExec::new(ordering.clone(), input).with_preserve_partitioning(true),
-    );
+    let sort: Arc<dyn ExecutionPlan> =
+        Arc::new(SortExec::new(ordering.clone(), input).with_preserve_partitioning(true));
 
     // ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val)
     let partition_by = vec![col("pk", &s)?];
@@ -123,17 +122,15 @@ fn build_window_topn_plan(
 /// Build a plan with no partition-by: ROW_NUMBER() OVER (ORDER BY val)
 fn build_window_topn_no_partition(limit_value: i64) -> Result<Arc<dyn ExecutionPlan>> {
     let s = schema();
-    let input: Arc<dyn ExecutionPlan> =
-        Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
+    let input: Arc<dyn ExecutionPlan> = Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
 
     // Sort by val ASC only (no partition key)
-    let ordering = LexOrdering::new(vec![
-        PhysicalSortExpr::new_default(col("val", &s)?).asc(),
-    ]).unwrap();
+    let ordering =
+        LexOrdering::new(vec![PhysicalSortExpr::new_default(col("val", &s)?).asc()])
+            .unwrap();
 
-    let sort: Arc<dyn ExecutionPlan> = Arc::new(
-        SortExec::new(ordering.clone(), input).with_preserve_partitioning(true),
-    );
+    let sort: Arc<dyn ExecutionPlan> =
+        Arc::new(SortExec::new(ordering.clone(), input).with_preserve_partitioning(true));
 
     // ROW_NUMBER() OVER (ORDER BY val) — no partition by
     let order_by = vec![PhysicalSortExpr::new_default(col("val", &s)?).asc()];
@@ -146,7 +143,7 @@ fn build_window_topn_no_partition(limit_value: i64) -> Result<Arc<dyn ExecutionP
             "row_number".to_string(),
             false,
         )?,
-        &[],       // empty partition_by
+        &[], // empty partition_by
         &order_by,
         Arc::new(WindowFrame::new_bounds(
             WindowFrameUnits::Rows,
@@ -174,17 +171,16 @@ fn build_window_topn_no_partition(limit_value: i64) -> Result<Arc<dyn ExecutionP
 /// Build a plan where filter is on a data column (not window output)
 fn build_non_window_filter_plan() -> Result<Arc<dyn ExecutionPlan>> {
     let s = schema();
-    let input: Arc<dyn ExecutionPlan> =
-        Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
+    let input: Arc<dyn ExecutionPlan> = Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
 
     let ordering = LexOrdering::new(vec![
         PhysicalSortExpr::new_default(col("pk", &s)?).asc(),
         PhysicalSortExpr::new_default(col("val", &s)?).asc(),
-    ]).unwrap();
+    ])
+    .unwrap();
 
-    let sort: Arc<dyn ExecutionPlan> = Arc::new(
-        SortExec::new(ordering.clone(), input).with_preserve_partitioning(true),
-    );
+    let sort: Arc<dyn ExecutionPlan> =
+        Arc::new(SortExec::new(ordering.clone(), input).with_preserve_partitioning(true));
 
     let partition_by = vec![col("pk", &s)?];
     let order_by = vec![PhysicalSortExpr::new_default(col("val", &s)?).asc()];
@@ -257,7 +253,8 @@ fn flipped_3_gteq_rn() -> Result<()> {
         let ordering = LexOrdering::new(vec![
             PhysicalSortExpr::new_default(col("pk", &s)?).asc(),
             PhysicalSortExpr::new_default(col("val", &s)?).asc(),
-        ]).unwrap();
+        ])
+        .unwrap();
 
         let sort: Arc<dyn ExecutionPlan> = Arc::new(
             SortExec::new(ordering.clone(), input).with_preserve_partitioning(true),
@@ -292,7 +289,8 @@ fn flipped_3_gteq_rn() -> Result<()> {
 
         // Flipped: 3 >= rn  (Literal GtEq Column)
         let rn_col = Arc::new(Column::new("row_number", 2));
-        let limit_lit = lit(ScalarValue::UInt64(Some(3)));        let predicate = Arc::new(BinaryExpr::new(limit_lit, Operator::GtEq, rn_col));
+        let limit_lit = lit(ScalarValue::UInt64(Some(3)));
+        let predicate = Arc::new(BinaryExpr::new(limit_lit, Operator::GtEq, rn_col));
         let filter: Arc<dyn ExecutionPlan> =
             Arc::new(FilterExec::try_new(predicate, window)?);
         filter
@@ -313,7 +311,10 @@ fn non_window_column_filter_no_change() -> Result<()> {
     let before = plan_str(plan.as_ref());
     let optimized = optimize(plan)?;
     let after = plan_str(optimized.as_ref());
-    assert_eq!(before, after, "Plan should not change when filter is on data column");
+    assert_eq!(
+        before, after,
+        "Plan should not change when filter is on data column"
+    );
     Ok(())
 }
 
@@ -323,7 +324,10 @@ fn config_disabled_no_change() -> Result<()> {
     let before = plan_str(plan.as_ref());
     let optimized = optimize_disabled(plan)?;
     let after = plan_str(optimized.as_ref());
-    assert_eq!(before, after, "Plan should not change when config is disabled");
+    assert_eq!(
+        before, after,
+        "Plan should not change when config is disabled"
+    );
     Ok(())
 }
 
@@ -345,17 +349,16 @@ fn no_partition_by_no_change() -> Result<()> {
 #[test]
 fn with_projection_between() -> Result<()> {
     let s = schema();
-    let input: Arc<dyn ExecutionPlan> =
-        Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
+    let input: Arc<dyn ExecutionPlan> = Arc::new(PlaceholderRowExec::new(Arc::clone(&s)));
 
     let ordering = LexOrdering::new(vec![
         PhysicalSortExpr::new_default(col("pk", &s)?).asc(),
         PhysicalSortExpr::new_default(col("val", &s)?).asc(),
-    ]).unwrap();
+    ])
+    .unwrap();
 
-    let sort: Arc<dyn ExecutionPlan> = Arc::new(
-        SortExec::new(ordering.clone(), input).with_preserve_partitioning(true),
-    );
+    let sort: Arc<dyn ExecutionPlan> =
+        Arc::new(SortExec::new(ordering.clone(), input).with_preserve_partitioning(true));
 
     let partition_by = vec![col("pk", &s)?];
     let order_by = vec![PhysicalSortExpr::new_default(col("val", &s)?).asc()];
@@ -418,4 +421,4 @@ fn with_projection_between() -> Result<()> {
           PlaceholderRowExec
     "#);
     Ok(())
-}
+}

datafusion/physical-optimizer/src/window_topn.rs

@@ -30,25 +30,25 @@
 //! with `BoundedWindowAggExec → PartitionedTopKExec(fetch=K)`, removing both
 //! the `FilterExec` and `SortExec`.
 //!
-//! See [`PartitionedTopKExec`](datafusion_physical_plan::sorts::partitioned_topk::PartitionedTopKExec)
+//! See [`PartitionedTopKExec`]
 //! for details on the replacement operator.
 
 use std::sync::Arc;
 
 use crate::PhysicalOptimizerRule;
+use arrow::datatypes::DataType;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{Result, ScalarValue};
 use datafusion_expr::Operator;
-use arrow::datatypes::DataType;
 use datafusion_physical_expr::expressions::{BinaryExpr, Column, Literal};
 use datafusion_physical_expr::window::StandardWindowExpr;
+use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::filter::FilterExec;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::sorts::partitioned_topk::PartitionedTopKExec;
 use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::windows::{BoundedWindowAggExec, WindowUDFExpr};
-use datafusion_physical_plan::{ExecutionPlan};
 
 /// Physical optimizer rule that converts per-partition `ROW_NUMBER` top-K
 /// queries into a more efficient plan using [`PartitionedTopKExec`].
@@ -106,9 +106,7 @@ impl WindowTopN {
     /// `FilterExec → [ProjectionExec] → BoundedWindowAggExec → SortExec`
     /// pattern and can be rewritten, or `None` if the node should be
     /// left unchanged.
-    fn try_transform(
-        plan: &Arc<dyn ExecutionPlan>,
-    ) -> Option<Arc<dyn ExecutionPlan>> {
+    fn try_transform(plan: &Arc<dyn ExecutionPlan>) -> Option<Arc<dyn ExecutionPlan>> {
         // Step 1: Match FilterExec at the top
         let filter = plan.downcast_ref::<FilterExec>()?;
 
@@ -159,7 +157,7 @@ impl WindowTopN {
             partition_prefix_len,
             limit_n,
         )
-            .ok()?;
+        .ok()?;
 
         // Step 8: Rebuild window with new child
         let new_window = Arc::clone(&child_as_arc(window_exec))
@@ -203,7 +201,7 @@ impl PhysicalOptimizerRule for WindowTopN {
                 },
             )
         })
-            .data()
+        .data()
     }
 
     fn name(&self) -> &str {
@@ -250,9 +248,9 @@ fn extract_window_limit(
         right.as_any().downcast_ref::<Literal>(),
     ) {
         let n = scalar_to_usize(lit_val.value())?;
-        return match op {
-            &Operator::LtEq => Some((col.index(), n)),
-            &Operator::Lt => Some((col.index(), n - 1)),
+        return match *op {
+            Operator::LtEq => Some((col.index(), n)),
+            Operator::Lt => Some((col.index(), n - 1)),
             _ => None,
         };
     }
@@ -263,9 +261,9 @@ fn extract_window_limit(
         right.as_any().downcast_ref::<Column>(),
     ) {
         let n = scalar_to_usize(lit_val.value())?;
-        return match op {
-            &Operator::GtEq => Some((col.index(), n)),
-            &Operator::Gt => Some((col.index(), n - 1)),
+        return match *op {
+            Operator::GtEq => Some((col.index(), n)),
+            Operator::Gt => Some((col.index(), n - 1)),
             _ => None,
         };
     }
@@ -293,9 +291,7 @@ fn scalar_to_usize(value: &ScalarValue) -> Option<usize> {
 /// Downcasts through `StandardWindowExpr` → `WindowUDFExpr` and checks
 /// that the UDF name is `"row_number"`. Returns `false` for all other
 /// window functions (e.g., `RANK`, `DENSE_RANK`, `SUM`).
-fn is_row_number(
-    expr: &Arc<dyn datafusion_physical_expr::window::WindowExpr>,
-) -> bool {
+fn is_row_number(expr: &Arc<dyn datafusion_physical_expr::window::WindowExpr>) -> bool {
     let Some(swe) = expr.as_any().downcast_ref::<StandardWindowExpr>() else {
         return false;
     };
@@ -331,4 +327,4 @@ fn find_window_below(
     }
 
     None
-}
+}

datafusion/physical-plan/src/sorts/partitioned_topk.rs

@@ -232,7 +232,6 @@ impl PartitionedTopKExec {
     /// The output is sorted by `sort_exprs` (partition keys then order keys),
     /// uses the same partitioning as the input, emits all output at once
     /// (`EmissionType::Final`), and is bounded.
-
     fn compute_properties(
         input: &Arc<dyn ExecutionPlan>,
         sort_exprs: LexOrdering,
@@ -360,9 +359,9 @@ impl ExecutionPlan for PartitionedTopKExec {
                 runtime,
                 metrics_set,
             )
-                .await
+            .await
         })
-            .try_flatten();
+        .try_flatten();
 
         Ok(Box::pin(RecordBatchStreamAdapter::new(
             self.input.schema(),
@@ -406,6 +405,7 @@ fn create_noop_dynamic_filter() -> Arc<RwLock<TopKDynamicFilters>> {
 ///
 /// - Time: O(N log K) where N = total rows, K = fetch
 /// - Memory: O(K × P × row_size) where P = number of distinct partitions
+#[expect(clippy::too_many_arguments)]
 async fn do_partitioned_topk(
     mut input: SendableRecordBatchStream,
     schema: SchemaRef,
@@ -495,4 +495,4 @@ async fn do_partitioned_topk(
         schema,
         futures::stream::iter(output_batches.into_iter().map(Ok)),
     )))
-}
+}