Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

IGNITE-22183 Recording of Index query events added #11342

Merged
merged 5 commits into from
May 15, 2024
Merged

IGNITE-22183 Recording of Index query events added #11342

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2219e1f
IGNITE-22183 Recording of Index query events added
oleg-vlsk May 7, 2024
106d635
IGNITE-22183 Recording of Index query object reading events moved to …
oleg-vlsk May 9, 2024
d9ec35c
IGNITE-22183 Additional check for 'evtsDisabled = true' added to Igni…
oleg-vlsk May 9, 2024
636762b
IGNITE-22183 Further optimization of IndexQueryProcessor#queryLocal
oleg-vlsk May 14, 2024
7e4ad9b
IGNITE-22183 Object read event recording moved from onNext() to onHas…
oleg-vlsk May 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
40 changes: 39 additions & 1 deletion .../core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexQueryProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.query.IndexQuery;
import org.apache.ignite.cache.query.IndexQueryCriterion;
import org.apache.ignite.events.CacheQueryReadEvent;
import org.apache.ignite.internal.cache.query.RangeIndexQueryCriterion;
import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
Expand All @@ -41,6 +42,7 @@
import org.apache.ignite.internal.processors.cache.CacheObjectUtils;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.query.CacheQueryType;
import org.apache.ignite.internal.processors.cache.query.IndexQueryDesc;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.util.GridCloseableIteratorAdapter;
Expand All @@ -52,6 +54,9 @@
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_OBJECT_READ;
import static org.apache.ignite.internal.processors.security.SecurityUtils.securitySubjectId;

