Fix full outer join rewrite and nested mv refresh bug

For rewrite:
1. fix join derivibility rewrite bug for multi joins query
2. fix ec 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>
Signed-off-by: Youngwb <yangwenbo_mailbox@163.com>
(cherry picked from commit 1db05ef)

fix format

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

remove comments

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

fix ut

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

fix case bug

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

add ut

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

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

@@ -27,6 +27,7 @@
 import com.starrocks.analysis.FunctionCallExpr;
 import com.starrocks.analysis.IsNullPredicate;
 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;
@@ -1101,6 +1102,23 @@ 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))) {
+                FunctionCallExpr functionCallExpr = (FunctionCallExpr) partitionExpr;
+                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
@@ -1272,6 +1290,7 @@ public void refreshMaterializedView(MvTaskRunContext mvContext, ExecPlan execPla
         }
         ctx.setStmtId(new AtomicInteger().incrementAndGet());
         ctx.setExecutionId(UUIDUtil.toTUniqueId(ctx.getQueryId()));
+        ctx.getSessionVariable().setEnableInsertStrict(false);
         try {
             executor.handleDMLStmtWithProfile(execPlan, insertStmt, beginTimeInNanoSecond);
         } catch (Exception e) {

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

@@ -34,6 +34,9 @@ public MvPlanContext getPlanContext(MaterializedView mv) {
             optimizerConfig.disableRule(RuleType.TF_MATERIALIZED_VIEW);
         }
         optimizerConfig.setMVRewritePlan(true);
-        return mvOptimizer.optimize(mv, new ConnectContext(), optimizerConfig);
+        ConnectContext connectContext = new ConnectContext();
+        connectContext.getSessionVariable().setOptimizerExecuteTimeout(
+                ConnectContext.get().getSessionVariable().getOptimizerExecuteTimeout());
+        return mvOptimizer.optimize(mv, connectContext, optimizerConfig);
     }
 }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/scalar/ColumnRefOperator.java

@@ -172,7 +172,7 @@ public boolean equivalent(Object obj) {
 
         ColumnRefOperator leftColumn = (ColumnRefOperator) this;
         ColumnRefOperator rightColumn = (ColumnRefOperator) obj;
-        return leftColumn.getName().equals(rightColumn.getName())
+        return leftColumn.getName().equalsIgnoreCase(rightColumn.getName())
                 && leftColumn.getType().equals(rightColumn.getType())
                 && leftColumn.isNullable() == rightColumn.isNullable();
     }

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,11 +65,15 @@ public int compare(ScalarOperator o1, ScalarOperator o2) {
                     } else if (o2 == null) {
                         return 1;
                     } 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);
+                        int ret = Integer.compare(n1.length(), n2.length());
+                        if (ret != 0) {
+                            return ret;
+                        }
+                        ret = n1.compareTo(n2);
                         return (ret == 0) ? s1.compareTo(s2) : ret;
                     }
                 }

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);
+    }
 }