IGNITE-28705 Use MessageSerializer for QueryEntity

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

@@ -206,6 +206,7 @@
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage;
 import org.apache.ignite.internal.processors.query.InlineSizesData;
+import org.apache.ignite.internal.processors.query.QueryEntityMessage;
 import org.apache.ignite.internal.processors.query.QueryField;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryCancelRequest;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryFailResponse;
@@ -575,6 +576,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(CacheContinuousQueryBatchAck.class);
         withSchema(CacheContinuousQueryEntry.class);
         withNoSchema(InlineSizesData.class);
+        withNoSchema(QueryEntityMessage.class);
 
         // [11200 - 11300]: Compute, distributed process messages.
         msgIdx = 11200;

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

@@ -27,7 +27,9 @@
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.encryption.GroupKeyEncrypted;
 import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
+import org.apache.ignite.internal.processors.query.QueryEntityMessage;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -57,13 +59,10 @@ public class StoredCacheData implements Serializable, CdcCacheEvent, Marshallabl
     @Order(0)
     transient byte[] ccfgBytes;
 
-    /** Query entities. */
-    @GridToStringInclude
-    private Collection<QueryEntity> qryEntities;
-
-    /** Serialized {@link #qryEntities}. */
+    /** Query entity messages. */
     @Order(1)
-    transient byte[] qryEntitiesBytes;
+    @GridToStringInclude
+    Collection<QueryEntityMessage> entityMsgs;
 
     /** SQL flag - {@code true} if cache was created with {@code CREATE TABLE}. */
     @Order(2)
