intake more comments

Author: Jiayu Liu

datafusion/src/execution/context.rs

@@ -1275,7 +1275,9 @@ mod tests {
             4,
         )
         .await?;
-        // result in one batch
+        // result in one batch, although e.g. having 2 batches do not change
+        // result semantics, having a len=1 assertion upfront keeps surprises
+        // at bay
         assert_eq!(results.len(), 1);
 
         let expected = vec![

datafusion/src/physical_plan/expressions/row_number.rs

@@ -29,14 +29,12 @@ use std::sync::Arc;
 
 /// row_number expression
 #[derive(Debug)]
-pub struct RowNumber {
-    name: String,
-}
+pub struct RowNumber {}
 
 impl RowNumber {
     /// Create a new ROW_NUMBER function
-    pub fn new(name: String) -> Self {
-        Self { name }
+    pub fn new() -> Self {
+        Self {}
     }
 }
 
@@ -49,15 +47,15 @@ impl BuiltInWindowFunctionExpr for RowNumber {
     fn field(&self) -> Result<Field> {
         let nullable = false;
         let data_type = DataType::UInt64;
-        Ok(Field::new(&self.name, data_type, nullable))
+        Ok(Field::new(&self.name(), data_type, nullable))
     }
 
     fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
         vec![]
     }
 
     fn name(&self) -> &str {
-        &self.name
+        "row_number"
     }
 
     fn create_accumulator(&self) -> Result<Box<dyn WindowAccumulator>> {
@@ -71,7 +69,7 @@ struct RowNumberAccumulator {
 }
 
 impl RowNumberAccumulator {
-    /// new count accumulator
+    /// new row_number accumulator
     pub fn new() -> Self {
         // row number is 1 based
         Self { row_number: 1 }
@@ -91,6 +89,8 @@ impl WindowAccumulator for RowNumberAccumulator {
         _values: &[ArrayRef],
     ) -> Result<Option<ArrayRef>> {
         let new_row_number = self.row_number + (num_rows as u64);
+        // TODO: probably would be nice to have a (optimized) kernel for this at some point to
+        // generate an array like this.
         let result = UInt64Array::from_iter_values(self.row_number..new_row_number);
         self.row_number = new_row_number;
         Ok(Some(Arc::new(result)))
@@ -116,7 +116,7 @@ mod tests {
         let schema = Schema::new(vec![Field::new("arr", DataType::Boolean, false)]);
         let batch = RecordBatch::try_new(Arc::new(schema), vec![arr])?;
 
-        let row_number = Arc::new(RowNumber::new("row_number".to_owned()));
+        let row_number = Arc::new(RowNumber::new());
 
         let mut acc = row_number.create_accumulator()?;
         let expr = row_number.expressions();
@@ -147,7 +147,7 @@ mod tests {
         let schema = Schema::new(vec![Field::new("arr", DataType::Boolean, false)]);
         let batch = RecordBatch::try_new(Arc::new(schema), vec![arr])?;
 
-        let row_number = Arc::new(RowNumber::new("row_number".to_owned()));
+        let row_number = Arc::new(RowNumber::new());
 
         let mut acc = row_number.create_accumulator()?;
         let expr = row_number.expressions();

datafusion/src/physical_plan/windows.rs

@@ -84,10 +84,10 @@ fn create_built_in_window_expr(
     fun: &BuiltInWindowFunction,
     _args: &[Arc<dyn PhysicalExpr>],
     _input_schema: &Schema,
-    name: String,
+    _name: String,
 ) -> Result<Arc<dyn BuiltInWindowFunctionExpr>> {
     match fun {
-        BuiltInWindowFunction::RowNumber => Ok(Arc::new(RowNumber::new(name))),
+        BuiltInWindowFunction::RowNumber => Ok(Arc::new(RowNumber::new())),
         _ => Err(DataFusionError::NotImplemented(format!(
             "Window function with {:?} not yet implemented",
             fun
@@ -330,8 +330,7 @@ fn window_aggregate_batch(
         .collect::<Result<Vec<_>>>()
 }
 
-/// returns a vector of ArrayRefs, where each entry corresponds to either the
-/// final value (mode = Final) or states (mode = Partial)
+/// returns a vector of ArrayRefs, where each entry corresponds to one window expr
 fn finalize_window_aggregation(
     window_accumulators: &[WindowAccumulatorItem],
 ) -> Result<Vec<Option<ScalarValue>>> {

datafusion/tests/sql.rs

@@ -809,7 +809,9 @@ async fn csv_query_window_with_empty_over() -> Result<()> {
     count(c3) over (), \
     max(c3) over (), \
     min(c3) over () \
-    from aggregate_test_100 limit 5";
+    from aggregate_test_100 \
+    order by c1 \
+    limit 5";
     let actual = execute(&mut ctx, sql).await;
     let expected = vec![
         vec!["2", "1", "781", "7.81", "100", "125", "-117"],