vector-store: implement filtering by non-primary-key columns by ewienik · Pull Request #501 · scylladb/vector-store

ewienik · 2026-07-02T14:35:10Z

This PR refactors db_index fullscan to retrieve filtering columns. It also refactors db_cdc to use CDC table only for operation type and primary_key - then ask directly database for target values and filtering column values. Tests on 500k showed that this process is faster or not-slower than the previous one. It helps to synchronize retrieval of filtering columns with future multi-column targets.

This PR adds single integration test and two e2e tests in validator.

This PR refactors validator's serde::test_decimal_key.

Fixes: VECTOR-708

ewienik · 2026-07-02T15:18:53Z

Changelog for 3c3defe

refactor using single Semaphore for both wide and fine readers for limiting
concurrency on cdc queries
fix integration tests
fix validator decimal test

range diff

1:  d4c8e9c35 = 1:  44051d14b vector-store: implement filtering columns with fullscan
2:  155f9d993 ! 2:  88542d6d2 vector-store: refactor db_cdc to query db for target and filtering values
    @@ crates/vector-store/src/db_cdc.rs: use std::future;
      use tokio::sync::mpsc;
      use tokio::sync::mpsc::Sender;
      use tokio::sync::watch;
    +@@ crates/vector-store/src/db_cdc.rs: pub(crate) enum CdcReaderConfig {
    +     Fine,
    + }
    + 
    +-impl From<CdcReaderConfig> for CdcReaderState {
    +-    fn from(config: CdcReaderConfig) -> Self {
    +-        match config {
    +-            CdcReaderConfig::Wide => Self::new("wide", CdcReaderParams::wide),
    +-            CdcReaderConfig::Fine => Self::new("fine", CdcReaderParams::fine),
    +-        }
    +-    }
    +-}
    +-
    + /// Spawns a CDC actor that watches for session changes and manages a CDC reader.
    ++#[allow(clippy::too_many_arguments)]
    + pub(crate) fn new(
    +     config_rx: watch::Receiver<Arc<Config>>,
    +     mut session_rx: watch::Receiver<Option<Arc<Session>>>,
    +@@ crates/vector-store/src/db_cdc.rs: pub(crate) fn new(
    +     metrics: Arc<Metrics>,
    +     internals: Sender<Internals>,
    +     tx_embeddings: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
    ++    semaphore: Arc<Semaphore>,
    +     config: CdcReaderConfig,
    + ) -> mpsc::Sender<DbCdc> {
    +     let (tx, mut rx) = mpsc::channel::<DbCdc>(perf::channel_size().into());
    +@@ crates/vector-store/src/db_cdc.rs: pub(crate) fn new(
    +     // Mark the receiver to ensure first session update is visible
    +     session_rx.mark_changed();
    + 
    +-    let mut reader: CdcReaderState = config.into();
    ++    let mut reader = CdcReaderState::new(config, semaphore);
    +     let name = reader.name;
    +     let actor_key = metadata.key();
    +     let span_key = actor_key.clone();
    +@@ crates/vector-store/src/db_cdc.rs: struct CdcReaderState {
    +     handler_task: Option<tokio::task::JoinHandle<Duration>>,
    +     shutdown_notify: Arc<Notify>,
    +     error_notify: Arc<Notify>,
    ++    semaphore: Arc<Semaphore>,
    +     start: Duration,
    +     name: &'static str,
    +     params_fn: fn(&Config) -> CdcReaderParams,
    + }
    + 
    + impl CdcReaderState {
    +-    fn new(name: &'static str, params_fn: fn(&Config) -> CdcReaderParams) -> Self {
    +-        Self {
    ++    fn new(config: CdcReaderConfig, semaphore: Arc<Semaphore>) -> Self {
    ++        let ctor = |name: &'static str, params_fn: fn(&Config) -> CdcReaderParams| Self {
    +             reader: None,
    +             handler_task: None,
    +             shutdown_notify: Arc::new(Notify::new()),
    +             error_notify: Arc::new(Notify::new()),
    ++            semaphore,
    +             start: cdc_now(),
    +             name,
    +             params_fn,
    ++        };
    ++        match config {
    ++            CdcReaderConfig::Wide => ctor("wide", CdcReaderParams::wide),
    ++            CdcReaderConfig::Fine => ctor("fine", CdcReaderParams::fine),
    +         }
    +     }
    + 
    +@@ crates/vector-store/src/db_cdc.rs: impl CdcReaderState {
    +             metadata.clone(),
    +             metrics,
    +             tx_embeddings.clone(),
    ++            Arc::clone(&self.semaphore),
    +             self.name,
    +         )
    +         .await
    +@@ crates/vector-store/src/db_cdc.rs: fn drain_pending_notifications(notify: &Notify) {
    + }
    + 
    + /// Creates a CDC log reader with the given parameters.
    ++#[allow(clippy::too_many_arguments)]
    + async fn create_cdc_reader(
    +     start: Duration,
    +     params: CdcReaderParams,
     @@ crates/vector-store/src/db_cdc.rs: async fn create_cdc_reader(
    +     metadata: IndexMetadata,
    +     metrics: Arc<Metrics>,
    +     tx_embeddings: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
    ++    semaphore: Arc<Semaphore>,
    +     reader_name: &str,
    + ) -> anyhow::Result<(
    +     scylla_cdc::log_reader::CDCLogReader,
          impl std::future::Future<Output = anyhow::Result<()>>,
      )> {
    -     let consumer_factory =
    +-    let consumer_factory =
     -        CdcConsumerFactory::new(Arc::clone(&session), &metadata, metrics, tx_embeddings)?;
    -+        CdcConsumerFactory::new(Arc::clone(&session), &metadata, metrics, tx_embeddings).await?;
    ++    let consumer_factory = CdcConsumerFactory::new(
    ++        Arc::clone(&session),
    ++        &metadata,
    ++        metrics,
    ++        tx_embeddings,
    ++        semaphore,
    ++    )
    ++    .await?;
      
          let cdc_start = start - CHECKPOINT_TIMESTAMP_OFFSET;
          info!(
    @@ crates/vector-store/src/db_cdc.rs: fn spawn_handler_task(
          tx: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
          metrics: Arc<Metrics>,
     +    semaphore: Arc<Semaphore>,
    -+}
    -+
    + }
    + 
    +-struct CdcConsumer(Arc<CdcConsumerData>);
     +impl CdcConsumerData {
     +    async fn process_upsert(
     +        &self,
    @@ crates/vector-store/src/db_cdc.rs: fn spawn_handler_task(
     +    primary_key: Arc<Vec<CqlValue>>,
     +    timestamp: Timestamp,
     +    operation: Operation,
    - }
    - 
    --struct CdcConsumer(Arc<CdcConsumerData>);
    ++}
    ++
     +impl CdcConsumer {
     +    async fn process_row(&self) {
     +        if matches!(self.operation, Operation::Upsert) {
    @@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerFactory(Arc<CdcConsumerData
              session: Arc<Session>,
              metadata: &IndexMetadata,
              metrics: Arc<Metrics>,
    +         tx: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
    ++        semaphore: Arc<Semaphore>,
    +     ) -> anyhow::Result<Self> {
    +         let cluster_state = session.get_cluster_state();
    +         let table = cluster_state
     @@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
                  .collect_nonempty_arc()
                  .ok_or_else(|| anyhow!("primary key must have at least one column"))?;
    @@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
                  kind: metadata.kind.clone(),
                  tx,
                  metrics,
    -+            semaphore: Arc::new(Semaphore::new(concurrency_limit())),
    ++            semaphore,
              })))
          }
      }
    +
    + ## crates/vector-store/src/db_index.rs ##
    +@@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
    + 
    +     let statements = Arc::new(Statements::new(statements_session_rx, metadata.clone()).await?);
    + 
    ++    let semaphore = Arc::new(Semaphore::new(concurrency_limit()));
     +
    +     // Create wide-framed CDC actor
    +     let cdc_wide = db_cdc::new(
    +         config_rx.clone(),
    +@@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
    +         Arc::clone(&metrics),
    +         internals.clone(),
    +         tx_embeddings.clone(),
    ++        Arc::clone(&semaphore),
    +         CdcReaderConfig::Wide,
    +     );
    + 
    +@@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
    +         metrics,
    +         internals,
    +         tx_embeddings.clone(),
    ++        semaphore,
    +         CdcReaderConfig::Fine,
    +     );
    + 
    +@@ crates/vector-store/src/db_index.rs: fn parse_indexed_value(value: CqlValue, kind: &IndexKind) -> anyhow::Result<DbIn
    +     }
    + }
    + 
     +fn concurrency_limit() -> usize {
     +    const RATIO: usize = 3;
     +    perf::num_workers().get() * RATIO
     +}
    ++
    + #[cfg(test)]
    + mod tests {
    + 
     
      ## crates/vector-store/src/db_index_backend.rs ##
     @@
3:  1304fb956 = 3:  df4f699cf vector-store: enable filtering on non-primary-key columns
4:  28cd72931 ! 4:  6a2f252d0 integration/tests: implement a simple test for filtering on non-primary-key columns
    @@ crates/vector-store/tests/integration/routing.rs: fn single_row_scan(pks: impl I
              Timestamp::from_millis(10),
          )])
      }
    +@@ crates/vector-store/tests/integration/routing.rs: async fn ann_routes_to_local_index_with_filter_columns_covering_restriction() {
    +         allow_filtering: true,
    +     };
    + 
    +-    // TODO: update this assertion to expect StatusCode::OK once filtering on
    +-    // non-primary-key columns is supported end-to-end. Currently the request
    +-    // is routed correctly but fails at the filter validation layer.
    +     let response = assert_ann_served_by(
    +         &client,
    +         &covering,
    +         post_ann_with_filter(&client, &non_covering, filter),
    +     )
    +     .await;
    +-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    ++    assert_eq!(response.status(), StatusCode::OK);
    + }
    + 
    + #[tokio::test]
    +@@ crates/vector-store/tests/integration/routing.rs: async fn ann_routes_to_global_index_with_filter_columns_covering_restriction() {
    +         allow_filtering: true,
    +     };
    + 
    +-    // TODO: update this assertion to expect StatusCode::OK once filtering on
    +-    // non-primary-key columns is supported end-to-end. Currently the request
    +-    // is routed correctly but fails at the filter validation layer.
    +     let response = assert_ann_served_by(
    +         &client,
    +         &covering,
    +         post_ann_with_filter(&client, &non_covering, filter),
    +     )
    +     .await;
    +-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    ++    assert_eq!(response.status(), StatusCode::OK);
    + }
    + 
    + #[tokio::test]
     
      ## crates/vector-store/tests/integration/usearch.rs ##
     @@ crates/vector-store/tests/integration/usearch.rs: pub(crate) async fn setup_store_with_quantization(
5:  da03f16c6 = 5:  d13db9d64 validator: implement simple tests for non-primary-key filtering columns
6:  69ff28e45 ! 6:  3c3defeb0 validator: fix serde::test_decimal_key
    @@ crates/validator/src/serde.rs: async fn test_decimal_key(actors: Arc<TestActors>
     +            _ => panic!("unexpected pk={pk}"),
     +        }
     +    }
    ++    assert!(expected_pks.is_empty());
      
          // Phase 2: after index creation (CDC path).
          // Same pattern as Phase 1.
    @@ crates/validator/src/serde.rs: async fn test_decimal_key(actors: Arc<TestActors>
     +            _ => panic!("unexpected pk={pk}"),
     +        }
     +    }
    ++    assert!(expected_pks.is_empty());
      
          session
              .query_unpaged(format!("DROP KEYSPACE {keyspace}"), ())

