[BugFix] Fix full outer join rewrite and nested mv refresh bug for main(

#34071) (#34299)

For rewrite:
1. fix join derivibility rewrite bug for multi joins query
2. fix equivalence class construction bugs to exclude outer join on predicate, which should be rewritten by ReplaceColumnRefRewriter
3. optimize predicate normalization to process column name's case
4. set mv plan timeout to new_planner_optimize_timeout

For refresh:
1. fix refresh on nested mv bug to process date format like %Y%m%d
2. set strict mode to false for mv refresh

Signed-off-by: ABingHuang <codekhuang@163.com>

fe/fe-core/src/main/java/com/starrocks/scheduler/PartitionBasedMvRefreshProcessor.java

@@ -26,6 +26,7 @@
 import com.starrocks.analysis.IsNullPredicate;
 import com.starrocks.analysis.LiteralExpr;
 import com.starrocks.analysis.SlotRef;
+import com.starrocks.analysis.StringLiteral;
 import com.starrocks.catalog.BaseTableInfo;
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.DataProperty;
@@ -1200,6 +1201,28 @@ private Expr generatePartitionPredicate(Set<String> tablePartitionNames, QuerySt
                 sourceTablePartitionRange.add(refBaseTableRangePartitionMap.get(partitionName));
             }
             sourceTablePartitionRange = MvUtils.mergeRanges(sourceTablePartitionRange);
+            // for nested mv, the base table may be another mv, which is partition by str2date(dt, '%Y%m%d')
+            // here we should convert date into '%Y%m%d' format
+            Expr partitionExpr = materializedView.getFirstPartitionRefTableExpr();
+            Pair<Table, Column> partitionTableAndColumn = materializedView.getBaseTableAndPartitionColumn();
+            boolean isConvertToDate = PartitionUtil.isConvertToDate(partitionExpr, partitionTableAndColumn.second);
+            if (isConvertToDate && partitionExpr instanceof FunctionCallExpr
+                    && !sourceTablePartitionRange.isEmpty() && MvUtils.isDateRange(sourceTablePartitionRange.get(0))) {
+                Optional<FunctionCallExpr> functionCallExprOpt = getStr2DateExpr(partitionExpr);
+                if (!functionCallExprOpt.isPresent()) {
+                    LOG.warn("invalid partition expr:{}", partitionExpr);
+                    return null;
+                }
+                FunctionCallExpr functionCallExpr = functionCallExprOpt.get();
+                Preconditions.checkState(functionCallExpr.getFnName().getFunction().equalsIgnoreCase(FunctionSet.STR2DATE));
+                String dateFormat = ((StringLiteral) functionCallExpr.getChild(1)).getStringValue();
+                List<Range<PartitionKey>> converted = Lists.newArrayList();
+                for (Range<PartitionKey> range : sourceTablePartitionRange) {
+                    Range<PartitionKey> varcharPartitionKey = MvUtils.convertToVarcharRange(range, dateFormat);
+                    converted.add(varcharPartitionKey);
+                }
+                sourceTablePartitionRange = converted;
+            }
             List<Expr> partitionPredicates =
                     MvUtils.convertRange(outputPartitionSlot, sourceTablePartitionRange);
             // range contains the min value could be null value
@@ -1231,6 +1254,20 @@ private Expr generatePartitionPredicate(Set<String> tablePartitionNames, QuerySt
         }
     }
 
