Fix partial aggregation skipping with Decimal aggregators (#11833)

* Fix partial aggregation skipping for Decimal

* fix bug

datafusion/physical-expr-common/src/aggregate/groups_accumulator/prim_op.rs

@@ -184,7 +184,8 @@ where
                     "initial_values underlying buffer must not be shared"
                 )
             })?
-            .map_err(DataFusionError::from)?;
+            .map_err(DataFusionError::from)?
+            .with_data_type(self.data_type.clone());
 
         Ok(vec![Arc::new(state_values)])
     }

datafusion/sqllogictest/test_files/aggregate_skip_partial.slt

@@ -322,3 +322,29 @@ FROM aggregate_test_100_null GROUP BY c2 ORDER BY c2;
 3 109 211 2.80575042963 2.80632930994
 4 -171 56 2.10740506649 1.939846396446
 5 -86 -76 1.8741710186 1.600569307804
+
+
+statement ok
+DROP TABLE aggregate_test_100_null;
+
+# Test for aggregate functions with different intermediate types
+# Need more than 10 values to trigger skipping
+statement ok
+CREATE TABLE decimal_table(i int, d decimal(10,3)) as
+VALUES (1, 1.1), (2, 2.2), (3, 3.3), (2, 4.4), (1, 5.5);
+
+statement ok
+CREATE TABLE t(id int) as values (1), (2), (3), (4), (5), (6), (7), (8), (9), (10);
+
+query IR
+SELECT i, sum(d)
+FROM decimal_table CROSS JOIN t
+GROUP BY i
+ORDER BY i;
+----
+1 66
+2 66
+3 33
+
+statement ok
+DROP TABLE decimal_table;