This is an automated email from the ASF dual-hosted git repository.
shishkovilja pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 4aaecc77230 IGNITE-28977 Remove SelfMarshallingMessage implementation
by ColocationGroup (#13468)
4aaecc77230 is described below
commit 4aaecc77230f5ea8c4a7666269a6dd787a59044b
Author: Vladimir Steshin <[email protected]>
AuthorDate: Thu Aug 13 15:39:58 2026 +0300
IGNITE-28977 Remove SelfMarshallingMessage implementation by
ColocationGroup (#13468)
---
.../query/calcite/message/QueryStartRequest.java | 4 +--
.../query/calcite/metadata/ColocationGroup.java | 35 +++++++++++++++-------
.../calcite/metadata/FragmentDescription.java | 35 +++++++++++++++++++++-
3 files changed, 60 insertions(+), 14 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
index 0ec34128837..31856081a2c 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java
@@ -105,7 +105,7 @@ public class QueryStartRequest implements
DeferredUnmarshalMessage, ExecutionCon
this.schema = schema;
this.root = root;
this.ver = ver;
- this.fragmentDesc = fragmentDesc;
+ this.fragmentDesc = fragmentDesc.preparedToSend();
this.totalFragmentsCnt = totalFragmentsCnt;
this.params = params;
this.paramsBytes = paramsBytes; // If we already have marshalled
params, use it.
@@ -148,7 +148,7 @@ public class QueryStartRequest implements
DeferredUnmarshalMessage, ExecutionCon
* @return Fragment description.
*/
public FragmentDescription fragmentDescription() {
- return fragmentDesc;
+ return fragmentDesc.receivedFragment();
}
/**
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/ColocationGroup.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/ColocationGroup.java
index 336e2210664..5c22d18c971 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/ColocationGroup.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/ColocationGroup.java
@@ -30,16 +30,23 @@ import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.SelfMarshallingMessage;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.GridIntIterator;
import org.apache.ignite.internal.util.GridIntList;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
-/** */
-public class ColocationGroup implements SelfMarshallingMessage {
+/**
+ * Query/fragment colocation group. As a {@link Message}, has to be prepared
to send to another node and restored after
+ * receiving from another node.
+ *
+ * @see #prepareToSend()
+ * @see #afterReceive()
+ */
+public class ColocationGroup implements Message {
/** */
@Order(0)
long[] srcIds;
@@ -49,7 +56,7 @@ public class ColocationGroup implements
SelfMarshallingMessage {
List<UUID> nodeIds;
/** */
- private List<List<UUID>> assignments;
+ private @Nullable List<List<UUID>> assignments;
/**
* Flag, indacating that assignment is formed by original cache assignment
for given topology.
@@ -59,7 +66,7 @@ public class ColocationGroup implements
SelfMarshallingMessage {
/** Marshalled assignments serialization call holder. */
@Order(2)
- int[] marshalledAssignments;
+ @Nullable int[] marshalledAssignments;
/** */
public static ColocationGroup forNodes(List<UUID> nodeIds) {
@@ -233,6 +240,9 @@ public class ColocationGroup implements
SelfMarshallingMessage {
if (assignments == null)
return this;
+ /** Protects {@link #afterReceive()}: assignments must not be
marshaled yet. */
+ assert marshalledAssignments == null : "Marshalled assignments are
already set.";
+
List<List<UUID>> assignments = new
ArrayList<>(this.assignments.size());
Set<UUID> nodes = new HashSet<>();
@@ -294,7 +304,7 @@ public class ColocationGroup implements
SelfMarshallingMessage {
* Returns List of partitions to scan on the given node.
*
* @param nodeId Cluster node ID.
- * @return List of partitions to scan on the given node.
+ * @return Partitions to scan on the given node.
*/
public int[] partitions(UUID nodeId) {
if (F.isEmpty(assignments))
@@ -311,8 +321,8 @@ public class ColocationGroup implements
SelfMarshallingMessage {
return parts.arrayCopy();
}
- /** {@inheritDoc} */
- @Override public void selfMarshal() {
+ /** Prepares colocation group as {@link Message} to send to another node.
*/
+ public void prepareToSend() {
if (!F.isEmpty(marshalledAssignments) || assignments == null ||
primaryAssignment)
return;
@@ -340,9 +350,10 @@ public class ColocationGroup implements
SelfMarshallingMessage {
marshalledAssignments = builder.build().buffer();
}
- /** {@inheritDoc} */
- @Override public void selfUnmarshal() {
- if (F.isEmpty(marshalledAssignments))
+ /** Properly unwraps colocation group as {@link Message} after receiving
from another node. */
+ public void afterReceive() {
+ /** {@link #assignments} are set in constructors or are updated when
{@link #marshalledAssignments} is {@code null}. */
+ if (marshalledAssignments == null || assignments != null)
return;
int bitsPerPart = Integer.SIZE -
Integer.numberOfLeadingZeros(nodeIds.size());
@@ -357,6 +368,8 @@ public class ColocationGroup implements
SelfMarshallingMessage {
assignments.add(nodeIdx >= nodeIds.size() ?
Collections.emptyList() :
Collections.singletonList(nodeIds.get(nodeIdx)));
}
+
+ marshalledAssignments = null;
}
/** */
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
index 4cbb2f34faa..619be0ac1a6 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/FragmentDescription.java
@@ -24,7 +24,13 @@ import org.apache.ignite.internal.Order;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
-/** */
+/**
+ * Query fragment description. As a {@link Message}, has to be prepared to
send to another node and restored after
+ * receiving from another node.
+ *
+ * @see #preparedToSend()
+ * @see #receivedFragment()
+ */
public class FragmentDescription implements Message {
/** */
@Order(0)
@@ -42,6 +48,9 @@ public class FragmentDescription implements Message {
@Order(3)
@Nullable ColocationGroup target;
+ /** Transient flag of {@link #receivedFragment()}-once-invoked. */
+ boolean received;
+
/** */
public FragmentDescription() {
// No-op.
@@ -58,6 +67,30 @@ public class FragmentDescription implements Message {
this.target = target.explicitMapping();
}
+ /** Prepares fragment description as {@link Message} to send to another
node. */
+ public FragmentDescription preparedToSend() {
+ if (target != null)
+ target.prepareToSend();
+
+ mapping.colocationGrps.forEach(ColocationGroup::prepareToSend);
+
+ return this;
+ }
+
+ /** Properly unwraps fragment description as {@link Message} after
receiving from another node. */
+ public FragmentDescription receivedFragment() {
+ if (!received) {
+ if (target != null)
+ target.afterReceive();
+
+ mapping.colocationGrps.forEach(ColocationGroup::afterReceive);
+
+ received = true;
+ }
+
+ return this;
+ }
+
/** */
public long fragmentId() {
return fragmentId;
