Skip to content

Commit a7cd169

Browse files
weizijunweizijun
committed
commit
1 parent 87f0c6d commit a7cd169

File tree

5 files changed

+70
-44
lines changed

5 files changed

+70
-44
lines changed

‎benchmarks/src/main/java/org/elasticsearch/benchmark/search/fetch/subphase/FetchSourcePhaseBenchmark.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -89,11 +89,11 @@ private BytesReference buildBigExample(String extraText) throws IOException {
8989
}
9090

9191
@Benchmark
92-
public BytesReference filterObjects() throws IOException {
92+
public BytesReference filterObjects() {
9393
SourceLookup lookup = new SourceLookup();
9494
lookup.setSource(sourceBytes);
95-
Object value = lookup.filter(fetchContext);
96-
return FetchSourcePhase.objectToBytes(value, XContentType.JSON, Math.min(1024, lookup.internalSourceRef().length()));
95+
lookup.setSourceContentType(XContentType.JSON);
96+
return lookup.filter(fetchContext);
9797
}
9898

9999
@Benchmark

‎server/src/main/java/org/elasticsearch/action/update/UpdateHelper.java

Lines changed: 2 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,11 @@
1010

1111
import org.apache.logging.log4j.LogManager;
1212
import org.apache.logging.log4j.Logger;
13-
import org.elasticsearch.ElasticsearchException;
1413
import org.elasticsearch.action.DocWriteResponse;
1514
import org.elasticsearch.action.delete.DeleteRequest;
1615
import org.elasticsearch.action.index.IndexRequest;
1716
import org.elasticsearch.client.Requests;
1817
import org.elasticsearch.common.bytes.BytesReference;
19-
import org.elasticsearch.common.io.stream.BytesStreamOutput;
2018
import org.elasticsearch.common.io.stream.Writeable;
2119
import org.elasticsearch.common.xcontent.XContentHelper;
2220
import org.elasticsearch.core.Nullable;
@@ -33,10 +31,8 @@
3331
import org.elasticsearch.script.ScriptService;
3432
import org.elasticsearch.script.UpdateScript;
3533
import org.elasticsearch.search.lookup.SourceLookup;
36-
import org.elasticsearch.xcontent.XContentBuilder;
3734
import org.elasticsearch.xcontent.XContentType;
3835

39-
import java.io.IOException;
4036
import java.util.Collections;
4137
import java.util.HashMap;
4238
import java.util.Map;
@@ -339,17 +335,8 @@ public static GetResult extractGetResult(
339335
if (request.fetchSource().includes().length > 0 || request.fetchSource().excludes().length > 0) {
340336
SourceLookup sourceLookup = new SourceLookup();
341337
sourceLookup.setSource(source);
342-
Object value = sourceLookup.filter(request.fetchSource());
343-
try {
344-
final int initialCapacity = sourceAsBytes != null ? Math.min(1024, sourceAsBytes.length()) : 1024;
345-
BytesStreamOutput streamOutput = new BytesStreamOutput(initialCapacity);
346-
try (XContentBuilder builder = new XContentBuilder(sourceContentType.xContent(), streamOutput)) {
347-
builder.value(value);
348-
sourceFilteredAsBytes = BytesReference.bytes(builder);
349-
}
350-
} catch (IOException e) {
351-
throw new ElasticsearchException("Error filtering source", e);
352-
}
338+
sourceLookup.setSourceContentType(sourceContentType);
339+
sourceFilteredAsBytes = sourceLookup.filter(request.fetchSource());
353340
}
354341

355342
// TODO when using delete/none, we can still return the source as bytes by generating it (using the sourceContentType)

‎server/src/main/java/org/elasticsearch/search/fetch/subphase/FetchSourceContext.java

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,24 +8,28 @@
88

99
package org.elasticsearch.search.fetch.subphase;
1010

11+
import org.elasticsearch.ElasticsearchException;
1112
import org.elasticsearch.common.ParsingException;
1213
import org.elasticsearch.common.Strings;
14+
import org.elasticsearch.common.bytes.BytesReference;
15+
import org.elasticsearch.common.io.stream.BytesStreamOutput;
1316
import org.elasticsearch.common.io.stream.StreamInput;
1417
import org.elasticsearch.common.io.stream.StreamOutput;
1518
import org.elasticsearch.common.io.stream.Writeable;
16-
import org.elasticsearch.common.xcontent.support.XContentMapValues;
19+
import org.elasticsearch.common.util.set.Sets;
1720
import org.elasticsearch.core.Booleans;
1821
import org.elasticsearch.rest.RestRequest;
1922
import org.elasticsearch.xcontent.ParseField;
2023
import org.elasticsearch.xcontent.ToXContentObject;
2124
import org.elasticsearch.xcontent.XContentBuilder;
2225
import org.elasticsearch.xcontent.XContentParser;
26+
import org.elasticsearch.xcontent.XContentParserConfiguration;
27+
import org.elasticsearch.xcontent.XContentType;
2328

2429
import java.io.IOException;
2530
import java.util.ArrayList;
2631
import java.util.Arrays;
2732
import java.util.List;
28-
import java.util.Map;
2933
import java.util.function.Function;
3034

3135
/**
@@ -41,7 +45,7 @@ public class FetchSourceContext implements Writeable, ToXContentObject {
4145
private final boolean fetchSource;
4246
private final String[] includes;
4347
private final String[] excludes;
44-
private Function<Map<String, ?>, Map<String, Object>> filter;
48+
private Function<BytesReference, BytesReference> filter;
4549

4650
public FetchSourceContext(boolean fetchSource, String[] includes, String[] excludes) {
4751
this.fetchSource = fetchSource;
@@ -246,12 +250,26 @@ public int hashCode() {
246250
}
247251

248252
/**
249-
* Returns a filter function that expects the source map as an input and returns
250-
* the filtered map.
253+
* Returns a filter function that expects the source as an input and returns
254+
* the filtered source.
251255
*/
252-
public Function<Map<String, ?>, Map<String, Object>> getFilter() {
256+
public Function<BytesReference, BytesReference> getFilter(final XContentType contentType) {
253257
if (filter == null) {
254-
filter = XContentMapValues.filter(includes, excludes);
258+
filter = (sourceBytes) -> {
259+
BytesStreamOutput streamOutput = new BytesStreamOutput(Math.min(1024, sourceBytes.length()));
260+
try (XContentBuilder builder = new XContentBuilder(XContentType.JSON.xContent(), streamOutput)) {
261+
XContentParserConfiguration parserConfig = XContentParserConfiguration.EMPTY.withFiltering(
262+
Sets.newHashSet(includes),
263+
Sets.newHashSet(excludes)
264+
);
265+
try (XContentParser parser = contentType.xContent().createParser(parserConfig, sourceBytes.streamInput())) {
266+
builder.copyCurrentStructure(parser);
267+
return BytesReference.bytes(builder);
268+
}
269+
} catch (IOException e) {
270+
throw new ElasticsearchException("Error filtering source");
271+
}
272+
};
255273
}
256274
return filter;
257275
}

‎server/src/main/java/org/elasticsearch/search/fetch/subphase/FetchSourcePhase.java

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import org.elasticsearch.ElasticsearchException;
1313
import org.elasticsearch.common.bytes.BytesReference;
1414
import org.elasticsearch.common.io.stream.BytesStreamOutput;
15+
import org.elasticsearch.common.xcontent.XContentHelper;
1516
import org.elasticsearch.search.SearchHit;
1617
import org.elasticsearch.search.fetch.FetchContext;
1718
import org.elasticsearch.search.fetch.FetchSubPhase;
@@ -67,17 +68,20 @@ private void hitExecute(FetchSourceContext fetchSourceContext, HitContext hitCon
6768
}
6869

6970
// Otherwise, filter the source and add it to the hit.
70-
Object value = source.filter(fetchSourceContext);
71+
BytesReference filterSource;
7172
if (nestedHit) {
72-
value = getNestedSource((Map<String, Object>) value, hitContext);
73-
}
74-
75-
try {
76-
final int initialCapacity = nestedHit ? 1024 : Math.min(1024, source.internalSourceRef().length());
77-
hitContext.hit().sourceRef(objectToBytes(value, source.sourceContentType(), initialCapacity));
78-
} catch (IOException e) {
79-
throw new ElasticsearchException("Error filtering source", e);
73+
Object value = source.filter(fetchSourceContext);
74+
value = getNestedSource((BytesReference) value, hitContext);
75+
try {
76+
final int initialCapacity = nestedHit ? 1024 : Math.min(1024, source.internalSourceRef().length());
77+
filterSource = objectToBytes(value, source.sourceContentType(), initialCapacity);
78+
} catch (IOException e) {
79+
throw new ElasticsearchException("Error filtering source", e);
80+
}
81+
} else {
82+
filterSource = source.filter(fetchSourceContext);
8083
}
84+
hitContext.hit().sourceRef(filterSource);
8185
}
8286

8387
@Override
@@ -109,7 +113,8 @@ public static BytesReference objectToBytes(Object value, XContentType xContentTy
109113
}
110114

111115
@SuppressWarnings("unchecked")
112-
private Map<String, Object> getNestedSource(Map<String, Object> sourceAsMap, HitContext hitContext) {
116+
private Map<String, Object> getNestedSource(BytesReference source, HitContext hitContext) {
117+
Map<String, Object> sourceAsMap = XContentHelper.convertToMap(source, true).v2();
113118
for (SearchHit.NestedIdentity o = hitContext.hit().getNestedIdentity(); o != null; o = o.getChild()) {
114119
sourceAsMap = (Map<String, Object>) sourceAsMap.get(o.getField().string());
115120
if (sourceAsMap == null) {

‎server/src/main/java/org/elasticsearch/search/lookup/SourceLookup.java

Lines changed: 25 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.elasticsearch.core.Tuple;
2020
import org.elasticsearch.index.fieldvisitor.FieldsVisitor;
2121
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
22+
import org.elasticsearch.xcontent.XContentFactory;
2223
import org.elasticsearch.xcontent.XContentType;
2324

2425
import java.io.IOException;
@@ -64,28 +65,42 @@ public Map<String, Object> source() {
6465
}
6566
if (sourceAsBytes != null) {
6667
Tuple<XContentType, Map<String, Object>> tuple = sourceAsMapAndType(sourceAsBytes);
67-
sourceContentType = tuple.v1();
6868
source = tuple.v2();
6969
return source;
7070
}
7171
try {
72-
FieldsVisitor sourceFieldVisitor = new FieldsVisitor(true);
73-
fieldReader.accept(docId, sourceFieldVisitor);
74-
BytesReference source = sourceFieldVisitor.source();
72+
BytesReference source = sourceAsBytes();
7573
if (source == null) {
7674
this.source = emptyMap();
77-
this.sourceContentType = null;
7875
} else {
7976
Tuple<XContentType, Map<String, Object>> tuple = sourceAsMapAndType(source);
80-
this.sourceContentType = tuple.v1();
8177
this.source = tuple.v2();
8278
}
8379
} catch (Exception e) {
84-
throw new ElasticsearchParseException("failed to parse / load source", e);
80+
throw new ElasticsearchParseException("failed to parse source", e);
8581
}
8682
return this.source;
8783
}
8884

85+
private BytesReference sourceAsBytes() {
86+
if (sourceAsBytes != null) {
87+
return sourceAsBytes;
88+
}
89+
90+
try {
91+
FieldsVisitor sourceFieldVisitor = new FieldsVisitor(true);
92+
fieldReader.accept(docId, sourceFieldVisitor);
93+
sourceAsBytes = sourceFieldVisitor.source();
94+
if (sourceContentType != null) {
95+
sourceContentType = XContentFactory.xContentType(sourceAsBytes.streamInput());
96+
}
97+
} catch (Exception e) {
98+
throw new ElasticsearchParseException("failed to load source", e);
99+
}
100+
101+
return sourceAsBytes;
102+
}
103+
89104
private static Tuple<XContentType, Map<String, Object>> sourceAsMapAndType(BytesReference source) throws ElasticsearchParseException {
90105
return XContentHelper.convertToMap(source, false);
91106
}
@@ -204,8 +219,9 @@ public Object extractValue(String path, @Nullable Object nullValue) {
204219
return XContentMapValues.extractValue(path, source(), nullValue);
205220
}
206221

207-
public Object filter(FetchSourceContext context) {
208-
return context.getFilter().apply(source());
222+
public BytesReference filter(FetchSourceContext context) {
223+
BytesReference source = sourceAsBytes();
224+
return context.getFilter(sourceContentType).apply(source);
209225
}
210226

211227
@Override

0 commit comments

Comments
 (0)