/**
* Processor of {@link IndexQuery}.
*/
Expand All @@ -74,7 +79,8 @@ public <K, V> IndexQueryResult<K, V> queryLocal(
IndexQueryDesc idxQryDesc,
@Nullable IgniteBiPredicate<K, V> filter,
IndexingQueryFilter cacheFilter,
boolean keepBinary
boolean keepBinary,
int taskHash
) throws IgniteCheckedException {
InlineIndexImpl idx = (InlineIndexImpl)findSortedIndex(cctx, idxQryDesc);

Expand Down Expand Up @@ -125,6 +131,38 @@ public <K, V> IndexQueryResult<K, V> queryLocal(

IgniteBiTuple<K, V> row = currVal;

if (cctx.events().isRecordable(EVT_CACHE_QUERY_OBJECT_READ) &&
timoninmaxim marked this conversation as resolved.
Show resolved Hide resolved
cctx.gridEvents().hasListener(EVT_CACHE_QUERY_OBJECT_READ)) {
timoninmaxim marked this conversation as resolved.
Show resolved Hide resolved

final K key = row.getKey();
final V val = row.getValue();

CacheObjectContext objCtx = cctx.cacheObjectContext();

K key0 = (K)CacheObjectUtils.unwrapBinaryIfNeeded(objCtx, key, keepBinary, false, null);
Copy link
Member

Choose a reason for hiding this comment

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

This code duplicates code in the onHasNext method.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Made this section of code more concise by using the unwrap() method.

V val0 = (V)CacheObjectUtils.unwrapBinaryIfNeeded(objCtx, val, keepBinary, false, null);

String taskName = cctx.kernalContext().task().resolveTaskName(taskHash);

cctx.gridEvents().record(new CacheQueryReadEvent<>(
cctx.localNode(),
"Index query entry read.",
EVT_CACHE_QUERY_OBJECT_READ,
CacheQueryType.INDEX.name(),
cctx.name(),
idxQryDesc.valType(),
null,
filter,
null,
null,
securitySubjectId(cctx),
taskName,
key0,
val0,
null,
null));
}

currVal = null;

return row;
Expand Down
5 changes: 4 additions & 1 deletion .../core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQueryType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,5 +43,8 @@ public enum CacheQueryType {
CONTINUOUS,

/** SPI query. */
SPI
SPI,

/** Index query. */
INDEX
}
18 changes: 17 additions & 1 deletion ...rc/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -629,13 +629,29 @@ private QueryResult<K, V> executeQuery(GridCacheQueryAdapter<?> qry, @Nullable O
break;

case INDEX:
if (cctx.events().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
cctx.gridEvents().record(new CacheQueryExecutedEvent<>(
cctx.localNode(),
"Index query executed.",
EVT_CACHE_QUERY_EXECUTED,
CacheQueryType.INDEX.name(),
cctx.name(),
qry.queryClassName(),
null,
qry.scanFilter(),
null,
null,
securitySubjectId(cctx),
taskName));
}

int[] parts = null;

if (qry.partition() != null)
parts = new int[]{qry.partition()};

IndexQueryResult<K, V> idxQryRes = qryProc.queryIndex(cacheName, qry.queryClassName(), qry.idxQryDesc(),
qry.scanFilter(), filter(qry, parts, parts != null), qry.keepBinary());
qry.scanFilter(), filter(qry, parts, parts != null), qry.keepBinary(), qry.taskHash());

iter = idxQryRes.iter();
res.metadata(idxQryRes.metadata());
Expand Down
6 changes: 4 additions & 2 deletions ...es/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3657,6 +3657,7 @@ public <K, V> GridCloseableIterator<IgniteBiTuple<K, V>> queryText(final String
* @param entryFilter Optional user defined cache entries filter.
* @param cacheFilter Ignite specific cache entries filters.
* @param keepBinary Keep binary flag.
* @param taskHash Hashcode of the task.
* @return Key/value rows.
* @throws IgniteCheckedException If failed.
*/
Expand All @@ -3666,7 +3667,8 @@ public <K, V> IndexQueryResult<K, V> queryIndex(
final IndexQueryDesc idxQryDesc,
@Nullable IgniteBiPredicate<K, V> entryFilter,
final IndexingQueryFilter cacheFilter,
boolean keepBinary
boolean keepBinary,
int taskHash
) throws IgniteCheckedException {
if (!busyLock.enterBusy())
throw new IllegalStateException("Failed to execute query (grid is stopping).");
Expand All @@ -3678,7 +3680,7 @@ public <K, V> IndexQueryResult<K, V> queryIndex(
new IgniteOutClosureX<IndexQueryResult<K, V>>() {
@Override public IndexQueryResult<K, V> applyx() throws IgniteCheckedException {
try {
return idxQryPrc.queryLocal(cctx, idxQryDesc, entryFilter, cacheFilter, keepBinary);
return idxQryPrc.queryLocal(cctx, idxQryDesc, entryFilter, cacheFilter, keepBinary, taskHash);
}
catch (IgniteCheckedException e) {
String msg = "Failed to execute IndexQuery: " + e.getMessage() + ". Query desc: " + idxQryDesc;
Expand Down
115 changes: 115 additions & 0 deletions ...st/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.query.IndexQuery;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cache.query.SqlFieldsQuery;
Expand All @@ -80,6 +81,7 @@
import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.processors.cache.query.QueryCursorEx;
import org.apache.ignite.internal.processors.query.GridQueryFieldMetadata;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.util.lang.GridPlainCallable;
import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
Expand All @@ -101,10 +103,13 @@
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.cache.query.IndexQueryCriteriaBuilder.gt;
import static org.apache.ignite.cache.query.IndexQueryCriteriaBuilder.lt;
import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_EXECUTED;
import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_OBJECT_READ;
import static org.apache.ignite.events.EventType.EVT_SQL_QUERY_EXECUTION;
import static org.apache.ignite.internal.processors.cache.query.CacheQueryType.FULL_TEXT;
import static org.apache.ignite.internal.processors.cache.query.CacheQueryType.INDEX;
import static org.apache.ignite.internal.processors.cache.query.CacheQueryType.SCAN;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause;
import static org.junit.Assert.assertArrayEquals;
Expand Down Expand Up @@ -1883,6 +1888,116 @@ public void testFieldsQueryEvents() throws Exception {
}
}

/**
* @throws Exception If failed.
*/
@Test
public void testIndexQueryEvents() throws Exception {
final Map<Integer, Type2> qryResults = new ConcurrentHashMap<>();
final IgniteCache<Integer, Type2> cache = jcache(Integer.class, Type2.class);
final boolean evtsDisabled = cache.getConfiguration(CacheConfiguration.class).isEventsDisabled();

final CountDownLatch readLatch = new CountDownLatch(evtsDisabled ? 0 : 2);
final CountDownLatch execLatch = new CountDownLatch(evtsDisabled ? 0 :
cacheMode() == REPLICATED ? 1 : gridCount());

IgnitePredicate[] objReadLsnrs = new IgnitePredicate[gridCount()];
IgnitePredicate[] qryExecLsnrs = new IgnitePredicate[gridCount()];

for (int i = 0; i < gridCount(); i++) {
IgnitePredicate<Event> objReadPred = new IgnitePredicate<Event>() {
@Override public boolean apply(Event evt) {
assert evt instanceof CacheQueryReadEvent;

if (evtsDisabled)
fail("Cache events are disabled");

CacheQueryReadEvent<Integer, Type2> qe = (CacheQueryReadEvent<Integer, Type2>)evt;

assertEquals(INDEX.name(), qe.queryType());
assertEquals(cache.getName(), qe.cacheName());
assertEquals("Type2", QueryUtils.typeName(qe.className()));
assertNotNull(qe.scanQueryFilter());
assertNull(qe.clause());
assertNull(qe.continuousQueryFilter());
assertNull(qe.arguments());

qryResults.put(qe.key(), qe.value());

readLatch.countDown();

return true;
}
};

grid(i).events().localListen(objReadPred, EVT_CACHE_QUERY_OBJECT_READ);
objReadLsnrs[i] = objReadPred;

IgnitePredicate<Event> qryExecPred = new IgnitePredicate<Event>() {
@Override public boolean apply(Event evt) {
assert evt instanceof CacheQueryExecutedEvent;

if (evtsDisabled)
fail("Cache events are disabled");

CacheQueryExecutedEvent qe = (CacheQueryExecutedEvent)evt;

assertEquals(INDEX.name(), qe.queryType());
assertEquals(cache.getName(), qe.cacheName());
assertEquals("Type2", QueryUtils.typeName(qe.className()));
assertNotNull(qe.scanQueryFilter());
assertNull(qe.clause());
assertNull(qe.continuousQueryFilter());
assertNull(qe.arguments());

execLatch.countDown();

return true;
}
};

grid(i).events().localListen(qryExecPred, EVT_CACHE_QUERY_EXECUTED);
qryExecLsnrs[i] = qryExecPred;
}

try {
cache.put(1, new Type2(1, "John"));
cache.put(2, new Type2(2, "Bill"));
cache.put(3, new Type2(3, "Sam"));
cache.put(4, new Type2(4, "Bill"));
cache.put(5, new Type2(5, "Bob"));

IndexQuery<Integer, Type2> qry = new IndexQuery<Integer, Type2>(Type2.class)
.setCriteria(gt("id", 1), lt("id", 5))
.setFilter((k, v) -> v.name().contains("Bill"));

if (cacheMode() == REPLICATED)
qry.setLocal(true);

QueryCursor<Cache.Entry<Integer, Type2>> cursor = cache.query(qry);

cursor.getAll();

assert readLatch.await(1000, MILLISECONDS);
assert execLatch.await(1000, MILLISECONDS);

if (!evtsDisabled) {
timoninmaxim marked this conversation as resolved.
Show resolved Hide resolved
assertEquals(2, qryResults.size());

assertEquals("Bill", qryResults.get(2).name());
assertEquals("Bill", qryResults.get(4).name());
}
else
assert qryResults.isEmpty();
}
finally {
for (int i = 0; i < gridCount(); i++) {
grid(i).events().stopLocalListen(objReadLsnrs[i]);
grid(i).events().stopLocalListen(qryExecLsnrs[i]);
}
}
}

/**
* @throws Exception If failed.
*/
Expand Down