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
+169 −5
Merged

IGNITE-22183 Recording of Index query events added #11342

Show file tree
Hide file tree
Changes from 1 commit
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
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
}
41 changes: 39 additions & 2 deletions ...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,6 +629,22 @@ 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)
Expand Down Expand Up @@ -1203,7 +1219,7 @@ protected void runQuery(GridCacheQueryInfo qryInfo) {
break;
}

if (type == SCAN || type == INDEX)
Copy link
Member

Choose a reason for hiding this comment

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

This change might affect IndexQuery performance due to useless generics handling. Let's check how ScanQuery record this event.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Recording of Index query object reading events moved to IndexQueryProcessor#queryLocal.

if (type == SCAN)
// Scan iterator may return already transformed entry
data.add(row0);
else {
Expand Down Expand Up @@ -1289,6 +1305,27 @@ protected void runQuery(GridCacheQueryInfo qryInfo) {
null));

break;

case INDEX:
cctx.gridEvents().record(new CacheQueryReadEvent<>(
cctx.localNode(),
"Index query entry read.",
EVT_CACHE_QUERY_OBJECT_READ,
CacheQueryType.INDEX.name(),
cctx.name(),
qry.queryClassName(),
null,
qry.scanFilter(),
null,
null,
securitySubjectId(cctx),
taskName,
key0,
val0,
null,
null));

break;
}
}

Expand All @@ -1312,7 +1349,7 @@ protected void runQuery(GridCacheQueryInfo qryInfo) {
continue;
}
else {
if (type == TEXT)
if (type == TEXT || type == INDEX)
// (K, V, score). Value transfers as BinaryObject.
data.add(row0);
else
Expand Down
113 changes: 113 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,114 @@ 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());
}
}
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