[fix][core][api]统一Aviator表达式语法和调用方式,所有processor和函数使用表达式不需要使用event.字段名。

13 files changed, 60 insertions, 190 deletions

diff --git a/groot-core/src/main/java/com/geedgenetworks/core/expressions/Calc.java b/groot-core/src/main/java/com/geedgenetworks/core/expressions/Calc.java
deleted file mode 100644
index 2b88853..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/expressions/Calc.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package com.geedgenetworks.core.expressions;
-
-import java.io.Serializable;
-import java.util.Map;
-
-/**
- * The Calc interface represents a calculation operation that can be evaluated with input data
- * provided as a map.
- *
- * @author chaoc
- * @since 1.0
- */
-public interface Calc extends Serializable {
-
-    /**
-     * Evaluates the calculation operation with the given input data.
-     *
-     * @param data A map containing input data for the calculation.
-     * @return The result of the calculation operation.
-     */
-    Object eval(final Map<String, Object> data);
-}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/expressions/EvalExecutor.java b/groot-core/src/main/java/com/geedgenetworks/core/expressions/EvalExecutor.java
deleted file mode 100644
index d3936f5..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/expressions/EvalExecutor.java
+++ /dev/null
@@ -1,35 +0,0 @@
-package com.geedgenetworks.core.expressions;
-
-import com.googlecode.aviator.AviatorEvaluator;
-import com.googlecode.aviator.Expression;
-import com.googlecode.aviator.exception.ExpressionRuntimeException;
-import com.googlecode.aviator.exception.ExpressionSyntaxErrorException;
-
-import java.util.Map;
-
-/**
- * Eval Function Executor.
- *
- * @author chaoc
- * @since 1.5
- */
-public final class EvalExecutor implements Calc {
-
-    private final Expression script;
-
-    public EvalExecutor(String script) {
-        try {
-            AviatorEvaluator.validate(script);
-            this.script =
-                    AviatorEvaluator.getInstance().compile(script);
-        } catch (ExpressionSyntaxErrorException
-                 | ExpressionRuntimeException e) {
-            throw new SyntaxErrorException("Eval script syntax error. ", e);
-        }
-    }
-
-    @Override
-    public Object eval(final Map<String, Object> data) {
-        return script.execute(data);
-    }
-}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/expressions/SyntaxErrorException.java b/groot-core/src/main/java/com/geedgenetworks/core/expressions/SyntaxErrorException.java
deleted file mode 100644
index f058615..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/expressions/SyntaxErrorException.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package com.geedgenetworks.core.expressions;
-
-/**
- * Exception class representing a syntax error that may occur during program execution.
- *
- * @author chaoc
- * @since 1.0
- */
-public class SyntaxErrorException extends RuntimeException {
-
-    public SyntaxErrorException() {}
-
-    public SyntaxErrorException(String message) {
-        super(message);
-    }
-
-    public SyntaxErrorException(String message, Throwable cause) {
-        super(message, cause);
-    }
-}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AbstractFirstAggregation.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AbstractFirstAggregation.java
index 0d9c3ca..00d7d9c 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AbstractFirstAggregation.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AbstractFirstAggregation.java
@@ -12,6 +12,7 @@ import com.geedgenetworks.api.common.udf.UDFContext;
 import com.geedgenetworks.api.common.udf.UdfEntity;
 import com.geedgenetworks.api.metrics.InternalMetrics;
 import com.geedgenetworks.api.event.Event;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.google.common.collect.Lists;
 import com.googlecode.aviator.AviatorEvaluator;
 import com.googlecode.aviator.AviatorEvaluatorInstance;
