Support RPN function validator

Signed-off-by: Breezewish <breezewish@pingcap.com>

src/coprocessor/dag/batch/interface.rs

@@ -135,7 +135,5 @@ pub struct BatchExecuteResult {
     ///                `Ok(true)`, there could be some remaining data in the `data` field which is
     ///                valid data and should be processed. The caller should NOT call `next_batch()`
     ///                any more.
-    // TODO: The name of this field is confusing and not obvious, that we need so many comments to
-    // explain what it is. We can change it to a better name or use an enum if necessary.
     pub is_drained: Result<bool, Error>,
 }

src/coprocessor/dag/rpn_expr/impl_compare.rs

@@ -207,13 +207,10 @@ pub fn coalesce<T: Evaluable>(args: &[&Option<T>]) -> Result<Option<T>> {
     Ok(None)
 }
 
-#[rpn_fn(varg)]
+#[rpn_fn(varg, min_args = 1)]
 #[inline]
 pub fn compare_in<T: Evaluable + Eq>(args: &[&Option<T>]) -> Result<Option<Int>> {
-    if args.is_empty() {
-        // TODO: Change to assert after validator function is implemented.
-        return Err(box_err!("Invalid IN() arguments"));
-    }
+    assert!(!args.is_empty());
     let base_val = args[0];
     match base_val {
         None => Ok(None),

src/coprocessor/dag/rpn_expr/impl_control.rs

@@ -14,26 +14,36 @@ fn if_null<T: Evaluable>(lhs: &Option<T>, rhs: &Option<T>) -> Result<Option<T>>
     Ok(rhs.clone())
 }
 
-#[rpn_fn(raw_varg)]
+#[rpn_fn(raw_varg, extra_validator = case_when_validator::<T>)]
 #[inline]
 pub fn case_when<T: Evaluable>(args: &[ScalarValueRef<'_>]) -> Result<Option<T>> {
     for chunk in args.chunks(2) {
         if chunk.len() == 1 {
-            // else statement
-            // TODO: Must verify type
+            // Else statement
             let ret: &Option<T> = Evaluable::borrow_scalar_value_ref(&chunk[0]);
             return Ok(ret.clone());
         }
         let cond: &Option<Int> = Evaluable::borrow_scalar_value_ref(&chunk[0]);
         if cond.unwrap_or(0) != 0 {
-            // TODO: Must verify type
             let ret: &Option<T> = Evaluable::borrow_scalar_value_ref(&chunk[1]);
             return Ok(ret.clone());
         }
     }
     Ok(None)
 }
 
+fn case_when_validator<T: Evaluable>(expr: &tipb::expression::Expr) -> Result<()> {
+    for chunk in expr.get_children().chunks(2) {
+        if chunk.len() == 1 {
+            super::function::validate_expr_return_type(&chunk[0], T::EVAL_TYPE)?;
+        } else {
+            super::function::validate_expr_return_type(&chunk[0], Int::EVAL_TYPE)?;
+            super::function::validate_expr_return_type(&chunk[1], T::EVAL_TYPE)?;
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/coprocessor/dag/rpn_expr/types/expr.rs

@@ -27,39 +27,30 @@ pub enum RpnExpressionNode {
 
 impl RpnExpressionNode {
     /// Gets the field type.
-    #[inline]
-    pub fn field_type(&self) -> Option<&FieldType> {
+    #[cfg(test)]
+    pub fn field_type(&self) -> &FieldType {
         match self {
-            RpnExpressionNode::FnCall { ref field_type, .. } => Some(field_type),
-            RpnExpressionNode::Constant { ref field_type, .. } => Some(field_type),
-            RpnExpressionNode::ColumnRef { .. } => None,
+            RpnExpressionNode::FnCall { field_type, .. } => field_type,
+            RpnExpressionNode::Constant { field_type, .. } => field_type,
+            RpnExpressionNode::ColumnRef { .. } => panic!(),
         }
     }
 
     /// Borrows the function instance for `FnCall` variant.
-    #[inline]
-    pub fn fn_call_func(&self) -> Option<RpnFnMeta> {
+    #[cfg(test)]
+    pub fn fn_call_func(&self) -> RpnFnMeta {
         match self {
-            RpnExpressionNode::FnCall { func_meta, .. } => Some(*func_meta),
-            _ => None,
+            RpnExpressionNode::FnCall { func_meta, .. } => *func_meta,
+            _ => panic!(),
         }
     }
 
     /// Borrows the constant value for `Constant` variant.
-    #[inline]
-    pub fn constant_value(&self) -> Option<&ScalarValue> {
-        match self {
-            RpnExpressionNode::Constant { ref value, .. } => Some(value),
-            _ => None,
-        }
-    }
-
-    /// Gets the column offset for `ColumnRef` variant.
-    #[inline]
-    pub fn column_ref_offset(&self) -> Option<usize> {
+    #[cfg(test)]
+    pub fn constant_value(&self) -> &ScalarValue {
         match self {
-            RpnExpressionNode::ColumnRef { ref offset, .. } => Some(*offset),
-            _ => None,
+            RpnExpressionNode::Constant { value, .. } => value,
+            _ => panic!(),
         }
     }
 }
@@ -113,6 +104,11 @@ impl RpnExpression {
             RpnExpressionNode::ColumnRef { offset } => &schema[*offset],
         }
     }
+
+    /// Unwraps into the underlying expression node vector.
+    pub fn into_inner(self) -> Vec<RpnExpressionNode> {
+        self.0
+    }
 }
 
 // For `RpnExpression::eval`, see `expr_eval` file.