IGNITE-28705 Use MessageSerializer for QueryEntity

modules/core/src/main/java/org/apache/ignite/cache/QueryEntity.java

@@ -97,7 +97,7 @@ public class QueryEntity implements Serializable {
     /** Fields default values. */
     private Map<String, Object> defaultFieldValues = new HashMap<>();
 
-    /** Precision(Maximum length) for fields. */
+    /** Precision (maximum length) for fields. */
     private Map<String, Integer> fieldsPrecision = new HashMap<>();
 
     /** Scale for fields. */

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -24,6 +24,7 @@
 import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
 import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyTypeSettings;
 import org.apache.ignite.internal.management.cache.PartitionKey;
+import org.apache.ignite.internal.management.cache.QueryEntity;
 import org.apache.ignite.internal.managers.checkpoint.GridCheckpointRequest;
 import org.apache.ignite.internal.managers.communication.CompressedMessage;
 import org.apache.ignite.internal.managers.communication.ErrorMessage;
@@ -578,6 +579,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(CacheContinuousQueryBatchAck.class);
         withSchema(CacheContinuousQueryEntry.class);
         withNoSchema(InlineSizesData.class);
+        withNoSchema(QueryEntity.class);
 
         // [11200 - 11300]: Compute, distributed process messages.
         msgIdx = 11200;

modules/core/src/main/java/org/apache/ignite/internal/management/cache/QueryEntity.java

@@ -17,21 +17,25 @@
 
 package org.apache.ignite.internal.management.cache;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.dto.IgniteDataTransferObject;
+import org.apache.ignite.internal.cache.query.QueryIndexMessage;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 
 /**
- * Data transfer object for {@link org.apache.ignite.cache.QueryEntity}.
+ * Message for {@link org.apache.ignite.cache.QueryEntity}.
  */
-public class QueryEntity extends IgniteDataTransferObject {
+public class QueryEntity implements Message, Serializable {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -69,11 +73,11 @@ public class QueryEntity extends IgniteDataTransferObject {
 
     /** Fields to create group indexes for. */
     @Order(8)
-    List<QueryIndex> grps;
+    List<QueryIndexMessage> grps;
 
     /**
      * @param qryEntities Collection of query entities.
-     * @return Data transfer object for query entities.
+     * @return Messages for query entities.
      */
     public static List<QueryEntity> list(Collection<org.apache.ignite.cache.QueryEntity> qryEntities) {
         List<QueryEntity> entities = new ArrayList<>();
@@ -87,14 +91,40 @@ public static List<QueryEntity> list(Collection<org.apache.ignite.cache.QueryEnt
     }
 
     /**
-     * Create data transfer object for given cache type metadata.
+     * @param qryEntities Collection of messages for query entities.
+     * @return Query entities from messages.
      */
+    public static Collection<org.apache.ignite.cache.QueryEntity> unwrapList(Collection<QueryEntity> qryEntities) {
+        Collection<org.apache.ignite.cache.QueryEntity> entities = new ArrayList<>();
+
+        if (!F.isEmpty(qryEntities)) {
+            for (QueryEntity entityDto : qryEntities) {
+                org.apache.ignite.cache.QueryEntity entity = new org.apache.ignite.cache.QueryEntity();
+
+                entity.setKeyType(entityDto.keyType);
+                entity.setValueType(entityDto.valType);
+                entity.setFields((LinkedHashMap<String, String>)entityDto.qryFlds);
+                entity.setKeyFields(new LinkedHashSet<>(entityDto.keyFields));
+                entity.setAliases(entityDto.aliases);
+                entity.setTableName(entityDto.tblName);
+                entity.setKeyFieldName(entityDto.keyFieldName);
+                entity.setValueFieldName(entityDto.valFieldName);
+                entity.setIndexes(F.viewReadOnly(entityDto.grps, QueryIndexMessage::queryIndex));
+
+                entities.add(entity);
+            }
+        }
+
+        return entities;
+    }
+
+    /** Empty constructor for a {@link MessageFactory}. */
     public QueryEntity() {
         // No-op.
     }
 
     /**
-     * Create data transfer object for given cache type metadata.
+     * Create message for given cache type metadata.
      *
      * @param q Actual cache query entities.
      */
@@ -104,7 +134,7 @@ private QueryEntity(org.apache.ignite.cache.QueryEntity q) {
         keyType = q.getKeyType();
         valType = q.getValueType();
 
-        keyFields = toList(q.getKeyFields());
+        keyFields = q.getKeyFields() != null ? new ArrayList<>(q.getKeyFields()) : null;
 
         LinkedHashMap<String, String> qryFields = q.getFields();
 
@@ -117,7 +147,7 @@ private QueryEntity(org.apache.ignite.cache.QueryEntity q) {
         grps = new ArrayList<>(qryIdxs.size());
 
         for (org.apache.ignite.cache.QueryIndex qryIdx : qryIdxs)
-            grps.add(new QueryIndex(qryIdx));
+            grps.add(new QueryIndexMessage(qryIdx));
 
         tblName = q.getTableName();
         keyFieldName = q.getKeyFieldName();
@@ -183,7 +213,7 @@ public String getValueFieldName() {
     /**
      * @return Fields to create group indexes for.
      */
-    public List<QueryIndex> getGroups() {
+    public List<QueryIndexMessage> getGroups() {
         return grps;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java

@@ -20,11 +20,11 @@
 import java.io.Serializable;
 import java.util.Collection;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.cache.QueryEntity;
 import org.apache.ignite.cdc.CdcCacheEvent;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.management.cache.QueryEntity;
 import org.apache.ignite.internal.managers.encryption.GroupKeyEncrypted;
 import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -58,12 +58,9 @@ public class StoredCacheData implements Serializable, CdcCacheEvent, Marshallabl
     transient byte[] ccfgBytes;
 
     /** Query entities. */
-    @GridToStringInclude
-    private Collection<QueryEntity> qryEntities;
-
-    /** Serialized {@link #qryEntities}. */
     @Order(1)
-    transient byte[] qryEntitiesBytes;
+    @GridToStringInclude
+    Collection<QueryEntity> qryEntities;
 
     /** SQL flag - {@code true} if cache was created with {@code CREATE TABLE}. */
     @Order(2)
@@ -97,7 +94,7 @@ public StoredCacheData(CacheConfiguration<?, ?> ccfg) {
         A.notNull(ccfg, "ccfg");
 
         this.ccfg = ccfg;
-        qryEntities = ccfg.getQueryEntities();
+        qryEntities = QueryEntity.list(ccfg.getQueryEntities());
     }
 
     /**
@@ -128,15 +125,15 @@ public void config(CacheConfiguration<?, ?> ccfg) {
     /**
      * @return Query entities.
      */
-    @Override public Collection<QueryEntity> queryEntities() {
-        return qryEntities;
+    @Override public Collection<org.apache.ignite.cache.QueryEntity> queryEntities() {
+        return QueryEntity.unwrapList(qryEntities);
     }
 
     /**
      * @param qryEntities Query entities.
      */
-    public void queryEntities(Collection<QueryEntity> qryEntities) {
-        this.qryEntities = qryEntities;
+    public void queryEntities(Collection<org.apache.ignite.cache.QueryEntity> qryEntities) {
+        this.qryEntities = QueryEntity.list(qryEntities);
     }
 
     /**
@@ -228,9 +225,6 @@ public StoredCacheData withSplittedCacheConfig(CacheConfigurationSplitter splitt
     @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
         if (ccfg != null)
             ccfgBytes = U.marshal(marsh, ccfg);
-
-        if (qryEntities != null)
-            qryEntitiesBytes = U.marshal(marsh, qryEntities);
     }
 
     /** {@inheritDoc} */
@@ -240,11 +234,5 @@ public StoredCacheData withSplittedCacheConfig(CacheConfigurationSplitter splitt
 
             ccfgBytes = null;
         }
-
-        if (qryEntitiesBytes != null) {
-            qryEntities = U.unmarshal(marsh, qryEntitiesBytes, clsLdr);
-
-            qryEntitiesBytes = null;
-        }
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/operation/SchemaAddQueryEntityOperation.java

@@ -19,26 +19,19 @@
 
 import java.util.Collection;
 import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.cache.QueryEntity;
-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.internal.management.cache.QueryEntity;
 
 /**
  * Enabling indexing on cache operation.
  */
-public class SchemaAddQueryEntityOperation extends SchemaAbstractOperation implements MarshallableMessage {
+public class SchemaAddQueryEntityOperation extends SchemaAbstractOperation {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private Collection<QueryEntity> entities;
-
-    /** Serialized form of query entities. */
     @Order(0)
-    transient byte[] qryEntitiesBytes;
+    Collection<QueryEntity> qryEntities;
 
     /** */
     @Order(1)
@@ -63,21 +56,21 @@ public SchemaAddQueryEntityOperation(
         UUID opId,
         String cacheName,
         String schemaName,
-        Collection<QueryEntity> entities,
+        Collection<org.apache.ignite.cache.QueryEntity> entities,
         int qryParallelism,
         boolean sqlEscape
     ) {
         super(opId, cacheName, schemaName);
-        this.entities = entities;
+        qryEntities = QueryEntity.list(entities);
         this.qryParallelism = qryParallelism;
         this.sqlEscape = sqlEscape;
     }
 
     /**
      * @return Collection of query entities.
      */
-    public Collection<QueryEntity> entities() {
-        return entities;
+    public Collection<org.apache.ignite.cache.QueryEntity> entities() {
+        return QueryEntity.unwrapList(qryEntities);
     }
 
     /**
@@ -93,20 +86,4 @@ public int queryParallelism() {
     public boolean isSqlEscape() {
         return sqlEscape;
     }
-
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        if (entities != null)
-            qryEntitiesBytes = U.marshal(marsh, entities);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (qryEntitiesBytes != null) {
-            entities = U.unmarshal(marsh, qryEntitiesBytes, clsLdr);
-
-            qryEntitiesBytes = null;
-        }
-    }
-
 }