IGNITE-22150 .NET: Add PartitionManager API

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

@@ -23,7 +23,6 @@
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -85,11 +84,6 @@ public CompletableFuture<Map<Partition, ClusterNode>> primaryReplicasAsync() {
                 w -> w.out().packInt(tbl.tableId()),
                 r -> {
                     ClientMessageUnpacker in = r.in();
-
-                    if (in.tryUnpackNil()) {
-                        return Collections.<Partition, ClusterNode>emptyMap();
-                    }
-
                     int size = in.unpackInt();
 
                     Map<Partition, ClusterNode> res = new HashMap<>(size);

modules/platforms/dotnet/Apache.Ignite.Tests/Compute/ComputeTests.cs

@@ -66,6 +66,8 @@ public class ComputeTests : IgniteTestsBase
 
         public static readonly JobDescriptor<string, object> CheckedExceptionJob = new(PlatformTestNodeRunner + "$CheckedExceptionJob");
 
+        public static readonly JobDescriptor<long, int> PartitionJob = new(PlatformTestNodeRunner + "$PartitionJob");
+
         [Test]
         public async Task TestGetClusterNodes()
         {

modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs

@@ -377,6 +377,29 @@ protected override void Handle(Socket handler, CancellationToken cancellationTok
                         Send(handler, requestId, Array.Empty<byte>());
                         continue;
                     }
+
+                    case ClientOp.PrimaryReplicasGet:
+                    {
+                        using var arrayBufferWriter = new PooledArrayBuffer();
+                        var writer = new MsgPackWriter(arrayBufferWriter);
+
+                        writer.Write(PartitionAssignment.Length);
+
+                        for (var index = 0; index < PartitionAssignment.Length; index++)
+                        {
+                            var nodeId = PartitionAssignment[index];
+
+                            writer.Write(index); // Partition id.
+                            writer.Write(4); // Prop count.
+                            writer.Write(nodeId); // Id.
+                            writer.Write(nodeId); // Name.
+                            writer.Write("localhost"); // Host.
+                            writer.Write(10900 + index); // Port.
+                        }
+
+                        Send(handler, requestId, arrayBufferWriter);
+                        continue;
+                    }
                 }
 
                 // Fake error message for any other op code.

modules/platforms/dotnet/Apache.Ignite.Tests/IgniteTestsBase.cs

@@ -35,6 +35,7 @@ namespace Apache.Ignite.Tests
     public class IgniteTestsBase
     {
         protected const string TableName = "TBL1";
+        protected const int TablePartitionCount = 10;
 
         protected const string TableAllColumnsName = "TBL_ALL_COLUMNS";
         protected const string TableAllColumnsNotNullName = "TBL_ALL_COLUMNS_NOT_NULL";

modules/platforms/dotnet/Apache.Ignite.Tests/Table/PartitionManagerTests.cs

@@ -0,0 +1,167 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Tests.Table;
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Compute;
+using Ignite.Compute;
+using Ignite.Table;
+using Internal.Table;
+using NUnit.Framework;
+
+/// <summary>
+/// Tests for <see cref="IPartitionManager"/>.
+/// </summary>
+public class PartitionManagerTests : IgniteTestsBase
+{
+    [Test]
+    public async Task TestGetPrimaryReplicas()
+    {
+        var replicas = await Table.PartitionManager.GetPrimaryReplicasAsync();
+        var replicasNodes = replicas.Values.Distinct().OrderBy(x => x.Address.Port).ToList();
+        var replicasPartitions = replicas.Keys.Select(x => ((HashPartition)x).PartitionId).OrderBy(x => x).ToList();
+
+        var expectedNodes = (await Client.GetClusterNodesAsync()).OrderBy(x => x.Address.Port).ToList();
+
+        CollectionAssert.AreEqual(expectedNodes, replicasNodes, "Primary replicas should be distributed among all nodes");
+
+        CollectionAssert.AreEqual(
+            Enumerable.Range(0, TablePartitionCount),
+            replicasPartitions,
+            "Primary replicas map should have all partitions");
+    }
+
+    [Test]
+    public async Task TestGetPrimaryReplicasReturnsCachedPartitionInstances()
+    {
+        var partitions1 = await GetPartitions();
+        var partitions2 = await GetPartitions();
+
+        for (int i = 0; i < partitions1.Count; i++)
+        {
+            Assert.AreSame(partitions1[i], partitions2[i]);
+        }
+
+        async Task<List<HashPartition>> GetPartitions()
+        {
+            var replicas = await Table.PartitionManager.GetPrimaryReplicasAsync();
+            return replicas.Keys.Cast<HashPartition>().OrderBy(x => x.PartitionId).ToList();
+        }
+    }
+
+    [Test]
+    public async Task TestGetPrimaryReplica()
+    {
+        var nodes = await Client.GetClusterNodesAsync();
+
+        for (int partId = 0; partId < TablePartitionCount; partId++)
+        {
+            var partition = new HashPartition(partId);
+            var replica = await Table.PartitionManager.GetPrimaryReplicaAsync(partition);
+
+            CollectionAssert.Contains(nodes, replica);
+        }
+    }
+
+    [Test]
+    public void TestGetPrimaryReplicaNegativePartitionIdThrows()
+    {
+        var ex = Assert.ThrowsAsync<ArgumentException>(
+            async () => await Table.PartitionManager.GetPrimaryReplicaAsync(new HashPartition(-1)));
+
+        Assert.AreEqual("Partition id can't be negative: HashPartition { PartitionId = -1 }", ex.Message);
+    }
+
+    [Test]
+    public void TestGetPrimaryReplicaPartitionIdOutOfRangeThrows()
+    {
+        var ex = Assert.ThrowsAsync<ArgumentException>(
+            async () => await Table.PartitionManager.GetPrimaryReplicaAsync(new HashPartition(10)));
+
+        Assert.AreEqual("Partition id can't be greater than 9: HashPartition { PartitionId = 10 }", ex.Message);
+    }
+
+    [Test]
+    public void TestGetPrimaryReplicaUnknownPartitionClassThrows()
+    {
+        var ex = Assert.ThrowsAsync<ArgumentException>(
+            async () => await Table.PartitionManager.GetPrimaryReplicaAsync(new MyPartition()));
+
+        Assert.AreEqual($"Unsupported partition type: {typeof(MyPartition)}", ex.Message);
+    }
+
+    [Test]
+    public async Task TestGetPartitionForKey([Values(true, false)] bool poco)
+    {
+        var jobTarget = JobTarget.AnyNode(await Client.GetClusterNodesAsync());
+
+        for (int id = 0; id < 30; id++)
+        {
+            var partition = poco
+                ? await Table.PartitionManager.GetPartitionAsync(GetPoco(id))
+                : await Table.PartitionManager.GetPartitionAsync(GetTuple(id));
+
+            var partitionJobExec = await Client.Compute.SubmitAsync(jobTarget, ComputeTests.PartitionJob, id);
+            var expectedPartition = await partitionJobExec.GetResultAsync();
+
+            Assert.AreEqual(expectedPartition, ((HashPartition)partition).PartitionId);
+        }
+    }
+
+    [Test]
+    public async Task TestGetPartitionReturnsCachedInstance()
+    {
+        var partition1 = await Table.PartitionManager.GetPartitionAsync(GetTuple(1));
+        var partition2 = await Table.PartitionManager.GetPartitionAsync(GetTuple(1));
+
+        Assert.AreSame(partition1, partition2);
+    }
+
+    [Test]
+    public async Task TestPrimaryReplicaCacheInvalidation()
+    {
+        using var server = new FakeServer
+        {
+            PartitionAssignmentTimestamp = 123,
+            PartitionAssignment = new[] { "n1", "n2" }
+        };
+
+        using var client = await server.ConnectClientAsync();
+        var table = await client.Tables.GetTableAsync(FakeServer.ExistingTableName);
+        var partition = new HashPartition(0);
+
+        var replica1 = await table!.PartitionManager.GetPrimaryReplicaAsync(partition);
+        Assert.AreEqual("n1", replica1.Name);
+
+        server.PartitionAssignmentTimestamp = 124;
+        server.PartitionAssignment = new[] { "n2", "n1" };
+
+        await client.Tables.GetTablesAsync(); // Trigger cache invalidation with any response.
+
+        var replica2 = await table.PartitionManager.GetPrimaryReplicaAsync(partition);
+        Assert.AreEqual("n2", replica2.Name);
+    }
+
+    private class MyPartition : IPartition
+    {
+        // No-op.
+    }
+}

modules/platforms/dotnet/Apache.Ignite/ClientOperationType.cs

@@ -146,6 +146,11 @@ public enum ClientOperationType
         /// </summary>
         ComputeChangePriority,
 
+        /// <summary>
+        /// Get primary replicas (<see cref="IPartitionManager.GetPrimaryReplicasAsync"/>).
+        /// </summary>
+        PrimaryReplicasGet,
+
         /// <summary>
         /// Send data streamer batch (<see cref="IDataStreamerTarget{T}"/>).
         /// </summary>

modules/platforms/dotnet/Apache.Ignite/Internal/IgniteClientInternal.cs

@@ -88,13 +88,7 @@ IList<IClusterNode> Read()
 
                 for (var i = 0; i < count; i++)
                 {
-                    var fieldCount = r.ReadInt32();
-                    Debug.Assert(fieldCount == 4, "fieldCount == 4");
-
-                    res.Add(new ClusterNode(
-                        id: r.ReadString(),
-                        name: r.ReadString(),
-                        endpoint: new IPEndPoint(IPAddress.Parse(r.ReadString()), r.ReadInt32())));
+                    res.Add(ClusterNode.Read(ref r));
                 }
 
                 return res;

modules/platforms/dotnet/Apache.Ignite/Internal/Network/ClusterNode.cs

@@ -18,8 +18,11 @@
 namespace Apache.Ignite.Internal.Network
 {
     using System;
+    using System.Diagnostics;
+    using System.Linq;
     using System.Net;
     using Ignite.Network;
+    using Proto.MsgPack;
 
     /// <summary>
     /// Cluster node.
@@ -74,5 +77,30 @@ public bool Equals(ClusterNode? other)
 
         /// <inheritdoc/>
         public override int GetHashCode() => HashCode.Combine(Id, Name, Address);
+
+        /// <summary>
+        /// Read node from reader.
+        /// </summary>
+        /// <param name="r">Reader.</param>
+        /// <returns>Cluster node.</returns>
+        internal static ClusterNode Read(ref MsgPackReader r)
+        {
+            var fieldCount = r.ReadInt32();
+            Debug.Assert(fieldCount == 4, "fieldCount == 4");
+
+            var id = r.ReadString();
+            var name = r.ReadString();
+            var addr = r.ReadString();
+            var port = r.ReadInt32();
+
+            // TODO IGNITE-22695 .NET: ClusterNode.Address does not support host names
+            var ipAddress = IPAddress.TryParse(addr, out var ip)
+                ? ip
+                : Dns.GetHostEntry(addr).AddressList.FirstOrDefault() ?? IPAddress.Loopback;
+
+            var endPoint = new IPEndPoint(ipAddress, port);
+
+            return new ClusterNode(id, name, endPoint);
+        }
     }
 }

modules/platforms/dotnet/Apache.Ignite/Internal/Proto/ClientOp.cs

@@ -133,6 +133,9 @@ internal enum ClientOp
         /** Send streamer batch. */
         StreamerBatchSend = 62,
 
+        /** Get primary replicas. */
+        PrimaryReplicasGet = 65,
+
         /** Send streamer batch with receiver. */
         StreamerWithReceiverBatchSend = 66
     }

modules/platforms/dotnet/Apache.Ignite/Internal/Proto/ClientOpExtensions.cs

@@ -69,6 +69,7 @@ internal static class ClientOpExtensions
                 ClientOp.PartitionAssignmentGet => null,
                 ClientOp.SqlParamMeta => null,
                 ClientOp.StreamerBatchSend => ClientOperationType.StreamerBatchSend,
+                ClientOp.PrimaryReplicasGet => ClientOperationType.PrimaryReplicasGet,
                 ClientOp.StreamerWithReceiverBatchSend => ClientOperationType.StreamerBatchSend,
 
                 // Do not return null from default arm intentionally so we don't forget to update this when new ClientOp values are added.

modules/platforms/dotnet/Apache.Ignite/Internal/Table/HashPartition.cs

@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Internal.Table;
+
+using Ignite.Table;
+
+/// <summary>
+/// Hash partition.
+/// </summary>
+/// <param name="PartitionId">Partition id.</param>
+internal sealed record HashPartition(int PartitionId) : IPartition; // Not a struct to avoid interface boxing.