Skip to content

Commit e4bda80

Browse files
ivakeggapmoriarty
ivakegg
and
apmoriarty
authored
Task/cherry pick moriarty fix (#3413)
Co-authored-by: Moriarty <22225248+apmoriarty@users.noreply.github.com>
1 parent b7dfc8b commit e4bda80

31 files changed

Lines changed: 1518 additions & 135 deletions

‎warehouse/query-core/src/main/java/datawave/core/iterators/FieldedRegexExpansionIterator.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.apache.accumulo.core.iterators.OptionDescriber;
2020
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
2121
import org.apache.accumulo.core.iterators.user.SeekingFilter;
22+
import org.apache.commons.lang3.tuple.Pair;
2223
import org.apache.hadoop.io.Text;
2324
import org.slf4j.Logger;
2425
import org.slf4j.LoggerFactory;
@@ -132,6 +133,10 @@ public FilterResult filter(Key k, Value v) {
132133
log.debug("tk: {}", k.toStringNoTime());
133134
}
134135

136+
if (TimeoutExceptionIterator.exceededTimedValue(Pair.of(k, v))) {
137+
return new FilterResult(true, AdvanceResult.NEXT);
138+
}
139+
135140
// parse key and reset hint
136141
parser.parse(k);
137142
hint = HINT_TYPE.NONE;

‎warehouse/query-core/src/main/java/datawave/core/iterators/TimeoutExceptionIterator.java‎

Lines changed: 71 additions & 75 deletions
Original file line numberDiff line numberDiff line change
@@ -15,21 +15,25 @@
1515
import org.apache.accumulo.core.iterators.WrappingIterator;
1616

1717
/**
18-
* Purpose: Timeout catching exception Iterator that will work in conjunction with an iterator that returns the timeout exception
18+
* Purpose: Timeout catching exception Iterator that will work in conjunction with an iterator that returns the timeout exception Once a timeout is detected,
19+
* the last key will be returned again with a value set to EXCEPTEDVALUE.
1920
*/
2021
public class TimeoutExceptionIterator extends WrappingIterator {
2122

2223
/**
23-
* Last key that we've received from the iterators below us
24+
* Various states of this iterator
2425
*/
25-
protected Key lastKey = null;
26+
private enum STATE {
27+
NORMAL, // life as usual
28+
TIMEOUT, // we have detected a timeout
29+
COMPLETE // nothing left to do
30+
}
2631

27-
protected Value lastValue = null;
32+
// The current state of the iterator
33+
STATE state = STATE.NORMAL;
2834

29-
/**
30-
* boolean to identify that we've exceeded the time
31-
*/
32-
boolean exceededTime = false;
35+
// The next key to return
36+
Key nextKey;
3337

3438
// Exceeded timeout value exception marker
3539
public static final Value EXCEPTEDVALUE = new Value(new byte[] {0x0d, 0x0e, 0x0a, 0x0d, 0x0b, 0x0e, 0x0e, 0x0f});
@@ -38,105 +42,97 @@ public static boolean exceededTimedValue(Entry<Key,Value> kv) {
3842
return kv.getValue().equals(EXCEPTEDVALUE);
3943
}
4044

45+
public static boolean exceededTimedValue(Value value) {
46+
return value.equals(EXCEPTEDVALUE);
47+
}
48+
4149
@Override
4250
public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
4351
super.init(source, options, env);
4452
}
4553

46-
/**
47-
* Set the return key
48-
*/
49-
protected void setReturnKey() {
50-
lastValue = EXCEPTEDVALUE;
51-
exceededTime = true;
52-
}
53-
5454
@Override
5555
public boolean hasTop() {
56-
if (exceededTime) {
57-
return null != lastKey;
56+
switch (state) {
57+
case NORMAL:
58+
return super.hasTop();
59+
case TIMEOUT:
60+
// we have one more key to return
61+
return true;
62+
default:
63+
return false;
5864
}
59-
return super.hasTop();
6065
}
6166

6267
@Override
6368
public Value getTopValue() {
64-
if (exceededTime) {
65-
if (null != lastValue) {
66-
Value returnValue = lastValue;
67-
lastValue = null;
68-
return returnValue;
69-
} else {
69+
switch (state) {
70+
case NORMAL:
71+
return super.getTopValue();
72+
case TIMEOUT:
73+
return EXCEPTEDVALUE;
74+
default:
7075
return null;
71-
}
7276
}
73-
lastValue = super.getTopValue();
74-
return lastValue;
7577
}
7678

7779
@Override
7880
public Key getTopKey() {
79-
if (exceededTime) {
80-
if (null != lastKey) {
81-
Key returnKey = lastKey;
82-
lastKey = null;
83-
return returnKey;
84-
} else {
81+
switch (state) {
82+
case NORMAL:
83+
case TIMEOUT:
84+
return nextKey;
85+
default:
8586
return null;
86-
}
8787
}
88-
89-
lastKey = super.getTopKey();
90-
return lastKey;
9188
}
9289

9390
@Override
9491
public void next() throws IOException {
95-
if (exceededTime) {
96-
return;
97-
}
98-
99-
try {
100-
super.next();
101-
} catch (IteratorTimeoutException e) {
102-
setReturnKey();
103-
} catch (RuntimeException e) {
104-
if (e.getCause() instanceof IteratorTimeoutException) {
105-
setReturnKey();
106-
} else {
107-
throw e;
108-
}
109-
}
92+
move(super::next);
11093
}
11194

11295
@Override
11396
public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
114-
115-
if (!range.isInfiniteStartKey()) {
116-
if (range.isStartKeyInclusive()) {
117-
lastKey = range.getStartKey();
118-
} else {
119-
lastKey = range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME);
120-
}
97+
// setup a next key in case we timeout immediately
98+
if (range.getStartKey() == null) {
99+
nextKey = new Key();
100+
} else if (range.isStartKeyInclusive()) {
101+
nextKey = range.getStartKey();
121102
} else {
122-
lastKey = new Key();
103+
nextKey = range.getStartKey().followingKey(PartialKey.ROW_COLFAM_COLQUAL_COLVIS_TIME);
123104
}
124-
lastValue = new Value();
125105

126-
if (exceededTime) {
127-
return;
128-
}
106+
move(() -> super.seek(range, columnFamilies, inclusive));
107+
}
129108

130-
try {
131-
super.seek(range, columnFamilies, inclusive);
132-
} catch (IteratorTimeoutException e) {
133-
setReturnKey();
134-
} catch (RuntimeException e) {
135-
if (e.getCause() instanceof IteratorTimeoutException) {
136-
setReturnKey();
137-
} else {
138-
throw e;
139-
}
109+
private void move(IOAction a) throws IOException {
110+
switch (state) {
111+
case NORMAL:
112+
try {
113+
a.call();
114+
if (super.hasTop()) {
115+
nextKey = super.getTopKey();
116+
} else {
117+
state = STATE.COMPLETE;
118+
}
119+
} catch (IteratorTimeoutException e) {
120+
state = STATE.TIMEOUT;
121+
} catch (RuntimeException e) {
122+
if (e.getCause() instanceof IteratorTimeoutException) {
123+
state = STATE.TIMEOUT;
124+
} else {
125+
state = STATE.COMPLETE;
126+
throw e;
127+
}
128+
}
129+
break;
130+
case TIMEOUT:
131+
state = STATE.COMPLETE;
140132
}
141133
}
134+
135+
private interface IOAction {
136+
void call() throws IOException;
137+
}
142138
}

‎warehouse/query-core/src/main/java/datawave/query/config/ShardQueryConfiguration.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ public class ShardQueryConfiguration extends GenericQueryConfiguration implement
104104
private boolean allTermsIndexOnly;
105105
private long maxIndexScanTimeMillis = Long.MAX_VALUE;
106106
private long maxAnyFieldScanTimeMillis = Long.MAX_VALUE;
107+
private boolean useNewIndexLookups = false;
107108

108109
// Allows this query to parse the root uids from TLD uids found in the global shard index. This effectively ignores hits in child documents.
109110
private boolean parseTldUids = false;
@@ -628,6 +629,7 @@ public void copyFrom(ShardQueryConfiguration other) {
628629
this.setAllTermsIndexOnly(other.isAllTermsIndexOnly());
629630
this.setMaxIndexScanTimeMillis(other.getMaxIndexScanTimeMillis());
630631
this.setMaxAnyFieldScanTimeMillis(other.getMaxAnyFieldScanTimeMillis());
632+
this.setUseNewIndexLookups(other.isUseNewIndexLookups());
631633
this.setCollapseUids(other.getCollapseUids());
632634
this.setCollapseUidsThreshold(other.getCollapseUidsThreshold());
633635
this.setEnforceUniqueTermsWithinExpressions(other.getEnforceUniqueTermsWithinExpressions());
@@ -3223,6 +3225,7 @@ public boolean equals(Object o) {
32233225
isUseQueryTreeScanHintRules() == that.isUseQueryTreeScanHintRules() &&
32243226
getMaxLinesToPrint() == that.getMaxLinesToPrint() &&
32253227
getMaxAnyFieldScanTimeMillis() == that.getMaxAnyFieldScanTimeMillis() &&
3228+
isUseNewIndexLookups() == that.isUseNewIndexLookups() &&
32263229
isDisableIteratorUniqueFields() == that.isDisableIteratorUniqueFields() &&
32273230
isUseShardedIndex() == that.isUseShardedIndex() &&
32283231
getDayIndexThreshold() == that.getDayIndexThreshold() &&
@@ -3460,6 +3463,7 @@ public int hashCode() {
34603463
isUseQueryTreeScanHintRules(),
34613464
getMaxLinesToPrint(),
34623465
getMaxAnyFieldScanTimeMillis(),
3466+
isUseNewIndexLookups(),
34633467
isDisableIteratorUniqueFields(),
34643468
isUseShardedIndex(),
34653469
getDayIndexThreshold(),
@@ -3605,4 +3609,12 @@ public String getOriginalJexlQuery() {
36053609
public void setOriginalJexlQuery(String originalJexlQuery) {
36063610
this.originalJexlQuery = originalJexlQuery;
36073611
}
3612+
3613+
public boolean isUseNewIndexLookups() {
3614+
return useNewIndexLookups;
3615+
}
3616+
3617+
public void setUseNewIndexLookups(boolean useNewIndexLookups) {
3618+
this.useNewIndexLookups = useNewIndexLookups;
3619+
}
36083620
}

‎warehouse/query-core/src/main/java/datawave/query/jexl/lookups/BaseRegexIndexLookup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ protected String getHintKey(String tableName) {
5050

5151
protected IteratorSetting createTimeoutIterator() {
5252
long maxTime = (long) (config.getMaxIndexScanTimeMillis() * 1.25);
53-
IteratorSetting iterator = new IteratorSetting(1, TimeoutIterator.class);
53+
IteratorSetting iterator = new IteratorSetting(5, TimeoutIterator.class);
5454
iterator.addOption(TimeoutIterator.MAX_SESSION_TIME, Long.valueOf(maxTime).toString());
5555
return iterator;
5656
}

‎warehouse/query-core/src/main/java/datawave/query/jexl/lookups/BoundedRangeIndexLookup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ public synchronized void submit() {
8989
IteratorSetting fairnessIterator = null;
9090
if (config.getMaxIndexScanTimeMillis() > 0) {
9191
// The fairness iterator solves the problem whereby we have runaway iterators as a result of an evaluation that never finds anything
92-
fairnessIterator = new IteratorSetting(1, TimeoutIterator.class);
92+
fairnessIterator = new IteratorSetting(5, TimeoutIterator.class);
9393

9494
long maxTime = config.getMaxIndexScanTimeMillis();
9595
if (maxTime < Long.MAX_VALUE / 2)

‎warehouse/query-core/src/main/java/datawave/query/jexl/lookups/FieldedRegexIndexLookup.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -89,12 +89,15 @@ public void submit() {
8989
}
9090

9191
} catch (ExceededThresholdException e) {
92-
log.info("ExceededThresholdException", e);
92+
log.warn("ExceededThresholdException", e);
9393
exceededValueThreshold.set(true);
9494
indexLookupMap.get(field).setThresholdExceeded();
9595
} catch (Exception e) {
9696
exceptionSeen.set(true);
97-
log.error(e.getMessage(), e);
97+
indexLookupMap.setExceptionSeen(true);
98+
indexLookupMap.setTimeoutExceeded(true); // stub this out
99+
indexLookupMap.get(field).setThresholdExceeded();
100+
log.error("Unexpected exception seen", e);
98101
} finally {
99102
latch.countDown();
100103
}

‎warehouse/query-core/src/main/java/datawave/query/jexl/lookups/RegexIndexLookup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ public synchronized void submit() {
122122
IteratorSetting fairnessIterator = null;
123123
if (config.getMaxIndexScanTimeMillis() > 0) {
124124
// The fairness iterator solves the problem whereby we have runaway iterators as a result of an evaluation that never finds anything
125-
fairnessIterator = new IteratorSetting(1, TimeoutIterator.class);
125+
fairnessIterator = new IteratorSetting(5, TimeoutIterator.class);
126126

127127
long maxTime = (long) (config.getMaxIndexScanTimeMillis() * 1.25);
128128
fairnessIterator.addOption(TimeoutIterator.MAX_SESSION_TIME, Long.valueOf(maxTime).toString());

‎warehouse/query-core/src/main/java/datawave/query/jexl/lookups/UnfieldedRegexIndexLookup.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import datawave.core.iterators.TimeoutExceptionIterator;
2020
import datawave.core.iterators.UnfieldedRegexExpansionIterator;
2121
import datawave.query.config.ShardQueryConfiguration;
22-
import datawave.query.exceptions.DatawaveFatalQueryException;
2322
import datawave.query.tables.ScannerFactory;
2423
import datawave.util.time.DateHelper;
2524

@@ -79,6 +78,7 @@ public void submit() {
7978

8079
if (TimeoutExceptionIterator.exceededTimedValue(entry)) {
8180
indexLookupMap.setTimeoutExceeded(true);
81+
indexLookupMap.clear(); // reset state so that ANYFIELD is marked NOFIELD
8282
break;
8383
}
8484

@@ -117,9 +117,6 @@ protected IteratorSetting createRegexIterator() {
117117
@Override
118118
public IndexLookupMap lookup() {
119119
await();
120-
if (indexLookupMap.isTimeoutExceeded()) {
121-
throw new DatawaveFatalQueryException("Unfielded regex expansion timed out");
122-
}
123120
return indexLookupMap;
124121
}
125122
}

‎warehouse/query-core/src/main/java/datawave/query/jexl/visitors/RegexIndexExpansionVisitor.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -282,7 +282,11 @@ public Object visit(ASTERNode node, Object data) {
282282
throw new DatawaveFatalQueryException(e);
283283
}
284284

285-
return buildIndexLookup(node, false, false, () -> createFieldedRegexIndexLookup(node));
285+
if (config.isUseNewIndexLookups()) {
286+
return buildIndexLookup(node, false, false, () -> createFieldedRegexIndexLookup(node));
287+
} else {
288+
return buildIndexLookup(node, false, false, () -> createLookup(node));
289+
}
286290
}
287291

288292
@Override

‎warehouse/query-core/src/main/java/datawave/query/jexl/visitors/UnfieldedIndexExpansionVisitor.java‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525

2626
import datawave.query.Constants;
2727
import datawave.query.config.ShardQueryConfiguration;
28+
import datawave.query.exceptions.DatawaveFatalQueryException;
2829
import datawave.query.exceptions.EmptyUnfieldedTermExpansionException;
2930
import datawave.query.jexl.JexlASTHelper;
3031
import datawave.query.jexl.JexlNodeFactory;
@@ -168,7 +169,11 @@ public Object visit(ASTNENode node, Object data) {
168169
public Object visit(ASTERNode node, Object data) {
169170
String field = JexlASTHelper.getIdentifier(node);
170171
if (field.equals(Constants.ANY_FIELD)) {
171-
return buildIndexLookup(node, true, negated, () -> createUnfieldedRegexIndexLookup(node));
172+
if (config.isUseNewIndexLookups()) {
173+
return buildIndexLookup(node, true, negated, () -> createUnfieldedRegexIndexLookup(node));
174+
} else {
175+
return buildIndexLookup(node, true, negated, () -> createLookup(node));
176+
}
172177
}
173178
// in the future a single index expansion visitor could handle all cases
174179
return copy(node);
@@ -180,7 +185,11 @@ public Object visit(ASTNRNode node, Object data) {
180185
try {
181186
String field = JexlASTHelper.getIdentifier(node);
182187
if (field.equals(Constants.ANY_FIELD)) {
183-
return buildIndexLookup(node, true, negated, () -> createUnfieldedRegexIndexLookup(node));
188+
if (config.isUseNewIndexLookups()) {
189+
return buildIndexLookup(node, true, negated, () -> createUnfieldedRegexIndexLookup(node));
190+
} else {
191+
return buildIndexLookup(node, true, negated, () -> createLookup(node));
192+
}
184193
}
185194
// in the future a single index expansion visitor could handle all cases
186195
return copy(node);
@@ -291,6 +300,17 @@ protected IndexLookup createUnfieldedRegexIndexLookup(JexlNode node) {
291300
return new UnfieldedRegexIndexLookup(config, scannerFactory, executor, pattern, description.range, description.isForReverseIndex, expansionFields);
292301
}
293302

303+
@Override
304+
protected IndexLookup createLookup(JexlNode node) {
305+
try {
306+
// Using the datatype filter when expanding this term isn't really
307+
// necessary
308+
return ShardIndexQueryTableStaticMethods.expandQueryTerms(node, config, scannerFactory, expansionFields, helper, executor);
309+
} catch (TableNotFoundException e) {
310+
throw new DatawaveFatalQueryException(e);
311+
}
312+
}
313+
294314
/**
295315
* Get the set of fields used to restrict this index expansion operation
296316
*

0 commit comments

Comments
 (0)