[Enhancement](udf) support volatility for scalar UDFs

Author: linrrzqqq

fe/fe-catalog/src/main/java/org/apache/doris/catalog/Function.java

@@ -114,6 +114,8 @@ public enum BinaryType {
     protected String runtimeVersion;
     @SerializedName("fc")
     protected String functionCode;
+    @SerializedName("vol")
+    protected FunctionVolatility volatility = FunctionVolatility.IMMUTABLE;
 
     // Only used for serialization
     protected Function() {
@@ -174,6 +176,7 @@ public Function(Function other) {
         this.expirationTime = other.expirationTime;
         this.runtimeVersion = other.runtimeVersion;
         this.functionCode = other.functionCode;
+        this.volatility = other.getVolatility();
     }
 
     public Function clone() {
@@ -301,6 +304,14 @@ public void setFunctionCode(String functionCode) {
         this.functionCode = functionCode;
     }
 
+    public FunctionVolatility getVolatility() {
+        return volatility == null ? FunctionVolatility.IMMUTABLE : volatility;
+    }
+
+    public void setVolatility(FunctionVolatility volatility) {
+        this.volatility = volatility == null ? FunctionVolatility.IMMUTABLE : volatility;
+    }
+
     // TODO(cmy): Currently we judge whether it is UDF by wheter the 'location' is set.
     // Maybe we should use a separate variable to identify,
     // but additional variables need to modify the persistence information.

fe/fe-catalog/src/main/java/org/apache/doris/catalog/FunctionVolatility.java

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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
+//
+//   http://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 org.apache.doris.catalog;
+
+import java.util.Locale;
+
+/** Function volatility controls which optimizer rewrites are safe for a function call. */
+public enum FunctionVolatility {
+    IMMUTABLE,
+    STABLE,
+    VOLATILE;
+
+    public static FunctionVolatility fromString(String value) {
+        if (value == null) {
+            return IMMUTABLE;
+        }
+        try {
+            return FunctionVolatility.valueOf(value.trim().toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException("Invalid volatility: '" + value
+                    + "'. Expected one of: immutable, stable, volatile", e);
+        }
+    }
+
+    public String toSql() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionToSqlConverter.java

@@ -75,6 +75,13 @@ public static String toSql(ScalarFunction fn, boolean ifNotExists) {
                     .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\"");
             boolean isReturnNull = fn.getNullableMode() == NullableMode.ALWAYS_NULLABLE;
             sb.append(",\n  \"ALWAYS_NULLABLE\"=").append("\"" + isReturnNull + "\"");
+            sb.append(",\n  \"VOLATILITY\"=").append("\"" + fn.getVolatility().toSql() + "\"");
+        } else if (fn.getBinaryType() == Function.BinaryType.PYTHON_UDF) {
+            sb.append(",\n  \"FILE\"=")
+                    .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\"");
+            boolean isReturnNull = fn.getNullableMode() == NullableMode.ALWAYS_NULLABLE;
+            sb.append(",\n  \"ALWAYS_NULLABLE\"=").append("\"" + isReturnNull + "\"");
+            sb.append(",\n  \"VOLATILITY\"=").append("\"" + fn.getVolatility().toSql() + "\"");
         } else {
             sb.append(",\n  \"OBJECT_FILE\"=")
                     .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\"");
@@ -125,6 +132,11 @@ public static String toSql(AggregateFunction fn, boolean ifNotExists) {
                     .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\",");
             boolean isReturnNull = fn.getNullableMode() == NullableMode.ALWAYS_NULLABLE;
             sb.append("\n  \"ALWAYS_NULLABLE\"=").append("\"" + isReturnNull + "\",");
+        } else if (fn.getBinaryType() == Function.BinaryType.PYTHON_UDF) {
+            sb.append("\n  \"FILE\"=")
+                    .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\",");
+            boolean isReturnNull = fn.getNullableMode() == NullableMode.ALWAYS_NULLABLE;
+            sb.append("\n  \"ALWAYS_NULLABLE\"=").append("\"" + isReturnNull + "\",");
         } else {
             sb.append("\n  \"OBJECT_FILE\"=")
                     .append("\"" + (fn.getLocation() == null ? "" : fn.getLocation().toString()) + "\",");

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PushDownFilterThroughProject.java

@@ -45,7 +45,7 @@ public Plan visitPhysicalFilter(PhysicalFilter<? extends Plan> filter, CascadesC
         PhysicalProject<? extends Plan> project = (PhysicalProject<? extends Plan>) child;
         Map<Slot, Expression> childAlias = project.getAliasToProducer();
         if (filter.getInputSlots().stream().map(childAlias::get).filter(Objects::nonNull)
-                .anyMatch(Expression::containsUniqueFunction)) {
+                .anyMatch(Expression::containsVolatileExpression)) {
             return filter;
         }
         PhysicalFilter<? extends Plan> newFilter = filter.withConjunctsAndChild(

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java

@@ -1419,7 +1419,7 @@ private List<Expression> bindGroupByUniqueId(List<Expression> groupByExpressions
         // 2. for 'group by a + random(), a + random() + 1', the two 'random()' will be different.
         int containsUniqueGroupByCount = 0;
         for (Expression groupByExpr : groupByExpressions) {
-            if (groupByExpr.containsUniqueFunction()) {
+            if (groupByExpr.containsVolatileExpression()) {
                 containsUniqueGroupByCount++;
             }
         }
@@ -1433,7 +1433,7 @@ private List<Expression> bindGroupByUniqueId(List<Expression> groupByExpressions
                 groupByExpressions.size());
         for (Expression groupByExpr : groupByExpressions) {
             Expression newGroupByExpr = groupByExpr;
-            if (groupByExpr.containsUniqueFunction()) {
+            if (groupByExpr.containsVolatileExpression()) {
                 Expression ignoreUniqueIdExpr = ExpressionUtils.setIgnoreUniqueIdForUniqueFunc(groupByExpr, true);
                 Expression previousGroupByExpr = ignoreUniqueIdGroupByExprs.get(ignoreUniqueIdExpr);
                 if (previousGroupByExpr == null) {
@@ -1476,7 +1476,7 @@ private <T extends Expression> List<T> bindExprsUniqueIdWithGroupBy(List<T> expr
     //    c) let E3 = rewrite E2 with enable unique ids. then E3 is the bind unique id expression for E.
     private <T extends Expression> T bindExprUniqueIdWithGroupBy(T expression,
             Map<Expression, Expression> bindUniqueIdReplaceMap) {
-        if (!expression.containsUniqueFunction() || bindUniqueIdReplaceMap.isEmpty()) {
+        if (!expression.containsVolatileExpression() || bindUniqueIdReplaceMap.isEmpty()) {
             return expression;
         }
 
@@ -1522,7 +1522,7 @@ private Map<Expression, Expression> getBelowAggregateGroupByUniqueFuncReplaceMap
     private Map<Expression, Expression> getGroupByUniqueFuncReplaceMap(List<Expression> groupByByExpressions) {
         Map<Expression, Expression> replaceMap = Maps.newHashMap();
         for (Expression expression : groupByByExpressions) {
-            if (expression.containsUniqueFunction()) {
+            if (expression.containsVolatileExpression()) {
                 Expression ignoreUniqueIdExpr = ExpressionUtils.setIgnoreUniqueIdForUniqueFunc(expression, true);
                 // for sql:
                 //    select distinct a + random(),  a + random()
@@ -1554,7 +1554,7 @@ private Plan bindRepeat(MatchingContext<LogicalRepeat<Plan>> ctx) {
                     = ImmutableList.builderWithExpectedSize(boundGroupingSet.size());
             for (Expression groupBy : boundGroupingSet) {
                 Expression newGroupBy = groupBy;
-                if (groupBy.containsUniqueFunction()) {
+                if (groupBy.containsVolatileExpression()) {
                     Expression ignoreUniqueIdGroupBy = ExpressionUtils.setIgnoreUniqueIdForUniqueFunc(groupBy, true);
                     Expression previousGroupBy = ignoreUniqueIdGroupByExpressions.get(ignoreUniqueIdGroupBy);
                     if (previousGroupBy == null) {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/InPredicateExtractNonConstant.java

@@ -50,7 +50,7 @@ public List<ExpressionPatternMatcher<? extends Expression>> buildRules() {
                 matchesType(InPredicate.class)
                         .when(inPredicate ->
                                 inPredicate.getOptions().size() <= InPredicateDedup.REWRITE_OPTIONS_MAX_SIZE
-                                && !inPredicate.getCompareExpr().containsUniqueFunction())
+                                && !inPredicate.getCompareExpr().containsVolatileExpression())
                         .then(this::rewrite)
                         .toRule(ExpressionRuleType.IN_PREDICATE_EXTRACT_NON_CONSTANT)
         );

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/PushIntoCaseWhenBranch.java

@@ -166,7 +166,7 @@ private Optional<Expression> tryPushIntoNvl(Expression parent, int childIndex, N
         // so there will exist twice 'first' in the rewritten IF expression, which may increase the computation cost.
         // if the plan is not filter and not join, then push down action may not have positive effect,
         // considering this, we give up the rewrite if the plan is not condition plan or first contains unique function.
-        if (first.containsUniqueFunction() || !isConditionPlan) {
+        if (first.containsVolatileExpression() || !isConditionPlan) {
             return Optional.empty();
         }
         If ifExpr = new If(new IsNull(first), second, first);
@@ -182,7 +182,7 @@ private Optional<Expression> tryPushIntoNullIf(Expression parent, int childIndex
         // so there will exist twice 'first' in the rewritten IF expression, which may increase the computation cost.
         // if the plan is not filter and not join, then push down action may not have positive effect,
         // considering this, we give up the rewrite if the plan is not condition plan or first contains unique function.
-        if (first.containsUniqueFunction() || !isConditionPlan) {
+        if (first.containsVolatileExpression() || !isConditionPlan) {
             return Optional.empty();
         }
         If ifExpr = new If(new EqualTo(first, second), new NullLiteral(nullIf.getDataType()), first);

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AddProjectForUniqueFunction.java

@@ -26,8 +26,8 @@
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.expressions.VolatileExpression;
 import org.apache.doris.nereids.trees.expressions.functions.Function;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.UniqueFunction;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
@@ -274,22 +274,24 @@ public <T extends Expression> Optional<Pair<List<T>, LogicalProject<Plan>>> rewr
      */
     @VisibleForTesting
     public List<NamedExpression> tryGenUniqueFunctionAlias(Collection<? extends Expression> targets) {
-        Map<UniqueFunction, Integer> unqiueFunctionCounter = Maps.newLinkedHashMap();
+        Map<Expression, Integer> unqiueFunctionCounter = Maps.newLinkedHashMap();
         for (Expression target : targets) {
             target.foreach(e -> {
                 Expression expr = (Expression) e;
-                if (expr instanceof UniqueFunction) {
-                    unqiueFunctionCounter.merge((UniqueFunction) expr, 1, Integer::sum);
+                if (expr instanceof VolatileExpression && ((VolatileExpression) expr).isVolatile()) {
+                    unqiueFunctionCounter.merge(expr, 1, Integer::sum);
                 }
             });
         }
 
         ImmutableList.Builder<NamedExpression> builder
                 = ImmutableList.builderWithExpectedSize(unqiueFunctionCounter.size());
-        for (Entry<UniqueFunction, Integer> entry : unqiueFunctionCounter.entrySet()) {
+        for (Entry<Expression, Integer> entry : unqiueFunctionCounter.entrySet()) {
             if (entry.getValue() > 1) {
                 ExprId exprId = StatementScopeIdGenerator.newExprId();
-                String name = "$_" + entry.getKey().getName() + "_" + exprId.asInt() + "_$";
+                String functionName = entry.getKey() instanceof Function
+                        ? ((Function) entry.getKey()).getName() : "volatile";
+                String name = "$_" + functionName + "_" + exprId.asInt() + "_$";
                 builder.add(new Alias(exprId, entry.getKey(), name));
             }
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CollectFilterAboveConsumer.java

@@ -38,7 +38,7 @@ public Rule build() {
             LogicalCTEConsumer cteConsumer = filter.child();
             Set<Expression> exprs = filter.getConjuncts();
             for (Expression expr : exprs) {
-                if (expr.containsUniqueFunction()) {
+                if (expr.containsVolatileExpression()) {
                     continue;
                 }
                 Expression rewrittenExpr = expr.rewriteUp(e -> {

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/JoinExtractOrFromCaseWhen.java

@@ -108,7 +108,7 @@ private boolean needRewrite(LogicalJoin<Plan, Plan> join) {
 
     // 1. expr contains slots from both sides;
     private boolean isConditionNeedRewrite(Expression expr, Set<Slot> leftSlots, Set<Slot> rightSlots) {
-        if (expr.containsUniqueFunction()) {
+        if (expr.containsVolatileExpression()) {
             return false;
         }
         Set<Slot> exprSlots = expr.getInputSlots();