[fix](Nereids) physical property deriver on some node is not right (#30819)

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

@@ -97,7 +97,16 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
 
     @Override
     public PhysicalProperties visit(Plan plan, PlanContext context) {
-        return PhysicalProperties.ANY;
+        if (childrenOutputProperties.isEmpty()) {
+            return PhysicalProperties.ANY;
+        } else {
+            DistributionSpec firstChildSpec = childrenOutputProperties.get(0).getDistributionSpec();
+            if (firstChildSpec instanceof DistributionSpecHash) {
+                return PhysicalProperties.createAnyFromHash((DistributionSpecHash) firstChildSpec);
+            } else {
+                return PhysicalProperties.ANY;
+            }
+        }
     }
 
     /* ********************************************************************************************
@@ -116,7 +125,7 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
     @Override
     public PhysicalProperties visitPhysicalCTEConsumer(
             PhysicalCTEConsumer cteConsumer, PlanContext context) {
-        Preconditions.checkState(childrenOutputProperties.size() == 0);
+        Preconditions.checkState(childrenOutputProperties.isEmpty(), "cte consumer should be leaf node");
         return PhysicalProperties.MUST_SHUFFLE;
     }
 
@@ -279,7 +288,7 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
                                 leftHashSpec.getShuffleType()));
                     }
                 case FULL_OUTER_JOIN:
-                    return PhysicalProperties.ANY;
+                    return PhysicalProperties.createAnyFromHash(leftHashSpec);
                 default:
                     throw new AnalysisException("unknown join type " + hashJoin.getJoinType());
             }
@@ -348,7 +357,12 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
     @Override
     public PhysicalProperties visitPhysicalRepeat(PhysicalRepeat<? extends Plan> repeat, PlanContext context) {
         Preconditions.checkState(childrenOutputProperties.size() == 1);
-        return PhysicalProperties.ANY.withOrderSpec(childrenOutputProperties.get(0).getOrderSpec());
+        DistributionSpec childDistributionSpec = childrenOutputProperties.get(0).getDistributionSpec();
+        PhysicalProperties output = childrenOutputProperties.get(0);
+        if (childDistributionSpec instanceof DistributionSpecHash) {
+            output = PhysicalProperties.createAnyFromHash((DistributionSpecHash) childDistributionSpec);
+        }
+        return output.withOrderSpec(childrenOutputProperties.get(0).getOrderSpec());
     }
 
     @Override
@@ -386,7 +400,12 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
         for (int i = 0; i < childrenDistribution.size(); i++) {
             DistributionSpec childDistribution = childrenDistribution.get(i);
             if (!(childDistribution instanceof DistributionSpecHash)) {
-                return PhysicalProperties.ANY;
+                if (i != 0) {
+                    // NOTICE: if come here, the first child output must be DistributionSpecHash
+                    return PhysicalProperties.createAnyFromHash((DistributionSpecHash) childrenDistribution.get(0));
+                } else {
+                    return new PhysicalProperties(childDistribution);
+                }
             }
             DistributionSpecHash distributionSpecHash = (DistributionSpecHash) childDistribution;
             int[] offsetsOfCurrentChild = new int[distributionSpecHash.getOrderedShuffledColumns().size()];
@@ -396,7 +415,8 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
                 if (offset >= 0) {
                     offsetsOfCurrentChild[offset] = j;
                 } else {
-                    return PhysicalProperties.ANY;
+                    // NOTICE: if come here, the first child output must be DistributionSpecHash
+                    return PhysicalProperties.createAnyFromHash((DistributionSpecHash) childrenDistribution.get(0));
                 }
             }
             if (offsetsOfFirstChild == null) {
@@ -404,7 +424,8 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
                 offsetsOfFirstChild = offsetsOfCurrentChild;
             } else if (!Arrays.equals(offsetsOfFirstChild, offsetsOfCurrentChild)
                     || firstType != ((DistributionSpecHash) childDistribution).getShuffleType()) {
-                return PhysicalProperties.ANY;
+                // NOTICE: if come here, the first child output must be DistributionSpecHash
+                return PhysicalProperties.createAnyFromHash((DistributionSpecHash) childrenDistribution.get(0));
             }
         }
         // bucket
@@ -422,7 +443,7 @@ public class ChildOutputPropertyDeriver extends PlanVisitor<PhysicalProperties,
         } else {
             // current be could not run const expr on appropriate node,
             // so if we have constant exprs on union, the output of union always any
-            return PhysicalProperties.ANY;
+            return visit(union, context);
         }
     }
 

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

@@ -93,6 +93,14 @@ public class PhysicalProperties {
         return new PhysicalProperties(distributionSpecHash);
     }
 
+    public static PhysicalProperties createAnyFromHash(DistributionSpecHash childSpec) {
+        if (childSpec.getShuffleType() == ShuffleType.NATURAL) {
+            return PhysicalProperties.STORAGE_ANY;
+        } else {
+            return PhysicalProperties.ANY;
+        }
+    }
+
     public PhysicalProperties withOrderSpec(OrderSpec orderSpec) {
         return new PhysicalProperties(distributionSpec, orderSpec);
     }