fix bugs

Signed-off-by: shuming.li <[email protected]>

fe/conf/fe.conf

@@ -0,0 +1,90 @@
+# 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.
+
+#####################################################################
+## The uppercase properties are read and exported by bin/start_fe.sh.
+## To see all Frontend configurations,
+## see fe/src/com/starrocks/common/Config.java
+
+# the output dir of stderr/stdout/gc
+LOG_DIR = ${STARROCKS_HOME}/log
+
+DATE = "$(date +%Y%m%d-%H%M%S)"
+#JAVA_OPTS="-Dlog4j2.formatMsgNoLookups=true -Xmx8192m -XX:+UseMembar -XX:SurvivorRatio=8 -XX:MaxTenuringThreshold=7 -XX:+PrintGCDateStamps -XX:+PrintGCDetails -XX:+UseConcMarkSweepGC -XX:+UseParNewGC -XX:-CMSParallelRemarkEnabled -XX:CMSInitiatingOccupancyFraction=80 -XX:SoftRefLRUPolicyMSPerMB=0 -Xloggc:${LOG_DIR}/fe.gc.log.$DATE"
+
+#JAVA_OPTS="-Dlog4j2.formatMsgNoLookups=true -Xmx8192m -XX:+UseG1GC -Xlog:gc*:${LOG_DIR}/fe.gc.log.$DATE:time -XX:ErrorFile=${LOG_DIR}/hs_err_pid%p.log -Djava.security.policy=${STARROCKS_HOME}/conf/udf_security.policy"
+JAVA_OPTS="-Dlog4j2.formatMsgNoLookups=true -Xmx8192m -XX:+UseG1GC -XX:ErrorFile=${LOG_DIR}/hs_err_pid%p.log -Djava.security.policy=${STARROCKS_HOME}/conf/udf_security.policy"
+# For jdk 9+, this JAVA_OPTS will be used as default JVM options
+JAVA_OPTS_FOR_JDK_9="-Dlog4j2.formatMsgNoLookups=true -Xmx8192m -XX:SurvivorRatio=8 -XX:MaxTenuringThreshold=7 -Xlog:gc*:${LOG_DIR}/fe.gc.log.$DATE:time"
+#JAVA_OPTS_FOR_JDK_9="-Dlog4j2.formatMsgNoLookups=true -Xmx8192m -XX:SurvivorRatio=8 -XX:MaxTenuringThreshold=7 -XX:-CMSParallelRemarkEnabled -XX:CMSInitiatingOccupancyFraction=80 -XX:SoftRefLRUPolicyMSPerMB=0 -Xlog:gc*:${LOG_DIR}/fe.gc.log.$DATE:time"
+
+##
+## the lowercase properties are read by main program.
+##
+
+# DEBUG, INFO, WARN, ERROR, FATAL
+#sys_log_level = DEBUG
+sys_log_level = INFO
+#sys_log_level = DEBUG
+
+# store metadata, create it if it is not exist.
+# Default value is ${STARROCKS_HOME}/meta
+#meta_dir = ${STARROCKS_HOME}/meta
+
+http_port = 8131
+rpc_port = 9121
+query_port = 9131
+edit_log_port = 9112
+mysql_service_nio_enabled = true
+
+# Enable jaeger tracing by setting jaeger_grpc_endpoint
+# jaeger_grpc_endpoint = http://localhost:14250
+
+# Choose one if there are more than one ip except loopback address. 
+# Note that there should at most one ip match this list.
+# If no ip match this rule, will choose one randomly.
+# use CIDR format, e.g. 10.10.10.0/24
+# Default value is empty.
+# priority_networks = 10.10.10.0/24;192.168.0.0/16
+
+# Advanced configurations 
+# log_roll_size_mb = 1024
+# sys_log_dir = ${STARROCKS_HOME}/log
+# sys_log_roll_num = 10
+# sys_log_verbose_modules = *
+# audit_log_dir = ${STARROCKS_HOME}/log
+# audit_log_modules = slow_query, query
+# audit_log_roll_num = 10
+# meta_delay_toleration_second = 10
+# qe_max_connection = 1024
+# max_conn_per_user = 100
+# qe_query_timeout_second = 300
+# qe_slow_log_ms = 5000
+#metadata_failure_recovery = true
+#enable_strict_storage_medium_check=false
+
+#enable_replicas_on_the_same_host=true
+#enable_experimental_mv=true
+#enable_incremental_mv=true
+enable_statistic_collect=false
+#enable_optimizer_trace_log=true
+enable_incremental_mv=true
+default_replication_num = 1
+max_routine_load_task_num_per_be = 20
+
+priority_networks = 172.26.92.0/24
+tablet_stat_update_interval_second=180

