IGNITE-24730 SQL Calcite: Fix correlated distribution restoring failure - Fixes #11947.

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/IgniteFilter.java

@@ -171,6 +171,16 @@ public IgniteFilter(RelInput input) {
                 // Remap correlate fields to input fields.
                 int corrFieldsCnt = ((RexNode)corr.getNode()).getType().getFieldCount();
                 int inputFieldsCnt = getRowType().getFieldCount();
+
+                // Calcite during planning sometimes can produce hash distribution for correlated nested loop joins
+                // on keys which are not part of RexCorrelVariable type.
+                // These keys can't be refered by RexFieldAccess and original hash distribution can't be restored,
+                // so, just ignore correlated distribution with these keys.
+                for (int i : corrDistr.getKeys()) {
+                    if (i >= corrFieldsCnt)
+                        return null;
+                }
+
                 Mapping mapping = Mappings.create(MappingType.PARTIAL_FUNCTION, corrFieldsCnt, inputFieldsCnt);
 
                 List<RexNode> conds = RelOptUtil.conjunctions(RexUtil.toCnf(getCluster().getRexBuilder(), condition));

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

@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.processors.query.calcite.integration.tpch;
 
 import org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTest;
-import org.junit.Ignore;
 import org.junit.Test;
 
 /** */
@@ -27,7 +26,6 @@ public class TpchQ20Test extends AbstractBasicIntegrationTest {
      * Test the Q20 TPC-H query can be planned and executed.
      */
     @Test
-    @Ignore("https://issues.apache.org/jira/browse/IGNITE-24730")
     public void testQ20() throws Exception {
         TpchHelper.createTables(client);
 

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/planner/CorrelatedSubqueryPlannerTest.java

@@ -35,6 +35,7 @@
 import org.apache.ignite.internal.processors.query.calcite.prepare.IgnitePlanner;
 import org.apache.ignite.internal.processors.query.calcite.prepare.PlannerPhase;
 import org.apache.ignite.internal.processors.query.calcite.prepare.PlanningContext;
+import org.apache.ignite.internal.processors.query.calcite.rel.AbstractIgniteJoin;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteCorrelatedNestedLoopJoin;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
@@ -281,6 +282,14 @@ public void testCorrelatedDistribution() throws Exception {
             ")) FROM th1";
         assertPlan(sql, schema, colocatedPredicate);
 
+        // Correlate on top of another join.
+        sql = "SELECT a FROM ta1 WHERE ta1.a IN (SELECT a FROM th1) AND EXISTS (SELECT 1 FROM ta2 WHERE ta2.b = ta1.a)";
+        assertPlan(sql, schema, hasChildThat(isInstanceOf(IgniteCorrelatedNestedLoopJoin.class)
+            .and(input(0, isInstanceOf(AbstractIgniteJoin.class)))
+            .and(input(1, isInstanceOf(IgniteColocatedAggregateBase.class)
+                .and(hasChildThat(isInstanceOf(IgniteExchange.class)).negate())
+        ))));
+
         // Condition on not colocated column.
         sql = "SELECT a FROM ta1 WHERE EXISTS (SELECT a FROM ta2 WHERE ta2.a = ta1.a)";
         assertPlan(sql, schema, colocatedPredicate.negate());