feat(expr): support substring function

src/array/ops.rs

@@ -234,6 +234,34 @@ impl ArrayImpl {
         })
     }
 
+    pub fn substring(&self, start: &Self, length: &Self) -> Result<Self, ConvertError> {
+        let (A::Utf8(a), A::Int32(b), A::Int32(c)) = (self, start, length) else {
+            return Err(ConvertError::NoTernaryOp("substring".into(), self.type_string(), start.type_string(), length.type_string()));
+        };
+        Ok(A::new_utf8(ternary_op(
+            a.as_ref(),
+            b.as_ref(),
+            c.as_ref(),
+            |a, b, c| {
+                let chars = a.chars().count() as i32;
+                let mut start = match *b {
+                    0.. => *b - 1,
+                    _ => chars + *b,
+                };
+                let mut end = start.saturating_add(*c);
+                if start > end {
+                    (start, end) = (end, start);
+                }
+                let skip = start.max(0);
+                let take = (end - skip).max(0);
+                a.chars()
+                    .skip(skip as usize)
+                    .take(take as usize)
+                    .collect::<String>()
+            },
+        )))
+    }
+
     /// Perform binary operation.
     pub fn binary_op(
         &self,
@@ -608,6 +636,26 @@ where
     Ok(builder.finish())
 }
 
+fn ternary_op<A, B, C, O, F, V>(a: &A, b: &B, c: &C, f: F) -> O
+where
+    A: Array,
+    B: Array,
+    C: Array,
+    O: Array,
+    V: Borrow<O::Item>,
+    F: Fn(&A::Item, &B::Item, &C::Item) -> V,
+{
+    let mut builder = O::Builder::with_capacity(a.len());
+    for e in a.iter().zip(b.iter()).zip(c.iter()) {
+        if let ((Some(a), Some(b)), Some(c)) = e {
+            builder.push(Some(f(a, b, c).borrow()));
+        } else {
+            builder.push(None);
+        }
+    }
+    builder.finish()
+}
+
 /// Optimized operations.
 ///
 /// Assume both bitvecs have the same length.

src/binder/expr.rs

@@ -41,6 +41,11 @@ impl Binder {
             } => self.bind_between(*expr, negated, *low, *high),
             Expr::Interval(interval) => self.bind_interval(interval),
             Expr::Extract { field, expr } => self.bind_extract(field, *expr),
+            Expr::Substring {
+                expr,
+                substring_from,
+                substring_for,
+            } => self.bind_substring(*expr, substring_from, substring_for),
             _ => todo!("bind expression: {:?}", expr),
         }?;
         self.check_type(id)?;
@@ -194,6 +199,24 @@ impl Binder {
         Ok(self.egraph.add(Node::Extract([field, expr])))
     }
 
+    fn bind_substring(
+        &mut self,
+        expr: Expr,
+        from: Option<Box<Expr>>,
+        for_: Option<Box<Expr>>,
+    ) -> Result {
+        let expr = self.bind_expr(expr)?;
+        let from = match from {
+            Some(expr) => self.bind_expr(*expr)?,
+            None => self.egraph.add(Node::Constant(DataValue::Int32(1))),
+        };
+        let for_ = match for_ {
+            Some(expr) => self.bind_expr(*expr)?,
+            None => self.egraph.add(Node::Constant(DataValue::Int32(i32::MAX))),
+        };
+        Ok(self.egraph.add(Node::Substring([expr, from, for_])))
+    }
+
     fn bind_function(&mut self, func: Function) -> Result {
         let mut args = vec![];
         for arg in func.args {

src/executor/evaluator.rs

@@ -85,6 +85,12 @@ impl<'a> Evaluator<'a> {
                 let Expr::Field(field) = self.expr[*field] else { panic!("not a field") };
                 a.extract(field)
             }
+            Substring([str, start, length]) => {
+                let str = self.next(*str).eval(chunk)?;
+                let start = self.next(*start).eval(chunk)?;
+                let length = self.next(*length).eval(chunk)?;
+                str.substring(&start, &length)
+            }
             Desc(a) | Ref(a) => self.next(*a).eval(chunk),
             // for aggs, evaluate its children
             RowCount => Ok(ArrayImpl::new_null(

src/planner/explain.rs

@@ -170,6 +170,14 @@ impl<'a> Explain<'a> {
                     ("to", self.expr(c).pretty()),
                 ],
             ),
+            Substring([str, start, len]) => Pretty::childless_record(
+                "Substring",
+                vec![
+                    ("str", self.expr(str).pretty()),
+                    ("start", self.expr(start).pretty()),
+                    ("length", self.expr(len).pretty()),
+                ],
+            ),
 
             // aggregations
             RowCount | RowNumber => enode.to_string().into(),

src/planner/mod.rs

@@ -66,6 +66,7 @@ define_language! {
         "extract" = Extract([Id; 2]),           // (extract field expr)
             Field(DateTimeField),
         "replace" = Replace([Id; 3]),           // (replace expr pattern replacement)
+        "substring" = Substring([Id; 3]),       // (substring expr start length)
 
         // aggregations
         "max" = Max(Id),

src/planner/rules/type_.rs

@@ -99,6 +99,12 @@ pub fn analyze_type(enode: &Expr, x: impl Fn(&Id) -> Type, catalog: &RootCatalog
         Extract([_, a]) => merge(enode, [x(a)?], |[a]| {
             matches!(a, Kind::Date | Kind::Interval).then_some(Kind::Int32)
         }),
+        Substring([str, start, len]) => {
+            merge(enode, [x(str)?, x(start)?, x(len)?], |[str, start, len]| {
+                (str == Kind::String && start == Kind::Int32 && len == Kind::Int32)
+                    .then_some(Kind::String)
+            })
+        }
 
         // number agg
         Max(a) | Min(a) => x(a),

src/types/mod.rs

@@ -107,7 +107,7 @@ impl From<&crate::parser::DataType> for DataTypeKind {
             // Real => Self::Float32,
             Float(_) | Double => Self::Float64,
             SmallInt(_) => Self::Int16,
-            Int(_) => Self::Int32,
+            Int(_) | Integer(_) => Self::Int32,
             BigInt(_) => Self::Int64,
             Boolean => Self::Bool,
             Decimal(info) => match info {
@@ -293,6 +293,8 @@ pub enum ConvertError {
     NoUnaryOp(String, &'static str),
     #[error("no function {0}({1}, {2})")]
     NoBinaryOp(String, &'static str, &'static str),
+    #[error("no function {0}({1}, {2}, {3})")]
+    NoTernaryOp(String, &'static str, &'static str, &'static str),
     #[error("no cast {0} -> {1}")]
     NoCast(&'static str, DataTypeKind),
 }