(datafusion) branch main updated: fix: last value accumulator merge indexing (#23905)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git


The following commit(s) were added to refs/heads/main by this push:
     new cdc412455a fix: last value accumulator merge indexing (#23905)
cdc412455a is described below

commit cdc412455af0af373e03a5a867c8c387fd1e2824
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Wed Jul 29 07:27:31 2026 +0800

    fix: last value accumulator merge indexing (#23905)
    
    ## Which issue does this PR close?
    
    - Related to apache/datafusion-comet#4131.
    
    ## Rationale for this change
    
    `TrivialLastValueAccumulator::merge_batch` filtered out unset states but
    then read row `0` from the remaining value array. As a result, merging
    multiple `LAST_VALUE` partial states returned the first valid state
    instead of the last one.
    
    ## What changes are included in this PR?
    
    - Select the final valid state row when merging unordered `LAST_VALUE`
    states.
    - Extend the existing merge test to verify the evaluated result.
    
    ## Are these changes tested?
    
    Yes:
    
    ```shell
    cargo test -p datafusion-functions-aggregate test_first_last_state_after_merge
---
 datafusion/functions-aggregate/src/first_last.rs | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs
index e2da7ec753..ea45e42e84 100644
--- a/datafusion/functions-aggregate/src/first_last.rs
+++ b/datafusion/functions-aggregate/src/first_last.rs
@@ -1193,7 +1193,7 @@ impl Accumulator for TrivialLastValueAccumulator {
         if let Some(last) = filtered_states.last()
             && !last.is_empty()
         {
-            self.last = ScalarValue::try_from_array(last, 0)?;
+            self.last = ScalarValue::try_from_array(last, last.len() - 1)?;
             self.is_set = true;
         }
         Ok(())
@@ -1525,10 +1525,24 @@ mod tests {
 
         let merged_state = last_accumulator.state()?;
         assert_eq!(merged_state.len(), state1.len());
+        assert_eq!(last_accumulator.evaluate()?, ScalarValue::Int64(Some(10)));
 
         Ok(())
     }
 
+    #[test]
+    fn test_trivial_last_value_merge_all_flags_false() -> Result<()> {
+        let mut acc = TrivialLastValueAccumulator::try_new(&DataType::Int64, false)?;
+        let states: Vec<ArrayRef> = vec![
+            Arc::new(Int64Array::from(vec![None, None])),
+            Arc::new(BooleanArray::from(vec![false, false])),
+        ];
+
+        acc.merge_batch(&states)?;
+        assert_eq!(acc.evaluate()?, ScalarValue::Int64(None));
+        Ok(())
+    }
+
     #[test]
     fn test_first_group_acc() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org