feat(storage): change the prefix_hint to dist_key_hint for bloom_filter

src/batch/src/executor/insert.rs

@@ -307,7 +307,7 @@ mod tests {
                 epoch,
                 None,
                 ReadOptions {
-                    prefix_hint: None,
+                    dist_key_hint: None,
                     check_bloom_filter: false,
                     ignore_range_tombstone: false,
                     table_id: Default::default(),

src/common/src/util/ordered/serde.rs

@@ -49,6 +49,17 @@ impl OrderedRowSerde {
         }
     }
 
+    pub fn dist_key_serde(&self, start_index: usize, end_index: usize) -> Cow<'_, Self> {
+        if end_index - start_index == self.order_types.len() {
+            Cow::Borrowed(self)
+        } else {
+            Cow::Owned(Self {
+                schema: self.schema[start_index..end_index].to_vec(),
+                order_types: self.order_types[start_index..end_index].to_vec(),
+            })
+        }
+    }
+
     pub fn serialize(&self, row: impl Row2, append_to: &mut Vec<u8>) {
         self.serialize_datums(row.iter(), append_to)
     }
@@ -104,6 +115,34 @@ impl OrderedRowSerde {
 
         Ok(len)
     }
+
+    /// return the distribution key start position in serialized key and the distribution key
+    /// length.
+    pub fn deserialize_dist_key_position_with_column_indices(
+        &self,
+        key: &[u8],
+        column_indices: impl Iterator<Item = usize>,
+        dist_key_start_index: usize,
+    ) -> memcomparable::Result<(usize, usize)> {
+        use crate::types::ScalarImpl;
+        let mut dist_key_start_position: usize = 0;
+        let mut len: usize = 0;
+        for index in column_indices {
+            let data_type = &self.schema[index];
+            let order_type = &self.order_types[index];
+            let data = &key[len..];
+            let mut deserializer = memcomparable::Deserializer::new(data);
+            deserializer.set_reverse(*order_type == OrderType::Descending);
+
+            let field_length = ScalarImpl::encoding_data_size(data_type, &mut deserializer)?;
+            len += field_length;
+            if index < dist_key_start_index {
+                dist_key_start_position += field_length;
+            }
+        }
+
+        Ok((dist_key_start_position, (len - dist_key_start_position)))
+    }
 }
 
 #[cfg(test)]
@@ -237,6 +276,68 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_deserialize_dist_key_position_with_column_indices() {
+        let order_types = vec![
+            OrderType::Descending,
+            OrderType::Ascending,
+            OrderType::Descending,
+            OrderType::Ascending,
+        ];
+
+        let schema = vec![
+            DataType::Varchar,
+            DataType::Int16,
+            DataType::Varchar,
+            DataType::Varchar,
+        ];
+        let serde = OrderedRowSerde::new(schema, order_types);
+        let row1 = Row::new(vec![
+            Some(Utf8("aaa".to_string())),
+            Some(Int16(5)),
+            Some(Utf8("bbb".to_string())),
+            Some(Utf8("ccc".to_string())),
+        ]);
+        let rows = vec![row1];
+        let mut array = vec![];
+        for row in &rows {
+            let mut row_bytes = vec![];
+            serde.serialize(row, &mut row_bytes);
+            array.push(row_bytes);
+        }
+
+        {
+            let dist_key_indices = [1, 2];
+            let dist_key_start_index = 1;
+            let (dist_key_start_position, dist_key_len) = serde
+                .deserialize_dist_key_position_with_column_indices(
+                    &array[0],
+                    0..dist_key_start_index + dist_key_indices.len(),
+                    dist_key_start_index,
+                )
+                .unwrap();
+
+            let schema = vec![DataType::Varchar];
+            let order_types = vec![OrderType::Descending];
+            let deserde = OrderedRowSerde::new(schema, order_types);
+            let prefix_slice = &array[0][0..dist_key_start_position];
+            assert_eq!(
+                deserde.deserialize(prefix_slice).unwrap(),
+                Row::new(vec![Some(Utf8("aaa".to_string()))])
+            );
+
+            let schema = vec![DataType::INT16, DataType::VARCHAR];
+            let order_types = vec![OrderType::Ascending, OrderType::Descending];
+            let deserde = OrderedRowSerde::new(schema, order_types);
+            let dist_key_slice =
+                &array[0][dist_key_start_position..dist_key_start_position + dist_key_len];
+            assert_eq!(
+                deserde.deserialize(dist_key_slice).unwrap(),
+                Row::new(vec![Some(Int16(5)), Some(Utf8("bbb".to_string()))])
+            );
+        }
+    }
+
     #[test]
     fn test_encoding_data_size() {
         use std::mem::size_of;

src/ctl/src/cmd_impl/hummock/list_kv.rs

@@ -34,7 +34,7 @@ pub async fn list_kv(epoch: u64, table_id: u32) -> anyhow::Result<()> {
                 None,
                 ReadOptions {
                     ignore_range_tombstone: false,
-                    prefix_hint: None,
+                    dist_key_hint: None,
                     table_id: TableId { table_id },
                     retention_seconds: None,
                     check_bloom_filter: false,

src/frontend/src/catalog/table_catalog.rs

@@ -411,7 +411,7 @@ mod tests {
                 vnode_col_idx: None,
                 value_indices: vec![0],
                 definition: "".into(),
-                handle_pk_conflict: false
+                handle_pk_conflict: false,
             }
         );
         assert_eq!(table, TableCatalog::from(table.to_prost(0, 0)));

src/frontend/src/optimizer/plan_node/stream_materialize.rs

@@ -168,14 +168,16 @@ impl StreamMaterialize {
 
         let ctx = input.ctx();
         let properties = ctx.inner().with_options.internal_table_subset();
+        let distribution_key = base.dist.dist_column_indices().to_vec();
+
         let table = TableCatalog {
             id: TableId::placeholder(),
             associated_source_id: None,
             name: mv_name,
             columns,
             pk: pk_list,
             stream_key: pk_indices.clone(),
-            distribution_key: base.dist.dist_column_indices().to_vec(),
+            distribution_key,
             is_index,
             appendonly: input.append_only(),
             owner: risingwave_common::catalog::DEFAULT_SUPER_USER_ID,