This is an automated email from the ASF dual-hosted git repository. amashenkov pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push: new 7bfa2a0fe8 IGNITE-20503 Sql. Support big clusters by mapping service (#3966) 7bfa2a0fe8 is described below commit 7bfa2a0fe812f67c0a14a5e5110836e373f2cd60 Author: Andrew V. Mashenkov <amashen...@users.noreply.github.com> AuthorDate: Thu Jun 27 16:45:56 2024 +0300 IGNITE-20503 Sql. Support big clusters by mapping service (#3966) --- .../sql/engine/exec/mapping/ExecutionTarget.java | 17 +- .../sql/engine/exec/mapping/FragmentMapper.java | 6 +- .../sql/engine/exec/mapping/MappingContext.java | 7 +- .../AbstractTarget.java | 151 +++++++---- .../AllOfTarget.java | 10 +- .../LargeClusterFactory.java} | 56 ++-- .../OneOfTarget.java | 30 +-- .../PartitionedTarget.java | 46 ++-- .../SomeOfTarget.java | 25 +- .../exec/mapping/smallcluster/AbstractTarget.java | 42 ++- .../exec/mapping/smallcluster/AllOfTarget.java | 5 - .../exec/mapping/smallcluster/OneOfTarget.java | 8 +- .../mapping/smallcluster/PartitionedTarget.java | 7 +- .../mapping/smallcluster/SmallClusterFactory.java | 19 +- .../exec/mapping/smallcluster/SomeOfTarget.java | 5 - .../sql/engine/benchmarks/MappingBenchmark.java | 180 +++++++++++++ .../mapping/ExecutionTargetFactorySelfTest.java | 284 +++++++++++++++++++++ 17 files changed, 700 insertions(+), 198 deletions(-) diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTarget.java index 47f629fc2c..b988eace26 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTarget.java @@ -35,9 +35,7 @@ public interface ExecutionTarget { * <p>Colocation is a process of finding intersection of the given two targets. For example, * lets assume that we have two targets T1 and T2. T1 may be execute on one of the nodes * [N1, N2, N3]. T2 may be executed on one of the nodes [N2, N3, N4, N5]. The result of - * colocation of T1 and T2 will be target OneOf[N2, N3]. Please note, that result of this - * example requires finalisation, since we've got two nodes [N2, N3], but target should - * be executed on only one of them. + * colocation of T1 and T2 will be target OneOf[N2, N3]. * * @param other A target to colocate with. * @return A colocated target. @@ -47,22 +45,11 @@ public interface ExecutionTarget { /** * Removes options from current target which are not colocated with other target. - * + * * <p>If target has several options, remove those are not presented in given target to improve colocation. * * @param other Target with which we need to colocate current target. * @return Returns new target in case current has been adjusted, return {@code this} instance otherwise. */ ExecutionTarget trimTo(ExecutionTarget other); - - /** - * Finalises target by choosing exactly one node for targets with multiple options. - * - * <p>Some targets may have several options, so we have to pick one in order to get - * correct results. Call to this methods resolves this ambiguity by truncating all - * but one option. Which exactly option will be left is implementation defined. - * - * @return Finalised target. - */ - ExecutionTarget finalise(); } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMapper.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMapper.java index 271a7055bb..d252b7a118 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMapper.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/FragmentMapper.java @@ -804,10 +804,8 @@ class FragmentMapper { @Override public List<ColocationGroup> createColocationGroups() { - ExecutionTarget finalised = target.finalise(); - - List<String> nodes = context.targetFactory().resolveNodes(finalised); - Int2ObjectMap<NodeWithConsistencyToken> assignments = context.targetFactory().resolveAssignments(finalised); + List<String> nodes = context.targetFactory().resolveNodes(target); + Int2ObjectMap<NodeWithConsistencyToken> assignments = context.targetFactory().resolveAssignments(target); return List.of( new ColocationGroup( diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingContext.java index 4d559694aa..b230df95ab 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingContext.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/MappingContext.java @@ -19,6 +19,7 @@ package org.apache.ignite.internal.sql.engine.exec.mapping; import java.util.List; import org.apache.calcite.plan.RelOptCluster; +import org.apache.ignite.internal.sql.engine.exec.mapping.largecluster.LargeClusterFactory; import org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster.SmallClusterFactory; import org.apache.ignite.internal.sql.engine.util.Commons; @@ -37,11 +38,7 @@ class MappingContext { this.localNode = localNode; this.nodes = nodes; - if (nodes.size() > 64) { - throw new UnsupportedOperationException("https://issues.apache.org/jira/browse/IGNITE-20503"); - } - - this.targetFactory = new SmallClusterFactory(nodes); + this.targetFactory = nodes.size() > 64 ? new LargeClusterFactory(nodes) : new SmallClusterFactory(nodes); } public RelOptCluster cluster() { diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AbstractTarget.java similarity index 57% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AbstractTarget.java index 293d697774..6d8c14c32e 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AbstractTarget.java @@ -15,15 +15,16 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; - -import static org.apache.ignite.internal.util.IgniteUtils.isPow2; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; import it.unimi.dsi.fastutil.ints.Int2ObjectMap; import it.unimi.dsi.fastutil.ints.Int2ObjectMaps; import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap; import java.util.ArrayList; +import java.util.Arrays; +import java.util.BitSet; import java.util.List; +import org.apache.calcite.util.BitSets; import org.apache.ignite.internal.sql.engine.exec.NodeWithConsistencyToken; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; @@ -35,25 +36,30 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; * colocation method. */ abstract class AbstractTarget implements ExecutionTarget { - final long nodes; + final BitSet nodes; + + AbstractTarget(BitSet nodes) { + assert !nodes.isEmpty(); - AbstractTarget(long nodes) { this.nodes = nodes; } List<String> nodes(List<String> nodeNames) { - if (isPow2(nodes)) { - int idx = Long.numberOfTrailingZeros(nodes); + int cardinality = nodes.cardinality(); + + if (cardinality == 1) { + int idx = nodes.nextSetBit(0); return List.of(nodeNames.get(idx)); } - int count = Long.bitCount(nodes); - List<String> result = new ArrayList<>(count); + List<String> result = new ArrayList<>(cardinality); - for (int bit = 1, idx = 0; bit <= nodes; bit <<= 1, idx++) { - if ((nodes & bit) != 0) { - result.add(nodeNames.get(idx)); + for (int idx = nodes.nextSetBit(0); idx >= 0; idx = nodes.nextSetBit(idx + 1)) { + result.add(nodeNames.get(idx)); + + if (idx == Integer.MAX_VALUE) { + break; // or (i+1) would overflow } } @@ -70,11 +76,11 @@ abstract class AbstractTarget implements ExecutionTarget { Int2ObjectMap<NodeWithConsistencyToken> result = new Int2ObjectOpenHashMap<>(partitionedTarget.partitionsNodes.length); for (int partNo = 0; partNo < partitionedTarget.partitionsNodes.length; partNo++) { - long partitionNodes = partitionedTarget.partitionsNodes[partNo]; + BitSet partitionNodes = partitionedTarget.partitionsNodes[partNo]; - assert isPow2(partitionNodes); + assert partitionNodes.cardinality() == 1; - int idx = Long.numberOfTrailingZeros(partitionNodes); + int idx = partitionNodes.nextSetBit(0); result.put(partNo, new NodeWithConsistencyToken( nodeNames.get(idx), @@ -85,7 +91,16 @@ abstract class AbstractTarget implements ExecutionTarget { return result; } - abstract boolean finalised(); + /** + * Finalises target by choosing exactly one node for targets with multiple options. + * + * <p>Some targets may have several options, so we have to pick one in order to get + * correct results. Call to this methods resolves this ambiguity by truncating all + * but one option. Which exactly option will be left is implementation defined. + * + * @return Finalised target. + */ + abstract ExecutionTarget finalise(); abstract ExecutionTarget colocate(AllOfTarget other) throws ColocationMappingException; @@ -96,7 +111,7 @@ abstract class AbstractTarget implements ExecutionTarget { abstract ExecutionTarget colocate(SomeOfTarget other) throws ColocationMappingException; static ExecutionTarget colocate(AllOfTarget allOf, AllOfTarget otherAllOf) throws ColocationMappingException { - if (allOf.nodes != otherAllOf.nodes) { + if (!allOf.nodes.equals(otherAllOf.nodes) || otherAllOf.nodes.cardinality() == 0) { throw new ColocationMappingException("Targets are not colocated"); } @@ -104,11 +119,10 @@ abstract class AbstractTarget implements ExecutionTarget { } static ExecutionTarget colocate(AllOfTarget allOf, OneOfTarget oneOf) throws ColocationMappingException { - if ((allOf.nodes & oneOf.nodes) == 0) { - throw new ColocationMappingException("Targets are not colocated"); - } + int target = allOf.nodes.nextSetBit(0); - if (!isPow2(allOf.nodes)) { + // When colocated, AllOfTarget must contains a single node that matches one of OneOfTarget nodes. + if (target == -1 || allOf.nodes.nextSetBit(target + 1) != -1 || !oneOf.nodes.get(target)) { throw new ColocationMappingException("Targets are not colocated"); } @@ -120,9 +134,7 @@ abstract class AbstractTarget implements ExecutionTarget { } static ExecutionTarget colocate(AllOfTarget allOf, SomeOfTarget someOf) throws ColocationMappingException { - long newNodes = allOf.nodes & someOf.nodes; - - if (allOf.nodes != newNodes) { + if (!BitSets.contains(someOf.nodes, allOf.nodes) || allOf.nodes.isEmpty()) { throw new ColocationMappingException("Targets are not colocated"); } @@ -130,9 +142,10 @@ abstract class AbstractTarget implements ExecutionTarget { } static ExecutionTarget colocate(OneOfTarget oneOf, OneOfTarget anotherOneOf) throws ColocationMappingException { - long newNodes = oneOf.nodes & anotherOneOf.nodes; + BitSet newNodes = (BitSet) oneOf.nodes.clone(); + newNodes.and(anotherOneOf.nodes); - if (newNodes == 0) { + if (newNodes.isEmpty()) { throw new ColocationMappingException("Targets are not colocated"); } @@ -140,24 +153,45 @@ abstract class AbstractTarget implements ExecutionTarget { } static ExecutionTarget colocate(OneOfTarget oneOf, PartitionedTarget partitioned) throws ColocationMappingException { - if ((oneOf.nodes & partitioned.nodes) == 0) { - throw new ColocationMappingException("Targets are not colocated"); + if (partitioned.nodes.cardinality() == 1 && oneOf.nodes.get(partitioned.nodes.nextSetBit(0))) { + return partitioned; // All partitions on single node. } - if (!isPow2((partitioned.nodes))) { - throw new ColocationMappingException("Targets are not colocated"); + boolean changed = false; + BitSet newNodes = (BitSet) oneOf.nodes.clone(); + for (int partNo = 0; partNo < partitioned.partitionsNodes.length; partNo++) { + if (newNodes.equals(partitioned.partitionsNodes[partNo])) { + continue; + } + + changed = true; + + newNodes.and(partitioned.partitionsNodes[partNo]); + + if (newNodes.isEmpty()) { + throw new ColocationMappingException("Targets are not colocated"); + } } - return partitioned; + if (!changed) { + return partitioned; + } + + BitSet[] newPartitionsNodes = new BitSet[partitioned.partitionsNodes.length]; + Arrays.fill(newPartitionsNodes, newNodes); + boolean finalised = newNodes.cardinality() == 1; + + return new PartitionedTarget(finalised, newPartitionsNodes, partitioned.enlistmentConsistencyTokens); } static ExecutionTarget colocate(OneOfTarget oneOf, SomeOfTarget someOf) throws ColocationMappingException { - long newNodes = oneOf.nodes & someOf.nodes; - - if (newNodes == 0) { + if (!oneOf.nodes.intersects(someOf.nodes)) { throw new ColocationMappingException("Targets are not colocated"); } + BitSet newNodes = (BitSet) oneOf.nodes.clone(); + newNodes.and(someOf.nodes); + return new OneOfTarget(newNodes); } @@ -166,54 +200,71 @@ abstract class AbstractTarget implements ExecutionTarget { throw new ColocationMappingException("Partitioned targets with mot matching numbers of partitioned are not colocated"); } + boolean changed = false; boolean finalised = true; - long[] newPartitionsNodes = new long[partitioned.partitionsNodes.length]; + BitSet[] newPartitionsNodes = new BitSet[partitioned.partitionsNodes.length]; for (int partNo = 0; partNo < partitioned.partitionsNodes.length; partNo++) { - long newNodes = partitioned.partitionsNodes[partNo] & otherPartitioned.partitionsNodes[partNo]; + if (partitioned.partitionsNodes[partNo].equals(otherPartitioned.partitionsNodes[partNo])) { + newPartitionsNodes[partNo] = partitioned.partitionsNodes[partNo]; - if (newNodes == 0) { - throw new ColocationMappingException("Targets are not colocated"); + continue; } - if (partitioned.enlistmentConsistencyTokens[partNo] != otherPartitioned.enlistmentConsistencyTokens[partNo]) { - throw new ColocationMappingException("Partitioned targets have different terms"); + changed = true; + BitSet newNodes = (BitSet) partitioned.partitionsNodes[partNo].clone(); + newNodes.and(otherPartitioned.partitionsNodes[partNo]); + + if (newNodes.isEmpty()) { + throw new ColocationMappingException("Targets are not colocated"); } newPartitionsNodes[partNo] = newNodes; - finalised = finalised && isPow2(newNodes); + finalised = finalised && newNodes.cardinality() == 1; } - return new PartitionedTarget(finalised, newPartitionsNodes, partitioned.enlistmentConsistencyTokens); + if (!Arrays.equals(partitioned.enlistmentConsistencyTokens, otherPartitioned.enlistmentConsistencyTokens)) { + throw new ColocationMappingException("Partitioned targets have different terms"); + } + + if (changed) { + return new PartitionedTarget(finalised, newPartitionsNodes, partitioned.enlistmentConsistencyTokens); + } + + return partitioned; } static ExecutionTarget colocate(PartitionedTarget partitioned, SomeOfTarget someOf) throws ColocationMappingException { boolean finalised = true; - long[] newPartitionsNodes = new long[partitioned.partitionsNodes.length]; + BitSet[] newPartitionsNodes = new BitSet[partitioned.partitionsNodes.length]; for (int partNo = 0; partNo < partitioned.partitionsNodes.length; partNo++) { - long newNodes = partitioned.partitionsNodes[partNo] & someOf.nodes; + BitSet newNodes = (BitSet) partitioned.partitionsNodes[partNo].clone(); + newNodes.and(someOf.nodes); - if (newNodes == 0) { + if (newNodes.isEmpty()) { throw new ColocationMappingException("Targets are not colocated"); } newPartitionsNodes[partNo] = newNodes; - finalised = finalised && isPow2(newNodes); + finalised = finalised && newNodes.cardinality() == 1; } return new PartitionedTarget(finalised, newPartitionsNodes, partitioned.enlistmentConsistencyTokens); } static ExecutionTarget colocate(SomeOfTarget someOf, SomeOfTarget otherSomeOf) throws ColocationMappingException { - long newNodes = someOf.nodes & otherSomeOf.nodes; + BitSet newNodes = (BitSet) someOf.nodes.clone(); + newNodes.and(otherSomeOf.nodes); - if (newNodes == 0) { + if (newNodes.isEmpty()) { throw new ColocationMappingException("Targets are not colocated"); } return new SomeOfTarget(newNodes); } - static long pickOne(long nodes) { - return Long.lowestOneBit(nodes); + static BitSet pickOne(BitSet nodes) { + int node = nodes.nextSetBit(0); + + return node == -1 ? BitSets.of() : BitSets.of(node); } } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AllOfTarget.java similarity index 93% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AllOfTarget.java index d2bae3afcf..70763f7972 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/AllOfTarget.java @@ -15,8 +15,9 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; +import java.util.BitSet; import java.util.List; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; @@ -28,15 +29,10 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory * <p>See javadoc of {@link ExecutionTargetFactory#allOf(List)} for details. */ class AllOfTarget extends AbstractTarget { - AllOfTarget(long nodes) { + AllOfTarget(BitSet nodes) { super(nodes); } - @Override - boolean finalised() { - return true; - } - @Override public ExecutionTarget finalise() { return this; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/LargeClusterFactory.java similarity index 68% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/LargeClusterFactory.java index e369c311d3..0cf12b78ba 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/LargeClusterFactory.java @@ -15,13 +15,13 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; import it.unimi.dsi.fastutil.ints.Int2ObjectMap; -import it.unimi.dsi.fastutil.objects.Object2LongMap; -import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap; +import it.unimi.dsi.fastutil.objects.Object2IntMap; +import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; +import java.util.BitSet; import java.util.List; -import java.util.stream.Collectors; import org.apache.ignite.internal.affinity.Assignment; import org.apache.ignite.internal.affinity.TokenizedAssignments; import org.apache.ignite.internal.sql.engine.exec.NodeWithConsistencyToken; @@ -31,24 +31,21 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory /** * A factory that able to create targets for cluster with up to 64 nodes. */ -public class SmallClusterFactory implements ExecutionTargetFactory { +public class LargeClusterFactory implements ExecutionTargetFactory { private final List<String> nodes; - private final Object2LongMap<String> nodeNameToId; + private final Object2IntMap<String> nodeNameToId; /** Constructor. */ - public SmallClusterFactory(List<String> nodes) { - if (nodes.size() > 64) { - throw new IllegalArgumentException("Supported up to 64 nodes, but was " + nodes.size()); - } - - // to make mapping stable - this.nodes = nodes.stream().sorted().collect(Collectors.toList()); + public LargeClusterFactory(List<String> nodes) { + this.nodes = nodes; - nodeNameToId = new Object2LongOpenHashMap<>(nodes.size()); + nodeNameToId = new Object2IntOpenHashMap<>(nodes.size()); + nodeNameToId.defaultReturnValue(-1); int idx = 0; - for (String name : this.nodes) { - nodeNameToId.putIfAbsent(name, 1L << idx++); + for (String name : nodes) { + int ret = nodeNameToId.putIfAbsent(name, idx++); + assert ret == -1 : "invalid node"; } } @@ -69,7 +66,7 @@ public class SmallClusterFactory implements ExecutionTargetFactory { @Override public ExecutionTarget partitioned(List<TokenizedAssignments> assignments) { - long[] partitionNodes = new long[assignments.size()]; + BitSet[] partitionNodes = new BitSet[assignments.size()]; long[] enlistmentConsistencyTokens = new long[assignments.size()]; int idx = 0; @@ -77,11 +74,15 @@ public class SmallClusterFactory implements ExecutionTargetFactory { for (TokenizedAssignments assignment : assignments) { finalised = finalised && assignment.nodes().size() < 2; + BitSet nodes = new BitSet(assignment.nodes().size()); for (Assignment a : assignment.nodes()) { - partitionNodes[idx] |= nodeNameToId.getOrDefault(a.consistentId(), 0); - enlistmentConsistencyTokens[idx] = assignment.token(); + int node = nodeNameToId.getOrDefault(a.consistentId(), -1); + assert node >= 0 : "invalid node"; + nodes.set(node); } + partitionNodes[idx] = nodes; + enlistmentConsistencyTokens[idx] = assignment.token(); idx++; } @@ -90,27 +91,30 @@ public class SmallClusterFactory implements ExecutionTargetFactory { @Override public List<String> resolveNodes(ExecutionTarget target) { - target = target.finalise(); - assert target instanceof AbstractTarget : target == null ? "<null>" : target.getClass().getCanonicalName(); + target = ((AbstractTarget) target).finalise(); + return ((AbstractTarget) target).nodes(nodes); } @Override public Int2ObjectMap<NodeWithConsistencyToken> resolveAssignments(ExecutionTarget target) { - target = target.finalise(); - assert target instanceof AbstractTarget : target == null ? "<null>" : target.getClass().getCanonicalName(); + target = ((AbstractTarget) target).finalise(); + return ((AbstractTarget) target).assignments(nodes); } - private long nodeListToMap(List<String> nodes) { - long nodesMap = 0; + private BitSet nodeListToMap(List<String> nodes) { + BitSet nodesMap = new BitSet(nodes.size()); for (String name : nodes) { - nodesMap |= nodeNameToId.getOrDefault(name, 0); + int id = nodeNameToId.getOrDefault(name, -1); + assert id >= 0 : "invalid node"; + + nodesMap.set(id); } return nodesMap; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/OneOfTarget.java similarity index 83% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/OneOfTarget.java index 3098c7dd97..4c9c8de635 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/OneOfTarget.java @@ -15,10 +15,9 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; - -import static org.apache.ignite.internal.util.IgniteUtils.isPow2; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; +import java.util.BitSet; import java.util.List; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; @@ -30,19 +29,13 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory * <p>See javadoc of {@link ExecutionTargetFactory#oneOf(List)} for details. */ class OneOfTarget extends AbstractTarget { - - OneOfTarget(long nodes) { + OneOfTarget(BitSet nodes) { super(nodes); } - @Override - boolean finalised() { - return isPow2(nodes); - } - @Override public ExecutionTarget finalise() { - if (finalised()) { + if (nodes.cardinality() == 1) { return this; } @@ -60,15 +53,20 @@ class OneOfTarget extends AbstractTarget { public ExecutionTarget trimTo(ExecutionTarget other) { assert other instanceof AbstractTarget : other == null ? "<null>" : other.getClass().getCanonicalName(); - long otherNodes = ((AbstractTarget) other).nodes; + if (nodes.cardinality() == 1) { + return this; + } - long newNodes = nodes & otherNodes; + BitSet otherNodes = ((AbstractTarget) other).nodes; - if (newNodes == nodes || newNodes == 0) { - return this; + if (!nodes.equals(otherNodes) && nodes.intersects(otherNodes)) { + BitSet newNodes = (BitSet) nodes.clone(); + newNodes.and(otherNodes); + + return new OneOfTarget(newNodes); } - return new OneOfTarget(newNodes); + return this; } @Override diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/PartitionedTarget.java similarity index 75% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/PartitionedTarget.java index 9a0680eb3e..60f238a167 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/PartitionedTarget.java @@ -15,9 +15,10 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; -import it.unimi.dsi.fastutil.ints.Int2LongFunction; +import it.unimi.dsi.fastutil.ints.Int2ObjectFunction; +import java.util.BitSet; import java.util.List; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; @@ -30,10 +31,10 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory */ class PartitionedTarget extends AbstractTarget { private final boolean finalised; - final long[] partitionsNodes; + final BitSet[] partitionsNodes; final long[] enlistmentConsistencyTokens; - PartitionedTarget(boolean finalised, long[] partitionsNodes, long[] enlistmentConsistencyTokens) { + PartitionedTarget(boolean finalised, BitSet[] partitionsNodes, long[] enlistmentConsistencyTokens) { super(computeNodes(partitionsNodes)); this.finalised = finalised; @@ -41,18 +42,13 @@ class PartitionedTarget extends AbstractTarget { this.enlistmentConsistencyTokens = enlistmentConsistencyTokens; } - @Override - boolean finalised() { - return finalised; - } - @Override public ExecutionTarget finalise() { if (finalised) { return this; } - long[] newPartitionsNodes = new long[partitionsNodes.length]; + BitSet[] newPartitionsNodes = new BitSet[partitionsNodes.length]; for (int partNo = 0; partNo < partitionsNodes.length; partNo++) { newPartitionsNodes[partNo] = pickOne(partitionsNodes[partNo]); @@ -76,18 +72,18 @@ class PartitionedTarget extends AbstractTarget { return this; } - long[] newPartitionsNodes = new long[partitionsNodes.length]; + BitSet[] newPartitionsNodes = new BitSet[partitionsNodes.length]; boolean changed = false; - Int2LongFunction partitionNodesResolver = partitionNodeResolver(other); + Int2ObjectFunction<BitSet> partitionNodesResolver = partitionNodeResolver(other); for (int i = 0; i < partitionsNodes.length; i++) { - long newNodes = partitionsNodes[i] & partitionNodesResolver.get(i); + BitSet newNodes = partitionsNodes[i]; + BitSet otherNodes = partitionNodesResolver.get(i); - if (newNodes == 0) { - newNodes = partitionsNodes[i]; - } + if (!newNodes.equals(otherNodes) && newNodes.intersects(otherNodes)) { + newNodes = (BitSet) newNodes.clone(); + newNodes.and(otherNodes); - if (newNodes != partitionsNodes[i]) { changed = true; } @@ -101,16 +97,16 @@ class PartitionedTarget extends AbstractTarget { return this; } - private Int2LongFunction partitionNodeResolver(ExecutionTarget other) { - Int2LongFunction partitionNodesResolver; + private Int2ObjectFunction<BitSet> partitionNodeResolver(ExecutionTarget other) { + Int2ObjectFunction<BitSet> partitionNodesResolver; - if (other instanceof PartitionedTarget + if (other instanceof PartitionedTarget && ((PartitionedTarget) other).partitionsNodes.length == partitionsNodes.length) { PartitionedTarget otherPartitioned = (PartitionedTarget) other; partitionNodesResolver = partId -> otherPartitioned.partitionsNodes[partId]; } else { - long otherNodes = ((AbstractTarget) other).nodes; + BitSet otherNodes = ((AbstractTarget) other).nodes; partitionNodesResolver = partId -> otherNodes; } @@ -138,10 +134,10 @@ class PartitionedTarget extends AbstractTarget { return colocate(this, other); } - private static long computeNodes(long[] partitionsNodes) { - long nodes = 0; - for (long nodesOfPartition : partitionsNodes) { - nodes |= nodesOfPartition; + private static BitSet computeNodes(BitSet[] partitionsNodes) { + BitSet nodes = new BitSet(); + for (BitSet nodesOfPartition : partitionsNodes) { + nodes.or(nodesOfPartition); } return nodes; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/SomeOfTarget.java similarity index 84% copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/SomeOfTarget.java index 8810657f31..531412681e 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/largecluster/SomeOfTarget.java @@ -15,8 +15,9 @@ * limitations under the License. */ -package org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster; +package org.apache.ignite.internal.sql.engine.exec.mapping.largecluster; +import java.util.BitSet; import java.util.List; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; @@ -28,15 +29,10 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory * <p>See javadoc of {@link ExecutionTargetFactory#someOf(List)} for details. */ class SomeOfTarget extends AbstractTarget { - SomeOfTarget(long nodes) { + SomeOfTarget(BitSet nodes) { super(nodes); } - @Override - boolean finalised() { - return true; - } - @Override public ExecutionTarget finalise() { return this; @@ -53,15 +49,20 @@ class SomeOfTarget extends AbstractTarget { public ExecutionTarget trimTo(ExecutionTarget other) { assert other instanceof AbstractTarget : other == null ? "<null>" : other.getClass().getCanonicalName(); - long otherNodes = ((AbstractTarget) other).nodes; + if (nodes.cardinality() == 1) { + return this; + } + + BitSet otherNodes = ((AbstractTarget) other).nodes; - long newNodes = nodes & otherNodes; + if (!nodes.equals(otherNodes) && nodes.intersects(otherNodes)) { + BitSet newNodes = (BitSet) nodes.clone(); + newNodes.and(otherNodes); - if (newNodes == nodes || newNodes == 0) { - return this; + return new SomeOfTarget(newNodes); } - return new SomeOfTarget(newNodes); + return this; } @Override diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java index 293d697774..1783adaa92 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AbstractTarget.java @@ -23,6 +23,7 @@ import it.unimi.dsi.fastutil.ints.Int2ObjectMap; import it.unimi.dsi.fastutil.ints.Int2ObjectMaps; import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.apache.ignite.internal.sql.engine.exec.NodeWithConsistencyToken; import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; @@ -38,6 +39,8 @@ abstract class AbstractTarget implements ExecutionTarget { final long nodes; AbstractTarget(long nodes) { + assert nodes != 0 : "Empty target is not allowed"; + this.nodes = nodes; } @@ -85,7 +88,16 @@ abstract class AbstractTarget implements ExecutionTarget { return result; } - abstract boolean finalised(); + /** + * Finalises target by choosing exactly one node for targets with multiple options. + * + * <p>Some targets may have several options, so we have to pick one in order to get + * correct results. Call to this methods resolves this ambiguity by truncating all + * but one option. Which exactly option will be left is implementation defined. + * + * @return Finalised target. + */ + abstract ExecutionTarget finalise(); abstract ExecutionTarget colocate(AllOfTarget other) throws ColocationMappingException; @@ -144,11 +156,24 @@ abstract class AbstractTarget implements ExecutionTarget { throw new ColocationMappingException("Targets are not colocated"); } - if (!isPow2((partitioned.nodes))) { - throw new ColocationMappingException("Targets are not colocated"); + if (isPow2((partitioned.nodes))) { + return partitioned; // All partitions on single node. + } + + long colocatedNodes = oneOf.nodes; + for (int partNo = 0; partNo < partitioned.partitionsNodes.length; partNo++) { + colocatedNodes &= partitioned.partitionsNodes[partNo]; + + if (colocatedNodes == 0) { + throw new ColocationMappingException("Targets are not colocated"); + } } - return partitioned; + boolean finalised = isPow2(colocatedNodes); + long[] newNodes = new long[partitioned.partitionsNodes.length]; + Arrays.fill(newNodes, colocatedNodes); + + return new PartitionedTarget(finalised, newNodes, partitioned.enlistmentConsistencyTokens); } static ExecutionTarget colocate(OneOfTarget oneOf, SomeOfTarget someOf) throws ColocationMappingException { @@ -175,14 +200,15 @@ abstract class AbstractTarget implements ExecutionTarget { throw new ColocationMappingException("Targets are not colocated"); } - if (partitioned.enlistmentConsistencyTokens[partNo] != otherPartitioned.enlistmentConsistencyTokens[partNo]) { - throw new ColocationMappingException("Partitioned targets have different terms"); - } - newPartitionsNodes[partNo] = newNodes; finalised = finalised && isPow2(newNodes); } + + if (!Arrays.equals(partitioned.enlistmentConsistencyTokens, otherPartitioned.enlistmentConsistencyTokens)) { + throw new ColocationMappingException("Partitioned targets have different terms"); + } + return new PartitionedTarget(finalised, newPartitionsNodes, partitioned.enlistmentConsistencyTokens); } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java index d2bae3afcf..4b30173ca3 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/AllOfTarget.java @@ -32,11 +32,6 @@ class AllOfTarget extends AbstractTarget { super(nodes); } - @Override - boolean finalised() { - return true; - } - @Override public ExecutionTarget finalise() { return this; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java index 3098c7dd97..29df4f102c 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/OneOfTarget.java @@ -30,19 +30,13 @@ import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory * <p>See javadoc of {@link ExecutionTargetFactory#oneOf(List)} for details. */ class OneOfTarget extends AbstractTarget { - OneOfTarget(long nodes) { super(nodes); } - @Override - boolean finalised() { - return isPow2(nodes); - } - @Override public ExecutionTarget finalise() { - if (finalised()) { + if (isPow2(nodes)) { return this; } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java index 9a0680eb3e..2e7fd14536 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/PartitionedTarget.java @@ -42,12 +42,7 @@ class PartitionedTarget extends AbstractTarget { } @Override - boolean finalised() { - return finalised; - } - - @Override - public ExecutionTarget finalise() { + ExecutionTarget finalise() { if (finalised) { return this; } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java index e369c311d3..732940c6d2 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SmallClusterFactory.java @@ -78,10 +78,13 @@ public class SmallClusterFactory implements ExecutionTargetFactory { finalised = finalised && assignment.nodes().size() < 2; for (Assignment a : assignment.nodes()) { - partitionNodes[idx] |= nodeNameToId.getOrDefault(a.consistentId(), 0); - enlistmentConsistencyTokens[idx] = assignment.token(); + long node = nodeNameToId.getOrDefault(a.consistentId(), -1); + assert node >= 0 : "invalid node"; + partitionNodes[idx] |= node; } + enlistmentConsistencyTokens[idx] = assignment.token(); + idx++; } @@ -90,19 +93,19 @@ public class SmallClusterFactory implements ExecutionTargetFactory { @Override public List<String> resolveNodes(ExecutionTarget target) { - target = target.finalise(); - assert target instanceof AbstractTarget : target == null ? "<null>" : target.getClass().getCanonicalName(); + target = ((AbstractTarget) target).finalise(); + return ((AbstractTarget) target).nodes(nodes); } @Override public Int2ObjectMap<NodeWithConsistencyToken> resolveAssignments(ExecutionTarget target) { - target = target.finalise(); - assert target instanceof AbstractTarget : target == null ? "<null>" : target.getClass().getCanonicalName(); + target = ((AbstractTarget) target).finalise(); + return ((AbstractTarget) target).assignments(nodes); } @@ -110,7 +113,9 @@ public class SmallClusterFactory implements ExecutionTargetFactory { long nodesMap = 0; for (String name : nodes) { - nodesMap |= nodeNameToId.getOrDefault(name, 0); + long node = nodeNameToId.getOrDefault(name, -1); + assert node >= 0 : "invalid node"; + nodesMap |= node; } return nodesMap; diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java index 8810657f31..3aa5a3300c 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/mapping/smallcluster/SomeOfTarget.java @@ -32,11 +32,6 @@ class SomeOfTarget extends AbstractTarget { super(nodes); } - @Override - boolean finalised() { - return true; - } - @Override public ExecutionTarget finalise() { return this; diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/benchmarks/MappingBenchmark.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/benchmarks/MappingBenchmark.java new file mode 100644 index 0000000000..f473849a19 --- /dev/null +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/benchmarks/MappingBenchmark.java @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine.benchmarks; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.apache.ignite.internal.affinity.Assignment; +import org.apache.ignite.internal.affinity.TokenizedAssignments; +import org.apache.ignite.internal.affinity.TokenizedAssignmentsImpl; +import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationMappingException; +import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget; +import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory; +import org.apache.ignite.internal.sql.engine.exec.mapping.largecluster.LargeClusterFactory; +import org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster.SmallClusterFactory; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; + +/** + * Mapping part benchmark. + * + * <p>Note: seems more accurate results can be obtained after boosting disabling: + * linux: echo "1" | sudo tee /sys/devices/system/cpu/intel_pstate/no_turbo + */ +@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(1) +@Threads(1) +@State(Scope.Benchmark) +public class MappingBenchmark { + private ExecutionTarget oneOfSmall; + private ExecutionTarget someOfSmall; + private ExecutionTarget part1Small; + private ExecutionTarget part2Small; + private ExecutionTarget part1SmallMore; + private ExecutionTarget part2SmallMore; + private ExecutionTarget allOfSmall; + + private ExecutionTarget oneOfBig; + private ExecutionTarget someOfBig; + private ExecutionTarget part1Big; + private ExecutionTarget part2Big; + private ExecutionTarget part1BigMore; + private ExecutionTarget part2BigMore; + private ExecutionTarget allOfBig; + + private ExecutionTargetFactory targetFactorySmall; + private ExecutionTargetFactory targetFactoryLarge; + + /** Prepare the plan of the query. */ + @Setup + public void setUp() { + List<String> nodes = List.of("n1", "n2", "n3", "n4", "n5"); + List<String> oneOfNodes = List.of("n1", "n3"); + List<TokenizedAssignments> partNodes = + List.of(tokenizedAssignment(1, "n3")); + List<TokenizedAssignments> partNodesMore = + List.of(tokenizedAssignment(1, "n3", "n4"), + tokenizedAssignment(2, "n4", "n5"), + tokenizedAssignment(3, "n5", "n3")); + List<TokenizedAssignments> partNodesMore2 = + List.of(tokenizedAssignment(1, "n5", "n3"), + tokenizedAssignment(2, "n3", "n4"), + tokenizedAssignment(3, "n4", "n5")); + + targetFactorySmall = new SmallClusterFactory(nodes); + targetFactoryLarge = new LargeClusterFactory(nodes); + + oneOfSmall = targetFactorySmall.oneOf(oneOfNodes); + someOfSmall = targetFactorySmall.someOf(nodes); + allOfSmall = targetFactorySmall.allOf(List.of("n3")); + part1Small = targetFactorySmall.partitioned(partNodes); + part2Small = targetFactorySmall.partitioned(partNodes); + part1SmallMore = targetFactorySmall.partitioned(partNodesMore); + part2SmallMore = targetFactorySmall.partitioned(partNodesMore2); + + oneOfBig = targetFactoryLarge.oneOf(oneOfNodes); + someOfBig = targetFactoryLarge.someOf(nodes); + allOfBig = targetFactoryLarge.allOf(List.of("n3")); + part1Big = targetFactoryLarge.partitioned(partNodes); + part2Big = targetFactoryLarge.partitioned(partNodes); + part1BigMore = targetFactoryLarge.partitioned(partNodesMore); + part2BigMore = targetFactoryLarge.partitioned(partNodesMore2); + } + + /** Measure small cluster mapping implementation. */ + @Benchmark + public void benchSmall(Blackhole bh) throws ColocationMappingException { + finalise(targetFactorySmall, oneOfSmall.colocateWith(oneOfSmall), bh); + finalise(targetFactorySmall, oneOfSmall.colocateWith(someOfSmall), bh); + finalise(targetFactorySmall, oneOfSmall.colocateWith(allOfSmall), bh); + finalise(targetFactorySmall, oneOfSmall.colocateWith(part1Small), bh); + + finalise(targetFactorySmall, someOfSmall.colocateWith(someOfSmall), bh); + finalise(targetFactorySmall, someOfSmall.colocateWith(part1Small), bh); + finalise(targetFactorySmall, someOfSmall.colocateWith(allOfSmall), bh); + + finalise(targetFactorySmall, part1Small.colocateWith(part2Small), bh); + } + + /** Measure small cluster mapping partitioned only implementation. */ + @Benchmark + public void benchSmallPartitionedOnly(Blackhole bh) throws ColocationMappingException { + finalise(targetFactorySmall, part1SmallMore.colocateWith(part2SmallMore), bh); + } + + /** Measure huge cluster mapping implementation. */ + @Benchmark + public void benchBig(Blackhole bh) throws ColocationMappingException { + finalise(targetFactoryLarge, oneOfBig.colocateWith(oneOfBig), bh); + finalise(targetFactoryLarge, oneOfBig.colocateWith(someOfBig), bh); + finalise(targetFactoryLarge, oneOfBig.colocateWith(allOfBig), bh); + finalise(targetFactoryLarge, oneOfBig.colocateWith(part1Big), bh); + + finalise(targetFactoryLarge, someOfBig.colocateWith(someOfBig), bh); + finalise(targetFactoryLarge, someOfBig.colocateWith(part1Big), bh); + finalise(targetFactoryLarge, someOfBig.colocateWith(allOfBig), bh); + + finalise(targetFactoryLarge, part1Big.colocateWith(part2Big), bh); + } + + /** Measure huge cluster mapping partitioned only implementation. */ + @Benchmark + public void benchBigPartitionedOnly(Blackhole bh) throws ColocationMappingException { + finalise(targetFactoryLarge, part1BigMore.colocateWith(part2BigMore), bh); + } + + /** + * Runs the benchmark. + * + * @param args args + * @throws Exception if something goes wrong + */ + public static void main(String[] args) throws Exception { + Options build = new OptionsBuilder() + .include(MappingBenchmark.class.getName()) + .build(); + + new Runner(build).run(); + } + + private static TokenizedAssignmentsImpl tokenizedAssignment(int token, String... peer) { + return new TokenizedAssignmentsImpl(Arrays.stream(peer).map(Assignment::forPeer).collect(Collectors.toSet()), token); + } + + private static void finalise(ExecutionTargetFactory factory, ExecutionTarget target, Blackhole bh) { + bh.consume(factory.resolveNodes(target)); + } +} \ No newline at end of file diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTargetFactorySelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTargetFactorySelfTest.java new file mode 100644 index 0000000000..be662b48d0 --- /dev/null +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/mapping/ExecutionTargetFactorySelfTest.java @@ -0,0 +1,284 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.sql.engine.exec.mapping; + +import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrows; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.allOf; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasItems; +import static org.hamcrest.Matchers.in; +import static org.hamcrest.Matchers.iterableWithSize; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.ignite.internal.affinity.Assignment; +import org.apache.ignite.internal.affinity.TokenizedAssignments; +import org.apache.ignite.internal.affinity.TokenizedAssignmentsImpl; +import org.apache.ignite.internal.sql.engine.exec.mapping.largecluster.LargeClusterFactory; +import org.apache.ignite.internal.sql.engine.exec.mapping.smallcluster.SmallClusterFactory; +import org.hamcrest.Matcher; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Test class to verify {@link ExecutionTargetFactory} implementations. + * + * @see SmallClusterFactory + * @see LargeClusterFactory + */ +public class ExecutionTargetFactorySelfTest { + private static final List<String> ALL_NODES = List.of("node1", "node2", "node3", "node4", "node5"); + private static final List<String> NODE_SET = List.of("node2", "node4", "node5"); + private static final List<String> NODE_SET2 = List.of("node2", "node3", "node5"); + private static final List<String> NODE_SUBSET = List.of("node2", "node5"); + private static final List<String> SINGLE_NODE_SET = List.of("node4"); + private static final List<String> INVALID_NODE_SET = List.of("node0"); + + private static List<ExecutionTargetFactory> clusterFactory() { + return List.of( + new SmallClusterFactory(ALL_NODES), + new LargeClusterFactory(ALL_NODES) + ); + } + + @SuppressWarnings({"ResultOfObjectAllocationIgnored", "ThrowableNotThrown"}) + @Test + void smallClusterFactory() { + List<String> nodes = IntStream.range(0, 65).mapToObj(i -> "node").collect(Collectors.toList()); + + assertThrows(IllegalArgumentException.class, () -> new SmallClusterFactory(nodes), "Supported up to 64 nodes"); + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void targetsResolution(ExecutionTargetFactory f) { + assertThat(f.resolveNodes(f.allOf(NODE_SET)), equalTo(NODE_SET)); + assertThat(f.resolveNodes(f.allOf(NODE_SET)), equalTo(NODE_SET)); + assertThat(f.resolveNodes(f.someOf(NODE_SET)), hasItems(in(NODE_SET))); + assertThat(f.resolveNodes(f.oneOf(NODE_SET)), containsSingleFrom(NODE_SET)); + assertThat(f.resolveNodes(f.partitioned(assignmentFromPrimaries(NODE_SET))), equalTo(NODE_SET)); + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void targetValidation(ExecutionTargetFactory f) { + assertThrows(AssertionError.class, () -> f.allOf(List.of()), null); + assertThrows(AssertionError.class, () -> f.someOf(List.of()), null); + assertThrows(AssertionError.class, () -> f.oneOf(List.of()), null); + assertThrows(AssertionError.class, () -> f.partitioned(List.of()), null); + + assertThrows(Throwable.class, () -> f.allOf(INVALID_NODE_SET), "invalid node"); + assertThrows(Throwable.class, () -> f.someOf(INVALID_NODE_SET), "invalid node"); + assertThrows(Throwable.class, () -> f.oneOf(INVALID_NODE_SET), "invalid node"); + assertThrows(Throwable.class, () -> f.partitioned(assignmentFromPrimaries(INVALID_NODE_SET)), "invalid node"); + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void allOfTargets(ExecutionTargetFactory f) throws Exception { + // Self colocation + assertColocated(f, f.allOf(NODE_SET), f.allOf(NODE_SET), equalTo(NODE_SET)); + assertNotColocated(f.allOf(NODE_SET), f.allOf(NODE_SUBSET)); + assertNotColocated(f.allOf(NODE_SUBSET), f.allOf(NODE_SET)); + + // Colocation with SomeOf + assertColocated(f, f.allOf(NODE_SET), f.someOf(NODE_SET), equalTo(NODE_SET)); + assertNotColocated(f.allOf(NODE_SET), f.someOf(NODE_SUBSET)); + assertColocated(f, f.allOf(NODE_SUBSET), f.someOf(NODE_SET), equalTo(NODE_SUBSET)); + + // Colocation with OneOf + assertNotColocated(f.allOf(NODE_SET), f.oneOf(NODE_SET)); + assertNotColocated(f.allOf(NODE_SET), f.oneOf(NODE_SUBSET)); + assertNotColocated(f.allOf(NODE_SUBSET), f.oneOf(NODE_SET)); + assertColocated(f, f.allOf(SINGLE_NODE_SET), f.oneOf(NODE_SET), equalTo(SINGLE_NODE_SET)); + + // Colocation with Partitioned + assertNotColocated(f.allOf(NODE_SET), f.partitioned(assignmentFromPrimaries(NODE_SET)), + "AllOf target and Partitioned can't be colocated"); + assertNotColocated(f.allOf(SINGLE_NODE_SET), f.partitioned(assignmentFromPrimaries(NODE_SET)), + "AllOf target and Partitioned can't be colocated"); + assertNotColocated(f.allOf(SINGLE_NODE_SET), f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET)), + "AllOf target and Partitioned can't be colocated"); + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void someOfTargets(ExecutionTargetFactory f) throws Exception { + // Self colocation + assertColocated(f, f.someOf(NODE_SET), f.someOf(NODE_SET), equalTo(NODE_SET)); + assertColocated(f, f.someOf(NODE_SET), f.someOf(NODE_SUBSET), equalTo(NODE_SUBSET)); + assertColocated(f, f.someOf(NODE_SUBSET), f.someOf(NODE_SET), equalTo(NODE_SUBSET)); + assertNotColocated(f.someOf(SINGLE_NODE_SET), f.someOf(NODE_SET2)); // Disjoint sets. + + // Colocation with AllOf + assertColocated(f, f.someOf(NODE_SET), f.allOf(NODE_SET), equalTo(NODE_SET)); + assertColocated(f, f.someOf(NODE_SET), f.allOf(NODE_SUBSET), equalTo(NODE_SUBSET)); + assertNotColocated(f.someOf(NODE_SUBSET), f.allOf(NODE_SET)); + + // Colocation with OneOf + assertColocated(f, f.someOf(NODE_SET), f.oneOf(NODE_SET), containsSingleFrom(NODE_SET)); + assertColocated(f, f.someOf(NODE_SUBSET), f.oneOf(NODE_SET), containsSingleFrom(NODE_SUBSET)); + assertColocated(f, f.someOf(NODE_SET), f.oneOf(NODE_SUBSET), containsSingleFrom(NODE_SUBSET)); + assertNotColocated(f.someOf(SINGLE_NODE_SET), f.oneOf(NODE_SET2)); // Disjoint sets. + + // Colocation with Partitioned + assertColocated(f, f.someOf(NODE_SET), f.partitioned(assignmentFromPrimaries(NODE_SET)), equalTo(NODE_SET)); + assertColocated(f, f.someOf(NODE_SET), f.partitioned(assignmentFromPrimaries(NODE_SUBSET)), equalTo(NODE_SUBSET)); + assertNotColocated(f.someOf(NODE_SUBSET), f.partitioned(assignmentFromPrimaries(NODE_SET))); + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void oneOfTargets(ExecutionTargetFactory f) throws Exception { + // Self colocation + assertColocated(f, f.oneOf(NODE_SET), f.oneOf(NODE_SET), containsSingleFrom(NODE_SET)); + assertColocated(f, f.oneOf(NODE_SET), f.oneOf(shuffle(NODE_SET)), containsSingleFrom(NODE_SET)); + assertColocated(f, f.oneOf(NODE_SUBSET), f.oneOf(NODE_SET), containsSingleFrom(NODE_SUBSET)); + assertColocated(f, f.oneOf(NODE_SET), f.oneOf(NODE_SUBSET), containsSingleFrom(NODE_SUBSET)); + assertNotColocated(f.oneOf(NODE_SET2), f.oneOf(SINGLE_NODE_SET)); // Disjoint sets. + + // Colocation with AllOf + assertNotColocated(f.oneOf(NODE_SET), f.allOf(NODE_SET)); + assertColocated(f, f.oneOf(NODE_SET), f.allOf(SINGLE_NODE_SET), equalTo(SINGLE_NODE_SET)); + assertNotColocated(f.oneOf(NODE_SUBSET), f.allOf(NODE_SET)); + assertNotColocated(f.oneOf(NODE_SET2), f.allOf(SINGLE_NODE_SET)); // Disjoint sets. + + // Colocation with someOf + assertColocated(f, f.oneOf(NODE_SET), f.someOf(NODE_SET), containsSingleFrom(NODE_SET)); + assertColocated(f, f.oneOf(NODE_SUBSET), f.someOf(NODE_SET), containsSingleFrom(NODE_SUBSET)); + assertColocated(f, f.oneOf(NODE_SET), f.someOf(NODE_SUBSET), containsSingleFrom(NODE_SUBSET)); + assertNotColocated(f.oneOf(SINGLE_NODE_SET), f.someOf(NODE_SET2)); // Disjoint sets. + + // Colocation with Partitioned + assertNotColocated(f.oneOf(NODE_SET), f.partitioned(assignmentFromPrimaries(NODE_SET))); + assertColocated(f, f.oneOf(NODE_SET), f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET)), equalTo(SINGLE_NODE_SET)); + assertNotColocated(f.oneOf(NODE_SET2), f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET))); // Disjoint sets. + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void partitionedPrimaryTargets(ExecutionTargetFactory f) throws Exception { + // Self colocation + assertColocated(f, f.partitioned(assignmentFromPrimaries(NODE_SET)), f.partitioned(assignmentFromPrimaries(NODE_SET)), + equalTo(NODE_SET)); + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.partitioned(shuffle(assignmentFromPrimaries(NODE_SET)))); + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SUBSET)), f.partitioned(assignmentFromPrimaries(NODE_SET)), + "Partitioned targets with mot matching numbers of partitioned are not colocated"); + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.partitioned(assignmentFromPrimaries(NODE_SUBSET)), + "Partitioned targets with mot matching numbers of partitioned are not colocated"); + + assertNotColocated(f.partitioned(singleWithToken("node1", 1)), f.partitioned(singleWithToken("node1", 2)), + "Partitioned targets have different terms"); + + // Colocation with AllOf + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.allOf(NODE_SET), + "AllOf target and Partitioned can't be colocated"); + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.allOf(SINGLE_NODE_SET), + "AllOf target and Partitioned can't be colocated"); + + // Colocation with someOf + assertColocated(f, f.partitioned(assignmentFromPrimaries(NODE_SET)), f.someOf(NODE_SET), equalTo(NODE_SET)); + assertColocated(f, f.partitioned(assignmentFromPrimaries(NODE_SUBSET)), f.someOf(NODE_SET), equalTo(NODE_SUBSET)); + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.someOf(NODE_SUBSET)); + assertNotColocated(f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET)), f.someOf(NODE_SET2)); // Disjoint sets. + + // Colocation with oneOf + assertNotColocated(f.partitioned(assignmentFromPrimaries(NODE_SET)), f.oneOf(NODE_SET)); + assertColocated(f, f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET)), f.oneOf(NODE_SET), equalTo(SINGLE_NODE_SET)); + assertNotColocated(f.partitioned(assignmentFromPrimaries(SINGLE_NODE_SET)), f.oneOf(NODE_SET2)); // Disjoint + } + + @ParameterizedTest + @MethodSource("clusterFactory") + void partitionedWithBackupsTargets(ExecutionTargetFactory f) throws Exception { + // Self colocation + assertColocated(f, f.partitioned(assignment(NODE_SET, NODE_SET2)), f.partitioned(assignment(NODE_SET, NODE_SET2)), + hasItems(in(NODE_SUBSET))); + assertColocated(f, f.partitioned(assignment(NODE_SET, NODE_SET2)), f.partitioned(assignment(NODE_SET2, NODE_SET)), + hasItems(in(NODE_SUBSET))); + + // Colocation with AllOf + assertNotColocated(f.partitioned(assignment(NODE_SET, NODE_SET2)), f.allOf(NODE_SUBSET), + "AllOf target and Partitioned can't be colocated"); + assertNotColocated(f.partitioned(assignment(NODE_SET, NODE_SET2)), f.allOf(SINGLE_NODE_SET), + "AllOf target and Partitioned can't be colocated"); + + // Colocation with someOf + assertColocated(f, f.partitioned(assignment(NODE_SET, NODE_SET2)), f.someOf(NODE_SET), hasItems((in(NODE_SET)))); + assertNotColocated(f.partitioned(assignment(NODE_SET, NODE_SET2)), f.someOf(SINGLE_NODE_SET)); // Disjoint sets. + + // Colocation with oneOf + assertColocated(f, f.partitioned(assignment(NODE_SET, NODE_SET2)), f.someOf(NODE_SET), hasItems((in(NODE_SUBSET)))); + assertColocated(f, f.partitioned(assignment(NODE_SET, NODE_SET2)), f.oneOf(NODE_SET), containsSingleFrom(NODE_SUBSET)); + assertNotColocated(f.partitioned(assignment(NODE_SET, NODE_SET2)), f.oneOf(SINGLE_NODE_SET)); // Disjoint + } + + private static List<TokenizedAssignments> assignmentFromPrimaries(List<String> nodes) { + return nodes.stream() + .map(n -> new TokenizedAssignmentsImpl(Set.of(Assignment.forPeer(n)), 1)) + .collect(Collectors.toList()); + } + + private static List<TokenizedAssignments> assignment(List<String> part1Nodes, List<String> part2Nodes) { + Set<Assignment> part1 = part1Nodes.stream().map(Assignment::forPeer).collect(Collectors.toSet()); + Set<Assignment> part2 = part2Nodes.stream().map(Assignment::forPeer).collect(Collectors.toSet()); + + return List.of( + new TokenizedAssignmentsImpl(part1, 1), + new TokenizedAssignmentsImpl(part2, 2) + ); + } + + private static List<TokenizedAssignments> singleWithToken(String name, int token) { + return List.of(new TokenizedAssignmentsImpl(Set.of(Assignment.forPeer(name)), token)); + } + + private static <T> ArrayList<T> shuffle(List<T> nodeSetWithTokens) { + ArrayList<T> shuffled = new ArrayList<>(nodeSetWithTokens); + Collections.reverse(shuffled); + return shuffled; + } + + private static Matcher<Iterable<String>> containsSingleFrom(List<String> items) { + return allOf(iterableWithSize(1), hasItems(in(items))); + } + + private static void assertColocated( + ExecutionTargetFactory factory, + ExecutionTarget target, + ExecutionTarget other, + Matcher<Iterable<String>> matcher + ) throws ColocationMappingException { + assertThat(factory.resolveNodes(target.colocateWith(other)), matcher); + } + + private static void assertNotColocated(ExecutionTarget target, ExecutionTarget other) { + assertNotColocated(target, other, "Targets are not colocated"); + } + + @SuppressWarnings("ThrowableNotThrown") + private static void assertNotColocated(ExecutionTarget target, ExecutionTarget other, String errorMessageFragment) { + assertThrows(ColocationMappingException.class, () -> target.colocateWith(other), errorMessageFragment); + } +}