[ENH] Wireup regex in filter operator (#4452)

## Description of changes

_Summarize the changes made by this PR._

- Improvements & Bug fixes
  - Return prefix as part of `get_range` method
- New functionality
  - Wire up regex evaluation process in filter operator

## Test plan

_How are these changes tested?_

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js, `cargo test` for rust

## Documentation Changes

_Are all docstrings for user-facing APIs updated if required? Do we need to make documentation changes in the [docs section](https://github.com/chroma-core/chroma/tree/main/docs/docs.trychroma.com)?_
<!-- Summary by @propel-code-bot -->

---

**Implementing Regex Matching in Filter Operator**

This PR implements regex evaluation within the filter operator, enabling document-level regex matching. It changes blockfile reader APIs to expose prefixes as part of query results and leverages full-text indexes for more efficient regex evaluation. The implementation includes optimizations for exact pattern matching that avoids re-scanning documents when possible.

**Key Changes:**
• Add regex pattern evaluation in filter operator using `ChromaRegex` from `chroma_types`
• Update blockfile reader `APIs` to return prefixes along with keys and values
• Implement `NgramLiteralProvider` for `FullTextIndexReader` to optimize regex searches
• Add benchmark for regex matching with various patterns
• Return prefix as part of `get_range` and `get_range_stream` methods in blockstore


**Affected Areas:**
• Worker filter operator implementation
• Blockstore reader/writer ``API``
• `FullText` index implementation
• Distributed segment implementation
• Metadata segment implementation


**Potential Impact:**

**Functionality**: Adds regex matching capability for document filtering, providing users with more powerful query options

**Performance**: Optimizes regex matching by using full-text indexes for preliminary candidate selection before regex validation, improving search performance for complex patterns

**Security**: No significant security implications

**Scalability**: Efficient regex implementation should maintain good performance as document collections grow

**Review Focus:**
• Regex implementation efficiency in filter.rs
• Error handling for regex pattern compilation and matching
• ``API`` changes to return prefix in blockstore methods
• Benchmark methodology for regex performance evaluation


<details>
<summary><strong>Testing Needed</strong></summary>

• Benchmark different regex patterns against large document collections
• Test complex regex patterns with edge cases
• Verify consistent results between optimized and brute-force implementations
• Test with documents of varying sizes and content

</details>

<details>
<summary><strong>Code Quality Assessment</strong></summary>

**rust/worker/src/execution/operators/filter.rs**: Well-structured implementation with good error handling, but has a potential performance bottleneck with sequential document fetching

**rust/blockstore/src/memory/reader_writer.rs**: Multiple unwrap() calls could be replaced with expect() for better error messages

**rust/index/src/fulltext/types.rs**: Possible lifetime parameter issues in the lookup_ngram_range method


</details>

<details>
<summary><strong>Best Practices</strong></summary>

**Performance**:
• Consider parallelizing document fetches with `buffer_unordered` for concurrent I/O

**Error Handling**:
• Replace unwrap() with expect() for better error messages
• Consider proper error propagation instead of unwrapping


</details>

<details>
<summary><strong>Possible Issues</strong></summary>

• Potential panics from unwrap() calls in several places
• Sequential document fetching in filter operator might cause performance issues with large result sets
• Lifetime parameter conflicts in lookup_ngram_range method
• Fallback bitmap handling in assertions could cause test failures

</details>

---
*This summary was automatically generated by @propel-code-bot*

Author: Sicheng-Pan

rust/blockstore/src/arrow/block/types.rs

@@ -361,7 +361,7 @@ impl Block {
         &'me self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> impl Iterator<Item = (K, V)> + 'me
+    ) -> impl Iterator<Item = (&'me str, K, V)> + 'me
     where
         PrefixRange: RangeBounds<&'prefix str>,
         KeyRange: RangeBounds<K>,
@@ -410,8 +410,16 @@ impl Block {
             Bound::Unbounded => self.len(),
         };
 
+        let prefix_array = self
+            .data
+            .column(0)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap();
+
         (start_index..end_index).map(move |index| {
             (
+                prefix_array.value(index),
                 K::get(self.data.column(1), index),
                 V::get(self.data.column(2), index),
             )

rust/blockstore/src/arrow/blockfile.rs

@@ -496,7 +496,7 @@ impl<'me, K: ArrowReadableKey<'me> + Into<KeyWrapper>, V: ArrowReadableValue<'me
         &'me self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> impl Stream<Item = Result<(K, V), Box<dyn ChromaError>>> + Send + 'me
+    ) -> impl Stream<Item = Result<(&'me str, K, V), Box<dyn ChromaError>>> + Send + 'me
     where
         PrefixRange: RangeBounds<&'prefix str> + Clone + Send + 'me,
         KeyRange: RangeBounds<K> + Clone + Send + 'me,
@@ -533,7 +533,7 @@ impl<'me, K: ArrowReadableKey<'me> + Into<KeyWrapper>, V: ArrowReadableValue<'me
         &'me self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> Result<Vec<(K, V)>, Box<dyn ChromaError>>
+    ) -> Result<Vec<(&'me str, K, V)>, Box<dyn ChromaError>>
     where
         PrefixRange: RangeBounds<&'prefix str> + Clone,
         KeyRange: RangeBounds<K> + Clone,
@@ -543,7 +543,7 @@ impl<'me, K: ArrowReadableKey<'me> + Into<KeyWrapper>, V: ArrowReadableValue<'me
             .sparse_index
             .get_block_ids_range(prefix_range.clone(), key_range.clone());
 
-        let mut result: Vec<(K, V)> = vec![];
+        let mut result: Vec<(&str, K, V)> = vec![];
         for block_id in block_ids {
             let block_opt = match self.get_block(block_id, StorageRequestPriority::P0).await {
                 Ok(Some(block)) => Some(block),
@@ -894,7 +894,7 @@ mod tests {
                 Ok(c) => {
                     let mut kv_map = HashMap::new();
                     for entry in c {
-                        kv_map.insert(format!("{}/{}", prefix_query, entry.0), entry.1);
+                        kv_map.insert(format!("{}/{}", prefix_query, entry.1), entry.2);
                     }
                     for j in 1..=5 {
                         let prefix = format!("{}/{}", "prefix", j);
@@ -1016,7 +1016,7 @@ mod tests {
 
             let mut kv_map = HashMap::new();
             for entry in materialized_range {
-                kv_map.insert(entry.0, entry.1);
+                kv_map.insert(entry.1, entry.2);
             }
             for i in 1..num_keys {
                 let key = format!("{}/{}", "key", i);

rust/blockstore/src/memory/reader_writer.rs

@@ -101,7 +101,7 @@ impl<
         &'storage self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> Result<impl Iterator<Item = (K, V)> + 'storage, Box<dyn ChromaError>>
+    ) -> Result<impl Iterator<Item = (&'storage str, K, V)> + 'storage, Box<dyn ChromaError>>
     where
         PrefixRange: RangeBounds<&'prefix str>,
         KeyRange: RangeBounds<K>,
@@ -120,7 +120,7 @@ impl<
 
         Ok(values
             .into_iter()
-            .map(|(key, value)| (K::try_from(&key.key).unwrap(), value)))
+            .map(|(key, value)| (key.prefix.as_str(), K::try_from(&key.key).unwrap(), value)))
     }
 
     pub(crate) fn count(&self) -> Result<usize, Box<dyn ChromaError>> {
@@ -308,10 +308,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == "key1" && *value == "value1"));
+            .any(|(_, key, value)| *key == "key1" && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == "key2" && *value == "value2"));
+            .any(|(_, key, value)| *key == "key2" && *value == "value2"));
     }
 
     #[test]
@@ -348,13 +348,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -375,10 +375,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -420,13 +420,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -450,10 +450,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -487,13 +487,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -512,10 +512,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -549,13 +549,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -574,10 +574,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -611,13 +611,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -636,10 +636,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
     }
 
     #[test]
@@ -673,13 +673,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -698,10 +698,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
     }
 
     #[test]
@@ -735,13 +735,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3 && *value == "value3"));
     }
 
     #[test]
@@ -760,10 +760,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2 && *value == "value2"));
     }
 
     #[test]
