vector-store: implement local vector indexes based on nonpk columns by ewienik · Pull Request #517 · scylladb/vector-store

ewienik · 2026-07-13T14:18:15Z

This PR adds support for nonpk partition key columns in local vector indexes. The solution is based on passing such column values using DbIndexedOperation::Upsert(Box<[DbIndexedValue]>) - the first are values for DbIndexedValue::Vector() or DbIndexedValue(Document) - target values to build index. The second are values for nonpk partition key column. The third are filtering values.

The PartitionKey for Index is built in table module - from values coming in DbIndexedRow. db_index and db_cdc actors retrieve additional data during fullscan or query after CDC. Table is responsible for synchronizing data incoming from fullscan or cdc (data coming from monitor_items).

This PR contains also refactor of IndexMetadata - to store primary_key_columns and partion_key_count (instead of calling for them from db_index actor).

This PR adds unit tests and validator e2e tests for new functionality.

This PR is based on #501 (last 6 commits are valid)

Fixes: VECTOR-721

In table module there exist similar names: partition_key_count and partition_key_columns - the first is for partition key for a table, second is for partition key of the index. Let's rename the first to better name its purpose.

This commit adds new enum DbIndexedValue::Filtering(CqlValue) to distinguish the third type of indexed value (vector, document, filtering). It implements also retrieving filtering columns values from DbIndexedRow::values - after the target value is taken. This commit refactors table::Column to support insert/remove methods. These methods cannot be combined as we need to check the CqlValue enum type before inserting. For this reason the intermediate list of column values is Box<[(Timestamp, Option<CqlValue>)]> instead of Box<[Timestamped<CqlValue>]> - for easier handling upserting/deletion of values. This commit refactors ColumnVec::update to check timestamp before updating.

…ations This commit adds DbIndexedOperation struct to distinguish two types of operations on a single db row: Upsert or Delete. It refactors DbIndexedRow::values as operation: DbIndexedOperation. Using such operations we can refactor TableAdd trait into TableModify with two methods: upsert and delete. With previous implementation it was hard to understand if we delete a row or only update with no values. It refactors Table::add into two methods - upsert and delete and provides a common function update_index. It refactors monitor_items::add function into two - upsert and delete and provides a common function process_operations.

This commit adds unit tests for split_values_filtering to check valid value or tombstone value and values with filtering. This commit adds a unit test to check if table properly upsert/delete filtering column. This commit adds unit test to validate monitor_items process DbIndexedOperation::Delete.

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.

Currently, primary_key_columns is taken from db_index. As this data is needed to calculate non-pk index partition key this commit put it into IndexMetadata for easier maintenance. This commit removes unneeded DbIndex::GetPrimaryKeyColumns message.

This commit refactors IndexMetadata and adds a new field partition_key_count to be aligned with primary_key_columns - as this value describes the partition_key part in primary_key_columns. This commit removes unneeded DbIndex::GetPartitionKeyCount message.

This commit adds support for nonpk partition key columns in local vector indexes by passing such column values using DbIndexedOperation::Upsert(Box<[DbIndexedValue]>) - the first are values for DbIndexedValue::Vector() or DbIndexedValue(Document) - target values to build index. The second are values for nonpk partition key column. The third are filtering values. The PartitionKey for Index is built in table module - from values coming in DbIndexedRow. db_index and db_cdc actors retrieve additional data during fullscan or query after CDC. Table is responsible for synchronizing data incoming from fullscan or cdc (data coming from monitor_items). This commit fixes parsing index options to support all possible table columns. Tests are prepared in following commits.

It seems that preparing available_columns in table module is unnecessary, as available columns are checked in httproutes.

This commit updates flow test to check all possible combination of local vector indexes keys: partition keys, clustering keys, regular keys.

This commit introduces validator tests for creating local indexes based on clustering or regular columns. It adds tests for supporting such indexes in cdc and for testing filtering on such indexes.

