Merge pull request #13786 from s1monw/move_term_vector_to_node_level

Move ShardTermVectorService to be on indices level as TermVectorService

core/src/main/java/org/elasticsearch/action/termvectors/TransportShardMultiTermsVectorAction.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.action.termvectors;
 
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
@@ -81,7 +80,7 @@ protected MultiTermVectorsShardResponse shardOperation(MultiTermVectorsShardRequ
             try {
                 IndexService indexService = indicesService.indexServiceSafe(request.index());
                 IndexShard indexShard = indexService.shardSafe(shardId.id());
-                TermVectorsResponse termVectorsResponse = indexShard.termVectorsService().getTermVectors(termVectorsRequest, shardId.getIndex());
+                TermVectorsResponse termVectorsResponse = indexShard.getTermVectors(termVectorsRequest);
                 termVectorsResponse.updateTookInMillis(termVectorsRequest.startTime());
                 response.add(request.locations.get(i), termVectorsResponse);
             } catch (Throwable t) {

core/src/main/java/org/elasticsearch/action/termvectors/TransportTermVectorsAction.java

@@ -83,7 +83,7 @@ protected void resolveRequest(ClusterState state, InternalRequest request) {
     protected TermVectorsResponse shardOperation(TermVectorsRequest request, ShardId shardId) {
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
         IndexShard indexShard = indexService.shardSafe(shardId.id());
-        TermVectorsResponse response = indexShard.termVectorsService().getTermVectors(request, shardId.getIndex());
+        TermVectorsResponse response = indexShard.getTermVectors(request);
         response.updateTookInMillis(request.startTime());
         return response;
     }

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -32,6 +32,8 @@
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.admin.indices.optimize.OptimizeRequest;
 import org.elasticsearch.action.admin.indices.upgrade.post.UpgradeRequest;
+import org.elasticsearch.action.termvectors.TermVectorsRequest;
+import org.elasticsearch.action.termvectors.TermVectorsResponse;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -91,7 +93,7 @@
 import org.elasticsearch.index.store.StoreStats;
 import org.elasticsearch.index.suggest.stats.ShardSuggestMetric;
 import org.elasticsearch.index.suggest.stats.SuggestStats;
-import org.elasticsearch.index.termvectors.ShardTermVectorsService;
+import org.elasticsearch.index.termvectors.TermVectorsService;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.index.translog.TranslogConfig;
 import org.elasticsearch.index.translog.TranslogStats;
@@ -137,7 +139,7 @@ public class IndexShard extends AbstractIndexShardComponent {
     private final ShardFieldData shardFieldData;
     private final PercolatorQueriesRegistry percolatorQueriesRegistry;
     private final ShardPercolateService shardPercolateService;
-    private final ShardTermVectorsService termVectorsService;
+    private final TermVectorsService termVectorsService;
     private final IndexFieldDataService indexFieldDataService;
     private final IndexService indexService;
     private final ShardSuggestMetric shardSuggestMetric = new ShardSuggestMetric();
@@ -202,7 +204,7 @@ public class IndexShard extends AbstractIndexShardComponent {
     public IndexShard(ShardId shardId, IndexSettingsService indexSettingsService, IndicesLifecycle indicesLifecycle, Store store, StoreRecoveryService storeRecoveryService,
                       ThreadPool threadPool, MapperService mapperService, IndexQueryParserService queryParserService, IndexCache indexCache, IndexAliasesService indexAliasesService,
                       IndicesQueryCache indicesQueryCache, CodecService codecService,
-                      ShardTermVectorsService termVectorsService, IndexFieldDataService indexFieldDataService, IndexService indexService,
+                      TermVectorsService termVectorsService, IndexFieldDataService indexFieldDataService, IndexService indexService,
                       @Nullable IndicesWarmer warmer, SnapshotDeletionPolicy deletionPolicy, SimilarityService similarityService, EngineFactory factory,
                       ClusterService clusterService, ShardPath path, BigArrays bigArrays, IndexSearcherWrappingService wrappingService) {
         super(shardId, indexSettingsService.getSettings());
@@ -226,7 +228,7 @@ public IndexShard(ShardId shardId, IndexSettingsService indexSettingsService, In
         this.indexAliasesService = indexAliasesService;
         this.indexingService = new ShardIndexingService(shardId, indexSettings);
         this.getService = new ShardGetService(this, mapperService);
-        this.termVectorsService = termVectorsService.setIndexShard(this);
+        this.termVectorsService = termVectorsService;
         this.searchService = new ShardSearchStats(indexSettings);
         this.shardWarmerService = new ShardIndexWarmerService(shardId, indexSettings);
         this.indicesQueryCache = indicesQueryCache;
@@ -284,10 +286,6 @@ public ShardGetService getService() {
         return this.getService;
     }
 
-    public ShardTermVectorsService termVectorsService() {
-        return termVectorsService;
-    }
-
     public ShardSuggestMetric getSuggestMetric() {
         return shardSuggestMetric;
     }
@@ -636,6 +634,10 @@ public SegmentsStats segmentStats() {
         return segmentsStats;
     }
 
+    public TermVectorsResponse getTermVectors(TermVectorsRequest request) {
+        return this.termVectorsService.getTermVectors(this, request);
+    }
+
     public WarmerStats warmerStats() {
         return shardWarmerService.stats();
     }

core/src/main/java/org/elasticsearch/index/shard/IndexShardModule.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.index.engine.IndexSearcherWrappingService;
 import org.elasticsearch.index.engine.EngineFactory;
 import org.elasticsearch.index.engine.InternalEngineFactory;
-import org.elasticsearch.index.termvectors.ShardTermVectorsService;
 
 /**
  * The {@code IndexShardModule} module is responsible for binding the correct
@@ -68,7 +67,6 @@ protected void configure() {
 
         bind(EngineFactory.class).to(engineFactoryImpl);
         bind(StoreRecoveryService.class).asEagerSingleton();
-        bind(ShardTermVectorsService.class).asEagerSingleton();
         bind(IndexSearcherWrappingService.class).asEagerSingleton();
         // this injects an empty set in IndexSearcherWrappingService, otherwise guice can't construct IndexSearcherWrappingService
         Multibinder<IndexSearcherWrapper> multibinder

core/src/main/java/org/elasticsearch/index/shard/ShadowIndexShard.java

@@ -40,7 +40,7 @@
 import org.elasticsearch.index.settings.IndexSettingsService;
 import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.index.store.Store;
-import org.elasticsearch.index.termvectors.ShardTermVectorsService;
+import org.elasticsearch.index.termvectors.TermVectorsService;
 import org.elasticsearch.indices.IndicesLifecycle;
 import org.elasticsearch.indices.IndicesWarmer;
 import org.elasticsearch.indices.cache.query.IndicesQueryCache;
@@ -62,7 +62,7 @@ public ShadowIndexShard(ShardId shardId, IndexSettingsService indexSettingsServi
                             ThreadPool threadPool, MapperService mapperService,
                             IndexQueryParserService queryParserService, IndexCache indexCache,
                             IndexAliasesService indexAliasesService, IndicesQueryCache indicesQueryCache,
-                            CodecService codecService, ShardTermVectorsService termVectorsService, IndexFieldDataService indexFieldDataService,
+                            CodecService codecService, TermVectorsService termVectorsService, IndexFieldDataService indexFieldDataService,
                             IndexService indexService, @Nullable IndicesWarmer warmer,
                             SnapshotDeletionPolicy deletionPolicy, SimilarityService similarityService,
                             EngineFactory factory, ClusterService clusterService,

core/src/main/java/org/elasticsearch/index/termvectors/ShardTermVectorsService.java → core/src/main/java/org/elasticsearch/index/termvectors/TermVectorsService.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
@@ -42,10 +43,7 @@
 import org.elasticsearch.index.mapper.*;
 import org.elasticsearch.index.mapper.core.StringFieldMapper;
 import org.elasticsearch.index.mapper.internal.UidFieldMapper;
-import org.elasticsearch.index.settings.IndexSettings;
-import org.elasticsearch.index.shard.AbstractIndexShardComponent;
 import org.elasticsearch.index.shard.IndexShard;
-import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.dfs.AggregatedDfs;
 
 import java.io.IOException;
@@ -56,27 +54,20 @@
 /**
  */
 
-public class ShardTermVectorsService extends AbstractIndexShardComponent {
+public class TermVectorsService  {
 
-    private IndexShard indexShard;
     private final MappingUpdatedAction mappingUpdatedAction;
     private final TransportDfsOnlyAction dfsAction;
 
     @Inject
-    public ShardTermVectorsService(ShardId shardId, @IndexSettings Settings indexSettings, MappingUpdatedAction mappingUpdatedAction, TransportDfsOnlyAction dfsAction) {
-        super(shardId, indexSettings);
+    public TermVectorsService(MappingUpdatedAction mappingUpdatedAction, TransportDfsOnlyAction dfsAction) {
         this.mappingUpdatedAction = mappingUpdatedAction;
         this.dfsAction = dfsAction;
     }
 
-    // sadly, to overcome cyclic dep, we need to do this and inject it ourselves...
-    public ShardTermVectorsService setIndexShard(IndexShard indexShard) {
-        this.indexShard = indexShard;
-        return this;
-    }
 
-    public TermVectorsResponse getTermVectors(TermVectorsRequest request, String concreteIndex) {
-        final TermVectorsResponse termVectorsResponse = new TermVectorsResponse(concreteIndex, request.type(), request.id());
+    public TermVectorsResponse getTermVectors(IndexShard indexShard, TermVectorsRequest request) {
+        final TermVectorsResponse termVectorsResponse = new TermVectorsResponse(indexShard.shardId().index().name(), request.type(), request.id());
         final Term uidTerm = new Term(UidFieldMapper.NAME, Uid.createUidAsBytes(request.type(), request.id()));
 
         Engine.GetResult get = indexShard.get(new Engine.Get(request.realtime(), uidTerm).version(request.version()).versionType(request.versionType()));
@@ -94,7 +85,7 @@ public TermVectorsResponse getTermVectors(TermVectorsRequest request, String con
 
         /* handle potential wildcards in fields */
         if (request.selectedFields() != null) {
-            handleFieldWildcards(request);
+            handleFieldWildcards(indexShard, request);
         }
 
         final Engine.Searcher searcher = indexShard.acquireSearcher("term_vector");
@@ -103,7 +94,7 @@ public TermVectorsResponse getTermVectors(TermVectorsRequest request, String con
             Versions.DocIdAndVersion docIdAndVersion = get.docIdAndVersion();
             /* from an artificial document */
             if (request.doc() != null) {
-                termVectorsByField = generateTermVectorsFromDoc(request, !docFromTranslog);
+                termVectorsByField = generateTermVectorsFromDoc(indexShard, request, !docFromTranslog);
                 // if no document indexed in shard, take the queried document itself for stats
                 if (topLevelFields == null) {
                     topLevelFields = termVectorsByField;
@@ -122,7 +113,7 @@ else if (docIdAndVersion != null) {
                 }
                 // fields without term vectors
                 if (selectedFields != null) {
-                    termVectorsByField = addGeneratedTermVectors(get, termVectorsByField, request, selectedFields);
+                    termVectorsByField = addGeneratedTermVectors(indexShard, get, termVectorsByField, request, selectedFields);
                 }
                 termVectorsResponse.setDocVersion(docIdAndVersion.version);
                 termVectorsResponse.setExists(true);
@@ -158,7 +149,7 @@ else if (docIdAndVersion != null) {
         return termVectorsResponse;
     }
 
-    private void handleFieldWildcards(TermVectorsRequest request) {
+    private void handleFieldWildcards(IndexShard indexShard, TermVectorsRequest request) {
         Set<String> fieldNames = new HashSet<>();
         for (String pattern : request.selectedFields()) {
             fieldNames.addAll(indexShard.mapperService().simpleMatchToIndexNames(pattern));
@@ -178,7 +169,7 @@ private boolean isValidField(MappedFieldType fieldType) {
         return true;
     }
 
-    private Fields addGeneratedTermVectors(Engine.GetResult get, Fields termVectorsByField, TermVectorsRequest request, Set<String> selectedFields) throws IOException {
+    private Fields addGeneratedTermVectors(IndexShard indexShard, Engine.GetResult get, Fields termVectorsByField, TermVectorsRequest request, Set<String> selectedFields) throws IOException {
         /* only keep valid fields */
         Set<String> validFields = new HashSet<>();
         for (String field : selectedFields) {
@@ -201,7 +192,7 @@ private Fields addGeneratedTermVectors(Engine.GetResult get, Fields termVectorsB
         /* generate term vectors from fetched document fields */
         GetResult getResult = indexShard.getService().get(
                 get, request.id(), request.type(), validFields.toArray(Strings.EMPTY_ARRAY), null, false);
-        Fields generatedTermVectors = generateTermVectors(getResult.getFields().values(), request.offsets(), request.perFieldAnalyzer(), validFields);
+        Fields generatedTermVectors = generateTermVectors(indexShard, getResult.getFields().values(), request.offsets(), request.perFieldAnalyzer(), validFields);
 
         /* merge with existing Fields */
         if (termVectorsByField == null) {
@@ -211,7 +202,7 @@ private Fields addGeneratedTermVectors(Engine.GetResult get, Fields termVectorsB
         }
     }
 
-    private Analyzer getAnalyzerAtField(String field, @Nullable Map<String, String> perFieldAnalyzer) {
+    private Analyzer getAnalyzerAtField(IndexShard indexShard, String field, @Nullable Map<String, String> perFieldAnalyzer) {
         MapperService mapperService = indexShard.mapperService();
         Analyzer analyzer;
         if (perFieldAnalyzer != null && perFieldAnalyzer.containsKey(field)) {
@@ -235,7 +226,7 @@ private Set<String> getFieldsToGenerate(Map<String, String> perAnalyzerField, Fi
         return selectedFields;
     }
 
-    private Fields generateTermVectors(Collection<GetField> getFields, boolean withOffsets, @Nullable Map<String, String> perFieldAnalyzer, Set<String> fields)
+    private Fields generateTermVectors(IndexShard indexShard, Collection<GetField> getFields, boolean withOffsets, @Nullable Map<String, String> perFieldAnalyzer, Set<String> fields)
             throws IOException {
         /* store document in memory index */
         MemoryIndex index = new MemoryIndex(withOffsets);
@@ -245,7 +236,7 @@ private Fields generateTermVectors(Collection<GetField> getFields, boolean withO
                 // some fields are returned even when not asked for, eg. _timestamp
                 continue;
             }
-            Analyzer analyzer = getAnalyzerAtField(field, perFieldAnalyzer);
+            Analyzer analyzer = getAnalyzerAtField(indexShard, field, perFieldAnalyzer);
             for (Object text : getField.getValues()) {
                 index.addField(field, text.toString(), analyzer);
             }
@@ -254,9 +245,9 @@ private Fields generateTermVectors(Collection<GetField> getFields, boolean withO
         return MultiFields.getFields(index.createSearcher().getIndexReader());
     }
 
-    private Fields generateTermVectorsFromDoc(TermVectorsRequest request, boolean doAllFields) throws Throwable {
+    private Fields generateTermVectorsFromDoc(IndexShard indexShard, TermVectorsRequest request, boolean doAllFields) throws Throwable {
         // parse the document, at the moment we do update the mapping, just like percolate
-        ParsedDocument parsedDocument = parseDocument(indexShard.shardId().getIndex(), request.type(), request.doc());
+        ParsedDocument parsedDocument = parseDocument(indexShard, indexShard.shardId().getIndex(), request.type(), request.doc());
 
         // select the right fields and generate term vectors
         ParseContext.Document doc = parsedDocument.rootDoc();
@@ -282,10 +273,10 @@ private Fields generateTermVectorsFromDoc(TermVectorsRequest request, boolean do
             String[] values = doc.getValues(field.name());
             getFields.add(new GetField(field.name(), Arrays.asList((Object[]) values)));
         }
-        return generateTermVectors(getFields, request.offsets(), request.perFieldAnalyzer(), seenFields);
+        return generateTermVectors(indexShard, getFields, request.offsets(), request.perFieldAnalyzer(), seenFields);
     }
 
-    private ParsedDocument parseDocument(String index, String type, BytesReference doc) throws Throwable {
+    private ParsedDocument parseDocument(IndexShard indexShard, String index, String type, BytesReference doc) throws Throwable {
         MapperService mapperService = indexShard.mapperService();
 
         // TODO: make parsing not dynamically create fields not in the original mapping

core/src/main/java/org/elasticsearch/indices/IndicesModule.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.index.query.*;
 import org.elasticsearch.index.query.functionscore.FunctionScoreQueryParser;
 import org.elasticsearch.index.query.MoreLikeThisQueryParser;
+import org.elasticsearch.index.termvectors.TermVectorsService;
 import org.elasticsearch.indices.analysis.HunspellService;
 import org.elasticsearch.indices.analysis.IndicesAnalysisService;
 import org.elasticsearch.indices.cache.query.IndicesQueryCache;
@@ -147,6 +148,7 @@ protected void configure() {
         bind(UpdateHelper.class).asEagerSingleton();
         bind(MetaDataIndexUpgradeService.class).asEagerSingleton();
         bind(IndicesFieldDataCacheListener.class).asEagerSingleton();
+        bind(TermVectorsService.class).asEagerSingleton();
     }
 
     protected void bindQueryParsersExtension() {

core/src/test/java/org/elasticsearch/search/fetch/FetchSubPhasePluginIT.java

@@ -28,8 +28,10 @@
 import org.elasticsearch.action.termvectors.TermVectorsResponse;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.index.termvectors.TermVectorsService;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.search.SearchHitField;
 import org.elasticsearch.search.SearchModule;
@@ -166,7 +168,7 @@ public void hitExecute(SearchContext context, HitContext hitContext) {
                 hitField = new InternalSearchHitField(NAMES[0], new ArrayList<>(1));
                 hitContext.hit().fields().put(NAMES[0], hitField);
             }
-            TermVectorsResponse termVector = context.indexShard().termVectorsService().getTermVectors(new TermVectorsRequest(context.indexShard().indexService().index().getName(), hitContext.hit().type(), hitContext.hit().id()), context.indexShard().indexService().index().getName());
+            TermVectorsResponse termVector = context.indexShard().getTermVectors(new TermVectorsRequest(context.indexShard().indexService().index().getName(), hitContext.hit().type(), hitContext.hit().id()));
             try {
                 Map<String, Integer> tv = new HashMap<>();
                 TermsEnum terms = termVector.getFields().terms(field).iterator();