fe/conf/hadoop_env.sh

@@ -0,0 +1,15 @@
+# This file is licensed under the Elastic License 2.0. Copyright 2021-present, StarRocks Limited.
+
+export HADOOP_CLASSPATH=${STARROCKS_HOME}/lib/hadoop/common/*:${STARROCKS_HOME}/lib/hadoop/common/lib/*:${STARROCKS_HOME}/lib/hadoop/hdfs/*:${STARROCKS_HOME}/lib/hadoop/hdfs/lib/*
+export HADOOP_USER_NAME=${USER}
+
+# the purpose is to use local hadoop configuration first.
+# under HADOOP_CONF_DIR(eg. /etc/ecm/hadoop-conf), there are hadoop/hdfs/hbase conf files.
+# and by putting HADOOP_CONF_DIR at front of HADOOP_CLASSPATH, local hadoop conf file will be searched & used first.
+
+# local hadoop configuration is usually well-tailored and optimized, we'd better to leverage that.
+# for example, if local hdfs has enabled short-circuit read, then we can use short-circuit read and save io time
+
+if [ ${HADOOP_CONF_DIR}"X" != "X" ]; then
+    export HADOOP_CLASSPATH=${HADOOP_CONF_DIR}:${HADOOP_CLASSPATH}
+fi

fe/fe-core/src/main/java/com/starrocks/alter/RollupJobV2.java

@@ -456,7 +456,8 @@ private Expr analyzeExpr(SelectAnalyzer.RewriteAliasVisitor visitor,
         Expr newExpr = defineExpr.clone(smap);
         newExpr = newExpr.accept(visitor, null);
         newExpr = Expr.analyzeAndCastFold(newExpr);
-        if (!newExpr.getType().equals(type)) {
+        Type newType = newExpr.getType();
+        if (!type.isFullyCompatible(newType)) {
             newExpr = new CastExpr(type, newExpr);
         }
         return newExpr;

fe/fe-core/src/main/java/com/starrocks/catalog/ArrayType.java

@@ -88,6 +88,16 @@ public void toThrift(TTypeDesc container) {
         itemType.toThrift(container);
     }
 
+    @Override
+    public boolean isWildcardDecimal() {
+        if (!(itemType instanceof ScalarType)) {
+            return false;
+        }
+        ScalarType scalarType = (ScalarType) itemType;
+        return (itemType.isDecimalV2() || itemType.isDecimalV3())
+                && (scalarType.getPrecision() == -1 && scalarType.getScalarPrecision() == -1);
+    }
+
     @Override
     public boolean isFullyCompatible(Type other) {
         if (!other.isArrayType()) {

fe/fe-core/src/main/java/com/starrocks/catalog/combinator/AggStateCombinator.java

@@ -59,7 +59,6 @@ public static Optional<AggStateCombinator> of(AggregateFunction aggFunc) {
 
             AggStateDesc aggStateDesc = new AggStateDesc(aggFunc);
             aggStateFunc.setAggStateDesc(aggStateDesc);
-
             // `agg_state` function's type will contain agg state desc.
             intermediateType.setAggStateDesc(aggStateDesc);
             LOG.info("Register agg state function: {}", aggStateFunc.functionName());

fe/fe-core/src/main/java/com/starrocks/catalog/combinator/AggStateMergeCombinator.java

@@ -56,7 +56,13 @@ public static Optional<AggStateMergeCombinator> of(AggregateFunction aggFunc) {
                     new AggStateMergeCombinator(functionName, imtermediateType, aggFunc.getReturnType());
             aggStateMergeFunc.setBinaryType(TFunctionBinaryType.BUILTIN);
             aggStateMergeFunc.setPolymorphic(aggFunc.isPolymorphic());
-            aggStateMergeFunc.setAggStateDesc(new AggStateDesc(aggFunc));
+            AggStateDesc aggStateDesc;
+            if (aggFunc.getAggStateDesc() != null) {
+                aggStateDesc = aggFunc.getAggStateDesc().clone();
+            } else {
+                aggStateDesc = new AggStateDesc(aggFunc);
+            }
+            aggStateMergeFunc.setAggStateDesc(aggStateDesc);
             LOG.info("Register agg state function: {}", aggStateMergeFunc.functionName());
             return Optional.of(aggStateMergeFunc);
         } catch (Exception e) {

fe/fe-core/src/main/java/com/starrocks/catalog/combinator/AggStateUnionCombinator.java

@@ -56,7 +56,13 @@ public static Optional<AggStateUnionCombinator> of(AggregateFunction aggFunc) {
                     new AggStateUnionCombinator(functionName, intermediateType);
             aggStateUnionFunc.setBinaryType(TFunctionBinaryType.BUILTIN);
             aggStateUnionFunc.setPolymorphic(aggFunc.isPolymorphic());
-            aggStateUnionFunc.setAggStateDesc(new AggStateDesc(aggFunc));
+            AggStateDesc aggStateDesc;
+            if (aggFunc.getAggStateDesc() != null) {
+                aggStateDesc = aggFunc.getAggStateDesc().clone();
+            } else {
+                aggStateDesc = new AggStateDesc(aggFunc);
+            }
+            aggStateUnionFunc.setAggStateDesc(aggStateDesc);
             LOG.info("Register agg state function: {}", aggStateUnionFunc.functionName());
             return Optional.of(aggStateUnionFunc);
         } catch (Exception e) {

fe/fe-core/src/main/java/com/starrocks/catalog/combinator/AggStateUtils.java

@@ -157,7 +157,7 @@ public static Function getAnalyzedCombinatorFunction(ConnectContext session,
                 return null;
             }
             AggregateFunction aggFunc = (AggregateFunction) argFn;
-            if (aggFunc.getNumArgs() == 1 && argumentTypes[0].isDecimalOfAnyVersion()) {
+            if (aggFunc.getNumArgs() == 1) {
                 // only copy argument if it's a decimal type
                 AggregateFunction argFnCopy = (AggregateFunction) aggFunc.copy();
                 argFnCopy.setArgsType(argumentTypes);
@@ -209,7 +209,9 @@ private static AggregateFunction getAggStateFunction(ConnectContext session,
         if (!(fn instanceof AggregateFunction)) {
             return null;
         }
-        return (AggregateFunction) fn;
+        AggregateFunction result = (AggregateFunction) fn.copy();
+        result.setAggStateDesc(aggStateDesc);
+        return result;
     }
 
     private static Type[] getNewArgumentTypes(Type[] origArgTypes, String argFnName, Type arg0Type) {

fe/fe-core/src/main/java/com/starrocks/sql/ast/CreateMaterializedViewStmt.java

@@ -49,7 +49,6 @@
 import com.starrocks.analysis.OrderByElement;
 import com.starrocks.analysis.SlotRef;
 import com.starrocks.analysis.TableName;
-import com.starrocks.catalog.AggregateFunction;
 import com.starrocks.catalog.AggregateType;
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Function;
@@ -547,38 +546,47 @@ private MVColumnItem buildAggColumnItem(ConnectContext context,
         FunctionCallExpr node = (FunctionCallExpr) selectListItem.getExpr();
         String functionName = node.getFnName().getFunction();
         Preconditions.checkState(node.getChildren().size() == 1, "Aggregate function only support one child");
+
         if (!FN_NAME_TO_PATTERN.containsKey(functionName)) {
+            if (Strings.isNullOrEmpty(selectListItem.getAlias())) {
+                throw new SemanticException("Create materialized view non-slot ref expression should have an alias:" +
+                        selectListItem.getExpr());
+            }
+
             Expr defineExpr = node.getChild(0);
             List<Type> argTypes = node.getChildren().stream().map(Expr::getType).collect(Collectors.toList());
-
+            Type arg0Type = argTypes.get(0);
+            if (arg0Type.getAggStateDesc() == null) {
+                throw new UnsupportedMVException("Unsupported function:" + functionName + ", cannot find agg state desc from " +
+                        "arg0");
+            }
             FunctionParams params = new FunctionParams(false, Lists.newArrayList());
             Type[] argumentTypes = argTypes.toArray(Type[]::new);
             Boolean[] isArgumentConstants = argTypes.stream().map(x -> false).toArray(Boolean[]::new);
-            Function function = FunctionAnalyzer.getAnalyzedAggregateFunction(ConnectContext.get(), functionName,
+            Function function = FunctionAnalyzer.getAnalyzedAggregateFunction(context, functionName,
                     params, argumentTypes, isArgumentConstants, NodePosition.ZERO);
-            if (Strings.isNullOrEmpty(selectListItem.getAlias())) {
-                throw new SemanticException("Create materialized view non-slot ref expression should have an alias:" +
-                        selectListItem.getExpr());
-            }
-            String mvColumnName = MVUtils.getMVColumnName(selectListItem.getAlias());
-            if (function == null) {
+            if (function == null || !(function instanceof AggStateUnionCombinator)) {
                 throw new UnsupportedMVException("Unsupported function:" + functionName);
             }
-            if (!(function instanceof AggregateFunction)) {
-                throw new UnsupportedMVException("Unsupported function:" + functionName + " which is not an aggregate function");
+            AggStateUnionCombinator aggFunction = (AggStateUnionCombinator) function;
+            String mvColumnName = MVUtils.getMVColumnName(selectListItem.getAlias());
+            AggStateDesc aggStateDesc = aggFunction.getAggStateDesc();
+            Type type = aggFunction.getIntermediateTypeOrReturnType();
+            Type finalType = AnalyzerUtils.transformTableColumnType(type, false);
+            if (type.isWildcardDecimal()) {
+                throw new UnsupportedMVException("Unsupported wildcard decimal type in materialized view:" + type + ", " +
+                        "function:" + node);
             }
-            if (!(function instanceof AggStateUnionCombinator)) {
-                throw new UnsupportedMVException("Unsupported function:" + functionName + " which is not an " +
-                        "agg_state_union aggregate function");
+            if (aggStateDesc.getArgTypes().stream().anyMatch(t -> t.isWildcardDecimal())) {
+                throw new UnsupportedMVException("Unsupported wildcard decimal type in materialized view:" + type + ", " +
+                        "function:" + node);
             }
-            AggStateUnionCombinator aggStateUnion = (AggStateUnionCombinator) function;
-            AggStateDesc aggStateDesc = aggStateUnion.getAggStateDesc();
-            Type type = function.getReturnType();
             Set<String> baseColumnNames = baseSlotRefs.stream().map(slot -> slot.getColumnName())
                     .collect(Collectors.toSet());
             AggregateType mvAggregateType = AggregateType.AGG_STATE_UNION;
-            return new MVColumnItem(mvColumnName, type, mvAggregateType, aggStateDesc, false,
-                    defineExpr, node.isNullable(), baseColumnNames);
+
+            return new MVColumnItem(mvColumnName, finalType, mvAggregateType, aggStateDesc, false,
+                    defineExpr, aggStateDesc.getResultNullable(), baseColumnNames);
         } else {
             return buildAggColumnItemWithPattern(selectListItem, baseSlotRefs);
         }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/AggregatedMaterializedViewRewriter.java

@@ -659,13 +659,13 @@ private Map<ColumnRefOperator, CallOperator> rewriteAggregates(Map<ColumnRefOper
             // Aggregate must be CallOperator
             Pair<CallOperator, EquivalentShuttleContext> rewritten = getRollupAggregate(context, equationRewriter,
                     queryColumnSet, aggCall);
-            CallOperator newAggregate = rewritten.first;
-            if (newAggregate == null) {
+            if (rewritten == null || rewritten.first == null) {
                 OptimizerTraceUtil.logMVRewriteFailReason(mvRewriteContext.getMVName(),
                         "Rewrite aggregate function failed, cannot get rollup function: {}",
                         aggCall.toString());
                 return null;
             }
+            CallOperator newAggregate = rewritten.first;
             EquivalentShuttleContext eqContext = rewritten.second;
 
             ColumnRefOperator origColRef = entry.getKey();