ewienik · 2026-07-16T11:14:31Z

Changelog for a96084a

rebase to master

range diff

1:  44051d14b ! 1:  465b0ad53 vector-store: implement filtering columns with fullscan
    @@ crates/vector-store/src/db_index.rs: mod tests {
      }
     
      ## crates/vector-store/src/db_index_backend.rs ##
    -@@ crates/vector-store/src/db_index_backend.rs: impl DbIndexBackend {
    +@@ crates/vector-store/src/db_index_backend.rs: fn take_alternator_attr(attrs: CqlValue, target: &str) -> anyhow::Result<Option<
      ///
      /// For CQL-native tables, selects the vector column directly.
      /// For Alternator tables, selects from the `:attrs` map column.
2:  88542d6d2 ! 2:  5d66d945c vector-store: refactor db_cdc to query db for target and filtering values
    @@ Commit message
         This commit adds retrieval of filtering column values - it adds supporting
         functions to the db_index_backend.
     
    - ## crates/vector-store/src/db_cdc.rs ##
    -@@ crates/vector-store/src/db_cdc.rs: use crate::ColumnName;
    - use crate::Config;
    - use crate::DbIndexedOperation;
    - use crate::DbIndexedRow;
    --use crate::DbIndexedValue;
    - use crate::IndexKey;
    - use crate::IndexKind;
    - use crate::IndexMetadata;
    - use crate::Metrics;
    - use crate::NonemptyArc;
    --use crate::NonemptyBox;
    - use crate::NonemptyIteratorExt;
    --use crate::db_index_backend::CdcValueStatus;
    --use crate::db_index_backend::DbIndexBackend;
    -+use crate::PrimaryKey;
    -+use crate::Timestamp;
    -+use crate::db_index;
    -+use crate::db_index_backend;
    - use crate::internals::Internals;
    - use crate::internals::InternalsExt;
    - use crate::perf;
    --use crate::timestamp::Timestamped;
    - use ::time::OffsetDateTime;
    - use anyhow::Context;
    - use anyhow::anyhow;
    -@@ crates/vector-store/src/db_cdc.rs: use anyhow::bail;
    - use async_trait::async_trait;
    - use futures::FutureExt;
    - use scylla::client::session::Session;
    -+use scylla::statement::prepared::PreparedStatement;
    - use scylla::value::CqlValue;
    -+use scylla::value::Row;
    - use scylla_cdc::consumer::CDCRow;
    - use scylla_cdc::consumer::Consumer;
    - use scylla_cdc::consumer::ConsumerFactory;
    -@@ crates/vector-store/src/db_cdc.rs: use std::future;
    - use std::sync::Arc;
    + ## crates/vector-store/src/db_cdc/actor.rs ##
    +@@ crates/vector-store/src/db_cdc/actor.rs: use std::sync::Arc;
      use std::time::Duration;
      use std::time::SystemTime;
    -+use tap::Pipe;
      use tokio::sync::Notify;
     +use tokio::sync::Semaphore;
      use tokio::sync::mpsc;
      use tokio::sync::mpsc::Sender;
      use tokio::sync::watch;
    -@@ crates/vector-store/src/db_cdc.rs: pub(crate) enum CdcReaderConfig {
    -     Fine,
    +@@ crates/vector-store/src/db_cdc/actor.rs: impl CdcReaderConfig {
      }
      
    --impl From<CdcReaderConfig> for CdcReaderState {
    --    fn from(config: CdcReaderConfig) -> Self {
    --        match config {
    --            CdcReaderConfig::Wide => Self::new("wide", CdcReaderParams::wide),
    --            CdcReaderConfig::Fine => Self::new("fine", CdcReaderParams::fine),
    --        }
    --    }
    --}
    --
      /// Spawns a CDC actor that watches for session changes and manages a CDC reader.
     +#[allow(clippy::too_many_arguments)]
      pub(crate) fn new(
          config_rx: watch::Receiver<Arc<Config>>,
          mut session_rx: watch::Receiver<Option<Arc<Session>>>,
    -@@ crates/vector-store/src/db_cdc.rs: pub(crate) fn new(
    -     metrics: Arc<Metrics>,
    +@@ crates/vector-store/src/db_cdc/actor.rs: pub(crate) fn new(
          internals: Sender<Internals>,
    +     metrics: Arc<Metrics>,
          tx_embeddings: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
     +    semaphore: Arc<Semaphore>,
          config: CdcReaderConfig,
      ) -> mpsc::Sender<DbCdc> {
          let (tx, mut rx) = mpsc::channel::<DbCdc>(perf::channel_size().into());
    -@@ crates/vector-store/src/db_cdc.rs: pub(crate) fn new(
    -     // Mark the receiver to ensure first session update is visible
    -     session_rx.mark_changed();
    - 
    --    let mut reader: CdcReaderState = config.into();
    -+    let mut reader = CdcReaderState::new(config, semaphore);
    +@@ crates/vector-store/src/db_cdc/actor.rs: pub(crate) fn new(
    +         Arc::clone(&metrics),
    +         metadata.keyspace_name.clone(),
    +         metadata.index_name.clone(),
    ++        semaphore,
    +     );
          let name = reader.name;
          let actor_key = metadata.key();
    -     let span_key = actor_key.clone();
    -@@ crates/vector-store/src/db_cdc.rs: struct CdcReaderState {
    +@@ crates/vector-store/src/db_cdc/actor.rs: struct CdcReaderState {
          handler_task: Option<tokio::task::JoinHandle<Duration>>,
          shutdown_notify: Arc<Notify>,
          error_notify: Arc<Notify>,
    @@ crates/vector-store/src/db_cdc.rs: struct CdcReaderState {
          start: Duration,
          name: &'static str,
          params_fn: fn(&Config) -> CdcReaderParams,
    - }
    - 
    - impl CdcReaderState {
    --    fn new(name: &'static str, params_fn: fn(&Config) -> CdcReaderParams) -> Self {
    --        Self {
    -+    fn new(config: CdcReaderConfig, semaphore: Arc<Semaphore>) -> Self {
    -+        let ctor = |name: &'static str, params_fn: fn(&Config) -> CdcReaderParams| Self {
    +@@ crates/vector-store/src/db_cdc/actor.rs: impl CdcReaderState {
    +         metrics: Arc<Metrics>,
    +         keyspace: KeyspaceName,
    +         index_name: IndexName,
    ++        semaphore: Arc<Semaphore>,
    +     ) -> Self {
    +         let state = Self {
                  reader: None,
                  handler_task: None,
                  shutdown_notify: Arc::new(Notify::new()),
    @@ crates/vector-store/src/db_cdc.rs: struct CdcReaderState {
                  start: cdc_now(),
                  name,
                  params_fn,
    -+        };
    -+        match config {
    -+            CdcReaderConfig::Wide => ctor("wide", CdcReaderParams::wide),
    -+            CdcReaderConfig::Fine => ctor("fine", CdcReaderParams::fine),
    -         }
    -     }
    - 
    -@@ crates/vector-store/src/db_cdc.rs: impl CdcReaderState {
    +@@ crates/vector-store/src/db_cdc/actor.rs: impl CdcReaderState {
    +             Arc::clone(session),
                  metadata.clone(),
    -             metrics,
                  tx_embeddings.clone(),
     +            Arc::clone(&self.semaphore),
    +             Arc::clone(&self.metrics),
                  self.name,
              )
    -         .await
    -@@ crates/vector-store/src/db_cdc.rs: fn drain_pending_notifications(notify: &Notify) {
    +@@ crates/vector-store/src/db_cdc/actor.rs: fn drain_pending_notifications(notify: &Notify) {
      }
      
      /// Creates a CDC log reader with the given parameters.
    @@ crates/vector-store/src/db_cdc.rs: fn drain_pending_notifications(notify: &Notif
      async fn create_cdc_reader(
          start: Duration,
          params: CdcReaderParams,
    -@@ crates/vector-store/src/db_cdc.rs: async fn create_cdc_reader(
    +     session: Arc<Session>,
          metadata: IndexMetadata,
    -     metrics: Arc<Metrics>,
          tx_embeddings: mpsc::Sender<(DbIndexedRow, AsyncInProgress)>,
     +    semaphore: Arc<Semaphore>,
    +     metrics: Arc<Metrics>,
          reader_name: &str,
      ) -> anyhow::Result<(
    -     scylla_cdc::log_reader::CDCLogReader,
    -     impl std::future::Future<Output = anyhow::Result<()>>,
    - )> {
    --    let consumer_factory =
    --        CdcConsumerFactory::new(Arc::clone(&session), &metadata, metrics, tx_embeddings)?;
    -+    let consumer_factory = CdcConsumerFactory::new(
    -+        Arc::clone(&session),
    -+        &metadata,
    -+        metrics,
    -+        tx_embeddings,
    +@@ crates/vector-store/src/db_cdc/actor.rs: async fn create_cdc_reader(
    +         &metadata,
    +         Arc::clone(&metrics),
    +         tx_embeddings,
    +-    )?;
     +        semaphore,
     +    )
     +    .await?;
      
          let cdc_start = start - CHECKPOINT_TIMESTAMP_OFFSET;
          info!(
    -@@ crates/vector-store/src/db_cdc.rs: fn spawn_handler_task(
    -     )
    - }
    +@@ crates/vector-store/src/db_cdc/actor.rs: mod tests {
    +             metrics,
    +             "ks".into(),
    +             "idx".into(),
    ++            Semaphore::new(1).into(),
    +         )
    +     }
    + 
    +
    + ## crates/vector-store/src/db_cdc/consumer.rs ##
    +@@ crates/vector-store/src/db_cdc/consumer.rs: use crate::AsyncInProgress;
    + use crate::ColumnName;
    + use crate::DbIndexedOperation;
    + use crate::DbIndexedRow;
    +-use crate::DbIndexedValue;
    + use crate::IndexKey;
    + use crate::IndexKind;
    + use crate::IndexMetadata;
    + use crate::Metrics;
    + use crate::NonemptyArc;
    +-use crate::NonemptyBox;
    + use crate::NonemptyIteratorExt;
    +-use crate::Timestamped;
    +-use crate::db_index_backend::CdcValueStatus;
    +-use crate::db_index_backend::DbIndexBackend;
    ++use crate::PrimaryKey;
    ++use crate::Timestamp;
    ++use crate::db_index;
    ++use crate::db_index_backend;
    ++use anyhow::Context;
    + use anyhow::anyhow;
    + use anyhow::bail;
    + use async_trait::async_trait;
    + use scylla::client::session::Session;
    ++use scylla::statement::prepared::PreparedStatement;
    + use scylla::value::CqlValue;
    ++use scylla::value::Row;
    + use scylla_cdc::consumer::CDCRow;
    + use scylla_cdc::consumer::Consumer;
    + use scylla_cdc::consumer::ConsumerFactory;
    + use scylla_cdc::consumer::OperationType;
    + use std::sync::Arc;
    ++use tap::Pipe;
    ++use tokio::sync::Semaphore;
    + use tokio::sync::mpsc;
    ++use tracing::debug;
    ++use tracing::error;
      
     -fn extract_indexed_value(
     -    backend: &DbIndexBackend,
    @@ crates/vector-store/src/db_cdc.rs: fn spawn_handler_task(
     +            OperationType::PartitionDelete | OperationType::RowDelete => Operation::Delete,
      
                  OperationType::RowUpdate | OperationType::RowInsert | OperationType::PostImage => {
    --                let value = match source.take_cdc_value(&mut row) {
    +-                let value = match source.take_cdc_value(&mut row)? {
     -                    CdcValueStatus::Deleted => None,
     -                    CdcValueStatus::Skip => return Ok(()),
     -                    CdcValueStatus::NewValue(value) => {
    @@ crates/vector-store/src/db_cdc.rs: fn spawn_handler_task(
                  }
      
                  OperationType::PreImage
    -@@ crates/vector-store/src/db_cdc.rs: impl Consumer for CdcConsumer {
    +@@ crates/vector-store/src/db_cdc/consumer.rs: impl Consumer for CdcConsumer {
              };
      
              let Some(primary_key) = self
    @@ crates/vector-store/src/db_cdc.rs: impl Consumer for CdcConsumer {
              Ok(())
          }
      }
    -@@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerFactory(Arc<CdcConsumerData>);
    +@@ crates/vector-store/src/db_cdc/consumer.rs: pub(super) struct CdcConsumerFactory(Arc<CdcConsumerData>);
      #[async_trait]
      impl ConsumerFactory for CdcConsumerFactory {
          async fn new_consumer(&self) -> Box<dyn Consumer> {
    @@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerFactory(Arc<CdcConsumerData
      }
      
      impl CdcConsumerFactory {
    --    fn new(
    -+    async fn new(
    +-    pub(super) fn new(
    ++    pub(super) async fn new(
              session: Arc<Session>,
              metadata: &IndexMetadata,
              metrics: Arc<Metrics>,
    @@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerFactory(Arc<CdcConsumerData
          ) -> anyhow::Result<Self> {
              let cluster_state = session.get_cluster_state();
              let table = cluster_state
    -@@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
    +@@ crates/vector-store/src/db_cdc/consumer.rs: impl CdcConsumerFactory {
                  .collect_nonempty_arc()
                  .ok_or_else(|| anyhow!("primary key must have at least one column"))?;
      
     -        let backend = DbIndexBackend::from(metadata);
    --
     +        let target_columns = metadata.target_columns.clone();
     +        let filtering_columns: Arc<[_]> = metadata
     +            .filtering_columns
    @@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
     +                    stmt.set_is_idempotent(true);
     +                    stmt
     +                });
    + 
              Ok(Self(Arc::new(CdcConsumerData {
     +            session,
     +            st_select_values,
    @@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
          let cdc_wide = db_cdc::new(
              config_rx.clone(),
     @@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
    -         Arc::clone(&metrics),
              internals.clone(),
    +         metrics.clone(),
              tx_embeddings.clone(),
     +        Arc::clone(&semaphore),
              CdcReaderConfig::Wide,
          );
      
     @@ crates/vector-store/src/db_index.rs: pub(crate) async fn new(
    -         metrics,
              internals,
    +         metrics,
              tx_embeddings.clone(),
     +        semaphore,
              CdcReaderConfig::Fine,
    @@ crates/vector-store/src/db_index_backend.rs
      use crate::TableIdentifier;
      use crate::TableName;
     -use crate::Vector;
    --use crate::vector;
      use futures::TryStreamExt;
      use regex::Regex;
      use scylla::client::session::Session;
    @@ crates/vector-store/src/db_index_backend.rs
      /// Alternator tables store all user attributes in a single map column `:attrs`.
      /// Because Alternator (DynamoDB-compatible) is schemaless, different items can
     @@ crates/vector-store/src/db_index_backend.rs: use std::sync::OnceLock;
    - /// of dedicated typed columns.
    + /// are text keys and attribute values are serialized blobs.
      const ALTERNATOR_ATTRS_COLUMN: &str = ":attrs";
      
     -pub(crate) enum CdcValueStatus {
    @@ crates/vector-store/src/db_index_backend.rs: use std::sync::OnceLock;
     -    }
     -
     -    pub fn extract_vector(&self, value: CqlValue) -> anyhow::Result<Option<Vector>> {
    --        match self {
    --            Self::Cql { .. } => Vector::try_from(value).map(Some),
    --            Self::Alternator { target_columns } => vector::AlternatorAttrs {
    --                attrs: value,
    --                target_column: target_columns.first().as_ref(),
    --            }
    --            .try_into(),
    --        }
    +-        Vector::try_from(value).map(Some)
     -    }
     -
     -    pub fn extract_document(&self, value: CqlValue) -> anyhow::Result<Option<String>> {
    @@ crates/vector-store/src/db_index_backend.rs: use std::sync::OnceLock;
     -        }
     -    }
     -
    --    pub fn take_cdc_value(&self, row: &mut CDCRow<'_>) -> CdcValueStatus {
    +-    pub fn take_cdc_value(&self, row: &mut CDCRow<'_>) -> anyhow::Result<CdcValueStatus> {
     -        match self {
     -            Self::Cql { target_columns } => {
     -                let column = target_columns.first().as_ref();
     -                if row.is_value_deleted(column) {
    --                    return CdcValueStatus::Deleted;
    +-                    return Ok(CdcValueStatus::Deleted);
     -                }
    --                match row.take_value(column) {
    +-                Ok(match row.take_value(column) {
     -                    Some(value) => CdcValueStatus::NewValue(value),
     -                    None => CdcValueStatus::Skip,
    --                }
    +-                })
     -            }
     -            Self::Alternator { target_columns } => {
     -                // Check take_value before is_value_deleted.
    @@ crates/vector-store/src/db_index_backend.rs: use std::sync::OnceLock;
     -                // so is_value_deleted can be true even when a new value is present.
     -                // We must check for a value first to avoid misreporting it as deleted.
     -                let column = ALTERNATOR_ATTRS_COLUMN;
    --                match row.take_value(column) {
    --                    Some(value) => CdcValueStatus::NewValue(value),
    --                    None if row.is_value_deleted(column) => CdcValueStatus::Deleted,
    +-                let target = target_columns.first().as_ref();
    +-                let value = match row.take_value(column) {
    +-                    Some(attrs) => take_alternator_attr(attrs, target)?,
    +-                    None => None,
    +-                };
    +-
    +-                match value {
    +-                    Some(value) => Ok(CdcValueStatus::NewValue(value)),
    +-                    None if row.is_value_deleted(column) => Ok(CdcValueStatus::Deleted),
     -                    None => {
    --                        let target_deleted = row.take_deleted_elements(column).iter().any(|el| {
    --                            el.as_text().map(String::as_str)
    --                                == Some(target_columns.first().as_ref())
    --                        });
    +-                        let target_deleted = row
    +-                            .take_deleted_elements(column)
    +-                            .iter()
    +-                            .any(|el| alternator_attr_key_matches_target(el, target));
     -                        if target_deleted {
    --                            CdcValueStatus::Deleted
    +-                            Ok(CdcValueStatus::Deleted)
     -                        } else {
    --                            CdcValueStatus::Skip
    +-                            Ok(CdcValueStatus::Skip)
     -                        }
     -                    }
     -                }
    @@ crates/vector-store/src/db_index_backend.rs: use std::sync::OnceLock;
     -    }
     -}
     -
    +-fn alternator_attr_key_matches_target(key: &CqlValue, target: &str) -> bool {
    +-    key.as_text().map(String::as_str) == Some(target)
    +-}
    +-
    +-fn take_alternator_attr(attrs: CqlValue, target: &str) -> anyhow::Result<Option<CqlValue>> {
    +-    let CqlValue::Map(entries) = attrs else {
    +-        anyhow::bail!("expected Map for :attrs column, got {attrs:?}");
    +-    };
    +-
    +-    Ok(entries
    +-        .into_iter()
    +-        .find_map(|(key, value)| alternator_attr_key_matches_target(&key, target).then_some(value)))
    +-}
    +-
     -/// Builds the CQL range scan query appropriate for the given keyspace.
     -///
     -/// For CQL-native tables, selects the vector column directly.
    @@ crates/vector-store/src/db_index_backend.rs: pub(crate) fn range_scan_query<'a>(
      /// Retrieves the vector dimensions for the given index, dispatching to the
      /// appropriate strategy based on whether the keyspace is Alternator- or CQL-backed.
      pub(crate) async fn get_dimensions(
    +@@ crates/vector-store/src/db_index_backend.rs: mod tests {
    +             "writetime must use the same escaped attribute access: {query}"
    +         );
    +     }
    +-
    +-    #[test]
    +-    fn alternator_attr_text_key_matches_target() {
    +-        assert!(alternator_attr_key_matches_target(
    +-            &CqlValue::Text("vec".into()),
    +-            "vec"
    +-        ));
    +-    }
    +-
    +-    #[test]
    +-    fn alternator_attr_non_text_key_does_not_match_target() {
    +-        assert!(!alternator_attr_key_matches_target(
    +-            &CqlValue::Blob(b"vec".to_vec()),
    +-            "vec"
    +-        ));
    +-    }
    +-
    +-    #[test]
    +-    fn take_alternator_attr_returns_target_value() {
    +-        let attrs = CqlValue::Map(vec![
    +-            (CqlValue::Text("unrelated".into()), CqlValue::Blob(vec![1])),
    +-            (CqlValue::Text("vec".into()), CqlValue::Blob(vec![2])),
    +-        ]);
    +-
    +-        assert_eq!(
    +-            take_alternator_attr(attrs, "vec").unwrap(),
    +-            Some(CqlValue::Blob(vec![2]))
    +-        );
    +-    }
    +-
    +-    #[test]
    +-    fn take_alternator_attr_skips_unrelated_attrs() {
    +-        let attrs = CqlValue::Map(vec![(
    +-            CqlValue::Text("unrelated".into()),
    +-            CqlValue::Blob(vec![1]),
    +-        )]);
    +-
    +-        assert_eq!(take_alternator_attr(attrs, "vec").unwrap(), None);
    +-    }
    +-
    +-    #[test]
    +-    fn take_alternator_attr_rejects_non_map() {
    +-        assert!(take_alternator_attr(CqlValue::Blob(vec![1]), "vec").is_err());
    +-    }
    + }
3:  df4f699cf = 3:  56c6c3a8c vector-store: enable filtering on non-primary-key columns
4:  6a2f252d0 = 4:  4f53ee1ab integration/tests: implement a simple test for filtering on non-primary-key columns
5:  d13db9d64 = 5:  ec6b4757a validator: implement simple tests for non-primary-key filtering columns
6:  3c3defeb0 = 6:  a96084a88 validator: fix serde::test_decimal_key

ewienik · 2026-07-16T14:15:21Z

Changelog for 8939dee

rebase to master

range diff

1:  465b0ad53 = 1:  880050a34 vector-store: implement filtering columns with fullscan
2:  5d66d945c = 2:  50d802f7a vector-store: refactor db_cdc to query db for target and filtering values
3:  56c6c3a8c = 3:  ed3f7035f vector-store: enable filtering on non-primary-key columns
4:  4f53ee1ab = 4:  cb0c5590c integration/tests: implement a simple test for filtering on non-primary-key columns
5:  ec6b4757a = 5:  bf54c00a9 validator: implement simple tests for non-primary-key filtering columns
6:  a96084a88 = 6:  8939deeaa validator: fix serde::test_decimal_key

m-szymon

I need to digest this PR a little more, so for now just two comments.

m-szymon · 2026-07-17T08:57:27Z

+        .filter_map(|timestamp_value| {
+            timestamp_value
+                .and_then(|(timestamp, value)| match (timestamp, value) {
+                    (Some(timestamp), value) => Ok(Some((timestamp, value))),
+                    (None, None) => Ok(None),
+                    (None, Some(_)) => {
+                        bail!("parse_values: missing timestamp for value");
+                    }
+                })
+                .transpose()
+        })


I am confused what this is supposed to filter.
If we filter anything, indexes of values no longer match column indexes.
Does (None, None) case have test?

I think there is an error -> there should be (None, _) => bail! - we need to have timestamp. I'll fix this.

fixed in 89af466

m-szymon · 2026-07-17T10:18:56Z

+        tokio::spawn(async move {
+            let _permit = permit;
+            if let Err(err) = consumer_data.process_upsert(primary_key, timestamp).await {
+                error!("Error processing upsert: {err}");
+            }


What if SELECT fails for any reason? In initial scan failures are recoverable, but here we lose that update, right?

Both select for fullscan and cdc query are idempotent - so it limits potential fails. Fullscan is partially recoverable. Here we don't repeat query - the question is how to retry it correctly - to limit such retries. Another question is what to do with failed insertions to the index - that can happen at the last stage of processing input stream, where we don't have data. I'm going to address fullscan issues in https://scylladb.atlassian.net/browse/VECTOR-726 as we need some way of looping over retries - I was thinking that we can fix both fullscan and cdc retry in that issue.

This commit extend fullscan select to support retrieving filtering columns: SELECT {primary_key_list}, (column-1) writetime(column-1), .. FROM {keyspace}.{table} This commit refactors db_index::range_scan_stream by adding parse_values and parse_primary_key functions. Parsing filtering columns is done in chunks of two values from db row.

…lues This commit refactors db_cdc to use only information about primary_key from CDC table. If the operation is DELETE it sends primary_key to the monitor_items to delete. If the operation is UPSERT it concurrently queries db for current target and filtering columns values. There is a limit for concurrency setup as 3 * num_workers. This commit adds retrieval of filtering column values - it adds supporting functions to the db_index_backend.

This commit enables filtering using HTTP API, so it allows filtering by CQL queries.

…ry-key columns This commit adds single ann_filter_filtering_columns_int_eq integration test. It tests if filtering by filtering columns is working. It doesn't test other types of restrictions as they are tested with filtering using primary key columns. This commit refactors creating db store for integration tests to support filtering columns.

This commit adds two tests - one for global indexes, second for local indexes - that check if filtering by additional filtering columns provided in the index definition is working. This commit refactors also tests in create_index group to support test new functionality.

It seems that this test is unreliable - sometimes ck values can be mixed in cdc or fullscan. This commit refactors the test to check possible values for ck.

ewienik · 2026-07-17T15:34:10Z

Changelog for 89af466

fix parse_values and add unit tests

range diff

1:  880050a34 ! 1:  d2937823d vector-store: implement filtering columns with fullscan
    @@ crates/vector-store/src/db_index.rs: impl Statements {
     +        let columns_len_expected = self.primary_key_columns.len().get()
     +            + (self.target_columns.len().get() + self.filtering_columns.len()) * 2;
     +        let target_columns_offset = self.primary_key_columns.len().get();
    -+        let target_columns_len = self.target_columns.len().get();
    ++        let target_columns_len = self.target_columns.len();
              let kind = self.kind.clone();
      
              // wait for an active session
    @@ crates/vector-store/src/db_index.rs: impl Statements {
     +pub(crate) fn parse_values(
     +    columns: impl IntoIterator<Item = Option<CqlValue>>,
     +    default_timestamp: Option<Timestamp>,
    -+    target_columns_len: usize,
    ++    target_columns_len: NonZeroUsize,
     +    kind: &IndexKind,
     +) -> anyhow::Result<NonemptyBox<Timestamped<DbIndexedValue>>> {
     +    let values = columns
    @@ crates/vector-store/src/db_index.rs: impl Statements {
     +            let value = chunk.next().ok_or(anyhow!(
     +                "parse_values: unable to get column value from chunk"
     +            ))?;
    -+            let value = if idx < target_columns_len {
    ++            let value = if idx < target_columns_len.get() {
     +                // target columns
     +                if let Some(value) = value {
     +                    Some(parse_indexed_value(value, kind)?)
    @@ crates/vector-store/src/db_index.rs: impl Statements {
     +            timestamp_value
     +                .and_then(|(timestamp, value)| match (timestamp, value) {
     +                    (Some(timestamp), value) => Ok(Some((timestamp, value))),
    -+                    (None, None) => Ok(None),
    -+                    (None, Some(_)) => {
    -+                        bail!("parse_values: missing timestamp for value");
    -+                    }
    ++                    (None, _) => bail!("parse_values: missing timestamp for value"),
     +                })
     +                .transpose()
     +        })
     +        .map_ok(|(timestamp, value)| Timestamped::new(timestamp, value))
     +        .collect::<anyhow::Result<Box<_>>>()?;
    -+    NonemptyBox::try_from(values)
    -+        .map_err(|err| anyhow!("parse_values: no values in the row: {err}"))
    ++    if target_columns_len.get() > values.len() {
    ++        bail!(
    ++            "parse_values: target len ({target_columns_len}) is greater than values len ({values_len})",
    ++            values_len = values.len()
    ++        );
    ++    }
    ++    Ok(NonemptyBox::try_from(values).unwrap())
     +}
     +
     +fn parse_primary_key(columns: impl IntoIterator<Item = Option<CqlValue>>) -> Option<PrimaryKey> {
    @@ crates/vector-store/src/db_index.rs: impl Statements {
                  }
              },
          }
    +@@ crates/vector-store/src/db_index.rs: mod tests {
    +     use crate::IndexOptionsVs;
    +     use crate::Quantization;
    +     use crate::SpaceType;
    ++    use std::assert_matches;
    + 
    +     fn vs_kind() -> IndexKind {
    +         IndexKind::Vs(IndexOptionsVs {
     @@ crates/vector-store/src/db_index.rs: mod tests {
              ]);
              let result = parse_indexed_value(cql, &vs_kind());
    @@ crates/vector-store/src/db_index.rs: mod tests {
              let result = parse_indexed_value(cql, &fts_kind());
     -        assert_eq!(result, None);
     +        assert!(result.is_err());
    ++    }
    ++
    ++    #[test]
    ++    fn parse_values_with_missing_timestamp() {
    ++        let columns = vec![
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(1.0)])),
    ++            Some(CqlValue::BigInt(1234567890)),
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(2.0)])),
    ++            None, // missing timestamp
    ++        ];
    ++
    ++        let result = parse_values(
    ++            columns.clone(),
    ++            None,
    ++            NonZeroUsize::new(1).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert!(result.is_err());
    ++        assert_matches!(
    ++            result
    ++                .unwrap_err()
    ++                .to_string(),
    ++                err if err.contains("missing timestamp for value")
    ++        );
    ++
    ++        let result = parse_values(
    ++            columns,
    ++            Some(Timestamp::from_millis(1)),
    ++            NonZeroUsize::new(1).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert!(result.is_ok());
    ++    }
    ++
    ++    #[test]
    ++    fn parse_values_with_wrong_timestamp() {
    ++        let columns = vec![
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(1.0)])),
    ++            Some(CqlValue::Int(1234567890)),
    ++        ];
    ++        let result = parse_values(
    ++            columns.clone(),
    ++            None,
    ++            NonZeroUsize::new(1).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert_matches!(
    ++            result
    ++                .unwrap_err()
    ++                .to_string(),
    ++                err if err.contains("bad type of a writetime")
    ++        );
    ++    }
    ++
    ++    #[test]
    ++    fn parse_values_with_wrong_value() {
    ++        let columns = vec![Some(CqlValue::Int(1)), Some(CqlValue::BigInt(1234567890))];
    ++        let result = parse_values(
    ++            columns.clone(),
    ++            None,
    ++            NonZeroUsize::new(1).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert_matches!(
    ++            result
    ++                .unwrap_err()
    ++                .to_string(),
    ++                err if err.contains("unsupported CQL type")
    ++        );
    ++    }
    ++
    ++    #[test]
    ++    fn parse_values_with_wrong_columns_len() {
    ++        let columns = vec![
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(1.0)])),
    ++            Some(CqlValue::BigInt(1234567890)),
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(1.0)])),
    ++        ];
    ++        let result = parse_values(
    ++            columns.clone(),
    ++            None,
    ++            NonZeroUsize::new(1).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert_matches!(
    ++            result
    ++                .unwrap_err()
    ++                .to_string(),
    ++                err if err.contains("unable to get timestamp value from chunk")
    ++        );
    ++    }
    ++
    ++    #[test]
    ++    fn parse_values_with_wrong_target_len() {
    ++        let columns = vec![
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(1.0)])),
    ++            Some(CqlValue::BigInt(1234567890)),
    ++            Some(CqlValue::Vector(vec![CqlValue::Float(2.0)])),
    ++            Some(CqlValue::BigInt(1234567891)),
    ++        ];
    ++        let result = parse_values(
    ++            columns.clone(),
    ++            None,
    ++            NonZeroUsize::new(3).unwrap(),
    ++            &vs_kind(),
    ++        );
    ++        assert_matches!(
    ++            result
    ++                .unwrap_err()
    ++                .to_string(),
    ++                err if err.contains("target len (3) is greater than values len (2)")
    ++        );
          }
      }
     
2:  50d802f7a ! 2:  5b902261c vector-store: refactor db_cdc to query db for target and filtering values
    @@ crates/vector-store/src/db_cdc/consumer.rs: use crate::AsyncInProgress;
     +            return Ok(());
     +        };
     +
    -+        let target_columns_len = self.target_columns.len().get();
    -+        let columns_len_expected = (target_columns_len + self.filtering_columns.len()) * 2;
    ++        let target_columns_len = self.target_columns.len();
    ++        let columns_len_expected = (target_columns_len.get() + self.filtering_columns.len()) * 2;
     +        if row.columns.len() != columns_len_expected {
     +            let msg = format!(
     +                "Unexpected number of columns in row: expected {columns_len_expected}, got {received_len}",
3:  ed3f7035f = 3:  4993fc3fe vector-store: enable filtering on non-primary-key columns
4:  cb0c5590c = 4:  d6235c44b integration/tests: implement a simple test for filtering on non-primary-key columns
5:  bf54c00a9 = 5:  d5e5dc89e validator: implement simple tests for non-primary-key filtering columns
6:  8939deeaa = 6:  89af4661b validator: fix serde::test_decimal_key

ewienik requested review from QuerthDP, knowack1 and m-szymon July 2, 2026 14:35

github-actions Bot added P3 area/vector labels Jul 2, 2026

ewienik force-pushed the vector-708-implement-filtering branch from 69ff28e to 3c3defe Compare July 2, 2026 15:18

ewienik mentioned this pull request Jul 13, 2026

vector-store: implement local vector indexes based on nonpk columns #517

Draft

ewienik force-pushed the vector-708-implement-filtering branch from 3c3defe to a96084a Compare July 16, 2026 11:14

ewienik force-pushed the vector-708-implement-filtering branch from a96084a to 8939dee Compare July 16, 2026 14:15

github-actions Bot added P2 and removed P3 labels Jul 16, 2026

ewienik marked this pull request as ready for review July 16, 2026 14:28

m-szymon reviewed Jul 17, 2026

View reviewed changes

ewienik added 6 commits July 17, 2026 17:28

vector-store: enable filtering on non-primary-key columns

4993fc3

This commit enables filtering using HTTP API, so it allows filtering by CQL queries.

validator: fix serde::test_decimal_key

89af466

It seems that this test is unreliable - sometimes ck values can be mixed in cdc or fullscan. This commit refactors the test to check possible values for ck.

ewienik force-pushed the vector-708-implement-filtering branch from 8939dee to 89af466 Compare July 17, 2026 15:34

ewienik requested a review from m-szymon July 17, 2026 15:35

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

vector-store: implement filtering by non-primary-key columns#501

vector-store: implement filtering by non-primary-key columns#501
ewienik wants to merge 6 commits into
scylladb:masterfrom
ewienik:vector-708-implement-filtering

ewienik commented Jul 2, 2026 •

edited

Loading

Uh oh!

ewienik commented Jul 2, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

m-szymon left a comment

Uh oh!

m-szymon Jul 17, 2026

Uh oh!

ewienik Jul 17, 2026

Uh oh!

ewienik Jul 17, 2026

Uh oh!

m-szymon Jul 17, 2026

Uh oh!

ewienik Jul 17, 2026

Uh oh!

ewienik commented Jul 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

ewienik commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ewienik commented Jul 2, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

m-szymon left a comment

Choose a reason for hiding this comment

Uh oh!

m-szymon Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

ewienik Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

ewienik Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

m-szymon Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

ewienik Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

ewienik commented Jul 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ewienik commented Jul 2, 2026 •

edited

Loading