ewienik · 2026-07-16T11:15:52Z

Changelog for 7e4ad48

rebase to master

range diff

1:  117214984 = 1:  61d58daf7 vector-store: add primary_key_columns into IndexMetadata
2:  a0d436462 ! 2:  0fbe8e07e vector-store: add partition_key_count into IndexMetadata
    @@ crates/vector-store/src/table/mod.rs: fn partition_key(
      #[derive(Debug)]
      pub struct Table {
          primary_key_columns: NonemptyArc<ColumnName>,
    --    partition_key_count: usize,
    -+    partition_key_count: NonZeroUsize,
    +-    partition_primary_key_count: usize,
    ++    partition_primary_key_count: NonZeroUsize,
          needs_ck_normalization: bool,
          primary_ids: BTreeMap<PrimaryKey, PrimaryId>,
          free_primary_ids: FreePrimaryIds,
    @@ crates/vector-store/src/table/mod.rs: impl Table {
          pub(crate) fn new(
              index_key: IndexKey,
              primary_key_columns: NonemptyArc<ColumnName>,
    --        partition_key_count: usize,
    -+        partition_key_count: NonZeroUsize,
    +-        partition_primary_key_count: usize,
    ++        partition_primary_key_count: NonZeroUsize,
              partition_key_columns: Option<NonemptyArc<ColumnName>>,
              column_targets_count: NonZeroUsize,
              filtering_columns: Arc<[ColumnName]>,
              table_columns: Arc<HashMap<ColumnName, NativeType>>,
          ) -> anyhow::Result<Self> {
    --        let partition_key_count = partition_key_count.min(primary_key_columns.len().get());
    -+        if partition_key_count > primary_key_columns.len() {
    +-        let partition_primary_key_count =
    +-            partition_primary_key_count.min(primary_key_columns.len().get());
    ++        if partition_primary_key_count > primary_key_columns.len() {
     +            bail!(
    -+                "Partition key count ({partition_key_count}) cannot be greater than primary key columns count ({primary_count}) for index {index_key:?}",
    ++                "Partition key count ({partition_primary_key_count}) \
    ++                cannot be greater than primary key columns \
    ++                count ({primary_count}) for index {index_key:?}",
     +                primary_count = primary_key_columns.len(),
     +            );
     +        }
    @@ crates/vector-store/src/table/mod.rs: impl Table {
                  })
                  .collect::<anyhow::Result<Vec<_>>>()?;
              columns.extend(column_with_values);
    --        let needs_ck_normalization = primary_key_columns.as_slice()[partition_key_count..]
    -+        let needs_ck_normalization = primary_key_columns.as_slice()[partition_key_count.get()..]
    +-        let needs_ck_normalization = primary_key_columns.as_slice()[partition_primary_key_count..]
    ++        let needs_ck_normalization = primary_key_columns.as_slice()
    ++            [partition_primary_key_count.get()..]
                  .iter()
                  .any(|col| matches!(table_columns.get(col), Some(NativeType::Decimal)));
              let mut table = Self {
    @@ crates/vector-store/src/table/mod.rs: impl Table {
              let normalized: PrimaryKey = (0..key.len())
                  .map(|idx| {
                      let value = key.get(idx).expect("primary key column exists");
    --                if idx >= self.partition_key_count {
    -+                if idx >= self.partition_key_count.get() {
    +-                if idx >= self.partition_primary_key_count {
    ++                if idx >= self.partition_primary_key_count.get() {
                          normalize(value)
                      } else {
                          value
    @@ crates/vector-store/src/table/mod.rs: mod tests {
                      partition_key_columns.clone(),
                      NonZeroUsize::new(1).unwrap(),
                      Arc::new([]),
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +         let mut table = Table::new(
    +             index_key.clone(),
    +             NonemptyArc::new(["p"]).unwrap(),
    +-            1,
    ++            NonZeroUsize::new(1).unwrap(),
    +             None,
    +             NonZeroUsize::new(1).unwrap(),
    +             Arc::new([]),
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             let mut table = Table::new(
    +                 index_key.clone(),
    +                 primary_key_columns,
    +-                1,
    ++                NonZeroUsize::new(1).unwrap(),
    +                 partition_key_columns,
    +                 NonZeroUsize::new(1).unwrap(),
    +                 Arc::new(["f".into()]),
     
      ## crates/vector-store/tests/integration/db_basic.rs ##
     @@ crates/vector-store/tests/integration/db_basic.rs: use scylla::value::CqlTimeuuid;
3:  901a08111 ! 3:  bfd09cac3 vector-store: add support for nonpk partition key columns
    @@ crates/vector-store/src/db.rs: fn from_target_option(
              DbIndexPartitioning::Local(
                  target
     
    - ## crates/vector-store/src/db_cdc.rs ##
    -@@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerData {
    + ## crates/vector-store/src/db_cdc/consumer.rs ##
    +@@ crates/vector-store/src/db_cdc/consumer.rs: struct CdcConsumerData {
          st_select_values: PreparedStatement,
          index_key: IndexKey,
          primary_key_columns: NonemptyArc<ColumnName>,
    @@ crates/vector-store/src/db_cdc.rs: struct CdcConsumerData {
          target_columns: NonemptyArc<ColumnName>,
          filtering_columns: Arc<[ColumnName]>,
          kind: IndexKind,
    -@@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerData {
    +@@ crates/vector-store/src/db_cdc/consumer.rs: impl CdcConsumerData {
              };
      
              let target_columns_len = self.target_columns.len().get();
    @@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerData {
              if row.columns.len() != columns_len_expected {
                  let msg = format!(
                      "Unexpected number of columns in row: expected {columns_len_expected}, got {received_len}",
    -@@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
    +@@ crates/vector-store/src/db_cdc/consumer.rs: impl CdcConsumerFactory {
                  .cloned()
                  .collect();
      
    @@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
                  primary_key_columns.iter(),
              );
              let st_select_values =
    -@@ crates/vector-store/src/db_cdc.rs: impl CdcConsumerFactory {
    +@@ crates/vector-store/src/db_cdc/consumer.rs: impl CdcConsumerFactory {
                  st_select_values,
                  index_key: metadata.key(),
                  primary_key_columns,
    @@ crates/vector-store/src/lib.rs: impl IndexMetadata {
      #[derive(Clone, Debug, PartialEq, Eq, Hash)]
     
      ## crates/vector-store/src/table/mod.rs ##
    +@@ crates/vector-store/src/table/mod.rs: impl FreePartitionIds {
    +             .pop_front()
    +             .ok_or(anyhow!("Partition ids should be reserved"))
    +     }
    +-
    +-    fn return_id(&mut self, id: PartitionId) {
    +-        self.0.push_back(id);
    +-    }
    + }
    + 
    + /// An enum that represents the data of an index specific to the index type.
     @@ crates/vector-store/src/table/mod.rs: enum IndexData {
          Local {
              /// Column names for which the index is built. The order of column names is important, as it
    @@ crates/vector-store/src/table/mod.rs: enum IndexData {
      
              map: BTreeMap<PartitionKey, PartitionId>,
              free_ids: FreePartitionIds,
    +@@ crates/vector-store/src/table/mod.rs: impl IndexData {
    +                 keys,
    +                 sizes,
    +                 map,
    +-                free_ids,
    +                 ..
    +             } => {
    +-                let Some(Some(partition_id)) = ids.get_mut(primary_id).map(|id| id.take()) else {
    ++                let Some(partition_id) = ids.get(primary_id).copied().flatten() else {
    +                     return false;
    +                 };
    + 
    +@@ crates/vector-store/src/table/mod.rs: impl IndexData {
    +                 if is_empty && let Some(key) = keys.get_mut(partition_id).and_then(|key| key.take())
    +                 {
    +                     map.remove(&key);
    +-                    free_ids.return_id(partition_id);
    +                 }
    +                 is_empty
    +             }
     @@ crates/vector-store/src/table/mod.rs: impl Index {
      
          fn new_local(
    @@ crates/vector-store/src/table/mod.rs: impl Index {
              primary_id: PrimaryId,
              primary_key: &PrimaryKey,
              primary_key_columns: &[ColumnName],
    -+        values: &NonemptyBox<Timestamped<DbIndexedValue>>,
    ++        values: Option<&NonemptyBox<Timestamped<DbIndexedValue>>>,
          ) -> anyhow::Result<PartitionId> {
              match &mut self.data {
                  IndexData::Global => Ok(PartitionId::global(index_id)),
    @@ crates/vector-store/src/table/mod.rs: impl Index {
                      }
     -                let partition_key =
     -                    partition_key(primary_key, primary_key_columns, key_columns.as_slice())?;
    -+                let left = column_targets_count.get();
    -+                let right = left + nonpk_partition_key_columns.len();
    -+                if right > values.len().get() {
    -+                    bail!(
    -+                        "Not enough values to construct partition key: \
    -+                        expected at least {right}, got {len}",
    -+                        len = values.len()
    -+                    );
    -+                }
    ++                let values = if nonpk_partition_key_columns.is_empty() {
    ++                    &[]
    ++                } else {
    ++                    let Some(values) = values else {
    ++                        // There is operation only on primary key, so let's create a new partition
    ++                        // id for it and don't fill other fields in the IndexData as a sign of a
    ++                        // tombstone.
    ++                        let partition_id = free_ids.take_id()?;
    ++                        partition_id_storage.replace(partition_id);
    ++                        return Ok(partition_id);
    ++                    };
    ++                    let left = column_targets_count.get();
    ++                    let right = left + nonpk_partition_key_columns.len();
    ++                    if right > values.len().get() {
    ++                        bail!(
    ++                            "Not enough values to construct partition key: \
    ++                            expected at least {right}, got {len}",
    ++                            len = values.len()
    ++                        );
    ++                    }
    ++                    &values.as_slice()[left..right]
    ++                };
     +                let partition_key = partition_key(
     +                    primary_key,
     +                    primary_key_columns,
     +                    partition_key_columns.as_slice(),
     +                    nonpk_partition_key_columns,
    -+                    &values.as_slice()[left..right],
    ++                    values,
     +                )?;
                      match map.entry(partition_key.clone()) {
                          Entry::Occupied(entry) => {
    @@ crates/vector-store/src/table/mod.rs: impl TableModify for Table {
                  primary_id,
                  &primary_key,
                  self.primary_key_columns.as_slice(),
    -+            &values,
    ++            Some(&values),
              )?;
      
     +        let nonpk_partition_key_columns_count = match &index.data {
    @@ crates/vector-store/src/table/mod.rs: impl TableModify for Table {
              )
          }
      
    +@@ crates/vector-store/src/table/mod.rs: impl TableModify for Table {
    +             primary_id,
    +             &primary_key,
    +             self.primary_key_columns.as_slice(),
    ++            None,
    +         )?;
    + 
    +         let filtering = index
     @@ crates/vector-store/src/table/mod.rs: impl TableSearch for Table {
              match &index.data {
                  IndexData::Global => Some((PartitionId::global(*index_id), restrictions)),
    @@ crates/vector-store/src/table/mod.rs: impl TableSearch for Table {
                      })
                      .and_then(|(partition_key, restrictions)| {
                          map.get(&partition_key)
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             Some(DbIndexedValue::Vector(vec![1.0].into())),
    +         );
    +         let values = NonemptyBox::new([value]).unwrap();
    +-        let split = split_values_filtering(values).unwrap();
    ++        let split = split_values_filtering(values, 0).unwrap();
    +         assert_eq!(
    +             split.values,
    +             Some(SplittingValues::Vector(vec![1.0].into()))
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +     fn split_values_filtering_only_values_tombstone() {
    +         let value = Timestamped::new(Timestamp::from_millis(2), None);
    +         let values = NonemptyBox::new([value]).unwrap();
    +-        let split = split_values_filtering(values).unwrap();
    ++        let split = split_values_filtering(values, 0).unwrap();
    +         assert!(split.values.is_none());
    +         assert_eq!(split.timestamps.len().get(), 1);
    +         assert_eq!(
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             Some(DbIndexedValue::Filtering(CqlValue::Int(42))),
    +         );
    +         let values = NonemptyBox::new([value, filtering]).unwrap();
    +-        let split = split_values_filtering(values).unwrap();
    ++        let split = split_values_filtering(values, 0).unwrap();
    +         assert_eq!(
    +             split.values,
    +             Some(SplittingValues::Vector(vec![2.0].into()))
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             Some(DbIndexedValue::Filtering(CqlValue::Int(1))),
    +         );
    +         let values = NonemptyBox::new([value]).unwrap();
    +-        assert!(split_values_filtering(values).is_err());
    ++        assert!(split_values_filtering(values, 0).is_err());
    + 
    +         // Vector in the filtering place is not allowed
    +         let value1 = Timestamped::new(
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             Some(DbIndexedValue::Vector(vec![2.0].into())),
    +         );
    +         let values = NonemptyBox::new([value1, value2]).unwrap();
    +-        assert!(split_values_filtering(values).is_err());
    ++        assert!(split_values_filtering(values, 0).is_err());
    + 
    +         // Document in the filtering place is not allowed
    +         let value1 = Timestamped::new(
    +@@ crates/vector-store/src/table/mod.rs: mod tests {
    +             Some(DbIndexedValue::Document("doc2".to_string())),
    +         );
    +         let values = NonemptyBox::new([value1, value2]).unwrap();
    +-        assert!(split_values_filtering(values).is_err());
    ++        assert!(split_values_filtering(values, 0).is_err());
    +     }
    + 
    +     #[test]
4:  f9d308796 = 4:  af346c66f vector-store: remove unneded _available_columns
5:  5b924aa0a = 5:  f549cf5b5 vector-store: refactor table unit tests
6:  7b754ac7c = 6:  7e4ad48a4 validator: implement e2e tests for local index with non-pk columns

ewienik requested review from QuerthDP, knowack1 and m-szymon July 13, 2026 14:18

github-actions Bot added P2 area/vector labels Jul 13, 2026

ewienik added 16 commits July 16, 2026 11:31

vector-store: refactor name partition_primary_key_count

48f997c

In table module there exist similar names: partition_key_count and partition_key_columns - the first is for partition key for a table, second is for partition key of the index. Let's rename the first to better name its purpose.

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

56c6c3a

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

validator: fix serde::test_decimal_key

a96084a

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.

vector-store: remove unneded _available_columns

af346c6

It seems that preparing available_columns in table module is unnecessary, as available columns are checked in httproutes.

vector-store: refactor table unit tests

f549cf5

This commit updates flow test to check all possible combination of local vector indexes keys: partition keys, clustering keys, regular keys.

validator: implement e2e tests for local index with non-pk columns

7e4ad48

This commit introduces validator tests for creating local indexes based on clustering or regular columns. It adds tests for supporting such indexes in cdc and for testing filtering on such indexes.

ewienik force-pushed the vector-721-local-nonpk-indexes branch from 7b754ac to 7e4ad48 Compare July 16, 2026 11:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

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

vector-store: implement local vector indexes based on nonpk columns#517
ewienik wants to merge 16 commits into
scylladb:masterfrom
ewienik:vector-721-local-nonpk-indexes

ewienik commented Jul 13, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

ewienik commented Jul 13, 2026

Uh oh!

ewienik commented Jul 16, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant