support storing, filtering and retrieving non-key columns

README.md

@@ -32,6 +32,7 @@ service supports also `.env` files.
 | `VECTOR_STORE_CDC_FINE_SAFETY_INTERVAL`    | Fine-grained CDC reader's safety interval for low-latency updates (ie. `100ms`)                                                                                                      | `100ms`                  |
 | `VECTOR_STORE_CDC_FINE_SLEEP_INTERVAL`     | Fine-grained CDC reader's sleep interval for low-latency updates (ie. `500ms`)                                                                                                       | `500ms`                  |
 | `VECTOR_STORE_MONITOR_INDEXES_INTERVAL`    | How often to poll Scylla for schema changes (new/removed vector indexes). The value is in human readable format (ie. `100ms`)                                                        | `1s`                     |
+| `VECTOR_STORE_INDEX_STATUS_UPDATE_INTERVAL` | How often to sync index status (e.g., BOOTSTRAPPING->SERVING) into the engine's cached state. The value is in human readable format (ie. `100ms`) | `1s`            |
 | `VECTOR_STORE_USEARCH_SIMULATOR`           | Enable simulator for USearch. Provides human readable delays for simulated operations (`search:add-remove:reserve`).                                                                 |                          |
 | `VECTOR_STORE_ALTER_INDEX_SIMULATOR`       | Enable simulator for missing `ALTER INDEX`. When enable indexes aren't deleted and their version is not checked.                                                                     | `false`                  |
 

api/openapi.json

@@ -407,6 +407,13 @@
           "limit": {
             "$ref": "#/components/schemas/Limit"
           },
+          "return_columns": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/ColumnName"
+            },
+            "description": "Filtering-column names whose stored values should be returned alongside\nthe primary keys.  Empty (the default) means return no column values."
+          },
           "vector": {
             "$ref": "#/components/schemas/Vector"
           }
@@ -420,6 +427,18 @@
           "similarity_scores"
         ],
         "properties": {
+          "column_values": {
+            "type": "object",
+            "description": "Per-column stored values for the filtering columns requested in\n`return_columns`.  Each entry maps a column name to a Vec of\n`Option<Value>` — one entry per returned nearest neighbour, in the\nsame order as `similarity_scores`.  `None` means the value was not\npresent for that row (e.g. the attribute did not exist when the row\nwas indexed).  Absent when `return_columns` was empty.",
+            "additionalProperties": {
+              "type": "array",
+              "items": {}
+            },
+            "propertyNames": {
+              "type": "string",
+              "description": "Name of the column in a db table."
+            }
+          },
           "distances": {
             "type": "array",
             "items": {

crates/httpapi/src/lib.rs

@@ -270,13 +270,25 @@ pub struct PostIndexAnnRequest {
     pub filter: Option<PostIndexAnnFilter>,
     #[serde(default)]
     pub limit: Limit,
+    /// Filtering-column names whose stored values should be returned alongside
+    /// the primary keys.  Empty (the default) means return no column values.
+    #[serde(default)]
+    pub return_columns: Vec<ColumnName>,
 }
 
 #[derive(serde::Deserialize, serde::Serialize, utoipa::ToSchema)]
 pub struct PostIndexAnnResponse {
     pub primary_keys: HashMap<ColumnName, Vec<Value>>,
     pub distances: Vec<Distance>,
     pub similarity_scores: Vec<SimilarityScore>,
+    /// Per-column stored values for the filtering columns requested in
+    /// `return_columns`.  Each entry maps a column name to a Vec of
+    /// `Option<Value>` — one entry per returned nearest neighbour, in the
+    /// same order as `similarity_scores`.  `None` means the value was not
+    /// present for that row (e.g. the attribute did not exist when the row
+    /// was indexed).  Absent when `return_columns` was empty.
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub column_values: HashMap<ColumnName, Vec<Option<Value>>>,
 }
 
 #[derive(

crates/httpclient/src/lib.rs

@@ -88,6 +88,7 @@ impl HttpClient {
             vector,
             filter,
             limit,
+            return_columns: vec![],
         };
         self.post_ann_data(keyspace_name, index_name, &request)
             .await

crates/testclient/src/lib.rs

@@ -59,6 +59,7 @@ impl TestClient {
                 vector,
                 filter,
                 limit,
+                return_columns: vec![],
             })
             .await
     }

crates/validator/src/cdc.rs

@@ -51,6 +51,11 @@ pub(crate) async fn new() -> TestCase<TestActors> {
             timeout,
             cql_per_row_ttl_expires_from_index,
         )
+        .with_test(
+            "cql_explicit_vector_column_deletion_removes_from_index",
+            timeout,
+            cql_explicit_vector_column_deletion_removes_from_index,
+        )
 }
 
 #[framed]
@@ -659,3 +664,94 @@ async fn cql_per_row_ttl_expires_from_index(actors: TestActors) {
 
     info!("finished");
 }
