feat: validate predicates on column_values

read_buffer/src/chunk.rs

@@ -1255,6 +1255,16 @@ mod test {
         // sketchy_sensor won't be returned because it has a NULL value for the
         // only matching row.
         assert_eq!(result, to_set(&["counter", "region", "time"]));
+
+        // Error when invalid predicate provided.
+        assert!(matches!(
+            chunk.column_names(
+                Predicate::new(vec![BinaryExpr::from(("time", "=", "not a number"))]),
+                Selection::Some(&["region", "env"]),
+                BTreeSet::new()
+            ),
+            Err(Error::TableError { .. })
+        ));
     }
 
     fn to_map(arr: Vec<(&str, &[&str])>) -> BTreeMap<String, BTreeSet<String>> {
@@ -1346,5 +1356,15 @@ mod test {
             chunk.column_values(Predicate::default(), Selection::All, BTreeMap::new()),
             Err(Error::UnsupportedOperation { .. })
         ));
+
+        // Error when invalid predicate provided.
+        assert!(matches!(
+            chunk.column_values(
+                Predicate::new(vec![BinaryExpr::from(("time", "=", "not a number"))]),
+                Selection::Some(&["region", "env"]),
+                BTreeMap::new()
+            ),
+            Err(Error::TableError { .. })
+        ));
     }
 }

read_buffer/src/table.rs

@@ -495,6 +495,7 @@ impl Table {
         mut dst: BTreeMap<String, BTreeSet<String>>,
     ) -> Result<BTreeMap<String, BTreeSet<String>>> {
         let (meta, row_groups) = self.filter_row_groups(predicate);
+        meta.validate_exprs(predicate.iter())?;
 
         // Validate that only supported columns present in `columns`.
         for (name, (ct, _)) in columns.iter().zip(meta.schema_for_column_names(columns)) {