IGNITE-23601 Create broadcast partitioned compute method

modules/api/src/main/java/org/apache/ignite/compute/IgniteCompute.java

@@ -244,6 +244,102 @@ default <T, R> Map<ClusterNode, R> executeBroadcast(
         return map;
     }
 
+    /**
+     * Submits a {@link ComputeJob} of the given class for an execution on nodes where the primary replicas of a partitions of the specified
+     * table are located. The partition objects are passed to the job in the {@link JobExecutionContext#partitions()}. At most one job is
+     * executed on a node.
+     *
+     * @param tableName Table name.
+     * @param descriptor Job descriptor.
+     * @param arg Job argument.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
+     * @return Future that will be completed with the map from the node to the job execution objects when jobs are submitted for the
+     *         execution.
+     */
+    <T, R> CompletableFuture<Map<ClusterNode, JobExecution<R>>> submitBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable T arg
+    );
+
+    /**
+     * Executes a {@link ComputeJob} of the given class on nodes where the primary replicas of a partitions of the specified table are
+     * located. The partition objects are passed to the job in the {@link JobExecutionContext#partitions()}. At most one job is executed on
+     * a node.
+     *
+     * @param tableName Table name.
+     * @param descriptor Job descriptor.
+     * @param arg Job argument.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
+     * @return Future that will be completed with the map from the node to the job execution results.
+     */
+    default <T, R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastPartitionedAsync(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable T arg
+    ) {
+        return executeBroadcastPartitionedAsync(tableName, descriptor, null, arg);
+    }
+
+    /**
+     * Executes a {@link ComputeJob} of the given class on nodes where the primary replicas of a partitions of the specified table are
+     * located. The partition objects are passed to the job in the {@link JobExecutionContext#partitions()}. At most one job is executed on
+     * a node.
+     *
+     * @param tableName Table name.
+     * @param descriptor Job descriptor.
+     * @param arg Job argument.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
+     * @return Future that will be completed with the map from the node to the job execution results.
+     */
+    <T, R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastPartitionedAsync(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    );
+
+    /**
+     * Executes a {@link ComputeJob} of the given class on nodes where the primary replicas of a partitions of the specified table are
+     * located. The partition objects are passed to the job in the {@link JobExecutionContext#partitions()}. At most one job is executed on
+     * a node.
+     *
+     * @param tableName Table name.
+     * @param descriptor Job descriptor.
+     * @param arg Job argument.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
+     * @return Map from the node to the job execution results.
+     */
+    default <T, R> Map<ClusterNode, R> executeBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable T arg
+    ) {
+        return executeBroadcastPartitioned(tableName, descriptor, null, arg);
+    }
+
+    /**
+     * Executes a {@link ComputeJob} of the given class on nodes where the primary replicas of a partitions of the specified table are
+     * located. The partition indices are passed to the job in the {@link JobExecutionContext#partitions()}.
+     *
+     * @param tableName Table name.
+     * @param descriptor Job descriptor.
+     * @param arg Job argument.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
+     * @return Map from the node to the job execution results.
+     */
+    <T, R> Map<ClusterNode, R> executeBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    );
+
     /**
      * Submits a {@link MapReduceTask} of the given class for an execution.
      *

modules/api/src/main/java/org/apache/ignite/compute/JobExecutionContext.java

@@ -17,7 +17,10 @@
 
 package org.apache.ignite.compute;
 
+import java.util.List;
 import org.apache.ignite.Ignite;
+import org.apache.ignite.table.partition.Partition;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Context of the {@link ComputeJob} execution.
@@ -36,4 +39,13 @@ public interface JobExecutionContext {
      * @return {@code true} when the job was cancelled.
      */
     boolean isCancelled();
+
+    /**
+     * List of partitions associated with this job. Not a {@code null} only when
+     * {@link IgniteCompute#submitBroadcastPartitioned(String, JobDescriptor, Object)} method is used to submit jobs. In this case, the list
+     * contains partitions that are local on the node executing the job.
+     *
+     * @return list of partitions associated with this job.
+     */
+    @Nullable List<Partition> partitions();
 }

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

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.client.compute;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
 import static org.apache.ignite.lang.ErrorGroups.Client.TABLE_ID_NOT_FOUND_ERR;
 
 import java.util.ArrayList;
