x

Author: forsaken628

src/query/expression/src/aggregate/hash_index.rs

@@ -15,7 +15,6 @@
 use std::fmt::Debug;
 
 use super::payload_row::CompareState;
-use super::CompareItem;
 use super::PartitionedPayload;
 use super::ProbeState;
 use super::RowPtr;
@@ -164,14 +163,9 @@ impl HashIndex {
         row_count: usize,
         mut adapter: impl TableAdapter,
     ) -> usize {
-        for (i, item) in state.no_match_vector[..row_count].iter_mut().enumerate() {
-            let hash = state.group_hashes[i];
-            *item = CompareItem {
-                row: i.into(),
-                salt: Entry::hash_to_salt(hash),
-                slot: self.init_slot(hash),
-                row_ptr: RowPtr::null(),
-            };
+        for (i, row) in state.no_match_vector[..row_count].iter_mut().enumerate() {
+            *row = i.into();
+            state.slots[i] = self.init_slot(state.group_hashes[i]);
         }
 
         let mut new_group_count = 0;
@@ -183,21 +177,18 @@ impl HashIndex {
             let mut no_match_count = 0;
 
             // 1. inject new_group_count, new_entry_count, need_compare_count, no_match_count
-            for item in state.no_match_vector[..remaining_entries].iter_mut() {
-                let (slot, is_new) = self.find_or_insert(item.slot, item.salt);
-                item.slot = slot;
+            for row in state.no_match_vector[..remaining_entries].iter().copied() {
+                let slot = &mut state.slots[row];
+                let is_new;
+
+                let salt = Entry::hash_to_salt(state.group_hashes[row]);
+                (*slot, is_new) = self.find_or_insert(*slot, salt);
 
                 if is_new {
-                    state.empty_vector[new_entry_count] = item.row;
-                    state.slots[new_entry_count] = slot;
+                    state.empty_vector[new_entry_count] = row;
                     new_entry_count += 1;
                 } else {
-                    state.group_compare_vector[need_compare_count] = CompareItem {
-                        row: item.row,
-                        slot: item.slot,
-                        salt: item.salt,
-                        row_ptr: self.mut_entry(slot).get_pointer(),
-                    };
+                    state.group_compare_vector[need_compare_count] = row;
                     need_compare_count += 1;
                 }
             }
@@ -208,40 +199,36 @@ impl HashIndex {
 
                 adapter.append_rows(state, new_entry_count);
 
-                for (i, row) in state.empty_vector[..new_entry_count]
-                    .iter()
-                    .copied()
-                    .enumerate()
-                {
-                    let entry = self.mut_entry(state.slots[i]);
+                for row in state.empty_vector[..new_entry_count].iter().copied() {
+                    let entry = self.mut_entry(state.slots[row]);
                     entry.set_pointer(state.addresses[row]);
                     debug_assert_eq!(entry.get_pointer(), state.addresses[row]);
                 }
             }
 
             // 3. set address of compare vector
             if need_compare_count > 0 {
-                for item in &mut state.group_compare_vector[..need_compare_count] {
-                    let entry = self.mut_entry(item.slot);
+                for row in state.group_compare_vector[..need_compare_count]
+                    .iter()
+                    .copied()
+                {
+                    let entry = self.mut_entry(state.slots[row]);
 
                     debug_assert!(entry.is_occupied());
-                    debug_assert_eq!(
-                        entry.get_salt(),
-                        (state.group_hashes[item.row] >> 48) as u16
-                    );
-                    item.row_ptr = entry.get_pointer();
-                    state.addresses[item.row] = item.row_ptr;
+                    debug_assert_eq!(entry.get_salt(), (state.group_hashes[row] >> 48) as u16);
+                    state.addresses[row] = entry.get_pointer();
                 }
 
                 // 4. compare
                 no_match_count = adapter.compare(state, need_compare_count, no_match_count);
             }
 
             // 5. Linear probing, just increase iter_times
-            for item in &mut state.no_match_vector[..no_match_count] {
-                item.slot += 1;
-                if item.slot >= self.capacity {
-                    item.slot = 0;
+            for row in state.no_match_vector[..no_match_count].iter().copied() {
+                let slot = &mut state.slots[row];
+                *slot += 1;
+                if *slot >= self.capacity {
+                    *slot = 0;
                 }
             }
             remaining_entries = no_match_count;
@@ -269,7 +256,9 @@ impl<'a> TableAdapter for AdapterImpl<'a> {
         need_compare_count: usize,
         no_match_count: usize,
     ) -> usize {
+        // todo: compare hash first if NECESSARY
         CompareState {
+            address: &state.addresses,
             compare: &mut state.group_compare_vector,
             matched: &mut state.match_vector,
             no_matched: &mut state.no_match_vector,
@@ -349,12 +338,11 @@ mod tests {
     impl TableAdapter for &mut TestTableAdapter {
         fn append_rows(&mut self, state: &mut ProbeState, new_entry_count: usize) {
             for row in state.empty_vector[..new_entry_count].iter() {
-                let row_index = row.to_index();
-                let (key, hash) = self.incoming[row_index];
+                let (key, hash) = self.incoming[*row];
                 let value = key + 20;
 
                 self.payload.push((key, hash, value));
-                state.addresses[*row] = self.get_row_ptr(true, row_index);
+                state.addresses[*row] = self.get_row_ptr(true, row.to_usize());
             }
         }
 
@@ -364,18 +352,21 @@ mod tests {
             need_compare_count: usize,
             mut no_match_count: usize,
         ) -> usize {
-            for item in &state.group_compare_vector[..need_compare_count] {
-                let incoming = self.incoming[item.row.to_index()];
+            for row in state.group_compare_vector[..need_compare_count]
+                .iter()
+                .copied()
+            {
+                let incoming = self.incoming[row];
 
-                let (key, hash, _) = self.get_payload(item.row_ptr);
+                let (key, hash, _) = self.get_payload(state.addresses[row]);
 
                 const POINTER_MASK: u64 = 0x0000FFFFFFFFFFFF;
                 assert_eq!(incoming.1 | POINTER_MASK, hash | POINTER_MASK);
                 if incoming.0 == key {
                     continue;
                 }
 
-                state.no_match_vector[no_match_count] = item.clone();
+                state.no_match_vector[no_match_count] = row;
                 no_match_count += 1;
             }
 

src/query/expression/src/aggregate/payload.rs

@@ -250,7 +250,7 @@ impl Payload {
                     for row in select_vector {
                         unsafe {
                             address[*row]
-                                .write_u8(write_offset, bitmap.get_bit(row.to_index()) as u8);
+                                .write_u8(write_offset, bitmap.get_bit(row.to_usize()) as u8);
                         }
                     }
                 }
@@ -291,15 +291,15 @@ impl Payload {
             let (array_layout, padded_size) = layout.repeat(select_vector.len()).unwrap();
             // Bump only allocates but does not drop, so there is no use after free for any item.
             let place = self.arena.alloc_layout(array_layout);
-            for (idx, place) in select_vector
+            for (row, place) in select_vector
                 .iter()
                 .copied()
                 .enumerate()
-                .map(|(i, row)| (row.to_index(), unsafe { place.add(padded_size * i) }))
+                .map(|(i, row)| (row, unsafe { place.add(padded_size * i) }))
             {
                 let place = StateAddr::from(place);
-                address[idx].set_state_addr(&self.row_layout, &place);
-                let page = &mut self.pages[page_index[idx]];
+                address[row].set_state_addr(&self.row_layout, &place);
+                let page = &mut self.pages[page_index[row]];
                 for (aggr, loc) in self.aggrs.iter().zip(states_loc.iter()) {
                     aggr.init_state(AggrState::new(place, loc));
                     page.state_offsets += 1;

src/query/expression/src/aggregate/payload_row.rs

@@ -17,7 +17,6 @@ use databend_common_column::bitmap::Bitmap;
 use databend_common_io::prelude::bincode_deserialize_from_slice;
 use databend_common_io::prelude::bincode_serialize_into_buf;
 
-use super::CompareItem;
 use super::RowID;
 use super::RowLayout;
 use super::RowPtr;
@@ -117,19 +116,19 @@ pub(super) unsafe fn serialize_column_to_rowformat(
                 }
             } else {
                 for row in select_vector {
-                    address[*row].write_u8(offset, v.get_bit(row.to_index()) as u8);
+                    address[*row].write_u8(offset, v.get_bit(row.to_usize()) as u8);
                 }
             }
         }
         Column::Binary(v) | Column::Bitmap(v) | Column::Variant(v) | Column::Geometry(v) => {
             for row in select_vector {
-                let data = arena.alloc_slice_copy(v.index_unchecked(row.to_index()));
+                let data = arena.alloc_slice_copy(v.index_unchecked(row.to_usize()));
                 address[*row].write_bytes(offset, data);
             }
         }
         Column::String(v) => {
             for row in select_vector {
-                let data = arena.alloc_str(v.index_unchecked(row.to_index()));
+                let data = arena.alloc_str(v.index_unchecked(row.to_usize()));
                 address[*row].write_bytes(offset, data.as_bytes());
             }
         }
@@ -150,7 +149,7 @@ pub(super) unsafe fn serialize_column_to_rowformat(
         // for complex column
         other => {
             for row in select_vector {
-                let s = other.index_unchecked(row.to_index()).to_owned();
+                let s = other.index_unchecked(row.to_usize()).to_owned();
                 scratch.clear();
                 bincode_serialize_into_buf(scratch, &s).unwrap();
 
@@ -170,15 +169,16 @@ unsafe fn serialize_fixed_size_column_to_rowformat<T>(
     T: AccessType<Scalar: Copy>,
 {
     for row in select_vector {
-        let val = T::index_column_unchecked_scalar(column, row.to_index());
+        let val = T::index_column_unchecked_scalar(column, row.to_usize());
         address[*row].write(offset, &val);
     }
 }
 
 pub struct CompareState<'a> {
-    pub(super) compare: &'a mut [CompareItem; BATCH_SIZE],
-    pub(super) matched: &'a mut [CompareItem; BATCH_SIZE],
-    pub(super) no_matched: &'a mut [CompareItem; BATCH_SIZE],
+    pub(super) address: &'a [RowPtr; BATCH_SIZE],
+    pub(super) compare: &'a mut [RowID; BATCH_SIZE],
+    pub(super) matched: &'a mut [RowID; BATCH_SIZE],
+    pub(super) no_matched: &'a mut [RowID; BATCH_SIZE],
 }
 
 impl<'s> CompareState<'s> {
@@ -207,6 +207,7 @@ impl<'s> CompareState<'s> {
             );
 
             self = CompareState::<'s> {
+                address: self.address,
                 compare: self.matched,
                 matched: self.compare,
                 no_matched: self.no_matched,
@@ -355,32 +356,32 @@ impl<'s> CompareState<'s> {
         let mut match_count = 0;
         if let Some(validity) = validity {
             let is_all_set = validity.null_count() == 0;
-            for item in &self.compare[..count] {
-                let row = item.row.to_index();
-                let is_set2 = unsafe { item.row_ptr.read::<u8>(validity_offset) != 0 };
-                let is_set = is_all_set || unsafe { validity.get_bit_unchecked(row) };
+            for row in self.compare[..count].iter().copied() {
+                let row_index = row.to_usize();
+                let is_set2 = unsafe { self.address[row].read::<u8>(validity_offset) != 0 };
+                let is_set = is_all_set || unsafe { validity.get_bit_unchecked(row_index) };
 
                 let equal = if is_set && is_set2 {
-                    compare_fn(row, &item.row_ptr)
+                    compare_fn(row_index, &self.address[row])
                 } else {
                     is_set == is_set2
                 };
 
                 if equal {
-                    self.matched[match_count] = item.clone();
+                    self.matched[match_count] = row;
                     match_count += 1;
                 } else {
-                    self.no_matched[no_match_count] = item.clone();
+                    self.no_matched[no_match_count] = row;
                     no_match_count += 1;
                 }
             }
         } else {
-            for item in &self.compare[..count] {
-                if compare_fn(item.row.to_index(), &item.row_ptr) {
-                    self.matched[match_count] = item.clone();
+            for row in self.compare[..count].iter().copied() {
+                if compare_fn(row.to_usize(), &self.address[row]) {
+                    self.matched[match_count] = row;
                     match_count += 1;
                 } else {
-                    self.no_matched[no_match_count] = item.clone();
+                    self.no_matched[no_match_count] = row;
                     no_match_count += 1;
                 }
             }
@@ -396,16 +397,17 @@ impl<'s> CompareState<'s> {
         (count, mut no_match_count): (usize, usize),
     ) -> (usize, usize) {
         let mut match_count = 0;
-        for item in &self.compare[..count] {
-            let value = unsafe { AnyType::index_column_unchecked(col, item.row.to_index()) };
-            let scalar = unsafe { item.row_ptr.read_bytes(col_offset) };
+        for row in self.compare[..count].iter().copied() {
+            let row_index = row.to_usize();
+            let value = unsafe { AnyType::index_column_unchecked(col, row_index) };
+            let scalar = unsafe { self.address[row].read_bytes(col_offset) };
             let scalar: Scalar = bincode_deserialize_from_slice(scalar).unwrap();
 
             if scalar.as_ref() == value {
-                self.matched[match_count] = item.clone();
+                self.matched[match_count] = row;
                 match_count += 1;
             } else {
-                self.no_matched[no_match_count] = item.clone();
+                self.no_matched[no_match_count] = row;
                 no_match_count += 1;
             }
         }