IGNITE-24721 SQL Calcite: Fix distribution for rehashing to the right hand of JOIN - Fixes #11913.

Signed-off-by: Aleksey Plekhanov <plehanov.alex@gmail.com>

Author: alex-plekhanov

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/AbstractIgniteJoin.java

@@ -174,7 +174,7 @@ protected AbstractIgniteJoin(RelOptCluster cluster, RelTraitSet traitSet, RelNod
         }
 
         if (rightDistr.getType() == HASH_DISTRIBUTED && right2leftProjectedDistr != random()) {
-            outTraits = nodeTraits.replace(rightDistr);
+            outTraits = nodeTraits.replace(right2leftProjectedDistr);
             leftTraits = left.replace(right2leftProjectedDistr);
             rightTraits = right.replace(rightDistr);
 

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/JoinRehashIntegrationTest.java

@@ -60,7 +60,7 @@ public void testNullAffinityKeys() {
 
         // Add null values.
         for (int i = 0; i < 10; i++)
-            sql("INSERT INTO order_items VALUES(?, null, null, null)", "null_key_" + i);
+            sql("INSERT INTO order_items (id) VALUES(?)", "null_key_" + i);
 
         String sql = "SELECT sum(i.price * i.amount)" +
             " FROM order_items i JOIN orders o ON o.id=i.orderId" +
@@ -73,11 +73,31 @@ public void testNullAffinityKeys() {
             .check();
     }
 
+    /** */
+    @Test
+    public void testRehashOnRightHand() {
+        prepareTables();
+
+        assertQuery("SELECT /*+ ENFORCE_JOIN_ORDER */ o.id, i.id " +
+            "FROM order_items oi JOIN items i ON (i.id = oi.itemId) JOIN orders o ON (oi.orderId = o.id) " +
+            "WHERE oi.orderId between 2 and 3 and oi.itemId between 4 and 5")
+            .matches(QueryChecker.containsSubPlan("IgniteExchange(distribution=[affinity"))
+            .returns(2, 4).returns(2, 5).returns(3, 4).returns(3, 5)
+            .check();
+    }
+
     /** Prepare tables orders and order_items with data. */
     private void prepareTables() {
+        sql("CREATE TABLE items (\n" +
+            "    id int,\n" +
+            "    name varchar,\n" +
+            "    PRIMARY KEY (id))\n" +
+            "    WITH \"cache_name=items,backups=1," + atomicity() + "\"");
+
         sql("CREATE TABLE order_items (\n" +
             "    id varchar,\n" +
             "    orderId int,\n" +
+            "    itemId int,\n" +
             "    price decimal,\n" +
             "    amount int,\n" +
             "    PRIMARY KEY (id))\n" +
@@ -90,12 +110,16 @@ private void prepareTables() {
             "    WITH \"cache_name=orders,backups=1," + atomicity() + "\"");
 
         sql("CREATE INDEX order_items_orderId ON order_items (orderId ASC)");
+        sql("CREATE INDEX order_items_itemId ON order_items (itemId ASC)");
         sql("CREATE INDEX orders_region ON orders (region ASC)");
 
+        for (int i = 0; i < 20; i++)
+            sql("INSERT INTO items VALUES(?, ?)", i, "item" + i);
+
         for (int i = 0; i < 30; i++) {
             sql("INSERT INTO orders VALUES(?, ?)", i, "region" + i % 10);
             for (int j = 0; j < 20; j++)
-                sql("INSERT INTO order_items VALUES(?, ?, ?, ?)", i + "_" + j, i, i / 10.0, j % 10);
+                sql("INSERT INTO order_items VALUES(?, ?, ?, ?, ?)", i + "_" + j, i, j, i / 10.0, j % 10);
         }
     }
 }