[refactor](Nereids) refactor physical properties and plan translator

fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java

@@ -370,6 +370,7 @@ private PhysicalPlan chooseBestPlan(Group rootGroup, PhysicalProperties physical
 
             Plan plan = groupExpression.getPlan().withChildren(planChildren);
             if (!(plan instanceof PhysicalPlan)) {
+                // TODO need add some log
                 throw new AnalysisException("Result plan must be PhysicalPlan");
             }
             // add groupExpression to plan so that we could print group id in plan.treeString()

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java

@@ -75,7 +75,6 @@
 import org.apache.doris.nereids.trees.expressions.WhenClause;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNullable;
-import org.apache.doris.nereids.trees.expressions.functions.BoundFunction;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
@@ -365,14 +364,11 @@ public Expr visitWindowFunction(WindowFunction function, PlanTranslatorContext c
 
     @Override
     public Expr visitScalarFunction(ScalarFunction function, PlanTranslatorContext context) {
-        List<Expression> nereidsArguments = adaptFunctionArgumentsForBackends(function);
-
-        List<Expr> arguments = nereidsArguments
-                .stream()
+        List<Expr> arguments = function.getArguments().stream()
                 .map(arg -> arg.accept(this, context))
                 .collect(Collectors.toList());
 
-        List<Type> argTypes = nereidsArguments.stream()
+        List<Type> argTypes = function.getArguments().stream()
                 .map(Expression::getDataType)
                 .map(AbstractDataType::toCatalogDataType)
                 .collect(Collectors.toList());
@@ -594,12 +590,4 @@ private static org.apache.doris.analysis.AssertNumRowsElement.Assertion translat
                 throw new AnalysisException("UnSupported type: " + assertion);
         }
     }
-
-    /**
-     * some special arguments not need exists in the nereids, and backends need it, so we must add the
-     * special arguments for backends, e.g. the json data type string in the json_object function.
-     */
-    private List<Expression> adaptFunctionArgumentsForBackends(BoundFunction function) {
-        return function.getArguments();
-    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java

@@ -222,8 +222,8 @@ private boolean calculateEnforce(List<PhysicalProperties> requestChildrenPropert
         // it's certain that lowestCostChildren is equals to arity().
         ChildrenPropertiesRegulator regulator = new ChildrenPropertiesRegulator(groupExpression,
                 lowestCostChildren, outputChildrenProperties, requestChildrenProperties, context);
-        double enforceCost = regulator.adjustChildrenProperties();
-        if (enforceCost < 0) {
+        boolean success = regulator.adjustChildrenProperties();
+        if (!success) {
             // invalid enforce, return.
             return false;
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Analyzer.java

@@ -35,7 +35,6 @@
 import org.apache.doris.nereids.rules.analysis.ResolveOrdinalInOrderByAndGroupBy;
 import org.apache.doris.nereids.rules.analysis.SubqueryToApply;
 import org.apache.doris.nereids.rules.analysis.UserAuthentication;
-import org.apache.doris.nereids.rules.rewrite.HideOneRowRelationUnderUnion;
 
 import java.util.List;
 
@@ -70,8 +69,7 @@ public class Analyzer extends AbstractBatchJobExecutor {
                 // please see rule BindSlotReference or BindFunction for example
                 new ProjectWithDistinctToAggregate(),
                 new ResolveOrdinalInOrderByAndGroupBy(),
-                new ReplaceExpressionByChildOutput(),
-                new HideOneRowRelationUnderUnion()
+                new ReplaceExpressionByChildOutput()
             ),
             topDown(
                 new FillUpMissingSlots(),

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -63,6 +63,7 @@
 import org.apache.doris.nereids.rules.rewrite.InferPredicates;
 import org.apache.doris.nereids.rules.rewrite.InlineCTE;
 import org.apache.doris.nereids.rules.rewrite.MergeFilters;
+import org.apache.doris.nereids.rules.rewrite.MergeOneRowRelationIntoUnion;
 import org.apache.doris.nereids.rules.rewrite.MergeProjects;
 import org.apache.doris.nereids.rules.rewrite.MergeSetOperations;
 import org.apache.doris.nereids.rules.rewrite.NormalizeAggregate;
@@ -71,6 +72,8 @@
 import org.apache.doris.nereids.rules.rewrite.PruneOlapScanPartition;
 import org.apache.doris.nereids.rules.rewrite.PruneOlapScanTablet;
 import org.apache.doris.nereids.rules.rewrite.PushFilterInsideJoin;
+import org.apache.doris.nereids.rules.rewrite.PushProjectIntoOneRowRelation;
+import org.apache.doris.nereids.rules.rewrite.PushProjectThroughUnion;
 import org.apache.doris.nereids.rules.rewrite.PushdownFilterThroughProject;
 import org.apache.doris.nereids.rules.rewrite.PushdownFilterThroughWindow;
 import org.apache.doris.nereids.rules.rewrite.PushdownLimit;
@@ -223,10 +226,13 @@ public class Rewriter extends AbstractBatchJobExecutor {
             // this rule should invoke after ColumnPruning
             custom(RuleType.ELIMINATE_UNNECESSARY_PROJECT, EliminateUnnecessaryProject::new),
 
-            topic("Intersection optimization",
+            topic("Set operation optimization",
                     // Do MergeSetOperation first because we hope to match pattern of Distinct SetOperator.
+                    topDown(new PushProjectThroughUnion(), new MergeProjects()),
                     bottomUp(new MergeSetOperations()),
-                    bottomUp(new BuildAggForUnion())
+                    bottomUp(new PushProjectIntoOneRowRelation()),
+                    topDown(new MergeOneRowRelationIntoUnion()),
+                    topDown(new BuildAggForUnion())
             ),
 
             topic("Window optimization",

fe/fe-core/src/main/java/org/apache/doris/nereids/memo/GroupExpression.java

@@ -248,9 +248,14 @@ public Cost getCostValueByProperties(PhysicalProperties property) {
         return lowestCostTable.get(property).first;
     }
 
-    public void putOutputPropertiesMap(PhysicalProperties outputPropertySet,
-            PhysicalProperties requiredPropertySet) {
-        this.requestPropertiesMap.put(requiredPropertySet, outputPropertySet);
+    public void putOutputPropertiesMap(PhysicalProperties outputProperties,
+            PhysicalProperties requiredProperties) {
+        this.requestPropertiesMap.put(requiredProperties, outputProperties);
+    }
+
+    public void putOutputPropertiesMapIfAbsent(PhysicalProperties outputProperties,
+            PhysicalProperties requiredProperties) {
+        this.requestPropertiesMap.putIfAbsent(requiredProperties, outputProperties);
     }
 
     /**

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

@@ -142,6 +142,9 @@ public PhysicalPlan visitPhysicalHashJoin(PhysicalHashJoin<? extends Plan, ? ext
                                 } else {
                                     origSlot = (SlotReference) targetExpr;
                                 }
+                                if (!aliasTransferMap.containsKey(origSlot)) {
+                                    continue;
+                                }
                                 Slot olapScanSlot = aliasTransferMap.get(origSlot).second;
                                 PhysicalRelation scan = aliasTransferMap.get(origSlot).first;
                                 if (type == TRuntimeFilterType.IN_OR_BLOOM