refactor(storage): use new higher rank lifetime assumption to avoid unnecessary clone

.cargo/config.toml

@@ -31,7 +31,10 @@ rustflags = [
 
 # Flags for all targets.
 [target.'cfg(all())']
-rustflags = ["--cfg", "tokio_unstable"]
+rustflags = ["--cfg", "tokio_unstable", "-Zhigher-ranked-assumptions"]
+
+[build]
+rustdocflags = ["-Zhigher-ranked-assumptions"]
 
 # We have large git dependencies. This can make cloning faster.
 # https://doc.rust-lang.org/nightly/cargo/reference/unstable.html#git

ci/scripts/doc.sh

@@ -9,14 +9,14 @@ echo "--- Set openssl static link env vars"
 configure_static_openssl
 
 echo "--- Build documentation"
-RUSTDOCFLAGS="-Dwarnings" cargo doc --document-private-items --no-deps
+RUSTDOCFLAGS="-Dwarnings -Zhigher-ranked-assumptions" cargo doc --document-private-items --no-deps
 
 echo "--- Show sccache stats"
 sccache --show-stats
 sccache --zero-stats
 
 echo "--- Run doctest"
-RUSTDOCFLAGS="-Clink-arg=-fuse-ld=lld" cargo test --doc
+cargo test --doc
 
 echo "--- Show sccache stats"
 sccache --show-stats

src/storage/hummock_test/src/hummock_vector_tests.rs

@@ -82,7 +82,7 @@ async fn test_flat_vector() {
 
     let epoch1_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch1_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 
@@ -95,7 +95,7 @@ async fn test_flat_vector() {
 
     let epoch2_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch2_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 
@@ -152,7 +152,7 @@ async fn test_flat_vector() {
 
         let output = read_snapshot_epoch
             .nearest(
-                query.clone(),
+                query.to_ref(),
                 VectorNearestOptions {
                     top_n,
                     measure: DistanceMeasurement::InnerProduct,
@@ -189,7 +189,7 @@ async fn test_flat_vector() {
     vector_writer.init_for_test(epoch3).await.unwrap();
     let epoch3_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch3_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 
@@ -264,7 +264,7 @@ async fn test_hnsw_vector() {
 
     let epoch1_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch1_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 
@@ -277,7 +277,7 @@ async fn test_hnsw_vector() {
 
     let epoch2_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch2_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 
@@ -334,7 +334,7 @@ async fn test_hnsw_vector() {
             let top_n = 10;
             let output = read_snapshot_epoch
                 .nearest(
-                    query.clone(),
+                    query.to_ref(),
                     VectorNearestOptions {
                         top_n,
                         measure: DistanceMeasurement::InnerProduct,
@@ -374,7 +374,7 @@ async fn test_hnsw_vector() {
     vector_writer.init_for_test(epoch3).await.unwrap();
     let epoch3_vectors = (0..100).map(|_| next_input()).collect_vec();
     for (vec, info) in &epoch3_vectors {
-        vector_writer.insert(vec.clone(), info.clone()).unwrap();
+        vector_writer.insert(vec.to_ref(), info.clone()).unwrap();
         vector_writer.try_flush().await.unwrap();
     }
 

src/storage/src/hummock/store/hummock_storage.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use arc_swap::ArcSwap;
 use bytes::Bytes;
 use itertools::Itertools;
+use risingwave_common::array::VectorRef;
 use risingwave_common::catalog::TableId;
 use risingwave_common::dispatch_distance_measurement;
 use risingwave_common::util::epoch::is_max_epoch;
@@ -260,11 +261,11 @@ impl HummockStorageReadSnapshot {
     /// If `Ok(Some())` is returned, the key is found. If `Ok(None)` is returned,
     /// the key is not found. If `Err()` is returned, the searching for the key
     /// failed due to other non-EOF errors.
-    async fn get_inner<O>(
-        &self,
+    async fn get_inner<'a, O>(
+        &'a self,
         key: TableKey<Bytes>,
         read_options: ReadOptions,
-        on_key_value_fn: impl KeyValueFn<O>,
+        on_key_value_fn: impl KeyValueFn<'a, O>,
     ) -> StorageResult<Option<O>> {
         let key_range = (Bound::Included(key.clone()), Bound::Included(key.clone()));
 
@@ -639,12 +640,12 @@ pub struct HummockStorageReadSnapshot {
 }
 
 impl StateStoreGet for HummockStorageReadSnapshot {
-    fn on_key_value<O: Send + 'static>(
-        &self,
+    fn on_key_value<'a, O: Send + 'a>(
+        &'a self,
         key: TableKey<Bytes>,
         read_options: ReadOptions,
-        on_key_value_fn: impl KeyValueFn<O>,
-    ) -> impl StorageFuture<'_, Option<O>> {
+        on_key_value_fn: impl KeyValueFn<'a, O>,
+    ) -> impl StorageFuture<'a, Option<O>> {
         self.get_inner(key, read_options, on_key_value_fn)
     }
 }
@@ -687,11 +688,11 @@ impl StateStoreRead for HummockStorageReadSnapshot {
 }
 
 impl StateStoreReadVector for HummockStorageReadSnapshot {
-    async fn nearest<O: Send + 'static>(
-        &self,
-        vec: Vector,
+    async fn nearest<'a, O: Send + 'a>(
+        &'a self,
+        vec: VectorRef<'a>,
         options: VectorNearestOptions,
-        on_nearest_item_fn: impl OnNearestItemFn<O>,
+        on_nearest_item_fn: impl OnNearestItemFn<'a, O>,
     ) -> StorageResult<Vec<O>> {
         let version = match self.epoch {
             HummockReadEpoch::Committed(epoch)

src/storage/src/hummock/store/local_hummock_storage.rs

@@ -107,12 +107,12 @@ pub struct LocalHummockStorage {
 }
 
 impl LocalHummockFlushedSnapshotReader {
-    async fn get_flushed<O>(
-        hummock_version_reader: &HummockVersionReader,
+    async fn get_flushed<'a, O>(
+        hummock_version_reader: &'a HummockVersionReader,
         read_version: &HummockReadVersionRef,
         user_key: UserKey<Bytes>,
         read_options: ReadOptions,
-        on_key_value_fn: impl crate::store::KeyValueFn<O>,
+        on_key_value_fn: impl KeyValueFn<'a, O>,
     ) -> StorageResult<Option<O>> {
         let table_key_range = (
             Bound::Included(user_key.table_key.clone()),
@@ -266,11 +266,11 @@ pub struct LocalHummockFlushedSnapshotReader {
 }
 
 impl StateStoreGet for LocalHummockFlushedSnapshotReader {
-    async fn on_key_value<O: Send + 'static>(
-        &self,
+    async fn on_key_value<'a, O: Send + 'a>(
+        &'a self,
         key: TableKey<Bytes>,
         read_options: ReadOptions,
-        on_key_value_fn: impl KeyValueFn<O>,
+        on_key_value_fn: impl KeyValueFn<'a, O>,
     ) -> StorageResult<Option<O>> {
         let key = UserKey::new(self.table_id, key);
         Self::get_flushed(
@@ -308,11 +308,11 @@ impl StateStoreRead for LocalHummockFlushedSnapshotReader {
 }
 
 impl StateStoreGet for LocalHummockStorage {
-    async fn on_key_value<O: Send + 'static>(
-        &self,
+    async fn on_key_value<'a, O: Send + 'a>(
+        &'a self,
         key: TableKey<Bytes>,
         read_options: ReadOptions,
-        on_key_value_fn: impl KeyValueFn<O>,
+        on_key_value_fn: impl KeyValueFn<'a, O>,
     ) -> StorageResult<Option<O>> {
         let key = UserKey::new(self.table_id, key);
         match self.mem_table.buffer.get(&key.table_key) {

src/storage/src/hummock/store/vector_writer.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use risingwave_common::array::VectorRef;
 use risingwave_common::catalog::TableId;
 use risingwave_common::util::epoch::EpochPair;
 use risingwave_hummock_sdk::HummockEpoch;
@@ -171,7 +172,7 @@ impl StateStoreWriteEpochControl for HummockVectorWriter {
 }
 
 impl StateStoreWriteVector for HummockVectorWriter {
-    fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()> {
+    fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()> {
         Ok(self
             .state
             .as_mut()

src/storage/src/hummock/store/version.rs

@@ -22,6 +22,7 @@ use bytes::Bytes;
 use futures::future::try_join_all;
 use itertools::Itertools;
 use parking_lot::RwLock;
+use risingwave_common::array::VectorRef;
 use risingwave_common::bitmap::Bitmap;
 use risingwave_common::catalog::TableId;
 use risingwave_common::hash::VirtualNode;
@@ -67,7 +68,7 @@ use crate::monitor::{
     GetLocalMetricsGuard, HummockStateStoreMetrics, IterLocalMetricsGuard, StoreLocalStatistic,
 };
 use crate::store::{
-    OnNearestItemFn, ReadLogOptions, ReadOptions, Vector, VectorNearestOptions, gen_min_epoch,
+    OnNearestItemFn, ReadLogOptions, ReadOptions, VectorNearestOptions, gen_min_epoch,
 };
 use crate::vector::hnsw::nearest;
 use crate::vector::{MeasureDistanceBuilder, NearestBuilder};
@@ -588,14 +589,14 @@ impl HummockVersionReader {
 const SLOW_ITER_FETCH_META_DURATION_SECOND: f64 = 5.0;
 
 impl HummockVersionReader {
-    pub async fn get<O>(
-        &self,
+    pub async fn get<'a, O>(
+        &'a self,
         table_key: TableKey<Bytes>,
         epoch: u64,
         table_id: TableId,
         read_options: ReadOptions,
         read_version_tuple: ReadVersionTuple,
-        on_key_value_fn: impl crate::store::KeyValueFn<O>,
+        on_key_value_fn: impl crate::store::KeyValueFn<'a, O>,
     ) -> StorageResult<Option<O>> {
         let (imms, uncommitted_ssts, committed_version) = read_version_tuple;
 
@@ -1201,13 +1202,13 @@ impl HummockVersionReader {
         .await
     }
 
-    pub async fn nearest<M: MeasureDistanceBuilder, O: Send>(
-        &self,
+    pub async fn nearest<'a, M: MeasureDistanceBuilder, O: Send>(
+        &'a self,
         version: PinnedVersion,
         table_id: TableId,
-        target: Vector,
+        target: VectorRef<'a>,
         options: VectorNearestOptions,
-        on_nearest_item_fn: impl OnNearestItemFn<O>,
+        on_nearest_item_fn: impl OnNearestItemFn<'a, O>,
     ) -> HummockResult<Vec<O>> {
         let Some(index) = version.vector_indexes.get(&table_id) else {
             return Ok(vec![]);
@@ -1221,7 +1222,7 @@ impl HummockVersionReader {
         }
         match &index.inner {
             VectorIndexImpl::Flat(flat) => {
-                let mut builder = NearestBuilder::<'_, O, M>::new(target.to_ref(), options.top_n);
+                let mut builder = NearestBuilder::<'_, O, M>::new(target, options.top_n);
                 for vector_file in &flat.vector_store_info.vector_files {
                     let meta = self.sstable_store.get_vector_file_meta(vector_file).await?;
                     for (i, block_meta) in meta.block_metas.iter().enumerate() {
@@ -1246,7 +1247,7 @@ impl HummockVersionReader {
                 let (items, _stats) = nearest::<O, M>(
                     &vector_store,
                     &*graph,
-                    target.to_ref(),
+                    target,
                     on_nearest_item_fn,
                     options.hnsw_ef_search,
                     options.top_n,

src/storage/src/hummock/vector/writer/hnsw.rs

@@ -18,6 +18,7 @@ use bytes::{Bytes, BytesMut};
 use prost::Message;
 use rand::SeedableRng;
 use rand::rngs::StdRng;
+use risingwave_common::array::VectorRef;
 use risingwave_common::dispatch_distance_measurement;
 use risingwave_common::vector::distance::DistanceMeasurement;
 use risingwave_hummock_sdk::HummockObjectId;
@@ -30,7 +31,6 @@ use crate::hummock::vector::file::FileVectorStore;
 use crate::hummock::vector::writer::VectorObjectIdManagerRef;
 use crate::hummock::{HummockResult, SstableStoreRef};
 use crate::opts::StorageOpts;
-use crate::store::Vector;
 use crate::vector::hnsw::{
     HnswBuilderOptions, HnswGraphBuilder, VectorAccessor, insert_graph, new_node,
 };
@@ -90,13 +90,13 @@ impl HnswFlatIndexWriter {
         })
     }
 
-    pub(crate) fn insert(&mut self, vec: Vector, info: Bytes) -> HummockResult<()> {
+    pub(crate) fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> HummockResult<()> {
         self.vector_store
             .building_vectors
             .as_mut()
             .expect("for write")
             .file_builder
-            .add(vec.to_ref(), &info);
+            .add(vec, &info);
         Ok(())
     }
 

src/storage/src/hummock/vector/writer/mod.rs

@@ -19,6 +19,7 @@ use std::sync::Arc;
 use bytes::Bytes;
 use futures::FutureExt;
 use hnsw::HnswFlatIndexWriter;
+use risingwave_common::array::VectorRef;
 use risingwave_common::vector::distance::DistanceMeasurement;
 use risingwave_hummock_sdk::vector_index::{
     FlatIndex, FlatIndexAdd, VectorFileInfo, VectorIndex, VectorIndexAdd, VectorIndexImpl,
@@ -29,7 +30,6 @@ use risingwave_hummock_sdk::{HummockObjectId, HummockRawObjectId};
 use crate::hummock::vector::file::VectorFileBuilder;
 use crate::hummock::{HummockResult, ObjectIdManager, SstableStoreRef};
 use crate::opts::StorageOpts;
-use crate::vector::Vector;
 
 #[async_trait::async_trait]
 pub trait VectorObjectIdManager: Send + Sync {
@@ -105,7 +105,7 @@ impl VectorWriterImpl {
         })
     }
 
-    pub(crate) fn insert(&mut self, vec: Vector, info: Bytes) -> HummockResult<()> {
+    pub(crate) fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> HummockResult<()> {
         match self {
             VectorWriterImpl::Flat(writer) => writer.insert(vec, info),
             VectorWriterImpl::HnswFlat(writer) => writer.insert(vec, info),
@@ -163,8 +163,8 @@ impl FlatIndexWriter {
         }
     }
 
-    pub(crate) fn insert(&mut self, vec: Vector, info: Bytes) -> HummockResult<()> {
-        self.vector_file_builder.add(vec.to_ref(), info.as_ref());
+    pub(crate) fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> HummockResult<()> {
+        self.vector_file_builder.add(vec, info.as_ref());
         Ok(())
     }