Skip to content

add plugin mechanism for handling source storage and retrieval - index level #1815

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

Closed
imotov wants to merge 1 commit into from
Closed

add plugin mechanism for handling source storage and retrieval - index level #1815

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.get.GetField;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.mapper.MapperService;
import org.elasticsearch.index.mapper.internal.ParentFieldMapper;
import org.elasticsearch.index.mapper.internal.RoutingFieldMapper;
import org.elasticsearch.index.mapper.internal.SourceFieldMapper;
Expand Down Expand Up @@ -158,6 +159,7 @@ protected void shardOperation(final UpdateRequest request, final ActionListener<
protected void shardOperation(final UpdateRequest request, final ActionListener<UpdateResponse> listener, final int retryCount) throws ElasticSearchException {
IndexService indexService = indicesService.indexServiceSafe(request.index());
IndexShard indexShard = indexService.shardSafe(request.shardId());
final MapperService mapperService = indexService.mapperService();

long getDate = System.currentTimeMillis();
final GetResult getResult = indexShard.getService().get(request.type(), request.id(),
Expand Down Expand Up @@ -231,7 +233,7 @@ protected void shardOperation(final UpdateRequest request, final ActionListener<
public void onResponse(IndexResponse response) {
UpdateResponse update = new UpdateResponse(response.index(), response.type(), response.id(), response.version());
update.matches(response.matches());
update.getResult(extractGetResult(request, response.version(), updatedSourceAsMap, updateSourceContentType, updateSourceBytes));
update.getResult(extractGetResult(request, response.version(), updatedSourceAsMap, updateSourceContentType, updateSourceBytes, mapperService));
listener.onResponse(update);
}

Expand Down Expand Up @@ -260,7 +262,7 @@ public void run() {
@Override
public void onResponse(DeleteResponse response) {
UpdateResponse update = new UpdateResponse(response.index(), response.type(), response.id(), response.version());
update.getResult(extractGetResult(request, response.version(), updatedSourceAsMap, updateSourceContentType, null));
update.getResult(extractGetResult(request, response.version(), updatedSourceAsMap, updateSourceContentType, null, mapperService));
listener.onResponse(update);
}

Expand All @@ -283,7 +285,7 @@ public void run() {
});
} else if ("none".equals(operation)) {
UpdateResponse update = new UpdateResponse(getResult.index(), getResult.type(), getResult.id(), getResult.version());
update.getResult(extractGetResult(request, getResult.version(), updatedSourceAsMap, updateSourceContentType, null));
update.getResult(extractGetResult(request, getResult.version(), updatedSourceAsMap, updateSourceContentType, null, mapperService));
listener.onResponse(update);
} else {
logger.warn("Used update operation [{}] for script [{}], doing nothing...", operation, request.script);
Expand All @@ -292,14 +294,14 @@ public void run() {
}

@Nullable
protected GetResult extractGetResult(final UpdateRequest request, long version, final Map<String, Object> source, XContentType sourceContentType, @Nullable final BytesHolder sourceAsBytes) {
protected GetResult extractGetResult(final UpdateRequest request, long version, final Map<String, Object> source, XContentType sourceContentType, @Nullable final BytesHolder sourceAsBytes, MapperService mapperService) {
if (request.fields() == null || request.fields().length == 0) {
return null;
}
boolean sourceRequested = false;
Map<String, GetField> fields = null;
if (request.fields() != null && request.fields().length > 0) {
SourceLookup sourceLookup = new SourceLookup();
SourceLookup sourceLookup = new SourceLookup(mapperService);
sourceLookup.setNextSource(source);
for (String field : request.fields()) {
if (field.equals("_source")) {
Expand Down
18 changes: 12 additions & 6 deletions src/main/java/org/elasticsearch/index/get/ShardGetService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ public GetResult innerGet(String type, String id, String[] gFields, boolean real
// break between having loaded it from translog (so we only have _source), and having a document to load
if (get.docIdAndVersion() != null) {
Map<String, GetField> fields = null;
byte[] source = null;
BytesHolder source = null;
UidField.DocIdAndVersion docIdAndVersion = get.docIdAndVersion();
ResetFieldSelector fieldSelector = buildFieldSelectors(docMapper, gFields);
if (fieldSelector != null) {
Expand All @@ -151,7 +151,7 @@ public GetResult innerGet(String type, String id, String[] gFields, boolean real
} catch (IOException e) {
throw new ElasticSearchException("Failed to get type [" + type + "] and id [" + id + "]", e);
}
source = extractSource(doc, docMapper);
source = extractSource(type, id, doc, docMapper);

for (Object oField : doc.getFields()) {
Fieldable field = (Fieldable) oField;
Expand Down Expand Up @@ -198,6 +198,9 @@ public GetResult innerGet(String type, String id, String[] gFields, boolean real
SearchScript searchScript = scriptService.search(searchLookup, "mvel", field, null);
searchScript.setNextReader(docIdAndVersion.reader);
searchScript.setNextDocId(docIdAndVersion.docId);
if(source != null) {
searchLookup.source().setNextSource(source.bytes(), source.offset(), source.length());
}

try {
value = searchScript.run();
Expand All @@ -214,6 +217,9 @@ public GetResult innerGet(String type, String id, String[] gFields, boolean real
searchLookup = new SearchLookup(mapperService, indexCache.fieldData());
searchLookup.setNextReader(docIdAndVersion.reader);
searchLookup.setNextDocId(docIdAndVersion.docId);
if(source != null) {
searchLookup.source().setNextSource(source.bytes(), source.offset(), source.length());
}
}
value = searchLookup.source().extractValue(field);
}
Expand All @@ -233,7 +239,7 @@ public GetResult innerGet(String type, String id, String[] gFields, boolean real
}
}

return new GetResult(shardId.index().name(), type, id, get.version(), get.exists(), source == null ? null : new BytesHolder(source), fields);
return new GetResult(shardId.index().name(), type, id, get.version(), get.exists(), source, fields);
} else {
Translog.Source source = get.source();

Expand Down Expand Up @@ -355,11 +361,11 @@ private static ResetFieldSelector buildFieldSelectors(DocumentMapper docMapper,
return fieldSelector;
}

private static byte[] extractSource(Document doc, DocumentMapper documentMapper) {
byte[] source = null;
private static BytesHolder extractSource(String type, String id, Document doc, DocumentMapper documentMapper) {
BytesHolder source = null;
Fieldable sourceField = doc.getFieldable(documentMapper.sourceMapper().names().indexName());
if (sourceField != null) {
source = documentMapper.sourceMapper().nativeValue(sourceField);
source = documentMapper.sourceMapper().extractSource(type, id, sourceField);
doc.removeField(documentMapper.sourceMapper().names().indexName());
}
return source;
Expand Down
15 changes: 10 additions & 5 deletions src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.elasticsearch.index.mapper.object.ObjectMapper;
import org.elasticsearch.index.mapper.object.RootObjectMapper;
import org.elasticsearch.index.settings.IndexSettings;
import org.elasticsearch.index.source.ExternalSourceProviderService;

import java.io.IOException;
import java.util.Map;
Expand All @@ -59,16 +60,20 @@ public class DocumentMapperParser extends AbstractIndexComponent {

private final Object typeParsersMutex = new Object();

final ExternalSourceProviderService externalSourceProviderService;

private volatile ImmutableMap<String, Mapper.TypeParser> typeParsers;
private volatile ImmutableMap<String, Mapper.TypeParser> rootTypeParsers;

public DocumentMapperParser(Index index, AnalysisService analysisService) {
this(index, ImmutableSettings.Builder.EMPTY_SETTINGS, analysisService);
public DocumentMapperParser(Index index, AnalysisService analysisService, ExternalSourceProviderService externalSourceProviderService) {
this(index, ImmutableSettings.Builder.EMPTY_SETTINGS, analysisService, externalSourceProviderService);
}

public DocumentMapperParser(Index index, @IndexSettings Settings indexSettings, AnalysisService analysisService) {
public DocumentMapperParser(Index index, @IndexSettings Settings indexSettings, AnalysisService analysisService,
ExternalSourceProviderService externalSourceProviderService) {
super(index, indexSettings);
this.analysisService = analysisService;
this.externalSourceProviderService = externalSourceProviderService;
typeParsers = new MapBuilder<String, Mapper.TypeParser>()
.put(ByteFieldMapper.CONTENT_TYPE, new ByteFieldMapper.TypeParser())
.put(ShortFieldMapper.CONTENT_TYPE, new ShortFieldMapper.TypeParser())
Expand Down Expand Up @@ -123,7 +128,7 @@ public void putRootTypeParser(String type, Mapper.TypeParser typeParser) {
}

public Mapper.TypeParser.ParserContext parserContext() {
return new Mapper.TypeParser.ParserContext(analysisService, typeParsers);
return new Mapper.TypeParser.ParserContext(analysisService, typeParsers, externalSourceProviderService);
}

public DocumentMapper parse(String source) throws MapperParsingException {
Expand Down Expand Up @@ -157,7 +162,7 @@ public DocumentMapper parse(@Nullable String type, String source, String default
}
}

Mapper.TypeParser.ParserContext parserContext = new Mapper.TypeParser.ParserContext(analysisService, typeParsers);
Mapper.TypeParser.ParserContext parserContext = new Mapper.TypeParser.ParserContext(analysisService, typeParsers, externalSourceProviderService);

DocumentMapper.Builder docBuilder = doc(index.name(), indexSettings, (RootObjectMapper.Builder) rootObjectTypeParser.parse(type, mapping, parserContext));

Expand Down
10 changes: 9 additions & 1 deletion src/main/java/org/elasticsearch/index/mapper/Mapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.index.analysis.AnalysisService;
import org.elasticsearch.index.source.ExternalSourceProviderService;

import java.io.IOException;
import java.util.Map;
Expand Down Expand Up @@ -80,9 +81,12 @@ public static class ParserContext {

private final ImmutableMap<String, TypeParser> typeParsers;

public ParserContext(AnalysisService analysisService, ImmutableMap<String, TypeParser> typeParsers) {
private final ExternalSourceProviderService externalSourceProviderService;

public ParserContext(AnalysisService analysisService, ImmutableMap<String, TypeParser> typeParsers, ExternalSourceProviderService externalSourceProviderService) {
this.analysisService = analysisService;
this.typeParsers = typeParsers;
this.externalSourceProviderService = externalSourceProviderService;
}

public AnalysisService analysisService() {
Expand All @@ -92,6 +96,10 @@ public AnalysisService analysisService() {
public TypeParser typeParser(String type) {
return typeParsers.get(Strings.toUnderscoreCase(type));
}

public ExternalSourceProviderService sourceProviderService() {
return externalSourceProviderService;
}
}

Mapper.Builder parse(String name, Map<String, Object> node, ParserContext parserContext) throws MapperParsingException;
Expand Down
6 changes: 4 additions & 2 deletions src/main/java/org/elasticsearch/index/mapper/MapperService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
import org.elasticsearch.index.mapper.object.ObjectMapper;
import org.elasticsearch.index.search.nested.NonNestedDocsFilter;
import org.elasticsearch.index.settings.IndexSettings;
import org.elasticsearch.index.source.ExternalSourceProviderService;
import org.elasticsearch.indices.InvalidTypeNameException;
import org.elasticsearch.indices.TypeMissingException;

Expand Down Expand Up @@ -97,10 +98,11 @@ public class MapperService extends AbstractIndexComponent implements Iterable<Do
private final SmartIndexNameSearchAnalyzer searchAnalyzer;

@Inject
public MapperService(Index index, @IndexSettings Settings indexSettings, Environment environment, AnalysisService analysisService) {
public MapperService(Index index, @IndexSettings Settings indexSettings, Environment environment,
AnalysisService analysisService, ExternalSourceProviderService externalSourceProviderService) {
super(index, indexSettings);
this.analysisService = analysisService;
this.documentParser = new DocumentMapperParser(index, indexSettings, analysisService);
this.documentParser = new DocumentMapperParser(index, indexSettings, analysisService, externalSourceProviderService);
this.searchAnalyzer = new SmartIndexNameSearchAnalyzer(analysisService.defaultSearchAnalyzer());

this.dynamic = componentSettings.getAsBoolean("dynamic", true);
Expand Down
Loading