Support Union in decoupled mode

pom.xml

@@ -1262,6 +1262,12 @@
                 <version>${hamcrest.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.reflections</groupId>
+                <artifactId>reflections</artifactId>
+                <version>0.9.12</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>pl.pragmatists</groupId>
                 <artifactId>JUnitParams</artifactId>

processing/pom.xml

@@ -380,7 +380,6 @@
     <dependency>
       <groupId>org.reflections</groupId>
       <artifactId>reflections</artifactId>
-      <version>0.9.12</version>
       <scope>test</scope>
     </dependency>
     <dependency>

sql/pom.xml

@@ -262,6 +262,11 @@
       <artifactId>jdbi</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.reflections</groupId>
+      <artifactId>reflections</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

sql/src/main/java/org/apache/druid/sql/calcite/planner/CalciteRulesManager.java

@@ -248,20 +248,21 @@ public List<Program> programs(final PlannerContext plannerContext)
         ),
         Programs.sequence(
             druidPreProgram,
-            buildBaseRuleSetProgram(plannerContext),
-            new LoggingProgram("After baseRuleSet program", isDebug),
+            buildDecoupledLogicalOptimizationProgram(plannerContext),
+            new LoggingProgram("After DecoupledLogicalOptimizationProgram program", isDebug),
             Programs.ofRules(logicalConventionRuleSet(plannerContext)),
             new LoggingProgram("After logical volcano planner program", isDebug)
         )
     );
   }
 
-  private Program buildBaseRuleSetProgram(PlannerContext plannerContext)
+  private Program buildDecoupledLogicalOptimizationProgram(PlannerContext plannerContext)
   {
     final HepProgramBuilder builder = HepProgram.builder();
     builder.addMatchLimit(CalciteRulesManager.HEP_DEFAULT_MATCH_LIMIT);
     builder.addGroupBegin();
     builder.addRuleCollection(baseRuleSet(plannerContext));
+    builder.addRuleInstance(CoreRules.UNION_MERGE);
     builder.addGroupEnd();
     return Programs.of(builder.build(), true, DefaultRelMetadataProvider.INSTANCE);
   }