[ISSUE #6916] support startsWith and endsWith in sql filter. (#6915)

* support startsWith and endsWith in sql filter.

* add tests

* add tests

Author: yuz10

filter/src/main/java/org/apache/rocketmq/filter/expression/ComparisonExpression.java

@@ -153,6 +153,174 @@ public static BooleanExpression createNotContains(Expression left, String search
         return new NotContainsExpression(left, search);
     }
 
+    static class StartsWithExpression extends UnaryExpression implements BooleanExpression {
+
+        String search;
+
+        public StartsWithExpression(Expression right, String search) {
+            super(right);
+            this.search = search;
+        }
+
+        public String getExpressionSymbol() {
+            return "STARTSWITH";
+        }
+
+        public Object evaluate(EvaluationContext message) throws Exception {
+
+            if (search == null || search.length() == 0) {
+                return Boolean.FALSE;
+            }
+
+            Object rv = this.getRight().evaluate(message);
+
+            if (rv == null) {
+                return Boolean.FALSE;
+            }
+
+            if (!(rv instanceof String)) {
+                return Boolean.FALSE;
+            }
+
+            return ((String)rv).startsWith(search) ? Boolean.TRUE : Boolean.FALSE;
+        }
+
+        public boolean matches(EvaluationContext message) throws Exception {
+            Object object = evaluate(message);
+            return object != null && object == Boolean.TRUE;
+        }
+    }
+
+    static class NotStartsWithExpression extends UnaryExpression implements BooleanExpression {
+
+        String search;
+
+        public NotStartsWithExpression(Expression right, String search) {
+            super(right);
+            this.search = search;
+        }
+
+        public String getExpressionSymbol() {
+            return "NOT STARTSWITH";
+        }
+
+        public Object evaluate(EvaluationContext message) throws Exception {
+
+            if (search == null || search.length() == 0) {
+                return Boolean.FALSE;
+            }
+
+            Object rv = this.getRight().evaluate(message);
+
+            if (rv == null) {
+                return Boolean.FALSE;
+            }
+
+            if (!(rv instanceof String)) {
+                return Boolean.FALSE;
+            }
+
+            return ((String)rv).startsWith(search) ? Boolean.FALSE : Boolean.TRUE;
+        }
+
+        public boolean matches(EvaluationContext message) throws Exception {
+            Object object = evaluate(message);
+            return object != null && object == Boolean.TRUE;
+        }
+    }
+
+    public static BooleanExpression createStartsWith(Expression left, String search) {
+        return new StartsWithExpression(left, search);
+    }
+
+    public static BooleanExpression createNotStartsWith(Expression left, String search) {
+        return new NotStartsWithExpression(left, search);
+    }
+
+    static class EndsWithExpression extends UnaryExpression implements BooleanExpression {
+
+        String search;
+
+        public EndsWithExpression(Expression right, String search) {
+            super(right);
+            this.search = search;
+        }
+
+        public String getExpressionSymbol() {
+            return "ENDSWITH";
+        }
+
+        public Object evaluate(EvaluationContext message) throws Exception {
+
+            if (search == null || search.length() == 0) {
+                return Boolean.FALSE;
+            }
+
+            Object rv = this.getRight().evaluate(message);
+
+            if (rv == null) {
+                return Boolean.FALSE;
+            }
+
+            if (!(rv instanceof String)) {
+                return Boolean.FALSE;
+            }
+
+            return ((String)rv).endsWith(search) ? Boolean.TRUE : Boolean.FALSE;
+        }
+
+        public boolean matches(EvaluationContext message) throws Exception {
+            Object object = evaluate(message);
+            return object != null && object == Boolean.TRUE;
+        }
+    }
+
+    static class NotEndsWithExpression extends UnaryExpression implements BooleanExpression {
+
+        String search;
+
+        public NotEndsWithExpression(Expression right, String search) {
+            super(right);
+            this.search = search;
+        }
+
+        public String getExpressionSymbol() {
+            return "NOT ENDSWITH";
+        }
+
+        public Object evaluate(EvaluationContext message) throws Exception {
+
+            if (search == null || search.length() == 0) {
+                return Boolean.FALSE;
+            }
+
+            Object rv = this.getRight().evaluate(message);
+
+            if (rv == null) {
+                return Boolean.FALSE;
+            }
+
+            if (!(rv instanceof String)) {
+                return Boolean.FALSE;
+            }
+
+            return ((String)rv).endsWith(search) ? Boolean.FALSE : Boolean.TRUE;
+        }
+
+        public boolean matches(EvaluationContext message) throws Exception {
+            Object object = evaluate(message);
+            return object != null && object == Boolean.TRUE;
+        }
+    }
+
+    public static BooleanExpression createEndsWith(Expression left, String search) {
+        return new EndsWithExpression(left, search);
+    }
+
+    public static BooleanExpression createNotEndsWith(Expression left, String search) {
+        return new NotEndsWithExpression(left, search);
+    }
+
     @SuppressWarnings({"rawtypes", "unchecked"})
     public static BooleanExpression createInFilter(Expression left, List elements) {