@@ -97,15 +96,15 @@ public StoredCacheData(CacheConfiguration<?, ?> ccfg) {
         A.notNull(ccfg, "ccfg");
 
         this.ccfg = ccfg;
-        qryEntities = ccfg.getQueryEntities();
+        entityMsgs = F.viewReadOnly(ccfg.getQueryEntities(), QueryEntityMessage::new);
     }
 
     /**
      * @param cacheData Cache data.
      */
     public StoredCacheData(StoredCacheData cacheData) {
         ccfg = cacheData.ccfg;
-        qryEntities = cacheData.qryEntities;
+        entityMsgs = cacheData.entityMsgs;
         sql = cacheData.sql;
         cacheConfigurationEnrichment = cacheData.cacheConfigurationEnrichment;
         grpKeyEncrypted = cacheData.grpKeyEncrypted;
@@ -129,14 +128,14 @@ public void config(CacheConfiguration<?, ?> ccfg) {
      * @return Query entities.
      */
     @Override public Collection<QueryEntity> queryEntities() {
-        return qryEntities;
+        return F.viewReadOnly(entityMsgs, QueryEntityMessage::queryEntity);
     }
 
     /**
      * @param qryEntities Query entities.
      */
     public void queryEntities(Collection<QueryEntity> qryEntities) {
-        this.qryEntities = qryEntities;
+        entityMsgs = F.viewReadOnly(qryEntities, QueryEntityMessage::new);
     }
 
     /**
@@ -228,9 +227,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 +236,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/QueryEntityMessage.java

@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.cache.query.QueryIndexMessage;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+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;
+import org.apache.ignite.spi.discovery.ObjectData;
+
+/** Message for {@link QueryEntity}. */
+public class QueryEntityMessage implements Serializable, Message {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Key type. */
+    @Order(0)
+    String keyType;
+
+    /** Value type. */
+    @Order(1)
+    String valType;
+
+    /** Key name. */
+    @Order(2)
+    String keyFieldName;
+
+    /** Value name. */
+    @Order(3)
+    String valFieldName;
+
+    /** Fields available for query. A map from field name to type name. */
+    @Order(4)
+    @GridToStringInclude
+    LinkedHashMap<String, String> fields;
+
+    /** Set of field names that belong to the key. */
+    @Order(5)
+    @GridToStringInclude
+    Set<String> keyFields;
+
+    /** Aliases. */
+    @Order(6)
+    @GridToStringInclude
+    Map<String, String> aliases;
+
+    /** Collection of query indexes. */
+    @Order(7)
+    @GridToStringInclude
+    Collection<QueryIndexMessage> idxs;
+
+    /** Table name. */
+    @Order(8)
+    String tableName;
+
+    /** Fields that must have non-null value. NB: DO NOT remove underscore to avoid clashes with QueryEntityEx. */
+    @Order(9)
+    Set<String> notNullFields;
+
+    /** Fields default values. */
+    @Order(10)
+    Map<String, ObjectData> dfltFieldValues;
+
+    /** Precision (maximum length) for fields. */
+    @Order(11)
+    Map<String, Integer> fieldsPrecision;
+
+    /** Scale for fields. */
+    @Order(12)
+    Map<String, Integer> fieldsScale;
+
+    /** Empty constructor for {@link MessageFactory}. */
+    public QueryEntityMessage() {
+        // No-op.
+    }
+
+    /** Copies {@code qryEntity}. */
+    public QueryEntityMessage(QueryEntity qryEntity) {
+        assert qryEntity != null;
+
+        keyType = qryEntity.getKeyType();
+        valType = qryEntity.getValueType();
+        keyFieldName = qryEntity.getKeyFieldName();
+        valFieldName = qryEntity.getValueFieldName();
+        fields = qryEntity.getFields();
+        keyFields = qryEntity.getKeyFields();
+        aliases = qryEntity.getAliases();
+        idxs = F.viewReadOnly(qryEntity.getIndexes(), QueryIndexMessage::new);
+        tableName = qryEntity.getTableName();
+        notNullFields = qryEntity.getNotNullFields();
+        dfltFieldValues = F.viewReadOnly(qryEntity.getDefaultFieldValues(), val -> new ObjectData((Serializable)val));
+        fieldsPrecision = qryEntity.getFieldsPrecision();
+        fieldsScale = qryEntity.getFieldsScale();
+    }
+
+    /** @return Copy of {@code msg} as {@link QueryEntity}. */
+    public static QueryEntity queryEntity(QueryEntityMessage msg) {
+        assert msg != null;
+
+        QueryEntity res = new QueryEntity();
+
+        res.setKeyType(msg.keyType);
+        res.setValueType(msg.valType);
+        res.setKeyFieldName(msg.keyFieldName);
+        res.setValueFieldName(msg.valFieldName);
+        res.setFields(msg.fields);
+        res.setKeyFields(msg.keyFields);
+        res.setAliases(msg.aliases);
+        res.setIndexes(F.viewReadOnly(msg.idxs, QueryIndexMessage::queryIndex));
+        res.setTableName(msg.tableName);
+        res.setNotNullFields(msg.notNullFields);
+        res.setDefaultFieldValues(F.viewReadOnly(msg.dfltFieldValues, ObjectData::unwrap));
+        res.setFieldsPrecision(msg.fieldsPrecision);
+        res.setFieldsScale(msg.fieldsScale);
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(QueryEntityMessage.class, this);
+    }
+}

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

@@ -19,26 +19,21 @@
 
 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.processors.query.QueryEntityMessage;
+import org.apache.ignite.internal.util.typedef.F;
 
 /**
  * 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<QueryEntityMessage> entityMsgs;
 
     /** */
     @Order(1)
@@ -68,7 +63,7 @@ public SchemaAddQueryEntityOperation(
         boolean sqlEscape
     ) {
         super(opId, cacheName, schemaName);
-        this.entities = entities;
+        entityMsgs = F.viewReadOnly(entities, QueryEntityMessage::new);
         this.qryParallelism = qryParallelism;
         this.sqlEscape = sqlEscape;
     }
@@ -77,7 +72,7 @@ public SchemaAddQueryEntityOperation(
      * @return Collection of query entities.
      */
     public Collection<QueryEntity> entities() {
-        return entities;
+        return F.viewReadOnly(entityMsgs, QueryEntityMessage::queryEntity);
     }
 
     /**
@@ -93,20 +88,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;
-        }
-    }
-
 }