From ac96b5d6b98fb29b5c5283f0b687370fed0bad63 Mon Sep 17 00:00:00 2001
From: kamille <caoruiqiu.crq@antgroup.com>
Date: Tue, 8 Oct 2024 19:01:46 +0800
Subject: [PATCH] impl equal to.

---
 .../aggregates/group_values/group_column.rs   | 48 +++++++++++++++++--
 1 file changed, 45 insertions(+), 3 deletions(-)

diff --git a/datafusion/physical-plan/src/aggregates/group_values/group_column.rs b/datafusion/physical-plan/src/aggregates/group_values/group_column.rs
index e39afa5ba7396..538349c8522de 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/group_column.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/group_column.rs
@@ -445,7 +445,9 @@ impl ByteGroupValueViewBuilder {
         let value: &[u8] = arr.value(row).as_ref();
 
         let value_len = value.len();
-        let view = if value_len > 12 {
+        let view = if value_len <= 12 {
+            make_view(value, 0, 0)
+        } else {
             // Ensure big enough block to hold the value firstly
             self.ensure_in_progress_big_enough(value_len);
 
@@ -455,8 +457,6 @@ impl ByteGroupValueViewBuilder {
             self.in_progress.extend_from_slice(value);
 
             make_view(value, block_id, offset)
-        } else {
-            make_view(value, 0, 0)
         };
 
         // Append view
@@ -477,6 +477,48 @@ impl ByteGroupValueViewBuilder {
             self.completed.push(buffer);
         }
     }
+
+    fn equal_to_inner<B>(&self, lhs_row: usize, array: &ArrayRef, rhs_row: usize) -> bool
+    where
+        B: ByteViewType,
+    {
+        let array = array.as_byte_view::<B>();
+
+        // Check if nulls equal firstly
+        let exist_null = self.nulls.is_null(lhs_row);
+        let input_null = array.is_null(rhs_row);
+        if let Some(result) = nulls_equal_to(exist_null, input_null) {
+            return result;
+        }
+
+        // Otherwise, we need to check their values
+        let exist_view = self.views[lhs_row];
+        let exist_view_len = exist_view as u32;
+
+        let input_view = array.views()[rhs_row];
+        let input_view_len = exist_view as u32;
+
+        // The check logic
+        //   - Check len equality
+        //   - If non-inlined, check prefix and then check value in buffer
+        //     when needed    
+        //   - If inlined, check inlined value
+        if exist_view_len != input_view_len {
+            return false;
+        }
+
+        if exist_view_len > 12 {
+            let exist_prefix = unsafe { GenericByteViewArray::<B>::inline_value(&exist_view, 4) };
+            let input_prefix = unsafe { GenericByteViewArray::<B>::inline_value(&input_view, 4) };
+
+            if exist_prefix != input_prefix {
+                return false;
+            }
+
+            let exist_full: &[u8] = unsafe { array.value_unchecked(lhs_row).as_ref() };
+            let input_full: &[u8] = unsafe { right.value_unchecked(right_idx).as_ref() };            
+        }
+    }
 }
 
 /// Determines if the nullability of the existing and new input array can be used