[Enhancement] MV support self-union-join

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/MVPCTRefreshPlanBuilder.java

@@ -46,6 +46,7 @@
 import com.starrocks.sql.ast.SelectListItem;
 import com.starrocks.sql.ast.SelectRelation;
 import com.starrocks.sql.ast.TableRelation;
+import com.starrocks.sql.ast.UnionRelation;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.logging.log4j.Logger;
 
@@ -169,7 +170,10 @@ private InsertStmt buildInsertPlan(InsertStmt insertStmt,
             // If there are multiple table relations, don't push down partition predicate into table relation
             // If `enable_mv_refresh_query_rewrite` is enabled, table relation should not set partition names
             // since it will deduce `hasTableHints` to true and causes rewrite failed.
-            boolean isPushDownBelowTable = (relations.size() == 1);
+            boolean isSameTable = relations.stream().allMatch(e ->
+                    e.getName().equals(relations.iterator().next().getName()));
+            // If the query relation is a union relation and the table relation is the same table, need to push down.
+            boolean isPushDownBelowTable = (relations.size() == 1 || (isSameTable && queryRelation instanceof UnionRelation));
             if (isPushDownBelowTable) {
                 boolean ret = pushDownPartitionPredicates(table, tableRelation, refTablePartitionSlotRefs,
                         tablePartitionNames, isEnableMVRefreshQueryRewrite);

fe/fe-core/src/test/java/com/starrocks/analysis/RefreshMaterializedViewTest.java

@@ -968,6 +968,40 @@ public void testMvOnUnion_Unaligned() throws Exception {
         starRocksAssert.dropMaterializedView(mvName);
     }
 
+    @Test
+    public void testUnionSelf() throws Exception {
+        starRocksAssert.withTable("CREATE TABLE IF NOT EXISTS mv_union_t1 (\n" +
+                "    leg_id VARCHAR(100) NOT NULL,\n" +
+                "    cabin_class VARCHAR(1) NOT NULL,\n" +
+                "    datekey DATE NOT NULL,\n" +
+                "    v1 int(11) NULL\n" +
+                ")\n" +
+                "DUPLICATE KEY(leg_id, cabin_class)\n" +
+                "PARTITION BY RANGE(datekey) (" +
+                "   PARTITION p1_20240321 VALUES LESS THAN ('2024-03-21'), \n" +
+                "   PARTITION p1_20240322 VALUES LESS THAN ('2024-03-22') \n" +
+                ") ");
+
+        starRocksAssert.withRefreshedMaterializedView("CREATE MATERIALIZED VIEW mv_union_1 \n" +
+                "PARTITION BY p_time\n" +
+                "REFRESH ASYNC\n" +
+                "AS \n" +
+                "select date_trunc(\"day\", a.datekey) as p_time FROM mv_union_t1 a group by p_time \n" +
+                "UNION ALL\n" +
+                "select date_trunc(\"day\", b.datekey) as p_time FROM mv_union_t1 b group by p_time \n");
+
+        String mvName = "mv_union_1";
+        MaterializedView mv = starRocksAssert.getMv("test", mvName);
+        Assert.assertTrue(starRocksAssert.waitRefreshFinished(mv.getId()));
+        Assert.assertEquals(
+                Sets.newHashSet("p00010101_20240321", "p20240321_20240322"),
+                mv.getPartitionNames());
+
+        // cleanup
+        starRocksAssert.dropTable("mv_union_t1");
+        starRocksAssert.dropMaterializedView(mvName);
+    }
+
     /**
      * Intersected UNION partition must be same, otherwise will report error
      */