IGNITE-27012 Partition API improvements.

modules/api/src/main/java/org/apache/ignite/table/Table.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.table;
 
 import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.table.partition.PartitionDistribution;
 import org.apache.ignite.table.partition.PartitionManager;
 
 /**
@@ -59,8 +60,16 @@ default String name() {
      *
      * @return Partition manager.
      */
+    @Deprecated
     PartitionManager partitionManager();
 
+    /**
+     * Gets the partition distribution.
+     *
+     * @return Partition distribution.
+     */
+    PartitionDistribution partitionDistribution();
+
     /**
      * Gets a record view of the table using the specified record class mapper.
      *

modules/api/src/main/java/org/apache/ignite/table/partition/Partition.java

@@ -20,8 +20,9 @@
 import java.io.Serializable;
 
 /**
- * Marker interface which represents a partition reference.
+ * Interface which represents a partition reference.
  */
 public interface Partition extends Serializable {
-
+    /** ID of the partition. */
+    int partitionId();
 }

modules/api/src/main/java/org/apache/ignite/table/partition/PartitionDistribution.java

@@ -0,0 +1,64 @@
+/*
+ * 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.table.partition;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.mapper.Mapper;
+
+/**
+ * The partition distribution provides the ability to obtain information about table partitions.
+ * This interface can be used to get all partitions of a table, the location of the primary replica of a partition,
+ * the partition for a specific table key.
+ */
+public interface PartitionDistribution {
+    /**
+     * Returns location of primary replica for provided partition.
+     *
+     * @param partition Partition instance.
+     * @return Cluster node where primary replica of provided partition is located.
+     */
+    CompletableFuture<ClusterNode> primaryReplicaAsync(Partition partition);
+
+    /**
+     * Returns map with all partitions and their locations.
+     *
+     * @return Map from partition to cluster node where primary replica of the partition is located.
+     */
+    CompletableFuture<Map<Partition, ClusterNode>> primaryReplicasAsync();
+
+    /**
+     * Returns partition instance for provided table key.
+     *
+     * @param key Table key.
+     * @param mapper Table key mapper.
+     * @param <K> Key type.
+     * @return Partition instance which contains provided key.
+     */
+    <K> CompletableFuture<Partition> partitionAsync(K key, Mapper<K> mapper);
+
+    /**
+     * Returns partition instance for provided table key.
+     *
+     * @param key Table key tuple.
+     * @return Partition instance which contains provided key.
+     */
+    CompletableFuture<Partition> partitionAsync(Tuple key);
+}

modules/api/src/main/java/org/apache/ignite/table/partition/PartitionManager.java

@@ -17,49 +17,12 @@
 
 package org.apache.ignite.table.partition;
 
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.network.ClusterNode;
-import org.apache.ignite.table.Tuple;
-import org.apache.ignite.table.mapper.Mapper;
-
 /**
- * The partition manager provides the ability to obtain information about table partitions.
- * This interface can be used to get all partitions of a table,
- * the location of the primary replica of a partition,
- * the partition for a specific table key.
+ * The partition manager provides the ability to obtain information about table partitions. This interface can be used to get all partitions
+ * of a table, the location of the primary replica of a partition, the partition for a specific table key.
+ *
+ * @deprecated Replaced by {@link PartitionDistribution}.
  */
