From a69e46efc6dfbf2d002d65f1a74b9555b41c6f5d Mon Sep 17 00:00:00 2001
From: Edd Robinson <me@edd.io>
Date: Fri, 24 Sep 2021 14:34:36 +0100
Subject: [PATCH] feat: validate predicates on column_values

---
 read_buffer/src/chunk.rs | 20 ++++++++++++++++++++
 read_buffer/src/table.rs |  1 +
 2 files changed, 21 insertions(+)

diff --git a/read_buffer/src/chunk.rs b/read_buffer/src/chunk.rs
index 331eb1c690..e80e8f585e 100644
--- a/read_buffer/src/chunk.rs
+++ b/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 { .. })
+        ));
     }
 }
diff --git a/read_buffer/src/table.rs b/read_buffer/src/table.rs
index 4d0add5b26..da6d8233a4 100644
--- a/read_buffer/src/table.rs
+++ b/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)) {