You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "PakhomovAlexander (via GitHub)" <gi...@apache.org> on 2024/08/21 21:01:25 UTC
[PR] IGNITE-22787 WIP [ignite-3]
PakhomovAlexander opened a new pull request, #4260:
URL: https://github.com/apache/ignite-3/pull/4260
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#issuecomment-2310497843
I've fixed .NET tests in #4286
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1731372146
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/task/DelegatingTaskExecution.java:
##########
@@ -60,4 +62,9 @@ public CompletableFuture<R> resultAsync() {
public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return delegate.thenCompose(execution -> execution.changePriorityAsync(newPriority));
}
+
+ @Override
+ public @Nullable Marshaller<R, byte[]> resultMarshaller() {
+ return delegate.join().resultMarshaller();
Review Comment:
Delegate future is supposed to be completed at this point, right? Let's add an assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1734795095
##########
modules/api/src/main/java/org/apache/ignite/compute/TaskDescriptor.java:
##########
@@ -56,6 +65,19 @@ public List<DeploymentUnit> units() {
return units;
}
+ /**
+ * Marshaller for split job argument.
+ *
+ * @return Marshaller for split job argument.
+ */
+ public Marshaller<T, byte[]> splitJobArgumentMarshaller() {
+ return splitJobArgumentMarshaller;
+ }
+
Review Comment:
```suggestion
/**
* Marshaller for reduce job result.
*
* @return Marshaller for reduce job result.
*/
```
##########
modules/api/src/main/java/org/apache/ignite/compute/TaskDescriptor.java:
##########
@@ -113,6 +137,21 @@ public Builder<T, R> units(DeploymentUnit... units) {
return this;
}
+ /**
+ * Sets the marshaller for split job argument.
+ *
+ * @param splitJobArgumentMarshaller Marshaller for split job argument.
+ * @return This builder.
+ */
+ public Builder<T, R> splitJobArgumentMarshaller(Marshaller<T, byte[]> splitJobArgumentMarshaller) {
+ this.splitJobArgumentMarshaller = splitJobArgumentMarshaller;
+ return this;
+ }
+
Review Comment:
```suggestion
/**
* Sets the marshaller for reduce job result.
*
* @param reduceJobResultMarshaller Marshaller for reduce job result.
* @return This builder.
*/
```
##########
modules/api/src/main/java/org/apache/ignite/compute/task/MapReduceTask.java:
##########
@@ -53,4 +54,14 @@ public interface MapReduceTask<I, M, T, R> {
* @return Final task result future.
*/
CompletableFuture<R> reduceAsync(TaskExecutionContext taskContext, Map<UUID, T> results);
+
+ /** The marshaller that is called to unmarshal split job argument if not null. */
+ default @Nullable Marshaller<I, byte[]> splitJobInputMarshaller() {
+ return null;
+ }
+
+ /** The marshaller that is called to marshal split job argument if not null. */
Review Comment:
```suggestion
/** The marshaller that is called to marshal reduce job result if not null. */
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander merged PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1734808082
##########
modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientComputeJobUnpacker.java:
##########
@@ -63,7 +63,7 @@ public final class ClientComputeJobUnpacker {
case NATIVE:
if (marshaller != null) {
throw new UnmarshallingException(
- "Can not unpack object because the marshaller is provided but the object was packed without marshaller."
+ "Can not unpack object because the marshaller is provided but the object were packed without marshaller."
Review Comment:
```suggestion
"Can not unpack object because the marshaller is provided but the object was packed without marshaller."
```
##########
modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientComputeJobUnpacker.java:
##########
@@ -74,7 +74,7 @@ public final class ClientComputeJobUnpacker {
case MARSHALLED_OBJECT:
if (marshaller == null) {
throw new UnmarshallingException(
- "Can not unpack object because the marshaller is not provided but the object was packed with marshaller."
+ "Can not unpack object because the marshaller is not provided but the object were packed with marshaller."
Review Comment:
```suggestion
"Can not unpack object because the marshaller is not provided but the object was packed with marshaller."
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1735590325
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java:
##########
@@ -82,11 +85,17 @@ static void packJobIds(ClientMessagePacker out, List<UUID> ids) {
}
static CompletableFuture<Object> sendTaskResult(TaskExecution<Object> execution, NotificationSender notificationSender) {
+ TaskExecution<Object> t = execution;
return execution.resultAsync().whenComplete((val, err) ->
- execution.stateAsync().whenComplete((state, errState) ->
+ t.stateAsync().whenComplete((state, errState) ->
execution.statesAsync().whenComplete((states, errStates) ->
notificationSender.sendNotification(w -> {
- w.packObjectAsBinaryTuple(val);
+ if (t instanceof MarshallerProvider) {
+ Marshaller<Object, byte[]> resultMarshaller = ((MarshallerProvider<Object>) t).resultMarshaller();
+ ClientComputeJobPacker.packJobResult(val, resultMarshaller, w);
+ } else {
+ throw new IllegalArgumentException(t.getClass() + " does not implement MarshallerProvider");
+ }
Review Comment:
```suggestion
Marshaller<Object, byte[]> resultMarshaller = ((MarshallerProvider<Object>) t).resultMarshaller();
ClientComputeJobPacker.packJobResult(val, resultMarshaller, w);
```
No need to replace `ClassCastException` with our own logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1731373599
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/task/TaskExecutionInternal.java:
##########
@@ -81,6 +84,8 @@ public class TaskExecutionInternal<I, M, T, R> implements TaskExecution<R> {
private final AtomicBoolean isCancelled;
+ private final AtomicReference<Marshaller<R, byte[]>> reduceResultMarshallerRef = new AtomicReference<>();
Review Comment:
We only use `get` and `set` from the `AtomicReference`, so it can be replaced with a `volatile` field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1731368818
##########
modules/api/src/main/java/org/apache/ignite/compute/task/MapReduceTask.java:
##########
@@ -53,4 +54,14 @@ public interface MapReduceTask<I, M, T, R> {
* @return Final task result future.
*/
CompletableFuture<R> reduceAsync(TaskExecutionContext taskContext, Map<UUID, T> results);
+
+ /** The marshaller that is called to unmarshall split job argument if not null. */
Review Comment:
Here and below:
* Marshaller (noun) - 2 `l`s
* To marshal (verb) - 1 `l`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1731478364
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java:
##########
@@ -82,11 +85,17 @@ static void packJobIds(ClientMessagePacker out, List<UUID> ids) {
}
static CompletableFuture<Object> sendTaskResult(TaskExecution<Object> execution, NotificationSender notificationSender) {
+ TaskExecution<Object> t = execution;
return execution.resultAsync().whenComplete((val, err) ->
- execution.stateAsync().whenComplete((state, errState) ->
+ t.stateAsync().whenComplete((state, errState) ->
execution.statesAsync().whenComplete((states, errStates) ->
notificationSender.sendNotification(w -> {
- w.packObjectAsBinaryTuple(val);
+ if (t instanceof MarshallerProvider) {
+ Marshaller<Object, byte[]> resultMarshaller = ((MarshallerProvider<Object>) t).resultMarshaller();
+ ClientComputeJobPacker.packJobResult(val, resultMarshaller, w);
+ } else {
+ w.packObjectAsBinaryTuple(val);
Review Comment:
In fact, it must be removed, otherwise the client protocol becomes broken.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22787 Support Marshallers in MapReduce API [ignite-3]
Posted by "ptupitsyn (via GitHub)" <gi...@apache.org>.
ptupitsyn commented on code in PR #4260:
URL: https://github.com/apache/ignite-3/pull/4260#discussion_r1731419059
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java:
##########
@@ -82,11 +85,17 @@ static void packJobIds(ClientMessagePacker out, List<UUID> ids) {
}
static CompletableFuture<Object> sendTaskResult(TaskExecution<Object> execution, NotificationSender notificationSender) {
+ TaskExecution<Object> t = execution;
return execution.resultAsync().whenComplete((val, err) ->
- execution.stateAsync().whenComplete((state, errState) ->
+ t.stateAsync().whenComplete((state, errState) ->
execution.statesAsync().whenComplete((states, errStates) ->
notificationSender.sendNotification(w -> {
- w.packObjectAsBinaryTuple(val);
+ if (t instanceof MarshallerProvider) {
+ Marshaller<Object, byte[]> resultMarshaller = ((MarshallerProvider<Object>) t).resultMarshaller();
+ ClientComputeJobPacker.packJobResult(val, resultMarshaller, w);
+ } else {
+ w.packObjectAsBinaryTuple(val);
Review Comment:
This branch is not reachable in practice, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org