From 63c815770edefaf9221b7c131c007dcc0dd91f92 Mon Sep 17 00:00:00 2001 From: William Wen Date: Wed, 15 Oct 2025 14:49:51 +0800 Subject: [PATCH 1/6] temp save --- .cargo/config.toml | 2 +- .../src/hummock/store/hummock_storage.rs | 14 +++++++------- .../src/hummock/store/local_hummock_storage.rs | 18 +++++++++--------- src/storage/src/hummock/store/version.rs | 6 +++--- src/storage/src/memory.rs | 12 ++++++------ src/storage/src/monitor/monitored_store.rs | 8 ++++---- src/storage/src/monitor/traced_store.rs | 6 +++--- src/storage/src/panic_store.rs | 8 ++++---- src/storage/src/store.rs | 12 ++++++------ src/storage/src/store_impl.rs | 12 ++++++------ src/storage/src/table/batch_table/mod.rs | 4 +--- src/stream/src/common/table/state_table.rs | 5 +---- 12 files changed, 51 insertions(+), 56 deletions(-) diff --git a/.cargo/config.toml b/.cargo/config.toml index 1dfbe381a5563..a56c617640b4b 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -31,7 +31,7 @@ rustflags = [ # Flags for all targets. [target.'cfg(all())'] -rustflags = ["--cfg", "tokio_unstable"] +rustflags = ["--cfg", "tokio_unstable", "-Zhigher-ranked-assumptions"] # We have large git dependencies. This can make cloning faster. # https://doc.rust-lang.org/nightly/cargo/reference/unstable.html#git diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs index b938254e964d3..f2b6126646fe9 100644 --- a/src/storage/src/hummock/store/hummock_storage.rs +++ b/src/storage/src/hummock/store/hummock_storage.rs @@ -260,11 +260,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( - &self, + async fn get_inner<'a, O>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { let key_range = (Bound::Included(key.clone()), Bound::Included(key.clone())); @@ -639,12 +639,12 @@ pub struct HummockStorageReadSnapshot { } impl StateStoreGet for HummockStorageReadSnapshot { - fn on_key_value( - &self, + fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, - ) -> impl StorageFuture<'_, Option> { + on_key_value_fn: impl KeyValueFn<'a, O>, + ) -> impl StorageFuture<'a, Option> { self.get_inner(key, read_options, on_key_value_fn) } } diff --git a/src/storage/src/hummock/store/local_hummock_storage.rs b/src/storage/src/hummock/store/local_hummock_storage.rs index 84e4c60692c4b..22c60b8ebaa56 100644 --- a/src/storage/src/hummock/store/local_hummock_storage.rs +++ b/src/storage/src/hummock/store/local_hummock_storage.rs @@ -107,12 +107,12 @@ pub struct LocalHummockStorage { } impl LocalHummockFlushedSnapshotReader { - async fn get_flushed( - hummock_version_reader: &HummockVersionReader, + async fn get_flushed<'a, O>( + hummock_version_reader: &'a HummockVersionReader, read_version: &HummockReadVersionRef, user_key: UserKey, read_options: ReadOptions, - on_key_value_fn: impl crate::store::KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { 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( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { 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( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { let key = UserKey::new(self.table_id, key); match self.mem_table.buffer.get(&key.table_key) { diff --git a/src/storage/src/hummock/store/version.rs b/src/storage/src/hummock/store/version.rs index 38c07f5a1c274..485d17422c424 100644 --- a/src/storage/src/hummock/store/version.rs +++ b/src/storage/src/hummock/store/version.rs @@ -588,14 +588,14 @@ impl HummockVersionReader { const SLOW_ITER_FETCH_META_DURATION_SECOND: f64 = 5.0; impl HummockVersionReader { - pub async fn get( - &self, + pub async fn get<'a, O>( + &'a self, table_key: TableKey, epoch: u64, table_id: TableId, read_options: ReadOptions, read_version_tuple: ReadVersionTuple, - on_key_value_fn: impl crate::store::KeyValueFn, + on_key_value_fn: impl crate::store::KeyValueFn<'a, O>, ) -> StorageResult> { let (imms, uncommitted_ssts, committed_version) = read_version_tuple; diff --git a/src/storage/src/memory.rs b/src/storage/src/memory.rs index facc08050a6b2..7982b7fc04cf1 100644 --- a/src/storage/src/memory.rs +++ b/src/storage/src/memory.rs @@ -687,11 +687,11 @@ pub struct RangeKvStateStoreReadSnapshot { } impl StateStoreGet for RangeKvStateStoreReadSnapshot { - async fn on_key_value( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, _read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { self.inner .get_keyed_row_impl(key, self.epoch, self.table_id) @@ -1007,11 +1007,11 @@ impl RangeKvLocalStateStore { } impl StateStoreGet for RangeKvLocalStateStore { - async fn on_key_value( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, _read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { if let Some((key, value)) = match self.mem_table.buffer.get(&key) { None => self diff --git a/src/storage/src/monitor/monitored_store.rs b/src/storage/src/monitor/monitored_store.rs index d745438bb42e5..c315064a27d9d 100644 --- a/src/storage/src/monitor/monitored_store.rs +++ b/src/storage/src/monitor/monitored_store.rs @@ -139,12 +139,12 @@ impl MonitoredStateStore { } impl StateStoreGet for MonitoredTableStateStore { - fn on_key_value( - &self, + fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, - ) -> impl StorageFuture<'_, Option> { + on_key_value_fn: impl KeyValueFn<'a, O>, + ) -> impl StorageFuture<'a, Option> { let table_id = self.table_id(); let key_len = key.len(); self.monitored_on_key_value( diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs index ba2e56ea7c023..75d53127d3e23 100644 --- a/src/storage/src/monitor/traced_store.rs +++ b/src/storage/src/monitor/traced_store.rs @@ -145,11 +145,11 @@ impl TracedStateStore { } impl StateStoreGet for TracedStateStore { - async fn on_key_value( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { if let Some((key, value)) = self .traced_get_keyed_row( diff --git a/src/storage/src/panic_store.rs b/src/storage/src/panic_store.rs index de774e924cb51..14511e86f6e9a 100644 --- a/src/storage/src/panic_store.rs +++ b/src/storage/src/panic_store.rs @@ -30,12 +30,12 @@ use crate::store::*; pub struct PanicStateStore; impl StateStoreGet for PanicStateStore { - fn on_key_value( - &self, + fn on_key_value<'a, O: Send + 'static>( + &'a self, _key: TableKey, _read_options: ReadOptions, - _on_key_value_fn: impl KeyValueFn, - ) -> impl StorageFuture<'_, Option> { + _on_key_value_fn: impl KeyValueFn<'a, O>, + ) -> impl StorageFuture<'a, Option> { async { panic!("should not read from PanicStateStore") } } } diff --git a/src/storage/src/store.rs b/src/storage/src/store.rs index 346295a6be84d..9fd8e42b2c453 100644 --- a/src/storage/src/store.rs +++ b/src/storage/src/store.rs @@ -257,16 +257,16 @@ pub trait StateStoreReadLog: StaticSendSync { ) -> impl StorageFuture<'_, Self::ChangeLogIter>; } -pub trait KeyValueFn = - for<'kv> FnOnce(FullKey<&'kv [u8]>, &'kv [u8]) -> StorageResult + Send + 'static; +pub trait KeyValueFn<'a, O> = + for<'kv> FnOnce(FullKey<&'kv [u8]>, &'kv [u8]) -> StorageResult + Send + 'a; pub trait StateStoreGet: StaticSendSync { - fn on_key_value( - &self, + fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, - ) -> impl StorageFuture<'_, Option>; + on_key_value_fn: impl KeyValueFn<'a, O>, + ) -> impl StorageFuture<'a, Option>; } pub trait StateStoreRead: StateStoreGet + StaticSendSync { diff --git a/src/storage/src/store_impl.rs b/src/storage/src/store_impl.rs index 609509738e081..30928dae7b1fb 100644 --- a/src/storage/src/store_impl.rs +++ b/src/storage/src/store_impl.rs @@ -331,11 +331,11 @@ pub mod verify { } impl StateStoreGet for VerifyStateStore { - async fn on_key_value( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { let actual: Option<(FullKey, Bytes)> = self .actual @@ -1421,11 +1421,11 @@ mod dyn_state_store { where StateStorePointer

: AsRef + StaticSendSync, { - async fn on_key_value( - &self, + async fn on_key_value<'a, O: Send + 'static>( + &'a self, key: TableKey, read_options: ReadOptions, - on_key_value_fn: impl KeyValueFn, + on_key_value_fn: impl KeyValueFn<'a, O>, ) -> StorageResult> { let option = self.as_ref().get_keyed_row(key, read_options).await?; option diff --git a/src/storage/src/table/batch_table/mod.rs b/src/storage/src/table/batch_table/mod.rs index 5efadc219a6ac..23a42e3d82e65 100644 --- a/src/storage/src/table/batch_table/mod.rs +++ b/src/storage/src/table/batch_table/mod.rs @@ -409,11 +409,9 @@ impl BatchTableInner { }, ) .await?; - // TODO: may avoid the clone here when making the `on_key_value_fn` non-static - let row_serde = self.row_serde.clone(); match read_snapshot .on_key_value(serialized_pk, read_options, move |key, value| { - let row = row_serde.deserialize(value)?; + let row = self.row_serde.deserialize(value)?; Ok((key.epoch_with_gap.pure_epoch(), row)) }) .await? diff --git a/src/stream/src/common/table/state_table.rs b/src/stream/src/common/table/state_table.rs index 480ec6da69768..a021c2677a0f3 100644 --- a/src/stream/src/common/table/state_table.rs +++ b/src/stream/src/common/table/state_table.rs @@ -967,12 +967,9 @@ impl StateTableRowStore { ..Default::default() }; - // TODO: avoid clone when `on_key_value_fn` can be non-static - let row_serde = self.row_serde.clone(); - self.state_store .on_key_value(key_bytes, read_options, move |_, value| { - let row = row_serde.deserialize(value)?; + let row = self.row_serde.deserialize(value)?; Ok(OwnedRow::new(row)) }) .await From c6751bfadfe9713966f736de671cbc3067192b46 Mon Sep 17 00:00:00 2001 From: William Wen Date: Thu, 16 Oct 2025 17:57:28 +0800 Subject: [PATCH 2/6] remove clone for vector nearest --- src/storage/src/hummock/store/hummock_storage.rs | 6 +++--- src/storage/src/hummock/store/version.rs | 6 +++--- src/storage/src/memory.rs | 12 ++++++------ src/storage/src/monitor/monitored_store.rs | 8 ++++---- src/storage/src/monitor/traced_store.rs | 8 ++++---- src/storage/src/panic_store.rs | 6 +++--- src/storage/src/store.rs | 10 +++++----- src/storage/src/store_impl.rs | 14 +++++++------- 8 files changed, 35 insertions(+), 35 deletions(-) diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs index f2b6126646fe9..30a6b230e2221 100644 --- a/src/storage/src/hummock/store/hummock_storage.rs +++ b/src/storage/src/hummock/store/hummock_storage.rs @@ -687,11 +687,11 @@ impl StateStoreRead for HummockStorageReadSnapshot { } impl StateStoreReadVector for HummockStorageReadSnapshot { - async fn nearest( - &self, + async fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { let version = match self.epoch { HummockReadEpoch::Committed(epoch) diff --git a/src/storage/src/hummock/store/version.rs b/src/storage/src/hummock/store/version.rs index 485d17422c424..8a50ed3778290 100644 --- a/src/storage/src/hummock/store/version.rs +++ b/src/storage/src/hummock/store/version.rs @@ -1201,13 +1201,13 @@ impl HummockVersionReader { .await } - pub async fn nearest( - &self, + pub async fn nearest<'a, M: MeasureDistanceBuilder, O: Send>( + &'a self, version: PinnedVersion, table_id: TableId, target: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> HummockResult> { let Some(index) = version.vector_indexes.get(&table_id) else { return Ok(vec![]); diff --git a/src/storage/src/memory.rs b/src/storage/src/memory.rs index 7982b7fc04cf1..9c0b7c6d36ae1 100644 --- a/src/storage/src/memory.rs +++ b/src/storage/src/memory.rs @@ -728,19 +728,19 @@ impl StateStoreRead for RangeKvStateStoreReadSnapshot { } impl StateStoreReadVector for RangeKvStateStoreReadSnapshot { - async fn nearest( - &self, + async fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { - fn nearest_impl( - store: &InMemVectorStore, + fn nearest_impl<'a, M: MeasureDistanceBuilder, O>( + store: &'a InMemVectorStore, epoch: u64, table_id: TableId, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> Vec { let mut builder = NearestBuilder::<'_, O, M>::new(vec.to_ref(), options.top_n); builder.add( diff --git a/src/storage/src/monitor/monitored_store.rs b/src/storage/src/monitor/monitored_store.rs index c315064a27d9d..cdc57d8d29316 100644 --- a/src/storage/src/monitor/monitored_store.rs +++ b/src/storage/src/monitor/monitored_store.rs @@ -207,12 +207,12 @@ impl StateStoreReadLog for MonitoredStateStore { } impl StateStoreReadVector for MonitoredTableStateStore { - fn nearest( - &self, + fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, - ) -> impl StorageFuture<'_, Vec> { + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, + ) -> impl StorageFuture<'a, Vec> { // TODO: monitor self.inner.nearest(vec, options, on_nearest_item_fn) } diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs index 75d53127d3e23..3eec6e2359add 100644 --- a/src/storage/src/monitor/traced_store.rs +++ b/src/storage/src/monitor/traced_store.rs @@ -336,12 +336,12 @@ impl StateStore for TracedStateStore { } impl StateStoreReadVector for TracedStateStore { - fn nearest( - &self, + fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, - ) -> impl StorageFuture<'_, Vec> { + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, + ) -> impl StorageFuture<'a, Vec> { self.inner.nearest(vec, options, on_nearest_item_fn) } } diff --git a/src/storage/src/panic_store.rs b/src/storage/src/panic_store.rs index 14511e86f6e9a..df43aa22fdeec 100644 --- a/src/storage/src/panic_store.rs +++ b/src/storage/src/panic_store.rs @@ -152,11 +152,11 @@ impl StateStoreWriteVector for PanicStateStore { } impl StateStoreReadVector for PanicStateStore { - async fn nearest( - &self, + async fn nearest<'a, O: Send + 'static>( + &'a self, _vec: Vector, _options: VectorNearestOptions, - _on_nearest_item_fn: impl OnNearestItemFn, + _on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { panic!() } diff --git a/src/storage/src/store.rs b/src/storage/src/store.rs index 9fd8e42b2c453..a734f20460df4 100644 --- a/src/storage/src/store.rs +++ b/src/storage/src/store.rs @@ -437,15 +437,15 @@ pub struct VectorNearestOptions { pub hnsw_ef_search: usize, } -pub trait OnNearestItemFn = OnNearestItem + Send + Sync + 'static; +pub trait OnNearestItemFn<'a, O> = OnNearestItem + Send + Sync + 'a; pub trait StateStoreReadVector: StaticSendSync { - fn nearest( - &self, + fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, - ) -> impl StorageFuture<'_, Vec>; + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, + ) -> impl StorageFuture<'a, Vec>; } /// If `prefetch` is true, prefetch will be enabled. Prefetching may increase the memory diff --git a/src/storage/src/store_impl.rs b/src/storage/src/store_impl.rs index 30928dae7b1fb..01226211efe1c 100644 --- a/src/storage/src/store_impl.rs +++ b/src/storage/src/store_impl.rs @@ -368,12 +368,12 @@ pub mod verify { impl StateStoreReadVector for VerifyStateStore { - fn nearest( - &self, + fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, - ) -> impl StorageFuture<'_, Vec> { + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, + ) -> impl StorageFuture<'a, Vec> { self.actual.nearest(vec, options, on_nearest_item_fn) } } @@ -1296,11 +1296,11 @@ mod dyn_state_store { where StateStorePointer

: AsRef + StaticSendSync, { - async fn nearest( - &self, + async fn nearest<'a, O: Send + 'static>( + &'a self, vec: Vector, options: VectorNearestOptions, - on_nearest_item_fn: impl OnNearestItemFn, + on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { let output = self.as_ref().nearest(vec, options).await?; Ok(output From 8cb1b12fabd0edb521a2496dbac7b48c58f2d45e Mon Sep 17 00:00:00 2001 From: William Wen Date: Thu, 16 Oct 2025 18:08:11 +0800 Subject: [PATCH 3/6] remove static requirement on O --- src/storage/src/hummock/store/hummock_storage.rs | 4 ++-- src/storage/src/hummock/store/local_hummock_storage.rs | 4 ++-- src/storage/src/memory.rs | 6 +++--- src/storage/src/monitor/monitored_store.rs | 4 ++-- src/storage/src/monitor/traced_store.rs | 4 ++-- src/storage/src/panic_store.rs | 4 ++-- src/storage/src/store.rs | 4 ++-- src/storage/src/store_impl.rs | 8 ++++---- 8 files changed, 19 insertions(+), 19 deletions(-) diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs index 30a6b230e2221..5db8e0a0a53b1 100644 --- a/src/storage/src/hummock/store/hummock_storage.rs +++ b/src/storage/src/hummock/store/hummock_storage.rs @@ -639,7 +639,7 @@ pub struct HummockStorageReadSnapshot { } impl StateStoreGet for HummockStorageReadSnapshot { - fn on_key_value<'a, O: Send + 'static>( + fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -687,7 +687,7 @@ impl StateStoreRead for HummockStorageReadSnapshot { } impl StateStoreReadVector for HummockStorageReadSnapshot { - async fn nearest<'a, O: Send + 'static>( + async fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, diff --git a/src/storage/src/hummock/store/local_hummock_storage.rs b/src/storage/src/hummock/store/local_hummock_storage.rs index 22c60b8ebaa56..f06d5a825a9c9 100644 --- a/src/storage/src/hummock/store/local_hummock_storage.rs +++ b/src/storage/src/hummock/store/local_hummock_storage.rs @@ -266,7 +266,7 @@ pub struct LocalHummockFlushedSnapshotReader { } impl StateStoreGet for LocalHummockFlushedSnapshotReader { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -308,7 +308,7 @@ impl StateStoreRead for LocalHummockFlushedSnapshotReader { } impl StateStoreGet for LocalHummockStorage { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, diff --git a/src/storage/src/memory.rs b/src/storage/src/memory.rs index 9c0b7c6d36ae1..1bb035fdae647 100644 --- a/src/storage/src/memory.rs +++ b/src/storage/src/memory.rs @@ -687,7 +687,7 @@ pub struct RangeKvStateStoreReadSnapshot { } impl StateStoreGet for RangeKvStateStoreReadSnapshot { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, _read_options: ReadOptions, @@ -728,7 +728,7 @@ impl StateStoreRead for RangeKvStateStoreReadSnapshot { } impl StateStoreReadVector for RangeKvStateStoreReadSnapshot { - async fn nearest<'a, O: Send + 'static>( + async fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, @@ -1007,7 +1007,7 @@ impl RangeKvLocalStateStore { } impl StateStoreGet for RangeKvLocalStateStore { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, _read_options: ReadOptions, diff --git a/src/storage/src/monitor/monitored_store.rs b/src/storage/src/monitor/monitored_store.rs index cdc57d8d29316..fc3245e99c58a 100644 --- a/src/storage/src/monitor/monitored_store.rs +++ b/src/storage/src/monitor/monitored_store.rs @@ -139,7 +139,7 @@ impl MonitoredStateStore { } impl StateStoreGet for MonitoredTableStateStore { - fn on_key_value<'a, O: Send + 'static>( + fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -207,7 +207,7 @@ impl StateStoreReadLog for MonitoredStateStore { } impl StateStoreReadVector for MonitoredTableStateStore { - fn nearest<'a, O: Send + 'static>( + fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs index 3eec6e2359add..0673a695912d4 100644 --- a/src/storage/src/monitor/traced_store.rs +++ b/src/storage/src/monitor/traced_store.rs @@ -145,7 +145,7 @@ impl TracedStateStore { } impl StateStoreGet for TracedStateStore { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -336,7 +336,7 @@ impl StateStore for TracedStateStore { } impl StateStoreReadVector for TracedStateStore { - fn nearest<'a, O: Send + 'static>( + fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, diff --git a/src/storage/src/panic_store.rs b/src/storage/src/panic_store.rs index df43aa22fdeec..3f242f1a57e2f 100644 --- a/src/storage/src/panic_store.rs +++ b/src/storage/src/panic_store.rs @@ -30,7 +30,7 @@ use crate::store::*; pub struct PanicStateStore; impl StateStoreGet for PanicStateStore { - fn on_key_value<'a, O: Send + 'static>( + fn on_key_value<'a, O: Send + 'a>( &'a self, _key: TableKey, _read_options: ReadOptions, @@ -152,7 +152,7 @@ impl StateStoreWriteVector for PanicStateStore { } impl StateStoreReadVector for PanicStateStore { - async fn nearest<'a, O: Send + 'static>( + async fn nearest<'a, O: Send + 'a>( &'a self, _vec: Vector, _options: VectorNearestOptions, diff --git a/src/storage/src/store.rs b/src/storage/src/store.rs index a734f20460df4..9b78c56a4184c 100644 --- a/src/storage/src/store.rs +++ b/src/storage/src/store.rs @@ -261,7 +261,7 @@ pub trait KeyValueFn<'a, O> = for<'kv> FnOnce(FullKey<&'kv [u8]>, &'kv [u8]) -> StorageResult + Send + 'a; pub trait StateStoreGet: StaticSendSync { - fn on_key_value<'a, O: Send + 'static>( + fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -440,7 +440,7 @@ pub struct VectorNearestOptions { pub trait OnNearestItemFn<'a, O> = OnNearestItem + Send + Sync + 'a; pub trait StateStoreReadVector: StaticSendSync { - fn nearest<'a, O: Send + 'static>( + fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, diff --git a/src/storage/src/store_impl.rs b/src/storage/src/store_impl.rs index 01226211efe1c..96ee46b4259b9 100644 --- a/src/storage/src/store_impl.rs +++ b/src/storage/src/store_impl.rs @@ -331,7 +331,7 @@ pub mod verify { } impl StateStoreGet for VerifyStateStore { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, @@ -368,7 +368,7 @@ pub mod verify { impl StateStoreReadVector for VerifyStateStore { - fn nearest<'a, O: Send + 'static>( + fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, @@ -1296,7 +1296,7 @@ mod dyn_state_store { where StateStorePointer

: AsRef + StaticSendSync, { - async fn nearest<'a, O: Send + 'static>( + async fn nearest<'a, O: Send + 'a>( &'a self, vec: Vector, options: VectorNearestOptions, @@ -1421,7 +1421,7 @@ mod dyn_state_store { where StateStorePointer

: AsRef + StaticSendSync, { - async fn on_key_value<'a, O: Send + 'static>( + async fn on_key_value<'a, O: Send + 'a>( &'a self, key: TableKey, read_options: ReadOptions, From 7fac8dbd1db271d6a391ded780dc9d9d2dbb7e1a Mon Sep 17 00:00:00 2001 From: William Wen Date: Fri, 17 Oct 2025 15:57:04 +0800 Subject: [PATCH 4/6] avoid passing owned vector --- .../hummock_test/src/hummock_vector_tests.rs | 16 ++++++++-------- .../src/hummock/store/hummock_storage.rs | 4 ++-- src/storage/src/hummock/store/vector_writer.rs | 3 ++- src/storage/src/hummock/store/version.rs | 10 +++++----- src/storage/src/hummock/vector/writer/hnsw.rs | 6 +++--- src/storage/src/hummock/vector/writer/mod.rs | 8 ++++---- src/storage/src/memory.rs | 13 +++++++------ src/storage/src/monitor/monitored_store.rs | 6 +++--- src/storage/src/monitor/traced_store.rs | 4 ++-- src/storage/src/panic_store.rs | 5 +++-- src/storage/src/store.rs | 6 +++--- src/storage/src/store_impl.rs | 17 +++++++++-------- .../table/batch_table/vector_index_reader.rs | 4 ++-- src/stream/src/executor/vector_index/mod.rs | 2 -- 14 files changed, 53 insertions(+), 51 deletions(-) diff --git a/src/storage/hummock_test/src/hummock_vector_tests.rs b/src/storage/hummock_test/src/hummock_vector_tests.rs index e2faebb6c367e..98752f2488f24 100644 --- a/src/storage/hummock_test/src/hummock_vector_tests.rs +++ b/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(); } diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs index 5db8e0a0a53b1..c8e5277c33948 100644 --- a/src/storage/src/hummock/store/hummock_storage.rs +++ b/src/storage/src/hummock/store/hummock_storage.rs @@ -35,7 +35,7 @@ use risingwave_rpc_client::HummockMetaClient; use thiserror_ext::AsReport; use tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; use tokio::sync::oneshot; - +use risingwave_common::array::VectorRef; use super::local_hummock_storage::LocalHummockStorage; use super::version::{CommittedVersion, HummockVersionReader, read_filter_for_version}; use crate::compaction_catalog_manager::CompactionCatalogManagerRef; @@ -689,7 +689,7 @@ impl StateStoreRead for HummockStorageReadSnapshot { impl StateStoreReadVector for HummockStorageReadSnapshot { async fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { diff --git a/src/storage/src/hummock/store/vector_writer.rs b/src/storage/src/hummock/store/vector_writer.rs index e9a659b890aa5..fa51dc92d6723 100644 --- a/src/storage/src/hummock/store/vector_writer.rs +++ b/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() diff --git a/src/storage/src/hummock/store/version.rs b/src/storage/src/hummock/store/version.rs index 8a50ed3778290..8ac0e37850858 100644 --- a/src/storage/src/hummock/store/version.rs +++ b/src/storage/src/hummock/store/version.rs @@ -39,7 +39,7 @@ use risingwave_hummock_sdk::{EpochWithGap, HummockEpoch, LocalSstableInfo}; use risingwave_pb::hummock::LevelType; use sync_point::sync_point; use tracing::warn; - +use risingwave_common::array::VectorRef; use crate::error::StorageResult; use crate::hummock::event_handler::LocalInstanceId; use crate::hummock::iterator::change_log::ChangeLogIterator; @@ -67,7 +67,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}; @@ -1205,7 +1205,7 @@ impl HummockVersionReader { &'a self, version: PinnedVersion, table_id: TableId, - target: Vector, + target: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> HummockResult> { @@ -1221,7 +1221,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 +1246,7 @@ impl HummockVersionReader { let (items, _stats) = nearest::( &vector_store, &*graph, - target.to_ref(), + target, on_nearest_item_fn, options.hnsw_ef_search, options.top_n, diff --git a/src/storage/src/hummock/vector/writer/hnsw.rs b/src/storage/src/hummock/vector/writer/hnsw.rs index 916e7a518c4be..c543974451bec 100644 --- a/src/storage/src/hummock/vector/writer/hnsw.rs +++ b/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(()) } diff --git a/src/storage/src/hummock/vector/writer/mod.rs b/src/storage/src/hummock/vector/writer/mod.rs index 55a56fb671af3..52ffa16685982 100644 --- a/src/storage/src/hummock/vector/writer/mod.rs +++ b/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(()) } diff --git a/src/storage/src/memory.rs b/src/storage/src/memory.rs index 1bb035fdae647..882339630d493 100644 --- a/src/storage/src/memory.rs +++ b/src/storage/src/memory.rs @@ -35,7 +35,8 @@ use risingwave_hummock_sdk::{HummockEpoch, HummockReadEpoch}; use thiserror_ext::AsReport; use tokio::task::yield_now; use tracing::error; - +use risingwave_common::array::VectorRef; +use risingwave_common::types::ScalarRef; use crate::error::StorageResult; use crate::hummock::HummockError; use crate::hummock::utils::{ @@ -730,7 +731,7 @@ impl StateStoreRead for RangeKvStateStoreReadSnapshot { impl StateStoreReadVector for RangeKvStateStoreReadSnapshot { async fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { @@ -738,11 +739,11 @@ impl StateStoreReadVector for RangeKvStateStoreReadSnapshot { store: &'a InMemVectorStore, epoch: u64, table_id: TableId, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> Vec { - let mut builder = NearestBuilder::<'_, O, M>::new(vec.to_ref(), options.top_n); + let mut builder = NearestBuilder::<'_, O, M>::new(vec, options.top_n); builder.add( store .read() @@ -1285,8 +1286,8 @@ impl StateStoreWriteEpochControl for RangeKvLocalStateStore { } impl StateStoreWriteVector for RangeKvLocalStateStore { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()> { - self.vectors.push((vec, info)); + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()> { + self.vectors.push((vec.to_owned_scalar(), info)); Ok(()) } } diff --git a/src/storage/src/monitor/monitored_store.rs b/src/storage/src/monitor/monitored_store.rs index fc3245e99c58a..48f4057fd09aa 100644 --- a/src/storage/src/monitor/monitored_store.rs +++ b/src/storage/src/monitor/monitored_store.rs @@ -28,7 +28,7 @@ use risingwave_hummock_sdk::{HummockEpoch, HummockReadEpoch, SyncResult}; use thiserror_ext::AsReport; use tokio::time::Instant; use tracing::{Instrument, error}; - +use risingwave_common::types::VectorRef; use super::{MonitoredStateStoreGetStats, MonitoredStateStoreIterStats, MonitoredStorageMetrics}; use crate::error::StorageResult; use crate::hummock::sstable_store::SstableStoreRef; @@ -209,7 +209,7 @@ impl StateStoreReadLog for MonitoredStateStore { impl StateStoreReadVector for MonitoredTableStateStore { fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> impl StorageFuture<'a, Vec> { @@ -300,7 +300,7 @@ impl StateStoreWriteEpochControl for MonitoredTa } impl StateStoreWriteVector for MonitoredTableStateStore { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()> { + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()> { // TODO: monitor self.inner.insert(vec, info) } diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs index 0673a695912d4..db61bed080ff2 100644 --- a/src/storage/src/monitor/traced_store.rs +++ b/src/storage/src/monitor/traced_store.rs @@ -28,7 +28,7 @@ use risingwave_hummock_trace::{ TracedBytes, TracedSealCurrentEpochOptions, init_collector, should_use_trace, }; use thiserror_ext::AsReport; - +use risingwave_common::array::VectorRef; use crate::error::StorageResult; use crate::hummock::sstable_store::SstableStoreRef; use crate::hummock::{HummockStorage, ObjectIdManagerRef}; @@ -338,7 +338,7 @@ impl StateStore for TracedStateStore { impl StateStoreReadVector for TracedStateStore { fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> impl StorageFuture<'a, Vec> { diff --git a/src/storage/src/panic_store.rs b/src/storage/src/panic_store.rs index 3f242f1a57e2f..9c0b75610b429 100644 --- a/src/storage/src/panic_store.rs +++ b/src/storage/src/panic_store.rs @@ -16,6 +16,7 @@ use std::marker::PhantomData; use std::sync::Arc; use bytes::Bytes; +use risingwave_common::array::VectorRef; use risingwave_common::bitmap::Bitmap; use risingwave_common::hash::VirtualNode; use risingwave_hummock_sdk::HummockReadEpoch; @@ -146,7 +147,7 @@ impl StateStoreWriteEpochControl for PanicStateStore { } impl StateStoreWriteVector for PanicStateStore { - fn insert(&mut self, _vec: Vector, _info: Bytes) -> StorageResult<()> { + fn insert(&mut self, _vec: VectorRef<'_>, _info: Bytes) -> StorageResult<()> { panic!() } } @@ -154,7 +155,7 @@ impl StateStoreWriteVector for PanicStateStore { impl StateStoreReadVector for PanicStateStore { async fn nearest<'a, O: Send + 'a>( &'a self, - _vec: Vector, + _vec: VectorRef<'a>, _options: VectorNearestOptions, _on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { diff --git a/src/storage/src/store.rs b/src/storage/src/store.rs index 9b78c56a4184c..e8abd45add720 100644 --- a/src/storage/src/store.rs +++ b/src/storage/src/store.rs @@ -22,7 +22,7 @@ use bytes::Bytes; use futures::{Stream, TryStreamExt}; use futures_async_stream::try_stream; use prost::Message; -use risingwave_common::array::Op; +use risingwave_common::array::{Op, VectorRef}; use risingwave_common::bitmap::Bitmap; use risingwave_common::catalog::{TableId, TableOption}; use risingwave_common::hash::VirtualNode; @@ -428,7 +428,7 @@ pub trait StateStoreWriteEpochControl: StaticSendSync { } pub trait StateStoreWriteVector: StateStoreWriteEpochControl + StaticSendSync { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()>; + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()>; } pub struct VectorNearestOptions { @@ -442,7 +442,7 @@ pub trait OnNearestItemFn<'a, O> = OnNearestItem + Send + Sync + 'a; pub trait StateStoreReadVector: StaticSendSync { fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> impl StorageFuture<'a, Vec>; diff --git a/src/storage/src/store_impl.rs b/src/storage/src/store_impl.rs index 96ee46b4259b9..c61abd5f3ef3e 100644 --- a/src/storage/src/store_impl.rs +++ b/src/storage/src/store_impl.rs @@ -291,7 +291,7 @@ pub mod verify { use risingwave_hummock_sdk::HummockReadEpoch; use risingwave_hummock_sdk::key::{FullKey, TableKey, TableKeyRange}; use tracing::log::warn; - + use risingwave_common::array::VectorRef; use crate::error::StorageResult; use crate::hummock::HummockStorage; use crate::store::*; @@ -370,7 +370,7 @@ pub mod verify { { fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> impl StorageFuture<'a, Vec> { @@ -932,6 +932,7 @@ mod dyn_state_store { use std::sync::Arc; use bytes::Bytes; + use risingwave_common::array::VectorRef; use risingwave_common::bitmap::Bitmap; use risingwave_common::hash::VirtualNode; use risingwave_hummock_sdk::HummockReadEpoch; @@ -1244,12 +1245,12 @@ mod dyn_state_store { #[async_trait::async_trait] pub trait DynStateStoreWriteVector: DynStateStoreWriteEpochControl + StaticSendSync { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()>; + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()>; } #[async_trait::async_trait] impl DynStateStoreWriteVector for S { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()> { + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()> { self.insert(vec, info) } } @@ -1257,7 +1258,7 @@ mod dyn_state_store { pub type BoxDynStateStoreWriteVector = StateStorePointer>; impl StateStoreWriteVector for BoxDynStateStoreWriteVector { - fn insert(&mut self, vec: Vector, info: Bytes) -> StorageResult<()> { + fn insert(&mut self, vec: VectorRef<'_>, info: Bytes) -> StorageResult<()> { self.0.insert(vec, info) } } @@ -1268,7 +1269,7 @@ mod dyn_state_store { pub trait DynStateStoreReadVector: StaticSendSync { async fn nearest( &self, - vec: Vector, + vec: VectorRef<'_>, options: VectorNearestOptions, ) -> StorageResult>; } @@ -1277,7 +1278,7 @@ mod dyn_state_store { impl DynStateStoreReadVector for S { async fn nearest( &self, - vec: Vector, + vec: VectorRef<'_>, options: VectorNearestOptions, ) -> StorageResult> { use risingwave_common::types::ScalarRef; @@ -1298,7 +1299,7 @@ mod dyn_state_store { { async fn nearest<'a, O: Send + 'a>( &'a self, - vec: Vector, + vec: VectorRef<'a>, options: VectorNearestOptions, on_nearest_item_fn: impl OnNearestItemFn<'a, O>, ) -> StorageResult> { diff --git a/src/storage/src/table/batch_table/vector_index_reader.rs b/src/storage/src/table/batch_table/vector_index_reader.rs index 9158ed7fc8a41..9c9a62d3112bc 100644 --- a/src/storage/src/table/batch_table/vector_index_reader.rs +++ b/src/storage/src/table/batch_table/vector_index_reader.rs @@ -21,7 +21,7 @@ use risingwave_common::array::{ }; use risingwave_common::catalog::TableId; use risingwave_common::row::RowDeserializer; -use risingwave_common::types::{DataType, ScalarImpl, ScalarRef, StructType}; +use risingwave_common::types::{DataType, ScalarImpl, StructType}; use risingwave_common::util::value_encoding::BasicDeserializer; use risingwave_common::vector::distance::DistanceMeasurement; use risingwave_hummock_sdk::HummockReadEpoch; @@ -164,7 +164,7 @@ impl VectorIndexSnapshot<'_, S> { let row_results: Vec> = self .snapshot .nearest( - vector.to_owned_scalar(), + vector, VectorNearestOptions { top_n: self.reader.top_n, measure: self.reader.measure, diff --git a/src/stream/src/executor/vector_index/mod.rs b/src/stream/src/executor/vector_index/mod.rs index ba2cdd098b55c..aecdf3bbf3acb 100644 --- a/src/stream/src/executor/vector_index/mod.rs +++ b/src/stream/src/executor/vector_index/mod.rs @@ -18,7 +18,6 @@ use itertools::Itertools; use risingwave_common::array::Op; use risingwave_common::catalog::TableId; use risingwave_common::row::{Row, RowExt}; -use risingwave_common::types::ScalarRef; use risingwave_common::util::value_encoding::{BasicSerializer, ValueRowSerializer}; use risingwave_storage::StateStore; use risingwave_storage::store::{ @@ -99,7 +98,6 @@ impl VectorIndexWriteExecutor { continue; }; let vector = vector_datum.into_vector(); - let vector = vector.to_owned_scalar(); let info = self .serializer .serialize(row.project(&info_column_indices)) From c6bac47ee6135de605bfedb97fea35201551812c Mon Sep 17 00:00:00 2001 From: William Wen Date: Fri, 17 Oct 2025 19:12:50 +0800 Subject: [PATCH 5/6] pass doc --- .cargo/config.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.cargo/config.toml b/.cargo/config.toml index a56c617640b4b..e59bb43bdf223 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -33,6 +33,9 @@ rustflags = [ [target.'cfg(all())'] 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 # Actually we also want to prevent submodule cloning completely From 61adc6160683222ec6fdc57524b17c6cd0116991 Mon Sep 17 00:00:00 2001 From: William Wen Date: Mon, 20 Oct 2025 13:43:32 +0800 Subject: [PATCH 6/6] fmt and fix doc ci --- ci/scripts/doc.sh | 4 ++-- src/storage/src/hummock/store/hummock_storage.rs | 3 ++- src/storage/src/hummock/store/version.rs | 3 ++- src/storage/src/memory.rs | 5 +++-- src/storage/src/monitor/monitored_store.rs | 3 ++- src/storage/src/monitor/traced_store.rs | 3 ++- src/storage/src/store_impl.rs | 3 ++- 7 files changed, 15 insertions(+), 9 deletions(-) diff --git a/ci/scripts/doc.sh b/ci/scripts/doc.sh index 0459d4915bf07..f7b4e34d49e43 100755 --- a/ci/scripts/doc.sh +++ b/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 diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs index c8e5277c33948..5c20b9cdbd9ac 100644 --- a/src/storage/src/hummock/store/hummock_storage.rs +++ b/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; @@ -35,7 +36,7 @@ use risingwave_rpc_client::HummockMetaClient; use thiserror_ext::AsReport; use tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; use tokio::sync::oneshot; -use risingwave_common::array::VectorRef; + use super::local_hummock_storage::LocalHummockStorage; use super::version::{CommittedVersion, HummockVersionReader, read_filter_for_version}; use crate::compaction_catalog_manager::CompactionCatalogManagerRef; diff --git a/src/storage/src/hummock/store/version.rs b/src/storage/src/hummock/store/version.rs index 8ac0e37850858..f19647bf00bde 100644 --- a/src/storage/src/hummock/store/version.rs +++ b/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; @@ -39,7 +40,7 @@ use risingwave_hummock_sdk::{EpochWithGap, HummockEpoch, LocalSstableInfo}; use risingwave_pb::hummock::LevelType; use sync_point::sync_point; use tracing::warn; -use risingwave_common::array::VectorRef; + use crate::error::StorageResult; use crate::hummock::event_handler::LocalInstanceId; use crate::hummock::iterator::change_log::ChangeLogIterator; diff --git a/src/storage/src/memory.rs b/src/storage/src/memory.rs index 882339630d493..b4aabcb856ee1 100644 --- a/src/storage/src/memory.rs +++ b/src/storage/src/memory.rs @@ -22,10 +22,12 @@ use std::sync::{Arc, LazyLock}; use bytes::Bytes; use itertools::Itertools; use parking_lot::RwLock; +use risingwave_common::array::VectorRef; use risingwave_common::bitmap::{Bitmap, BitmapBuilder}; use risingwave_common::catalog::{TableId, TableOption}; use risingwave_common::dispatch_distance_measurement; use risingwave_common::hash::{VirtualNode, VnodeBitmapExt}; +use risingwave_common::types::ScalarRef; use risingwave_common::util::epoch::{EpochPair, MAX_EPOCH}; use risingwave_hummock_sdk::key::{ FullKey, TableKey, TableKeyRange, UserKey, prefixed_range_with_vnode, @@ -35,8 +37,7 @@ use risingwave_hummock_sdk::{HummockEpoch, HummockReadEpoch}; use thiserror_ext::AsReport; use tokio::task::yield_now; use tracing::error; -use risingwave_common::array::VectorRef; -use risingwave_common::types::ScalarRef; + use crate::error::StorageResult; use crate::hummock::HummockError; use crate::hummock::utils::{ diff --git a/src/storage/src/monitor/monitored_store.rs b/src/storage/src/monitor/monitored_store.rs index 48f4057fd09aa..6135bfc976ec3 100644 --- a/src/storage/src/monitor/monitored_store.rs +++ b/src/storage/src/monitor/monitored_store.rs @@ -23,12 +23,13 @@ use futures::{Future, FutureExt, TryFutureExt}; use risingwave_common::bitmap::Bitmap; use risingwave_common::catalog::TableId; use risingwave_common::hash::VirtualNode; +use risingwave_common::types::VectorRef; use risingwave_hummock_sdk::key::{TableKey, TableKeyRange}; use risingwave_hummock_sdk::{HummockEpoch, HummockReadEpoch, SyncResult}; use thiserror_ext::AsReport; use tokio::time::Instant; use tracing::{Instrument, error}; -use risingwave_common::types::VectorRef; + use super::{MonitoredStateStoreGetStats, MonitoredStateStoreIterStats, MonitoredStorageMetrics}; use crate::error::StorageResult; use crate::hummock::sstable_store::SstableStoreRef; diff --git a/src/storage/src/monitor/traced_store.rs b/src/storage/src/monitor/traced_store.rs index db61bed080ff2..b7b7a6b8f9856 100644 --- a/src/storage/src/monitor/traced_store.rs +++ b/src/storage/src/monitor/traced_store.rs @@ -18,6 +18,7 @@ use std::sync::Arc; use bytes::Bytes; use futures::future::BoxFuture; use futures::{Future, FutureExt}; +use risingwave_common::array::VectorRef; use risingwave_common::bitmap::Bitmap; use risingwave_common::catalog::TableId; use risingwave_common::hash::VirtualNode; @@ -28,7 +29,7 @@ use risingwave_hummock_trace::{ TracedBytes, TracedSealCurrentEpochOptions, init_collector, should_use_trace, }; use thiserror_ext::AsReport; -use risingwave_common::array::VectorRef; + use crate::error::StorageResult; use crate::hummock::sstable_store::SstableStoreRef; use crate::hummock::{HummockStorage, ObjectIdManagerRef}; diff --git a/src/storage/src/store_impl.rs b/src/storage/src/store_impl.rs index c61abd5f3ef3e..6be788235ea75 100644 --- a/src/storage/src/store_impl.rs +++ b/src/storage/src/store_impl.rs @@ -286,12 +286,13 @@ pub mod verify { use std::sync::Arc; use bytes::Bytes; + use risingwave_common::array::VectorRef; use risingwave_common::bitmap::Bitmap; use risingwave_common::hash::VirtualNode; use risingwave_hummock_sdk::HummockReadEpoch; use risingwave_hummock_sdk::key::{FullKey, TableKey, TableKeyRange}; use tracing::log::warn; - use risingwave_common::array::VectorRef; + use crate::error::StorageResult; use crate::hummock::HummockStorage; use crate::store::*;