IGNITE-28907 Support merging QueryEntity metadata in CacheConfiguration

modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java

@@ -53,6 +53,7 @@
 import org.apache.ignite.cache.store.CacheStoreSessionListener;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.internal.binary.BinaryUtils;
+import org.apache.ignite.internal.processors.query.QueryEntityMerger;
 import org.apache.ignite.internal.processors.query.QueryUtils;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.A;
@@ -1968,26 +1969,23 @@ public CacheConfiguration<K, V> setIndexedTypes(Class<?>... indexedTypes) {
             Class<?> keyCls = newIndexedTypes[i];
             Class<?> valCls = newIndexedTypes[i + 1];
 
-            QueryEntity newEntity = new QueryEntity(keyCls, valCls);
+            QueryEntity incomingEntity = new QueryEntity(keyCls, valCls);
 
-            boolean dup = false;
+            QueryEntity existingEntity = findQueryEntity(incomingEntity.findValueType());
 
-            for (QueryEntity entity : qryEntities) {
-                if (Objects.equals(entity.findValueType(), newEntity.findValueType())) {
-                    dup = true;
+            if (existingEntity == null)
+                qryEntities.add(incomingEntity);
+            else {
+                QueryEntity mergedEntity = QueryEntityMerger.merge(getName(), existingEntity, incomingEntity);
 
-                    break;
-                }
+                replaceQueryEntity(existingEntity, mergedEntity);
             }
 
-            if (!dup)
-                qryEntities.add(newEntity);
-
             // Set key configuration if needed.
             String affFieldName = BinaryUtils.affinityFieldName(keyCls);
 
             if (affFieldName != null) {
-                CacheKeyConfiguration newKeyCfg = new CacheKeyConfiguration(newEntity.getKeyType(), affFieldName);
+                CacheKeyConfiguration newKeyCfg = new CacheKeyConfiguration(incomingEntity.getKeyType(), affFieldName);
 
                 if (F.isEmpty(keyCfg))
                     keyCfg = new CacheKeyConfiguration[] { newKeyCfg };
@@ -2080,25 +2078,21 @@ public CacheConfiguration<K, V> setPartitionLossPolicy(PartitionLossPolicy partL
      * @return {@code this} for chaining.
      */
     public CacheConfiguration<K, V> setQueryEntities(Collection<QueryEntity> qryEntities) {
-        if (this.qryEntities == null) {
-            this.qryEntities = new ArrayList<>(qryEntities);
+        if (this.qryEntities == null)
+            this.qryEntities = new ArrayList<>();
 
-            return this;
-        }
+        for (QueryEntity incomingEntity : qryEntities) {
+            String valType = incomingEntity.findValueType();
 
-        for (QueryEntity entity : qryEntities) {
-            boolean found = false;
+            QueryEntity existingEntity = findQueryEntity(valType);
 
-            for (QueryEntity existing : this.qryEntities) {
-                if (Objects.equals(entity.findValueType(), existing.findValueType())) {
-                    found = true;
+            if (existingEntity == null)
+                this.qryEntities.add(incomingEntity);
+            else {
+                QueryEntity mergedEntity = QueryEntityMerger.merge(getName(), existingEntity, incomingEntity);
 
-                    break;
-                }
+                replaceQueryEntity(existingEntity, mergedEntity);
             }
-
-            if (!found)
-                this.qryEntities.add(entity);
         }
 
         return this;
@@ -2484,6 +2478,29 @@ public CacheConfiguration<K, V> setIndexPath(String idxPath) {
         return S.toString(CacheConfiguration.class, this);
     }
 
+    /** */
+    private QueryEntity findQueryEntity(String valType) {
+        if (qryEntities == null)
+            return null;
+
+        for (QueryEntity entity : qryEntities) {
+            if (Objects.equals(entity.findValueType(), valType))
+                return entity;
+        }
+
+        return null;
+    }
+
+    /** */
+    private void replaceQueryEntity(QueryEntity oldEntity, QueryEntity newEntity) {
+        Collection<QueryEntity> updated = new ArrayList<>(qryEntities.size());
+
+        for (QueryEntity entity : qryEntities)
+            updated.add(entity == oldEntity ? newEntity : entity);
+
+        qryEntities = updated;
+    }
+
     /**
      *  Filter that accepts all nodes.
      */

modules/core/src/main/java/org/apache/ignite/internal/processors/query/QueryEntityMerger.java

@@ -0,0 +1,261 @@
+/*
+ * 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.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 java.util.Objects;
+import java.util.Set;
+import javax.cache.CacheException;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.internal.util.typedef.F;
+
+/** Utility for merging compatible {@link QueryEntity} metadata. */
+public final class QueryEntityMerger {
+    /** */
+    private final String cacheName;
+
+    /** */
+    private QueryEntityMerger(String cacheName) {
+        this.cacheName = cacheName;
+    }
+
+    /**
+     * Merges incoming query entity metadata into existing entity.
+     *
+     * @param cacheName Cache name.
+     * @param existing Existing query entity.
+     * @param incoming Incoming query entity.
+     * @return Merged query entity.
+     * @throws CacheException If entities contain conflicting metadata.
+     */
+    public static QueryEntity merge(String cacheName, QueryEntity existing, QueryEntity incoming) {
+        return new QueryEntityMerger(cacheName).merge0(existing, incoming);
+    }
+
+    /** */
+    private QueryEntity merge0(QueryEntity ex, QueryEntity in) {
+        if (!Objects.equals(ex.findValueType(), in.findValueType())) {
+            throw new CacheException(
+                "Failed to merge query entities because value types differ " +
+                    "[cacheName=" + cacheName +
+                    ", existingValueType=" + ex.findValueType() +
+                    ", incomingValueType=" + in.findValueType() + ']'
+            );
+        }
+
+        QueryEntity res = new QueryEntity(ex);
+
+        res.setKeyType(mergeKeyType(ex, in));
+
+        res.setValueType(mergeProperty("valueType", ex.getValueType(), in.getValueType()));
+        res.setTableName(mergeProperty("tableName", ex.getTableName(), in.getTableName()));
+        res.setKeyFieldName(mergeProperty("keyFieldName", ex.getKeyFieldName(), in.getKeyFieldName()));
+        res.setValueFieldName(mergeProperty("valueFieldName", ex.getValueFieldName(), in.getValueFieldName()));
+
+        res.setFields(mergeFields(ex.getFields(), in.getFields()));
+
+        res.setKeyFields(mergeSet(ex.getKeyFields(), in.getKeyFields()));
+        res.setNotNullFields(mergeSet(ex.getNotNullFields(), in.getNotNullFields()));
+
+        res.setAliases(mergeMap("aliases", ex.getAliases(), in.getAliases()));
+        res.setDefaultFieldValues(mergeMap("defaultFieldValues", ex.getDefaultFieldValues(), in.getDefaultFieldValues()));
+        res.setFieldsPrecision(mergeMap("fieldsPrecision", ex.getFieldsPrecision(), in.getFieldsPrecision()));
+        res.setFieldsScale(mergeMap("fieldsScale", ex.getFieldsScale(), in.getFieldsScale()));
+
+        res.setIndexes(mergeIndexes(res, ex.getIndexes(), in.getIndexes()));
+
+        return res;
+    }
+
+    /** */
+    private String mergeKeyType(QueryEntity ex, QueryEntity in) {
+        String exKeyType = ex.findKeyType();
+        String inKeyType = in.findKeyType();
+
+        if (exKeyType != null && inKeyType != null && !Objects.equals(exKeyType, inKeyType)) {
+            throw mergeConflict(
+                "keyType",
+                exKeyType,
+                inKeyType
+            );
+        }
+
+        return ex.getKeyType() != null ? ex.getKeyType() : in.getKeyType();
+    }
+
+    /** */
+    private <T> T mergeProperty(String propName, T existingVal, T incomingVal) {
+        if (existingVal == null)
+            return incomingVal;
+
+        if (incomingVal == null)
+            return existingVal;
+
+        if (Objects.equals(existingVal, incomingVal))
+            return existingVal;
+
+        throw mergeConflict(propName, existingVal, incomingVal);
+    }
+
+    /** */
+    private LinkedHashMap<String, String> mergeFields(
+        Map<String, String> existingFields,
+        Map<String, String> incomingFields
+    ) {
+        if (existingFields == null && incomingFields == null)
+            return null;
+
+        LinkedHashMap<String, String> res = new LinkedHashMap<>();
+
+        if (existingFields != null)
+            res.putAll(existingFields);
+
+        if (incomingFields == null)
+            return res;
+
+        for (Map.Entry<String, String> entry : incomingFields.entrySet()) {
+            String field = entry.getKey();
+            String incomingType = entry.getValue();
+
+            if (!res.containsKey(field)) {
+                res.put(field, incomingType);
+
+                continue;
+            }
+
+            String existingType = res.get(field);
+
+            if (!Objects.equals(existingType, incomingType))
+                throw mergeConflict("fieldType[" + field + ']', existingType, incomingType);
+        }
+
+        return res;
+    }
+
+    /** */
+    private <T> Map<String, T> mergeMap(String propName, Map<String, T> existingVals, Map<String, T> incomingVals) {
+        if (existingVals == null && incomingVals == null)
+            return null;
+
+        Map<String, T> res = new HashMap<>();
+
+        if (existingVals != null)
+            res.putAll(existingVals);
+
+        if (incomingVals == null)
+            return res;
+
+        for (Map.Entry<String, T> entry : incomingVals.entrySet()) {
+            String field = entry.getKey();
+            T incomingVal = entry.getValue();
+
+            if (!res.containsKey(field)) {
+                res.put(field, incomingVal);
+
+                continue;
+            }
+
+            T existingVal = res.get(field);
+
+            if (!Objects.equals(existingVal, incomingVal))
+                throw mergeConflict(propName + '[' + field + ']', existingVal, incomingVal);
+        }
+
+        return res;
+    }
+
+    /** */
+    private <T> Set<T> mergeSet(Set<T> existing, Set<T> incoming) {
+        if (F.isEmpty(existing) && F.isEmpty(incoming))
+            return null;
+
+        Set<T> res = new LinkedHashSet<>();
+
+        if (existing != null)
+            res.addAll(existing);
+
+        if (incoming != null)
+            res.addAll(incoming);
+
+        return res;
+    }
+
+    /** */
+    private Collection<QueryIndex> mergeIndexes(
+        QueryEntity entity,
+        Collection<QueryIndex> existingIndexes,
+        Collection<QueryIndex> incomingIndexes
+    ) {
+        if (F.isEmpty(existingIndexes) && F.isEmpty(incomingIndexes))
+            return null;
+
+        List<QueryIndex> res = new ArrayList<>();
+
+        Map<String, QueryIndex> indexesByName = new HashMap<>();
+
+        if (existingIndexes != null) {
+            for (QueryIndex idx : existingIndexes) {
+                String idxName = QueryUtils.indexName(entity, idx);
+
+                res.add(idx);
+
+                indexesByName.put(idxName, idx);
+            }
+        }
+
+        if (incomingIndexes == null)
+            return res;
+
+        for (QueryIndex incomingIdx : incomingIndexes) {
+            String idxName = QueryUtils.indexName(entity, incomingIdx);
+
+            QueryIndex existingIdx = indexesByName.get(idxName);
+
+            if (existingIdx == null) {
+                res.add(incomingIdx);
+
+                indexesByName.put(idxName, incomingIdx);
+
+                continue;
+            }
+
+            if (!existingIdx.equals(incomingIdx))
+                throw mergeConflict("index[" + idxName + ']', existingIdx, incomingIdx);
+        }
+
+        return res;
+    }
+
+    /** */
+    private CacheException mergeConflict(String propName, Object existingVal, Object incomingVal) {
+        return new CacheException(
+            "Failed to merge query entities due to conflicting metadata " +
+                "[cacheName=" + cacheName +
+                ", property=" + propName +
+                ", existingValue=" + existingVal +
+                ", incomingValue=" + incomingVal + ']'
+        );
+    }
+}

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsCorruptedIndexTest.java

@@ -90,7 +90,7 @@ public class IgnitePdsCorruptedIndexTest extends GridCommonAbstractTest {
         CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>(CACHE)
             .setBackups(1)
             .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
-            .setIndexedTypes(Integer.class, IndexedObject.class, Long.class, IndexedObject.class)
+            .setIndexedTypes(Integer.class, IndexedObject.class)
             .setAffinity(new RendezvousAffinityFunction(false, 32));
 
         cfg.setCacheConfiguration(ccfg);