@@ -72,7 +73,6 @@ public class AbstractFirstAggregation extends ProcessFunction<Event, Accumulator
             Map<String, String> udfClassReflect = getClassReflect(udfClassNameLists);
 
             for (UDFContext udfContext : udfContexts) {
-                Expression filterExpression = null;
                 UdfEntity udfEntity = new UdfEntity();
                 // 平台注册的函数包含任务中配置的函数则对函数进行实例化
                 if (udfClassReflect.containsKey(udfContext.getFunction())) {
@@ -80,14 +80,9 @@ public class AbstractFirstAggregation extends ProcessFunction<Event, Accumulator
                     AggregateFunction aggregateFunction = (AggregateFunction) cls.getConstructor().newInstance();
                     // 函数如果包含filter,对表达式进行编译
                     if (udfContext.getFilter() != null) {
-                        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-                        instance.setCachedExpressionByDefault(true);
-                        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-                        instance.setFunctionMissing(null);
-                        filterExpression = instance.compile(udfContext.getFilter(), true);
+                        udfEntity.setAviatorExecutor(new AviatorExecutor(udfContext.getFilter()));
                     }
                     udfEntity.setAggregateFunction(aggregateFunction);
-                    udfEntity.setFilterExpression(filterExpression);
                     udfEntity.setName(udfContext.getFunction());
                     udfEntity.setClassName(udfClassReflect.get(udfContext.getFunction()));
                     udfEntity.setUdfContext(udfContext);
@@ -171,7 +166,7 @@ public class AbstractFirstAggregation extends ProcessFunction<Event, Accumulator
         accumulator.setInEvents(accumulator.getInEvents() + 1);
         for (UdfEntity udafEntity : functions) {
             try {
-                boolean result = udafEntity.getFilterExpression() != null ? filterExecute(udafEntity.getFilterExpression(), udafEntity.getFilterExpression().newEnv("event", event.getExtractedFields())) : true;
+                boolean result = udafEntity.getAviatorExecutor() != null ? udafEntity.getAviatorExecutor().filter(event.getExtractedFields()) : true;
                 if (result) {
                     udafEntity.getAggregateFunction().add(event, accumulator);
                 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AggregateProcessorFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AggregateProcessorFunction.java
index 2fb0ff1..ea05c1c 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AggregateProcessorFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/AggregateProcessorFunction.java
@@ -1,6 +1,7 @@
 package com.geedgenetworks.core.processor.aggregate;
 
 import com.alibaba.fastjson.JSON;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.common.config.Accumulator;
 import com.geedgenetworks.common.config.Constants;
 import com.geedgenetworks.common.exception.CommonErrorCode;
@@ -42,7 +43,6 @@ public class AggregateProcessorFunction implements org.apache.flink.api.common.f
         Map<String, String> udfClassReflect = getClassReflect(udfClassNameLists);
         try {
             for (UDFContext udfContext : udfContexts) {
-                Expression filterExpression = null;
                 UdfEntity udfEntity = new UdfEntity();
                 // 平台注册的函数包含任务中配置的函数则对函数进行实例化
                 if (udfClassReflect.containsKey(udfContext.getFunction())) {
@@ -50,14 +50,10 @@ public class AggregateProcessorFunction implements org.apache.flink.api.common.f
                     AggregateFunction aggregateFunction = (AggregateFunction) cls.getConstructor().newInstance();
                     // 函数如果包含filter,对表达式进行编译
                     if (udfContext.getFilter() != null) {
-                        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-                        instance.setCachedExpressionByDefault(true);
-                        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-                        instance.setFunctionMissing(null);
-                        filterExpression = instance.compile(udfContext.getFilter(), true);
+                        udfEntity.setAviatorExecutor(new AviatorExecutor(udfContext.getFilter()));
+
                     }
                     udfEntity.setAggregateFunction(aggregateFunction);
-                    udfEntity.setFilterExpression(filterExpression);
                     udfEntity.setName(udfContext.getFunction());
                     udfEntity.setClassName(udfClassReflect.get(udfContext.getFunction()));
                     udfEntity.setUdfContext(udfContext);
@@ -92,7 +88,7 @@ public class AggregateProcessorFunction implements org.apache.flink.api.common.f
         accumulator.setInEvents(accumulator.getInEvents()+1);
         for (UdfEntity udafEntity : functions) {
             try {
-                boolean result = udafEntity.getFilterExpression() != null ? filterExecute(udafEntity.getFilterExpression(), udafEntity.getFilterExpression().newEnv("event", event.getExtractedFields())) : true;
+                boolean result = udafEntity.getAviatorExecutor() != null ? udafEntity.getAviatorExecutor().filter(event.getExtractedFields()) : true;
                 if (result) {
                     udafEntity.getAggregateFunction().add(event, accumulator);
                 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/SecondAggregateProcessorFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/SecondAggregateProcessorFunction.java
index 86cf3f6..db5336a 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/SecondAggregateProcessorFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/aggregate/SecondAggregateProcessorFunction.java
@@ -1,6 +1,7 @@
 package com.geedgenetworks.core.processor.aggregate;
 
 import com.alibaba.fastjson.JSON;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.common.config.Accumulator;
 import com.geedgenetworks.common.config.Constants;
 import com.geedgenetworks.common.exception.CommonErrorCode;
@@ -47,15 +48,7 @@ public class SecondAggregateProcessorFunction implements org.apache.flink.api.co
                     Class<?> cls = Class.forName(udfClassReflect.get(udfContext.getFunction()));
                     AggregateFunction aggregateFunction = (AggregateFunction) cls.getConstructor().newInstance();
                     // 函数如果包含filter,对表达式进行编译
-                    if (udfContext.getFilter() != null) {
-                        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-                        instance.setCachedExpressionByDefault(true);
-                        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-                        instance.setFunctionMissing(null);
-                        filterExpression = instance.compile(udfContext.getFilter(), true);
-                    }
                     udfEntity.setAggregateFunction(aggregateFunction);
-                    udfEntity.setFilterExpression(filterExpression);
                     udfEntity.setName(udfContext.getFunction());
                     udfEntity.setClassName(udfClassReflect.get(udfContext.getFunction()));
                     udfEntity.setUdfContext(udfContext);
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/filter/FilterFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/filter/FilterFunction.java
index 8a08cb3..7107161 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/filter/FilterFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/filter/FilterFunction.java
@@ -1,13 +1,16 @@
 package com.geedgenetworks.core.processor.filter;
 
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.common.utils.ColumnUtil;
 import com.geedgenetworks.api.metrics.InternalMetrics;
 import com.geedgenetworks.api.event.Event;
+import com.geedgenetworks.api.expressions.SyntaxErrorException;
 import com.googlecode.aviator.AviatorEvaluator;
 import com.googlecode.aviator.AviatorEvaluatorInstance;
 import com.googlecode.aviator.Expression;
 import com.googlecode.aviator.Options;
 import com.googlecode.aviator.exception.ExpressionRuntimeException;
+import com.googlecode.aviator.exception.ExpressionSyntaxErrorException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.common.functions.RichFilterFunction;
 import org.apache.flink.configuration.Configuration;
@@ -16,8 +19,7 @@ import java.util.Map;
 @Slf4j
 public class FilterFunction extends RichFilterFunction<Event> {
     private final FilterConfig filterConfig;
-    private static Expression compiledExp;
-    private static String expression;
+    private AviatorExecutor aviatorExecutor;
     private transient InternalMetrics internalMetrics;
 
     public FilterFunction(FilterConfig filterConfig) {
@@ -27,12 +29,8 @@ public class FilterFunction extends RichFilterFunction<Event> {
     @Override
     public void open(Configuration parameters) throws Exception {
         this.internalMetrics = new InternalMetrics(getRuntimeContext());
-        expression = filterConfig.getProperties().getOrDefault("expression", "").toString();
-        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-        instance.setCachedExpressionByDefault(true);
-        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-        instance.setFunctionMissing(null);
-        compiledExp = instance.compile(expression, true);
+        String expression = filterConfig.getProperties().getOrDefault("expression", "").toString();
+        this.aviatorExecutor = new AviatorExecutor(expression);
     }
 
     @Override
@@ -47,11 +45,11 @@ public class FilterFunction extends RichFilterFunction<Event> {
                         ColumnUtil.columnSelector(
                                 value.getExtractedFields(), filterConfig.getOutput_fields()));
             }
-            isFilter = aviatorExcute(value.getExtractedFields());
+            isFilter = aviatorExecutor.filter(value.getExtractedFields());
 
         } catch (ExpressionRuntimeException e){
             isFilter = false;
-            log.error("Invalid filter expression: {}, {}", expression, e);
+            log.error("Invalid filter expression ! "+e);
             internalMetrics.incrementErrorEvents();
         }
         catch (RuntimeException ignored) {
@@ -66,19 +64,5 @@ public class FilterFunction extends RichFilterFunction<Event> {
         }
         return isFilter;
     }
-    public static boolean aviatorExcute(Map<String, Object> map) {
 
-        boolean bool = true;
-        if (!"".equals(expression)) {
-            Object object = compiledExp.execute(compiledExp.newEnv("event", map));
-            if(object != null){
-                bool = (boolean) object;
-            }
-            else {
-                throw new ExpressionRuntimeException();
-            }
-        }
-        return bool;
-
-    }
 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/projection/ProjectionProcessFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/projection/ProjectionProcessFunction.java
index 03a8f5a..e079790 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/projection/ProjectionProcessFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/projection/ProjectionProcessFunction.java
@@ -2,6 +2,7 @@ package com.geedgenetworks.core.processor.projection;
 
 import com.alibaba.fastjson.JSON;
 
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.common.config.Constants;
 import com.geedgenetworks.common.config.*;
 import com.geedgenetworks.common.exception.CommonErrorCode;
@@ -56,7 +57,6 @@ public class ProjectionProcessFunction extends ProcessFunction<Event, Event> {
             CommonConfig commonConfig = JSON.parseObject(configuration.toMap().get(Constants.SYSPROP_GROOTSTREAM_CONFIG), CommonConfig.class);
             KnowledgeBaseScheduler.startSchedulerForKnowledgeBase(Integer.parseInt(commonConfig.getPropertiesConfig().getOrDefault(Constants.SYSPROP_GROOTSTREAM_KB_SCHEDULER_INTERVAL_NAME, "5")));
             for (UDFContext udfContext : udfContexts) {
-                Expression filterExpression = null;
                 UdfEntity udfEntity = new UdfEntity();
                 // 平台注册的函数包含任务中配置的函数则对函数进行实例化
                 if (udfClassReflect.containsKey(udfContext.getFunction())) {
@@ -65,14 +65,9 @@ public class ProjectionProcessFunction extends ProcessFunction<Event, Event> {
                     scalarFunction.open(getRuntimeContext(), udfContext);
                     // 函数如果包含filter,对表达式进行编译
                     if (udfContext.getFilter() != null) {
-                        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-                        instance.setCachedExpressionByDefault(true);
-                        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-                        instance.setFunctionMissing(null);
-                        filterExpression = instance.compile(udfContext.getFilter(), true);
+                        new AviatorExecutor(udfContext.getFilter());
                     }
                     udfEntity.setScalarFunction(scalarFunction);
-                    udfEntity.setFilterExpression(filterExpression);
                     udfEntity.setName(udfContext.getFunction());
                     udfEntity.setClassName(udfClassReflect.get(udfContext.getFunction()));
                     functions.add(udfEntity);
@@ -99,7 +94,7 @@ public class ProjectionProcessFunction extends ProcessFunction<Event, Event> {
                 if (event.isDropped()) {
                     break;
                 } else {
-                    boolean result = udfEntity.getFilterExpression() != null ? filterExecute(udfEntity.getFilterExpression(), udfEntity.getFilterExpression().newEnv("event", event.getExtractedFields())) : true;
+                    boolean result = udfEntity.getAviatorExecutor() != null ? udfEntity.getAviatorExecutor().filter(event.getExtractedFields()) : true;
                     if (result) {
                         udfEntity.getScalarFunction().evaluate(event);
                     }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/split/RuleContext.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/split/RuleContext.java
index f1cd8e8..33d135a 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/split/RuleContext.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/split/RuleContext.java
@@ -1,6 +1,8 @@
 package com.geedgenetworks.core.processor.split;
 
 import com.geedgenetworks.api.event.Event;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
+import com.googlecode.aviator.AviatorEvaluator;
 import com.googlecode.aviator.Expression;
 import lombok.Data;
 import org.apache.flink.util.OutputTag;
@@ -12,7 +14,7 @@ public class RuleContext implements Serializable {
 
     private String tag;
     private String expression;
-    private Expression compiledExpression;
+    private AviatorExecutor aviatorExecutor;
     private OutputTag<Event> outputTag ;
 
 
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/split/SplitFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/split/SplitFunction.java
index 2cbf74d..d9232d7 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/split/SplitFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/split/SplitFunction.java
@@ -1,5 +1,6 @@
 package com.geedgenetworks.core.processor.split;
 
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.api.metrics.InternalMetrics;
 import com.geedgenetworks.api.event.Event;
 import com.googlecode.aviator.AviatorEvaluator;
@@ -34,13 +35,7 @@ public class SplitFunction extends ProcessFunction<Event, Event> {
         this.internalMetrics = new InternalMetrics(getRuntimeContext());
         this.rules = splitConfig.getRules();
         for(RuleContext rule : rules){
-            String expression = rule.getExpression();
-            AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-            instance.setCachedExpressionByDefault(true);
-            instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-            instance.setFunctionMissing(null);
-            Expression compiledExp = instance.compile(expression, true);
-            rule.setCompiledExpression(compiledExp);
+            rule.setAviatorExecutor(new AviatorExecutor(rule.getExpression()));
             OutputTag<Event> outputTag = new OutputTag<>(rule.getTag()){};
             rule.setOutputTag(outputTag);
         }
@@ -52,7 +47,7 @@ public class SplitFunction extends ProcessFunction<Event, Event> {
         try {
             internalMetrics.incrementInEvents();
             for (RuleContext route : rules){
-                boolean result = route.getExpression() != null ? (filterExecute(route.getCompiledExpression(), route.getCompiledExpression().newEnv("event", event.getExtractedFields()))) : true;
+                boolean result = route.getAviatorExecutor() != null ? route.getAviatorExecutor().filter(event.getExtractedFields()): true;
                 if (result) {
                     ctx.output(route.getOutputTag(), event);
                 }
@@ -63,15 +58,4 @@ public class SplitFunction extends ProcessFunction<Event, Event> {
         }
     }
 
-    public static Boolean filterExecute(Expression expression, Map<String, Object> map) {
-
-        boolean result;
-        Object object = expression.execute(map);
-        if (object != null) {
-            result = (Boolean) object;
-        } else {
-            throw new ExpressionRuntimeException();
-        }
-        return result;
-    }
 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/processor/table/TableProcessorFunction.java b/groot-core/src/main/java/com/geedgenetworks/core/processor/table/TableProcessorFunction.java
index f061028..a054fcd 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/processor/table/TableProcessorFunction.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/processor/table/TableProcessorFunction.java
@@ -1,6 +1,7 @@
 package com.geedgenetworks.core.processor.table;
 
 import com.alibaba.fastjson.JSON;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.common.config.Constants;
 import com.geedgenetworks.common.exception.CommonErrorCode;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
@@ -57,14 +58,9 @@ public  class TableProcessorFunction extends RichFlatMapFunction<Event, Event> {
                     tableFunction.open(getRuntimeContext(), udfContext);
                     // 函数如果包含filter,对表达式进行编译
                     if (udfContext.getFilter() != null) {
-                        AviatorEvaluatorInstance instance = AviatorEvaluator.getInstance();
-                        instance.setCachedExpressionByDefault(true);
-                        instance.setOption(Options.OPTIMIZE_LEVEL, AviatorEvaluator.EVAL);
-                        instance.setFunctionMissing(null);
-                        filterExpression = instance.compile(udfContext.getFilter(), true);
+                        new AviatorExecutor(udfContext.getFilter());
                     }
                     udfEntity.setTableFunction(tableFunction);
-                    udfEntity.setFilterExpression(filterExpression);
                     udfEntity.setName(udfContext.getFunction());
                     udfEntity.setClassName(udfClassReflect.get(udfContext.getFunction()));
                     functions.add(udfEntity);
@@ -90,7 +86,7 @@ public  class TableProcessorFunction extends RichFlatMapFunction<Event, Event> {
             List<Event> newEvents = new ArrayList<>();
             for(int i=0;i<events.size();i++) {
                 try {
-                        boolean result = udfEntity.getFilterExpression() != null ? filterExecute(udfEntity.getFilterExpression(), udfEntity.getFilterExpression().newEnv("event", events.get(i).getExtractedFields())) : true;
+                    boolean result = udfEntity.getAviatorExecutor() != null ? udfEntity.getAviatorExecutor().filter(event.getExtractedFields()) : true;
                         if (!events.get(i).isDropped() && result) {
                             newEvents.addAll(udfEntity.getTableFunction().evaluate(events.get(i)));
                         }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Eval.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Eval.java
index ca1f731..dba08c6 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Eval.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Eval.java
@@ -2,8 +2,8 @@ package com.geedgenetworks.core.udf;
 
 import com.geedgenetworks.common.exception.CommonErrorCode;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
-import com.geedgenetworks.core.expressions.Calc;
-import com.geedgenetworks.core.expressions.EvalExecutor;
+import com.geedgenetworks.api.expressions.Calc;
+import com.geedgenetworks.api.expressions.AviatorExecutor;
 import com.geedgenetworks.api.common.udf.ScalarFunction;
 import com.geedgenetworks.api.common.udf.UDFContext;
 import com.geedgenetworks.api.event.Event;
@@ -30,7 +30,7 @@ public class Eval implements ScalarFunction {
         String expr = (String) udfContext.getParameters().get("value_expression");
         List<String> outputField = udfContext.getOutputFields();
         output = outputField.get(0);
-        calc = new EvalExecutor(expr);
+        calc = new AviatorExecutor(expr);
     }
 
     @Override
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/expressions/EvalExecutorTest.java b/groot-core/src/test/java/com/geedgenetworks/core/expressions/AviatorExecutorTest.java
index e1e9924..64f0fb8 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/expressions/EvalExecutorTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/expressions/AviatorExecutorTest.java
@@ -1,5 +1,7 @@
 package com.geedgenetworks.core.expressions;
 
+import com.geedgenetworks.api.expressions.AviatorExecutor;
+import com.geedgenetworks.api.expressions.SyntaxErrorException;
 import com.googlecode.aviator.exception.ExpressionRuntimeException;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -11,7 +13,7 @@ import java.util.Map;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-class EvalExecutorTest {
+class AviatorExecutorTest {
 
     private static Map<String, Object> origin;
 
@@ -40,15 +42,15 @@ class EvalExecutorTest {
     @Test
     public void testValueExpression() throws Exception {
         // constant
-        assertEquals(new EvalExecutor("'name'").eval(origin), "name");
-        assertEquals(new EvalExecutor("5").eval(origin), 5L);
-        assertEquals(new EvalExecutor("true").eval(origin), true);
-        assertEquals(new EvalExecutor("false").eval(origin), false);
+        assertEquals(new AviatorExecutor("'name'").eval(origin), "name");
+        assertEquals(new AviatorExecutor("5").eval(origin), 5L);
+        assertEquals(new AviatorExecutor("true").eval(origin), true);
+        assertEquals(new AviatorExecutor("false").eval(origin), false);
 
         // variable
-        assertEquals(new EvalExecutor("common_client_ip").eval(origin),
+        assertEquals(new AviatorExecutor("common_client_ip").eval(origin),
                 origin.get("common_client_ip"));
-        assertEquals(new EvalExecutor("common_client_port").eval(origin),
+        assertEquals(new AviatorExecutor("common_client_port").eval(origin),
                 origin.get("common_client_port"));
     }
 
@@ -61,15 +63,15 @@ class EvalExecutorTest {
         int commonC2sByteNum = (int) origin.get("common_c2s_byte_num");
         int commonS2cByteNum = (int) origin.get("common_s2c_byte_num");
         assertEquals(
-                new EvalExecutor("common_client_port > common_server_port " +
+                new AviatorExecutor("common_client_port > common_server_port " +
                         "? common_c2s_pkt_num : common_s2c_pkt_num").eval(origin),
                 commonClientPort > commonServerPort ? commonC2sPktNum : commonS2cPktNum);
         assertEquals(
-                new EvalExecutor("common_client_port > common_server_port " +
+                new AviatorExecutor("common_client_port > common_server_port " +
                         "? 'C2S:' + common_c2s_pkt_num : 'S2C:' + common_s2c_pkt_num").eval(origin),
                 commonClientPort > commonServerPort ? "C2S:" + commonC2sPktNum : "S2C:" + commonS2cPktNum);
         assertEquals(
-                new EvalExecutor("common_c2s_pkt_num > common_s2c_pkt_num " +
+                new AviatorExecutor("common_c2s_pkt_num > common_s2c_pkt_num " +
                         "&& common_c2s_byte_num > common_s2c_byte_num ? common_client_port : common_server_port").eval(origin),
                 commonC2sPktNum > commonS2cPktNum
                         && commonC2sByteNum > commonS2cByteNum ? commonClientPort : commonServerPort
@@ -81,59 +83,59 @@ class EvalExecutorTest {
     public void testArithmeticExpression() throws Exception {
         final Map<String, Object> map = Collections.emptyMap();
         assertEquals(
-                1L, new EvalExecutor("5 & 3").eval(map));
+                1L, new AviatorExecutor("5 & 3").eval(map));
         assertEquals(
-                7L, new EvalExecutor("5 | 3").eval(map));
+                7L, new AviatorExecutor("5 | 3").eval(map));
         assertEquals(
-                20L, new EvalExecutor("5 << 2").eval(map));
+                20L, new AviatorExecutor("5 << 2").eval(map));
         assertEquals(
-                1L, new EvalExecutor("5 >> 2").eval(map));
+                1L, new AviatorExecutor("5 >> 2").eval(map));
         assertEquals(
-                8L, new EvalExecutor("5 + 3").eval(map));
+                8L, new AviatorExecutor("5 + 3").eval(map));
         assertEquals(
-                2L, new EvalExecutor("5 - 3").eval(map));
+                2L, new AviatorExecutor("5 - 3").eval(map));
         assertEquals(
-                15L, new EvalExecutor("5 * 3").eval(map));
+                15L, new AviatorExecutor("5 * 3").eval(map));
         assertEquals(
-                2L, new EvalExecutor("5 / 2").eval(map));
+                2L, new AviatorExecutor("5 / 2").eval(map));
         assertEquals(
-                2L, new EvalExecutor("5 % 3").eval(map));
+                2L, new AviatorExecutor("5 % 3").eval(map));
 
 
         final Map<String, Object> map1 = new HashMap<>();
         map1.put("flags", 8);
 
         assertEquals("B",
-                new EvalExecutor(" ( flags & 8 ) == 9 ? 'A' : 'B'").eval(map1));
+                new AviatorExecutor(" ( flags & 8 ) == 9 ? 'A' : 'B'").eval(map1));
         assertEquals("A",
-                new EvalExecutor(" ( 5 | flags ) == 13 ? 'A' : 'B'").eval(map1));
+                new AviatorExecutor(" ( 5 | flags ) == 13 ? 'A' : 'B'").eval(map1));
     }
 
     @Test
     public void testFieldArithmeticExpression() throws Exception {
         assertEquals(1L,
-                new EvalExecutor("common_c2s_pkt_num - common_s2c_pkt_num").eval(origin));
+                new AviatorExecutor("common_c2s_pkt_num - common_s2c_pkt_num").eval(origin));
 
-        assertEquals(17L, new EvalExecutor("common_c2s_pkt_num + 10 + 1").eval(origin));
+        assertEquals(17L, new AviatorExecutor("common_c2s_pkt_num + 10 + 1").eval(origin));
 
     }
 
     @Test
     public void testNullField() {
-        var evalExecutor = new EvalExecutor("( a > nil ? a : 0 ) + ( b > nil ? b : 10 )");
+        var evalExecutor = new AviatorExecutor("( a > nil ? a : 0 ) + ( b > nil ? b : 10 )");
         assertEquals(10L, evalExecutor.eval(origin));
     }
 
     @Test
     public void testError() {
         assertThrows(SyntaxErrorException.class,
-                () -> new EvalExecutor("1L"));
+                () -> new AviatorExecutor("1L"));
         assertThrows(SyntaxErrorException.class,
-                () -> new EvalExecutor("2 = 3"));
+                () -> new AviatorExecutor("2 = 3"));
 
         final Map<String, Object> map = new HashMap<>(origin);
         map.put("has_address", true);
-        var convert = new EvalExecutor("common_c2s_pkt_num + has_address");
+        var convert = new AviatorExecutor("common_c2s_pkt_num + has_address");
         assertThrows(ExpressionRuntimeException.class, () -> convert.eval(map));
     }