Dictionary encoding: Optimize hash aggregation with i64 len prefixes

Author: Rich-T-kid

datafusion/physical-plan/src/aggregates/group_values/single_group_by/dictionary.rs

@@ -40,22 +40,20 @@ macro_rules! decode_list {
                 Some(raw_vector) => {
                     let mut offset = 0;
                     while offset < raw_vector.len() {
-                        let len = i32::from_ne_bytes(
-                            raw_vector[offset..offset + 4]
+                        let len = i64::from_ne_bytes(
+                            raw_vector[offset..offset + 8]
                                 .try_into()
-                                .expect("slice of length 4"),
+                                .expect("slice of length 8"),
                         );
-                        offset += 4;
+                        offset += 8;
                         if len == -1 {
                             builder.values().append_null();
                         } else {
                             let s = std::str::from_utf8(
                                 &raw_vector[offset..offset + len as usize],
                             )
                             .map_err(|e| {
-                                datafusion_common::DataFusionError::Internal(format!(
-                                    "Invalid utf8 in list element: {e}"
-                                ))
+                                Internal(format!("Invalid utf8 in list element: {e}"))
                             })?;
                             builder.values().append_value(s);
                             offset += len as usize;
@@ -75,9 +73,7 @@ macro_rules! decode_scalar_string {
             match raw_bytes {
                 Some(raw_vector) => {
                     let s = std::str::from_utf8(raw_vector).map_err(|e| {
-                        datafusion_common::DataFusionError::Internal(format!(
-                            "Invalid utf8 in GroupValuesDictionary: {e}"
-                        ))
+                        Internal(format!("Invalid utf8 in GroupValuesDictionary: {e}"))
                     })?;
                     builder.append_value(s);
                 }
@@ -163,10 +159,10 @@ impl<K: ArrowDictionaryKeyType + Send> GroupValuesDictionary<K> {
                 for i in start..end {
                     if child.is_null(i) {
                         // acts as a marker for transform_into_array to write a null
-                        bytes.extend_from_slice(&(-1i32).to_ne_bytes());
+                        bytes.extend_from_slice(&(-1i64).to_ne_bytes());
                     } else {
                         let raw = Self::get_raw_bytes(child, i);
-                        bytes.extend_from_slice(&(raw.len() as i32).to_ne_bytes());
+                        bytes.extend_from_slice(&(raw.len() as i64).to_ne_bytes());
                         bytes.extend_from_slice(&raw);
                     }
                 }