Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 };
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I plan to use JSpecify Nullaway frameqork in future, thus we need to be accurate with return types if function is not annotated with @Nullable plz fix it.


for (QueryEntity entity : qryEntities) {
if (Objects.equals(entity.findValueType(), valType))
return entity;
}

return null;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

and it

}

/** */
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.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2158,7 +2158,14 @@ public static <K, V> CacheConfiguration<K, V> patchCacheConfiguration(
boolean isSqlEscape,
int qryParallelism
) {
return new CacheConfiguration<>(oldCfg)
CacheConfiguration<K, V> newCfg = new CacheConfiguration<>(oldCfg);

Collection<QueryEntity> oldEntities = oldCfg.getQueryEntities();

newCfg.clearQueryEntities();
newCfg.setQueryEntities(oldEntities);

return newCfg
.setQueryEntities(entities)
.setSqlSchema(sqlSchema)
.setSqlEscapeAll(isSqlEscape)
Expand Down
Original file line number Diff line number Diff line change
@@ -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 {
/** */
public static final String CONFLICT_MESSAGE_TEMPLATE = "Failed to merge query entities due to conflicting metadata " +
"[cacheName=%s, property=%s, existingValue=%s, incomingValue=%s]";

/** */
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);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I reproduced the QueryEntityEx metadata loss and investigated the schema-add and cache-configuration merge paths.

The reported problem is real: with the current PR implementation, QueryEntity res = new QueryEntity(ex) can turn a QueryEntityEx into a plain QueryEntity. Which leads to losing sql, preserveKeysOrder, fillAbsentPKsWithDefaults, and the other extended metadata.

However, after tracing the actual lifecycle, it looks like the schema-add case does not require defining merge rules for two different QueryEntityEx instances. The downgrade was caused by a different issue in patchCacheConfiguration().

The reproducer is testSchemaAddPreservesQueryEntityExMetadata(). It uses a statically configured cache with no query entities and then executes a query that creates a table. This DDL operation creates a QueryEntityEx with:

  • sql = true
  • preserveKeysOrder = true
  • fillAbsentPKsWithDefaults = true

The schema-add lifecycle patches the cache configuration twice:

SchemaAddQueryEntityOperation
>>>
GridCacheContext.onSchemaAddQueryEntity() -> patchCacheConfiguration()
>>>
SchemaFinishDiscovery -> DynamicCacheDescriptor.schemaChangeFinish() → patchCacheConfiguration()

Initially I expected this to be a legitimate merge of two extended entities, but it is not. The first and the second patch operate on different CacheConfiguration objects, but CacheConfiguration copy construction keeps the same mutable qryEntities collection. Therefore the first patch modifies the collection that is still visible from the descriptor configuration.

So the "merge" in this scenario is actually an accidentally repeated usage of the same schema-add entity. It is not merging two independently defined QueryEntityEx configurations.

I added an isolated regression test for the underlying issue:

GridCacheUtilsTest.testPatchCacheConfigurationDoesNotModifyOriginalQueryEntities

The test prepares an oldCfg with a materialized empty query-entity collection, calls GridCacheUtils.patchCacheConfiguration(), and verifies that adding the new entity to the patched configuration does not modify oldCfg.

On master / the previous implementation it fails on:

assertTrue(oldCfg.getQueryEntities().isEmpty());

because both configurations share the same mutable qryEntities collection.

After making patchCacheConfiguration() use an independent query-entity collection, this test passes.

More importantly, after this change testSchemaAddPreservesQueryEntityExMetadata also passes. The schema-finish patch now sees an empty descriptor configuration and simply installs the incoming QueryEntityEx. It no longer tries to merge two identical entities.

I also checked whether we have a supported runtime path that would require resolving a real conflict between two different QueryEntityEx objects. The relevant cases I checked were:

  1. The normal QueryEntity conflicts targeted by this PR are still real and are covered by CacheConfigurationQueryEntityMergeTest.
  2. Adding SQL metadata dynamically to an already indexed cache is rejected, for example:
javax.cache.CacheException: Cache is already indexed: sql-cache
  1. Joining an active cluster with a static cache configuration that would require schema/configuration merging is rejected:
org.apache.ignite.spi.IgniteSpiException: Failed to join node to the active cluster (the config of the cache 'TEST_CACHE' has to be merged which is impossible on active grid). Deactivate grid and retry node join or clean the joining node.
  1. I also tested a more specific SQL-vs-API case: create the cache/table through CREATE TABLE on the existing node and configure the same cache statically on the joining node. This is rejected with:
org.apache.ignite.IgniteCheckedException: Cache configuration mismatch (local cache was created via Ignite API, while remote cache was created via CREATE TABLE): ...

Again, QueryEntityMerger is not reached.
5. It is possible to force QueryEntityEx + QueryEntity manually by obtaining the internal CacheConfiguration through node.context().cache().cacheConfiguration(...) and calling setQueryEntities() directly. In that artificial scenario the current merger can indeed downgrade the entity.
However, this mutates Ignite's internal runtime configuration directly. And as I understand it is not a supported public cache-configuration update path.

I also reviewed the internal setQueryEntities() paths. Normalization replaces the entity collection after clearing it; creation of a new SQL cache installs entities into an initially empty configuration; schema-add is expected to add entities to an empty schema; and the join cases above are validated before conflicting configurations can reach the merger.

Therefore I did not add rules for merging conflicting QueryEntityEx-specific fields. This would introduce new semantics, for example, for two different extended entities we would have to decide what these combinations mean:

  • sql: false vs true
  • implicitPk: false vs true
  • preserveKeysOrder: false vs true
  • fillAbsentPKsWithDefaults: false vs true
  • primaryKeyInlineSize: 10 vs 20
  • affinityKeyInlineSize: 10 vs 20

It is not clear whether they should be conflicts, whether one side should overtake or whether some boolean fields should be combined. I could not find a supported runtime lifecycle that requires making those decisions.

So the fix I propose is:

  • keep QueryEntityMerger responsible for actual QueryEntity metadata merging
  • make patchCacheConfiguration() independent from the original configuration's mutable query-entity collection
  • keep testSchemaAddPreservesQueryEntityExMetadata as the integration regression test
  • add GridCacheUtilsTest.testPatchCacheConfigurationDoesNotModifyOriginalQueryEntities
    as the focused regression test for the root cause.
    With this change, the QueryEntityEx created by the schema-add operation is processed correctly without introducing merge rules for its extended metadata.

If there is a supported path where two independently created QueryEntityEx instances are expected to be merged, I can add handling for that path as well, but I could not reproduce or identify such a case in the current cache/schema lifecycle.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

probably it`s better to rewrite like :
QueryEntity res = ex instanceof QueryEntityEx ? new QueryEntityEx(ex) : new QueryEntity(ex);
and roll back code in
org.apache.ignite.internal.processors.cache.GridCacheUtils#patchCacheConfiguration
I think it become more clear, wdyt ?


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;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see that result of this operation is used in : org.apache.ignite.cache.QueryEntity#setFields but there is no @Nullable annotation, thus you need to return smth like: return U.newLinkedHashMap(0);


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;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
return null;
return Map.of();


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;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
return null;
return Set.of();


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;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
return null;
return List.of();


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(
String.format(CONFLICT_MESSAGE_TEMPLATE, cacheName, propName, existingVal, incomingVal)
);
}
}
Loading
Loading