IGNITE-22789 Throw an exception when join is used for partitioned cache with affinity key and replicated cache with different partitions count (#11449)

zstan · web-flow · commit 5c3cf86f3c41 · 2024-07-24T08:20:48.000+03:00
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlQuerySplitter.java
@@ -48,6 +48,7 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.h2.command.Prepared;
 import org.h2.command.dml.Query;
+import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.processors.query.h2.opt.join.CollocationModel.isCollocated;
 import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlConst.TRUE;
@@ -1271,6 +1272,12 @@ private void splitSelect(GridSqlAst parent, int childIdx) throws IgniteCheckedEx
         SqlAstTraverser traverser = new SqlAstTraverser(mapQry, distributedJoins, log);
         traverser.traverse();
 
+        @Nullable SqlAstTraverser.MixedModeCachesJoinIssue mixedJoinIssue = traverser.hasOuterJoinMixedCacheModeIssue();
+
+        if (mixedJoinIssue != null && mixedJoinIssue.error()) {
+            throw new CacheException(mixedJoinIssue.errorMessage());
+        }
+
         map.columns(collectColumns(mapExps));
         map.sortColumns(mapQry.sort());
         map.partitioned(traverser.hasPartitionedTables());
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/SqlAstTraverser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/SqlAstTraverser.java
@@ -18,9 +18,11 @@
 package org.apache.ignite.internal.processors.query.h2.sql;
 
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Traverse over query AST to find info about partitioned table usage.
@@ -44,6 +46,9 @@ class SqlAstTraverser {
     /** Whether query has joins between replicated and partitioned tables. */
     private boolean hasOuterJoinReplicatedPartitioned;
 
+    /** */
+    private @Nullable MixedModeCachesJoinIssue hasOuterJoinMixedCacheModeIssue;
+
     /** Whether top-level table is replicated. */
     private boolean isRootTableReplicated;
 
@@ -86,6 +91,11 @@ public boolean hasReplicatedWithPartitionedAndSubQuery() {
         return (isRootTableReplicated && hasSubQueries && hasPartitionedTables);
     }
 
+    /** */
+    public @Nullable MixedModeCachesJoinIssue hasOuterJoinMixedCacheModeIssue() {
+        return hasOuterJoinMixedCacheModeIssue;
+    }
+
     /**
      * Traverse AST while join operation isn't found. Check it if found.
      *
@@ -168,8 +178,24 @@ else if (ast instanceof GridSqlTable)
         if (left == null || right == null)
             return;
 
-        if (join.isLeftOuter() && !left.isPartitioned() && right.isPartitioned())
-            hasOuterJoinReplicatedPartitioned = true;
+        if (join.isLeftOuter() && !left.isPartitioned() && right.isPartitioned()) {
+            if (left.cacheContext().affinity().partitions() != right.cacheContext().affinity().partitions()) {
+                hasOuterJoinMixedCacheModeIssue = new MixedModeCachesJoinIssue("Cache [cacheName=" + left.cacheName() +
+                        ", partitionsCount=" + left.cacheContext().affinity().partitions() +
+                        "] can`t be joined with [cacheName=" + right.cacheName() +
+                        ", partitionsCount=" + right.cacheContext().affinity().partitions() +
+                        "] due to different affinity configuration. Join between PARTITIONED and REPLICATED caches is possible "
+                        + "only with the same partitions number configuration.");
+            }
+            // the only way to compare predicate classes, not work for different class loaders.
+            else if (!Objects.equals(className(left.cacheInfo().config().getNodeFilter()), className(right.cacheInfo().config()
+                    .getNodeFilter()))) {
+                hasOuterJoinMixedCacheModeIssue = new MixedModeCachesJoinIssue("Cache [cacheName=" + left.cacheName() + "] "
+                        + "can`t be joined with [cacheName=" + right.cacheName() + "] due to different node filters configuration.");
+            }
+            else
+                hasOuterJoinReplicatedPartitioned = true;
+        }
 
         // Skip check if at least one of tables isn't partitioned.
         if (!(left.isPartitioned() && right.isPartitioned()))
@@ -179,6 +205,11 @@ else if (ast instanceof GridSqlTable)
             checkPartitionedJoin(join, where, left, right, log);
     }
 
+    /** Object class name. */
+    @Nullable private static String className(@Nullable Object obj) {
+        return obj != null ? obj.getClass().getName() : null;
+    }
+
     /**
      * Checks whether an AST contains valid join operation between partitioned tables.
      * Join condition should be an equality operation of affinity keys of tables. Conditions can be splitted between
@@ -242,7 +273,7 @@ private String getAlias(GridSqlElement el) {
     private Set<String> affKeys(boolean pk, GridH2Table tbl) {
         Set<String> affKeys = new HashSet<>();
 
-        // User explicitly specify an affinity key. Otherwise use primary key.
+        // User explicitly specify an affinity key. Otherwise, use primary key.
         if (!pk)
             affKeys.add(tbl.getAffinityKeyColumn().columnName);
         else {
@@ -279,7 +310,7 @@ private boolean checkPartitionedCondition(GridSqlElement condition,
         if (GridSqlOperationType.EQUAL == op.operationType())
             checkEqualityOperation(op, leftTbl, leftAffKeys, pkLeft, rightTbl, rightAffKeys, pkRight);
 
-        // Check affinity condition is covered fully. If true then return. Otherwise go deeper.
+        // Check affinity condition is covered fully. If true then return. Otherwise, go deeper.
         if (affinityCondIsCovered(leftAffKeys, rightAffKeys))
             return true;
 
@@ -342,4 +373,29 @@ private void checkEqualityOperation(GridSqlOperation equalOp,
     private boolean affinityCondIsCovered(Set<String> leftAffKeys, Set<String> rightAffKeys) {
         return leftAffKeys.isEmpty() && rightAffKeys.isEmpty();
     }
+
+    /** Mixed cache mode join issues. */
+    static class MixedModeCachesJoinIssue {
+        /** */
+        private final boolean err;
+
+        /** */
+        private final String msg;
+
+        /** Constructor. */
+        MixedModeCachesJoinIssue(String errMsg) {
+            err = true;
+            msg = errMsg;
+        }
+
+        /** Return {@code true} if error present. */
+        boolean error() {
+            return err;
+        }
+
+        /** Return appropriate error message. */
+        String errorMessage() {
+            return msg;
+        }
+    }
 }
diff --git a/modules/indexing/src/test/java/org/apache/ignite/sqltests/BaseSqlTest.java b/modules/indexing/src/test/java/org/apache/ignite/sqltests/BaseSqlTest.java
@@ -1224,7 +1224,6 @@ public void testRightJoin() {
     /**
      * Check that FULL OUTER JOIN (which is currently unsupported) causes valid error message.
      */
-    @SuppressWarnings("ThrowableNotThrown")
     @Test
     public void testFullOuterJoinIsNotSupported() {
         testAllNodes(node -> {
@@ -1245,7 +1244,6 @@ public void testFullOuterJoinIsNotSupported() {
     /**
      * Check that distributed FULL OUTER JOIN (which is currently unsupported) causes valid error message.
      */
-    @SuppressWarnings("ThrowableNotThrown")
     @Test
     public void testFullOuterDistributedJoinIsNotSupported() {
         testAllNodes(node -> {
diff --git a/modules/indexing/src/test/java/org/apache/ignite/sqltests/ReplicatedSqlCustomPartitionsTest.java b/modules/indexing/src/test/java/org/apache/ignite/sqltests/ReplicatedSqlCustomPartitionsTest.java
@@ -17,25 +17,38 @@
 
 package org.apache.ignite.sqltests;
 
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
 
 /**
  * Includes all base sql test plus tests that make sense in replicated mode with a non-default number of partitions.
  */
 public class ReplicatedSqlCustomPartitionsTest extends ReplicatedSqlTest {
     /** Test partitions count. */
-    private static final int NUM_OF_PARTITIONS = 509;
+    static final int NUM_OF_PARTITIONS = 509;
+
+    /** */
+    static final String DEP_PART_TAB_DIFF = "DepartmentPartDiff";
+
+    /** */
+    static final String DEP_PART_TAB_DIFF_NF = "DepartmentPartDiffNf";
 
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
         return super.getConfiguration(igniteInstanceName)
             .setCacheConfiguration(
                 new CacheConfiguration("partitioned" + NUM_OF_PARTITIONS + "*")
                     .setAffinity(new RendezvousAffinityFunction(false, NUM_OF_PARTITIONS)),
+                new CacheConfiguration("partitioned" + NUM_OF_PARTITIONS + "_DIFF*")
+                    .setAffinity(new RendezvousAffinityFunction(false, NUM_OF_PARTITIONS + 1)),
+                new CacheConfiguration("partitioned" + NUM_OF_PARTITIONS + "_DIFF_NF*")
+                    .setAffinity(new RendezvousAffinityFunction(false, NUM_OF_PARTITIONS))
+                        .setNodeFilter(clusterNode -> true),
                 new CacheConfiguration("replicated" + NUM_OF_PARTITIONS + "*")
                     .setCacheMode(CacheMode.REPLICATED)
                     .setAffinity(new RendezvousAffinityFunction(false, NUM_OF_PARTITIONS))
@@ -55,6 +68,14 @@ public class ReplicatedSqlCustomPartitionsTest extends ReplicatedSqlTest {
         createDepartmentTable(DEP_PART_TAB, "template=partitioned" + NUM_OF_PARTITIONS);
 
         fillDepartmentTable(DEP_PART_TAB);
+
+        createDepartmentTable(DEP_PART_TAB_DIFF, "template=partitioned" + NUM_OF_PARTITIONS + "_DIFF");
+
+        fillDepartmentTable(DEP_PART_TAB_DIFF);
+
+        createDepartmentTable(DEP_PART_TAB_DIFF_NF, "template=partitioned" + NUM_OF_PARTITIONS + "_DIFF_NF");
+
+        fillDepartmentTable(DEP_PART_TAB_DIFF_NF);
     }
 
     /**
@@ -73,4 +94,42 @@ public void testLeftJoinReplicatedPartitioned() {
     public void testRightJoinPartitionedReplicated() {
         checkRightJoinDepartmentEmployee(DEP_PART_TAB);
     }
+
+    /**
+     * Check LEFT JOIN with collocated data of replicated and partitioned tables with different affinity.
+     * This test relies on having the same number of partitions in replicated and partitioned caches
+     */
+    @Test
+    public void testLeftJoinReplicatedPartitionedDiffPartitionsErr() {
+        GridTestUtils.assertThrows(log, () -> checkLeftJoinEmployeeDepartment(DEP_PART_TAB_DIFF), IgniteException.class,
+                "only with the same partitions number configuration");
+    }
+
+    /**
+     * Check RIGHT JOIN with collocated data of partitioned and replicated tables with different affinity.
+     */
+    @Test
+    public void testRightJoinPartitionedReplicatedDiffPartitionsErr() {
+        GridTestUtils.assertThrows(log, () -> checkRightJoinDepartmentEmployee(DEP_PART_TAB_DIFF), IgniteException.class,
+                "only with the same partitions number configuration");
+    }
+
+    /**
+     * Check LEFT JOIN with collocated data of replicated and partitioned tables with different node filter.
+     * This test relies on having the same number of partitions in replicated and partitioned caches
+     */
+    @Test
+    public void testLeftJoinReplicatedPartitionedDiffNodeFilterErr() {
+        GridTestUtils.assertThrows(log, () -> checkLeftJoinEmployeeDepartment(DEP_PART_TAB_DIFF_NF), IgniteException.class,
+                "due to different node filters configuration");
+    }
+
+    /**
+     * Check RIGHT JOIN with collocated data of partitioned and replicated tables with different node filter.
+     */
+    @Test
+    public void testRightJoinPartitionedReplicatedDiffNodeFilterErr() {
+        GridTestUtils.assertThrows(log, () -> checkRightJoinDepartmentEmployee(DEP_PART_TAB_DIFF_NF), IgniteException.class,
+                "due to different node filters configuration");
+    }
 }
