IGNITE-28520 Auto-generate message serialization / marshalling / deployment

.mvn/jvm.config

@@ -0,0 +1,10 @@
+--add-opens=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.code=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.comp=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.jvm=ALL-UNNAMED
+--add-opens=jdk.compiler/com.sun.tools.javac.processing=ALL-UNNAMED

modules/binary/api/src/main/java/org/apache/ignite/internal/processors/cache/CacheObject.java

@@ -120,13 +120,13 @@ public interface CacheObject {
      * @param ldr Class loader.
      * @throws IgniteCheckedException If failed.
      */
-    public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException;
+    public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException;
 
     /**
      * @param ctx Context.
      * @throws IgniteCheckedException If failed.
      */
-    public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException;
+    public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException;
 
     /**
      * @return {@code True} if postWriteHashCode call required, {@code false} otherwise.

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java

@@ -367,12 +367,12 @@ private <T> T uncachedValue(Class<?> cls) throws BinaryObjectException {
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         this.ctx = ctx.binaryContext();
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
+    @Override public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
         // No-op.
     }
 

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java

@@ -220,13 +220,13 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
     @Override public CacheObject prepareForCache(CacheObjectValueContext ctx) {
         BinaryObjectImpl res = detached() ? this : detach(false);
 
-        res.prepareMarshal(ctx);
+        res.marshal(ctx);
 
         return res;
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         assert arr != null || valBytes != null;
 
         if (arr == null)
@@ -238,7 +238,7 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) {
+    @Override public void marshal(CacheObjectValueContext ctx) {
         assert arr != null || valBytes != null;
 
         if (valBytes == null)

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectOffheapImpl.java

@@ -453,12 +453,12 @@ else if (fieldOffLen == BinaryUtils.OFFSET_2)
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         throw new UnsupportedOperationException();
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
+    @Override public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
         throw new UnsupportedOperationException();
     }
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteMessageFactory.java

@@ -36,18 +36,18 @@ public class CalciteMessageFactory extends AbstractMarshallableMessageFactoryPro
 
     /** {@inheritDoc} */
     @Override public void registerAll(MessageFactory factory) {
-        register(factory, QueryStartRequest.class, (short)300, schemaAwareMarsh, resolvedClsLdr);
-        register(factory, QueryStartResponse.class, (short)301, dfltMarsh, dftlClsLdr);
-        register(factory, CalciteErrorMessage.class, (short)302, dfltMarsh, resolvedClsLdr);
-        register(factory, QueryBatchMessage.class, (short)303, dfltMarsh, dftlClsLdr);
-        register(factory, QueryBatchAcknowledgeMessage.class, (short)304, dfltMarsh, dftlClsLdr);
-        register(factory, QueryInboxCloseMessage.class, (short)305, dfltMarsh, dftlClsLdr);
-        register(factory, QueryCloseMessage.class, (short)306, dfltMarsh, dftlClsLdr);
-        register(factory, GenericValueMessage.class, (short)307, schemaAwareMarsh, resolvedClsLdr);
-        register(factory, FragmentMapping.class, (short)308, dfltMarsh, dftlClsLdr);
-        register(factory, ColocationGroup.class, (short)309, dfltMarsh, dftlClsLdr);
-        register(factory, FragmentDescription.class, (short)310, dfltMarsh, dftlClsLdr);
-        register(factory, QueryTxEntry.class, (short)311, dfltMarsh, dftlClsLdr);
+        register(factory, QueryStartRequest.class, (short)300, schemaAwareMarsh);
+        register(factory, QueryStartResponse.class, (short)301, dfltMarsh);
+        register(factory, CalciteErrorMessage.class, (short)302, dfltMarsh);
+        register(factory, QueryBatchMessage.class, (short)303, dfltMarsh);
+        register(factory, QueryBatchAcknowledgeMessage.class, (short)304, dfltMarsh);
+        register(factory, QueryInboxCloseMessage.class, (short)305, dfltMarsh);
+        register(factory, QueryCloseMessage.class, (short)306, dfltMarsh);
+        register(factory, GenericValueMessage.class, (short)307, schemaAwareMarsh);
+        register(factory, FragmentMapping.class, (short)308, dfltMarsh);
+        register(factory, ColocationGroup.class, (short)309, dfltMarsh);
+        register(factory, FragmentDescription.class, (short)310, dfltMarsh);
+        register(factory, QueryTxEntry.class, (short)311, dfltMarsh);
     }
 
     /** */

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java

@@ -18,10 +18,10 @@
 package org.apache.ignite.internal.processors.query.calcite.message;
 
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 
 /** */
 public final class GenericValueMessage implements MarshallableMessage {
@@ -48,13 +48,13 @@ public Object value() {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
         if (val != null && serialized == null)
             serialized = U.marshal(marsh, val);
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (serialized != null && val == null)
             val = U.unmarshal(marsh, serialized, clsLdr);
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageServiceImpl.java

@@ -39,6 +39,7 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageMarshaller;
 
 /**
  *
@@ -154,11 +155,8 @@ public FailureProcessor failureProcessor() {
     @Override public void send(UUID nodeId, Message msg) throws IgniteCheckedException {
         if (localNodeId().equals(nodeId))
             onMessage(nodeId, msg);
-        else {
-            prepareMarshal(msg);
-
+        else
             ioManager().sendToGridTopic(nodeId, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.CALLER_THREAD);
-        }
     }
 
     /** {@inheritDoc} */
@@ -181,24 +179,10 @@ public FailureProcessor failureProcessor() {
         }
     }
 
-    /** */
-    protected void prepareMarshal(Message msg) throws IgniteCheckedException {
-        try {
-            if (msg instanceof CalciteContextMarshallableMessage)
-                ((CalciteContextMarshallableMessage)msg).prepareMarshal(ctx);
-        }
-        catch (Exception e) {
-            failureProcessor().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
-            throw e;
-        }
-    }
-
     /** */
     protected void prepareUnmarshal(Message msg) throws IgniteCheckedException {
         try {
-            if (msg instanceof CalciteContextMarshallableMessage)
-                ((CalciteContextMarshallableMessage)msg).finishUnmarshal(ctx, clsLdr);
+            MessageMarshaller.unmarshal(ctx.kernalContext().messageFactory(), msg, ctx.kernalContext(), null, clsLdr);
         }
         catch (Exception e) {
             failureProcessor().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
@@ -221,22 +205,24 @@ protected void onMessage(UUID nodeId, Message msg) {
             );
     }
 
-    /** */
+    /** Listener for messages arriving from remote nodes; they are marshalled and must be finish-unmarshalled here. */
     private void onMessage(UUID nodeId, Object msg, byte plc) {
-        if (msg instanceof Message && CalciteMessageFactory.isCalciteMessage((Message)msg))
+        if (msg instanceof Message && CalciteMessageFactory.isCalciteMessage((Message)msg)) {
+            try {
+                prepareUnmarshal((Message)msg);
+            }
+            catch (IgniteCheckedException e) {
+                throw U.convertException(e);
+            }
+
             onMessage(nodeId, (Message)msg);
+        }
     }
 
     /** */
     private void onMessageInternal(UUID nodeId, Message msg) {
-        try {
-            prepareUnmarshal(msg);
+        MessageListener lsnr = Objects.requireNonNull(lsnrs.get(msg.getClass()));
 
-            MessageListener lsnr = Objects.requireNonNull(lsnrs.get(msg.getClass()));
-            lsnr.onMessage(nodeId, msg);
-        }
-        catch (IgniteCheckedException e) {
-            throw U.convertException(e);
-        }
+        lsnr.onMessage(nodeId, msg);
     }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java

@@ -21,19 +21,16 @@
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.jetbrains.annotations.Nullable;
 
-/**
- *
- */
-public class QueryStartRequest implements MarshallableMessage, CalciteContextMarshallableMessage, ExecutionContextAware {
+/** Message sent to remote nodes to start a query fragment execution. */
+public class QueryStartRequest implements MarshallableMessage, ExecutionContextAware {
     /** */
     @Order(0)
     String schema;
@@ -215,32 +212,16 @@ public boolean keepBinaryMode() {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
         if (paramsBytes == null && params != null)
             paramsBytes = U.marshal(marsh, params);
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        if (qryTxEntries != null) {
-            for (QueryTxEntry e : qryTxEntries)
-                e.prepareMarshal(ctx);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (params == null && paramsBytes != null)
             params = U.unmarshal(marsh, paramsBytes, clsLdr);
 
         paramsBytes = null;
     }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (qryTxEntries != null) {
-            for (QueryTxEntry e : qryTxEntries)
-                e.finishUnmarshal(ctx, clsLdr);
-        }
-    }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryTxEntry.java

@@ -20,15 +20,14 @@
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.function.Function;
-import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.configuration.TransactionConfiguration;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.cache.CacheObject;
-import org.apache.ignite.internal.processors.cache.CacheObjectContext;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.KeyCacheObject;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.plugin.extensions.communication.CacheIdAware;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
  * Class to pass to remote nodes transaction changes.
@@ -38,7 +37,7 @@
  * @see ExecutionContext#transactionChanges(int, int[], Function, Comparator)
  * @see QueryStartRequest#queryTransactionEntries()
  */
-public class QueryTxEntry implements CalciteContextMarshallableMessage {
+public class QueryTxEntry implements Message, CacheIdAware {
     /** Cache id. */
     @Order(0)
     int cacheId;
@@ -75,8 +74,8 @@ public QueryTxEntry(int cacheId, KeyCacheObject key, CacheObject val, GridCacheV
         this.ver = ver;
     }
 
-    /** @return Cache id. */
-    public int cacheId() {
+    /** {@inheritDoc} */
+    @Override public int cacheId() {
         return cacheId;
     }
 
@@ -94,24 +93,4 @@ public CacheObject value() {
     public GridCacheVersion version() {
         return ver;
     }
-
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        CacheObjectContext coctx = ctx.cacheContext(cacheId).cacheObjectContext();
-
-        key.prepareMarshal(coctx);
-
-        if (val != null)
-            val.prepareMarshal(coctx);
-    }
-
-    /** {@inheritDoc}  */
-    @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader ldr) throws IgniteCheckedException {
-        CacheObjectContext coctx = ctx.cacheContext(cacheId).cacheObjectContext();
-
-        key.finishUnmarshal(coctx, ldr);
-
-        if (val != null)
-            val.finishUnmarshal(coctx, ldr);
-    }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/ColocationGroup.java

@@ -30,7 +30,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
 import org.apache.ignite.internal.processors.query.calcite.util.Commons;
@@ -39,6 +38,7 @@
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 
 /** */
 public class ColocationGroup implements MarshallableMessage {
@@ -314,8 +314,8 @@ public int[] partitions(UUID nodeId) {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        if (assignments == null || primaryAssignment)
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
+        if (!F.isEmpty(marshalledAssignments) || assignments == null || primaryAssignment)
             return;
 
         Map<UUID, Integer> nodeIdxs = new HashMap<>();
@@ -343,7 +343,7 @@ public int[] partitions(UUID nodeId) {
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (F.isEmpty(marshalledAssignments))
             return;
 

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java

@@ -399,6 +399,11 @@ private TestMessageServiceImpl(GridKernalContext kernal, TestIoManager mgr) {
         @Override public boolean alive(UUID nodeId) {
             return true;
         }
+
+        /** {@inheritDoc} */
+        @Override protected void prepareUnmarshal(Message msg) {
+            // No-op: TestIoManager delivers messages in-memory, no deserialization needed.
+        }
     }
 
     /**