Skip to content

Commit

Permalink
GEODE-9602: Changes after review
Browse files Browse the repository at this point in the history
  • Loading branch information
@albertogpz
albertogpz committed Sep 21, 2021
1 parent d32b31c commit 96a15bc
Show file tree
Hide file tree
Showing 5 changed files with 62 additions and 75 deletions.
5 changes: 3 additions & 2 deletions ...ntegrationTest/java/org/apache/geode/cache/query/internal/QueryObserverCallbacksTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
Expand Down Expand Up @@ -251,8 +252,8 @@ public void testBeforeAndAfterIterationEvaluateNoWhere() throws Exception {
"select count(*) from " + SEPARATOR + "portfolio p");

query.execute();
verify(myQueryObserver, times(0)).beforeIterationEvaluation(any(), any());
verify(myQueryObserver, times(0)).afterIterationEvaluation(any());
verify(myQueryObserver, never()).beforeIterationEvaluation(any(), any());
verify(myQueryObserver, never()).afterIterationEvaluation(any());
}

@Test
Expand Down
1 change: 1 addition & 0 deletions ...e/src/integrationTest/java/org/apache/geode/cache/query/internal/QueryTraceJUnitTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,7 @@ public void testTraceOnLocalRegionWithTracePrefixNoComments() throws Exception {
BeforeQueryExecutionHook hook = (BeforeQueryExecutionHook) DefaultQuery.testHook;
assertThat(hook.getObserver()).isInstanceOf(IndexTrackingQueryObserver.class);

// The query should return all elements in region.
assertEquals(region.size(), results.size());
QueryObserverHolder.reset();
}
Expand Down
23 changes: 12 additions & 11 deletions geode-core/src/main/java/org/apache/geode/cache/query/internal/QueryObserverHolder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
*/
package org.apache.geode.cache.query.internal;

import java.util.concurrent.atomic.AtomicReference;

import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;