@@ -797,13 +797,13 @@ mod tests {
         assert_eq!(values.len(), 3);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 3.0 && *value == "value3"));
+            .any(|(_, key, value)| *key == 3.0 && *value == "value3"));
     }
 
     #[test]
@@ -822,10 +822,10 @@ mod tests {
         assert_eq!(values.len(), 2);
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 1.0 && *value == "value1"));
+            .any(|(_, key, value)| *key == 1.0 && *value == "value1"));
         assert!(values
             .iter()
-            .any(|(key, value)| *key == 2.0 && *value == "value2"));
+            .any(|(_, key, value)| *key == 2.0 && *value == "value2"));
     }
 
     #[test]

rust/blockstore/src/types/reader.rs

@@ -67,7 +67,9 @@ impl<
         &'referred_data self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> impl Stream<Item = Result<(K, V), Box<dyn ChromaError>>> + 'referred_data + Send
+    ) -> impl Stream<Item = Result<(&'referred_data str, K, V), Box<dyn ChromaError>>>
+           + 'referred_data
+           + Send
     where
         PrefixRange: RangeBounds<&'prefix str> + Clone + Send + 'referred_data,
         KeyRange: RangeBounds<K> + Clone + Send + 'referred_data,
@@ -92,7 +94,7 @@ impl<
         &'referred_data self,
         prefix_range: PrefixRange,
         key_range: KeyRange,
-    ) -> Result<Vec<(K, V)>, Box<dyn ChromaError>>
+    ) -> Result<Vec<(&'referred_data str, K, V)>, Box<dyn ChromaError>>
     where
         PrefixRange: RangeBounds<&'prefix str> + Clone,
         KeyRange: RangeBounds<K> + Clone,

rust/blockstore/tests/blockfile_writer_test.rs

@@ -284,8 +284,8 @@ mod tests {
                 for (blockfile_entry, expected_entry) in
                     all_entries.iter().zip(ref_last_commit.iter())
                 {
-                    assert_eq!(blockfile_entry.0, expected_entry.0 .1); // key matches
-                    assert_eq!(blockfile_entry.1, expected_entry.1); // value matches
+                    assert_eq!(blockfile_entry.1, expected_entry.0 .1); // key matches
+                    assert_eq!(blockfile_entry.2, expected_entry.1); // value matches
                 }
             }