+
+/// Regression test: explicitly deleting the vector column (`DELETE v FROM t WHERE pk = ?`)
+/// must remove the item from the vector index, not leave it as "unchanged".
+///
+/// Previously the CQL CDC path only treated full row/partition deletions as
+/// removals.  An explicit column tombstone produced `embedding = None`
+/// ("unchanged") instead of `Some(None)` ("delete from index"), so the stale
+/// vector remained searchable.
+#[framed]
+async fn cql_explicit_vector_column_deletion_removes_from_index(actors: TestActors) {
+    info!("started");
+
+    let (session, clients) = prepare_connection_single_vs(&actors).await;
+    let client = clients.first().unwrap();
+    let keyspace = create_keyspace(&session).await;
+    let table = create_table(&session, "pk INT PRIMARY KEY, v VECTOR<FLOAT, 3>", None).await;
+
+    session
+        .query_unpaged(
+            format!("INSERT INTO {table} (pk, v) VALUES (1, [1.0, 0.0, 0.0])"),
+            (),
+        )
+        .await
+        .expect("failed to insert data");
+    session
+        .query_unpaged(
+            format!("INSERT INTO {table} (pk, v) VALUES (2, [0.0, 1.0, 0.0])"),
+            (),
+        )
+        .await
+        .expect("failed to insert data");
+
+    let index = create_index(CreateIndexQuery::new(&session, &clients, &table, "v")).await;
+
+    let status = wait_for_index(client, &index).await;
+    assert_eq!(
+        status.count, 2,
+        "Index should have 2 vectors after full scan"
+    );
+
+    // Explicitly delete only the vector column of pk=1 (not the whole row).
+    // This is a column tombstone, not a row deletion — the CQL CDC path
+    // must recognise it via is_value_deleted() and remove the vector from the index.
+    session
+        .query_unpaged(format!("DELETE v FROM {table} WHERE pk = 1"), ())
+        .await
+        .expect("failed to delete vector column");
+
+    // Wait for the index count to drop to 1.
+    wait_for(
+        || async {
+            let status = client.index_status(&index.keyspace, &index.index).await;
+            matches!(status, Ok(s) if s.count == 1)
+        },
+        "Waiting for index count to drop to 1 after explicit column deletion",
+        FINE_GRAINED_CDC_MAX_LATENCY,
+    )
+    .await;
+
+    // ANN query should only return pk=2.
+    let result = wait_for_value(
+        || async {
+            let result = get_opt_query_results(
+                format!("SELECT pk FROM {table} ORDER BY v ANN OF [0.0, 1.0, 0.0] LIMIT 10"),
+                &session,
+            )
+            .await;
+            result.filter(|r| r.rows_num() == 1)
+        },
+        "Waiting for ANN query to return only pk=2",
+        FINE_GRAINED_CDC_MAX_LATENCY,
+    )
+    .await;
+    let pks: Vec<i32> = result
+        .rows::<(i32,)>()
+        .expect("failed to get rows")
+        .map(|r| r.expect("row error").0)
+        .collect();
+    assert_eq!(
+        pks,
+        vec![2],
+        "Only pk=2 should remain after explicit vector column deletion"
+    );
+
+    session
+        .query_unpaged(format!("DROP KEYSPACE {keyspace}"), ())
+        .await
+        .expect("failed to drop keyspace");
+
+    info!("finished");
+}

crates/validator/src/filtering.rs

@@ -100,6 +100,11 @@ pub(crate) async fn new() -> TestCase<TestActors> {
             timeout,
             local_ann_with_timestamp_gte_filter,
         )
+        .with_test(
+            "ann_filter_by_non_pk_filtering_column",
+            timeout,
+            ann_filter_by_non_pk_filtering_column,
+        )
 }
 
 /// Test ANN search filtered by partition key equality.
@@ -1044,6 +1049,87 @@ async fn local_ann_with_timestamp_gte_filter(actors: TestActors) {
     info!("finished");
 }
 