Expand Down Expand Up @@ -49,31 +51,30 @@ public class QueryObserverHolder {
* The current observer which will be notified of all query events.
*/
@MakeNotStatic
private static QueryObserver _instance = NO_OBSERVER;
private static final AtomicReference<QueryObserver> _instance =
new AtomicReference<>(NO_OBSERVER);

/**
* Set the given observer to be notified of query events. Returns the current observer.
*/
public static synchronized QueryObserver setInstance(QueryObserver observer) {
public static QueryObserver setInstance(QueryObserver observer) {
Support.assertArg(observer != null, "setInstance expects a non-null argument!");
QueryObserver oldObserver = _instance;
_instance = observer;
return oldObserver;
return _instance.getAndSet(observer);
}

public static synchronized boolean hasObserver() {
return _instance != NO_OBSERVER;
public static boolean hasObserver() {
return _instance.get() != NO_OBSERVER;
}

/** Return the current QueryObserver instance */
public static synchronized QueryObserver getInstance() {
return _instance;
public static QueryObserver getInstance() {
return _instance.get();
}

/**
* Only for test purposes.
*/
public static synchronized void reset() {
_instance = NO_OBSERVER;
public static void reset() {
_instance.set(NO_OBSERVER);
}
}
65 changes: 26 additions & 39 deletions geode-core/src/main/java/org/apache/geode/cache/query/internal/index/AbstractIndex.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1824,19 +1824,8 @@ void addValuesToCollection(Collection result, CompiledValue iterOp, RuntimeItera
synchronized (value) {
for (Object o1 : ((Iterable) value)) {
boolean ok = true;
if (reUpdateInProgress) {
// Compare the value in index with value in RegionEntry.
ok = verifyEntryAndIndexValue(entry, o1, context);
}
try {
if (ok && runtimeItr != null) {
runtimeItr.setCurrent(o1);
observer.beforeIterationEvaluation(iterOp, o1);
ok = QueryUtils.applyCondition(iterOp, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
ok = applyCondition(iterOp, runtimeItr, context, observer, o1, entry,
reUpdateInProgress, ok);
if (ok) {
applyProjection(projAttrib, context, result, o1, intermediateResults,
isIntersection);
Expand All @@ -1849,19 +1838,8 @@ void addValuesToCollection(Collection result, CompiledValue iterOp, RuntimeItera
} else {
for (Object o1 : ((Iterable) value)) {
boolean ok = true;
if (reUpdateInProgress) {
// Compare the value in index with value in RegionEntry.
ok = verifyEntryAndIndexValue(entry, o1, context);
}
try {
if (ok && runtimeItr != null) {
runtimeItr.setCurrent(o1);
observer.beforeIterationEvaluation(iterOp, o1);
ok = QueryUtils.applyCondition(iterOp, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
ok = applyCondition(iterOp, runtimeItr, context, observer, o1, entry,
reUpdateInProgress, ok);
if (ok) {
applyProjection(projAttrib, context, result, o1, intermediateResults,
isIntersection);
Expand All @@ -1873,19 +1851,8 @@ void addValuesToCollection(Collection result, CompiledValue iterOp, RuntimeItera
}
} else {
boolean ok = true;
if (reUpdateInProgress) {
// Compare the value in index with in RegionEntry.
ok = verifyEntryAndIndexValue(entry, value, context);
}
try {
if (ok && runtimeItr != null) {
runtimeItr.setCurrent(value);
observer.beforeIterationEvaluation(iterOp, value);
ok = QueryUtils.applyCondition(iterOp, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
ok = applyCondition(iterOp, runtimeItr, context, observer, value, entry,
reUpdateInProgress, ok);
if (ok) {
if (context.isCqQueryContext()) {
result.add(new CqEntry(((RegionEntry) e.getKey()).getKey(), value));
Expand All @@ -1899,6 +1866,26 @@ void addValuesToCollection(Collection result, CompiledValue iterOp, RuntimeItera
}
}

private boolean applyCondition(CompiledValue iterOp, RuntimeIterator runtimeItr,
ExecutionContext context, QueryObserver observer, Object value, RegionEntry entry,
boolean reUpdateInProgress, boolean ok) throws FunctionDomainException,
TypeMismatchException, NameResolutionException, QueryInvocationTargetException {
if (reUpdateInProgress) {
// Compare the value in index with in RegionEntry.
ok = verifyEntryAndIndexValue(entry, value, context);
}
try {
if (ok && runtimeItr != null) {
runtimeItr.setCurrent(value);
observer.beforeIterationEvaluation(iterOp, value);
ok = QueryUtils.applyCondition(iterOp, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
return ok;
}

private boolean verifyLimit(Collection result, int limit, ExecutionContext context) {
if (limit > 0) {
if (!context.isDistinct()) {
Expand Down
43 changes: 20 additions & 23 deletions geode-core/src/main/java/org/apache/geode/cache/query/internal/index/CompactRangeIndex.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -805,18 +805,7 @@ private void addToResultsFromEntries(Object lowerBoundKey, Object upperBoundKey,
value = iterator.next();
if (value != null) {
boolean ok = true;

if (runtimeItr != null) {
runtimeItr.setCurrent(value);
}
try {
if (ok && runtimeItr != null) {
observer.beforeIterationEvaluation(iterOps, value);
ok = QueryUtils.applyCondition(iterOps, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
ok = applyCondition(iterOps, runtimeItr, context, observer, value, ok);
if (ok) {
applyCqOrProjection(projAttrib, context, result, value, intermediateResults,
isIntersection, indexEntry.getDeserializedRegionKey());
Expand Down Expand Up @@ -845,17 +834,7 @@ private void addToResultsFromEntries(Object lowerBoundKey, Object upperBoundKey,

ok = evaluateEntry((IndexInfo) indexInfo, context, null);
}
if (runtimeItr != null) {
runtimeItr.setCurrent(value);
}
try {
if (ok && runtimeItr != null) {
observer.beforeIterationEvaluation(iterOps, value);
ok = QueryUtils.applyCondition(iterOps, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
ok = applyCondition(iterOps, runtimeItr, context, observer, value, ok);
if (ok) {
if (context != null && context.isCqQueryContext()) {
result.add(new CqEntry(indexEntry.getDeserializedRegionKey(), value));
Expand All @@ -879,6 +858,24 @@ private void addToResultsFromEntries(Object lowerBoundKey, Object upperBoundKey,
}
}

private boolean applyCondition(CompiledValue iterOps, RuntimeIterator runtimeItr,
ExecutionContext context, QueryObserver observer, Object value, boolean ok)
throws FunctionDomainException, TypeMismatchException, NameResolutionException,
QueryInvocationTargetException {
if (runtimeItr != null) {
runtimeItr.setCurrent(value);
}
try {
if (ok && runtimeItr != null) {
observer.beforeIterationEvaluation(iterOps, value);
ok = QueryUtils.applyCondition(iterOps, context);
}
} finally {
observer.afterIterationEvaluation(ok);
}
return ok;
}

public List expandValue(ExecutionContext context, Object lowerBoundKey, Object upperBoundKey,
int lowerBoundOperator, int upperBoundOperator, Object value) {
try {
Expand Down

0 comments on commit 96a15bc

Please sign in to comment.