feat: skip invisible row for chunk::get_hash_values

src/batch/src/task/hash_shuffle_channel.rs

@@ -59,6 +59,8 @@ fn generate_hash_values(chunk: &DataChunk, hash_info: &HashInfo) -> BatchResult<
                 .map(|idx| *idx as usize)
                 .collect::<Vec<_>>(),
             hasher_builder,
+            // TODO(st1page): https://github.com/risingwavelabs/risingwave/issues/15705
+            false,
         )
         .iter_mut()
         .map(|hash_value| hash_value.value() as usize % output_count)

src/common/src/array/compact_chunk.rs

@@ -110,7 +110,7 @@ impl StreamChunkCompactor {
             .map(|c| {
                 let hash_values = c
                     .data_chunk()
-                    .get_hash_values(&key_indices, Crc32FastBuilder)
+                    .get_hash_values(&key_indices, Crc32FastBuilder, true)
                     .into_iter()
                     .map(|hash| hash.value())
                     .collect_vec();

src/common/src/array/data_chunk.rs

@@ -344,18 +344,25 @@ impl DataChunk {
         Ok(outputs)
     }
 
-    /// Compute hash values for each row.
+    /// Compute hash values for each row. The number of the returning `HashCodes` is `self.capacity()`.
+    /// When `skip_invisible_row` is true, the `HashCode` for the invisible rows is arbitrary.
     pub fn get_hash_values<H: BuildHasher>(
         &self,
         column_idxes: &[usize],
         hasher_builder: H,
+        skip_invisible_row: bool,
     ) -> Vec<HashCode<H>> {
-        let mut states = Vec::with_capacity(self.capacity());
-        states.resize_with(self.capacity(), || hasher_builder.build_hasher());
+        let len = self.capacity();
+        let mut states = Vec::with_capacity(len);
+        states.resize_with(len, || hasher_builder.build_hasher());
         // Compute hash for the specified columns.
         for column_idx in column_idxes {
             let array = self.column_at(*column_idx);
-            array.hash_vec(&mut states[..]);
+            if skip_invisible_row {
+                array.hash_vec(&mut states[..], self.visibility());
+            } else {
+                array.hash_vec(&mut states[..], &Bitmap::ones(len));
+            }
         }
         finalize_hashers(&states[..])
             .into_iter()

src/common/src/array/mod.rs

@@ -281,10 +281,10 @@ pub trait Array:
         }
     }
 
-    fn hash_vec<H: Hasher>(&self, hashers: &mut [H]) {
+    fn hash_vec<H: Hasher>(&self, hashers: &mut [H], vis: &Bitmap) {
         assert_eq!(hashers.len(), self.len());
-        for (idx, state) in hashers.iter_mut().enumerate() {
-            self.hash_at(idx, state);
+        for idx in vis.iter_ones() {
+            self.hash_at(idx, &mut hashers[idx]);
         }
     }
 
@@ -554,8 +554,8 @@ impl ArrayImpl {
         dispatch_array_variants!(self, inner, { inner.hash_at(idx, state) })
     }
 
-    pub fn hash_vec<H: Hasher>(&self, hashers: &mut [H]) {
-        dispatch_array_variants!(self, inner, { inner.hash_vec(hashers) })
+    pub fn hash_vec<H: Hasher>(&self, hashers: &mut [H], vis: &Bitmap) {
+        dispatch_array_variants!(self, inner, { inner.hash_vec(hashers, vis) })
     }
 
     /// Select some elements from `Array` based on `visibility` bitmap.
@@ -711,6 +711,7 @@ mod test_util {
     use std::hash::{BuildHasher, Hasher};
 
     use super::Array;
+    use crate::buffer::Bitmap;
     use crate::util::iter_util::ZipEqFast;
 
     pub fn hash_finish<H: Hasher>(hashers: &[H]) -> Vec<u64> {
@@ -732,8 +733,9 @@ mod test_util {
                 arr.hash_at(i, state)
             }
         });
+        let vis = Bitmap::ones(len);
         arrs.iter()
-            .for_each(|arr| arr.hash_vec(&mut states_vec[..]));
+            .for_each(|arr| arr.hash_vec(&mut states_vec[..], &vis));
         itertools::cons_tuples(
             expects
                 .iter()

src/common/src/hash/consistent_hash/vnode.rs

@@ -146,7 +146,8 @@ impl VirtualNode {
         }
 
         data_chunk
-            .get_hash_values(keys, Crc32FastBuilder)
+            // TODO(st1page): https://github.com/risingwavelabs/risingwave/issues/15705
+            .get_hash_values(keys, Crc32FastBuilder, false)
             .into_iter()
             .map(|hash| hash.into())
             .collect()

src/common/src/hash/key_v2.rs

@@ -288,7 +288,7 @@ impl<S: KeyStorage, N: NullBitmap> HashKey for HashKeyImpl<S, N> {
     type Bitmap = N;
 
     fn build_many(column_indices: &[usize], data_chunk: &DataChunk) -> Vec<Self> {
-        let hash_codes = data_chunk.get_hash_values(column_indices, XxHash64Builder);
+        let hash_codes = data_chunk.get_hash_values(column_indices, XxHash64Builder, true);
 
         let mut serializers = {
             let buffers = if S::Buffer::alloc() {