+    private Optional<FunctionCallExpr> getStr2DateExpr(Expr partitionExpr) {
+        List<Expr> matches = Lists.newArrayList();
+        partitionExpr.collect(expr -> isStr2Date(expr), matches);
+        if (matches.size() != 1) {
+            return Optional.empty();
+        }
+        return  Optional.of(matches.get(0).cast());
+    }
+
+    private boolean isStr2Date(Expr expr) {
+        return expr instanceof FunctionCallExpr
+                && ((FunctionCallExpr) expr).getFnName().getFunction().equalsIgnoreCase(FunctionSet.STR2DATE);
+    }
+
     private boolean checkBaseTablePartitionChange() {
         // check snapshotBaseTables and current tables in catalog
         for (Pair<BaseTableInfo, Table> tablePair : snapshotBaseTables.values()) {

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/MvPlanContextBuilder.java

@@ -22,6 +22,9 @@ public class MvPlanContextBuilder {
     public MvPlanContext getPlanContext(MaterializedView mv) {
         // build mv query logical plan
         MaterializedViewOptimizer mvOptimizer = new MaterializedViewOptimizer();
-        return mvOptimizer.optimize(mv, new ConnectContext());
+        ConnectContext connectContext = new ConnectContext();
+        connectContext.getSessionVariable().setOptimizerExecuteTimeout(
+                ConnectContext.get().getSessionVariable().getOptimizerExecuteTimeout());
+        return mvOptimizer.optimize(mv, connectContext);
     }
 }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rewrite/scalar/MvNormalizePredicateRule.java

@@ -47,7 +47,7 @@ public int compare(ScalarOperator o1, ScalarOperator o2) {
             } else if (o2 == null) {
                 return 1;
             } else {
-                return o1.toString().compareTo(o2.toString());
+                return o1.toString().toLowerCase().compareTo(o2.toString().toLowerCase());
             }
         }
     };
@@ -65,14 +65,14 @@ public int compare(ScalarOperator o1, ScalarOperator o2) {
                     if (o1.isColumnRef() && o2.isColumnRef()) {
                         ColumnRefOperator c1 = (ColumnRefOperator) o1;
                         ColumnRefOperator c2 = (ColumnRefOperator) o2;
-                        int ret = c1.getName().compareTo(c2.getName());
+                        int ret = c1.getName().toLowerCase().compareTo(c2.getName().toLowerCase());
                         if (ret != 0) {
                             return ret;
                         }
                         return Integer.compare(c1.getId(), c2.getId());
                     } else {
-                        String s1 = o1.toString();
-                        String s2 = o2.toString();
+                        String s1 = o1.toString().toLowerCase();
+                        String s2 = o2.toString().toLowerCase();
                         String n1 = s1.replaceAll("\\d+: ", "");
                         String n2 = s2.replaceAll("\\d+: ", "");
                         int ret = n1.compareTo(n2);

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/mv/JoinDeriveContext.java

@@ -25,18 +25,22 @@ public class JoinDeriveContext {
     private final JoinOperator mvJoinType;
     // join columns for left and right join tables
     private final List<List<ColumnRefOperator>> joinColumns;
+    // join columns for left and right join tables
+    private final List<List<ColumnRefOperator>> childOutputColumns;
 
     private final List<Pair<ColumnRefOperator, ColumnRefOperator>> compensatedEquivalenceColumns;
 
     public JoinDeriveContext(
             JoinOperator queryJoinType,
             JoinOperator mvJoinType,
             List<List<ColumnRefOperator>> joinColumns,
-            List<Pair<ColumnRefOperator, ColumnRefOperator>> compensatedEquivalenceColumns) {
+            List<Pair<ColumnRefOperator, ColumnRefOperator>> compensatedEquivalenceColumns,
+            List<List<ColumnRefOperator>> childOutputColumns) {
         this.queryJoinType = queryJoinType;
         this.mvJoinType = mvJoinType;
         this.joinColumns = joinColumns;
         this.compensatedEquivalenceColumns = compensatedEquivalenceColumns;
+        this.childOutputColumns = childOutputColumns;
     }
 
     public JoinOperator getQueryJoinType() {
@@ -58,4 +62,12 @@ public List<ColumnRefOperator> getRightJoinColumns() {
     public List<Pair<ColumnRefOperator, ColumnRefOperator>> getCompensatedEquivalenceColumns() {
         return compensatedEquivalenceColumns;
     }
+
+    public List<ColumnRefOperator> getLeftChildOutputColumns() {
+        return childOutputColumns.get(0);
+    }
+
+    public List<ColumnRefOperator> getRightChildOutputColumns() {
+        return childOutputColumns.get(1);
+    }
 }