@@ -236,6 +237,39 @@ public <T, R> R executeMapReduce(TaskDescriptor<T, R> taskDescriptor, @Nullable
         return sync(executeMapReduceAsync(taskDescriptor, cancellationToken, arg));
     }
 
+    @Override
+    public <T, R> CompletableFuture<Map<ClusterNode, JobExecution<R>>> submitBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable T arg
+    ) {
+        Objects.requireNonNull(tableName);
+        Objects.requireNonNull(descriptor);
+        // TODO https://issues.apache.org/jira/browse/IGNITE-23936
+        return failedFuture(new UnsupportedOperationException("Not implemented"));
+    }
+
+    @Override
+    public <T, R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastPartitionedAsync(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    ) {
+        // TODO https://issues.apache.org/jira/browse/IGNITE-23936
+        return failedFuture(new UnsupportedOperationException("Not implemented"));
+    }
+
+    @Override
+    public <T, R> Map<ClusterNode, R> executeBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    ) {
+        return sync(executeBroadcastPartitionedAsync(tableName, descriptor, cancellationToken, arg));
+    }
+
     private <T, R> CompletableFuture<SubmitTaskResult> doExecuteMapReduceAsync(TaskDescriptor<T, R> taskDescriptor, @Nullable T arg) {
         return ch.serviceAsync(
                 ClientOp.COMPUTE_EXECUTE_MAPREDUCE,

modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.client.fakes;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
 import static org.apache.ignite.compute.JobStatus.COMPLETED;
 import static org.apache.ignite.compute.JobStatus.EXECUTING;
 import static org.apache.ignite.compute.JobStatus.FAILED;
@@ -122,7 +123,7 @@ public <R> JobExecution<R> executeAsyncWithFailover(
             Class<ComputeJob<Object, R>> jobClass = ComputeUtils.jobClass(jobClassLoader, jobClassName);
             ComputeJob<Object, R> job = ComputeUtils.instantiateJob(jobClass);
             CompletableFuture<R> jobFut = job.executeAsync(
-                    new JobExecutionContextImpl(ignite, new AtomicBoolean(), this.getClass().getClassLoader()), args);
+                    new JobExecutionContextImpl(ignite, new AtomicBoolean(), this.getClass().getClassLoader(), null), args);
 
             return jobExecution(jobFut != null ? jobFut : nullCompletedFuture());
         }
@@ -201,6 +202,35 @@ public <T, R> R executeMapReduce(TaskDescriptor<T, R> taskDescriptor, @Nullable
         return sync(executeMapReduceAsync(taskDescriptor, cancellationToken, arg));
     }
 
+    @Override
+    public <T, R> CompletableFuture<Map<ClusterNode, JobExecution<R>>> submitBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable T arg
+    ) {
+        return failedFuture(new UnsupportedOperationException("Not implemented"));
+    }
+
+    @Override
+    public <T, R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastPartitionedAsync(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    ) {
+        return failedFuture(new UnsupportedOperationException("Not implemented"));
+    }
+
+    @Override
+    public <T, R> Map<ClusterNode, R> executeBroadcastPartitioned(
+            String tableName,
+            JobDescriptor<T, R> descriptor,
+            @Nullable CancellationToken cancellationToken,
+            @Nullable T arg
+    ) {
+        return sync(executeBroadcastPartitionedAsync(tableName, descriptor, cancellationToken, arg));
+    }
+
     private <R> JobExecution<R> completedExecution(R result) {
         return jobExecution(completedFuture(result));
     }

modules/compute/jobs.gradle

@@ -82,6 +82,7 @@ processIntegrationTestResources {
 
 dependencies {
     jobsImplementation project(':ignite-api')
+    jobsImplementation project(':ignite-core')
     unit1Implementation project(':ignite-api')
     unit2Implementation project(':ignite-api')