diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseComplexFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseComplexFunctionCallTransformer.java
new file mode 100644
index 00000000000000..714cfd0011998a
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseComplexFunctionCallTransformer.java
@@ -0,0 +1,22 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser;
+
+import com.starrocks.analysis.Expr;
+
+
+public abstract class BaseComplexFunctionCallTransformer {
+
+    protected abstract Expr transform(String functionName, Expr... args);
+}
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseFunctionCallTransformer.java
new file mode 100644
index 00000000000000..ca0fa56f202b05
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/BaseFunctionCallTransformer.java
@@ -0,0 +1,100 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.starrocks.connector.parser;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.starrocks.analysis.Expr;
+import com.starrocks.analysis.FunctionCallExpr;
+
+import java.util.List;
+import java.util.Map;
+
+public abstract class BaseFunctionCallTransformer {
+    public static Map<String, List<FunctionCallTransformer>> TRANSFORMER_MAP = Maps.newHashMap();
+
+    public BaseFunctionCallTransformer() {
+        registerAllFunctionTransformer();
+    }
+
+    protected abstract void registerAllFunctionTransformer();
+
+    protected abstract Expr convert(String fnName, List<Expr> children);
+
+    public static Expr convertRegisterFn(String fnName, List<Expr> children) {
+        List<FunctionCallTransformer> transformers = TRANSFORMER_MAP.get(fnName);
+        if (transformers == null) {
+            return null;
+        }
+
+        FunctionCallTransformer matcher = null;
+        for (FunctionCallTransformer transformer : transformers) {
+            if (transformer.match(children)) {
+                matcher = transformer;
+            }
+        }
+        if (matcher == null) {
+            return null;
+        }
+        return matcher.transform(children);
+    }
+
+    protected static FunctionCallExpr buildStarRocksFunctionCall(String starRocksFnName,
+                                                               List<Class<? extends Expr>> starRocksArgumentsClass) {
+        List<Expr> arguments = Lists.newArrayList();
+        for (int index = 0; index < starRocksArgumentsClass.size(); ++index) {
+            // For a FunctionCallExpr, do not know the actual arguments here, so we use a PlaceholderExpr to replace it.
+            arguments.add(new PlaceholderExpr(index + 1, starRocksArgumentsClass.get(index)));
+        }
+        return new FunctionCallExpr(starRocksFnName, arguments);
+    }
+
+
+    protected static void registerFunctionTransformer(String originFnName, int originFnArgNums, String starRocksFnName,
+                                                    List<Class<? extends Expr>> starRocksArgumentsClass) {
+        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, starRocksArgumentsClass);
+        registerFunctionTransformer(originFnName, originFnArgNums, starRocksFunctionCall);
+    }
+
+    protected static void registerFunctionTransformer(String originFnName, String starRocksFnName) {
+        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, Lists.newArrayList());
+        registerFunctionTransformer(originFnName, 0, starRocksFunctionCall);
+    }
+
+    protected static void registerFunctionTransformerWithVarArgs(String originFnName, String starRocksFnName,
+                                                               List<Class<? extends Expr>> starRocksArgumentsClass) {
+        Preconditions.checkState(starRocksArgumentsClass.size() == 1);
+        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, starRocksArgumentsClass);
+        registerFunctionTransformerWithVarArgs(originFnName, starRocksFunctionCall);
+    }
+
+    protected static void registerFunctionTransformer(String originFnName, int originFnArgNums,
+                                                    FunctionCallExpr starRocksFunctionCall) {
+        FunctionCallTransformer transformer = new FunctionCallTransformer(starRocksFunctionCall, originFnArgNums);
+
+        List<FunctionCallTransformer> transformerList = TRANSFORMER_MAP.computeIfAbsent(originFnName,
+                k -> Lists.newArrayList());
+        transformerList.add(transformer);
+    }
+
+    protected static void registerFunctionTransformerWithVarArgs(String originFnName, FunctionCallExpr starRocksFunctionCall) {
+        FunctionCallTransformer transformer = new FunctionCallTransformer(starRocksFunctionCall, true);
+
+        List<FunctionCallTransformer> transformerList = TRANSFORMER_MAP.computeIfAbsent(originFnName,
+                k -> Lists.newArrayList());
+        transformerList.add(transformer);
+    }
+}
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/FunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/FunctionCallTransformer.java
similarity index 99%
rename from fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/FunctionCallTransformer.java
rename to fe/fe-core/src/main/java/com/starrocks/connector/parser/FunctionCallTransformer.java
index f6ce80b30c11b2..c5fe16cd2dd75f 100644
--- a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/FunctionCallTransformer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/FunctionCallTransformer.java
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.starrocks.connector.parser.trino;
+package com.starrocks.connector.parser;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/PlaceholderExpr.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/PlaceholderExpr.java
similarity index 96%
rename from fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/PlaceholderExpr.java
rename to fe/fe-core/src/main/java/com/starrocks/connector/parser/PlaceholderExpr.java
index a9205345126340..d31b8339b31b32 100644
--- a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/PlaceholderExpr.java
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/PlaceholderExpr.java
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.starrocks.connector.parser.trino;
+package com.starrocks.connector.parser;
 
 import com.starrocks.analysis.Expr;
 import com.starrocks.sql.ast.AstVisitor;
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/AstBuilder.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/AstBuilder.java
new file mode 100644
index 00000000000000..eaacbe808b6511
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/AstBuilder.java
@@ -0,0 +1,74 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import com.starrocks.analysis.ArithmeticExpr;
+import com.starrocks.analysis.Expr;
+import com.starrocks.analysis.FunctionCallExpr;
+import com.starrocks.analysis.FunctionName;
+import com.starrocks.analysis.FunctionParams;
+import com.starrocks.analysis.HintNode;
+import com.starrocks.analysis.IntLiteral;
+import com.starrocks.analysis.ParseNode;
+import com.starrocks.sql.parser.NodePosition;
+import com.starrocks.sql.parser.StarRocksParser;
+import com.starrocks.sql.parser.SyntaxSugars;
+import org.antlr.v4.runtime.ParserRuleContext;
+
+import java.util.IdentityHashMap;
+import java.util.List;
+
+public class AstBuilder extends com.starrocks.sql.parser.AstBuilder {
+
+    protected AstBuilder(long sqlMode) {
+        super(sqlMode);
+    }
+
+    public AstBuilder(long sqlMode, IdentityHashMap<ParserRuleContext, List<HintNode>> hintMap) {
+        super(sqlMode, hintMap);
+    }
+
+    @Override
+    public ParseNode visitFunctionCallExpression(StarRocksParser.FunctionCallExpressionContext ctx) {
+        return visitChildren(ctx);
+    }
+
+    public ParseNode visitSimpleFunctionCall(StarRocksParser.SimpleFunctionCallContext context) {
+        String fullFunctionName = getQualifiedName(context.qualifiedName()).toString();
+        NodePosition pos = createPos(context);
+
+        List<Expr> arguments = visit(context.expression(), Expr.class);
+        FunctionName fnName = FunctionName.createFnName(fullFunctionName);
+        String functionName = fnName.getFunction();
+
+        Pinot2SRFunctionCallTransformer transformer = new Pinot2SRFunctionCallTransformer();
+        Expr convertedFunctionCall = transformer.convert(functionName, arguments);
+
+        if (convertedFunctionCall != null) {
+            if (functionName.equalsIgnoreCase("fromdatetime")) {
+                ArithmeticExpr toMillis = new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY,
+                        convertedFunctionCall, new IntLiteral(1000));
+                return toMillis;
+            }
+            return convertedFunctionCall;
+        } else {
+            FunctionCallExpr functionCallExpr = new FunctionCallExpr(fnName,
+                    new FunctionParams(false, arguments), pos);
+            if (context.over() != null) {
+                return buildOverClause(functionCallExpr, context.over(), pos);
+            }
+            return SyntaxSugars.parse(functionCallExpr);
+        }
+    }
+}
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/ComplexFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/ComplexFunctionCallTransformer.java
new file mode 100644
index 00000000000000..f788321de9842c
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/ComplexFunctionCallTransformer.java
@@ -0,0 +1,287 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.starrocks.analysis.ArithmeticExpr;
+import com.starrocks.analysis.CompoundPredicate;
+import com.starrocks.analysis.DecimalLiteral;
+import com.starrocks.analysis.Expr;
+import com.starrocks.analysis.FunctionCallExpr;
+import com.starrocks.analysis.FunctionParams;
+import com.starrocks.analysis.IntLiteral;
+import com.starrocks.analysis.StringLiteral;
+import com.starrocks.catalog.FunctionSet;
+import com.starrocks.catalog.Type;
+import com.starrocks.connector.parser.BaseComplexFunctionCallTransformer;
+import com.starrocks.sql.analyzer.SemanticException;
+import com.starrocks.sql.ast.IntervalLiteral;
+import com.starrocks.sql.ast.UnitIdentifier;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ComplexFunctionCallTransformer extends BaseComplexFunctionCallTransformer {
+
+    @Override
+    public Expr transform(String functionName, Expr... args) {
+        List<Expr> argumentsList = Arrays.asList(args);
+
+        if (functionName.equalsIgnoreCase("datetimeconvert")) {
+            return transformDateTimeConvert(argumentsList);
+        } else if (functionName.equalsIgnoreCase("datetrunc")) {
+            return transformDateTrunc(argumentsList);
+        } else if (functionName.equalsIgnoreCase("text_match")) {
+            return transformTextMatch(argumentsList);
+        }
+
+        return null;
+    }
+
+    private static Expr transformDateTimeConvert(List<Expr> args) {
+        if (args.size() < 4 || args.size() > 5) {
+            throw new SemanticException("The datetimeconvert function must include between 4 and 5 parameters, inclusive.");
+        }
+        // DATETIMECONVERT(columnName, inputFormat, outputFormat, outputGranularity)
+        // DATETIMECONVERT(columnName, inputFormat, outputFormat, outputGranularity, timeZone)
+        // format is <time size>:<time unit>:<time format>:<pattern>, only works for time zie to be 1
+        StringLiteral outputGranularity = (StringLiteral) args.get(3);
+        List<String> granularity = PinotParserUtils.parseTime(outputGranularity.getValue());
+        StringLiteral outputFormat = (StringLiteral) args.get(2);
+        List<String> outputFormatList = PinotParserUtils.parseFormat(outputFormat.getValue());
+        String timeFormat = outputFormatList.get(2);
+
+        if (timeFormat.contains("EPOCH")) {
+            return transformDateTimeConvertEpoch(args, granularity, outputFormatList);
+        } else {
+            return transformDateTimeConvertPattern(args, granularity, outputFormatList);
+        }
+    }
+
+    private static Expr transformDateTimeConvertEpoch(List<Expr> args, List<String> granularity, List<String> outputFormatList) {
+        IntervalLiteral intervalLiteral = new IntervalLiteral(new IntLiteral(Integer.parseInt(granularity.get(0))),
+                new UnitIdentifier(granularity.get(1)));
+        FunctionCallExpr timeSlice = new FunctionCallExpr(FunctionSet.TIME_SLICE,
+                getArgumentsForTimeSlice(args.get(0),
+                        intervalLiteral.getValue(), intervalLiteral.getUnitIdentifier().getDescription().toLowerCase(),
+                        "floor"));
+
+        FunctionCallExpr timeSliceTZ = timeSlice;
+        if (args.size() == 5) {
+            timeSliceTZ = new FunctionCallExpr(FunctionSet.CONVERT_TZ,
+                    new FunctionParams(ImmutableList.of(timeSlice, args.get(4), new StringLiteral("UTC"))));
+        }
+
+        FunctionCallExpr unixTimestamp = new FunctionCallExpr(FunctionSet.UNIX_TIMESTAMP,
+                new FunctionParams(ImmutableList.of(timeSliceTZ)));
+        ArithmeticExpr outputTimeUnit = new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, unixTimestamp,
+                new DecimalLiteral(BigDecimal.valueOf(PinotParserUtils.getMultiplier(outputFormatList.get(1)))));
+        return new FunctionCallExpr(FunctionSet.FLOOR, new FunctionParams(ImmutableList.of(outputTimeUnit)));
+    }
+
+    private static Expr transformDateTimeConvertPattern(List<Expr> args, List<String> granularity,
+                                                        List<String> outputFormatList) {
+        //parse the time pattern of the output
+        String[] timePattern = PinotParserUtils.parseDateFormat(outputFormatList.get(3));
+        String formatValue = PinotParserUtils.convertToStrftimeFormat(timePattern[0]);
+        String timeZone = timePattern[1] == null ? "UTC" : timePattern[1];
+        FunctionCallExpr convertTz = new FunctionCallExpr(FunctionSet.CONVERT_TZ,
+                new FunctionParams(ImmutableList.of(args.get(0),
+                        new StringLiteral("UTC"), new StringLiteral(timeZone))));
+
+        IntervalLiteral intervalLiteral = new IntervalLiteral(new IntLiteral(Integer.parseInt(granularity.get(0))),
+                new UnitIdentifier(granularity.get(1)));
+        FunctionCallExpr timeSlice = new FunctionCallExpr(FunctionSet.TIME_SLICE,
+                getArgumentsForTimeSlice(convertTz,
+                        intervalLiteral.getValue(), intervalLiteral.getUnitIdentifier().getDescription().toLowerCase(),
+                        "floor"));
+
+        FunctionCallExpr timeSliceTZ = timeSlice;
+        if (args.size() == 5) {
+            timeSliceTZ = new FunctionCallExpr(FunctionSet.CONVERT_TZ,
+                    new FunctionParams(ImmutableList.of(timeSlice, args.get(4), new StringLiteral("UTC"))));
+        }
+
+        return new FunctionCallExpr(FunctionSet.DATE_FORMAT,
+                new FunctionParams(ImmutableList.of(timeSliceTZ, new StringLiteral(formatValue))));
+    }
+
+    private static Expr transformDateTrunc(List<Expr> args) {
+        if (args.size() < 2 || args.size() > 5) {
+            throw new SemanticException("The datetrunc function must include between 2 and 5 parameters, inclusive.");
+        }
+        // DATETRUNC(unit, timeValue)  or DATETRUNC(unit, timeValue, inputTimeUnitStr) output is milliseconds
+        // Transform to unix_timestamp(date_trunc('day',event_timestamp)) * 1000
+        // DATETRUNC(unit, timeValue, inputTimeUnitStr, timeZone) output is milliseconds
+        // Transform to unix_timestamp(convertz_tz(date_trunc('day', event_timestamp), "UTC", timeZone)) * 1000
+        // DATETRUNC(unit, timeValue, inputTimeUnitStr, timeZone, outputTimeUnitStr)
+        // Transform to unix_timestamp(convertz_tz(date_trunc('day', event_timestamp), "UTC", timeZone)),
+        // Output can be NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS
+        if (args.size() < 4) {
+            return transformDateTruncBasic(args);
+        } else if (args.size() == 4) {
+            return transformDateTruncWithTimeZone(args);
+        } else if (args.size() == 5) {
+            return transformDateTruncWithOutputTimeUnit(args);
+        }
+        return null;
+    }
+
+    private static Expr transformDateTruncBasic(List<Expr> args) {
+        FunctionCallExpr dateTrunc = new FunctionCallExpr(FunctionSet.DATE_TRUNC,
+                new FunctionParams(ImmutableList.of(args.get(0), args.get(1))));
+        FunctionCallExpr unixTimestamp = new FunctionCallExpr(FunctionSet.UNIX_TIMESTAMP,
+                new FunctionParams(ImmutableList.of(dateTrunc)));
+        return new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, unixTimestamp,
+                new IntLiteral(1000));
+    }
+
+    private static Expr transformDateTruncWithTimeZone(List<Expr> args) {
+        FunctionCallExpr dateTrunc = new FunctionCallExpr(FunctionSet.DATE_TRUNC,
+                new FunctionParams(ImmutableList.of(args.get(0), args.get(1))));
+        FunctionCallExpr convertTz = new FunctionCallExpr(FunctionSet.CONVERT_TZ,
+                new FunctionParams(ImmutableList.of(dateTrunc, args.get(3),
+                        new StringLiteral("UTC"))));
+        FunctionCallExpr unixTimestamp = new FunctionCallExpr(FunctionSet.UNIX_TIMESTAMP,
+                new FunctionParams(ImmutableList.of(convertTz)));
+        return new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, unixTimestamp,
+                new IntLiteral(1000));
+    }
+
+    private static Expr transformDateTruncWithOutputTimeUnit(List<Expr> args) {
+        FunctionCallExpr dateTrunc = new FunctionCallExpr(FunctionSet.DATE_TRUNC,
+                new FunctionParams(ImmutableList.of(args.get(0), args.get(1))));
+        FunctionCallExpr convertTz = new FunctionCallExpr(FunctionSet.CONVERT_TZ,
+                new FunctionParams(ImmutableList.of(dateTrunc, args.get(3),
+                        new StringLiteral("UTC"))));
+        FunctionCallExpr unixTimestamp = new FunctionCallExpr(FunctionSet.UNIX_TIMESTAMP,
+                new FunctionParams(ImmutableList.of(convertTz)));
+        StringLiteral outputTimeUnitStr = (StringLiteral) args.get(4);
+        ArithmeticExpr outputTimeUnit = new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, unixTimestamp,
+                new DecimalLiteral(BigDecimal.valueOf(PinotParserUtils.getMultiplier(outputTimeUnitStr.getStringValue()
+                ))));
+        return new FunctionCallExpr(FunctionSet.FLOOR, new FunctionParams(ImmutableList.of(outputTimeUnit)));
+    }
+
+    private static Expr transformTextMatch(List<Expr> args) {
+        if (args.size() < 2) {
+            throw new SemanticException("The text_match function must include at least 2 parameters.");
+        }
+        List<String> parsedInput = parseInputString(((StringLiteral) args.get(1)).getValue());
+        return buildPredicate(parsedInput, args);
+    }
+
+    public static Expr buildPredicate(List<String> parsedList, List<Expr> argumentsList) {
+        Expr current = null;
+
+        for (int i = 0; i < parsedList.size(); i++) {
+            String element = parsedList.get(i);
+
+            if (!element.equals("AND") && !element.equals("OR")) {
+                FunctionCallExpr regexFunc = new FunctionCallExpr(FunctionSet.REGEXP,
+                        new FunctionParams(ImmutableList.of(argumentsList.get(0),
+                            new StringLiteral(element))));
+                if (current == null) {
+                    current = regexFunc;
+                } else {
+                    // The default logic is OR
+                    current = new CompoundPredicate(CompoundPredicate.Operator.OR, current, regexFunc);
+                }
+            } else if (element.equals("AND") || element.equals("OR")) {
+                String operator = element;
+                String nextElement = parsedList.get(++i);
+                FunctionCallExpr rightFunction =  new FunctionCallExpr(FunctionSet.REGEXP,
+                        new FunctionParams(ImmutableList.of(argumentsList.get(0), new StringLiteral(nextElement))));
+                current = new CompoundPredicate(operator.equals("AND") ? CompoundPredicate.Operator.AND :
+                        CompoundPredicate.Operator.OR, current, rightFunction);
+            }
+        }
+
+        return current;
+    }
+
+    public static List<String> parseInputString(String input) {
+        List<String> result = new ArrayList<>();
+
+        // to separate the phrase and term
+        Pattern pattern = Pattern.compile("\"[^\"]+\"|\\S+");
+        Matcher matcher = pattern.matcher(input);
+
+        while (matcher.find()) {
+            result.add(matcher.group());
+        }
+
+        List<String> finalResult = new ArrayList<>();
+        if (input.matches(".*\\b(AND|OR)\\b.*")) {
+            for (int i = 0; i < result.size(); i++) {
+                String token = result.get(i);
+                if (token.equalsIgnoreCase("AND") || token.equalsIgnoreCase("OR")) {
+                    finalResult.add(token.toUpperCase());
+                } else {
+                    addTokenToList(finalResult, token.trim());
+                }
+            }
+        } else {
+            for (int i = 0; i < result.size(); i++) {
+                if (i > 0) {
+                    finalResult.add("OR");
+                }
+                addTokenToList(finalResult, result.get(i).trim());
+            }
+        }
+
+        return finalResult;
+    }
+
+    private static void addTokenToList(List<String> list, String token) {
+        if (token.startsWith("\"") && token.endsWith("\"")) {
+            list.add("\\b" + token.substring(1, token.length() - 1) + "\\b");
+        } else if (token.endsWith("*")) {
+            list.add("^" + token.replace("*", ".*"));
+        } else if (token.startsWith("/") && token.endsWith("/")) {
+            list.add(token.substring(1, token.length() - 1));
+        } else {
+            list.add("\\b" + token + "\\b");
+        }
+    }
+
+    private static List<Expr> getArgumentsForTimeSlice(Expr time, Expr value, String ident, String boundary) {
+        List<Expr> exprs = Lists.newLinkedList();
+        exprs.add(time);
+        addArgumentUseTypeInt(value, exprs);
+        exprs.add(new StringLiteral(ident));
+        exprs.add(new StringLiteral(boundary));
+
+        return exprs;
+    }
+
+    private static void addArgumentUseTypeInt(Expr value, List<Expr> exprs) {
+        // IntLiteral may use TINYINT/SMALLINT/INT/BIGINT type
+        // but time_slice only support INT type when executed in BE
+        try {
+            if (value instanceof IntLiteral) {
+                exprs.add(new IntLiteral(((IntLiteral) value).getValue(), Type.INT));
+            } else {
+                exprs.add(value);
+            }
+        } catch (Exception e) {
+            throw new IllegalArgumentException(String.format("Cast argument %s to int type failed.", value.toSql()));
+        }
+    }
+}
\ No newline at end of file
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/Pinot2SRFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/Pinot2SRFunctionCallTransformer.java
new file mode 100644
index 00000000000000..33d69ee3f907fb
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/Pinot2SRFunctionCallTransformer.java
@@ -0,0 +1,135 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import com.starrocks.analysis.DecimalLiteral;
+import com.starrocks.analysis.Expr;
+import com.starrocks.analysis.FunctionCallExpr;
+import com.starrocks.analysis.IntLiteral;
+import com.starrocks.analysis.StringLiteral;
+import com.starrocks.analysis.VariableExpr;
+import com.starrocks.connector.parser.BaseFunctionCallTransformer;
+import com.starrocks.connector.parser.PlaceholderExpr;
+import com.starrocks.sql.analyzer.SemanticException;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+public class Pinot2SRFunctionCallTransformer extends BaseFunctionCallTransformer {
+
+    public Pinot2SRFunctionCallTransformer() {
+        super();
+    }
+
+    @Override
+    public Expr convert(String fnName, List<Expr> children) {
+        //for some functions, the arguments may need some format or number alignment
+        List<Expr> processedChildren = preprocessChildren(fnName, children);
+
+        Expr result = convertRegisterFn(fnName, processedChildren);
+
+        if (result == null) {
+            ComplexFunctionCallTransformer complexFunctionCallTransformer = new ComplexFunctionCallTransformer();
+            result = complexFunctionCallTransformer.transform(fnName, children.toArray(new Expr[0]));
+        }
+        return result;
+    }
+
+    private static List<Expr> preprocessChildren(String fnName, List<Expr> children) {
+        switch (fnName) {
+            case "todatetime":
+            case "fromdatetime":
+                if (children.size() < 2 || children.size() > 3) {
+                    throw new SemanticException("The todatetime/fromdatetime function must include between " +
+                            "2 and 3 parameters, inclusive.");
+                }
+                // transform date format
+                children.set(1, transformDateFormat(children.get(1)));
+                break;
+
+            default:
+                if (fnName.equalsIgnoreCase("percentiletdigest")) {
+                    children.set(1, transformPercentileValue(children.get(1)));
+                }
+                break;
+        }
+        return children;
+    }
+
+    private static Expr transformDateFormat(Expr child) {
+        if (child instanceof StringLiteral) {
+            String value = ((StringLiteral) child).getValue();
+            // convert to strftime format
+            if (PinotParserUtils.isJavaDateFormat(value)) {
+                String formatValue = PinotParserUtils.convertToStrftimeFormat(value);
+                return new StringLiteral(formatValue);
+            }
+        }
+        return child;
+    }
+
+    private static Expr transformPercentileValue(Expr child) {
+        if (child instanceof IntLiteral) {
+            double value = ((IntLiteral) child).getValue();
+            return new DecimalLiteral(new BigDecimal(Double.toString(value / 100)));
+        } else if (child instanceof DecimalLiteral) {
+            BigDecimal value = ((DecimalLiteral) child).getValue();
+            return new DecimalLiteral(value.divide(new BigDecimal(100)));
+        }
+
+        return child;
+    }
+
+    @Override
+    protected void registerAllFunctionTransformer() {
+        registerDateFunctionTransformer();
+        registerStringFunctionTransformer();
+        registerAggregateFunctionTransformer();
+        // todo: support more function transform
+    }
+
+    private static void registerStringFunctionTransformer() {
+        // regexp_like -> regexp
+        registerFunctionTransformer("regexp_like", 2, "regexp", List.of(Expr.class, Expr.class));
+    }
+
+    private static void registerDateFunctionTransformer() {
+        // todatetime -> date_format
+        registerFunctionTransformer("todatetime", 2, "date_format", List.of(Expr.class, Expr.class));
+
+        // todatetime (time, pattern, timeZone) -> convert_tz, str_to_date
+        registerFunctionTransformer("todatetime", 3, new FunctionCallExpr("convert_tz", List.of(
+                new FunctionCallExpr("date_format", List.of(
+                        new PlaceholderExpr(1, Expr.class), new PlaceholderExpr(2, Expr.class))),
+                new VariableExpr("time_zone"),
+                new PlaceholderExpr(3, Expr.class)
+        )));
+
+        // fromdatetime -> str_to_date
+        registerFunctionTransformer("fromdatetime", 2, new FunctionCallExpr("unix_timestamp",
+                List.of(new FunctionCallExpr("str_to_date", List.of(new PlaceholderExpr(1, Expr.class),
+                        new PlaceholderExpr(2, Expr.class))))));
+    }
+
+    private static void registerAggregateFunctionTransformer() {
+        //distinctcounthll -> approx_count_distinct
+        registerFunctionTransformer("distinctcounthll", 1, "approx_count_distinct", List.of(Expr.class));
+
+        //percentiletdigest -> percentile_approx
+        registerFunctionTransformer("percentiletdigest", 2, "percentile_approx", List.of(Expr.class, Expr.class));
+
+        //percentiletdigest -> percentile_approx
+        registerFunctionTransformer("percentiletdigest", 3, "percentile_approx", List.of(Expr.class, Expr.class, Expr.class));
+    }
+}
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/PinotParserUtils.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/PinotParserUtils.java
new file mode 100644
index 00000000000000..a5c07a248fd235
--- /dev/null
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/pinot/PinotParserUtils.java
@@ -0,0 +1,246 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import com.starrocks.analysis.CastExpr;
+import com.starrocks.analysis.Expr;
+import com.starrocks.analysis.FunctionCallExpr;
+import com.starrocks.analysis.StringLiteral;
+import com.starrocks.catalog.FunctionSet;
+import com.starrocks.common.util.DateUtils;
+
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+
+public class PinotParserUtils {
+
+    /**
+     * Functions that return DATETIME as the result type
+     */
+    private static Set<String> DATETIME_RETURNING_FUNCTIONS = new HashSet<>();
+    private static final Set<String> JAVA_DATE_FORMATS = createDateFormatSet();
+    private static final Map<String, String> UNIT_MAPPING = createUnitMapping();
+
+    static {
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.TIMESTAMPADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.DATE_TRUNC);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.STR_TO_DATE);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.DATE_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.DATE_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.DAYS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.DAYS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.CURRENT_TIMESTAMP);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.YEARS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.YEARS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.NOW);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.WEEKS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.WEEKS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.TIMESTAMP);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.TIME_SLICE);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.SECONDS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.SECONDS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MONTHS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MONTHS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MINUTES_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MINUTES_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MICROSECONDS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.MICROSECONDS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.HOURS_ADD);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.HOURS_SUB);
+        DATETIME_RETURNING_FUNCTIONS.add(FunctionSet.CONVERT_TZ);
+    }
+
+    private static Set<String> createDateFormatSet() {
+        Set<String> formats = new HashSet<>();
+        formats.add("yyyy");
+        formats.add("yy");
+        formats.add("MM");
+        formats.add("M");
+        formats.add("dd");
+        formats.add("d");
+        formats.add("HH");
+        formats.add("H");
+        formats.add("hh");
+        formats.add("h");
+        formats.add("mm");
+        formats.add("ss");
+        formats.add("SSS");
+        formats.add("a");
+        formats.add("E");
+        formats.add("MMM");
+        formats.add("MMMM");
+        return Collections.unmodifiableSet(formats);
+    }
+
+    private static Map<String, String> createUnitMapping() {
+        Map<String, String> mapping = new HashMap<>();
+        mapping.put("DAYS", "DAY");
+        mapping.put("HOURS", "HOUR");
+        mapping.put("MINUTES", "MINUTE");
+        mapping.put("SECONDS", "SECOND");
+        mapping.put("MILLISECONDS", "MILLISECOND");
+        mapping.put("MICROSECONDS", "MICROSECOND");
+        mapping.put("NANOSECONDS", "NANOSECOND");
+        return mapping;
+    }
+
+    public static boolean isDateTimeType(Expr expr) {
+        // type of expr could be Type.INVALID till now, hence we need to examine many other possible cases
+        if (expr.getType().isDatetime()) {
+            return true;
+        } else if (expr instanceof FunctionCallExpr) {
+            return DATETIME_RETURNING_FUNCTIONS.contains(((FunctionCallExpr) expr).getFnName().getFunction().toLowerCase());
+        } else if (expr instanceof CastExpr) {
+            return ((CastExpr) expr).getTargetTypeDef().getType().isDatetime();
+        } else if (expr instanceof com.starrocks.analysis.StringLiteral) {
+            String literal = ((StringLiteral) expr).getStringValue();
+            try {
+                DateUtils.DATE_TIME_FORMATTER_UNIX.parse(literal);
+            } catch (DateTimeParseException e) {
+                try {
+                    DateUtils.DATE_TIME_MS_FORMATTER_UNIX.parse(literal);
+                } catch (DateTimeParseException eMs) {
+                    return false;
+                }
+                return true;
+            }
+            return true;
+        }
+        return false;
+    }
+
+    public static boolean isJavaDateFormat(String format) {
+        for (String javaFormat : JAVA_DATE_FORMATS) {
+            if (format.contains(javaFormat)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public static String convertToStrftimeFormat(String javaDateFormat) {
+        String cleanedFormat = javaDateFormat.replace("Z", "").replace("'", "");
+
+        Map<String, String> formatMapping = new LinkedHashMap<>();
+        formatMapping.put("yyyy", "%Y");
+        formatMapping.put("MM", "%m");
+        formatMapping.put("dd", "%d");
+        formatMapping.put("HH", "%H");
+        formatMapping.put("mm", "%i");
+        formatMapping.put("ss", "%S");
+        formatMapping.put("SSS", "%f");
+        formatMapping.put("yy", "%y");
+        formatMapping.put("hh", "%I");
+        formatMapping.put("h", "%I");
+        formatMapping.put("a", "%p");
+        formatMapping.put("E", "%a");
+        formatMapping.put("MMM", "%b");
+        formatMapping.put("MMMM", "%B");
+
+        String strftimeFormat = cleanedFormat;
+        for (Map.Entry<String, String> entry : formatMapping.entrySet()) {
+            strftimeFormat = strftimeFormat.replace(entry.getKey(), entry.getValue());
+        }
+        return strftimeFormat;
+    }
+
+    public static List<String> parseTime(String input) {
+        String[] parts = input.split(":");
+        if (parts.length != 2) {
+            throw new IllegalArgumentException("Invalid input format. Expected <time size>:<time unit>");
+        }
+
+        String timeSize = parts[0].trim();
+        String timeUnit = parts[1].trim().toUpperCase();
+
+        if (UNIT_MAPPING.containsKey(timeUnit)) {
+            timeUnit = UNIT_MAPPING.get(timeUnit);
+        } else {
+            throw new IllegalArgumentException("Invalid time unit: " + timeUnit);
+        }
+
+        List<String> result = new ArrayList<>();
+        result.add(timeSize);
+        result.add(timeUnit);
+        return result;
+    }
+
+    public static List<String> parseFormat(String format) {
+        String[] parts = format.split(":", 4);
+        if (parts.length < 3) {
+            throw new IllegalArgumentException("Invalid input format. Expected <time size>:<time unit>:<time format>:<pattern>");
+        }
+
+        String timeSize = parts[0].trim();
+        String timeUnit = parts[1].trim().toUpperCase(Locale.ROOT);
+        String timeFormat = parts[2].trim().toUpperCase(Locale.ROOT);
+        String pattern = parts.length == 4 ? parts[3] : null;
+
+        List<String> result = new ArrayList<>();
+        result.add(timeSize);
+        result.add(timeUnit);
+        result.add(timeFormat);
+        result.add(pattern);
+        return result;
+    }
+
+    public static double getMultiplier(String outputTimeUnitStr) {
+        switch (outputTimeUnitStr.toUpperCase()) {
+            case "NANOSECONDS":
+                return 1000000000.0;
+            case "MICROSECONDS":
+                return 1000000.0;
+            case "MILLISECONDS":
+                return 1000.0;
+            case "SECONDS":
+                return 1.0;
+            case "MINUTES":
+                return 1.0 / 60;
+            case "HOURS":
+                return 1.0 / 3600;
+            case "DAYS":
+                return 1.0 / 86400;
+            default:
+                throw new IllegalArgumentException("Invalid outputTimeUnitStr: " + outputTimeUnitStr);
+        }
+    }
+
+    public static String[] parseDateFormat(String pattern) {
+        String regex =  "^([\\d\\- :yMdHm]{10,16})( tz\\((.*?)\\))?$";
+        Pattern p = Pattern.compile(regex);
+        Matcher m = p.matcher(pattern);
+
+        String dateFormat = null;
+        String timeZone = null;
+
+        if (m.matches()) {
+            dateFormat = m.group(1);
+            timeZone = m.group(3);
+        }
+
+        return new String[] {dateFormat, timeZone};
+    }
+
+}
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/AstBuilder.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/AstBuilder.java
index f95fc0b5bdf0e0..2d43b49ca7242e 100644
--- a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/AstBuilder.java
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/AstBuilder.java
@@ -732,7 +732,8 @@ protected ParseNode visitFunctionCall(FunctionCall node, ParseTreeContext contex
         List<Expr> arguments = visit(node.getArguments(), context, Expr.class);
 
         Expr callExpr;
-        Expr convertedFunctionCall = Trino2SRFunctionCallTransformer.convert(node.getName().toString(), arguments);
+        Trino2SRFunctionCallTransformer trino2SRFunctionCallTransformer = new Trino2SRFunctionCallTransformer();
+        Expr convertedFunctionCall = trino2SRFunctionCallTransformer.convert(node.getName().toString(), arguments);
         if (convertedFunctionCall != null) {
             callExpr = convertedFunctionCall;
         } else if (DISTINCT_FUNCTION.contains(node.getName().toString())) {
@@ -1022,7 +1023,8 @@ protected ParseNode visitCoalesceExpression(CoalesceExpression node, ParseTreeCo
     @Override
     protected ParseNode visitExtract(Extract node, ParseTreeContext context) {
         String fieldString = node.getField().toString().toLowerCase();
-        Expr expr = Trino2SRFunctionCallTransformer.convert(fieldString,
+        Trino2SRFunctionCallTransformer trino2SRFunctionCallTransformer = new Trino2SRFunctionCallTransformer();
+        Expr expr = trino2SRFunctionCallTransformer.convert(fieldString,
                 Lists.newArrayList((Expr) visit(node.getExpression(), context)));
         return Objects.requireNonNullElseGet(expr, () -> new FunctionCallExpr(fieldString,
                 new FunctionParams(Lists.newArrayList((Expr) visit(node.getExpression(), context)))));
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/ComplexFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/ComplexFunctionCallTransformer.java
index c7cbbd74ae8c72..18e08ffd838e2a 100644
--- a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/ComplexFunctionCallTransformer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/ComplexFunctionCallTransformer.java
@@ -30,13 +30,16 @@
 import com.starrocks.analysis.TimestampArithmeticExpr;
 import com.starrocks.catalog.FunctionSet;
 import com.starrocks.catalog.Type;
+import com.starrocks.connector.parser.BaseComplexFunctionCallTransformer;
 import com.starrocks.sql.analyzer.SemanticException;
 import com.starrocks.sql.ast.MapExpr;
 
 import java.util.Collections;
 
-public class ComplexFunctionCallTransformer {
-    public static Expr transform(String functionName, Expr... args) {
+public class ComplexFunctionCallTransformer extends BaseComplexFunctionCallTransformer {
+
+    @Override
+    public Expr transform(String functionName, Expr... args) {
         if (functionName.equalsIgnoreCase("date_add")) {
             if (args.length == 3 && args[0] instanceof StringLiteral) {
                 StringLiteral unit = (StringLiteral) args[0];
diff --git a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/Trino2SRFunctionCallTransformer.java b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/Trino2SRFunctionCallTransformer.java
index 1f8eccc06a7c13..50683812fad387 100644
--- a/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/Trino2SRFunctionCallTransformer.java
+++ b/fe/fe-core/src/main/java/com/starrocks/connector/parser/trino/Trino2SRFunctionCallTransformer.java
@@ -14,9 +14,6 @@
 
 package com.starrocks.connector.parser.trino;
 
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.starrocks.analysis.CaseExpr;
 import com.starrocks.analysis.CaseWhenClause;
 import com.starrocks.analysis.Expr;
@@ -24,9 +21,11 @@
 import com.starrocks.analysis.IntLiteral;
 import com.starrocks.analysis.StringLiteral;
 import com.starrocks.analysis.VariableExpr;
+import com.starrocks.connector.parser.BaseFunctionCallTransformer;
+import com.starrocks.connector.parser.FunctionCallTransformer;
+import com.starrocks.connector.parser.PlaceholderExpr;
 
 import java.util.List;
-import java.util.Map;
 
 /**
  * This Class is the entry to to convert a Trino function to StarRocks built-in function, we use
@@ -35,41 +34,24 @@
  * FunctionCallTransformer, all stored in TRANSFORMER_MAP.
  */
 
-public class Trino2SRFunctionCallTransformer {
-    // function name -> list of function transformer
-    public static Map<String, List<FunctionCallTransformer>> TRANSFORMER_MAP = Maps.newHashMap();
+public class Trino2SRFunctionCallTransformer extends BaseFunctionCallTransformer {
 
-    static {
-        registerAllFunctionTransformer();
+    public Trino2SRFunctionCallTransformer() {
+        super();
     }
 
-    public static Expr convert(String fnName, List<Expr> children) {
+    @Override
+    public Expr convert(String fnName, List<Expr> children) {
         Expr result = convertRegisterFn(fnName, children);
         if (result == null) {
-            result = ComplexFunctionCallTransformer.transform(fnName, children.toArray(new Expr[0]));
+            ComplexFunctionCallTransformer complexFunctionCallTransformer = new ComplexFunctionCallTransformer();
+            result = complexFunctionCallTransformer.transform(fnName, children.toArray(new Expr[0]));
         }
         return result;
     }
 
-    public static Expr convertRegisterFn(String fnName, List<Expr> children) {
-        List<FunctionCallTransformer> transformers = TRANSFORMER_MAP.get(fnName);
-        if (transformers == null) {
-            return null;
-        }
-
-        FunctionCallTransformer matcher = null;
-        for (FunctionCallTransformer transformer : transformers) {
-            if (transformer.match(children)) {
-                matcher = transformer;
-            }
-        }
-        if (matcher == null) {
-            return null;
-        }
-        return matcher.transform(children);
-    }
-
-    private static void registerAllFunctionTransformer() {
+    @Override
+    protected void registerAllFunctionTransformer() {
         registerAggregateFunctionTransformer();
         registerArrayFunctionTransformer();
         registerDateFunctionTransformer();
@@ -390,49 +372,4 @@ private static void registerMathFunctionTransformer() {
                 List.of(new PlaceholderExpr(1, Expr.class), new IntLiteral(0))));
 
     }
-
-    private static void registerFunctionTransformer(String trinoFnName, int trinoFnArgNums, String starRocksFnName,
-                                                    List<Class<? extends Expr>> starRocksArgumentsClass) {
-        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, starRocksArgumentsClass);
-        registerFunctionTransformer(trinoFnName, trinoFnArgNums, starRocksFunctionCall);
-    }
-
-    private static void registerFunctionTransformer(String trinoFnName, String starRocksFnName) {
-        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, Lists.newArrayList());
-        registerFunctionTransformer(trinoFnName, 0, starRocksFunctionCall);
-    }
-
-    private static void registerFunctionTransformerWithVarArgs(String trinoFnName, String starRocksFnName,
-                                                               List<Class<? extends Expr>> starRocksArgumentsClass) {
-        Preconditions.checkState(starRocksArgumentsClass.size() == 1);
-        FunctionCallExpr starRocksFunctionCall = buildStarRocksFunctionCall(starRocksFnName, starRocksArgumentsClass);
-        registerFunctionTransformerWithVarArgs(trinoFnName, starRocksFunctionCall);
-    }
-
-    private static void registerFunctionTransformer(String trinoFnName, int trinoFnArgNums,
-                                                    FunctionCallExpr starRocksFunctionCall) {
-        FunctionCallTransformer transformer = new FunctionCallTransformer(starRocksFunctionCall, trinoFnArgNums);
-
-        List<FunctionCallTransformer> transformerList = TRANSFORMER_MAP.computeIfAbsent(trinoFnName,
-                k -> Lists.newArrayList());
-        transformerList.add(transformer);
-    }
-
-    private static void registerFunctionTransformerWithVarArgs(String trinoFnName, FunctionCallExpr starRocksFunctionCall) {
-        FunctionCallTransformer transformer = new FunctionCallTransformer(starRocksFunctionCall, true);
-
-        List<FunctionCallTransformer> transformerList = TRANSFORMER_MAP.computeIfAbsent(trinoFnName,
-                k -> Lists.newArrayList());
-        transformerList.add(transformer);
-    }
-
-    private static FunctionCallExpr buildStarRocksFunctionCall(String starRocksFnName,
-                                                               List<Class<? extends Expr>> starRocksArgumentsClass) {
-        List<Expr> arguments = Lists.newArrayList();
-        for (int index = 0; index < starRocksArgumentsClass.size(); ++index) {
-            // For a FunctionCallExpr, do not know the actual arguments here, so we use a PlaceholderExpr to replace it.
-            arguments.add(new PlaceholderExpr(index + 1, starRocksArgumentsClass.get(index)));
-        }
-        return new FunctionCallExpr(starRocksFnName, arguments);
-    }
 }
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitor.java b/fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitor.java
index 8bf71a9220e4b8..80bd9a6eff34ff 100644
--- a/fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitor.java
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitor.java
@@ -52,7 +52,7 @@
 import com.starrocks.analysis.UserVariableExpr;
 import com.starrocks.analysis.UserVariableHint;
 import com.starrocks.analysis.VariableExpr;
-import com.starrocks.connector.parser.trino.PlaceholderExpr;
+import com.starrocks.connector.parser.PlaceholderExpr;
 import com.starrocks.sql.ShowTemporaryTableStmt;
 import com.starrocks.sql.ast.feedback.AddPlanAdvisorStmt;
 import com.starrocks.sql.ast.feedback.ClearPlanAdvisorStmt;
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java b/fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java
index a2184ba77e7d89..15d6de5e7a0c63 100644
--- a/fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java
@@ -597,6 +597,10 @@ public AstBuilder create(long sqlMode) {
         public AstBuilder create(long sqlMode, IdentityHashMap<ParserRuleContext, List<HintNode>> hintMap) {
             return new AstBuilder(sqlMode, hintMap);
         }
+        public com.starrocks.connector.parser.pinot.AstBuilder createPinotAst(long sqlMode, IdentityHashMap<ParserRuleContext,
+                List<HintNode>> hintMap) {
+            return new com.starrocks.connector.parser.pinot.AstBuilder(sqlMode, hintMap);
+        }
     }
 
     public List<Parameter> getParameters() {
@@ -7201,7 +7205,7 @@ public ParseNode visitWindowFunction(StarRocksParser.WindowFunctionContext conte
         return functionCallExpr;
     }
 
-    private AnalyticExpr buildOverClause(FunctionCallExpr functionCallExpr, StarRocksParser.OverContext context,
+    protected AnalyticExpr buildOverClause(FunctionCallExpr functionCallExpr, StarRocksParser.OverContext context,
                                          NodePosition pos) {
         functionCallExpr.setIsAnalyticFnCall(true);
         List<OrderByElement> orderByElements = new ArrayList<>();
@@ -8275,7 +8279,7 @@ private TableName getTableName(StarRocksParser.QualifiedNameContext context) {
         return qualifiedNameToTableName(getQualifiedName(context));
     }
 
-    private QualifiedName getQualifiedName(StarRocksParser.QualifiedNameContext context) {
+    protected QualifiedName getQualifiedName(StarRocksParser.QualifiedNameContext context) {
         List<String> parts = new ArrayList<>();
         NodePosition pos = createPos(context);
         for (ParseTree c : context.children) {
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/parser/SqlParser.java b/fe/fe-core/src/main/java/com/starrocks/sql/parser/SqlParser.java
index d92d2d670e544e..fa7e351f3e501d 100644
--- a/fe/fe-core/src/main/java/com/starrocks/sql/parser/SqlParser.java
+++ b/fe/fe-core/src/main/java/com/starrocks/sql/parser/SqlParser.java
@@ -64,6 +64,8 @@ public static List<StatementBase> parse(String sql, SessionVariable sessionVaria
         try {
             if (sessionVariable.getSqlDialect().equalsIgnoreCase("trino")) {
                 return parseWithTrinoDialect(sql, sessionVariable);
+            } else if (sessionVariable.getSqlDialect().equalsIgnoreCase("pinot")) {
+                return parseWithPinotDialect(sql, sessionVariable);
             } else {
                 return parseWithStarRocksDialect(sql, sessionVariable);
             }
@@ -73,6 +75,31 @@ public static List<StatementBase> parse(String sql, SessionVariable sessionVaria
         }
     }
 
+    private static List<StatementBase> parseWithPinotDialect(String sql, SessionVariable sessionVariable) {
+        List<StatementBase> statements = Lists.newArrayList();
+        Pair<ParserRuleContext, StarRocksParser> pair = invokeParser(sql, sessionVariable, StarRocksParser::sqlStatements);
+        StarRocksParser.SqlStatementsContext sqlStatementsContext = (StarRocksParser.SqlStatementsContext) pair.first;
+        List<StarRocksParser.SingleStatementContext> singleStatementContexts = sqlStatementsContext.singleStatement();
+        for (int idx = 0; idx < singleStatementContexts.size(); ++idx) {
+            // collect hint info
+            HintCollector collector = new HintCollector((CommonTokenStream) pair.second.getTokenStream(), sessionVariable);
+            collector.collect(singleStatementContexts.get(idx));
+            AstBuilder astBuilder = GlobalStateMgr.getCurrentState().getSqlParser().astBuilderFactory
+                    .createPinotAst(sessionVariable.getSqlMode(), collector.getContextWithHintMap());
+
+            StatementBase statement = (StatementBase) astBuilder.visitSingleStatement(singleStatementContexts.get(idx));
+            if (astBuilder.getParameters() != null && astBuilder.getParameters().size() != 0
+                    && !(statement instanceof PrepareStmt)) {
+                // for prepare stm1 from  '', here statement is inner statement
+                statement = new PrepareStmt("", statement, astBuilder.getParameters());
+            } else {
+                statement.setOrigStmt(new OriginStatement(sql, idx));
+            }
+            statements.add(statement);
+        }
+        return statements;
+    }
+
     private static List<StatementBase> parseWithTrinoDialect(String sql, SessionVariable sessionVariable) {
         List<StatementBase> statements = Lists.newArrayList();
         try {
diff --git a/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotQueryTest.java b/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotQueryTest.java
new file mode 100644
index 00000000000000..74df54bcaa988c
--- /dev/null
+++ b/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotQueryTest.java
@@ -0,0 +1,409 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class PinotQueryTest extends PinotTestBase {
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        PinotTestBase.beforeClass();
+        starRocksAssert.getCtx().getSessionVariable().setCboPushDownAggregateMode(-1);
+    }
+
+    @Test
+    public void testSelectClause() throws Exception {
+        String sql = "select * from test.t0;";
+        assertPlanContains(sql, "OUTPUT EXPRS:1: v1 | 2: v2 | 3: v3");
+
+        sql = "select * from test.t0 limit 100;";
+        assertPlanContains(sql, "limit: 100");
+
+        sql = "select \"v1\", \"v2\" from test.t0";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 4> : 'v1'\n" +
+                "  |  <slot 5> : 'v2'");
+    }
+
+    @Test
+    public void testSelectAggregation() throws Exception {
+        String sql = "select count(*) from test.t0";
+        assertPlanContains(sql, "2:AGGREGATE (update finalize)\n" +
+                "  |  output: count(*)");
+
+        sql = "select max(v1) from test.t0";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: max(1: v1)");
+
+        sql = "select sum(v2) from test.t0";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: sum(2: v2)");
+    }
+
+    @Test
+    public void testSelectGroupingOnAggregation() throws Exception {
+        String sql = "select min(v1), max(v2) from test.t0 group by v3";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: min(1: v1), max(2: v2)\n" +
+                "  |  group by: 3: v3");
+    }
+
+    @Test
+    public void testSelectOrderingOnAggregation() throws Exception {
+        String sql = "select min(v1), max(v2) from test.t0 group by v3 order by min(v1)";
+
+        assertPlanContains(sql, "3:SORT\n" +
+                "  |  order by: <slot 4> 4: min ASC\n" +
+                "  |  offset: 0");
+    }
+
+    @Test
+    public void testSelectFilter() throws Exception {
+        String sql = "select * from test.t0 where v1 > 10";
+        assertPlanContains(sql, "PREDICATES: 1: v1 > 10");
+
+        sql = "select * from test.t0 where v1 > 10 and v2 < 20";
+        assertPlanContains(sql, "PREDICATES: 1: v1 > 10, 2: v2 < 20");
+
+        sql = "select * from test.t0 where v1 > 10 or v2 < 20";
+        assertPlanContains(sql, "PREDICATES: (1: v1 > 10) OR (2: v2 < 20)");
+
+        sql = "SELECT COUNT(*) \n" +
+                "FROM test.tall\n" +
+                "  WHERE ta = 'foo'\n" +
+                "  AND td BETWEEN 1 AND 20\n" +
+                "  OR (tg < 42 AND ta IN ('hello', 'goodbye') AND ta NOT IN (42, 69));";
+
+        assertPlanContains(sql, "PREDICATES: ((1: ta = 'foo') AND ((4: td >= 1) AND (4: td <= 20))) OR (((7: tg < 42) " +
+                "AND (1: ta IN ('hello', 'goodbye'))) AND (1: ta NOT IN ('42', '69'))), " +
+                "1: ta IN ('foo', 'hello', 'goodbye')");
+    }
+
+    @Test
+    public void testOrderOnSelection() throws Exception {
+        String sql = "select * from test.t0 order by v1";
+        assertPlanContains(sql, "1:SORT\n" +
+                "  |  order by: <slot 1> 1: v1 ASC\n" +
+                "  |  offset: 0");
+
+        sql = "select * from test.t0 order by v1 desc";
+        assertPlanContains(sql, "1:SORT\n" +
+                "  |  order by: <slot 1> 1: v1 DESC\n" +
+                "  |  offset: 0");
+
+        sql = "select * from test.t0 order by v1 desc, v2 asc";
+        assertPlanContains(sql, "1:SORT\n" +
+                "  |  order by: <slot 1> 1: v1 DESC, <slot 2> 2: v2 ASC\n" +
+                "  |  offset: 0");
+    }
+
+    @Test
+    public void testPaginationOnSelection() throws Exception {
+        String sql = "select v1, v2 from test.t0 where v2 > 10 order by v2 desc limit 10, 20";
+        assertPlanContains(sql, "2:MERGING-EXCHANGE\n" +
+                "     offset: 10\n" +
+                "     limit: 20");
+    }
+
+    @Test
+    public void testCaseWhen() throws Exception {
+        String sql = "select case when v1 > 10 then 1 else 0 end from test.t0";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 4> : if(1: v1 > 10, 1, 0)");
+
+        sql = "select sum(case when v1 > 10 then 1 when v1 < 10 then 0 else 2 end) as total_cost from test.t0";
+        assertPlanContains(sql, "");
+    }
+
+    @Test
+    public void testJoin() throws Exception {
+        String sql = "select * from test.t0 join test.t1 on t0.v1 = t1.v4";
+        assertPlanContains(sql, "4:HASH JOIN\n" +
+                "  |  join op: INNER JOIN (PARTITIONED)\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 4: v4");
+
+        sql = "select * from test.t0 join test.t1 on t0.v1 = t1.v4 join test.t2 on t0.v1 = t2.v7";
+        assertPlanContains(sql, "7:HASH JOIN\n" +
+                "  |  join op: INNER JOIN (BUCKET_SHUFFLE(S))\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 7: v7\n" +
+                "  |  \n" +
+                "  |----6:EXCHANGE\n" +
+                "  |    \n" +
+                "  4:HASH JOIN\n" +
+                "  |  join op: INNER JOIN (PARTITIONED)\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 4: v4");
+
+        sql = "select * from test.t0 left join test.t1 on t0.v1 = t1.v4";
+        assertPlanContains(sql, " 4:HASH JOIN\n" +
+                "  |  join op: LEFT OUTER JOIN (PARTITIONED)\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 4: v4");
+
+        sql = "select * from test.t0 right join test.t1 on t0.v1 = t1.v4";
+        assertPlanContains(sql, "4:HASH JOIN\n" +
+                "  |  join op: RIGHT OUTER JOIN (PARTITIONED)\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 4: v4");
+
+        sql = "select * from test.t0 full join test.t1 on t0.v1 = t1.v4";
+        assertPlanContains(sql, "4:HASH JOIN\n" +
+                "  |  join op: FULL OUTER JOIN (PARTITIONED)\n" +
+                "  |  colocate: false, reason: \n" +
+                "  |  equal join conjunct: 1: v1 = 4: v4");
+
+        sql = "select * from test.t0 join test.t1 on t0.v1 = t1.v4 where t0.v1 > 10";
+        assertPlanContains(sql, "PREDICATES: 1: v1 > 10");
+    }
+
+    @Test
+    public void testDateTruncFunction() throws Exception {
+        String sql = "select dateTrunc('week', th) AS ts\n" +
+                "FROM test.tall\n";
+
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : unix_timestamp(date_trunc('week', 8: th)) * 1000");
+
+        sql = "select dateTrunc('week', th, 'MILLISECONDS') AS ts\n" +
+                "FROM test.tall";
+
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : unix_timestamp(date_trunc('week', 8: th)) * 1000");
+
+        sql = "select dateTrunc(\n" +
+                "  'week', \n" +
+                "  th, \n" +
+                "  'MILLISECONDS', \n" +
+                "  'UTC', \n" +
+                "  'SECONDS'\n" +
+                ") AS ts\n" +
+                "FROM test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(convert_tz(date_trunc('week', 8: th), 'UTC', 'UTC')) " +
+                "AS DECIMAL128(18,0)) * 1.0 AS DOUBLE))");
+
+        sql = "select dateTrunc(\n" +
+                "  'week', \n" +
+                "  th, \n" +
+                "  'MILLISECONDS', \n" +
+                "  'CET', \n" +
+                "  'SECONDS'\n" +
+                ") AS ts\n" +
+                "FROM test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(convert_tz(date_trunc('week', 8: th), " +
+                "'CET', 'UTC')) AS DECIMAL128(18,0)) * 1.0 AS DOUBLE))");
+
+        sql = "select dateTrunc(\n" +
+                "  'quarter', \n" +
+                "  th, \n" +
+                "  'MILLISECONDS', \n" +
+                "  'America/Los_Angeles', \n" +
+                "  'HOURS'\n" +
+                ") AS ts\n" +
+                "FROM test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(convert_tz(date_trunc('quarter', 8: th), " +
+                "'America/Los_Angeles', 'UTC')) AS DECIMAL128(18,0)) * 0.0002777777777777778 AS DOUBLE))");
+    }
+
+    @Test
+    public void testDateTimeConvertFunction() throws Exception {
+        String sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:DAYS:EPOCH', \n" +
+                "         '1:DAYS'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(time_slice(8: th, 1, 'day', 'floor')) " +
+                "AS DECIMAL128(18,0)) * 0.000011574074074074073 AS DOUBLE))");
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:DAYS'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(time_slice(8: th, 1, 'day', 'floor')) " +
+                "AS DECIMAL128(18,0)) * 1000.0 AS DOUBLE))");
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:DAYS',\n" +
+                "         'Europe/Berlin'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(convert_tz(time_slice(8: th, 1, 'day', 'floor'), " +
+                "'Europe/Berlin', 'UTC')) AS DECIMAL128(18,0)) * 1000.0 AS DOUBLE))");
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '15:MINUTES'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : floor(CAST(CAST(unix_timestamp(time_slice(8: th, 15, 'minute', 'floor')) " +
+                "AS DECIMAL128(18,0)) * 1000.0 AS DOUBLE))");
+
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd', \n" +
+                "         '1:DAYS'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(time_slice(convert_tz(8: th, 'UTC', 'UTC'), 1, 'day', 'floor'), '%Y-%m-%d')");
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd HH:mm tz(Pacific/Kiritimati)', \n" +
+                "         '1:MILLISECONDS'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(time_slice(convert_tz(8: th, 'UTC', 'Pacific/Kiritimati'), 1, " +
+                "'millisecond', 'floor'), '%Y-%m-%d %H:%i')");
+
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd HH:mm tz(Pacific/Kiritimati)', \n" +
+                "         '1:DAYS'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(time_slice(convert_tz(8: th, 'UTC', 'Pacific/Kiritimati'), 1, " +
+                "'day', 'floor'), '%Y-%m-%d %H:%i')");
+
+        sql = "select DATETIMECONVERT(\n" +
+                "         th, \n" +
+                "         '1:MILLISECONDS:EPOCH', \n" +
+                "         '1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd HH:mm', \n" +
+                "         '1:DAYS',\n" +
+                "         'Europe/Berlin'\n" +
+                "       ) AS convertedTime\n" +
+                "from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(convert_tz(time_slice(convert_tz(8: th, 'UTC', 'UTC'), 1, " +
+                "'day', 'floor'), 'Europe/Berlin', 'UTC'), '%Y-%m-%d %H:%i')");
+    }
+
+    @Test
+    public void testToDateTimeFunction() throws Exception {
+        String sql = "SELECT ToDateTime(th, 'yyyy-MM-dd') AS dateTimeString from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(8: th, '%Y-%m-%d')");
+
+        sql = "SELECT ToDateTime(\n" +
+                "    th, \n" +
+                "    'yyyy-MM-dd hh:mm:ss a'\n" +
+                "    ) AS dateTimeString from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : date_format(8: th, '%Y-%m-%d %I:%i:%S %p')");
+
+        sql = "SELECT ToDateTime(\n" +
+                "    th, \n" +
+                "    'yyyy-MM-dd HH:mm:ss Z',\n" +
+                "    'CET'\n" +
+                "    ) AS dateTimeString from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : convert_tz(CAST(date_format(8: th, '%Y-%m-%d %H:%i:%S ') " +
+                "AS DATETIME), 'Asia/Shanghai', 'CET')");
+    }
+
+    @Test
+    public void testFromDateTimeFunction() throws Exception {
+        String sql = "select FromDateTime(th, 'yyyy-MM-dd') AS epochMillis from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : unix_timestamp(str_to_date(CAST(8: th AS VARCHAR), '%Y-%m-%d')) * 1000");
+
+        sql = "select FromDateTime(\n" +
+                "    th, \n" +
+                "    'yyyy-MM-dd hh:mm:ss a'\n" +
+                "    ) AS epochMillis from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : unix_timestamp(str_to_date(CAST(8: th AS VARCHAR), '%Y-%m-%d %I:%i:%S %p')) * 1000");
+
+        sql = "select FromDateTime(\n" +
+                "    th, \n" +
+                "    'yyyy-MM-dd''T''HH:mm:ss'\n" +
+                "    ) AS epochMillis from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : unix_timestamp(str_to_date(CAST(8: th AS VARCHAR), '%Y-%m-%dT%H:%i:%S')) * 1000");
+    }
+
+    @Test
+    public void testAggregationFunction() throws Exception {
+
+        String sql = "select DISTINCTCOUNTHLL(ta) AS value from test.tall";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: approx_count_distinct(1: ta)");
+
+        sql = "select PERCENTILETDIGEST(tg, 50, 1000) AS value from test.tall";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: percentile_approx(CAST(7: tg AS DOUBLE), 0.5, 1000.0)");
+
+        sql = "select PERCENTILETDIGEST(tg, 99.9) AS value from test.tall";
+        assertPlanContains(sql, "1:AGGREGATE (update finalize)\n" +
+                "  |  output: percentile_approx(CAST(7: tg AS DOUBLE), 0.999)");
+    }
+
+    @Test
+    public void testTextMatchFunction() throws Exception {
+        String sql = "select text_match(ta, '\"distributed system\"') AS value from test.tall";
+        assertPlanContains(sql, " 1:Project\n" +
+                "  |  <slot 12> : regexp(1: ta, '\\\\bdistributed system\\\\b')");
+
+        sql = "select text_match(ta, 'Java') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : regexp(1: ta, '\\\\bJava\\\\b')");
+
+        sql = "select text_match(ta, 'Java AND C++') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : (regexp(1: ta, '\\\\bJava\\\\b')) AND (regexp(1: ta, '\\\\bC++\\\\b'))");
+
+        sql = "select text_match(ta, 'stream*') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : regexp(1: ta, '^stream.*')");
+
+        sql = "select text_match(ta, '/Exception.*/') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : regexp(1: ta, 'Exception.*')");
+
+        sql = "select text_match(ta, 'Java C++') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : (regexp(1: ta, '\\\\bJava\\\\b')) OR (regexp(1: ta, '\\\\bC++\\\\b'))");
+
+        sql = "select text_match(ta, '\"Machine learning\" AND gpu AND python') AS value from test.tall";
+        assertPlanContains(sql, "1:Project\n" +
+                "  |  <slot 12> : ((regexp(1: ta, '\\\\bMachine learning\\\\b')) AND " +
+                "(regexp(1: ta, '\\\\bgpu\\\\b'))) AND (regexp(1: ta, '\\\\bpython\\\\b'))");
+    }
+}
\ No newline at end of file
diff --git a/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotTestBase.java b/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotTestBase.java
new file mode 100644
index 00000000000000..5ee84071cf3ba3
--- /dev/null
+++ b/fe/fe-core/src/test/java/com/starrocks/connector/parser/pinot/PinotTestBase.java
@@ -0,0 +1,171 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.starrocks.connector.parser.pinot;
+
+import com.starrocks.common.FeConstants;
+import com.starrocks.common.Pair;
+import com.starrocks.planner.TpchSQL;
+import com.starrocks.qe.ConnectContext;
+import com.starrocks.sql.StatementPlanner;
+import com.starrocks.sql.ast.StatementBase;
+import com.starrocks.sql.optimizer.LogicalPlanPrinter;
+import com.starrocks.sql.optimizer.dump.QueryDumpInfo;
+import com.starrocks.sql.plan.ExecPlan;
+import com.starrocks.thrift.TExplainLevel;
+import com.starrocks.utframe.StarRocksAssert;
+import com.starrocks.utframe.UtFrameUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.rules.ErrorCollector;
+
+import java.util.List;
+
+public class PinotTestBase {
+    public static ConnectContext connectContext;
+    public static StarRocksAssert starRocksAssert;
+
+    @Rule
+    public ErrorCollector collector = new ErrorCollector();
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        UtFrameUtils.createMinStarRocksCluster();
+        FeConstants.enablePruneEmptyOutputScan = false;
+        connectContext = UtFrameUtils.createDefaultCtx();
+        starRocksAssert = new StarRocksAssert(connectContext);
+        String dbName = "test";
+        starRocksAssert.withDatabase(dbName).useDatabase(dbName);
+        starRocksAssert.withTable("CREATE TABLE `t0` (\n" +
+                "  `v1` bigint NULL COMMENT \"\",\n" +
+                "  `v2` bigint NULL COMMENT \"\",\n" +
+                "  `v3` bigint NULL\n" +
+                ") ENGINE=OLAP\n" +
+                "DUPLICATE KEY(`v1`, `v2`, v3)\n" +
+                "DISTRIBUTED BY HASH(`v1`) BUCKETS 3\n" +
+                "PROPERTIES (\n" +
+                "\"replication_num\" = \"1\",\n" +
+                "\"in_memory\" = \"false\"\n" +
+                ");");
+
+        starRocksAssert.withTable("CREATE TABLE `t1` (\n" +
+                "  `v4` bigint NULL COMMENT \"\",\n" +
+                "  `v5` bigint NULL COMMENT \"\",\n" +
+                "  `v6` bigint NULL\n" +
+                ") ENGINE=OLAP\n" +
+                "DUPLICATE KEY(`v4`, `v5`, v6)\n" +
+                "DISTRIBUTED BY HASH(`v4`) BUCKETS 3\n" +
+                "PROPERTIES (\n" +
+                "\"replication_num\" = \"1\",\n" +
+                "\"in_memory\" = \"false\"\n" +
+                ");");
+
+        starRocksAssert.withTable("CREATE TABLE `t2` (\n" +
+                "  `v7` bigint NULL COMMENT \"\",\n" +
+                "  `v8` bigint NULL COMMENT \"\",\n" +
+                "  `v9` bigint NULL\n" +
+                ") ENGINE=OLAP\n" +
+                "DUPLICATE KEY(`v7`, `v8`, v9)\n" +
+                "DISTRIBUTED BY HASH(`v7`) BUCKETS 3\n" +
+                "PROPERTIES (\n" +
+                "\"replication_num\" = \"1\",\n" +
+                "\"in_memory\" = \"false\"\n" +
+                ");");
+
+        starRocksAssert.withTable("CREATE TABLE `t3` (\n" +
+                "`day` int NULL COMMENT \"\") \n" +
+                "PROPERTIES ('replication_num' = '1')");
+
+        starRocksAssert.withTable("CREATE TABLE `tall` (\n" +
+                "  `ta` varchar(20) NULL COMMENT \"\",\n" +
+                "  `tb` smallint(6) NULL COMMENT \"\",\n" +
+                "  `tc` int(11) NULL COMMENT \"\",\n" +
+                "  `td` bigint(20) NULL COMMENT \"\",\n" +
+                "  `te` float NULL COMMENT \"\",\n" +
+                "  `tf` double NULL COMMENT \"\",\n" +
+                "  `tg` bigint(20) NULL COMMENT \"\",\n" +
+                "  `th` datetime NULL COMMENT \"\",\n" +
+                "  `ti` date NULL COMMENT \"\",\n" +
+                "  `tj` decimal(9, 3) NULL COMMENT \"\",\n" +
+                "  `tk` varbinary NULL COMMENT \"\"\n" +
+                ") ENGINE=OLAP\n" +
+                "DUPLICATE KEY(`ta`)\n" +
+                "COMMENT \"OLAP\"\n" +
+                "DISTRIBUTED BY HASH(`ta`) BUCKETS 3\n" +
+                "PROPERTIES (\n" +
+                "\"replication_num\" = \"1\",\n" +
+                "\"in_memory\" = \"false\"\n" +
+                ");");
+
+        starRocksAssert.withTable(TpchSQL.REGION);
+        starRocksAssert.withTable(TpchSQL.SUPPLIER);
+        starRocksAssert.withTable(TpchSQL.PARTSUPP);
+        starRocksAssert.withTable(TpchSQL.ORDERS);
+        starRocksAssert.withTable(TpchSQL.CUSTOMER);
+        starRocksAssert.withTable(TpchSQL.NATION);
+        starRocksAssert.withTable(TpchSQL.PART);
+        starRocksAssert.withTable(TpchSQL.LINEITEM);
+
+        starRocksAssert.withTable("create table test_array(c0 INT, " +
+                "c1 array<varchar(65533)>, " +
+                "c2 array<int>) " +
+                " duplicate key(c0) distributed by hash(c0) buckets 1 " +
+                "properties('replication_num'='1');");
+
+        FeConstants.runningUnitTest = true;
+        starRocksAssert.withTable("create table test_struct(c0 INT, " +
+                "c1 struct<a array<struct<b int>>>," +
+                "c2 struct<a int,b double>) " +
+                " duplicate key(c0) distributed by hash(c0) buckets 1 " +
+                "properties('replication_num'='1');");
+
+        starRocksAssert.withTable("create table test_map(c0 int, " +
+                "c1 map<int,int>, " +
+                "c2 array<map<int,int>>, " +
+                "c3 map<varchar(65533), date>) " +
+                "engine=olap distributed by hash(c0) buckets 10 " +
+                "properties('replication_num'='1');");
+
+        FeConstants.runningUnitTest = false;
+
+        connectContext.getSessionVariable().setSqlDialect("pinot");
+        connectContext.getSessionVariable().setCboPushDownGroupingSet(false);
+    }
+
+    protected void assertPlanContains(String sql, String... explain) throws Exception {
+        String explainString = getFragmentPlan(sql);
+        for (String expected : explain) {
+            Assert.assertTrue("expected is: " + expected + " but plan is \n" + explainString,
+                    StringUtils.containsIgnoreCase(explainString.toLowerCase(), expected));
+        }
+    }
+
+    public String getFragmentPlan(String sql) throws Exception {
+        return getPlanAndFragment(connectContext, sql).second.getExplainString(TExplainLevel.NORMAL);
+    }
+
+    public static Pair<String, ExecPlan> getPlanAndFragment(ConnectContext connectContext, String originStmt)
+            throws Exception {
+        connectContext.setDumpInfo(new QueryDumpInfo(connectContext));
+
+        List<StatementBase> statements =
+                com.starrocks.sql.parser.SqlParser.parse(originStmt, connectContext.getSessionVariable());
+        connectContext.getDumpInfo().setOriginStmt(originStmt);
+        StatementBase statementBase = statements.get(0);
+
+        ExecPlan execPlan = StatementPlanner.plan(statementBase, connectContext);
+        return new Pair<>(LogicalPlanPrinter.print(execPlan.getPhysicalPlan()), execPlan);
+    }
+}
\ No newline at end of file