+/// Test ANN search filtered by a non-primary-key CQL filtering column.
+///
+/// The index is created with `category` as a filtering column.  Rows have two
+/// category values (0 and 1).  Querying with `WHERE category = 0` must return
+/// only the rows whose category is 0.
+#[framed]
+async fn ann_filter_by_non_pk_filtering_column(actors: TestActors) {
+    info!("started");
+
+    let (session, clients) = prepare_connection(&actors).await;
+
+    let keyspace = create_keyspace(&session).await;
+    let table = create_table(
+        &session,
+        "pk INT PRIMARY KEY, category INT, v VECTOR<FLOAT, 3>",
+        None,
+    )
+    .await;
+
+    // Insert 6 rows: pk 0..5, alternating category 0 and 1.
+    for pk in 0..6_i32 {
+        session
+            .query_unpaged(
+                format!("INSERT INTO {table} (pk, category, v) VALUES (?, ?, ?)"),
+                (pk, pk % 2, &vec![pk as f32, 0.0, 0.0]),
+            )
+            .await
+            .expect("failed to insert data");
+    }
+
+    let index = create_index(
+        CreateIndexQuery::new(&session, &clients, &table, "v").filter_columns(["category"]),
+    )
+    .await;
+
+    for client in &clients {
+        let index_status = wait_for_index(client, &index).await;
+        assert_eq!(index_status.count, 6, "Expected 6 vectors to be indexed");
+    }
+
+    let result = wait_for_value(
+        || async {
+            let result = get_opt_query_results(
+                format!(
+                    "SELECT pk, category FROM {table} \
+                     WHERE category = 0 \
+                     ORDER BY v ANN OF [0.0, 0.0, 0.0] LIMIT 10 \
+                     ALLOW FILTERING"
+                ),
+                &session,
+            )
+            .await;
+            result.filter(|r| r.rows_num() == 3)
+        },
+        "Waiting for category=0 filtered ANN query to return 3 rows",
+        DEFAULT_OPERATION_TIMEOUT,
+    )
+    .await;
+
+    let rows: Vec<(i32, i32)> = result
+        .rows::<(i32, i32)>()
+        .expect("failed to get rows")
+        .map(|row| row.expect("failed to get row"))
+        .collect();
+
+    assert_eq!(rows.len(), 3, "Expected exactly 3 rows with category=0");
+    for (pk, category) in &rows {
+        assert_eq!(
+            *category, 0,
+            "Expected all rows to have category=0, got pk={pk} category={category}"
+        );
+    }
+
+    session
+        .query_unpaged(format!("DROP KEYSPACE {keyspace}"), ())
+        .await
+        .expect("failed to drop a keyspace");
+
+    info!("finished");
+}
+
 #[framed]
 async fn ann_filter_by_clustering_key_only_requires_allow_filtering(actors: TestActors) {
     info!("started");

crates/vector-store/benches/pipeline.rs

@@ -265,8 +265,9 @@ fn scan_fn_mpsc(
                     .send((
                         DbEmbedding {
                             primary_key,
-                            embedding,
+                            embedding: Some(embedding),
                             timestamp,
+                            column_values: Default::default(),
                         },
                         in_progress,
                     ))
@@ -1139,7 +1140,7 @@ where
             })
         })
         .collect_vec();
-    stream::iter(tasks.into_iter())
+    stream::iter(tasks)
         .then(|task| async move { task.await.unwrap() })
         .fold(Duration::ZERO, |acc, x| async move { acc + x })
         .await

crates/vector-store/src/config_manager.rs

@@ -491,6 +491,12 @@ pub async fn load_config(env: impl Fn(&str) -> anyhow::Result<String>) -> anyhow
         .transpose()?
         .map(|v| v.into());
 
+    config.engine_status_update_interval = env("VECTOR_STORE_INDEX_STATUS_UPDATE_INTERVAL")
+        .ok()
+        .map(|v| v.parse::<humantime::Duration>())
+        .transpose()?
+        .map(|v| v.into());
+
     config.cql_uri_translation_map = env("VECTOR_STORE_CQL_URI_TRANSLATION_MAP")
         .ok()
         .map(|v| serde_json::from_str(&v))

crates/vector-store/src/db.rs

@@ -42,7 +42,6 @@
 use scylla::client::session::Session;
 use scylla::client::session::TlsContext;
 use scylla::client::session_builder::SessionBuilder;
-use scylla::cluster::metadata::ColumnType;
 use scylla::cluster::metadata::Table;
 use scylla::statement::prepared::PreparedStatement;
 use scylla::value::CqlTimeuuid;
@@ -748,10 +747,10 @@
                         .tables
                         .get(&table_name)
                         .ok_or_else(|| {
                             anyhow!("table {table_name} does not exist").context(InvalidMetadata)
                         })?;
                     Ok(options.remove("target").and_then(|target| {
-                        from_target_option(table, target)
+                        from_target_option(table, &KeyspaceName::from(keyspace_name.as_str()), target)
                             .map(
                                 |(index_type, target_column, filtering_columns)| DbCustomIndex {
                                     keyspace: keyspace_name.into(),
@@ -1015,6 +1014,7 @@
 
 fn from_target_option(
     table: &Table,
+    keyspace_name: &KeyspaceName,
     value: String,
 ) -> anyhow::Result<(DbIndexType, ColumnName, Vec<ColumnName>)> {
     let Some(target) = parse_target_option(table, &value)? else {
@@ -1023,13 +1023,7 @@
     };
 
     let validate_target_type = |target_name: &str| -> anyhow::Result<()> {
-        let column = table.columns.get(target_name).ok_or_else(|| {
-            anyhow!("invalid target option: column {target_name} does not exist in a table")
-        })?;
-        if !matches!(column.typ, ColumnType::Vector { .. }) {
-            bail!("invalid target option: column {target_name} is not a vector column in a table");
-        }
-        Ok(())
+        db_index_backend::validate_target_type(table, keyspace_name, target_name)
     };
 
     validate_target_type(&target.target_column)?;