fix(cubestore): fix crash in partition filter (#2748)

Caused by an error in handling of OR expressions.
Also fix occasional incomplete partition filters on OR expressions.

rust/cubestore/src/queryplanner/partition_filter.rs

@@ -236,19 +236,22 @@ impl Builder<'_> {
         rs: Iter,
     ) -> Vec<MinMaxCondition> {
         let mut res = None;
-        for mut r in rs {
+        for r in rs {
             if r.is_empty() {
                 return Vec::new();
             }
-            if res.is_none() {
-                res = Some(r);
-                continue;
-            }
-            if PartitionFilter::SIZE_LIMIT < res.as_mut().unwrap().len() + r.len() {
-                assert!(r.len() <= PartitionFilter::SIZE_LIMIT);
-                r.truncate(PartitionFilter::SIZE_LIMIT - r.len());
+            let res = match &mut res {
+                Some(res) => res,
+                res @ None => {
+                    *res = Some(r);
+                    continue;
+                }
+            };
+            if PartitionFilter::SIZE_LIMIT < res.len() + r.len() {
+                // TODO: keep some information.
+                return Vec::new();
             }
-            res.as_mut().unwrap().extend(r);
+            res.extend(r);
         }
 
         res.unwrap_or_default()
@@ -961,7 +964,7 @@ mod tests {
 
         let f = extract("(a <= 1 or b <= 2) and (a <= 2 or b <= 3) and (a <= 4 or b <= 5) and (a <= 6 or b <= 7) and (a <= 8 or b <= 9) and (a <= 10 or b <= 11)");
         // Must bail out to avoid too much compute.
-        assert_eq!(f.min_max.len(), 50)
+        assert_eq!(f.min_max.len(), 0)
     }
 
     #[test]
@@ -988,6 +991,36 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_limits_no_panic() {
+        let s = schema(&[
+            ("a", DataType::Int64),
+            ("b", DataType::Int64),
+            ("c", DataType::Int64),
+            ("d", DataType::Int64),
+            ("e", DataType::Int64),
+        ]);
+        let extract = |sql| PartitionFilter::extract(&s, &[parse(sql, &s)]);
+
+        let filter = extract(
+            "a IN (1,2,3,4,5,6,7,8,9) \
+                 AND b = 1 \
+                 AND c = 1 \
+                 AND d IN (1,2,3,4,5,6,7) \
+                 AND e IN (1, 2)",
+        );
+        assert_ne!(filter.min_max.len(), 0);
+        assert!(filter.can_match(
+            Some(&vec![TableValue::Int(1); 5]),
+            Some(&vec![TableValue::Int(1); 5])
+        ));
+        assert!(filter.can_match(Some(&vals(&[9, 1, 1, 7, 2])), Some(&vals(&[9, 1, 1, 7, 2]))));
+
+        fn vals(is: &[i64]) -> Vec<TableValue> {
+            is.iter().map(|i| TableValue::Int(*i)).collect()
+        }
+    }
+
     fn schema(s: &[(&str, DataType)]) -> Schema {
         Schema::new(
             s.iter()