-public interface PartitionManager {
-    /**
-     * Returns location of primary replica for provided partition.
-     *
-     * @param partition Partition instance.
-     * @return Cluster node where primary replica of provided partition is located.
-     */
-    CompletableFuture<ClusterNode> primaryReplicaAsync(Partition partition);
-
-    /**
-     * Returns map with all partitions and their locations.
-     *
-     * @return Map from partition to cluster node where primary replica of the partition is located.
-     */
-    CompletableFuture<Map<Partition, ClusterNode>> primaryReplicasAsync();
-
-    /**
-     * Returns partition instance for provided table key.
-     *
-     * @param key Table key.
-     * @param mapper Table key mapper.
-     * @param <K> Key type.
-     * @return Partition instance which contains provided key.
-     */
-    <K> CompletableFuture<Partition> partitionAsync(K key, Mapper<K> mapper);
-
-    /**
-     * Returns partition instance for provided table key.
-     *
-     * @param key Table key tuple.
-     * @return Partition instance which contains provided key.
-     */
-    CompletableFuture<Partition> partitionAsync(Tuple key);
+@Deprecated(since = "3.2")
+public interface PartitionManager extends PartitionDistribution {
 }

modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/partition/ClientTablePartitionPrimaryReplicasNodesGetRequest.java

@@ -25,7 +25,6 @@
 import org.apache.ignite.client.handler.ResponseWriter;
 import org.apache.ignite.client.handler.requests.table.ClientTablePartitionPrimaryReplicasGetRequest;
 import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
-import org.apache.ignite.internal.table.partition.HashPartition;
 import org.apache.ignite.network.ClusterNode;
 import org.apache.ignite.table.IgniteTables;
 import org.apache.ignite.table.partition.Partition;
@@ -54,9 +53,7 @@ public static CompletableFuture<ResponseWriter> process(
                 .thenApply(partitions -> out -> {
                     out.packInt(partitions.size());
                     for (Entry<Partition, ClusterNode> e : partitions.entrySet()) {
-                        HashPartition partition = (HashPartition) e.getKey();
-
-                        out.packInt(partition.partitionId());
+                        out.packInt(e.getKey().partitionId());
 
                         packClusterNode(e.getValue(), out);
                     }

modules/client/src/main/java/org/apache/ignite/client/ClientOperationType.java

@@ -26,7 +26,7 @@
 import org.apache.ignite.table.DataStreamerTarget;
 import org.apache.ignite.table.IgniteTables;
 import org.apache.ignite.table.RecordView;
-import org.apache.ignite.table.partition.PartitionManager;
+import org.apache.ignite.table.partition.PartitionDistribution;
 import org.apache.ignite.tx.Transaction;
 
 /**
@@ -196,7 +196,7 @@ public enum ClientOperationType {
     SQL_EXECUTE_BATCH,
 
     /**
-     * Get all primary replicas mapping to cluster nodes ({@link PartitionManager#primaryReplicasAsync()}).
+     * Get all primary replicas mapping to cluster nodes ({@link PartitionDistribution#primaryReplicasAsync()}).
      */
     PRIMARY_REPLICAS_GET
 }

modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java

@@ -66,7 +66,6 @@
 import org.apache.ignite.internal.client.table.PartitionAwarenessProvider;
 import org.apache.ignite.internal.compute.BroadcastJobExecutionImpl;
 import org.apache.ignite.internal.compute.FailedExecution;
-import org.apache.ignite.internal.table.partition.HashPartition;
 import org.apache.ignite.internal.util.ExceptionUtils;
 import org.apache.ignite.internal.util.ViewUtils;
 import org.apache.ignite.lang.CancelHandleHelper;
@@ -445,7 +444,7 @@ private static <T, R> CompletableFuture<SubmitResult> executePartitioned(
             UUID taskId,
             @Nullable T arg
     ) {
-        int partitionId = ((HashPartition) partition).partitionId();
+        int partitionId = partition.partitionId();
         return t.doSchemaOutOpAsync(
                 ClientOp.COMPUTE_EXECUTE_PARTITIONED,
                 (schema, outputChannel, unused) -> {

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientPartitionManager.java

@@ -55,10 +55,6 @@ class ClientPartitionManager implements PartitionManager {
 
     @Override
     public CompletableFuture<ClusterNode> primaryReplicaAsync(Partition partition) {
-        if (!(partition instanceof HashPartition)) {
-            throw new IllegalArgumentException("Unsupported partition type: " + partition);
-        }
-
         ClusterNode clusterNode = getClusterNode(partition);
 
         if (clusterNode != null) {

modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java

@@ -72,6 +72,7 @@
 import org.apache.ignite.table.Table;
 import org.apache.ignite.table.Tuple;
 import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.table.partition.PartitionDistribution;
 import org.apache.ignite.table.partition.PartitionManager;
 import org.apache.ignite.tx.Transaction;
 import org.apache.ignite.tx.TransactionOptions;
@@ -169,6 +170,11 @@ public PartitionManager partitionManager() {
         return clientPartitionManager;
     }
 
+    @Override
+    public PartitionDistribution partitionDistribution() {
+        return clientPartitionManager;
+    }
+
     /** {@inheritDoc} */
     @Override
     public <R> RecordView<R> recordView(Mapper<R> recMapper) {

modules/client/src/test/java/org/apache/ignite/client/ClientPartitionManagerTest.java

@@ -29,7 +29,7 @@
 import org.apache.ignite.network.ClusterNode;
 import org.apache.ignite.table.Table;
 import org.apache.ignite.table.partition.Partition;
-import org.apache.ignite.table.partition.PartitionManager;
+import org.apache.ignite.table.partition.PartitionDistribution;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -55,7 +55,7 @@ public void setUp() {
     @Test
     public void testPrimaryReplicasCacheInvalidation() {
         Table table = client.tables().table(TABLE_NAME);
-        PartitionManager partMgr = table.partitionManager();
+        PartitionDistribution partMgr = table.partitionDistribution();
         HashPartition part0 = new HashPartition(0);
         HashPartition part2 = new HashPartition(2);
 

modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeBaseTest.java

@@ -80,7 +80,6 @@
 import org.apache.ignite.deployment.DeploymentUnit;
 import org.apache.ignite.internal.ClusterPerClassIntegrationTest;
 import org.apache.ignite.internal.ConfigOverride;
-import org.apache.ignite.internal.table.partition.HashPartition;
 import org.apache.ignite.internal.testframework.IgniteTestUtils;
 import org.apache.ignite.lang.CancelHandle;
 import org.apache.ignite.lang.CancellationToken;
@@ -807,7 +806,7 @@ void partitionedBroadcast() {
 
         Map<Partition, ClusterNode> replicas = node(0).tables().table("test").partitionManager().primaryReplicasAsync().join();
         Map<Integer, ClusterNode> partitionIdToNode = replicas.entrySet().stream()
-                .collect(toMap(entry -> ((HashPartition) entry.getKey()).partitionId(), Entry::getValue));
+                .collect(toMap(entry -> entry.getKey().partitionId(), Entry::getValue));
 
         // When run job that will return its partition id
         JobDescriptor<Void, Integer> job = JobDescriptor.builder(GetPartitionJob.class).units(units()).build();

modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/InteractiveJobs.java

@@ -36,7 +36,6 @@
 import org.apache.ignite.compute.JobExecutionContext;
 import org.apache.ignite.internal.network.ClusterNodeImpl;
 import org.apache.ignite.internal.network.InternalClusterNode;
-import org.apache.ignite.internal.table.partition.HashPartition;
 import org.apache.ignite.network.ClusterNode;
 
 /**
@@ -261,7 +260,7 @@ public CompletableFuture<String> executeAsync(JobExecutionContext context, Strin
                         case RETURN_WORKER_NAME:
                             return completedFuture(workerNodeName);
                         case RETURN_PARTITION_ID:
-                            return completedFuture(Integer.toString(((HashPartition) context.partition()).partitionId()));
+                            return completedFuture(Integer.toString(context.partition().partitionId()));
                         case GET_WORKER_NAME:
                             NODE_CHANNELS.get(workerNodeName).add(workerNodeName);
                             break;

modules/compute/src/jobs/java/org/apache/ignite/internal/compute/GetPartitionJob.java

@@ -22,12 +22,11 @@
 import java.util.concurrent.CompletableFuture;
 import org.apache.ignite.compute.ComputeJob;
 import org.apache.ignite.compute.JobExecutionContext;
-import org.apache.ignite.internal.table.partition.HashPartition;
 
 /** Compute job that returns a partition id for the specified partition passed in the context. */
 public class GetPartitionJob implements ComputeJob<Void, Integer> {
     @Override
     public CompletableFuture<Integer> executeAsync(JobExecutionContext context, Void arg) {
-        return completedFuture(((HashPartition) context.partition()).partitionId());
+        return completedFuture(context.partition().partitionId());
     }
 }

modules/compute/src/main/java/org/apache/ignite/internal/compute/PartitionNextWorkerSelector.java

@@ -21,7 +21,6 @@
 import org.apache.ignite.internal.network.TopologyService;
 import org.apache.ignite.internal.placementdriver.PlacementDriver;
 import org.apache.ignite.internal.replicator.ZonePartitionId;
-import org.apache.ignite.internal.table.partition.HashPartition;
 import org.apache.ignite.table.partition.Partition;
 
 /**
@@ -40,7 +39,7 @@ class PartitionNextWorkerSelector extends PrimaryReplicaNextWorkerSelector {
     ) {
         super(placementDriver, topologyService, clock);
 
-        this.partitionGroupId = new ZonePartitionId(zoneId, ((HashPartition) partition).partitionId());
+        this.partitionGroupId = new ZonePartitionId(zoneId, partition.partitionId());
     }
 
     @Override

modules/core/src/main/java/org/apache/ignite/internal/table/partition/HashPartition.java

@@ -32,6 +32,7 @@ public HashPartition(int partitionId) {
         this.partitionId = partitionId;
     }
 
+    @Override
     public int partitionId() {
         return partitionId;
     }

modules/runner/src/integrationTest/java/org/apache/ignite/internal/app/AsyncApiOperation.java

@@ -117,10 +117,10 @@ enum AsyncApiOperation {
 
     MAPPED_RECORD_VIEW_GET(refs -> refs.mappedRecordView.getAsync(null, new Record(1, ""))),
 
-    PARTITION_MANAGER_PRIMARY_REPLICA(refs -> refs.partitionManager.primaryReplicaAsync(new HashPartition(0))),
-    PARTITION_MANAGER_PRIMARY_REPLICAS(refs -> refs.partitionManager.primaryReplicasAsync()),
-    PARTITION_MANAGER_PARTITION_BY_KEY(refs -> refs.partitionManager.partitionAsync(1, Mapper.of(Integer.class))),
-    PARTITION_MANAGER_PARTITION_BY_TUPLE(refs -> refs.partitionManager.partitionAsync(KEY_TUPLE)),
+    PARTITION_MANAGER_PRIMARY_REPLICA(refs -> refs.partitionDistribution.primaryReplicaAsync(new HashPartition(0))),
+    PARTITION_MANAGER_PRIMARY_REPLICAS(refs -> refs.partitionDistribution.primaryReplicasAsync()),
+    PARTITION_MANAGER_PARTITION_BY_KEY(refs -> refs.partitionDistribution.partitionAsync(1, Mapper.of(Integer.class))),
+    PARTITION_MANAGER_PARTITION_BY_TUPLE(refs -> refs.partitionDistribution.partitionAsync(KEY_TUPLE)),
 
     TRANSACTIONS_BEGIN(refs -> refs.transactions.beginAsync()),
     TRANSACTIONS_BEGIN_WITH_OPTS(refs -> refs.transactions.beginAsync(null)),